feat: promote accepted proposals from API

This commit is contained in:
Josh Creek
2026-09-01 10:29:41 +01:00
parent e0ba6c6ead
commit 03ff8e485e
16 changed files with 270 additions and 49 deletions
+2 -1
View File
@@ -140,7 +140,8 @@ product policy are in [`docs/MATCHMAKING.md`](docs/MATCHMAKING.md).
adapter now performs proposal insertion, participant insertion, and every adapter now performs proposal insertion, participant insertion, and every
queue-ticket promotion in one rollback-safe SERIALIZABLE callback; accepted queue-ticket promotion in one rollback-safe SERIALIZABLE callback; accepted
proposals now atomically promote their exact team/slot map and tickets into an proposals now atomically promote their exact team/slot map and tickets into an
`ALLOCATING` match; the runnable `ALLOCATING` match; final unanimous proposal acceptance now invokes this
replay-safe promotion through the API; the runnable
casual matcher can optionally use a Redis candidate projection and repairs an casual matcher can optionally use a Redis candidate projection and repairs an
empty/lost index from PostgreSQL before claiming durably; live DB/Redis failover empty/lost index from PostgreSQL before claiming durably; live DB/Redis failover
testing remains. testing remains.
+2 -2
View File
@@ -1173,7 +1173,7 @@ the local/CI/community transport, not a silent production fallback.
| 8.2 `[D:8.1]` | **DONE.** Encode the launch SLOs from `docs/MATCHMAKING.md`: RTT, allocation/connect latency, 99.9% allocation/result success, API latency and tick health | [`docs/MATCHMAKING-SLOs.md`](docs/MATCHMAKING-SLOs.md) defines each metric, denominator, percentile/window, owner, alert threshold and release evidence | | 8.2 `[D:8.1]` | **DONE.** Encode the launch SLOs from `docs/MATCHMAKING.md`: RTT, allocation/connect latency, 99.9% allocation/result success, API latency and tick health | [`docs/MATCHMAKING-SLOs.md`](docs/MATCHMAKING-SLOs.md) defines each metric, denominator, percentile/window, owner, alert threshold and release evidence |
| 8.3 `[D:8.1]` | **DONE.** Publish versioned OpenAPI + WebSocket contracts for Steam login/session, profile/rating, queue create/heartbeat/cancel/resume, proposal accept/decline, assignment/status, server registration/roster/result/shutdown | [`server/contracts/v1/`](server/contracts/v1/) contains machine-readable REST/events contracts and dependency-free structural tests; REST resync is specified by the contract; `server/api/service.go` also exposes the documented `/api/v1` route names (including server-assigned idempotent queue ticket IDs and DELETE cancellation) alongside the existing client `/v1` routes, covered by `TestDocumentedContractRoutesAdaptToServiceAPI` | | 8.3 `[D:8.1]` | **DONE.** Publish versioned OpenAPI + WebSocket contracts for Steam login/session, profile/rating, queue create/heartbeat/cancel/resume, proposal accept/decline, assignment/status, server registration/roster/result/shutdown | [`server/contracts/v1/`](server/contracts/v1/) contains machine-readable REST/events contracts and dependency-free structural tests; REST resync is specified by the contract; `server/api/service.go` also exposes the documented `/api/v1` route names (including server-assigned idempotent queue ticket IDs and DELETE cancellation) alongside the existing client `/v1` routes, covered by `TestDocumentedContractRoutesAdaptToServiceAPI` |
| 8.4 `[D:8.3]` | **DONE.** Define opaque IDs, legal queue/match state transitions, revisions and idempotency keys | [`server/contracts/v1/state-transitions.json`](server/contracts/v1/state-transitions.json) locks terminal states, legal edges, stale-revision handling and same-key replay/conflict behavior; contract tests cover the invariants | | 8.4 `[D:8.3]` | **DONE.** Define opaque IDs, legal queue/match state transitions, revisions and idempotency keys | [`server/contracts/v1/state-transitions.json`](server/contracts/v1/state-transitions.json) locks terminal states, legal edges, stale-revision handling and same-key replay/conflict behavior; contract tests cover the invariants |
| 8.5 `[D:8.4]` | **IN PROGRESS.** Initial PostgreSQL migration now defines durable idempotency keys, queue ownership/active-participation fencing, identities, sessions/revocations, ranked seasons, ratings/events, matches/participants, penalties, results, audits and outbox; follow-up migrations persist server-derived queue probe RTT metadata and the allocator GameServer/allocation registry | `server/migrations/0001_initial.sql`, `0003_queue_probe_metadata.sql`, `0004_allocator_registry.sql`, `migrations/runner.go`, `cmd/migrate` and static checks cover the durable tables, uniqueness/check constraints, Redis-as-cache boundary and serialized forward migration recording; opt-in `scripts/run_postgres_integration.sh` now runs the runner and real queue/assignment ownership, idempotency, revision and expiry checks through pgx; rollback/down migration, the remaining serializable adapters and cache-loss repair remain | | 8.5 `[D:8.4]` | **IN PROGRESS.** Initial PostgreSQL migration now defines durable idempotency keys, queue ownership/active-participation fencing, identities, sessions/revocations, ranked seasons, ratings/events, matches/participants, penalties, results, audits and outbox; follow-up migrations persist server-derived queue probe RTT metadata, the allocator GameServer/allocation registry, and matcher-selected proposal region/protocol/team/slot plans | `server/migrations/0001_initial.sql`, `0003_queue_probe_metadata.sql`, `0004_allocator_registry.sql`, `0005_proposal_match_plans.sql`, `migrations/runner.go`, `cmd/migrate` and static checks cover the durable tables, uniqueness/check constraints, Redis-as-cache boundary and serialized forward migration recording; opt-in `scripts/run_postgres_integration.sh` now runs the runner and real queue/assignment ownership, idempotency, revision and expiry checks through pgx; rollback/down migration, remaining serializable adapters and cache-loss repair remain |
| 8.6 `[D:8.3,8.4]` | **IN PROGRESS.** Add allocated-mode `ServerConfig` compatibility fields as opt-in defaults | `ServerConfig` now validates allocation mode, match/server IDs, playlist version, client build, future assignment expiry, image digest, transport and EU/NA region; `server_boot.gd` fails closed for the not-yet-wired Steam SDR transport, constrains allocated processes to one match, and emits allocation identity/transport in `server_started`; signed-authorisation admission, dynamic endpoint wiring and full manifest/runtime tests remain | | 8.6 `[D:8.3,8.4]` | **IN PROGRESS.** Add allocated-mode `ServerConfig` compatibility fields as opt-in defaults | `ServerConfig` now validates allocation mode, match/server IDs, playlist version, client build, future assignment expiry, image digest, transport and EU/NA region; `server_boot.gd` fails closed for the not-yet-wired Steam SDR transport, constrains allocated processes to one match, and emits allocation identity/transport in `server_started`; signed-authorisation admission, dynamic endpoint wiring and full manifest/runtime tests remain |
#### 8B — Authentication and secure control plane #### 8B — Authentication and secure control plane
@@ -1196,7 +1196,7 @@ the local/CI/community transport, not a silent production fallback.
| 8.15 `[D:7.8,8.3]` | **IN PROGRESS.** Pure Go probe validation treats Steam location as opaque, requires nonce/freshness/region and server-computed RTT, and implements discrepancy quarantine/release; authenticated HTTP now accepts only opaque location/nonce input through a server-owned probe provider and records validated RTT into the active player's durable queue ticket; durable queue projections have a server-derived RTT JSON field for matcher reads | `server/domain/probes.go`, `server/migrations/0003_queue_probe_metadata.sql`, adversarial fixtures and `server/api/service.go`/`store/queue_sql.go` cover stale/wrong/forged evidence, the 25 ms/30% threshold, three-sample quarantine, five-clean release, authenticated provider arguments, rejection of client RTT fields, player/ticket/expiry binding, persistence failure, playlist-scoped candidate reads and bounded metadata decoding; Steam coordinator, regional probe adapters and multi-region probe population remain | | 8.15 `[D:7.8,8.3]` | **IN PROGRESS.** Pure Go probe validation treats Steam location as opaque, requires nonce/freshness/region and server-computed RTT, and implements discrepancy quarantine/release; authenticated HTTP now accepts only opaque location/nonce input through a server-owned probe provider and records validated RTT into the active player's durable queue ticket; durable queue projections have a server-derived RTT JSON field for matcher reads | `server/domain/probes.go`, `server/migrations/0003_queue_probe_metadata.sql`, adversarial fixtures and `server/api/service.go`/`store/queue_sql.go` cover stale/wrong/forged evidence, the 25 ms/30% threshold, three-sample quarantine, five-clean release, authenticated provider arguments, rejection of client RTT fields, player/ticket/expiry binding, persistence failure, playlist-scoped candidate reads and bounded metadata decoding; Steam coordinator, regional probe adapters and multi-region probe population remain |
| 8.16 `[D:8.14,8.15]` | **IN PROGRESS.** Pure Go candidate/team selection implements the <=100 ms ceiling, pairwise widening tolerance, anchor inclusion, deterministic set/region scoring and balanced team partitioning; queue-backed formation now consumes the server-owned projection, fences duplicate player identities and rejects playlist/build/protocol mixing; the matcher worker performs bounded formation and delegates the final claim to the durable proposal transaction; the runnable matcher now supports explicitly enabled ranked six-player polling with durable verified-Steam identity lookup | `server/domain/matcher.go`, `teams.go`, `server/matcher/worker.go`, `server/store/queue_sql.go` and adversarial fixtures cover no-common-region, tolerance boundaries, lexical ties, mean-rating balance, malformed candidates, duplicate identities, compatibility mismatches, incomplete batches, source failure, durable claim failure, queue-backed oldest-anchor formation and incomplete ranked identity metadata; arena selection and long-running worker integration remain | | 8.16 `[D:8.14,8.15]` | **IN PROGRESS.** Pure Go candidate/team selection implements the <=100 ms ceiling, pairwise widening tolerance, anchor inclusion, deterministic set/region scoring and balanced team partitioning; queue-backed formation now consumes the server-owned projection, fences duplicate player identities and rejects playlist/build/protocol mixing; the matcher worker performs bounded formation and delegates the final claim to the durable proposal transaction; the runnable matcher now supports explicitly enabled ranked six-player polling with durable verified-Steam identity lookup | `server/domain/matcher.go`, `teams.go`, `server/matcher/worker.go`, `server/store/queue_sql.go` and adversarial fixtures cover no-common-region, tolerance boundaries, lexical ties, mean-rating balance, malformed candidates, duplicate identities, compatibility mismatches, incomplete batches, source failure, durable claim failure, queue-backed oldest-anchor formation and incomplete ranked identity metadata; arena selection and long-running worker integration remain |
| 8.17 `[D:8.14,8.16]` | **IN PROGRESS.** Pure Go proposal policy sends a 10-second response window to every selected human, requires unanimous acceptance, applies exact decline/timeout cooldowns and ranked escalation; authenticated API exposes revisioned accept/decline mutations; formed matches now pass through a playlist-aware proposal boundary | `server/domain/proposal.go`, `formation.go` and `server/api/service.go` plus adversarial fixtures cover partial/unanimous response, expiry, replay/conflict, stale API revision, casual lineup preparation and ranked metadata validation; queue precedence and allocation integration remain | | 8.17 `[D:8.14,8.16]` | **IN PROGRESS.** Pure Go proposal policy sends a 10-second response window to every selected human, requires unanimous acceptance, applies exact decline/timeout cooldowns and ranked escalation; authenticated API exposes revisioned accept/decline mutations; formed matches now pass through a playlist-aware proposal boundary | `server/domain/proposal.go`, `formation.go` and `server/api/service.go` plus adversarial fixtures cover partial/unanimous response, expiry, replay/conflict, stale API revision, casual lineup preparation and ranked metadata validation; queue precedence and allocation integration remain |
| 8.18 `[D:8.5,8.14,8.17]` | **IN PROGRESS.** Go store layer defines PostgreSQL SERIALIZABLE whole-transaction retries and queue candidate/proposal claim SQL using `FOR UPDATE SKIP LOCKED` plus durable uniqueness/revision fences; proposal creation now inserts proposal/participants and promotes every ticket in one rollback-safe transaction with player- and playlist-bound claim predicates, queue creation has a durable idempotency/owner-read adapter, participant-scoped proposal recovery now expires OPEN proposals and pending participants transactionally at read time, and proposal accept/decline now uses participant/proposal locks, revision fencing and durable idempotency; an accepted proposal can now atomically create the exact `ALLOCATING` match/team/slot topology and promote all claimed tickets to `ACCEPTED`; runnable matcher polling supports an optional Redis candidate projection that repairs empty/lost cache state from authoritative PostgreSQL before the durable final claim | `server/store/serializable.go`, `queue_sql.go`, `proposal_sql.go`, `proposal_recovery_sql.go`, `match_sql.go`, `redis_candidates.go`, `server/matcher/worker.go` and tests cover retry classification, claim-boundary invariants, player/ticket/playlist mapping, durable queue replay/conflict, owner-scoped queue/proposal recovery, expiry at read and mutation boundaries, response replay/conflict, exact match-promotion replay/conflict, fixed team/slot persistence, zero-row claim aborts, atomic statement ordering, incomplete matcher batches, source failures and empty-index Redis repair; opt-in PostgreSQL execution now covers queue create/replay/fencing, assignment persistence, proposal claim/promotion, accepted-proposal match promotion, participant recovery, unanimous response and rollback of partial claims; worker invocation, allocation runtime and concurrent two-matcher integration tests remain | | 8.18 `[D:8.5,8.14,8.17]` | **IN PROGRESS.** Go store layer defines PostgreSQL SERIALIZABLE whole-transaction retries and queue candidate/proposal claim SQL using `FOR UPDATE SKIP LOCKED` plus durable uniqueness/revision fences; proposal creation now persists its matcher-selected region/protocol/team/slot plan alongside proposal/participant claims, queue creation has a durable idempotency/owner-read adapter, participant-scoped proposal recovery now expires OPEN proposals and pending participants transactionally at read time, and proposal accept/decline now uses participant/proposal locks, revision fencing and durable idempotency; final unanimous API acceptance invokes replay-safe promotion of the exact stored `ALLOCATING` match/team/slot topology and claimed tickets to `ACCEPTED`; runnable matcher polling supports an optional Redis candidate projection that repairs empty/lost cache state from authoritative PostgreSQL before the durable final claim | `server/store/serializable.go`, `queue_sql.go`, `proposal_sql.go`, `proposal_recovery_sql.go`, `match_sql.go`, `redis_candidates.go`, `server/matcher/worker.go`, `server/api/service.go` and tests cover retry classification, claim-boundary invariants, player/ticket/playlist mapping, durable queue replay/conflict, owner-scoped queue/proposal recovery, expiry at read and mutation boundaries, final-response promotion retry, exact match-promotion replay/conflict, fixed team/slot persistence, zero-row claim aborts, atomic statement ordering, incomplete matcher batches, source failures and empty-index Redis repair; opt-in PostgreSQL execution now covers queue create/replay/fencing, assignment persistence, proposal claim/promotion, accepted-proposal match promotion, participant recovery, unanimous response and rollback of partial claims; allocation runtime and concurrent two-matcher integration tests remain |
| 8.19 `[D:8.18]` | **IN PROGRESS.** Pure Go casual lineup requires 26 humans with at least one per team, fills missing slots with explicit bots, permits kickoff-only bot-slot backfill and assigns no backfill penalty/rating update; proposal preparation now derives the lineup from formed teams | `server/domain/casual.go`, `formation.go` cover both-team minimum, bot shape, live-play rejection, zero-penalty backfill and casual proposal composition; queue candidate selection, opt-in 10 s backfill proposals, reconnect/leave penalties and live integration remain | | 8.19 `[D:8.18]` | **IN PROGRESS.** Pure Go casual lineup requires 26 humans with at least one per team, fills missing slots with explicit bots, permits kickoff-only bot-slot backfill and assigns no backfill penalty/rating update; proposal preparation now derives the lineup from formed teams | `server/domain/casual.go`, `formation.go` cover both-team minimum, bot shape, live-play rejection, zero-penalty backfill and casual proposal composition; queue candidate selection, opt-in 10 s backfill proposals, reconnect/leave penalties and live integration remain |
| 8.20 `[D:8.18]` | **IN PROGRESS.** Pure Go ranked admission requires six unique verified solo humans, rejects bots/backfill/parties, and allows only random-enabled non-elevated arenas; proposal preparation requires matching metadata for every formed player | `server/domain/ranked.go`, `formation.go` cover count, identity, party, bot/backfill, arena eligibility and formed-player metadata rejection; `ArenaRegistry` integration, allocation wiring and innocent-ticket restoration remain | | 8.20 `[D:8.18]` | **IN PROGRESS.** Pure Go ranked admission requires six unique verified solo humans, rejects bots/backfill/parties, and allows only random-enabled non-elevated arenas; proposal preparation requires matching metadata for every formed player | `server/domain/ranked.go`, `formation.go` cover count, identity, party, bot/backfill, arena eligibility and formed-player metadata rejection; `ArenaRegistry` integration, allocation wiring and innocent-ticket restoration remain |
| 8.21 `[D:8.5,8.20]` | **IN PROGRESS.** Pure Go rating core implements canonical Glicko-2, daily inactivity, ranked 1/3 and casual 1/N human-opponent weights, deterministic opponent ordering, and authoritative draw/overtime/abandon scoring; certified result completion now applies per-player updates inside the durable transaction with lexical row locks and revision increments | `server/domain/rating.go`, `server/store/result_sql.go` and tests cover canonical/inactivity/weight/invalid-input, draw/OT/abandon, ordered participant snapshots, lock/value re-read and rating update SQL; live PostgreSQL rating, seasons and concurrent result transaction tests remain | | 8.21 `[D:8.5,8.20]` | **IN PROGRESS.** Pure Go rating core implements canonical Glicko-2, daily inactivity, ranked 1/3 and casual 1/N human-opponent weights, deterministic opponent ordering, and authoritative draw/overtime/abandon scoring; certified result completion now applies per-player updates inside the durable transaction with lexical row locks and revision increments | `server/domain/rating.go`, `server/store/result_sql.go` and tests cover canonical/inactivity/weight/invalid-input, draw/OT/abandon, ordered participant snapshots, lock/value re-read and rating update SQL; live PostgreSQL rating, seasons and concurrent result transaction tests remain |
+38 -23
View File
@@ -63,6 +63,14 @@ type ProposalBackend interface {
type ProposalMutationBackend interface { type ProposalMutationBackend interface {
Respond(context.Context, string, string, string, bool, uint64, time.Time) (domain.Proposal, error) Respond(context.Context, string, string, string, bool, uint64, time.Time) (domain.Proposal, error)
} }
type ProposalPromoter interface {
Promote(context.Context, domain.Proposal, time.Time) error
}
type ProposalPromoterFunc func(context.Context, domain.Proposal, time.Time) error
func (f ProposalPromoterFunc) Promote(ctx context.Context, proposal domain.Proposal, now time.Time) error {
return f(ctx, proposal, now)
}
type AssignmentView struct { type AssignmentView struct {
MatchID string `json:"match_id"` MatchID string `json:"match_id"`
@@ -83,29 +91,30 @@ type AssignmentView struct {
type AssignmentProvider func(context.Context, string, string, time.Time) (AssignmentView, error) type AssignmentProvider func(context.Context, string, string, time.Time) (AssignmentView, error)
type Service struct { type Service struct {
Sessions *domain.SessionStore Sessions *domain.SessionStore
SessionBackend SessionBackend SessionBackend SessionBackend
SessionIssuer SessionIssuer SessionIssuer SessionIssuer
SteamLogin SteamLoginProvider SteamLogin SteamLoginProvider
Queue *domain.Queue Queue *domain.Queue
Candidate CandidateProvider Candidate CandidateProvider
CandidateV2 CandidateProviderV2 CandidateV2 CandidateProviderV2
QueueBackend QueueBackend QueueBackend QueueBackend
CandidateIndex CandidateIndex CandidateIndex CandidateIndex
Probe ProbeProvider Probe ProbeProvider
ProbeRecorder ProbeRecorder ProbeRecorder ProbeRecorder
WorkloadVerify WorkloadVerifier WorkloadVerify WorkloadVerifier
ResultSubmitter ResultSubmitter ResultSubmitter ResultSubmitter
Assignment AssignmentProvider Assignment AssignmentProvider
Now func() time.Time Now func() time.Time
Proposals map[string]*domain.Proposal Proposals map[string]*domain.Proposal
ProposalBackend ProposalBackend ProposalBackend ProposalBackend
RankedProfiles map[string]domain.RankedProfile ProposalPromoter ProposalPromoter
TierPolicy domain.TierPolicy RankedProfiles map[string]domain.RankedProfile
RateLimiter *RateLimiter TierPolicy domain.TierPolicy
proposalMu sync.Mutex RateLimiter *RateLimiter
eventsMu sync.Mutex proposalMu sync.Mutex
events *eventHub eventsMu sync.Mutex
events *eventHub
} }
func (s *Service) Handler() http.Handler { func (s *Service) Handler() http.Handler {
@@ -598,6 +607,12 @@ func (s *Service) proposalMutation(w http.ResponseWriter, r *http.Request) {
writeDomainError(w, err) writeDomainError(w, err)
return return
} }
if updated.State == domain.Accepted && s.ProposalPromoter != nil {
if err := s.ProposalPromoter.Promote(r.Context(), updated, now); err != nil {
writeError(w, http.StatusServiceUnavailable, "match_promotion_unavailable")
return
}
}
s.publishProposalEvent(updated, now) s.publishProposalEvent(updated, now)
writeJSON(w, http.StatusOK, toProposalResponse(updated)) writeJSON(w, http.StatusOK, toProposalResponse(updated))
} }
+55
View File
@@ -41,6 +41,18 @@ type resultSubmitterSpy struct {
result domain.MatchResult result domain.MatchResult
} }
type proposalPromoterSpy struct {
calls int
proposal domain.Proposal
err error
}
func (p *proposalPromoterSpy) Promote(_ context.Context, proposal domain.Proposal, _ time.Time) error {
p.calls++
p.proposal = proposal
return p.err
}
func (r *resultSubmitterSpy) SubmitResult(_ context.Context, key string, result domain.MatchResult, _ domain.WorkloadBinding, _ []byte, _ time.Time) error { func (r *resultSubmitterSpy) SubmitResult(_ context.Context, key string, result domain.MatchResult, _ domain.WorkloadBinding, _ []byte, _ time.Time) error {
r.calls++ r.calls++
r.key, r.result = key, result r.key, r.result = key, result
@@ -409,6 +421,49 @@ func TestStateChangingAPIActionsPublishTargetedEvents(t *testing.T) {
} }
} }
func TestFinalProposalAcceptancePromotesDurableMatchAndFailsRetryably(t *testing.T) {
now := time.Unix(1000, 0).UTC()
proposal, err := domain.NewProposal("proposal-promote-123456", domain.Casual, []string{"player-1", "player-2"}, now)
if err != nil {
t.Fatal(err)
}
backend := &proposalBackendSpy{proposal: proposal}
promoter := &proposalPromoterSpy{}
sessions := domain.NewSessionStore()
session1, token1, err := sessions.Issue("player-1", time.Hour, now)
if err != nil {
t.Fatal(err)
}
session2, token2, err := sessions.Issue("player-2", time.Hour, now)
if err != nil {
t.Fatal(err)
}
service := &Service{Sessions: sessions, ProposalBackend: backend, ProposalPromoter: promoter, Now: func() time.Time { return now }}
respond := func(credential, key, revision string) int {
req := httptest.NewRequest(http.MethodPost, "/v1/proposals/"+proposal.ProposalID+"/accept", nil)
req.Header.Set("Authorization", "Bearer "+credential)
req.Header.Set("Idempotency-Key", key)
req.Header.Set("If-Match-Revision", revision)
recorder := httptest.NewRecorder()
service.proposalMutation(recorder, req)
return recorder.Code
}
credential1 := session1.SessionID + ":" + token1
credential2 := session2.SessionID + ":" + token2
if status := respond(credential1, "proposal-promote-first", "0"); status != http.StatusOK || promoter.calls != 0 {
t.Fatalf("first acceptance status/calls = %d/%d", status, promoter.calls)
}
if status := respond(credential2, "proposal-promote-final", "1"); status != http.StatusOK || promoter.calls != 1 || promoter.proposal.State != domain.Accepted {
t.Fatalf("final acceptance status/promoter = %d/%+v", status, promoter)
}
promoter.err = errors.New("database unavailable")
// A duplicate response is replayed by the durable proposal backend and
// retries promotion instead of asking the player to accept again.
if status := respond(credential2, "proposal-promote-final", "1"); status != http.StatusServiceUnavailable || promoter.calls != 2 {
t.Fatalf("promotion retry status/calls = %d/%d", status, promoter.calls)
}
}
func TestProposalRecoveryUsesDurableBackendAndRemainsParticipantScoped(t *testing.T) { func TestProposalRecoveryUsesDurableBackendAndRemainsParticipantScoped(t *testing.T) {
now := time.Unix(1000, 0).UTC() now := time.Unix(1000, 0).UTC()
proposal, err := domain.NewProposal("proposal-1234567890123456", domain.Casual, []string{"player-1", "player-2"}, now) proposal, err := domain.NewProposal("proposal-1234567890123456", domain.Casual, []string{"player-1", "player-2"}, now)
+12
View File
@@ -46,3 +46,15 @@ func (p postgresProposalBackend) Respond(ctx context.Context, playerID, proposal
func ProposalProviderFromStore(db *sql.DB) ProposalBackend { func ProposalProviderFromStore(db *sql.DB) ProposalBackend {
return postgresProposalBackend{db: db} return postgresProposalBackend{db: db}
} }
// ProposalPromoterFromStore turns a durably accepted proposal into its exact
// matcher-selected ALLOCATING match. The store chooses a deterministic match
// ID so an API retry after a transient failure cannot duplicate the match.
func ProposalPromoterFromStore(db *sql.DB) ProposalPromoter {
return ProposalPromoterFunc(func(ctx context.Context, proposal domain.Proposal, now time.Time) error {
if proposal.State != domain.Accepted {
return domain.ErrIllegalTransition
}
return store.PromoteStoredAcceptedProposal(ctx, db, proposal.ProposalID, now)
})
}
+13
View File
@@ -4,6 +4,8 @@ import (
"context" "context"
"testing" "testing"
"time" "time"
"github.com/cosmic-clash/cosmic-clash/server/domain"
) )
func TestAssignmentProviderFromStorePreservesPlayerScopedRecoveryBoundary(t *testing.T) { func TestAssignmentProviderFromStorePreservesPlayerScopedRecoveryBoundary(t *testing.T) {
@@ -25,3 +27,14 @@ func TestProposalProviderFromStoreFailsClosedWithoutDatabase(t *testing.T) {
t.Fatal("nil store was treated as an available proposal source") t.Fatal("nil store was treated as an available proposal source")
} }
} }
func TestProposalPromoterFromStoreFailsClosedWithoutDatabase(t *testing.T) {
promoter := ProposalPromoterFromStore(nil)
if promoter == nil {
t.Fatal("proposal promoter was not created")
}
proposal := domain.Proposal{ProposalID: "proposal-1", State: domain.Accepted}
if err := promoter.Promote(context.Background(), proposal, time.Unix(1000, 0)); err == nil {
t.Fatal("nil store was treated as an available proposal promoter")
}
}
+8 -7
View File
@@ -81,13 +81,14 @@ func newAPIHandler(db *sql.DB, indexes ...api.CandidateIndex) http.Handler {
candidateIndex = indexes[0] candidateIndex = indexes[0]
} }
return (&api.Service{ return (&api.Service{
SessionBackend: store.PostgresSessions{DB: db}, SessionBackend: store.PostgresSessions{DB: db},
QueueBackend: store.PostgresQueue{DB: db}, QueueBackend: store.PostgresQueue{DB: db},
ProposalBackend: api.ProposalProviderFromStore(db), ProposalBackend: api.ProposalProviderFromStore(db),
Assignment: api.AssignmentProviderFromStore(db), ProposalPromoter: api.ProposalPromoterFromStore(db),
CandidateIndex: candidateIndex, Assignment: api.AssignmentProviderFromStore(db),
ProbeRecorder: store.PostgresQueue{DB: db}, CandidateIndex: candidateIndex,
Now: func() time.Time { return time.Now().UTC() }, ProbeRecorder: store.PostgresQueue{DB: db},
Now: func() time.Time { return time.Now().UTC() },
}).Handler() }).Handler()
} }
+30
View File
@@ -2,6 +2,7 @@ package domain
import ( import (
"fmt" "fmt"
"sort"
"time" "time"
) )
@@ -68,5 +69,34 @@ func PrepareProposal(id string, playlist Playlist, formation MatchFormation, ran
if err != nil { if err != nil {
return PreparedProposal{}, err return PreparedProposal{}, err
} }
if formation.Selection.Region == "" || len(formation.Selection.Players) == 0 || formation.Selection.Players[0].ProtocolVersion < 1 {
return PreparedProposal{}, fmt.Errorf("formed match metadata is incomplete")
}
proposal.Region = formation.Selection.Region
proposal.Protocol = formation.Selection.Players[0].ProtocolVersion
for _, player := range formation.Selection.Players {
if player.ProtocolVersion != proposal.Protocol {
return PreparedProposal{}, fmt.Errorf("formed match has mixed protocols")
}
}
assignProposalSlots(&proposal, formation.Teams)
return PreparedProposal{Proposal: proposal, CasualLineup: lineup}, nil return PreparedProposal{Proposal: proposal, CasualLineup: lineup}, nil
} }
func assignProposalSlots(proposal *Proposal, teams Teams) {
assign := func(team int, players []Candidate) {
ordered := append([]Candidate(nil), players...)
sort.Slice(ordered, func(i, j int) bool { return ordered[i].PlayerID < ordered[j].PlayerID })
for index, player := range ordered {
for participant := range proposal.Participants {
if proposal.Participants[participant].PlayerID == player.PlayerID {
proposal.Participants[participant].Team = team
proposal.Participants[participant].Slot = team*3 + index
break
}
}
}
}
assign(0, teams.Team0)
assign(1, teams.Team1)
}
+4 -1
View File
@@ -10,7 +10,7 @@ func testFormation(t *testing.T, count int) MatchFormation {
now := time.Unix(1000, 0) now := time.Unix(1000, 0)
players := make([]Candidate, count) players := make([]Candidate, count)
for i := range players { for i := range players {
players[i] = Candidate{TicketID: string(rune('a' + i)), PlayerID: string(rune('p' + i)), Rating: 1500, EnqueuedAt: now, PredictedRTT: map[string]float64{"EU": 40}} players[i] = Candidate{TicketID: string(rune('a' + i)), PlayerID: string(rune('p' + i)), ProtocolVersion: 1, Rating: 1500, EnqueuedAt: now, PredictedRTT: map[string]float64{"EU": 40}}
} }
selection, err := SelectCandidates(players[0], players[1:], count, now) selection, err := SelectCandidates(players[0], players[1:], count, now)
if err != nil { if err != nil {
@@ -31,6 +31,9 @@ func TestPrepareProposalBuildsCasualLineupBeforeCreatingProposal(t *testing.T) {
if prepared.Proposal.Playlist != Casual || len(prepared.Proposal.Participants) != 2 || len(prepared.CasualLineup) != 6 { if prepared.Proposal.Playlist != Casual || len(prepared.Proposal.Participants) != 2 || len(prepared.CasualLineup) != 6 {
t.Fatalf("prepared casual proposal = %+v", prepared) t.Fatalf("prepared casual proposal = %+v", prepared)
} }
if prepared.Proposal.Region != "EU" || prepared.Proposal.Protocol != 1 || prepared.Proposal.Participants[0].Slot == prepared.Proposal.Participants[1].Slot || prepared.Proposal.Participants[0].Team == prepared.Proposal.Participants[1].Team {
t.Fatalf("prepared proposal did not retain deterministic topology: %+v", prepared.Proposal)
}
humans := 0 humans := 0
teams := map[int]bool{} teams := map[int]bool{}
for _, slot := range prepared.CasualLineup { for _, slot := range prepared.CasualLineup {
+4
View File
@@ -34,11 +34,15 @@ const (
type ProposalParticipant struct { type ProposalParticipant struct {
PlayerID string PlayerID string
Response Response Response Response
Team int
Slot int
} }
type Proposal struct { type Proposal struct {
ProposalID string ProposalID string
Playlist Playlist Playlist Playlist
Region string
Protocol int
Participants []ProposalParticipant Participants []ProposalParticipant
State State State State
Revision uint64 Revision uint64
+1 -1
View File
@@ -27,7 +27,7 @@ func candidates() []domain.Candidate {
now := time.Unix(1000, 0).UTC() now := time.Unix(1000, 0).UTC()
result := make([]domain.Candidate, 4) result := make([]domain.Candidate, 4)
for i := range result { for i := range result {
result[i] = domain.Candidate{TicketID: "ticket-" + string(rune('1'+i)), PlayerID: "player-" + string(rune('1'+i)), Playlist: domain.Casual, EnqueuedAt: now.Add(time.Duration(i) * time.Second), PredictedRTT: map[string]float64{"EU": 20}} result[i] = domain.Candidate{TicketID: "ticket-" + string(rune('1'+i)), PlayerID: "player-" + string(rune('1'+i)), Playlist: domain.Casual, ProtocolVersion: 1, EnqueuedAt: now.Add(time.Duration(i) * time.Second), PredictedRTT: map[string]float64{"EU": 20}}
} }
return result return result
} }
@@ -0,0 +1,15 @@
-- Preserve the matcher-selected topology through the proposal response window.
-- These fields are nullable for already-created proposals during a rolling
-- deployment; new matcher-created proposals always populate them before they
-- can be promoted to an ALLOCATING match.
ALTER TABLE proposals
ADD COLUMN match_region TEXT CHECK (match_region IN ('EU', 'NA')),
ADD COLUMN match_protocol INTEGER CHECK (match_protocol > 0);
ALTER TABLE proposal_participants
ADD COLUMN team INTEGER CHECK (team IN (0, 1)),
ADD COLUMN slot INTEGER CHECK (slot BETWEEN 0 AND 5);
CREATE UNIQUE INDEX proposal_participants_unique_slot
ON proposal_participants (proposal_id, slot)
WHERE slot IS NOT NULL;
+38
View File
@@ -63,6 +63,44 @@ const AcceptedMatchParticipantInsertSQL = `INSERT INTO match_participants
(match_id, player_id, ticket_id, slot, team) (match_id, player_id, ticket_id, slot, team)
VALUES ($1, $2, $3, $4, $5)` VALUES ($1, $2, $3, $4, $5)`
const StoredProposalMatchPlanSQL = `SELECT match_region, match_protocol
FROM proposals
WHERE proposal_id = $1 AND state = 'ACCEPTED'`
const StoredProposalMatchPlayersSQL = `SELECT player_id, team, slot
FROM proposal_participants
WHERE proposal_id = $1 AND response = 'ACCEPTED'
ORDER BY player_id`
// PromoteStoredAcceptedProposal materializes the exact topology persisted by
// the matcher once every player has accepted. The deterministic match ID makes
// a request retry converge after an API/worker interruption.
func PromoteStoredAcceptedProposal(ctx context.Context, db *sql.DB, proposalID string, now time.Time) error {
if db == nil || proposalID == "" || now.IsZero() {
return fmt.Errorf("invalid stored proposal promotion arguments")
}
plan := AcceptedMatchPlan{MatchID: "match-" + proposalID, ProposalID: proposalID}
if err := db.QueryRowContext(ctx, StoredProposalMatchPlanSQL, proposalID).Scan(&plan.Region, &plan.Protocol); err != nil {
return err
}
rows, err := db.QueryContext(ctx, StoredProposalMatchPlayersSQL, proposalID)
if err != nil {
return err
}
defer rows.Close()
for rows.Next() {
var player MatchPlayer
if err := rows.Scan(&player.PlayerID, &player.Team, &player.Slot); err != nil {
return err
}
plan.Players = append(plan.Players, player)
}
if err := rows.Err(); err != nil {
return err
}
return CreateMatchFromAcceptedProposal(ctx, db, plan, now)
}
// CreateMatchFromAcceptedProposal atomically promotes the exact accepted // CreateMatchFromAcceptedProposal atomically promotes the exact accepted
// roster into an ALLOCATING match. An existing match ID is an idempotent retry // roster into an ALLOCATING match. An existing match ID is an idempotent retry
// only if every durable field and participant assignment matches the request. // only if every durable field and participant assignment matches the request.
+36 -4
View File
@@ -10,8 +10,8 @@ import (
) )
const ProposalInsertSQL = `INSERT INTO proposals const ProposalInsertSQL = `INSERT INTO proposals
(proposal_id, playlist, state, expires_at, revision) (proposal_id, playlist, state, expires_at, revision, match_region, match_protocol)
VALUES ($1, $2, 'OPEN', $3, 0)` VALUES ($1, $2, 'OPEN', $3, 0, NULLIF($4, ''), NULLIF($5, 0))`
// CreateProposal atomically claims the queue tickets and creates the proposal. // CreateProposal atomically claims the queue tickets and creates the proposal.
// Every statement runs inside the same SERIALIZABLE retry callback; callers // Every statement runs inside the same SERIALIZABLE retry callback; callers
@@ -20,8 +20,11 @@ func CreateProposal(ctx context.Context, db *sql.DB, proposal domain.Proposal, t
if proposal.ProposalID == "" || len(proposal.Participants) == 0 { if proposal.ProposalID == "" || len(proposal.Participants) == 0 {
return fmt.Errorf("invalid proposal transaction") return fmt.Errorf("invalid proposal transaction")
} }
if !validProposalMatchPlan(proposal) {
return fmt.Errorf("invalid proposal match plan")
}
return RunSerializable(ctx, db, DefaultSerializableAttempts, func(ctx context.Context, tx *sql.Tx) error { return RunSerializable(ctx, db, DefaultSerializableAttempts, func(ctx context.Context, tx *sql.Tx) error {
if _, err := tx.ExecContext(ctx, ProposalInsertSQL, proposal.ProposalID, proposal.Playlist, proposal.ExpiresAt); err != nil { if _, err := tx.ExecContext(ctx, ProposalInsertSQL, proposal.ProposalID, proposal.Playlist, proposal.ExpiresAt, proposal.Region, proposal.Protocol); err != nil {
return err return err
} }
for _, participant := range proposal.Participants { for _, participant := range proposal.Participants {
@@ -29,7 +32,7 @@ func CreateProposal(ctx context.Context, db *sql.DB, proposal domain.Proposal, t
if participant.PlayerID == "" || ticketID == "" { if participant.PlayerID == "" || ticketID == "" {
return fmt.Errorf("missing proposal ticket mapping") return fmt.Errorf("missing proposal ticket mapping")
} }
if _, err := tx.ExecContext(ctx, ProposalParticipantInsertSQL, proposal.ProposalID, participant.PlayerID, ticketID); err != nil { if _, err := tx.ExecContext(ctx, ProposalParticipantInsertSQL, proposal.ProposalID, participant.PlayerID, ticketID, nullablePlanField(proposal.Region != "", participant.Team), nullablePlanField(proposal.Region != "", participant.Slot)); err != nil {
return err return err
} }
result, err := tx.ExecContext(ctx, QueueTicketProposeSQL, ticketID, participant.PlayerID, string(proposal.Playlist), now) result, err := tx.ExecContext(ctx, QueueTicketProposeSQL, ticketID, participant.PlayerID, string(proposal.Playlist), now)
@@ -47,3 +50,32 @@ func CreateProposal(ctx context.Context, db *sql.DB, proposal domain.Proposal, t
return nil return nil
}) })
} }
func validProposalMatchPlan(proposal domain.Proposal) bool {
if proposal.Region == "" && proposal.Protocol == 0 {
return true // Legacy/direct callers have no matcher formation to persist.
}
if (proposal.Region != "EU" && proposal.Region != "NA") || proposal.Protocol < 1 {
return false
}
seenSlots := make(map[int]struct{}, len(proposal.Participants))
teams := [2]int{}
for _, participant := range proposal.Participants {
if participant.Team < 0 || participant.Team > 1 || participant.Slot < 0 || participant.Slot > 5 {
return false
}
if _, exists := seenSlots[participant.Slot]; exists {
return false
}
seenSlots[participant.Slot] = struct{}{}
teams[participant.Team]++
}
return teams[0] > 0 && teams[1] > 0
}
func nullablePlanField(enabled bool, value int) any {
if !enabled {
return nil
}
return value
}
+10 -8
View File
@@ -8,14 +8,16 @@ import (
func TestQueueSQLUsesDurableIdempotencyAndOwnerScopedRecovery(t *testing.T) { func TestQueueSQLUsesDurableIdempotencyAndOwnerScopedRecovery(t *testing.T) {
for query, fragments := range map[string][]string{ for query, fragments := range map[string][]string{
QueueIdempotencyInsertSQL: {"idempotency_keys", "ON CONFLICT (scope, idempotency_key) DO NOTHING", "payload_digest"}, QueueIdempotencyInsertSQL: {"idempotency_keys", "ON CONFLICT (scope, idempotency_key) DO NOTHING", "payload_digest"},
QueueIdempotencySelectSQL: {"scope = $1", "idempotency_key = $2", "FOR UPDATE"}, QueueIdempotencySelectSQL: {"scope = $1", "idempotency_key = $2", "FOR UPDATE"},
QueueTicketSelectSQL: {"ticket_id = $1", "player_id = $2"}, QueueTicketSelectSQL: {"ticket_id = $1", "player_id = $2"},
QueueTicketInsertSQL: {"player_id", "playlist", "client_build", "protocol_version"}, QueueTicketInsertSQL: {"player_id", "playlist", "client_build", "protocol_version"},
QueueTicketHeartbeatSQL: {"player_id = $2", "revision = $3", "expires_at > $4", "RETURNING"}, QueueTicketHeartbeatSQL: {"player_id = $2", "revision = $3", "expires_at > $4", "RETURNING"},
QueueTicketCancelSQL: {"player_id = $2", "revision = $3", "state NOT IN", "RETURNING"}, QueueTicketCancelSQL: {"player_id = $2", "revision = $3", "state NOT IN", "RETURNING"},
QueueCandidateProjectionSQL: {"playlist = $1", "predicted_rtt", "expires_at > $2", "LIMIT $3"}, QueueCandidateProjectionSQL: {"playlist = $1", "predicted_rtt", "expires_at > $2", "LIMIT $3"},
RankedParticipantSQL: {"steam_id", "player_id = ANY($1)", "ORDER BY player_id"}, RankedParticipantSQL: {"steam_id", "player_id = ANY($1)", "ORDER BY player_id"},
ProposalInsertSQL: {"match_region", "match_protocol", "NULLIF($4, '')"},
ProposalParticipantInsertSQL: {"team", "slot", "'PENDING'"},
} { } {
for _, fragment := range fragments { for _, fragment := range fragments {
if !contains(query, fragment) { if !contains(query, fragment) {
+2 -2
View File
@@ -75,8 +75,8 @@ ORDER BY enqueued_at, ticket_id
LIMIT $2 LIMIT $2
FOR UPDATE SKIP LOCKED` FOR UPDATE SKIP LOCKED`
ProposalParticipantInsertSQL = `INSERT INTO proposal_participants (proposal_id, player_id, ticket_id, response) ProposalParticipantInsertSQL = `INSERT INTO proposal_participants (proposal_id, player_id, ticket_id, response, team, slot)
VALUES ($1, $2, $3, 'PENDING')` VALUES ($1, $2, $3, 'PENDING', $4, $5)`
QueueTicketProposeSQL = `UPDATE queue_tickets SET state = 'PROPOSED', revision = revision + 1 QueueTicketProposeSQL = `UPDATE queue_tickets SET state = 'PROPOSED', revision = revision + 1
WHERE ticket_id = $1 AND player_id = $2 AND playlist = $3 AND state = 'QUEUED' AND expires_at > $4` WHERE ticket_id = $1 AND player_id = $2 AND playlist = $3 AND state = 'QUEUED' AND expires_at > $4`