CosmicClash/server/matcher/worker_test.go

package matcher

import (
	"context"
	"errors"
	"fmt"
	"sync"
	"testing"
	"time"

	"github.com/cosmic-clash/cosmic-clash/server/domain"
)

type creatorSpy struct {
	calls int
	err   error
	last  domain.Proposal
	ids   map[string]string
}

func (c *creatorSpy) CreateProposal(_ context.Context, proposal domain.Proposal, ids map[string]string, _ time.Time) error {
	c.calls++
	c.last = proposal
	c.ids = ids
	return c.err
}

func candidates() []domain.Candidate {
	now := time.Unix(1000, 0).UTC()
	result := make([]domain.Candidate, 4)
	for i := range result {
		result[i] = domain.Candidate{TicketID: "ticket-" + string(rune('1'+i)), PlayerID: "player-" + string(rune('1'+i)), Playlist: domain.Casual, ProtocolVersion: 1, EnqueuedAt: now.Add(time.Duration(i) * time.Second), PredictedRTT: map[string]float64{"EU": 20}}
	}
	return result
}

func workerFor(source CandidateSource, creator ProposalCreator) Worker {
	return Worker{Source: source, Creator: creator, Playlist: domain.Casual, Size: 4, Now: func() time.Time { return time.Unix(1000, 0).UTC() }, NextID: func() string { return "proposal-1234567890123456" }, Prepare: func(id string, playlist domain.Playlist, formation domain.MatchFormation, now time.Time) (domain.PreparedProposal, error) {
		return domain.PrepareProposal(id, playlist, formation, nil, domain.RankedArena{}, now)
	}}
}

func TestRunOnceDelegatesFinalClaimAndBindsTickets(t *testing.T) {
	creator := &creatorSpy{}
	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)
	}
	if len(creator.ids) != 4 || creator.ids["player-1"] != "ticket-1" {
		t.Fatalf("ticket bindings=%v", creator.ids)
	}
}

func TestRunOnceFailsClosedOnSourceOrDurableClaimFailure(t *testing.T) {
	creator := &creatorSpy{err: errors.New("serialization conflict")}
	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, domain.Playlist, int) ([]domain.Candidate, error) {
		return candidates(), nil
	}
	if _, err := worker.RunOnce(context.Background()); err == nil {
		t.Fatal("durable claim failure was swallowed")
	}
	if creator.calls != 1 {
		t.Fatalf("creator calls=%d", creator.calls)
	}
}

func TestRunOnceDoesNotClaimAnIncompleteBatch(t *testing.T) {
	creator := &creatorSpy{}
	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)
	}
}

// TestRunSurvivesPerPassErrorsAndKeepsRetrying reproduces a real production
// bug found via a live integration test (multiplayer-next.md 8.40): two real
// players queued with no verified common region formed exactly this
// "source succeeds, formation fails" shape, and the matcher process died
// entirely rather than waiting for a compatible batch -- silently taking
// matchmaking down for every other player behind them too, not just the
// incompatible pair. Run must survive a per-pass RunOnce error and try
// again next interval rather than returning immediately.
func TestRunSurvivesPerPassErrorsAndKeepsRetrying(t *testing.T) {
	var mu sync.Mutex
	attempts := 0
	creatorCalls := 0
	worker := workerFor(func(context.Context, time.Time, domain.Playlist, int) ([]domain.Candidate, error) {
		mu.Lock()
		defer mu.Unlock()
		attempts++
		if attempts == 1 {
			// Same failure shape as domain.FormFromQueue's "no compatible
			// candidates" -- RunOnce still returns a non-nil error here, only
			// Run's handling of it is what this test is about.
			return nil, errors.New("no common region")
		}
		return candidates(), nil
	}, ProposalCreatorFunc(func(context.Context, domain.Proposal, map[string]string, time.Time) error {
		mu.Lock()
		defer mu.Unlock()
		creatorCalls++
		return nil
	}))
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()
	done := make(chan error, 1)
	go func() { done <- worker.Run(ctx, 5*time.Millisecond) }()
	deadline := time.After(1 * time.Second)
	for {
		mu.Lock()
		calls := creatorCalls
		mu.Unlock()
		if calls > 0 {
			break
		}
		select {
		case err := <-done:
			t.Fatalf("Run returned early on a per-pass error instead of retrying: %v", err)
		case <-deadline:
			t.Fatal("Run never recovered from the first pass's error")
		default:
			time.Sleep(time.Millisecond)
		}
	}
	mu.Lock()
	defer mu.Unlock()
	if creatorCalls != 1 {
		t.Fatalf("creator calls=%d, want exactly 1 once formation finally succeeded", creatorCalls)
	}
}

// TestRunOnceRequestsMoreCandidatesThanASingleFormationNeeds guards the
// companion half of the wedge fix below: excluding a failed formation and
// retrying is a no-op if Source was only ever asked for exactly w.Size
// candidates in the first place, since nothing is left afterward. RunOnce
// must ask Source for headroom beyond one formation's worth.
func TestRunOnceRequestsMoreCandidatesThanASingleFormationNeeds(t *testing.T) {
	var requestedLimit int
	worker := workerFor(func(_ context.Context, _ time.Time, _ domain.Playlist, limit int) ([]domain.Candidate, error) {
		requestedLimit = limit
		return candidates(), nil
	}, &creatorSpy{})
	if _, err := worker.RunOnce(context.Background()); err != nil {
		t.Fatalf("RunOnce: %v", err)
	}
	if requestedLimit <= worker.Size {
		t.Fatalf("Source was asked for limit=%d, want more than worker.Size=%d so a failed formation has a remainder to retry against", requestedLimit, worker.Size)
	}
}

