diff --git a/cmd/pad/cmd_server.go b/cmd/pad/cmd_server.go index 90599bdc..bddc79ba 100644 --- a/cmd/pad/cmd_server.go +++ b/cmd/pad/cmd_server.go @@ -700,8 +700,20 @@ func serveCmd() *cobra.Command { if cfg.EventsPublishEpoch { phase = 2 } + // The heartbeat rollout is logged for the SAME reason and is a + // SEPARATE, independently-flipped migration (BUG-2738): an + // instance can be on id-space phase 2 and heartbeat phase 1 or + // any other combination. pad_event_subscription_cycled_total + // reads differently per phase — on heartbeat phase 1 a QUIET + // workspace's wedged route is undetectable, so a zero there + // means less than it does on phase 2. + heartbeatPhase := 1 + if cfg.EventsHeartbeat { + heartbeatPhase = 2 + } slog.Info("Event bus using Redis pub/sub", "addr", opts.Addr, "db", opts.DB, - "namespace", redisKeys.Namespace(), "id_space_phase", phase) + "namespace", redisKeys.Namespace(), "id_space_phase", phase, + "heartbeat_phase", heartbeatPhase) } else { eventBus = newObservedEventBus(cfg, nil, redisKeys, m) slog.Info("Event bus using in-memory (single instance)") @@ -1346,7 +1358,7 @@ func humanBytes(n int64) string { // internal/idspace), so it ignores the field. func newObservedEventBus(cfg *config.Config, rc *redis.Client, redisKeys redisns.Keys, m *metrics.Metrics) events.EventBus { if rc != nil { - bus := events.NewRedisBusWithKeys(rc, redisKeys, cfg.EventsPublishEpoch) + bus := events.NewRedisBusWithKeys(rc, redisKeys, cfg.EventsPublishEpoch, cfg.EventsHeartbeat) bus.SetObserver(metrics.NewEventsObserver(m)) return bus } diff --git a/cmd/pad/event_bus_wiring_test.go b/cmd/pad/event_bus_wiring_test.go index 016c0082..08a8cf0e 100644 --- a/cmd/pad/event_bus_wiring_test.go +++ b/cmd/pad/event_bus_wiring_test.go @@ -198,3 +198,107 @@ func TestThePublishEpochFlipReachesTheRedisBus(t *testing.T) { } }) } + +// TestTheHeartbeatFlipReachesTheRedisBus is the same claim for BUG-2738's +// phase-2 flip, and it needs its own test for the same reason the epoch one +// does: internal/events proves a bus constructed with publishHeartbeat=true +// emits liveness frames and runs idle detection, and every one of those tests +// passes if newObservedEventBus hardcodes `false` here — the deployment would +// simply never detect a wedged connection, which is indistinguishable from a +// deployment that has none. +// +// Both directions are asserted because a helper that ignored its config and +// hardcoded EITHER value would pass a one-directional test. +// +// Asserted on the FRAME rather than on a cycle, deliberately: publishing is +// observable in one interval on a test cadence, while a cycle needs the idle +// threshold to elapse. They are one switch (see config.EventsHeartbeat), so +// the frame is a faithful proxy — and the detector's own gate is pinned in +// internal/events by TestAQuietWorkspaceIsNotCycledOnPhase1 and its +// counterfactual. +func TestTheHeartbeatFlipReachesTheRedisBus(t *testing.T) { + channel := redisns.Default.Name("events:") + "ws-1" + + for _, tc := range []struct { + name string + heartbeat bool + wantHeartbeat bool + }{ + {name: "phase 1", heartbeat: false, wantHeartbeat: false}, + {name: "phase 2", heartbeat: true, wantHeartbeat: true}, + } { + t.Run(tc.name, func(t *testing.T) { + mr := miniredis.RunT(t) + client := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + t.Cleanup(func() { _ = client.Close() }) + + ps := client.Subscribe(context.Background(), channel) + t.Cleanup(func() { _ = ps.Close() }) + if _, err := ps.Receive(context.Background()); err != nil { + t.Fatalf("subscribe: %v", err) + } + incoming := ps.Channel() + + bus := newObservedEventBus(&config.Config{EventsHeartbeat: tc.heartbeat}, client, redisns.Default, metrics.New()) + t.Cleanup(bus.Close) + + redisBus, ok := bus.(*events.RedisBus) + if !ok { + t.Fatalf("expected a *events.RedisBus, got %T", bus) + } + // A subscription to publish heartbeats FOR. Heartbeats are scoped + // to the workspaces this instance subscribes to, so without this + // there is nothing to emit in either arm and the test would pass on + // both. + ch, _, outcome := redisBus.Subscribe(context.Background(), "ws-1") + if outcome != events.SubscribeOK { + t.Fatalf("subscribe: %v", outcome) + } + t.Cleanup(func() { redisBus.Unsubscribe(ch) }) + + // DRIVEN DIRECTLY, NOT WAITED FOR. Shortening the cadence and + // sleeping made the negative arm a race against the scheduler: a + // phase-1 bus that is correctly silent and a phase-2 goroutine that + // merely has not run yet look identical, so the test could pass or + // fail for reasons unrelated to the flip. One synchronous pass + // removes the timing entirely. + redisBus.PublishHeartbeatsForTest() + + // The barrier is then an ORDINARY event on the same channel: Redis + // delivers in publish order on one connection, so if a frame were + // emitted it is already ahead of this. + redisBus.Publish(events.Event{Type: events.ItemCreated, WorkspaceID: "ws-1", ItemID: "item-7"}) + + deadline := time.After(5 * time.Second) + for { + select { + case msg := <-incoming: + if strings.HasPrefix(msg.Payload, "hb|") { + if !tc.wantHeartbeat { + t.Fatalf("EventsHeartbeat=%v published a liveness frame %q: every un-upgraded peer resyncs all its clients", + tc.heartbeat, msg.Payload) + } + return // phase 2: the frame reached Redis, which is the claim + } + if strings.Contains(msg.Payload, `"item_id":"item-7"`) { + if tc.wantHeartbeat { + t.Fatal("EventsHeartbeat=true published no liveness frame ahead of the barrier event: the flip is not reaching the bus") + } + return // phase 1: the barrier arrived with no frame ahead of it + } + case <-deadline: + t.Fatal("timed out waiting for the barrier event") + } + } + }) + } + + t.Run("in-process shape ignores it", func(t *testing.T) { + bus := newObservedEventBus(&config.Config{EventsHeartbeat: true}, nil, redisns.Default, metrics.New()) + t.Cleanup(bus.Close) + bus.Publish(events.Event{Type: events.ItemCreated, WorkspaceID: "ws-1"}) + if got := bus.EventsSince("ws-1", 0); len(got) != 1 { + t.Fatalf("the in-process bus must publish normally regardless of the flip, got %d events", len(got)) + } + }) +} diff --git a/docs/deployment.md b/docs/deployment.md index 2fafea2b..c8b6dc7b 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -88,6 +88,7 @@ All configuration is via environment variables or a config file (`~/.pad/config. | `PAD_SSE_MAX_PER_WORKSPACE` | `100` | Per-workspace maximum connections on `/api/v1/events`, **per instance** | | `PAD_SSE_MAX_PER_USER` | `50` | Per-user maximum streaming connections across both endpoints, **per instance** | | `PAD_EVENTS_PUBLISH_EPOCH` | `false` | Phase 2 of the event ID-space migration: publish the `\|\|` wire form. **Only set this once every instance runs a binary that accepts it** — see *Event ID-space migration* below. Ignored without Redis. | +| `PAD_EVENTS_HEARTBEAT` | `false` | Phase 2 of the half-open-connection detection rollout: publish a bus-internal liveness frame on each subscribed workspace channel every 30s. **Only set this once every instance runs a binary that recognises it** — see *Half-open connection detection* below. Setting it early makes every un-upgraded instance resync all its clients every 30 seconds. Ignored without Redis. | #### Streaming connection limits @@ -300,19 +301,24 @@ reading the metrics below, and for anyone writing a third-party consumer: whole message into memory before Pad sees it; bound it with Redis's `proto-max-bulk-len` and with who holds `PUBLISH`. - **Two gaps in that detection remain, and an operator should know both.** A - message lost in transit with the connection intact — no flap, no decode - failure, just a message that never arrived (BUG-2735): on the watch stream a - LATER notification exposes it as an ID gap, while on the activity stream, - whose per-workspace IDs are non-consecutive by construction, nothing local - ever does. And a HALF-OPEN connection — a route that stopped carrying - traffic without closing, so nothing ever resubscribes and no message ever - arrives to be non-consecutive with (BUG-2738). Do not assume go-redis's - pub/sub health check covers the second: `PubSub.Ping` writes the command and - never reads a reply, so it reports healthy for as long as the socket accepts - writes. Detecting it needs application-level idle tracking, which needs a - threshold, which is a deployment decision rather than an implementation - detail. + **One gap in that detection remains everywhere, and a second remains on the + watch stream only.** A message lost in transit with the connection intact — + no flap, no decode failure, just a message that never arrived (BUG-2735): on + the watch stream a LATER notification exposes it as an ID gap, while on the + activity stream, whose per-workspace IDs are non-consecutive by construction, + nothing local ever does. That one is open on both. + + A HALF-OPEN connection — a route that stopped carrying traffic without + closing, so nothing ever resubscribes and no message ever arrives to be + non-consecutive with — is **closed on the activity stream** as of BUG-2738 + and **still open on the watch stream**, which has the same defect by the same + mechanism and has not been ported yet. Do not assume go-redis's pub/sub + health check covers it on either: `PubSub.Ping` writes the command and never + reads a reply, so it reports healthy for as long as the socket accepts + writes. What closes it on the activity stream is application-level idle + tracking with a heartbeat that makes the threshold answerable — see *Half-open + connection detection* — and until the same lands on the watch stream, a wedged + route there is still silent. **A third residual affects RESUMES rather than open streams** (BUG-2743): if the watch counter restarts without the epoch rotating — evicted under @@ -412,8 +418,9 @@ Alert on these instead: | `pad_event_resume_gaps_total` | The ACTIVITY stream's (`/api/v1/events`) twin of the watch resume counter above. **Expect a step around a deploy, with the RATE settling back to baseline** (the counter itself only ever increases) — each instance starts with no replay coverage, so an early resume against a workspace it has not seen yet is a warranted resync. It counts RESUMES, not clients: a deploy with no reconnects does not move it at all, and a client that reconnects several times is counted several times. A rate that does not settle is the thing to alert on | | `pad_event_midstream_resyncs_total` | Activity-stream subscribers told MID-STREAM that they missed events, on a connection that stayed open. New in BUG-2730, and the counter to watch when judging whether that fix is costing more resyncs than it is worth. It counts ANNOUNCEMENTS, not causes and not distinct clients: a reset that drops buffers moves it once per live subscriber (and that ratio against `pad_event_sequence_resets_total` is the fan-out); a burst of drops on ONE connection moves it once, because signals coalesce and are rate-limited per connection; and a coverage loss on a workspace with no buffer yet moves it while every cause counter stays flat, because there was no coverage to end but the subscribers still have a hole | | `pad_watchevents_midstream_resyncs_total` (see also, listed above) | Same meaning for the watch stream. Its causes are a slow-subscriber drop and a received sequence gap or reset; a gap announces to EVERY subscriber on the instance, so it can exceed all of its cause counters | -| `pad_event_sequence_resets_total` | Activity replay coverage dropped, by reason. `subscription_resumed` — a pub/sub connection dropped and resubscribed, dropping that workspace's buffer; expect it during a Redis failover and expect it to stop afterwards. `epoch_change` — the shared counter's ID space changed generation, dropping every buffer; expect a handful per cutover. `counter_backward` — an ID arrived at or below a buffer's high-water mark with no generation change; see *Event ID-space migration* for what to expect per phase. `epoch_regressed` — a LOWER generation was seen, so this instance stopped vouching for its buffers. One alongside an `epoch_change` is a message that was in flight when the generation rotated; a RUN of them means the counter itself went backwards — usually Redis lost writes, and since BUG-2740 possibly a repaired generation key (see *A repaired generation counter*). `undecodable_message` — a message on these channels could not be parsed, so that workspace's coverage ended; expect zero, and suspect a namespace collision. `subscription_unconfirmed` — a subscription was admitted before Redis acknowledged the SUBSCRIBE and the acknowledgement then arrived, so the span in between is one that stream cannot account for; it reaches THIS counter only when a buffer existed to drop, so read `pad_event_subscription_unconfirmed_total` for the dependable count | +| `pad_event_sequence_resets_total` | Activity replay coverage dropped, by reason. `subscription_resumed` — a pub/sub connection dropped and resubscribed, dropping that workspace's buffer; expect it during a Redis failover and expect it to stop afterwards. `epoch_change` — the shared counter's ID space changed generation, dropping every buffer; expect a handful per cutover. `counter_backward` — an ID arrived at or below a buffer's high-water mark with no generation change; see *Event ID-space migration* for what to expect per phase. `epoch_regressed` — a LOWER generation was seen, so this instance stopped vouching for its buffers. One alongside an `epoch_change` is a message that was in flight when the generation rotated; a RUN of them means the counter itself went backwards — usually Redis lost writes, and since BUG-2740 possibly a repaired generation key (see *A repaired generation counter*). `undecodable_message` — a message on these channels could not be parsed, so that workspace's coverage ended; expect zero, and suspect a namespace collision. `subscription_unconfirmed` — a subscription was admitted before Redis acknowledged the SUBSCRIBE and the acknowledgement then arrived, so the span in between is one that stream cannot account for; it reaches THIS counter only when a buffer existed to drop, so read `pad_event_subscription_unconfirmed_total` for the dependable count. `idle_timeout` — a subscription received nothing at all (no event, no heartbeat, no acknowledgement) for longer than the idle timeout, so this instance stopped vouching for its buffer. It means **coverage ended, not that the connection was replaced**: the replacement is attempted afterwards and installs nothing if the instance is shutting down or the workspace loses its last subscriber, so only `pad_event_subscription_cycled_total` proves a replacement. Unlike `subscription_resumed` it does NOT establish that events went missing, only that the socket stopped proving it works, and like `subscription_unconfirmed` it reaches this counter only when a buffer existed to drop | | `pad_event_events_dropped_total` | Activity events not delivered to a live subscriber, by reason — today only `slow_subscriber` (that connection's 64-deep channel was full). Per-SUBSCRIBER: every subscriber that was keeping up received the event. Pairs with `pad_event_midstream_resyncs_total`, though not one-for-one in either direction — see that row. New in BUG-2730, along with the fix that stops the drop being silent, so a deploy that starts reporting these is not necessarily a regression — it may be the first time they were countable | +| `pad_event_subscription_cycled_total` | Activity-stream workspace subscriptions torn down **and replaced** because nothing arrived on them — no event, no heartbeat, no acknowledgement — within the idle timeout. It counts replacements, not teardowns: a cycle that installed nothing because the instance was shutting down or the workspace lost its last subscriber does not increment it, so a restart cannot manufacture this signal. Detects a **half-open connection**: no FIN, no RST, just a route that stopped working, which go-redis cannot see because its pub/sub health check writes a PING and never reads the reply. **Expect zero.** Read this rather than `pad_event_sequence_resets_total{reason="idle_timeout"}`, which moves only when a buffer existed to drop and so under-reports exactly the early-wedge case this detector exists for. A non-zero rate means connections to Redis are being silently blackholed — a NAT idle timeout, a stateful firewall, an overlay network dropping long-lived flows; check TCP keepalive on the path before changing the interval. **On heartbeat phase 1 this counter is structurally zero** — detection is part of phase 2, so a zero there says nothing at all about whether any route has wedged. Read `heartbeat_phase` off the startup log before drawing any conclusion from it | | `pad_event_subscription_unconfirmed_total` | Activity-stream subscriptions admitted before Redis acknowledged the SUBSCRIBE, because the wait for it timed out (BUG-2747). **Expect zero.** Counts ESTABLISHMENTS, not clients — one workspace subscription that timed out increments it once however many subscribers were waiting on it. Nothing is known to have been lost; what it says is that a stream was admitted whose coverage this instance cannot describe, and that every subscriber waiting on it will be told to reconcile when the acknowledgement lands. A non-zero rate means the SUBSCRIBE round trip is slow or stalling — read it alongside SSE connect latency rather than alongside `pad_event_sequence_resets_total` | | `pad_event_receive_loop_exits_total` | A workspace's activity subscription loop stopped. Unlike the watch stream's twin this does **not** stay at zero — it is expected at shutdown and whenever a workspace's last local subscriber leaves. Read it as a rate against a stable subscriber count | | `pad_session_presence_failures_total` | Presence operations failing — **read the `op` label**, the risks differ and run in opposite directions: `register`/`renew` may under-report (a live session unlisted and untargetable), `deregister` may over-report (a dead session left listed, and a push aimed at it reaches nobody), `list` returns a 503, `prune` is benign. A failure means the operation reported an error — Redis can fail a pipeline after applying it, so the write may have landed anyway | @@ -756,6 +763,167 @@ rather than probabilities, and neither reachable by a process that has to bind a listener and open a database before it can publish anything. A clock stepped **backwards** across a restart degrades the other way, into extra `sync_required` responses rather than wrong replays. +#### Half-open connection detection (`PAD_EVENTS_HEARTBEAT`) + +**The problem this fixes.** A TCP connection can stop carrying traffic without +closing — no FIN, no RST, just a route that stopped working. A NAT table +expiring, a stateful firewall dropping an idle flow, an overlay network +silently rerouting. The instance behind it blocks on a read that will never +return, receives nothing, and its replay buffer goes on looking complete. Every +resume for that workspace is then answered "caught up" from a coverage window +that ended when the route did — silent loss, with nothing in any metric. + +**Why go-redis does not cover it.** Its pub/sub health check writes a `PING` +and never reads a reply, so its error stays nil for as long as the socket +accepts writes — which a half-open socket does until its send buffer fills. The +channel path sets no read deadline either. Measured, not assumed: against a TCP +proxy that silently stopped forwarding, with the health check running, there +was no reconnect in 24 seconds. + +**What the fix does.** Every subscription records when it last received +anything — an event, a subscription acknowledgement, or a heartbeat. When that +goes stale past the idle timeout, the instance ends the workspace's replay +coverage (so the next resume answers `sync_required` rather than "caught up") +**and replaces the connection**. Dropping coverage alone would not recover: the +resync it demands is served from the same dead socket, and the detector fires +again on the next pass — a loop metering the failure rather than fixing it. + +**Why a heartbeat, rather than just a threshold on real traffic.** "Is this +workspace quiet, or is the route dead?" cannot be answered from traffic — it +depends on your publish rate, and no constant is right for every deployment. +Publishing our own frame replaces it with "did our heartbeat arrive?", which is +answerable everywhere. The instance publishes one frame per subscribed workspace +every **30 seconds** (T), and cycles a subscription that has received nothing for +**90 seconds** (3T). Three intervals rather than two so a single lost or late +frame is not a cycle. Detection latency measured from the last frame that got +through is 90–120s — the scan runs on its own 30s cadence, which adds up to one +interval on top of the threshold. Measured from the moment the route actually +died it is wider, roughly 60–120s: the publisher runs on an independent +schedule, so the last frame through may have been sent anywhere in the interval +before the fault. + +**Detection is part of phase 2, not phase 1.** Publishing and detecting are one +capability with one switch, because an instance detects off its *own* frames — +it publishes to the workspace channels it subscribes to and receives them back, +so it never depends on peers having flipped. A phase-1 instance therefore +detects nothing; it only recognises the frame so that a phase-2 peer costs it +nothing. Splitting them was tried and is wrong: with no heartbeat and no +events, a perfectly healthy *quiet* workspace crosses the threshold every +90–120s and gets cycled, which is a resync storm on the default configuration +every deployment lands in first. + +**It rolls out in two phases, and the order is not optional.** + +| Phase | What you do | What instances publish | What they do with a frame | +|-------|-------------|------------------------|---------------------------| +| 1 | Roll the new binary everywhere. Leave `PAD_EVENTS_HEARTBEAT` unset. | No heartbeats | Recognise and ignore it. **No idle detection.** | +| 2 | Set `PAD_EVENTS_HEARTBEAT=true` and roll again. | One frame per subscribed workspace per 30s | Recognise and ignore it. **Idle detection active.** | + +**What happens if you run them out of order.** The frame has to travel on the +workspace's *event* channel, because that channel's connection is the thing +whose liveness is in question — a probe anywhere else proves the wrong thing. +An instance running a **pre-phase-1** binary cannot classify it: the frame falls +through to the event decoder, fails to parse, and is treated as a hole in +coverage. That instance drops the workspace's replay buffer **and tells every +one of its live subscribers to resync** — every 30 seconds, for every workspace, +for as long as the deployment is mixed. The blast radius is the instances you +have *not* upgraded, which no amount of care in the new code can reach. This is +noisier than the ID-space migration's equivalent mistake and it is the reason +the default is off. + +Both rolls are zero-loss in the other direction: phase-1 instances recognise the +frame from the release that introduces it, so during the phase-2 roll a mix of +publishing and non-publishing instances is exactly the case ignore-the-frame +exists for. + +**Rolling back to phase 1** is safe and takes effect immediately: make the +effective value **false** and roll. Peers ignore the frame throughout, and idle +detection stops with it — you are back to the pre-BUG-2738 behaviour, which is +a wedged route going unnoticed, not a worse one. The same two wrinkles as +the ID-space migration apply, for the same reasons: + +- **Setting the value to false is not the same as unsetting the environment + variable.** `events_heartbeat` can also be set in `~/.pad/config.toml`, and + the file's value stands when the environment variable is absent. Clear both, + or set the environment variable explicitly to `false`. +- **Downgrading past phase 1 is a SECOND step, in the reverse order.** A + pre-phase-1 binary still cannot classify the frame. Roll every instance to + phase 1 (new binary, flip off), let it finish, *then* downgrade the binary. + +**The frame is validated, not just prefix-matched.** A liveness frame is +`hb|` plus optional short tokens, under a length cap. Anything else +that happens to begin with `hb|` is treated exactly as any other unreadable +payload: that workspace's coverage ends and +`pad_event_sequence_resets_total{reason="undecodable_message"}` moves, which is +the signal that says *suspect a namespace collision*. A forged frame cannot +fake liveness in any case — liveness means "this socket carried traffic", and a +frame that arrives demonstrates that whoever sent it. + +There is no Redis or database migration in either direction, and the frames are +never persisted: a heartbeat consumes no event ID, carries no epoch, is never +buffered or replayed, never reaches a subscriber, and is never counted as an +event. That last part is load-bearing rather than tidy — three of this bus's +reset reasons (`counter_backward`, `epoch_change`, `epoch_regressed`) are +derived from the shared ID counter, so a probe that consumed IDs would +manufacture the resets it exists to avoid. + +**Which phase an instance publishes in is in its startup log**, as +`heartbeat_phase=1` or `heartbeat_phase=2` on the "Event bus using Redis pub/sub" +line, alongside `id_space_phase`. The two migrations are independent — any +combination is valid. An unparseable `PAD_EVENTS_HEARTBEAT` is ignored and logs +a warning naming the value. + +**What this covers, and what it does not.** It is a *receive-side* detector, +not a round-trip health check. It measures whether frames arrive on a +workspace's subscription, so: + +- A subscription whose *outbound* direction is broken but which still receives + looks healthy — correctly, since nothing is being lost. +- The **PUBLISH path is not covered and cannot be.** `PUBLISH` travels on the + client's ordinary connection pool while a subscription holds a connection + from a separate pub/sub pool; those are different sockets with different + fates, and a reconnect of one repairs nothing about the other. An instance + whose publish path is wedged loses its own events for every other instance, + and this feature will not tell you. +- **The replacement is attempted, not guaranteed.** If the path is still + blackholed when the cycle re-dials, the new connection cannot receive either + and the detector fires again on the next pass. Coverage stays ended + throughout, so nothing is ever falsely claimed — but delivery resuming is a + statement about your network, not about Pad. One case where the replacement + can fail on a *healthy* path is tracked as BUG-2764: go-redis discards the + error from the initial `SUBSCRIBE`, so a failed subscribe yields a connection + that looks live and is subscribed to nothing. The detector cycles it again on + the next pass, which is why this self-heals on phase 2 and does not on phase 1. + +**What to watch.** `pad_event_subscription_cycled_total` — expect zero. Read it +rather than the `idle_timeout` reset label, which only moves when there was a +buffer to drop and therefore misses the early-wedge case. A non-zero rate is a +network fact about the path between your instances and Redis, not a Pad +condition: compare it against TCP keepalive settings on that path before +changing the interval, because a shorter interval treats the symptom and a +longer one widens the window the detector exists to bound. + +**A residual an operator should know about, not fixed here.** When many +workspaces are cycled at once — a NAT table flush, a firewall rule change, an +overlay network dropping every long-lived flow — every connected subscriber of +every affected workspace is told to resync in the same instant. The SSE +connections stay open, so this is *not* a reconnect storm and the admission +limits are not involved; what it produces is a burst of `/changes` requests +against the database, coalesced per browser tab but with no jitter and no +global budget. This is not new with the heartbeat: a Redis failover already +signals every workspace at once through `subscription_resumed`. What is new is +a second trigger of the same class. Tracked separately; if you run a large +fleet, watch database load alongside +`pad_event_sequence_resets_total` after any network event that could wedge many +routes simultaneously. Tracked as BUG-2761. + +**Cost.** Each workspace has its own Redis subscription — and therefore its own +connection — so liveness is genuinely per-workspace and there is no cheaper +shared probe. An instance subscribed to N workspaces publishes N frames every +30s; at N=1000 that is roughly 33 publishes/sec, which is noise for Redis. If +fleet workspace counts ever make it matter, the fix is connection +consolidation, not a longer interval. + ### Security | Variable | Default | Description | diff --git a/internal/config/config.go b/internal/config/config.go index afcd228f..0d7c20d1 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -143,6 +143,50 @@ type Config struct { // docs/deployment.md for the full procedure in both directions. EventsPublishEpoch bool `toml:"events_publish_epoch"` + // EventsHeartbeat turns on PHASE 2 of the activity-bus heartbeat rollout + // (BUG-2738): this instance PUBLISHES a bus-internal liveness frame on each + // workspace channel it is subscribed to, every 30s. + // + // WHY A HEARTBEAT AT ALL. A half-open Redis connection — no FIN, no RST, + // just a route that stopped working — leaves an instance blocked on a read + // forever while its replay buffer goes on looking complete, so every resume + // is answered "caught up" from a coverage window that ended when the route + // did. go-redis cannot see it: its pub/sub health check writes a PING and + // never reads the reply. Idle detection can, but only if silence is + // diagnostic — and on a quiet workspace it is not. Publishing our own + // traffic replaces "is this workspace quiet or is the route dead?" with + // "did our heartbeat arrive?", which is answerable on every deployment. + // + // IT IS A TWO-PHASE FLIP, AND THE ORDER IS NOT OPTIONAL — for the same + // mechanical reason as EventsPublishEpoch, but with a WORSE failure if you + // get it wrong. The frame must travel on the workspace's EVENT channel, + // because that is the connection whose liveness is in question. An instance + // running an OLDER binary cannot classify it: it falls through to the event + // decoder, fails, and — since BUG-2739 — treats the failure as a hole in + // coverage, dropping that workspace's replay buffer AND telling every one + // of its live subscribers to resync. Every 30 seconds. For every workspace. + // For as long as the deployment is mixed. Phase 1: roll the new binary + // everywhere with this false; it recognises and ignores the frame, and does + // nothing else — no publishing and no detection. Phase 2: set it true and + // roll again, which turns both on together. + // + // PUBLISHING AND DETECTING ARE ONE SWITCH. An instance detects off its own + // frames — it publishes to the channels it subscribes to and receives them + // back — so a phase-1 instance does no idle detection at all. Splitting + // them was tried and is wrong: with neither heartbeat nor events, a healthy + // QUIET workspace crosses the threshold every 90-120s and is cycled, which + // is a resync storm on the default configuration. + // + // Rolling BACK is safe and immediate: set the EFFECTIVE value false and + // roll. Peers ignore the frame throughout, and detection stops with it — + // back to the pre-BUG-2738 behaviour, not to a worse one. The same two wrinkles as + // EventsPublishEpoch apply — unsetting the environment variable is not the + // same as setting it false, because config.toml's value stands when the + // variable is absent; and downgrading PAST phase 1 is a second step in the + // reverse order, because a pre-phase-1 binary still cannot classify the + // frame. See docs/deployment.md for the procedure in both directions. + EventsHeartbeat bool `toml:"events_heartbeat"` + // Push carries per-USER push/consent preferences (PLAN-2613 S2). A // pointer so an absent `[push]` table stays nil and Save() (via the // omitempty tag) never writes an empty table into everyone's @@ -407,6 +451,30 @@ func Load() (*Config, error) { "value", v) } } + if v := os.Getenv("PAD_EVENTS_HEARTBEAT"); v != "" { + if on, err := strconv.ParseBool(v); err == nil { + cfg.EventsHeartbeat = on + } else { + // LOUD, and note that the SAFE DIRECTION IS THE OPPOSITE OF + // PAD_EVENTS_PUBLISH_EPOCH's (BUG-2738). There, leaving the flip + // OFF was the data-LOSING direction and the guard existed to stop + // a typo carrying a deployment FORWARD into a phase its peers + // could not read. Here OFF is the SAFE direction: an instance that + // publishes no heartbeat does no detection at all, which is the + // behaviour that existed before this feature, while one that + // publishes into a mixed fleet resyncs every client of every + // un-upgraded instance every 30 seconds. So the ignore is + // the conservative outcome in both cases and the reasoning is + // inverted — do not copy the epoch flag's rationale onto this one. + // + // It must still be LOUD for the epoch flag's reason, which does + // carry over: an operator who typed "yes" believes phase 2 is on, + // the value is ignored, and a silent ignore makes that + // indistinguishable from a phase-1 deployment in every metric. + slog.Warn("PAD_EVENTS_HEARTBEAT is not a boolean and was ignored; this instance keeps its current heartbeat phase", + "value", v) + } + } if v := os.Getenv("PAD_SSE_MAX_PER_USER"); v != "" { if max, err := strconv.Atoi(v); err == nil { cfg.SSEMaxPerUser = max diff --git a/internal/config/redis_stream_config_test.go b/internal/config/redis_stream_config_test.go index 1bbc13c0..d311e7d1 100644 --- a/internal/config/redis_stream_config_test.go +++ b/internal/config/redis_stream_config_test.go @@ -186,3 +186,163 @@ func TestEventsPublishEpochPrecedenceBetweenEnvAndFile(t *testing.T) { } }) } + +// --------------------------------------------------------------------------- +// BUG-2738's phase-2 flip. Structurally these mirror the EventsPublishEpoch +// tests above, and the ASSERTIONS are the same — but the RATIONALE is +// inverted, which is why they are written out rather than folded into a table +// with the epoch's comments attached. See TestEventsHeartbeatIgnoresANonBooleanValue. +// --------------------------------------------------------------------------- + +// TestEventsHeartbeatEnvMapping is the wiring. The bus's heartbeat behaviour +// has its own tests and every one of them passes with Load() never populating +// this field — the deployment would simply stay on phase 1 forever, which is +// indistinguishable from a correct phase-1 deployment in every metric. +func TestEventsHeartbeatEnvMapping(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + t.Setenv("PAD_EVENTS_HEARTBEAT", "true") + + cfg, err := Load() + if err != nil { + t.Fatalf("Load: %v", err) + } + if !cfg.EventsHeartbeat { + t.Error("EventsHeartbeat = false, want true from PAD_EVENTS_HEARTBEAT") + } + // The neighbour, so a copy-paste that pointed two env vars at one field + // cannot pass. These two flags are the same SHAPE and are set in the same + // procedures, which is exactly when that mistake happens. + if cfg.EventsPublishEpoch { + t.Error("PAD_EVENTS_HEARTBEAT must not move EventsPublishEpoch") + } +} + +// TestEventsHeartbeatIgnoresANonBooleanValue. +// +// READ THIS COMMENT BEFORE "FIXING" THIS TEST TO MATCH ITS EPOCH TWIN. The +// assertion is identical to TestEventsPublishEpochIgnoresANonBooleanValue and +// the reason for it is the OPPOSITE one. +// +// For the epoch flip, OFF was the data-LOSING direction: an instance stuck on +// phase 1 published a wire form nothing could misread, and the hazard was a +// typo carrying a deployment FORWARD into a phase its peers could not parse. +// +// Here OFF is the SAFE direction. An instance that publishes no heartbeat does +// no idle detection at all — exactly the behaviour that existed before this +// feature, so the worst case of a wrong OFF is that a wedged route goes +// unnoticed, which is where every deployment already was. An instance that publishes into a MIXED +// fleet makes every un-upgraded peer fail to decode the frame, drop that +// workspace's replay buffer, and tell every one of its live subscribers to +// resync — every 30 seconds, per workspace, for the length of the roll. The +// blast radius is the instances you have NOT upgraded, which no amount of care +// in the new code can reach. +// +// So: same assertion, opposite hazard. Copying the epoch's rationale onto this +// test would leave a comment arguing for the wrong thing, and the next person +// to touch it would "fix" the behaviour to match its own comment. +func TestEventsHeartbeatIgnoresANonBooleanValue(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + t.Setenv("PAD_EVENTS_HEARTBEAT", "yes-please") + + cfg, err := Load() + if err != nil { + t.Fatalf("Load: %v", err) + } + if cfg.EventsHeartbeat { + t.Error("an unparseable value must not turn the flip on; publishing into a mixed fleet resyncs every client of every un-upgraded instance") + } +} + +// The precise contract, stated because the sentence above is easy to read as +// something stronger than it is (codex round 9): an unparseable value is +// IGNORED, not read as false. From a default config that leaves the flip off, +// which is what the test above checks. From a config file that set it true it +// leaves it TRUE — deliberately, and the same as the epoch flag: a typo must +// not move a migration in either direction, and silently rolling an operator +// back to phase 1 would disable detection on a fleet that had opted in without +// anything saying so. The warning is what tells them. The file-true case is +// asserted in TestEventsHeartbeatPrecedenceBetweenEnvAndFile; this comment +// exists so nobody "fixes" the ignore into a fail-closed reset. + +// The default MUST be off, for the reason above: phase 2 emits a frame older +// instances treat as a hole in coverage, so defaulting it on would break a +// rolling upgrade for every deployment that upgrades without reading the +// release notes. +func TestEventsHeartbeatDefaultsOff(t *testing.T) { + if _, set := os.LookupEnv("PAD_EVENTS_HEARTBEAT"); set { + t.Setenv("PAD_EVENTS_HEARTBEAT", "") + } + if cfg := DefaultConfig(); cfg.EventsHeartbeat { + t.Error("default EventsHeartbeat = true, want false — phase 2 must be opted into after every instance recognises the frame") + } +} + +// The TOML tag, for the same reason its epoch twin has one: the env-var test +// above says nothing about the file, and a wrong or missing +// `toml:"events_heartbeat"` would keep every other test green while an +// operator who set the flag in ~/.pad/config.toml silently stayed on phase 1. +func TestEventsHeartbeatRoundTripsThroughTheConfigFile(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + t.Setenv("PAD_EVENTS_HEARTBEAT", "") + + cfg := DefaultConfig() + cfg.EventsHeartbeat = true + if err := cfg.Save(); err != nil { + t.Fatalf("save: %v", err) + } + + reloaded, err := Load() + if err != nil { + t.Fatalf("reload: %v", err) + } + if !reloaded.EventsHeartbeat { + t.Error("events_heartbeat did not survive a save/load round trip through config.toml") + } +} + +// The rollback procedure tells an operator to make the EFFECTIVE value false +// and warns that unsetting the environment variable is not the same thing. +// Both halves are asserted here for the same reason they are for the epoch +// flag — and here rollback is the direction an operator reaches for in a +// hurry, because the failure they are rolling back FROM is a fleet-wide resync +// storm. +func TestEventsHeartbeatPrecedenceBetweenEnvAndFile(t *testing.T) { + writeFileValue := func(t *testing.T) { + t.Helper() + cfg := DefaultConfig() + cfg.EventsHeartbeat = true + if err := cfg.Save(); err != nil { + t.Fatalf("save: %v", err) + } + } + + t.Run("an explicit env false overrides a true in the file", func(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + writeFileValue(t) + t.Setenv("PAD_EVENTS_HEARTBEAT", "false") + + cfg, err := Load() + if err != nil { + t.Fatalf("reload: %v", err) + } + if cfg.EventsHeartbeat { + t.Error("an explicit env-var false must win over the config file — this is the documented rollback") + } + }) + + t.Run("an unparseable env value leaves the file's value standing", func(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + writeFileValue(t) + t.Setenv("PAD_EVENTS_HEARTBEAT", "off-ish") + + cfg, err := Load() + if err != nil { + t.Fatalf("reload: %v", err) + } + // IGNORED, not read as false. A typo must not move a migration in + // either direction; the warning is what tells the operator. + if !cfg.EventsHeartbeat { + t.Error("an unparseable env value must leave the configured value alone, not reset it") + } + }) +} diff --git a/internal/events/observer.go b/internal/events/observer.go index fe99afd2..0d2ab088 100644 --- a/internal/events/observer.go +++ b/internal/events/observer.go @@ -116,6 +116,49 @@ type Observer interface { // stalling — the same Redis condition BUG-2748 makes an availability // hazard. SubscriptionUnconfirmed() + + // SubscriptionCycled reports that a workspace's Redis subscription received + // NOTHING — no event, no heartbeat, no acknowledgement — for longer than + // the bus's idle timeout, so its connection was torn down and replaced + // (BUG-2738). + // + // IT IS COUNTED SEPARATELY FROM SequenceReset FOR A REASON THAT IS NOT + // STYLISTIC. An idle cycle calls dropWorkspaceCoverage, which reports + // SequenceReset with reason idle_timeout — but ONLY when a buffer existed + // to drop, and the case this detector exists for is disproportionately the + // case with no buffer: a route that wedged early, on a quiet workspace, + // having delivered nothing this instance could buffer. Reading cycles off + // the reset counter alone would therefore under-report exactly the + // incidents it was built to find. This counter is the dependable one; the + // idle_timeout reset label is corroboration. + // + // Expect zero — and note that on heartbeat phase 1 it is zero STRUCTURALLY, + // because the detector does not run at all there, so a zero says nothing + // about whether any route has wedged. On phase 2, a non-zero rate means + // connections between this instance and Redis are being silently + // blackholed — a NAT idle timeout, a stateful firewall, an overlay network + // dropping long-lived flows. Compare against TCP keepalive settings on the + // path before tuning the interval, because tuning the interval treats the + // symptom. + SubscriptionCycled() + + // HeartbeatPublishFailed reports that this instance could not publish a + // liveness heartbeat for one workspace (BUG-2738). + // + // IT IS THE DETECTOR SAYING IT CANNOT SEE, not a finding about any peer. + // While it is firing, idle detection for that workspace is SUSPENDED — + // silence cannot be read as evidence when we could not ask — so a non-zero + // rate here means half-open detection is degraded or off for those + // workspaces, however healthy pad_event_subscription_cycled_total looks. + // + // PUBLISH and pub/sub use different connection pools, so this is a signal + // about the OUTBOUND path specifically: pool exhaustion, a wedged outbound + // route, or Redis refusing writes. An instance in this state is also + // failing to deliver its own events to every other instance, which is a + // larger problem than the one this feature exists to find. + // + // Expect zero. + HeartbeatPublishFailed() } // Drop reasons. Bounded by construction so they are safe as metric labels. @@ -191,8 +234,43 @@ const ( // deliberate asymmetry between the metric and the client signal. The // dependable counter for this condition is Observer.SubscriptionUnconfirmed. ResetReasonSubscriptionUnconfirmed = "subscription_unconfirmed" + + // ResetReasonIdleTimeout means a workspace's Redis subscription received + // nothing at all for longer than the bus's idle timeout, so this instance + // STOPPED VOUCHING FOR ITS BUFFER (BUG-2738). + // + // IT DOES NOT SAY THE CONNECTION WAS REPLACED, and an earlier version of + // this comment claimed it did (codex round 6). This reason is emitted + // before the re-establishment is attempted, and the attempt can install + // nothing — the bus closes, or the last subscriber leaves while we dial. + // Observer.SubscriptionCycled is the one that means "replaced"; this one + // means "coverage ended". + // + // WHAT IT ESTABLISHES IS NOT THAT EVENTS WERE LOST, unlike + // subscription_resumed: nothing was observed going missing. What it says is + // that the socket stopped proving it works, and a socket that cannot be + // proved cannot back a coverage claim. The silence includes this instance's + // own heartbeats, which is what makes it diagnostic rather than a guess + // about how busy the workspace is — and is why the detector only runs on + // heartbeat phase 2. On phase 1 this reason is structurally never emitted. + // + // It reaches this counter only when a buffer existed to drop. Read + // Observer.SubscriptionCycled for the dependable count — the no-buffer case + // is over-represented here for the reason recorded there. + ResetReasonIdleTimeout = "idle_timeout" ) +// THE ONE THING AN OBSERVER CALLBACK MUST NOT DO is call a Subscribe path on +// the bus that is reporting to it. +// +// Callbacks run synchronously, and several of the paths that report — a late +// subscription acknowledgement, an idle-fired cycle — do so while holding that +// workspace's establishment record. A Subscribe arriving there waits on a +// record only the reporting goroutine can retire, and the reporting goroutine +// is waiting on the callback: neither moves again. Publishing, reading, and +// unsubscribing from a callback are all fine and are exercised by this +// package's tests; subscribing is the one door that is closed. +// // observable is the shared, nil-safe Observer holder both bus implementations // embed. Reporting before SetObserver is called — every bus in every test that // does not opt in — is a no-op. @@ -240,6 +318,18 @@ func (o *observable) reportSubscriptionUnconfirmed() { } } +func (o *observable) reportSubscriptionCycled() { + if obs := o.observer(); obs != nil { + obs.SubscriptionCycled() + } +} + +func (o *observable) reportHeartbeatPublishFailed() { + if obs := o.observer(); obs != nil { + obs.HeartbeatPublishFailed() + } +} + func (o *observable) reportDropped(reason string) { if obs := o.observer(); obs != nil { obs.EventDropped(reason) diff --git a/internal/events/observer_test.go b/internal/events/observer_test.go index e4d2f928..2e60c5ec 100644 --- a/internal/events/observer_test.go +++ b/internal/events/observer_test.go @@ -6,12 +6,14 @@ import ( ) type recordingObserver struct { - mu sync.Mutex - resumeGaps []string - resets []string - loopExits int - drops []string - unconfirmed int + mu sync.Mutex + resumeGaps []string + resets []string + loopExits int + drops []string + unconfirmed int + cycled int + probeFailures int } func (o *recordingObserver) ResumeGap(workspaceID string) { @@ -44,6 +46,36 @@ func (o *recordingObserver) SubscriptionUnconfirmed() { o.unconfirmed++ } +func (o *recordingObserver) HeartbeatPublishFailed() { + o.mu.Lock() + defer o.mu.Unlock() + o.probeFailures++ +} + +func (o *recordingObserver) probeFailureCount() int { + o.mu.Lock() + defer o.mu.Unlock() + return o.probeFailures +} + +func (o *recordingObserver) SubscriptionCycled() { + o.mu.Lock() + defer o.mu.Unlock() + o.cycled++ +} + +func (o *recordingObserver) loopExitCount() int { + o.mu.Lock() + defer o.mu.Unlock() + return o.loopExits +} + +func (o *recordingObserver) cycledCount() int { + o.mu.Lock() + defer o.mu.Unlock() + return o.cycled +} + func (o *recordingObserver) unconfirmedCount() int { o.mu.Lock() defer o.mu.Unlock() @@ -190,3 +222,7 @@ func (o callbackObserver) EventDropped(string) {} func (o callbackObserver) ReceiveLoopExited() {} func (o callbackObserver) SubscriptionUnconfirmed() {} + +func (o callbackObserver) SubscriptionCycled() {} + +func (o callbackObserver) HeartbeatPublishFailed() {} diff --git a/internal/events/redis_blackhole_test.go b/internal/events/redis_blackhole_test.go new file mode 100644 index 00000000..43feeff3 --- /dev/null +++ b/internal/events/redis_blackhole_test.go @@ -0,0 +1,229 @@ +package events + +import ( + "bytes" + "context" + "net" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/alicebob/miniredis/v2" + "github.com/redis/go-redis/v9" + + "github.com/PerpetualSoftware/pad/internal/redisns" +) + +// blackholeProxy is a TCP proxy that can stop delivering server→client bytes on +// the connections it already holds, while continuing to accept and forward +// client→server on them and serving new connections normally. +// +// THAT ASYMMETRY IS THE WHOLE POINT, and it is what no other instrument in this +// package can produce. miniredis is a working Redis, so every unit test here +// has to SIMULATE a wedge by advancing a clock and stamping fields. This +// reproduces the real thing: a route that stopped carrying traffic without +// closing — no FIN, no RST, writes still accepted — which is precisely the +// failure go-redis's health check cannot see, because PubSub.Ping writes the +// command and never reads a reply. +// +// New connections keep working, so the replacement subscription can succeed +// and the test can assert RECOVERY rather than only detection. +type proxiedConn struct { + dark *atomic.Bool + isPubSub *atomic.Bool +} + +type blackholeProxy struct { + ln net.Listener + backend string + mu sync.Mutex + conns []proxiedConn +} + +func newBlackholeProxy(t *testing.T, backend string) *blackholeProxy { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + p := &blackholeProxy{ln: ln, backend: backend} + t.Cleanup(func() { _ = ln.Close() }) + + go func() { + for { + client, err := ln.Accept() + if err != nil { + return + } + server, err := net.Dial("tcp", backend) + if err != nil { + _ = client.Close() + return + } + + // PER-CONNECTION, NOT A GLOBAL FLAG, and the first version of this + // proxy got that wrong in a way that made the test vacuous: a + // global "dead" bool consulted at read time meant that re-enabling + // delivery for FUTURE connections also revived the ones that were + // supposed to be dark, so nothing was ever wedged and the test + // failed for the wrong reason. The connections open when blackhole() + // is called are the ones that go silent, permanently; anything + // dialled afterwards is healthy. + dark := &atomic.Bool{} + isPubSub := &atomic.Bool{} + p.mu.Lock() + p.conns = append(p.conns, proxiedConn{dark: dark, isPubSub: isPubSub}) + p.mu.Unlock() + + // OUTBOUND ALWAYS FLOWS, and the connection is CLASSIFIED as it + // does. go-redis puts PUBLISH on its ordinary connection pool and + // each subscription on a connection from a separate pub/sub pool; + // darkening both would break the probe as well as the delivery, + // and the test could then pass on an implementation that treats a + // failed probe as evidence of a dead peer — the exact defect the + // premise check exists to prevent (codex round 14). Only a + // connection that has carried a SUBSCRIBE goes dark. + go func() { + buf := make([]byte, 4096) + for { + n, err := client.Read(buf) + if n > 0 { + if bytes.Contains(bytes.ToLower(buf[:n]), []byte("subscribe")) { + isPubSub.Store(true) + } + if _, werr := server.Write(buf[:n]); werr != nil { + return + } + } + if err != nil { + return + } + } + }() + go func() { + defer func() { _ = client.Close(); _ = server.Close() }() + buf := make([]byte, 4096) + for { + n, err := server.Read(buf) + if n > 0 && !dark.Load() { + if _, werr := client.Write(buf[:n]); werr != nil { + return + } + } + if err != nil { + return + } + } + }() + } + }() + return p +} + +func (p *blackholeProxy) addr() string { return p.ln.Addr().String() } + +// blackhole stops inbound delivery on every connection currently open, for +// good. Writes on those connections keep succeeding, which is what makes this +// a half-open route rather than a disconnection — and is exactly the state +// go-redis reports as healthy, because PubSub.Ping writes without reading. +// +// Connections opened afterwards are unaffected, so the replacement subscription +// can succeed and the test can assert RECOVERY rather than only detection. +func (p *blackholeProxy) blackhole() { + p.mu.Lock() + defer p.mu.Unlock() + for _, c := range p.conns { + if c.isPubSub.Load() { + c.dark.Store(true) + } + } +} + +// TestAWedgedRouteIsDetectedEndToEnd is the integration test for BUG-2738's +// central claim, and the only test here that exercises a REAL half-open socket +// rather than a simulated one (codex round 13, P3). +// +// Everything else in this package drives the mechanism through a fake clock: +// necessary, because the threshold is 90 seconds by construction and miniredis +// always answers, but it means every one of those tests assumes the wedge +// rather than producing it. This one produces it — the bus's own heartbeats +// keep reaching Redis while nothing comes back — and asserts both halves of +// the claim: the connection is cycled, and delivery resumes on the replacement. +// +// It runs on real time with a compressed cadence, so it is deliberately the +// slowest test in the file. +func TestAWedgedRouteIsDetectedEndToEnd(t *testing.T) { + mr := miniredis.RunT(t) + proxy := newBlackholeProxy(t, mr.Addr()) + + client := redis.NewClient(&redis.Options{Addr: proxy.addr()}) + t.Cleanup(func() { _ = client.Close() }) + + b := NewRedisBusWithKeys(client, redisns.Default, false, true) + obs := &recordingObserver{} + b.SetObserver(obs) + t.Cleanup(b.Close) + + ch, _, outcome := b.Subscribe(context.Background(), "ws-1") + if outcome != SubscribeOK { + t.Fatalf("subscribe: %v", outcome) + } + defer b.Unsubscribe(ch) + + // Prove the route works before breaking it, so a test that never delivered + // anything cannot pass by looking like a successful detection. + publisher := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + t.Cleanup(func() { _ = publisher.Close() }) + b.Publish(Event{Type: ItemCreated, WorkspaceID: "ws-1", ItemID: "before"}) + select { + case ev := <-ch: + if ev.ItemID != "before" { + t.Fatalf("fixture: unexpected event %+v", ev) + } + case <-time.After(5 * time.Second): + t.Fatal("fixture: the route never worked, so wedging it proves nothing") + } + + // Break the receive direction only. Writes keep succeeding, so the bus goes + // on publishing heartbeats it will never see come back — exactly the state + // no health check in go-redis can observe. + proxy.blackhole() + + b.setMaintenanceCadence(50*time.Millisecond, 200*time.Millisecond) + + deadline := time.Now().Add(20 * time.Second) + // THE PROBE MUST KEEP SUCCEEDING while nothing comes back — that pairing IS + // the half-open case, and without asserting it this test would also pass on + // an implementation that cycles because it could not publish at all + // (codex round 14). Only the subscription's connection is darkened, so the + // publish path stays healthy and this stays at zero. + defer func() { + if got := obs.probeFailureCount(); got != 0 { + t.Fatalf("%d heartbeat publishes failed: this run exercised the cannot-probe path, not a half-open route", got) + } + }() + for obs.cycledCount() == 0 { + if time.Now().After(deadline) { + t.Fatalf("a wedged route was never detected in 20s (probe failures: %d): go-redis cannot see this and neither can we", + obs.probeFailureCount()) + } + time.Sleep(20 * time.Millisecond) + } + + // ...and the replacement actually delivers, which is the half that + // distinguishes recovery from a resync loop. + for { + if time.Now().After(deadline) { + t.Fatal("the workspace was cycled but the replacement never delivered anything") + } + b.Publish(Event{Type: ItemUpdated, WorkspaceID: "ws-1", ItemID: "after"}) + select { + case ev := <-ch: + if ev.ItemID == "after" { + return + } + case <-time.After(200 * time.Millisecond): + } + } +} diff --git a/internal/events/redis_bus.go b/internal/events/redis_bus.go index 01a6f0d9..843f5d7a 100644 --- a/internal/events/redis_bus.go +++ b/internal/events/redis_bus.go @@ -504,6 +504,49 @@ type RedisBus struct { // connections and amplifying the outage. confirmTimeout time.Duration + // heartbeatInterval is T and idleTimeout is 3T: how often this instance + // publishes a liveness frame per subscribed workspace, and how long a + // subscription may receive nothing at all before its coverage ends and its + // connection is replaced (BUG-2738). Tunables with the ruled defaults; see + // DefaultHeartbeatInterval and cycleIdleSubscriptions. + heartbeatInterval time.Duration + idleTimeout time.Duration + + // heartbeatKick and idleKick wake their loops when the cadence above + // changes, so a new interval takes effect at once instead of after the old + // one expires. Buffered depth 1 and written non-blockingly: each is a + // signal that the values moved, not a queue of changes. + // + // ONE PER LOOP because the two run in separate goroutines (see + // maintenanceLoop); a shared channel would be consumed by whichever was + // waiting and leave the other on the stale cadence. + heartbeatKick chan struct{} + idleKick chan struct{} + + // maintenanceStopped is closed when maintenanceLoop returns. + // + // IT EXISTS BECAUSE THE LOOP'S TEARDOWN IS OTHERWISE UNOBSERVABLE, which + // makes it untestable and therefore unprotected. Close drains wsSubs, so a + // loop that ignored b.ctx entirely would find no workspaces and publish + // nothing — indistinguishable from a loop that stopped, while it went on + // waking every interval for the life of the process. Same reason + // Observer.ReceiveLoopExited exists for the receive goroutines. + maintenanceStopped chan struct{} + + // publishHeartbeat selects whether this instance EMITS liveness frames: + // PHASE 2 of the heartbeat rollout. Receiving instances recognise and + // ignore them from the release that introduced this field, so emission is + // the half that is gated — see config.EventsHeartbeat for why the order is + // not optional. Constructor parameter with no default, the same shape as + // publishEpoch, so every call site states which phase it is in. + publishHeartbeat bool + + // nowFunc overrides the clock behind idle detection. Nil in every real + // construction; tests set it so a 90s threshold can be crossed without + // sleeping through one. Distinct from nowUnix, which seams a different + // clock for a different reason (BUG-2740's generation repair). + nowFunc func() time.Time + // afterSubscribeRegister is a TEST SEAM, nil in production. It runs // inside SubscribeAndReplaySince's critical section, after the subscriber // is registered and before the replay is read — the only point at which @@ -553,6 +596,36 @@ type RedisBus struct { // correct abandon. Receives the workspace. afterRegisterBeforeEstablish func(workspaceID string) + // afterProbePublish is a TEST SEAM, nil in production. It runs in + // publishHeartbeats after a heartbeat has been published for one workspace + // and BEFORE its lastProbeOK is stamped. Receives the workspace. + // + // POSITIONAL: that gap is exactly where a slow publish lets the + // subscription it was sent for be replaced, and stamping the replacement + // would credit it with a probe it never received. It is the only place a + // test can make that interleave happen on purpose. + afterProbePublish func(workspaceID string) + + // afterIdleScan is a TEST SEAM, nil in production. It runs in + // cycleIdleSubscriptions after the scan has selected its victims and + // RELEASED b.mu, and before any of them is cycled. + // + // POSITIONAL, like its siblings. That gap is the whole subject of the + // freshness re-check in cycleOne: in production it is widened by the + // concurrency cap and by GC pauses, and it is the only place a test can + // make a selected workspace start receiving again before its turn. + afterIdleScan func() + + // afterCycleEstablish is a TEST SEAM, nil in production. It runs in + // cycleOne after establishSubscription returns and BEFORE the cycle decides + // whether to count a replacement. + // + // POSITIONAL: that is the one point at which an UNRELATED caller's fresh + // subscription can be mistaken for this cycle's replacement, which is the + // misattribution the explicit installed result exists to prevent. Receives + // the workspace. + afterCycleEstablish func(workspaceID string) + // beforeInstallSubscription is a TEST SEAM, nil in production. It runs in // establishSubscription after the dial and BEFORE the lock that decides // whether to install or abandon, so a test can make either abandon reason @@ -628,6 +701,48 @@ type redisSub struct { // recognised as belonging to a subscription that has already ended. gen int64 + // lastSeen is when this subscription last received ANYTHING from Redis: + // an event, a heartbeat, or a subscription confirmation. Guarded by b.mu. + // + // WHAT IS BEING MEASURED IS WHETHER THE SOCKET CARRIES TRAFFIC, not + // whether the workspace is busy — which is why every inbound frame stamps + // it rather than only the ones that turn into events, and why it is + // stamped at INSTALL time too. A zero value would read as 1970 and cycle a + // subscription that has simply not been given the chance to receive + // anything yet; see cycleIdleSubscriptions' rule 2. + lastSeen time.Time + + // lastProbeOK is when this instance last SUCCEEDED in publishing a + // heartbeat for this workspace. Guarded by b.mu. + // + // IT IS THE DETECTOR'S PREMISE, not bookkeeping (codex round 13). Idle + // detection reasons "we published a frame and nothing came back, so the + // receive path is dead". That inference is only valid if the publish + // actually happened. PUBLISH travels on the client's connPool while the + // subscription holds a connection from the separate pubSubPool, so a + // publish-side failure — pool exhaustion, a wedged outbound path — says + // nothing whatever about whether this subscription can receive. Without + // this field the detector reads its own inability to probe as evidence + // that the peer is dead, and tears down a healthy connection on a schedule. + // + // Stamped at install, and the mutation matrix says that stamp is REDUNDANT + // under the ordering rule — recorded rather than left as an unearned + // justification. An earlier version of this comment claimed a zero value + // would "permanently disqualify a subscription from ever being cycled". + // That was true of the age-based premise it was written for; it is not true + // now. The rule is `lastProbeOK.After(lastSeen)`, and a zero value fails + // that test exactly as an install stamp equal to lastSeen does — in both + // cases the subscription is simply not cycled until its first successful + // probe, which is the intended behaviour either way. Removing the stamp + // changes no outcome and no test. + // + // It is kept because it makes the field's invariant true by construction — + // an installed subscription always carries a real timestamp, so any future + // rule that reasons about this value's AGE rather than its order gets a + // sane one instead of 1970. That is the same trap the age-based rule fell + // into, one field over. + lastProbeOK time.Time + // confirmed is closed by receiveMessages when Redis acknowledges the // SUBSCRIBE for this subscription (BUG-2747). Subscribe waits on it — up to // confirmTimeout, after which it admits anyway and says so, see @@ -673,7 +788,7 @@ type pendingSub struct { // NewRedisBus creates a new Redis-backed EventBus. // The provided redis.Client should already be configured and connected. func NewRedisBus(client *redis.Client) *RedisBus { - return NewRedisBusWithKeys(client, redisns.Default, false) + return NewRedisBusWithKeys(client, redisns.Default, false, false) } // NewRedisBusWithKeys is NewRedisBus with an explicit key namespace @@ -681,28 +796,65 @@ func NewRedisBus(client *redis.Client) *RedisBus { // shared with the watch bus and the presence registry so all three // keyspaces carry the same namespace or none. // +// TWO INDEPENDENT ROLLOUT FLAGS, in this order, and they are NOT +// interchangeable despite being adjacent booleans of the same type — the +// hazard being that a maintenance edit swaps or drops one silently +// (codex round 9). publishEpoch is BUG-2736's ID-space migration; publishHeartbeat +// is BUG-2738's half-open-connection detection. Any combination is valid and +// each has its own phase in the startup log. +// +// publishHeartbeat turns on PHASE 2 of the heartbeat rollout: this instance +// publishes a bus-internal liveness frame per subscribed workspace AND runs +// idle detection. Those are one switch on purpose — an instance detects off +// its own frames, so detecting without publishing cycles healthy quiet +// workspaces; see cycleIdleSubscriptions and config.EventsHeartbeat. +// // publishEpoch selects the wire form this instance EMITS (BUG-2736). It is a // constructor parameter with no default rather than a setter, so every call // site states which phase of the rollout it is in and none can flip a bus that // is already publishing. See config.EventsPublishEpoch for the order the two // phases must be rolled in. -func NewRedisBusWithKeys(client *redis.Client, keys redisns.Keys, publishEpoch bool) *RedisBus { +func NewRedisBusWithKeys(client *redis.Client, keys redisns.Keys, publishEpoch, publishHeartbeat bool) *RedisBus { ctx, cancel := context.WithCancel(context.Background()) - return &RedisBus{ - client: client, - keys: keys, - publishEpoch: publishEpoch, - subscribers: make(map[string]map[chan Event]*subscriber), - workspaceOf: make(map[chan Event]string), - wsCounts: make(map[string]int), - wsSubs: make(map[string]*redisSub), - pendingSubs: make(map[string]*pendingSub), - replayBuffers: make(map[string]*replayBuffer), - replaySize: DefaultReplayBufferSize, - confirmTimeout: defaultSubscribeConfirmTimeout, - ctx: ctx, - cancel: cancel, + b := &RedisBus{ + client: client, + keys: keys, + publishEpoch: publishEpoch, + publishHeartbeat: publishHeartbeat, + subscribers: make(map[string]map[chan Event]*subscriber), + workspaceOf: make(map[chan Event]string), + wsCounts: make(map[string]int), + wsSubs: make(map[string]*redisSub), + pendingSubs: make(map[string]*pendingSub), + replayBuffers: make(map[string]*replayBuffer), + replaySize: DefaultReplayBufferSize, + confirmTimeout: defaultSubscribeConfirmTimeout, + heartbeatInterval: DefaultHeartbeatInterval, + idleTimeout: DefaultIdleTimeout, + heartbeatKick: make(chan struct{}, 1), + idleKick: make(chan struct{}, 1), + maintenanceStopped: make(chan struct{}), + ctx: ctx, + cancel: cancel, } + // NOT STARTED AT ALL ON PHASE 1 (codex round 4, P3). Both halves are gated + // on publishHeartbeat and would be guaranteed no-ops there, so the loop + // would be two goroutines and two timers per process waking every 30s for + // the life of a deployment that has asked for none of it — and the DEFAULT + // deployment is phase 1. The flag is constructor-only, so this decision can + // be taken once and cannot go stale. + // + // The in-function gates stay regardless: they are the correctness ones + // (see cycleIdleSubscriptions for what a phase-1 detector does to a quiet + // workspace), and direct callers — the tests — reach them without a loop. + if publishHeartbeat { + go b.maintenanceLoop() + } else { + // Nothing will ever run, so the teardown signal is already true; a + // caller waiting on it must not hang. + close(b.maintenanceStopped) + } + return b } // Subscribe registers a local subscriber for the given workspace. @@ -1232,17 +1384,52 @@ func (b *RedisBus) eventsSinceLocked(workspaceID string, sinceID int64) []Event } // Close shuts down all Redis subscriptions and closes local subscriber channels. +// +// IT DOES NOT JOIN THE MAINTENANCE GOROUTINES (BUG-2738, codex round 3), and +// that is a choice rather than an omission. Their publish half makes +// synchronous Redis calls bounded by go-redis's own Dial/Read/WriteTimeout — +// exactly the calls that stall on the wedged route this whole feature exists +// to detect — so joining them would let a dead network hold shutdown open for +// as long as those timeouts take. maintenanceStopped is available for a caller +// that genuinely wants to wait; nothing in production does. +// +// What holds instead is that a cycle already past its own ctx check cannot +// leave anything behind: establishSubscription re-checks b.ctx under its +// deciding lock and abandons there, closing the PubSub and retiring the record +// in the same critical section, and the dial dies with b.ctx through +// mergeCancellation (except under TLS, where DialTimeout bounds it — see that +// function). Pinned by TestClosingTheBusDuringACycleInstallsNothing. func (b *RedisBus) Close() { b.cancel() // signal all subscription goroutines to stop b.mu.Lock() - defer b.mu.Unlock() - + // COLLECTED UNDER THE LOCK, CLOSED AFTER IT (codex round 13). Same reason + // stopRedisSubscription hands its close off: PubSub.Close takes go-redis's + // mutex, which the health check can hold across reconnect work, so closing + // here would block shutdown inside the lock that every fan-out and every + // Subscribe contends for — with subscriber channels still open behind it. + // Round 12 fixed the cycle path and left this one, which is the same defect + // on the path that runs once per process. + // + // UNTESTED, DELIBERATELY. Moving a close off a lock is a CONTENTION + // property: the only assertion that distinguishes it is a timing one — how + // long some other goroutine waited for b.mu — and a timing assertion in + // this suite is a flaky assertion. The mutation matrix says so plainly + // (closing under the lock survives every test), and that survival is + // recorded here rather than papered over with a test that would pass + // either way. + closing := make([]*redis.PubSub, 0, len(b.wsSubs)) for wsID, sub := range b.wsSubs { sub.cancel() - sub.pubsub.Close() + closing = append(closing, sub.pubsub) delete(b.wsSubs, wsID) } + defer func() { + for _, ps := range closing { + _ = ps.Close() + } + }() + defer b.mu.Unlock() for wsID, byWorkspace := range b.subscribers { for ch := range byWorkspace { @@ -1290,7 +1477,12 @@ func (b *RedisBus) WorkspaceSubscriberCount(workspaceID string) int { // could subscribe, unsubscribe, close, or receive a fanned-out event. // // Exactly one caller per workspace reaches here; the rest wait on pending. -func (b *RedisBus) establishSubscription(ctx context.Context, workspaceID string, establisher *subscriber, pending *pendingSub) { +// Returns whether a subscription was actually INSTALLED. The idle cycle needs +// that answer and cannot infer it: reading the live generation afterwards +// misattributes an unrelated caller's fresh subscription as this cycle's +// replacement, and misses a real replacement that has already lost its last +// subscriber (codex round 13). +func (b *RedisBus) establishSubscription(ctx context.Context, workspaceID string, establisher *subscriber, pending *pendingSub) (installed bool) { channel := b.keys.Name(redisChannelSuffix) + workspaceID // DIALLED ON THE CALLER'S CONTEXT *AND* THE BUS'S, so a client that leaves // mid-dial stops paying for it (BUG-2749) without taking away Close()'s @@ -1368,7 +1560,14 @@ func (b *RedisBus) establishSubscription(ctx context.Context, workspaceID string // still non-zero and the subscription is installed for them, which is the // hand-off the filing asked about — expressed as a count rather than as a // transfer of ownership. - if ctx.Err() != nil { + // A NIL ESTABLISHER IS THE BUS ESTABLISHING FOR ITSELF (BUG-2738, rule 4). + // The idle detector re-establishes on b.ctx with no subscriber + // registration of its own, so there is nothing to deregister — and b.ctx + // is never cancelled until Close, at which point the count/closed check + // below is what abandons. Guarding the nil here rather than handing the + // detector a synthetic subscriber keeps wsCounts meaning "clients", which + // is what every arbitration in this file reads it as. + if establisher != nil && ctx.Err() != nil { b.unsubscribeLocked(establisher.ch) } if b.wsCounts[workspaceID] == 0 || b.ctx.Err() != nil { @@ -1377,15 +1576,24 @@ func (b *RedisBus) establishSubscription(ctx context.Context, workspaceID string subCancel() _ = pubsub.Close() close(pending.done) - return + return false } b.subGen++ gen := b.subGen sub := &redisSub{ - pubsub: pubsub, - cancel: subCancel, - gen: gen, - confirmed: make(chan struct{}), + pubsub: pubsub, + cancel: subCancel, + gen: gen, + // STAMPED AT INSTALL, not left at the zero value (BUG-2738, rule 2 of + // cycleIdleSubscriptions). A zero time reads as 1970, so a subscription + // that has simply not received anything yet would be older than any + // threshold and the idle detector would cycle it on its next tick — + // hardest in exactly the case BUG-2747 exists for, an unconfirmed + // admission where no acknowledgement ever arrives to stamp it. The + // clock starts when the socket does. + lastSeen: b.now(), + lastProbeOK: b.now(), + confirmed: make(chan struct{}), } b.wsSubs[workspaceID] = sub b.mu.Unlock() @@ -1441,7 +1649,7 @@ func (b *RedisBus) establishSubscription(ctx context.Context, workspaceID string } b.finishPending(workspaceID, pending) }() - return + return true case <-timer.C: b.markUnconfirmedAdmission(workspaceID, gen) } @@ -1452,6 +1660,7 @@ func (b *RedisBus) establishSubscription(ctx context.Context, workspaceID string } b.finishPending(workspaceID, pending) + return true } // mergeCancellation returns a context that ends when EITHER input does. @@ -1570,9 +1779,21 @@ func (b *RedisBus) stopRedisSubscription(workspaceID string) { return } sub.cancel() - sub.pubsub.Close() delete(b.wsSubs, workspaceID) + // CLOSED OFF THE LOCK (codex round 12). PubSub.Close takes go-redis's own + // mutex, which its health check can be holding across reconnect work — so + // closing here would put a network-bound wait inside b.mu, and b.mu is the + // lock every fan-out and every Subscribe on this instance contends for. + // The idle detector made that matter: teardown used to happen only when a + // workspace lost its last subscriber, and now happens on every cycle. + // + // Fire-and-forget is safe because nothing references this PubSub any more: + // the map entry is gone and the receive loop has already been signalled by + // cancel() above, which is what actually stops delivery. Close only + // releases the connection. + go func(ps *redis.PubSub) { _ = ps.Close() }(sub.pubsub) + // WHEN WE STOP RECEIVING, THE HONEST STATE IS NO BUFFER, NOT A STALE // CONTIGUOUS ONE (BUG-2731). This is the invariant a future optimization // will be tempted to violate — keeping the buffer "in case they come @@ -1619,9 +1840,13 @@ func (b *RedisBus) stopRedisSubscription(workspaceID string) { // (Receive → ReceiveTimeout(ctx, 0)). // // So an instance behind a wedged route sits there receiving nothing while its -// buffer keeps looking valid. Detecting that needs application-level idle -// tracking, which is BUG-2730's family and its own decision, because it needs -// a threshold. Do not assume the health check covers it. +// buffer keeps looking valid. THAT IS NOW COVERED, but NOT by anything in this +// function's choice of channel constructor: BUG-2738 added application-level +// idle tracking on top. Every inbound frame stamps sub.lastSeen below, and +// cycleIdleSubscriptions ends coverage and replaces the connection when the +// stamp goes stale. Do not assume the health check covers it; it still does +// not, and a future change that drops the stamping silently un-fixes BUG-2738 +// while leaving this loop looking untouched. func (b *RedisBus) receiveMessages(ctx context.Context, pubsub *redis.PubSub, workspaceID string, gen int64) { defer b.reportReceiveLoopExited() @@ -1635,6 +1860,17 @@ func (b *RedisBus) receiveMessages(ctx context.Context, pubsub *redis.PubSub, wo if !ok { return } + // STAMPED FOR EVERY FRAME, ahead of the type switch and ahead of + // any decode (BUG-2738). What idle detection measures is whether + // the SOCKET carries traffic, so a frame that turns out to be + // undecodable, or to name another workspace, or to be a + // resubscription notice, is still proof the route works — and each + // of those paths `continue`s, so stamping inside the switch would + // miss them. A message we could not read means coverage is broken, + // which dropWorkspaceCoverage handles; it does NOT mean the + // connection is dead, and cycling it would be the wrong remedy. + b.stampLastSeen(workspaceID, gen) + switch msg := raw.(type) { case *redis.Subscription: if msg.Kind != "subscribe" && msg.Kind != "psubscribe" { @@ -1659,7 +1895,17 @@ func (b *RedisBus) receiveMessages(ctx context.Context, pubsub *redis.PubSub, wo b.dropWorkspaceCoverage(workspaceID, ResetReasonSubscriptionResumed, gen) case *redis.Message: - epoch, event, err := decodePayload(msg.Payload) + kind, epoch, event, err := decodePayload(msg.Payload) + if kind == payloadHeartbeat { + // PHASE 1 IS EXACTLY THIS: recognise and ignore. The frame + // has already done its whole job by arriving — the stamp + // above is the entire effect. It consumes no id, drops no + // buffer, reaches no subscriber and moves no counter, so an + // instance that publishes none is still a correct receiver + // for one that does. That is what makes the two-phase roll + // zero-loss. + continue + } if err != nil { // A MESSAGE WE CANNOT READ IS A HOLE IN THIS WORKSPACE'S // COVERAGE (codex round 11). Dropping it and carrying on @@ -1719,16 +1965,26 @@ func (b *RedisBus) receiveMessages(ctx context.Context, pubsub *redis.PubSub, wo // any other workspace's channel, and dropping the rest would be a resync // charged to clients whose stream never broke. func (b *RedisBus) dropWorkspaceCoverage(workspaceID, reason string, gen int64) { - var report string - defer func() { - if report != "" { - b.reportReset(report) - } - }() - b.mu.Lock() - defer b.mu.Unlock() + report := b.dropWorkspaceCoverageLocked(workspaceID, reason, gen) + b.mu.Unlock() + if report != "" { + b.reportReset(report) + } +} +// dropWorkspaceCoverageLocked is dropWorkspaceCoverage with the lock already +// held. It returns the reason to report, or "" for nothing to report; the +// caller reports it AFTER releasing b.mu, because an Observer callback may call +// back into the bus. +// +// SPLIT OUT SO A CALLER CAN MAKE THE DROP PART OF A LARGER ATOMIC DECISION +// (BUG-2738, codex round 11). The idle cycle has to validate the subscription, +// end its coverage and tear it down without releasing the lock in between — +// otherwise a heartbeat arriving in one of those gaps makes it drop coverage +// for a workspace that had just recovered. +func (b *RedisBus) dropWorkspaceCoverageLocked(workspaceID, reason string, gen int64) string { + var report string // THE GENERATION CHECK BELONGS HERE TOO, not only in fan-out (codex round // 7). A receive loop can notice its connection died LONG after the // workspace was unsubscribed and resubscribed under it: last viewer @@ -1739,7 +1995,7 @@ func (b *RedisBus) dropWorkspaceCoverage(workspaceID, reason string, gen int64) // before its subscription began, and the reset counter names an incident // that did not happen to it. if sub, ok := b.wsSubs[workspaceID]; !ok || sub.gen != gen { - return + return report } if _, ok := b.replayBuffers[workspaceID]; !ok { @@ -1761,7 +2017,7 @@ func (b *RedisBus) dropWorkspaceCoverage(workspaceID, reason string, gen int64) // (there was none) and the signal measures CLIENTS WHO MAY HAVE // MISSED SOMETHING (there are some). b.signalWorkspaceLocked(workspaceID) - return + return report } delete(b.replayBuffers, workspaceID) // TELL THE SUBSCRIBERS THAT ARE STILL HOLDING THE STREAM OPEN (BUG-2730). @@ -1775,6 +2031,7 @@ func (b *RedisBus) dropWorkspaceCoverage(workspaceID, reason string, gen int64) // directly above: no other workspace's channel is implicated. b.signalWorkspaceLocked(workspaceID) report = reason + return report } // signalWorkspaceLocked raises the gap flag for every live subscriber of one @@ -1797,6 +2054,23 @@ func (b *RedisBus) signalAllLocked() { } } +// stampLastSeen records that this workspace's subscription just received +// something from Redis. +// +// GENERATION-CHECKED like every other bookkeeping write keyed by workspace: a +// receive loop can outlive its subscription (stopRedisSubscription only +// signals it, never joins it), and a straggler frame from a dead generation +// must not refresh the liveness of the one that replaced it. Without this +// check a wedged old loop's final buffered frames could keep a NEW subscription +// looking alive. +func (b *RedisBus) stampLastSeen(workspaceID string, gen int64) { + b.mu.Lock() + defer b.mu.Unlock() + if sub, ok := b.wsSubs[workspaceID]; ok && sub.gen == gen { + sub.lastSeen = b.now() + } +} + // currentSubGen reports the generation of the workspace's live subscription, // or 0 if it has none. Test seam: a real message carries its generation down // from receiveMessages, and a test driving the fan-out directly needs a way to @@ -1922,6 +2196,17 @@ const ( floorKeep ) +// payloadKind distinguishes the frames that arrive on a workspace's event +// channel. A heartbeat is BUS-INTERNAL: never an event, never buffered, never +// replayed, never fanned out, and never counted. It exists only so that +// silence on a socket becomes diagnostic (BUG-2738). +type payloadKind int + +const ( + payloadEvent payloadKind = iota + payloadHeartbeat +) + // decodePayload parses the "||" wire form publishScript emits, // and ALSO accepts a bare JSON body with no prefix. // @@ -1943,43 +2228,58 @@ const ( // The leading '{' check is what stops a JSON body that happens to contain two // '|' characters from being mistaken for a prefixed payload — an epoch is // never a JSON object. -func decodePayload(payload string) (int64, Event, error) { +func decodePayload(payload string) (payloadKind, int64, Event, error) { + // CLASSIFIED BEFORE ANYTHING IS SPLIT OR UNMARSHALLED, and the ORDER is + // what makes the frame safe (BUG-2738). "hb|1" has one separator and would + // otherwise fall through to the bare-JSON branch and fail to unmarshal; + // a future two-field frame would split into three and fail to parse an + // epoch. Either way it would reach the caller as an error, and an error + // here ends the workspace's coverage — so a liveness probe would + // manufacture the resync it exists to prevent. + // + // THE KIND IS RETURNED RATHER THAN HANDLED AT THE CALL SITE so that no + // future caller of this decoder can reintroduce that. The wire format is + // this function's to know. + if isHeartbeat(payload) { + return payloadHeartbeat, 0, Event{}, nil + } + if parts := strings.SplitN(payload, "|", 3); len(parts) == 3 && !strings.HasPrefix(parts[0], "{") { epochPart, idPart, body := parts[0], parts[1], parts[2] epoch, err := strconv.ParseInt(epochPart, 10, 64) if err != nil { - return 0, Event{}, fmt.Errorf("payload epoch prefix %q is not an integer: %w", epochPart, err) + return payloadEvent, 0, Event{}, fmt.Errorf("payload epoch prefix %q is not an integer: %w", epochPart, err) } if epoch <= 0 { // Zero is this package's sentinel for "no ID-space information", // so a message may not carry it as a real generation — otherwise a // malformed publisher could make every receiver stop reconciling // while looking perfectly healthy. - return 0, Event{}, fmt.Errorf("payload epoch prefix %d is not a positive generation", epoch) + return payloadEvent, 0, Event{}, fmt.Errorf("payload epoch prefix %d is not a positive generation", epoch) } id, err := strconv.ParseInt(idPart, 10, 64) if err != nil { - return 0, Event{}, fmt.Errorf("payload id prefix %q is not an integer: %w", idPart, err) + return payloadEvent, 0, Event{}, fmt.Errorf("payload id prefix %q is not an integer: %w", idPart, err) } var event Event if err := json.Unmarshal([]byte(body), &event); err != nil { - return 0, Event{}, fmt.Errorf("payload body is not an Event: %w", err) + return payloadEvent, 0, Event{}, fmt.Errorf("payload body is not an Event: %w", err) } event.ID = id if err := requirePositiveID(event.ID); err != nil { - return 0, Event{}, err + return payloadEvent, 0, Event{}, err } - return epoch, event, nil + return payloadEvent, epoch, event, nil } var event Event if err := json.Unmarshal([]byte(payload), &event); err != nil { - return 0, Event{}, fmt.Errorf("payload is neither || nor a bare Event: %w", err) + return payloadEvent, 0, Event{}, fmt.Errorf("payload is neither || nor a bare Event: %w", err) } if err := requirePositiveID(event.ID); err != nil { - return 0, Event{}, err + return payloadEvent, 0, Event{}, err } - return 0, event, nil + return payloadEvent, 0, event, nil } // requirePositiveID is applied to BOTH wire forms, and being applied to both diff --git a/internal/events/redis_generation_guard_test.go b/internal/events/redis_generation_guard_test.go index b46b1669..d56f9d8c 100644 --- a/internal/events/redis_generation_guard_test.go +++ b/internal/events/redis_generation_guard_test.go @@ -37,7 +37,7 @@ func newSeededFlippedBus(t *testing.T) (*RedisBus, *miniredis.Miniredis) { mr := miniredis.RunT(t) client := redis.NewClient(&redis.Options{Addr: mr.Addr()}) t.Cleanup(func() { _ = client.Close() }) - b := NewRedisBusWithKeys(client, redisns.Default, true) + b := NewRedisBusWithKeys(client, redisns.Default, true, false) b.nowUnix = func() int64 { return fixedSeed } t.Cleanup(b.Close) return b, mr @@ -91,7 +91,7 @@ func TestACorruptedGenerationCounterIsRepairedRatherThanFatal(t *testing.T) { // puts the sequence past 1 so the later publish reaches the // branch under test. b.Publish(Event{Type: ItemUpdated, WorkspaceID: "ws-1"}) - if _, _, err := decodePayload(next()); err != nil { + if _, _, _, err := decodePayload(next()); err != nil { t.Fatalf("fixture: the first publish must succeed, got %v", err) } @@ -112,7 +112,7 @@ func TestACorruptedGenerationCounterIsRepairedRatherThanFatal(t *testing.T) { b.Publish(Event{Type: ItemUpdated, WorkspaceID: "ws-1", ItemID: "item-7"}) - epoch, ev, err := decodePayload(next()) + _, epoch, ev, err := decodePayload(next()) if err != nil { t.Fatalf("the publish must survive a %s generation key (%s): %v", tc.name, tc.abort, err) } @@ -165,7 +165,7 @@ func TestAHealthyGenerationCounterIsIncrementedNotReseeded(t *testing.T) { ctx := context.Background() b.Publish(Event{Type: ItemUpdated, WorkspaceID: "ws-1"}) - first, _, err := decodePayload(next()) + _, first, _, err := decodePayload(next()) if err != nil { t.Fatalf("first publish: %v", err) } @@ -178,7 +178,7 @@ func TestAHealthyGenerationCounterIsIncrementedNotReseeded(t *testing.T) { t.Fatalf("clear the epoch: %v", err) } b.Publish(Event{Type: ItemUpdated, WorkspaceID: "ws-1", ItemID: "item-2"}) - second, _, err := decodePayload(next()) + _, second, _, err := decodePayload(next()) if err != nil { t.Fatalf("second publish: %v", err) } @@ -259,7 +259,7 @@ func TestEveryRotationBranchGuardsTheGenerationCounter(t *testing.T) { // Get the sequence past 1 so the branches that need a live // sequence can be reached; the first case then clears it again. b.Publish(Event{Type: ItemUpdated, WorkspaceID: "ws-1"}) - if _, _, err := decodePayload(next()); err != nil { + if _, _, _, err := decodePayload(next()); err != nil { t.Fatalf("fixture: the first publish must succeed, got %v", err) } @@ -274,7 +274,7 @@ func TestEveryRotationBranchGuardsTheGenerationCounter(t *testing.T) { b.Publish(Event{Type: ItemUpdated, WorkspaceID: "ws-1", ItemID: "item-9"}) - epoch, ev, err := decodePayload(next()) + _, epoch, ev, err := decodePayload(next()) if err != nil { t.Fatalf("the %s branch must survive a corrupted generation counter: %v", tc.branch, err) } @@ -316,7 +316,7 @@ func TestThePublishedGenerationMatchesTheStoredOneAboveExactDoubleRange(t *testi ctx := context.Background() b.Publish(Event{Type: ItemUpdated, WorkspaceID: "ws-1"}) - if _, _, err := decodePayload(next()); err != nil { + if _, _, _, err := decodePayload(next()); err != nil { t.Fatalf("fixture: the first publish must succeed, got %v", err) } @@ -337,7 +337,7 @@ func TestThePublishedGenerationMatchesTheStoredOneAboveExactDoubleRange(t *testi } b.Publish(Event{Type: ItemUpdated, WorkspaceID: "ws-1", ItemID: "item-11"}) - published, _, err := decodePayload(next()) + _, published, _, err := decodePayload(next()) if err != nil { t.Fatalf("publish at the guard's limit: %v", err) } @@ -399,7 +399,7 @@ func TestTheGenerationCeilingIsOneUnderTheEpochCeiling(t *testing.T) { ctx := context.Background() b.Publish(Event{Type: ItemUpdated, WorkspaceID: "ws-1", ItemID: "a"}) - if _, _, err := decodePayload(next()); err != nil { + if _, _, _, err := decodePayload(next()); err != nil { t.Fatalf("fixture: %v", err) } @@ -411,7 +411,7 @@ func TestTheGenerationCeilingIsOneUnderTheEpochCeiling(t *testing.T) { } b.Publish(Event{Type: ItemUpdated, WorkspaceID: "ws-1", ItemID: "b"}) - published, _, err := decodePayload(next()) + _, published, _, err := decodePayload(next()) if err != nil { t.Fatalf("publish: %v", err) } @@ -613,7 +613,7 @@ func TestABrokenClockDoesNotProduceAnUnpublishableEpoch(t *testing.T) { ctx := context.Background() b.Publish(Event{Type: ItemUpdated, WorkspaceID: "ws-1", ItemID: "a"}) - if _, _, err := decodePayload(next()); err != nil { + if _, _, _, err := decodePayload(next()); err != nil { t.Fatalf("fixture: %v", err) } @@ -632,7 +632,7 @@ func TestABrokenClockDoesNotProduceAnUnpublishableEpoch(t *testing.T) { } b.Publish(Event{Type: ItemUpdated, WorkspaceID: "ws-1", ItemID: "b"}) - epoch, ev, err := decodePayload(next()) + _, epoch, ev, err := decodePayload(next()) if err != nil { t.Fatalf("a repair under a zero clock must still publish a decodable event: %v", err) } diff --git a/internal/events/redis_heartbeat.go b/internal/events/redis_heartbeat.go new file mode 100644 index 00000000..dab2f1e8 --- /dev/null +++ b/internal/events/redis_heartbeat.go @@ -0,0 +1,798 @@ +package events + +import ( + "log/slog" + "strings" + "sync" + "time" +) + +// The heartbeat exists because NOTHING ELSE IN THIS PACKAGE CAN SEE A HALF-OPEN +// CONNECTION (BUG-2738). A route that stops working without closing — a NAT +// table expiring, a firewall dropping an idle flow, a silently rerouted path — +// leaves this instance blocked on a read that will never return, receiving +// nothing, while its replay buffer goes on looking complete. Every resume is +// then answered "caught up" from a coverage window that ended when the route +// did. +// +// go-redis's pub/sub health check does NOT cover it, and that was measured +// rather than reasoned. PubSub.Ping calls writeCmd and returns without ever +// reading a reply (v9.22.0, pubsub.go), so its error stays nil for as long as +// the socket accepts writes — which a half-open socket does until its send +// buffer fills. The channel path sets no read deadline either (Receive calls +// ReceiveTimeout(ctx, 0)). Probed against a TCP proxy that silently stopped +// forwarding: no reconnect in 24 seconds. Do not replace this with a Ping. +// +// WHAT THIS DETECTS AND WHAT IT DOES NOT, stated as precisely as the mechanism +// actually supports (codex round 10, which was asked to refute the claim rather +// than to look for defects, and partly succeeded). All three limits below were +// checked against go-redis v9.22.0 rather than reasoned about: +// +// - IT IS A RECEIVE-SIDE DETECTOR, not a round-trip health check. What it +// measures is whether frames ARRIVE on this workspace's subscription. A +// subscription that receives fine but whose outbound direction is broken +// looks healthy here — correctly, since nothing is being lost. +// +// - IT DOES NOT COVER THE PUBLISH PATH, and cannot: PUBLISH travels on the +// client's connPool while a subscription holds a connection from the +// separate pubSubPool (redis.go:363, :1956). Those are different sockets +// with different fates, so a wedged publish path is invisible to this, and +// a reconnect of one repairs nothing about the other. An instance whose +// publishes fail loses ITS OWN events for everyone; that is a different +// failure needing a different signal. +// +// - REPLACEMENT IS ATTEMPTED, NOT GUARANTEED. If the network path is still +// blackholed when the cycle re-dials, the replacement cannot receive +// either, and the detector fires again on the next pass. That is the +// honest behaviour — coverage stays ended, so nothing is claimed falsely — +// but "delivery resumes" is a statement about the network, not about this +// code. See BUG-2764 for a case where the replacement can fail silently +// even on a healthy path. +// +// WHAT MAKES THE THRESHOLD ANSWERABLE. "Is this workspace quiet, or is the +// route dead?" cannot be answered from traffic, because it depends on the +// deployment's publish rate and no constant is right for every one of them. +// Publishing our OWN traffic replaces it with "did our heartbeat arrive?", +// which is app-controlled and the same on every deployment. That is the whole +// reason the interval is not a tuned number: it is not measuring a workspace, +// it is measuring a socket. +const ( + // DefaultHeartbeatInterval is T: how often an instance publishes one + // liveness frame per workspace it is subscribed to. Dave's ruling + // (day-49): 30s. + DefaultHeartbeatInterval = 30 * time.Second + + // DefaultIdleTimeout is 3T: how long a subscription may receive NOTHING — + // no message, no heartbeat, no subscription confirmation — before its + // coverage ends and its connection is cycled. + // + // Three intervals rather than two so that a single lost or late heartbeat + // is not a cycle. + // + // THE LATENCY ARITHMETIC, corrected after the loops were split (codex + // round 8; the earlier wording described a shared ticker that no longer + // exists). Measured FROM lastSeen, detection lands in [3T, 4T) — the scan + // runs on its own T-cadence, so it adds up to one interval on top of the + // threshold. Measured from FAULT ONSET it is wider and less tidy, roughly + // [2T, 4T): the publisher has its own independent phase, so the last frame + // to get through may have been sent anywhere in the interval before the + // route died. A pass that overruns widens both ends further. Quote the + // from-lastSeen figure when reasoning about the code and the from-onset + // one when telling an operator how long an incident hides. + DefaultIdleTimeout = 3 * DefaultHeartbeatInterval +) + +// heartbeatPrefix marks a BUS-INTERNAL liveness frame on a workspace's event +// channel. +// +// IT TRAVELS ON THE EVENT CHANNEL ON PURPOSE, and that is the entire cost of +// this design: what needs proving is that THIS channel's connection still +// carries traffic, so a probe on any other channel proves the wrong thing. +// That is also why this is a wire-format change and why it rolls out in two +// phases — see config.EventsHeartbeat. +// +// A PREFIX RATHER THAN AN EXACT PAYLOAD, so a later version of the frame can +// carry fields without needing a third roll: a phase-1 binary already ignores +// a v2 frame it knows nothing about. +// +// It cannot be confused with either event form. decodePayload classifies on +// this prefix BEFORE it splits or unmarshals anything, and no epoch generation +// begins with "hb" — the prefixed event form's first field is parsed as an +// integer, and the bare form is JSON. +const heartbeatPrefix = "hb|" + +// heartbeatPayload is what this version emits. The suffix is a format version, +// not a timestamp: a receiver derives arrival time from its own clock, because +// a publisher's clock is not comparable to it (the same reason +// redisEpochGenSuffix is a generation and not a wall clock). +const heartbeatPayload = heartbeatPrefix + "1" + +// heartbeatMaxLen bounds a frame this package will accept as one of its own. +// A liveness frame carries a version and, at most, a few short tokens; anything +// larger is something else wearing the prefix. +const heartbeatMaxLen = 64 + +// isHeartbeat reports whether a payload is a bus-internal liveness frame. +// +// THE SHAPE IS VALIDATED, NOT JUST THE PREFIX (codex round 5, P2), and the +// first draft got this wrong in a way worth recording. Accepting any "hb|…" +// created a silently-ignored class on the workspace event channel where +// previously EVERY unreadable payload ended coverage loudly and moved +// undecodable_message — the counter whose documented job is "suspect a +// namespace collision". A foreign or buggy publisher whose payload happened to +// start with "hb|" would have slipped through that signal without a trace. +// +// What is NOT a problem, and was considered: a forged frame cannot fake +// liveness. Liveness here means "this socket carried traffic", and a frame +// that ARRIVES demonstrates exactly that whoever sent it — which is why +// stampLastSeen fires for undecodable frames too. There is no claim about +// event coverage in a heartbeat to forge. +// +// So the rule is conservative in the direction that keeps the loud path loud: +// "hb|" then a decimal version, then optional "|"-separated tokens from a +// narrow charset, under a length cap. A disciplined future frame still needs +// no third roll; arbitrary bytes wearing the prefix go back to being a +// coverage-ending decode failure. +func isHeartbeat(payload string) bool { + if len(payload) > heartbeatMaxLen || !strings.HasPrefix(payload, heartbeatPrefix) { + return false + } + fields := strings.Split(payload[len(heartbeatPrefix):], "|") + if fields[0] == "" { + return false + } + for _, r := range fields[0] { + if r < '0' || r > '9' { + return false + } + } + for _, f := range fields[1:] { + for _, r := range f { + switch { + case r >= '0' && r <= '9', r >= 'a' && r <= 'z', r >= 'A' && r <= 'Z': + case r == '.' || r == '_' || r == '-' || r == ':': + default: + return false + } + } + } + return true +} + +// now reads the bus's clock. The seam is nil in every real construction; tests +// set it so an idle threshold can be crossed without sleeping through one. +func (b *RedisBus) now() time.Time { + if b.nowFunc != nil { + return b.nowFunc() + } + return time.Now() +} + +// maintenanceLoop runs the two halves of BUG-2738's machinery. +// +// TWO GOROUTINES, NOT ONE LOOP DOING BOTH, and the separation is the whole +// point rather than tidiness (codex round 1, P1). publishHeartbeats makes N +// SYNCHRONOUS Redis publishes, one per subscribed workspace. Against the +// failure this feature exists to detect — a route that has stopped carrying +// traffic — those publishes are exactly the ones that block, and go-redis +// bounds them by its own Dial/Read/WriteTimeout rather than by any context we +// could pass. Sharing a goroutine would therefore let a stalled publisher +// delay idle detection for as long as those timeouts take, on the very +// instance whose connections have wedged: the detector would sleep through the +// incident it was built to find, and the more workspaces an instance carried +// the longer it would sleep. +// +// A stalled publisher is not otherwise a problem — it produces silence, which +// is precisely what the detector reads. It only had to stop being the +// detector's problem too. +// +// Started by the constructor and ended by Close through b.ctx. Both halves are +// separately callable and the tests drive them directly, which is why the +// wiring has its own test: a direct-call test vouches for the function, not +// for its binding (team CONVE-19). +// A TIMER RE-READ EACH PASS, NOT A TICKER CONSTRUCTED ONCE. A ticker would +// capture heartbeatInterval at goroutine start, which makes the field +// write-once-at-construction in practice while looking like an ordinary +// tunable — and makes any later write to it a data race against this +// goroutine. Re-reading under b.mu each pass costs one uncontended lock per +// interval and makes the field genuinely what its comment says it is. +func (b *RedisBus) maintenanceLoop() { + defer close(b.maintenanceStopped) + var wg sync.WaitGroup + wg.Add(2) + go func() { defer wg.Done(); b.tickForever(b.heartbeatKick, b.publishHeartbeats) }() + go func() { defer wg.Done(); b.tickForever(b.idleKick, b.cycleIdleSubscriptions) }() + wg.Wait() +} + +// tickForever runs work on the configured interval until the bus closes. +// +// Each loop gets its OWN kick channel: a single shared one would be consumed +// by whichever goroutine happened to be waiting, leaving the other serving out +// a cadence that had already changed. +// SCHEDULED FROM A DEADLINE, NOT FROM THE END OF THE LAST PASS (codex round 5, +// P2). Restarting the timer after work() makes the real period T plus however +// long the pass took, and for the PUBLISHER that is self-defeating: an instance +// whose publishes are slow emits heartbeats further apart, its own subscription +// sees them further apart, and it can cross its own 3T threshold and cycle +// connections that were never wedged. The slowness would manufacture the +// incident. +// +// When a pass overruns badly the schedule is reset to now rather than firing +// the missed ticks back-to-back: there is no value in a burst of heartbeats, +// and a burst of idle scans would hammer a Redis that is already struggling. +func (b *RedisBus) tickForever(kick <-chan struct{}, work func()) { + next := time.Now() + for { + b.mu.Lock() + interval := b.heartbeatInterval + b.mu.Unlock() + + next = nextTick(next, interval, time.Now()) + if wait := time.Until(next); wait > 0 { + timer := time.NewTimer(wait) + select { + case <-b.ctx.Done(): + timer.Stop() + return + case <-kick: + // The cadence changed under us; drop this wait, re-read it, and + // re-base the schedule so the new interval starts now rather + // than from a deadline computed under the old one. + timer.Stop() + next = time.Now() + continue + case <-timer.C: + } + } + + // CHECKED AGAIN AFTER THE WAIT, and this is NOT redundant with the + // ctx.Done arm above even though removing either one alone survives + // every test (mutation matrix; team lesson: a pair that only dies + // together is a question, not a clearance). They cover disjoint + // moments and each is independently right: + // + // - The select arm is the exit while WAITING, which is where this + // goroutine spends essentially all of its life. Without it a closed + // bus leaves both loops sleeping out a full interval before + // noticing, every interval, forever. + // - This check is the exit after the timer has already fired, so a + // bus that closed DURING the previous pass does not start another + // one. Without it, work() runs once more against a cancelled + // context: the publish half writes to Redis on a dead ctx and the + // idle half takes b.mu after Close has drained it. + // + // Removing BOTH is detected, by TestClosingTheBusStopsTheMaintenanceLoop. + if b.ctx.Err() != nil { + return + } + work() + } +} + +// PublishHeartbeatsForTest runs one publish pass, for tests in OTHER packages. +// +// cmd/pad's wiring test needs to know whether the config flip reached this +// bus's constructor, and driving the pass DIRECTLY is what makes that test +// deterministic (codex round 7). Shortening the cadence and waiting instead +// makes the negative arm — a phase-1 bus must publish nothing — a race against +// the scheduler: under -race or a loaded CI box the goroutine may simply not +// have run yet, which is indistinguishable from a bus that is correctly +// silent. The loop's own wiring is covered inside this package, where the +// unexported cadence setter is available. +// +// Named so no production caller reaches for it. +func (b *RedisBus) PublishHeartbeatsForTest() { b.publishHeartbeats() } + +// setMaintenanceCadence changes T and the idle threshold on a running bus. +// +// IT EXISTS SO THE WIRING CAN BE TESTED AT THE CADENCE, not only the halves at +// the call (team CONVE-19: a direct-call test vouches for the component, not +// its binding). Without the kick, a test that shortens the interval races the +// loop's first read — lose that race and the test waits a full default +// interval — so the only deterministic alternative would be a test-only +// constructor, which would vouch for a construction production never uses. +// +// The kick is buffered and non-blocking: a change that arrives while the loop +// is mid-pass is picked up on its next read, which is the same interval either +// way. +func (b *RedisBus) setMaintenanceCadence(interval, idleTimeout time.Duration) { + b.mu.Lock() + b.heartbeatInterval = interval + b.idleTimeout = idleTimeout + b.mu.Unlock() + for _, kick := range []chan struct{}{b.heartbeatKick, b.idleKick} { + select { + case kick <- struct{}{}: + default: + } + } +} + +// publishHeartbeats emits one liveness frame per workspace this instance is +// currently subscribed to. No-op until phase 2 (see config.EventsHeartbeat). +// +// IT DOES NOT GO THROUGH Publish, and that is load-bearing rather than +// stylistic. Publish mints an ID from the shared Redis counter; a heartbeat +// that consumed one would inflate the ID space that three of this bus's reset +// reasons are derived from — counter_backward, epoch_change and epoch_regressed +// all reason about that counter's values — so a liveness probe would start +// manufacturing the very resets it exists to avoid. A heartbeat carries no ID, +// no epoch, and is never buffered, replayed, fanned out, or counted as an +// event. +// +// COST, WITH THE SENTENCE THAT STOPS SOMEONE OPTIMISING THE WRONG LAYER: the +// heartbeat inherits per-workspace granularity from the existing +// one-PubSub-per-workspace structure — establishSubscription mints a separate +// connection per workspace, so liveness is genuinely per-workspace and there is +// no cheaper shared probe. An instance subscribed to N workspaces publishes N +// frames every T; at N=1000 and T=30s that is ~33 publishes/sec. If fleet +// workspace counts ever make this matter, the fix is CONNECTION CONSOLIDATION, +// not heartbeat thinning: thinning the interval widens the silent window this +// exists to bound, while consolidation reduces the number of sockets that need +// proving at all. +func (b *RedisBus) publishHeartbeats() { + b.mu.Lock() + if !b.publishHeartbeat { + b.mu.Unlock() + return + } + // THE GENERATION IS PART OF THE SNAPSHOT (codex round 14). The publish + // below happens off the lock and can take as long as go-redis's timeouts + // allow, during which this workspace's subscription may be torn down and + // replaced — by a cycle, or by its last subscriber leaving and a new one + // arriving. Stamping "whatever occupies this workspace now" would then + // credit a probe to a subscription that never received one, and the next + // pass's failures could cycle it while it looked recently probed. Same + // hazard stampLastSeen already guards against, on the same map. + probes := make([]idleProbe, 0, len(b.wsSubs)) + for ws, sub := range b.wsSubs { + probes = append(probes, idleProbe{workspaceID: ws, gen: sub.gen}) + } + b.mu.Unlock() + + for _, p := range probes { + ws := p.workspaceID + channel := b.keys.Name(redisChannelSuffix) + ws + if err := b.client.Publish(b.ctx, channel, heartbeatPayload).Err(); err != nil { + // Logged and dropped, never retried: retrying here would only make + // a wedged publish path look healthier than it is. + // + // AND DELIBERATELY NOT STAMPED. lastProbeOK stays where it was, so + // the detector stops treating this workspace's silence as evidence + // — see that field. A failure to PROBE is not a finding about the + // peer, and counting it as one tears down healthy connections + // whenever this instance's outbound path is the broken one. + slog.Warn("events: failed to publish a liveness heartbeat; this workspace's idle detection is suspended until a probe succeeds, because silence cannot be read as a finding when we could not ask", + "channel", channel, "error", err) + b.reportHeartbeatPublishFailed() + continue + } + if b.afterProbePublish != nil { + b.afterProbePublish(ws) + } + + b.mu.Lock() + if sub, ok := b.wsSubs[ws]; ok && sub.gen == p.gen { + sub.lastProbeOK = b.now() + } + b.mu.Unlock() + } +} + +// liveGen reports the generation of the workspace's installed subscription, and +// whether there is one at all. +// +// Distinct from currentSubGen, which answers zero for both "no subscription" +// and a genuine zero — a distinction the cycle needs, because "nothing was +// installed" and "something was installed" are the two outcomes it reports on. +func (b *RedisBus) liveGen(workspaceID string) (int64, bool) { + b.mu.Lock() + defer b.mu.Unlock() + sub, ok := b.wsSubs[workspaceID] + if !ok { + return 0, false + } + return sub.gen, true +} + +// idleProbe is one workspace selected for a heartbeat, with the generation of +// the subscription the probe is FOR. The generation travels with it so a slow +// publish cannot credit a subscription that replaced the one it was sent for. +type idleProbe struct { + workspaceID string + gen int64 +} + +// idleCycle is one workspace selected for cycling, with the establishment +// record its selection minted. +type idleCycle struct { + workspaceID string + gen int64 + pending *pendingSub +} + +// cycleIdleSubscriptions ends coverage for every workspace whose subscription +// has received nothing for idleTimeout, and REPLACES that subscription. +// +// DROP-AND-CYCLE, NOT DROP ALONE, and the difference is the whole remedy. A +// half-open route stays half-open: dropping coverage makes the next resume +// honest, but the instance is still attached to a dead socket, so the resync it +// just demanded is served from the same dead subscription and the detector +// fires again on the next tick. That is a resync loop metering the failure at +// 3T intervals, not a recovery. Cycling is what restores delivery. +// +// THE IDLE DETECTOR IS A THIRD ACTOR IN THIS REGION, and every invariant here +// was designed around the other two. Until now only request goroutines mutated +// wsSubs/pendingSubs, plus Close; this one is a background mutator with no +// request behind it. Four hazards, each with the rule that answers it: +// +// 1. CYCLING ACROSS AN IN-FLIGHT ESTABLISHMENT. If pendingSubs holds a record, +// an establishment is already running and may be about to install over what +// a cycle just tore down — or the cycle installs a second subscription and +// the single-establisher wall is breached from a direction it was never +// guarded against. RULE: the detector takes the same wall. It refuses to +// cycle while a record exists and waits for the next tick, and when it does +// cycle it MINTS the record itself, under b.mu, before tearing anything +// down. Refusing is the cheaper correct answer: an establishment in flight +// is itself evidence of imminent traffic. +// +// 2. A FRESHLY INSTALLED SUBSCRIPTION LOOKS IDLE. A zero lastSeen reads as +// 1970 and fires the detector on the next tick. RULE: lastSeen is stamped +// at INSTALL time, not only on inbound frames — see establishSubscription. +// This matters most in exactly the case BUG-2747 exists for, an unconfirmed +// admission where no confirmation ever arrives to stamp it, so the naive +// version would cycle hardest on the workspaces already having a bad time. +// +// 3. CYCLING A WORKSPACE NOBODY WANTS. wsCounts may reach zero between the +// tick's read and the cycle. RULE: re-check under the SAME lock that +// performs the teardown — the deregister-before-arbitration ordering +// BUG-2749 established, applied to a new caller. THESE CHECKS ARE AN +// OPTIMISATION RATHER THAN A CORRECTNESS GUARD, and the mutation matrix is +// what says so rather than an argument: removing the whole second read +// survives every test here, because establishSubscription's own abandon +// path already refuses to install for an empty workspace and retires the +// record in the same critical section. What the checks buy is a dial not +// paid for. See cycleOne for the per-term reading. +// +// 4. NO REQUEST CONTEXT TO ESTABLISH ON. RULE: the re-establishment runs on +// b.ctx, which establishSubscription's cancellation path already tolerates +// (never cancelled until Close). That path's comments are all written in +// terms of "the caller"; here the caller is the bus, and it passes a nil +// establisher because it has no subscriber registration of its own to +// unwind. +func (b *RedisBus) cycleIdleSubscriptions() { + now := b.now() + + var due []idleCycle + b.mu.Lock() + // DETECTION IS GATED ON PUBLISHING, and getting this wrong is the defect + // codex round 1 found in the first draft of this unit (P2, and it is a P1 + // in effect). Idle detection ran on every instance from phase 1, justified + // in a comment as "detecting off whatever traffic the deployment already + // carries" — which is true only of a BUSY workspace. On a QUIET one, phase + // 1 has no traffic to detect off and no heartbeat either, so a perfectly + // healthy subscription crosses the threshold every 90-120s and is cycled: + // coverage dropped, every live subscriber told to resync, forever, on the + // DEFAULT configuration every deployment lands in first. That is the exact + // load-posture inversion this family keeps having to avoid, shipped as the + // default. + // + // An instance detects off its OWN frames — it publishes to the workspace + // channels it subscribes to and receives them back — so it never depends + // on peers having flipped. Publishing and detecting are therefore one + // capability with one switch, and phase 1 is exactly "recognise the frame + // so a phase-2 peer costs you nothing". + if !b.publishHeartbeat { + b.mu.Unlock() + return + } + idleTimeout := b.idleTimeout + for ws, sub := range b.wsSubs { + if _, inFlight := b.pendingSubs[ws]; inFlight { + continue // rule 1 + } + if b.wsCounts[ws] == 0 { + // RULE 3, FIRST READ. Also an optimisation rather than a guard, and + // for a sharper reason than the second read's: reaching zero takes + // the subscription down with it (Unsubscribe's count-to-zero branch + // calls stopRedisSubscription), so a workspace at zero has no + // wsSubs entry and this loop never sees it. Removing this line + // survives every test. Kept as a cheap statement of the intended + // precondition rather than as a load-bearing check. + continue + } + // NO `lastSeen.IsZero()` SKIP HERE, and its absence is deliberate. + // Treating an unstamped subscription as "not idle" reads as a safe + // belt-and-braces guard next to rule 2, and is the exact opposite: it + // would make a subscription that has NEVER received anything + // permanently uncyclable — which is the BUG-2747 unconfirmed + // admission, the one case the plan singles out as mattering most. + // A route that wedges before the acknowledgement arrives would then be + // undetectable forever, in the population already having the worst + // time. The install-time stamp (rule 2) is what makes a zero value + // unreachable for an installed subscription; a guard here would only + // mask it. Found by the mutation matrix: with the skip present, + // removing the install stamp survived every test; with it gone, that + // mutation is caught. + // + // RE-ADDING THE SKIP IS ITSELF UNDETECTABLE, and that is the correct + // reading rather than a coverage gap: rule 2 makes a zero lastSeen + // unreachable, so the branch would never be taken — until the day rule + // 2 regressed, which is the day it would hide the regression. An + // unreachable guard that only acts when a real one has already broken + // is worse than no guard, because it converts a caught defect into a + // silent one. This comment is the enforcement; there is no test that + // can be. + if now.Sub(sub.lastSeen) < idleTimeout { + continue + } + // THE PREMISE HAS TO HOLD BEFORE THE CONCLUSION IS DRAWN. Silence only + // means "the receive path is dead" if we actually managed to send + // something into it AFTER the silence began; see redisSub.lastProbeOK. + // + // EXPRESSED AS AN ORDERING, not as an age, and the honest reason is + // weaker than the one this comment first gave. Codex round 16 argued + // an age-based form ("has a probe succeeded within the threshold") + // failed to suspend detection where this one would; the mutation + // matrix then declined to confirm it — reverting to the age form, and + // even removing cycleOne's copy too, breaks no test, and no case could + // be constructed that separates them. On any healthy path the two + // stamps advance TOGETHER, because a probe whose frame arrives sets + // both; they diverge only on the wedge, where both forms cycle. + // + // It is kept because it says exactly what the rule means — we have + // sent something into this subscription more recently than anything + // came out of it — and is never weaker. Not because it was shown to + // fix a reachable defect. A fresh subscription has the two equal, so it + // is never cycled before its first successful probe. + // + // CHECKED HERE AND AGAIN IN cycleOne. Removing either alone leaves the + // tests green, and so does removing both, for the reason above; the + // pair is justified by what it expresses, not by the matrix. + // + // The two placements still cover different moments: this one keeps a + // workspace off the due list at all, so no establishment record is + // minted and no joiner is made to wait, while cycleOne's covers the + // probe failing AFTER selection — a window the concurrency cap makes + // real. Neither subsumes the other. + if !sub.lastProbeOK.After(sub.lastSeen) { + continue + } + // Minting the record HERE is what makes rule 1 hold in the other + // direction too: from this moment a subscriber arriving for this + // workspace joins the establishment we are about to run instead of + // finding the doomed subscription live and being admitted into it. + // subscribeAndReplay checks pendingSubs BEFORE wsSubs precisely so + // that this overlap is safe. + pending := &pendingSub{done: make(chan struct{})} + b.pendingSubs[ws] = pending + due = append(due, idleCycle{workspaceID: ws, gen: sub.gen, pending: pending}) + } + b.mu.Unlock() + + // BOUNDED-PARALLEL, NOT SERIAL (codex round 5, P2). Each cycle re-dials, + // and a dial against a struggling Redis is bounded by go-redis's own + // timeouts rather than by anything here — so a serial pass makes recovery + // take N x that timeout, and the workspaces at the end of the map wait the + // longest while still reporting themselves uncovered. The failure that puts + // many workspaces on this list at once is precisely a Redis failover, so + // the serial case is the common one, not the exotic one. + // + // Each entry already owns its own establishment record, minted under the + // lock above, so they are independent by construction: rule 1 keeps any + // other caller off a workspace being cycled, and two entries never name the + // same one. + // + // The cap is a deliberate middle: unbounded goroutines would answer a Redis + // outage by opening one dial per workspace at once, which is the shape that + // turns a slow dependency into an outage of our own. + if b.afterIdleScan != nil { + b.afterIdleScan() + } + + sem := make(chan struct{}, maxConcurrentCycles) + var wg sync.WaitGroup + for _, c := range due { + wg.Add(1) + sem <- struct{}{} + go func(c idleCycle) { + defer wg.Done() + defer func() { <-sem }() + b.cycleOne(c, idleTimeout) + }(c) + } + // WAITED ON, so one pass cannot overlap the next and so a direct caller — + // every test here — observes a finished pass rather than a started one. + wg.Wait() +} + +// nextTick returns the deadline for the pass after one that was scheduled for +// prev, given the configured interval and the current time. +// +// SEPARATED OUT SO THE ARITHMETIC CAN BE TESTED WITHOUT A CLOCK (mutation +// matrix: restoring the drift survived every test, because the only way to +// observe it in the loop is to time it, and a timing test is a flaky test). +// +// The schedule is deadline-based rather than sleep-after-work, because the +// latter makes the real period T plus however long the pass took. For the +// publisher that is self-defeating: an instance whose publishes are slow emits +// heartbeats further apart, its own subscription sees them further apart, and +// it can cross its own 3T threshold and cycle connections that were never +// wedged — the slowness manufacturing the incident. +// +// When a pass overruns by more than a whole interval the schedule is RESET to +// now rather than firing the missed ticks back to back. A burst of heartbeats +// buys nothing, and a burst of idle scans would hammer a Redis that is already +// struggling — which is precisely the condition that made the pass overrun. +func nextTick(prev time.Time, interval time.Duration, now time.Time) time.Time { + next := prev.Add(interval) + if now.Sub(next) > interval { + return now.Add(interval) + } + return next +} + +// maxConcurrentCycles bounds how many replacement dials one idle pass has in +// flight. Eight because the work is entirely network-bound and the point is to +// stop N sequential dial timeouts from serialising recovery, not to saturate +// anything: at the 30s cadence this is eight concurrent connects at most once +// per interval, against a Redis that is by definition already in trouble when +// the number is large. +const maxConcurrentCycles = 8 + +// cycleOne ends one workspace's coverage and re-establishes its subscription. +// +// EVERY VALIDATION, THE COVERAGE DROP AND THE TEARDOWN HAPPEN UNDER ONE LOCK, +// and that is the fix for a false positive codex round 11 found — the property +// this whole design cares about most, because a false positive costs a +// coverage drop and a resync for every subscriber of a healthy workspace. +// +// The scan selects victims and releases the lock; this runs afterwards, and +// "afterwards" can be a long time. The concurrency cap means a workspace can +// wait behind several batches of slow replacement dials, and a GC or CPU pause +// can leave a backlog of heartbeats undrained in the receive loop. In that +// window the subscription can start receiving again — and the earlier version +// cycled it anyway, because its re-checks covered generation, subscriber count +// and bus liveness but never re-asked the question the scan had asked. +// +// The staleness re-check below is therefore not defensive tidying: it is the +// difference between "idle when we looked" and "idle now", and the gap between +// those two was widened by this unit's own concurrency cap. +func (b *RedisBus) cycleOne(c idleCycle, idleTimeout time.Duration) { + b.mu.Lock() + sub, live := b.wsSubs[c.workspaceID] + + // RULE 3, SECOND READ, plus the freshness re-check. Between the scan and + // here, the last subscriber may have left (taking the subscription down + // with it), the workspace may have been re-established under a new + // generation, the bus may have closed, or the connection may simply have + // started working again. + // + // WHAT THE MUTATION MATRIX SAYS ABOUT THE FIRST THREE TERMS, recorded + // because the honest reading is not the flattering one. Removing the + // liveness term, the generation term, the count term, or all of them + // survives every test in this package — establishSubscription re-reads + // wsCounts under its own deciding lock and abandons, retiring the record in + // that same section (BUG-2749), so dropping them costs a dial that is + // immediately thrown away rather than a wrong outcome. They are kept + // because they do not DEPEND on that coupling. The generation term is + // additionally unreachable while we hold the establishment record, by rule + // 1's own mechanism. Do not read those survivals as dead code to delete, + // and do not read them as tested defence in depth. + // + // The FRESHNESS term is different in kind: it is load-bearing, it has its + // own test, and removing it is detected. + switch { + case !live || sub.gen != c.gen || b.wsCounts[c.workspaceID] == 0 || b.ctx.Err() != nil: + b.retirePendingLocked(c.workspaceID, c.pending) + b.mu.Unlock() + close(c.pending.done) + return + case !sub.lastProbeOK.After(sub.lastSeen): + // The probe started failing, or something arrived, while this cycle sat + // in the queue. Same ordering rule as the scan's check: with no + // successful probe SINCE the last thing we received, we have no + // evidence about the receive path, so tearing it down would be a + // guess. + b.retirePendingLocked(c.workspaceID, c.pending) + b.mu.Unlock() + close(c.pending.done) + return + case b.now().Sub(sub.lastSeen) < idleTimeout: + // It recovered while this cycle sat in the queue. Nothing to end and + // nothing to replace: leaving it alone is the whole point. + // + // THIS AND THE PREMISE CASE ABOVE DIE ONLY TOGETHER in the matrix, and + // they are NOT redundant — they catch different shapes of the same + // recovery, which is why removing either alone leaves the recovery test + // green: + // + // - Recovered and NOT re-probed since: the arrival pushed lastSeen + // past lastProbeOK, so the premise case fires and this one is never + // reached. That is the common shape and the one the test produces. + // - Recovered AND re-probed since: the publisher runs on its own + // goroutine at its own cadence, so it can land a successful probe + // between the arrival and this decision. lastProbeOK is then ahead + // of lastSeen again, the premise case passes, and only this one + // stops a healthy subscription being torn down. + // + // Deleting this because "the matrix says it survives" would remove the + // second shape's only guard. + b.retirePendingLocked(c.workspaceID, c.pending) + b.mu.Unlock() + close(c.pending.done) + slog.Info("events: a workspace queued for an idle cycle started receiving again before its turn; leaving its subscription alone", + "workspace", c.workspaceID) + return + } + + // The drop must precede the teardown, because it authenticates against the + // LIVE subscription's generation and stopRedisSubscription deletes that + // entry. Both now happen without releasing the lock in between, so there is + // no window in which coverage is ended for a workspace this function then + // decides not to cycle. + report := b.dropWorkspaceCoverageLocked(c.workspaceID, ResetReasonIdleTimeout, c.gen) + b.stopRedisSubscription(c.workspaceID) + b.mu.Unlock() + + // LOGGED AFTER THE UNLOCK, and after the decision is final (codex rounds 6 + // and 12). Two separate reasons, both learned the hard way: + // + // - After the DECISION, so the log cannot describe a cycle that then + // abandons. It still says ATTEMPTING to replace, because + // establishSubscription can install nothing if the bus closes or the + // workspace empties while it dials — an operator correlating this line + // with pad_event_subscription_cycled_total would otherwise find the log + // without the counter and go hunting a bug that is not there. + // - After the UNLOCK, because slog runs the installed handler + // synchronously and b.mu is the lock every fan-out and every Subscribe + // on this instance contends for. A slow or custom handler would stall + // all of them, and one that called back into the bus would deadlock. + slog.Warn("events: no traffic on this workspace's Redis subscription within the idle timeout; ending its replay coverage and attempting to replace the connection, resumes across the silence will report sync_required", + "workspace", c.workspaceID, "idle_timeout", idleTimeout) + + // Reported with the lock released: an Observer callback may call back into + // the bus (see the Observer interface for the one thing it may not do). + if report != "" { + b.reportReset(report) + } + + // RULE 4: b.ctx, and a nil establisher. establishSubscription owns the + // record from here — it installs or abandons, and retires the record in the + // same critical section either way, so no joiner is stranded by a cycle any + // more than by a cancelled caller (BUG-2749). + installed := b.establishSubscription(b.ctx, c.workspaceID, nil, c.pending) + if !installed { + // THE OUTCOME IS LOGGED, not only the attempt (codex round 16). The + // line above says "attempting"; without this an on-call correlating it + // with pad_event_subscription_cycled_total finds a log with no counter + // and no explanation, on the one path where that is expected. + slog.Warn("events: the idle cycle installed no replacement subscription; the workspace was left uncovered because the bus is closing or it lost its last subscriber", + "workspace", c.workspaceID) + } + + if b.afterCycleEstablish != nil { + b.afterCycleEstablish(c.workspaceID) + } + + // AND ONLY IF A REPLACEMENT ACTUALLY LANDED (codex round 3). The counter's + // documented meaning is "torn down AND replaced", and establishSubscription + // has two reasons to install nothing: the bus closed under us, or the + // workspace emptied while we dialled. Reporting unconditionally would count + // those as cycles, which is wrong in the direction that matters — an + // operator reading a non-zero rate concludes connections are being + // blackholed, and a shutdown would manufacture that signal. The teardown is + // still visible through the idle_timeout reset reason when a buffer existed + // to drop. + // + // TAKEN FROM THE ESTABLISHMENT ITSELF, not inferred from the live + // generation afterwards (codex round 13). Inference is wrong in both + // directions: if this cycle installed nothing and an unrelated caller + // established the workspace before the check, that caller's subscription + // was counted as this cycle's replacement; and a real replacement that + // immediately lost its last subscriber was missed. + if installed { + b.reportSubscriptionCycled() + } +} diff --git a/internal/events/redis_heartbeat_test.go b/internal/events/redis_heartbeat_test.go new file mode 100644 index 00000000..85eb278d --- /dev/null +++ b/internal/events/redis_heartbeat_test.go @@ -0,0 +1,1862 @@ +package events + +import ( + "context" + "fmt" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/alicebob/miniredis/v2" + "github.com/redis/go-redis/v9" + + "github.com/PerpetualSoftware/pad/internal/redisns" +) + +// testClock is a hand-driven clock for idle detection. +// +// A REAL CLOCK CANNOT TEST THIS. The threshold is 90 seconds by construction — +// it is three publish intervals, not a tuned number — so every test here would +// either sleep through one or be rewritten against a threshold production does +// not use. Driving the clock tests the SHIPPED constants. +type testClock struct { + mu sync.Mutex + t time.Time +} + +func (c *testClock) now() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + return c.t +} + +func (c *testClock) advance(d time.Duration) { + c.mu.Lock() + c.t = c.t.Add(d) + c.mu.Unlock() +} + +// newHeartbeatBus builds a phase-1 bus (recognises heartbeats, publishes none) +// on a hand-driven clock, with an observer attached. +func newHeartbeatBus(t *testing.T, publishHeartbeat bool) (*RedisBus, *miniredis.Miniredis, *testClock, *recordingObserver) { + t.Helper() + mr := miniredis.RunT(t) + client := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + t.Cleanup(func() { _ = client.Close() }) + + b := NewRedisBusWithKeys(client, redisns.Default, false, publishHeartbeat) + clock := &testClock{t: time.Now()} + // Set before anything subscribes, so the INSTALL stamp is on this clock + // too — the seam is read only from now(), and the maintenance loop's first + // read is a full interval away. + b.nowFunc = clock.now + obs := &recordingObserver{} + b.SetObserver(obs) + t.Cleanup(b.Close) + return b, mr, clock, obs +} + +// resetHookObserver runs a callback on SequenceReset and ignores everything +// else. It is how a test reaches the one window inside cycleOne that has no +// seam of its own — see TestAnIdleCycleAbandonsAWorkspaceEmptiedUnderIt. +type resetHookObserver struct { + onReset func(reason string) +} + +func (o resetHookObserver) ResumeGap(string) {} +func (o resetHookObserver) EventDropped(string) {} +func (o resetHookObserver) ReceiveLoopExited() {} +func (o resetHookObserver) SubscriptionUnconfirmed() {} +func (o resetHookObserver) SubscriptionCycled() {} +func (o resetHookObserver) HeartbeatPublishFailed() {} +func (o resetHookObserver) SequenceReset(reason string) { o.onReset(reason) } + +func (b *RedisBus) channelFor(workspaceID string) string { + return b.keys.Name(redisChannelSuffix) + workspaceID +} + +// rawSubscriber returns a channel of raw payload strings on one workspace's +// Redis channel, bypassing the bus entirely — the only way to assert what this +// instance PUBLISHES rather than what it does with what it receives. +func rawSubscriber(t *testing.T, addr, channel string) <-chan string { + t.Helper() + client := redis.NewClient(&redis.Options{Addr: addr}) + t.Cleanup(func() { _ = client.Close() }) + ps := client.Subscribe(context.Background(), channel) + t.Cleanup(func() { _ = ps.Close() }) + if _, err := ps.Receive(context.Background()); err != nil { + t.Fatalf("raw subscribe to %s: %v", channel, err) + } + out := make(chan string, 32) + go func() { + for msg := range ps.Channel() { + out <- msg.Payload + } + }() + return out +} + +// wedge simulates the failure this whole unit exists to detect: the outbound +// probe keeps SUCCEEDING and nothing ever comes back. +// +// It has to be simulated rather than produced, because miniredis is a working +// Redis — a heartbeat published against it is delivered straight back to our +// own subscription and the workspace never looks idle. So the test advances +// the clock past the threshold (nothing arrived) and stamps lastProbeOK as a +// successful publish pass would (the probe went out). That pair IS a half-open +// route: writes accepted, reads dead. +// +// Without the stamp these tests would exercise the "we could not even probe" +// case instead, where the detector deliberately does nothing — which is how +// the premise check announced itself when it landed: every cycling test went +// red at once. +func wedge(t *testing.T, b *RedisBus, clock *testClock, d time.Duration) { + t.Helper() + clock.advance(d) + b.mu.Lock() + for _, sub := range b.wsSubs { + sub.lastProbeOK = clock.now() + } + b.mu.Unlock() +} + +// waitFor polls until cond holds, so a test asserts an outcome rather than a +// sleep. Fails the test rather than hanging. +func waitFor(t *testing.T, what string, cond func() bool) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + if cond() { + return + } + time.Sleep(time.Millisecond) + } + t.Fatalf("timed out waiting for %s", what) +} + +// --------------------------------------------------------------------------- +// THE JOINT (BUG-2738). The idle detector is a THIRD actor in a region whose +// invariants were all designed around request goroutines plus Close. This is +// the test the plan named before the code was written, mirroring BUG-2749's +// joiner test against a different third actor. +// --------------------------------------------------------------------------- + +// TestAJoinerIsServedAcrossAnIdleFiredCycle is the joint test. +// +// A subscriber arrives for a workspace at the exact moment the idle detector is +// replacing that workspace's connection. It must end up served by a LIVE +// subscription — never stranded on an establishment promise nobody keeps, and +// never handed a channel behind a subscription the cycle has already torn down. +// +// The two failure shapes it discriminates between, both of which look like a +// working test run from the outside: +// +// - If the cycle tore down wsSubs WITHOUT first minting the establishment +// record, the joiner would find no live subscription and no pending one, +// appoint itself establisher, and open a SECOND Redis subscription for the +// workspace — two connections, two receive loops, every event twice. +// - If the cycle minted the record and then abandoned it without retiring it, +// the joiner would wait on a promise nobody keeps, return with a channel +// wired to nothing, and — because its own registration keeps wsCounts +// non-zero — no later caller would establish either. +func TestAJoinerIsServedAcrossAnIdleFiredCycle(t *testing.T) { + b, mr, clock, _ := newHeartbeatBus(t, true) + + first, _, outcome := b.Subscribe(context.Background(), "ws-1") + if outcome != SubscribeOK { + t.Fatalf("first subscriber: outcome %v", outcome) + } + defer b.Unsubscribe(first) + + genBefore, ok := b.liveGen("ws-1") + if !ok { + t.Fatal("fixture: no subscription was installed for ws-1") + } + + // Arm the seam only for the CYCLE's establishment: the first subscriber's + // is already finished. + var joinerCh chan Event + var joinerOutcome SubscribeOutcome + joinerDone := make(chan struct{}) + var armed atomic.Bool + armed.Store(true) + var once sync.Once + b.beforeInstallSubscription = func(workspaceID string) { + if !armed.Load() { + return + } + once.Do(func() { + go func() { + defer close(joinerDone) + joinerCh, _, joinerOutcome = b.SubscribeIfAllowed(context.Background(), workspaceID, 0) + }() + // The joiner must be REGISTERED and waiting before the cycle's + // establishment proceeds; otherwise this test would be exercising + // an ordinary sequential subscribe and would pass against every + // defect above. + waitFor(t, "the joiner to register against the in-flight cycle", func() bool { + b.mu.Lock() + defer b.mu.Unlock() + _, pending := b.pendingSubs[workspaceID] + return b.wsCounts[workspaceID] >= 2 && pending + }) + }) + } + + wedge(t, b, clock, 2*DefaultIdleTimeout) + b.cycleIdleSubscriptions() + armed.Store(false) + + // BOUNDED, and the bound is part of the instrument. An implementation that + // drops coverage without re-establishing never reaches the seam at all, so + // the joiner is never spawned and this channel never closes — an unbounded + // receive would turn that mutation into a hung test run rather than a + // named failure, which is not a detection anyone can act on. + select { + case <-joinerDone: + case <-time.After(10 * time.Second): + t.Fatal("the cycle never established a replacement, so no joiner ever ran: this is drop-only, not drop-and-cycle") + } + if joinerOutcome != SubscribeOK { + t.Fatalf("the joiner was refused across the cycle: outcome %v", joinerOutcome) + } + if joinerCh == nil { + t.Fatal("the joiner returned no channel") + } + defer b.Unsubscribe(joinerCh) + + genAfter, ok := b.liveGen("ws-1") + if !ok { + t.Fatal("no subscription is installed for ws-1 after the cycle: the joiner is holding a channel wired to nothing") + } + if genAfter == genBefore { + t.Fatal("the subscription was never replaced; this test did not exercise a cycle") + } + + // EXACTLY ONE REDIS SUBSCRIPTION, which the delivery assertion below does + // NOT establish on its own (codex round 8). If the cycle had torn wsSubs + // down WITHOUT first minting the establishment record, the joiner would + // have found neither a live subscription nor a pending one, appointed + // itself establisher, and opened a SECOND connection and receive loop for + // this workspace — and every subscriber would still have received the event + // below, once per channel, because fan-out is per subscriber. The only + // thing that separates one subscription from two is counting them. + channel := b.channelFor("ws-1") + waitFor(t, "the cycled subscription to settle to exactly one", func() bool { + return mr.PubSubNumSub(channel)[channel] == 1 + }) + + // And the joiner's channel is behind a LIVE subscription rather than a + // torn-down one. + b.Publish(Event{Type: ItemCreated, WorkspaceID: "ws-1"}) + for name, ch := range map[string]chan Event{"joiner": joinerCh, "first subscriber": first} { + select { + case <-ch: + case <-time.After(3 * time.Second): + t.Fatalf("%s received nothing after the cycle: its channel is behind a subscription that no longer exists", name) + } + // Delivered ONCE. Two receive loops on one channel deliver every event + // twice, which is the other half of the two-subscription failure and is + // invisible to a test that only checks something arrived. + select { + case dup := <-ch: + t.Fatalf("%s received the event twice (%+v): there is more than one receive loop on this workspace", name, dup) + case <-time.After(200 * time.Millisecond): + } + } +} + +// TestAResumingJoinerIsToldSyncRequiredAcrossACycle answers codex round 2's P2 +// with the case that discriminates. +// +// The concern raised was that a subscriber arriving DURING a cycle gets no gap +// signal — dropWorkspaceCoverage signals only the subscribers present when it +// runs. That is true, and for a RESUMING caller it is not the mechanism that +// protects it: the registration MARK is. It registers while the workspace has +// no buffer at all, so its mark cannot match the buffer that exists by the +// time it reads, and eventsSinceMarkLocked answers nil — the strongest form of +// "this instance cannot vouch". Its caller then reports a resume gap and the +// SSE layer answers sync_required. +// +// MUTATION-CONFIRMED, and the first two attempts at this test were not. +// Replacing eventsSinceMarkLocked with the unmarked eventsSinceLocked fails +// here with the joiner handed the post-cycle event as though it followed its +// cursor — which is the defect this asserts against. Note that deleting the +// `mark.buffer == nil` term ALONE survives: inside that function the keep +// arithmetic already reduces to zero for a nil mark, so that term is redundant +// with its neighbour rather than load-bearing. The mark being CONSULTED AT ALL +// is what matters. +// +// A FRESH caller (sinceID == 0) is deliberately NOT signalled, and that is not +// an oversight: it holds no prior position, so there is no span it could be +// missing. It is also admitted only after the replacement subscription is +// acknowledged — it waits on the cycle's establishment record, which +// finishPending closes after the confirmation — and on the unconfirmed path it +// is told to reconcile when the acknowledgement eventually lands. Signalling +// it anyway would be a resync demanded of a client with nothing to reconcile, +// which is the load inversion this unit has already had to fix once. +func TestAResumingJoinerIsToldSyncRequiredAcrossACycle(t *testing.T) { + b, _, clock, _ := newHeartbeatBus(t, true) + + first, _, _ := b.Subscribe(context.Background(), "ws-1") + defer b.Unsubscribe(first) + + b.Publish(Event{Type: ItemCreated, WorkspaceID: "ws-1"}) + var seen Event + select { + case seen = <-first: + case <-time.After(3 * time.Second): + t.Fatal("fixture: no event, so there is no cursor to resume from") + } + + var joinerMissed []Event + var joinerOutcome SubscribeOutcome + var joinerCh chan Event + joinerDone := make(chan struct{}) + var once sync.Once + var armed atomic.Bool + armed.Store(true) + b.beforeInstallSubscription = func(workspaceID string) { + if !armed.Load() { + return + } + once.Do(func() { + go func() { + defer close(joinerDone) + joinerCh, joinerMissed, _, joinerOutcome = + b.SubscribeAndReplaySince(context.Background(), workspaceID, seen.ID, 0) + }() + waitFor(t, "the resuming joiner to register mid-cycle", func() bool { + b.mu.Lock() + defer b.mu.Unlock() + return b.wsCounts[workspaceID] >= 2 + }) + }) + } + + // A BUFFER MUST EXIST AGAIN BEFORE THE JOINER READS ITS REPLAY, or this + // test proves nothing. Mutation-checked: without this, the cycle leaves no + // buffer at all, eventsSinceMarkLocked returns nil from its FIRST term + // (`!ok`), and removing the mark check entirely still passes — the test + // would be asserting the empty case rather than the one it is named for. + // Publishing here puts a FRESH buffer in place, so the only thing that can + // still answer nil is the mark not matching it. + b.afterSubscriptionConfirmed = func() { + if !armed.Load() { + return + } + b.Publish(Event{Type: ItemUpdated, WorkspaceID: "ws-1"}) + waitFor(t, "the post-cycle event to rebuild the workspace's buffer", func() bool { + b.mu.Lock() + defer b.mu.Unlock() + return b.replayBuffers["ws-1"] != nil + }) + } + + wedge(t, b, clock, DefaultIdleTimeout+time.Second) + b.cycleIdleSubscriptions() + armed.Store(false) + + select { + case <-joinerDone: + case <-time.After(10 * time.Second): + t.Fatal("the resuming joiner never returned") + } + + b.mu.Lock() + rebuilt := b.replayBuffers["ws-1"] != nil + b.mu.Unlock() + if !rebuilt { + t.Fatal("fixture: no buffer was rebuilt before the joiner read its replay, so this test could not have discriminated") + } + if joinerOutcome != SubscribeOK { + t.Fatalf("the resuming joiner was refused: %v", joinerOutcome) + } + defer b.Unsubscribe(joinerCh) + + if joinerMissed != nil { + t.Fatalf("a caller resuming from %d across a cycle was answered with %d replayed events instead of sync_required: it was told it was caught up across a span this instance cannot vouch for", + seen.ID, len(joinerMissed)) + } +} + +// --------------------------------------------------------------------------- +// Phase 1: recognise and ignore. +// --------------------------------------------------------------------------- + +// TestAHeartbeatDoesNotEndAWorkspacesCoverage is the phase-1 contract, and it +// is the half that makes the two-phase roll zero-loss. +// +// Fails without the classification in decodePayload: "hb|1" has one separator, +// so it falls through to the bare-JSON branch, fails to unmarshal, and — since +// BUG-2739 — is treated as a hole in coverage. The buffer would be dropped and +// every live subscriber told to resync, every interval, for the length of a +// mixed deployment. +func TestAHeartbeatDoesNotEndAWorkspacesCoverage(t *testing.T) { + b, mr, _, obs := newHeartbeatBus(t, false) + + ch, _, outcome := b.Subscribe(context.Background(), "ws-1") + if outcome != SubscribeOK { + t.Fatalf("subscribe: %v", outcome) + } + defer b.Unsubscribe(ch) + + b.Publish(Event{Type: ItemCreated, WorkspaceID: "ws-1"}) + var first Event + select { + case first = <-ch: + case <-time.After(3 * time.Second): + t.Fatal("fixture: the first event never arrived, so there is no buffer to protect") + } + + client := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + defer func() { _ = client.Close() }() + if err := client.Publish(context.Background(), b.channelFor("ws-1"), heartbeatPayload).Err(); err != nil { + t.Fatalf("publish heartbeat: %v", err) + } + + // ORDERING BARRIER, not a sleep: both are published on the same channel, so + // Redis delivers them to this subscription in order. Receiving the second + // event proves the heartbeat has already been through receiveMessages. + b.Publish(Event{Type: ItemUpdated, WorkspaceID: "ws-1"}) + select { + case <-ch: + case <-time.After(3 * time.Second): + t.Fatal("the event published after the heartbeat never arrived") + } + + if reasons := obs.resetReasons(); len(reasons) != 0 { + t.Fatalf("a heartbeat ended this workspace's coverage: resets %v", reasons) + } + if missed := b.EventsSince("ws-1", first.ID); len(missed) != 1 { + t.Fatalf("the replay buffer no longer covers the span across the heartbeat: EventsSince returned %d events, want 1", len(missed)) + } +} + +// TestAHeartbeatReachesNoSubscriber pins the other half of "bus-internal": a +// heartbeat must never be delivered as an event, which would put a zero-valued +// Event with no type and no id in front of a client. +func TestAHeartbeatReachesNoSubscriber(t *testing.T) { + b, mr, _, _ := newHeartbeatBus(t, false) + + ch, _, _ := b.Subscribe(context.Background(), "ws-1") + defer b.Unsubscribe(ch) + + client := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + defer func() { _ = client.Close() }() + if err := client.Publish(context.Background(), b.channelFor("ws-1"), heartbeatPayload).Err(); err != nil { + t.Fatalf("publish heartbeat: %v", err) + } + b.Publish(Event{Type: ItemCreated, WorkspaceID: "ws-1"}) + + select { + case got := <-ch: + if got.Type != ItemCreated { + t.Fatalf("a heartbeat was delivered to a subscriber as an event: %+v", got) + } + case <-time.After(3 * time.Second): + t.Fatal("the real event never arrived") + } +} + +// TestAHeartbeatIsRecognisedByPrefixNotByExactPayload pins the forward +// compatibility the prefix buys: a LATER version of the frame, carrying fields +// this binary knows nothing about, must still be ignored rather than ending +// coverage. Without it, extending the frame would need a third two-phase roll. +func TestAHeartbeatIsRecognisedByPrefixNotByExactPayload(t *testing.T) { + for _, payload := range []string{ + heartbeatPayload, + heartbeatPrefix + "2", + heartbeatPrefix + "2|instance-a|1699999999", + } { + kind, _, _, err := decodePayload(payload) + if err != nil { + t.Fatalf("decodePayload(%q) errored: %v — an error here ends a workspace's coverage", payload, err) + } + if kind != payloadHeartbeat { + t.Fatalf("decodePayload(%q) kind = %v, want payloadHeartbeat", payload, kind) + } + } +} + +// TestAPayloadWearingThePrefixStillEndsCoverage is the other half of the frame +// contract, and the one codex round 5 asked for. +// +// The prefix buys forward compatibility, but it must not become a hole in the +// loud path: before this feature, EVERY payload this instance could not read +// ended the workspace's coverage and moved undecodable_message, the counter +// whose documented job is "suspect a namespace collision". A foreign or buggy +// publisher whose bytes happen to begin with "hb|" must not slip through that +// signal. +// +// So: a disciplined frame (version, then narrow tokens, under a length cap) is +// ignored; anything else wearing the prefix goes back to being a decode +// failure. Note the asymmetry this deliberately does NOT close — a frame that +// arrives has still proved the socket carries traffic, whoever sent it, so +// liveness cannot be forged in the first place. There is no coverage claim +// inside a heartbeat to fake. +func TestAPayloadWearingThePrefixStillEndsCoverage(t *testing.T) { + for _, payload := range []string{ + "hb|", // no version + "hb|x", // version is not a number + "hb|1|