store

package
v0.0.0-...-b8a15ae Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 24, 2026 License: Apache-2.0 Imports: 29 Imported by: 0

Documentation

Overview

Package store owns the SQLite database: connection pools, PRAGMAs, migrations and every SQL query in the project.

The write model

One file, two pools. The writer pool has exactly one connection, so database/sql is the write queue: it serialises writers, applies backpressure and honours context deadlines, and no SQLITE_BUSY can come from paceq's own writers because there are never two. Every writer transaction begins IMMEDIATE, which avoids the lock upgrade that busy_timeout cannot retry. The reader pool opens read-only with query_only, so the engine rather than convention enforces that reads cannot write.

The payoff is that read, compute in Go, then write is safe inside one transaction without optimistic locking. Nobody else can have written in between.

Five rules

Breaking one of these is a review stopper.

  1. No process execution, file I/O or network I/O inside a write transaction, ever. The single write connection is held for the whole callback, and lock hold time is the real scarcity in this system.
  2. All mutation goes through methods on *Store. The writer handle is private, and an architecture test fails the build if a database handle reaches the exported API.
  3. Admission control is read, compute, write inside one IMMEDIATE transaction, not a clever atomic statement.
  4. RETURNING rows are consumed in full before the transaction runs anything else. The writer pool has one connection to lose.
  5. The database never runs on a network or FUSE filesystem. SQLite file locking is undefined there and the corruption shows up weeks later.

Schema conventions

These hold for every table in the schema, and a new table that breaks one is a review stopper. Retrofitting any of them means rebuilding every table.

  1. Every table is STRICT. A column declared INTEGER holds integers, and a string that looks like a number is an error rather than a silent conversion. This needs SQLite 3.37, which the driver ships.
  2. Every time column is INTEGER unix milliseconds UTC. Sortable, indexable, no parsing, no timezone ambiguity. A timezone is stored separately as an IANA name, and only where a time has to be interpreted rather than ordered.
  3. Every status is TEXT with a CHECK listing the values it may take. The premise of this product is that reading the tables with the sqlite3 shell explains what happened, which integer codes do not, and the CHECK mirrors the state machine in internal/model so both ends enforce it.
  4. Structured values are canonical JSON in a TEXT column: object keys sorted, no insignificant whitespace. Canonical form is what makes a hash of the text stable and two rows comparable.

The schema a migrated database ends up with is checked in as schema.golden.sql, and TestGoldenSchema fails on any change to it. Reviewing a schema change means reading that diff.

auto_vacuum is INCREMENTAL, set by Open while the database still has no schema. It is the one setting here that cannot be changed afterwards without a full VACUUM holding an exclusive lock, so it is decided at creation. Databases created before paceq set it keep NONE, and the doctor check reports them.

The state directory

A state directory holds one lock file and one database, and one process owns both. OpenState takes an exclusive flock on the lock file before the database is opened for writing, so a second paceq is refused before it touches a page of a file somebody else owns, and it is told which process to stop. The lock lives in the kernel: it is released when the process dies, however it dies, so there is no stale lock to clean up and the lock file is kept between runs.

The lock covers a state directory, not a database file. Two processes pointed at the same database through different state directories both start, which the role leases in M2-02 are what make safe.

StartSession records who is running, from when, and on which boot. A session row still open at the next start belongs to a run that never got to say goodbye, and is marked crashed. The boot id, read from /proc/sys/kernel/random/boot_id, is the strongest evidence in the system: a changed one means the machine restarted, so no process paceq started can have survived. Platforms without it degrade to lease expiry, which is slower and still correct.

Migrations

Migrate applies the SQL files embedded from the migrations directory, in version order, on the write connection. Forward only, one migration per transaction, sha256 pinned per applied file, and PRAGMA user_version as the fence that stops an old binary from writing to a newer database. The rules a migration file has to follow are in migrations/README.md.

PRAGMA values are read back from both pools at startup. A mismatch fails Open, naming the setting. Misconfigured durability is a startup error, never a warning and never a quiet degradation.

Rules 1 and 4 cannot be checked mechanically and are enforced in review. Rules 2, 3 and 5 are covered by tests in this package and in internal/arch.

Index

Constants

View Source 
const (
	// DefaultRunLeaseTTL is how long a claim lasts. Sixty seconds, renewed
	// every twenty.
	DefaultRunLeaseTTL = 60 * time.Second

	// DefaultClockSkewAllowance is how long past the expiry the reaper waits
	// before it takes a run. Ten seconds.
	DefaultClockSkewAllowance = 10 * time.Second

	// DefaultRequeueBackoff is how long a reaped run waits before it is due
	// again. The crash already cost a ttl; the backoff keeps a job that kills
	// its executor from burning the queue the moment it is requeued.
	DefaultRequeueBackoff = 30 * time.Second

	// DefaultMaxCrashCount is the poison quarantine line (02 section 5.7):
	// once a run has outlived this many executors, the next reap fails it for
	// good instead of requeueing it.
	DefaultMaxCrashCount = 5
)

The run lease. Ownership of a run is explicit and time bounded: one statement claims, one transaction renews every run the owner holds at once, the reaper takes what expired and hands it to a new holder, and every write from a holder carries the fencing token so a frozen worker can never overwrite its successor's verdict (issue #60).

The timing constants are the ones the issue fixes. The ttl is renewed every tick a third its length, so two lost renewals are tolerated before leadership of a run is even in question. The skew allowance is what keeps the reaper late on purpose: the owner measures its budget on a monotonic clock and gives up early; the reaper waits out the wall clock plus skew and takes over sent.

View Source 
const (
	OutcomeTriggered = "triggered"
	OutcomeSkipped   = "skipped"
	OutcomeError     = "error"
)

Tick outcomes as stored in ticks.outcome.

View Source 
const (

	// DirMode and DatabaseMode are the only modes paceq accepts on its own
	// state. Anything wider means the state was readable by another user, which
	// is a refusal rather than something to correct quietly.
	DirMode      fs.FileMode = 0o700
	DatabaseMode fs.FileMode = 0o600
)
View Source 
const DatabaseFileName = "state.db"

DatabaseFileName is the database inside a state directory. The name is fixed so a state directory is self describing: one lock, one database, one owner.

View Source 
const DefaultDeferBackoff = 500 * time.Millisecond

DefaultDeferBackoff is how far a queued overlap defers its run. It is a policy number with one constraint: when the blocking run ends, the deferred run must already be due, so a release costs a wake rather than a wait. Half a second keeps every release inside the one second the acceptance criterion allows, with the whole backoff left as slack.

View Source 
const UnexplainedReasonSQL = `` /* 680-byte string literal not displayed */

UnexplainedReasonSQL is the audit query behind the reason code rule (06 section 2.1): it lists every terminal run, step, tick and trigger that sits in the database without a usable reason code. "Without a usable code" means NULL, empty, or the literal UNKNOWN, which is the value a rotting catalogue reaches for; the catalogue itself holds no UNKNOWN code, so a stored one is always a bug.

The schema CHECKs refuse the NULL and empty cases at write time. This query exists for the cases a CHECK cannot see: rows written before a constraint existed, and the UNKNOWN case, which no CHECK names. It returns zero rows on a healthy database, and testutil.AssertNoUnknownReasons runs it as the teardown assertion of every integration test from M1 on. It later becomes part of `paceq fsck` (M1-12).

Variables

View Source 
var (
	// ErrRunNotFound is returned when no run has the id, or the id prefix.
	ErrRunNotFound = errors.New("no run has that id")

	// ErrAmbiguousRunID is returned when an id prefix matches more than one
	// run. The caller types more characters; nobody guesses on their behalf.
	ErrAmbiguousRunID = errors.New("the id prefix matches more than one run")

	// ErrConcurrencyKeyHeld is returned when an active run already holds the
	// concurrency key. It is an ordinary outcome, not a fault: the caller
	// records a skip with a reason code and moves on.
	ErrConcurrencyKeyHeld = errors.New("an active run already holds the concurrency key")
)

Errors callers act on rather than report. Everything else that comes out of this file is a failure and reads as one.

View Source 
var (
	ErrNotClaimable   = errors.New("the run cannot be claimed")
	ErrStepNotPending = errors.New("the step is not pending")
)

ErrNotClaimable is returned when a run cannot be claimed: it is not queued, it is not available yet, or a cancellation is waiting to be observed. It is an ordinary outcome for a worker polling for work, not a fault.

View Source 
var ErrLeaseLost = errors.New("the run lease was lost")

ErrLeaseLost is what every result write answers when the writer's lease has gone: another holder claimed after a reap, or the row moved on. A caller that sees it must discard its result and stop working on the run; the engine turns that into a run.result_discarded event and kills its process group.

View Source 
var ErrNotFound = errors.New("no such row")

ErrNotFound wraps every refusal of this file that means "no such row". A caller that only cares whether something existed matches this; one that cares which table it asked about reads the message.

View Source 
var ErrReadOnly = errors.New("the store is open read only")

ErrReadOnly is returned when a store opened through OpenReadOnly is asked to write. It lives beside the transaction machinery rather than beside any one writer, because refusing is the transaction machinery's own act.

Functions

func CheckMode 

func CheckMode(path string, want fs.FileMode) error

CheckMode refuses a path that is readable by anyone but its owner. Widening permissions back is deliberately not offered: correcting them quietly would hide that the state has been exposed, which is the fact an operator needs.

func KnownSchemaVersion 

func KnownSchemaVersion() (int, error)

KnownSchemaVersion is the highest schema version this build carries. It answers without a database, because version has to answer on a machine that has none.

func ProcessStartTicksReadable 

func ProcessStartTicksReadable() bool

ProcessStartTicksReadable reports whether this platform can read start ticks at all. It is how the sweep decides between verifying and failing closed without pretending.

func ReadProcessStartTicks 

func ReadProcessStartTicks(pid int) (int64, bool)

ReadProcessStartTicks returns field 22 of /proc/<pid>/stat: the kernel's record of when the process started, in clock ticks since boot. Together with the pid it forms the process identity the orphan sweep verifies before any signal (issue #62): a pid alone is recyclable, pid plus start ticks is not.

