Terminally fail retrying deliveries with a non-retry target type (closes #82) #104
12
README.md
12
README.md
@@ -636,6 +636,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
|
||||
|
||||
4
TODO.md
4
TODO.md
@@ -28,6 +28,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-07 Update golangci-lint to v2.12.2 (Docker image digest in
|
||||
`Dockerfile`, release-archive sha256 pins in `script/bootstrap`),
|
||||
adopt the canonical `.golangci.yml` (v2 `linters.settings` layout so
|
||||
|
||||
@@ -453,8 +453,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,
|
||||
@@ -475,6 +476,10 @@ func (e *Engine) recoverSingleRetry(
|
||||
|
||||
rs, ok := e.targets[target.Type].(rescheduler)
|
||||
if !ok {
|
||||
e.failUnretryableRetry(
|
||||
webhookDB, webhookID, d, &target,
|
||||
)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
@@ -649,8 +654,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,
|
||||
@@ -670,6 +675,10 @@ func (e *Engine) sweepSingleRetry(
|
||||
|
||||
rs, ok := e.targets[target.Type].(rescheduler)
|
||||
if !ok {
|
||||
e.failUnretryableRetry(
|
||||
webhookDB, webhookID, d, &target,
|
||||
)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
@@ -710,6 +719,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(
|
||||
|
||||
@@ -748,6 +748,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.
|
||||
|
||||
@@ -188,6 +188,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(ctx context.Context) {
|
||||
e.start(ctx)
|
||||
|
||||
Reference in New Issue
Block a user