store

package
v0.0.0-...-f1846c5 Latest Latest
Warning

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

 Go to latest Published: Sep 11, 2026 License: Apache-2.0 Imports: 34 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 (
	MetaKeyBackupLastAt        = "backup_last_at"
	MetaKeyBackupLastStatus    = "backup_last_status"
	MetaKeyBackupLastPath      = "backup_last_path"
	MetaKeyBackupLastError     = "backup_last_error"
	MetaKeyBackupLastDeepCheck = "backup_last_deep_check_at"

	MetaKeyGCCycleLastAt     = "gc_last_cycle_at"
	MetaKeyGCCycleLastStatus = "gc_last_cycle_status"
	MetaKeyGCCycleLastError  = "gc_last_cycle_error"
)

Meta keys this package writes. doctor reads the backup ones through BackupStatus, and the metrics surface reads the same rows later.

View Source 
const (
	BackupStatusVerified = "verified"
	BackupStatusFailed   = "failed"
)

Backup statuses as they land in meta. A copy that failed verification is a failed backup, not a backup with an asterisk: the file is removed and the status says so.

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"

	// OutcomeMissed is the gap walk's synthetic explanation of a fire-time
	// nobody was alive to evaluate. It is written by startup reconciliation,
	// never by the loop, and it carries no run: it is the one outcome a
	// catch-up attempt is allowed to take the slot back from.
	OutcomeMissed = "missed"

	// OutcomeShadowTriggered records that a fire-time WOULD have started a
	// run (#32): the decision chain ran to its triggered verdict, but shadow
	// mode replaced materialisation with this marker. No trigger row, no run
	// key, no run, no process ever follows it.
	OutcomeShadowTriggered = "shadow_triggered"
)

Tick outcomes as stored in ticks.outcome.

View Source 
const (
	// MetaShadowMode marks the state directory as served in shadow mode.
	// It is set on serve startup with --shadow and cleared on a normal
	// start, so `paceq status` from any process sees what instance is
	// running before it reads a single tick.
	MetaShadowMode = "shadow_mode"

	// MetaShadowSinceMs is when the current shadow period began, Unix ms.
	MetaShadowSinceMs = "shadow_since_ms"

	// MetaShadowObserve names the configured comparison source: journald,
	// file or none.
	MetaShadowObserve = "shadow_observe_source"
)
View Source 
const (
	// ShadowEstimateRuns bounds how many recent finished runs feed the
	// duration average. Twenty keeps one weird outlier from owning forever
	// while still tracking a job whose runtime genuinely changed.
	ShadowEstimateRuns = 20

	// ShadowOccupancyCap is the longest span one earlier shadow fire-time
	// may occupy. An estimate above it is treated as this cap instead of
	// letting a bad average claim minutes forever.
	ShadowOccupancyCap = 24 * time.Hour
)
View Source 
const (
	ObsSourceJournald = "journald"
	ObsSourceFile     = "file"
	ObsSourceManual   = "manual"
)

ShadowObservationSource enumerates where an observation came from.

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 DefaultListLimit = 200

DefaultListLimit bounds one listing. An unbounded audit query on a busy outbox would be a read-side outage waiting to happen.

View Source 
const ExplainPageSize = 50

ExplainPageSize is how many rows one explain query fetches per page. The caller pages by passing the last id it saw as before.

View Source 
const MaxListLimit = 10_000

MaxListLimit is the explicit ceiling even a larger request cannot pass.

View Source 
const MetaBackupLastSuccessAtMs = "backup_last_success_at_ms"

MetaBackupLastSuccessAtMs is the meta key carrying the unix-millisecond stamp of the last successful backup. Nothing writes it yet - the backup machinery itself is M6 work - and the metrics side exposes the family only while the key exists, the same no-series-without-truth rule expected_within follows.

View Source 
const MetaBackupVerifiedOk = "backup_verified_ok"

MetaBackupVerifiedOk is the meta key carrying whether the last backup passed its verification ("1") or failed it ("0").

View Source 
const MetaFsckLastSweepAtMs = "fsck_last_sweep_at_ms"

MetaFsckLastSweepAtMs is the meta key carrying the unix-millisecond stamp of the newest completed sweep, clean or not. It is the identity of that sweep: the violation gauge reads the event rows stamped with it, and the timestamp gauge reads it directly.

View Source 
const MetaGCLastSuccessAtMs = "gc_last_success_at_ms"

MetaGCLastSuccessAtMs is the meta key carrying the unix-millisecond stamp of the last completed retention sweep.

View Source 
const PruneBatchLimit = 200

PruneBatchLimit caps every retention DELETE at this many parent rows per transaction. The plan sketch named 500 on the assumption it keeps each transaction in the tens of milliseconds; measured here (60k seeded runs, warm cache, modernc.org/sqlite), a 500-row batch with its cascade holds the write lock for p50 47.8 ms / p99 69 ms - over the 50 ms budget this issue exists to defend. Two hundred rows measures p50 ~19 ms / p99 ~29 ms, which leaves real headroom for scheduler noise and CPU contention, so the smaller number is what ships: the lock budget is the invariant, the batch size is only the mechanism.

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 (
	ErrRunNotRetryable  = errors.New("this run has nothing to retry")
	ErrNothingToReopen  = errors.New("no step of this run is failed or skipped")
	ErrRunNotTerminal   = errors.New("the run has not finished")
	ErrStepNotInThisRun = errors.New("no such step in this run")
)

ErrRunNotRetryable is returned when the run has nothing a retry could redo. A succeeded run is redone by replay, which makes a new row and leaves this one as history.

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 ErrConflictingReuse = errors.New("choose one of --from and --failed")

ErrConflictingReuse is returned when a caller asks for both reuse rules at once. They are two answers to one question, and answering twice is how a command grows a third meaning nobody documented.

View Source 
var ErrEpochMismatch = errors.New("the result belongs to a lease that has moved on")

ErrEpochMismatch means the result's fencing token does not match the run's current lease epoch: the attempt was reclaimed, and the file is a voice from a run that moved on without it.

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 ErrNotificationDelivered = errors.New("delivered")

ErrNotificationDelivered is what RetryOutbox refuses with: a delivered row is history, and history does not resend itself.

View Source 
var ErrNotificationNotFound = errors.New("no such notification")

ErrNotificationNotFound is what reads and writes return when an outbox id does not name a row. The CLI turns it into exit 3.

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.

View Source 
var ErrRunNotHeld = errors.New("the run lease is not held at the fencing epoch")

ErrRunNotHeld is returned when a write named a run that is not held at the caller's fencing epoch.

View Source 
var ErrRunsHeld = errors.New("new runs are held")

ErrRunsHeld is what a manual run is refused with while the gate holds. The refusal is also recorded - the tick stands down and its trigger is stored rejected with the code - so explain answers for the run that never happened.

View Source 
var ErrUnknownAttempt = errors.New("no running attempt matches the result")

ErrUnknownAttempt means nothing running matches the result's identity: the run is gone, the step is not running, or the attempt number is not the one on the row. The file is not a lie — it is a leftover.

View Source 
var Invariants = []Invariant{
	{
		ID:       "I1",
		Severity: Serious,
		Title:    "a run in running holds a live lease",
		Remedy:   "the reaper clears this on its own within one lease TTL; paceq fsck --repair requeues the run now with epoch+1 and reason " + string(reason.RUNOrphanedReconciled),
	},
	{
		ID:       "I2",
		Severity: Serious,
		Title:    "a terminal run has no step still pending or running",
		Remedy:   "paceq fsck --repair cancels the stranded steps with a reason code, exactly as a shutdown would",
	},
	{
		ID:       "I3",
		Severity: Critical,
		Title:    "one run per dedup identity (source, epoch, run_key)",
		Remedy:   "the run_keys PRIMARY KEY (source_id, epoch, run_key) grants each identity one run, so a run named by two of them means a hand edit or corruption; keep a copy of the state directory and restore from backup",
	},
	{
		ID:       "I5",
		Severity: Serious,
		Title:    "a step's attempt counter sits inside its budget",
		Remedy:   "a row was written behind the transition layer; report it as a bug with paceq export run <id>",
	},
	{
		ID:       "I6",
		Severity: Critical,
		Title:    "at most one tick per (source_kind, source_name, scheduled_for)",
		Remedy:   "the ticks UNIQUE (source_kind, source_name, scheduled_for) constraint refuses the write, so a break means a hand edit or corruption; keep a copy of the state directory and restore from backup",
	},
	{
		ID:       "I8",
		Severity: Serious,
		Title:    "a running step has every step it needs succeeded",
		Remedy:   "the claim predicate gates on the same rule, so this is drift; report it as a bug with paceq export run <id>",
	},
	{
		ID:       "I9",
		Severity: Critical,
		Title:    "the step dependency graph is acyclic and names existing steps",
		Remedy:   "a cyclic run can never finish: cancel it with paceq cancel run <id> and fix the job spec; a dangling edge means the job spec was edited mid-run",
	},
	{
		ID:       "I10",
		Severity: Serious,
		Title:    "a run's stored state is what its steps aggregate to, unless the run itself failed",
		Remedy:   "paceq explain run <id> shows both stories; the reconciler converges a run that has not finished, and nothing revisits one that has, so report a terminal row as a bug with paceq export run <id>",
	},
	{
		ID:       "I11",
		Severity: Serious,
		Title:    "the fencing token never falls",
		Remedy:   "an unfenced write happened; report it as a bug with the events of the named run (paceq export run <id>)",
	},
	{
		ID:       "I12",
		Severity: Serious,
		Title:    "no job or concurrency key runs past its ceiling",
		Remedy:   "admission control and the claim both enforce this on the way in; report it as a bug with the job name",
	},
	{
		ID:       "I13",
		Severity: Warning,
		Title:    "timestamps are monotone: created <= started <= finished, and a present stamp is a real one",
		Remedy:   "a clock jump under way; check the machine's clock sync with paceq doctor",
	},
	{
		ID:       "I14",
		Severity: Warning,
		Title:    "a run held for the future carries its defer reason",
		Remedy:   "paceq fsck --repair stamps the reason as unspecified, which the CLI then reports as held for an unknown reason",
	},
	{
		ID:       "I15",
		Severity: Warning,
		Title:    "every event's from_state is the event before it, per run and per step",
		Remedy:   "a hole in the history makes explain incomplete; report it as a bug with paceq export run <id>",
	},
	{
		ID:       "reason",
		Severity: Warning,
		Title:    "a terminal row carries a usable reason code",
		Remedy:   "rows from an older paceq; paceq fsck --repair stamps them as legacy, and the schema CHECK that a terminal row carries reason_code IS NOT NULL refuses new ones",
	},
}

Invariants is the catalogue, ordered by ID. The full sweep covers every entry; QuickFsck covers the critical subset.

View Source 
var QuickFsckChecks = []string{"I3", "I6", "I9", "I12"}

QuickFsckChecks names the invariants the quick sweep evaluates, in the order it runs them. It answers "what did this sweep look at", which the full catalogue cannot: the sweep covers a subset of it. Any surface that reports a count of checked invariants reads it from here.

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 DeliveryBuckets 

func DeliveryBuckets() []float64

DeliveryBuckets exposes the bucket edges for the scrape side. Never mutate.

func FSMagic 

func FSMagic(dir string) (uint64, error)

