CosmicClash/server/matcher/worker.go

// Package matcher contains the provider-neutral orchestration around the
// durable proposal claim transaction.
package matcher

import (
	"context"
	"errors"
	"fmt"
	"time"

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

// These three are the only RunOnce failures Run treats as fatal to the whole
// worker: they're static misconfiguration, true on every future pass just as
// much as this one, so retrying cannot help. Every other RunOnce error --
// a source read hiccup, no common region among the current candidate pool,
// a losing race against another matcher replica, an incomplete batch -- is a
// single pass's worth of "no match formed this time," a routine and
// expected steady state that must not take matching down for every other
// player still waiting behind it.
var (
	ErrWorkerNotConfigured = errors.New("matcher worker is not configured")
	ErrUnsupportedPlaylist = errors.New("unsupported matcher playlist")
	ErrInvalidMatcherSize  = errors.New("invalid matcher size")
)

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
}

type ProposalCreatorFunc func(context.Context, domain.Proposal, map[string]string, time.Time) error

func (f ProposalCreatorFunc) CreateProposal(ctx context.Context, proposal domain.Proposal, ticketIDs map[string]string, now time.Time) error {
	return f(ctx, proposal, ticketIDs, now)
}

type PrepareFunc func(string, domain.Playlist, domain.MatchFormation, time.Time) (domain.PreparedProposal, error)

type Worker struct {
	Source   CandidateSource
	Creator  ProposalCreator
	Playlist domain.Playlist
	Size     int
	Now      func() time.Time
	NextID   func() string
	Prepare  PrepareFunc
	OnError  func(error)
}

// Run polls until cancellation. A failed attempt is returned so a supervisor
// can restart the role rather than silently dropping durable claim failures.
func (w Worker) Run(ctx context.Context, interval time.Duration) error {
	if interval <= 0 {
		return fmt.Errorf("matcher interval must be positive")
	}
	for {
		if _, err := w.RunOnce(ctx); err != nil {
			if w.OnError != nil {
				w.OnError(err)
			}
			if errors.Is(err, ErrWorkerNotConfigured) || errors.Is(err, ErrUnsupportedPlaylist) || errors.Is(err, ErrInvalidMatcherSize) {
				return err
			}
			// Not fatal -- fall through and retry next interval.
		}
		timer := time.NewTimer(interval)
		select {
		case <-ctx.Done():
			timer.Stop()
			return nil
		case <-timer.C:
		}
	}
}

// RunOnce performs one bounded matchmaking attempt. The source may be Redis
// backed, but the creator must be the durable transaction that claims tickets;
// a stale cache therefore fails safely and can be retried on the next pass.
func (w Worker) RunOnce(ctx context.Context) (bool, error) {
	if w.Source == nil || w.Creator == nil || w.Now == nil || w.NextID == nil || w.Prepare == nil {
		return false, ErrWorkerNotConfigured
	}
	if w.Playlist != domain.Casual && w.Playlist != domain.Ranked {
		return false, ErrUnsupportedPlaylist
	}
	if w.Size < 2 || w.Size > 6 {
		return false, ErrInvalidMatcherSize
	}
	now := w.Now()
	// Request headroom beyond exactly w.Size: the exclusion-retry loop below
	// needs other candidates to fall back to when the oldest-anchor formation
	// fails, and a pool capped at exactly w.Size leaves nothing to retry with
	// -- silently reintroducing the same head-of-line wedge the loop exists
	// to fix. Bounded (not unlimited) so a large playlist backlog doesn't turn
	// every pass into an expensive scan.
	candidates, err := w.Source(ctx, now, w.Playlist, w.candidatePoolSize())
	if err != nil {
		return false, err
	}
	if len(candidates) < w.Size {
		return false, nil
	}
	// A first full pass over every candidate validates playlist and identity
	// shape once, exactly as before -- these are hard input-format errors, not
	// "this particular formation didn't work out", so they still fail the pass
	// immediately rather than being retried below.
	for _, candidate := range candidates {
		if candidate.Playlist != w.Playlist {
			return false, fmt.Errorf("candidate playlist does not match worker")
		}
	}

	// FormFromQueue's anchor is always the oldest candidate, deterministically.
	// If domain.PrepareProposal then rejects that exact formation (mismatched
	// protocol, incomplete ranked identity metadata, a duplicate-SteamID pair,
	// etc.), retrying next interval reproduces the identical formation and
	// fails again -- forever, permanently head-of-line-blocking every other
	// waiting player behind that anchor, not just the players actually at
	// fault. Excluding the failed formation's players and retrying with the
	// remainder, bounded within this one pass, means one bad combination can
	// no longer wedge the whole playlist; Worker.Run's existing non-fatal
	// per-pass-error handling still applies if every attempt is exhausted.
	remaining := candidates
	var lastErr error
	for attempt := 0; attempt < maxFormationAttemptsPerPass && len(remaining) >= w.Size; attempt++ {
		queue := domain.NewQueue()
		for _, candidate := range remaining {
			if _, err := queue.Create(candidate.PlayerID, candidate.TicketID, "matcher-"+candidate.TicketID, candidate, now); err != nil {
				return false, err
			}
		}
		formation, err := domain.FormFromQueue(queue, w.Size, now)
		if err != nil {
			// No compatible batch exists at all within what's left of the pool
			// (e.g. no shared region) -- not specific to one formation, so
			// retrying within this pass cannot help either.
			return false, err
		}
		prepared, err := w.Prepare(w.NextID(), w.Playlist, formation, now)
		if err != nil {
			lastErr = err
			excluded := make(map[string]bool, len(formation.Selection.Players))
			for _, player := range formation.Selection.Players {
				excluded[player.PlayerID] = true
			}
			next := make([]domain.Candidate, 0, len(remaining))
			for _, candidate := range remaining {
				if !excluded[candidate.PlayerID] {
					next = append(next, candidate)
				}
			}
			remaining = next
			continue
		}
		return w.claim(ctx, formation, prepared, now)
	}
	return false, lastErr
}

// maxFormationAttemptsPerPass bounds how many distinct formations RunOnce
// will try excluding prior failures before deferring to the next interval.
// Each attempt is pure in-memory work (no durable claim happens until
// Prepare succeeds), so this is cheap; it exists to keep one pass bounded
// rather than to conserve resources.
const maxFormationAttemptsPerPass = 8

const (
	candidatePoolMultiplier = 10
	maxCandidatePoolSize    = 200
)

// candidatePoolSize is how many candidates RunOnce asks Source for. It is
// deliberately larger than w.Size (see RunOnce) and bounded independently of
// the playlist's actual backlog size.
func (w Worker) candidatePoolSize() int {
	poolSize := w.Size * candidatePoolMultiplier
	if poolSize > maxCandidatePoolSize {
		return maxCandidatePoolSize
	}
	return poolSize
}

func (w Worker) claim(ctx context.Context, formation domain.MatchFormation, prepared domain.PreparedProposal, now time.Time) (bool, error) {
	ticketIDs := make(map[string]string, len(prepared.Proposal.Participants))
	for _, participant := range prepared.Proposal.Participants {
		for _, candidate := range formation.Selection.Players {
			if candidate.PlayerID == participant.PlayerID {
				ticketIDs[participant.PlayerID] = candidate.TicketID
				break
			}
		}
	}
	if len(ticketIDs) != len(prepared.Proposal.Participants) {
		return false, fmt.Errorf("proposal participant is not in formed selection")
	}
	if err := w.Creator.CreateProposal(ctx, prepared.Proposal, ticketIDs, now); err != nil {
		return false, err
	}
	return true, nil
}