diff --git a/multiplayer-todo.md b/multiplayer-todo.md index a42e5b20..7d985404 100644 --- a/multiplayer-todo.md +++ b/multiplayer-todo.md @@ -1191,7 +1191,7 @@ the local/CI/community transport, not a silent production fallback. | # | Task | Acceptance | |---|---|---| -| 8.14 `[D:8.4,8.5,8.8]` | **IN PROGRESS.** Pure Go queue domain enforces one active ticket per verified player, 10 s heartbeat/30 s expiry, retry-safe create/heartbeat/cancel and deterministic candidate projection; store layer adds a rebuildable candidate-cache boundary | `server/domain/queue.go` and `server/store/candidates.go` cover ownership/expiry/idempotency, deterministic projection, cache loss and atomic rebuild; PostgreSQL row adapter, real Redis index/TTLs and restart/failover integration remain | +| 8.14 `[D:8.4,8.5,8.8]` | **IN PROGRESS.** Pure Go queue domain enforces one active ticket per verified player under concurrent mutation, 10 s heartbeat/30 s expiry, retry-safe create/heartbeat/cancel and deterministic candidate projection; store layer adds a rebuildable candidate-cache boundary | `server/domain/queue.go` and `server/store/candidates.go` cover ownership/expiry/idempotency, concurrent create fencing, deterministic projection, cache loss and atomic rebuild; PostgreSQL row adapter, real Redis index/TTLs and restart/failover integration remain | | 8.15 `[D:7.8,8.3]` | **IN PROGRESS.** Pure Go probe validation treats Steam location as opaque, requires nonce/freshness/region and server-computed RTT, and implements discrepancy quarantine/release | `server/domain/probes.go` and adversarial fixtures cover stale/wrong/forged evidence, the 25 ms/30% threshold, three-sample quarantine and five-clean release; Steam coordinator and regional probe adapters remain | | 8.16 `[D:8.14,8.15]` | **IN PROGRESS.** Pure Go candidate/team selection implements the <=100 ms ceiling, pairwise widening tolerance, anchor inclusion, deterministic set/region scoring and balanced team partitioning | `server/domain/matcher.go`, `teams.go` and adversarial fixtures cover no-common-region, tolerance boundaries, lexical ties, mean-rating balance and input-order independence; queue-backed candidate loading and full population fixtures remain | | 8.17 `[D:8.14,8.16]` | **IN PROGRESS.** Pure Go proposal policy sends a 10-second response window to every selected human, requires unanimous acceptance, applies exact decline/timeout cooldowns and ranked escalation | `server/domain/proposal.go` and adversarial fixtures cover partial/unanimous response, expiry, replay/conflict and ranked six-player/cooldown rules; casual 6→2 composition, queue precedence and allocation integration remain | diff --git a/server/domain/queue.go b/server/domain/queue.go index a2fdf699..81fd272e 100644 --- a/server/domain/queue.go +++ b/server/domain/queue.go @@ -6,6 +6,7 @@ import ( "fmt" "sort" "strings" + "sync" "time" ) @@ -37,6 +38,7 @@ type queueMutation struct { } type Queue struct { + mu sync.Mutex tickets map[string]QueueTicket byPlayer map[string]string mutations map[string]queueMutation @@ -50,6 +52,8 @@ func NewQueue() *Queue { // production adapter must perform the same check in one transaction and use // the same idempotency semantics. func (q *Queue) Create(playerID, ticketID, idempotencyKey string, candidate Candidate, now time.Time) (QueueTicket, error) { + q.mu.Lock() + defer q.mu.Unlock() digest := sha256.Sum256([]byte(createPayload(playerID, ticketID, candidate))) if prior, ok := q.mutations[idempotencyKey]; ok { if prior.digest != digest { @@ -74,6 +78,8 @@ func (q *Queue) Create(playerID, ticketID, idempotencyKey string, candidate Cand } func (q *Queue) Heartbeat(playerID, ticketID, idempotencyKey string, expectedRevision uint64, now time.Time) (QueueTicket, error) { + q.mu.Lock() + defer q.mu.Unlock() digest := sha256.Sum256([]byte(fmt.Sprintf("heartbeat:%s:%d", ticketID, expectedRevision))) if prior, ok := q.mutations[idempotencyKey]; ok { if prior.digest != digest { @@ -105,6 +111,8 @@ func (q *Queue) Heartbeat(playerID, ticketID, idempotencyKey string, expectedRev } func (q *Queue) Cancel(playerID, ticketID, idempotencyKey string, expectedRevision uint64, now time.Time) (QueueTicket, error) { + q.mu.Lock() + defer q.mu.Unlock() digest := sha256.Sum256([]byte(fmt.Sprintf("cancel:%s:%d", ticketID, expectedRevision))) if prior, ok := q.mutations[idempotencyKey]; ok { if prior.digest != digest { @@ -132,6 +140,12 @@ func (q *Queue) Cancel(playerID, ticketID, idempotencyKey string, expectedRevisi } func (q *Queue) Expire(now time.Time) []QueueTicket { + q.mu.Lock() + defer q.mu.Unlock() + return q.expireLocked(now) +} + +func (q *Queue) expireLocked(now time.Time) []QueueTicket { var expired []QueueTicket for id, ticket := range q.tickets { if (ticket.State == Queued || ticket.State == Proposed) && !now.Before(ticket.ExpiresAt) { @@ -147,7 +161,9 @@ func (q *Queue) Expire(now time.Time) []QueueTicket { } func (q *Queue) Candidates(now time.Time) []Candidate { - q.Expire(now) + q.mu.Lock() + defer q.mu.Unlock() + q.expireLocked(now) result := make([]Candidate, 0) for _, ticket := range q.tickets { if ticket.State == Queued { diff --git a/server/domain/queue_test.go b/server/domain/queue_test.go index 4b8e2896..5e5e770f 100644 --- a/server/domain/queue_test.go +++ b/server/domain/queue_test.go @@ -3,6 +3,7 @@ package domain import ( "errors" "reflect" + "sync" "testing" "time" ) @@ -77,3 +78,32 @@ func TestQueueCreateIdempotencyIncludesCandidatePayload(t *testing.T) { t.Fatalf("changed create payload error = %v", err) } } + +func TestQueueConcurrentCreateKeepsOneActiveTicketPerPlayer(t *testing.T) { + q := NewQueue() + now := time.Unix(1000, 0) + var wg sync.WaitGroup + results := make(chan error, 2) + for i := 0; i < 2; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + id := string(rune('a' + i)) + _, err := q.Create("same-player", "ticket-"+id, "create-"+id+"-123456", Candidate{TicketID: "ticket-" + id, PlayerID: "same-player", EnqueuedAt: now}, now) + results <- err + }(i) + } + wg.Wait() + close(results) + succeeded := 0 + for err := range results { + if err == nil { + succeeded++ + } else if !errors.Is(err, ErrPlayerQueued) { + t.Fatalf("unexpected concurrent create error: %v", err) + } + } + if succeeded != 1 { + t.Fatalf("concurrent creates succeeded %d times", succeeded) + } +}