run

func (*CheckpointWriter) Force ¶

func (w *CheckpointWriter) Force(rs *RunState, ownerGeneration int64) error

Force writes a full checkpoint and prunes, regardless of cadence.

func (*CheckpointWriter) Maybe ¶

func (w *CheckpointWriter) Maybe(rs *RunState, ownerGeneration int64) error

Maybe writes a checkpoint when the cadence threshold is met, else no-ops.

func (CompletionResult) Durable ¶

func (r CompletionResult) Durable() bool

Durable reports whether the result carries terminal rows the owner must write: a sequence stamped on the completing task, or owner-decided skips. A result that is not Durable must NOT be persisted — CompleteTaskOwner would stamp terminal_sequence = 0 on the row, and recovery reads the terminal tail with a strictly-greater predicate (`terminal_sequence > ?`), so a zero-stamped row is invisible to replay after a takeover.

func (DispatchableTask) ExecutionRef ¶

func (d DispatchableTask) ExecutionRef() uuid.UUID

ExecutionRef is the identity a dispatched task's row-level operations address: the instance when the task is fanned, the catalog task otherwise (which loadTaskRunByIDOrUnique resolves to the run's single row).

func (*LeaseStore) AcquireExpiredLeases ¶

func (ls *LeaseStore) AcquireExpiredLeases(ctx context.Context, newOwner string, ttl time.Duration) (int64, error)

AcquireExpiredLeases takes over every run lease whose holder let it expire (lease_expires_at <= now), reassigning it to newOwner with an incremented generation and a fresh expiry — in a single atomic UPDATE. The expiry predicate is the compare-and-swap: if two nodes sweep concurrently, the first commit moves those rows out of the expired set, so the second updates nothing (no double takeover). Returns the number of leases taken over; the caller's dispatch loop then sees them via OwnedRunsWithGenerations and recovers their in-memory state from the latest checkpoint + terminal tail.

This is the run-owner failover mechanism: in in-memory mode the DB's per-task predecessor counters are stale (the owner advanced the DAG in memory), so ClaimNext recovery does not apply — a peer must take ownership and replay.

func (*LeaseStore) AcquireLease ¶

func (ls *LeaseStore) AcquireLease(ctx context.Context, runID uuid.UUID, ownerNode string, ttl time.Duration) (int64, error)

AcquireLease writes a run_leases row for the given run, recording the owning node and expiry. If a row already exists (e.g., from a previous attempt), it is left unchanged — the initial write is treated as idempotent: whoever wrote it first is the owner.

Returns the generation written on success (always 1 for a fresh lease).

func (*LeaseStore) GetLease ¶

func (ls *LeaseStore) GetLease(ctx context.Context, runID uuid.UUID) (*models.RunLease, error)

GetLease returns the current lease record for runID, if any.

func (*LeaseStore) IsOwner ¶

func (ls *LeaseStore) IsOwner(ctx context.Context, ownerNode string, runID uuid.UUID) (bool, error)

IsOwner returns true if ownerNode currently holds a valid (non-expired) lease on runID. Used to validate requests before acting as owner.

func (*LeaseStore) OwnedRuns ¶

func (ls *LeaseStore) OwnedRuns(ctx context.Context, ownerNode string) ([]uuid.UUID, error)

OwnedRuns returns the IDs of all runs currently owned by ownerNode whose leases have not yet expired. Used by the renewal ticker to build its batch.

func (*LeaseStore) OwnedRunsWithGenerations ¶

func (ls *LeaseStore) OwnedRunsWithGenerations(ctx context.Context, ownerNode string) (map[uuid.UUID]int64, error)

OwnedRunsWithGenerations returns a map of run IDs to lease generations for every non-expired lease owned by ownerNode, in a single query. Used by the dispatch loop to avoid an N+1 GetLease pattern on every tick.

func (*LeaseStore) RenewOwnedLeases ¶

func (ls *LeaseStore) RenewOwnedLeases(ctx context.Context, ownerNode string, newExpiresAt time.Time) (int64, error)

RenewOwnedLeases extends lease_expires_at in a single UPDATE for every non-expired lease owned by ownerNode — no upstream SELECT required. Returns the number of rows actually renewed (which is also the count of currently owned, non-expired leases).

Use this on the per-node renewal ticker; it replaces an OwnedRuns + RenewRunLeases pair with one round-trip.

func (*LeaseStore) RenewRunLeases ¶

func (ls *LeaseStore) RenewRunLeases(ctx context.Context, ownerNode string, runIDs []uuid.UUID, newExpiresAt time.Time) (int64, error)

RenewRunLeases performs a single batched UPDATE extending lease_expires_at for every run in runIDs that is still owned by ownerNode. The WHERE clause on owner_node is the safety net that prevents renewing a lease that was taken over by another node between the decision and the write.

Returns the number of rows actually updated.

func (*OwnerManager) Adopt ¶

func (m *OwnerManager) Adopt(runID uuid.UUID, generation int64) error

Adopt seeds a fresh in-memory state for a run this node created and owns at the given generation. Topology is loaded from the catalog (outside any lock). Idempotent: a second Adopt for an already-tracked run is a no-op.

func (*OwnerManager) Complete ¶

func (m *OwnerManager) Complete(runID, taskID uuid.UUID, status TaskStatus, result, errMsg, claimedBy string, output map[string]string, branchSelections []string) (CompleteResult, error)

Complete applies a worker-reported terminal outcome to the owned run: it resolves any branch skips, advances the in-memory DAG, durably writes the terminal rows (completed task + skips) via CompleteTaskOwner, and checkpoints on cadence. Returns the newly-ready tasks and whether the run is complete. Owned is false when this node does not own the run, signalling the caller to fall back to the SQL path.

Completions can be delivered more than once — a worker re-POSTs the identical envelope when this handler answers 503 for transient dqlite contention, and by then the DAG has already advanced in memory even though the durable write did not land. ApplyCompletion replays what that first delivery decided, so the retry re-persists the same terminal rows at the same sequences. The write is gated on CompletionResult.Durable: a result with nothing this owner stamped must never be persisted, because CompleteTaskOwner would write terminal_sequence = 0 and recovery reads the terminal tail with `terminal_sequence > ?`, making the row invisible after a takeover.

All run work is serialized by the run's own lock; finalize/drop run after that lock is released, so the brief map lock is never held during a DB call.

func (*OwnerManager) CompleteInstance ¶

func (m *OwnerManager) CompleteInstance(runID, taskID, taskRunID uuid.UUID, status TaskStatus, result, errMsg, claimedBy string, output map[string]string, branchSelections []string, partitions []pkgtask.Partition) (CompleteResult, error)

func (*OwnerManager) Drop ¶

func (m *OwnerManager) Drop(runID uuid.UUID)

Drop releases the run's in-memory state (on completion or lease loss). A final checkpoint is forced so a subsequent takeover replays the least tail.

The checkpoint is written BEFORE the run is forgotten, and that order is load-bearing against a concurrent partition retry. Forgetting first leaves a window in which Release finds the run untracked, returns without marking it stale and without deleting its checkpoints — and then this forced "run is complete" snapshot lands after the store's DeleteCheckpoints and survives, so the next recovery restores a complete run and never dispatches the reset instance. Writing first means Release either still sees the run (marks it stale, so this write is a no-op) or runs afterwards and deletes what was written.

func (*OwnerManager) MarkDispatched ¶

func (m *OwnerManager) MarkDispatched(runID, taskID uuid.UUID, worker string, attempt int, leaseExpiresAtMs int64)

MarkDispatched records that a ready task was pushed to a worker.

func (*OwnerManager) Owns ¶

func (m *OwnerManager) Owns(runID uuid.UUID) bool

Owns reports whether this node is tracking in-memory state for the run.

func (*OwnerManager) Ready ¶

func (m *OwnerManager) Ready(runID uuid.UUID) []uuid.UUID

Ready returns the run's current ready queue in dispatch order, or nil if the run is not owned by this node.

func (*OwnerManager) ReadyForDispatch ¶

func (m *OwnerManager) ReadyForDispatch(runID uuid.UUID) []DispatchableTask

ReadyForDispatch returns the run's ready tasks paired with their current attempt, for the dispatch loop to push. Nil if the run is not owned here.

func (*OwnerManager) ReclaimExpiredClaims ¶

func (m *OwnerManager) ReclaimExpiredClaims(runID uuid.UUID) []uuid.UUID

ReclaimExpiredClaims returns this run's in-flight tasks whose worker claim lease has lapsed to the ready queue, and reports which it re-queued.

It closes the one hole the two existing reapers leave between them. A worker that dies mid-task leaves its row `running` with a dead claim_expires_at. Claimer.ReclaimExpired will not touch it — its live-lease guard deliberately skips rows belonging to a run whose owner is alive, so the reaper can never race the owner's dispatch loop. The owner, meanwhile, only ever re-queued in-flight work on TAKEOVER (Recover → requeueRunning); for a run whose owner is perfectly healthy, the instance stayed `running` in memory forever, consuming a fanOut.maxParallel slot and blocking the run from ever completing.