FSMagic reports the filesystem magic of the directory, for tools that report rather than refuse. The classification is the open path's rule, shared so doctor and the daemon can never disagree about what is network.

func IsBusy 

func IsBusy(err error) bool

IsBusy reports whether err is one of the two SQLITE_BUSY outcomes: a contended database, not a broken one. It is exported because internal/arch forbids every package outside this one from importing the driver, so a caller that has to tell a transient refusal from a failure it should report as a bug has no other way to ask.

func IsNetworkFSMagic 

func IsNetworkFSMagic(magic uint64) bool

IsNetworkFSMagic reports whether the magic names a filesystem where SQLite locking is undefined.

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.

The parse itself lives in internal/procfs, the one reader of the kernel's process facts, shared with the exec shim's kill verification (#39) so the two can never disagree about what a /proc line says.

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.

func RenderCatalogueDoc 

func RenderCatalogueDoc() string

RenderCatalogueDoc is the invariant reference page, projected from the catalogue and nothing else. The generator writes it (go generate ./internal/store) and the staleness test compares it, so the documentation cannot drift from the sweep: a remedy changed in code and not regenerated fails the build.

func RetryPolicyOf 

func RetryPolicyOf(r *spec.Retry) retry.Policy

RetryPolicyOf translates a frozen spec's retry block into the pure calculator's vocabulary, filling in the defaults the parser promises. A block missing fields in a hand written document therefore behaves exactly like one the decoder normalised: exponential growth from thirty seconds, capped at ten minutes, full jitter.

func StatusQueries 

func StatusQueries() []string

StatusQueries lists the exact statement texts the status overview executes. The query-plan test walks this list and refuses any plan that scans a history table, so an index regression fails a test instead of an operator's morning.

func StepVerdict 

func StepVerdict(outcome string, cancelled bool) (event string, code reason.Code, ok bool)

StepVerdict is the one table from what happened to an attempt to what the step machine is told about it. Every writer of a step verdict reads its row here: the live executor with what it watched itself, recovery with what the dead attempt's shim wrote down. One table is the only way the same exit fact can be made to read the same way on both, which is precisely what the spool file exists to promise — a verdict that survives an executor's death is only worth having if it is the verdict that executor would have committed (#204).

The arguments are the two facts the answer turns on and nothing else. The outcome word is the spool format's vocabulary, because that is the vocabulary that outlives the process that observed it. cancelled says the signal which ended the attempt was this system answering a cancellation; it is the only row that branches, and it is the row that matters, because the machine retries a failure and never retries a cancellation, and a run with a failed step outranks a run with a cancelled one (I10). Getting it wrong therefore reruns work an operator asked to stop and calls the run a failure.

ok is false for a word the vocabulary has no row for. Only a file can carry one, and the reader of that file archives it rather than guessing.

func ValidListState 

func ValidListState(state string) bool

ValidListState reports whether f asked for a state the contract names.

func VerifyDatabaseFile 

func VerifyDatabaseFile(ctx context.Context, path string, deep bool) error

VerifyDatabaseFile opens path read-only in a fresh handle and runs quick_check, or integrity_check when deep is set. "ok" passes; anything else, including a file that cannot be opened at all, fails. This is the only way a backup counts as taken.

Types

type AggregateMismatch 

type AggregateMismatch struct {
	RunID     string
	State     string
	Aggregate model.RunState
}

AggregateMismatch names one run whose stored state disagrees with what its own steps aggregate to: the exact row issue #20's fifth acceptance item exists for, and the fact behind fsck's I10 finding.

type Artifact 

type Artifact struct {
	StepName  string
	Name      string
	URI       string
	SizeBytes *int64 // nil means the step did not say
	Checksum  string // empty means the step did not say
	MediaType string

	// CreatedAt is set on write and filled on read; it is the verdict's
	// stamp, so a reference ages with the attempt that published it.
	CreatedAt time.Time
}

Artifact is one published reference. MediaType rides in the row's meta_json, which is the schema's slot for exactly this kind of claim.

type ArtifactOwner 

type ArtifactOwner struct {
	Name     string
	StepName string
	Idx      int
}

ArtifactOwner says who currently holds an artifact name inside one run, and where that holder sits in the spec order. The engine consults it before a verdict so a colliding reference resolves deterministically instead of surprising the UNIQUE constraint.

type ArtifactRef 

type ArtifactRef struct {
	RunID    string
	StepName string
	Name     string
	URI      string
	Checksum string
}

ArtifactRef is one staged or copied artifact reference, as the replay proofs compare them: which step made it, what it is called, where the bytes live and how they are fingerprinted. Content never moves; only these facts.

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 BackupInfo 

type BackupInfo struct {
	LastAt        time.Time
	Status        string
	Path          string
	Error         string
	LastDeepCheck time.Time
	// HasBackup is false when this database has never attempted a backup.
	HasBackup bool
}

BackupInfo is what doctor shows about backups: when the last attempt ran, whether the copy verified, where it lives, and why it failed if it did.

func (BackupInfo) Age 

func (b BackupInfo) Age(now time.Time) time.Duration

Age returns how old the last backup attempt is relative to now.

func (BackupInfo) Verified 

func (b BackupInfo) Verified() bool

Verified reports whether the last attempt produced a verified copy.

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 ClaimedStep 

type ClaimedStep struct {
	RunID       string
	Name        string
	Attempt     int
	MaxParallel int
}

ClaimedStep is one step this process just took for its run: everything the executor needs to run and then record it, plus the run's parallel cap read off the same row for sizing a local pool.

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 CutoverEvent 

type CutoverEvent struct {
	ID         string
	Action     string // "cutover" or "rollback"
	JobName    string
	LineNumber int
	LineText   string
	Actor      string
	BackupPath string
	Forced     bool
	CreatedAt  time.Time
}

CutoverEvent is one line a cutover or rollback changed, as the revision trail keeps it. The trail is the operator action's answer to the same question run_events answers for runs: what happened, to what, by whom, and when - written after the crontab itself is safe, because a trail that lies about an operation that failed is worse than a gap.

type DedupVerdict 

type DedupVerdict struct {
	RunKey string
	Seen   bool
	RunID  string
}

DedupVerdict is the dry-run answer for one run key: does this key already have a run in the sensor's current dedup epoch, and which run is it?

type DeliverySnapshot 

type DeliverySnapshot struct {
	BucketCounts []uint64 // aligned with DeliveryBuckets(); +Inf computed as TotalCount
	SumSeconds   float64
	TotalCount   uint64
}

DeliverySnapshot is one scrape-side reading of the histogram: cumulative bucket counts ready to print, plus sum and total for the complementary series.

type ExplainJobFacts 

type ExplainJobFacts struct {
	Paused        bool
	MaxConcurrent int
	Found         bool
}

ExplainJobFacts are the job-level facts a report's summary needs.

type ExplainRun 

type ExplainRun struct {
	ID          string
	JobName     string
	Origin      string
	State       string
	ReasonCode  string
	ReasonText  string
	TriggerID   string
	CreatedAt   time.Time
	StartedAt   time.Time
	FinishedAt  time.Time
	DeferReason string
}

ExplainRun is one run as a timeline child needs it: identity, outcome, timing, and the trigger that caused it.

type ExplainSource 

type ExplainSource struct {
	Kind string // "schedule" or "sensor"
	Name string // source_name as stored on the tick row
}

ExplainSource names one tick producer: a schedule (whose ticks carry "job/name" as their source name) or a sensor (whose ticks carry the bare sensor name). The pairing matters: the two producers spell their source names differently in the ticks table, and a filter that ignores the kind can match the wrong rows.

type ExplainTick 

type ExplainTick struct {
	ID            string
	SourceKind    string
	SourceName    string
	ScheduledFor  time.Time // zero when the tick carries no fire-time
	StartedAt     time.Time
	LastStartedAt time.Time
	FinishedAt    time.Time // zero while still open or not yet closed
	DurationMS    int64
	HasDuration   bool
	RepeatCount   int
	Outcome       string
	ReasonCode    string
	ReasonText    string
	ReasonData    string
	TriggerCount  int
	DedupedCount  int
	CursorBefore  string
	CursorAfter   string
}

ExplainTick is one recorded evaluation as explain reads it back: what was due, when it ran, what it became, and why it produced nothing.

type ExplainTrigger 

type ExplainTrigger struct {
	ID         string
	TickID     string
	JobName    string
	RunKey     string
	CreatedAt  time.Time
	Outcome    string // accepted | deduped | rejected
	ReasonCode string
	ReasonText string
	RunID      string // the run an accepted or deduped trigger led to
}

ExplainTrigger is one trigger decision as explain reads it back.

type ExportStepDep 

type ExportStepDep struct {
	StepName  string `json:"step_name"`
	DependsOn string `json:"depends_on"`
}

ExportStepDep is one edge of the run's step graph.

type ExportTick 

type ExportTick struct {
	ID           string     `json:"id"`
	SourceKind   string     `json:"source_kind"`
	SourceName   string     `json:"source_name"`
	ScheduledFor *time.Time `json:"scheduled_for,omitempty"`
	StartedAt    time.Time  `json:"started_at"`
	FinishedAt   *time.Time `json:"finished_at,omitempty"`
	DurationMS   *int64     `json:"duration_ms,omitempty"`
	RepeatCount  int        `json:"repeat_count"`
	Outcome      string     `json:"outcome"`
	ReasonCode   string     `json:"reason_code"`
	ReasonText   string     `json:"reason_text"`
}

ExportTick is the tick that fired the trigger. Optional fields stay pointers so an absent stamp exports as absent, not as the epoch.

type ExportTrigger 

type ExportTrigger struct {
	ID         string    `json:"id"`
	TickID     string    `json:"tick_id"`
	JobName    string    `json:"job_name"`
	RunKey     string    `json:"run_key"`
	ParamsJSON string    `json:"params_json"`
	CreatedAt  time.Time `json:"created_at"`
	Outcome    string    `json:"outcome"`
	ReasonCode string    `json:"reason_code"`
	ReasonText string    `json:"reason_text"`
}

ExportTrigger is the trigger row that produced (or refused) the run.

type ExportVersion 

type ExportVersion struct {
	ID         string    `json:"id"`
	Version    int       `json:"version"`
	SpecHash   string    `json:"spec_hash"`
	SpecJSON   string    `json:"spec_json"`
	CreatedAt  time.Time `json:"created_at"`
	SourcePath string    `json:"source_path"`
}

ExportVersion is the frozen job version the run executed against: exactly what apply recorded, hash and all.

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 HeldError 

type HeldError struct {
	Hold RunHold
}

HeldError carries the gate's verdict to the command that asked for a run it could not have. Is(ErrRunsHeld) matches it.

func (*HeldError) Error 

func (e *HeldError) Error() string

func (*HeldError) Unwrap 

func (e *HeldError) Unwrap() error

type InputCollision 

type InputCollision struct {
	Name   string `json:"name"`
	Winner string `json:"winner"`
	Loser  string `json:"loser"`
}

InputCollision records one name two contributors claimed, winner first.

type Inputs 

type Inputs struct {
	Artifacts  map[string]InputsRef
	Params     map[string]any
	Collisions []InputCollision
}

Inputs is the whole payload for one step's environment.

func (*Inputs) Marshal 

func (in *Inputs) Marshal() (string, error)

Marshal renders the payload in the frozen shape: 

{"artifacts":{name:{step,uri,media_type,checksum,size_bytes}},"params":{}}

encoding/json sorts map keys, so the same facts always read back the same way, byte for byte.

type InputsRef 

type InputsRef struct {
	StepName  string `json:"step"`
	URI       string `json:"uri"`
	MediaType string `json:"media_type,omitempty"`
	Checksum  string `json:"checksum,omitempty"`
	SizeBytes *int64 `json:"size_bytes"`
}

InputsRef is one reference as a downstream step reads it: whose it was and what it claims about itself. Field order here is the wire order of the frozen shape.

type IntegrityFinding 

type IntegrityFinding struct {
	// Invariant is the catalogue ID: "I3", "reason".
	Invariant string
	// Severity is the catalogue grade the row was found under.
	Severity Severity
	// Violations is how many rows broke the invariant. Never zero: a clean
	// invariant produces no finding.
	Violations int
	// Subjects names the first breaking rows, capped at integritySubjectCap.
	Subjects []string
}

IntegrityFinding is one invariant's outcome from one sweep.

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 Invariant 

type Invariant struct {
	// ID is the check's name in every report: "I1", "I3", "reason".
	ID string
	// Severity grades what the finding means.
	Severity Severity
	// Title is the rule in one line.
	Title string
	// Remedy is what to do about a finding. Obligatory: a report that names
	// a problem without a next step is only half a report.
	Remedy string
}

Invariant is one named rule the sweep checks.

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

	// Schedules is the schedule sync that ran in the same transaction, for
	// the same reason: a job file that declares a cron schedule leaves the
	// apply with a schedule row the scheduler loop can already find.
	Schedules SyncResult
}

JobApplyResult says what apply did with one job spec.

type JobCommandRef 

type JobCommandRef struct {
	JobName string
	Command string
}

JobCommandRef pairs an exec step command with its job name.

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 JobSource 

type JobSource struct {
	Name       string
	SourcePath string
}

JobSource is one job and the file it was loaded from, for the commands that need to walk back to a job's origin.

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

	// Shadow is the job's top-level shadow declaration (#32): every schedule
	// of this job records its ticks and executes nothing. It is resolved into
	// each schedule row rather than stored on the job, so the row stays the
	// one answer the tick path, schedules ls and the shadow report all read.
	Shadow bool

	// 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

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

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 Loss 

type Loss string

Loss names what already owned a fire-time a pass could not claim. Losing to a rival holder is the ordinary follower answer and says nothing; losing to the gap walk's evidence means a decision was overruled by a row that only records that nobody was there, and a caller may want to say so. 

const (
	// LossNone is the zero value: this call lost nothing.
	LossNone Loss = ""

	// LossEvaluation says a real evaluation holds the fire-time, made by an
	// earlier pass of this loop or by a rival holder mid handover.
	LossEvaluation Loss = "evaluation"

	// LossMissed says the gap walk's synthetic explanation holds the
	// fire-time and would not give it up.
	LossMissed Loss = "missed"
)

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 MetricsHook 

type MetricsHook interface {
	// ObserveTick records one decided evaluation. kind is "schedule" or
	// "sensor", name is the schedule or sensor name without its job
	// prefix, outcome is the status that landed in ticks.outcome, and
	// reasonCode is the closed vocabulary that explains a tick which
	// produced no run ("" for one that did).
	ObserveTick(kind, name, outcome, reasonCode string)

	// ObserveLeaseReclaims records n runs whose leases the reaper took
	// away from a dead holder.
	ObserveLeaseReclaims(n int)

	// ObserveNotificationGaveUp records one notification given up on for
	// good. It is an event, not a row state: an operator retry clears the
	// row's failed_at and a delivered row is eventually pruned, so no
	// query over the outbox can count how often we gave up.
	ObserveNotificationGaveUp()
}

MetricsHook receives the event-time observations the in-memory side of /metrics is built from (#40): tick outcomes as their transactions commit, and lease reclaims as the reaper decides them. The interface lives here so the store stays free of any dependency on a metrics package; the implementation lives in the consumer. Nil means nobody is listening, which is the normal state of every process that never serves /metrics.

The hook is called after the transaction has committed, never inside it: a counter must not be able to record something the database refused.

type MetricsInstigatorState 

type MetricsInstigatorState struct {
	Name                string
	Paused              bool
	Cadence             time.Duration // sensors: their interval; schedules: filled by the caller
	CadenceKnown        bool
	LastTick            time.Time // zero when the row never ticked
	NextTick            time.Time
	CursorUpdatedAt     time.Time // sensors only; zero when never moved
	ConsecutiveFailures int64     // sensors only
}

MetricsInstigatorState is the scrape-time state of one schedule or sensor row: what monitoring calls an instigator. Name is the row's own name (the schedule name within its job, or the sensor name), matching the label the tick counters use.

type MetricsIntegrityViolation 

type MetricsIntegrityViolation struct {
	Invariant  string
	Severity   string
	Violations int64
}

MetricsIntegrityViolation is one cell of the newest recorded sweep: how many rows broke one invariant, and at what grade.

type MetricsJobSLA 

type MetricsJobSLA struct {
	Job     string
	Within  time.Duration
	Version string
}

MetricsJobSLA is one job's declared freshness expectation, read back out of the frozen current version rather than mirrored into a column of its own: the versioned spec is the one place the expectation lives.

type MetricsJobStamp 

type MetricsJobStamp struct {
	Job string
	At  time.Time
}

MetricsJobStamp is one timestamped fact about one job.

type MetricsJobStateCount 

type MetricsJobStateCount struct {
	Job   string
	State string
	Count int64
}

MetricsJobStateCount is one cell of the runs grid: how many runs of one job sit in one state right now.

type MetricsSourceLag 

type MetricsSourceLag struct {
	Kind      string // "schedule" or "sensor"
	Name      string
	LagMillis int64
}

MetricsSourceLag is how late the newest decided evaluation of one source fired relative to the fire-time it was claiming.

type MetricsSourceStamp 

type MetricsSourceStamp struct {
	Kind string
	Name string
	At   time.Time
}

MetricsSourceStamp is the newest activity stamp of one source.

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

	// MaxParallel is how many steps of this run may be in flight at once, the
	// budget the claim predicate reads off the run row. Zero means the spec
	// default, which is what a job that declares nothing gets.
	MaxParallel 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 NotificationFilter 

type NotificationFilter struct {
	Since   time.Time // only rows created at or after this instant
	State   string    // "", "pending", "delivered" or "failed"
	Subject string    // exact job/sensor match
	Limit   int       // capped hard below; zero means DefaultListLimit
}

ListNotifications answers `pulseq notifications list`. Newest first, the way an audit reads. All filters are optional; an empty filter lists everything still on disk.

type NotificationSummary 

type NotificationSummary struct {
	ID          int64      `json:"id"`
	Topic       string     `json:"topic"`
	Subject     string     `json:"subject"`
	Target      string     `json:"target"`
	State       string     `json:"state"` // pending | delivered | failed
	CreatedAt   time.Time  `json:"created_at"`
	AvailableAt time.Time  `json:"available_at"`
	Delivered   *time.Time `json:"delivered_at,omitempty"`
	Failed      *time.Time `json:"failed_at,omitempty"`
	Attempts    int        `json:"attempts"`
	LastError   string     `json:"last_error,omitempty"`
	Payload     string     `json:"payload"`
}

NotificationSummary is the CLI-facing view of one outbox row (#29): stable fields, RFC-timestamped facts, the payload verbatim.

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

	// Metrics receives the event-time observations the in-memory counters
	// behind /metrics are built from (#40): tick outcomes as they commit and
	// lease reclaims as they happen. Nil means nobody is listening, which is
	// the normal state of every non-daemon process. The interface lives here
	// rather than in a metrics package so the store stays free of any
	// dependency direction it does not already have.
	Metrics MetricsHook
}

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 OutboxMsg 

type OutboxMsg struct {
	ID       int64
	Topic    string
	Subject  string
	Target   string
	Payload  string
	Attempts int

	// Suppressed is how many later events in the same group collapsed into
	// this delivery ("+49 lignende"). Zero when nothing grouped here.
	Suppressed int64
	// WindowOpenedAt is when the group's delivery window opened; zero for
	// ungrouped messages.
	WindowOpenedAt time.Time
}

OutboxMsg is one claimed notification on its way to its notifier. It carries the window facts the dispatcher merges into the wire payload: an event that was collapsed into this delivery reports how many similar events it swallowed and when the window opened.

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 Policies 

type Policies struct {
	// LogShardDays removes whole log date shards older than this.
	// Key: retention.log_shard_days. Default 14.
	LogShardDays int

	// RunsDays deletes terminal runs older than this; children cascade.
	// Key: retention.runs_days. Default 90.
	RunsDays int

	// RunsKeepMin keeps at least this many newest finished runs per job,
	// regardless of age. Key: retention.runs_keep_min. Default 50.
	RunsKeepMin int

	// TicksSkippedDays deletes ticks with outcome 'skipped' older than this.
	// Key: retention.ticks_skipped_days. Default 7.
	TicksSkippedDays int

	// TicksDays deletes other ticks older than this.
	// Key: retention.ticks_days. Default 90.
	TicksDays int

	// TicksKeepMin keeps at least this many newest ticks per source.
	// Key: retention.ticks_keep_min_per_source. Default 200.
	TicksKeepMin int

	// RunKeysDays deletes dedup keys first seen before this horizon. It is
	// the longest one on purpose, and deleting a key means the trigger it
	// deduplicated can fire again. Key: retention.run_keys_days. Default 365.
	RunKeysDays int

	// SessionsDays deletes stopped daemon sessions older than this, except
	// the keep-minimum below and sessions outages still cite.
	// Key: retention.daemon_sessions_days. Default 90.
	SessionsDays int

	// SessionsKeepMin keeps at least this many newest daemon sessions.
	// Key: retention.daemon_sessions_keep_min. Default 50.
	SessionsKeepMin int

	// OutboxDeliveredDays deletes DELIVERED notification rows older than
	// this (#29). Delivered rows are history until then, and failed rows
	// are kept for ever regardless. Key: retention.outbox_delivered_days.
	// Default 30 (06 section 9.4 via the issue's design note; 07's seven
	// day sketch was superseded by the owner's decision).
	OutboxDeliveredDays int

	// BackupRetain is how many verified backup generations are kept.
	// Key: backup.retain. Default 14.
	BackupRetain int
}

Policies are the retention configuration keys and their defaults. Every value in the table below is a named key an operator can set; these are the values used when nothing is set (06 section 9.4 merged with 07 section 6.2). 

job_versions        never auto-deleted (too small to matter, history must keep meaning)
outages             kept forever (small and precious)
batch limit         200 rows per transaction (PruneBatchLimit; measured
                    against the 50 ms lock budget - see retention.go)
batch pause         50 ms between batches

The batch shape is deliberately not configurable: it is the mechanism that keeps the write lock short, not a tuning knob.

func DefaultPolicies 

func DefaultPolicies() Policies

DefaultPolicies returns the shipped configuration defaults. Tests pin this table field by field: a default that drifts is a silent policy change for every installation that never set the key.

func (Policies) WithDefaults 

func (p Policies) WithDefaults() Policies

withDefaults fills zero fields from DefaultPolicies, so a caller may set only the keys it cares about.

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 ReopenOpts 

type ReopenOpts struct {
	// OnlyStep restricts the reopen to this step plus its transitive
	// downstream closure in the frozen graph of THIS run. Nil reopens every
	// failed and skipped step. A succeeded step named here is refused: its
	// work is what the retry builds on.
	OnlyStep *string

	// RetryBudget is how many attempts every reopened step gains. Zero or
	// less means one. The attempt counter itself is never touched: history
	// stays tight from 1 across generations (02 section 3.3).
	RetryBudget int

	// Forced records that the caller passed --force past the unknown
	// outcome warning. It changes nothing here; it rides on the event so
	// the history shows the reopen was a deliberate double effect risk.
	Forced bool
}

ReopenOpts says how much of the run an operator reopen reopens.

type ReopenResult 

type ReopenResult struct {
	NewEpoch int64    `json:"new_epoch"`
	Reopened []string `json:"reopened"`
}

ReopenResult says what one reopen did: the fencing token the run now sits behind, and which steps are pending again, in spec order.

type RepairConfirmError 

type RepairConfirmError struct {
	// Critical holds the critical findings that require the confirmation.
	Critical []Violation
}

RepairConfirmError reports critical findings under an unconfirmed repair. The CLI turns it into the preservation-and-confirm guidance the startup refusal names; the errors.As chain exists so no caller has to string-match.

func (*RepairConfirmError) Error 

func (e *RepairConfirmError) Error() string

type RepairOutcome 

type RepairOutcome struct {
	// Invariant is the catalogue ID that was repaired.
	Invariant string
	// Repaired counts rows the repair put back in line.
	Repaired int
	// Skipped counts rows the repair left for the operator, with the reason
	// in Note.
	Skipped int
	// Note says why anything was skipped. Empty when nothing was.
	Note string
}

RepairOutcome is what repairing one invariant did.

type ReplayOpts 

type ReplayOpts struct {
	// From spares this step plus everything it transitively depends on in
	// the frozen graph of the source run. The direction matters: naming w
	// in x -> y -> w spares x and y, because they are what w sits on.
	From *string

	// FailedOnly spares exactly the steps that succeeded in the source run.
	FailedOnly bool

	// Actor is who typed the command, recorded on the queued event. Empty
	// becomes "system".
	Actor string
}

ReplayOpts says which steps a replay spares. Neither field means the whole graph runs again.

type ReplayResult 

type ReplayResult struct {
	NewRunID string   `json:"run_id"`
	Reused   []string `json:"reused"`
	Rerun    []string `json:"rerun"`
}

ReplayResult says what one replay made: the new run, which of its steps start as succeeded references, and which have to earn their outputs again. Both lists are in spec order, and together they hold every step exactly once.

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 RetentionPlan 

type RetentionPlan struct {
	Runs         int64 `json:"runs"`
	SkippedTicks int64 `json:"skipped_ticks"`
	Ticks        int64 `json:"ticks"`
	RunKeys      int64 `json:"run_keys"`
	Sessions     int64 `json:"daemon_sessions"`
	// Notifications counts DELIVERED outbox rows past their horizon (#29);
	// failed rows are kept for ever and never estimated as deletable.
	Notifications int64 `json:"notifications"`
}

RetentionPlan is the per-rule estimate `prune --dry-run` shows. Each count answers the same predicate the real delete would, including the keep-minimum protection, without the batch limit.

func (RetentionPlan) Total 

func (p RetentionPlan) Total() int64

Total sums the database-side deletions the plan would perform.

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

	// ReplayOf is the run this one replays, empty when it replays nothing.
	// It is what ties a replay back to the attempt it was made from.
	ReplayOf string

	// 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 RunExport 

type RunExport struct {
	Run        Run             `json:"run"`
	Steps      []Step          `json:"steps"`
	StepDeps   []ExportStepDep `json:"step_deps"`
	Events     []RunEvent      `json:"run_events"`
	Artifacts  []Artifact      `json:"artifacts"`
	Trigger    *ExportTrigger  `json:"trigger,omitempty"`
	Tick       *ExportTick     `json:"tick,omitempty"`
	JobVersion *ExportVersion  `json:"job_version,omitempty"`
}

RunExport is the complete evidence of one run, gathered for `pulseq export run` (06 section 9.4): the run row, its steps with their deps and artifacts, the event chain, the trigger and tick that caused it, and the frozen job version it executed against. Everything here is plain data so the archive stays readable without paceq.

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 RunHold 

type RunHold struct {
	Code reason.Code
	Text string
	Data map[string]any
}

RunHold is what an installed gate reports while new runs are refused: the catalogue code the refusal is stored under, the operator-facing line beside it, and the measurements that make the decision auditable from explain.

type RunHoldFunc 

type RunHoldFunc func() *RunHold

RunHoldFunc is the gate itself. Returning nil, or a hold without a code, admits runs as usual. The daemon installs one that reads the disk-guard's state atomically, so the cost inside the admission transaction is one pointer load - the decision stays inside the one IMMEDIATE transaction that already owns the tick, which is the whole admission idea (07 §4.2).

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 SLAEpisodeChange 

type SLAEpisodeChange struct {
	Job       string
	Breaching bool
	Notes     []model.Notification // targets expanded by the caller at transition time
}

SLAEpisodeChange is one job's freshness verdict, applied under the episode guard. Breaching with no episode row opens one and emits Notes once; recovery deletes the row silently; steady-state breach changes nothing - one notification per breach episode, not per check.

type ScheduleExpressionRow 

type ScheduleExpressionRow struct {
	SourceName string
	JobName    string
	Name       string
	Expr       string
	Timezone   string
	Catchup    string
}

ScheduleExpressionRow carries the planning side of one shadowed schedule, so the report can re-read its expression under different interpretations.

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

	// Shadow puts the schedule into shadow mode on apply: ticks are
	// recorded, nothing executes (#32).
	Shadow bool

	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

	// Shadow marks the schedule as running in shadow mode (#32): every
	// evaluation materialises its tick exactly as it would normally, but no
	// run is created and nothing executes. A global --shadow on serve
	// shadows every schedule regardless of this flag.
	Shadow bool

	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 SensorNameTakenError 

type SensorNameTakenError struct {
	// Sensor is the contested name.
	Sensor string
	// Owner is the job whose row holds the name today.
	Owner string
	// Taker is the job whose spec declares the name.
	Taker string
}

SensorNameTakenError is an apply that would move a sensor row from the job that owns it to a different job. The name is the row's primary key across every job, and the cursor, the dedup epoch and the breaker counters live on that row, so a name that changes hands takes the old owner's evaluation position with it. It is a distinct type so the CLI can report it as the validation refusal it is, under the diagnostic code the in-batch check already uses, rather than as an internal failure.

func (*SensorNameTakenError) Error 

func (e *SensorNameTakenError) Error() string

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 SensorSeedInput 

type SensorSeedInput struct {
	Name     string
	JobName  string
	ExecJSON string
	Paused   bool
}

SensorSeedInput is the definition columns a sensor row needs, for the seed path that materialises a sensor before the apply seam (M3-01) owns it. It is deliberately minimal: the CLI test harness and early fixtures use it to get rows into the table, and it doubles as the single-insert building block the apply path can reuse. Drift state (cursor, dedup_epoch, failures, paused) is left alone on a re-seed, matching the apply guarantee.

type SensorSummary 

type SensorSummary struct {
	Name                string
	JobName             string
	Kind                string
	ExecJSON            string
	IntervalMS          int64
	MinIntervalMS       int64
	TimeoutMS           int64
	MaxTriggersPerTick  int
	Paused              bool
	PausedReason        string
	Cursor              *string
	CursorVersion       int64
	DedupEpoch          int64
	ConsecutiveFailures int
	BreakerOpenedAt     time.Time
	NextEvalAt          int64
	LastTickAt          *int64
	LastOutcome         string
}

SensorSummary is one sensor as the CLI lists it: the definition columns a pause/resume decision needs and the state an operator wants at a glance.

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 is the cursor the evaluation reported. cursorAdvanceOf
	// decides whether this outcome commits it; a value reported by an
	// evaluation that elected nothing is not written anywhere.
	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

	// ReasonData is the JSON detail object beside the code, and it belongs on
	// any tick that has one: a truncated batch is triggered, carries no reason
	// code, and still has to say how many triggers it dropped.
	ReasonData string

	// BreakerExempt marks an errored evaluation that must not burn circuit
	// breaker budget: exit 75 (EX_TEMPFAIL) and exit 64 (EX_USAGE) are read
	// off the result by the caller that classified it. It is read only when
	// Outcome is OutcomeError.
	BreakerExempt bool

	// 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

	// Rejected is how many triggers a held concurrency key refused (#17):
	// no run behind them, only the trigger row recording the refusal.
	Rejected int

	// Coalesced is true when the sensor's previous tick took this evaluation
	// in as another repeat_count step instead of keeping a row of its own.
	Coalesced bool
}

SensorTickCommitResult reports what one atomic commit did.

type SensorTickView 

type SensorTickView struct {
	StartedAt     time.Time
	LastStartedAt time.Time
	RepeatCount   int
	Outcome       string
	ReasonCode    string
	TriggerCount  int
	DedupedCount  int
}

SensorTickView is one recorded sensor tick as the CLI shows it: what happened, when, and how many runs it produced.

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 Severity 

type Severity int

Severity says how much damage a violation implies. 

const (
	// Warning marks cosmetic or historic drift: nothing lies about what
	// happened, but the row would have been written differently today.
	Warning Severity = iota
	// Serious marks state drift the daemon reconciles out of on its own:
	// alarming, event-worthy, but not a reason to refuse startup.
	Serious
	// Critical marks a broken uniqueness rule or a cyclic graph: the state
	// is not one the code can reason about, so startup is refused until an
	// operator confirms.
	Critical
)

func SeverityOf 

func SeverityOf(id string) Severity

SeverityOf looks a check's severity up in the catalogue for callers outside the package, so the daemon's hourly sweep grades findings with the same table the fsck command uses.

func (Severity) String 

func (s Severity) String() string

type ShadowAggregateRow 

type ShadowAggregateRow struct {
	// SourceName is the full source key: "<job>/<schedule>".
	SourceName string

	Outcome    string
	ReasonCode string

	Ticks   int64 // rows in the group
	Repeats int64 // repeat_count sum, so coalesced skips keep their weight

	FirstScheduled *time.Time
	LastScheduled  *time.Time
}

ShadowAggregateRow groups ticks by schedule and verdict over one window. Everything the per-job diff needs flows through here; no report-level SQL lives outside the store.

type ShadowEvidenceSummary 

type ShadowEvidenceSummary struct {
	// TickCount counts every shadow_triggered evaluation including repeats.
	TickCount int64

	// FirstEvidence / LastEvidence bound the window of shadow activity by
	// write time (started_at), not fire time, so uptime answers honestly
	// even on schedules that have not fired yet.
	FirstEvidence *time.Time
	LastEvidence  *time.Time
}

ShadowEvidenceSummary is the thin overview `paceq shadow status` renders.

type ShadowObservation 

type ShadowObservation struct {
	ObservedAt time.Time
	Source     string
	Raw        string
	Command    string
	CronUser   string
	// JobName is set when the command matched an imported job; matching is
	// best effort and stays unset otherwise.
	JobName string
	HasJob  bool
}

ShadowObservation is one cron start seen outside paceq.

type ShadowRuntimeInfo 

type ShadowRuntimeInfo struct {
	Running bool
	Since   time.Time
	Observe string
}

ShadowRuntimeInfo is what `status` and friends read about the running (or last-running) shadow instance. Running reports the persisted flag; Since is when that flag was last raised.

type ShadowTickPoint 

type ShadowTickPoint struct {
	SourceName   string
	ScheduledFor time.Time
}

ShadowTickPoint is one would-run instant, the pairing side of the diff.

type SpoolOutcome 

type SpoolOutcome struct {
	RunID      string
	Step       string
	Attempt    int
	ClaimEpoch int64
	BootID     string
	Outcome    StepOutcome
}

SpoolOutcome is one consumed result file, translated into what the store needs to settle it: the identity the file claims, the fencing token it carried, the boot it was written on, and the verdict itself.

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 StatusJobRow 

type StatusJobRow struct {
	JobName string
	Paused  bool

	RunID      string
	RunState   string // terminal state of the newest finished run
	ReasonCode string
	StartedAt  time.Time
	FinishedAt time.Time

	// DurationMS is the duration the run wrote at finish time. HasDuration
	// distinguishes "took zero milliseconds" from "no duration recorded",
	// which matters to a table cell.
	DurationMS  int64
	HasDuration bool
}

StatusJobRow is what the overview says about one job: its newest finished run beside it. A job that has never finished a run carries an empty RunID; a paused job says so through Paused rather than by disappearing.

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

	// OutcomeSource names where the verdict came from (#39): 'direct' when
	// the executor waited on the process itself, 'spool' when it was read
	// from the attempt's result spool after a crash, 'reconciled' when
	// recovery had no source and had to assume. Empty means the row was
	// written before the column existed, which reads as the direct era.
	OutcomeSource 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

	// Artifacts are the references the step published through its
	// $PACEQ_OUTPUT (#13). Only a succeeded step has any; they are filled
	// by GetRun and RunsArtifacts and never written here.
	Artifacts []Artifact
}

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

	// NoFurtherAttempt is the caller's veto on the retry transition: this
	// ending forecloses another attempt whatever budget the row still
	// holds. The run's own deadline is the case it exists for. A kill by a
	// spent run budget leaves no room for the attempt the step's policy
	// would otherwise buy, and the veto is how the one party that knows
	// why the attempt ended tells the one party that owns the counters.
	//
	// It only ever narrows: false leaves the budget on the row to decide
	// alone, which is what a caller with no policy to consult means.
	// Recovery and the spool committer both write verdicts without one.
	NoFurtherAttempt bool

	// Artifacts are the references the step published through its output
	// file (#13). They are written inside this verdict's transaction and
	// only when the machine lands the step on succeeded: a failed step
	// publishes nothing.
	Artifacts []Artifact

	// OutcomeSource names where this verdict's facts came from (#39):
	// 'direct' from the executor's own wait, 'spool' from the attempt's
	// result file, 'reconciled' from recovery's honest guess. Empty keeps
	// the column NULL, the pre-shim era's shape.
	OutcomeSource string

	// Actor names who wrote this verdict on the events it produces, the
	// step's own and the skips its failure closes. Empty is the run's
	// holder, which is what every event carried before a writer other
	// than the executor took the same path (#213).
	Actor string
}

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.

