fix(events): surface the SUBSCRIBE error and refuse callers instead of admitting a dead stream (BUG-2764)

docs/deployment.md

@@ -420,9 +420,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. `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_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, the workspace loses its last subscriber, or Redis refuses the `SUBSCRIBE` (BUG-2764 — logged with the error), 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, and take it from the **"Event bus using Redis pub/sub"** line: since BUG-2769 the watch bus logs a `heartbeat_phase` of its own, on its own line, under a separate flag, and it has no bearing on this counter |
+| `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, the workspace lost its last subscriber, or Redis refused the `SUBSCRIBE` does not increment it, so a restart cannot manufacture this signal and a refused replacement does not count as one. 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, and take it from the **"Event bus using Redis pub/sub"** line: since BUG-2769 the watch bus logs a `heartbeat_phase` of its own, on its own line, under a separate flag, and it has no bearing on this counter |
 | `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 |
@@ -911,10 +911,19 @@ workspace's subscription, so:
   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.
+  can fail on a *healthy* path used to be invisible (BUG-2764, fixed): go-redis
+  discards the error from a `SUBSCRIBE` issued through `Client.Subscribe`, so a
+  failed subscribe yielded a connection that looked live and was subscribed to
+  nothing, and only a later reconnect or the detector's next pass (phase 2)
+  ever replaced it. Pad now issues the `SUBSCRIBE` where its error is
+  visible: a failed one installs nothing and is logged with the error. On the
+  **activity stream** (`/api/v1/events`) its callers are then refused with a
+  503 (`subscription_failed`, `Retry-After`) rather than admitted into a
+  stream that would carry nothing — on both phases, with no detector
+  involved. The **watch stream** cannot refuse yet: its bus has no failure
+  outcome, so a watch client on an instance whose single subscription could
+  not be established is still admitted and hears nothing (BUG-2800); phase 2
+  re-establishes on the next maintenance pass, phase 1 does not.
 
 **What to watch.** On the activity bus,
 `pad_event_subscription_cycled_total` — expect zero. Read it rather than that

internal/events/bus.go

@@ -156,6 +156,16 @@ const (
 	// be reported against it: it is not a refusal, not a resume this instance
 	// failed to serve, and not evidence about Redis.
 	SubscribeCancelled
+
+	// SubscribeFailed means this instance could not open a Redis subscription
+	// for the workspace — the SUBSCRIBE could not be delivered (BUG-2764), or
+	// the bus is shutting down — and the caller's registration has been
+	// undone. Unlike an acknowledgement that merely arrives late (admitted,
+	// counted, reconciled when it lands), this is a stream that would never
+	// carry anything, so it is refused: the client can reconnect, and a
+	// refusal it can see beats a silence it cannot. MemoryBus never returns
+	// it.
+	SubscribeFailed
 )
 
 func (o SubscribeOutcome) String() string {
@@ -166,6 +176,8 @@ func (o SubscribeOutcome) String() string {
 		return "workspace_limit"
 	case SubscribeCancelled:
 		return "cancelled"
+	case SubscribeFailed:
+		return "failed"
 	default:
 		return "unknown"
 	}

internal/events/redis_bus.go

@@ -3,6 +3,7 @@ package events
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"log/slog"
 	"strconv"
@@ -1144,6 +1145,58 @@ func (b *RedisBus) subscribeAndReplay(ctx context.Context, workspaceID string, s
 	}
 
 	b.mu.Lock()
+	// A CALLER WHOSE LOOP ENDED WITH NOTHING BEHIND ITS CHANNEL HAS FAILED
+	// (BUG-2764). The two abandon reasons that predate this check only fire
+	// when there is nobody to tell — the workspace emptied, or the bus is
+	// closing — so a channel wired to nothing was harmless. A failed
+	// SUBSCRIBE abandons with callers still registered, and this is the only
+	// place every one of them passes through, establisher and joiner alike,
+	// after the loop's built-in retry.
+	//
+	// AN IN-FLIGHT RECORD IS WAITED ON, NOT TAKEN AS SUCCESS (codex round 1
+	// P2): an idle cycle re-establishing in its dial window owns one, and
+	// this caller's registration is served only if that attempt installs —
+	// so the caller waits for it and looks again, and returns failed only
+	// when there is neither a live subscription nor anyone still trying.
+	//
+	// A CLOSED BUS IS A REFUSAL TOO (codex round 1 P2). Close drains the
+	// subscriber maps; a caller that registers after that holds a channel
+	// nothing will ever close, and a handler holding it would outlive the
+	// shutdown. Nothing this instance can promise is worth a 200 here.
+	for {
+		if b.ctx.Err() != nil {
+			b.unsubscribeLocked(sub.ch)
+			b.mu.Unlock()
+			resuming = false
+			return nil, nil, nil, SubscribeFailed
+		}
+		// PENDING BEFORE WSSUBS, the same order section 1 uses and for the
+		// same reason (codex round 2 P1): a replacement installs its wsSubs
+		// entry BEFORE Redis acknowledges it and holds its record until the
+		// acknowledgement lands, so "live" alone can be a subscription in its
+		// unconfirmed window. A record in flight is waited on first; only a
+		// live entry with nobody establishing is a subscription to return.
+		p, inFlight := b.pendingSubs[workspaceID]
+		if !inFlight {
+			if _, live := b.wsSubs[workspaceID]; live {
+				break
+			}
+			b.unsubscribeLocked(sub.ch)
+			b.mu.Unlock()
+			resuming = false
+			return nil, nil, nil, SubscribeFailed
+		}
+		b.mu.Unlock()
+		select {
+		case <-p.done:
+		case <-ctx.Done():
+			b.Unsubscribe(sub.ch)
+			resuming = false
+			return nil, nil, nil, SubscribeCancelled
+		case <-b.ctx.Done():
+		}
+		b.mu.Lock()
+	}
 	defer b.mu.Unlock()
 	if b.afterSubscribeRegister != nil {
 		b.afterSubscribeRegister()
@@ -1529,16 +1582,39 @@ func (b *RedisBus) establishSubscription(ctx context.Context, workspaceID string
 	//
 	// The cancellation check below is repeated after the dial anyway: it costs
 	// nothing and does not depend on the dialer being the one we think it is.
-	pubsub := b.client.Subscribe(dialCtx, channel)
+	//
+	// ISSUED WHERE THE ERROR IS VISIBLE (BUG-2764). Client.Subscribe with
+	// channels discards the write's error — go-redis v9.22.0, redis.go:
+	// `_ = pubsub.Subscribe(ctx, channels...)` — so a SUBSCRIBE that failed to
+	// reach Redis (dial refused, connection dying mid-command, the caller's
+	// context ending mid-dial) came back as a PubSub indistinguishable from a
+	// healthy one. It was then installed, its acknowledgement never arrived,
+	// and its callers were admitted as an unconfirmed subscription promised a
+	// reconcile that could never come: a stream subscribed to nothing, for
+	// the life of the process on heartbeat phase 1. With no channels the
+	// first call neither dials nor writes; the second does both and returns
+	// what happened.
+	pubsub := b.client.Subscribe(dialCtx)
+	subscribeErr := pubsub.Subscribe(dialCtx, channel)
 	subCtx, subCancel := context.WithCancel(b.ctx)
 
 	if b.beforeInstallSubscription != nil {
 		b.beforeInstallSubscription(workspaceID)
 	}
 
 	b.mu.Lock()
-	// TWO REASONS TO ABANDON, and both must retire the establishment record in
-	// THIS critical section (codex round 2, both P1s).
+	// THREE REASONS TO ABANDON, and all must retire the establishment record
+	// in THIS critical section (codex round 2, both P1s; BUG-2764 for the
+	// third).
+	//
+	// SUBSCRIBE failed: the client could not deliver the command (a refused
+	// dial, a write that failed, a caller's context ending mid-dial), so
+	// there is no subscription to install and the PubSub would only carry a
+	// receive loop that can never be acknowledged. Nobody is admitted into
+	// it: once the loop's one built-in retry has had its go, a caller whose
+	// loop ends with no live subscription and no record returns
+	// SubscribeFailed. The record is retired here for exactly the reason the
+	// other two retire theirs.
 	//
 	// Nobody left: everyone who wanted this workspace disconnected while we
 	// were dialling. Installing now would leave a receive loop and a Redis
@@ -1583,12 +1659,23 @@ func (b *RedisBus) establishSubscription(ctx context.Context, workspaceID string
 	if establisher != nil && ctx.Err() != nil {
 		b.unsubscribeLocked(establisher.ch)
 	}
-	if b.wsCounts[workspaceID] == 0 || b.ctx.Err() != nil {
+	if subscribeErr != nil || b.wsCounts[workspaceID] == 0 || b.ctx.Err() != nil {
 		b.retirePendingLocked(workspaceID, pending)
 		b.mu.Unlock()
 		subCancel()
 		_ = pubsub.Close()
 		close(pending.done)
+		// Logged after the unlock, and only for the reason that is a fault:
+		// an emptied workspace and a closing bus are both quiet by design. A
+		// cancelled caller's dial ends in a context error here too — that is
+		// its departure, not Redis failing, and its own path reports it.
+		// CLASSIFIED BY THE ERROR, not by the contexts (codex round 1 P2): a
+		// genuine Redis failure that happens to race the caller leaving is
+		// still a Redis failure, and gating on ctx.Err() would swallow it.
+		if subscribeErr != nil && !errors.Is(subscribeErr, context.Canceled) && !errors.Is(subscribeErr, context.DeadlineExceeded) {
+			slog.Error("events: Redis refused or dropped the SUBSCRIBE; no subscription was installed and its callers are being refused rather than admitted into a stream that would carry nothing",
+				"workspace", workspaceID, "error", subscribeErr)
+		}
 		return false
 	}
 	b.subGen++