mirror of
https://github.com/jcreek/CosmicClash.git
synced 2026-09-11 08:23:45 +00:00
feat(multiplayer): expose server shutdown acknowledgement
This commit is contained in:
Josh Creek
+35
-3
@@ -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
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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).
|
||||
|
||||
@@ -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(),
|
||||
|
||||
@@ -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(),
|
||||
|
||||
@@ -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"]}}}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
})
|
||||
}
|
||||
@@ -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")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user