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))