fix(server): fan outbox events out to every control-plane replica

The Deployment runs two replicas, but WebSocket subscribers live only in
each process's in-memory hub. Every replica races to read the same
global unpublished outbox rows, and publishing succeeded even when the
winning replica held no matching local subscriber -- that replica then
set the single global published_at. A client connected to the other
replica never received the event, and delivery degraded further with
each replica added. REST recovery eventually converged, but short-lived
proposal transitions could be observed late or not at all.

Publish committed events through PostgreSQL LISTEN/NOTIFY so the replica
that owns the subscriber's connection delivers it, regardless of which
replica drained the row. The listener holds its own pgx connection --
LISTEN is session state, so a pooled database/sql connection cannot
carry it -- and reconnects with backoff, since losing it would silently
downgrade that replica's subscribers to REST-only recovery.

The fan-out is optional: without EventFanout configured, behaviour is
unchanged local-hub publication, which stays correct for a single
replica and for tests. Only outbox-sourced events are routed through it;
the in-request-path publishes remain local, as those are a latency
optimisation for the caller's own connection.

Fan-out needs a wire shape of its own because ControlPlaneEvent hides
PlayerID from clients, and the recipient is exactly what a peer replica
needs to route on.

server/api/events.go

@@ -224,12 +224,52 @@ func (s *Service) getEventHub() *eventHub {
 }
 
 // PublishControlPlaneEvent routes an already-authorized event to the matching
-// authenticated player connection. Durable callers should publish from their
-// outbox after commit; this in-memory hub is deliberately non-authoritative.
+// authenticated player connection on THIS replica. Durable callers should
+// publish from their outbox after commit; this in-memory hub is deliberately
+// non-authoritative.
 func (s *Service) PublishControlPlaneEvent(event ControlPlaneEvent) error {
 	return s.getEventHub().publish(event)
 }
 
