Keep failed outbox events retryable
This commit is contained in:
parent
a0f39f58bd
commit
0b6a57dc6c
3 changed files with 23 additions and 1 deletions
|
|
@ -419,7 +419,8 @@ class PostgresUserEngineStore:
|
||||||
"""
|
"""
|
||||||
SELECT payload
|
SELECT payload
|
||||||
FROM user_engine_outbox_events
|
FROM user_engine_outbox_events
|
||||||
WHERE claimed_at IS NULL AND delivered_at IS NULL AND failed_at IS NULL
|
WHERE delivered_at IS NULL
|
||||||
|
AND COALESCE(payload->>'dead_lettered_at', '') = ''
|
||||||
ORDER BY occurred_at, event_id
|
ORDER BY occurred_at, event_id
|
||||||
"""
|
"""
|
||||||
)
|
)
|
||||||
|
|
|
||||||
|
|
@ -24,6 +24,17 @@ class PostgresStoreAdapterTests(unittest.TestCase):
|
||||||
self.assertFalse(store.ready)
|
self.assertFalse(store.ready)
|
||||||
self.assertIsNone(store.schema_version)
|
self.assertIsNone(store.schema_version)
|
||||||
|
|
||||||
|
def test_pending_query_keeps_failed_non_dead_letter_events_retryable(self):
|
||||||
|
connection = _FakePostgresConnection()
|
||||||
|
store = PostgresUserEngineStore(connection)
|
||||||
|
|
||||||
|
store.pending_outbox()
|
||||||
|
|
||||||
|
normalized = " ".join(connection.last_sql.lower().split())
|
||||||
|
self.assertIn("delivered_at is null", normalized)
|
||||||
|
self.assertIn("payload->>'dead_lettered_at'", normalized)
|
||||||
|
self.assertNotIn("failed_at is null", normalized)
|
||||||
|
|
||||||
|
|
||||||
class _FakePostgresConnection:
|
class _FakePostgresConnection:
|
||||||
def __init__(self) -> None:
|
def __init__(self) -> None:
|
||||||
|
|
@ -37,6 +48,7 @@ class _FakePostgresConnection:
|
||||||
list[dict[str, Any]],
|
list[dict[str, Any]],
|
||||||
list[dict[str, Any]],
|
list[dict[str, Any]],
|
||||||
] | None = None
|
] | None = None
|
||||||
|
self.last_sql = ""
|
||||||
|
|
||||||
def cursor(self) -> "_FakePostgresCursor":
|
def cursor(self) -> "_FakePostgresCursor":
|
||||||
return _FakePostgresCursor(self)
|
return _FakePostgresCursor(self)
|
||||||
|
|
@ -70,6 +82,7 @@ class _FakePostgresCursor:
|
||||||
self._rows: list[Any] = []
|
self._rows: list[Any] = []
|
||||||
|
|
||||||
def execute(self, sql: str, params: Iterable[Any] | None = None) -> None:
|
def execute(self, sql: str, params: Iterable[Any] | None = None) -> None:
|
||||||
|
self.connection.last_sql = sql
|
||||||
normalized = " ".join(sql.lower().split())
|
normalized = " ".join(sql.lower().split())
|
||||||
values = tuple(params or ())
|
values = tuple(params or ())
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -221,3 +221,11 @@ fails closed when its dedicated token is absent. Tests assert that the two
|
||||||
Authorization headers differ. The full suite passes 132 tests with three
|
Authorization headers differ. The full suite passes 132 tests with three
|
||||||
external-provider skips. Production rollout still waits on governed delivery
|
external-provider skips. Production rollout still waits on governed delivery
|
||||||
of both scoped credentials.
|
of both scoped credentials.
|
||||||
|
|
||||||
|
2026-08-14 live retry finding: the PostgreSQL adapter excluded every row with
|
||||||
|
`failed_at` set, while the in-memory contract correctly retains failures until
|
||||||
|
`dead_lettered_at`. A first receiver error therefore required manual replay
|
||||||
|
instead of receiving the documented bounded automatic retry. The query now
|
||||||
|
selects every undelivered, non-dead-letter event; regression coverage checks
|
||||||
|
that failed rows are not filtered out. The full suite passes 133 tests with
|
||||||
|
three external-provider skips.
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue