engine

Documentation

Overview

Package engine orchestrates runs and steps: claiming work, holding leases and driving state transitions.

Index

• Constants
• type Engine
  • func (e *Engine) DrainRun(ctx context.Context, runID string, ref store.LeaseRef, code reason.Code) (bool, error)
  • func (e *Engine) ExecuteRun(ctx context.Context, runID string) (string, error)
  • func (e *Engine) Fsck(ctx context.Context) ([]store.Violation, error)
  • func (e *Engine) HeldLease(runID string) (store.LeaseRef, bool)
  • func (e *Engine) ReapExpiredRuns(ctx context.Context) ([]store.ReapedRun, error)
  • func (e *Engine) Recover(ctx context.Context, runID string) (string, error)
  • func (e *Engine) RunLeaseRenewals(ctx context.Context) error

Constants

const DefaultPollInterval = 2 * time.Second

DefaultPollInterval is how often a running step's cancellation request is re-read while the process runs. Between steps the request is read directly, so this only bounds how long a cancel can stay invisible mid-run.

const DefaultReapInterval = 10 * time.Second

DefaultReapInterval is how often a leader sweeps for expired run leases. Ten seconds against a sixty second ttl: recovery is bounded by the ttl plus one interval plus the skew, never by the ttl alone.

Types

type Engine

type Engine struct {
	// Store is the database. All reads and writes go through it; the
	// engine holds no handle of its own and opens no transaction.
	Store *store.Store

	// LogRoot is where step logs are written. Paths stored in the
	// database are relative to it.
	LogRoot logsink.Root

	// Clock drives every timing decision: poll ticks, deadlines, the
	// stamps handed to the runner. Never time directly.
	Clock clock.Clock

	// Owner is the name this executor claims leases under. Every write
	// the engine makes on a claimed run must come from the same name.
	Owner string

	// PollInterval bounds how long a cancellation can wait to be seen
	// while a step runs. Zero means DefaultPollInterval.
	PollInterval time.Duration

	// StepTimeoutDefault applies when neither the job nor the step names
	// a timeout. Zero means runner.DefaultTimeout.
	StepTimeoutDefault time.Duration

	LeaseTTL time.Duration

	// RenewInterval is how often held leases renew, in one batched
	// transaction for all of them. Zero means a third of the ttl, which is
	// the ratio that tolerates two lost renewals before ownership is even in
	// question.
	RenewInterval time.Duration

	// ClockSkewAllowance is how long past the expiry the reaper waits before
	// it takes a run from its holder. Zero means the store's default.
	ClockSkewAllowance time.Duration

	// RequeueBackoff is how long a reaped run waits before it is due again.
	// Zero means the store's default.
	RequeueBackoff time.Duration

	// MaxCrashCount is the poison quarantine line: a run that has outlived
	// this many executors fails on the next reap instead of requeueing.
	// Zero means the store's default of five.
	MaxCrashCount int

	// Rnd is the source full-jitter draws from. Nil means one seeded
	// from system entropy on first use. Tests inject a seeded source so
	// backoff sequences replay exactly.
	Rnd *rand.Rand
	// contains filtered or unexported fields
}

Engine executes runs. It owns no state of its own beyond its wiring: every fact about a run lives in the store, and every decision about a transition lives in internal/model. What is left here is sequencing and the one thing nothing else may do, running processes, which happens strictly between transactions, never inside one.

func (*Engine) DrainRun

func (e *Engine) DrainRun(ctx context.Context, runID string, ref store.LeaseRef, code reason.Code) (bool, error)

DrainRun hands one claimed run back at a clean stop. It is the store's transactional handback with this process's own token attached. Whatever the row answers, this process is done answering for the run once the call returns: a handed-back run belongs to the queue, and a refused handback was never ours to give. The entry only leaves when the store call itself landed, so a transient error keeps the claim alive for the next try.

func (*Engine) ExecuteRun

func (e *Engine) ExecuteRun(ctx context.Context, runID string) (string, error)

ExecuteRun takes one queued run and drives it to a terminal state: claim, then steps one at a time in index order, each requiring its every upstream step to have succeeded, then the run's own verdict.

The claim is where ownership becomes real: one statement hands back the fencing token, and every write this attempt makes afterwards carries that token. A write refused for a lost lease ends the attempt with its result discarded and an event saying so; worst case is duplicate work, never duplicate state.

The shape of the loop is where the other hard rules live. Every state change is a store method, and each of those commits the state change with exactly one event row. The process runs strictly between those transactions: the transaction that starts the step is closed before the command is spawned, and the transaction that records its verdict is opened after it is reaped.

func (*Engine) Fsck

func (e *Engine) Fsck(ctx context.Context) ([]store.Violation, error)

Fsck sweeps the database for broken invariants and returns everything it finds. The checks themselves are SQL and live beside the rest of the SQL, in internal/store; this wrapper is the entry point the hidden command and the crash harness call. A healthy database returns an empty list.

func (*Engine) HeldLease

func (e *Engine) HeldLease(runID string) (store.LeaseRef, bool)

HeldLease reports the fencing token this process believes it holds for a run. The belief may be stale; every caller hands it straight back to the store, which is where it gets checked.

func (*Engine) ReapExpiredRuns

func (e *Engine) ReapExpiredRuns(ctx context.Context) ([]store.ReapedRun, error)

ReapExpiredRuns runs one sweep against the store with this engine's timing. The returned slice names every run the sweep took. The looping belongs to whichever process drives the sweeps; in a daemon that is the reaper loop under the reaper role lease, so leadership and cadence live in one place.

func (*Engine) Recover

func (e *Engine) Recover(ctx context.Context, runID string) (string, error)

Recover closes out what a dead executor left behind and makes the run claimable again. It is the restart half of the guarantee the crash harness (#75) proves: a run interrupted by SIGKILL converges on restart, without an invariant violation and without inventing a verdict.

Three steps, in this order:

1. Wait for the abandoned lease to expire. A lease that still lives may belong to a process that is only slow, and two executors on one run is the one outcome fencing exists to prevent.
2. Close every step the dead attempt left running. Each goes through the machine as a failed attempt with STEP_FAILED_EXECUTOR_LOST: the verdict was lost with the executor, so recovery records exactly that instead of guessing. A step with attempts left comes back pending for its next attempt; otherwise it fails and the run will follow.
3. Requeue the run through the store's lease_expired transition, which bumps the epoch, counts the crash and writes defer_reason.

A run that is not running needs nothing and is returned untouched, so a caller may call Recover before every ExecuteRun without checking first.

func (*Engine) RunLeaseRenewals

func (e *Engine) RunLeaseRenewals(ctx context.Context) error

RunLeaseRenewals keeps every claim this process holds alive until the context ends. The daemon runs it beside its executors; a foreground run command runs it around ExecuteRun. Errors from individual ticks never end the loop: losing one renewal is survivable by design, and ending the loop would guarantee losing all of them.