Files
memby/server/internal/credits/flight.go
2026-08-15 09:23:26 +12:00

57 lines
1.8 KiB
Go

package credits
import "sync"
// A minimal single-flight, because four signals can name the same episode.
//
// Live playback, a predicted next episode, another viewer's prediction and a refresh cycle
// all legitimately want the same media version at the same moment, and each of them arriving
// as its own scan is the one way this subsystem could become expensive. The queue already
// deduplicates by item; this closes the gap for anything that reaches Process directly — a
// forced scan from the console, the benchmark, or a live push racing the worker.
//
// Written rather than taken from golang.org/x/sync so the gateway keeps its two direct
// dependencies. It is thirty lines and the semantics are not subtle: the first caller does
// the work, everybody else waits and receives the same answer.
type flightGroup struct {
mu sync.Mutex
calls map[string]*flightCall
}
type flightCall struct {
done chan struct{}
value any
err error
}
func newFlightGroup() *flightGroup {
return &flightGroup{calls: map[string]*flightCall{}}
}
// Do runs fn unless an identical key is already in flight, in which case it waits for that
// one and returns its result.
func (g *flightGroup) Do(key string, fn func() (any, error)) (any, error) {
g.mu.Lock()
if existing, found := g.calls[key]; found {
g.mu.Unlock()
<-existing.done
return existing.value, existing.err
}
call := &flightCall{done: make(chan struct{})}
g.calls[key] = call
g.mu.Unlock()
// The deferred cleanup matters more than it looks: a detector that panics must not leave
// a key permanently in flight, which would make that episode unscannable for the life of
// the process and give every later caller a channel that never closes.
defer func() {
g.mu.Lock()
delete(g.calls, key)
g.mu.Unlock()
close(call.done)
}()
call.value, call.err = fn()
return call.value, call.err
}