feat(realtime): cross-device realtime sync — protocol v1.0.0 + server/daemon/apple - #3
Closed
QuintinShaw wants to merge 72 commits into
Closed
feat(realtime): cross-device realtime sync — protocol v1.0.0 + server/daemon/apple#3QuintinShaw wants to merge 72 commits into
QuintinShaw wants to merge 72 commits into
Conversation
Add a Swift mirror of proto/realtime/ (envelope + 14 message types), SpaceState with the SPEC.md section 6.4 deterministic folding rules, a RealtimeResumeGate implementing the section 6.2/6.3 resume/delta/snapshot gating as a transport-agnostic state machine, and a RealtimeClient actor that drives one WebSocket connection through hello -> subscribe -> resume, interest-lease renewal, ping/pong, and reconnect with backoff. Viewer-role only, matching SitrepApp's use as an observer rather than a task source.
Add fixture-driven SitrepKit tests against proto/realtime/fixtures/ (loaded directly from the frozen protocol directory, never copied): every valid fixture and every scenario message decodes and round-trips; every invalid fixture fails to decode or is rejected by the client authorization matrix. RealtimeResumeGateScenarioTests drives the resume/delta/snapshot gate with the four viewer-relevant scenario directories in file order. SpaceStateFoldingTests asserts delta-by-delta replay and a direct snapshot converge on identical task/automation state per SPEC.md section 6.4.
Wire SitrepApp's AppModel to SitrepKit's RealtimeClient: tasks, metrics, messages, and automations now update live from SpaceState deltas while the app is foreground, replacing the always-on 3s HTTP poll. The poll survives only as a low-frequency (>=30s) fallback that engages after repeated WebSocket backoff failures and disengages once the realtime connection recovers. scenePhase drives connection lifecycle deterministically: .active (re)connects and revives anything silently killed while suspended, .background closes the connection and lets the interest lease lapse. Add RealtimeUIBridge in SitrepKit adapting the realtime wire model to the existing poll-era UI types (TaskState/MetricState/EventLogEntry/ AutomationInfo) so NowView/DashboardView/HistoryView need no changes. Add public initializers to AutomationInfo/Executor/Schedule to support that construction from outside SitrepKit. MainTabView gains a subtle sync-status strip (reusing the presence-pill dot+caption language, not a new visual system) that only appears while not live, and a one-shot, tap-or-6s-dismiss banner for SPEC.md section 9.4 `superseded`.
Add internal/realtime/wire, a Go mirror of proto/realtime's frozen JSON Schemas: the strict envelope (type/id/ts/body), one struct per message body, and validation matching the schema's required fields, enums, ranges, and conditional requirements. Round-trips every fixture under proto/realtime/fixtures/valid and fixtures/scenarios through decode -> re-encode and asserts semantic equality with the original, and asserts every fixture under fixtures/invalid is rejected (including the sender_role-wrapped authorization-matrix cases via a new Authorize function mirroring SPEC.md section 10.1). proto/realtime/ itself is untouched; this package only depends on it as a read-only source of truth.
Add internal/realtime/outbox, a SQLite-backed (modernc.org/sqlite, pure Go, no cgo) local queue for task.event/message.event delivery. It owns the per-(device, space) device_seq counter from SPEC.md section 5.1: allocating the next sequence number and durably persisting the event it belongs to happen in one transaction, so a crash between the two is impossible to observe as a gap or a double-allocation. Enqueue/Ack/ Pending give a source device exactly what section 5.3/5.4 need: keep every unacked event until its ack arrives, and replay outstanding ones oldest-first after a reconnect or a process restart. Tests cover monotonic and concurrent-safe allocation, the crash-safety property (an aborted transaction must not skip a sequence number), ack-then-delete, oldest-first ordering, and full survival across a simulated process restart (close and reopen the same database file).
Add the metric batcher for internal/realtime/client: merges multiple
pending updates to the same metric_id into a single latest sample
before emission (SPEC.md section 12), and gates emission at a bounded
cadence — no metric_id is re-included more than once per 500ms (the
protocol's 2 Hz per-metric ceiling), and the whole batcher only flushes
at its own current interval, switchable between a normal and a slower
"throttled" cadence for the command{throttle}/{resume_rate} pair from
SPEC.md section 7. A sample offered while nothing is due simply
overwrites the pending entry for its metric_id rather than queueing,
matching metric.frame's best-effort, never-persisted, never-retried
nature (section 12): only the latest value survives to be sent.
It is a pure, clock-driven data structure with no timer of its own, so
every merge/rate-limit/throttle-switch behavior is tested by calling
Flush with an explicit instant — no sleeping real durations.
Per protocol-owner ruling, the cross-implementation route convention for the realtime WebSocket channel is /v3/realtime, matching the server implementation — not the /v2/realtime path this client previously assumed from the REST surface's prefix. Update the derived URL and the doc comment accordingly.
Add internal/realtime/client, the reconnecting WebSocket source
connection (github.com/coder/websocket, pure Go): the mandatory
hello{offer}/hello{accept} sequence with role "source", exponential
backoff with jitter on any drop (doubling from a 1s base, capped at
60s, +/-20%; stretched to the cap instead of reconnecting aggressively
after a superseded error, since that means our own newer connection
already won or the credential is compromised), the ping/pong heartbeat
with silent-peer detection, and command handling: TTL validation
against the local clock with the +/-30s skew allowance from SPEC.md
section 8, command_id dedup, and dispatch of throttle/resume_rate into
the metric batcher (other actions forward to an optional callback for
future process-control wiring).
Every reliable send (SendTaskEvent/SendMessageEvent) goes through the
outbox first and is replayed oldest-first, in a fresh envelope but with
its original device_seq, immediately after each reconnect's hello
completes (SPEC.md section 5.4) — on top of the outbox's own
already-tested crash safety, this is where that guarantee actually
reaches the wire.
Add internal/realtime/rttest, a minimal scriptable mock WebSocket
server used only by this package's tests to drive hello/ack/drop/
command scenarios without a real server dependency; it intentionally
does not persist state or model space_revision and must never be
described as a self-hosted server implementation.
Backoff is a pure, unexported-state-free function tested without any
sleeping; the reconnect/replay/throttle/command behaviors are tested
end-to-end against rttest.
Wire internal/realtime/client into the existing HTTP uplink and the resident `sitrep agent` process, off by default: - config: SITREP_REALTIME (or config.json's realtime_enabled) opts in; SITREP_REALTIME_URL overrides the wss://<server>/v2/realtime endpoint this package otherwise derives from the existing server URL. - uplink.Config gains an optional Realtime field. When set, Offer() routes every kind with a realtime-protocol equivalent (task lifecycle -> task.event, message.send -> message.event, metric.update -> metric.frame) to it instead of the HTTP /v2/ingest batch, so the same event is never sent both ways; task.log (no realtime equivalent) always continues over HTTP. Nil Realtime (the default) leaves this package's behavior byte-for-byte unchanged. - cmdAgent builds one shared realtime client (and its SQLite outbox, next to the existing config.json) for the whole process when the flag is on, reused across every automation run. This mirrors the protocol's "at most one connection per device" model (SPEC.md section 9.4); the one-shot `sitrep run` CLI is intentionally left on the existing HTTP-only path; see the handoff notes for why. Covered by a new integration test asserting the flag routes task/ message/metric events to a mock realtime server and never double-sends them over HTTP, and that the flag left off reproduces prior behavior exactly.
The protocol owner settled the realtime WebSocket route on /v3/realtime (the path the server implementation exposes; the Apple client targets the same). Update RealtimeURLFor and its documentation from the provisional /v2/realtime convention, and add a test pinning the derived endpoint (scheme mapping, trailing-slash normalization, and the explicit RealtimeURL override).
Implement SpaceHub, a per-space SQLite-backed Durable Object that speaks the frozen proto/realtime/SPEC.md v1 protocol: hello/subscribe/ resume connection gating, per-(device,space) device_seq dedup for task.event/message.event, deterministic event folding into tasks/ messages/automations tables, chunked snapshot and chained catch-up delta replies, server-minted config.event, interest leases with lazy 1<->0 edge detection driving throttle/resume_rate, and in-memory-only best-effort metric.frame relay. Uses ctx.acceptWebSocket with per-connection attachments (never DO-instance memory) so identity and handshake state survive hibernation/eviction, and ctx.setWebSocketAutoResponse for the ping/pong heartbeat pair. Also adds the hand-written TS protocol types and runtime guards (server/src/realtime/types.ts, guards.ts) mirroring proto/realtime/ schemas field-for-field, plus pure chunking/row-mapping helpers.
Add the SPACE_HUB Durable Object binding (new_sqlite_classes migration v2, alongside UserStore's v1) and wire a /v3/realtime WebSocket upgrade route plus a minimal /v3/automations HTTP control-plane companion (upsert/pause/remove, mints a config.event via SpaceHub#mintConfigEvent) into the Workers adapter. The Worker does exactly three things per request: reuse the existing st2 token authenticate() middleware (extended to also cover /v3/*, no behavior change for /v2/*) to resolve role/space/deviceId, reject an invalid or unresolvable token with 401 before ever touching SpaceHub, and forward the upgrade to that space's SpaceHub stub exactly once with the resolved identity attached as trusted headers. /v2/* stays untouched as the existing fallback.
Add the second test runway: @cloudflare/vitest-pool-workers running inside local workerd (test:workers), alongside the existing node --test suite (test:unit); `npm test` runs both. Lockfile updated and committed. Coverage: fixture round-trip decode/encode against every proto/realtime/fixtures/valid and invalid fixture (including the role-wrapped authorization-only ones) and pure chunking-boundary tests under test:unit; connection identity recovery via fresh connections/stub references, duplicate device_seq dedup (no double apply, no double revision, still acks), the full resume decision table (snapshot/current-empty-delta/revision_unavailable/retention- miss-falls-back-to-snapshot), push_outbox idempotency, metric.frame's zero-SQL-write guarantee and multi-viewer broadcast, chunked-snapshot ordering under a racing live event, interest-lease throttle/ resume_rate edges, connection gating (hello_required, resume-before- subscribe malformed, no live delta before a connection's own resume reply), connection supersession, and the Worker-layer invalid-token/ reconnect-storm guard that never instantiates a SpaceHub under test:workers.
Cap hot-path per-frame logging in SpaceHub to <=1% of messages (injectable sampler, so tests can force full logging), while every protocol_error and ws_error path continues to log unconditionally. Avoids a console.log per inbound frame on a busy connection.
Add docs/design/realtime-server.md: a per-message-type table of SQL writes / revision++ / outbox / broadcast aligned with proto/realtime/SPEC.md section 14, a Durable Object billing-axis analysis (requests/duration/storage), and the specific conditions under which SpaceHub can hibernate (no timers/alarms anywhere; lazy lease-edge detection; attachment-only connection identity).
Address three adversarial-review findings in the SitrepKit client layer: - Terminal errors stop reconnecting (F2): a server error carrying fatal:true and retryable:false (superseded, unauthenticated, version_unsupported) now ends the run loop with the .failed phase preserved instead of feeding the backoff/reconnect cycle; a later start() — user action or the next foreground pass — begins a fresh cycle. Handshake- and subscribe-stage server errors route through handleError so their notices surface and their retryable/fatal semantics reach the loop. A run-generation token stops a stale loop from clobbering its successor's task handle. - Chunked-snapshot interleaving is malformed (F3): per SPEC.md 6.2, any non-snapshot envelope arriving while snapshot chunks are being reassembled (ping/pong never reach dispatch) now discards the buffered chunks and closes the connection for a fresh resume, in both the gate and the client dispatch path. - Malformed frames are contained, not fatal (F5): a frame that fails to decode is skipped per SPEC.md 13 (malformed is non-fatal) and only a run of 5 consecutive malformed frames trips a circuit breaker that reconnects; the DeltaEvent wrapper now ignores unknown sibling fields, matching the body-level tolerance of SPEC.md 15. Also adds Phase.allowsReliableStateOverwrite, the policy the app layer uses to keep HTTP snapshot results from clobbering live delta state (applied in the follow-up commit).
MetricSample.Validate now enforces the metric_id pattern from
common.schema.json#/$defs/metric_id (^[a-z0-9_.-]{1,64}$), which also
makes an empty id invalid. Previously only the value/label caps and the
timestamp bound were checked, so a malformed id could slip into an
outbound metric.frame. Add a test covering valid ids and the empty/
uppercase/whitespace/non-ascii/over-length rejections.
Cap the outbox's total row count (default 5000, OpenWithMaxRows for an explicit value). At the cap Enqueue fails with ErrOutboxFull inside the same transaction as the insert — no device_seq is consumed, no row is written — so the existing enqueue-failure fallback in the uplink routes the event over the HTTP ingest path instead: reliability does not degrade while the server is unreachable, and local disk usage stays bounded. Once acks drain the backlog, capacity returns and subsequent events take the realtime path again. Policy documented on ErrOutboxFull and routeToRealtime. Tests: outbox-level (ErrOutboxFull at cap, no seq consumed, capacity restored by an ack) and an end-to-end uplink test (overflow event travels HTTP while the outbox is full, realtime resumes after the backlog drains, non-overflow events never appear on HTTP).
Two adversarial-review findings in the app wiring: - HTTP refresh no longer bypasses realtime state (F1): while the WebSocket phase is .live, refresh() updates only REST-only concepts (presence, lastSyncAt) and leaves the four reliable collections (tasks/metrics/events/automations) to deltas. The liveness check runs AFTER the await returns, so an in-flight response — a foreground force-refresh, the post-updateMetric refresh, or a fallback-poll response landing after .recovered — can never clobber newer delta state. An HTTP failure while live no longer raises the error banner. - Foreground cycles resume incrementally (F4): AppModel keeps the last SpaceState (with revision C) for the process lifetime and seeds the rebuilt RealtimeClient with it, so returning to the foreground resumes with last_revision C — an incremental delta or empty you-are-current reply — instead of pulling a full snapshot every time. The preserved state is dropped on disconnect and on credential change, where its revision sequence no longer applies.
Tighten and extend the SitrepKit suite per review (F6): - Every schema-invalid fixture now asserts the SPECIFIC constraint it trips (substring of the thrown error), so a fixture failing for the wrong reason is a test failure; an unmapped new fixture fails loudly. - New RealtimeRegressionTests cover the review fixes: only .live blocks HTTP overwrite of reliable state and the check re-runs after the await (F1, incl. the in-flight race); a delta or any other envelope interleaved during a chunked snapshot is a malformed sequence that discards the buffer (F3); a reconnect seeded with the preserved SpaceState resumes incrementally from C and RealtimeClient honors its initialState (F4); the delta event wrapper ignores unknown sibling fields (F5). - New folding boundary test: an event without step/title leaves the previous values unchanged, while done still clears step (SPEC.md 6.4).
Two handshake outcomes previously fed the reconnect loop forever even though a retry could only fail identically: - an error reply that is fatal AND retryable:false (version_unsupported, unauthenticated, ... — SPEC.md section 13's flags are authoritative) - a hello accept naming a protocol version the offer never contained, which section 9.2's intersection rule makes a failed negotiation Both now surface a "giving up, not retryable" log and terminate the reconnect loop; recovery is an explicit restart (a new Client). Tests cover both paths by counting connections at the mock server: exactly one attempt, no loop.
SPEC.md sections 5.3/13 require every pair in an ack to carry the
receiving connection's authenticated device_id, and a pair for any
other device makes the WHOLE envelope malformed. The client previously
skipped foreign pairs but still honored the matching ones; now it drops
the entire ack without retiring anything from the outbox (resend keeps
this safe — the events stay queued for a later well-formed ack),
reports the violation back with error{malformed, retryable, non-fatal},
and keeps the connection.
Extract the error-envelope write into a shared sendError helper (also
used by the command_expired path). Test: the mock server sends an ack
mixing one matching and one foreign pair — nothing is deleted — then a
well-formed ack on the same connection still retires the event.
Fix the agreed semantics in a regression test: a source that was told to throttle keeps its throttled metric cadence across a connection drop (it does not optimistically resume the fast rate on reconnect), and the server-side remedy — re-sending the space's current interest state after the new connection's hello completes — takes effect immediately: a resume_rate on the second connection restores the normal cadence.
The hello offer schema declares uniqueItems on protocol_versions and the subscribe/interest.renew schema declares it on topics; the hand-written guards accepted duplicates, deviating from the frozen proto/realtime schemas they claim to mirror. Enforce both.
Wrap SpaceHub's webSocketMessage dispatch (and webSocketClose) in a
top-level guard: an unexpected handler throw now logs in full and
answers the frame with error{internal_error, retryable, non-fatal}
whose message carries no stack traces or internal paths (SPEC.md
section 13), instead of surfacing as an uncaught DO exception. Also
remove reply()'s never-taken internal_error log exclusion so every
outbound protocol error is logged unconditionally. Covered by a test
that injects a handler fault and verifies the connection survives.
SPEC.md section 7's throttle/resume_rate are pure edge triggers, so a
source that was offline when the space's lease count crossed its 1<->0
edge would stay stuck at its last-known rate forever. Immediately
after a source's hello accept, SpaceHub now unicasts the CURRENT
interest state as command{origin:server} (throttle when no unexpired
lease exists, resume_rate otherwise) with a fresh command_id — no new
message type and zero protocol schema changes; flagged in-code as a
v1.1 clarification candidate per the protocol owner's ruling. Tests
cover both the initial throttle and the reconnect-after-missed-edge
resume_rate, and the shared helpers drain the new post-hello command.
SPEC.md section 12's discard-stale-samples rule was keyed off a per-WebSocket rate-limiter map, so a reconnecting or second source could replay an older sample past a fresher one. The baseline now lives in the space-level metricsCache (keyed by metric_id alone, same lifetime and best-effort semantics as the rest of that cache — reset after eviction is acceptable per section 6.2). Test interleaves two source connections on one metric_id and asserts the older ts is dropped and a genuinely fresher one still flows.
Four rulings from the adversarial review, all on the /v3/automations surface: - Idempotency-Key is now bound to request content: mintConfigEvent stores a canonical-JSON fingerprint of (kind, automation_id, automation) and compares it on replay; the same key reused for a different operation returns 409 instead of silently replaying the first operation's result. - http_idempotency growth is bounded lazily on its own write path (no alarm): entries older than 24h are dropped and the table is capped at the 500 most recent rows; documented in the cost model. - Role matrix finalized per the protocol owner (aligned with "watcher schedules are editable from every device, creation only from trusted devices"): POST = owner/admin; PATCH = owner/admin/viewer (viewer editing schedules is intentional); DELETE now also allows viewer, matching /v2 canView. - DELETE of a nonexistent automation returns 404 before minting, so a no-op removal cannot burn a revision (same pattern as PATCH); the /v3 HTTP handlers also gained the internal_error guard (500 with no internal details, full server-side log). Tests cover the 409 conflict, viewer PATCH/DELETE success + POST 403, and the 404-without-revision-burn invariant.
Note in the cost model why the 10 s handshake timeout (SPEC.md section 9.1, SHOULD) is deliberately not enforced server-side (it would require the per-connection timer machinery this design excludes to stay hibernation-friendly, and the exposure of an offer-less connection is bounded), and why the advisory sequence_gap error (section 5.1, MAY) is not emitted.
The runGeneration token previously protected only the runTask handle: a stale connectOnce() unwinding after a stop()/start() cycle on the same RealtimeClient instance would run its deferred teardownConnection() and cancel the NEWER generation's socket, lease-renewal task, and watchdog. Unreachable in the app today (it rebuilds the client each cycle) but a hole in the class's own public contract. Teardown of the shared resources now requires the requesting generation to still be current (ConnectionTeardownPolicy, pure and unit-tested); a stale generation limits itself to closing the socket it created and holds by reference. scheduleLeaseRenewal and startHeartbeat gain the same guard so a stale dispatch cannot replace the current generation's tasks either. Adds a stop-then-start same-instance regression test against a refused local port (no real network).
"observability": { "enabled": true } alone doesn't bound per-invocation
telemetry, so the documented cost model wasn't real. Disable the
platform's automatic invocation logs and keep structured logs unsampled
(logs.head_sampling_rate: 1) so logAlways() security events (superseded,
protocol errors) are never head-sampled away — volume is bounded instead
by SpaceHub's existing in-code logSampler. Traces, which have no
logAlways equivalent, are head-sampled at 1%. Update the cost model doc
with the final config and revised estimate.
SpaceSnapshot gains realtime_enabled (decode-if-present, default false) so old/undeployed servers that never send the field are treated as "no realtime" rather than inferred from URL shape. RealtimeCapabilityGate is pure decision logic (connect/disconnect/none) kept in SitrepKit so the P0 gate is unit-testable without a live socket, and RealtimeClient itself stays capability-agnostic.
The app previously derived /v3/realtime from the REST URL and connected unconditionally once paired and foregrounded, so a server without realtime enabled could serve a valid legacy /v2 snapshot and then have the UI clobbered by an empty/failed connection state. AppModel now owns a RealtimeCapabilityGate, fed by realtime_enabled on every successful refresh(). startRealtime() is gated on it, so the connection only ever opens once a refresh has confirmed capability; cold start (before any refresh) and legacy/absent responses default to off. If a later refresh reports false/absent while connected, the existing generation-scoped teardown tears the connection down and the app falls back to plain HTTP refresh with no reconnect attempts until a later refresh says true again. Re-pairing (saveSettings) and disconnect() reset the gate since a new identity must prove capability again. MainTabView's sync strip needed no change: gating off keeps connectionPhase at .idle, which already renders as plain HTTP mode.
DO constructors re-run on every hibernation wake, and migrate() was unconditionally re-issuing 10+ CREATE TABLE/INDEX IF NOT EXISTS statements each time — conflicting with the sparse-connection cost goals in docs/design/realtime-server.md. Add a _schema_migrations version table: migrate() now does one lightweight SELECT and only runs (and writes) the DDL block when the store is behind, including the fresh-DO case where the version table doesn't exist yet.
metricsCache had per-frame size/rate limits but no cap on distinct metric_ids, so a compromised or misbehaving source rotating metric_ids could grow DO memory without bound until the instance crashes. Add METRIC_CACHE_MAX_METRICS (256, protocol.ts) and evict the least-recently-updated metric_id when a new one would exceed it. Eviction is safe per SPEC.md 6.2 (snapshot.metrics is a best-effort cache); evicted metrics are simply absent from the next snapshot.
handlePreHello sent the superseded peer's error frame before calling logAlways(), so a throwing send() would skip the security log entirely. Reorder so the 100%-sampled logAlways fires first; the wire frame and ordering toward the client are unchanged in the success path (still error frame, then close).
… schema versioning, metric cache cap, log ordering)
Cloudflare dashboard variable overrides always arrive as strings at
runtime, even though wrangler.jsonc declares REALTIME_ENABLED as a
boolean. Boolean("false") is true, so an operator typing "false" in the
dashboard would silently re-enable the realtime kill switch. Add
parseRealtimeEnabledFlag() as a strict allow-list (true / "true" / "1",
case-insensitive, trimmed) and route the worker adapter through it;
widen the Secrets type to string | boolean to match the real runtime
shape. Covered by new unit tests in test/app.test.ts.
schemaVersion()'s bare catch treated any SQL failure — including corruption or I/O errors — as "fresh DO, run migrations" (version 0). The DDL is IF NOT EXISTS so a genuinely fresh DO re-running it is benign, but misdiagnosing real corruption this way hides the failure. Only swallow the specific "no such table" error that means the migrations table doesn't exist yet; rethrow anything else.
…tinel Enqueue's BuildBody closures call body.Validate() before the insert, but a validation failure was indistinguishable from a transient BeginTx/COUNT/ INSERT/Commit error (SQLITE_BUSY, disk I/O) once it came back as a plain error. Wrap it in ErrInvalidBody so callers can tell "this body can never validate, drop it" apart from "this Enqueue attempt failed, retry it" via errors.Is.
Two bugs in the outbox-backpressure path: - BLOCKER: sendReliable always attempted Enqueue for a new event, even with older items already waiting in rtRetry. A freed outbox row between an older event's failed attempt and a newer event's Offer() let the newer one grab a lower device_seq, inverting wire order once the older one was retried. Fix: a dedicated rtMu now guards the whole check-then-act step in sendReliable (queue non-empty -> append behind it; queue empty -> attempt Enqueue) and retryRealtime's whole drain, so the two can never interleave. - MAJOR: any non-ErrOutboxFull Enqueue failure was logged and dropped, which silently discarded a reliable task.done/message.send on a transient SQLite error (SQLITE_BUSY, disk I/O). isRetryable now treats only rtclient.ErrInvalidBody as permanent (retrying the same malformed body can't help); everything else, including unrecognized errors, defaults to retry. Adds regression tests for both: a deterministic FIFO-interleaving test that drives sendReliable/retryRealtime directly (no sleeps), a transient- error test asserting eventual delivery, and a permanent-error test asserting drop-not-retry.
REALTIME_ENABLED=false only suppressed realtime_enabled in /v2/snapshot;
the /v3/realtime upgrade route accepted connections regardless, so a
daemon with its local flag on kept writing to SpaceHub while viewers
read UserStore (dual authority). The route now rejects with 403
{"error":"realtime_disabled"} using the same parseRealtimeEnabledFlag
single source of truth, checked after authenticate() (so an
unauthenticated caller still gets its normal 401, not a 403 that would
leak flag state) and before the SpaceHub stub.fetch() call, so a
disabled deployment never wakes a DO.
/v3/automations POST/PATCH/DELETE get the identical gate: they mutate
the same SpaceHub DO via mintConfigEvent, so leaving them open would
just relocate the dual-authority bug to the automations control plane
instead of fixing it.
No active-connection kill is needed: flag changes ship via redeploy,
which evicts every DO instance and drops its live WebSockets as a side
effect (noted at the check site).
…/automations Adds realtime-gate.workers.ts: flag off -> /v3/realtime and /v3/automations POST/PATCH/DELETE all return 403 realtime_disabled without waking a SpaceHub DO; flag on -> the upgrade still works (explicit regression check); unauthenticated + flag off -> still 401, not 403. wrangler.jsonc's REALTIME_ENABLED stays false as the production kill-switch default, but the rest of the /v3/realtime protocol suite (handshake, broadcast, reliability, token-gate, ...) exercises the enabled path and would otherwise all start failing with 403. Override it to true for the test pool only via vitest.config.ts's miniflare bindings; this file flips env.REALTIME_ENABLED back to false per-test to reach the disabled branch.
…verflow table The outbox's row cap used to reject Enqueue outright (ErrOutboxFull), pushing backpressure onto an in-memory retry queue in the uplink that a short-lived process (agent.go's Uplink.Close()) never drained past a single final flush. Enqueue now durably persists an overflowing event into a new outbox_overflow table in the same SQLite database instead, returning the informational ErrOverflowed. PromoteOverflow moves the oldest overflow row into the seq-bearing outbox (allocating device_seq only at that moment) whenever an ack frees capacity or the store reopens after a restart; the FIFO guard against a fresh event stealing a seq ahead of an older overflowed one is enforced inside Enqueue's own transaction.
…flow-aware residual queue rtRetry used to hold every event that hit outbox-full locally, in a slice Close() never fully drained — the exact loss the no-fork rule was meant to prevent for a short-lived per-automation Uplink. Now that outbox.Enqueue persists synchronously (overflowing durably instead of failing), sendReliable only needs a tiny residual queue for the genuinely rare case where Enqueue's own bounded retries are exhausted (a persistent local-disk failure) — ordinary outbox-full backpressure never reaches it.
The server can now reject /v3/realtime pre-upgrade with HTTP 403
{"error":"realtime_disabled"} when its own REALTIME_ENABLED is off.
coder/websocket's Dial keeps the rejected *http.Response on error, so a 403
there is recognized as the new sentinel ErrRealtimeDisabled rather than an
ordinary transient dial failure — it no longer feeds the tight reconnect
backoff, and instead waits a long, fixed, named interval before re-probing.
Uplink.pollRealtimeMode polls Client.ServerDisabled() once per flush tick and
switches the whole session to legacy /v2 HTTP routing while it's true —
draining every event already sitting durably in the outbox/overflow tables
to /v2, in FIFO order, so nothing already-enqueued gets stranded in a store
no viewer reads. Authority genuinely moves to UserStore for this case; it is
not the forked-reliable-path failure mode the earlier P0 fix guarded against.
When the client's periodic re-probe succeeds again, new events resume
routing to realtime immediately (legacy mode drained as it went, so there is
no backlog to replay).
…e Client cmd/sitrep/agent.go builds one *rtclient.Client for the whole process but constructs a fresh uplink.Uplink (and its own pollRealtimeMode loop) per concurrently-running automation. Each Uplink's drainOutboxToLegacy read the shared outbox's Pending()/OverflowPending() and sent the row before either had a chance to Ack/DeleteOverflow it, so two automations racing while the server rejected realtime could deliver the same reliable event to /v2/ingest twice. Move drain ownership onto rtclient.Client (already the single shared object owning the outbox and the single-writer send loop): DrainToLegacy takes a per-row LegacySink and serializes every row's read-sink-retire sequence under a new legacyDrainMu, held one row at a time so a reconnecting client is never blocked behind a whole backlog. The same lock guards connectAndServe's post-hello replayPending, closing the narrower switch-back race between a lagging legacy drain and the newly reconnected client's own replay of the same outbox. Uplink.drainOutboxToLegacy now just delegates translation/send to rt.DrainToLegacy via a LegacySink closure; retiring rows is entirely the Client's responsibility.
…drain Add TestDrainToLegacyConcurrentCallersDeliverEachRowExactlyOnce: many goroutines call Client.DrainToLegacy concurrently against one Client/outbox pre-loaded with a deterministic backlog, simulating several automations' Uplinks polling the same shared realtime Client at once (the normal case commit 1114ff9's review flagged as racy). Asserts every row reaches the sink exactly once, in device_seq order, across all callers combined. Meaningful under -race; passes at -count=20.
…vents
SendTaskEvent/SendMessageEvent returned outbox.ErrOverflowed to the
caller like any other Enqueue failure, even though ErrOverflowed's own
doc comment calls it informational (the event is already durably
persisted into outbox_overflow) and says callers must not retry or
reroute on it. uplink.Uplink.sendReliable's doc comment already assumed
this never reached it ("ordinary outbox-full backpressure durably
overflows inside Enqueue and never reaches here") — but nothing enforced
it, so an overflowed event fell into sendReliable's persist-failure/
residual-retry branch and got re-Enqueued on every flush tick until
capacity freed, each attempt durably creating another outbox_overflow row
for the same logical event (Enqueue never deduplicates): unbounded
duplicate rows in overflow's uncapped table, only one of which is ever
promoted and delivered.
Swallow ErrOverflowed in both Send*Event methods so it never reaches the
caller as an error, matching the invariant both doc comments already
assumed.
Add TestOverflowSurvivesProcessRestartAndDeliversInOrder: fill a cap-1 outbox, overflow a second reliable event, shut the Uplink/Client/Store down exactly as a process exit would, open a brand new Store handle on the same file (never reusing the old one), ack the still-unacked older event once reconnected, and assert the overflowed event is promoted in its original order and delivered over realtime — with the /v2 HTTP spy seeing nothing at all throughout.
…ontention Add TestEnqueueRecoversFromRealTransientLockContention: a second, real sql.DB connection to the same outbox file takes SQLite's write lock (BEGIN IMMEDIATE) for a duration that lands inside Enqueue's second internal attempt window, so enqueueAttempt observes a genuine SQLITE_BUSY (not a faked send/build error) on its first attempt and recovers via its own bounded retry (enqueueMaxAttempts/ enqueueRetryDelay) once the lock actually releases. Asserts the event ends up durable and readable, never dropped. Add the openWithBusyTimeoutMS test seam (OpenWithMaxRows now delegates to it with the production 5000ms) so the test can use a short busy_timeout and stay fast instead of waiting out the real 5s default.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
proto/realtime/): transport-agnostic envelope, 14 message types, device_seq dedup, space_revision delta chain, snapshot chunking, interest leases, command channel. Frozen at main@a8ed3f4; 81 fixtures + scenario invariants (proto/realtime/tools)./v3/realtimeon existing Worker, new SQLite-backedSpaceHubDO (Hibernatable WebSockets, serialized attachments, auto-response ping/pong), push_outbox only;/v2untouched.docs/design/realtime-integration.md(contract audit + real E2E via wrangler dev).Validation
go test ./... -raceclean, 39 top-level testsswift test; SitrepApp xcodebuild + SitrepMenuBar build greenRisk
realtime_enabled, outbox-full backpressure model, observability sampling config, DO migration version table, metricsCache LRU cap, superseded log ordering). Merge blocked until fixed + re-reviewed.