func SpoolVerdict 

func SpoolVerdict(res spool.Result) (StepOutcome, error)

SpoolVerdict reads one result file as the verdict the step machine takes. The judgment itself is not made here: it comes from StepVerdict, the table the live executor reads too, so the same exit fact cannot read one way when the executor saw it and another way when this file is all that is left of it. What this function adds is what only a file can say — its stamps, its exit code, and that the verdict was read rather than watched. Log facts stay empty on purpose: the shim cannot know how the log ended, and recovery does not guess.

An outcome word the format does not define is an admission error; the caller archives the file rather than guessing.

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) ActiveConcurrencyKeyViolations 

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

ActiveConcurrencyKeyViolations returns every key held by more than one active run: the I12 variant of 02 section 4.3. The partial unique index makes this state unreachable through the code, which is exactly why fsck looks for it anyway: a violation here means a dropped index, a hand edit or a broken migration, not an ordinary bug.

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) ActiveRunsExist 

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

ActiveRunsExist reports whether any run is queued or running right now. The WAL checkpoint stays away while the answer is yes: truncating the WAL mid-flight would fight the writers it belongs to (07 section 6.4).

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 and schedule 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) ApplySLAEpisodes 

func (s *Store) ApplySLAEpisodes(ctx context.Context, changes []SLAEpisodeChange, now time.Time) error