ok is false when the process does not exist any more, has already been reaped, or cannot be read by this user. Every caller fails closed on false: an unreadable process is nobody's to kill.

Types

type AttemptProcess 

type AttemptProcess struct {
	RunID      string
	Step       string
	PID        int
	StartTicks int64
}

AttemptProcess is one attempt's recorded process identity: which pid led its process group, and what /proc said its start time was when it was spawned. The pair is what lets the orphan sweep tell a surviving child of a dead executor from a recycled pid that happens to carry the same number.

Its write lives beside the other step transitions; this file only reads baselines back for the sweep.

type AutoVacuumMode 

type AutoVacuumMode int

AutoVacuumMode is the file level setting that decides whether SQLite ever returns freed pages to the filesystem. It is stored in the database header, so it survives restarts and belongs to the file rather than the connection. 

const (
	// AutoVacuumNone never shrinks the file. Deleted pages are reused, and the
	// file only ever grows.
	AutoVacuumNone AutoVacuumMode = 0
	// AutoVacuumFull moves pages on every commit, which puts the cost on the
	// hot path.
	AutoVacuumFull AutoVacuumMode = 1
	// AutoVacuumIncremental keeps a free page list that PRAGMA
	// incremental_vacuum releases in bounded batches, off the hot path.
	AutoVacuumIncremental AutoVacuumMode = 2
)

func (AutoVacuumMode) String 

func (m AutoVacuumMode) String() string

type BeginSensorTickInput 

type BeginSensorTickInput struct {
	SensorName      string
	CursorBefore    string
	DaemonSessionID string
	Now             time.Time
}

BeginSensorTickInput names the sensor evaluation that is about to start.

type BeginSensorTickResult 

type BeginSensorTickResult struct {
	// TickID is the intention row CommitSensorTick will close.
	TickID string

	// CursorVersion is the CAS guard read at the start. CommitSensorTick
	// only advances the cursor when this value still holds, so a commit from
	// an evaluation that was already overtaken is refused instead of writing
	// an old result over a new one.
	CursorVersion int64
}

BeginSensorTickResult is everything the commit half needs to guard the evaluation it is about to record.

type CancelRequest 

type CancelRequest struct {
	CancelRequestedAt time.Time
	CancelRequestedBy string
	CancelReason      string
}

CancelRequest is the durable cancellation record read back after it was written. The request is not a transition and carries no event: the event belongs to whoever observes the request and stops the run (02 section 5.8).

type ClaimSpec 

type ClaimSpec struct {
	// Owner is the executor name every later write must come from.
	Owner string

	// TTL is how long each claim lasts. Zero means DefaultRunLeaseTTL.
	TTL time.Duration

	// Limit caps the batch. Zero means one.
	Limit int

	// Only restricts the claim to these ids. An executor claiming the exact
	// run it was handed uses this; the empty slice claims whatever is due.
	Only []string
}

ClaimSpec says who claims and how much.

type ClaimedRun 

type ClaimedRun struct {
	ID           string
	JobName      string
	JobVersionID string
	RunKey       string
	Attempt      int
	LeaseEpoch   int64
	ParamsJSON   string
}

ClaimedRun is one run this process just took: everything ExecuteRun needs to drive it, with the fencing token first among equals.

type CursorInput 

type CursorInput struct {
	Name   string
	Cursor string
}

CursorInput names a sensor and the value to move its cursor to. Moving the cursor is deliberately distinct from a reset: it touches cursor and the cursor guard, and nothing about the dedup table. This is the store-level half of `cursor set` (the M3-06 CLI arrives later).

type FinishReason 

type FinishReason struct {
	Code reason.Code
	Data string
}

FinishReason is how a run ends: the reason code the machine validated and the canonical detail object beside it, such as which step failed.

type GapSchedule 

type GapSchedule struct {
	JobName  string
	Name     string
	Expr     string
	Timezone string
}

GapSchedule is one enabled schedule as the gap walk needs it: enough to recompute where its fire times fell inside an outage.

type IntegrityReport 

type IntegrityReport struct {
	// Integrity is pragma integrity_check's verdict, "ok" when the file
	// is sound.
	Integrity string

	// ForeignKeyIssues counts foreign_key_check rows: references that
	// point at nothing. Zero is the only healthy number.
	ForeignKeyIssues int
}

IntegrityReport is what SQLite itself says about the database file after a crash. WAL recovery on open fixes torn transactions; these two pragmas are the proof that it did.

type JobApplyResult 

type JobApplyResult struct {
	// JobName is the job the input named.
	JobName string

	// VersionID is the id of the version this spec maps to after the apply:
	// the new row when one was written, the existing row otherwise.
	VersionID string

	// Version is that row's number.
	Version int

	// Created reports whether this apply wrote a new job_versions row. A false
	// here is the idempotent case: the file was already loaded as it stands,
	// and the database came out of the transaction byte for byte the same.
	Created bool

	// Sensors is the sensor sync that ran in the same transaction, so a job's
	// new definition and its sensor rows land and are reported together.
	Sensors SyncResult
}

JobApplyResult says what apply did with one job spec.

type JobRunSummary 

type JobRunSummary struct {
	JobName    string
	RunID      string
	State      string
	ReasonCode string
	CreatedAt  time.Time
	StartedAt  time.Time
	FinishedAt time.Time
	StepsTotal int
	StepsDone  int

	// The deferral facts (#68), read straight off the newest run so status
	// can say "utsatt" with its reason without a second query.
	AvailableAt time.Time
	DeferReason string
	ReasonData  string
}

JobRunSummary is what paceq status shows for one job: the job's newest run and how far its steps got. A job that has never run carries an empty RunID and state.

type JobVersion 

type JobVersion struct {
	ID         string
	JobName    string
	Version    int
	SpecHash   string
	SpecJSON   string
	SourcePath string
	CreatedAt  time.Time
}

JobVersion is one immutable snapshot of a job spec.

type JobVersionInput 

type JobVersionInput struct {
	// JobName is the stable human identity of the job, and its primary key.
	JobName string

	Description string

	// SourcePath is where the spec was read from, for messages. Empty when the
	// spec came from somewhere that has no path.
	SourcePath string

	// MaxConcurrent is how many runs of this job may be active at once. Zero
	// means one, which is the default a job that says nothing gets.
	MaxConcurrent int

	// SpecHash is the digest of the canonical spec, "sha256:<hex>". It is the
	// whole of idempotent reload: the same digest is the same version.
	SpecHash string

	// SpecJSON is the canonical spec itself, the intermediate representation a
	// run is materialised from.
	SpecJSON string

	// Sensors are the sensor definitions of the job, materialised as rows in
	// the same transaction that writes the version. Nil or empty means the
	// job declares no sensors, and any sensor rows it owned are removed.
	Sensors []spec.Sensor
}

JobVersionInput is one job spec as the loader read it. It carries no id and no version number: which version this spec is, is a fact about the database and not about the file.

type JobView 

type JobView struct {
	Name          string
	Description   string
	MaxConcurrent int

	// Paused is the operator pause flag. A paused job still admits what a
	// person starts; it only stands its schedules down.
	Paused bool

	SourcePath string

	// CurrentVersion is the version number the job currently points at,
	// zero when no version has ever been applied to it.
	CurrentVersion int

	CreatedAt time.Time
	UpdatedAt time.Time
}

JobView is one job as `paceq jobs show` reads it back: the identity, the ceiling that admission control enforces, and where the job came from.

type LeaseEvent 

type LeaseEvent struct {
	At     time.Time
	Lease  string
	Holder string
	Epoch  int64
	Code   reason.Code
}

LeaseEvent is one recorded moment in a lease's life: taken, lost or taken over from a dead holder. It is the stored explanation beside the structured log line, written through AppendLeaseEvent.

type LeaseGrant 

type LeaseGrant struct {
	Name       string
	Holder     string
	Epoch      int64
	AcquiredAt time.Time
	ExpiresAt  time.Time
}

LeaseGrant is one live lease row as the database computed it.

A lease is the right for one holder to act as the singleton of one role: "scheduler" decides when jobs fire, "reaper" reaps what died, and so on. Epoch is the fencing token: it grows by exactly one every time the lease changes hands through an expiry, and never moves on a renewal, so a stale holder can always be told apart from the current one by comparing numbers.

type LeaseInput 

type LeaseInput struct {
	// Owner is the executor's name. Every later write to this run has to
	// come from the same name, which is what makes the lease a fence and
	// not a decoration.
	Owner string

	// TTL is how long the claim lasts. Zero means DefaultRunLeaseTTL.
	TTL time.Duration
}

LeaseInput says who claims a run and for how long.

type LeaseRef 

type LeaseRef struct {
	Owner string
	Epoch int64
}

LeaseRef is who claims to hold a run's lease, and at which fencing token. Every result write proves its standing by carrying one.

The zero LeaseRef means the system itself, writing against an executor that is already gone: recovery closing a dead attempt's steps. That path is real, and it is gated the other way round: the store refuses it while the lease is still live, because a live lease means someone else is driving.

type LeaseRenewal 

type LeaseRenewal struct {
	ID                string
	LeaseEpoch        int64
	CancelRequestedAt time.Time
	CancelRequestedBy string
}

LeaseRenewal is one held run's answer to the heartbeat: the token the owner still holds, and any cancellation request that arrived since. The answer is the whole two way channel (11 section 4.3): cancellation needs no push, no extra polling and no second mechanism, because every renewal carries it.

type LockedError 

type LockedError struct {
	Path string
	Err  error
	// Owner is the session row of the process holding the lock, when the
	// database could be read and named one. Its message says so when it is nil.
	Owner *Session
}

LockedError is returned when another process already owns the state directory. It is a distinct type so a caller can tell "somebody else is running" from "this directory is broken".

func (*LockedError) Error 

func (e *LockedError) Error() string

Error explains the refusal the way every paceq error does: what went wrong, where, and what to do next. The owner is named from the session row when there is one, because "already running" without a pid is not actionable.

func (*LockedError) Unwrap 

func (e *LockedError) Unwrap() error

type LogMeta 

