mirror of
https://github.com/jcreek/CosmicClash.git
synced 2026-09-13 12:12:03 +00:00
fix: bind matcher source to playlist
This commit is contained in:
@@ -46,8 +46,8 @@ func main() {
|
|||||||
}
|
}
|
||||||
now := func() time.Time { return time.Now().UTC() }
|
now := func() time.Time { return time.Now().UTC() }
|
||||||
worker := matcher.Worker{
|
worker := matcher.Worker{
|
||||||
Source: func(ctx context.Context, at time.Time, limit int) ([]domain.Candidate, error) {
|
Source: func(ctx context.Context, at time.Time, playlist domain.Playlist, limit int) ([]domain.Candidate, error) {
|
||||||
return store.ListQueuedCandidates(ctx, db, at, limit)
|
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 {
|
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)
|
return store.CreateProposal(ctx, db, proposal, ticketIDs, at)
|
||||||
|
|||||||
@@ -10,7 +10,7 @@ import (
|
|||||||
"github.com/cosmic-clash/cosmic-clash/server/domain"
|
"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 {
|
type ProposalCreator interface {
|
||||||
CreateProposal(context.Context, domain.Proposal, map[string]string, time.Time) error
|
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")
|
return false, fmt.Errorf("invalid matcher size")
|
||||||
}
|
}
|
||||||
now := w.Now()
|
now := w.Now()
|
||||||
candidates, err := w.Source(ctx, now, w.Size)
|
candidates, err := w.Source(ctx, now, w.Playlist, w.Size)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return false, err
|
return false, err
|
||||||
}
|
}
|
||||||
@@ -77,6 +77,9 @@ func (w Worker) RunOnce(ctx context.Context) (bool, error) {
|
|||||||
}
|
}
|
||||||
queue := domain.NewQueue()
|
queue := domain.NewQueue()
|
||||||
for _, candidate := range candidates {
|
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 {
|
if _, err := queue.Create(candidate.PlayerID, candidate.TicketID, "matcher-"+candidate.TicketID, candidate, now); err != nil {
|
||||||
return false, err
|
return false, err
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -40,7 +40,9 @@ func workerFor(source CandidateSource, creator ProposalCreator) Worker {
|
|||||||
|
|
||||||
func TestRunOnceDelegatesFinalClaimAndBindsTickets(t *testing.T) {
|
func TestRunOnceDelegatesFinalClaimAndBindsTickets(t *testing.T) {
|
||||||
creator := &creatorSpy{}
|
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())
|
formed, err := worker.RunOnce(context.Background())
|
||||||
if err != nil || !formed || creator.calls != 1 {
|
if err != nil || !formed || creator.calls != 1 {
|
||||||
t.Fatalf("formed=%v err=%v calls=%d", formed, err, creator.calls)
|
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) {
|
func TestRunOnceFailsClosedOnSourceOrDurableClaimFailure(t *testing.T) {
|
||||||
creator := &creatorSpy{err: errors.New("serialization conflict")}
|
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")
|
return nil, errors.New("redis unavailable")
|
||||||
}, creator)
|
}, creator)
|
||||||
if _, err := worker.RunOnce(context.Background()); err == nil {
|
if _, err := worker.RunOnce(context.Background()); err == nil {
|
||||||
t.Fatal("source failure was swallowed")
|
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 {
|
if _, err := worker.RunOnce(context.Background()); err == nil {
|
||||||
t.Fatal("durable claim failure was swallowed")
|
t.Fatal("durable claim failure was swallowed")
|
||||||
}
|
}
|
||||||
@@ -69,9 +73,34 @@ func TestRunOnceFailsClosedOnSourceOrDurableClaimFailure(t *testing.T) {
|
|||||||
|
|
||||||
func TestRunOnceDoesNotClaimAnIncompleteBatch(t *testing.T) {
|
func TestRunOnceDoesNotClaimAnIncompleteBatch(t *testing.T) {
|
||||||
creator := &creatorSpy{}
|
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())
|
formed, err := worker.RunOnce(context.Background())
|
||||||
if err != nil || formed || creator.calls != 0 {
|
if err != nil || formed || creator.calls != 0 {
|
||||||
t.Fatalf("formed=%v err=%v calls=%d", formed, err, creator.calls)
|
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)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -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,
|
const QueueCandidateProjectionSQL = `SELECT ticket_id, player_id, playlist, client_build,
|
||||||
protocol_version, enqueued_at
|
protocol_version, enqueued_at
|
||||||
FROM queue_tickets
|
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
|
ORDER BY enqueued_at, ticket_id
|
||||||
LIMIT $2`
|
LIMIT $3`
|
||||||
|
|
||||||
// ListQueuedCandidates is an authoritative, expiry-filtered source for the
|
// ListQueuedCandidates is an authoritative, expiry-filtered source for the
|
||||||
// matcher projection. It deliberately does not claim rows; CreateProposal is
|
// matcher projection. It deliberately does not claim rows; CreateProposal is
|
||||||
// the transaction that performs the competing claim with SKIP LOCKED fences.
|
// 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) {
|
func ListQueuedCandidates(ctx context.Context, db *sql.DB, playlist domain.Playlist, now time.Time, limit int) ([]domain.Candidate, error) {
|
||||||
if db == nil || now.IsZero() || limit < 1 || limit > 1000 {
|
if db == nil || (playlist != domain.Casual && playlist != domain.Ranked) || now.IsZero() || limit < 1 || limit > 1000 {
|
||||||
return nil, fmt.Errorf("invalid queued candidate arguments")
|
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 {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user