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

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

* fix(watchevents): surface the SUBSCRIBE error at construction and on resubscribe (BUG-2764)

* docs(events): the idle-cycle prose and metric Help now name the third install-nothing reason (BUG-2764)

* fix(events): retry uncovered workspaces, wait on in-flight records, refuse after Close (BUG-2764 codex round 1)

* fix(events): post-loop check waits on an in-flight record before trusting a live entry (BUG-2764 codex round 2)

* test(watchevents): assert what the failed subscribe leaves, not how long it takes (BUG-2764 codex round 3)

* docs(events): prose says what the code guarantees — delivery failure or shutdown, replacement refusals do not count as cycles (BUG-2764 codex round 4)

* test(server): the 503 mapping is asserted on both subscribe branches; docs scope the refusal to the activity stream (BUG-2764 codex round 5, BUG-2800)

* fix(events): uncovered-workspace retry logs are quiet once the bus is closing (BUG-2764 codex round 6)

* fix(events): the uncovered-retry log promises a retry only while subscribers remain (BUG-2764 codex round 7)

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,7 +1582,20 @@ 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 {
@@ -1537,8 +1603,18 @@ func (b *RedisBus) establishSubscription(ctx context.Context, workspaceID string
 	}
 
 	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++

internal/events/redis_heartbeat.go

@@ -569,6 +569,37 @@ func (b *RedisBus) cycleIdleSubscriptions() {
 		b.pendingSubs[ws] = pending
 		due = append(due, idleCycle{workspaceID: ws, gen: sub.gen, pending: pending})
 	}
+	// UNCOVERED WORKSPACES ARE RE-ESTABLISHED ON THE SAME PASS (BUG-2764,
+	// codex round 1 P1). A workspace can have subscribers counted and no
+	// subscription behind them: its last establishment — a cycle's
+	// replacement, or a caller's — abandoned because Redis refused the
+	// SUBSCRIBE. Before BUG-2764 that failure INSTALLED the dead PubSub, so
+	// this loop saw it as idle and cycled it again next pass; now nothing is
+	// installed, the loop above walks wsSubs and never sees the workspace,
+	// and its already-admitted subscribers would sit on a live-looking stream
+	// until a new caller happened to arrive and establish. This pass is that
+	// retry, on the same cadence and under the same record discipline: the
+	// record is minted here so a caller arriving now joins this attempt
+	// rather than starting its own.
+	//
+	// Not a cycle: nothing is torn down and no coverage is dropped (there is
+	// none to drop), so it is dispatched to reestablishUncovered rather than
+	// cycleOne and does not move pad_event_subscription_cycled_total.
+	var uncovered []idleCycle
+	for ws, count := range b.wsCounts {
+		if count == 0 {
+			continue
+		}
+		if _, live := b.wsSubs[ws]; live {
+			continue
+		}
+		if _, inFlight := b.pendingSubs[ws]; inFlight {
+			continue
+		}
+		pending := &pendingSub{done: make(chan struct{})}
+		b.pendingSubs[ws] = pending
+		uncovered = append(uncovered, idleCycle{workspaceID: ws, pending: pending})
+	}
 	b.mu.Unlock()
 
 	// BOUNDED-PARALLEL, NOT SERIAL (codex round 5, P2). Each cycle re-dials,
@@ -602,11 +633,51 @@ func (b *RedisBus) cycleIdleSubscriptions() {
 			b.cycleOne(c, idleTimeout)
 		}(c)
 	}