type LogMeta struct {
	RelPath   string
	Bytes     int64
	Truncated bool
	ErrorTail string
}

LogMeta is what the log sink reported for one attempt: where the file is relative to the log root, how big it grew, whether the quota cut it, and the last lines of output. It lands beside the verdict, in the same transaction, because a verdict whose evidence went missing is exactly the drift the write model refuses.

type ManualTriggerInput 

type ManualTriggerInput struct {
	// JobName is the job whose current version runs.
	JobName string

	// Actor is who typed the command, recorded on the queued event and on
	// the tick's session trail. Empty becomes "system".
	Actor string

	// ParamsJSON is the parameter object for the run, empty for none.
	ParamsJSON string
}

ManualTriggerInput is one person's decision to run a job now.

type ManualTriggerResult 

type ManualTriggerResult struct {
	TickID    string
	TriggerID string
	Run       Run
}

ManualTriggerResult names everything one manual decision created. The three rows are born in one transaction, so any later reader sees all of them or none of them.

type MissedTick 

type MissedTick struct {
	SourceName   string
	ScheduledFor time.Time
}

MissedTick is one schedule slot inside a gap that nobody evaluated.

type NewRun 

type NewRun struct {
	JobName      string
	JobVersionID string

	// TriggerID is the decision this run came out of, empty for a run nobody
	// triggered, such as a manual start.
	TriggerID string

	// Origin says which mechanism produced the run: schedule, sensor, manual,
	// retry, replay or backfill. The schema refuses anything else.
	Origin string

	// RunKey is the dedup key the trigger carried, empty when there is none.
	RunKey string

	// ConcurrencyKey caps how many runs sharing it may be active. Empty means
	// unlimited. A second active run on one key fails with
	// ErrConcurrencyKeyHeld, decided by the database and not by a read here.
	ConcurrencyKey string

	// ScheduledFor is the logical slot a schedule or backfill run belongs to,
	// zero for everything else. It is not when the run will start.
	ScheduledFor time.Time

	// AvailableAt is the earliest the run may be claimed. Zero means now. A
	// time in the future needs DeferReason: a run held back always says why.
	AvailableAt time.Time
	DeferReason string

	// ParamsJSON is the canonical parameter object, empty for none.
	ParamsJSON string

	// MaxAttempts is how many times the run may be attempted. Zero means one.
	MaxAttempts int

	// ReplayOf is the run this one replays, empty when it replays nothing.
	ReplayOf string

	// Actor is who caused this, recorded on the queued event. Empty is system.
	Actor string

	// Steps are the run's steps in spec order, which is the order M1 runs them
	// in. Their dependency edges are frozen here and are never read from the
	// spec again.
	Steps []NewStep
}

NewRun is a run to materialise, with the steps it will run.

type NewStep 

type NewStep struct {
	Name string

	// DependsOn are the steps that have to finish first. The names have to be
	// steps of the same run; nothing checks that here, because M1 runs steps in
	// order and M4-02 is what reads the edges.
	DependsOn []string

	// MaxAttempts is how many times the step may be attempted. Zero means one.
	MaxAttempts int
}

NewStep is one step of a run to materialise.

type Options 

type Options struct {
	// Synchronous is "normal" (default) or "full". Nothing else is accepted.
	Synchronous string

	// AllowNetworkFS skips the refusal to run on a network or FUSE filesystem.
	AllowNetworkFS bool

	// Clock is the clock the retry backoff waits on. A nil Clock means
	// clock.System(). Tests that want the backoff to be instant pass their own.
	Clock clock.Clock
}

Options configures Open.

type Outage 

type Outage struct {
	ID          int64
	From        time.Time
	To          time.Time
	DetectedAt  time.Time
	Kind        string
	PrevSession string
	MissedTicks int
}

Outage is one stored downtime record.

type OutageInput 

type OutageInput struct {
	// From and To bound the unaccounted period. From is the last heartbeat of
	// the session that died; To is when the replacement started.
	From time.Time
	To   time.Time

	// Kind is the schema's word for what happened: 'crash' or 'boot' from
	// startup reconciliation, 'clock_jump' from jump detection.
	Kind string

	// PrevSession is the daemon_sessions row whose death opened the gap, or
	// empty when no session row survived to name (a first start after a wipe).
	PrevSession string
}

OutageInput names the downtime an outage row explains.

type PermissionError 

type PermissionError struct {
	Path string
	Got  fs.FileMode
	Want fs.FileMode
}

PermissionError is a path another user can read. It is a distinct type so a caller can tell a refusal it can explain and act on from a failure it cannot, and so the CLI can map it to its own exit code rather than report it as an internal error.

func (*PermissionError) Error 

func (e *PermissionError) Error() string

type ReapOptions 

type ReapOptions struct {
	// Skew is the wall clock allowance past the expiry before a lease counts
	// as dead. Zero means DefaultClockSkewAllowance.
	Skew time.Duration

	// Backoff is how long a requeued run waits before it is due again.
	Backoff time.Duration

	// MaxCrashCount is the poison line. Zero means DefaultMaxCrashCount.
	MaxCrashCount int

	// Limit caps the sweep. Zero means reapLimit.
	Limit int

	// IgnoreLease widens the sweep from "leases that expired" to "every
	// running run". It exists for startup reconciliation after a machine
	// restart (#62): a changed boot id proves every child process dead, so
	// lease_expires_at has stopped being evidence about anything and waiting
	// it out would be pure delay. Nothing else may set it. The decision arms
	// are the same either way; only candidate selection changes, and the
	// fencing token still rises in every arm.
	IgnoreLease bool
}

ReapOptions tunes one sweep. Zero fields mean the defaults above.

type ReapedRun 

type ReapedRun struct {
	ID         string
	State      string
	ReasonCode string
	CrashCount int
	Attempt    int
	LeaseEpoch int64
}

ReapedRun is one run the reaper took, with where it went.

type ResetResult 

type ResetResult struct {
	Sensor     string
	OldEpoch   int64
	NewEpoch   int64
	Cursor     *string
	ForgotKeys bool
}

ResetResult reports what one reset did: the epoch it started from and where the bump took it, so the caller (the CLI, later) can tell the operator.

type ResetSensorInput 

type ResetSensorInput struct {
	Name          string
	SetCursor     *string
	ForgetRunKeys bool
}

ResetSensorInput names the sensor being reset and the scope of the reset. A nil Cursor sets the cursor to NULL (the full "start over" form); a value replays from a chosen point. ForgetRunKeys erases the sensor's run_key rows; it is the only path that deletes them, and is never implicit.

type RetryPlan 

type RetryPlan struct {
	// NextAttemptAt is when the step becomes runnable again. Zero falls
	// back to now, the M1 behaviour before backoff existed.
	NextAttemptAt time.Time

	// ReasonCode names the scheduled retry on the row and its event.
	// Empty keeps the outcome's own code.
	ReasonCode reason.Code

	// DetailJSON replaces the event's detail object on the pending path,
	// where the facts that matter are the attempt number and the
	// backoff, not the exit verdict.
	DetailJSON string
}

RetryPlan is the schedule a caller computed for a further attempt of a step whose machine transition went back to pending. The machine still decides whether the step goes back to pending; the plan only fills in the when and the words. Backoff arithmetic lives with the caller, which holds both the policy and the clock.

type Run 

type Run struct {
	ID             string
	JobName        string
	JobVersionID   string
	TriggerID      string
	Origin         string
	RunKey         string
	State          string
	ConcurrencyKey string
	AvailableAt    time.Time
	DeferReason    string
	ScheduledFor   time.Time
	ParamsJSON     string
	Attempt        int
	MaxAttempts    int

	// LeaseOwner and LeaseEpoch say who is executing the run now. The epoch
	// is the fencing token: it rises on every claim and on every reap, so a
	// writer whose lease was taken can always be told apart from the current
	// holder by comparing numbers. Every result write proves its standing by
	// carrying both.
	LeaseOwner     string
	LeaseEpoch     int64
	LeaseExpiresAt time.Time

	// HeartbeatAt is the last renewal of the lease. It moves in one batched
	// transaction for every run an owner holds.
	HeartbeatAt time.Time

	// CrashCount is how many executors have died holding this run's lease.
	// Crossing the reaper's crash budget ends the run in the poison
	// quarantine instead of offering it to yet another executor.
	CrashCount int

	// The cancellation request, durable before anything is killed. Empty
	// time means nobody asked.
	CancelRequestedAt time.Time
	CancelRequestedBy string
	CancelReason      string

	ReasonCode string
	ReasonText string

	// ReasonData is the canonical detail object beside the reason code: for
	// a failed run, which step failed and why the run ended as it did.
	ReasonData string

	Error      string
	CreatedAt  time.Time
	StartedAt  time.Time
	FinishedAt time.Time
	UpdatedAt  time.Time
}

Run is one attempt at one job version.

type RunDetail 

type RunDetail struct {
	Run
	Steps []Step
}

RunDetail is a run with its steps, which is what showing one run needs. Events are not here: they are the explain query, and they are read on their own because most callers do not want them.

type RunEvent 

type RunEvent struct {
	RunID string

	// StepName is set on an event about a step, empty on one about the run.
	StepName string

	// At is when it happened. Zero means now.
	At time.Time

	// Kind names the transition: run.queued, step.started, step.retry_scheduled
	// and so on.
	Kind       string
	FromState  string
	ToState    string
	ReasonCode string

	// Actor is who caused it: system, reaper, operator, cli:<uid>. Empty is
	// system.
	Actor string

	// DetailJSON is the canonical detail object. Empty is an empty object.
	DetailJSON string
}

RunEvent is one state transition, as explain will read it back.

type RunFilter 

type RunFilter struct {
	JobName string

	// States restricts the listing to these run states. Empty means all.
	States []string

	// Before is the keyset cursor: only runs whose id sorts below it. Ids are
	// ULIDs, so this walks the history backwards in time. Empty starts at the
	// newest run.
	Before string

	// Limit caps the page. Zero is 50 and anything above 500 is 500.
	Limit int
}

RunFilter narrows a listing. The zero filter lists the newest runs of every job.

type RunSummary 

type RunSummary struct {
	ID         string
	JobName    string
	Origin     string
	State      string
	ReasonCode string
	CreatedAt  time.Time
	StartedAt  time.Time
	FinishedAt time.Time

	// The deferral facts (#68): available_at decides whether a queued row
	// reads as deferred, defer_reason names why it waits, and reason_data
	// carries the blocking run id the display points at.
	AvailableAt time.Time
	DeferReason string
	ReasonData  string
}

RunSummary is one line of a listing.

type ScheduleInput 

type ScheduleInput struct {
	JobName         string
	Name            string
	Kind            string
	Expr            string
	Timezone        string
	SpringForward   string
	FallBack        string
	Catchup         string
	CatchupLimit    int
	CatchupWindowMS int64

	// Overlap is the policy for ticks that fire while the job's concurrency
	// limit is held: "skip" (default) or "queue".
	Overlap string

	Paused     bool
	NextTickAt time.Time
}

ScheduleInput names a schedule to create or replace. Fields left empty take the schema defaults.

type ScheduleRow 

type ScheduleRow struct {
	ID              string
	JobName         string
	Name            string
	Kind            string
	Expr            string
	Timezone        string
	SpringForward   string
	FallBack        string
	Catchup         string
	CatchupLimit    int
	CatchupWindowMS int64

	// Overlap is what a tick does when the job's max_concurrent is already
	// held: "skip" stands down and records why, "queue" materialises the run
	// deferred into the future with a defer_reason. Empty means skip.
	Overlap string

	Paused     bool
	LastTickAt *time.Time
	NextTickAt time.Time
	CreatedAt  time.Time
	UpdatedAt  time.Time
	// contains filtered or unexported fields
}

ScheduleRow is one schedule as the due query returns it. Every column the loop needs to decide what fired comes back at once, so processing one schedule costs one read.

type SensorRow 

type SensorRow struct {
	Name          string
	JobName       string
	Kind          string
	IntervalMS    int64
	MinIntervalMS int64
	TimeoutMS     int64
	Paused        bool
	Cursor        string
	CursorVersion int64
	DedupEpoch    int64
	NextEvalAt    int64
}

SensorRow is one sensor as the commit path reads it: the cursor the evaluation started from, the version that guards it, and the dedup epoch that makes a cursor reset safe.

type SensorSummary 

type SensorSummary struct {
	Name               string
	JobName            string
	Kind               string
	ExecJSON           string
	SpecJSON           string
	IntervalMS         int64
	MinIntervalMS      int64
	TimeoutMS          int64
	MaxTriggersPerTick int
	Paused             bool
	NextEvalAt         time.Time
	CreatedAt          time.Time
	UpdatedAt          time.Time
}

SensorSummary is the read-side view of one sensor row as the readers here return it.

type SensorTickCommitInput 

type SensorTickCommitInput struct {
	// TickID is the intention row opened by BeginSensorTick.
	TickID string

	SensorName string
	JobName    string

	// CursorVersion is the guard BeginSensorTick read. It is compared inside
	// the transaction; a stale value refuses the commit.
	CursorVersion int64

	// CursorAfter moves the sensor cursor when Outcome is triggered. It is
	// the value the whole evaluation built towards.
	CursorAfter string

	// DedupEpoch is the sensor's dedup epoch, the same value BeginSensorTick
	// saw on the row. It prevents a cursor reset from replaying an old key
	// into an already-ringed table.
	DedupEpoch int64

	// Triggers are the run keys this evaluation decided to fire. Each is
	// registered against the run_keys gate first, so a replayed evaluation
	// folds into the run that already exists instead of creating a twin.
	Triggers []SensorTrigger

	// Outcome is OutcomeTriggered, OutcomeSkipped or OutcomeError. Only a
	// triggered evaluation moves the cursor and creates runs.
	Outcome string

	// ReasonCode and ReasonText belong on a skipped or error tick. A skipped
	// sensor's own reason travels verbatim (plan: the sensor says why it
	// skipped and paceq does not paraphrase).
	ReasonCode reason.Code
	ReasonText string

	// NextEvalAt is when the sensor becomes due again. Set on every commit.
	NextEvalAt int64

	// DurationMs is how long the evaluation took, for the tick row.
	DurationMs int64

	Now time.Time
}

SensorTickCommitInput is everything a completed sensor evaluation decided, ready to be written or refused as one unit.

type SensorTickCommitResult 

type SensorTickCommitResult struct {
	// Accepted is how many new runs were created.
	Accepted int

	// Deduped is how many triggers folded into an existing run.
	Deduped int

	// RunIDs are the ids of the accepted runs, in trigger order.
	RunIDs []string

	// Fenced is true when the commit was refused because the sensor already
	// advanced past the cursor_version this evaluation started from. Nothing
	// was written except the tick being marked error with TICK_MISSED_LEASE_LOST.
	Fenced bool
}

SensorTickCommitResult reports what one atomic commit did.

type SensorTrigger 

type SensorTrigger struct {
	RunKey     string
	ParamsJSON string
}

SensorTrigger is one trigger a sensor evaluation produced: the run key that deduplicates it and the parameters that vary per trigger. Each becomes a triggers row and, on a new run key, a runs row.

type Session 

type Session struct {
	ID         string
	Version    string
	BootID     string
	PID        int
	StartedAt  time.Time
	LastSeenAt time.Time
	StoppedAt  time.Time
	StopReason string
}

Session is one run of paceq against this database. Its id is the instance identity: a ULID minted at startup, not a process id, because process ids are recycled and a recycled one would make a dead run look like a live one both in the history and in a lease.

func (Session) Stopped 

func (s Session) Stopped() bool

Stopped reports whether the run this session describes has ended.

type StateLock 

type StateLock struct {
	// contains filtered or unexported fields
}

StateLock is the exclusive claim on one state directory. It is held by an open file description for as long as the process lives, so no second paceq can write to the same state, and the kernel releases it even when the process is killed.

func AcquireStateLock 

func AcquireStateLock(dir string) (*StateLock, error)

AcquireStateLock takes the exclusive lock on dir, creating the directory when it does not exist. It never waits: a daemon that blocks at startup looks hung, so a held lock is an immediate error naming the process that holds it.

The lock is taken before the database is opened for writing, which is what makes two writers impossible rather than merely unlikely.

func (*StateLock) Path 

func (l *StateLock) Path() string

Path is the lock file this lock is held on.

func (*StateLock) Release 

func (l *StateLock) Release() error

Release drops the lock and closes the file. The file itself stays: closing the descriptor is what releases the lock, and unlinking it would let a second process create a fresh inode and lock that instead.

type Step 

type Step struct {
	Name        string
	Index       int
	State       string
	Attempt     int
	MaxAttempts int

	// ExitCode is meaningful only when HasExitCode is set. A step killed by a
	// signal, or one that never started, has no exit code at all, and zero is a
	// perfectly ordinary success.
	ExitCode    int
	HasExitCode bool

	Signal     string
	StartedAt  time.Time
	FinishedAt time.Time
	DurationMS int64
	ReasonCode string
	ReasonText string
	ReasonData string
	Error      string
	LogPath    string

	// LogBytes, LogTruncated and ErrorTail are the log facts that live in
	// the database: how big the log grew, whether the quota cut it, and the
	// last few KiB of output. The tail is what explain shows after the log
	// file itself is gone.
	LogBytes     int64
	LogTruncated bool
	ErrorTail    string

	// NextAttemptAt is when a pending step that failed with attempts left
	// becomes runnable again. M1 computes no retries, so nothing sets it
	// yet; the claim gate reads it all the same, because M1-09 will fill
	// it without touching a single reader.
	NextAttemptAt time.Time
}

Step is one step of a run.

type StepDep 

type StepDep struct {
	RunID     string
	StepName  string
	DependsOn string
}

StepDep is one edge as it was frozen when the run was materialised. M1 runs steps in index order and never reads these; they exist so the record of why a run waited survives, and so M4 inherits its graph from rows rather than from a spec that may since have changed.

type StepOutcome 

type StepOutcome struct {
	// Event is one of the step machine's input names: step_succeeded,
	// step_failed, cancel_observed or upstream_failed.
	Event string

	// ReasonCode explains a terminal outcome. The machine refuses a
	// terminal transition without one.
	ReasonCode reason.Code

	// ExitCode is what the process exited with. Nil means there is none:
	// a signalled step and a step that never ran have no exit code, and
	// zero is a perfectly ordinary success.
	ExitCode *int

	Signal     string
	FinishedAt time.Time

	// LogMeta carries the attempt's log facts. Empty means the attempt
	// produced no log, which is what a skip is.
	LogMeta LogMeta

	// DetailJSON is the canonical detail object on the event and on the
	// row's reason_data, empty for none.
	DetailJSON string

	// Retry schedules a further attempt when this event sends the step
	// back to pending. Nil means runnable again at once. It never changes
	// the transition itself; the machine decides from the row.
	Retry *RetryPlan
}

StepOutcome is one event the engine hands to a step: the runner came back, or an upstream ended in a way that closes this step, or a cancellation was observed. Event names the model event, and the machine decides whether the step may take it.

type Store 

type Store struct {
	// contains filtered or unexported fields
}

Store owns every connection to the database file. Both handles are private: no package outside internal/store may hold a writable database handle.

func Open 

func Open(ctx context.Context, path string, opt Options) (*Store, error)

Open opens the database at path with a single-connection writer pool and a read-only reader pool, then verifies that every PRAGMA is actually in effect. It creates the file when it does not exist and works against an empty, unmigrated database.

func OpenReadOnly 

func OpenReadOnly(ctx context.Context, path string, opt Options) (*Store, error)

OpenReadOnly opens only the read pool of a database: no writer connection, no state lock, no migration and none of the startup writes. It exists for the commands that must work while the daemon holds the state lock, such as paceq logs, which reads files and read only database rows and nothing else.

A write through the returned store fails with ErrReadOnly instead of reaching the database.

func OpenState 

func OpenState(ctx context.Context, dir string, opt Options) (*Store, error)

OpenState claims a state directory and opens the database in it.

