Documentation
¶
Overview ¶
Package piston supplies step candidates from one shard. One piston works one shard, firing the same cycle over and over against its own database on its own clock, with no barrier against its peers; an engine with N shards runs N of them.
A piston is a CONSUMER of its database, never the owner: the handle is passed in already open and is closed by whoever opened it, so there is no Open, no Close, and no say over pool sizes. It owns the supply cycle, the two queries behind it, and its instruments; it borrows the planner and the candidate cache, both shared with every other piston on the replica.
Run blocks and drives the cycle until its context ends:
cycle (paced by the pipeline) -> record -> repeat
Liveness reports whether that loop is turning, for an owner that publishes this replica's liveness somewhere the fleet can see it.
SetIdle(true) skips the cycle entirely, which is the await-only replica: it keeps holding connections, but claims no work and reports itself idle. Going idle withdraws the shard from the shared planner and empties its cache partition, so nothing is left claiming a band this piston no longer reports on.
Every Set* is live: the owner may re-derive any of them while Run is in flight.
Index ¶
- Constants
- type ContextFunc
- type PartitionFunc
- type Piston
- func (p *Piston) Cycle(ctx context.Context) pipeline.Result
- func (p *Piston) FetchSteps(ctx context.Context, shard, band int, keys []string, perKey int) (map[string][]int, error)
- func (p *Piston) Idle() bool
- func (p *Piston) Interval() time.Duration
- func (p *Piston) Liveness() (turns uint64, busy, idle bool)
- func (p *Piston) MinGap() time.Duration
- func (p *Piston) Run(ctx context.Context)
- func (p *Piston) ScanBand(ctx context.Context, shard int) (band int, tallies []planner.Tally, err error)
- func (p *Piston) SetContextFunc(fn ContextFunc)
- func (p *Piston) SetIdle(idle bool)
- func (p *Piston) SetInterval(d time.Duration)
- func (p *Piston) SetLogger(l *slog.Logger)
- func (p *Piston) SetMeter(m metric.Meter) error
- func (p *Piston) SetMinGap(d time.Duration)
- func (p *Piston) SetPartitionFunc(fn PartitionFunc)
- func (p *Piston) SetSeams(s *seamster.Seamster)
- func (p *Piston) SetStealAfter(periods int)
- func (p *Piston) Shard() int
- func (p *Piston) StealAfter() int
Constants ¶
const CheckpointCycleDone = "refillCycleDone"
CheckpointCycleDone fires once per cycle that PUSHED - see SetSeams. Exported for the same catalogue reason as FaultScanErr.
It is fired only when the cycle reached its push, which is exactly when this shard's cache partition has been reconciled against the plan: the two error paths (a failed tally, a failed fetch) return before pushing and deliberately leave the partition alone, while an empty plan pushes nothing and CLEARS it. So a visit means "this shard's partition now reflects the plan", which is the thing a test can neither observe from outside nor wait out on a clock - each piston turns on its own cadence, and a shard whose goroutine is starved or blocked on a slow round trip can hold an unreconciled partition arbitrarily long while its peers turn normally.
Fired BOTH unscoped and scoped by shard (a scoped fire does not wake an unscoped waiter, so a waiter for "any shard cycled" and one for "shard 3 cycled" need separate fires). Counting scoped visits is the way to wait for a SPECIFIC shard, since with several shards the unscoped count says nothing about which.
const CheckpointStole = "refillStole"
CheckpointStole fires once per fetch that took at least one step from OUTSIDE this replica's residue class - see SetSeams. Exported for the same catalogue reason as FaultScanErr.
It earns its place on the same boundary rule as CheckpointCycleDone: it reports an effect on state the package borrows, at the moment the effect happens, and no clock substitutes for it. A test proving that a slow peer's work is picked up cannot wait out a duration - the steal fires on the first cycle after the gate arms and the grace elapses, which is a function of the pipeline's cadence, the peer's degradation and the backlog, none of which the test controls. Without it the only assertion available is "the flows eventually finished", which passes just as well against a build where stealing does nothing and the dispatch-window eviction did the work several seconds later - i.e. it cannot tell the mechanism under test from the mechanism it replaces.
Fired BOTH unscoped and scoped by shard, for the same reason CheckpointCycleDone is.
const FaultScanErr = "refillScanErr"
FaultScanErr makes ScanBand fail without touching the database - see SetSeams. The name is exported so the owning application's fault catalogue can alias it rather than re-spell the string.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type ContextFunc ¶
ContextFunc derives the context a cycle's queries run on - see SetContextFunc.
type PartitionFunc ¶
PartitionFunc reports the replica partition - see SetPartitionFunc.
type Piston ¶
type Piston struct {
// contains filtered or unexported fields
}
Piston runs one shard's supply cycle and heartbeat.
Run must be driven by a single goroutine. Every Set* is safe to call from another one, at any time.
func New ¶
New returns a piston for one shard over an already-open database handle. The planner and cache are shared with this replica's other pistons and are not owned here.
func (*Piston) Cycle ¶
Cycle runs exactly ONE supply cycle - paced as always - records it, and returns what happened. Run is this in a loop; a caller drives it directly when it wants a cycle at a moment of its own choosing rather than on the piston's cadence.
NOT safe to call concurrently with Run, or with itself: the pipeline's cadence timestamps are single-goroutine state. A caller that drives cycles by hand should idle the piston first.
func (*Piston) FetchSteps ¶
func (p *Piston) FetchSteps(ctx context.Context, shard, band int, keys []string, perKey int) (map[string][]int, error)
FetchSteps implements pipeline.Source. It loads, per chosen fairness key, up to perKey of this shard's oldest due steps at the given band, keyed and ordered oldest-first - the order the plan replay expects.
perKey is a UNIFORM cap, the max per-key demand across this shard's slice rather than each key's exact demand, which keeps the fetch one IN-list query (an exact per-key cap would need a per-key VALUES/LATERAL join, non-trivial across four dialects). The cost is at most len(keys)*perKey rows, and both factors are bounded by the cache capacity - so the fetch is bounded by capacity^2 regardless of how many fairness keys exist. That independence from key cardinality is the whole point: at high cardinality perKey is ~1, so the fetch is ~capacity.
The band is bound (priority=?), not re-mined from a MIN subquery: the plan committed to this band, and re-mining could pick a lower one that arrived between phases and mismatch the chosen keys. A bound priority does not defeat the selection index - only a bound status would, which is why status stays inlined.
func (*Piston) Liveness ¶
Liveness reports whether this piston is turning, for an owner that publishes the fact somewhere the fleet can see it: the count of cycles completed so far, whether one is running right now, and whether the piston is idling.
A COUNTER rather than a "since you last asked" flag, and that is the load-bearing part. A consuming getter would create a contract - call it exactly once per publication, from exactly one caller - and any second caller, a metric or a test, would silently clear the evidence and leave a healthy piston reading as stalled. Holding the previous count is the reader's business, so this is a pure read that may be called any number of times.
A cycle inside its QUERIES counts, because a scan can legitimately run for tens of seconds on a deep backlog where the executor cannot early-stop, and a piston in the middle of one is plainly still serving. It is deliberately the queries and not the whole cycle: a cycle spends most of a healthy wall clock asleep in its pace, so counting that would make busy permanently true and turn this into "the loop is alive" - which a piston whose every scan FAILS would satisfy just as well, keeping a residue class of steps it never selects. That is the exact stranding the evidence exists to prevent.
func (*Piston) Run ¶
Run drives the piston until ctx is cancelled. It blocks; a caller runs it in a goroutine and waits on its own WaitGroup.
It publishes nothing about this replica's liveness itself - Liveness is a pure read an owner samples on its own cadence, which is what keeps how often that is published independent of how long a cycle takes. The independence is a correctness requirement rather than tidiness: phase one's `rn <= capacity` cut early-stops only on Postgres 15+, so on MySQL/SQL Server/SQLite a deep backlog is still O(backlog), measured in the tens of seconds at a few million due rows. A liveness signal gated on a cycle RETURNING would let one such scan drop a perfectly healthy replica out of its own fleet - which should mean "the process is stuck", nothing less.
func (*Piston) ScanBand ¶
func (p *Piston) ScanBand(ctx context.Context, shard int) (band int, tallies []planner.Tally, err error)
ScanBand implements pipeline.Source. It returns this shard's minimum due priority band and one aggregate row per fairness key at that band - O(distinct keys), never O(backlog).
The per-key count is CAPPED at the cache capacity rather than exact: it is MAX(rn) under an `rn <= capacity` cut, not COUNT(*) OVER. The cap is lossless, since no key can be assigned more than the whole batch, and it is what lets the scan stop touching a key's rows past capacity instead of counting the whole partition - the O(backlog) cost that made a single-key flood scan millions of rows every cycle.
The partition filters the ROWS this replica tallies but deliberately NOT the MIN(priority) subquery: the band is a cluster-wide fact, so mining it from one replica's slice would let replicas disagree on which band is open. A replica holding nothing at the global band therefore tallies zero rows - correct, since its own work is at a worse band that must not be served until the better one drains.
The shard argument is the pipeline's own and equals Shard(); it is accepted for symmetry with the planner and cache calls, which are keyed the same way.
func (*Piston) SetContextFunc ¶
func (p *Piston) SetContextFunc(fn ContextFunc)
SetContextFunc supplies a derivation applied to the context at the START of each cycle, so whatever the caller needs its queries to carry - a priority, an admission time, a deadline - is on both of them without this package knowing what any of it means. Nil (the default) leaves the context alone.
Once per CYCLE rather than once per query, because the two queries of a cycle are one unit of work: what a caller stamps is that unit, and re-deriving between them would make the fetch look like a newer arrival than the scan it belongs to.
func (*Piston) SetIdle ¶
SetIdle puts the piston in or out of idle. An idling piston runs no cycle and claims no work, so Liveness reports it idle and its owner can keep this replica counted for the connections it holds while excluding it from anything that divides work.
The default is NOT idle: a fresh piston dispatches, which is the common case, and a zero value that silently did nothing would be the worse default.
"Idle" here is a configured MODE, distinct from the refill sense of the word (nothing is due), which is a circumstance a dispatching piston meets all the time.
GOING IDLE WITHDRAWS THIS SHARD, and it must: the planner's contract is that every shard either tallies or clears each cycle, and an idle piston does neither, so its last tally would stand forever. The planner is shared with this replica's other pistons, so that stale claim on the best band is the documented wedge - every live piston finds none of its own keys at that band and dispatches nothing, indefinitely. Benign only if every piston is idle, and this setter is per-piston, so the API permits the bad case. The cache partition goes for the same reason an empty plan clears it: nothing here is dispatchable, so every cached candidate is a dead hint a worker would pop and burn a claim round-trip on.
Both are the same positive statement an empty plan makes, so both use the same two calls.
func (*Piston) SetInterval ¶
SetInterval sets the cycle period, start of one cycle's scan to the next. See pipeline.SetInterval.
func (*Piston) SetMeter ¶
SetMeter resolves this piston's instruments from an already-created meter. Nil restores no-ops.
A Meter rather than a MeterProvider so the OWNER picks the instrumentation scope once, for every module it assembles. Each package deriving its own scope from a provider would split one engine's metrics across several scopes in the export whenever a name drifted, and nothing here needs provider-level capability anyway.
Instrument names are a public surface that dashboards bind to - do not rename them.
func (*Piston) SetMinGap ¶
SetMinGap sets the minimum quiet time between cycles. See pipeline.SetMinGap.
func (*Piston) SetPartitionFunc ¶
func (p *Piston) SetPartitionFunc(fn PartitionFunc)
SetPartitionFunc supplies the replica partition: the (replicas, ordinal) pair that restricts this replica's selection to its own residue class of step ids, so replicas sharing a database select disjoint candidates instead of racing for the same rows. ok=false selects everything, which is correct for a solo replica or an unknown ordinal.
A function rather than a value because the pair changes as the fleet does, and a captured one would leave this piston selecting a class that no longer exists.
func (*Piston) SetSeams ¶
SetSeams supplies the owner's fault-injection seams, so a test that drives a whole engine can make this piston's scan fail (FaultScanErr) without reaching for the database. Nil restores an inert one, which is also the default - a piston built and never told otherwise consults nothing.
The seams are the OWNER's, not this package's, for the same reason the meter is: one catalogue of fault names per application, armed in one place, however many modules consult it. A Seamster built disabled makes every consult a bool read, so a production piston pays nothing for the call site.
This package deliberately has no seam of its OWN, and the distinction matters. A seam inside pure logic would be a signal that a dependency should have been injected instead; this one perturbs the DATABASE query, which is exactly the boundary a test cannot otherwise reach - the pipeline's error policy (clear the shard from planning, leave the cache alone) is only reachable by making a real scan fail. The pipeline itself gets none: its faults are all reachable through the Source, which is this type.
func (*Piston) SetStealAfter ¶
SetStealAfter sets how many cycle periods a step outside this replica's residue class must have been DUE before this replica selects it anyway - and only while its own class is empty. Zero or negative disables stealing, restoring strict residue partitioning.
WHAT IT IS FOR. Partitioning hands each replica a disjoint class of step ids, which is what keeps peers from racing for the same rows - but it also means a replica that is slow rather than DEAD keeps its class while being unable to serve it, and nobody else will look at those steps. Measured on a three-replica fleet with one replica crippled: throughput collapsed to a third of what the same fleet did with that replica REMOVED from the divisor entirely, with its class aging past 30s while its healthy peers sat at a third of a core. Two independent cripplings - a one-worker capacity cap and 10ms of injected latency - produced the same cap, so it is a property of the partitioning rather than of how a replica goes slow.
It fails open on every axis: stealing only ever ADMITS rows, the claim CAS still grants every step, and a replica holding work of its own never steals at all.
func (*Piston) StealAfter ¶
StealAfter is the current steal grace in cycle periods; zero means stealing is off.
Source Files