mirror of
https://github.com/jcreek/CosmicClash.git
synced 2026-09-10 16:04:04 +00:00
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.
This commit is contained in:
+51
-4
@@ -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")
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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}
|
||||
}
|
||||
|
||||
@@ -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},
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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 {
|
||||
|
||||
Reference in New Issue
Block a user