The order is the point. The lock is taken before the database is opened for writing, so a second paceq is refused before it can touch a single page of a file another process owns. The returned store holds the lock until Close.

func (*Store) AbandonedChains 

func (s *Store) AbandonedChains(ctx context.Context) ([]string, error)

AbandonedChains names every half written trigger chain: a tick that claims it triggered but has no trigger row, an accepted trigger with no run, a run without its steps, a run without its queued event. Each pattern is a chain a SIGKILL could have cut in the middle if materialisation were not one transaction. An empty result says no aborted chain left anything behind.

Skipped ticks and deduped triggers are deliberately not findings: standing down is a decision those rows record, not a broken promise.

func (*Store) AcquireOrRenew 

func (s *Store) AcquireOrRenew(ctx context.Context, name, holder string, ttl time.Duration) (LeaseGrant, bool, error)

AcquireOrRenew takes the named lease for holder, or renews the holding one holder already has. It is idempotent: calling it again with the same inputs moves nothing but expires_at.

ok is false when another holder owns a lease that is still alive; the grant returned then is empty and the caller is the follower. ttl counts from the moment the database applies the statement, which this store reads from its own injected clock. The method deliberately takes no now parameter: lease time is always computed here, never compared across processes.

func (*Store) ActiveAttempts 

func (s *Store) ActiveAttempts(ctx context.Context) ([]AttemptProcess, error)

ActiveAttempts lists the baselines whose step and whose run are still running right now. A pid found on this list with matching start ticks is a legitimate live worker, and the orphan sweep leaves it alone.

func (*Store) ActiveRunViolations 

func (s *Store) ActiveRunViolations(ctx context.Context) ([]Violation, error)

ActiveRunViolations returns every job whose running rows outnumber its max_concurrent: the I12 invariant of 02 section 4.3. It is part of the full fsck sweep, and it is the check the startup quick subset calls as well once #62's reconciliation lands there - the call site is one line, because this method already reads exactly what that subset needs.

func (*Store) ActiveRunsForJob 

func (s *Store) ActiveRunsForJob(ctx context.Context, job string) (int, string, error)

ActiveRunsForJob reads the occupancy of one job. It exists for tests and for the fsck invariant; production decisions never call it, because a read outside the transaction that decides is exactly the shape this feature exists to avoid.

func (*Store) AppendLeaseEvent 

func (s *Store) AppendLeaseEvent(ctx context.Context, e LeaseEvent) error

AppendLeaseEvent records one moment of a lease's life: taken, lost or taken over. The code comes from the closed reason catalogue, so the row explains itself the way run_events rows do.

func (*Store) AppendRunEvent 

func (s *Store) AppendRunEvent(ctx context.Context, e RunEvent) error

AppendRunEvent records one transition on its own. Transitions that happen inside a write of their own carry their event in that same transaction instead, which is what keeps the history from disagreeing with the state.

func (*Store) ApplyJobs 

func (s *Store) ApplyJobs(ctx context.Context, inputs []JobVersionInput) ([]JobApplyResult, error)

ApplyJobs records a batch of job specs and leaves every job pointing at the version its spec hashes to. The whole batch is one transaction on purpose: apply either lands complete or not at all, because a half applied catalog is the one state an operator cannot reason about. Any failure rolls every job in the batch back, including the sensor rows each job's spec declares.

Idempotency lives in the schema, not in a check: job_versions carries UNIQUE (job_name, spec_hash), the insert comes first, and a conflict means the version was already there. Two applies racing on the same file therefore end in one row no matter which commit lands first.

The inputs are finished facts: parsing happened before this was called, and nothing inside the transaction reads a file or a clock. Ids are minted before the transaction opens, for the same reason UpsertJobVersion mints its own there.

func (*Store) AutoVacuum 

func (s *Store) AutoVacuum(ctx context.Context) (AutoVacuumMode, error)

AutoVacuum is the mode the database file was created with. doctor reports it, because a database left at NONE never gives disk back after retention and the way out costs a full VACUUM.

func (*Store) BeginSensorTick 

func (s *Store) BeginSensorTick(ctx context.Context, in BeginSensorTickInput) (BeginSensorTickResult, error)

BeginSensorTick writes the intention row for one sensor evaluation: a tick in 'running' carrying the cursor it started from. A crash after this point leaves a visible interrupted tick that reconciliation closes; there is no such thing as a sensor evaluation the database silently forgets started.

The write is deliberately separate from CommitSensorTick. The sensor runs between them, outside any transaction (plan 00 section 3.11: no process execution or network I/O inside a write transaction).

func (*Store) BootChanged 

func (s *Store) BootChanged() bool

BootChanged reports whether the machine has restarted since the last session ran. It answers for the session StartSession opened, and is false until then.

The evidence lasts for one start. StartSession reads the recorded boot id and overwrites it in the same transaction, so the first start after a reboot is the only one that sees the change: every later start on that boot reports false. Whoever reconciles has to act on it then, or the fact is gone.

False is not the same as "no restart": a platform without a boot id reports false, and then a surviving process can only be ruled out by waiting for its lease to expire.

func (*Store) CancelRequested 

func (s *Store) CancelRequested(ctx context.Context, runID string) (bool, string, error)

CancelRequested reports whether a cancellation is waiting, and who asked. The executor reads this between steps and on its poll while a step runs.

func (*Store) CheckpointTruncate 

func (s *Store) CheckpointTruncate(ctx context.Context) error

CheckpointTruncate runs PRAGMA wal_checkpoint(TRUNCATE) through the writer pool: every committed frame moves into the database and the wal file shrinks to zero bytes. The daemon calls it after its last write, before Close, so the next start opens a small file instead of replaying a long wal.

An error here is reported, not fatal: a checkpoint that could not finish costs startup speed, never correctness.

func (*Store) ClaimRun 

func (s *Store) ClaimRun(ctx context.Context, runID string, in LeaseInput) (string, int64, error)

ClaimRun claims exactly this run for the caller. It is the single run form of ClaimRuns and shares its one statement.

The returned state is "running" with the new fencing token, or "cancelled" when a cancellation was waiting before the run ever started: the sweep in the same transaction closed it, and there was never a lease to hand out. Anything else is ErrNotClaimable.

func (*Store) ClaimRuns 

func (s *Store) ClaimRuns(ctx context.Context, spec ClaimSpec) ([]ClaimedRun, error)

ClaimRuns claims up to Limit due runs for Owner in one transaction: the cancel sweep for queued runs asked to stop, then the claim statement, then one run.started event per claimed row. Everything lands together or not at all, which is why a crash mid-claim leaves every run exactly as it was.

Runs that are not due yet, already claimed, or asked to stop come back some other cycle; the result only names what this call took.

func (*Store) ClaimableRunIDs 

func (s *Store) ClaimableRunIDs(ctx context.Context) ([]string, error)

ClaimableRunIDs names the queued runs that are due now, in claim order: by available_at first, then id, which is the order the claim index keeps. A parked run whose time has not arrived is invisible here, exactly as the claim gate would have it.

func (*Store) Close 

func (s *Store) Close() error

Close releases every connection in both pools, and the state lock when this store took one. The lock goes last: it may not be handed on while a connection to the database it protects is still open.

func (*Store) CommitSensorTick 

func (s *Store) CommitSensorTick(ctx context.Context, in SensorTickCommitInput) (SensorTickCommitResult, error)

CommitSensorTick records one finished sensor evaluation, all or nothing: the tick outcome, every trigger, every run and its event, the run_keys dedup gate, and the cursor advance share ONE BEGIN IMMEDIATE transaction. A crash at any point between BEGIN IMMEDIATE and COMMIT leaves the database as if the evaluation had never committed, so a restart re-evaluates from the old cursor and the dedup gate makes the replay a no-op. There is no window in which runs exist without their cursor, or a cursor exists without its runs.

The cursor advance is guarded. CursorVersion was read at evaluation start; if the sensor has since been advanced by another evaluation (a takeover, a reset, a concurrent pass), the CAS matches zero rows, the commit is refused and the tick records TICK_MISSED_LEASE_LOST. An old result never overwrites a newer one.

func (*Store) CreateRunWithSteps 

func (s *Store) CreateRunWithSteps(ctx context.Context, in NewRun) (Run, error)

CreateRunWithSteps materialises a run: the run row, every step, the frozen dependency edges and the queued event, in one transaction. A crash anywhere in it leaves no rows at all, so there is no such thing as a run without its steps or a transition without its event.

A concurrency key held by another active run returns ErrConcurrencyKeyHeld. The database decides that, through the partial unique index, and this method never reads the key first: a read followed by a write is a decision made on information that can already be out of date.

func (*Store) CurrentJobVersion 

func (s *Store) CurrentJobVersion(ctx context.Context, jobName string) (JobVersion, error)

CurrentJobVersion returns the version a job currently points at. This is a read outside any transaction: a caller that decides based on it races an apply by design, and the callers that must not race choose versions inside MaterializeManualTrigger instead.

func (*Store) DrainRun 

func (s *Store) DrainRun(ctx context.Context, runID string, ref LeaseRef, code reason.Code) (bool, error)

DrainRun hands a claimed run back to the queue at a clean stop. It is the whole handback in one transaction: the running steps go back to pending with their attempts restored (05 section 3.2, point 4), because the attempt was cut short by the daemon's own stop and produced no verdict, and then the run itself is requeued without counting a crash, because the executor left on purpose.

The epoch still rises, so any writer from the drained attempt stays fenced out, and available_at moves to now so the next executor can claim at once. The CAS decides whether anything is owed: a caller whose lease has moved on gets handed=false and writes nothing.

func (*Store) DueSchedules 

func (s *Store) DueSchedules(ctx context.Context, nowMilli int64, max int) ([]ScheduleRow, error)

DueSchedules returns up to max unpaused schedules whose next tick is due at or before nowMilli, oldest next tick first.

func (*Store) DumpForIdempotence 

func (s *Store) DumpForIdempotence(ctx context.Context) ([]string, error)

DumpForIdempotence renders every user table as ordered, comparable text, leaving out only the daemon session rows whose timestamps move on every start by design. Two calls bracketing a pass of OnStartup that wrote nothing come back equal, which is exactly the assertion AC7 and AC12 of issue #62 rest on: bit-identical after the first pass, untouched when healthy.

