From d15d16d59302003b20e879901b76021a6da9549f Mon Sep 17 00:00:00 2001 From: Josh Creek <8179928+jcreek@users.noreply.github.com> Date: Tue, 1 Sep 2026 16:30:35 +0100 Subject: [PATCH] feat(multiplayer): publish allocation state events --- multiplayer-next.md | 2 + server/api/outbox.go | 56 ++++++++++++++++++++++++++++ server/api/outbox_test.go | 19 ++++++++++ server/cmd/control-plane/main.go | 1 + server/cmd/testkit-api/main.go | 1 + server/store/allocation_match_sql.go | 44 ++++++++++++++++++++-- server/store/outbox.go | 13 +++++++ server/store/outbox_test.go | 1 + 8 files changed, 133 insertions(+), 4 deletions(-) diff --git a/multiplayer-next.md b/multiplayer-next.md index 35939f76..6d9b24a8 100644 --- a/multiplayer-next.md +++ b/multiplayer-next.md @@ -1416,3 +1416,5 @@ The no-show policy now has an explicit domain translation layer (`PlanInitialCon The durable no-show boundary is now implemented by `ApplyInitialConnectPlan`: it locks the match and roster, validates that the plan covers every active participant, records deterministic no-show cooldown penalties, fails no-show tickets, requeues innocent tickets on cancellation or advances connected tickets to `LIVE` for eligible casual bot start, and emits a replayable state-change outbox event under the same serializable transaction. Idempotency keys reject conflicting retries. Focused store tests, race tests, and vet pass; the real PostgreSQL integration remains an environment-dependent gate. The maintenance command now invokes a bounded `ReconcileInitialConnect` sweep for `ASSIGNMENT_READY`/`ASSIGNED`/`CONNECTING` matches, carrying ranked no-show history into the domain ladder and skipping non-actionable WAIT plans. This closes the local control-plane trigger for task 8.35; actual allocated-server bot spawning, shutdown signaling, and live Agones integration remain separate gates. + +Allocation registration now writes a participant-targeted, revisioned `state_changed` outbox event for both `PROCESS_READY` and `ASSIGNMENT_READY` transitions. The production and test API binaries run a type-scoped dispatcher with delivery-before-ack semantics, so allocation lifecycle events survive WebSocket outages without competing with proposal or result consumers. Store/API adversarial tests cover event-type isolation, target validation, and revision mismatches; live allocator/Agones delivery remains an integration gate. diff --git a/server/api/outbox.go b/server/api/outbox.go index 71aacf7e..9f231238 100644 --- a/server/api/outbox.go +++ b/server/api/outbox.go @@ -64,6 +64,32 @@ func RunResultOutboxDispatcher(ctx context.Context, db *sql.DB, service *Service } } +// RunStateOutboxDispatcher delivers committed allocation/no-show lifecycle +// transitions to each participant without acknowledging proposal or result +// events owned by the other dispatchers. +func RunStateOutboxDispatcher(ctx context.Context, db *sql.DB, service *Service) { + if db == nil || service == nil { + return + } + ticker := time.NewTicker(100 * time.Millisecond) + defer ticker.Stop() + dispatcher := store.NewOutboxDispatcher(db, func(deliveryCtx context.Context, event store.OutboxEvent) error { + return deliverStateOutboxEvent(deliveryCtx, event, service) + }) + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + events, err := store.ReadUnpublishedStateOutbox(ctx, db, 100) + if err != nil { + continue + } + _ = dispatchOutboxEvents(ctx, dispatcher, events) + } + } +} + func dispatchOutboxEvents(ctx context.Context, dispatcher *store.OutboxDispatcher, events []store.OutboxEvent) error { if len(events) == 0 { return nil @@ -136,3 +162,33 @@ func deliverResultOutboxEvent(ctx context.Context, db *sql.DB, event store.Outbo } return nil } + +func deliverStateOutboxEvent(_ context.Context, event store.OutboxEvent, service *Service) error { + if event.EventType != "state_changed" || event.AggregateID == "" || event.Revision == 0 || len(event.Payload) == 0 { + return fmt.Errorf("invalid state outbox event") + } + var envelope struct { + Event string `json:"event"` + Revision uint64 `json:"revision"` + ResourceID string `json:"resource_id"` + OccurredAt time.Time `json:"occurred_at"` + State string `json:"state"` + MatchID string `json:"match_id"` + PlayerIDs []string `json:"player_ids"` + } + if err := json.Unmarshal(event.Payload, &envelope); err != nil { + return fmt.Errorf("decode state outbox event: %w", err) + } + if envelope.Event != "state_changed" || envelope.ResourceID != event.AggregateID || envelope.Revision != event.Revision || envelope.State == "" || len(envelope.PlayerIDs) == 0 { + return fmt.Errorf("invalid state outbox payload") + } + for _, playerID := range envelope.PlayerIDs { + 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 { + return err + } + } + return nil +} diff --git a/server/api/outbox_test.go b/server/api/outbox_test.go index f273744b..04311ea8 100644 --- a/server/api/outbox_test.go +++ b/server/api/outbox_test.go @@ -61,3 +61,22 @@ func TestDeliverResultOutboxEventRejectsMalformedRows(t *testing.T) { } } } + +func TestDeliverStateOutboxEventValidatesRevisionAndTargets(t *testing.T) { + service := &Service{} + first := service.getEventHub().subscribe("player-a") + defer service.getEventHub().unsubscribe(first) + payload := []byte(`{"event":"state_changed","revision":4,"resource_id":"match-1","occurred_at":"1970-01-01T00:16:40Z","state":"ASSIGNMENT_READY","match_id":"match-1","player_ids":["player-a"]}`) + if err := deliverStateOutboxEvent(context.Background(), store.OutboxEvent{EventType: "state_changed", AggregateID: "match-1", Revision: 4, Payload: payload}, service); err != nil { + t.Fatalf("valid state event rejected: %v", err) + } + select { + case <-first.queue: + case <-time.After(time.Second): + t.Fatal("participant did not receive state event") + } + bad := []byte(`{"event":"state_changed","revision":3,"resource_id":"match-1","state":"LIVE","player_ids":["player-a"]}`) + if err := deliverStateOutboxEvent(context.Background(), store.OutboxEvent{EventType: "state_changed", AggregateID: "match-1", Revision: 4, Payload: bad}, service); err == nil { + t.Fatal("revision-mismatched state event accepted") + } +} diff --git a/server/cmd/control-plane/main.go b/server/cmd/control-plane/main.go index bc6b6fe9..f39ff868 100644 --- a/server/cmd/control-plane/main.go +++ b/server/cmd/control-plane/main.go @@ -70,6 +70,7 @@ func main() { defer stop() go api.RunProposalOutboxDispatcher(ctx, db, service) go api.RunResultOutboxDispatcher(ctx, db, service) + go api.RunStateOutboxDispatcher(ctx, db, service) select { case err := <-serveErr: if err != nil && err != http.ErrServerClosed { diff --git a/server/cmd/testkit-api/main.go b/server/cmd/testkit-api/main.go index 151df17c..bf777a39 100644 --- a/server/cmd/testkit-api/main.go +++ b/server/cmd/testkit-api/main.go @@ -86,6 +86,7 @@ func main() { defer stop() go api.RunProposalOutboxDispatcher(ctx, db, service) go api.RunResultOutboxDispatcher(ctx, db, service) + go api.RunStateOutboxDispatcher(ctx, db, service) select { case err := <-serveErr: if err != nil && err != http.ErrServerClosed { diff --git a/server/store/allocation_match_sql.go b/server/store/allocation_match_sql.go index 6813e049..f24b931f 100644 --- a/server/store/allocation_match_sql.go +++ b/server/store/allocation_match_sql.go @@ -5,6 +5,7 @@ import ( "context" "crypto/sha256" "database/sql" + "encoding/json" "fmt" "time" @@ -45,7 +46,7 @@ const BindAllocatedMatchParticipantsSQL = `WITH bound AS ( SELECT 1 FROM allocations WHERE allocation_id = $2 AND match_id = $1 AND server_id = $3 AND state = 'ALLOCATED' ) - RETURNING match_id + RETURNING match_id, revision ), participants AS ( SELECT mp.ticket_id, mp.player_id FROM match_participants mp @@ -77,7 +78,13 @@ const AdvanceServerRegistrationSQL = `WITH matched AS ( WHERE q.ticket_id = mp.ticket_id AND q.player_id = mp.player_id AND q.state = $3 RETURNING q.ticket_id ) -SELECT (SELECT count(*) FROM matched), (SELECT count(*) FROM match_participants WHERE match_id = $1), (SELECT count(*) FROM advanced)` +SELECT (SELECT count(*) FROM matched), (SELECT count(*) FROM match_participants WHERE match_id = $1), (SELECT count(*) FROM advanced), COALESCE((SELECT revision FROM matched), -1)` + +const serverRegistrationParticipantIDsSQL = `SELECT player_id FROM match_participants WHERE match_id = $1 ORDER BY player_id` + +const serverRegistrationOutboxSQL = `INSERT INTO outbox + (event_id, aggregate_type, aggregate_id, revision, event_type, payload) +VALUES ($1, 'match', $2, $3, 'state_changed', $4)` const ServerRegistrationIdempotencyScope = "server.register" @@ -120,13 +127,42 @@ func AdvanceServerRegistration(ctx context.Context, db *sql.DB, binding domain.W return nil } var matched, participants, advanced int - if err := tx.QueryRowContext(ctx, AdvanceServerRegistrationSQL, binding.MatchID, binding.ServerID, from, to, binding.AllocationID, now, protocol).Scan(&matched, &participants, &advanced); err != nil { + var revision int64 + if err := tx.QueryRowContext(ctx, AdvanceServerRegistrationSQL, binding.MatchID, binding.ServerID, from, to, binding.AllocationID, now, protocol).Scan(&matched, &participants, &advanced, &revision); err != nil { return err } if matched != 1 || participants == 0 || advanced != participants { return domain.ErrConflict } - return nil + rows, err := tx.QueryContext(ctx, serverRegistrationParticipantIDsSQL, binding.MatchID) + if err != nil { + return err + } + playerIDs := make([]string, 0, participants) + for rows.Next() { + var playerID string + if err := rows.Scan(&playerID); err != nil { + rows.Close() + return err + } + playerIDs = append(playerIDs, playerID) + } + if err := rows.Err(); err != nil { + rows.Close() + return err + } + if err := rows.Close(); err != nil { + return err + } + payload, err := json.Marshal(map[string]any{ + "event": "state_changed", "revision": revision, "resource_id": binding.MatchID, + "occurred_at": now, "state": string(to), "match_id": binding.MatchID, "player_ids": playerIDs, + }) + if err != nil { + return err + } + _, err = tx.ExecContext(ctx, serverRegistrationOutboxSQL, fmt.Sprintf("match:%s:%d", binding.MatchID, revision), binding.MatchID, revision, payload) + return err }) } diff --git a/server/store/outbox.go b/server/store/outbox.go index 16a1fb7e..eff11c1a 100644 --- a/server/store/outbox.go +++ b/server/store/outbox.go @@ -42,6 +42,13 @@ WHERE published_at IS NULL AND event_type = 'match_completed' ORDER BY created_at, event_id LIMIT $1` +const OutboxUnpublishedStateSelectSQL = `SELECT event_id, aggregate_type, aggregate_id, revision, + event_type, payload, created_at, published_at +FROM outbox +WHERE published_at IS NULL AND event_type = 'state_changed' +ORDER BY created_at, event_id +LIMIT $1` + const MatchParticipantIDsSQL = `SELECT player_id FROM match_participants WHERE match_id = $1 @@ -125,6 +132,12 @@ func ReadUnpublishedResultOutbox(ctx context.Context, db *sql.DB, limit int) ([] return readUnpublishedOutbox(ctx, db, limit, OutboxUnpublishedResultSelectSQL) } +// ReadUnpublishedStateOutbox returns lifecycle state events, leaving proposal +// and result rows to their dedicated consumers. +func ReadUnpublishedStateOutbox(ctx context.Context, db *sql.DB, limit int) ([]OutboxEvent, error) { + return readUnpublishedOutbox(ctx, db, limit, OutboxUnpublishedStateSelectSQL) +} + func ReadMatchParticipantIDs(ctx context.Context, db *sql.DB, matchID string) ([]string, error) { if db == nil || matchID == "" { return nil, fmt.Errorf("invalid match participant read arguments") diff --git a/server/store/outbox_test.go b/server/store/outbox_test.go index 56e146ca..1471b57c 100644 --- a/server/store/outbox_test.go +++ b/server/store/outbox_test.go @@ -13,6 +13,7 @@ func TestOutboxSQLPreservesReplayableOrderedReadAndPublishAck(t *testing.T) { OutboxUnpublishedSelectSQL: {"published_at IS NULL", "ORDER BY created_at, event_id", "LIMIT $1"}, OutboxUnpublishedProposalSelectSQL: {"published_at IS NULL", "event_type = 'proposal_changed'", "ORDER BY created_at, event_id", "LIMIT $1"}, OutboxUnpublishedResultSelectSQL: {"published_at IS NULL", "event_type = 'match_completed'", "ORDER BY created_at, event_id", "LIMIT $1"}, + OutboxUnpublishedStateSelectSQL: {"published_at IS NULL", "event_type = 'state_changed'", "ORDER BY created_at, event_id", "LIMIT $1"}, MatchParticipantIDsSQL: {"SELECT player_id", "match_participants", "match_id = $1", "ORDER BY player_id"}, OutboxMarkPublishedSQL: {"published_at = $2", "event_id = $1", "published_at IS NULL"}, } {