// Package work bounds concurrent expensive operations.
//
// Nothing in quarry limited concurrency: every zit subprocess, every
// buffered archive, ran in the request goroutine with the arrival rate
// as the only bound. That is fine until it is not — the file page
// saturated at 32 requests per second and then went BACKWARDS, serving
// 24 while latency quadrupled, because each request forked a process
// competing with the others.
//
// A Lane is the smallest thing that fixes it: a concurrency limit plus a
// queue cap, and a refusal past the cap. Refusing matters as much as
// limiting. A request accepted into a queue it cannot be served from in
// time consumes the capacity that would have served someone else, so
// past the cap a Lane sheds rather than accepts — the caller gets a 503
// it can retry, not a timeout it cannot distinguish from a hang.
//
// Lanes are separate rather than one global pool because the operations
// have wildly different costs: a 256 MiB archive and a 3 ms name lookup
// in one pool means the archive decides how many name lookups happen.
package work
import (
"context"
"errors"
"sync/atomic"
"craftdesign.group/quarry/internal/fault"
)
// ErrBusy reports that a lane is at its queue cap. Callers map it to a
// retryable refusal; ops.ErrUnavailable is the ops-layer spelling.
//
// It stays a bare sentinel because errors.Is compares against it. What
// reaches the journal is busy(), which wraps it with the numbers that
// make the refusal actionable — the sentinel alone said "work: lane is
// saturated" 833 times, naming no lane, no depth and no repository,
// while the lane raising it knew all three.
var ErrBusy = errors.New("work: lane is saturated")
// busy is ErrBusy with this lane's load attached. Building it costs an
// allocation on a path that is already shedding, which is the cheapest
// moment in the system to spend one.
func (l *Lane) busy() error {
st := l.Stats()
return fault.Wrap(ErrBusy, l.name+" lane is full").
Cost("work is being refused rather than queued; callers are told to retry").
Do("raise this lane's limit or queue, or add a node — a shed count that keeps climbing is capacity, not a bug").
With("lane", st.Name, "running", st.InFlight, "limit", st.Limit,
"waiting", st.Waiting, "queue", st.Queue, "shed", st.Shed)
}
// Lane admits at most `limit` operations at once and queues at most
// `queue` more before refusing.
type Lane struct {
name string
slots chan struct{}
waiting atomic.Int64
maxWait int64
// Counters for the admin surface. The reason this package exists is
// to make load legible, so the numbers are part of the contract, not
// debug instrumentation.
admitted atomic.Int64
shed atomic.Int64
inFlight atomic.Int64
}
// NewLane builds a lane. limit is how many run concurrently; queue is
// how many may WAIT beyond that before the lane sheds. A queue of zero
// means "run limit at a time, refuse the rest immediately".
func NewLane(name string, limit, queue int) *Lane {
if limit < 1 {
limit = 1
}
if queue < 0 {
queue = 0
}
l := &Lane{name: name, slots: make(chan struct{}, limit), maxWait: int64(queue)}
return l
}
// Do runs fn if the lane can admit it, blocking while a slot frees.
// It returns ErrBusy without running fn when the queue is full, and the
// context's error if the caller gives up while waiting.
func (l *Lane) Do(ctx context.Context, fn func() error) error {
if l == nil {
return fn()
}
// Reserve a waiting position first, so the check and the wait cannot
// race: without this, N goroutines could each observe a short queue
// and then all wait.
if l.waiting.Add(1) > l.maxWait+int64(cap(l.slots)) {
l.waiting.Add(-1)
l.shed.Add(1)
return l.busy()
}
defer l.waiting.Add(-1)
select {
case l.slots <- struct{}{}:
case <-ctx.Done():
return ctx.Err()
}
defer func() { <-l.slots }()
l.admitted.Add(1)
l.inFlight.Add(1)
defer l.inFlight.Add(-1)
return fn()
}
// Stats is a lane's load, for the fleet admin surface.
type Stats struct {
Name string `json:"name"`
Limit int `json:"limit"`
Queue int `json:"queue"`
InFlight int64 `json:"inFlight"`
Waiting int64 `json:"waiting"`
Admitted int64 `json:"admitted"`
// Shed counts refusals. A nonzero and growing value is the signal
// that this node is out of capacity — the number that turns "it
// feels slow" into "add a node".
Shed int64 `json:"shed"`
}
func (l *Lane) Stats() Stats {
if l == nil {
return Stats{}
}
waiting := l.waiting.Load() - l.inFlight.Load()
if waiting < 0 {
waiting = 0
}
return Stats{
Name: l.name, Limit: cap(l.slots), Queue: int(l.maxWait),
InFlight: l.inFlight.Load(), Waiting: waiting,
Admitted: l.admitted.Load(), Shed: l.shed.Load(),
}
}