fix: send failed Kafka events to the DLQ and retry them in the group that failed

Merged
HA
13d ago
+296-164

Closes #288.

A handler that raised lost its event. Nothing has called DLQManager.handle_message since #207, and AckPolicy.ACK commits the offset either way. FastStream has no dead letter queue for Kafka, so this adds the intake as a router middleware and keeps the existing Mongo-backed DLQ and dlq-processor.

DeadLetterMiddleware sits on the k8s worker, result processor, saga orchestrator and notification routers. When a handler raises, it saves the event with its topic, consumer group, error and retry count, then re-raises, so logging and the telemetry error count stay as they were. The SSE bridge doesn't get it, since a late SSE push would show the browser a stale status.

A retry goes back to the original topic, and every group on that topic reads it. So the dlq-processor now publishes it with x-dlq-consumer-group and x-dlq-retry-count headers, and RetryTargetMiddleware on every broker makes the other groups skip it. A retry that fails again comes back with its count, and the existing policy reschedules or discards it.

Two other things had to change for retries to work. The idempotency middleware now releases its key when the handler raises instead of marking it failed, because a failed key turned the retry away as a duplicate (the HTTP path already releases). And DLQ rows are keyed by event id plus consumer group now, since two groups can fail on the same execution_completed and the old unique index on event id let the second overwrite the first.

The processor also marks a row retried before publishing, and puts it back to scheduled if the publish fails. Before, a retry that failed within milliseconds could save its new failure and then get overwritten by the late retried write.

I deleted the manual retry, batch retry and manual discard methods, their result models and set_retry_policy. Nothing has called them since the DLQ routes went away in #207, and keeping them meant threading the consumer group through dead code. That leaves dlq_processing_errors_total with nothing recording it while the dashboards still query it, which is the kind of drift #317 is about.

Deploy note: Beanie never drops indexes, so production keeps the old unique idx_dlq_event_id until someone runs db.dlq_messages.dropIndex("idx_dlq_event_id") once. Until then a second group failing on the same event can't save its row. The collection should be empty, since nothing has written to it since #207.

Checked with ruff, mypy strict, the orphan check, the unit tests (the new ones push events through TestKafkaBroker with the real middlewares, and cover the retry headers and the failed-publish rollback), the Grafana contract test, mkdocs build --strict, and the non-k8s e2e suite in docker.

Summary by cubic

Sends failed Kafka events to the Mongo-backed DLQ and retries them in the consumer group that failed, closing #288. Previously a handler that raised lost its event since AckPolicy.ACK commits the offset either way and nothing called DLQManager.handle_message.

  • DeadLetterMiddleware on the k8s worker, result processor, saga orchestrator and notification routers saves a failed event with its topic, consumer group, error and retry count, then re-raises so logging and the telemetry error count are unchanged. The SSE bridge gets no middleware, since a late SSE push would show stale status.
  • The dlq-processor republishes retries with x-dlq-consumer-group and x-dlq-retry-count headers, and RetryTargetMiddleware on every broker makes the other groups on the topic skip them. A retry that fails again comes back with its count.
  • The idempotency middleware now releases its key when a handler raises instead of marking it failed, so a retry isn't rejected as a duplicate.
  • DLQ rows are now keyed by event id plus consumer group, so two groups failing on the same event each keep their own row.
  • The processor marks a row retried before publishing and puts it back to scheduled if the publish fails, so a fast-failing retry isn't overwritten.
  • Removed the unused manual retry, batch retry and manual discard methods and set_retry_policy; dlq_processing_errors_total has no recorder left.

Migration

One manual step needed in production: Beanie never drops indexes, so the old unique idx_dlq_event_id on event.event_id stays and would block a second group from saving the same event. Run db.dlq_messages.dropIndex("idx_dlq_event_id") once. The collection should be empty, since nothing has written to it since #207.