|
| 1 | +# Skip stale events on `/subscribe` initial backfill |
| 2 | + |
| 3 | +**Status:** approved, implemented in PR #8 |
| 4 | +**Date:** 2026-05-09 |
| 5 | +**Affects:** `internal/subscribe`, `internal/server` (handler wiring) |
| 6 | + |
| 7 | +## Problem |
| 8 | + |
| 9 | +`/subscribe/<source>` opens an SSE stream that first replays every event since the caller's cursor and then tails live. The replay is byte-for-byte: each event carries the original provider headers (Render's Standard Webhooks set: `webhook-id`, `webhook-timestamp`, `webhook-signature`). |
| 10 | + |
| 11 | +Standard Webhooks consumers reject messages whose `webhook-timestamp` is older than a tolerance window (5 minutes by default; hardcoded in the `standard-webhooks` Ruby gem and several other implementations). Any event sitting in the relay's store longer than that window will be rejected by a verifying consumer when replayed — verification fails with `"Message timestamp too old"` even though the HMAC is valid. |
| 12 | + |
| 13 | +The conflict is structural. The relay durably stores webhooks; consumers verify timestamps strictly. As soon as automatic replay crosses the consumer's tolerance, every redelivery 401s. The Rails app behind `hooksctl forward` is exhibiting exactly this: |
| 14 | + |
| 15 | +```text |
| 16 | +Render webhook: signature verification failed: Message timestamp too old |
| 17 | +Filter chain halted as :verify_render_signature rendered or redirected |
| 18 | +Completed 401 Unauthorized |
| 19 | +``` |
| 20 | + |
| 21 | +## Decision |
| 22 | + |
| 23 | +Stop automatic catch-up of stale events. Manual replay is unaffected. |
| 24 | + |
| 25 | +A fresh `/subscribe/<source>` connection's **initial backfill** filters out events whose `provider_timestamp` is older than the source's effective `SkewWindow` (the per-source value from `hooks.yaml`, falling back to the verifier's 5-minute default — the same `effective_skew` ingest already enforces). The cursor advances past skipped events so reconnects do not re-evaluate them. **Live tail is unaffected** — drains triggered by the notifier or the keepalive ticker do not filter. |
| 26 | + |
| 27 | +This is a deliberate inversion of the relay's documented "replay anything missed while disconnected" promise. A consumer offline for longer than `effective_skew` will silently miss events that landed during the gap. The trade-off is accepted because: |
| 28 | + |
| 29 | +- Standard Webhooks consumers cannot accept stale messages without weakening their own replay-attack defense. |
| 30 | +- The relay still owns the events. Operators recover any specific delivery via the inspector's "Replay to listeners" action, `hooksctl replay`, or direct DB inspection — those paths are unchanged. |
| 31 | + |
| 32 | +(Note: `/audit` is **not** part of this recovery surface. `internal/audit` records operator-action volume, not webhook volume; skipped events do not produce audit rows.) |
| 33 | + |
| 34 | +## Behavior in detail |
| 35 | + |
| 36 | +### What changes |
| 37 | + |
| 38 | +`/subscribe/<source>` initial backfill — the first drain call before the live select loop — filters events by age. |
| 39 | + |
| 40 | +| Aspect | Behavior | |
| 41 | +|---|---| |
| 42 | +| Age field | `provider_timestamp` (the original webhook timestamp; matches what the consumer's signature-verifier will check). Filter is one-sided. | |
| 43 | +| Threshold | `effective_skew(source)` — the same value ingest uses. If `hooks.yaml` sets a non-zero `skew_window`, that value; otherwise the verifier's 5-minute default. **Zero is not a disable switch** — it propagates to the same default ingest already applies. | |
| 44 | +| Comparison | `now - provider_timestamp > effective_skew` skips; equal-to-window passes (matches `delta > skew` at `internal/sources/render.go:102`). | |
| 45 | +| Future timestamps | Always pass. A producer-clock-fast event that admitted at ingest (delta inside `[-skew, +skew]`) yields a negative `now - provider_timestamp` on backfill and emits unconditionally. | |
| 46 | +| Zero `provider_timestamp` | Always emit. Failing open: a future verifier that omits the field shouldn't silently swallow events. Render's verifier always populates it today (`internal/sources/render.go:84`), so this is a forward-compat guard. | |
| 47 | +| Cursor on skip | Advances to the skipped event's sequence so future reconnects with `?since=<seq>` start past it, and so a follow-up live drain does not re-emit the skipped events. **Load-bearing** — see TDD case 4. | |
| 48 | +| Skipped event in store | Unchanged. Remains queryable, replayable, and pruneable per existing retention rules. | |
| 49 | +| Logging | One `slog.Debug` line per skip: `slog.String("source", ...)`, `slog.Int64("seq", ...)`, `slog.String("delivery_id", ...)`, `slog.Duration("age", ...)`, `slog.Duration("skew_window", ...)`. No body, signature, or token bytes. | |
| 50 | + |
| 51 | +### What does not change |
| 52 | + |
| 53 | +- **Live tail.** Once initial backfill returns, drains triggered by `Notifier.Publish` or the keepalive ticker do **not** filter. Live ingest events are fresh by definition (they just passed the same `effective_skew` check at ingest), and the inspector "Replay to listeners" button uses `Notifier.Publish` to wake currently-connected SSE subscribers — that path stays open. The keepalive ticker is started after the initial drain returns (`internal/subscribe/handler.go:138` is after the initial drain call), so the ticker cannot fire mid-initial-drain. |
| 54 | + |
| 55 | +- **Push subscriptions.** `internal/push.Manager` workers and `Push.ReplayOne` operate on a separate code path with a separate cursor model. Untouched. |
| 56 | + |
| 57 | +- **Manual inspector replay** (`POST /events/{source}/{sequence}/replay`). Untouched. `Push.ReplayOne` continues to deliver to push subscribers regardless of age. The SSE side of manual replay relies on `Notifier.Publish(source, seq)` waking any current subscriber's live drain, which then `ReadSince(cursor, ...)`. That delivers the replayed event only when `subscriber.cursor < replayed.seq` — i.e. the subscriber hadn't seen it yet. If `subscriber.cursor >= replayed.seq` (the common case after a successful initial backfill), `ReadSince` returns nothing and the replay is a no-op for that SSE subscriber. This is existing behavior, not a regression. Edge case: a fresh SSE subscriber whose initial backfill is in flight when a stale event is manually replayed will skip it during backfill, since initial backfill filters. Operator recovers via reconnect (live tail) or `hooksctl replay`. |
| 58 | + |
| 59 | +- **`hooksctl forward`.** No client change. The server change is sufficient. No new CLI flags, no parsed log messages, no new wire fields. |
| 60 | + |
| 61 | +- **`?since=latest`.** Cursor jumps to `LatestSequence(source)`; initial drain reads zero events; the filter never fires. No behavior change. |
| 62 | + |
| 63 | +- **Ingest-time skew check.** Existing behavior in `internal/sources/render.go` is correct and unchanged. |
| 64 | + |
| 65 | +- **Retention / pruning.** Stale events live in the store under each source's configured retention; only their automatic redelivery is suppressed. |
| 66 | + |
| 67 | +## Architecture |
| 68 | + |
| 69 | +### Plumbing |
| 70 | + |
| 71 | +`subscribe.Handler` currently holds: |
| 72 | + |
| 73 | +```go |
| 74 | +type Handler struct { |
| 75 | + Sources map[string]bool |
| 76 | + // ... |
| 77 | +} |
| 78 | +``` |
| 79 | + |
| 80 | +It needs the per-source `effective_skew` to apply the filter. Replace with: |
| 81 | + |
| 82 | +```go |
| 83 | +type Handler struct { |
| 84 | + // Allowed sources mapped to the effective skew window for that source |
| 85 | + // (= configured SkewWindow, or verifier default if zero/unset). Source |
| 86 | + // membership is determined by key presence (`d, ok := h.Sources[s]`), |
| 87 | + // not by value — the value is a real duration that may legitimately |
| 88 | + // be any non-negative number including zero (zero treated as "use |
| 89 | + // verifier default" upstream of this map; the map only ever sees the |
| 90 | + // resolved effective value). |
| 91 | + Sources map[string]time.Duration |
| 92 | + Now func() time.Time // injected for tests; defaults to time.Now |
| 93 | + // ... |
| 94 | +} |
| 95 | +``` |
| 96 | + |
| 97 | +`internal/server.Build` constructs the handler and reads `hooks.yaml` into a `config.Config` whose `Source` entries carry `SkewWindow`. Build is responsible for resolving zero/unset to the verifier default before populating the map; `subscribe.Handler` itself never sees zero. (Resolving at the seam keeps the verifier-default constant in `internal/sources` where it already lives.) |
| 98 | + |
| 99 | +`subscribe.New`'s signature changes from `sources []string` to `sources map[string]time.Duration`. The package is internal and the production caller is a single line in `internal/server.Build`; tests are updated alongside the change. No backward-compat shim. |
| 100 | + |
| 101 | +### Filter location: split into `initialDrain` + `liveDrain` |
| 102 | + |
| 103 | +The current single `drain` becomes two functions. The type system, not a comment, enforces "live tail does not filter." |
| 104 | + |
| 105 | +```go |
| 106 | +// stream(): |
| 107 | +cursor, err := h.initialDrain(ctx, w, flusher, source, cursor) // filters by age |
| 108 | +... |
| 109 | +for { |
| 110 | + select { |
| 111 | + case <-ctx.Done(): ... |
| 112 | + case <-ch: |
| 113 | + cursor, err = h.liveDrain(ctx, w, flusher, source, cursor) // unfiltered |
| 114 | + case <-ticker.C: |
| 115 | + if _, err := io.WriteString(w, ": keepalive\n\n"); err != nil { ... } |
| 116 | + flusher.Flush() |
| 117 | + cursor, err = h.liveDrain(ctx, w, flusher, source, cursor) // unfiltered |
| 118 | + } |
| 119 | +} |
| 120 | +``` |
| 121 | + |
| 122 | +Both call a shared inner `readBatchAndEmit(...)` helper that does the SQL + SSE write loop; only `initialDrain` consults `h.Sources[source]` and `h.Now()` to compute `cutoff = h.Now().Add(-effectiveSkew)` per drain pass and skip events whose `provider_timestamp.Before(cutoff)`. `cutoff` is computed once per drain pass, not per event, so a single drain doesn't shift its decision midstream. |
| 123 | + |
| 124 | +When `initialDrain` skips an event, it advances `cursor = ev.Sequence` *without* writing to the wire. This is the load-bearing invariant: if cursor isn't advanced, subsequent live drains call `ReadSince(oldCursor)` and re-emit the skipped events on the *unfiltered* path. |
| 125 | + |
| 126 | +## TDD outline |
| 127 | + |
| 128 | +Tests live in `internal/subscribe/handler_test.go` alongside the existing suite. `Handler.Now` is the clock-control seam. |
| 129 | + |
| 130 | +1. **Initial backfill skips a stale event.** Seed one event with `provider_timestamp = now - 10m`, source effective skew = 5m. Connect; read SSE stream until live transition (e.g. observe a keepalive comment, or assert no payload after a short read). Assert no SSE message was emitted for the stale seq. |
| 131 | + |
| 132 | +2. **Initial backfill delivers a fresh event.** Same shape, `provider_timestamp = now - 1m`. Assert the event is emitted. |
| 133 | + |
| 134 | +3. **Mixed batch, only fresh emitted; idempotent on reconnect.** Seed `[stale, fresh, stale, fresh]`. Connect with `?since=0`; assert only the two fresh events are emitted. Reconnect with `?since=0`; assert the same two fresh events emit and the stale ones remain suppressed. |
| 135 | + |
| 136 | +4. **All-stale batch still advances cursor (regression guard for the load-bearing invariant).** Seed `[stale, stale]`. Connect; finish initial backfill; while still connected, ingest one fresh event and `Notifier.Publish(source, freshSeq)`. Assert the live drain emits **only** the fresh event — not the two stale ones. (If `initialDrain` failed to advance cursor on skip, `liveDrain`'s `ReadSince(0)` would re-pick the stale events on the unfiltered path.) |
| 137 | + |
| 138 | +5. **Live tail does not filter.** Connect, finish initial backfill, then write a stale-`provider_timestamp` event directly to the store and `Notifier.Publish(source, seq)`. Assert it is emitted via the live drain. (Models manual-replay-of-stale-event-to-live-SSE-subscriber.) |
| 139 | + |
| 140 | +6. **Boundary at exactly `effective_skew`.** Event with `provider_timestamp = now - effective_skew` (delta == skew): emit. Matches `delta > skew` at `internal/sources/render.go:102`. |
| 141 | + |
| 142 | +7. **Future-timestamp event passes.** `provider_timestamp = now + 1m`. Emit. Documents the one-sided filter. |
| 143 | + |
| 144 | +8. **Zero `provider_timestamp` passes.** Insert event with `time.Time{}`. Emit. Forward-compat guard. |
| 145 | + |
| 146 | +9. **Skip is observable.** Capture the handler's logger (use a `slog.Handler` test double); seed one stale event; assert one debug-level entry with attrs `source` (string), `seq` (int64), `delivery_id` (string), `age` (duration), `skew_window` (duration). Plaintext body / signature do not appear. |
| 147 | + |
| 148 | +10. **Unknown source still 404s after map shape change.** Direct construction with a `map[string]time.Duration{"render": 5m}`; request `/subscribe/stripe`. 404. (Renames the previous "default when missing" case and exercises the key-presence-not-value-zero membership rule.) |
| 149 | + |
| 150 | +Test fixtures: existing `setup()` helper updated to construct `Handler.Sources = map[string]time.Duration{"render": 5 * time.Minute}` so all pre-existing tests remain valid (events stamped with `time.Now()` are always within 5m by construction). |
| 151 | + |
| 152 | +## Out of scope |
| 153 | + |
| 154 | +- A query-parameter override (`?max_age=...`) for client-side tuning. The server policy is uniform. |
| 155 | +- Push-side mirror behavior. Push delivery already cursors per-subscription independently and has different replay semantics; revisiting it is its own design. |
| 156 | +- A new YAML field, env var, or CLI flag. The change reuses `SkewWindow`. |
| 157 | +- Notifying the consumer that events were skipped (SSE comment, response header, `hooksctl forward` log line). If demand emerges, the SSE comment surface is the lowest-impact follow-up; not designed here. |
| 158 | +- Audit emission for skipped events. `/audit` is operator-action volume, not webhook volume. |
| 159 | + |
| 160 | +## Risks and mitigations |
| 161 | + |
| 162 | +| Risk | Mitigation | |
| 163 | +|------|------------| |
| 164 | +| Operator surprise — events present in the store but never replayed. | Debug log on every skip (per spec, observable via `--dev` or `HOOKS_LOG_LEVEL=debug`). Inspector and `hooksctl replay` remain authoritative recovery surfaces. README, `docs/quickstart.md`, and CLAUDE.md updated to describe the new policy (see "Done when"). | |
| 165 | +| Cursor not advanced past skipped events → live drain re-emits stale events on the unfiltered path. | TDD case 4 (all-stale batch + live emit) exercises this directly. Implementation contract: every skip in `initialDrain` updates `cursor`. | |
| 166 | +| Test clock divergence — `time.Now` vs injected `Now`. | Single source of truth via `Handler.Now`; production wiring leaves it nil and falls back to `time.Now`. | |
| 167 | +| Operator sets `skew_window: "0s"` expecting no enforcement, gets 5m default everywhere. | Pre-existing ingest behavior; this spec preserves it consistently rather than introducing a second convention. Documented in the Threshold table row. | |
| 168 | + |
| 169 | +## Done when |
| 170 | + |
| 171 | +- All ten TDD cases pass. |
| 172 | +- Existing `internal/subscribe` tests remain green. |
| 173 | +- `make lint && make test` clean. |
| 174 | +- Documentation strings updated to reflect that automatic catch-up is bounded by the skew window: |
| 175 | + - `README.md:3` — "including replay of anything missed while disconnected" (qualify with the new policy). |
| 176 | + - `README.md:107` — "`forward` first replays any events you missed (none on first run), then tails live." |
| 177 | + - `docs/quickstart.md:138` — "replays anything missed since the last cursor, then tails live." |
| 178 | + - `internal/subscribe/handler.go` package doc — describe `initialDrain` filter behavior. |
| 179 | + - `CLAUDE.md` `internal/subscribe` bullet — one sentence on the new policy and where the threshold comes from. |
0 commit comments