wisp.place-monorepo
Monorepo for wisp.place. A static site hosting service built on top of the AT Protocol. wisp.place
Star 104
Fork 7
Watch

perf(firehose): scan storage stats only on the leader, and bound the frame queue #88

Merged nekomimi.pet
targeting main from perf/firehose-standby-and-queue
Merged opened by and assigned to nekomimi.pet
targeting main from perf/firehose-standby-and-queue

Two small firehose fixes, both about memory and load on edges (phase 0 of the Rust-rewrite plan).

Problem #

1. Every standby edge lists the whole bucket every hour. main() calls startStorageStatsRefresh() right after the health server starts (apps/firehose-service/src/index.ts:603 on main), before leadership is known (:606). The refresh is a full paginated S3 ListObjectsV2 scan, once at start and then every STORAGE_STATS_REFRESH_INTERVAL_MS = 1 h (src/lib/storage-stats-cache.ts:7). Every edge runs a firehose container but only the leader ingests, so every standby does 1 scan at boot + 24/day for capacity numbers nobody reads off a standby (deploy/observability/probe.py doesn't read storage at all; health readiness ignores it, index.ts:183-191).

2. The bun-firehose frame queue has no bound. The WebSocket message handler pushes every frame into MessageQueue (packages/@wispplace/bun-firehose/src/subscription.ts:152-158, src/queue.ts:6-31) no matter how far behind the consumer is. A browser-style WebSocket can't pause reads, so whenever the consumer awaits real I/O (DID key resolution, a full cursor tracker, slow handlers) during catch-up/replay, the relay's whole backlog piles up in memory. Measured below: the queue held 98.5k of 100k replayed frames (+194 MiB RSS), and 295k of 300k (+525–549 MiB RSS), so it grows with the backlog.

Solution #

1. Start the storage-stat refresh inside startProtectedWork, right after startFirehose, so it runs only once this instance is the active worker (supervisor acquired, in-process election win, or single-instance mode). It stops on in-process step-down (stopStorageStatsRefresh() at the top of the release callback) and on shutdown (unchanged). In supervisor mode losing authority always means shutdown, which already stops it. On a standby, /health storage is now {lastSuccessAgeMs: null, lastErrorKind: null, stale: true, refreshing: false} with no tier stats, which says "no capacity data here". Nothing reads these fields for alerting: probe.py only reads leadership, ready, firehose.* (contract §9), and readiness never used storage. Rejected: keeping the scan on standbys at a longer interval (still pointless work), or faking stale: false on standbys (would lie).

2. BunSubscription gets maxQueuedBytes (default DEFAULT_MAX_QUEUED_BYTES = 16 MiB) for the bytes of frames received but not yet consumed. When a new frame would go past it (and the queue isn't empty, so one huge frame still gets through):

  • the connection is closed and any frames still in flight are ignored,
  • frames already queued are still delivered, in order (the consumer loop now drains the queue after a close instead of discarding it),
  • then it reconnects at once (no backoff, no onReconnectError, so the relay failure budget isn't charged) from getParams(). For the firehose that's getCurrentSeq() = the ordered tracker's resumableCursor (§2.2), which is ≤ every delivered seq, so the relay replays everything after it: at-least-once, ordered, duplicates dropped by the tracker as before (firehose-cursor.ts:146-155).
  • If getParams() has no cursor yet (fresh start before the first event), it resumes after the last seq it delivered instead of going live, so nothing is skipped.
  • onDisconnect fires once per connection, right at overflow, so a late close event can't flip isConnected to false after the next connection opened.

Rejected: real TCP backpressure (pause reads): neither Bun's client WebSocket (1.3.14/1.4.2) nor the WHATWG API has pause/resume; it would mean writing our own WebSocket+TLS client, which is more a job for the Rust rewrite. Dropping frames or blocking the event loop: loses events or stalls health.

Side effect worth knowing: frames queued when a connection closes for any reason (silence timeout, error) are now delivered before reconnecting rather than discarded and replayed. Order and at-least-once are the same either way; it just means less replay. (On main, the 300k bench below hits exactly this: the 15 s silence timer fires while ~17 s of backlog is still queued, so the queue is thrown away and replayed after a 2 s backoff.)

Before / After #

Everything was run locally on darwin arm64 with Bun 1.3.14. The bench was rerun on Linux with Bun 1.4.2 (oven/bun:1.4.2-alpine, the production image base), which gave the same picture. "Before" means origin/main 8324c85's index.ts / subscription.ts with this PR's scripts. Both scripts are committed.

1. Storage-stat scans per worker #

apps/firehose-service/scripts/storage-stats-scans.ts [standby|single] [holdMs] spawns the real src/index.ts against a fake S3 (counts list-type=2 requests) and, for standby, a fake supervisor that only ever says standby. It holds 3 s, then reads /health. A refresh scans once at start and then hourly, so the first-seconds count is the per-hour count.

cd apps/firehose-service && bun scripts/storage-stats-scans.ts standby && bun scripts/storage-stats-scans.ts single
role before: ListObjects after: ListObjects after /health.storage
standby (supervisor, never acquires) 1 (then +1/h → 25 in the first 24 h) 0 lastSuccessAgeMs: null, stale: true, refreshing: false
active (single-instance) 1 1 cold: {bytes:0, items:0}, stale: false

On a fleet with N edges that's N−1 fewer full-bucket listings per hour.

2. Frame queue under a slow consumer #

packages/@wispplace/bun-firehose/bench/backpressure.ts [frames] [frameBytes] [workUs] runs a relay in a child process that replays seqs after the requested cursor as fast as the socket accepts. The consumer spends workUs of CPU plus one event-loop turn per frame (like a handler awaiting I/O), and its getParams returns the last seq it handled. Peak queue length comes from wrapping MessageQueue.prototype.push; RSS/heap are sampled every 10 ms against a post-GC baseline. It exits non-zero on any gap, and checks every seq 1..N arrives in order.

cd packages/@wispplace/bun-firehose && bun bench/backpressure.ts 100000 1024 50 && bun bench/backpressure.ts 300000 1024 50

3 runs each, 1 KiB frames, workUs 50:

frames peak queued frames peak RSS growth peak heap growth connects gaps / dups seconds
100k before 98 453–98 626 192.7–198.5 MiB 119.9–120.6 MiB 1 0 / 0 5.36–5.38
100k after 15 932 74.7–84.1 MiB 31.8–39.4 MiB 7 0 / 0 5.32–5.33
300k before 295 425–295 705 524.5–548.7 MiB 360.4–380.4 MiB 2 0 / 0 17.33–17.57
300k after 15 932 83.6–87.5 MiB 37.8–39.3 MiB 19 0 / 0 15.88–16.13

Linux / Bun 1.4.2, 100k: before 98 751–99 077 queued, +200.9–204.2 MiB RSS, +118.4–118.6 MiB heap; after 15 932 queued, +75.9–97.7 MiB RSS, +38.5–44.4 MiB heap, 0 gaps.

After the fix, the peak is flat in backlog size: it's bounded by maxQueuedBytes and no longer by how far behind the consumer is. The leftover RSS/heap growth is mostly garbage from decoded frames that hasn't been collected yet, which shows up in both versions. Throughput is the same.

Tests #

  • New packages/@wispplace/bun-firehose/src/subscription.test.ts › BunSubscription backpressure (5 000 × 512 B frames, 64 KiB cap): every seq in order with no loss, peak queue ≤ cap, reconnects from a cursor; and the same with no consumer cursor (resumes after the last delivered seq). Both fail on main (connects = 1, the cap is ignored) and pass here.
  • New apps/firehose-service/src/index.storage-stats.test.ts (runs the script above): standby → 0 scans and the standby storage shape; active → 1 scan. The standby test fails on main (Expected: 0, Received: 1).
  • NODE_ENV=test bun test --isolate apps/firehose-service packages/@wispplace/bun-firehose: 416 pass, 83 skip (PG/Redis/S3 integration suites), 0 fail. bun-firehose suite also passes on Bun 1.4.2/Linux.
  • bun run check: the only error is the existing blog/src/pages/rss.xml.ts "Cannot find module 'astro'", which this PR doesn't touch. biome check is clean on the changed files.

Not measured yet #

  • Production effect. Staging is down (valefar storage pool), so there are no real relay or edge numbers. To confirm on staging: firehose container memory during a catch-up (e.g. the 26 h replay in records/2026-10-01-usersky-stale/), how often [BunSubscription] … unconsumed, reconnecting from cursor shows up, and S3 request metrics per edge.
  • The supervisor acquired path wasn't driven end to end, because the leader needs Redis for its cursor read. It goes through the same startProtectedWork as the single-instance path the test covers.
  • Overflow cost. Each overflow costs a reconnect (TLS handshake on a real relay) plus replaying from the committed cursor, which can sit behind the delivered seqs by up to the tracker's pending window (§2.2). Contract §14.1 warns that throttled reads could trip a relay slow-consumer disconnect; this design never stops reading, it closes on its own terms. If overflows turn out to be frequent in production, raise the 16 MiB default (it's a constructor option; BunFirehose doesn't expose it yet).
  • firehose.connected reads false for the short drain after an overflow (it's honest: the socket is closed). probe.py's ingestion check could catch that window if it samples during it.
Commits 0