From 129b0c7ef0b39214ea266f5f73f60ae4c0458843 Mon Sep 17 00:00:00 2001 From: Josh Creek <8179928+jcreek@users.noreply.github.com> Date: Sat, 5 Sep 2026 10:29:23 +0100 Subject: [PATCH] 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 | 44 ++++++++++- server/api/outbox.go | 6 +- server/api/service.go | 5 ++ server/cmd/control-plane/main.go | 22 ++++++ server/store/event_fanout.go | 93 +++++++++++++++++++++++ server/store/postgres_integration_test.go | 68 +++++++++++++++++ 6 files changed, 233 insertions(+), 5 deletions(-) create mode 100644 server/store/event_fanout.go diff --git a/server/api/events.go b/server/api/events.go index cc4caa26..3b9b31d3 100644 --- a/server/api/events.go +++ b/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, diff --git a/server/api/outbox.go b/server/api/outbox.go index a797e3f0..71e01b18 100644 --- a/server/api/outbox.go +++ b/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 } } diff --git a/server/api/service.go b/server/api/service.go index cb08ffc1..87beb393 100644 --- a/server/api/service.go +++ b/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 diff --git a/server/cmd/control-plane/main.go b/server/cmd/control-plane/main.go index 9c43c28f..daf32a0d 100644 --- a/server/cmd/control-plane/main.go +++ b/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) diff --git a/server/store/event_fanout.go b/server/store/event_fanout.go new file mode 100644 index 00000000..f95851e2 --- /dev/null +++ b/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)) + } +} diff --git a/server/store/postgres_integration_test.go b/server/store/postgres_integration_test.go index bf42563b..00b8b471 100644 --- a/server/store/postgres_integration_test.go +++ b/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") + } +}