From b5ba48bc1b31d620cebb71c623c407004114b3e4 Mon Sep 17 00:00:00 2001 From: Thomas Rademaker Date: Mon, 20 Apr 2026 13:52:57 -0400 Subject: [PATCH] go-live firehose resilience --- appview/firehose.go | 115 +++++++++++++++++++++++- appview/firehose/RUNBOOK.md | 172 ++++++++++++++++++++++++++++++++++++ appview/firehose_test.go | 108 ++++++++++++++++++++++ cmd/effem-appview/main.go | 21 ++--- 4 files changed, 400 insertions(+), 16 deletions(-) create mode 100644 appview/firehose/RUNBOOK.md create mode 100644 appview/firehose_test.go diff --git a/appview/firehose.go b/appview/firehose.go index 9004511..10e33bd 100644 --- a/appview/firehose.go +++ b/appview/firehose.go @@ -23,12 +23,91 @@ import ( ) const ( - effemNSPrefix = "xyz.effem." - cursorPersistInterval = 5 * time.Second + effemNSPrefix = "xyz.effem." + + // cursorPersistInterval caps how many events we may re-process after a + // crash. At 1s we stay cheap (a single-row upsert) while keeping the + // post-crash replay window tiny. + cursorPersistInterval = 1 * time.Second cursorPersistTimeout = 10 * time.Second - cursorMaxFailures = 5 + + // cursorMaxFailures is consecutive failed writes before we tear the + // firehose down and let RunFirehoseConsumerWithRetry reconnect. 30 ticks + // at 1s is ~30 seconds of DB unresponsiveness — matches the previous + // 5-failure × 5-second tolerance window. + cursorMaxFailures = 30 + + // firehoseStaleWarnAfter is how long without events triggers the watchdog + // warning. The relay fires many events per second; any quiet gap is bad. + firehoseStaleWarnAfter = 60 * time.Second + + // firehoseStaleWarnInterval throttles repeat warnings while a stall + // persists so the log stream doesn't scream once per tick. + firehoseStaleWarnInterval = time.Minute + + // firehoseBackoffInitial and firehoseBackoffMax bound the reconnect loop. + firehoseBackoffInitial = time.Second + firehoseBackoffMax = 60 * time.Second + + // firehoseStableThreshold is the connection duration after which we treat + // a disconnect as a transient blip and reset backoff to the initial value. + firehoseStableThreshold = 30 * time.Second ) +// RunFirehoseConsumerWithRetry runs the firehose consumer in a reconnect loop +// with exponential backoff. It only returns when ctx is cancelled — every +// other exit reason is treated as a transient failure and retried. +// +// Rationale: the relay occasionally drops WebSocket connections for housekeeping, +// and Railway itself can momentarily interrupt networking during host moves. +// Before this wrapper a single disconnect would kill the whole process. +func (srv *Server) RunFirehoseConsumerWithRetry(ctx context.Context) { + backoff := firehoseBackoffInitial + + for { + start := time.Now() + err := srv.RunFirehoseConsumer(ctx) + if ctx.Err() != nil { + return + } + + // A connection that ran long enough before failing is a transient + // blip, not a broken upstream. Reset so the next blip reconnects fast. + if time.Since(start) > firehoseStableThreshold { + backoff = firehoseBackoffInitial + } + + srv.logger.Error("firehose disconnected, reconnecting", + "err", err, + "backoff", backoff, + "connection_duration_s", int(time.Since(start).Seconds()), + ) + + select { + case <-ctx.Done(): + return + case <-time.After(backoff): + } + backoff = nextFirehoseBackoff(backoff, firehoseBackoffMax) + } +} + +// nextFirehoseBackoff doubles the current backoff, clamping at max. Extracted +// so the math is unit-testable without real network IO. +func nextFirehoseBackoff(current, max time.Duration) time.Duration { + if current <= 0 { + return firehoseBackoffInitial + } + if current >= max { + return max + } + next := current * 2 + if next > max { + return max + } + return next +} + func (srv *Server) RunFirehoseConsumer(ctx context.Context) error { ctx, cancel := context.WithCancel(ctx) defer cancel() @@ -179,6 +258,9 @@ func (srv *Server) persistCursorLoop(ctx context.Context, cancel context.CancelF defer ticker.Stop() consecutiveFailures := 0 + // Throttle stall warnings so a long quiet period doesn't flood the log. + // Local to this goroutine so no synchronization needed. + var lastStaleWarn time.Time for { select { @@ -194,6 +276,9 @@ func (srv *Server) persistCursorLoop(ctx context.Context, cancel context.CancelF return case <-ticker.C: seq := atomic.LoadInt64(&srv.lastSeq) + + srv.maybeWarnFirehoseStale(seq, &lastStaleWarn) + if seq <= 0 { continue } @@ -215,6 +300,30 @@ func (srv *Server) persistCursorLoop(ctx context.Context, cancel context.CancelF } } +// maybeWarnFirehoseStale emits a throttled warning if the time since the +// last firehose event exceeds firehoseStaleWarnAfter. It does nothing before +// the first event is received (zero lastSeqTime means "we never saw +// anything," not "we've been starved for eternity") to avoid spamming on +// cold start. +func (srv *Server) maybeWarnFirehoseStale(seq int64, lastWarn *time.Time) { + t := loadLastSeqTime(&srv.lastSeqTime) + if t.IsZero() { + return + } + age := time.Since(t) + if age <= firehoseStaleWarnAfter { + return + } + if time.Since(*lastWarn) < firehoseStaleWarnInterval { + return + } + srv.logger.Warn("firehose stale: no events since last seq", + "last_seq", seq, + "age_s", int(age.Seconds()), + ) + *lastWarn = time.Now() +} + func (srv *Server) loadFirehoseCursor(ctx context.Context) (int64, error) { var cur database.FirehoseCursor err := srv.db.WithContext(ctx).First(&cur).Error diff --git a/appview/firehose/RUNBOOK.md b/appview/firehose/RUNBOOK.md new file mode 100644 index 0000000..0786fb4 --- /dev/null +++ b/appview/firehose/RUNBOOK.md @@ -0,0 +1,172 @@ +# Firehose Operations Runbook + +Operational playbooks for the AT Protocol firehose consumer that lives in +`appview/firehose.go`. Paired alerts in `appview/ALERTS.md`; metrics in +`appview/metrics/`. + +## Normal behavior + +- The consumer connects to the relay (default `wss://bsky.network`) and + subscribes via `com.atproto.sync.subscribeRepos`. +- Every processed event updates `lastSeq` and `lastSeqTime` on the `Server`. +- Every 1 s (`cursorPersistInterval`), the cursor goroutine upserts + `firehose_cursor.seq` so restarts replay at most ~1 s of events. +- `RunFirehoseConsumerWithRetry` catches WebSocket drops and reconnects with + exponential backoff from 1 s up to 60 s. A connection that held for longer + than 30 s resets the backoff on its next failure. +- `/_health` reports `firehose_up`, `firehose_seq`, `firehose_age_s`. +- Prometheus gauge `effem_firehose_lag_seconds` tracks the age of the + most recent event at scrape time. + +## Incidents + +### 1. Firehose stall (watchdog warnings in the log) + +**Symptom**: `firehose stale: no events since last seq` warnings, throttled +to once per minute. `effem_firehose_lag_seconds` climbs without resetting. + +1. **Confirm upstream health**. The relay could be down or congested. Hit it + directly: + + ```sh + websocat 'wss://bsky.network/xrpc/com.atproto.sync.subscribeRepos' | head -c 4096 + ``` + + If the relay is silent for several seconds, the problem is upstream — no + action on our end except to wait and keep watching. + +2. **Confirm our connection is healthy**. If `effem_firehose_connected == 1` + and we're still stale, the WebSocket is alive but no events are flowing. + That's rare and almost always upstream. + +3. **If the connection dropped** (`effem_firehose_connected == 0`): the + retry loop should reconnect automatically. Check logs for + `firehose disconnected, reconnecting` lines with escalating backoffs. + Nothing to do unless reconnects are failing repeatedly. + +### 2. Reconnect loop (repeated "firehose disconnected" errors) + +**Symptom**: sustained `firehose disconnected, reconnecting` lines, backoff +capping at 60 s. + +1. **Diagnose the dial error** in the log line's `err` field: + - `context canceled` during shutdown → expected, ignore. + - DNS / connection refused → network issue on the Railway edge. Check + outbound connectivity from a Railway shell. + - `FutureCursor` / `ConsumerTooSlow` / `InvalidCursor` → the cursor has + drifted out of the relay's backfill window. See **§3 cursor-expired**. + - Relay-side 5xx → upstream; wait and watch. + +2. **If the dial keeps failing at the network layer**: redeploy — sometimes + Railway networking gets stuck on a host and a fresh pod recovers + connectivity. + +### 3. Cursor-expired recovery + +**Symptom**: relay rejects `subscribeRepos` with `FutureCursor`, +`ConsumerTooSlow`, or a similar cursor-invalid error. Usually happens after +the AppView has been down longer than the relay's backfill window. + +The consumer will retry forever with the bad cursor. You must manually clear +it. + +**Procedure**: + +1. Stop the AppView (Railway → Deploy → Stop, or scale to 0). + +2. Connect to the AppView Postgres: + + ```sh + railway run --service effem-db -- psql "$EFFEM_DATABASE_URL" + ``` + +3. Clear the cursor row. Do **not** delete the table — just zero the sequence: + + ```sql + UPDATE firehose_cursor SET seq = 0; + -- if the row is missing entirely, the consumer will start from live on + -- its own; no action needed + ``` + +4. Restart the AppView. The consumer will connect without a cursor, which + tells the relay to start from "live." You'll have a **gap** in indexed + data covering the outage window. + +5. Watch the logs for `firehose consumer running` and the first + `effem record event` line. `/_health` should flip to `status: "ok"` + within a few seconds. + +**Gap backfill (optional)**: + +Indexed records can be recovered by replaying each active repo: + +``` +com.atproto.sync.listRepos → for each DID → com.atproto.sync.getRepo +→ unpack CAR → feed matching xyz.effem.* records through indexer.IndexRecord +``` + +The indexer upserts are idempotent, so backfilling records that also arrive +via the live firehose is safe. There's no production tool for this yet; +write a one-off script (`cmd/effem-backfill/`) if the gap matters. + +### 4. Cursor persistence failing repeatedly + +**Symptom**: repeated `failed to persist cursor` warnings, eventually +followed by `cursor persistence failed repeatedly, signaling firehose +shutdown` and the retry loop reconnecting. + +This means the database itself is unreachable or rejecting writes. The +firehose will keep cycling: reconnect → consume → fail to persist → +shutdown → reconnect, every ~30 s. HTTP reads are unaffected during this +cycle (different connection), but no new data lands in the DB. + +1. Check `/_health`: if `db: false`, the DB is down or the pool is + exhausted. Focus on the database, not the firehose. +2. Inspect Railway Postgres status and connection pool usage. +3. If the DB is full (disk quota), free space or scale up before restarting. + +### 5. Events flowing but stats counts look wrong + +**Symptom**: `/xrpc/xyz.effem.podcast.getStats` (or similar) returns stale +numbers despite the firehose seq advancing. + +1. Confirm the indexer isn't erroring: check + `effem_indexer_errors_total` by `{collection, action}`. Non-zero rates + indicate records are being dropped silently (warnings, not fatals). +2. Check the record path that's failing — common causes are CID mismatch + (relay sent a different CID than the CAR), record CBOR decode errors, or + a schema drift in the lexicon. +3. If a single collection is affected, look at the indexer handler for + that collection in `appview/indexer/.go`. + +## Knobs (constants in `firehose.go`) + +| Name | Default | Notes | +|---|---|---| +| `cursorPersistInterval` | 1 s | Trade-off: smaller = less replay, more DB writes. | +| `cursorPersistTimeout` | 10 s | Per-write timeout. | +| `cursorMaxFailures` | 30 | Consecutive failures before tearing the firehose down. At 1 s interval this is ~30 s of DB-unavailable tolerance. | +| `firehoseStaleWarnAfter` | 60 s | Warn when no events arrive for this long. | +| `firehoseStaleWarnInterval` | 60 s | Throttle between repeated stall warnings. | +| `firehoseBackoffInitial` | 1 s | First reconnect backoff after a disconnect. | +| `firehoseBackoffMax` | 60 s | Maximum reconnect backoff. | +| `firehoseStableThreshold` | 30 s | Connection duration beyond which a disconnect resets backoff. | + +## Multi-instance notes (post-beta) + +Running multiple AppView instances against the same relay is safe today — +every indexer upsert is idempotent (`ON CONFLICT DO UPDATE` or equivalent). +Each instance maintains its own `firehose_cursor` row and converges on the +same indexed state. + +Things **not** to do: +- Coordinate cursors across instances (more failure surface than it's worth). +- Shard the firehose by collection or repo (our index is already bounded by + the `effem.` namespace filter in `handleCommit`). + +Things to revisit before scaling out: +- The 1 s cursor persistence × N instances × 1 row = N writes/s on a single + `firehose_cursor` row's table. At small N this is fine; at large N + consider adding an `instance_id` column and primary-keying on it. +- `pi_cache` writes already dedupe via conflict clause, but the + write-amplification grows linearly with instance count. diff --git a/appview/firehose_test.go b/appview/firehose_test.go new file mode 100644 index 0000000..2d9faa4 --- /dev/null +++ b/appview/firehose_test.go @@ -0,0 +1,108 @@ +package appview + +import ( + "bytes" + "log/slog" + "strings" + "testing" + "time" +) + +func TestNextFirehoseBackoff(t *testing.T) { + t.Parallel() + max := 60 * time.Second + cases := []struct { + name string + current time.Duration + want time.Duration + }{ + {"initial double", time.Second, 2 * time.Second}, + {"mid doubling", 8 * time.Second, 16 * time.Second}, + {"next step clamps to max", 32 * time.Second, max}, + {"already at max", max, max}, + {"above max clamps down", 2 * max, max}, + {"zero snaps back to initial", 0, firehoseBackoffInitial}, + {"negative snaps back to initial", -time.Second, firehoseBackoffInitial}, + } + for _, tc := range cases { + tc := tc + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + if got := nextFirehoseBackoff(tc.current, max); got != tc.want { + t.Fatalf("want %s, got %s", tc.want, got) + } + }) + } +} + +func TestMaybeWarnFirehoseStaleSkipsBeforeFirstEvent(t *testing.T) { + t.Parallel() + srv := &Server{logger: slog.Default()} + var lastWarn time.Time + + srv.maybeWarnFirehoseStale(0, &lastWarn) + + if !lastWarn.IsZero() { + t.Fatalf("should not warn before first event, got lastWarn=%v", lastWarn) + } +} + +func TestMaybeWarnFirehoseStaleWarnsAndThrottles(t *testing.T) { + t.Parallel() + + var logBuf bytes.Buffer + logger := slog.New(slog.NewTextHandler(&logBuf, &slog.HandlerOptions{Level: slog.LevelWarn})) + srv := &Server{logger: logger} + + // Make lastSeqTime older than the stale threshold so the watchdog fires. + srv.lastSeqTime.Store(time.Now().Add(-(firehoseStaleWarnAfter + time.Second))) + var lastWarn time.Time + + srv.maybeWarnFirehoseStale(42, &lastWarn) + firstLog := logBuf.String() + if !strings.Contains(firstLog, "firehose stale") { + t.Fatalf("expected stale warning in log, got:\n%s", firstLog) + } + firstWarn := lastWarn + if firstWarn.IsZero() { + t.Fatal("lastWarn should be stamped after a warning") + } + + // A second call immediately after should be suppressed by the throttle. + logBuf.Reset() + srv.maybeWarnFirehoseStale(43, &lastWarn) + if logBuf.Len() != 0 { + t.Fatalf("expected throttle to suppress second warning, got:\n%s", logBuf.String()) + } + if !lastWarn.Equal(firstWarn) { + t.Fatalf("throttled call must not update lastWarn; was %v, now %v", firstWarn, lastWarn) + } + + // Pretend enough time has elapsed that the throttle releases. + lastWarn = time.Now().Add(-(firehoseStaleWarnInterval + time.Second)) + logBuf.Reset() + srv.maybeWarnFirehoseStale(44, &lastWarn) + if !strings.Contains(logBuf.String(), "firehose stale") { + t.Fatalf("expected warning after throttle released, got:\n%s", logBuf.String()) + } +} + +func TestMaybeWarnFirehoseStaleSilentWhenFresh(t *testing.T) { + t.Parallel() + + var logBuf bytes.Buffer + logger := slog.New(slog.NewTextHandler(&logBuf, &slog.HandlerOptions{Level: slog.LevelWarn})) + srv := &Server{logger: logger} + + srv.lastSeqTime.Store(time.Now()) + var lastWarn time.Time + + srv.maybeWarnFirehoseStale(7, &lastWarn) + + if logBuf.Len() != 0 { + t.Fatalf("should be silent when events are fresh, got:\n%s", logBuf.String()) + } + if !lastWarn.IsZero() { + t.Fatal("should not stamp lastWarn when silent") + } +} diff --git a/cmd/effem-appview/main.go b/cmd/effem-appview/main.go index bddc54d..45f6ea0 100644 --- a/cmd/effem-appview/main.go +++ b/cmd/effem-appview/main.go @@ -2,7 +2,6 @@ package main import ( "context" - "errors" "fmt" "log/slog" "os" @@ -150,26 +149,22 @@ func run(cctx *cli.Context) error { return fmt.Errorf("creating server: %w", err) } - firehoseErrCh := make(chan error, 1) + firehoseDone := make(chan struct{}) go func() { - err := srv.RunFirehoseConsumer(ctx) - if err != nil && !errors.Is(err, context.Canceled) { - slog.Error("firehose consumer stopped", "err", err) - firehoseErrCh <- err - } else { - firehoseErrCh <- nil - } + // RunFirehoseConsumerWithRetry owns reconnects and only returns when + // ctx is cancelled. If it exits, treat it as a signal to shut down + // the whole process (ctx is likely already done, but cancel() here + // is a belt-and-suspenders guard). + srv.RunFirehoseConsumerWithRetry(ctx) cancel() + close(firehoseDone) }() apiErr := srv.RunAPI(ctx) - firehoseErr := <-firehoseErrCh + <-firehoseDone if apiErr != nil { return apiErr } - if firehoseErr != nil { - return fmt.Errorf("firehose consumer failed: %w", firehoseErr) - } return nil } -- 2.51.2