ApplySLAEpisodes applies a checker tick's verdicts. One transaction for the batch keeps flip-flop races between ticks impossible and writes nothing for jobs whose verdict matches their current episode.

func (*Store) ArtifactsOf 

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

ArtifactsOf lists one run's artifact references, oldest first. It reads; nothing is written, and no state is touched.

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) BackupStatus 

func (s *Store) BackupStatus(ctx context.Context) (BackupInfo, error)

BackupStatus reads the backup meta keys. A database that has never backed up returns the zero Info with HasBackup false - that absence is itself a finding doctor must be able to report.

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) BusyTotal 

func (s *Store) BusyTotal() uint64

BusyTotal returns how many SQLITE_BUSY outcomes the write pool has seen since the store was opened. A monotone load, never reset: Prometheus owns the rate arithmetic.

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) ClaimNextStep 

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

ClaimNextStep claims, if the predicate allows, the next step of one run: the pending step with every frozen upstream succeeded, past its retry gate, inside the run's parallel cap, and still owned by the caller's fencing token. It reports whether a step was claimed, never an error for the ordinary nothing-now outcome.

The claim is the reservation: it flips the step to running and writes the step.started event in the same BEGIN IMMEDIATE transaction, so a crash mid claim leaves the run as if the step had never been surveyed.