+	for _, c := range uncovered {
+		wg.Add(1)
+		sem <- struct{}{}
+		go func(c idleCycle) {
+			defer wg.Done()
+			defer func() { <-sem }()
+			b.reestablishUncovered(c)
+		}(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()
 }
 
+// reestablishUncovered opens a subscription for a workspace that has
+// subscribers and no subscription (BUG-2764) — the retry for an
+// establishment that abandoned because Redis refused the SUBSCRIBE. The
+// caller minted c.pending under the lock; establishSubscription owns it from
+// here and retires it whether it installs or abandons, exactly as on the
+// cycle path, so no joiner is stranded by a failed retry either.
+func (b *RedisBus) reestablishUncovered(c idleCycle) {
+	// QUIET ON SHUTDOWN (codex round 6 on BUG-2764): a pass queued before
+	// Close began would otherwise log, per workspace, that subscribers exist
+	// and that the next pass will retry — both false once the bus is
+	// closing, and at 200 workspaces that is a wall of misleading lines in
+	// the one log window an operator is reading. establishSubscription still
+	// runs so the record is retired and the PubSub closed either way.
+	if b.ctx.Err() == nil {
+		slog.Warn("events: a workspace has subscribers but no Redis subscription behind them (its last establishment failed); re-establishing",
+			"workspace", c.workspaceID)
+	}
+	if !b.establishSubscription(b.ctx, c.workspaceID, nil, c.pending) && b.ctx.Err() == nil {
+		// The cause is logged by establishSubscription when it is Redis
+		// refusing; a workspace that emptied meanwhile is quiet by design —
+		// and is NOT retried, since the next pass skips zero-count
+		// workspaces (codex round 7), so the line says so rather than
+		// promising a retry that only happens while subscribers remain.
+		slog.Warn("events: re-establishing an uncovered workspace installed nothing; while it still has subscribers the next idle pass will retry",
+			"workspace", c.workspaceID)
+	}
+	if b.afterCycleEstablish != nil {
+		b.afterCycleEstablish(c.workspaceID)
+	}
+}
+
 // nextTick returns the deadline for the pass after one that was scheduled for
 // prev, given the configured interval and the current time.
 //
@@ -768,7 +839,7 @@ func (b *RedisBus) cycleOne(c idleCycle, idleTimeout time.Duration) {
 		// 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",
+		slog.Warn("events: the idle cycle installed no replacement subscription; the workspace was left uncovered because the bus is closing, it lost its last subscriber, or Redis refused the SUBSCRIBE (logged separately)",
 			"workspace", c.workspaceID)
 	}
 
@@ -778,8 +849,9 @@ func (b *RedisBus) cycleOne(c idleCycle, idleTimeout time.Duration) {
 
 	// 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
+	// has three reasons to install nothing: the bus closed under us, the
+	// workspace emptied while we dialled, or Redis refused the SUBSCRIBE
+	// (BUG-2764). 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

internal/events/redis_subscribe_failed_test.go

@@ -0,0 +1,357 @@
+package events
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"net"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/alicebob/miniredis/v2"
+	"github.com/redis/go-redis/v9"
+
+	"github.com/PerpetualSoftware/pad/internal/redisns"
+)
+
+// subscribeWriteFailer is a Dialer whose connections fail the WRITE of a
+// SUBSCRIBE frame, deterministically, for the first `failures` such writes.
+//
+// WHY A CONN WRAPPER AND NOT A PROXY CLOSE (BUG-2764). The defect is that
+// go-redis's Client.Subscribe discards the error of the SUBSCRIBE write, so the
+// instrument has to make that write FAIL. A proxy that closes the connection
+// on seeing the command races the kernel: the client's write has usually
+// already succeeded into the socket buffer by the time the proxy reads it, and
+// what fails is the later read, which is a different error on a different
+// path. A net.Conn whose Write returns the error is the failure at the point
+// the fix inspects, every time.
+//
+// Every connection is wrapped, including the bus's own pooled ones; only a
+// frame carrying the SUBSCRIBE command is refused, so publishes and pings are
+// untouched.
+type subscribeWriteFailer struct {
+	failures atomic.Int64 // how many SUBSCRIBE writes to fail; <0 means all of them; settable mid-test
+	failed   atomic.Int64
+}
+
+var errInjectedSubscribeWrite = errors.New("injected: SUBSCRIBE write failed")
+
+func (f *subscribeWriteFailer) dial(ctx context.Context, network, addr string) (net.Conn, error) {
+	c, err := (&net.Dialer{}).DialContext(ctx, network, addr)
+	if err != nil {
+		return nil, err
+	}
+	return &subscribeFailingConn{Conn: c, f: f}, nil
+}
+
+type subscribeFailingConn struct {
+	net.Conn
+	f *subscribeWriteFailer
+}
+
+func (c *subscribeFailingConn) Write(p []byte) (int, error) {
+	if bytes.Contains(bytes.ToLower(p), []byte("\r\n$9\r\nsubscribe\r\n")) {
+		if n := c.f.failures.Load(); n < 0 || c.f.failed.Load() < n {
+			c.f.failed.Add(1)
+			return 0, errInjectedSubscribeWrite
+		}
+	}
+	return c.Conn.Write(p)
+}
+
+func newFailingSubscribeBus(t *testing.T, failures int64) (*RedisBus, *subscribeWriteFailer, *recordingObserver) {
+	t.Helper()
+	mr := miniredis.RunT(t)
+	f := &subscribeWriteFailer{}
+	f.failures.Store(failures)
+	client := redis.NewClient(&redis.Options{Addr: mr.Addr(), Dialer: f.dial})
+	t.Cleanup(func() { _ = client.Close() })
+	if err := client.Ping(context.Background()).Err(); err != nil {
+		t.Fatalf("ping: %v", err)
+	}
+	b := NewRedisBus(client)
+	t.Cleanup(b.Close)
+	obs := &recordingObserver{}
+	b.SetObserver(obs)
+	return b, f, obs
+}
+
+// TestAFailedSubscribeRefusesItsCallerInsteadOfAdmittingIt is BUG-2764's
+// regression test.
+//
+// Before the fix a SUBSCRIBE whose write failed came back from go-redis as a
+// PubSub indistinguishable from a healthy one; the bus installed it, waited
+// for an acknowledgement that could never come, and admitted the caller as an
+// unconfirmed subscription — SubscribeOK, a live-looking channel, a counter
+// promising a reconcile when the acknowledgement landed. Nothing ever
+// arrived. Every assertion below is stated as what that WRONG behaviour would
+// leave behind (CONVE-12), because "nothing happened" is compatible with the
+// bug.
+func TestAFailedSubscribeRefusesItsCallerInsteadOfAdmittingIt(t *testing.T) {
+	b, f, obs := newFailingSubscribeBus(t, -1)
+
+	ch, _, outcome := b.SubscribeIfAllowed(context.Background(), "ws-1", 0)
+
+	// PREMISE: the instrument fired. The loop retries establishment once, so
+	// a refusal with fewer than two failed writes is a refusal for some other
+	// reason.
+	if got := f.failed.Load(); got < 2 {
+		t.Fatalf("SUBSCRIBE writes failed = %d, want at least 2 (initial attempt + the loop's retry); the instrument never armed", got)
+	}
+
+	if outcome != SubscribeFailed {
+		t.Fatalf("outcome = %v, want failed — the caller was admitted into a subscription Redis never received", outcome)
+	}
+	if ch != nil {
+		t.Fatal("a refused caller was handed a channel")
+	}
+
+	b.mu.Lock()
+	_, live := b.wsSubs["ws-1"]
+	_, inFlight := b.pendingSubs["ws-1"]
+	count := b.wsCounts["ws-1"]
+	b.mu.Unlock()
+	if live {
+		t.Fatal("a subscription was installed for a SUBSCRIBE that failed to reach Redis")
+	}
+	if inFlight {
+		t.Fatal("an establishment record was left behind: the next caller would wait on it forever")
+	}
+	if count != 0 {
+		t.Fatalf("workspace subscriber count = %d after a refusal, want 0 — the refused caller is still counted against the workspace limit", count)
+	}
+	// The bug's signature: an unconfirmed admission that will never be
+	// reconciled. A refusal reports none.
+	if got := obs.unconfirmedCount(); got != 0 {
+		t.Fatalf("SubscriptionUnconfirmed reported %d times for a refused subscription, want 0 — the caller was admitted rather than refused", got)
+	}
+}
+
+// TestATransientSubscribeFailureIsRetriedOnceAndServesEveryCaller pins the
+// other half: a SUBSCRIBE that fails once is not a refusal. The establish
+// loop's built-in second pass re-establishes, and both the establisher and a
+// joiner that registered during the failed attempt end up on a live
+// subscription — each receiving a published event EXACTLY once.
+//
+// The joiner is placed by the beforeInstallSubscription seam, which runs
+// after the SUBSCRIBE has been issued and before the deciding lock: the one
+// window in which it finds the failing establishment's record and genuinely
+// waits on it.
+func TestATransientSubscribeFailureIsRetriedOnceAndServesEveryCaller(t *testing.T) {
+	b, f, obs := newFailingSubscribeBus(t, 1)
+
+	type result struct {
+		ch      chan Event
+		outcome SubscribeOutcome
+	}
+	joined := make(chan result, 1)
+	var placed atomic.Bool
+	b.beforeInstallSubscription = func(string) {
+		if !placed.CompareAndSwap(false, true) {
+			return
+		}
+		go func() {
+			ch, _, outcome := b.SubscribeIfAllowed(context.Background(), "ws-1", 0)
+			joined <- result{ch, outcome}
+		}()
+		// Let the joiner register and find the record before the abandon
+		// retires it.
+		deadline := time.Now().Add(3 * time.Second)
+		for b.WorkspaceSubscriberCount("ws-1") < 2 {
+			if time.Now().After(deadline) {
+				t.Error("the joiner never registered; this test never exercised the joiner path")
+				break
+			}
+			time.Sleep(time.Millisecond)
+		}
+	}
+
+	ch, _, outcome := b.SubscribeIfAllowed(context.Background(), "ws-1", 0)
+	if got := f.failed.Load(); got != 1 {
+		t.Fatalf("SUBSCRIBE writes failed = %d, want exactly 1; the instrument did not behave as configured", got)
+	}
+	if outcome != SubscribeOK {
+		t.Fatalf("establisher outcome = %v, want ok — one transient failure became a refusal", outcome)
+	}
+	defer b.Unsubscribe(ch)
+
+	var jr result
+	select {
+	case jr = <-joined:
+	case <-time.After(5 * time.Second):
+		t.Fatal("the joiner never returned — the failed establishment's record was not retired and stranded it")
+	}
+	if jr.outcome != SubscribeOK {
+		t.Fatalf("joiner outcome = %v, want ok", jr.outcome)
+	}
+	defer b.Unsubscribe(jr.ch)
+
+	if got := obs.unconfirmedCount(); got != 0 {
+		t.Fatalf("SubscriptionUnconfirmed reported %d times, want 0 — a caller was admitted into the FAILED attempt rather than the retry", got)
+	}
+
+	b.Publish(Event{Type: ItemCreated, WorkspaceID: "ws-1"})
+	for name, c := range map[string]chan Event{"establisher": ch, "joiner": jr.ch} {
+		select {
+		case <-c:
+		case <-time.After(3 * time.Second):
+			t.Fatalf("the %s never received an event published after both were admitted: it is holding a channel wired to nothing", name)
+		}
+		select {
+		case dup := <-c:
+			t.Fatalf("the %s received a second copy (%d): the failed attempt's subscription is alive alongside the retry's", name, dup.ID)
+		case <-time.After(250 * time.Millisecond):
+		}
+	}
+}
+
+// TestAJoinerOfAFailedEstablishmentIsRefusedToo covers the population the
+// establisher's refusal does not reach on its own: a caller that registered
+// while the failing establishment was in flight and waited on its record. It
+// takes the establishment over once, fails the same way, and must return
+// SubscribeFailed rather than SubscribeOK with a dead channel — and must
+// leave the workspace with no subscribers counted, so the next caller's
+// admission is not charged for two ghosts.
+func TestAJoinerOfAFailedEstablishmentIsRefusedToo(t *testing.T) {
+	b, _, _ := newFailingSubscribeBus(t, -1)
+
+	type result struct {
+		ch      chan Event
+		outcome SubscribeOutcome
+	}
+	joined := make(chan result, 1)
+	var placed atomic.Bool
+	b.beforeInstallSubscription = func(string) {
+		if !placed.CompareAndSwap(false, true) {
+			return
+		}
+		go func() {
+			ch, _, outcome := b.SubscribeIfAllowed(context.Background(), "ws-1", 0)
+			joined <- result{ch, outcome}
+		}()
+		deadline := time.Now().Add(3 * time.Second)
+		for b.WorkspaceSubscriberCount("ws-1") < 2 {
+			if time.Now().After(deadline) {
+				t.Error("the joiner never registered; this test never exercised the joiner path")
+				break
+			}
+			time.Sleep(time.Millisecond)
+		}
+	}
+
+	_, _, outcome := b.SubscribeIfAllowed(context.Background(), "ws-1", 0)
+	if outcome != SubscribeFailed {
+		t.Fatalf("establisher outcome = %v, want failed", outcome)
+	}
+
+	var jr result
+	select {
+	case jr = <-joined:
+	case <-time.After(10 * time.Second):
+		t.Fatal("the joiner never returned — stranded on a record nobody retired")
+	}
+	if jr.outcome != SubscribeFailed {
+		t.Fatalf("joiner outcome = %v, want failed — it was handed a channel wired to nothing", jr.outcome)
+	}
+	if jr.ch != nil {
+		t.Fatal("a refused joiner was handed a channel")
+	}
+	if got := b.WorkspaceSubscriberCount("ws-1"); got != 0 {
+		t.Fatalf("workspace subscriber count = %d after both refusals, want 0", got)
+	}
+}
+
+// TestSubscribeAfterCloseIsRefused pins the closed-bus arm of the post-loop
+// check (codex round 1 on BUG-2764). Close drains the subscriber maps; a
+// caller registering afterwards used to be handed SubscribeOK and a channel
+// nothing would ever close, so an SSE handler holding it outlived the
+// shutdown. The wrong behaviour is a channel; the right one is a refusal and
+// no registration left behind.
+func TestSubscribeAfterCloseIsRefused(t *testing.T) {
+	mr := miniredis.RunT(t)
+	client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
+	t.Cleanup(func() { _ = client.Close() })
+	b := NewRedisBus(client)
+	b.Close()
+
+	ch, _, outcome := b.SubscribeIfAllowed(context.Background(), "ws-1", 0)
+	if outcome != SubscribeFailed {
+		t.Fatalf("outcome after Close = %v, want failed", outcome)
+	}
+	if ch != nil {
+		t.Fatal("a caller was handed a channel by a closed bus")
+	}
+	if got := b.WorkspaceSubscriberCount("ws-1"); got != 0 {
+		t.Fatalf("subscriber count after a refused post-Close subscribe = %d, want 0", got)
+	}
+}
+
+// TestAWorkspaceLeftUncoveredByAFailedReplacementIsReestablishedOnTheNextPass
+// pins codex round 1's P1 on BUG-2764. Before the fix, an idle cycle whose
+// replacement SUBSCRIBE failed INSTALLED the dead PubSub, so the detector saw
+// it as idle and cycled it again next pass. With the failure now abandoning
+// instead, the workspace has subscribers counted and nothing behind them, and
+// a scanner that walks only wsSubs would never look at it again: the
+// already-admitted subscribers sit on a live-looking stream until a new caller
+// happens to establish. The uncovered pass is that retry, and this test is
+// driven entirely through the public surface plus the cycle entry point.
+func TestAWorkspaceLeftUncoveredByAFailedReplacementIsReestablishedOnTheNextPass(t *testing.T) {
+	mr := miniredis.RunT(t)
+	f := &subscribeWriteFailer{}
+	client := redis.NewClient(&redis.Options{Addr: mr.Addr(), Dialer: f.dial})
+	t.Cleanup(func() { _ = client.Close() })
+	if err := client.Ping(context.Background()).Err(); err != nil {
+		t.Fatalf("ping: %v", err)
+	}
+	// PHASE 2: the detector, and therefore the retry, only run there.
+	b := NewRedisBusWithKeys(client, redisns.Default, false, true)
+	clock := &testClock{t: time.Now()}
+	b.nowFunc = clock.now
+	t.Cleanup(b.Close)
+
+	ch, _, outcome := b.Subscribe(context.Background(), "ws-1")
+	if outcome != SubscribeOK {
+		t.Fatalf("healthy subscribe outcome = %v, want ok", outcome)
+	}
+	defer b.Unsubscribe(ch)
+
+	// Redis starts refusing SUBSCRIBEs; the route then wedges and the cycle
+	// tears the subscription down and fails to replace it.
+	f.failures.Store(-1)
+	wedge(t, b, clock, b.idleTimeout+time.Second)
+	b.cycleIdleSubscriptions()
+
+	// PREMISE: the state under test exists — a counted subscriber, no
+	// subscription, no record in flight, and a refusal actually happened.
+	b.mu.Lock()
+	_, live := b.wsSubs["ws-1"]
+	_, inFlight := b.pendingSubs["ws-1"]
+	count := b.wsCounts["ws-1"]
+	b.mu.Unlock()
+	if live || inFlight || count != 1 || f.failed.Load() == 0 {
+		t.Fatalf("after the failed replacement: live=%v inFlight=%v count=%d refused=%d; the uncovered state was never reached, so this test could not have discriminated",
+			live, inFlight, count, f.failed.Load())
+	}
+
+	// Redis recovers. The next pass must find the workspace without a
+	// subscription behind its subscribers and re-establish — with NO new
+	// caller arriving to do it for them.
+	f.failures.Store(0)
+	b.cycleIdleSubscriptions()
+
+	b.mu.Lock()
+	_, live = b.wsSubs["ws-1"]
+	b.mu.Unlock()
+	if !live {
+		t.Fatal("the uncovered workspace was not re-established on the next pass: its subscribers stay deaf until a new caller happens to arrive")
+	}
+	// Served, not merely re-installed: the ORIGINAL subscriber's channel
+	// receives again.
+	b.Publish(Event{Type: ItemCreated, WorkspaceID: "ws-1"})
+	select {
+	case <-ch:
+	case <-time.After(3 * time.Second):
+		t.Fatal("an event published after re-establishment never reached the subscriber that was left uncovered")
+	}
+}

internal/metrics/metrics.go

@@ -594,7 +594,7 @@ func New() *Metrics {
 
 	eventSequenceResetsTotal := prometheus.NewCounterVec(prometheus.CounterOpts{
 		Name: "pad_event_sequence_resets_total",
-		Help: "Times activity-event replay coverage was dropped, by reason: subscription_resumed (a Redis connection flap, one workspace's buffer), epoch_change (the shared counter's ID space changed generation, every buffer), counter_backward (an ID at or below a buffer's high-water mark with no generation change), epoch_regressed (the generation counter went backwards and stayed there — usually a Redis failover to a replica that lost writes, and since BUG-2740 also a corrupted generation key having been repaired and reseeded from wall-clock seconds; read the key to tell them apart, a repaired one looks like a unix timestamp), undecodable_message (a pub/sub message could not be parsed, so that workspace's coverage ended), subscription_unconfirmed (a subscription was admitted before Redis acknowledged the SUBSCRIBE and the acknowledgement then arrived; reaches this counter only when a buffer existed to drop — see pad_event_subscription_unconfirmed_total), 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 can install nothing if the instance is shutting down or the workspace loses its last subscriber, so only pad_event_subscription_cycled_total proves a replacement. It establishes that the socket stopped proving it works, NOT that events were observed going missing, and like subscription_unconfirmed it reaches this counter only when a buffer existed to drop).",
+		Help: "Times activity-event replay coverage was dropped, by reason: subscription_resumed (a Redis connection flap, one workspace's buffer), epoch_change (the shared counter's ID space changed generation, every buffer), counter_backward (an ID at or below a buffer's high-water mark with no generation change), epoch_regressed (the generation counter went backwards and stayed there — usually a Redis failover to a replica that lost writes, and since BUG-2740 also a corrupted generation key having been repaired and reseeded from wall-clock seconds; read the key to tell them apart, a repaired one looks like a unix timestamp), undecodable_message (a pub/sub message could not be parsed, so that workspace's coverage ended), subscription_unconfirmed (a subscription was admitted before Redis acknowledged the SUBSCRIBE and the acknowledgement then arrived; reaches this counter only when a buffer existed to drop — see pad_event_subscription_unconfirmed_total), 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 can install 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. It establishes that the socket stopped proving it works, NOT that events were observed going missing, and like subscription_unconfirmed it reaches this counter only when a buffer existed to drop).",
 	}, []string{"reason"})
 
 	eventReceiveLoopExitsTotal := prometheus.NewCounter(prometheus.CounterOpts{

internal/server/handlers_events.go

@@ -261,6 +261,16 @@ func (s *Server) handleSSE(w http.ResponseWriter, r *http.Request) {
 		slog.Debug("SSE client disconnected during subscription establishment",
 			"workspace", ws.Slug)
 		return
+	case events.SubscribeFailed:
+		// This instance could not open the workspace's Redis subscription
+		// (BUG-2764), or is shutting down: the stream would carry nothing,
+		// so refuse it and let the client reconnect rather than hold a
+		// connection that looks live. A Redis failure is logged by the bus
+		// with its error; a shutdown is quiet by design.
+		slog.Warn("SSE subscribe refused: no Redis subscription could be established", "workspace", ws.Slug)
+		w.Header().Set("Retry-After", "5")
+		writeError(w, http.StatusServiceUnavailable, "subscription_failed", "Event subscription could not be established; retry shortly")
+		return
 	default:
 		slog.Error("SSE subscribe returned an unknown outcome; refusing rather than guessing",
 			"workspace", ws.Slug, "outcome", outcome.String())

internal/server/handlers_events_failed_test.go

@@ -0,0 +1,90 @@
+package server
+
+import (
+	"context"
+	"encoding/json"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/PerpetualSoftware/pad/internal/events"
+)
+
+// failingEstablishBus stands in for a bus whose Redis SUBSCRIBE cannot be
+// issued (BUG-2764): every subscribe returns SubscribeFailed and hands back no
+// channel.
+type failingEstablishBus struct {
+	events.EventBus
+}
+
+func (b *failingEstablishBus) SubscribeIfAllowed(context.Context, string, int) (chan events.Event, <-chan struct{}, events.SubscribeOutcome) {
+	return nil, nil, events.SubscribeFailed
+}
+
+func (b *failingEstablishBus) SubscribeAndReplaySince(context.Context, string, int64, int) (chan events.Event, []events.Event, <-chan struct{}, events.SubscribeOutcome) {
+	return nil, nil, nil, events.SubscribeFailed
+}
+
+// TestAFailedSubscriptionIsRefusedWithARetryableStatus is the BINDING
+// assertion for BUG-2764 (CONVE-19): internal/events vouches for the bus
+// returning SubscribeFailed, and says nothing about what the SSE handler does
+// with it. Through the router, not by direct call.
+//
+// The wrong behaviours it is written against: the pre-fix handler had no case
+// for the outcome and fell to the default arm — a 500 "internal_error" that
+// reads as a server bug rather than a transient the client should retry —
+// and a handler that admitted the caller anyway would answer 200 and hold a
+// stream that carries nothing. Both are distinguishable from a 503 with a
+// Retry-After and the named code. The admission slot must also be released:
+// a refusal that kept it would let a Redis outage exhaust the per-instance
+// bound with connections that were never served.
+func TestAFailedSubscriptionIsRefusedWithARetryableStatus(t *testing.T) {
+	srv := testServerWithEvents(t)
+	srv.SetEventBus(&failingEstablishBus{EventBus: events.New()})
+	ts := httptest.NewServer(srv)
+	defer ts.Close()
+
+	slug := createTestWorkspace(t, ts.URL, "Test")
+
+	// BOTH BRANCHES of the handler's subscribe (codex round 5 on BUG-2764):
+	// a fresh connection goes through SubscribeIfAllowed, a resume through
+	// SubscribeAndReplaySince, and each has its own call site that could
+	// have missed the mapping.
+	for name, lastID := range map[string]string{"fresh": "", "resume": "7"} {
+		req, err := http.NewRequest(http.MethodGet, ts.URL+"/api/v1/events?workspace="+slug, nil)
+		if err != nil {
+			t.Fatalf("%s: new request: %v", name, err)
+		}
+		if lastID != "" {
+			req.Header.Set("Last-Event-ID", lastID)
+		}
+		resp, err := http.DefaultClient.Do(req)
+		if err != nil {
+			t.Fatalf("%s: GET: %v", name, err)
+		}
+		body, _ := io.ReadAll(resp.Body)
+		_ = resp.Body.Close()
+
+		if resp.StatusCode != http.StatusServiceUnavailable {
+			t.Fatalf("%s: status = %d, want 503 (body %q)", name, resp.StatusCode, body)
+		}
+		if got := resp.Header.Get("Retry-After"); got == "" {
+			t.Fatalf("%s: no Retry-After on a refusal the client is meant to retry", name)
+		}
+		var payload struct {
+			Error struct {
+				Code string `json:"code"`
+			} `json:"error"`
+		}
+		if err := json.Unmarshal(body, &payload); err != nil {
+			t.Fatalf("%s: body is not the JSON error shape: %v (%q)", name, err, body)
+		}
+		if payload.Error.Code != "subscription_failed" {
+			t.Fatalf("%s: error code = %q, want subscription_failed (body %q)", name, payload.Error.Code, body)
+		}
+		if held := srv.admission().heldTotal(); held != 0 {
+			t.Fatalf("%s: admission slots held after the refusal = %d, want 0", name, held)
+		}
+	}
+}

internal/watchevents/heartbeat.go

@@ -410,7 +410,18 @@ func (b *RedisBus) cycleIfIdle() {
 
 // resubscribe installs a fresh subscription and receive loop.
 func (b *RedisBus) resubscribe() error {
-	pubsub := b.client.Subscribe(b.ctx, b.keys.Name(redisWatchChannelSuffix))
+	// ISSUED WHERE THE ERROR IS VISIBLE (BUG-2764): Client.Subscribe with
+	// channels discards the SUBSCRIBE write's error (go-redis v9.22.0,
+	// redis.go), so a refused or dropped command used to come back as a
+	// healthy-looking PubSub and surface only as the confirmation below
+	// timing out five seconds later, with a deadline error that named the
+	// wrong thing. With no channels the first call neither dials nor writes;
+	// the second does both and says what happened.
+	pubsub := b.client.Subscribe(b.ctx)
+	if err := pubsub.Subscribe(b.ctx, b.keys.Name(redisWatchChannelSuffix)); err != nil {
+		_ = pubsub.Close()
+		return err
+	}
 
 	confirmCtx, cancelConfirm := context.WithTimeout(b.ctx, 5*time.Second)
 	defer cancelConfirm()

internal/watchevents/redis_bus.go

@@ -453,45 +453,78 @@ func NewRedisBusWithKeys(client *redis.Client, size int, keys redisns.Keys, publ
 		heartbeatKick:     make(chan struct{}, 1),
 		idleKick:          make(chan struct{}, 1),
 	}
-	// Eager subscription — see the type comment (2).
-	b.pubsub = client.Subscribe(ctx, keys.Name(redisWatchChannelSuffix))
-
-	// WAIT FOR THE SUBSCRIBE TO BE CONFIRMED before returning. go-redis
-	// establishes the subscription asynchronously, so without this the
-	// constructor hands back a bus that is not yet listening — and Redis
-	// pub/sub is at-most-once, so everything published in that window is
-	// lost to this instance, silently.
+	// Eager subscription — see the type comment (2). Issued in two calls so
+	// the SUBSCRIBE write's error is visible (BUG-2764; see resubscribe): a
+	// refused command used to be indistinguishable from a healthy one here
+	// and surfaced only as the confirmation wait below timing out. The bus is
+	// still constructed on failure — the same non-fatal contract as an
+	// unconfirmed subscription — but the cause is logged at once and the
+	// five-second wait for an acknowledgement that cannot come is skipped.
 	//
-	// Found as a test flake (a second bus constructed and published to
-	// immediately received nothing), which is exactly the shape a rolling
-	// deploy has: a replica comes up, and traffic reaches it before its
-	// subscription is live. The window is small and entirely real.
-	//
-	// A failure here is logged rather than fatal: the receive loop's
-	// ChannelWithSubscriptions() re-subscribes on reconnect, so a bus that
-	// missed its first confirmation still recovers — it just cannot promise
-	// it was listening from the moment it was constructed.
-	//
-	// THIS RECEIVE IS LOAD-BEARING FOR receiveMessages, which has no
-	// "skip the first confirmation" flag precisely because this consumes the
-	// initial one. Removing it makes every bus announce a hole at startup.
-	// See that function's comment, and TestNoCoverageIsDroppedAtStartup,
-	// which fails if this line goes away.
-	subCtx, subCancel := context.WithTimeout(ctx, 5*time.Second)
-	defer subCancel()
-	if _, err := b.pubsub.Receive(subCtx); err != nil {
-		slog.Warn("watchevents: Redis subscription not confirmed at construction; "+
-			"notifications published before it establishes will be missed by this instance",
+	// THE FAILED PUBSUB IS NOT KEPT (codex round 1 P1). go-redis's own
+	// reconnect after a failed write leaves a fresh connection with no
+	// subscription on it, and this bus's only retry — cycleIfIdle's
+	// resubscribe — fires on `b.pubsub == nil`. Retaining the dead PubSub
+	// and starting a receive loop on it would satisfy every liveness stamp
+	// while receiving nothing, and would leave the retry gate closed for the
+	// life of the process. Closing it and leaving the slot empty is what
+	// makes the next maintenance pass re-establish (phase 2); on phase 1
+	// there is no maintenance pass, which is the pre-existing posture for
+	// every failure this bus cannot see, and now at least a logged one.
+	b.pubsub = client.Subscribe(ctx)
+	if err := b.pubsub.Subscribe(ctx, keys.Name(redisWatchChannelSuffix)); err != nil {
+		slog.Warn("watchevents: Redis refused or dropped the SUBSCRIBE at construction; "+
+			"this instance receives no notifications until the subscription is re-established",
 			"error", err, "channel", keys.Name(redisWatchChannelSuffix))
+		_ = b.pubsub.Close()
+		b.pubsub = nil
+	} else {
+		// WAIT FOR THE SUBSCRIBE TO BE CONFIRMED before returning. go-redis
+		// establishes the subscription asynchronously, so without this the
+		// constructor hands back a bus that is not yet listening — and Redis
+		// pub/sub is at-most-once, so everything published in that window is
+		// lost to this instance, silently.
+		//
+		// Found as a test flake (a second bus constructed and published to
+		// immediately received nothing), which is exactly the shape a rolling
+		// deploy has: a replica comes up, and traffic reaches it before its
+		// subscription is live. The window is small and entirely real.
+		//
+		// A failure here is logged rather than fatal: the receive loop's
+		// ChannelWithSubscriptions() re-subscribes on reconnect, so a bus that
+		// missed its first confirmation still recovers — it just cannot promise
+		// it was listening from the moment it was constructed.
+		//
+		// THIS RECEIVE IS LOAD-BEARING FOR receiveMessages, which has no
+		// "skip the first confirmation" flag precisely because this consumes the
+		// initial one. Removing it makes every bus announce a hole at startup.
+		// See that function's comment, and TestNoCoverageIsDroppedAtStartup,
+		// which fails if this line goes away. Skipped only when the SUBSCRIBE
+		// itself failed above: there is then no initial acknowledgement to
+		// consume, and the first one the loop ever sees — after a
+		// re-establishment — IS a resubscription, so announcing a hole for it
+		// is the truth.
+		subCtx, subCancel := context.WithTimeout(ctx, 5*time.Second)
+		if _, err := b.pubsub.Receive(subCtx); err != nil {
+			slog.Warn("watchevents: Redis subscription not confirmed at construction; "+
+				"notifications published before it establishes will be missed by this instance",
+				"error", err, "channel", keys.Name(redisWatchChannelSuffix))
+		}
+		subCancel()
 	}
 
-	subLoopCtx, subLoopCancel := context.WithCancel(ctx)
-	b.subCancel = subLoopCancel
-	b.lastSeen = b.now()
-	b.lastProbeOK = b.now()
-	b.wg.Add(1)
-	b.subGen++
-	go b.receiveMessages(subLoopCtx, b.pubsub, b.subGen)
+	// No receive loop without a subscription to receive on: resubscribe
+	// starts one when it installs, and reads the current generation rather
+	// than minting its own, so nothing is owed here on the failed path.
+	if b.pubsub != nil {
+		subLoopCtx, subLoopCancel := context.WithCancel(ctx)
+		b.subCancel = subLoopCancel
+		b.lastSeen = b.now()
+		b.lastProbeOK = b.now()
+		b.wg.Add(1)
+		b.subGen++
+		go b.receiveMessages(subLoopCtx, b.pubsub, b.subGen)
+	}
 
 	// NOT STARTED AT ALL ON PHASE 1 (codex round 2). Both halves are gated on
 	// publishHeartbeat and would be guaranteed no-ops there, so the loop would

internal/watchevents/redis_subscribe_failed_test.go

@@ -0,0 +1,105 @@
+package watchevents
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"net"
+	"sync/atomic"
+	"testing"
+
+	"github.com/alicebob/miniredis/v2"
+	"github.com/redis/go-redis/v9"
+)
+
+// subscribeWriteFailer is a Dialer whose connections fail the WRITE of a
+// SUBSCRIBE frame — the failure go-redis's Client.Subscribe discards
+// (BUG-2764). A conn wrapper rather than a proxy close, because a close races
+// the kernel's socket buffer and what then fails is a later read on a
+// different path; the write error is the thing the fix inspects. Twin of the
+// helper in internal/events, kept local: the two packages share no test
+// code.
+type subscribeWriteFailer struct {
+	failed atomic.Int64
+}
+
+var errInjectedSubscribeWrite = errors.New("injected: SUBSCRIBE write failed")
+
+func (f *subscribeWriteFailer) dial(ctx context.Context, network, addr string) (net.Conn, error) {
+	c, err := (&net.Dialer{}).DialContext(ctx, network, addr)
+	if err != nil {
+		return nil, err
+	}
+	return &subscribeFailingConn{Conn: c, f: f}, nil
+}
+
+type subscribeFailingConn struct {
+	net.Conn
+	f *subscribeWriteFailer
+}
+
+func (c *subscribeFailingConn) Write(p []byte) (int, error) {
+	if bytes.Contains(bytes.ToLower(p), []byte("\r\n$9\r\nsubscribe\r\n")) {
+		c.f.failed.Add(1)
+		return 0, errInjectedSubscribeWrite
+	}
+	return c.Conn.Write(p)
+}
+
+// TestAFailedSubscribeIsReportedAsItselfAndAtOnce pins BUG-2764's watchevents
+// half. Before the fix a SUBSCRIBE whose write failed came back as a
+// healthy-looking PubSub; the only sign was the confirmation wait expiring
+// five seconds later as a read timeout — an error that named the wait, not
+// the cause. Both sites are covered by what they LEAVE, not by how long they
+// take: the constructor must leave the slot empty (the unfixed one kept the
+// dead PubSub), and resubscribe must return the injected error itself (the
+// unfixed one returns the confirmation wait's timeout).
+func TestAFailedSubscribeIsReportedAsItselfAndAtOnce(t *testing.T) {
+	mr := miniredis.RunT(t)
+	f := &subscribeWriteFailer{}
+	client := redis.NewClient(&redis.Options{Addr: mr.Addr(), Dialer: f.dial})
+	t.Cleanup(func() { _ = client.Close() })
+	if err := client.Ping(context.Background()).Err(); err != nil {
+		t.Fatalf("ping: %v", err)
+	}
+
+	b := NewRedisBus(client)
+	t.Cleanup(b.Close)
+	if f.failed.Load() == 0 {
+		t.Fatal("the constructor's SUBSCRIBE write was never refused; this test could not have discriminated")
+	}
+	// Nothing was installed: a retained dead PubSub would keep the retry
+	// gate (cycleIfIdle's `b.pubsub == nil`) closed for the life of the
+	// process (codex round 1 P1). This is also what distinguishes the fixed
+	// constructor from the unfixed one, which waited out its 5s confirmation
+	// bound and kept the PubSub — so no timing bound is asserted here: a
+	// wall-clock cap would be a second way for the test to fail under load
+	// that says nothing about the mechanism (codex round 3).
+	if b.currentPubSub() != nil {
+		t.Fatal("the constructor kept a PubSub whose SUBSCRIBE failed; the retry gate can never open")
+	}
+
+	// FABRICATES THE CYCLE'S TEARDOWN STATE ON PURPOSE: resubscribe installs
+	// only into an empty slot, and driving a full idle cycle here would test
+	// the cycle, not the subscribe. The retired PubSub is closed outside the
+	// lock exactly as the cycle does it.
+	b.mu.Lock()
+	old := b.pubsub
+	b.pubsub, b.subCancel = nil, nil
+	b.mu.Unlock()
+	if old != nil {
+		_ = old.Close()
+	}
+
+	before := f.failed.Load()
+	err := b.resubscribe()
+	if f.failed.Load() == before {
+		t.Fatal("resubscribe never issued a SUBSCRIBE; this test could not have discriminated")
+	}
+	if !errors.Is(err, errInjectedSubscribeWrite) {
+		t.Fatalf("resubscribe error = %v, want the injected SUBSCRIBE write error — the failure was reported as something else (the unfixed code returns the confirmation wait's read timeout, five seconds later)", err)
+	}
+	if b.currentPubSub() != nil {
+		t.Fatal("a subscription was installed for a SUBSCRIBE that failed to reach Redis")
+	}
+}