feat(opevents): attempt.success and attempt.failed operator events

docs/content/features/operator-events.mdoc

@@ -31,8 +31,12 @@ 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) |
+| `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 |
 
+The `attempt.success` and `attempt.failed` topics fire once per delivery attempt, so they dominate event volume — including under `*`. Subscribe to them deliberately and size your sink for your delivery throughput.
+
 {% tabs tabGroup="deployment" %}
 {% tab label="Managed" %}
 Operator event delivery and topic selection are managed through [Hookdeck Monitoring settings](https://dashboard.hookdeck.com/settings/project/monitoring).
@@ -194,6 +198,25 @@ Emitted when a delivery exhausts all retry attempts. Deduplicated per event+dest
 }
 ```
 
+### `attempt.success` / `attempt.failed`
+
+Emitted once per delivery attempt, keyed by outcome. The two topics share one payload shape; `attempt.status` carries the outcome. `attempt.failed` fires on every failed attempt including retries — use `attempt.attempt_number` and `event.eligible_for_retry` to distinguish retryable failures from final ones.
+
+```json
+{
+  "tenant_id": "tenant_123",
+  "event": {},
+  "attempt": {},
+  "destination": {
+    "id": "des_456",
+    "tenant_id": "tenant_123",
+    "type": "webhook",
+    "topics": ["order.created"],
+    "disabled_at": null
+  }
+}
+```
+
 ### `tenant.subscription.updated`
 
 Emitted when destination changes affect tenant-level subscribed topics or destination count.
@@ -210,7 +233,7 @@ Emitted when destination changes affect tenant-level subscribed topics or destin
 
 ## Delivery Guarantees
 
-`alert.*` topics are delivered with an at-least-once guarantee. For other topics (e.g. `tenant.subscription.updated`), delivery is on a best-effort basis with up to 3 attempts. Consumers should deduplicate using the event `id`.
+`alert.*` and `attempt.*` topics are delivered with an at-least-once guarantee. For other topics (e.g. `tenant.subscription.updated`), delivery is on a best-effort basis with up to 3 attempts. Consumers should deduplicate using the event `id`.
 
 ## Related Configuration
 

internal/alert/evaluator.go

@@ -109,6 +109,13 @@ func NewEvaluator(store AlertStore, retryMaxLimit int, opts ...Option) *Evaluato
 	return e
 }
 