func (*Store) ClaimOutbox 

func (s *Store) ClaimOutbox(ctx context.Context, limit int, now time.Time, visibility time.Duration) ([]OutboxMsg, error)

ClaimOutbox marks up to limit due rows as being delivered right now and returns them. The update commits first; sending happens strictly outside, which is why a claimed row that crashes stays bounded-visible rather than lost: once visibility elapses the row is due again and claims anew.

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) CommitSpooledOutcome 

func (s *Store) CommitSpooledOutcome(ctx context.Context, in SpoolOutcome) error

CommitSpooledOutcome settles one attempt from its spool result. The checks and the write are one transaction, so the file's whole admission test is atomic with the commit:

  • the run must exist and its lease epoch must equal the one the result carries — the same CAS the reaper's fencing gives every other writer; a mismatch is ErrEpochMismatch and the caller archives the file;
  • the step must be running at the attempt the result names — anything else is ErrUnknownAttempt, including a verdict that already landed, which is what makes consuming twice harmless;
  • the lease must be dead, unless the result was written on another boot: a changed boot id is the strongest evidence there is, and it voids a still-unexpired lease the same way it voids the orphan sweep's wait. A live lease with no boot change means a live executor owns the attempt, and the file stays where it is.

A passed check commits through the same verdict write every other outcome takes, with the caller's outcome source ('spool') on the row.

func (*Store) CountRows 

func (s *Store) CountRows(ctx context.Context, table string) (int64, error)

CountRows returns the row count of one of the tables retention touches. The retention tests assert on it before and after a pass.

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) DaemonSession 

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

DaemonSession answers which paceq daemon serves this state directory: the open session row, and whether the process it names is still there.

The session row is the fact a socket file is not. A socket is an artefact of one optional flag, so a correctly running daemon may have none, while StartSession always writes this row and closes whatever the last run left behind. A row still open therefore belongs either to a live daemon or to one that died without saying goodbye, and the process check is what separates them. The boot id is what makes the process id trustworthy: a pid recorded before a restart names whatever holds that number now.

func (*Store) DoubleCompletedRunKeys 

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

DoubleCompletedRunKeys names every dedup key that two or more succeeded runs share. It is the explicit form of the crash suite's fourth invariant (issue #20): at-least-once execution may duplicate work, never duplicate a completed run behind one key. Fsck does not carry this check, because the invariant is about the relationship between the run_keys decision table and the runs it produced, and the harness asserts it directly after every crash.

A key belongs to exactly one decision, so one succeeded run is the most a healthy database can hold behind it. Failed or cancelled runs do not count: an operator may legitimately see a failed attempt and rerun under a new decision, and a reopened run stays one row.

Empty keys are skipped. A manual run carries no key at all, and an empty string column would otherwise fold every such run into one group.

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) DueSensors 

func (s *Store) DueSensors(ctx context.Context, nowMilli int64, limit int) ([]SensorSummary, error)

DueSensors returns up to limit unpaused sensors that are due at or before nowMilli and past their min_interval floor, most overdue 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) EstimateRetention 

func (s *Store) EstimateRetention(ctx context.Context, p Policies, now time.Time) (RetentionPlan, error)

EstimateRetention counts, per rule, the rows a full retention pass over this database would delete today. It reads through the read-only pool and writes nothing.

func (*Store) ExplainActiveRuns 

func (s *Store) ExplainActiveRuns(ctx context.Context, jobName string) (int, error)

ExplainActiveRuns counts a job's queued and running runs: what the concurrency line of a summary reports. The partial index ux_runs_conc_key's sibling idx_runs_active covers exactly this predicate.

func (*Store) ExplainJobFacts 

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

ExplainJobFacts reads whether a job exists, whether it is paused, and its concurrency ceiling: the summary line of a job report.

func (*Store) ExplainNewestRun 

func (s *Store) ExplainNewestRun(ctx context.Context, jobName, state string) (RunSummary, bool, error)

ExplainNewestRun reads the newest run of a job, optionally restricted to one state (pass "" for any): the last outcome and last success of a summary.

func (*Store) ExplainNextScheduleTick 

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

ExplainNextScheduleTick reads when a job's schedules next come due (the earliest unpaused next_tick_at), or nothing when none is scheduled.

func (*Store) ExplainOutagesSince 

func (s *Store) ExplainOutagesSince(ctx context.Context, since time.Time) ([]Outage, error)

ExplainOutagesSince reads the downtime records that overlap the window, newest first. Outages is already the whole-table listing for tests; this is the windowed form a timeline renders.

func (*Store) ExplainRunEvents 

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

ExplainRunEvents reads one run's event chain, oldest first. This is the read side of AppendRunEvent, and the query explain (M5-01) exists for.

func (*Store) ExplainRunsByJob 

func (s *Store) ExplainRunsByJob(ctx context.Context, jobName, before string, limit int) ([]ExplainRun, error)

ExplainRunsByJob reads one page of a job's runs, newest first, strictly older than before. The plan searches idx_runs_history; there is no OFFSET.

func (*Store) ExplainRunsByPrefix 

func (s *Store) ExplainRunsByPrefix(ctx context.Context, prefix string, limit int) ([]RunSummary, error)

ExplainRunsByPrefix resolves a run id prefix the way git resolves an object: ids are ULIDs, so the prefix is a range scan on the primary key. Every match is returned up to limit, so the caller can name the whole candidate list when the prefix is ambiguous instead of the first two.

func (*Store) ExplainTickByID 

func (s *Store) ExplainTickByID(ctx context.Context, tickID string) (ExplainTick, bool, error)

ExplainTickByID reads one recorded evaluation by its whole id: the provenance line of a run report (which tick fired the trigger that queued this run).

func (*Store) ExplainTicks 

