diff --git a/multiplayer-next.md b/multiplayer-next.md index b38a2a16..bc81c6c9 100644 --- a/multiplayer-next.md +++ b/multiplayer-next.md @@ -1209,7 +1209,7 @@ the local/CI/community transport, not a silent production fallback. | 8.22 `[D:8.21]` | **IN PROGRESS.** Pure Go ranked profile exposes the first ten games as provisional, derives tiers only through validated backend-owned rating bands, and keeps casual ratings outside the API; authenticated HTTP now returns the authoritative ranked view. **"Authoritative ranked view" was durable-adapter-shaped but had no durable adapter**: `rankedProfile`/`profile` only ever read an in-memory map, so every real `GET /v1/profile/ranked` 404'd regardless of a player's actual rating. `RankedProfileProvider` (interface) + `store.PostgresRankedProfiles` close it, preferred over the map when set so existing tests/literals are unaffected; `LastSeasonID`/`SeasonHistory` deliberately left unset (no season pointer on `ratings`, needs its own query/semantics) | `server/domain/rating.go`, `tier_test.go` and `server/api/service.go` cover provisional override, exact band boundaries, malformed policy rejection, session authentication and ranked-only response fields; `server/store/ranked_profile_sql.go`, verified against real PostgreSQL via curl (a fresh identity correctly 404s through the real adapter) and via §8.40's integration test (`control_plane_smoke.gd` now asserts this exact 404 round-trips before queueing); persisted tier policy, client UI and reconnect transport remain | | 8.23 `[D:8.21]` | **IN PROGRESS.** Pure Go ranked-only season policy compresses 25% toward 1500, clamps RD to 200–350, preserves volatility/history, is idempotent by season ID, and defines exact 12-week windows/due detection; migration and Go store now persist a per-player/per-season marker and rating update atomically; `cmd/maintenance` runs bounded due-season batches with signal-bound shutdown | `server/domain/rating.go`, `season_test.go`, `server/migrations/0001_initial.sql` and `server/store/maintenance_sql.go` cover compression, floor/cap, duplicate replay, window boundary, completed-season idempotence, bounded enumeration, row locking and conflict-safe rollover markers; opt-in PostgreSQL execution now covers the durable rating update, marker creation and duplicate replay without a second compression — re-run live for the first time as part of the wider integration-suite verification below, after fixing a test setup gap (a missing `seasons` row tripped the `ranked_season_rollovers` foreign key before the rollover logic itself ran); live maintenance/DB execution remains | | 8.24 `[D:8.9,8.20,8.21]` | **IN PROGRESS.** Pure Go ranked connection policy binds match/server/player/team/slot/protocol, supports 60 s reclaim with server-owned generations, fences old connections, and applies the rolling 7-day 5 m/15 m/1 h/24 h abandon ladder; canonical signed authorisation issuance/verification now gates admission | `server/domain/reconnect.go`, `join_auth.go` and adversarial fixtures cover repeated backend-independent reclaim, all signed claim binding, tampering, failed verification, old-generation fencing, grace boundary and deterministic cooldown audit ordering; persistent lease fencing, join transport and full match/result integration remain | -| 8.25 `[D:8.10,8.24]` | **IN PROGRESS.** Pure Go result policy binds match/server/workload identity, hashes canonical payloads, makes identical retries idempotent, leaves conflicts inert, separates integrity eligibility, classifies roster/simulation/result/fairness evidence, validates annotation signatures/digests, and exposes 5 m alert/30 m review delivery thresholds; Go store SQL now executes receipt → match lock → certified rating updates → completion → receipt acknowledgment → outbox atomically, and exposes bounded ordered outbox reads plus publish acknowledgements for replayable fan-out; `OutboxDispatcher` now delivers in order and acknowledges only after successful fan-out; the API validates workload-bound server result submissions and the PostgreSQL adapter repeats domain validation before invoking this durable boundary | `server/domain/result.go`, `workload.go`, `server/workload/jwt.go`, `server/api/service.go`, `server/store/result_sql.go` and `outbox.go` plus adversarial fixtures cover credential binding, duplicate/conflict, annotation forgery, server/match mismatch, invalid direct-adapter payloads, delivery-outage-versus-integrity classification, commit ordering, idempotent SQL reconciliation, ordered rating locks, certified-update gating, unpublished-event replay/ack boundaries and delivery-before-ack failure ordering; opt-in PostgreSQL execution now covers result-pending completion, durable receipt/outbox publication, ack removal, identical replay and conflicting replay rejection, and a real concurrent-goroutine identical-submission race confirming exactly-once rating application; production credential verification, Agones annotation persistence/reconciliation and integrity evidence adapters remain | +| 8.25 `[D:8.10,8.24]` | **IN PROGRESS.** Pure Go result policy binds match/server/workload identity, hashes canonical payloads, makes identical retries idempotent, leaves conflicts inert, separates integrity eligibility, classifies roster/simulation/result/fairness evidence, validates annotation signatures/digests, and exposes 5 m alert/30 m review delivery thresholds; Go store SQL now executes receipt → match lock → certified rating updates → completion → receipt acknowledgment → outbox atomically, and exposes bounded ordered outbox reads plus publish acknowledgements for replayable fan-out; `OutboxDispatcher` now delivers in order and acknowledges only after successful fan-out; the API validates workload-bound server result submissions and the PostgreSQL adapter repeats domain validation before invoking this durable boundary; production now also runs a filtered `match_completed` dispatcher that turns each committed result into targeted `COMPLETED` state events for every durable participant, without acknowledging proposal rows | `server/domain/result.go`, `workload.go`, `server/workload/jwt.go`, `server/api/service.go`, `server/api/outbox.go`, `server/store/result_sql.go` and `outbox.go` plus adversarial fixtures cover credential binding, duplicate/conflict, annotation forgery, server/match mismatch, invalid direct-adapter payloads, delivery-outage-versus-integrity classification, commit ordering, idempotent SQL reconciliation, ordered rating locks, certified-update gating, unpublished-event replay/ack boundaries, event-type isolation and delivery-before-ack failure ordering; opt-in PostgreSQL execution now covers result-pending completion, durable receipt/outbox publication, ack removal, identical replay and conflicting replay rejection, and a real concurrent-goroutine identical-submission race confirming exactly-once rating application; live result WebSocket fan-out, production credential verification, Agones annotation persistence/reconciliation and integrity evidence adapters remain | #### 8D — Agones, allocation and regional scaling diff --git a/server/api/outbox.go b/server/api/outbox.go index 357e02fc..71aacf7e 100644 --- a/server/api/outbox.go +++ b/server/api/outbox.go @@ -38,6 +38,32 @@ func RunProposalOutboxDispatcher(ctx context.Context, db *sql.DB, service *Servi } } +// RunResultOutboxDispatcher delivers committed match results as targeted +// COMPLETED state events. It owns only match_completed rows; proposal rows +// remain with RunProposalOutboxDispatcher. +func RunResultOutboxDispatcher(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 deliverResultOutboxEvent(deliveryCtx, db, event, service) + }) + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + events, err := store.ReadUnpublishedResultOutbox(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 @@ -83,3 +109,30 @@ func deliverProposalOutboxEvent(_ context.Context, event store.OutboxEvent, serv } return nil } + +func deliverResultOutboxEvent(ctx context.Context, db *sql.DB, event store.OutboxEvent, service *Service) error { + if event.EventType != "match_completed" || event.AggregateID == "" || event.Revision == 0 || len(event.Payload) == 0 { + return fmt.Errorf("invalid result outbox event") + } + var payload map[string]any + if err := json.Unmarshal(event.Payload, &payload); err != nil || payload == nil { + return fmt.Errorf("decode result outbox event: %w", err) + } + players, err := store.ReadMatchParticipantIDs(ctx, db, event.AggregateID) + if err != nil { + return err + } + if len(players) == 0 { + return fmt.Errorf("result outbox event has no participants") + } + for _, playerID := range players { + if err := service.PublishControlPlaneEvent(ControlPlaneEvent{ + Event: "state_changed", Revision: event.Revision, ResourceID: event.AggregateID, + OccurredAt: event.CreatedAt, State: "COMPLETED", MatchID: event.AggregateID, + PlayerID: playerID, + }); err != nil { + return err + } + } + return nil +} diff --git a/server/api/outbox_test.go b/server/api/outbox_test.go index d9b03a4d..f273744b 100644 --- a/server/api/outbox_test.go +++ b/server/api/outbox_test.go @@ -49,3 +49,15 @@ func TestDeliverProposalOutboxEventRejectsMalformedOrUntargetedRows(t *testing.T }) } } + +func TestDeliverResultOutboxEventRejectsMalformedRows(t *testing.T) { + for _, event := range []store.OutboxEvent{ + {EventType: "proposal_changed", AggregateID: "match-1", Revision: 1, Payload: []byte(`{}`)}, + {EventType: "match_completed", AggregateID: "", Revision: 1, Payload: []byte(`{}`)}, + {EventType: "match_completed", AggregateID: "match-1", Revision: 1, Payload: []byte(`not-json`)}, + } { + if err := deliverResultOutboxEvent(nil, nil, event, &Service{}); err == nil { + t.Fatalf("invalid result event accepted: %+v", event) + } + } +} diff --git a/server/cmd/control-plane/main.go b/server/cmd/control-plane/main.go index f23a47a5..194ef1cb 100644 --- a/server/cmd/control-plane/main.go +++ b/server/cmd/control-plane/main.go @@ -69,6 +69,7 @@ func main() { ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() go api.RunProposalOutboxDispatcher(ctx, db, service) + go api.RunResultOutboxDispatcher(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 cb285725..651f5977 100644 --- a/server/cmd/testkit-api/main.go +++ b/server/cmd/testkit-api/main.go @@ -82,6 +82,7 @@ func main() { ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() go api.RunProposalOutboxDispatcher(ctx, db, service) + go api.RunResultOutboxDispatcher(ctx, db, service) select { case err := <-serveErr: if err != nil && err != http.ErrServerClosed { diff --git a/server/store/outbox.go b/server/store/outbox.go index 9dd208c9..16a1fb7e 100644 --- a/server/store/outbox.go +++ b/server/store/outbox.go @@ -35,6 +35,18 @@ WHERE published_at IS NULL AND event_type = 'proposal_changed' ORDER BY created_at, event_id LIMIT $1` +const OutboxUnpublishedResultSelectSQL = `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 = 'match_completed' +ORDER BY created_at, event_id +LIMIT $1` + +const MatchParticipantIDsSQL = `SELECT player_id +FROM match_participants +WHERE match_id = $1 +ORDER BY player_id` + const OutboxMarkPublishedSQL = `UPDATE outbox SET published_at = $2 WHERE event_id = $1 AND published_at IS NULL` @@ -106,6 +118,36 @@ func ReadUnpublishedProposalOutbox(ctx context.Context, db *sql.DB, limit int) ( return readUnpublishedOutbox(ctx, db, limit, OutboxUnpublishedProposalSelectSQL) } +// ReadUnpublishedResultOutbox returns only durable match-completion events. +// Proposal and result consumers acknowledge separate event types so one +// transient fan-out outage cannot hide rows owned by another consumer. +func ReadUnpublishedResultOutbox(ctx context.Context, db *sql.DB, limit int) ([]OutboxEvent, error) { + return readUnpublishedOutbox(ctx, db, limit, OutboxUnpublishedResultSelectSQL) +} + +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") + } + rows, err := db.QueryContext(ctx, MatchParticipantIDsSQL, matchID) + if err != nil { + return nil, err + } + defer rows.Close() + var players []string + for rows.Next() { + var playerID string + if err := rows.Scan(&playerID); err != nil { + return nil, err + } + players = append(players, playerID) + } + if err := rows.Err(); err != nil { + return nil, err + } + return players, nil +} + func readUnpublishedOutbox(ctx context.Context, db *sql.DB, limit int, query string) ([]OutboxEvent, error) { if db == nil || limit < 1 || limit > 1000 { return nil, fmt.Errorf("invalid outbox read arguments") diff --git a/server/store/outbox_test.go b/server/store/outbox_test.go index 8e6a6135..56e146ca 100644 --- a/server/store/outbox_test.go +++ b/server/store/outbox_test.go @@ -12,6 +12,8 @@ func TestOutboxSQLPreservesReplayableOrderedReadAndPublishAck(t *testing.T) { for query, fragments := range map[string][]string{ 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"}, + MatchParticipantIDsSQL: {"SELECT player_id", "match_participants", "match_id = $1", "ORDER BY player_id"}, OutboxMarkPublishedSQL: {"published_at = $2", "event_id = $1", "published_at IS NULL"}, } { for _, fragment := range fragments { @@ -67,4 +69,7 @@ func TestOutboxAdaptersRejectUnsafeArgumentsWithoutDatabase(t *testing.T) { if err := MarkOutboxPublished(nil, nil, "", time.Unix(1000, 0)); err == nil { t.Fatal("empty event acknowledgement accepted") } + if _, err := ReadMatchParticipantIDs(nil, nil, ""); err == nil { + t.Fatal("invalid participant read arguments accepted") + } }