The SQL lives here because all SQL lives here; callers get back plain strings they compare themselves.

func (*Store) FailHangingTicks 

func (s *Store) FailHangingTicks(ctx context.Context, startedBefore time.Time) ([]string, error)

FailHangingTicks closes every tick still marked running whose evaluation started before the given instant. A running tick older than this process's session belongs to a daemon that died mid evaluation; nothing else could ever have written its verdict. Each closed tick records why: the code says the daemon crashed, not that the sensor failed.

The returned slice names what closed, in no particular order, so a caller logs exactly as much as happened. An empty result means there was nothing to close, which is the common case for the periodic safety net.

func (*Store) FinishRun 

func (s *Store) FinishRun(ctx context.Context, runID string, ref LeaseRef, fr FinishReason) (string, error)

FinishRun closes a run whose steps are all terminal. The aggregate comes from the machine's guards, which read the step rows inside the same transaction, so the verdict can never disagree with the steps it describes: any failed step fails the run, and a run with a step still open is refused outright. The lease is released here.

The verdict is a CAS on the fence: the UPDATE carries owner and epoch, and zero rows affected means the lease was taken over while this writer worked. The refusal comes back as ErrLeaseLost and nothing at all is written: worst case is duplicate work, never duplicate state.

func (*Store) Fsck 

func (s *Store) Fsck(ctx context.Context) ([]Violation, error)

Fsck sweeps the whole database and returns every violation it finds, empty when the state is sound. It reads only: a checker that writes would be part of whatever it is checking.

func (*Store) GapSchedules 

func (s *Store) GapSchedules(ctx context.Context) ([]GapSchedule, error)

GapSchedules lists the schedules that were live to fire during a gap. Paused schedules are left out on purpose: their slots were not owed, and writing missed ticks beside them would turn an operator's pause into apparent downtime.

func (*Store) GetRun 

func (s *Store) GetRun(ctx context.Context, idOrPrefix string) (RunDetail, error)

GetRun reads one run and its steps. The argument is a whole id or any prefix of one, the way a git object is named: ids are ULIDs, so a prefix is a range scan on the primary key and not a scan of the table.

A prefix matching more than one run is ErrAmbiguousRunID rather than the first match. Picking one would mean cancelling or replaying whichever run the database happened to order first.

The run and its steps are two queries against the read pool, and the pool has many connections, so they are two snapshots and not one. A step that changes between them shows up as a step further along than the run says. That is the deliberate price of never opening an explicit read transaction: a read snapshot held open across a listing is what stops WAL checkpointing and grows the file without bound. Anything that has to see one instant reads it inside a write transaction instead.

func (*Store) GetSchedule 

func (s *Store) GetSchedule(ctx context.Context, jobName, name string) (ScheduleRow, error)

GetSchedule reads one schedule by job name and schedule name.

func (*Store) InjectActiveRunOverflow 

func (s *Store) InjectActiveRunOverflow(ctx context.Context) (string, error)

InjectActiveRunOverflow plants I12: two running runs on one job whose max_concurrent is 1. The seeded night job has one queued run; the injection sets it to running and creates a second running run, so the active count exceeds the ceiling.

func (*Store) InjectBackwardsTimestamp 

func (s *Store) InjectBackwardsTimestamp(ctx context.Context) (string, error)

InjectBackwardsTimestamp moves a step's finish before its start: I13.

func (*Store) InjectBrokenEventChain 

func (s *Store) InjectBrokenEventChain(ctx context.Context) (string, error)

InjectBrokenEventChain rewrites the second event's from_state, so the chain no longer picks up where the first event ended: I15.

func (*Store) InjectDependencyCycle 

func (s *Store) InjectDependencyCycle(ctx context.Context) (string, error)

InjectDependencyCycle plants I9: two steps of one run pointing at each other, so no order of execution can ever be legal. Like the duplicate run key, nothing in the schema forbids it; only the health check can see it.

func (*Store) InjectDuplicateRunKey 

func (s *Store) InjectDuplicateRunKey(ctx context.Context) (string, error)

InjectDuplicateRunKey plants I3: one job whose run_key names two runs. There is no UNIQUE index to dodge on purpose, because this invariant lives in application law, not in the schema; that is exactly why the health check has to re-read it at startup.

func (*Store) InjectFailedStepUnderSucceededRun 

func (s *Store) InjectFailedStepUnderSucceededRun(ctx context.Context) (string, error)

InjectFailedStepUnderSucceededRun marks one step failed while the run still says succeeded, so the stored state and the steps' aggregate disagree: I10.

func (*Store) InjectOrphanTick 

func (s *Store) InjectOrphanTick(ctx context.Context) (string, error)

InjectOrphanTick writes a manual tick that claims it triggered, with no trigger row behind it. AbandonedChains names it; no invariant check does, because the chain sweep is where this broken promise lives.

func (*Store) InjectTerminalStepPending 

func (s *Store) InjectTerminalStepPending(ctx context.Context) (string, error)

InjectTerminalStepPending flips one step of a terminal run back to pending: the exact row I2 exists for. The run's aggregate moves with it, so the sweep reports I2 and I10 for the same planted row.

func (*Store) InjectUnexplainedDeferral 

func (s *Store) InjectUnexplainedDeferral(ctx context.Context) (string, error)

InjectUnexplainedDeferral pushes a queued run's availability into the future and clears its defer_reason: a run held back that no longer says why. The CHECK constraint refuses this shape, so the injection lifts the checks for the one statement and puts them straight back: I14.

func (*Store) InjectUnexplainedTerminal 

func (s *Store) InjectUnexplainedTerminal(ctx context.Context) (string, error)

InjectUnexplainedTerminal clears a terminal run's reason code: the catalogue rule swept along with the invariants.

func (*Store) IntegrityCheck 

func (s *Store) IntegrityCheck(ctx context.Context) (IntegrityReport, error)

IntegrityCheck runs both pragmas and reports what they said. It reads only; SQLite answers integrity_check from the same connection pool every other reader uses, so nothing here disturbs a writer.

func (*Store) Job 

func (s *Store) Job(ctx context.Context, name string) (JobView, error)

Job reads one job back. An unknown name is ErrNotFound, the same answer every other single row read gives.

func (*Store) JobLastRuns 

func (s *Store) JobLastRuns(ctx context.Context, jobName string) ([]JobRunSummary, error)

JobLastRuns reads one row per job, with that job's newest run joined beside it. An empty jobName reads every job; a name narrows the answer to one. Jobs come back in name order, and a job without runs is still listed, because status is a per job report rather than a listing of runs.

The newest run is decided by id alone. Ids are ULIDs, so id order is time order and the subquery stays a single index lookup per job.

func (*Store) JobNames 

func (s *Store) JobNames(ctx context.Context) ([]string, error)

JobNames lists every job in name order. It is what a did you mean suggestion is drawn from when a command is given a job that does not exist.

func (*Store) JobPaused 

func (s *Store) JobPaused(ctx context.Context, jobName string) (bool, error)

JobPaused reports whether a job is paused. An unknown job is ErrNotFound.

func (*Store) JobVersionByID 

func (s *Store) JobVersionByID(ctx context.Context, versionID string) (JobVersion, error)

JobVersionByID returns exactly the version named, however old. This is how execution reads the bytes a run was frozen with.

func (*Store) JournalMode 

func (s *Store) JournalMode(ctx context.Context) (string, error)

JournalMode is the journal mode the database file is in. Open refuses anything but WAL, so a report built on this states what the file holds rather than what the code assumes.

func (*Store) KnownAttempts 

func (s *Store) KnownAttempts(ctx context.Context) ([]AttemptProcess, error)

KnownAttempts lists every baseline ever recorded, whatever happened to the step afterwards. This is the sweep's history half: a process still carrying a dead attempt's PACEQ_RUN_ID is identified by a baseline that outlived the attempt.

func (*Store) LatestSession 

func (s *Store) LatestSession(ctx context.Context) (Session, bool, error)

LatestSession is the most recent session row, open or closed. It is how an operator or the health surface asks "who ran here last, and how did it end": a closed row carries its stop_reason, an open one means the process died without saying goodbye.

func (*Store) LeaseEvents 

func (s *Store) LeaseEvents(ctx context.Context, name string, limit int) ([]LeaseEvent, error)

LeaseEvents returns the recorded moments of one lease, newest first. limit caps the read; pass a small number unless something really needs the history.

func (*Store) LeaseHolder 

func (s *Store) LeaseHolder(ctx context.Context, name string) (LeaseGrant, bool, error)

LeaseHolder returns the current row for the named lease, whether anyone holds it or not. It is the read side for status views and for whoever has to decide how long ago a holder was last seen. It reads only: nothing in this method writes, so it is safe next to a running daemon.

func (*Store) ListAllSchedules 

func (s *Store) ListAllSchedules(ctx context.Context) ([]ScheduleRow, error)

ListAllSchedules returns every schedule row: unpaused first, then paused, alphabetically within each group.

func (*Store) ListJobVersions 

func (s *Store) ListJobVersions(ctx context.Context, jobName string) ([]JobVersion, error)

ListJobVersions reads every version of one job, newest first. A job's history is short, so the whole list comes back: a report that says "you are on version 4 of 7" needs the same rows a rollback check needs.

func (*Store) ListRuns 

func (s *Store) ListRuns(ctx context.Context, f RunFilter) ([]RunSummary, error)

ListRuns is one page of run history, newest first. Ids are ULIDs, so id order is time order and the cursor is the last id of the previous page.

There is no OFFSET anywhere in this package. An offset re-reads everything it skips, and a listing that holds a read snapshot open while it does so is what starves WAL checkpointing and grows the file without bound.

func (*Store) MaterializeManualTrigger 

func (s *Store) MaterializeManualTrigger(ctx context.Context, in ManualTriggerInput) (ManualTriggerResult, error)

MaterializeManualTrigger records a manual decision end to end: the tick, the trigger and the run with its steps and edges, in one transaction. It is the same chain a schedule or a sensor produces, which is what keeps explain complete for hand started runs from day one and lets M2 and M3 add new source kinds without a second code path.

