Artifact [3dee27b02c]

Artifact 3dee27b02cddd4b8da1b321163d83fa684f4fa76a9e26d646a70d4fce9c0a193:

• File internal/work/work.go — part of check-in [45bb75ade2] at 2026-08-17 21:26:48 on branch trunk — work: bound bulk operations; footer times keep their precision

  Nothing in quarry limited concurrency — every zit subprocess ran in the request goroutine with the arrival rate as the only bound. work.Lane is a concurrency limit plus a queue cap that SHEDS past it: a request accepted into a queue it cannot be served from consumes the capacity that would have served someone else.

  Wired to archive first, which buffers 256 MiB per request on an anonymous route. That is latent rather than live — archive 503s on zit#97 — which is why the bound belongs there now, so the endpoint is born bounded rather than acquiring a limit after someone finds it.

  Footer precision: '<1ms' was true and useless once most pages went sub-millisecond, collapsing the interesting range into one label. timing.Format keeps ~3 significant figures across five orders of magnitude, so 900µs and 90µs no longer read the same. Fossil has the mirror-image bug: three decimals of seconds prints everything faster than a millisecond as 0.000s. (user: dmestas, size: 3793) [annotate] [blame] [check-ins using]

// 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"
)

// ErrBusy reports that a lane is at its queue cap. Callers map it to a
// retryable refusal; ops.ErrUnavailable is the ops-layer spelling.
var ErrBusy = errors.New("work: lane is saturated")

// 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 ErrBusy
	}
	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(),
	}
}