func (s *Store) ExplainTicks(ctx context.Context, sources []ExplainSource, since time.Time, before string, limit int) ([]ExplainTick, error)

ExplainTicks reads one page of recorded evaluations for the named sources, newest id first, strictly older than before (empty before starts at the newest). The sources arrive as resolved kind/name pairs rather than as a subselect over schedules/sensors: the two kinds spell their tick source names differently, and binding them here keeps the plan a search on idx_ticks_source whatever mix the caller asks for.

func (*Store) ExplainTriggerByID 

func (s *Store) ExplainTriggerByID(ctx context.Context, triggerID string) (ExplainTrigger, bool, error)

ExplainTriggerByID reads one trigger by its whole id.

func (*Store) ExplainTriggersByTicks 

func (s *Store) ExplainTriggersByTicks(ctx context.Context, tickIDs []string) (map[string][]ExplainTrigger, error)

ExplainTriggersByTicks reads the triggers a set of ticks produced, keyed by their tick id so the caller can hang them under the right timeline entry.

func (*Store) ExportRun 

func (s *Store) ExportRun(ctx context.Context, idOrPrefix string) (RunExport, error)

ExportRun gathers one run's whole evidence tree. The prefix may be any unambiguous prefix of the run id; ambiguity is an error, never a guess.

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, notes ...model.Notification) (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.

Any notifications ride in this same transaction (#29): state change and alert row commit together or not at all, which is what makes "run failed, nobody heard" unreachable.

func (*Store) ForeignKeys 

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

ForeignKeys reports whether the database enforces its foreign keys. The DSN turns the rule on for every connection; a read-only open cannot set it and only verifies, so a database whose rows a foreign tool edited is still checked on read.

func (*Store) FreelistCount 

func (s *Store) FreelistCount(ctx context.Context) (int64, error)

FreelistCount reports PRAGMA freelist_count.

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) FsckRepair 

func (s *Store) FsckRepair(ctx context.Context, only []string, confirm bool) ([]RepairOutcome, error)

FsckRepair sweeps, then repairs what is safely repairable. only limits the work to named invariants (the --only list; empty means all). confirm is the operator's explicit --confirm, required while any critical finding stands.

The sweep and the repairs are two phases: the repairs re-check their own predicates, so the sweep is only the map of what to look at. The return carries one outcome per invariant that had findings, in no order.

func (*Store) FullVacuum 

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

FullVacuum rewrites the database file. It takes an exclusive lock and needs twice the disk, which is why nothing in the daemon ever calls it: it exists for `paceq db compact --i-know-this-blocks` and nothing else.

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) GetNotification 

func (s *Store) GetNotification(ctx context.Context, id int64) (NotificationSummary, error)

GetNotification answers `pulseq notifications show <id>`.

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) GetSensor 

func (s *Store) GetSensor(ctx context.Context, name string) (SensorSummary, error)

GetSensor reads one sensor row with its last tick, or ErrNotFound.

func (*Store) IncrementalVacuum 

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

IncrementalVacuum releases up to maxPages free pages back to the filesystem. The page cap is what keeps the lock window short and predictable: 2000 pages of ordinary size is roughly 8 MB per night (07 section 6.3). Requires auto_vacuum=INCREMENTAL, which Open arms on new databases; on a database without free pages it is a no-op.

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) InjectArtifact 

func (s *Store) InjectArtifact(ctx context.Context, runID, stepName, name, uri, checksum string) error

InjectArtifact plants one artifact row on a step of a run, so a test can stage what a successful command produced before a replay copies it.

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) InjectDoubleCompletedRunKey 

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

InjectDoubleCompletedRunKey plants the exact row invariant AC-4 of issue 20 exists for: two completed runs sharing one dedup key. Inside one dedup epoch the gate refuses the second registration, so a crash cannot leave this shape, which is why the battery must plant it to prove the checker sees it. Two paths reach the shape legitimately and are not corruption: an operator raising the epoch with sensors reset, and a source re-presenting a run key retention has pruned. The injection twins the oldest existing run under one planted key: both rows carry a real job version, real frozen steps and real edges, and the base keeps whatever history execution gave it. It returns the shared key.

func (*Store) InjectDuplicateRunKey 

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

InjectDuplicateRunKey plants I3: one run claimed by two dedup identities. The run_keys primary key keeps identities apart, and the trigger materialisation writes exactly one identity and one run inside one transaction, so nothing but a hand edit or a partial write can point two identities at one run. That is what the health check re-reads at startup.

Two runs sharing a (job_name, run_key) pair is deliberately not what this plants: a sensor reset and a pruned run key both produce that shape on purpose.

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) InjectRunKey 

func (s *Store) InjectRunKey(ctx context.Context, runID, sourceID, key string) error

InjectRunKey gives one finished run a dedup key as a sensor trigger would: the key on the run row and its registration in the run_keys table. A test stages both so a replay can prove it touches neither.

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) InsertShadowObservation 

func (s *Store) InsertShadowObservation(ctx context.Context, o ShadowObservation) (bool, error)

InsertShadowObservation stores one observed start. The UNIQUE(source, observed_at, raw) key makes re-reading overlapping windows free: a line already recorded folds away silently. Returns whether this call inserted.

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) JobCommandsForMatching 

func (s *Store) JobCommandsForMatching(ctx context.Context) ([]JobCommandRef, error)

JobCommandsForMatching lists every command of every current job version, one entry per exec step. The observed-command matcher walks these to name the import a cron start belongs to. A version whose spec no longer parses is skipped rather than fatal: history matching must not break because one definition rotted.

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) JobsWithoutFreshnessSLA 

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

JobsWithoutFreshnessSLA names the jobs whose current version declares no expected_within, the jobs monitoring cannot alarm on (06 SLO 6). The extraction reads one top-level key, the same cheap read the SLA metric makes; a version whose bytes cannot be read back is skipped, because apply refused it and a scrape must not hang on history.

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) LatestShadowObservationStamp 

func (s *Store) LatestShadowObservationStamp(ctx context.Context, source string) (time.Time, error)

LatestShadowObservationStamp is the newest observation time for one source, used as a fetch watermark; zero means nothing stored yet.

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) ListCutoverEvents 

func (s *Store) ListCutoverEvents(ctx context.Context, limit int) ([]CutoverEvent, error)

ListCutoverEvents reads the newest events, newest first. Status shows the recent ones so the trail is visible from the command line without SQLite; the limit bounds the query, not the trail, which retention owns like any other table.

func (*Store) ListJobSources 

func (s *Store) ListJobSources(ctx context.Context) ([]JobSource, error)

ListJobSources reads every job with the path its current version came from, in name order. A job applied by hand has an empty path: cutover cannot match such a job to a crontab line and must say so rather than guess, which is why the emptiness is data here and not a filter.

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) ListNotifications 

func (s *Store) ListNotifications(ctx context.Context, f NotificationFilter) ([]NotificationSummary, error)

ListNotifications implements the read half of the notifications noun group.

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) ListSensors 

func (s *Store) ListSensors(ctx context.Context) ([]SensorSummary, error)

ListSensors reads every sensor row with its last tick, in name order.

func (*Store) ListShadowObservations 

func (s *Store) ListShadowObservations(ctx context.Context, sinceMs int64, jobFilter string) ([]ShadowObservation, error)

ListShadowObservations returns observations in [sinceMs, ...) ordered by time, optionally narrowed to one job (jobFilter "job/name"; empty is all).

func (*Store) MarkLogShardPruned 

func (s *Store) MarkLogShardPruned(ctx context.Context, shard string) (int64, error)

MarkLogShardPruned clears every step's log_path that points into a removed date shard (#44). steps.log_path must never name a file that does not exist; log_bytes, log_truncated and error_tail survive the clearing, so explain can still show what an attempt printed after the file itself is gone. shard is the directory's name under the log root ("2026-01-02").

It lives in the transition layer, not beside the gate: it rewrites step rows, and the module's rule is that step-row writes belong to this file.

func (*Store) MarkOutboxDelivered 

func (s *Store) MarkOutboxDelivered(ctx context.Context, id int64, now time.Time) error

MarkOutboxDelivered closes a delivery. The row stays forever as history: the audit question "did we notify?" points at this timestamp.

func (*Store) MarkOutboxFailed 

func (s *Store) MarkOutboxFailed(ctx context.Context, id int64, now time.Time, errMsg string) error

MarkOutboxFailed gives up permanently: failed_at seals the row into the pending index's blind spot but keeps every fact on disk. Nothing ever deletes a failed notification silently; only retention horizons or an operator's retry move it afterwards.

Giving up is also an event, counted once here after the write lands. The row cannot carry that count: a retry clears failed_at again.

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) MaterializeReplay 

func (s *Store) MaterializeReplay(ctx context.Context, srcRunID string, opt ReplayOpts) (ReplayResult, error)

MaterializeReplay makes a new run out of a finished one. The source has to be terminal: a queued or running run is somebody's live work, and copying states still being written would freeze a moment that never existed. The new run is queued, so the ordinary claim predicate picks it up with nothing special done anywhere.

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) MetaValue 

func (s *Store) MetaValue(ctx context.Context, key string) (string, bool, error)

MetaValue reads one meta key. The second return is false when the key has never been written, which callers must treat as "never ran" rather than an error.

func (*Store) MetricsDBBytes 

func (s *Store) MetricsDBBytes(ctx context.Context) (int64, int64, error)

MetricsDBBytes returns the size of the main database file in bytes, measured through SQLite's own page accounting, and the size of its WAL companion as the filesystem sees it. Both feed the canary alerts: a WAL that only ever grows is the long lived reader that blocks checkpointing.

func (*Store) MetricsFsckLastRun 

func (s *Store) MetricsFsckLastRun(ctx context.Context) (time.Time, bool, error)

MetricsFsckLastRun returns when the newest sweep ran, clean or not. The second return is false while no sweep has ever run, and the gauge family stays absent: no series without truth. It reads the sweep stamp rather than the event log, because the log holds only the sweeps that found something.

func (*Store) MetricsIntegrityViolations 

func (s *Store) MetricsIntegrityViolations(ctx context.Context) ([]MetricsIntegrityViolation, error)

MetricsIntegrityViolations returns the findings of the newest sweep, in one statement. It selects on that sweep's stamp rather than on the newest row, so a clean sweep empties the family instead of leaving the last failure standing for ever. The gauge therefore answers "what is broken right now, as far as the last sweep knows" without running the sweep at scrape time - a scrape that swept would be the long lived reader 07 section 6.4 forbids.

func (*Store) MetricsJobSLAs 

func (s *Store) MetricsJobSLAs(ctx context.Context) ([]MetricsJobSLA, error)

MetricsJobSLAs returns the freshness expectation of every current job version that declares one, in one joined scan. Jobs without the field have no row, which is what "no series" needs; the extraction reads one top-level key instead of paying for the whole FromIR decode at scrape time.

func (*Store) MetricsLastSuccesses 

func (s *Store) MetricsLastSuccesses(ctx context.Context) ([]MetricsJobStamp, error)

MetricsLastSuccesses returns the newest succeeded finished_at per job, in one grouped scan. A job without a success has no row: absence is the truth, and a fabricated zero would alarm on every healthy newcomer.

func (*Store) MetricsLastTicks 

func (s *Store) MetricsLastTicks(ctx context.Context) ([]MetricsSourceStamp, error)