The steps come out of job_versions.spec_json, the immutable version the job currently points at. Nothing re-reads the YAML file, so an apply that lands after the decision cannot change what this run does.

A paused job still runs here. A pause governs automatic firing; a person typing the command has already decided.

func (*Store) MaterializeTick 

func (s *Store) MaterializeTick(ctx context.Context, in TickInput) (TickResult, error)

materializeTick claims one fire-time end to end. The order inside the transaction is the write model's: claim first, then the consequences, then progress, and every consequence aborts together when any statement refuses.

func (*Store) Migrate 

func (s *Store) Migrate(ctx context.Context) error

Migrate applies every migration the database is missing. It is idempotent and safe to call on every startup.

func (*Store) MissedTickEvidence 

func (s *Store) MissedTickEvidence(ctx context.Context) ([]TickEvidence, error)

MissedTickEvidence lists every synthetic missed tick. It is how tests and the explain surface read the gap's evidence back without touching SQL outside this package.

func (*Store) NextRetryWait 

func (s *Store) NextRetryWait(ctx context.Context, runID string) (time.Duration, bool, error)

NextRetryWait reports how long the executor must wait before some pending step of the run passes its retry gate: the gap to the soonest next_attempt_at, clamped to zero once it is due. Waiting is false when no pending step carries a time at all, which is the executor's signal that nothing here will ever become runnable again. This read is what keeps retry free of state of its own: the engine sleeps on the answer instead of owning a scheduler.

func (*Store) NextRunnableStep 

func (s *Store) NextRunnableStep(ctx context.Context, runID string) (string, bool, error)

NextRunnableStep names the next step the engine may start: the lowest index that is pending, past its retry gate, and whose every frozen upstream has succeeded. A step waiting on upstream that has not succeeded is skipped over, which is the degenerated claim predicate M4-02 replaces with the whole graph.

func (*Store) ObserveRunCancel 

func (s *Store) ObserveRunCancel(ctx context.Context, runID string, ref LeaseRef, actor string, code reason.Code) error

ObserveRunCancel effectuates a cancellation somebody requested earlier: the caller has already killed the process group outside any transaction (the machine lists that effect first, and the engine performs it before calling here), the steps that were running are cancelled by their own events, and this closes whatever pending steps remain and then the run itself.

The caller must still hold the lease at its token; the write is a CAS like every other result write, so a holder that lost the run while killing the process group cannot cancel over whatever the new owner is doing.

It refuses when nobody asked: a run nobody asked to cancel is never cancelled, whatever the caller claims. The event names the person who made the request, not the executor that observed it.

func (*Store) OpenSession 

func (s *Store) OpenSession(ctx context.Context) (Session, bool, error)

OpenSession is the session that has not ended, if there is one. It is what names the process holding the state lock, and what tells a restart that the last run never finished.

func (*Store) Outages 

func (s *Store) Outages(ctx context.Context) ([]Outage, error)

Outages lists every stored outage, oldest first. It is how tests and the explain surface read history back.

func (*Store) OutagesWithoutTicks 

func (s *Store) OutagesWithoutTicks(ctx context.Context) ([]Outage, error)

OutagesWithoutTicks lists outage rows whose synthetic tick evidence has not been written yet (or legitimately holds zero slots). It is how reconciliation finishes what a crash mid-write left half explained: the row itself is the marker, and a later pass completes it.

func (*Store) OverrideBootIDForTest 

func (s *Store) OverrideBootIDForTest(boot string)

OverrideBootIDForTest pins what the platform's boot id read reports from now on. Tests cannot edit /proc/sys/kernel/random/boot_id, and a changed boot id is the strongest evidence reconciliation has (issue #62), so the value itself is what gets replaced: call it before a start to stage one boot, and again with a different value to stage a machine restart. An empty value lifts the override.

It exists for the startup reconciliation tests and their crash harness rows; production code has no reason to call it. It takes a value rather than a callback on purpose: the store's exported surface accepts no function values.

func (*Store) Path 

func (s *Store) Path() string

Path is the database file this store was opened against.

func (*Store) PauseSchedule 

func (s *Store) PauseSchedule(ctx context.Context, jobName, name string) (ScheduleRow, error)

PauseSchedule sets paused=1 on the schedule and returns the new row. Idempotent: calling on an already-paused schedule succeeds without error and does not change updated_at.

func (*Store) PendingSteps 

func (s *Store) PendingSteps(ctx context.Context, runID string) ([]Step, error)

PendingSteps lists a run's steps that have not started, in index order.

func (*Store) QuickFsck 

func (s *Store) QuickFsck(ctx context.Context) ([]Violation, error)

QuickFsck returns every critical violation it finds, empty when the state is sound enough to serve.

func (*Store) ReapExpiredRuns 

func (s *Store) ReapExpiredRuns(ctx context.Context, opt ReapOptions) ([]ReapedRun, error)

ReapExpiredRuns takes every run whose lease expired past the skew allowance and decides its fate in the one transaction: a cancellation request on a dead owner completes as cancelled, a run past its crash budget fails into the poison quarantine, a run with no attempt budget left fails reconciled, and everything else is requeued with a backoff for a new holder. The fencing token rises in every arm, including the failures: a late zombie must not overwrite a verdict someone else owns, whatever the verdict was.

Each decision writes its own run_events row, by the reaper, in the same transaction as the state change (G10).

func (*Store) RecordAttemptProcess 

func (s *Store) RecordAttemptProcess(ctx context.Context, runID, step string, ref LeaseRef, pid int, startTicks int64) error

RecordAttemptProcess stamps which live process is carrying one running step: its pid and the kernel's start ticks for that pid, read at spawn time. The engine writes it through OnStart the instant a spawn succeeds, so the evidence is on file before any verdict could exist, and startup reconciliation can later prove a surviving process is (or is not) the child it was told about.

The write is fenced like every other holder's write: only the executor that owns the run's lease may name processes for it, so a replaced executor cannot relabel someone else's process as one of ours.

func (*Store) RecordMissedTicks 

func (s *Store) RecordMissedTicks(ctx context.Context, sessionID string, outageID int64, ticks []MissedTick) (int, error)

RecordMissedTicks writes the synthetic ticks for one outage and adds the count it actually inserted to the outage's total. Every insert is INSERT OR IGNORE: a slot that already has a tick row was really evaluated before the crash, and reconciliation never rewrites real history. That same ignore makes a repeated pass over the same slots a no-op, which is what makes the whole sweep safe to run twice.

The work is chunked so a fourteen day gap across a hundred schedules cannot hold the write lock for one long stretch.

func (*Store) RecordOrphanKill 

func (s *Store) RecordOrphanKill(ctx context.Context, runID string, pid int) error

RecordOrphanKill writes the audit event that says this process group was signalled and why. A kill nobody can read about later is indistinguishable from sabotage, so the sweep's every shot lands here first.

The event carries the run's current state on both ends of the transition, because a kill changes no state: it only explains a signal that was sent. Stamping the live state also keeps the event chain (I15) continuous even when another actor moved the run between the scan and this write.

func (*Store) RecordOutage 

func (s *Store) RecordOutage(ctx context.Context, in OutageInput) (int64, error)

RecordOutage stores the outage and returns its rowid. The detected_at stamp is this store's clock reading at write time, which is what bounds the SLO promise: an outage longer than the noise threshold exists within seconds of the startup that noticed it.

func (*Store) RecordStepOutcome 

func (s *Store) RecordStepOutcome(ctx context.Context, runID, name string, out StepOutcome, ref LeaseRef) error

RecordStepOutcome applies one event to one step. The machine decides whether the step may take it and what the transition demands; this writes the verdict, its log facts and its event in one transaction. A step that fails with attempts left goes back to pending for its next attempt: parked at next_attempt_at when the caller attached a RetryPlan, runnable again at once when it did not. The claim gate, not this method, decides when the next attempt may start.

The ref is the fence. A holder writes only while its token still matches; recovery passes the zero ref, which is refused against any live lease. A writer that lost the lease gets ErrLeaseLost and nothing on the row moves.

func (*Store) ReleaseLease 

func (s *Store) ReleaseLease(ctx context.Context, name, holder string) (bool, error)

ReleaseLease deletes the named lease row when holder still owns it, which is the clean shutdown path: the next process can take over immediately instead of waiting out the ttl.

released is false when there was nothing of holder's to delete: either the lease was never held, or it expired and another holder took it over. Neither case is an error, and neither may touch the new holder's row.

func (*Store) RenewRunLeases 

func (s *Store) RenewRunLeases(ctx context.Context, owner string, ttl time.Duration) ([]LeaseRenewal, error)

RenewRunLeases is the batch heartbeat: one transaction for all the runs the owner holds, no matter how many there are, so the write cost of staying alive does not scale with the workload. Callers diff the answer against what they believe they hold; a run missing from the answer, or answering with a different token, is a lease lost and must be self-fenced.

func (*Store) RequestCancel 

func (s *Store) RequestCancel(ctx context.Context, runID, by, why string) (CancelRequest, error)

RequestCancel records that somebody wants the run stopped, durably and before anything is killed. The first request stands: a second one changes nothing, so two people asking at once cannot disagree about who asked or why. Observing the request is a different act, done by whoever holds the lease, and that is the transition with the event.

func (*Store) RequeueCrashedRun 

func (s *Store) RequeueCrashedRun(ctx context.Context, runID string) error

