diff --git a/server/cmd/matcher/main.go b/server/cmd/matcher/main.go index f8ba5fbe..3ca4a295 100644 --- a/server/cmd/matcher/main.go +++ b/server/cmd/matcher/main.go @@ -46,8 +46,8 @@ func main() { } now := func() time.Time { return time.Now().UTC() } worker := matcher.Worker{ - Source: func(ctx context.Context, at time.Time, limit int) ([]domain.Candidate, error) { - return store.ListQueuedCandidates(ctx, db, at, limit) + Source: func(ctx context.Context, at time.Time, playlist domain.Playlist, limit int) ([]domain.Candidate, error) { + return store.ListQueuedCandidates(ctx, db, playlist, at, limit) }, Creator: matcher.ProposalCreatorFunc(func(ctx context.Context, proposal domain.Proposal, ticketIDs map[string]string, at time.Time) error { return store.CreateProposal(ctx, db, proposal, ticketIDs, at) diff --git a/server/matcher/worker.go b/server/matcher/worker.go index 5cf526d3..46259ea1 100644 --- a/server/matcher/worker.go +++ b/server/matcher/worker.go @@ -10,7 +10,7 @@ import ( "github.com/cosmic-clash/cosmic-clash/server/domain" ) -type CandidateSource func(context.Context, time.Time, int) ([]domain.Candidate, error) +type CandidateSource func(context.Context, time.Time, domain.Playlist, int) ([]domain.Candidate, error) type ProposalCreator interface { CreateProposal(context.Context, domain.Proposal, map[string]string, time.Time) error @@ -68,7 +68,7 @@ func (w Worker) RunOnce(ctx context.Context) (bool, error) { return false, fmt.Errorf("invalid matcher size") } now := w.Now() - candidates, err := w.Source(ctx, now, w.Size) + candidates, err := w.Source(ctx, now, w.Playlist, w.Size) if err != nil { return false, err } @@ -77,6 +77,9 @@ func (w Worker) RunOnce(ctx context.Context) (bool, error) { } queue := domain.NewQueue() for _, candidate := range candidates { + if candidate.Playlist != w.Playlist { + return false, fmt.Errorf("candidate playlist does not match worker") + } if _, err := queue.Create(candidate.PlayerID, candidate.TicketID, "matcher-"+candidate.TicketID, candidate, now); err != nil { return false, err } diff --git a/server/matcher/worker_test.go b/server/matcher/worker_test.go index a4f6b432..309cd7c4 100644 --- a/server/matcher/worker_test.go +++ b/server/matcher/worker_test.go @@ -40,7 +40,9 @@ func workerFor(source CandidateSource, creator ProposalCreator) Worker { func TestRunOnceDelegatesFinalClaimAndBindsTickets(t *testing.T) { creator := &creatorSpy{} - worker := workerFor(func(context.Context, time.Time, int) ([]domain.Candidate, error) { return candidates(), nil }, creator) + worker := workerFor(func(context.Context, time.Time, domain.Playlist, int) ([]domain.Candidate, error) { + return candidates(), nil + }, creator) formed, err := worker.RunOnce(context.Background()) if err != nil || !formed || creator.calls != 1 { t.Fatalf("formed=%v err=%v calls=%d", formed, err, creator.calls) @@ -52,13 +54,15 @@ func TestRunOnceDelegatesFinalClaimAndBindsTickets(t *testing.T) { func TestRunOnceFailsClosedOnSourceOrDurableClaimFailure(t *testing.T) { creator := &creatorSpy{err: errors.New("serialization conflict")} - worker := workerFor(func(context.Context, time.Time, int) ([]domain.Candidate, error) { + worker := workerFor(func(context.Context, time.Time, domain.Playlist, int) ([]domain.Candidate, error) { return nil, errors.New("redis unavailable") }, creator) if _, err := worker.RunOnce(context.Background()); err == nil { t.Fatal("source failure was swallowed") } - worker.Source = func(context.Context, time.Time, int) ([]domain.Candidate, error) { return candidates(), nil } + worker.Source = func(context.Context, time.Time, domain.Playlist, int) ([]domain.Candidate, error) { + return candidates(), nil + } if _, err := worker.RunOnce(context.Background()); err == nil { t.Fatal("durable claim failure was swallowed") } @@ -69,9 +73,34 @@ func TestRunOnceFailsClosedOnSourceOrDurableClaimFailure(t *testing.T) { func TestRunOnceDoesNotClaimAnIncompleteBatch(t *testing.T) { creator := &creatorSpy{} - worker := workerFor(func(context.Context, time.Time, int) ([]domain.Candidate, error) { return candidates()[:3], nil }, creator) + worker := workerFor(func(context.Context, time.Time, domain.Playlist, int) ([]domain.Candidate, error) { + return candidates()[:3], nil + }, creator) formed, err := worker.RunOnce(context.Background()) if err != nil || formed || creator.calls != 0 { t.Fatalf("formed=%v err=%v calls=%d", formed, err, creator.calls) } } + +func TestRunOnceRejectsMixedPlaylistAndDuplicateIdentityBatches(t *testing.T) { + creator := &creatorSpy{} + worker := workerFor(func(_ context.Context, _ time.Time, _ domain.Playlist, _ int) ([]domain.Candidate, error) { + batch := candidates() + batch[1].Playlist = domain.Ranked + return batch, nil + }, creator) + if _, err := worker.RunOnce(context.Background()); err == nil { + t.Fatal("mixed playlist was accepted") + } + worker.Source = func(_ context.Context, _ time.Time, _ domain.Playlist, _ int) ([]domain.Candidate, error) { + batch := candidates() + batch[1].PlayerID = batch[0].PlayerID + return batch, nil + } + if _, err := worker.RunOnce(context.Background()); err == nil { + t.Fatal("duplicate identity was accepted") + } + if creator.calls != 0 { + t.Fatalf("creator calls=%d", creator.calls) + } +} diff --git a/server/store/queue_sql.go b/server/store/queue_sql.go index 19d83cb8..6af6efb7 100644 --- a/server/store/queue_sql.go +++ b/server/store/queue_sql.go @@ -40,18 +40,18 @@ RETURNING ticket_id, player_id, playlist, state, client_build, protocol_version, const QueueCandidateProjectionSQL = `SELECT ticket_id, player_id, playlist, client_build, protocol_version, enqueued_at FROM queue_tickets -WHERE state = 'QUEUED' AND expires_at > $1 +WHERE state = 'QUEUED' AND playlist = $1 AND expires_at > $2 ORDER BY enqueued_at, ticket_id -LIMIT $2` +LIMIT $3` // ListQueuedCandidates is an authoritative, expiry-filtered source for the // matcher projection. It deliberately does not claim rows; CreateProposal is // the transaction that performs the competing claim with SKIP LOCKED fences. -func ListQueuedCandidates(ctx context.Context, db *sql.DB, now time.Time, limit int) ([]domain.Candidate, error) { - if db == nil || now.IsZero() || limit < 1 || limit > 1000 { +func ListQueuedCandidates(ctx context.Context, db *sql.DB, playlist domain.Playlist, now time.Time, limit int) ([]domain.Candidate, error) { + if db == nil || (playlist != domain.Casual && playlist != domain.Ranked) || now.IsZero() || limit < 1 || limit > 1000 { return nil, fmt.Errorf("invalid queued candidate arguments") } - rows, err := db.QueryContext(ctx, QueueCandidateProjectionSQL, now, limit) + rows, err := db.QueryContext(ctx, QueueCandidateProjectionSQL, string(playlist), now, limit) if err != nil { return nil, err }