+// fannedOutEvent is the fan-out wire shape. It cannot reuse ControlPlaneEvent
+// directly because that type hides PlayerID from clients (json:"-"), and the
+// recipient is precisely what a peer replica needs in order to route.
+type fannedOutEvent struct {
+	ControlPlaneEvent
+	PlayerID string `json:"player_id"`
+}
+
+// EncodeFannedOutEvent and DecodeFannedOutEvent are exported for the
+// control-plane binary, which owns the transport wiring.
+func EncodeFannedOutEvent(event ControlPlaneEvent) ([]byte, error) {
+	return json.Marshal(fannedOutEvent{ControlPlaneEvent: event, PlayerID: event.PlayerID})
+}
+
+func DecodeFannedOutEvent(payload []byte) (ControlPlaneEvent, error) {
+	var decoded fannedOutEvent
+	if err := json.Unmarshal(payload, &decoded); err != nil {
+		return ControlPlaneEvent{}, err
+	}
+	event := decoded.ControlPlaneEvent
+	event.PlayerID = decoded.PlayerID
+	if event.Event == "" || event.ResourceID == "" || event.PlayerID == "" {
+		return ControlPlaneEvent{}, fmt.Errorf("invalid fanned-out control-plane event")
+	}
+	return event, nil
+}
+
+// publishOutboxEvent is how the outbox dispatchers publish. When EventFanout
+// is configured it hands the event to the shared transport so every replica --
+// including whichever one holds the subscriber's WebSocket -- can deliver it.
+// Without it, behaviour is unchanged: local-hub only, correct for a single
+// replica and for tests.
+func (s *Service) publishOutboxEvent(event ControlPlaneEvent) error {
+	if s.EventFanout != nil {
+		return s.EventFanout(event)
+	}
+	return s.PublishControlPlaneEvent(event)
+}
+
 func (s *Service) publishTicketEvent(ticket domain.QueueTicket, now time.Time) {
 	_ = s.PublishControlPlaneEvent(ControlPlaneEvent{
 		Event: "state_changed", Revision: ticket.Revision, ResourceID: ticket.TicketID,

server/api/outbox.go

@@ -155,7 +155,7 @@ func deliverProposalOutboxEvent(_ context.Context, event store.OutboxEvent, serv
 		return fmt.Errorf("invalid proposal outbox event")
 	}
 	for _, playerID := range envelope.PlayerIDs {
-		if err := service.PublishControlPlaneEvent(ControlPlaneEvent{
+		if err := service.publishOutboxEvent(ControlPlaneEvent{
 			Event: envelope.Event, Revision: envelope.Revision, ResourceID: envelope.ResourceID,
 			OccurredAt: envelope.OccurredAt, State: envelope.State, PlayerID: playerID,
 		}); err != nil {
@@ -181,7 +181,7 @@ func deliverResultOutboxEvent(ctx context.Context, db *sql.DB, event store.Outbo
 		return fmt.Errorf("result outbox event has no participants")
 	}
 	for _, playerID := range players {
-		if err := service.PublishControlPlaneEvent(ControlPlaneEvent{
+		if err := service.publishOutboxEvent(ControlPlaneEvent{
 			Event: "state_changed", Revision: event.Revision, ResourceID: event.AggregateID,
 			OccurredAt: event.CreatedAt, State: "COMPLETED", MatchID: event.AggregateID,
 			PlayerID: playerID,
@@ -215,7 +215,7 @@ func deliverStateOutboxEvent(_ context.Context, event store.OutboxEvent, service
 		if playerID == "" {
 			return fmt.Errorf("state outbox event has empty participant")
 		}
-		if err := service.PublishControlPlaneEvent(ControlPlaneEvent{Event: "state_changed", Revision: envelope.Revision, ResourceID: envelope.ResourceID, OccurredAt: envelope.OccurredAt, State: envelope.State, MatchID: envelope.MatchID, PlayerID: playerID}); err != nil {
+		if err := service.publishOutboxEvent(ControlPlaneEvent{Event: "state_changed", Revision: envelope.Revision, ResourceID: envelope.ResourceID, OccurredAt: envelope.OccurredAt, State: envelope.State, MatchID: envelope.MatchID, PlayerID: playerID}); err != nil {
 			return err
 		}
 	}

server/api/service.go

@@ -118,6 +118,11 @@ type Service struct {
 	CandidateV2           CandidateProviderV2
 	QueueBackend          QueueBackend
 	CandidateIndex        CandidateIndex
+	// EventFanout, when set, publishes outbox-sourced events through a shared
+	// transport instead of only this replica's in-memory hub. Without it a
+	// client connected to a replica other than the one that drained the outbox
+	// row never receives the event.
+	EventFanout func(ControlPlaneEvent) error
 	Probe                 ProbeProvider
 	ProbeRecorder         ProbeRecorder
 	WorkloadVerify        WorkloadVerifier

server/cmd/control-plane/main.go

@@ -105,6 +105,28 @@ func main() {
 			}
 		}
 	}()
+	// Fan committed outbox events out to every replica. Subscribers live in
+	// each process's in-memory hub, but any replica may drain a given outbox
+	// row, so without this a client connected elsewhere never sees the event
+	// and delivery degrades as replicas are added.
+	service.EventFanout = func(event api.ControlPlaneEvent) error {
+		payload, err := api.EncodeFannedOutEvent(event)
+		if err != nil {
+			return err
+		}
+		return store.NotifyControlPlaneEvent(ctx, db, payload)
+	}
+	go store.ListenControlPlaneEvents(ctx, *dsn, func(payload []byte) {
+		event, err := api.DecodeFannedOutEvent(payload)
+		if err != nil {
+			return
+		}
+		// Publishing to a player with no local subscriber is a no-op, so every
+		// replica can handle every notification.
+		_ = service.PublishControlPlaneEvent(event)
+	}, func(err error) {
+		fmt.Fprintf(os.Stderr, "control-plane: event fan-out listener: %v\n", err)
+	})
 	go api.RunProposalOutboxDispatcher(ctx, db, service)
 	go api.RunResultOutboxDispatcher(ctx, db, service)
 	go api.RunStateOutboxDispatcher(ctx, db, service)

server/store/event_fanout.go

@@ -0,0 +1,93 @@
+package store
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+)
+
+// ControlPlaneEventChannel is the PostgreSQL LISTEN/NOTIFY channel used to fan
+// committed outbox events out to every control-plane replica.
+//
+// WebSocket subscribers live in each process's in-memory hub, but the outbox
+// is global: every replica raced to read the same unpublished rows, and the
+// winner set the single global published_at even when it held no matching
+// subscriber. A client connected to any other replica then never received the
+// event, and delivery degraded as replicas were added. Notifying through a
+// shared transport means the replica that owns the connection publishes it,
+// regardless of which replica drained the row.
+const ControlPlaneEventChannel = "cosmic_clash_control_plane_events"
+
+// MaxNotifyPayloadBytes is PostgreSQL's hard limit for a NOTIFY payload.
+// Control-plane events are a handful of short fields, so this is a guard
+// against a future field making delivery fail at runtime, not a live concern.
+const MaxNotifyPayloadBytes = 7999
+
+// NotifyControlPlaneEvent broadcasts one already-encoded event to every
+// listening replica. It is called after the event's durable commit, so a lost
+// notification degrades to the REST recovery path rather than losing state.
+func NotifyControlPlaneEvent(ctx context.Context, db *sql.DB, payload []byte) error {
+	if db == nil || len(payload) == 0 {
+		return fmt.Errorf("invalid control-plane event notification")
+	}
+	if len(payload) > MaxNotifyPayloadBytes {
+		return fmt.Errorf("control-plane event payload is %d bytes, over the %d byte NOTIFY limit", len(payload), MaxNotifyPayloadBytes)
+	}
+	_, err := db.ExecContext(ctx, `SELECT pg_notify($1, $2)`, ControlPlaneEventChannel, string(payload))
+	return err
+}
+
+// ListenControlPlaneEvents holds a dedicated connection and delivers every
+// notification to handle until ctx is cancelled. It reconnects on failure:
+// losing the listener would silently downgrade this replica's subscribers to
+// REST-only recovery, which is exactly the degradation being fixed.
+//
+// A dedicated pgx connection is required because LISTEN is session state and
+// database/sql may hand any pooled connection to any caller.
+func ListenControlPlaneEvents(ctx context.Context, dsn string, handle func([]byte), onError func(error)) {
+	if dsn == "" || handle == nil {
+		return
+	}
+	backoff := time.Second
+	for ctx.Err() == nil {
+		err := listenOnce(ctx, dsn, handle)
+		if ctx.Err() != nil {
+			return
+		}
+		if err != nil && onError != nil {
+			onError(err)
+		}
+		select {
+		case <-ctx.Done():
+			return
+		case <-time.After(backoff):
+		}
+		if backoff < 30*time.Second {
+			backoff *= 2
+		}
+	}
+}
+
+func listenOnce(ctx context.Context, dsn string, handle func([]byte)) error {
+	conn, err := pgx.Connect(ctx, dsn)
+	if err != nil {
+		return err
+	}
+	defer conn.Close(context.Background())
+	if _, err := conn.Exec(ctx, `LISTEN `+pgx.Identifier{ControlPlaneEventChannel}.Sanitize()); err != nil {
+		return err
+	}
+	for {
+		notification, err := conn.WaitForNotification(ctx)
+		if err != nil {
+			return err
+		}
+		if notification == nil || notification.Payload == "" {
+			continue
+		}
+		handle([]byte(notification.Payload))
+	}
+}

server/store/postgres_integration_test.go

@@ -2006,3 +2006,71 @@ func TestPostgreSQLBanWithoutRevocationStillBlocksExistingSessions(t *testing.T)
 		t.Fatal("a ban applied without revocation left the existing session usable")
 	}
 }
+
+// The WebSocket hub is per-process, but any control-plane replica may drain a
+// given outbox row, and the winner sets the single global published_at even
+// with no matching local subscriber. Fan-out through LISTEN/NOTIFY is what
+// lets the replica that actually owns the connection deliver the event.
+func TestPostgreSQLControlPlaneEventFanoutReachesEveryReplica(t *testing.T) {
+	db := openIntegrationPostgres(t)
+	applyIntegrationMigrations(t, db)
+
+	dsn := os.Getenv("COSMIC_CLASH_POSTGRES_DSN")
+	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
+	defer cancel()
+
+	// Two listeners stand in for two replicas; neither is the one that will
+	// perform the notify.
+	type replica struct {
+		name     string
+		received chan []byte
+	}
+	replicas := []*replica{
+		{name: "replica-a", received: make(chan []byte, 4)},
+		{name: "replica-b", received: make(chan []byte, 4)},
+	}
+	for _, r := range replicas {
+		target := r
+		go ListenControlPlaneEvents(ctx, dsn, func(payload []byte) {
+			select {
+			case target.received <- payload:
+			default:
+			}
+		}, nil)
+	}
+
+	// LISTEN is asynchronous; retry the notify until both listeners are
+	// attached rather than sleeping an arbitrary amount.
+	payload := []byte(`{"event":"state_changed","resource_id":"match-fanout","player_id":"player-fanout"}`)
+	deadline := time.Now().Add(15 * time.Second)
+	pending := map[string]bool{"replica-a": true, "replica-b": true}
+	for len(pending) > 0 && time.Now().Before(deadline) {
+		if err := NotifyControlPlaneEvent(ctx, db, payload); err != nil {
+			t.Fatalf("notify: %v", err)
+		}
+		for _, r := range replicas {
+			select {
+			case got := <-r.received:
+				if string(got) != string(payload) {
+					t.Fatalf("%s received %q, want %q", r.name, got, payload)
+				}
+				delete(pending, r.name)
+			case <-time.After(200 * time.Millisecond):
+			}
+		}
+	}
+	if len(pending) > 0 {
+		t.Fatalf("replicas never received the fanned-out event: %v", pending)
+	}
+}
+
+func TestNotifyControlPlaneEventRejectsOversizedPayloads(t *testing.T) {
+	db := openIntegrationPostgres(t)
+	oversized := make([]byte, MaxNotifyPayloadBytes+1)
+	for i := range oversized {
+		oversized[i] = 'x'
+	}
+	if err := NotifyControlPlaneEvent(context.Background(), db, oversized); err == nil {
+		t.Fatal("payload over the NOTIFY limit was accepted")
+	}
+}