MetricsLastTicks returns the newest last_started_at per source in one grouped scan. This is the single source for last-tick stamps: the schedules table carries its own copy of the same instant, and exposing both would be the double bookkeeping this endpoint refuses.

func (*Store) MetricsMetaValues 

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

MetricsMetaValues returns whichever of the backup/GC meta keys exist. The map simply lacks the absent ones, so the collector can apply the no-series-without-truth rule without a second round trip per key.

func (*Store) MetricsNotificationsGivenUp 

func (s *Store) MetricsNotificationsGivenUp(ctx context.Context) (int64, error)

MetricsNotificationsGivenUp counts the rows sitting in the given-up state right now (pulseq_notifications_given_up). It is a gauge and only a gauge: an operator retry clears failed_at, and a row that then delivers is pruned with the rest, so the number falls as readily as it rises. How often we gave up is a different question, answered by the event counter the notifier feeds through MetricsHook.

func (*Store) MetricsNotificationsPending 

func (s *Store) MetricsNotificationsPending(ctx context.Context) (int64, error)

MetricsNotificationsPending counts undelivered, ungiven-up rows for /metrics (pulseq_notifications_pending). The partial index answers in O(pending), so a quiet system's scrape reads nothing at all.

func (*Store) MetricsOutageSeconds 

func (s *Store) MetricsOutageSeconds(ctx context.Context) (float64, error)

MetricsOutageSeconds returns the provable downtime total: the summed span of every recorded outage, in seconds. Reading it from the outages ledger at scrape time makes it exact immediately after a restart, which an in-memory copy of the same number would not be.

func (*Store) MetricsRunsByStates 

func (s *Store) MetricsRunsByStates(ctx context.Context) ([]MetricsJobStateCount, error)

MetricsRunsByStates returns every non-empty cell of the runs grid in one grouped scan. Terminal states ride along so the run totals come from the same statement instead of a second pass.

func (*Store) MetricsScheduleStates 

func (s *Store) MetricsScheduleStates(ctx context.Context) ([]MetricsInstigatorState, error)

MetricsScheduleStates returns the state of every schedule row in one scan. Cadence stays unknown here: a cron expression has no stored interval, and computing one belongs to the collector, which owns the cron parser.

func (*Store) MetricsSensorStates 

func (s *Store) MetricsSensorStates(ctx context.Context) ([]MetricsInstigatorState, error)

MetricsSensorStates returns the state of every sensor row in one scan.

func (*Store) MetricsTickLags 

func (s *Store) MetricsTickLags(ctx context.Context) ([]MetricsSourceLag, error)

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) ObserveDelivery 

func (s *Store) ObserveDelivery(d time.Duration)

ObserveDelivery records one notifier send attempt wall time (success or failure alike): what an operator watches is how long alerting blocks the loop, whatever came of it.

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), and this closes out whatever steps are still open - running and pending alike, each through its own event - and then the run itself.

The run's state comes from those closed-out steps, not from the request (I10). A cancellation that arrived after the run had already failed records the failure, and one that arrived after the last step succeeded records the success; the request stays on the row either way, so the history still shows that somebody asked.

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) PageCount 

func (s *Store) PageCount(ctx context.Context) (int64, error)

PageCount reports PRAGMA page_count.

func (*Store) PageSize 

func (s *Store) PageSize(ctx context.Context) (int64, error)

PageSize reports PRAGMA page_size.

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) PauseSensor 

func (s *Store) PauseSensor(ctx context.Context, name, reason string) error

PauseSensor marks a sensor paused with the given reason (or "operator" when none is given). Idempotent: pausing an already paused sensor succeeds.

func (*Store) PeekDedup 

func (s *Store) PeekDedup(ctx context.Context, name string, epoch int64, runKeys []string) ([]DedupVerdict, error)

PeekDedup is the only check-first (read-then-decide WITHOUT a following write) operation in the system, and it is legal for exactly that reason: it feeds `sensors test`, which writes nothing. It must never be copied into a write path (plan 11 section 5.2). It answers, for each run key a dry-run evaluation produced, whether the key already has a run in the current epoch and which run that is, so the CLI can show the dedup verdict per trigger.

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) PruneDeliveredNotificationsBatch 

func (s *Store) PruneDeliveredNotificationsBatch(ctx context.Context, cutoff time.Time) (int64, error)

PruneDeliveredNotificationsBatch deletes delivered rows past the retention horizon, batched like every other deletion here. Failed rows never age out.

func (*Store) PruneOrphanedWindowsBatch 

func (s *Store) PruneOrphanedWindowsBatch(ctx context.Context) (int64, error)

PruneOrphanedWindowsBatch drops throttle bookkeeping whose opener row retention has taken away. Cheap: bounded by the windows table itself.

func (*Store) PruneRunKeysBatch 

func (s *Store) PruneRunKeysBatch(ctx context.Context, cutoff time.Time) (int64, error)

PruneRunKeysBatch deletes up to PruneBatchLimit dedup keys first seen before cutoff.

This is deliberately the last table retention touches, and deleting here has a consequence an operator must be able to read somewhere other than this file: a run_key is what stops an old trigger from firing twice. Once the key is gone, the same trigger can fire again - a schedule replaying after a cursor reset, or a sensor whose dedup epoch moved. That is the documented price of keeping the table bounded (SYNTESE section 3.22), and the restore/runbook documentation owns repeating it in plain language.

func (*Store) PruneRunsBatch 

func (s *Store) PruneRunsBatch(ctx context.Context, cutoff time.Time, keepMin int) (int64, error)

PruneRunsBatch deletes up to PruneBatchLimit terminal runs that finished before cutoff, except each job's newest keepMin finished runs, which are kept no matter how old they are. ON DELETE CASCADE removes the run's steps, step_deps, run_events and artifacts in the same transaction.

func (*Store) PruneSessionsBatch 

func (s *Store) PruneSessionsBatch(ctx context.Context, cutoff time.Time, keepMin int) (int64, error)

PruneSessionsBatch deletes stopped daemon sessions older than cutoff, except the newest keepMin sessions overall and any session an outage row still points at. The outage reference has no ON DELETE action, so a session an outage cites simply survives until the outage does - outages are kept forever anyway (06 section 9.4).

func (*Store) PruneSkippedTicksBatch 

func (s *Store) PruneSkippedTicksBatch(ctx context.Context, cutoff time.Time) (int64, error)

PruneSkippedTicksBatch deletes up to PruneBatchLimit skipped ticks older than cutoff. Skips are the highest volume, lowest value rows in the database once their coalescing window has passed (07 section 6.2), so they carry no keep-minimum: seven days of "nothing was due" is evidence enough. Deleting a tick cascades to its triggers.

Age here is last_started_at, not started_at. A skipped row is a run of identical evaluations, not one: coalesceSkipSQL bumps repeat_count and moves last_started_at, leaving started_at at the first evaluation of the run. started_at is therefore the age of the oldest evaluation a row absorbed, and a row that is still absorbing has an old one by construction; last_started_at is when the coalescing window closed. It is NOT NULL on every tick, so no COALESCE is needed. The ORDER BY id batching stays: it bounds the batch, and ULID ids order by creation.

func (*Store) PruneTicksBatch 

func (s *Store) PruneTicksBatch(ctx context.Context, cutoff time.Time, keepMin int) (int64, error)

PruneTicksBatch deletes up to PruneBatchLimit non-skipped ticks older than cutoff, except each source's newest keepMin ticks. A source is the (source_kind, source_name) pair, so a quarterly schedule keeps its newest 200 evaluations even when every one of them is older than the horizon.

started_at is the right age column here, unlike in PruneSkippedTicksBatch: coalesceSkipSQL is the only writer that moves last_started_at on its own and it only ever folds into a childless skip, so on every row this predicate can reach the two columns hold the same instant.

func (*Store) QuickFsck 

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

QuickFsck returns every violation the startup subset can see, empty when the state is sound enough to serve. Not every finding is critical: the catalogue grades each one, and CriticalViolations names the subset that refuses a boot.

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) ReconcileRunStates 

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

ReconcileRunStates drives every run whose steps have all ended but whose own row still says non-terminal to the state those steps aggregate to. It is the crash backstop (M4-03): an executor that died between its final step verdict and the run verdict leaves a run stranded as running over terminal steps, and this closes it on restart.

It is idempotent. A run that is already terminal, or whose aggregate is still running, is skipped; the second pass changes nothing. It also never reaches a run a live executor is still driving: a run whose lease has not expired still belongs to a process that may be about to finish it.

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) RecordBackup 

func (s *Store) RecordBackup(ctx context.Context, at time.Time, status, path, errMsg string, deepCheckAt time.Time) error

RecordBackup writes the four backup meta keys in one transaction, plus the deep-check stamp when deepCheckAt is non-zero. at/status/path/err describe the most recent attempt, whatever its outcome.

func (*Store) RecordCutoverEvents 

func (s *Store) RecordCutoverEvents(ctx context.Context, events []CutoverEvent) error

RecordCutoverEvents writes one transaction's worth of cutover events. All rows commit together or none do: a cutover of five lines is five rows of one operation, and a trail with three of them would read as a partial change the crontab never had.

func (*Store) RecordIntegritySweep 

func (s *Store) RecordIntegritySweep(ctx context.Context, at time.Time, findings []IntegrityFinding) error

RecordIntegritySweep writes one sweep - the fact that it ran and whatever it found - in one transaction, so a reader never sees half a sweep. A sweep that found nothing still records itself, because "nothing is broken" and "nobody has looked" are different answers and the gauges are asked for both. The store's clock stamps the sweep unless the caller names a moment; the hourly sweep stamps through its own clock so a test can pin the timeline.

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) RecordOpsNotifications 

func (s *Store) RecordOpsNotifications(ctx context.Context, notes []model.Notification) error

RecordOpsNotifications writes operator-facing alerts that did not originate in a run's transaction (#44): disk.low and wal.growth. The rows go through the same dedup and throttle windows as every other outbox write, inside one IMMEDIATE transaction, so a persistent episode is one notification no matter how many guard cycles confirm it.

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) RecordSpoolArchive 

func (s *Store) RecordSpoolArchive(ctx context.Context, runID, step, file, why string) error

RecordSpoolArchive writes the warn event that says a result file was moved aside instead of committed, and why. A rejected result nobody can read about later is indistinguishable from a lost one, which is exactly the silence the spool was built to end (02 R2). The run must exist — the event hangs off it; a file naming no known run is archived with only the log for company.

The event is run level, the step's name riding in the detail: the event-chain invariant (I15) reads each step's partition as one transition story, and an annotation about a rejected file is not a step transition. The orphan-kill event is run level for the same reason.

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 the transition and what it 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. A caller that knows this ending buys no further attempt says so with NoFurtherAttempt, and the step ends terminal however much budget the row still shows.

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) ReopenTerminalRunByOperator 

func (s *Store) ReopenTerminalRunByOperator(ctx context.Context, runID string, actor string, opt ReopenOpts) (ReopenResult, error)

ReopenTerminalRunByOperator reopens a terminal run on an operator's behalf. It is deliberately named for who may call it: no scheduler, reaper or other automation reaches T14, because nothing automatic can spell this method (enforced by the architecture test that walks every caller).

The machine decides the move; this file performs what it demands. The event lands as kind='operator_reopen' (02 section 3.1), carrying the actor, the reason code and the new token, because the fencing history (I11) has to end exactly where the row does.

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) RescheduleOutbox 

