mirror of
https://github.com/jcreek/CosmicClash.git
synced 2026-09-14 13:22:03 +00:00
fix(multiplayer): repair allocated compose verification
This commit is contained in:
+16
-1
@@ -177,7 +177,22 @@ func RatingEligible(receipt ResultReceipt) bool {
|
||||
}
|
||||
|
||||
func validateBinding(binding WorkloadBinding) error {
|
||||
if binding.Issuer == "" || binding.Audience == "" || binding.Namespace == "" || binding.ServiceAcct == "" || binding.PodUID == "" || binding.GameServerUID == "" || binding.AllocationID == "" || binding.MatchID == "" || binding.ServerID == "" {
|
||||
// A Kubernetes JWT supplies the six workload-identity fields below, while
|
||||
// the signed workload credential is deliberately bound through the durable
|
||||
// allocation record and therefore supplies only allocation/match/server.
|
||||
// Accept either complete authority model, but never a partial Kubernetes
|
||||
// identity that could accidentally look authenticated.
|
||||
if binding.AllocationID == "" || binding.MatchID == "" || binding.ServerID == "" {
|
||||
return ErrResultBinding
|
||||
}
|
||||
kubernetesIdentity := []string{binding.Issuer, binding.Audience, binding.Namespace, binding.ServiceAcct, binding.PodUID, binding.GameServerUID}
|
||||
present := 0
|
||||
for _, value := range kubernetesIdentity {
|
||||
if value != "" {
|
||||
present++
|
||||
}
|
||||
}
|
||||
if present != 0 && present != len(kubernetesIdentity) {
|
||||
return ErrResultBinding
|
||||
}
|
||||
return nil
|
||||
|
||||
@@ -47,6 +47,20 @@ func TestResultStoreRejectsMissingAuthoritativeTime(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestResultStoreAcceptsDurablyBoundSignedWorkloadIdentity(t *testing.T) {
|
||||
// Signed workload tokens resolve this three-part binding from the durable
|
||||
// allocation record; they intentionally carry no Kubernetes JWT claims.
|
||||
binding := WorkloadBinding{AllocationID: "allocation-1", MatchID: "match-1", ServerID: "server-1"}
|
||||
if _, err := NewResultStore(binding); err != nil {
|
||||
t.Fatalf("signed workload binding rejected: %v", err)
|
||||
}
|
||||
partial := binding
|
||||
partial.Issuer = "https://issuer"
|
||||
if _, err := NewResultStore(partial); !errors.Is(err, ErrResultBinding) {
|
||||
t.Fatalf("partial Kubernetes identity error = %v, want ErrResultBinding", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestConflictingResultIsInertAndIntegritySuppressesRating(t *testing.T) {
|
||||
now := time.Unix(1000, 0)
|
||||
binding := testBinding()
|
||||
|
||||
@@ -15,6 +15,32 @@ const migrationTableSQL = `CREATE TABLE IF NOT EXISTS schema_migrations (
|
||||
applied_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
)`
|
||||
|
||||
const migrationLockSQL = `SELECT pg_advisory_xact_lock(hashtext('cosmic-clash:migrations'))`
|
||||
|
||||
// ensureMigrationTable serializes the bootstrap DDL itself. PostgreSQL's
|
||||
// CREATE TABLE IF NOT EXISTS is not safe against concurrent first creation:
|
||||
// the relation-type catalog entry can still collide before either statement
|
||||
// observes the other table. Every long-lived role calls Apply at startup, so
|
||||
// take the same transaction-scoped advisory lock used for individual files
|
||||
// before issuing the bootstrap statement.
|
||||
func ensureMigrationTable(ctx context.Context, db *sql.DB) error {
|
||||
tx, err := db.BeginTx(ctx, nil)
|
||||
if err != nil {
|
||||
return fmt.Errorf("begin migration bootstrap: %w", err)
|
||||
}
|
||||
defer tx.Rollback()
|
||||
if _, err := tx.ExecContext(ctx, migrationLockSQL); err != nil {
|
||||
return fmt.Errorf("lock migration bootstrap: %w", err)
|
||||
}
|
||||
if _, err := tx.ExecContext(ctx, migrationTableSQL); err != nil {
|
||||
return fmt.Errorf("create migration table: %w", err)
|
||||
}
|
||||
if err := tx.Commit(); err != nil {
|
||||
return fmt.Errorf("commit migration bootstrap: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Apply executes numbered SQL files in lexical order. A transaction-level
|
||||
// advisory lock serializes concurrent API/worker starts, while each migration
|
||||
// is committed together with its schema_migrations marker so a failed
|
||||
@@ -31,8 +57,8 @@ func Apply(ctx context.Context, db *sql.DB, directory string) error {
|
||||
if len(paths) == 0 {
|
||||
return fmt.Errorf("no migrations found in %s", directory)
|
||||
}
|
||||
if _, err := db.ExecContext(ctx, migrationTableSQL); err != nil {
|
||||
return fmt.Errorf("create migration table: %w", err)
|
||||
if err := ensureMigrationTable(ctx, db); err != nil {
|
||||
return err
|
||||
}
|
||||
for _, path := range paths {
|
||||
version := filepath.Base(path)
|
||||
@@ -50,7 +76,7 @@ func Apply(ctx context.Context, db *sql.DB, directory string) error {
|
||||
_ = tx.Rollback()
|
||||
}
|
||||
}()
|
||||
if _, err := tx.ExecContext(ctx, `SELECT pg_advisory_xact_lock(hashtext('cosmic-clash:migrations'))`); err != nil {
|
||||
if _, err := tx.ExecContext(ctx, migrationLockSQL); err != nil {
|
||||
return fmt.Errorf("lock migration %s: %w", version, err)
|
||||
}
|
||||
var applied bool
|
||||
@@ -86,8 +112,8 @@ func Rollback(ctx context.Context, db *sql.DB, directory string, steps int) erro
|
||||
if db == nil || strings.TrimSpace(directory) == "" || steps <= 0 {
|
||||
return fmt.Errorf("database, migration directory and a positive step count are required")
|
||||
}
|
||||
if _, err := db.ExecContext(ctx, migrationTableSQL); err != nil {
|
||||
return fmt.Errorf("create migration table: %w", err)
|
||||
if err := ensureMigrationTable(ctx, db); err != nil {
|
||||
return err
|
||||
}
|
||||
rows, err := db.QueryContext(ctx, `SELECT version FROM schema_migrations ORDER BY version DESC LIMIT $1`, steps)
|
||||
if err != nil {
|
||||
@@ -123,7 +149,7 @@ func Rollback(ctx context.Context, db *sql.DB, directory string, steps int) erro
|
||||
_ = tx.Rollback()
|
||||
}
|
||||
}()
|
||||
if _, err := tx.ExecContext(ctx, `SELECT pg_advisory_xact_lock(hashtext('cosmic-clash:migrations'))`); err != nil {
|
||||
if _, err := tx.ExecContext(ctx, migrationLockSQL); err != nil {
|
||||
return fmt.Errorf("lock rollback %s: %w", version, err)
|
||||
}
|
||||
var applied bool
|
||||
|
||||
@@ -22,10 +22,10 @@ class ComposeManifestTest(unittest.TestCase):
|
||||
runner = (ROOT / "scripts/verify_allocated_compose.sh").read_text()
|
||||
allocated = (ROOT / "compose.allocated-smoke.yml").read_text()
|
||||
for marker in (
|
||||
"/v1/servers/compose-server/result",
|
||||
"/v1/servers/compose-server-0001/result",
|
||||
"compose-result-key-123456",
|
||||
"result_receipts",
|
||||
"/v1/servers/compose-server/shutdown",
|
||||
"/v1/servers/compose-server-0001/shutdown",
|
||||
"SERVER_SHUTDOWN",
|
||||
"/v1/session/steam",
|
||||
"compose-queue-key-123456",
|
||||
|
||||
@@ -81,9 +81,14 @@ func PromoteStoredAcceptedProposal(ctx context.Context, db *sql.DB, proposalID s
|
||||
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, &plan.ArenaPath); err != nil {
|
||||
// Casual proposals intentionally persist no arena path. Scan it as nullable
|
||||
// here just as the in-transaction promotion path does, so an API retry after
|
||||
// the atomic promotion does not turn a successful acceptance into a 503.
|
||||
var arenaPath sql.NullString
|
||||
if err := db.QueryRowContext(ctx, StoredProposalMatchPlanSQL, proposalID).Scan(&plan.Region, &plan.Protocol, &arenaPath); err != nil {
|
||||
return err
|
||||
}
|
||||
plan.ArenaPath = arenaPath.String
|
||||
rows, err := db.QueryContext(ctx, StoredProposalMatchPlayersSQL, proposalID)
|
||||
if err != nil {
|
||||
return err
|
||||
|
||||
@@ -778,6 +778,12 @@ func TestPostgreSQLProposalClaimAndResponseAreAtomic(t *testing.T) {
|
||||
if accepted.State != domain.Accepted || accepted.Revision != 2 {
|
||||
t.Fatalf("proposal did not close after unanimous acceptance: %+v", accepted)
|
||||
}
|
||||
// The API's post-commit recovery promoter must accept the nullable casual
|
||||
// arena path left by the atomic response transaction and converge on the
|
||||
// already-created match.
|
||||
if err := PromoteStoredAcceptedProposal(ctx, db, proposal.ProposalID, now.Add(time.Second)); err != nil {
|
||||
t.Fatalf("replay persisted casual promotion: %v", err)
|
||||
}
|
||||
var matchState string
|
||||
if err := db.QueryRow(`SELECT state FROM matches WHERE match_id = 'match-proposal-integration'`).Scan(&matchState); err != nil {
|
||||
t.Fatalf("atomic accepted match: %v", err)
|
||||
|
||||
@@ -226,6 +226,12 @@ func (s *Supervisor) Start(ctx context.Context) error {
|
||||
s.cmd = exec.CommandContext(ctx, s.config.Command[0], s.config.Command[1:]...)
|
||||
}
|
||||
s.cmd.Env = env
|
||||
// A server's structured stdout/stderr is its operational interface. The
|
||||
// zero value for exec.Cmd streams is /dev/null, which would make a child
|
||||
// startup failure invisible to Docker, Kubernetes, and the Compose
|
||||
// readiness harness while the supervisor can report only "exit status 1".
|
||||
s.cmd.Stdout = os.Stdout
|
||||
s.cmd.Stderr = os.Stderr
|
||||
if err := s.cmd.Start(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user