The durable claim_expires_at is authoritative, not the owner's own LeaseExpiresAtMs: a worker renews its claim lease directly and never tells the owner, so the in-memory copy is only a filter (see RunState.AnyLeaseOverdue) — it can over-report, never under-report, and the query settles it.

The reset and the re-queue happen under the run's own lock, so this cannot race the dispatch it feeds. Returns nil for a run this node does not own.

func (*OwnerManager) Recover ¶

func (m *OwnerManager) Recover(runID uuid.UUID, generation int64) (RecoveryResult, error)

Recover rebuilds a run's in-memory state after a lease takeover: it loads the topology, the latest checkpoint, and the post-checkpoint terminal rows, then reconstructs RunState. All of this runs outside any manager lock (the run is not yet published). The RecoveryResult tells the caller which tasks are ready and which running tasks were re-queued for dispatch.

Because the rebuild holds no lock, a store-side invalidation (a partition retry reopening this run) can land in the middle of it. The epoch snapshot taken here is what stops the rebuild from publishing its now-obsolete view over the top: put refuses it, and the loop rebuilds off the current rows. The generation checkpoint is written only AFTER a successful publish, so a refused rebuild leaves nothing durable behind either.

func (*OwnerManager) Release ¶

func (m *OwnerManager) Release(runID uuid.UUID) error

Release forgets a run WITHOUT checkpointing its in-memory state and discards every checkpoint it has on disk. It is for the case where that state is known to be stale — the store refused the owner's completion because a per-partition retry reopened work the state never saw — and a recovering owner must rebuild from the rows instead.

The order is load-bearing. The run is marked stale under its own lock first, so a completion that captured the pointer cannot checkpoint it afterwards; then the checkpoints are deleted while this owner still holds the run; only then is the run forgotten. Forgetting it first would open a window in which a recovery tick restores a checkpoint the invalidation has not reached yet, into a fresh owner that would never see the reset row.

func (*OwnerManager) WithReclaimInterval ¶

func (m *OwnerManager) WithReclaimInterval(d time.Duration) *OwnerManager

WithReclaimInterval overrides the floor between owner-side expired-claim queries for one run. Zero or negative restores the default.

func (*RunState) AnyLeaseOverdue ¶

func (rs *RunState) AnyLeaseOverdue(nowMs int64) bool

AnyLeaseOverdue reports whether some in-flight task's dispatch lease has lapsed — or was never recorded — according to the owner's own bookkeeping.

It is the cheap, allocation-free precondition on the owner-side claim reaper: a run with nothing in flight, or with every in-flight lease still fresh, skips the reap query entirely. It is a NECESSARY condition, never a sufficient one. The owner's copy of a lease is set once at dispatch and never renewed, while the worker renews the durable claim_expires_at without telling the owner — so this can say "overdue" about a task that is alive and well (harmless: the query then finds nothing), but it cannot say "fresh" about a task whose durable claim has already lapsed, because both start from the same deadline. A task with no recorded lease at all is treated as overdue so an unbookkept dispatch can never wedge the run.

func (*RunState) ApplyCompletion ¶

func (rs *RunState) ApplyCompletion(taskID uuid.UUID, status TaskStatus, branchSkipped []uuid.UUID) CompletionResult

ApplyCompletion records a worker-reported terminal outcome for taskID and advances the DAG. branchSkipped lists the task's immediate successors that a branch decision excluded (nil for non-branch tasks); they are skipped and the skip propagates. Remaining successors have their predecessor count decremented and, on reaching zero, are pushed ready or skipped per their trigger rule. Returns the sequence stamped on taskID plus the newly ready and skipped tasks, with Applied true.

A completion can be delivered more than once for the same task: a worker re-POSTs the identical /internal/complete envelope when the owner answers 503 after transient dqlite contention, and by then the DAG has already advanced in memory even though the durable write did not land. Such a repeat delivery does not advance anything (Applied false) but still returns the sequence and skips the first delivery decided, so the caller re-persists the *same* terminal rows. Returning a zero sequence here would let the caller stamp terminal_sequence = 0, which recovery's `terminal_sequence > ?` replay filters out — the row would be invisible to a future takeover. Ready is deliberately not replayed: the ready queue lives in this state and the dispatch loop polls it, so a re-delivery must not look like fresh work.