func (s *Store) RescheduleOutbox(ctx context.Context, id int64, availableAt time.Time, errMsg string) error

RescheduleOutbox puts a failed attempt back into rotation after backoff, recording the error that caused it.

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) ResumeSensor 

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

ResumeSensor clears a sensor's paused state, its reason, and the whole of its circuit breaker state. Both breaker columns go together: a resume that left breaker_opened_at standing would refuse the sensor work for the rest of its cooldown while every health surface reported it healthy, which is the defect an operator hits while recovering from the first one (#220).

func (*Store) RetryOutbox 

func (s *Store) RetryOutbox(ctx context.Context, id int64) (string, error)

RetryOutbox gives a failed row another chance: available_at moves to the store's own now, attempts keep their count - the operator asked for a retry, not a rewrite of history - and failed_at clears. A delivered row refuses: it did go out, and history does not resend itself.

func (*Store) RunAggregateMismatches 

func (s *Store) RunAggregateMismatches(ctx context.Context) ([]AggregateMismatch, error)

RunAggregateMismatches sweeps every run and returns the ones whose state is not the aggregate of their steps. It is the explicit form of the crash suite's step-and-run consistency rule: at-least-once execution may repeat an attempt, but the run row and its steps must always tell one story.

Fsck reports the same fact as I10 by calling this method, so the harness battery and the invariant engine cannot drift apart. What counts as agreement is model.RunStateAgrees, beside the fold itself, and which reason codes mean the run failed for something no step can express is reason.IsRunLevelFailure, beside the codes themselves, so this package holds no part of the rule and no caller can find one half without the other.

func (*Store) RunArtifactOwners 

func (s *Store) RunArtifactOwners(ctx context.Context, runID string) (map[string]ArtifactOwner, error)

RunArtifactOwners maps every published name of a run to its holder.

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) RunKeysSnapshot 

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

RunKeysSnapshot reads the whole run_keys table as comparable lines. Two snapshots taken either side of an operation prove byte for byte that the table did not move.

func (*Store) RunsArtifacts 

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

RunsArtifacts lists every reference a run published, in spec order and by name within a step, which is the order runs show and explain speak. It is a plain read: references are history, not state.

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) SeedOldFinishedRuns 

func (s *Store) SeedOldFinishedRuns(ctx context.Context, job, versionID string, count, stepsPerRun int, finishedEnd time.Time) error

SeedOldFinishedRuns writes count terminal runs for one job, finished at even intervals ending at finishedEnd, each with stepsPerRun step rows and one run_event per run. It exists for the retention lock-hold gate: that test needs tens of thousands of rows with children, and driving the real claim/finish chain that many times would measure fixture overhead rather than deletion cost. Nothing outside the retention tests may call it.

func (*Store) SeedSkippedTicks 

func (s *Store) SeedSkippedTicks(ctx context.Context, source string, count int, startedEnd time.Time) error

SeedSkippedTicks writes count skipped ticks for one source, started at even intervals ending at startedEnd. Same purpose as SeedOldFinishedRuns: bulk history for the retention tests, and nothing else.

func (*Store) SensorTicks 

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

SensorTicks lists the last N ticks of one sensor, newest first. This is the coalesced history `sensors show --limit N` reads; ticks were coalesced onto one row at write time, so repetition shows as a grow of repeat_count rather than a second row. RepeatCount and LastStartedAt travel with every row, because a limit over folded rows spans an amount of time only the count can tell the reader.

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) SetMeta 

func (s *Store) SetMeta(ctx context.Context, kv map[string]string) error

SetMeta writes key/value pairs into meta inside one transaction.

func (*Store) SetRunHold 

func (s *Store) SetRunHold(fn RunHoldFunc)

SetRunHold installs the gate; nil removes it. Production sets it once at daemon startup and never races it, but the swap is atomic anyway: nothing outside this file needs a lock to consult it.

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) SetSensorDue 

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

SetSensorDue makes a sensor due for evaluation now. It is the store-half of `sensors tick` through the daemon: the CLI asks the daemon, the daemon moves next_eval_at to now, and the evaluator runtime picks the sensor up on its next wake. The runtime wake itself is the M3 emitter's seam; this method only records the operator's request.

func (*Store) SetShadowRuntime 

func (s *Store) SetShadowRuntime(ctx context.Context, shadow bool, observeSource string) error

SetShadowRuntime persists what serve startup decided about shadow mode so commands in other processes can mark their output truthfully. A normal start clears both keys; --shadow sets them along with the chosen source.

func (*Store) ShadowAggregatesWindow 

func (s *Store) ShadowAggregatesWindow(ctx context.Context, sinceMs int64) ([]ShadowAggregateRow, error)

ShadowAggregatesWindow returns the grouped verdicts for the report.

func (*Store) ShadowEvidence 

func (s *Store) ShadowEvidence(ctx context.Context) (ShadowEvidenceSummary, error)

ShadowEvidence summarises how much shadow materialisation exists at all.

func (*Store) ShadowRuntime 

func (s *Store) ShadowRuntime(ctx context.Context) (ShadowRuntimeInfo, error)

ShadowRuntime reads the persisted shadow markers.

func (*Store) ShadowScheduleExpressions 

func (s *Store) ShadowScheduleExpressions(ctx context.Context) ([]ScheduleExpressionRow, error)

ShadowScheduleExpressions lists every schedule's planning shape.

func (*Store) ShadowTriggeredPoints 

func (s *Store) ShadowTriggeredPoints(ctx context.Context, sinceMs int64, jobFilter string) ([]ShadowTickPoint, error)

ShadowTriggeredPoints returns every would-run instant in the window, collapsed by source and instant (the UNIQUE gate guarantees they already are one row).

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) StatusJobs 

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

StatusJobs reads the per-job rows the overview renders. An empty jobName reads every job; a name narrows the answer to one. Jobs come back in name order whether or not they ever finished a run.

func (*Store) StatusNextTicks 

func (s *Store) StatusNextTicks(ctx context.Context) (map[string]int64, error)

StatusNextTicks reads the earliest pending fire-time per job, in milliseconds since the epoch. A job with no standing schedule is absent.

func (*Store) StatusOpenSession 

func (s *Store) StatusOpenSession(ctx context.Context) (version string, since time.Time, found bool, err error)

StatusOpenSession reads the daemon's newest open session. found is false when no session is open, which is the database's half of the daemon-down answer; the live socket dial is the caller's half.

func (*Store) StatusSensorErrorCounts 

func (s *Store) StatusSensorErrorCounts(ctx context.Context) (map[string]int, error)

StatusSensorErrorCounts reads, per job, how many of its sensors are in error. A job absent from the map has healthy sensors (or none).

func (*Store) StatusSensorIntervals 

func (s *Store) StatusSensorIntervals(ctx context.Context) (map[string]int64, error)

StatusSensorIntervals reads the shortest live evaluation interval per sensor-driven job, in milliseconds. A job with no unpaused sensor is absent.

func (*Store) StatusStateCounts 

func (s *Store) StatusStateCounts(ctx context.Context) (queued, running int, err error)

StatusStateCounts reads how many runs are queued and how many are running right now. Absent states count zero.

func (*Store) StatusStuckCounts 

func (s *Store) StatusStuckCounts(ctx context.Context, now time.Time) (map[string]int, error)

StatusStuckCounts reads, per job, how many running runs hold an expired lease as of now. A job absent from the map has nothing stuck.

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) SynchronousMode 

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

SynchronousMode is the database's synchronous setting, read back as the numeric PRAGMA value ("2" is FULL, "1" NORMAL). The DSN pins it per connection, so the read states what a scrape-time connection runs with rather than what a hand-edited config intended.

func (*Store) TakeDelivery 

func (s *Store) TakeDelivery() DeliverySnapshot

TakeDelivery returns the histogram state. Cumulative counters are read directly rather than reset: Prometheus owns rate arithmetic. The separate total counter is the +Inf bucket and stays exact even if two bucket reads straddle a concurrent add.

func (*Store) TakeWriteWaitMax 

func (s *Store) TakeWriteWaitMax() float64

TakeWriteWaitMax returns the wall time of the slowest write transaction since the previous take, in seconds, and resets it to zero. Scrape-side semantics: a max that only ever ratchets up would go stale the first time the database had a genuinely bad second and then never report an ordinary one again. The swap is atomic because writers and the scraper share it.

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.

func (*Store) UpsertSensor 

func (s *Store) UpsertSensor(ctx context.Context, in SensorSeedInput) error

UpsertSensor inserts a sensor definition row, or replaces just the definition columns on a re-seed without touching drift state.

func (*Store) UpstreamInputs 

func (s *Store) UpstreamInputs(ctx context.Context, runID, step string) (*Inputs, error)

UpstreamInputs builds the payload for one step of one run. It reads only; the frozen graph and the published references are history by the time a downstream step starts.

func (*Store) VacuumInto 

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

VacuumInto writes a consistent, defragmented snapshot of the whole database to dst in one statement. Readers and writers keep working while it runs; the snapshot is as of one transaction. It never replaces verification: a copy nobody has checked is a hypothesis, so callers must follow with VerifyDatabaseFile.

func (*Store) WalCheckpointTruncate 

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

WalCheckpointTruncate checkpoints the WAL and truncates it to zero length. Callers gate this behind ActiveRunsExist; the pragma itself does not check.

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

	// Shadow runs this evaluation under shadow mode (#32): the tick row is
	// claimed exactly as normally - same UNIQUE gate, same pause re-read,
	// same coalescing for skips, same cursor advance - but a would-trigger
	// stores outcome 'shadow_triggered' instead of creating any run.
	// Admission is still simulated against real occupancy plus the virtual
	// occupancy earlier shadow fire-times hold while their estimated
	// duration lasts, so overlap stand-downs are recorded with their real
	// reason codes. This flag is read ONLY inside MaterializeTick; the loop
	// fills it from the process-level shadow switch or the schedule row.
	Shadow bool
}

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, an identical earlier skip absorbing it, or a gap row replayed
	// into it. False means the fire-time was already materialised by
	// someone else; nothing was written and no error is reported.
	Claimed bool

	// LostTo names what held the fire-time when Claimed is false. It is
	// LossNone whenever Claimed is true.
	LostTo Loss

	// Replayed is true when this evaluation took its fire-time back from a
	// gap-detection 'missed' row rather than claiming a free slot: the
	// outage is still recorded, and the run it owed exists after all.
	Replayed 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

	// Severity grades the finding (M6-06): critical means the daemon refuses
	// to start, serious means reconcilable drift, warning is cosmetic. It is
	// looked up from the invariant catalogue, so a report can never grade a
	// check differently from the reference.
	Severity Severity

	// Remedy is the catalogue's next step for this check, so a finding can
	// be acted on without reading the source.
	Remedy string
}

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

func CriticalViolations 

func CriticalViolations(all []Violation) []Violation

CriticalViolations returns the findings the catalogue grades Critical: the subset whose presence refuses a startup. Every surface that answers "does this state refuse a boot" reads it from here, so the boot gate and the health report cannot answer differently, and regrading an invariant moves both at once.

The grade comes from the catalogue rather than the Violation's own field, because an untagged finding carries the zero Severity and would otherwise downgrade itself to a warning on the way past.

Source Files

View all Source files

Directories

Path Synopsis
Command gen writes docs/reference/invarianter.md from the invariant catalogue.
 Command gen writes docs/reference/invarianter.md from the invariant catalogue.

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.