RequeueCrashedRun puts a run whose executor died back in the queue. It is the store half of the restart story the crash harness (#75) proves: a run left running by a SIGKILL is requeued through the machine's own lease_expired transition, never reset behind its back.

The refusal is the safety catch. A lease that has not expired yet belongs to a process that may still be alive, and requeuing under it would let two executors drive one run. Only an expired lease counts as evidence that the previous owner is gone; the state directory lock adds its own guarantee on this platform, but the machine's guard does not rely on it.

The transition writes what the machine demands: the epoch goes up so any writer from the dead attempt is fenced out, crash_count counts the loss of the executor against the run, and defer_reason records why the requeued run sits waiting (I14). One event row tells the story, in the same transaction.

func (*Store) ResetSensor 

func (s *Store) ResetSensor(ctx context.Context, in ResetSensorInput) (ResetResult, error)

ResetSensor atomically raises the sensor's dedup_epoch by one, so every run key this sensor ever registered becomes a new fingerprint in the fresh epoch. It is the store-level backing for `sensors reset`: the CLI surface lands in M3-06, and this is the method it will call.

No read happens before the write: the bump is one UPDATE with a RETURNING readback, so two resets can never collide on the same new value (insert- first discipline, 11 section 5.2).

func (*Store) ResumeSchedule 

func (s *Store) ResumeSchedule(ctx context.Context, jobName, name string, nextTickAt time.Time) (ScheduleRow, error)

ResumeSchedule sets paused=0, writes the caller's pre-computed next_tick_at, and returns the new row. The caller computes nextTickAt from the cursor and cron expression to avoid re-parsing inside the transaction.

func (*Store) RunEvents 

func (s *Store) RunEvents(ctx context.Context, idOrPrefix string) ([]RunEvent, error)

RunEvents returns a run's transition history, oldest first. This is the explain backbone: one row per state change, written in the transaction that made the change.

func (*Store) RunExists 

func (s *Store) RunExists(ctx context.Context, runID string) (bool, error)

RunExists reports whether this database has ever heard of the run. The orphan sweep checks before it believes an environment variable: a foreign program that happens to carry PACEQ_RUN_ID must never be mistaken for one of ours.

func (*Store) RunIDByRunKey 

func (s *Store) RunIDByRunKey(ctx context.Context, sourceID, runKey string) (string, error)

RunIDByRunKey names the run a registered dedup key points at. The crash harness uses it to find the run behind an already-committed tick decision; explain will use it to answer "which run did this fire-time become".

func (*Store) ScheduleCursor 

func (s *Store) ScheduleCursor(ctx context.Context, jobName, name string) (*time.Time, time.Time, error)

ScheduleCursor returns a schedule's progress stamps: last_tick_at (nil when the schedule has never been evaluated) and next_tick_at.

func (*Store) ScheduleTicks 

func (s *Store) ScheduleTicks(ctx context.Context, jobName, name string) ([]TickView, error)

ScheduleTicks lists one schedule's recorded ticks, oldest fire-time first. This is the read side of everything MaterializeTick writes, and the table explain (M5-01) will read.

func (*Store) SchemaVersion 

func (s *Store) SchemaVersion(ctx context.Context) (int, error)

SchemaVersion is the schema version recorded in this database file. It is the number doctor reports, and the one that decides whether a binary may write to the file at all.

func (*Store) SetJobPaused 

func (s *Store) SetJobPaused(ctx context.Context, jobName string, paused bool) error

SetJobPaused flips the pause flag. It exists for the doctor and the CLI pause command; tests use it to stage paused jobs.

func (*Store) SetSensorCursor 

func (s *Store) SetSensorCursor(ctx context.Context, in CursorInput) error

SetSensorCursor moves a sensor's cursor without touching its dedup epoch. The old run keys keep dedupping, because the epoch they are tagged with is still the current one. This is the "cursor set" row of the reset table in the issue: spool the cursor without replay, the dedup gate still stops old keys (10 section 5 F4c).

func (*Store) StartSession 

func (s *Store) StartSession(ctx context.Context, version string) (Session, error)

StartSession opens this process's session row and closes whatever the last run left behind.

Three things happen in one transaction, because a reader that saw them apart would draw the wrong conclusion. Sessions still open belong to a run that never got to say goodbye, so they are marked crashed. The new row records who is running, on which boot, from when. The boot id is written to meta, so the next start can tell a restart of paceq from a restart of the machine.

A changed boot id is the strongest evidence this system has: the machine restarted, so no process paceq started can still be alive. BootChanged reports it after this call returns.

func (*Store) StartStep 

func (s *Store) StartStep(ctx context.Context, runID, name string, ref LeaseRef) error

StartStep opens the first attempt of a pending step: running, attempt up by one, started_at stamped, its event in the same transaction. The run has to be running and still held by the writer at the writer's fencing token: a step may not move while its run is queued, whatever the step machine alone would allow, and it may not move under a lease the writer has lost.

func (*Store) StepDeps 

func (s *Store) StepDeps(ctx context.Context, runID string) ([]StepDep, error)

StepDeps lists a run's frozen edges, ordered the way explain prints them.

func (*Store) StopSession 

func (s *Store) StopSession(ctx context.Context, sessionID string) error

StopSession records a clean shutdown. A session that ends any other way keeps its open row, which is what the next start reads as a crash.

func (*Store) SyncSensors 

func (s *Store) SyncSensors(ctx context.Context, job string, sensors []spec.Sensor) (SyncResult, error)

SyncSensors makes the sensor rows of one job identical to the job's spec, atomically, without touching drift state. New sensors become rows with cursor NULL, dedup_epoch 0, consecutive_failures 0 and next_eval_at now; a re-apply of an unchanged spec is a no-op against the definition columns, so no updated_at moves.

func (*Store) TouchSession 

func (s *Store) TouchSession(ctx context.Context, sessionID string) error

TouchSession moves the session's heartbeat forward. Nothing here decides how often that happens; the daemon owns the interval.

The heartbeat is what turns a crashed session into a bounded outage: the gap runs from the last heartbeat to the restart, and without it the whole run would be unaccounted for.

func (*Store) UnexplainedReasons 

func (s *Store) UnexplainedReasons(ctx context.Context) ([]UnexplainedReason, error)

UnexplainedReasons runs the audit query and returns every terminal run, step, tick and trigger stored without a usable reason code. A healthy database returns nothing, and `paceq fsck` prints exactly this list.

It reads through the read only pool like every other read: fsck has to run while another process holds the state lock, and a checker that needed the single writer would be part of whatever it is checking.

func (*Store) UpsertJobVersion 

func (s *Store) UpsertJobVersion(ctx context.Context, in JobVersionInput) (JobVersion, bool, error)

UpsertJobVersion records a job and the spec it currently holds, and reports whether that spec is new. Loading the same file twice is not an error and does not invent a version: the second load conflicts on (job_name, spec_hash), keeps the version that is already there and points the job at it.

The insert comes first and the conflict decides, rather than a read deciding whether to insert. Reading first would be a check with a write after it, and the answer can be stale by the time the write lands.

Pausing is left alone. It is an operator decision about a job, not a property of the file, and a reload that silently resumed a paused job would be the worst kind of surprise.

func (*Store) UpsertSchedule 

func (s *Store) UpsertSchedule(ctx context.Context, in ScheduleInput) (ScheduleRow, error)

UpsertSchedule inserts a schedule or replaces its definition in place, keeping id and timestamps stable across re-apply.

type SyncResult 

type SyncResult struct {
	Created   []string
	Updated   []string
	Unchanged []string
	Removed   []string
}

SyncResult says what SyncSensors did, so a caller can tell the story of an apply without re-reading the table. Created and Updated both mean the row's definition changed; Unchanged means it was the same on arrival; Removed names the sensors that left the job's spec and were deleted.

type TickEvidence 

type TickEvidence struct {
	SourceName string
	ReasonCode string
	ReasonData string
}

TickEvidence is one stored tick's explanation triple: what ran, what code explains it, and what data points at its cause.

type TickInput 

type TickInput struct {
	// Schedule names the row the evaluation belongs to.
	Schedule ScheduleRow

	// ScheduledFor is the fire-time in UTC. Together with source_kind
	// 'schedule' and the schedule's name it IS the idempotency key.
	ScheduledFor time.Time

	// Outcome is OutcomeTriggered, OutcomeSkipped or OutcomeError.
	Outcome string

	// ReasonCode explains every outcome that produced no run.
	ReasonCode reason.Code

	// ReasonText and ReasonData carry the human line and its JSON detail.
	ReasonText string
	ReasonData string

	// RunKey is required when Outcome is triggered.
	RunKey string

	// NextTickAt is where progress should move to when UpdateProgress holds.
	NextTickAt time.Time

	// UpdateProgress moves last_tick_at/next_tick_at to this fire-time. A
	// config failure sets it false: the schedule must stay due so a fixed
	// definition picks the work back up instead of skipping past it.
	UpdateProgress bool

	// Actor lands on the queued run event. Empty becomes "system".
	Actor string
}

TickInput is one decided evaluation: the loop has already computed what happened at this fire-time and why; this transaction only records it.

type TickResult 

type TickResult struct {
	// Claimed is true when this call recorded the evaluation: a new tick
	// row, or an identical earlier skip absorbing it. False means the
	// fire-time was already materialised by someone else; nothing was
	// written and no error is reported.
	Claimed bool

	// Coalesced is true when an identical previous skip took this
	// evaluation in as another repeat_count step instead of a new row.
	Coalesced bool

	// Deferred is true when the run was materialised under overlap: queue
	// with every concurrency slot held: queued, but available_at points
	// into the future and defer_reason says why.
	Deferred bool

	// Run describes the queued run when Claimed and the outcome triggered.
	Run Run
}

TickResult says what one decided evaluation became.

type TickView 

type TickView struct {
	ScheduledFor time.Time // zero when the tick carries no fire-time
	Outcome      string
	ReasonCode   string
	RepeatCount  int
	TriggerCount int
}

TickView is one recorded schedule tick as a reader wants it: what fired, what it became, and why it did not.

type UnexplainedReason 

type UnexplainedReason struct {
	// Kind is which table the row lives in: "run", "step", "tick" or
	// "trigger".
	Kind string
	// Key names the row the way an operator would: the id, or run and step
	// name together for a step.
	Key string
}

UnexplainedReason is one row the audit query returns: a terminal object sitting without a usable reason code.

type Violation 

type Violation struct {
	// Check names the invariant: "I2", "I10", "I13", "I14", "I15" or
	// "reason" for the catalogue rule.
	Check string

	// Subject names the row: "run <id>" or "run <id> step <name>".
	Subject string

	// Detail says what was expected and what was found.
	Detail string
}

Violation is one broken invariant: which check caught it and on what row.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.