Compare commits
1 Commits
190cabe0f2
...
issue-82-o
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
0384de4a7b |
12
README.md
12
README.md
@@ -636,6 +636,18 @@ This means:
|
|||||||
durable fallback that ensures no retry is permanently lost, even under
|
durable fallback that ensures no retry is permanently lost, even under
|
||||||
extreme backpressure.
|
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)
|
### Circuit Breaker (HTTP Targets with Retries)
|
||||||
|
|
||||||
HTTP targets with `max_retries` > 0 are protected by a **per-target circuit breaker** that
|
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
|
# 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
|
- 2026-08-07 Update golangci-lint to v2.12.2 (Docker image digest in
|
||||||
`Dockerfile`, release-archive sha256 pins in `script/bootstrap`),
|
`Dockerfile`, release-archive sha256 pins in `script/bootstrap`),
|
||||||
adopt the canonical `.golangci.yml` (v2 `linters.settings` layout so
|
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
|
// recoverSingleRetry hands an orphaned retrying delivery back
|
||||||
// to its target to recompute the remaining backoff, then
|
// to its target to recompute the remaining backoff, then
|
||||||
// reschedules it. Targets that do not own durable retries
|
// reschedules it. Targets that do not own durable retries
|
||||||
// (fire-and-forget) never produce retrying deliveries, so
|
// (fire-and-forget) never produce retrying deliveries, so a
|
||||||
// they are skipped.
|
// delivery found in that state has had its target's type
|
||||||
|
// changed underneath it and is terminally failed.
|
||||||
func (e *Engine) recoverSingleRetry(
|
func (e *Engine) recoverSingleRetry(
|
||||||
webhookDB *gorm.DB,
|
webhookDB *gorm.DB,
|
||||||
webhookID string,
|
webhookID string,
|
||||||
@@ -475,6 +476,10 @@ func (e *Engine) recoverSingleRetry(
|
|||||||
|
|
||||||
rs, ok := e.targets[target.Type].(rescheduler)
|
rs, ok := e.targets[target.Type].(rescheduler)
|
||||||
if !ok {
|
if !ok {
|
||||||
|
e.failUnretryableRetry(
|
||||||
|
webhookDB, webhookID, d, &target,
|
||||||
|
)
|
||||||
|
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -649,8 +654,8 @@ func (e *Engine) sweepWebhookRetries(
|
|||||||
|
|
||||||
// sweepSingleRetry re-enqueues an orphaned retrying delivery
|
// sweepSingleRetry re-enqueues an orphaned retrying delivery
|
||||||
// whose backoff window has elapsed, delegating the backoff
|
// whose backoff window has elapsed, delegating the backoff
|
||||||
// decision to the delivery's target. Targets that do not own
|
// decision to the delivery's target. A delivery whose target
|
||||||
// durable retries are skipped.
|
// no longer owns durable retries is terminally failed.
|
||||||
func (e *Engine) sweepSingleRetry(
|
func (e *Engine) sweepSingleRetry(
|
||||||
webhookDB *gorm.DB,
|
webhookDB *gorm.DB,
|
||||||
webhookID string,
|
webhookID string,
|
||||||
@@ -670,6 +675,10 @@ func (e *Engine) sweepSingleRetry(
|
|||||||
|
|
||||||
rs, ok := e.targets[target.Type].(rescheduler)
|
rs, ok := e.targets[target.Type].(rescheduler)
|
||||||
if !ok {
|
if !ok {
|
||||||
|
e.failUnretryableRetry(
|
||||||
|
webhookDB, webhookID, d, &target,
|
||||||
|
)
|
||||||
|
|
||||||
return
|
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
|
// processDelivery dispatches a delivery to the target that
|
||||||
// owns its type. Unknown target types fail the delivery.
|
// owns its type. Unknown target types fail the delivery.
|
||||||
func (e *Engine) processDelivery(
|
func (e *Engine) processDelivery(
|
||||||
|
|||||||
@@ -748,6 +748,193 @@ func TestRecoverWebhookDeliveries_RetryingDeliveries(
|
|||||||
case <-time.After(5 * time.Second):
|
case <-time.After(5 * time.Second):
|
||||||
t.Fatal("expected retry task from recovery")
|
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.
|
// iSeedFailedResult creates a failed delivery result.
|
||||||
|
|||||||
@@ -188,6 +188,13 @@ func (e *Engine) ExportRecoverInFlight(
|
|||||||
e.recoverInFlight(ctx)
|
e.recoverInFlight(ctx)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ExportSweepWebhookRetries exposes sweepWebhookRetries.
|
||||||
|
func (e *Engine) ExportSweepWebhookRetries(
|
||||||
|
ctx context.Context, webhookID string,
|
||||||
|
) {
|
||||||
|
e.sweepWebhookRetries(ctx, webhookID)
|
||||||
|
}
|
||||||
|
|
||||||
// ExportStart exposes start for testing.
|
// ExportStart exposes start for testing.
|
||||||
func (e *Engine) ExportStart(ctx context.Context) {
|
func (e *Engine) ExportStart(ctx context.Context) {
|
||||||
e.start(ctx)
|
e.start(ctx)
|
||||||
|
|||||||
Reference in New Issue
Block a user