diff --git a/multiplayer-next.md b/multiplayer-next.md index f074d54d..7bcc7bae 100644 --- a/multiplayer-next.md +++ b/multiplayer-next.md @@ -1430,3 +1430,5 @@ The documented `server_shutdown` reliable control message is now implemented in Clients now consume planned shutdowns: the reason is retained for presentation, an in-match client returns to the lobby after the notice, and the generic disconnect callback is fenced so it cannot overwrite that planned transition. Lobby clients surface the reason directly. The complete Godot harness remains green; real two-process drain delivery is still an external runtime gate. An adversarial UI review found the lobby’s generic disconnect handler still replaced that message with the main menu immediately afterward. Planned disconnects are now fenced in the lobby, and a lobby reached from an active match restores the retained reason on startup; unplanned disconnects keep the existing main-menu behavior. + +The workload-authenticated `POST /servers/{serverId}/shutdown` contract is now exposed for allocated servers. It validates the bound credential and reason, records an idempotent `SERVER_SHUTDOWN` audit event under a serializable transaction, and returns a stable acknowledgment on retry; match-state transitions remain owned by the no-show/result transactions. API/store tests cover authorization, validation, idempotency SQL, and audit wiring; live PostgreSQL delivery remains an integration gate. diff --git a/server/api/service.go b/server/api/service.go index 86041dde..3218ea55 100644 --- a/server/api/service.go +++ b/server/api/service.go @@ -40,6 +40,9 @@ type ResultSubmitter interface { type ServerRegistrar interface { RegisterServer(context.Context, domain.WorkloadBinding, int, bool, string, time.Time) error } +type ServerShutdowner interface { + ShutdownServer(context.Context, domain.WorkloadBinding, string, string, time.Time) error +} type QueueBackend interface { Create(context.Context, string, string, string, domain.QueueSpec, time.Time) (domain.QueueTicket, error) @@ -113,6 +116,7 @@ type Service struct { WorkloadVerify WorkloadVerifier ResultSubmitter ResultSubmitter ServerRegistrar ServerRegistrar + ServerShutdowner ServerShutdowner Assignment AssignmentProvider Roster RosterProvider Now func() time.Time @@ -432,7 +436,8 @@ 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 + // (/servers/{serverId}/result, /servers/{serverId}/register, or + // /servers/{serverId}/shutdown) — 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/") @@ -464,7 +469,7 @@ type serverRegistrationRequest struct { func (s *Service) serverMutation(w http.ResponseWriter, r *http.Request) { parts := strings.Split(strings.TrimPrefix(r.URL.Path, "/v1/servers/"), "/") - if len(parts) != 2 || parts[0] == "" || (parts[1] != "result" && parts[1] != "register" && parts[1] != "roster") { + if len(parts) != 2 || parts[0] == "" || (parts[1] != "result" && parts[1] != "register" && parts[1] != "roster" && parts[1] != "shutdown") { writeError(w, http.StatusNotFound, "not_found") return } @@ -472,7 +477,7 @@ func (s *Service) serverMutation(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusMethodNotAllowed, "method_not_allowed") return } - if s.WorkloadVerify == nil || (parts[1] == "result" && s.ResultSubmitter == nil) || (parts[1] == "register" && s.ServerRegistrar == nil) || (parts[1] == "roster" && s.Roster == nil) { + if s.WorkloadVerify == nil || (parts[1] == "result" && s.ResultSubmitter == nil) || (parts[1] == "register" && s.ServerRegistrar == nil) || (parts[1] == "roster" && s.Roster == nil) || (parts[1] == "shutdown" && s.ServerShutdowner == nil) { writeError(w, http.StatusServiceUnavailable, "server_unavailable") return } @@ -538,6 +543,33 @@ func (s *Service) serverMutation(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusNoContent) return } + if parts[1] == "shutdown" { + var input struct { + Reason string `json:"reason"` + } + if !decodeBody(w, r, &input) { + return + } + if input.Reason == "" || len(input.Reason) > 96 || strings.ContainsAny(input.Reason, "\r\n\t") { + s.logEvent(observability.Event{Event: "server_shutdown", MatchID: binding.MatchID, ServerID: parts[0], Stage: "rejected", OccurredAt: now}) + writeError(w, http.StatusUnprocessableEntity, "invalid_request") + return + } + if err := s.ServerShutdowner.ShutdownServer(r.Context(), binding, input.Reason, key, now); err != nil { + stage := "invalid" + if errors.Is(err, domain.ErrConflict) { + stage = "conflict" + writeError(w, http.StatusConflict, "conflict") + } else { + writeError(w, http.StatusUnprocessableEntity, "invalid_request") + } + s.logEvent(observability.Event{Event: "server_shutdown", MatchID: binding.MatchID, ServerID: parts[0], Stage: stage, OccurredAt: now}) + return + } + s.logEvent(observability.Event{Event: "server_shutdown", MatchID: binding.MatchID, ServerID: parts[0], Stage: "acknowledged", OccurredAt: now}) + w.WriteHeader(http.StatusNoContent) + return + } var input resultRequest if !decodeBody(w, r, &input) { return diff --git a/server/api/service_test.go b/server/api/service_test.go index c3f95206..5d10a0b4 100644 --- a/server/api/service_test.go +++ b/server/api/service_test.go @@ -52,6 +52,20 @@ type serverRegistrarSpy struct { err error } +type serverShutdownerSpy struct { + calls int + binding domain.WorkloadBinding + reason string + key string + err error +} + +func (s *serverShutdownerSpy) ShutdownServer(_ context.Context, binding domain.WorkloadBinding, reason, key string, _ time.Time) error { + s.calls++ + s.binding, s.reason, s.key = binding, reason, key + return s.err +} + 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 @@ -1358,6 +1372,40 @@ func TestServerRegistrationAPIRequiresBoundWorkloadAndValidDigest(t *testing.T) response.Body.Close() } +func TestServerShutdownAPIRequiresBoundWorkloadAndDelegatesAcknowledgement(t *testing.T) { + now := time.Unix(1000, 0).UTC() + binding := domain.WorkloadBinding{AllocationID: "allocation-1", MatchID: "match-1", ServerID: "server-1"} + shutdowner := &serverShutdownerSpy{} + service := &Service{Now: func() time.Time { return now }, WorkloadVerify: func(token string, at time.Time) (domain.WorkloadBinding, error) { + if token != "workload-token" || !at.Equal(now) { + t.Fatalf("verifier input=%q %v", token, at) + } + return binding, nil + }, ServerShutdowner: shutdowner} + server := httptest.NewServer(service.Handler()) + defer server.Close() + req, _ := http.NewRequest(http.MethodPost, server.URL+"/v1/servers/server-1/shutdown", strings.NewReader(`{"reason":"server_draining"}`)) + req.Header.Set("Authorization", "Bearer workload-token") + req.Header.Set("Idempotency-Key", "shutdown-key-123456") + response, err := http.DefaultClient.Do(req) + if err != nil || response.StatusCode != http.StatusNoContent { + t.Fatalf("status=%v err=%v", response.StatusCode, err) + } + response.Body.Close() + if shutdowner.calls != 1 || shutdowner.binding != binding || shutdowner.reason != "server_draining" || shutdowner.key != "shutdown-key-123456" { + t.Fatalf("shutdown=%+v", shutdowner) + } + + req, _ = http.NewRequest(http.MethodPost, server.URL+"/v1/servers/server-1/shutdown", strings.NewReader(`{"reason":"bad\nreason"}`)) + req.Header.Set("Authorization", "Bearer workload-token") + req.Header.Set("Idempotency-Key", "shutdown-key-123456") + response, err = http.DefaultClient.Do(req) + if err != nil || response.StatusCode != http.StatusUnprocessableEntity || shutdowner.calls != 1 { + t.Fatalf("invalid shutdown status=%v err=%v calls=%d", response.StatusCode, err, shutdowner.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 ba2e6292..51052e82 100644 --- a/server/api/store_adapters.go +++ b/server/api/store_adapters.go @@ -74,6 +74,19 @@ func ServerRegistrarFromStore(db *sql.DB) ServerRegistrar { return postgresServerRegistrar{db: db} } +type postgresServerShutdowner struct{ db *sql.DB } + +func (p postgresServerShutdowner) ShutdownServer(ctx context.Context, binding domain.WorkloadBinding, reason, idempotencyKey string, now time.Time) error { + return store.RecordServerShutdown(ctx, p.db, binding, reason, idempotencyKey, now) +} + +func ServerShutdownerFromStore(db *sql.DB) ServerShutdowner { + if db == nil { + return nil + } + return postgresServerShutdowner{db: db} +} + // WorkloadVerifierFromSignedToken builds WorkloadVerify from a control-plane // -owned signed token instead of a Kubernetes-projected JWT (see // workload/signed_token.go for why: it needs no live cluster to verify). diff --git a/server/cmd/control-plane/main.go b/server/cmd/control-plane/main.go index f39ff868..59e289e4 100644 --- a/server/cmd/control-plane/main.go +++ b/server/cmd/control-plane/main.go @@ -101,6 +101,7 @@ func newAPIService(db *sql.DB, workloadSecret string, indexes ...api.CandidateIn ProposalBackend: api.ProposalProviderFromStore(db), ProposalPromoter: api.ProposalPromoterFromStore(db), ServerRegistrar: api.ServerRegistrarFromStore(db), + ServerShutdowner: api.ServerShutdownerFromStore(db), ResultSubmitter: store.PostgresResults{DB: db}, RankedProfileProvider: store.PostgresRankedProfiles{DB: db}, TierPolicy: domain.DefaultTierPolicy(), diff --git a/server/cmd/testkit-api/main.go b/server/cmd/testkit-api/main.go index bf777a39..1e51a7f3 100644 --- a/server/cmd/testkit-api/main.go +++ b/server/cmd/testkit-api/main.go @@ -62,6 +62,7 @@ func main() { ProposalBackend: api.ProposalProviderFromStore(db), ProposalPromoter: api.ProposalPromoterFromStore(db), ServerRegistrar: api.ServerRegistrarFromStore(db), + ServerShutdowner: api.ServerShutdownerFromStore(db), ResultSubmitter: store.PostgresResults{DB: db}, RankedProfileProvider: store.PostgresRankedProfiles{DB: db}, TierPolicy: domain.DefaultTierPolicy(), diff --git a/server/contracts/v1/openapi.json b/server/contracts/v1/openapi.json index 9ecb8429..f3b185f0 100644 --- a/server/contracts/v1/openapi.json +++ b/server/contracts/v1/openapi.json @@ -49,6 +49,9 @@ }, "/servers/{serverId}/result": { "post": {"security": [{"serverCredential": []}], "operationId": "submitMatchResult", "parameters": [{"$ref": "#/components/parameters/ServerId"}, {"$ref": "#/components/parameters/IdempotencyKey"}], "requestBody": {"required": true, "content": {"application/json": {"schema": {"$ref": "#/components/schemas/MatchResult"}}}}, "responses": {"202": {"description": "Result accepted"}, "409": {"$ref": "#/components/responses/Conflict"}, "422": {"$ref": "#/components/responses/Invalid"}}} + }, + "/servers/{serverId}/shutdown": { + "post": {"security": [{"serverCredential": []}], "operationId": "acknowledgeServerShutdown", "parameters": [{"$ref": "#/components/parameters/ServerId"}, {"$ref": "#/components/parameters/IdempotencyKey"}], "requestBody": {"required": true, "content": {"application/json": {"schema": {"$ref": "#/components/schemas/ServerShutdown"}}}}, "responses": {"204": {"description": "Shutdown acknowledged"}, "409": {"$ref": "#/components/responses/Conflict"}, "422": {"$ref": "#/components/responses/Invalid"}}} } }, "components": { @@ -83,6 +86,7 @@ "Proposal": {"type": "object", "required": ["proposal_id", "revision", "state", "expires_at", "participants"], "additionalProperties": false, "properties": {"proposal_id": {"$ref": "#/components/schemas/OpaqueId"}, "revision": {"type": "integer", "minimum": 0}, "state": {"type": "string", "enum": ["OPEN", "ACCEPTED", "DECLINED", "EXPIRED", "CANCELLED"]}, "expires_at": {"type": "string", "format": "date-time"}, "participants": {"type": "array", "minItems": 2, "items": {"$ref": "#/components/schemas/OpaqueId"}}}}, "Assignment": {"type": "object", "required": ["match_id", "server_id", "player_id", "slot", "expires_at", "protocol_version", "transport", "endpoint", "join_authorisation"], "additionalProperties": false, "properties": {"match_id": {"$ref": "#/components/schemas/OpaqueId"}, "server_id": {"$ref": "#/components/schemas/OpaqueId"}, "player_id": {"$ref": "#/components/schemas/OpaqueId"}, "slot": {"type": "integer", "minimum": 0, "maximum": 5}, "expires_at": {"type": "string", "format": "date-time"}, "protocol_version": {"type": "integer", "minimum": 1}, "transport": {"type": "string", "enum": ["steam_sdr", "enet"]}, "endpoint": {"type": "string", "minLength": 3, "maxLength": 256}, "join_authorisation": {"type": "string"}}}, "ServerRegistration": {"type": "object", "required": ["match_id", "protocol_version", "image_digest", "assignment_ready"], "additionalProperties": false, "properties": {"match_id": {"$ref": "#/components/schemas/OpaqueId"}, "protocol_version": {"type": "integer", "minimum": 1}, "image_digest": {"type": "string", "pattern": "^sha256:[a-f0-9]{64}$"}, "assignment_ready": {"type": "boolean"}}}, + "ServerShutdown": {"type": "object", "required": ["reason"], "additionalProperties": false, "properties": {"reason": {"type": "string", "minLength": 1, "maxLength": 96}}}, "MatchResult": {"type": "object", "required": ["match_id", "result_nonce", "score", "integrity_state"], "additionalProperties": false, "properties": {"match_id": {"$ref": "#/components/schemas/OpaqueId"}, "result_nonce": {"type": "string", "minLength": 16, "maxLength": 128}, "score": {"type": "object", "required": ["team_0", "team_1"], "additionalProperties": false, "properties": {"team_0": {"type": "integer", "minimum": 0}, "team_1": {"type": "integer", "minimum": 0}}}, "integrity_state": {"type": "string", "enum": ["CERTIFIED", "SUPPRESSED", "REVIEW"]}}} } } diff --git a/server/store/server_shutdown_sql.go b/server/store/server_shutdown_sql.go new file mode 100644 index 00000000..38322c9c --- /dev/null +++ b/server/store/server_shutdown_sql.go @@ -0,0 +1,77 @@ +package store + +import ( + "bytes" + "context" + "crypto/sha256" + "database/sql" + "encoding/json" + "fmt" + "time" + + "github.com/cosmic-clash/cosmic-clash/server/domain" +) + +const ServerShutdownIdempotencyScope = "server.shutdown" + +const ServerShutdownIdempotencyInsertSQL = `INSERT INTO idempotency_keys + (scope, idempotency_key, payload_digest, result) +VALUES ($1, $2, $3, '{}'::jsonb) ON CONFLICT (scope, idempotency_key) DO NOTHING` + +const ServerShutdownIdempotencySelectSQL = `SELECT payload_digest +FROM idempotency_keys WHERE scope = $1 AND idempotency_key = $2 FOR UPDATE` + +const ServerShutdownMatchLockSQL = `SELECT match_id +FROM matches WHERE match_id = $1 AND server_id = $2 FOR UPDATE` + +const ServerShutdownAuditSQL = `INSERT INTO audit_events + (actor_type, actor_id, action, aggregate_type, aggregate_id, request_id, metadata) +VALUES ('SERVER', $1, 'SERVER_SHUTDOWN', 'match', $2, $3, $4)` + +// RecordServerShutdown acknowledges a workload-authenticated server's planned +// termination without guessing a match-state transition. Result/no-show +// transactions own those transitions; this boundary records the server's +// lifecycle signal exactly once and is safe to retry. +func RecordServerShutdown(ctx context.Context, db *sql.DB, binding domain.WorkloadBinding, reason, idempotencyKey string, now time.Time) error { + if db == nil || binding.MatchID == "" || binding.ServerID == "" || reason == "" || len(reason) > 96 || len(idempotencyKey) < 16 || len(idempotencyKey) > 128 || now.IsZero() { + return fmt.Errorf("invalid server shutdown") + } + digest := sha256.Sum256([]byte(fmt.Sprintf("%s\x00%s\x00%s", binding.MatchID, binding.ServerID, reason))) + return RunSerializable(ctx, db, DefaultSerializableAttempts, func(ctx context.Context, tx *sql.Tx) error { + inserted, err := tx.ExecContext(ctx, ServerShutdownIdempotencyInsertSQL, ServerShutdownIdempotencyScope, 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, ServerShutdownIdempotencySelectSQL, ServerShutdownIdempotencyScope, idempotencyKey).Scan(&prior); err != nil { + return err + } + if !bytes.Equal(prior, digest[:]) { + return domain.ErrConflict + } + return nil + } + var matchID string + if err := tx.QueryRowContext(ctx, ServerShutdownMatchLockSQL, binding.MatchID, binding.ServerID).Scan(&matchID); err != nil { + return err + } + metadata, err := json.Marshal(map[string]string{"reason": reason}) + if err != nil { + return err + } + if _, err := tx.ExecContext(ctx, ServerShutdownAuditSQL, binding.ServerID, matchID, idempotencyKey, metadata); err != nil { + return err + } + result, err := json.Marshal(map[string]string{"match_id": matchID, "status": "acknowledged"}) + if err != nil { + return err + } + _, err = tx.ExecContext(ctx, `UPDATE idempotency_keys SET result = $3 WHERE scope = $1 AND idempotency_key = $2`, ServerShutdownIdempotencyScope, idempotencyKey, result) + return err + }) +} diff --git a/server/store/server_shutdown_sql_test.go b/server/store/server_shutdown_sql_test.go new file mode 100644 index 00000000..76f5a157 --- /dev/null +++ b/server/store/server_shutdown_sql_test.go @@ -0,0 +1,43 @@ +package store + +import ( + "context" + "database/sql" + "strings" + "testing" + "time" + + "github.com/cosmic-clash/cosmic-clash/server/domain" +) + +func TestServerShutdownSQLUsesIdempotencyLockAndAudit(t *testing.T) { + checks := map[string][]string{ + "insert": {"idempotency_keys", "ON CONFLICT (scope, idempotency_key) DO NOTHING", "payload_digest"}, + "select": {"scope = $1", "idempotency_key = $2", "FOR UPDATE"}, + "match": {"match_id = $1", "server_id = $2", "FOR UPDATE"}, + "audit": {"SERVER_SHUTDOWN", "request_id", "metadata"}, + } + queries := map[string]string{"insert": ServerShutdownIdempotencyInsertSQL, "select": ServerShutdownIdempotencySelectSQL, "match": ServerShutdownMatchLockSQL, "audit": ServerShutdownAuditSQL} + for name, fragments := range checks { + for _, fragment := range fragments { + if !strings.Contains(queries[name], fragment) { + t.Fatalf("%s query missing %q: %s", name, fragment, queries[name]) + } + } + } +} + +func TestRecordServerShutdownRejectsInvalidArguments(t *testing.T) { + binding := domain.WorkloadBinding{MatchID: "match-1", ServerID: "server-1"} + now := time.Unix(1000, 0).UTC() + for name, values := range map[string][2]string{ + "missing reason": {"", "shutdown-key-123456"}, + "short key": {"planned", "short"}, + } { + t.Run(name, func(t *testing.T) { + if err := RecordServerShutdown(context.Background(), (*sql.DB)(nil), binding, values[0], values[1], now); err == nil { + t.Fatal("expected validation error") + } + }) + } +}