From c695b7714da619e6cebfcd1c327cdab3b88492b6 Mon Sep 17 00:00:00 2001 From: sneak Date: Thu, 15 Oct 2026 08:02:12 +0000 Subject: [PATCH] Leave a delivery cut off by a stop for the next start (closes #527) Stopping the engine cancels the context every http and slack request is sent with. An attempt cut off that way was recorded as failed, which for a target with no retries, or on a last retry, ended the delivery failed and nothing sent it again. The shared retry core now records nothing for a failed attempt when the engine's context is cancelled: no result, no circuit breaker outcome, no status change. The delivery stays pending or retrying and startup recovery sends it again. A target's own timeout is still a failed attempt. The README's Graceful shutdown bullet, Shutdown section and the at-least-once note say so. Model: opus-5-5 --- README.md | 33 +++-- internal/delivery/engine_lifecycle_test.go | 145 +++++++++++++++++++++ internal/delivery/target_http.go | 48 ++++++- internal/delivery/target_slack.go | 2 +- 4 files changed, 213 insertions(+), 15 deletions(-) diff --git a/README.md b/README.md index 80ed3db..01a1178 100644 --- a/README.md +++ b/README.md @@ -1325,9 +1325,11 @@ webhooker solves this by acting as a durable intermediary: deliberately left in a recoverable state rather than marked done — losing a delivery is the worse failure — so the pending sweep picks it up about fifteen minutes later, or the next restart does, and the target receives a - payload it already got. webhooker adds no delivery identifier of its own to - an outbound request, so **make your receiver idempotent** against whatever - the payload itself carries. + payload it already got. A send still in flight when the service stops is + left the same way: its attempt is not recorded, and the next start sends it + again (see Shutdown). webhooker adds no delivery identifier of its own to an + outbound request, so **make your receiver idempotent** against whatever the + payload itself carries. 3. **Observability** — Full request/response logging for every webhook received and every delivery attempted. Prometheus metrics expose volume, latency, and @@ -1768,7 +1770,7 @@ is what bounds it. #### DeliveryResult The result of a single delivery attempt. Every attempt (including retries) is -individually logged for full observability. +individually logged, except one cut off by a stop (see Shutdown). | Field | Type | Description | | --------------- | ------- | ----------------------------------------- | @@ -2206,12 +2208,14 @@ This means: buffer rather than spawning more goroutines. - **Independent results** — each worker records its own delivery result in the per-webhook database without coordination. -- **Graceful shutdown** — cancel the context, workers finish their current task - and exit. The stop hook waits for the pool via `lifecycle.WaitForShutdown`, - which bounds that wait by fx's stop timeout rather than blocking forever on a - wedged worker. On timeout it logs at `ERROR` and returns an error, and the - goroutines that did not finish are still running — an unclean shutdown is - reported rather than hidden. +- **Graceful shutdown** — the stop hook cancels the workers' context, and each + worker exits once its current task returns. Cancelling cuts off an `http` or + `slack` request still in flight; that attempt is not recorded, and the next + start sends the delivery again (see Shutdown). The stop hook waits for the + pool via `lifecycle.WaitForShutdown`, which bounds that wait by fx's stop + timeout rather than blocking forever on a wedged worker. On timeout it logs at + `ERROR` and returns an error, and the goroutines that did not finish are still + running — an unclean shutdown is reported rather than hidden. **Recovery paths:** @@ -3284,7 +3288,8 @@ stop-hook log: 3. `server` — the HTTP drain, bounded by `server.ShutdownTimeout` (**3 seconds**) and by what the hooks before it left, then a Sentry flush if `SENTRY_DSN` is set -4. `delivery.Engine` — waits for its workers, then closes the archive databases +4. `delivery.Engine` — cuts off any `http` or `slack` request in flight, waits + for its workers, then closes the archive databases 5. `healthcheck` 6. `WebhookDBManager` 7. the database close @@ -3295,6 +3300,12 @@ by the stop context, so a wedge there consumes the 5 seconds before the HTTP server hook is ever entered. The hooks after the server are microsecond-scale in normal operation. +The delivery engine does not wait for a delivery request in flight: it cancels +it. That attempt is not recorded and the delivery keeps its `pending` or +`retrying` status, so startup recovery sends it again on the next start (see +Recovery paths under Bounded Worker Pool). If the receiver already had it, it +receives it twice: the at-least-once duplicate described under Rationale. + The HTTP drain budget is deliberately **shorter** than the sequence budget. Were the two equal, a drain that used its whole budget would exhaust the sequence budget at the instant it finished, and every later hook — the delivery engine, diff --git a/internal/delivery/engine_lifecycle_test.go b/internal/delivery/engine_lifecycle_test.go index 69e0d43..df68f2c 100644 --- a/internal/delivery/engine_lifecycle_test.go +++ b/internal/delivery/engine_lifecycle_test.go @@ -2,11 +2,15 @@ package delivery_test import ( "context" + "net/http" + "net/http/httptest" "path/filepath" + "sync" "testing" "time" "github.com/google/uuid" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "go.uber.org/fx" "gorm.io/gorm/clause" @@ -272,6 +276,147 @@ func TestEngine_StopHookHonoursStopTimeout(t *testing.T) { requireStopHookExpires(t, lc.hooks[0], "delivery engine") } +// seedHeldDelivery gives the setup's webhook an http target with +// maxRetries, pointed at a receiver that keeps every request open until +// the client gives up on it, and seeds one delivery to that target at +// status. It returns the task that sends attempt attemptNum of the +// delivery, and a channel that is closed when a request reaches the +// receiver. +func seedHeldDelivery( + t *testing.T, + s iSetup, + maxRetries int, + status database.DeliveryStatus, + attemptNum int, +) (delivery.Task, <-chan struct{}) { + t.Helper() + + arrived := make(chan struct{}) + + var once sync.Once + + ts := httptest.NewServer(http.HandlerFunc( + func(_ http.ResponseWriter, r *http.Request) { + once.Do(func() { close(arrived) }) + + <-r.Context().Done() + }, + )) + t.Cleanup(ts.Close) + + cfg := iHTTPConfig(ts.URL) + targetID := uuid.New().String() + + iCreateWebhook(t, s.MainDB, s.WebhookID, "hook") + iCreateTarget( + t, s.MainDB, targetID, s.WebhookID, "held", + database.TargetTypeHTTP, cfg, maxRetries, + ) + + event := iSeedEvent(t, s.WebhookDB, s.WebhookID, `{"held":true}`) + d := iSeedDelivery(t, s.WebhookDB, event.ID, targetID, status) + + bodyStr := event.Body + + return iTask( + d, event, s.WebhookID, targetID, "held", cfg, + maxRetries, attemptNum, &bodyStr, + ), arrived +} + +// stopDuringAttempt waits until the receiver holds an attempt open, then +// stops the engine through its real stop hook. +func stopDuringAttempt( + t *testing.T, lc *recordingLifecycle, arrived <-chan struct{}, +) { + t.Helper() + + select { + case <-arrived: + case <-time.After(5 * time.Second): + t.Fatal("the delivery attempt never reached the receiver") + } + + require.NoError(t, lc.hooks[0].OnStop(context.Background())) +} + +// requireRequeued runs restart recovery on the stopped engine and +// requires it to queue task's attempt again on ch. +func requireRequeued( + t *testing.T, s iSetup, ch <-chan delivery.Task, task delivery.Task, +) { + t.Helper() + + s.Engine.ExportRecoverWebhookDeliveries( + context.Background(), s.WebhookID, + ) + + select { + case got := <-ch: + assert.Equal(t, task.DeliveryID, got.DeliveryID) + assert.Equal(t, task.AttemptNum, got.AttemptNum) + case <-time.After(5 * time.Second): + t.Fatal("restart recovery did not queue the delivery again") + } +} + +// TestEngine_StopLeavesFireAndForgetAttemptPending is the regression +// test for a stop that turned an attempt in flight into a terminal +// failure. Stopping the engine cancels the request; for a target with +// no retries that was the delivery's only attempt, recorded as failed +// with "context canceled", and nothing ever sent it again. The +// delivery must instead stay pending, with no result, for restart +// recovery to send. +func TestEngine_StopLeavesFireAndForgetAttemptPending(t *testing.T) { + t.Parallel() + + s := newISetup(t) + lc := startEngineViaHook(t, s.Engine) + + task, arrived := seedHeldDelivery( + t, s, 0, database.DeliveryStatusPending, 1, + ) + + s.Engine.Notify([]delivery.Task{task}) + stopDuringAttempt(t, lc, arrived) + + iAssertStatus( + t, s.WebhookDB, task.DeliveryID, database.DeliveryStatusPending, + ) + assert.Empty(t, iResults(t, s.WebhookDB, task.DeliveryID)) + + requireRequeued(t, s, s.Engine.ExportDeliveryCh(), task) +} + +// TestEngine_StopLeavesLastRetryAttemptRetrying is the same regression +// for a target with retries, on its last attempt: the cancelled attempt +// used to exhaust the retries and end the delivery failed. It must stay +// retrying with only its earlier attempt recorded, and restart recovery +// must send that last attempt again. +func TestEngine_StopLeavesLastRetryAttemptRetrying(t *testing.T) { + t.Parallel() + + const maxRetries = 2 + + s := newISetup(t) + lc := startEngineViaHook(t, s.Engine) + + task, arrived := seedHeldDelivery( + t, s, maxRetries, database.DeliveryStatusRetrying, maxRetries, + ) + iSeedFailedResult(t, s.WebhookDB, task.DeliveryID) + + s.Engine.ExportScheduleRetry(task, 0) + stopDuringAttempt(t, lc, arrived) + + iAssertStatus( + t, s.WebhookDB, task.DeliveryID, database.DeliveryStatusRetrying, + ) + assert.Len(t, iResults(t, s.WebhookDB, task.DeliveryID), 1) + + requireRequeued(t, s, s.Engine.ExportRetryCh(), task) +} + // deliverToArchive gives the setup's webhook a database target, // runs one delivery to it through the running engine, and returns // the target's ID and archive file path. The archive writer holds diff --git a/internal/delivery/target_http.go b/internal/delivery/target_http.go index 8733f31..d14e880 100644 --- a/internal/delivery/target_http.go +++ b/internal/delivery/target_http.go @@ -57,7 +57,11 @@ type httpCore struct { // recorded and no circuit breaker is consulted. A positive // maxRetries gates the attempt on the circuit breaker and // schedules a backed-off retry on failure. +// +// ctx is the context attempt sends its request with. Stopping the +// engine cancels it; see interruptedByStop. func (c *httpCore) deliver( + ctx context.Context, webhookDB *gorm.DB, d *database.Delivery, task *Task, @@ -66,23 +70,56 @@ func (c *httpCore) deliver( attempt func() attemptResult, ) { if maxRetries == 0 { - c.fireAndForget(webhookDB, d, attempt()) + c.fireAndForget(ctx, webhookDB, d, attempt()) return } c.withRetry( - webhookDB, d, task, sched, maxRetries, attempt, + ctx, webhookDB, d, task, sched, maxRetries, attempt, ) } +// interruptedByStop reports whether an attempt failed because the +// engine is stopping, which cancels the context its request was sent +// with. Such an attempt is not recorded: no DeliveryResult, no circuit +// breaker outcome, no status change. The delivery stays pending or +// retrying, and restart recovery sends it again on the next start. +// +// A target's own timeout leaves that context alone, so its attempt is +// still recorded as the failure it is. +func (c *httpCore) interruptedByStop( + ctx context.Context, + d *database.Delivery, + res attemptResult, +) bool { + if res.success || ctx.Err() == nil { + return false + } + + c.eng.log.Warn( + "delivery attempt interrupted by shutdown; not recorded, "+ + "it is sent again on the next start", + "delivery_id", d.ID, + "event_id", d.EventID, + "target_id", d.TargetID, + ) + + return true +} + func (c *httpCore) fireAndForget( + ctx context.Context, webhookDB *gorm.DB, d *database.Delivery, res attemptResult, ) { c.eng.observeAttempt(d.Target.Type, res.elapsed()) + if c.interruptedByStop(ctx, d, res) { + return + } + err := c.eng.recordResult( webhookDB, d, 1, res.success, res.statusCode, res.respBody, res.errMsg, @@ -110,6 +147,7 @@ func (c *httpCore) fireAndForget( } func (c *httpCore) withRetry( + ctx context.Context, webhookDB *gorm.DB, d *database.Delivery, task *Task, @@ -133,6 +171,10 @@ func (c *httpCore) withRetry( c.eng.observeAttempt(d.Target.Type, res.elapsed()) + if c.interruptedByStop(ctx, d, res) { + return + } + err := c.eng.recordResult( webhookDB, d, attemptNum, res.success, res.statusCode, res.respBody, res.errMsg, @@ -395,7 +437,7 @@ func (t *httpTarget) Deliver( } t.deliver( - webhookDB, d, task, sched, + ctx, webhookDB, d, task, sched, d.Target.MaxRetries, attempt, ) } diff --git a/internal/delivery/target_slack.go b/internal/delivery/target_slack.go index 3703b89..553b98e 100644 --- a/internal/delivery/target_slack.go +++ b/internal/delivery/target_slack.go @@ -82,7 +82,7 @@ func (t *slackTarget) Deliver( } t.deliver( - webhookDB, d, task, sched, + ctx, webhookDB, d, task, sched, d.Target.MaxRetries, attempt, ) }