A completion for a task this state does not know, or one made terminal by recovery replay (which adopts the row's stored sequence rather than stamping one), has no recorded effect: the result is not Durable and must not be persisted.

func (*RunState) ApplyExpansion ¶

func (rs *RunState) ApplyExpansion(exp *FanOutExpansion) []SkippedTask

ApplyExpansion materializes fanned successor groups and in-group adjacency.

It rewrites each instance's OutstandingPredecessors in place before materializing it (see rebaseExpansionOutstanding), so the caller's payload — which CompleteTaskOwner persists — carries the same counts this state applies.

func (*RunState) ApplyTerminalRow ¶

func (rs *RunState) ApplyTerminalRow(taskID uuid.UUID, status TaskStatus, storedSeq int64) []uuid.UUID

ApplyTerminalRow applies a terminal task_runs row observed during recovery replay. Unlike ApplyCompletion it does NOT allocate a new sequence (it adopts the row's stored sequence) and does NOT auto-skip unsatisfied successors — every skip was itself persisted as a terminal row and arrives in sequence order, so re-deriving skips here would double-handle them. It sets the task's terminal status, advances the cursor, and pushes any successor whose trigger rule is now satisfied. Returns the newly-ready successors. Idempotent for a task already terminal in the restored snapshot.

func (*RunState) CatalogTaskID ¶

func (rs *RunState) CatalogTaskID(id uuid.UUID) (uuid.UUID, bool)

CatalogTaskID maps a ready/running identity back to its catalog task. ok is false when the identity *is* a catalog task (an unfanned step), which is how callers tell an instance from a plain task without reaching into the state.

func (*RunState) Clone ¶

func (rs *RunState) Clone() *RunState

Clone returns a deep copy of the mutable state: every map, every slice, and every *OwnerTaskState is duplicated, so a mutation applied to the copy is invisible to the original. The topology is shared by reference — it is immutable for the run's lifetime.

It exists for one reason: the owner must not PUBLISH a DAG transition that its durable write then fails to commit. CompleteInstance used to apply ApplyExpansion + ApplyCompletion straight to the authoritative state and only then call CompleteTaskOwner. When that transaction failed, the producer was already terminal in memory, so the worker's redelivery took the already-terminal branch, skipped expansion re-planning, and re-persisted the completion WITHOUT the expansion — while the owner's ready queue held instance ids that exist in no row. Those dispatch forever against a missing row. Staging in a clone and swapping it in only after the commit makes the two converge: either both the rows and the state advanced, or neither did.

func (*RunState) CompletionIdentity ¶

func (rs *RunState) CompletionIdentity(taskID, taskRunID uuid.UUID) uuid.UUID

CompletionIdentity resolves the key THIS state stores for a completion that names both a catalog task and the row that executed it.

The two sides of the wire disagree about identity on purpose, and this is the single place that reconciles them:

• Dispatch is precise. ReadyForDispatch sets DispatchableTask.TaskRunID only for a real instance and leaves it uuid.Nil for an unfanned step, because that is exactly how this state is keyed — ExpandTask deletes the template node and inserts one node per TaskRunID, and everything else stays keyed by its catalog task id.
• Completion is not. ownerSink.send stamps req.TaskRunID = taskRun.ID for every route and every task (internal/worker/completion_sink.go), which is right for the SQL fallback — loadTaskRunByIDOrUnique takes a primary key or a catalog id interchangeably — and is what makes a fanned sibling's completion unambiguous.

Preferring TaskRunID unconditionally therefore looked an unfanned task's primary key up in a map keyed by its catalog id, missed, and ApplyCompletion reported Applied=false with sequence 0: nothing persisted, no successor released, and the run hung until its harness timed out. Every job has an unfanned task, so that stalled the entire owner in-memory lane.

Asking the state rather than trusting either side keeps both properties: a fanned instance still resolves to its own row (falling back to the catalog id there would resolve an arbitrary sibling), and an unfanned task resolves to the catalog id this state actually stored. The completions map is consulted too so a RE-DELIVERED completion lands on the same key the first delivery stamped and replays its sequence instead of starting a second one.

func (*RunState) ExpandTask ¶

func (rs *RunState) ExpandTask(taskID uuid.UUID, instances []ExpandedInstance)

ExpandTask replaces the single catalog-keyed entry for taskID with N instance-keyed entries (TaskRun IDs), raising total in the same critical section. Unfanned runs never call this.

func (*RunState) HasFailures ¶

func (rs *RunState) HasFailures() bool

HasFailures reports whether any task reached the failed terminal state — used to decide the run's final status when the DAG completes.

func (*RunState) IsComplete ¶

func (rs *RunState) IsComplete() bool

IsComplete reports whether every task in the run has reached a terminal state.

func (*RunState) MarkDispatched ¶

func (rs *RunState) MarkDispatched(taskID uuid.UUID, claimedBy string, attempt int, leaseExpiresAtMs int64)

MarkDispatched records that a ready task was pushed to a worker: it leaves the ready queue and becomes running with the given claim metadata.

func (*RunState) MarkStarted ¶

func (rs *RunState) MarkStarted(taskID uuid.UUID)

MarkStarted records that a dispatched task is genuinely executing — a container exists for it, so cancelling it can no longer prevent it from running. Idempotent; a no-op for an unknown or terminal task.

Deliberately unwired in production, and the reason is the whole point of SyncStartedFromRows below: the owner never observes a start. MarkDispatched fires when a PEER accepts the push, and the worker that eventually creates the container reports back only when the task finishes, so there is no callback this would hang off. The single reader of Started (cancellableBeforeStart, on the fail_fast branch) syncs from the durable rows immediately before it reads, which is what makes the flag authoritative exactly when it matters. This states the transition directly, for tests and for any future seam that does observe one.

func (*RunState) ReadyTasks ¶

func (rs *RunState) ReadyTasks() []uuid.UUID

ReadyTasks returns a copy of the current ready queue in dispatch order, applying per-group fanOut.maxParallel so in-flight instances never exceed the cap.

func (*RunState) RehydrateInGroupEdges ¶

func (rs *RunState) RehydrateInGroupEdges(rows []models.TaskRun, catalog []models.Task)

RehydrateInGroupEdges rebuilds fan-out maps from durable instance rows. Called on recovery after Restore, before replaying the terminal tail.

catalog is the run's catalog Task rows; they carry the FanOutConfig that fanOut.maxParallel is re-seeded from (the cap lives only in memory, so without this a takeover silently drops it for the rest of the run) and the step names the group-duration metric is labelled with. It may be nil, in which case only the edges and partition keys are rebuilt.

func (*RunState) RequeueExpiredRows ¶

func (rs *RunState) RequeueExpiredRows(rows []models.TaskRun) []uuid.UUID

RequeueExpiredRows returns the tasks named by rows — in-flight rows whose worker claim lease the store just reclaimed — to the ready queue with an incremented attempt, and reports which it actually re-queued in dispatch order.

It is the in-memory half of owner-side claim reaping. A worker that dies mid-task leaves its row `running` with a lapsed claim_expires_at, and the claimer's reaper deliberately skips rows belonging to a LIVE-owned run (the owner is supposed to re-dispatch them). Nothing did: the owner counted the instance in flight forever, so a fanOut.maxParallel group wedged and the run never completed. Rows are matched by instance identity first, falling back to the catalog task id for an unfanned step, exactly as SyncStartedFromRows does. Only tasks this state still believes are running are re-queued, so a row reset underneath a completion that already landed is ignored.

func (*RunState) RunningTasks ¶

func (rs *RunState) RunningTasks() []uuid.UUID

RunningTasks returns the IDs of tasks currently in the running state — used during recovery to identify in-flight work a dead owner had dispatched, which the new owner must re-dispatch.

func (*RunState) Sequence ¶

func (rs *RunState) Sequence() int64

Sequence returns the current terminal_sequence cursor (the highest stamped).

func (*RunState) SetGroupFailurePolicy ¶

func (rs *RunState) SetGroupFailurePolicy(catalogID uuid.UUID, policy string)

SetGroupFailurePolicy records a fanned step's fanOut.failurePolicy. The owner learns it from the catalog (at expansion and again on recovery), never from the checkpoint: two copies of one config can disagree after a partial write, and the catalog row is authoritative.

func (*RunState) Snapshot ¶

func (rs *RunState) Snapshot() ([]byte, error)

Snapshot serializes the mutable run state to a checkpoint blob (JSON in v1). The active-only / incremental size optimizations from the design are layered by the checkpoint writer; this produces a complete, self-contained snapshot.

func (*RunState) SyncStartedFromRows ¶

func (rs *RunState) SyncStartedFromRows(rows []models.TaskRun)

SyncStartedFromRows refreshes the Started flag from durable TaskRun rows.

The owner never observes a start directly: MarkDispatched is called when a peer ACCEPTS the push, and the worker that eventually creates the container reports back only when the task finishes. The row's runtime_id is the durable record of "a container exists" (StartTask / StartTaskClaimed write it with the engine's atom id, and only after Create returned), so it is the marker both lanes agree on — deliberately not started_at, which ClaimTaskForDispatch stamps at CLAIM time and which therefore cannot tell a queued task from a running one on the owner-push path.

Callers sync immediately before a decision that depends on the distinction (fail_fast cancellation), so the flag is authoritative exactly when it is read rather than continuously. Rows are matched by instance identity first, falling back to the catalog task id for an unfanned step, mirroring how this state keys its nodes.

func (*RunState) TakeResolvedGroups ¶

func (rs *RunState) TakeResolvedGroups() []ResolvedGroup

TakeResolvedGroups returns every fan-out group that has become fully terminal and has not been reported before, in deterministic dispatch order. It is the owner engine's observation point for caesium_fanout_group_duration_seconds: the caller records the duration and this state never reports that group again.

func (*RunState) TaskState ¶

func (rs *RunState) TaskState(id uuid.UUID) (OwnerTaskState, bool)

TaskState returns a copy of a task's current state, or false if unknown.

func (*Store) AbandonPartitionRetries ¶

func (s *Store) AbandonPartitionRetries(runID uuid.UUID, taskRunIDs []uuid.UUID, reason string) (int, error)

AbandonPartitionRetries resolves the given retry-reset instances, if still pending and unstarted, as skipped with the given reason and clears their retry provenance. It is the bounded end of the completion fence: a replacement engine that could not dispatch the retries it was started for must not spawn yet another engine for them. Each row is resolved explicitly so the run can finalize and the operator can see why.

The guard lives in the UPDATE itself, not in a prior read: an executor can claim and start one of these rows at any moment, and a skip that reached a running instance would clear its provenance and let the run finalize under a container that keeps producing side effects. Returns the number of instances actually resolved.

func (*Store) AbandonPendingPartitionRetries ¶

func (s *Store) AbandonPendingPartitionRetries(runID uuid.UUID, reason string) (int, error)

AbandonPendingPartitionRetries is AbandonPartitionRetries over every retry-reset instance of the run that is still pending. It is for an engine that failed before it could execute anything, where no retry can be told apart from another.

func (*Store) AdmitRun ¶

func (s *Store) AdmitRun(jobID uuid.UUID, triggerID *uuid.UUID, opts ...StartOption) (*JobRun, bool, error)

func (*Store) AdoptStartedRun ¶

func (s *Store) AdoptStartedRun(runID uuid.UUID)

AdoptStartedRun records that this process should clear active-run bookkeeping when Complete sees runID. It is used when a run is created transactionally by a short-lived store instance but executed by the default runtime store.

func (*Store) Bus ¶

func (s *Store) Bus() event.Bus

func (*Store) CacheHitTask ¶

func (s *Store) CacheHitTask(runID, taskID uuid.UUID, source CacheHitSource, result string, output map[string]string, branchSelections []string) (*CompleteTaskResult, error)

CacheHitTask marks a task as completed via cache hit (local mode). It mirrors the CompleteTaskWithResult flow but sets status to "cached".

func (*Store) CacheHitTaskClaimed ¶

func (s *Store) CacheHitTaskClaimed(runID, taskID uuid.UUID, source CacheHitSource, result, claimedBy string, output map[string]string, branchSelections []string) error

CacheHitTaskClaimed marks a claimed task as completed via cache hit (distributed mode).

func (*Store) CacheHitTaskClaimedWithPartitions ¶

func (s *Store) CacheHitTaskClaimedWithPartitions(runID, taskID uuid.UUID, source CacheHitSource, result, claimedBy string, output map[string]string, branchSelections []string, partitions []pkgtask.Partition) error

CacheHitTaskClaimedWithPartitions is the claim-fenced twin of CacheHitTaskWithPartitions, and the method internal/worker/completion_sink.go and internal/dispatch/dispatch.go resolve by INTERFACE ASSERTION (cacheHitPartitionStore). Because the binding is an assertion rather than a compile-time call, a signature drift here does not fail the build — it makes the assertion miss, which those sites report as "run store cannot persist producer partitions; fan-out group will not expand" and then continue without expanding. Keep the signature in lockstep with those declarations.

func (*Store) CacheHitTaskWithPartitions ¶

func (s *Store) CacheHitTaskWithPartitions(runID, taskID uuid.UUID, source CacheHitSource, result string, output map[string]string, branchSelections []string, partitions []pkgtask.Partition) (*CompleteTaskResult, error)

CacheHitTaskWithPartitions is CacheHitTask plus the producer's parsed partition list, expanded inside the SAME transaction that writes the cached terminal row.

A cache hit is a completion, and cacheHitTask is a different function from completeTask — so an expansion hook placed only on the completion route is unreachable whenever a producer's own work cache-hits, and the group silently collapses to its single template row. That is the *common* path, not an edge case: with per-unit fingerprints the whole point is that repeated work hits the cache, and a cached producer still replays its partition list out of the cache entry (internal/worker/runtime_executor.go reads entry.Partitions).

The expansion runs the identical rules as the completion route — it calls expandFanOutSuccessorsTx, so validation (cycles, dangling dependsOn keys, caps), onEmpty handling, in-group indegree seeding and producer-list persistence are one implementation, not a second copy that can drift.

func (*Store) CancelQueuedRun ¶

func (s *Store) CancelQueuedRun(ctx context.Context, jobID, queueID uuid.UUID) error

func (*Store) CancelRun ¶

func (s *Store) CancelRun(ctx context.Context, runID uuid.UUID) error

func (*Store) CheckpointDeltasSince ¶

func (s *Store) CheckpointDeltasSince(runID uuid.UUID, fromSeq int64) ([]models.RunCheckpoint, error)

CheckpointDeltasSince returns the incremental checkpoints with sequence_high strictly greater than fromSeq, ascending, so recovery can apply them in order after the most recent full snapshot. (v1 writes only full snapshots, so this normally returns empty; it exists so the recovery path is delta-ready.)

func (*Store) ClaimTaskForDispatch ¶

func (s *Store) ClaimTaskForDispatch(runID, taskID uuid.UUID, workerNode string, ownerGeneration int64, leaseTTL time.Duration, trustOwnerReadiness bool) error

ClaimTaskForDispatch is the Phase 2 dispatch-side equivalent of ClaimNext for a specific task. It atomically transitions a pending task from (status=pending, claimed_by="") → (status=running, claimed_by=workerNode) in a single UPDATE, mirroring what ClaimNext does but targeting a known task rather than picking the next available one.

The ownerGeneration argument is stamped onto owner_generation so subsequent coordination writes can fence against a stale owner. The WHERE clause includes `AND owner_generation <= ?` to encode the monotonic-generation invariant: a row last touched by the current owner or any *older* generation is claimable (this covers pre-Phase-2A rows at implicit generation 0, normal re-claims at the same generation, and — critically — failover, where a new owner at generation N+1 must re-claim an in-flight task its predecessor stamped at generation N). A row stamped by a *newer* generation means the claimer is itself stale, so the claim is rejected.

Returns ErrTaskClaimMismatch if the task was not in the expected state (already claimed, wrong status, wrong run, stale generation). The caller should fall back to writing the task with claimed_by="" and letting ClaimNext pick it up.

func (*Store) ClaimedTaskRunIDs ¶

func (s *Store) ClaimedTaskRunIDs(ctx context.Context, nodeID string, ids []uuid.UUID) ([]uuid.UUID, error)

ClaimedTaskRunIDs returns the subset of ids whose task_run row is STILL claimed by nodeID. It is the read half of claim-loss detection: the worker asks "of the tasks I am executing, which do I still own?" and stops the containers of the rest.

A read rather than the batched RenewLeases UPDATE, because the two answer different questions on different clocks. Renewal is due only when a claim is within lease_ttl/2 of expiry, and the OWNER's push path stamps claim_expires_at from its dispatch deadline (internal/dispatch/dispatch.go, CAESIUM_RUN_OWNER_DISPATCH_DEADLINE, 5m by default) rather than from CAESIUM_WORKER_LEASE_TTL — so on a run-owner lane a freshly claimed task is not renewal-due for minutes, and a worker that only learned about lost claims from RenewLeases' RowsAffected learned about them minutes late. A cancelled run's container must not outlive the cancel by the renewal cadence, so the question is asked on every tick with a single indexed SELECT instead.

The two ways a claim is lost both surface here: cancelRunTx blanks claimed_by on every non-terminal row of a cancelled run, and a reclaim after lease expiry overwrites claimed_by with the new owner's node id.

func (*Store) Complete ¶

func (s *Store) Complete(runID uuid.UUID, result error) error

func (*Store) CompleteIfActive ¶

func (s *Store) CompleteIfActive(runID uuid.UUID, result error) (bool, error)

CompleteIfActive is Complete that also reports whether THIS call finalized the run: false when the run was already terminal (the idempotent no-op). Callers that dispatch completion callbacks on their own — an engine finalizing a resume that failed before its normal completion path — need the distinction so a run another path finalized first is not notified twice.

func (*Store) CompleteTask ¶

func (s *Store) CompleteTask(runID, taskID uuid.UUID, result string, output map[string]string, branchSelections []string) error

func (*Store) CompleteTaskClaimed ¶

func (s *Store) CompleteTaskClaimed(runID, taskID uuid.UUID, result, claimedBy string, output map[string]string, branchSelections []string) error

func (*Store) CompleteTaskClaimedWithPartitions ¶

func (s *Store) CompleteTaskClaimedWithPartitions(runID, taskID uuid.UUID, result, claimedBy string, output map[string]string, branchSelections []string, partitions []pkgtask.Partition) error

func (*Store) CompleteTaskInstance ¶

func (s *Store) CompleteTaskInstance(taskRunID uuid.UUID, result string, output map[string]string, branchSelections []string, partitions []pkgtask.Partition) (*CompleteTaskResult, error)

CompleteTaskInstance completes a specific TaskRun (fan-out instance) by primary key.

func (*Store) CompleteTaskOwner ¶

func (s *Store) CompleteTaskOwner(
	runID, taskRef uuid.UUID,
	status TaskStatus,
	result, errMsg, claimedBy string,
	output map[string]string,
	branchSelections []string,
	completedSeq, ownerGen int64,
	skips []SkippedTask,
	expansion *FanOutExpansion,
) error

CompleteTaskOwner is the run-owner in-memory path's durable terminal write. The owner has already advanced the DAG in memory (run.RunState), so this only persists terminal rows — it does NOT decrement predecessors, evaluate trigger rules, or resolve branches in SQL. It writes the completed task's terminal row (succeeded/failed/cached) plus each owner-decided skip, stamping terminal_sequence and owner_generation so a recovering owner can replay in order. Claim-fenced by claimedBy.

Cache hits DO travel this path: a cache hit is a completion, and under per-partition fingerprints a cache-hit prerequisite is the common case in an ordered group, so the owner's Cached sink carries its TaskRunID through here like any other terminal transition. (An earlier docstring claimed cache hits stayed on CacheHitTaskClaimed; that stopped being true when the owner sink gained instance identity.)

func (*Store) CompleteTaskWithPartitions ¶

func (s *Store) CompleteTaskWithPartitions(runID, taskID uuid.UUID, result string, output map[string]string, branchSelections []string, partitions []pkgtask.Partition) (*CompleteTaskResult, error)

CompleteTaskWithPartitions is CompleteTaskWithResult plus the producer's parsed partition list, which is expanded inside the completion transaction.

func (*Store) CompleteTaskWithResult ¶

func (s *Store) CompleteTaskWithResult(runID, taskID uuid.UUID, result string, output map[string]string, branchSelections []string) (*CompleteTaskResult, error)

CompleteTaskWithResult completes a task and returns details about branch skips so the local executor can update its in-memory state.

func (*Store) CountActive ¶

func (s *Store) CountActive(jobID uuid.UUID) (int64, error)

func (*Store) DB ¶

func (s *Store) DB() *gorm.DB

func (*Store) DeleteCheckpoints ¶

func (s *Store) DeleteCheckpoints(runID uuid.UUID) error

DeleteCheckpoints removes all checkpoints for a run, called when a terminal run is archived (the durable task_runs rows remain the system of record).

func (*Store) DeleteQueuedRun ¶

func (s *Store) DeleteQueuedRun(ctx context.Context, queued *models.RunQueue) error

func (*Store) DequeueNextRun ¶

func (s *Store) DequeueNextRun(ctx context.Context, jobID uuid.UUID, claimedBy string) (*models.RunQueue, error)

func (*Store) DiffRuns ¶

func (s *Store) DiffRuns(ctx context.Context, jobID, leftRunID, rightRunID uuid.UUID) (*RunDiff, error)

DiffRuns compares the latest terminal task-runs in rightRunID against leftRunID for one job. It pairs rows by task name, diffs each pair's persisted HashInput blob via DiffHashInputBlobs, and returns a JSON-ready read model for API/CLI layers to render.

Pairing considers terminal task-runs only (Succeeded/Failed/Skipped/Cached). A task with no terminal task-run on one side is treated as absent there and reported in TasksAdded/TasksRemoved; this read model is intended for two completed runs, where every task has a terminal attempt.

func (*Store) EnsureTaskRunStartable ¶

func (s *Store) EnsureTaskRunStartable(runID, taskRef uuid.UUID, claimedBy string) error

EnsureTaskRunStartable reports whether a claimed task may still be started, returning ErrTaskClaimMismatch when it may not.

A worker holds its claimed task for as long as its pool takes to free a slot, and in that window the row can be resolved out from under it — fail_fast cancelling a sibling of a group that has already failed is the case this exists for (markInstanceCancelledBeforeStartTx revokes the claim as part of the cancel). The executor calls this immediately before creating the container, because engine.Create both creates AND starts it: checking only afterwards, in StartTaskClaimed, would mean the cancelled work had already run.

Two conditions, either of which means the task is no longer this worker's to start: the row reached a terminal status, or its claim now belongs to someone else (including "" after a release). The check is advisory by nature — it is not the same transaction as the container create — so StartTaskClaimed's guarded UPDATE remains the authoritative fence; this only keeps a doomed container from being created in the first place.

func (*Store) EventStore ¶

func (s *Store) EventStore() *event.Store

func (*Store) FailTask ¶

func (s *Store) FailTask(runID, taskID uuid.UUID, failure error) error

func (*Store) FailTaskClaimed ¶

func (s *Store) FailTaskClaimed(runID, taskID uuid.UUID, failure error, claimedBy string) error

func (*Store) FailTaskInstance ¶

func (s *Store) FailTaskInstance(runID, taskRunID uuid.UUID, failure error) error

FailTaskInstance marks exactly one TaskRun row failed by its primary key, runs the transitive in-group skip cascade, and advances the group's cross-step successors once every sibling is terminal.

It is now a thin alias for failTask: that function's taskRef parameter follows the TaskRun-primary-key-or-catalog-task-ID contract and is no longer reassigned mid-flight, so passing an instance's primary key addresses that row and nothing else. This wrapper survives as the name that says "by instance", and for its nil-id guard.

func (*Store) FanOutInstanceIdentities ¶

func (s *Store) FanOutInstanceIdentities(ctx context.Context, runID, taskID uuid.UUID) ([]InstanceIdentity, error)

FanOutInstanceIdentities returns every instance row of (runID, taskID) in partition-index order with its persisted output and effective identity hash.

Ordering is load-bearing: GroupIdentityHash folds its input in partition-index order, so a caller may pass the successful entries straight through.

func (*Store) FindRunning ¶

func (s *Store) FindRunning(jobID uuid.UUID) (*JobRun, error)

func (*Store) Get ¶

func (s *Store) Get(runID uuid.UUID) (*JobRun, error)

func (*Store) GetTaskLogSnapshot ¶

func (s *Store) GetTaskLogSnapshot(runID, taskID uuid.UUID) (*TaskLogSnapshot, error)

func (*Store) HasAnyFanOutConsumerForRun ¶

func (s *Store) HasAnyFanOutConsumerForRun(runID uuid.UUID) (bool, error)

HasAnyFanOutConsumerForRun reports whether ANY task in runID's job declares a fanOut config at all. It is the cheap pre-filter HasFanOutSuccessor's callers (internal/job/job.go, internal/worker/runtime_executor.go) run first: on an ordinary run that never uses fan-out — the overwhelming majority of cache hits — this single indexed join is the ENTIRE cost of the F7 cache-hit gate, and HasFanOutSuccessor's fuller per-producer check never runs at all.

func (*Store) HasFanOutSuccessor ¶

func (s *Store) HasFanOutSuccessor(runID, producerTaskID uuid.UUID) (bool, error)

func (*Store) InvalidateRunCheckpoints ¶

func (s *Store) InvalidateRunCheckpoints(runID uuid.UUID) error

InvalidateRunCheckpoints drops every owner checkpoint of a run outside a retry transaction. The in-memory owner uses it when the store refused its completion for a pending per-partition retry: the snapshots it wrote since that retry committed describe a run with nothing left to do, and a recovery that restored one would never discover the reset instance.

func (*Store) Latest ¶

func (s *Store) Latest(jobID uuid.UUID) (*JobRun, error)

func (*Store) LatestFullCheckpoint ¶

func (s *Store) LatestFullCheckpoint(runID uuid.UUID) (*models.RunCheckpoint, error)

LatestFullCheckpoint returns the highest-sequence full (non-incremental) snapshot for runID, or (nil, nil) when the run has no checkpoint yet.

func (*Store) LatestSuccessfulCronRun ¶

func (s *Store) LatestSuccessfulCronRun(jobID uuid.UUID) (*JobRun, error)

LatestSuccessfulCronRun returns the most recent cron-triggered run for a job that completed with status "succeeded". It returns gorm.ErrRecordNotFound when no such run exists.

func (*Store) LeaseStore ¶

func (s *Store) LeaseStore() *LeaseStore

LeaseStore returns the run lease store, or nil when owner mode is disabled.

func (*Store) List ¶

func (s *Store) List(jobID uuid.UUID) ([]*JobRun, error)

func (*Store) LoadDispatchedTaskRun ¶

func (s *Store) LoadDispatchedTaskRun(runID, taskID uuid.UUID, claimedBy string) (*models.TaskRun, error)

LoadDispatchedTaskRun loads the full task_runs row for a task that was just claimed for dispatch. The (claimedBy, status=running) predicate ensures the row really is the one this node claimed via ClaimTaskForDispatch and not a row another node has since reclaimed. Returns ErrTaskClaimMismatch if no matching running row exists. The dispatch handler uses this to obtain the full execution spec (image/command/engine/etc.) to hand to the worker pool.

func (*Store) LoadRunTopology ¶

func (s *Store) LoadRunTopology(runID uuid.UUID) (RunTopology, error)

LoadRunTopology reads a run's immutable DAG shape and builds the RunTopology the owner's in-memory RunState needs. Normal runs use the live job catalog; quarantined replay runs use the per-task execution descriptors captured on the TaskRun rows so a later apply cannot change replay dispatch order.

The two sources differ ONLY in how they enumerate nodes and edges; both feed the same buildRunTopology.

func (*Store) PendingPartitionRetries ¶

func (s *Store) PendingPartitionRetries(runID uuid.UUID) ([]models.TaskRun, error)

PendingPartitionRetries lists the retry-reset instances of a run that are still pending — exactly the rows Complete's fence refuses on. Only id and task_id are populated.

func (*Store) PendingTasksForDispatch ¶

func (s *Store) PendingTasksForDispatch(ctx context.Context, runID uuid.UUID, limit int) ([]models.TaskRun, error)

PendingTasksForDispatch returns up to limit task_runs rows for runID that are ready for owner-push dispatch: status=pending, claimed_by="", and outstanding_predecessors=0. The caller (the dispatch loop) uses this to find the next batch of tasks to push to workers each tick.

The result is ordered by created_at ASC so earlier-registered tasks are dispatched first, preserving FIFO ordering within a run. The limit cap prevents a huge fan-out from saturating a single tick.

func (*Store) PlanFanOutExpansion ¶

func (s *Store) PlanFanOutExpansion(runID, producerTaskID uuid.UUID, partitions []pkgtask.Partition) (*FanOutExpansion, error)

PlanFanOutExpansion validates partitions and assigns instance IDs without writing rows. The owner in-memory path applies this to RunState first, then persists the same IDs inside CompleteTaskOwner.

The producer is addressed by its catalog task id, which resolves its TaskRun only while that task is UNFANNED. Use PlanFanOutExpansionForRow when the producer may itself be a fanned instance.

func (*Store) PlanFanOutExpansionForRow ¶

func (s *Store) PlanFanOutExpansionForRow(runID, producerTaskID, producerRowRef uuid.UUID, partitions []pkgtask.Partition) (*FanOutExpansion, error)

PlanFanOutExpansionForRow is PlanFanOutExpansion with the producer's two identities stated separately, because the function genuinely needs both: producerTaskID is the CATALOG task (the Task row whose name fanOut.from matches, and the from_task_id the successor edges hang off), while producerRowRef names the TaskRun that actually ran.

Collapsing them broke every fanned producer. A fanned step that emits partitions of its own — and, more commonly, ANY instance completing through OwnerManager.CompleteInstance, which plans an expansion on every success — resolves its catalog id to N rows, so the single-row load failed with ErrAmbiguousTaskRun. The owner treats a planning error as the task having FAILED, so a partition that succeeded was recorded failed, and under the default fail_fast policy that killed its pending siblings and the run. The error surfaced only as a `record not found` line in the query log.

producerRowRef follows the usual TaskRun-primary-key-or-catalog-task-ID contract, so an unfanned producer may still pass its catalog id.

func (*Store) PredecessorDescriptorInputs ¶

func (s *Store) PredecessorDescriptorInputs(runID, taskID uuid.UUID) (map[uuid.UUID]map[string]string, map[uuid.UUID]string, error)

PredecessorDescriptorInputs returns predecessor outputs and effective hashes keyed by predecessor task id for immutable execution-descriptor capture.

Group-aware for the same reason PredecessorOutputs and PredecessorHashes are: the descriptor records the inputs a task actually consumed, so it must record the same aggregate the cache key was computed from. Unfanned predecessors are byte-identical to the pre-fan-out behavior (one row, its own output map, its own effective hash).

func (*Store) PredecessorHashes ¶

func (s *Store) PredecessorHashes(runID, taskID uuid.UUID) ([]string, error)

PredecessorHashes returns the execution hashes recorded on predecessor task runs that completed successfully in the current run. This keeps distributed cache hashing aligned with local execution, including transitive cache hits.

The hash returned per predecessor is its EFFECTIVE identity: effective_hash when a value-verified short-circuit was proven for that predecessor (its code changed but it produced byte-identical output, see cache.EquivalentPriorHash and TaskRun.EffectiveHash), otherwise its own hash. Reading the effective hash is what stops a no-op upstream change from cascading a re-run downstream: the predecessor presents its prior, proven-equivalent identity, so a downstream whose only changed input was this predecessor sees an unchanged hash and cache-hits. Falling back to hash (effective_hash empty) is the common case and is byte-identical to the pre-D2 behavior.

func (*Store) PredecessorOutputs ¶

func (s *Store) PredecessorOutputs(runID, taskID uuid.UUID) (map[string]map[string]string, error)

PredecessorOutputs returns a map of step-name → output key-values for all predecessors of the given task within a run. This is used by the distributed executor to inject CAESIUM_OUTPUT_* env vars before starting a task. PredecessorOutputs returns each predecessor step's outputs keyed by step name. One entry per PREDECESSOR (never per row): a fanned predecessor contributes the group aggregate, see predecessorGroupOutput.

func (*Store) PruneCheckpoints ¶

func (s *Store) PruneCheckpoints(runID uuid.UUID, keepFulls int) error

PruneCheckpoints retains the most recent keepFulls full snapshots for runID (and every checkpoint at or after the oldest retained full's sequence), deleting anything older. A no-op until more than keepFulls fulls exist.

func (*Store) PublishEvents ¶

func (s *Store) PublishEvents(events ...event.Event)

func (*Store) RateLimitTask ¶

func (s *Store) RateLimitTask(ctx context.Context, runID, taskRef uuid.UUID, retryAfter time.Time) error

RateLimitTask leaves a task pending until retryAfter so rate-limit rejections do not hold worker capacity or spin through immediate reclaims.

taskRef follows the TaskRun-primary-key-or-catalog-task-ID contract, so a fan-out instance parks by its own TaskRun ID and its siblings keep running.

G1 open question — settled here: the `status IN (pending, running)` predicate is KEPT, and it is deliberate rather than a leftover. Both claim paths flip a row to `running` at CLAIM time (ClaimTaskForDispatch, and the claimer's atomic UPDATE), and the rate-limit rejection is only discovered afterwards, before any container exists — so the row this parks is legitimately `running` and has nothing in flight to orphan. What made the predicate dangerous was never the status set; it was the `WHERE job_run_id = ? AND task_id = ?` predicate, which re-pended every RUNNING sibling of a fanned group and orphaned their live containers. Now that the write is keyed to one instance's primary key, matching `running` affects exactly the row whose own rate-limit acquisition was rejected. Callers must therefore pass the instance's TaskRun ID; passing a catalog task ID for an expanded group now fails loudly with ErrAmbiguousTaskRun instead of silently parking a sibling. RateLimitTask parks ONE task instance until retryAfter, releasing its claim so the next tick can re-acquire the rate-limit token.

taskRef follows the TaskRun-primary-key-or-catalog-task-ID contract; a fanned group must be addressed by instance, and a catalog task ID naming N siblings is refused (ErrAmbiguousTaskRun) rather than parking an arbitrary one.

The `status IN (pending, running)` predicate was G1's one open question — it is KEPT, deliberately. Matching `running` looks like it could re-pend a live instance and orphan its container, but the rate-limit rejection is discovered AFTER the claim has already flipped the row to running and BEFORE any container exists (acquireTaskRateLimit runs at dispatch, not mid-execution), so the running row this matches has nothing in flight. What was a real bug is now closed by the re-key: the old (job_run_id, task_id) predicate fanned the re-pend across every sibling, so parking one instance re-pended its RUNNING siblings and did orphan their containers. Pinned by TestRateLimitTaskParksOneInstance.

func (*Store) ReclaimOwnerExpiredClaims ¶

func (s *Store) ReclaimOwnerExpiredClaims(runID uuid.UUID, ownerGeneration int64) ([]models.TaskRun, error)

ReclaimOwnerExpiredClaims returns this run's in-flight rows whose worker claim lease has lapsed to the dispatchable pending state and reports the rows it reset.

It is the OWNER's half of expired-claim reaping, and it exists because the worker-side reaper deliberately declines to do it. Claimer.ReclaimExpired guards on `NOT EXISTS (... run_leases ... lease_expires_at > now)`: a task belonging to a live-owned run is left alone so the reaper can never race the owner's dispatch loop into double-executing it. In run-owner in-memory mode nothing then completed the thought — a worker that died mid-task left its row `running` with a dead claim, the owner counted it in flight forever, and a fanOut.maxParallel group wedged behind a slot that never came back. This runs under the owner's own per-run lock, so it is the one implementation of the reset and cannot race the loop it serves.

Fencing mirrors ClaimTaskForDispatch's: `owner_generation <= ownerGeneration` accepts rows this owner stamped and legacy generation-0 rows, and rejects rows a NEWER owner has already taken over — a lease this node lost must not have its claims reset from under the node that now holds it. The reset columns are exactly Claimer.ReclaimExpired's, including leaving `attempt` alone: the retry-policy attempt counter is not what a lease expiry consumes (the owner's in-memory attempt is bumped by RunState.RequeueExpiredRows for dispatch bookkeeping).

func (*Store) RecordEventTx ¶

func (s *Store) RecordEventTx(tx *gorm.DB, evt *event.Event) error

func (*Store) RegisterTask ¶

func (s *Store) RegisterTask(runID uuid.UUID, task *models.Task, atom *models.Atom, outstanding int) error

func (*Store) RegisterTasks ¶

func (s *Store) RegisterTasks(runID uuid.UUID, inputs []RegisterTaskInput) error

func (*Store) ReleaseQueuedRun ¶

func (s *Store) ReleaseQueuedRun(ctx context.Context, queueID uuid.UUID, claimedBy string) error

func (*Store) ReleaseTaskClaim ¶

func (s *Store) ReleaseTaskClaim(runID, taskID uuid.UUID, claimedBy string, ownerGeneration int64) error

ReleaseTaskClaim reverts a task this node claimed for dispatch back to the dispatchable pending state (status=running → pending, claimed_by="", claim_expires_at=nil, runtime_id="", started_at=nil). It is the rollback used by HandleDispatch when the local worker cannot accept a just-claimed task (buffer full / worker not running): rather than leave the task claimed-but-orphaned, the owner returns it to the pool so the next dispatch tick re-dispatches it (to this or another peer).

The owner_generation predicate keeps the release fenced: only the owner that stamped the row (or a legacy generation-0 row) can release it. The status and claimed_by predicates make the release a no-op (zero rows, no error) if the task already advanced — e.g. a completion landed in the race window.

func (*Store) RenewLeases ¶

func (s *Store) RenewLeases(ctx context.Context, nodeID string, ids []uuid.UUID, newExpiresAt time.Time) (int64, error)

RenewLeases extends claim_expires_at for all task runs identified by ids that are still claimed by nodeID. The WHERE clause on claimed_by ensures that any task whose claim was reassigned after expiry is not accidentally extended. An empty ids slice is a no-op (no database round-trip). Returns the number of rows actually updated so callers can credit metrics accurately and detect the case where a claim was reassigned between the renewal decision and the write.

func (*Store) ResetInFlightTasks ¶

func (s *Store) ResetInFlightTasks(runID uuid.UUID) error

func (*Store) ResolveBranchSkips ¶

func (s *Store) ResolveBranchSkips(runID, taskID uuid.UUID, branchSelections []string) ([]uuid.UUID, error)

ResolveBranchSkips returns the immediate successor task IDs a branch task excluded at runtime via its branch selections — i.e. the successors to skip. It returns nil for non-branch tasks (which skip nothing here). It errors if a selection names a step that is not a valid successor, matching completeTask's validation.

G7: live and replay differ ONLY in how they enumerate a branch's successors, so that is the one thing that forks (branchTargetsTx) and the selection algorithm below exists once. Previously both halves carried their own copy of "build the name map, validate each selection, invert the selection into skips", which is the shape where a fix lands on one path and quietly does not on the other.

func (*Store) RetryFromFailure ¶

func (s *Store) RetryFromFailure(runID uuid.UUID) (*JobRun, error)

RetryFromFailure resets a failed run so that previously-succeeded and cached tasks are preserved and only failed/pending/skipped tasks are re-executed. This is the manual/human retry entry point (caesium run retry); it performs no concurrency re-admission and does not consult Job.Paused — a human retrying a paused job's run is a deliberate human decision.

func (*Store) RetryFromFailureAdmitted ¶

func (s *Store) RetryFromFailureAdmitted(runID uuid.UUID) (*JobRun, error)

RetryFromFailureAdmitted is the admit-aware retry entry point used by the incident action executor's retry actions (retry_from_failure, snooze_retry, retry_callbacks re-run). It adds the two safety valves the plain RetryFromFailure store call lacks (design-agent-in-the-loop.md):

1. It refuses while the job is Paused (returns ErrJobPaused) — a human pause outranks an agent retry.
2. It re-admits against metadata.concurrency using QUEUE semantics regardless of the job's declared strategy: the run is flipped back to running only if a concurrency slot is free (returns ErrMaxConcurrentRunsReached otherwise). An agent retry must never replace-cancel a live run, nor race the next cron tick for a slot admission never granted.

func (*Store) RetryPartition ¶

func (s *Store) RetryPartition(ctx context.Context, runID, taskRunID uuid.UUID) (*TaskRun, bool, error)

RetryPartition resets ONE fan-out instance of a run for re-execution.

It is the store half of `POST …/tasks/:task_id/partitions/:index/retry`. The controller previously did a bare `db.Model(&row).Updates(...)`: no transaction, no terminal-only guard (so a RUNNING instance could be reset mid-flight, orphaning its container), no reset of claimed_by / started_at / runtime_id / claim_expires_at / the cache columns, no outstanding_predecessors re-seed (so an ordered instance came back ready even though its in-group dependency had also been reset), no run re-open (so retrying an instance of an already-terminal run left the run terminal and nothing ever dispatched it), and no event.

Semantics, mirroring retryFromFailure for exactly one row:

• terminal instances only (ErrTaskRunNotTerminal otherwise)
• the full retryResetColumns reset
• outstanding_predecessors re-seeded over NON-TERMINAL dependencies only
• the run re-opened when it had already finished
• checkpoints invalidated so owner in-memory mode sees the reset
• a task_ready event when the instance is immediately dispatchable

The group is deliberately NOT re-expanded (the producer is terminal and the recorded instances are reused), and dependents that already succeeded are NOT cascaded — E2 requires the CLI to say so rather than silently re-running a subtree.

The bool is true iff this transaction reopened a terminal run. Callers that start an in-process engine (the HTTP handler) MUST use this flag rather than a pre-tx status snapshot: a running local run can complete after that read and be reopened here, and kicking off from the stale "running" status would leave the reset instance pending forever. If the tx still saw the run running it does not reopen — the in-process loop is alive — and a second Run() would race it.

The returned TaskRun is loaded inside that same transaction after the reset. A post-commit refresh must not be the source of the kickoff signal: if it failed, returning (nil, false, err) would discard a committed reopen and the handler would skip kickoff.

func (*Store) RetryTask ¶

func (s *Store) RetryTask(runID, taskRef uuid.UUID, attempt int) error

RetryTask resets a failed task run back to pending and increments its Attempt counter. It is the LOCAL lane's in-run retry: internal/job calls it for each `retries:` attempt while it drives the DAG itself.

There is deliberately no claim-fenced twin. RetryTaskClaimed used to be one and was unusable: it re-pends the row, while StartTaskClaimed only starts a row whose status is `running`, so the attempt that followed a claimed retry could never start. The distributed lane uses RetryTaskClaimedInstance (internal/run/store_instance.go), which keeps the row `running` under the same claim — the truthful state for a worker that never released the task and is about to launch the next container itself.

func (*Store) RetryTaskClaimedInstance ¶

func (s *Store) RetryTaskClaimedInstance(runID, taskRunID uuid.UUID, attempt int, claimedBy string) error

RetryTaskClaimedInstance resets one instance for its next attempt WITHOUT releasing it, for a worker that is retrying a task it still holds.

It differs from RetryTaskClaimed in the one respect that made that method unusable: RetryTaskClaimed re-pends the row, and StartTaskClaimed will only start a row whose status is `running`, so the attempt that followed a retry could never start — it hit ErrTaskClaimMismatch, tore its container down and abandoned the task mid-budget, whatever the step's `retries` said. The worker never released the claim and is about to launch the next container itself, so `running` + the same claim is the truthful state between attempts; re-pending described a task waiting to be picked up that nothing was going to pick up.

The guard is the claim fence: only a row still RUNNING and still claimed by this worker is reset. Anything else (a fail_fast cancellation, a lease that expired and was re-claimed elsewhere) yields ErrTaskClaimMismatch, which the executor already treats as "abandon quietly".

func (*Store) RetryTaskInstance ¶

func (s *Store) RetryTaskInstance(runID, taskRunID uuid.UUID, attempt int) error

RetryTaskInstance resets one fan-out instance for another attempt.

It is the instance-keyed analogue of RetryTask, which resolves its row via loadUniqueTaskRun and therefore cannot address a group member. Only a non-terminal-or-failed row is reset: the guard keeps a retry from resurrecting an instance that a fail_fast cancellation or an in-group skip cascade has already resolved (the same class of bug as the local replace-cancel resurrection fixed in #275).

func (*Store) SaveSchemaViolations ¶

func (s *Store) SaveSchemaViolations(runID, taskRef uuid.UUID, violations []pkgtask.SchemaViolation) error

SaveSchemaViolations persists schema validation violations onto exactly one task run. taskRef follows the TaskRun-primary-key-or-catalog-task-ID contract: a fan-out instance must be addressed by its TaskRun ID. The old `WHERE job_run_id = ? AND task_id = ?` predicate broadcast one instance's violations across every sibling row, so a single bad partition made all N look schema-invalid. Resolving through loadTaskRunByIDOrUnique means an ambiguous catalog task ID now fails loudly (ErrAmbiguousTaskRun) instead of fanning the write.

func (*Store) SaveTaskLogSnapshot ¶

func (s *Store) SaveTaskLogSnapshot(runID, taskRef uuid.UUID, snapshot *TaskLogSnapshot) error

SaveTaskLogSnapshot persists the captured log snapshot onto exactly one task run. taskRef follows the TaskRun-primary-key-or-catalog-task-ID contract: a fan-out instance must be addressed by its TaskRun ID, otherwise the snapshot would be broadcast across every sibling row sharing (job_run_id, task_id).

func (*Store) SetBus ¶

func (s *Store) SetBus(bus event.Bus)

func (*Store) SetRunStateCache ¶

func (s *Store) SetRunStateCache(inv runStateInvalidator)

SetRunStateCache registers the in-memory run state layered over this store. Called by NewOwnerManager so the wiring lives next to the thing being wired rather than in the server bootstrap, where forgetting it would look like nothing at all.

func (*Store) SetTaskEffectiveHash ¶

func (s *Store) SetTaskEffectiveHash(runID, taskRef uuid.UUID, effectiveHash string) error

SetTaskEffectiveHash records the proven-equivalent prior identity a task presents to its downstream consumers when a value-verified short-circuit was proven (design Component 5 / D2). It writes ONLY the effective_hash column; the task's own Hash, output, and result are untouched, so its receipt and `caesium why` still reflect its true identity. Passing an empty effectiveHash is a no-op (the common case — no short-circuit), keeping the column nil and PredecessorHashes falling back to the true hash. This is the only writer of effective_hash, so a downstream reader observes either the proven prior identity or nothing. taskRef follows the TaskRun-primary-key-or-catalog-task-ID contract: a value-verified short-circuit is proven per INSTANCE (its own output was byte-identical), so it must be recorded on that instance's row.

func (*Store) SetTaskExitCode ¶

func (s *Store) SetTaskExitCode(runID, taskRef uuid.UUID, exitCode *int) error

SetTaskExitCode persists the raw process exit code the engine reported at task completion onto the task run. The incident classifier reads this alongside SchemaViolations/Result to bucket a failure into a failure_class. Best-effort: a nil-safe no-op when the row is gone. taskRef follows the TaskRun-primary-key-or-catalog-task-ID contract so a fan-out instance records its own exit code instead of overwriting its siblings'.

func (*Store) SetTaskHash ¶

func (s *Store) SetTaskHash(runID, taskRef uuid.UUID, hash string) error

SetTaskHash persists a task's identity hash. taskRef follows the TaskRun-primary-key-or-catalog-task-ID contract, so a fan-out instance is addressed by its own TaskRun ID.

func (*Store) SetTaskHashWithBlob ¶

func (s *Store) SetTaskHashWithBlob(runID, taskRef uuid.UUID, hash, resolvedImageDigest string, hashInputBlob []byte) error

SetTaskHashWithBlob persists the task identity hash, the resolved image digest folded into it, and the canonical secret-redacted decomposition of the HashInput (the blob) on the same write — the existing hash write-path. The digest and blob are optional: an empty digest or a nil/empty blob leaves the corresponding column untouched (so a literal-tag, blob-less run stays consistent). The blob lets `caesium why` later diff two runs field-by-field rather than only observing that the opaque hashes differ. taskRef follows the TaskRun-primary-key-or-catalog-task-ID contract. This matters more here than almost anywhere else: with loadUniqueTaskRun, a fanned step's per-instance identity write returned ErrAmbiguousTaskRun and the hash was NEVER persisted, so `caesium why --partition`, `receipt get` and `run retry --partition` had no identity to match and the local lane published no per-partition cache entry.

func (*Store) SetTaskHashWithDigest ¶

func (s *Store) SetTaskHashWithDigest(runID, taskRef uuid.UUID, hash, resolvedImageDigest string) error

SetTaskHashWithDigest persists the task identity hash together with the resolved image digest that was folded into it. The digest may be empty when pinning is off or resolution failed; in that case only the hash is written and the existing digest column (if any) is left untouched, keeping the row consistent with the literal-tag cache key.

func (*Store) SkipTask ¶

func (s *Store) SkipTask(runID, taskID uuid.UUID, reason string) error

SkipTask skips a STEP: every caller uses it to skip a successor whose trigger rule was not satisfied, so under fan-out it means the whole group. It routes through markInstanceSkippedTx per instance, giving each its own terminal_sequence and its own task_skipped event rather than one UPDATE across the group.

func (*Store) SkipTaskInstance ¶

func (s *Store) SkipTaskInstance(runID, taskRunID uuid.UUID, reason string) error

SkipTaskInstance resolves one still-pending instance as skipped with a reason.

fanOut.failurePolicy: fail_fast needs this: on the first sibling failure the pending siblings must be resolved, not merely left undispatched. Leaving them pending would hang the run until its timeout, the same failure mode the in-group skip cascade exists to prevent. A row that is already terminal is left untouched.

func (*Store) Start ¶

func (s *Store) Start(jobID uuid.UUID, triggerID *uuid.UUID, opts ...StartOption) (*JobRun, error)

func (*Store) StartForBackfill ¶

func (s *Store) StartForBackfill(jobID, backfillID uuid.UUID, params map[string]string) (*JobRun, error)

StartForBackfill creates a JobRun pre-linked to a backfill ID. The caller should then execute the job with run.WithContext(ctx, r.ID) so the executor resumes from this pre-created record rather than creating a new one.

func (*Store) StartQueuedRun ¶

func (s *Store) StartQueuedRun(ctx context.Context, queued *models.RunQueue) (*JobRun, error)

StartQueuedRun creates a fresh JobRun from an already-claimed run_queue row. If a slot disappears between dequeue and insert, the caller should release the queue row so a later drain can retry it.

func (*Store) StartTask ¶

func (s *Store) StartTask(runID, taskRef uuid.UUID, runtimeID string) error

StartTask marks a task run as running. taskRef is resolved by loadTaskRunByIDOrUnique, so it may be either a TaskRun primary key (required for a fan-out instance, where the catalog task ID matches N sibling rows) or a catalog task ID (unfanned steps, which still have exactly one row).

func (*Store) StartTaskClaimed ¶

func (s *Store) StartTaskClaimed(runID, taskRef uuid.UUID, runtimeID, claimedBy string) error

StartTaskClaimed is the distributed-lane counterpart of StartTask. taskRef follows the same TaskRun-primary-key-or-catalog-task-ID contract, so a worker executing a fan-out instance must pass the instance's TaskRun ID.

func (*Store) StartWithContext ¶

func (s *Store) StartWithContext(ctx context.Context, jobID uuid.UUID, triggerID *uuid.UUID, opts ...StartOption) (*JobRun, error)

func (*Store) TaskExecutionDescriptor ¶

func (s *Store) TaskExecutionDescriptor(ctx context.Context, runID, taskRef uuid.UUID) (*models.TaskExecutionDescriptor, error)

TaskExecutionDescriptor returns the frozen descriptor for one TaskRun.

taskRef takes the usual TaskRun-primary-key-or-catalog-task-id contract, and callers holding a row MUST pass its primary key. The previous `(job_run_id, task_id) … Take` form silently picked an arbitrary sibling once a step could be fanned, and the one caller is the worker's quarantined-replay path — which executes the container the descriptor describes. Now that quarantined replay re-materializes fanned groups, that lookup would hand instance `bravo`'s container the descriptor recorded for `alpha`.

func (*Store) TaskLogSnapshotForInstance ¶

func (s *Store) TaskLogSnapshotForInstance(ctx context.Context, runID, taskRunID uuid.UUID) (*TaskLogSnapshot, error)

TaskLogSnapshotForInstance returns the persisted log snapshot of ONE instance, addressed by TaskRun primary key. GetTaskLogSnapshot's (run, task) predicate returns an arbitrary sibling's log for a fanned group; this one cannot.

A row with no captured log yields (nil, nil), matching GetTaskLogSnapshot.

func (*Store) TaskQuarantine ¶

func (s *Store) TaskQuarantine(ctx context.Context, runID, taskID uuid.UUID) (bool, error)

func (*Store) TaskRunInstances ¶

func (s *Store) TaskRunInstances(ctx context.Context, runID, taskID uuid.UUID) ([]*TaskRun, error)

TaskRunInstances returns every TaskRun row for (runID, taskID) in stable partition order (partition_index, then created_at, then id). An unfanned task yields exactly one element; a fanned group yields N, including instances that were skipped for an in-group dependency failure.

Unlike the collapsed run-detail payload (collapseFanOutGroups), the returned rows keep their own TaskRun primary key in ID — the caller can address a single instance afterwards.

func (*Store) TaskRunsForTask ¶

func (s *Store) TaskRunsForTask(runID, taskID uuid.UUID) ([]models.TaskRun, error)

TaskRunsForTask returns every TaskRun instance of (runID, taskID) ordered by partition_index. It is the group-aware read every caller that used to arbitrarily `.First()` a sibling must use instead: an unfanned task yields exactly one row (identical to the old behavior), a fanned task yields all N so the caller can choose the instance it means (e.g. the failed one) or summarize the group. Returns an empty slice, never an error, when the task has no rows.

func (*Store) TerminalTaskRunsSince ¶

func (s *Store) TerminalTaskRunsSince(runID uuid.UUID, afterSeq int64) ([]models.TaskRun, error)

TerminalTaskRunsSince returns the run's terminal task_runs rows with a terminal_sequence strictly greater than afterSeq, ordered by terminal_sequence ascending — the "post-checkpoint tail" a recovering owner replays on top of the latest checkpoint. Uses the (job_run_id, terminal_sequence) index.

func (*Store) UpdateTaskExecutionDescriptorInputs ¶

func (s *Store) UpdateTaskExecutionDescriptorInputs(runID, taskID uuid.UUID, predecessorOutputs map[uuid.UUID]map[string]string, predecessorHashes map[uuid.UUID]string, computedHash, resolvedImageDigest string, hashInputBlob []byte) error

func (*Store) UpdateTaskExecutionDescriptorSecretRefs ¶

func (s *Store) UpdateTaskExecutionDescriptorSecretRefs(runID, taskID uuid.UUID, refs []models.TaskExecutionSecretRef) error

func (*Store) WhyTask ¶

func (s *Store) WhyTask(ctx context.Context, runID uuid.UUID, taskRef string) (*WhyExplanation, error)

WhyTask explains why the task identified by taskRef (a task UUID or a task name) in run runID executed / hit cache / re-ran. taskRef is matched first as a UUID against task_id, then as a task name within the run's job.

On a FANNED step it returns the group summary (see WhyTaskPartition for the per-instance answer).

func (*Store) WhyTaskPartition ¶

func (s *Store) WhyTaskPartition(ctx context.Context, runID uuid.UUID, taskRef, partition string) (*WhyExplanation, error)

WhyTaskPartition is WhyTask with an explicit fan-out instance selector.

• partition == "": an unfanned task is explained exactly as before; a fanned step returns the aggregate WhyGroup summary rather than an arbitrary sibling's explanation (the pre-fan-out code path `.First()`-ed one row of N, so the answer depended on database row order).
• partition != "": the named instance is explained with the full single-instance diff. ErrPartitionNotFound (listing the available values) when the group has no such partition.

func (*Store) WithLeaseStore ¶

func (s *Store) WithLeaseStore(ls *LeaseStore) *Store

WithLeaseStore enables run-owner lease writing. Call this from startup code when CAESIUM_RUN_OWNER_ENABLED=true.

func (*Store) WriteCheckpoint ¶

func (s *Store) WriteCheckpoint(runID uuid.UUID, sequenceHigh, ownerGeneration int64, blob []byte, incremental bool) error

WriteCheckpoint persists a run_checkpoints row for runID. sequenceHigh is the highest terminal_sequence the snapshot covers; ownerGeneration fences the write against a stale owner; blob is the serialized state (RunState.Snapshot); incremental marks a delta vs a full snapshot. Re-writing the same (run_id, sequence_high) overwrites — checkpoints are idempotent at a sequence.