memby/server/internal/credits/queue.go

package credits

import (
	"sync"
	"time"
)

// The queue, which lives entirely in RAM and is deliberately allowed to be lost.
//
// A durable job queue would need rows written for every state transition, and the whole
// point of this subsystem is that the database sees one write per *scan*, not one per
// candidate. Nothing here is expensive to rebuild: the priorities came from a single indexed
// query against Tracearr sessions, and on restart that query simply runs again. A persistent
// scheduler would cost more to maintain than it could ever save.

// Queue is a bounded priority queue of candidates, deduplicated by item.
//
// Bounded because saturation is a real state and the right response to it is to throw
// speculation away: a queue holding forty episodes nobody has reached is not a queue that
// will eventually catch up, it is one that has stopped describing demand.
type Queue struct {
	limit int

	mu      sync.Mutex
	items   map[string]Candidate
	claimed map[string]bool
}

func NewQueue(limit int) *Queue {
	if limit <= 0 {
		limit = 20
	}
	return &Queue{
		limit:   limit,
		items:   map[string]Candidate{},
		claimed: map[string]bool{},
	}
}

// Push offers a candidate. It is merged with anything already queued for the same item — the
// strongest case wins — and dropped if the queue is full and it is not strong enough to
// displace what is already there.
//
// Returns whether the candidate is now queued, which is what lets a caller log a discard
// rather than believe it queued something.
func (q *Queue) Push(candidate Candidate) bool {
	if candidate.ItemID == "" {
		return false
	}
	q.mu.Lock()
	defer q.mu.Unlock()

	// Already being scanned. Single-flight lives in the service, but refusing to re-queue
	// what a worker is holding keeps the queue honest about what is outstanding.
	if q.claimed[candidate.ItemID] {
		return false
	}
	if existing, found := q.items[candidate.ItemID]; found {
		if candidate.Priority > existing.Priority {
			existing.Priority = candidate.Priority
			existing.Reason = candidate.Reason
		}
		if candidate.UserCount > existing.UserCount {
			existing.UserCount = candidate.UserCount
		}
		if candidate.LastViewed.After(existing.LastViewed) {
			existing.LastViewed = candidate.LastViewed
		}
		q.items[candidate.ItemID] = existing
		return true
	}

	if len(q.items) >= q.limit {
		weakestID, weakest := q.weakestLocked()
		// Ties go to what is already queued. A candidate that has been waiting is one the
		// worker is closer to reaching, and swapping equals would let a busy refresh cycle
		// churn the queue without ever finishing anything.
		if weakestID == "" || candidate.Priority <= weakest.Priority {
			return false
		}
		delete(q.items, weakestID)
	}
	q.items[candidate.ItemID] = candidate
	return true
}

// Replace swaps the speculative contents of the queue for a freshly built set, which is what
// a refresh cycle does. Anything a worker has claimed is untouched — cancelling a scan that
// is already reading a file to replace it with a marginally better candidate would waste
// exactly the disk activity this package exists to avoid.
func (q *Queue) Replace(candidates []Candidate) {
	q.mu.Lock()
	claimed := q.claimed
	q.items = map[string]Candidate{}
	q.mu.Unlock()

	for _, candidate := range candidates {
		if claimed[candidate.ItemID] {
			continue
		}
		q.Push(candidate)
	}
}

// Claim takes the highest-priority candidate and marks it in flight. Release must follow.
func (q *Queue) Claim() (Candidate, bool) {
	q.mu.Lock()
	defer q.mu.Unlock()

	best, found := Candidate{}, false
	for _, candidate := range q.items {
		if !found || betterCandidate(candidate, best) {
			best, found = candidate, true
		}
	}
	if !found {
		return Candidate{}, false
	}
	delete(q.items, best.ItemID)
	q.claimed[best.ItemID] = true
	return best, true
}

// Release ends a claim. Called from a defer so a panicking detector cannot wedge an item out
// of the queue for the life of the process.
func (q *Queue) Release(itemID string) {
	q.mu.Lock()
	defer q.mu.Unlock()
	delete(q.claimed, itemID)
}

// Len is the number waiting, not counting anything in flight.
func (q *Queue) Len() int {
	q.mu.Lock()
	defer q.mu.Unlock()
	return len(q.items)
}

// SetLimit applies an operator change immediately. The strongest candidates survive a
// reduction, using the same ordering as Claim, so changing the setting cannot reshuffle
// the meaning of priority.
func (q *Queue) SetLimit(limit int) {
	if limit <= 0 {
		return
	}
	q.mu.Lock()
	defer q.mu.Unlock()
	q.limit = limit
	for len(q.items) > q.limit {
		weakestID, _ := q.weakestLocked()
		if weakestID == "" {
			break
		}
		delete(q.items, weakestID)
	}
}

// Snapshot is the queue in worker order, for the log line and the admin console.
func (q *Queue) Snapshot() []Candidate {
	q.mu.Lock()
	out := make([]Candidate, 0, len(q.items))
	for _, candidate := range q.items {
		out = append(out, candidate)
	}
	q.mu.Unlock()
	sortCandidates(out)
	return out
}

// weakestLocked finds the candidate to discard under saturation. Caller holds the lock.
func (q *Queue) weakestLocked() (string, Candidate) {
	weakestID, weakest, found := "", Candidate{}, false
	for id, candidate := range q.items {
		if !found || betterCandidate(weakest, candidate) {
			weakestID, weakest, found = id, candidate, true
		}
	}
	return weakestID, weakest
}

// betterCandidate is the one ordering rule, shared by Claim and the saturation discard so
// the thing taken first and the thing thrown away first can never disagree.
func betterCandidate(a, b Candidate) bool {
	if a.Priority != b.Priority {
		return a.Priority > b.Priority
	}
	if !a.LastViewed.Equal(b.LastViewed) {
		return a.LastViewed.After(b.LastViewed)
	}
	return a.ItemID < b.ItemID
}

// pending is a live-playback candidate waiting out its settling delay.
type pending struct {
	candidate Candidate
	due       time.Time
	cancel    func()
}