diff --git a/README.md b/README.md index 80ed3db..91bc88e 100644 --- a/README.md +++ b/README.md @@ -1317,20 +1317,22 @@ webhooker solves this by acting as a durable intermediary: nothing is lost if downstream targets are unavailable. 2. **Guaranteed delivery** — Events are queued for delivery to each configured - target. Failed deliveries are retried with configurable backoff. Every - delivery attempt is logged with status codes, response bodies, and timing. + target. Failed deliveries are retried with configurable backoff. Delivery + attempts are logged with status codes, response bodies, and timing. **That guarantee is at-least-once, not exactly-once.** When a send reaches its target but the write recording that outcome fails, the delivery is 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 + and for delivery attempts. Prometheus metrics expose volume, latency, and error rates. The web UI provides real-time visibility into event flow. 4. **Fan-out** — A single incoming webhook can be delivered to multiple targets @@ -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,16 @@ 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. A worker + may still take tasks already queued before it exits. An `http` or `slack` + request in flight is cut off, and one started after the cancel fails at once. + Neither is recorded or counted in the metrics, and the next start sends the + delivery again (see Shutdown). A `database` or `log` task runs as normal. 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:** @@ -2317,7 +2323,7 @@ getting anywhere. | Metric | Type | Meaning | | -------------------------------------- | --------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `webhooker_events_received_total` | counter | Events received and durably stored. Compare against the delivery counters on one dashboard | -| `webhooker_delivery_attempts_total` | counter | Delivery attempts actually dispatched to a target. A delivery a circuit breaker refused is not one: it is counted as a retry instead, but only when the refusal moves it into `retrying` | +| `webhooker_delivery_attempts_total` | counter | Delivery attempts actually dispatched to a target. A delivery a circuit breaker refused is not one: it is counted as a retry instead, but only when the refusal moves it into `retrying`. An attempt cut off by a stop is not counted (see Shutdown) | | `webhooker_deliveries_succeeded_total` | counter | Deliveries that reached `delivered` | | `webhooker_deliveries_failed_total` | counter | Deliveries that failed terminally and will not be retried | | `webhooker_delivery_retries_total` | counter | Deliveries put back into `retrying` | @@ -3284,7 +3290,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 +3302,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 cancels a delivery request in flight. That attempt is not +recorded or counted in the metrics, 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..7db4258 100644 --- a/internal/delivery/engine_lifecycle_test.go +++ b/internal/delivery/engine_lifecycle_test.go @@ -2,11 +2,16 @@ package delivery_test import ( "context" + "io" + "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 +277,154 @@ 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) { + // The server notices the client hanging up only once the + // body has been read; until then r.Context() never ends. + _, _ = io.Copy(io.Discard, r.Body) + + 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 and no count in +// the attempt metrics, for restart recovery to send. +func TestEngine_StopLeavesFireAndForgetAttemptPending(t *testing.T) { + t.Parallel() + + s := newISetup(t) + reg := mIsolate(t, s) + 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)) + assert.Zero(t, mCounter(t, reg, mAttempts, mTypeHTTP)) + assert.Zero(t, mHTTPDurations(t, reg)) + + 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.go b/internal/delivery/target.go index 5ccb7d4..157565a 100644 --- a/internal/delivery/target.go +++ b/internal/delivery/target.go @@ -32,7 +32,8 @@ type Scheduler interface { // Engine.observeAttempt, alongside the DeliveryResult it records for // it. Deliver is also entered for attempts that never happen — an // open circuit breaker refuses one — so the count cannot be taken -// from around this call. +// from around this call. An attempt cut off by a stop is neither +// counted nor recorded. type Target interface { Deliver( ctx context.Context, diff --git a/internal/delivery/target_http.go b/internal/delivery/target_http.go index 8733f31..009c070 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,21 +70,55 @@ 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 is made +// with. Such an attempt is not recorded: no DeliveryResult, no circuit +// breaker outcome, no status change, no count in the attempt metrics. +// The delivery stays pending or retrying, and startup 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, ) { + if c.interruptedByStop(ctx, d, res) { + return + } + c.eng.observeAttempt(d.Target.Type, res.elapsed()) err := c.eng.recordResult( @@ -110,6 +148,7 @@ func (c *httpCore) fireAndForget( } func (c *httpCore) withRetry( + ctx context.Context, webhookDB *gorm.DB, d *database.Delivery, task *Task, @@ -131,6 +170,10 @@ func (c *httpCore) withRetry( res := attempt() + if c.interruptedByStop(ctx, d, res) { + return + } + c.eng.observeAttempt(d.Target.Type, res.elapsed()) err := c.eng.recordResult( @@ -395,7 +438,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, ) }