From ead81298ed5c3a31cde5646f1f66528357db6e80 Mon Sep 17 00:00:00 2001 From: clawbot Date: Sun, 9 Aug 2026 05:51:51 +0000 Subject: [PATCH] Terminally fail retrying deliveries with a non-retry target type (closes #82) Restart recovery and the 60s retry sweep both looked an orphaned `retrying` delivery's target up in the registry and silently returned when it did not implement `rescheduler`. If a target's type was edited from a retry type (`http`/`slack`) to a fire-and-forget type (`database`/`log`) or an unknown one while a delivery was still retrying, that delivery stayed `retrying` forever. Both sites now hand the delivery to one shared helper, `failUnretryableRetry`, which records a `DeliveryResult` naming the current target type as the reason and marks the delivery `failed`. It logs at warn, not error: this is operator-caused state, not a system fault. Re-dispatching under the new type was rejected as it would perform a delivery the operator never asked for; the event itself stays in the per-webhook event database, so manual redelivery can recover it deliberately. Fire-and-forget targets never set status `retrying` under normal operation, so this path stays unreachable for them in practice. --- README.md | 12 ++ TODO.md | 4 + internal/delivery/engine.go | 70 ++++++- internal/delivery/engine_integration_test.go | 187 +++++++++++++++++++ internal/delivery/export_test.go | 7 + 5 files changed, 276 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index cc79a1a..40efaae 100644 --- a/README.md +++ b/README.md @@ -666,6 +666,18 @@ This means: durable fallback that ensures no retry is permanently lost, even under extreme backpressure. +**Changing a target's type does not migrate in-flight deliveries.** Only +`http` and `slack` targets own durable retries; `database` and `log` +targets are fire-and-forget and never produce a `retrying` delivery. If a +target's `type` is edited from a retrying type to a non-retrying (or +unknown) one while one of its deliveries is still `retrying`, both +recovery paths above terminally mark that delivery `failed` and record a +`DeliveryResult` naming the current target type as the reason, logging it +at warn level. The delivery is not re-dispatched under the new type — the +operator never asked for that delivery — and the event itself remains +stored in the per-webhook event database, so it can be redelivered +manually. + ### Circuit Breaker (HTTP Targets with Retries) HTTP targets with `max_retries` > 0 are protected by a **per-target circuit breaker** that diff --git a/TODO.md b/TODO.md index cfafb43..23642a2 100644 --- a/TODO.md +++ b/TODO.md @@ -29,6 +29,10 @@ databases currently grow without bound. # Completed Steps +- 2026-08-09 Restart recovery and the 60s retry sweep terminally fail an + orphaned `retrying` delivery whose target type no longer supports + retries, recording a `DeliveryResult` with the reason instead of + leaving the delivery stuck forever (#82) - 2026-08-09 Root the delivery engine's worker pool and the retention reaper's sweep loop at `context.Background()` rather than the fx `OnStart` hook context (#97), which carries fx's 15s start timeout and diff --git a/internal/delivery/engine.go b/internal/delivery/engine.go index d000bee..564fd87 100644 --- a/internal/delivery/engine.go +++ b/internal/delivery/engine.go @@ -508,8 +508,9 @@ func (e *Engine) recoverRetryingDeliveries( // recoverSingleRetry hands an orphaned retrying delivery back // to its target to recompute the remaining backoff, then // reschedules it. Targets that do not own durable retries -// (fire-and-forget) never produce retrying deliveries, so -// they are skipped. +// (fire-and-forget) never produce retrying deliveries, so a +// delivery found in that state has had its target's type +// changed underneath it and is terminally failed. func (e *Engine) recoverSingleRetry( webhookDB *gorm.DB, webhookID string, @@ -530,6 +531,10 @@ func (e *Engine) recoverSingleRetry( rs, ok := e.targets[target.Type].(rescheduler) if !ok { + e.failUnretryableRetry( + webhookDB, webhookID, d, &target, + ) + return } @@ -704,8 +709,8 @@ func (e *Engine) sweepWebhookRetries( // sweepSingleRetry re-enqueues an orphaned retrying delivery // whose backoff window has elapsed, delegating the backoff -// decision to the delivery's target. Targets that do not own -// durable retries are skipped. +// decision to the delivery's target. A delivery whose target +// no longer owns durable retries is terminally failed. func (e *Engine) sweepSingleRetry( webhookDB *gorm.DB, webhookID string, @@ -725,6 +730,10 @@ func (e *Engine) sweepSingleRetry( rs, ok := e.targets[target.Type].(rescheduler) if !ok { + e.failUnretryableRetry( + webhookDB, webhookID, d, &target, + ) + return } @@ -765,6 +774,59 @@ func (e *Engine) sweepSingleRetry( } } +// failUnretryableRetry terminally fails an orphaned retrying +// delivery whose target type no longer supports retries. Both +// restart recovery and the periodic sweep call it, so the +// terminal transition exists once. +// +// This is only reachable when a target's type has been changed +// out from under an in-flight retrying delivery (or the type is +// unknown to the registry): fire-and-forget targets never set +// status retrying themselves. Re-dispatching under the new type +// would be a delivery the operator never asked for, and leaving +// the row retrying strands it forever, so the delivery is +// failed with a recorded reason and can be redelivered +// manually. Logged at warn, not error: this is operator-caused +// state, not a system fault. +func (e *Engine) failUnretryableRetry( + webhookDB *gorm.DB, + webhookID string, + d *database.Delivery, + target *database.Target, +) { + e.log.Warn( + "failing orphaned retrying delivery: target "+ + "type no longer supports retries", + "webhook_id", webhookID, + "delivery_id", d.ID, + "target_id", target.ID, + "target_name", target.Name, + "target_type", target.Type, + ) + + reason := fmt.Sprintf( + "target type %q does not support retries; "+ + "delivery was left retrying by a previous "+ + "target type and has been failed terminally", + target.Type, + ) + + e.recordResult( + webhookDB, + d, + e.countAttempts(webhookDB, d.ID)+1, + false, + 0, + "", + reason, + 0, + ) + + e.updateDeliveryStatus( + webhookDB, d, database.DeliveryStatusFailed, + ) +} + // processDelivery dispatches a delivery to the target that // owns its type. Unknown target types fail the delivery. func (e *Engine) processDelivery( diff --git a/internal/delivery/engine_integration_test.go b/internal/delivery/engine_integration_test.go index 004a4d1..50aeb1a 100644 --- a/internal/delivery/engine_integration_test.go +++ b/internal/delivery/engine_integration_test.go @@ -741,6 +741,193 @@ func TestRecoverWebhookDeliveries_RetryingDeliveries( case <-time.After(5 * time.Second): t.Fatal("expected retry task from recovery") } + + // Regression guard: a target that still supports retries + // must be rescheduled, never terminally failed, and must + // not gain a synthetic result row. + iAssertStatus( + t, s.WebhookDB, d.ID, + database.DeliveryStatusRetrying, + ) + + assert.Len(t, iResults(t, s.WebhookDB, d.ID), 1) +} + +// --- Retrying deliveries whose target type changed --- + +// iSeedRetryingWithType seeds a retrying delivery with one +// recorded failed attempt against a target of the given type, +// standing in for a target whose type was edited in the main +// database while the delivery was still retrying. +func iSeedRetryingWithType( + t *testing.T, + s iSetup, + targetType database.TargetType, +) string { + t.Helper() + + targetID := uuid.New().String() + + iCreateTarget(t, s.MainDB, targetID, + s.WebhookID, "mutated-target", targetType, + iHTTPConfig("http://example.com/hook"), 5, + ) + + event := iSeedEvent( + t, s.WebhookDB, s.WebhookID, + `{"orphaned":"retry"}`, + ) + + d := iSeedDelivery( + t, s.WebhookDB, event.ID, targetID, + database.DeliveryStatusRetrying, + ) + + iSeedFailedResult(t, s.WebhookDB, d.ID) + + return d.ID +} + +// iResults loads a delivery's results in attempt order. +func iResults( + t *testing.T, db *gorm.DB, deliveryID string, +) []database.DeliveryResult { + t.Helper() + + var results []database.DeliveryResult + + require.NoError(t, db. + Where("delivery_id = ?", deliveryID). + Order("attempt_num"). + Find(&results).Error) + + return results +} + +// iAssertTerminallyFailed asserts the delivery ended failed +// with a result row recording why, and was not rescheduled. +func iAssertTerminallyFailed( + t *testing.T, + s iSetup, + deliveryID string, + targetType database.TargetType, +) { + t.Helper() + + iAssertStatus( + t, s.WebhookDB, deliveryID, + database.DeliveryStatusFailed, + ) + + results := iResults(t, s.WebhookDB, deliveryID) + require.Len(t, results, 2) + + last := results[1] + + assert.False(t, last.Success) + assert.Equal(t, 2, last.AttemptNum) + + assert.Contains( + t, last.Error, string(targetType), + ) + + assert.Contains( + t, last.Error, "does not support retries", + ) + + assert.Empty(t, s.Engine.ExportRetryCh()) +} + +func TestRecoverSingleRetry_TypeNoLongerRetries( + t *testing.T, +) { + t.Parallel() + + s := newISetup(t) + + iCreateWebhook( + t, s.MainDB, s.WebhookID, "mutated-type", + ) + + deliveryID := iSeedRetryingWithType( + t, s, database.TargetTypeLog, + ) + + s.Engine.ExportRecoverWebhookDeliveries( + context.Background(), s.WebhookID, + ) + + iAssertTerminallyFailed( + t, s, deliveryID, database.TargetTypeLog, + ) +} + +func TestSweepSingleRetry_TypeNoLongerRetries( + t *testing.T, +) { + t.Parallel() + + s := newISetup(t) + + iCreateWebhook( + t, s.MainDB, s.WebhookID, "mutated-type-sweep", + ) + + deliveryID := iSeedRetryingWithType( + t, s, database.TargetTypeDatabase, + ) + + s.Engine.ExportSweepWebhookRetries( + context.Background(), s.WebhookID, + ) + + iAssertTerminallyFailed( + t, s, deliveryID, database.TargetTypeDatabase, + ) +} + +func TestRecoverSingleRetry_UnknownTargetType( + t *testing.T, +) { + t.Parallel() + + s := newISetup(t) + + iCreateWebhook( + t, s.MainDB, s.WebhookID, "unknown-type", + ) + + unknown := database.TargetType("not-a-target-type") + + deliveryID := iSeedRetryingWithType(t, s, unknown) + + s.Engine.ExportRecoverWebhookDeliveries( + context.Background(), s.WebhookID, + ) + + iAssertTerminallyFailed(t, s, deliveryID, unknown) +} + +func TestSweepSingleRetry_UnknownTargetType( + t *testing.T, +) { + t.Parallel() + + s := newISetup(t) + + iCreateWebhook( + t, s.MainDB, s.WebhookID, "unknown-type-sweep", + ) + + unknown := database.TargetType("not-a-target-type") + + deliveryID := iSeedRetryingWithType(t, s, unknown) + + s.Engine.ExportSweepWebhookRetries( + context.Background(), s.WebhookID, + ) + + iAssertTerminallyFailed(t, s, deliveryID, unknown) } // iSeedFailedResult creates a failed delivery result. diff --git a/internal/delivery/export_test.go b/internal/delivery/export_test.go index 1aa2eb8..326cbc1 100644 --- a/internal/delivery/export_test.go +++ b/internal/delivery/export_test.go @@ -195,6 +195,13 @@ func (e *Engine) ExportRecoverInFlight( e.recoverInFlight(ctx) } +// ExportSweepWebhookRetries exposes sweepWebhookRetries. +func (e *Engine) ExportSweepWebhookRetries( + ctx context.Context, webhookID string, +) { + e.sweepWebhookRetries(ctx, webhookID) +} + // ExportStart exposes start for testing. func (e *Engine) ExportStart() { e.start()