Documentation
¶
Overview ¶
Package webhookdelivery is Layer 3 of the webhook pipeline (docs/design/webhook-delivery-river-migration.md): the River execution stage that POSTs a webhook_subscriber_deliveries (Layer 2) row to the customer endpoint and retries it. It replaces the hand-rolled SKIP-LOCKED claim + retry worker (internal/webhook/subscriber_retry.go) with a River Worker on the shared `webhook` queue; River owns claim/lease/retry/backoff, and Layer 2 becomes pure delivery state written here for the history API.
Delivery is at-least-once (the industry standard — Stripe/Svix/Postmark): River re-drives a crashed job, so an endpoint may receive a duplicate if the POST succeeds but the worker crashes before writing `delivered`. Consumers dedup on the event id in the signed payload.
Index ¶
- Constants
- type DeliverWorker
- type Deliverer
- type Jobs
- func (j *Jobs) EnqueueDelivery(ctx context.Context, pool *pgxpool.Pool, deliveryID string) error
- func (j *Jobs) EnqueueDeliveryTx(ctx context.Context, tx pgx.Tx, deliveryID string) (int64, error)
- func (j *Jobs) ReconcilePending(ctx context.Context, pool *pgxpool.Pool) (jobs.ReconcileResult, error)
- func (j *Jobs) RegisterJobs(w *river.Workers) []*river.PeriodicJob
- func (j *Jobs) SetEnqueuer(e jobs.Enqueuer)
- func (j *Jobs) WithMetrics(m Metrics) *Jobs
- type MaintenanceJobs
- type MaintenanceWorker
- type Metrics
- type ReconcileWorker
- type Sweeper
- type WebhookDeliverArgs
- type WebhookMaintenanceArgs
- type WebhookReader
- type WebhookReconcileArgs
Constants ¶
const MaxDeliveryAttempts = 8
MaxDeliveryAttempts is River's MaxAttempts for a delivery job — after this many failed attempts River discards and the worker marks Layer 2 'failed'.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type DeliverWorker ¶
type DeliverWorker struct {
river.WorkerDefaults[WebhookDeliverArgs]
// contains filtered or unexported fields
}
DeliverWorker delivers one Layer 2 row. Mirrors the legacy processOne: re-fetch the webhook per attempt (disabled → snooze, deleted → cancel), POST, write Layer 2 state. River owns retry/backoff via NextRetry below.
func NewDeliverWorker ¶
func NewDeliverWorker(subStore *webhook.SubscriberStore, deliverer Deliverer, webhooks WebhookReader) *DeliverWorker
NewDeliverWorker constructs the worker. Used by the Registrar and by tests.
func (*DeliverWorker) NextRetry ¶
func (w *DeliverWorker) NextRetry(job *river.Job[WebhookDeliverArgs]) time.Time
NextRetry overrides River's client-wide policy for webhook jobs only, returning the exact 29h21m envelope. job.Attempt is the just-failed attempt (1-based); River discards at MaxAttempts so this is called for attempts 1..MaxDeliveryAttempts-1.
func (*DeliverWorker) WithMetrics ¶ added in v1.4.1
func (w *DeliverWorker) WithMetrics(m Metrics) *DeliverWorker
WithMetrics swaps in a metrics backend. Nil-safe: unset (or nil) means no emission, so tests don't have to wire anything.
func (*DeliverWorker) Work ¶
func (w *DeliverWorker) Work(ctx context.Context, job *river.Job[WebhookDeliverArgs]) error
Work intentionally has no Timeout() override — a single HTTP POST (bounded by the deliverer's own client timeout) fits River's 60s default JobTimeout.
type Deliverer ¶
type Deliverer interface {
Deliver(ctx context.Context, url string, body []byte, secret, secretPrev, eventType, schemaVersion string) webhook.DeliveryOutcome
}
Deliverer is the POST surface. *webhook.SubscriberDeliverer satisfies it.
type Jobs ¶
type Jobs struct {
// contains filtered or unexported fields
}
Jobs is the webhook-delivery integration on the shared River client: a jobs.Registrar (contributes DeliverWorker + the reconcile periodic) plus the transactional enqueue entry point the outbox drain + redelivery API call. The shared client is injected via SetEnqueuer after jobs.New builds it (two-phase wiring, same as senderidentity).
func NewJobs ¶
func NewJobs(subStore *webhook.SubscriberStore, deliverer Deliverer, webhooks WebhookReader, pool *pgxpool.Pool) *Jobs
NewJobs builds the integration with its dependencies (no client yet). pool backs the periodic reconciler's scan.
func (*Jobs) EnqueueDelivery ¶
EnqueueDelivery enqueues a River delivery job for an ALREADY-INSERTED pending Layer 2 row, in its own transaction, and stamps job_id. This is for the direct- insert API surfaces that bypass the outbox drain — the /test webhook endpoint and the redelivery API — which create a subscriber_deliveries row targeting a single webhook. Without this, those rows have no River job and (post SubscriberRetryWorker deletion) would never deliver. Idempotent per row via the job_id IS NULL guard under a row lock.
func (*Jobs) EnqueueDeliveryTx ¶
EnqueueDeliveryTx enqueues a delivery job WITHIN the caller's transaction — the outbox pattern: the Layer 2 row insert and this job commit together, so a delivery record can never exist without a job (or vice versa). The caller only calls this when the Layer 2 insert actually inserted a row (dedup ON CONFLICT returned an id), so a deduped event enqueues nothing. Returns the river_job id for the caller to stamp on the Layer 2 row's job_id.
func (*Jobs) ReconcilePending ¶
func (j *Jobs) ReconcilePending(ctx context.Context, pool *pgxpool.Pool) (jobs.ReconcileResult, error)
ReconcilePending enqueues a River delivery job for every pending Layer 2 row that has no live job: job_id IS NULL, or (RescueDeadJobs) stamped with a river_job that is terminal or already pruned. The dead-job arm closes the CW-2 strand: when the final delivery attempt's terminal 'failed' write is lost (a sustained DB outage in exactly that window — the CRITICAL log in DeliverWorker.Work), River discards the job and the row sat 'pending' with a dead job_id, invisible to an IS-NULL-only reconciler, until its TTL. Re-driving is safe: delivery is at-least-once and the DeliverWorker no-ops on rows that already terminalized. The rescue arm is age-gated (RescueWhere, rescueQuietFor) to rows quiet for longer than the full retry envelope, so the per-tick scan never churns the live in-flight set. It runs BOTH at startup (the one-shot cutover from the legacy queue) AND on a live schedule (ReconcileWorker) so a stranded row — from the separate-tx /test/redelivery enqueue paths, an outbox-drain crash window, or a lost terminal write — is re-driven within reconcileInterval (IS-NULL arm) or rescueQuietFor + reconcileInterval (dead-job arm) rather than only on the next restart. A re-run never double-enqueues on the IS-NULL arm; the dead-job arm is at-least-once under concurrent reconcilers (see jobs.ReconcilePending's EvalPlanQual note). Returns the per-arm counts.
func (*Jobs) RegisterJobs ¶
func (j *Jobs) RegisterJobs(w *river.Workers) []*river.PeriodicJob
RegisterJobs adds the DeliverWorker + the reconcile worker, and returns the reconcile periodic. Implements jobs.Registrar.
func (*Jobs) SetEnqueuer ¶
SetEnqueuer injects the shared client so EnqueueDeliveryTx can insert jobs.
func (*Jobs) WithMetrics ¶ added in v1.4.1
WithMetrics wires the observability backend the DeliverWorker emits the webhook-attempt SLI on. Nil-safe; call before RegisterJobs.
type MaintenanceJobs ¶
type MaintenanceJobs struct {
// contains filtered or unexported fields
}
MaintenanceJobs is the jobs.Registrar for the webhook janitor: it contributes the MaintenanceWorker and a periodic that fires it on QueueMaintenance. No enqueuer needed — the schedule is the only trigger.
func NewMaintenanceJobs ¶
func NewMaintenanceJobs(sweeper Sweeper) *MaintenanceJobs
NewMaintenanceJobs builds the registrar around a Sweeper (the AutoDisableWorker).
func (*MaintenanceJobs) RegisterJobs ¶
func (m *MaintenanceJobs) RegisterJobs(w *river.Workers) []*river.PeriodicJob
RegisterJobs adds the maintenance worker + its periodic schedule. Mirrors senderidentity's reaper: routed to QueueMaintenance, no UniqueOpts (River's periodic scheduler already inserts at most one per interval and a completed run must not dedup-block the next), RunOnStart:false (first sweep after one interval).
type MaintenanceWorker ¶
type MaintenanceWorker struct {
river.WorkerDefaults[WebhookMaintenanceArgs]
// contains filtered or unexported fields
}
MaintenanceWorker runs the janitor passes once per scheduled job. Errors are swallowed inside Tick (logged there); Work returns nil so a transient DB blip never spins River's retry machinery for a best-effort idempotent sweep — the next interval picks it up.
func NewMaintenanceWorker ¶
func NewMaintenanceWorker(sweeper Sweeper) *MaintenanceWorker
NewMaintenanceWorker builds the worker around a Sweeper. Exported so tests can drive Work directly (RegisterJobs builds an identical one for the client).
func (*MaintenanceWorker) Work ¶
func (w *MaintenanceWorker) Work(ctx context.Context, _ *river.Job[WebhookMaintenanceArgs]) error
type Metrics ¶ added in v1.4.1
type Metrics interface {
// WebhookAttempt records one delivery attempt. outcome ∈ {delivered,
// retryable_failure, exhausted, webhook_deleted, skipped_disabled};
// statusClass is "1xx".."5xx" or "none" (no HTTP response).
// A negative seconds means "count the attempt, record no duration
// sample" — used for outcomes with no HTTP POST (webhook_deleted,
// skipped_disabled), which would otherwise drag the duration
// quantiles toward zero.
WebhookAttempt(outcome, statusClass string, seconds float64)
// WebhookTerminal records a delivery only after this worker successfully
// transitions its row to a terminal state. outcome ∈ {delivered,
// e2a_failure, endpoint_failure, excluded}; scope ∈ {initial, replay,
// test, unknown}.
WebhookTerminal(outcome, scope string, count int)
// WebhookDeliveryRescued counts delivery rows the reconciler's dead-job
// arm re-drove (fresh job replacing a terminal/pruned one). A climbing
// rate is the poison-row signal.
WebhookDeliveryRescued(count int)
// WebhookFirstAttemptLatency records event→first-attempt latency for
// one subscriber delivery (attempt start − webhook_events.created_at).
// Observed only on a first-delivery row's FIRST HTTP attempt — retries,
// replays, and the no-POST outcomes never observe.
WebhookFirstAttemptLatency(seconds float64)
}
Metrics is the narrow slice of telemetry.Metrics the delivery worker emits (same pattern as internal/janitor.Metrics). Injectable so tests use a fake recorder; satisfied by any telemetry backend.
type ReconcileWorker ¶
type ReconcileWorker struct {
river.WorkerDefaults[WebhookReconcileArgs]
// contains filtered or unexported fields
}
ReconcileWorker re-enqueues any pending delivery row with no live River job (never enqueued, or its job is terminal/pruned). It is the LIVE backstop for the separate-tx enqueue paths (/test, redelivery), any outbox-drain crash window, and a lost final-attempt terminal write — turning "recovered only on restart" (or never) into "recovered within reconcileInterval" (dead-job rescues: within rescueQuietFor + reconcileInterval). Never double-enqueues on the IS-NULL arm; the dead-job rescue arm is at-least-once under concurrent reconcilers (see jobs.ReconcilePending's EvalPlanQual note) — a duplicate POST is within the delivery contract.
func (*ReconcileWorker) Work ¶
func (w *ReconcileWorker) Work(ctx context.Context, _ *river.Job[WebhookReconcileArgs]) error
type Sweeper ¶
Sweeper runs the webhook maintenance passes (auto-disable chronically-failing webhooks + clear expired signing_secret_prev rows). Satisfied by *webhook.AutoDisableWorker — the River worker just drives its Tick on a schedule instead of a hand-rolled time.Ticker.
type WebhookDeliverArgs ¶
type WebhookDeliverArgs struct {
DeliveryID string `json:"delivery_id"`
}
WebhookDeliverArgs carries only the Layer 2 delivery id — the worker reads the payload + webhook from the DB (keeps river_job rows tiny; Layer 2 is source of truth).
func (WebhookDeliverArgs) Kind ¶
func (WebhookDeliverArgs) Kind() string
type WebhookMaintenanceArgs ¶
type WebhookMaintenanceArgs struct{}
WebhookMaintenanceArgs drives the periodic webhook janitor. No fields — the worker sweeps the whole table each run.
func (WebhookMaintenanceArgs) Kind ¶
func (WebhookMaintenanceArgs) Kind() string
type WebhookReader ¶
type WebhookReader interface {
GetWebhookByIDInternal(ctx context.Context, webhookID string) (*identity.Webhook, error)
}
WebhookReader is the narrow webhook-lookup surface the worker needs. *identity.Store satisfies it.
type WebhookReconcileArgs ¶
type WebhookReconcileArgs struct{}
WebhookReconcileArgs drives the periodic stranded-delivery reconciler.
func (WebhookReconcileArgs) Kind ¶
func (WebhookReconcileArgs) Kind() string
Source Files