From d937cb153cc63dbef6d925e8bc43d608563c1f57 Mon Sep 17 00:00:00 2001 From: Josh Creek <8179928+jcreek@users.noreply.github.com> Date: Tue, 1 Sep 2026 12:35:30 +0100 Subject: [PATCH] feat(multiplayer): add server process/assignment-ready registration API Add POST /v1/servers/{id}/register (and its /api/v1 contract alias), authenticated by the same workload binding as the result route. A game server reports its protocol version and image digest and asks to advance ALLOCATING -> PROCESS_READY -> ASSIGNMENT_READY; the store boundary (AdvanceServerRegistration) does this as one idempotent SERIALIZABLE transaction that also advances every participant's queue ticket, and gates the final transition on every participant having a live, unexpired assignment. Adversarial review of the surrounding routing turned up a pre-existing bug: contractServerMutation rejected any path containing '/', so the already-documented /api/v1/servers/{id}/result route (and this new /register route) 404'd for every real caller despite being declared in the OpenAPI contract. Fix it to delegate shape validation to serverMutation, matching how contractQueueMutation handles its own two-segment paths, and add a regression test covering both contract routes end to end. --- server/api/service.go | 55 +++++++++++++-- server/api/service_test.go | 81 +++++++++++++++++++++++ server/api/store_adapters.go | 13 ++++ server/cmd/control-plane/main.go | 1 + server/store/allocation_match_sql.go | 68 +++++++++++++++++++ server/store/allocation_match_sql_test.go | 3 + 6 files changed, 217 insertions(+), 4 deletions(-) diff --git a/server/api/service.go b/server/api/service.go index 9377db2e..a4698241 100644 --- a/server/api/service.go +++ b/server/api/service.go @@ -33,6 +33,9 @@ type WorkloadVerifier func(string, time.Time) (domain.WorkloadBinding, error) type ResultSubmitter interface { SubmitResult(context.Context, string, domain.MatchResult, domain.WorkloadBinding, []byte, time.Time) error } +type ServerRegistrar interface { + RegisterServer(context.Context, domain.WorkloadBinding, int, bool, string, time.Time) error +} type QueueBackend interface { Create(context.Context, string, string, string, domain.QueueSpec, time.Time) (domain.QueueTicket, error) @@ -104,6 +107,7 @@ type Service struct { ProbeRecorder ProbeRecorder WorkloadVerify WorkloadVerifier ResultSubmitter ResultSubmitter + ServerRegistrar ServerRegistrar Assignment AssignmentProvider Now func() time.Time Proposals map[string]*domain.Proposal @@ -369,8 +373,12 @@ func (s *Service) contractAssignment(w http.ResponseWriter, r *http.Request) { } func (s *Service) contractServerMutation(w http.ResponseWriter, r *http.Request) { + // Unlike contractAssignment, the documented shape here is two segments + // (/servers/{serverId}/result, /servers/{serverId}/register) — rejecting + // any "/" would 404 every real call. Delegate shape validation to + // serverMutation, which already enforces exactly {id}/{result|register}. path := strings.TrimPrefix(r.URL.Path, "/api/v1/servers/") - if path == "" || strings.Contains(path, "/") { + if path == "" { writeError(w, http.StatusNotFound, "not_found") return } @@ -389,18 +397,25 @@ type resultRequest struct { IntegrityState domain.IntegrityState `json:"integrity_state"` } +type serverRegistrationRequest struct { + MatchID string `json:"match_id"` + ProtocolVersion int `json:"protocol_version"` + ImageDigest string `json:"image_digest"` + AssignmentReady bool `json:"assignment_ready"` +} + func (s *Service) serverMutation(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { writeError(w, http.StatusMethodNotAllowed, "method_not_allowed") return } parts := strings.Split(strings.TrimPrefix(r.URL.Path, "/v1/servers/"), "/") - if len(parts) != 2 || parts[0] == "" || parts[1] != "result" { + if len(parts) != 2 || parts[0] == "" || (parts[1] != "result" && parts[1] != "register") { writeError(w, http.StatusNotFound, "not_found") return } - if s.WorkloadVerify == nil || s.ResultSubmitter == nil { - writeError(w, http.StatusServiceUnavailable, "result_unavailable") + if s.WorkloadVerify == nil || (parts[1] == "result" && s.ResultSubmitter == nil) || (parts[1] == "register" && s.ServerRegistrar == nil) { + writeError(w, http.StatusServiceUnavailable, "server_unavailable") return } key := r.Header.Get("Idempotency-Key") @@ -419,6 +434,26 @@ func (s *Service) serverMutation(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusUnauthorized, "unauthorized") return } + if parts[1] == "register" { + var input serverRegistrationRequest + if !decodeBody(w, r, &input) { + return + } + if input.MatchID == "" || input.MatchID != binding.MatchID || input.ProtocolVersion < 1 || !validImageDigest(input.ImageDigest) { + writeError(w, http.StatusUnprocessableEntity, "invalid_request") + return + } + if err := s.ServerRegistrar.RegisterServer(r.Context(), binding, input.ProtocolVersion, input.AssignmentReady, key, now); err != nil { + if errors.Is(err, domain.ErrConflict) { + writeError(w, http.StatusConflict, "conflict") + } else { + writeError(w, http.StatusUnprocessableEntity, "invalid_request") + } + return + } + w.WriteHeader(http.StatusNoContent) + return + } var input resultRequest if !decodeBody(w, r, &input) { return @@ -444,6 +479,18 @@ func (s *Service) serverMutation(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusAccepted) } +func validImageDigest(value string) bool { + if len(value) != len("sha256:")+64 || !strings.HasPrefix(value, "sha256:") { + return false + } + for _, ch := range value[len("sha256:"):] { + if !(ch >= '0' && ch <= '9') && !(ch >= 'a' && ch <= 'f') { + return false + } + } + return true +} + func (s *Service) queueMutation(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost && r.Method != http.MethodGet && r.Method != http.MethodDelete { writeError(w, http.StatusMethodNotAllowed, "method_not_allowed") diff --git a/server/api/service_test.go b/server/api/service_test.go index ec613d61..8858ae3d 100644 --- a/server/api/service_test.go +++ b/server/api/service_test.go @@ -41,6 +41,20 @@ type resultSubmitterSpy struct { result domain.MatchResult } +type serverRegistrarSpy struct { + calls int + binding domain.WorkloadBinding + protocol int + assignmentReady bool + err error +} + +func (s *serverRegistrarSpy) RegisterServer(_ context.Context, binding domain.WorkloadBinding, protocol int, assignmentReady bool, _ string, _ time.Time) error { + s.calls++ + s.binding, s.protocol, s.assignmentReady = binding, protocol, assignmentReady + return s.err +} + type proposalPromoterSpy struct { calls int proposal domain.Proposal @@ -1018,6 +1032,73 @@ func TestServerResultAPIRequiresBoundWorkloadAndDelegatesDurableSubmission(t *te response.Body.Close() } +func TestContractServerRoutesAdaptTwoSegmentPaths(t *testing.T) { + now := time.Unix(1000, 0).UTC() + binding := domain.WorkloadBinding{AllocationID: "allocation-1", MatchID: "match-1", ServerID: "server-1"} + submitter := &resultSubmitterSpy{} + registrar := &serverRegistrarSpy{} + service := &Service{Now: func() time.Time { return now }, WorkloadVerify: func(token string, _ time.Time) (domain.WorkloadBinding, error) { + if token != "workload-token" { + return domain.WorkloadBinding{}, errors.New("bad token") + } + return binding, nil + }, ResultSubmitter: submitter, ServerRegistrar: registrar} + server := httptest.NewServer(service.Handler()) + defer server.Close() + + registerBody := `{"match_id":"match-1","protocol_version":1,"image_digest":"sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","assignment_ready":false}` + req, _ := http.NewRequest(http.MethodPost, server.URL+"/api/v1/servers/server-1/register", strings.NewReader(registerBody)) + req.Header.Set("Authorization", "Bearer workload-token") + req.Header.Set("Idempotency-Key", "contract-register-key-1") + response, err := http.DefaultClient.Do(req) + if err != nil || response.StatusCode != http.StatusNoContent || registrar.calls != 1 { + t.Fatalf("register status=%v err=%v calls=%d", response.StatusCode, err, registrar.calls) + } + response.Body.Close() + + resultBody := `{"match_id":"match-1","result_nonce":"nonce-1234567890","score":{"team_0":3,"team_1":2},"integrity_state":"CERTIFIED"}` + req, _ = http.NewRequest(http.MethodPost, server.URL+"/api/v1/servers/server-1/result", strings.NewReader(resultBody)) + req.Header.Set("Authorization", "Bearer workload-token") + req.Header.Set("Idempotency-Key", "contract-result-key-123") + response, err = http.DefaultClient.Do(req) + if err != nil || response.StatusCode != http.StatusAccepted || submitter.calls != 1 { + t.Fatalf("result status=%v err=%v calls=%d", response.StatusCode, err, submitter.calls) + } + response.Body.Close() +} + +func TestServerRegistrationAPIRequiresBoundWorkloadAndValidDigest(t *testing.T) { + now := time.Unix(1000, 0).UTC() + binding := domain.WorkloadBinding{AllocationID: "allocation-1", MatchID: "match-1", ServerID: "server-1"} + registrar := &serverRegistrarSpy{} + service := &Service{Now: func() time.Time { return now }, WorkloadVerify: func(token string, _ time.Time) (domain.WorkloadBinding, error) { + if token != "workload-token" { + return domain.WorkloadBinding{}, errors.New("bad token") + } + return binding, nil + }, ServerRegistrar: registrar} + server := httptest.NewServer(service.Handler()) + defer server.Close() + body := `{"match_id":"match-1","protocol_version":1,"image_digest":"sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","assignment_ready":false}` + req, _ := http.NewRequest(http.MethodPost, server.URL+"/v1/servers/server-1/register", strings.NewReader(body)) + req.Header.Set("Authorization", "Bearer workload-token") + req.Header.Set("Idempotency-Key", "register-key-123456") + response, err := http.DefaultClient.Do(req) + if err != nil || response.StatusCode != http.StatusNoContent || registrar.calls != 1 || registrar.binding != binding || registrar.protocol != 1 || registrar.assignmentReady { + t.Fatalf("status=%v err=%v registrar=%+v", response.StatusCode, err, registrar) + } + response.Body.Close() + body = `{"match_id":"match-1","protocol_version":1,"image_digest":"bad","assignment_ready":false}` + req, _ = http.NewRequest(http.MethodPost, server.URL+"/v1/servers/server-1/register", strings.NewReader(body)) + req.Header.Set("Authorization", "Bearer workload-token") + req.Header.Set("Idempotency-Key", "register-key-123456") + response, err = http.DefaultClient.Do(req) + if err != nil || response.StatusCode != http.StatusUnprocessableEntity || registrar.calls != 1 { + t.Fatalf("invalid registration status=%v err=%v calls=%d", response.StatusCode, err, registrar.calls) + } + response.Body.Close() +} + func TestProbeAPIUsesServerEvidenceAndRejectsClientRTTField(t *testing.T) { now := time.Unix(1000, 0).UTC() sessions := domain.NewSessionStore() diff --git a/server/api/store_adapters.go b/server/api/store_adapters.go index e555926d..9359a210 100644 --- a/server/api/store_adapters.go +++ b/server/api/store_adapters.go @@ -58,3 +58,16 @@ func ProposalPromoterFromStore(db *sql.DB) ProposalPromoter { return store.PromoteStoredAcceptedProposal(ctx, db, proposal.ProposalID, now) }) } + +type postgresServerRegistrar struct{ db *sql.DB } + +func (p postgresServerRegistrar) RegisterServer(ctx context.Context, binding domain.WorkloadBinding, protocol int, assignmentReady bool, idempotencyKey string, now time.Time) error { + return store.AdvanceServerRegistration(ctx, p.db, binding, protocol, assignmentReady, idempotencyKey, now) +} + +func ServerRegistrarFromStore(db *sql.DB) ServerRegistrar { + if db == nil { + return nil + } + return postgresServerRegistrar{db: db} +} diff --git a/server/cmd/control-plane/main.go b/server/cmd/control-plane/main.go index 6c65eebf..7a22c32a 100644 --- a/server/cmd/control-plane/main.go +++ b/server/cmd/control-plane/main.go @@ -85,6 +85,7 @@ func newAPIHandler(db *sql.DB, indexes ...api.CandidateIndex) http.Handler { QueueBackend: store.PostgresQueue{DB: db}, ProposalBackend: api.ProposalProviderFromStore(db), ProposalPromoter: api.ProposalPromoterFromStore(db), + ServerRegistrar: api.ServerRegistrarFromStore(db), Assignment: api.AssignmentProviderFromStore(db), CandidateIndex: candidateIndex, ProbeRecorder: store.PostgresQueue{DB: db}, diff --git a/server/store/allocation_match_sql.go b/server/store/allocation_match_sql.go index e3df0e97..6813e049 100644 --- a/server/store/allocation_match_sql.go +++ b/server/store/allocation_match_sql.go @@ -3,6 +3,7 @@ package store import ( "bytes" "context" + "crypto/sha256" "database/sql" "fmt" "time" @@ -62,6 +63,73 @@ const ReleaseAllocatedMatchClaimSQL = `UPDATE matches SET allocation_id = NULL, allocation_claimed_at = NULL WHERE match_id = $1 AND state = 'ALLOCATING' AND allocation_id = $2 AND server_id IS NULL` +const AdvanceServerRegistrationSQL = `WITH matched AS ( + UPDATE matches + SET state = $4, revision = revision + 1 + WHERE match_id = $1 AND server_id = $2 AND state = $3 AND protocol_version = $7 + AND EXISTS (SELECT 1 FROM allocations WHERE match_id = $1 AND server_id = $2 AND allocation_id = $5 AND protocol_version = $7 AND state = 'ALLOCATED') + AND ($4 <> 'ASSIGNMENT_READY' OR (SELECT count(*) FROM assignments WHERE match_id = $1 AND expires_at > $6) = (SELECT count(*) FROM match_participants WHERE match_id = $1)) + RETURNING match_id +), advanced AS ( + UPDATE queue_tickets q + SET state = $4, revision = revision + 1 + FROM match_participants mp JOIN matched m ON m.match_id = mp.match_id + 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)` + +const ServerRegistrationIdempotencyScope = "server.register" + +const ServerRegistrationIdempotencyInsertSQL = `INSERT INTO idempotency_keys + (scope, idempotency_key, payload_digest, result) +VALUES ($1, $2, $3, '{}') +ON CONFLICT (scope, idempotency_key) DO NOTHING` + +const ServerRegistrationIdempotencySelectSQL = `SELECT payload_digest +FROM idempotency_keys +WHERE scope = $1 AND idempotency_key = $2 +FOR UPDATE` + +func AdvanceServerRegistration(ctx context.Context, db *sql.DB, binding domain.WorkloadBinding, protocol int, assignmentReady bool, idempotencyKey string, now time.Time) error { + if db == nil || binding.MatchID == "" || binding.ServerID == "" || binding.AllocationID == "" || protocol < 1 || len(idempotencyKey) < 16 || len(idempotencyKey) > 128 || now.IsZero() { + return fmt.Errorf("invalid server registration") + } + from, to := domain.Allocating, domain.ProcessReady + if assignmentReady { + from, to = domain.ProcessReady, domain.AssignmentReady + } + digest := sha256.Sum256([]byte(fmt.Sprintf("%s\x00%s\x00%s\x00%d\x00%t", binding.AllocationID, binding.MatchID, binding.ServerID, protocol, assignmentReady))) + return RunSerializable(ctx, db, DefaultSerializableAttempts, func(ctx context.Context, tx *sql.Tx) error { + inserted, err := tx.ExecContext(ctx, ServerRegistrationIdempotencyInsertSQL, ServerRegistrationIdempotencyScope, idempotencyKey, digest[:]) + if err != nil { + return err + } + changed, err := inserted.RowsAffected() + if err != nil { + return err + } + if changed == 0 { + var prior []byte + if err := tx.QueryRowContext(ctx, ServerRegistrationIdempotencySelectSQL, ServerRegistrationIdempotencyScope, idempotencyKey).Scan(&prior); err != nil { + return err + } + if !bytes.Equal(prior, digest[:]) { + return domain.ErrConflict + } + 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 { + return err + } + if matched != 1 || participants == 0 || advanced != participants { + return domain.ErrConflict + } + return nil + }) +} + // FindProviderAllocation verifies whether a recovered lease has already // crossed the durable provider boundary. A worker can then bind it without // issuing a second external allocation request after a crash. diff --git a/server/store/allocation_match_sql_test.go b/server/store/allocation_match_sql_test.go index 5a5f02ee..dd0c1ae1 100644 --- a/server/store/allocation_match_sql_test.go +++ b/server/store/allocation_match_sql_test.go @@ -13,6 +13,9 @@ func TestAllocationMatchClaimSQLFencesConcurrentWorkers(t *testing.T) { AllocatingMatchBuildSQL: {"match_participants", "queue_tickets", "ORDER BY q.client_build"}, BindAllocatedMatchParticipantsSQL: {"allocation_id = $2", "server_id IS NULL", "SET server_id = $3", "FROM allocations", "state = 'ALLOCATING'", "revision = revision + 1"}, ReleaseAllocatedMatchClaimSQL: {"allocation_id = $2", "allocation_id = NULL", "allocation_claimed_at = NULL"}, + AdvanceServerRegistrationSQL: {"state = $4", "protocol_version = $7", "ASSIGNMENT_READY", "revision = revision + 1"}, + ServerRegistrationIdempotencyInsertSQL: {"idempotency_keys", "ON CONFLICT (scope, idempotency_key) DO NOTHING", "payload_digest"}, + ServerRegistrationIdempotencySelectSQL: {"scope = $1", "idempotency_key = $2", "FOR UPDATE"}, } for query, fragments := range checks { for _, fragment := range fragments {