+// SignalsEnabled reports whether any signal can ever fire: consecutive-failure
+// tracking, or exhausted-retries with a positive retry limit. When false,
+// Evaluate never touches the store and always returns an empty verdict.
+func (e *Evaluator) SignalsEnabled() bool {
+	return e.consecutiveFailureEnabled || (e.exhaustedRetriesEnabled && e.retryMaxLimit > 0)
+}
+
 func (e *Evaluator) Evaluate(ctx context.Context, attempt Attempt) (Evaluation, error) {
 	if attempt.Success {
 		// Nothing is tracked when consecutive-failure tracking is disabled, so

internal/idempotence/idempotence.go

@@ -140,6 +140,11 @@ func (i *IdempotenceImpl) Exec(ctx context.Context, key string, exec func(contex
 	execCtx, span := i.tracer.Start(ctx, "Idempotence.Exec")
 	err = exec(execCtx)
 	if err != nil {
+		// Known edge case: when exec failed because ctx itself was canceled,
+		// this clear fails too and the key lingers as "processing" until the
+		// claim TTL (options.Timeout) expires it — retries in that window get
+		// ErrConflict, then self-heal. Rare and bounded, so not worth the fix
+		// yet; if it bites, clear on context.WithoutCancel(ctx).
 		clearErr := i.clearIdempotency(ctx, prefixedKey)
 		if clearErr != nil {
 			finalErr := errors.Join(err, clearErr)

internal/logmq/attempt_events_test.go

@@ -0,0 +1,54 @@
+package logmq_test
+
+// attempt.success / attempt.failed emission. The failed side is pinned all
+// over the characterization suite (every failed attempt's expected multiset
+// carries one attempt.failed); these cover the success side, which has its own
+// path — no replay gate, no mark, send-then-ack.
+
+import (
+	"testing"
+
+	"github.com/hookdeck/outpost/internal/models"
+	"github.com/stretchr/testify/assert"
+)
+
+// A successful attempt emits exactly one attempt.success and acks.
+func TestAttemptEvents_SuccessEmits(t *testing.T) {
+	t.Parallel()
+	h := newHarness(t, harnessConfig{
+		batcher: batcherConfig{itemCount: 1},
+		alert:   alertConfig{autoDisableCount: 2, thresholds: []int{50, 100}},
+	})
+
+	dest, tenant := "dest_ae1", "tenant_ae1"
+	cm, msg := newCountingMessage(makeEntry(dest, tenant, "att_ok", models.AttemptStatusSuccess))
+	h.add(msg)
+	h.waitTerminal([]*countingMessage{cm})
+
+	cm.requireAcked(t)
+	recs := h.sink.forDest(dest)
+	assert.Equal(t, []string{topicSuccess}, topics(recs))
+	assert.Equal(t, []string{"att_ok"}, attemptIDs(recs))
+}
+
+// A failed attempt.success send nacks the message: the success path owes its
+// event the same at-least-once treatment as the failure path (redelivery
+// re-runs the reset — idempotent — and re-sends).
+func TestAttemptEvents_SuccessEmitFailure_Nacks(t *testing.T) {
+	t.Parallel()
+	h := newHarness(t, harnessConfig{
+		batcher: batcherConfig{itemCount: 1},
+		alert:   alertConfig{autoDisableCount: 2, thresholds: []int{50, 100}},
+		doubles: doublesConfig{
+			sinkFailOn: map[string]bool{topicSuccess: true},
+		},
+	})
+
+	dest, tenant := "dest_ae2", "tenant_ae2"
+	cm, msg := newCountingMessage(makeEntry(dest, tenant, "att_ok", models.AttemptStatusSuccess))
+	h.add(msg)
+	h.waitTerminal([]*countingMessage{cm})
+
+	cm.requireNacked(t)
+	assert.Empty(t, h.sink.forDest(dest), "the failed send recorded nothing")
+}

internal/logmq/batchprocessor.go

@@ -39,6 +39,10 @@ type LogStore interface {
 // processor.
 type AlertEvaluator interface {
 	Evaluate(ctx context.Context, attempt alert.Attempt) (alert.Evaluation, error)
+	// SignalsEnabled reports whether any alert signal can ever fire. When
+	// false, Evaluate is a stateless no-op, so the pipeline skips the replay
+	// gate (there is no streak or verdict to protect from replays).
+	SignalsEnabled() bool
 }
 
 // DestinationDisabler disables destinations that hit the auto-disable
@@ -110,6 +114,11 @@ type BatchProcessor struct {
 	alerts      AlertPipeline
 	batcher     *batcher.Batcher[*mqs.Message]
 	emitTimeout time.Duration
+	// alertsEnabled and emitsAttemptEvents are derived once from the pipeline's
+	// static config. alertsEnabled=false skips the replay gate on the failed
+	// path (nothing to protect); both false skips per-entry work entirely.
+	alertsEnabled      bool
+	emitsAttemptEvents bool
 	// inflight tracks entry goroutines so Shutdown can drain them. Each is
 	// bounded by emitTimeout, so the wait is bounded too.
 	inflight     sync.WaitGroup
@@ -136,6 +145,11 @@ func NewBatchProcessor(ctx context.Context, logger *logging.Logger, logStore Log
 	if bp.emitTimeout <= 0 {
 		bp.emitTimeout = emitTimeout
 	}
+	if alerts.Evaluator != nil {
+		bp.alertsEnabled = alerts.Evaluator.SignalsEnabled()
+		bp.emitsAttemptEvents = alerts.Emitter.Enabled(opevents.TopicAttemptSuccess) ||
+			alerts.Emitter.Enabled(opevents.TopicAttemptFailed)
+	}
 
 	b, err := batcher.NewBatcher(batcher.Config[*mqs.Message]{
 		GroupCountThreshold: 2,
@@ -266,7 +280,9 @@ func (bp *BatchProcessor) processBatch(_ string, msgs []*mqs.Message) {
 	// and every fetched message reaches ack/nack well inside the broker's
 	// visibility window.
 	for i, entry := range entries {
-		if bp.alerts.Evaluator == nil {
+		// No pipeline, or one that can't produce anything (every alert signal
+		// off and no attempt topic subscribed): persisted is terminal.
+		if bp.alerts.Evaluator == nil || (!bp.alertsEnabled && !bp.emitsAttemptEvents) {
 			validMsgs[i].Ack()
 			continue
 		}
@@ -297,9 +313,12 @@ func (bp *BatchProcessor) processBatch(_ string, msgs []*mqs.Message) {
 // replay arriving after a success reset must not count toward the fresh
 // streak. The mark lands only after the attempt's events are delivered — a
 // nacked attempt re-runs in full on redelivery (counting stays correct: the
-// store is idempotent per attempt ID). A success just resets the tracker —
-// idempotent, so it needs no gate (and gating it would cost one Redis key per
-// successful attempt).
+// store is idempotent per attempt ID). A success resets the tracker and emits
+// attempt.success — both idempotent-enough to skip the gate (gating would cost
+// one Redis key per successful attempt, the dominant traffic, to dedup a rare
+// redelivery re-emit; opevents are at-least-once anyway). The gate exists for
+// alert state and alert-event dedup only, so when every signal is disabled the
+// failed path skips it too and just emits attempt.failed.
 func (bp *BatchProcessor) processEntry(ctx context.Context, entry *models.LogEntry, msg *mqs.Message) {
 	attempt := alert.Attempt{
 		TenantID:         entry.Destination.TenantID,
@@ -315,6 +334,29 @@ func (bp *BatchProcessor) processEntry(ctx context.Context, entry *models.LogEnt
 			bp.nackAlertFailure(ctx, err, entry, msg)
 			return
 		}
+		success := deliveryEvent{
+			event: opevents.AttemptSuccessEvent(opevents.NewAlertDestination(entry.Destination), entry.Event, entry.Attempt),
+		}
+		if bp.sendAll(ctx, []deliveryEvent{success}, entry) != nil {
+			msg.Nack()
+			return
+		}
+		msg.Ack()
+		return
+	}
+
+	// With every alert signal off there is no streak to protect and no verdict
+	// to compute, so the replay gate buys nothing — skip its two Redis round
+	// trips and emit attempt.failed ungated, same at-least-once treatment as
+	// attempt.success (a redelivery may re-emit; tolerated).
+	if !bp.alertsEnabled {
+		failed := deliveryEvent{
+			event: opevents.AttemptFailedEvent(opevents.NewAlertDestination(entry.Destination), entry.Event, entry.Attempt),
+		}
+		if bp.sendAll(ctx, []deliveryEvent{failed}, entry) != nil {
+			msg.Nack()
+			return
+		}
 		msg.Ack()
 		return
 	}
@@ -342,28 +384,9 @@ func (bp *BatchProcessor) processEntry(ctx context.Context, entry *models.LogEnt
 		return
 	}
 
-	// Deliver the attempt's events concurrently; arrival order within an
-	// attempt is not guaranteed. Any failure nacks with nothing marked, so
-	// redelivery re-runs the attempt in full — events already sent may go out
-	// again (at-least-once).
-	g, gctx := errgroup.WithContext(ctx)
-	for _, de := range events {
-		g.Go(func() error {
-			sendCtx, cancel := context.WithTimeout(gctx, bp.emitTimeout)
-			defer cancel()
-			if err := bp.send(sendCtx, de, entry); err != nil {
-				bp.logger.Ctx(ctx).Error("opevent delivery failed",
-					zap.Error(err),
-					zap.String("topic", de.event.Topic),
-					zap.String("attempt_id", entry.Attempt.ID),
-					zap.String("event_id", entry.Event.ID),
-					zap.String("destination_id", entry.Destination.ID))
-				return err
-			}
-			return nil
-		})
-	}
-	if g.Wait() != nil {
+	// Any failure nacks with nothing marked, so redelivery re-runs the attempt
+	// in full — events already sent may go out again (at-least-once).
+	if bp.sendAll(ctx, events, entry) != nil {
 		msg.Nack()
 		return
 	}
@@ -387,17 +410,38 @@ type deliveryEvent struct {
 	suppressKey string
 }
 
-// plan acts on an evaluation and builds the operator events owed for this
-// attempt — disabled, consecutive_failure, exhausted_retries. They are sent
-// concurrently, so slice order carries no meaning. The disable (a DB write)
-// happens here: it's an action, not a notification, and it must precede event
-// construction so the payloads carry the destination's latest state
-// (disabled).
-func (bp *BatchProcessor) plan(ctx context.Context, eval alert.Evaluation, entry *models.LogEntry) ([]deliveryEvent, error) {
-	if eval.ConsecutiveFailure == nil && !eval.RetriesExhausted {
-		return nil, nil
+// sendAll delivers an attempt's events concurrently, each under the emit
+// timeout; arrival order within an attempt is not guaranteed. The first
+// failure is returned (the rest still run to completion or cancellation).
+func (bp *BatchProcessor) sendAll(ctx context.Context, events []deliveryEvent, entry *models.LogEntry) error {
+	g, gctx := errgroup.WithContext(ctx)
+	for _, de := range events {
+		g.Go(func() error {
+			sendCtx, cancel := context.WithTimeout(gctx, bp.emitTimeout)
+			defer cancel()
+			if err := bp.send(sendCtx, de); err != nil {
+				bp.logger.Ctx(ctx).Error("opevent delivery failed",
+					zap.Error(err),
+					zap.String("topic", de.event.Topic),
+					zap.String("attempt_id", entry.Attempt.ID),
+					zap.String("event_id", entry.Event.ID),
+					zap.String("destination_id", entry.Destination.ID))
+				return err
+			}
+			return nil
+		})
 	}
+	return g.Wait()
+}
 
+// plan acts on an evaluation and builds the operator events owed for this
+// attempt — attempt.failed always, plus disabled, consecutive_failure, and
+// exhausted_retries per the verdict. They are sent concurrently, so slice
+// order carries no meaning. The disable (a DB write) happens here: it's an
+// action, not a notification, and it must precede event construction so the
+// payloads carry the destination's latest state (disabled) — attempt.failed
+// included, since they share the projection.
+func (bp *BatchProcessor) plan(ctx context.Context, eval alert.Evaluation, entry *models.LogEntry) ([]deliveryEvent, error) {
 	dest := opevents.NewAlertDestination(entry.Destination)
 	var events []deliveryEvent
 
@@ -441,25 +485,20 @@ func (bp *BatchProcessor) plan(ctx context.Context, eval alert.Evaluation, entry
 		events = append(events, de)
 	}
 
+	events = append(events, deliveryEvent{
+		event: opevents.AttemptFailedEvent(dest, entry.Event, entry.Attempt),
+	})
+
 	return events, nil
 }
 
-// send emits one event and audits the send, inside the event's suppression
-// window when it has one. A suppressed duplicate (Exec skips the emit) counts
-// as delivered and is not audited.
-func (bp *BatchProcessor) send(ctx context.Context, de deliveryEvent, entry *models.LogEntry) error {
+// send emits one event, inside the event's suppression window when it has
+// 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.
+func (bp *BatchProcessor) send(ctx context.Context, de deliveryEvent) error {
 	emit := func(ctx context.Context) error {
-		if err := bp.alerts.Emitter.Emit(ctx, de.event); err != nil {
-			return err
-		}
-		bp.logger.Ctx(ctx).Audit("opevent delivered",
-			zap.String("topic", de.event.Topic),
-			zap.String("attempt_id", entry.Attempt.ID),
-			zap.String("event_id", entry.Event.ID),
-			zap.String("tenant_id", de.event.TenantID),
-			zap.String("destination_id", entry.Destination.ID),
-			zap.String("destination_type", entry.Destination.Type))
-		return nil
+		return bp.alerts.Emitter.Emit(ctx, de.event)
 	}
 	if de.suppressKey == "" {
 		return emit(ctx)

internal/logmq/batchprocessor_test.go

@@ -321,6 +321,10 @@ type mockAlertEvaluator struct {
 	returnErr error
 }
 
+// SignalsEnabled is true so the pipeline treats the mock as a live evaluator
+// (the gated path these tests exercise).
+func (m *mockAlertEvaluator) SignalsEnabled() bool { return true }
+
 func (m *mockAlertEvaluator) Evaluate(ctx context.Context, attempt alert.Attempt) (alert.Evaluation, error) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
@@ -343,7 +347,7 @@ func testAlertPipeline(t *testing.T, evaluator logmq.AlertEvaluator) logmq.Alert
 	t.Helper()
 	return logmq.AlertPipeline{
 		Evaluator:      evaluator,
-		Emitter:        opevents.NewEmitter(&opevents.NoopSink{}, "test-deploy", []string{"*"}),
+		Emitter:        opevents.NewEmitter(&opevents.NoopSink{}, "test-deploy", []string{"*"}, testutil.CreateTestLogger(t)),
 		ProcessedIdemp: idempotence.New(testutil.CreateTestRedisClient(t)),
 	}
 }

internal/logmq/characterization_acknowledgement_test.go

@@ -72,15 +72,18 @@ func TestCharacterization_MixedBatchAccounting(t *testing.T) {
 	cmInvalid.requireNacked(t)
 	cmEmitFail.requireNacked(t)
 
-	// Only the alerting destination produced a (successful) record.
-	assert.Equal(t, []string{topicCF}, topics(h.sink.forDest(destAlert)))
+	// Successful records: one attempt event per processed attempt (the dup's
+	// kept copy included), plus the cf alert. The emit-fail destination
+	// recorded nothing — its sends all fail.
+	assert.ElementsMatch(t, []string{topicFailed, topicCF}, topics(h.sink.forDest(destAlert)))
+	assert.Equal(t, []string{topicSuccess}, topics(h.sink.forDest(destSuccess)))
+	assert.Equal(t, []string{topicSuccess}, topics(h.sink.forDest(destDup)), "the dup's kept copy emitted once")
 	assert.Empty(t, h.sink.forDest(destFail), "emit-fail destination produced no recorded event")
-	assert.Len(t, h.sink.snapshot(), 1, "exactly one event delivered to the sink")
+	assert.Len(t, h.sink.snapshot(), 4, "no records beyond the four accounted for")
 }
 
 // A single failed attempt below any threshold (count 1) → persisted, acked,
-// zero sink records (the "nothing to deliver" fast path: mark + ack, no
-// delivery task).
+// and only the attempt.failed event delivered — no alert topics.
 func TestCharacterization_BelowThresholdNoAlert(t *testing.T) {
 	t.Parallel()
 	h := newHarness(t, harnessConfig{
@@ -94,6 +97,7 @@ func TestCharacterization_BelowThresholdNoAlert(t *testing.T) {
 	h.waitTerminal([]*countingMessage{cm})
 
 	cm.requireAcked(t)
-	assert.Empty(t, h.sink.snapshot(), "count 1 is below the 50%% threshold (5)")
+	assert.Equal(t, []string{topicFailed}, topics(h.sink.snapshot()),
+		"count 1 is below the 50%% threshold (5): only the attempt event, no alert")
 	require.Len(t, h.listAttempt(destA), 1, "attempt persisted")
 }