internal/formation/driver.go
internal/formation · 355 lines · 11 declarations · source
Declarations
type Driver
type Driver struct {
subjects *Subjects
compaction *Compaction
notifications *Notifications
owed OwedScopes
worker *Worker
observations *pg.ObservationStore
retention *pg.RetentionStore
audit *pg.AuditStore
schema pg.Schema
policy Policy
log *slog.Logger
}
Driver runs the backlog without anybody asking it to.
Forming happens behind the append, which is only true if something forms. A worker drains a scope once and returns; this is the thing that decides which scopes, how often, and what to do when a pass makes no progress.
Why it discovers scopes rather than being told them
A scope comes into existence when the first turn is written to it, and nothing announces that. A driver configured with a list of scopes would silently ignore every project created after it started — which is every project, in a system where creating one is writing to it.
Why it is safe to run several
Each scope is drained under an advisory lock, so a second driver finds the scope busy and moves on to the next one rather than forming the same turn twice. That is what makes the Helm chart's several replicas a deployment decision rather than a code change, and it is asserted by a test rather than assumed.
func NewDriver
func NewDriver(worker *Worker, observations *pg.ObservationStore, retention *pg.RetentionStore,
audit *pg.AuditStore, schema pg.Schema, policy Policy, log *slog.Logger, responsive ...func()) *Driver
NewDriver builds the driver over a worker that already knows how to drain one scope.
type Pass
type Pass struct {
Scopes int
Formed int
Busy int
Errored int
Reports int
Segments int
// Notified is how many notifications landed this pass, and Owed how many were newly written
// down. They are counted separately because one is somebody else's endpoint working and the
// other is this deployment noticing it had news.
Notified int
Owed int
// Derived is how many projects were given the derived passes although nothing was waiting to be
// formed in them.
Derived int
}
Pass is one sweep over every scope that has work waiting.
It returns what it did rather than only an error, because "the driver is running" and "the driver is making progress" are different questions and only the second one matters.
method Driver.WithPasses
func (d *Driver) WithPasses(subjects *Subjects, compaction *Compaction) *Driver
WithPasses attaches the subject pass, which partitions a scope and writes its reports, and the compaction pass, which rolls up each subject's history. Both run per scope after its backlog is drained, bounded by the policy, so a deployment writes reports and segments without an operator running anything. Either may be nil.
type OwedScopes
type OwedScopes interface {
ScopesOwingReports(ctx context.Context, writer string, limit int) ([]string, error)
}
OwedScopes names the projects that owe derived work with no backlog to announce them.
method Driver.WithOwedScopes
func (d *Driver) WithOwedScopes(owed OwedScopes) *Driver
WithOwedScopes lets the driver reach a project whose backlog is empty and whose reports are not. Nil, which is what a deployment without it passes, leaves the driver serving backlogs alone.
method Driver.WithNotifications
func (d *Driver) WithNotifications(notifications *Notifications) *Driver
WithNotifications attaches the pass that tells a project its memory moved. Nil, or a deployment whose operator has named no destinations, means no outbound call is ever made.
method Driver.Once
func (d *Driver) Once(ctx context.Context) (Pass, error)
Once drains every scope with a backlog, and returns having tried each of them.
A scope that fails does not stop the pass. The failure is already recorded on the turn that caused it — an attempt counted, a reason kept, and eventually a parked turn — so stopping here would mean one project's bad turn holding up every other project's memory, which is the same mistake at a larger scale than the one parking exists to fix.
method Driver.projections
func (d *Driver) projections(ctx context.Context, scope string, pass *Pass)
projections runs the passes that derive from formed turns: subjects and their reports, then compaction. A failure in either is logged and counted; it is not the scope's formation failing.
method Driver.SweepRetention
func (d *Driver) SweepRetention(ctx context.Context) error
SweepRetention removes what has expired, across every project that has any.
Separate from a formation pass rather than folded into it, because the two answer to different clocks: formation runs when there is a backlog, and retention runs whether or not anything was written. Folding them together would mean a quiet project never expiring anything.
It reports what it removed rather than only whether it ran. A sweep that deletes quietly is indistinguishable from data loss, and the count is what makes "memory disappeared" answerable.
method Driver.Run
func (d *Driver) Run(ctx context.Context) error
Run drives passes until the context ends.
It waits between passes rather than polling continuously, and the wait is the same whether the last pass did work or not. A driver that spun when there was work would be a busy loop against the database on the one path that is already the slowest thing in the system.