Documentation
¶
Overview ¶
Package sweep runs a service's reclaimers, the bounded passes that remove what its libraries leave behind, as one go-core reactor. A reclaimer is a library's own, such as blobfs's sweep or go-messaging's retention; this package imports none, and each satisfies Reclaimer structurally. A wiring defect panics, per the database package; a defect in the list of reclaimers is an error.
Construction ¶
New builds a Sweeper over the service's sqlate session from an explicit list of reclaimers, each a Registration with its name and its Policy, the interval between its cycles and the budget of each pass. The list is the only way in: the package keeps no registry. New refuses an empty list, an empty or repeated name, a nil reclaimer, a non-positive interval or budget, and a non-positive shutdown timeout, before any I/O.
An Option configures the rest:
- Unlocked runs without claiming the reclaimers' locks, on a dialect without sqlate.TryLocker or with it. Without it, such a dialect is ErrNoTryLocker.
- Gated sets the Gate each pass holds shared, such as go-core's reactor.Gate, so an operation that holds it exclusively, such as a schema change, runs with the passes paused.
- OnPass sets the Hook each Pass is reported to. Without it, a pass that returned an error is logged at warn on the sweeper's logger, and no other pass is.
Cycles ¶
The Sweeper's reactor runs on a go-core reactor.Wake source ticking at the smallest policy interval. Each delivery runs, one at a time in the list's order, every reclaimer whose interval has passed since its last cycle and every reclaimer Sweeper.Nudge named since then; every reclaimer is due at start. A nudge of a name the list does not hold panics.
A reclaimer's cycle first claims the advisory lock "sweep.<name>" on a pinned connection with sqlate.TryLocker. A lock another process holds runs no pass and reports nothing: that process is reclaiming. Once claimed, the cycle passes at the policy's budget while a pass reports more, reporting each pass. A pass's error is reported and ends neither the cycle nor the reactor: the cycle goes on while the pass reported more, and a pass that reported none ends it, error or not. A cycle runs for at most its reclaimer's interval: once that has passed since the claim, a cycle still reporting more yields the reactor to the other due reclaimers and stays due. The next delivery, which the yield asks for, runs every due reclaimer again in the list's order, so a backlog keeps draining interleaved with theirs. A reclaimer that reports more without progress still costs passes back to back for that interval, which is why a reclaimer reports more only while a further pass can make progress. The drain ends a cycle sooner. The claim is released when the cycle ends, on every path: unlocked, then the connection closed. A failure to claim or to release the lock is logged at warn, and a connection whose unlock failed is discarded rather than pooled, so the session's end releases the lock.
Drain ¶
The Sweeper is a lifecycle Starter, Stopper, ReadinessChecker, and Monitored through Sweeper.Start, Sweeper.Shutdown, Sweeper.Ready, and Sweeper.Err, which are its reactor's, so a composition root makes it a graph node's value and the coordinator starts, drains, and watches it. Once the drain begins, no further pass starts, and a wait on the gate ends without one; the pass in flight finishes. A pass still running when the reactor cancels its handlers (half the shutdown timeout New is given) ends the cycle unreported, and the cancellation is the reactor's error. The Sweeper keeps no backlog state: what a drain leaves, the next start finds.
Index ¶
Constants ¶
This section is empty.
Variables ¶
var ErrNoTryLocker = errors.New("sweep: dialect has no try-lock capability")
ErrNoTryLocker reports a dialect without the sqlate.TryLocker capability, which a Sweeper needs to claim each reclaimer's lock without waiting; Unlocked opts out.
Functions ¶
This section is empty.
Types ¶
type Gate ¶
type Gate interface {
}
Gate is a quiesce gate each pass holds shared, so an operation that holds it exclusively runs with the passes paused: go-core's reactor.Gate is one. Shared waits until the gate admits a shared holder and returns its release, or fails, holding nothing; a wait ctx ends returns ctx's error.
type Hook ¶
Hook receives each Pass, with the context it ran under. It must not block for long: the sweeper waits on it before the next pass.
type Option ¶
type Option func(*options)
Option configures a Sweeper.
func Gated ¶
Gated sets the gate each pass holds shared, from before the pass begins until it returns. Without it, or with a nil gate, passes hold none. A wait on g that the reactor's drain or the handler's context ends runs no pass and reports nothing. Any other failure of g is reported as the pass's error, with no pass run, and ends the cycle.
func OnPass ¶
OnPass sets the hook each Pass is reported to. Without it, or with a nil hook, the sweeper logs each pass that returned an error at warn on its logger and nothing else.
func Unlocked ¶
func Unlocked() Option
Unlocked lets the sweeper run on a dialect without the sqlate.TryLocker capability, or with it, without claiming the locks. Its intended shape is a single process, or a consumer that serializes the sweep across processes some other way; concurrent sweepers without one each run every reclaimer, which their idempotent passes tolerate but no budget bounds.
type Pass ¶
type Pass struct {
// Name is the reclaimer's registered name.
Name string
// More is what the pass reported: whether more may remain.
More bool
// Err is the pass's error, or, under [Gated], the gate's failure other
// than its wait ending, in which case no pass ran.
Err error
// Duration is how long the pass took, its wait on the gate included.
Duration time.Duration
}
Pass is one reclaimer pass, reported to the Hook once the pass returns.
type Policy ¶
type Policy struct {
// Interval is how long after a cycle the reclaimer is due again; it
// must be positive.
Interval time.Duration
// Budget is the bound handed to each Pass; it must be positive.
Budget int
}
Policy is a reclaimer's schedule: the least time between the starts of two of its cycles, and the budget each of its passes is given.
type Reclaimer ¶
Reclaimer is one library's bounded reclaim: a pass removes at most budget units of its garbage and reports whether more may remain. A pass is idempotent and finds its work in the database, so a pass that fails is retried by a later cycle. A pass that reports more is followed by another in the same cycle, whether or not it failed, until the cycle has run for its interval, when the reclaimer yields to the others due and stays due. A reclaimer therefore reports more only while a further pass can make progress: one that reports more on every pass spends each of its cycles, an interval at a time, on passes that remove nothing. blobfs's sweep and go-messaging's retention satisfy it structurally; this package imports neither.
type Registration ¶
Registration is one reclaimer in the list New builds a Sweeper from: its name, unique in the list, which Sweeper.Nudge takes and the advisory lock "sweep.<name>" is named for; the reclaimer; and its policy.
type Sweeper ¶
type Sweeper struct {
// contains filtered or unexported fields
}
Sweeper runs a fixed list of reclaimers as one go-core reactor. Each delivery runs every reclaimer that is due, one at a time in the list's order: a reclaimer is due when its interval has passed since its last cycle, when Sweeper.Nudge named it since its last cycle, and at start.
The reactor's source is a reactor.Wake whose interval is the smallest policy interval, so each delivery is a tick of that interval or a nudge. A reclaimer counts as due once its interval, less half a tick, has passed since the delivery that last ran it, which absorbs the drift between a nudge's time and the ticks; a reclaimer whose interval is not a multiple of the smallest therefore runs on the first tick at or after its interval, up to one tick late.
The Sweeper is a lifecycle Starter, Stopper, ReadinessChecker, and Monitored through its reactor's methods; it keeps no backlog state, and its readiness is the reactor's alone.
func New ¶
func New(db *sqlate.DB, regs []Registration, shutdown time.Duration, logger *slog.Logger, opts ...Option) (*Sweeper, error)
New builds a Sweeper over db from the reclaimers in regs, logging to logger. shutdown is the process's drain timeout: the reactor cancels a pass still running at half of it (reactor.CancelWithin). New refuses, with an error and before any I/O, an empty list; a registration with an empty name, a name another uses, a nil reclaimer, or a non-positive interval or budget; a non-positive shutdown; and a dialect without sqlate.TryLocker, as ErrNoTryLocker, unless Unlocked. A nil db or logger is a wiring defect and panics.
Every reclaimer is due at start: New nudges the source, which delivers as soon as the reactor starts.
func (*Sweeper) Err ¶
Err yields the reactor's failure while running; see reactor.Reactor.Err.
func (*Sweeper) Nudge ¶
Nudge marks the reclaimer registered as name due, and asks the reactor for a delivery as soon as it is free; nudges before a delivery coalesce into it, and a nudge while the reclaimer's cycle runs earns it one more cycle after. It never blocks. A name the Sweeper was not built with is a defect in the caller and panics.
func (*Sweeper) Ready ¶
Ready reports the reactor's readiness; see reactor.Reactor.Ready.
func (*Sweeper) Shutdown ¶
Shutdown drains the reactor: no further pass starts, the pass in flight finishes, and one still running when the CancelWithin bound passes is cancelled. See reactor.Reactor.Shutdown.
Source Files