Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
468db3c814 |
@@ -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.
|
||||||
|
|||||||
@@ -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)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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(
|
||||||
|
|||||||
@@ -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)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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 {
|
||||||
|
|||||||
@@ -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)
|
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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,
|
||||||
|
|||||||
@@ -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(
|
||||||
|
|||||||
@@ -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,
|
||||||
|
|||||||
Reference in New Issue
Block a user