Retry Sink.Store with backoff and dead-letter on exhaustion, across the pipeline and every sink

[mauriceyap]

When an ingester repeatedly fails to write a message to its sink (events Redis, the lookout postgres database, or the scheduler postgres database), it currently either retries forever or drops the message. This PR gives every ingester a proper dead letter queue: failures are retried with jittered backoff up to a configurable limit, and if the message still can't be stored, it's serialized and published to a dead-letter topic for later inspection or replay, instead of being retried indefinitely or silently lost. It also fixes a related bug where certain lookup errors caused a whole batch to be dropped rather than retried.

This touches both the shared pipeline and every sink because the two are compile-coupled. Sink[T] gains a Serialize method that every implementation must provide before anything builds, so the pipeline change and the per-sink work land together in one PR.

Details

Sink[T] gains a Serialize(msg T) ([]byte, error) method, implemented by RedisEventStore, LookoutDb, and SchedulerDb (each renders its instructions as JSON for DLQ inspection/manual replay), and the pipeline and every sink land together so the tree never sits in a red, non-compiling intermediate state.

The pipeline's store loop now retries Store up to DeadLetterMaxAttempts times (falling back to a defaultDeadLetterMaxAttempts of 5 when unset, so an unconfigured PulsarConfig still retries rather than dead-lettering on the first failure) with jittered exponential backoff (via the new newBackOff()/backoff/v4 helper), and on exhaustion serializes the message and publishes it to the configured dead-letter topic instead of dropping it, then still ACKs so the consumer doesn't redeliver forever.

If DeadLetterTopic is unset, no dead-letter publisher is created at all and an exhausted message is left unacked for redelivery instead (same as the shutdown case), rather than attempting to publish to an empty topic. Two new metrics (RecordPulsarMessageStoreRetry, RecordPulsarMessageDeadLettered) track this. Removes the now-redundant ingest.WithRetry helper (superseded by util.RetryUntilSuccessOrExhausted from PR 1). Context-cancellation during retry is handled by leaving the message unacked for redelivery rather than dead-lettering it.

Each sink's own internal retry loop is removed in the same change. Retry-then-dead-letter policy now lives entirely in the shared IngestionPipeline ack-path, and each Store implementation is updated to attempt once and classify unretryable failures via util.NewNonRetryableError so the pipeline skips straight to dead-lettering instead of burning its retry budget. NewLookoutDb and NewSchedulerDb drop their now-unused backoff/retry-count constructor parameters; every call site is updated accordingly. SchedulerDb.Serialize includes a type-switch (serializeDbOperation) covering every concrete DbOperation, guarded by a coverage test so new operation types can't silently fall through to an empty payload.

filterEventsForTerminalJobs now propagates lookup errors instead of silently ingesting unfiltered updates. Under the old pipeline this still had a gap: an ordinary, non-cancellation lookup error returned before any update was written, then hit the pipeline's ordinary-error branch, which acked and dropped the whole batch. That gap is closed structurally here, not with an extra special case - every Store error (including this one) now goes through the retry-then-dead-letter path, so the batch is retried and, if the error persists, dead-lettered and preserved rather than acked and silently dropped.

[mergify]

Tick the box to add this pull request to the merge queue (same as @mergifyio queue).

• Queue this pull request

[datadog-armadaproject]

⚠️ Warnings

🚦 1 Pipeline job failed

CI | All jobs succeeded     

Useful? React with 👍 / 👎

_{This comment will be updated automatically if new data arrives.
🔗 Commit SHA: 7187716 | Docs | Datadog PR Page | Give us feedback!}

[greptile-apps]

Greptile Summary

The PR centralizes bounded sink retries and dead-letter handling in the shared ingestion pipeline, while adapting the Redis, Lookout, and scheduler sinks to single-attempt storage and serialization.

• Adds configurable retry exhaustion, DLQ publication, and retry/dead-letter metrics.
• Adds JSON serialization for each sink’s failed instruction batches.
• Removes sink-local retry loops and propagates Lookout terminal-job lookup errors.
• Updates constructors, call sites, and tests for the shared sink contract.

Confidence Score: 4/5

The PR is not yet safe to merge because persistent Pulsar acknowledgement failures can still block ingestion-pipeline shutdown indefinitely.

The acknowledgement path still retries AckID using a background context, so a persistent failure prevents the store goroutine from reaching wg.Done() and leaves Run blocked in wg.Wait().

Files Needing Attention: internal/common/ingest/ingestion_pipeline.go

Important Files Changed

Filename	Overview
internal/common/ingest/ingestion_pipeline.go	Centralizes store retries, DLQ publication, metrics, and acknowledgement behavior; the previously reported unbounded background acknowledgement retry remains.
internal/common/config/pulsar.go	Documents and validates the optional dead-letter retry-attempt configuration.
internal/eventingester/store/eventstore.go	Removes Redis-local retries, classifies non-retryable errors, and serializes failed batches for the DLQ.
internal/lookoutingester/lookoutdb/insertion.go	Moves Lookout retry responsibility to the pipeline, propagates terminal-job lookup failures, and adds DLQ serialization.
internal/scheduleringester/schedulerdb.go	Adapts scheduler database storage to pipeline-owned retry and adds serialized DLQ output.

Flowchart

%%{init: {'theme': 'neutral'}}%%
flowchart LR
  P[Pulsar message batch] --> S[Sink.Store]
  S -->|success| A[Acknowledge source messages]
  S -->|retryable failure| R[Backoff and retry]
  R --> S
  S -->|exhausted or non-retryable| D[Serialize and publish to DLQ]
  D -->|success| A
  D -->|cancelled or no DLQ| U[Leave unacknowledged for redelivery]

Loading

_{Reviews (10): Last reviewed commit: "rename" | Re-trigger Greptile}

[nikola-jokic]

Overall looks good, I have few comments but they are not blockers

[nikola-jokic]

Why not use:

select {
case <-ctx.Done():
  return ctx.Err()
case <-time.After(wait):

The sleep will just delay as you pointed out the shutdown, which doesn't make much sense to wait for the full duration just to return after it. Also, time.After no longer has a leak where the counters are not de-allocated, as of go 1.23

[nikola-jokic]

If I read this correctly, you can preallocate here with len(update)

[nikola-jokic]

Not too important: I'd suggest extracting the struct as private package level struct for easier tests later. It doesn't have to be exported, but it would avoid having tests defining:

var data struct {
  // redefined struct fields
}

It is unnecessary duplication when you can simply do:

type redisEventStoreBatchUpdateMessage struct {
   // fields
}