From 0d6b4f63ee63ba78b03cd0d8481200bcc06992d1 Mon Sep 17 00:00:00 2001 From: Alex Luong Date: Fri, 17 Jul 2026 13:22:21 +0700 Subject: [PATCH] fix(alert): suppress exhausted_retries per destination, not per event ALERT_EXHAUSTED_RETRIES_WINDOW_SECONDS documents "the first exhaustion per destination emits an alert and subsequent ones within the window are suppressed", but the suppression key included the event ID, so every distinct event exhausting on a destination alerted regardless of the window. Key is now tenant+destination (tenant-scoped for the same reason as #967: user-supplied destination IDs can collide across tenants). An in-flight conflict on the shared key (two events exhausting concurrently on one destination) now counts as suppressed instead of nacking, so redelivery doesn't duplicate the entry's attempt.failed. Key format change resets live windows on deploy: worst case one extra alert per destination. Co-Authored-By: Claude Fable 5 --- docs/content/features/operator-events.mdoc | 4 +- internal/logmq/batchprocessor.go | 31 +++++-- internal/logmq/delivery_suppression_test.go | 93 +++++++++++++++++---- 3 files changed, 99 insertions(+), 29 deletions(-) diff --git a/docs/content/features/operator-events.mdoc b/docs/content/features/operator-events.mdoc index d4dfe932..eb265c5d 100644 --- a/docs/content/features/operator-events.mdoc +++ b/docs/content/features/operator-events.mdoc @@ -30,7 +30,7 @@ Available topics: |-------|---------| | `alert.destination.consecutive_failure` | Consecutive failure count reaches 50%, 70%, 90%, or 100% of `ALERT_CONSECUTIVE_FAILURE_COUNT` | | `alert.destination.disabled` | Destination auto-disabled at 100% failure threshold | -| `alert.attempt.exhausted_retries` | Delivery exhausts all retry attempts (deduplicated per event+destination) | +| `alert.attempt.exhausted_retries` | Delivery exhausts all retry attempts (at most one alert per destination within the deduplication window) | | `attempt.success` | Every successful delivery attempt | | `attempt.failed` | Every failed delivery attempt, including retries | | `tenant.subscription.updated` | Destination created/updated/deleted and tenant topics or destination count changed | @@ -181,7 +181,7 @@ Emitted when a destination is auto-disabled after reaching the 100% threshold (o ### `alert.attempt.exhausted_retries` -Emitted when a delivery exhausts all retry attempts. Deduplicated per event+destination within `ALERT_EXHAUSTED_RETRIES_WINDOW_SECONDS`. +Emitted when a delivery exhausts all retry attempts. At most one alert per destination within `ALERT_EXHAUSTED_RETRIES_WINDOW_SECONDS`; the alert payload carries the first exhausted event in the window. Set the window to `0` to alert on every exhaustion. ```json { diff --git a/internal/logmq/batchprocessor.go b/internal/logmq/batchprocessor.go index b2f5c535..281c98c3 100644 --- a/internal/logmq/batchprocessor.go +++ b/internal/logmq/batchprocessor.go @@ -8,6 +8,7 @@ import ( "time" "github.com/hookdeck/outpost/internal/alert" + "github.com/hookdeck/outpost/internal/idempotence" "github.com/hookdeck/outpost/internal/logging" "github.com/hookdeck/outpost/internal/models" "github.com/hookdeck/outpost/internal/mqs" @@ -85,9 +86,10 @@ type AlertPipeline struct { // processed failed attempt is skipped instead of re-counting/re-alerting. // Required. ProcessedIdemp ReplayGate - // ExhaustedIdemp is the per-(event,destination) suppression window for - // exhausted-retries alerts. Nil means no suppression (alert on every - // exhaustion). + // ExhaustedIdemp is the per-(tenant,destination) suppression window for + // exhausted-retries alerts: at most one alert per destination within the + // window, regardless of which events exhaust. Nil means no suppression + // (alert on every exhaustion). ExhaustedIdemp SuppressionWindow } @@ -482,7 +484,7 @@ func (bp *BatchProcessor) plan(ctx context.Context, eval alert.Evaluation, entry event: opevents.ExhaustedRetriesEvent(dest, entry.Event, entry.Attempt), } if bp.alerts.ExhaustedIdemp != nil { - de.suppressKey = exhaustedRetriesKey(entry.Event.ID, dest.ID) + de.suppressKey = exhaustedRetriesKey(dest.TenantID, dest.ID) } events = append(events, de) } @@ -498,6 +500,11 @@ func (bp *BatchProcessor) plan(ctx context.Context, eval alert.Evaluation, entry // one. A suppressed duplicate (Exec skips the emit) counts as delivered. The // emitter owns the delivery audit log — it fires iff an event actually went // out, so filtered topics and suppressed duplicates leave no line. +// +// An in-flight conflict (another exhaustion on the same destination is +// emitting concurrently) also counts as delivered: nacking would re-emit the +// entry's other events (attempt.failed) on redelivery, and the window's +// contract is one alert per destination anyway. func (bp *BatchProcessor) send(ctx context.Context, de deliveryEvent) error { emit := func(ctx context.Context) error { return bp.alerts.Emitter.Emit(ctx, de.event) @@ -505,7 +512,11 @@ func (bp *BatchProcessor) send(ctx context.Context, de deliveryEvent) error { if de.suppressKey == "" { return emit(ctx) } - return bp.alerts.ExhaustedIdemp.Exec(ctx, de.suppressKey, emit) + err := bp.alerts.ExhaustedIdemp.Exec(ctx, de.suppressKey, emit) + if errors.Is(err, idempotence.ErrConflict) { + return nil + } + return err } // nackAlertFailure logs an alert-pipeline failure and nacks. InsertMany is @@ -527,8 +538,10 @@ func processedKey(attemptID string) string { return "logmq:processed:" + attemptID } -// exhaustedRetriesKey is the per-(event,destination) suppression key for -// exhausted-retries alerts. Format is stable — changing it resets live windows. -func exhaustedRetriesKey(eventID, destinationID string) string { - return "opevents:exhausted:" + eventID + ":" + destinationID +// exhaustedRetriesKey is the per-(tenant,destination) suppression key for +// exhausted-retries alerts. Tenant scoping prevents cross-tenant collisions on +// user-supplied destination IDs. Format is stable — changing it resets live +// windows. +func exhaustedRetriesKey(tenantID, destinationID string) string { + return "opevents:exhausted:" + tenantID + ":" + destinationID } diff --git a/internal/logmq/delivery_suppression_test.go b/internal/logmq/delivery_suppression_test.go index 62fe960c..2235964b 100644 --- a/internal/logmq/delivery_suppression_test.go +++ b/internal/logmq/delivery_suppression_test.go @@ -16,7 +16,7 @@ import ( ) // makeExhaustedEntry builds a failed attempt past the retry limit for a fixed -// event+destination, so its exhausted-retries suppression key is deterministic. +// tenant+destination, so its exhausted-retries suppression key is deterministic. // cf alerting is kept quiet by the callers' high thresholds/auto-disable count. func makeExhaustedEntry(destID, tenantID, eventID, attemptID string, attemptNumber int) models.LogEntry { event := testutil.EventFactory.Any( @@ -46,8 +46,9 @@ func exhaustedAlertConfig() alertConfig { return alertConfig{thresholds: []int{100}, autoDisableCount: 100, retryMaxLimit: 3} } -// Two exhaustions of the same event+destination within the window → only the -// first delivers; the second is suppressed by the idempotence key. +// Two exhaustions on the same destination within the window → only the first +// delivers; the second is suppressed by the idempotence key. Same event here +// (the replay case); distinct events are covered by PerDestination. func TestDelivery_ExhaustedRetries_WindowSuppression(t *testing.T) { t.Parallel() idemp := idempotence.New(testutil.CreateTestRedisClient(t), idempotence.WithSuccessfulTTL(10*time.Second)) @@ -71,12 +72,11 @@ func TestDelivery_ExhaustedRetries_WindowSuppression(t *testing.T) { cm1.requireAcked(t) cm2.requireAcked(t) assert.ElementsMatch(t, []string{topicFailed, topicFailed, topicExhaust}, topics(h.sink.forDest(dest)), - "second exhaustion of the same event+destination is suppressed within the window") + "second exhaustion on the same destination is suppressed within the window") } -// With no suppression window (idemp nil == WindowSeconds 0), every exhaustion of -// the same event+destination delivers — the key is present but there's no -// idempotence to enforce it. +// With no suppression window (idemp nil == WindowSeconds 0), every exhaustion +// delivers — the key is present but there's no idempotence to enforce it. func TestDelivery_ExhaustedRetries_NoWindowEmitsEvery(t *testing.T) { t.Parallel() h := newHarness(t, harnessConfig{ @@ -95,35 +95,92 @@ func TestDelivery_ExhaustedRetries_NoWindowEmitsEvery(t *testing.T) { cm1.requireAcked(t) cm2.requireAcked(t) assert.ElementsMatch(t, []string{topicFailed, topicFailed, topicExhaust, topicExhaust}, topics(h.sink.forDest(dest)), - "with no window, every exhaustion of the same event+destination delivers") + "with no window, every exhaustion delivers") } -// Distinct events exhausting on the same destination carry distinct keys → each -// delivers its own alert. -func TestDelivery_ExhaustedRetries_PerEvent(t *testing.T) { +// Distinct events exhausting on the same destination share the window key → +// at most one alert per destination within the window. +func TestDelivery_ExhaustedRetries_PerDestination(t *testing.T) { t.Parallel() idemp := idempotence.New(testutil.CreateTestRedisClient(t), idempotence.WithSuccessfulTTL(10*time.Second)) + h := newHarness(t, harnessConfig{ + batcher: batcherConfig{itemCount: 1}, + alert: exhaustedAlertConfig(), + doubles: doublesConfig{idemp: idemp}, + }) + + // Paced one at a time: concurrent Execs on the shared window key would hit + // the in-flight conflict path instead of the suppression this test pins. + dest, tenant := "dest_pd", "tenant_pd" + cm1, msg1 := newCountingMessage(makeExhaustedEntry(dest, tenant, "evt_pd_1", "att_pd_1", 4)) + cm2, msg2 := newCountingMessage(makeExhaustedEntry(dest, tenant, "evt_pd_2", "att_pd_2", 4)) + h.add(msg1) + h.waitTerminal([]*countingMessage{cm1}) + h.add(msg2) + h.waitTerminal([]*countingMessage{cm2}) + + cm1.requireAcked(t) + cm2.requireAcked(t) + assert.ElementsMatch(t, []string{topicFailed, topicFailed, topicExhaust}, topics(h.sink.forDest(dest)), + "a second event exhausting on the same destination within the window is suppressed") +} + +// The window key is tenant-scoped: the same destination ID under two tenants +// gets independent windows, so one tenant's alert never suppresses another's. +func TestDelivery_ExhaustedRetries_TenantIsolation(t *testing.T) { + t.Parallel() + idemp := idempotence.New(testutil.CreateTestRedisClient(t), idempotence.WithSuccessfulTTL(10*time.Second)) + h := newHarness(t, harnessConfig{ + batcher: batcherConfig{itemCount: 1}, + alert: exhaustedAlertConfig(), + doubles: doublesConfig{idemp: idemp}, + }) + + dest := "dest_ti" + cm1, msg1 := newCountingMessage(makeExhaustedEntry(dest, "tenant_ti_a", "evt_ti_a", "att_ti_a", 4)) + cm2, msg2 := newCountingMessage(makeExhaustedEntry(dest, "tenant_ti_b", "evt_ti_b", "att_ti_b", 4)) + h.add(msg1) + h.waitTerminal([]*countingMessage{cm1}) + h.add(msg2) + h.waitTerminal([]*countingMessage{cm2}) + + cm1.requireAcked(t) + cm2.requireAcked(t) + assert.ElementsMatch(t, []string{topicFailed, topicFailed, topicExhaust, topicExhaust}, topics(h.sink.forDest(dest)), + "tenants with the same destination ID alert independently") +} + +// Two events exhausting concurrently on one destination race on the shared +// window key: exactly one alert delivers, and the loser — whether it lands on +// the suppressed path or the in-flight conflict path — acks instead of nacking. +func TestDelivery_ExhaustedRetries_ConcurrentConflictAcks(t *testing.T) { + t.Parallel() + idemp := idempotence.New(testutil.CreateTestRedisClient(t), + idempotence.WithSuccessfulTTL(10*time.Second), + // Short conflict wait so the losing Exec resolves quickly. + idempotence.WithTimeout(200*time.Millisecond), + ) h := newHarness(t, harnessConfig{ batcher: batcherConfig{itemCount: 2}, alert: exhaustedAlertConfig(), doubles: doublesConfig{idemp: idemp}, }) - dest, tenant := "dest_pe", "tenant_pe" - cm1, msg1 := newCountingMessage(makeExhaustedEntry(dest, tenant, "evt_pe_1", "att_pe_1", 4)) - cm2, msg2 := newCountingMessage(makeExhaustedEntry(dest, tenant, "evt_pe_2", "att_pe_2", 4)) + dest, tenant := "dest_cc", "tenant_cc" + cm1, msg1 := newCountingMessage(makeExhaustedEntry(dest, tenant, "evt_cc_1", "att_cc_1", 4)) + cm2, msg2 := newCountingMessage(makeExhaustedEntry(dest, tenant, "evt_cc_2", "att_cc_2", 4)) h.add(msg1) h.add(msg2) h.waitTerminal([]*countingMessage{cm1, cm2}) cm1.requireAcked(t) cm2.requireAcked(t) - assert.ElementsMatch(t, []string{topicFailed, topicFailed, topicExhaust, topicExhaust}, topics(h.sink.forDest(dest)), - "each distinct event exhausting retries delivers its own alert") + assert.ElementsMatch(t, []string{topicFailed, topicFailed, topicExhaust}, topics(h.sink.forDest(dest)), + "concurrent exhaustions on one destination deliver exactly one alert") } -// Emit failure clears the window key, so a later exhaustion of the same -// event+destination re-delivers instead of being suppressed. +// Emit failure clears the window key, so a later exhaustion on the same +// destination re-delivers instead of being suppressed. func TestDelivery_ExhaustedRetries_EmitFailureClearsWindow(t *testing.T) { t.Parallel() idemp := idempotence.New(testutil.CreateTestRedisClient(t), idempotence.WithSuccessfulTTL(10*time.Second))