1 Commits
Author SHA1 Message Date
sneak 468db3c814 Leave a delivery cut off by a stop for the next start (closes #527)
check / check (push) Canceled after 0s
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
2026-10-15 08:06:13 +00:00
10 changed files with 32 additions and 86 deletions
+14 -16
View File
@@ -1317,8 +1317,8 @@ webhooker solves this by acting as a durable intermediary:
nothing is lost if downstream targets are unavailable. nothing is lost if downstream targets are unavailable.
2. **Guaranteed delivery** — Events are queued for delivery to each configured 2. **Guaranteed delivery** — Events are queued for delivery to each configured
target. Failed deliveries are retried with configurable backoff. Delivery target. Failed deliveries are retried with configurable backoff. Every
attempts are logged with status codes, response bodies, and timing. delivery attempt is logged with status codes, response bodies, and timing.
**That guarantee is at-least-once, not exactly-once.** When a send reaches **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 its target but the write recording that outcome fails, the delivery is
@@ -1332,7 +1332,7 @@ webhooker solves this by acting as a durable intermediary:
payload itself carries. payload itself carries.
3. **Observability** — Full request/response logging for every webhook received 3. **Observability** — Full request/response logging for every webhook received
and for delivery attempts. Prometheus metrics expose volume, latency, and and every delivery attempted. Prometheus metrics expose volume, latency, and
error rates. The web UI provides real-time visibility into event flow. 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 4. **Fan-out** — A single incoming webhook can be delivered to multiple targets
@@ -2208,16 +2208,14 @@ This means:
buffer rather than spawning more goroutines. buffer rather than spawning more goroutines.
- **Independent results** — each worker records its own delivery result in the - **Independent results** — each worker records its own delivery result in the
per-webhook database without coordination. per-webhook database without coordination.
- **Graceful shutdown** — the stop hook cancels the workers' context. A worker - **Graceful shutdown** — the stop hook cancels the workers' context, and each
may still take tasks already queued before it exits. An `http` or `slack` worker exits once its current task returns. Cancelling cuts off an `http` or
request in flight is cut off, and one started after the cancel fails at once. `slack` request still in flight; that attempt is not recorded, and the next
Neither is recorded or counted in the metrics, and the next start sends the start sends the delivery again (see Shutdown). The stop hook waits for the
delivery again (see Shutdown). A `database` or `log` task runs as normal. The pool via `lifecycle.WaitForShutdown`, which bounds that wait by fx's stop
stop hook waits for the pool via `lifecycle.WaitForShutdown`, which bounds timeout rather than blocking forever on a wedged worker. On timeout it logs at
that wait by fx's stop timeout rather than blocking forever on a wedged `ERROR` and returns an error, and the goroutines that did not finish are still
worker. On timeout it logs at `ERROR` and returns an error, and the goroutines running — an unclean shutdown is reported rather than hidden.
that did not finish are still running — an unclean shutdown is reported rather
than hidden.
**Recovery paths:** **Recovery paths:**
@@ -2323,7 +2321,7 @@ getting anywhere.
| Metric | Type | Meaning | | Metric | Type | Meaning |
| -------------------------------------- | --------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | -------------------------------------- | --------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `webhooker_events_received_total` | counter | Events received and durably stored. Compare against the delivery counters on one dashboard | | `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`. An attempt cut off by a stop is not counted (see Shutdown) | | `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_deliveries_succeeded_total` | counter | Deliveries that reached `delivered` | | `webhooker_deliveries_succeeded_total` | counter | Deliveries that reached `delivered` |
| `webhooker_deliveries_failed_total` | counter | Deliveries that failed terminally and will not be retried | | `webhooker_deliveries_failed_total` | counter | Deliveries that failed terminally and will not be retried |
| `webhooker_delivery_retries_total` | counter | Deliveries put back into `retrying` | | `webhooker_delivery_retries_total` | counter | Deliveries put back into `retrying` |
@@ -3302,8 +3300,8 @@ 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 server hook is ever entered. The hooks after the server are microsecond-scale in
normal operation. normal operation.
The delivery engine cancels a delivery request in flight. That attempt is not The delivery engine does not wait for a delivery request in flight: it cancels
recorded or counted in the metrics, and the delivery keeps its `pending` or 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 `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 Recovery paths under Bounded Worker Pool). If the receiver already had it, it
receives it twice: the at-least-once duplicate described under Rationale. receives it twice: the at-least-once duplicate described under Rationale.
+2 -13
View File
@@ -32,9 +32,8 @@ func targetTotals(
// TestUpdateDeliveryStatus_FinishTimeAndTargetTotals pins what a status // TestUpdateDeliveryStatus_FinishTimeAndTargetTotals pins what a status
// write records for the webhook page's statistics: the time a delivery // write records for the webhook page's statistics: the time a delivery
// finished, set only when it becomes delivered or failed and stored in // finished, set only when it becomes delivered or failed, and one more
// the same form as created_at, and one more on its target's delivered // on its target's delivered or failed total.
// or failed total.
func TestUpdateDeliveryStatus_FinishTimeAndTargetTotals(t *testing.T) { func TestUpdateDeliveryStatus_FinishTimeAndTargetTotals(t *testing.T) {
t.Parallel() t.Parallel()
@@ -76,16 +75,6 @@ func TestUpdateDeliveryStatus_FinishTimeAndTargetTotals(t *testing.T) {
if tt.finished { if tt.finished {
require.NotNil(t, stored.FinishedAt) require.NotNil(t, stored.FinishedAt)
assert.False(t, stored.FinishedAt.Before(before)) assert.False(t, stored.FinishedAt.Before(before))
var createdAt, finishedAt string
require.NoError(t, db.Raw(
`SELECT CAST(created_at AS TEXT),
CAST(finished_at AS TEXT)
FROM deliveries WHERE id = ?`, d.ID,
).Row().Scan(&createdAt, &finishedAt))
assertStoredTime(t, createdAt)
assertStoredTime(t, finishedAt)
} else { } else {
assert.Nil(t, stored.FinishedAt) assert.Nil(t, stored.FinishedAt)
} }
+3 -7
View File
@@ -1701,11 +1701,8 @@ func writeDeliveryStatus(
database.DeliveryStatusFailed, database.DeliveryStatusFailed,
}). }).
Updates(map[string]any{ Updates(map[string]any{
"status": status, "status": status,
// GORM's clock, the one that stamps created_at. A bare "finished_at": time.Now(),
// time.Now() carries a monotonic clock reading, which the
// driver would store as an " m=+…" suffix.
"finished_at": tx.NowFunc(),
}) })
if res.Error != nil || res.RowsAffected == 0 { if res.Error != nil || res.RowsAffected == 0 {
return res.Error return res.Error
@@ -1909,13 +1906,12 @@ func (e *Engine) takeForRedispatch(
return false return false
} }
// GORM's clock, as writeDeliveryStatus explains.
res := webhookDB. res := webhookDB.
Model(&database.Delivery{}). Model(&database.Delivery{}).
Where( Where(
"id = ? AND status = ?", deliveryID, eligible, "id = ? AND status = ?", deliveryID, eligible,
). ).
UpdateColumn("updated_at", webhookDB.NowFunc()) UpdateColumn("updated_at", time.Now())
if res.Error != nil { if res.Error != nil {
e.log.Error( e.log.Error(
+2 -5
View File
@@ -370,13 +370,12 @@ func requireRequeued(
// failure. Stopping the engine cancels the request; for a target with // failure. Stopping the engine cancels the request; for a target with
// no retries that was the delivery's only attempt, recorded as failed // no retries that was the delivery's only attempt, recorded as failed
// with "context canceled", and nothing ever sent it again. The // with "context canceled", and nothing ever sent it again. The
// delivery must instead stay pending, with no result and no count in // delivery must instead stay pending, with no result, for restart
// the attempt metrics, for restart recovery to send. // recovery to send.
func TestEngine_StopLeavesFireAndForgetAttemptPending(t *testing.T) { func TestEngine_StopLeavesFireAndForgetAttemptPending(t *testing.T) {
t.Parallel() t.Parallel()
s := newISetup(t) s := newISetup(t)
reg := mIsolate(t, s)
lc := startEngineViaHook(t, s.Engine) lc := startEngineViaHook(t, s.Engine)
task, arrived := seedHeldDelivery( task, arrived := seedHeldDelivery(
@@ -390,8 +389,6 @@ func TestEngine_StopLeavesFireAndForgetAttemptPending(t *testing.T) {
t, s.WebhookDB, task.DeliveryID, database.DeliveryStatusPending, t, s.WebhookDB, task.DeliveryID, database.DeliveryStatusPending,
) )
assert.Empty(t, iResults(t, s.WebhookDB, task.DeliveryID)) 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) requireRequeued(t, s, s.Engine.ExportDeliveryCh(), task)
} }
-12
View File
@@ -117,18 +117,6 @@ func seedDelivery(
return d return d
} }
// assertStoredTime asserts that text, a timestamp as SQLite holds it,
// is in the form GORM stores created_at in: the layout of
// time.Time.String, with no " m=+…" monotonic clock reading after it.
func assertStoredTime(t *testing.T, text string) {
t.Helper()
_, err := time.Parse(
"2006-01-02 15:04:05.999999999 -0700 MST", text,
)
assert.NoError(t, err, "stored as %q", text)
}
// httpDeliveryFixture holds the task and delivery needed // httpDeliveryFixture holds the task and delivery needed
// for HTTP delivery tests. // for HTTP delivery tests.
type httpDeliveryFixture struct { type httpDeliveryFixture struct {
+1 -11
View File
@@ -462,8 +462,7 @@ func TestScheduledRetryIsNotSweptDuringBackoff(t *testing.T) {
// TestRedispatchStampsTheRow pins the cadence control: a stranded // TestRedispatchStampsTheRow pins the cadence control: a stranded
// delivery that has just been handed out is not selected again by the // delivery that has just been handed out is not selected again by the
// next tick a minute later. The stamp is stored in the same form as // next tick a minute later.
// created_at.
func TestRedispatchStampsTheRow(t *testing.T) { func TestRedispatchStampsTheRow(t *testing.T) {
t.Parallel() t.Parallel()
@@ -494,13 +493,4 @@ func TestRedispatchStampsTheRow(t *testing.T) {
"a re-dispatched delivery must be stamped so the "+ "a re-dispatched delivery must be stamped so the "+
"next tick does not select it again", "next tick does not select it again",
) )
var createdAt, updatedAt string
require.NoError(t, s.WebhookDB.Raw(
`SELECT CAST(created_at AS TEXT), CAST(updated_at AS TEXT)
FROM deliveries WHERE id = ?`, d.ID,
).Row().Scan(&createdAt, &updatedAt))
assertStoredTime(t, createdAt)
assertStoredTime(t, updatedAt)
} }
+1 -2
View File
@@ -32,8 +32,7 @@ type Scheduler interface {
// Engine.observeAttempt, alongside the DeliveryResult it records for // Engine.observeAttempt, alongside the DeliveryResult it records for
// it. Deliver is also entered for attempts that never happen — an // it. Deliver is also entered for attempts that never happen — an
// open circuit breaker refuses one — so the count cannot be taken // open circuit breaker refuses one — so the count cannot be taken
// from around this call. An attempt cut off by a stop is neither // from around this call.
// counted nor recorded.
type Target interface { type Target interface {
Deliver( Deliver(
ctx context.Context, ctx context.Context,
+1 -5
View File
@@ -277,11 +277,7 @@ func (w *archiveWriter) write(
} }
} }
// GORM's clock, the one that stamps created_at in the other row.ArchivedAt = time.Now()
// databases, so this is stored in the same form. A bare time.Now()
// carries a monotonic clock reading, which the driver would store
// as an " m=+…" suffix.
row.ArchivedAt = w.db.NowFunc()
err := w.db.Create(&row).Error err := w.db.Create(&row).Error
if err != nil { if err != nil {
@@ -150,13 +150,6 @@ func TestArchiveWriter_WritesRow(t *testing.T) {
assert.Equal(t, row.ContentType, got[0].ContentType) assert.Equal(t, row.ContentType, got[0].ContentType)
assert.JSONEq(t, `{"hello":"world"}`, got[0].Body) assert.JSONEq(t, `{"hello":"world"}`, got[0].Body)
assert.False(t, got[0].ArchivedAt.IsZero()) assert.False(t, got[0].ArchivedAt.IsZero())
var archivedAt string
require.NoError(t, w.DB().Raw(
"SELECT CAST(archived_at AS TEXT) FROM archived_events",
).Row().Scan(&archivedAt))
assertStoredTime(t, archivedAt)
} }
func TestArchiveWriter_RecreatesAfterRemoval( func TestArchiveWriter_RecreatesAfterRemoval(
+8 -8
View File
@@ -81,11 +81,11 @@ func (c *httpCore) deliver(
} }
// interruptedByStop reports whether an attempt failed because the // interruptedByStop reports whether an attempt failed because the
// engine is stopping, which cancels the context its request is made // engine is stopping, which cancels the context its request was sent
// with. Such an attempt is not recorded: no DeliveryResult, no circuit // with. Such an attempt is not recorded: no DeliveryResult, no circuit
// breaker outcome, no status change, no count in the attempt metrics. // breaker outcome, no status change. The delivery stays pending or
// The delivery stays pending or retrying, and startup recovery sends // retrying, and startup recovery sends it again on the next start.
// it again on the next start. // The request did go out, so the attempt metrics still count it.
// //
// A target's own timeout leaves that context alone, so its attempt is // A target's own timeout leaves that context alone, so its attempt is
// still recorded as the failure it is. // still recorded as the failure it is.
@@ -115,12 +115,12 @@ func (c *httpCore) fireAndForget(
d *database.Delivery, d *database.Delivery,
res attemptResult, res attemptResult,
) { ) {
c.eng.observeAttempt(d.Target.Type, res.elapsed())
if c.interruptedByStop(ctx, d, res) { if c.interruptedByStop(ctx, d, res) {
return return
} }
c.eng.observeAttempt(d.Target.Type, res.elapsed())
err := c.eng.recordResult( err := c.eng.recordResult(
webhookDB, d, 1, res.success, webhookDB, d, 1, res.success,
res.statusCode, res.respBody, res.errMsg, res.statusCode, res.respBody, res.errMsg,
@@ -170,12 +170,12 @@ func (c *httpCore) withRetry(
res := attempt() res := attempt()
c.eng.observeAttempt(d.Target.Type, res.elapsed())
if c.interruptedByStop(ctx, d, res) { if c.interruptedByStop(ctx, d, res) {
return return
} }
c.eng.observeAttempt(d.Target.Type, res.elapsed())
err := c.eng.recordResult( err := c.eng.recordResult(
webhookDB, d, attemptNum, res.success, webhookDB, d, attemptNum, res.success,
res.statusCode, res.respBody, res.errMsg, res.statusCode, res.respBody, res.errMsg,