feat(alert): default/disable semantics for consecutive-failure & exhausted-retries alerts (#964)

* feat(alert): add Settings + enable gates for consecutive/exhausted alerts

Introduce alert.Settings (the resolved, operational alert config) plus two
monitor gates: WithConsecutiveFailureEnabled and WithExhaustedRetriesEnabled.

Both default to true, so behavior is unchanged until a caller opts out. When
consecutive-failure alerting is gated off the monitor neither tracks failures
nor auto-disables; when exhausted-retries is gated off it never emits, even
with retries enabled. Extracts the consecutive-failure path into a helper to
keep the replay/ordering semantics identical on the enabled path.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* feat(config): resolve alert config to alert.Settings with unset/empty/value rule

AlertConfig.ConsecutiveFailureCount and ExhaustedRetriesWindowSeconds become
*string so the parse layer can tell three states apart: unset uses the default
(100 / 3600), an empty string disables that alert dimension, and any other value
must parse to a non-negative integer.

AlertConfig.ToConfig resolves the raw values into the operational alert.Settings
(domain-owned, so nothing downstream imports config). Validate rejects malformed
values at startup. builder wires the resolved gates into the monitor and only
builds the exhausted-retries suppression window when enabled with a positive
window (0 = alert on every exhaustion).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* docs(config): document alert default/disable behavior in config reference

Update the Alerts section of the self-hosting config reference for the new
unset/empty/value rule: ALERT_CONSECUTIVE_FAILURE_COUNT defaults to 100 (empty
disables), and document the previously-undocumented
ALERT_EXHAUSTED_RETRIES_WINDOW_SECONDS (default 3600, empty disables, 0 = no
suppression).

Also correct stale entries: drop the removed ALERT_CALLBACK_URL, fix the
ALERT_AUTO_DISABLE_DESTINATION default (false, not true), and fix the YAML
example key (alert, not alerts).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* docs(openapi): describe alert behavior in ManagedConfig

Document the unset/empty/value behavior for ALERT_CONSECUTIVE_FAILURE_COUNT,
ALERT_EXHAUSTED_RETRIES_WINDOW_SECONDS and ALERT_AUTO_DISABLE_DESTINATION in the
ManagedConfig schema. Descriptions only — the properties are already typed as
string. SDKs are regenerated from this schema at release time.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(config): make empty-string alert disable work on the env-var surface

The *string representation had two problems found by manual QA:
1. caarlos0/env ignores a present-but-empty env var, so `ALERT_..._COUNT=`
   resolved to the default instead of disabling — the empty=off rule only
   worked via YAML, not env vars (the primary surface for the cloud product).
2. caarlos0/env crashes ("expected a pointer to a Struct") on any non-nil
   *string it walks, so setting these in a YAML config file would crash startup
   (env.Parse runs after the YAML load).

Replace *string with an OptionalString value type that implements both
TextUnmarshaler (bound by caarlos0/env as a scalar — no crash) and
yaml.Unmarshaler (so `key: ""` expresses the empty/off state). The one case
caarlos0/env cannot surface — a present-but-empty env var — is handled
explicitly via OSInterface.LookupEnv, which also gives env precedence over YAML.

Net: unset -> default, empty -> disabled, value -> value, identically on both
env and YAML, with env > yaml. Adds a full parse-path test covering the matrix
and precedence.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(e2e): use OptionalString for ConsecutiveFailureCount in regression test

Missed call site when migrating AlertConfig fields to OptionalString;
the raw int assignment broke the cmd/e2e build.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: alexluong

cmd/e2e/configs/basic.go

@@ -87,7 +87,7 @@ func Basic(t *testing.T, opts BasicOpts) config.Config {
 	c.LogBatchSize = 100
 	c.DeploymentID = opts.DeploymentID
 	c.Alert.AutoDisableDestination = true
-	c.Alert.ConsecutiveFailureCount = 20
+	c.Alert.ConsecutiveFailureCount = config.NewOptionalString("20")
 
 	// Setup cleanup
 	t.Cleanup(func() {

cmd/e2e/regressions_test.go

@@ -91,7 +91,7 @@ func TestE2E_Regression_AutoDisableWithoutCallbackURL(t *testing.T) {
 		LogStorage: configs.LogStorageTypePostgres,
 	})
 	cfg.Alert.AutoDisableDestination = true
-	cfg.Alert.ConsecutiveFailureCount = 20
+	cfg.Alert.ConsecutiveFailureCount = config.NewOptionalString("20")
 
 	require.NoError(t, cfg.Validate(config.Flags{}))
 	configs.ApplyMigrations(t, &cfg)

docs/apis/openapi.yaml

@@ -3094,10 +3094,25 @@ components:
       properties:
         ALERT_AUTO_DISABLE_DESTINATION:
           type: string
+          description: >-
+            If "true", automatically disables a destination once
+            ALERT_CONSECUTIVE_FAILURE_COUNT is reached. Has no effect when
+            consecutive-failure alerting is disabled.
         ALERT_CONSECUTIVE_FAILURE_COUNT:
           type: string
+          description: >-
+            Consecutive delivery failures before alerting on a destination (and
+            disabling it when ALERT_AUTO_DISABLE_DESTINATION is "true"). Omit for
+            the default of 100; set to an empty string to disable
+            consecutive-failure alerting entirely.
         ALERT_EXHAUSTED_RETRIES_WINDOW_SECONDS:
           type: string
+          description: >-
+            Suppression window in seconds for exhausted_retries alerts: the first
+            exhaustion per destination alerts and subsequent ones within the
+            window are suppressed ("0" = no suppression). Omit for the default of
+            3600; set to an empty string to disable exhausted_retries alerting
+            entirely.
         DELIVERY_TIMEOUT_SECONDS:
           type: string
         DESTINATIONS_AWS_KINESIS_METADATA_IN_PAYLOAD:

docs/content/self-hosting/configuration.mdoc

@@ -98,9 +98,9 @@ Choose one for event log persistence:
 
 | Variable | Default | Description |
 |----------|---------|-------------|
-| `ALERT_CALLBACK_URL` | — | URL to POST to when an alert fires |
-| `ALERT_CONSECUTIVE_FAILURE_COUNT` | `20` | Consecutive failures before alert triggers |
-| `ALERT_AUTO_DISABLE_DESTINATION` | `true` | Auto-disable destination when failure count reaches 100% |
+| `ALERT_CONSECUTIVE_FAILURE_COUNT` | `100` | Consecutive delivery failures before alerting on a destination (and disabling it when `ALERT_AUTO_DISABLE_DESTINATION` is `true`). Leave unset for the default of `100`; set to an empty string to disable consecutive-failure alerting entirely. |
+| `ALERT_AUTO_DISABLE_DESTINATION` | `false` | Auto-disable a destination once `ALERT_CONSECUTIVE_FAILURE_COUNT` is reached. Has no effect when consecutive-failure alerting is disabled. |
+| `ALERT_EXHAUSTED_RETRIES_WINDOW_SECONDS` | `3600` | Suppression window (seconds) for `exhausted_retries` alerts: the first exhaustion per destination alerts and subsequent ones within the window are suppressed (`0` = no suppression, alert on every exhaustion). Leave unset for the default of `3600`; set to an empty string to disable `exhausted_retries` alerting entirely. |
 
 ## Destinations
 
@@ -159,7 +159,7 @@ portal:
   organization_name: "Acme Corp"
   accent_color: "#6122E7"
 
-alerts:
-  callback_url: "https://yourapp.com/api/outpost-alerts"
+alert:
   consecutive_failure_count: 50
+  exhausted_retries_window_seconds: 3600
 ```

internal/alert/monitor.go

@@ -90,6 +90,23 @@ func WithExhaustedRetriesIdempotence(idemp idempotence.Idempotence) AlertOption
 	}
 }
 
+// WithConsecutiveFailureEnabled toggles consecutive-failure alerting. When set
+// to false the monitor never tracks or alerts on consecutive failures (and
+// therefore never auto-disables). Defaults to true.
+func WithConsecutiveFailureEnabled(enabled bool) AlertOption {
+	return func(m *alertMonitor) {
+		m.consecutiveFailureEnabled = enabled
+	}
+}
+
+// WithExhaustedRetriesEnabled toggles exhausted_retries alerting. When set to
+// false the monitor never emits exhausted_retries alerts. Defaults to true.
+func WithExhaustedRetriesEnabled(enabled bool) AlertOption {
+	return func(m *alertMonitor) {
+		m.exhaustedRetriesEnabled = enabled
+	}
+}
+
 // DeliveryAttempt represents a single delivery attempt
 type DeliveryAttempt struct {
 	Event       *models.Event
@@ -109,6 +126,9 @@ type alertMonitor struct {
 	alertThresholds         []int
 	retryMaxLimit           int
 	exhaustedRetryIdemp     idempotence.Idempotence
+
+	consecutiveFailureEnabled bool
+	exhaustedRetriesEnabled   bool
 }
 
 // NewAlertMonitor creates a new alert monitor. Emitter and retryMaxLimit are
@@ -119,10 +139,12 @@ func NewAlertMonitor(logger *logging.Logger, redisClient redis.Cmdable, emitter
 		panic("alert: NewAlertMonitor requires a non-nil emitter")
 	}
 	alertMonitor := &alertMonitor{
-		logger:          logger,
-		emitter:         emitter,
-		retryMaxLimit:   retryMaxLimit,
-		alertThresholds: []int{50, 70, 90, 100}, // default thresholds
+		logger:                    logger,
+		emitter:                   emitter,
+		retryMaxLimit:             retryMaxLimit,
+		alertThresholds:           []int{50, 70, 90, 100}, // default thresholds
+		consecutiveFailureEnabled: true,
+		exhaustedRetriesEnabled:   true,
 	}
 
 	for _, opt := range opts {
@@ -142,93 +164,32 @@ func NewAlertMonitor(logger *logging.Logger, redisClient redis.Cmdable, emitter
 
 func (m *alertMonitor) HandleAttempt(ctx context.Context, attempt DeliveryAttempt) error {
 	if attempt.Attempt.Status == models.AttemptStatusSuccess {
+		// Nothing is tracked when consecutive-failure alerting is disabled, so
+		// there is no count to reset.
+		if !m.consecutiveFailureEnabled {
+			return nil
+		}
 		return m.store.ResetConsecutiveFailureCount(ctx, attempt.Destination.TenantID, attempt.Destination.ID)
 	}
 
-	// Consecutive failure tracking
-	res, err := m.store.IncrementConsecutiveFailureCount(ctx, attempt.Destination.TenantID, attempt.Destination.ID, attempt.Attempt.ID)
-	if err != nil {
-		return fmt.Errorf("failed to get alert state: %w", err)
-	}
-
-	// Replayed attempt (MQ redelivery, producer re-publish) that already
-	// completed a full evaluation — skip re-emitting alerts. Attempts that
-	// were counted but never marked evaluated (eval failed mid-way and the
-	// message was nacked) fall through and re-run as recovery.
-	if !res.NewlyCounted && res.AlreadyEvaluated {
-		m.logger.Ctx(ctx).Debug("skipping replayed attempt: already evaluated",
-			zap.String("attempt_id", attempt.Attempt.ID),
-			zap.String("tenant_id", attempt.Destination.TenantID),
-			zap.String("destination_id", attempt.Destination.ID),
-		)
-		return nil
-	}
-
-	count := res.Count
-	level, shouldAlert := m.evaluator.ShouldAlert(count)
-	if shouldAlert {
-		// At 100% threshold, disable the destination and emit disabled alert.
-		// Both operations are idempotent on replay: DisableDestination is a no-op
-		// if already disabled, and consumers deduplicate events by ID.
-		if level == 100 && m.disabler != nil {
-			if err := m.disabler.DisableDestination(ctx, attempt.Destination.TenantID, attempt.Destination.ID); err != nil {
-				return fmt.Errorf("failed to disable destination: %w", err)
-			}
-
-			now := time.Now()
-			attempt.Destination.DisabledAt = &now
-
-			m.logger.Ctx(ctx).Audit("destination disabled",
-				zap.String("attempt_id", attempt.Attempt.ID),
-				zap.String("event_id", attempt.Event.ID),
-				zap.String("tenant_id", attempt.Destination.TenantID),
-				zap.String("destination_id", attempt.Destination.ID),
-				zap.String("destination_type", attempt.Destination.Type),
-			)
-
-			disabledData := DestinationDisabledData{
-				TenantID:    attempt.Destination.TenantID,
-				Destination: attempt.Destination,
-				DisabledAt:  now,
-				Reason:      "consecutive_failure",
-				Event:       attempt.Event,
-				Attempt:     attempt.Attempt,
-			}
-			if err := m.emitter.Emit(ctx, opevents.TopicAlertDestinationDisabled, attempt.Destination.TenantID, disabledData); err != nil {
-				return fmt.Errorf("failed to emit destination disabled alert: %w", err)
-			}
-		}
-
-		// Emit consecutive failure alert
-		cfData := ConsecutiveFailureData{
-			TenantID:    attempt.Destination.TenantID,
-			Event:       attempt.Event,
-			Attempt:     attempt.Attempt,
-			Destination: attempt.Destination,
-			ConsecutiveFailures: ConsecutiveFailures{
-				Current:   count,
-				Max:       m.autoDisableFailureCount,
-				Threshold: level,
-			},
+	if m.consecutiveFailureEnabled {
+		// A replayed attempt that already completed evaluation skips the rest of
+		// the pipeline (exhausted-retries check and the evaluated mark), matching
+		// the original single-pass behavior.
+		done, err := m.handleConsecutiveFailure(ctx, attempt)
+		if err != nil {
+			return err
 		}
-		if err := m.emitter.Emit(ctx, opevents.TopicAlertConsecutiveFailure, attempt.Destination.TenantID, cfData); err != nil {
-			return fmt.Errorf("failed to emit consecutive failure alert: %w", err)
+		if done {
+			return nil
 		}
-
-		m.logger.Ctx(ctx).Audit("alert sent",
-			zap.String("topic", opevents.TopicAlertConsecutiveFailure),
-			zap.String("attempt_id", attempt.Attempt.ID),
-			zap.String("event_id", attempt.Event.ID),
-			zap.String("tenant_id", attempt.Destination.TenantID),
-			zap.String("destination_id", attempt.Destination.ID),
-			zap.String("destination_type", attempt.Destination.Type),
-		)
 	}
 
 	// Exhausted retries check (independent of consecutive failure thresholds).
 	// Attempt is 1-indexed: with retryMaxLimit=10, attempt 11 is the final one.
-	// Skip if retryMaxLimit=0 (retries disabled — no exhausted state to report).
-	if m.retryMaxLimit > 0 && attempt.Event.EligibleForRetry && attempt.Attempt.AttemptNumber > m.retryMaxLimit {
+	// Skip if retryMaxLimit=0 (retries disabled — no exhausted state to report)
+	// or if exhausted-retries alerting is disabled.
+	if m.exhaustedRetriesEnabled && m.retryMaxLimit > 0 && attempt.Event.EligibleForRetry && attempt.Attempt.AttemptNumber > m.retryMaxLimit {
 		erData := ExhaustedRetriesData{
 			TenantID:    attempt.Destination.TenantID,
 			Event:       attempt.Event,
@@ -263,17 +224,110 @@ func (m *alertMonitor) HandleAttempt(ctx context.Context, attempt DeliveryAttemp
 		}
 	}
 
-	// Mark the attempt fully evaluated so replays skip re-emitting alerts.
+	// Mark the attempt fully evaluated so replays skip re-emitting alerts. Only
+	// relevant when consecutive-failure tracking ran — it is the consumer of the
+	// evaluated mark (the replay short-circuit above).
 	// Non-fatal: on failure the attempt simply re-evaluates on replay, which
 	// matches the previous behavior (emit/disable are idempotent-by-design).
-	if err := m.store.MarkAttemptEvaluated(ctx, attempt.Destination.TenantID, attempt.Destination.ID, attempt.Attempt.ID); err != nil {
-		m.logger.Ctx(ctx).Warn("failed to mark attempt evaluated",
-			zap.Error(err),
+	if m.consecutiveFailureEnabled {
+		if err := m.store.MarkAttemptEvaluated(ctx, attempt.Destination.TenantID, attempt.Destination.ID, attempt.Attempt.ID); err != nil {
+			m.logger.Ctx(ctx).Warn("failed to mark attempt evaluated",
+				zap.Error(err),
+				zap.String("attempt_id", attempt.Attempt.ID),
+				zap.String("tenant_id", attempt.Destination.TenantID),
+				zap.String("destination_id", attempt.Destination.ID),
+			)
+		}
+	}
+
+	return nil
+}
+
+// handleConsecutiveFailure runs consecutive-failure tracking, alerting and
+// auto-disable for a failed attempt. It returns done=true when the attempt is a
+// replay that already completed evaluation, signalling the caller to stop
+// processing (skip the exhausted-retries check and the evaluated mark).
+func (m *alertMonitor) handleConsecutiveFailure(ctx context.Context, attempt DeliveryAttempt) (done bool, err error) {
+	res, err := m.store.IncrementConsecutiveFailureCount(ctx, attempt.Destination.TenantID, attempt.Destination.ID, attempt.Attempt.ID)
+	if err != nil {
+		return false, fmt.Errorf("failed to get alert state: %w", err)
+	}
+
+	// Replayed attempt (MQ redelivery, producer re-publish) that already
+	// completed a full evaluation — skip re-emitting alerts. Attempts that
+	// were counted but never marked evaluated (eval failed mid-way and the
+	// message was nacked) fall through and re-run as recovery.
+	if !res.NewlyCounted && res.AlreadyEvaluated {
+		m.logger.Ctx(ctx).Debug("skipping replayed attempt: already evaluated",
 			zap.String("attempt_id", attempt.Attempt.ID),
 			zap.String("tenant_id", attempt.Destination.TenantID),
 			zap.String("destination_id", attempt.Destination.ID),
 		)
+		return true, nil
 	}
 
-	return nil
+	count := res.Count
+	level, shouldAlert := m.evaluator.ShouldAlert(count)
+	if !shouldAlert {
+		return false, nil
+	}
+
+	// At 100% threshold, disable the destination and emit disabled alert.
+	// Both operations are idempotent on replay: DisableDestination is a no-op
+	// if already disabled, and consumers deduplicate events by ID.
+	if level == 100 && m.disabler != nil {
+		if err := m.disabler.DisableDestination(ctx, attempt.Destination.TenantID, attempt.Destination.ID); err != nil {
+			return false, fmt.Errorf("failed to disable destination: %w", err)
+		}
+
+		now := time.Now()
+		attempt.Destination.DisabledAt = &now
+
+		m.logger.Ctx(ctx).Audit("destination disabled",
+			zap.String("attempt_id", attempt.Attempt.ID),
+			zap.String("event_id", attempt.Event.ID),
+			zap.String("tenant_id", attempt.Destination.TenantID),
+			zap.String("destination_id", attempt.Destination.ID),
+			zap.String("destination_type", attempt.Destination.Type),
+		)
+
+		disabledData := DestinationDisabledData{
+			TenantID:    attempt.Destination.TenantID,
+			Destination: attempt.Destination,
+			DisabledAt:  now,
+			Reason:      "consecutive_failure",
+			Event:       attempt.Event,
+			Attempt:     attempt.Attempt,
+		}
+		if err := m.emitter.Emit(ctx, opevents.TopicAlertDestinationDisabled, attempt.Destination.TenantID, disabledData); err != nil {
+			return false, fmt.Errorf("failed to emit destination disabled alert: %w", err)
+		}
+	}
+
+	// Emit consecutive failure alert
+	cfData := ConsecutiveFailureData{
+		TenantID:    attempt.Destination.TenantID,
+		Event:       attempt.Event,
+		Attempt:     attempt.Attempt,
+		Destination: attempt.Destination,
+		ConsecutiveFailures: ConsecutiveFailures{
+			Current:   count,
+			Max:       m.autoDisableFailureCount,
+			Threshold: level,
+		},
+	}
+	if err := m.emitter.Emit(ctx, opevents.TopicAlertConsecutiveFailure, attempt.Destination.TenantID, cfData); err != nil {
+		return false, fmt.Errorf("failed to emit consecutive failure alert: %w", err)
+	}
+
+	m.logger.Ctx(ctx).Audit("alert sent",
+		zap.String("topic", opevents.TopicAlertConsecutiveFailure),
+		zap.String("attempt_id", attempt.Attempt.ID),
+		zap.String("event_id", attempt.Event.ID),
+		zap.String("tenant_id", attempt.Destination.TenantID),
+		zap.String("destination_id", attempt.Destination.ID),
+		zap.String("destination_type", attempt.Destination.Type),
+	)
+
+	return false, nil
 }