// TestRunOnceExcludesAFailingFormationAndTriesTheRemainingCandidates covers
// a wedge distinct from the no-common-region crash-loop above:
// domain.FormFromQueue's anchor is always the oldest candidate, so if
// domain.PrepareProposal rejects that exact formation (ranked admission,
// mismatched protocol, incomplete identity metadata -- anything formation-
// specific rather than "no batch exists at all"), retrying next interval
// reproduces the identical formation and fails again forever, permanently
// head-of-line-blocking every other waiting player behind that anchor too,
// not just the players actually at fault. RunOnce must exclude the failed
// formation's players and try the remaining pool within the same pass.
func TestRunOnceExcludesAFailingFormationAndTriesTheRemainingCandidates(t *testing.T) {
	now := time.Unix(1000, 0).UTC()
	batch := make([]domain.Candidate, 8)
	for i := range batch {
		batch[i] = domain.Candidate{TicketID: fmt.Sprintf("ticket-%d", i), PlayerID: fmt.Sprintf("player-%d", i), Playlist: domain.Casual, ProtocolVersion: 1, EnqueuedAt: now.Add(time.Duration(i) * time.Second), PredictedRTT: map[string]float64{"EU": 20}}
	}
	creator := &creatorSpy{}
	prepareCalls := 0
	worker := Worker{
		Source:   func(context.Context, time.Time, domain.Playlist, int) ([]domain.Candidate, error) { return batch, nil },
		Creator:  creator,
		Playlist: domain.Casual,
		Size:     4,
		Now:      func() time.Time { return now },
		NextID:   func() string { return "proposal-1234567890123456" },
		Prepare: func(id string, playlist domain.Playlist, formation domain.MatchFormation, at time.Time) (domain.PreparedProposal, error) {
			prepareCalls++
			for _, player := range formation.Selection.Players {
				// The oldest four players (the deterministic anchor group) are
				// the "doomed" combination -- always reject them, every time.
				if player.PlayerID == "player-0" {
					return domain.PreparedProposal{}, errors.New("simulated formation-specific rejection")
				}
			}
			return domain.PrepareProposal(id, playlist, formation, nil, domain.RankedArena{}, at)
		},
	}
	formed, err := worker.RunOnce(context.Background())
	if err != nil || !formed {
		t.Fatalf("formed=%v err=%v, want the second (players 4-7) formation to succeed", formed, err)
	}
	if prepareCalls != 2 {
		t.Fatalf("Prepare calls=%d, want exactly 2 (the doomed anchor group, then the remainder)", prepareCalls)
	}
	if creator.calls != 1 {
		t.Fatalf("creator calls=%d, want exactly 1", creator.calls)
	}
	for _, doomed := range []string{"player-0", "player-1", "player-2", "player-3"} {
		if _, claimed := creator.ids[doomed]; claimed {
			t.Fatalf("doomed player %s must not have been claimed by the surviving proposal: %+v", doomed, creator.ids)
		}
	}
	if len(creator.ids) != 4 {
		t.Fatalf("claimed ticket count=%d, want 4", len(creator.ids))
	}
}

// TestRunOnceReturnsTheLastFormationErrorWhenEveryAttemptFails proves the
// exclusion loop is bounded and still surfaces a real error to Run's
// existing non-fatal per-pass handling, rather than silently reporting
// formed=false,err=nil when nothing could ever have worked this pass.
func TestRunOnceReturnsTheLastFormationErrorWhenEveryAttemptFails(t *testing.T) {
	now := time.Unix(1000, 0).UTC()
	batch := make([]domain.Candidate, 8)
	for i := range batch {
		batch[i] = domain.Candidate{TicketID: fmt.Sprintf("ticket-%d", i), PlayerID: fmt.Sprintf("player-%d", i), Playlist: domain.Casual, ProtocolVersion: 1, EnqueuedAt: now.Add(time.Duration(i) * time.Second), PredictedRTT: map[string]float64{"EU": 20}}
	}
	worker := Worker{
		Source:   func(context.Context, time.Time, domain.Playlist, int) ([]domain.Candidate, error) { return batch, nil },
		Creator:  &creatorSpy{},
		Playlist: domain.Casual,
		Size:     4,
		Now:      func() time.Time { return now },
		NextID:   func() string { return "proposal-1234567890123456" },
		Prepare: func(string, domain.Playlist, domain.MatchFormation, time.Time) (domain.PreparedProposal, error) {
			return domain.PreparedProposal{}, errors.New("every formation is doomed")
		},
	}
	formed, err := worker.RunOnce(context.Background())
	if formed || err == nil || err.Error() != "every formation is doomed" {
		t.Fatalf("formed=%v err=%v, want the last formation-specific error surfaced", formed, err)
	}
}

// TestRunStopsImmediatelyOnConfigurationErrors is the other half of the
// fix: a genuinely static misconfiguration (true on every future pass, not
// just this one) must still stop the worker rather than spin forever.
func TestRunStopsImmediatelyOnConfigurationErrors(t *testing.T) {
	worker := workerFor(func(context.Context, time.Time, domain.Playlist, int) ([]domain.Candidate, error) {
		return candidates(), nil
	}, &creatorSpy{})
	worker.Playlist = domain.Playlist("invalid")
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()
	err := worker.Run(ctx, 5*time.Millisecond)
	if !errors.Is(err, ErrUnsupportedPlaylist) {
		t.Fatalf("Run() error = %v, want ErrUnsupportedPlaylist", err)
	}
}