engine

package
v0.0.8 Latest Latest
Warning

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

 Go to latest Published: Aug 29, 2026 License: MIT Imports: 23 Imported by: 0

Documentation

Overview

Package engine is the durable-execution adapter for the pasture epoch lifecycle. It owns the shared modernc SQLite handle, registers and drives the pure-Go EpochStateMachine over durable steps, persists an EpochState projection each transition, and records forensic rows exactly once.

The state machine itself lives in pkg/protocol and has no substrate dependency; this package is the impure adapter around it.

Package engine — queue.go defines the DBOS queue for concurrency-limited slice and review sub-workflow dispatch.

Sub-workflows that drive individual implementation slices and review cycles are dispatched through a shared DBOS queue with a configurable per-executor concurrency limit K. Bounded concurrency is the primary control point for the single-writer WAL bottleneck: 30+ unbounded sub-workflows would thrash the shared SQLite connection, so K is tuned to the write throughput of the pasture.db file.

Queue lifecycle:

  • newSliceQueue registers a database-backed queue row and may be called before or after dbos.Launch. Engine.New calls it during construction, so the queue exists before the first enqueue.
  • Sub-workflows are enqueued via Engine.EnqueueSlice / Engine.EnqueueReview, each of which calls dbos.RunWorkflow with dbos.WithQueue on the registered slice queue.
  • DBOS dequeues and starts sub-workflows up to K at a time; excess are held in the queues table until a running sub-workflow completes and frees a slot.
  • K can be changed on a running engine through Engine.SetSliceConcurrency, which rewrites the stored queue settings that every worker reads.

Crash recovery and the queue:

Recovery never restarts a workflow in place — it puts the workflow back on a queue (dbos/recovery.go). Which queue decides which limits apply to it:

  • A workflow returns to the queue it ran on. Slice and review work goes back onto the slice queue, so the concurrency limit K and the configured polling cadence still govern it. An epoch control workflow goes back onto the control queue, so its limit of one per process still governs it: production only ever starts one by enqueueing it there (see dbosController.StartEpoch in internal/handlers/controller.go), so it always has a queue of its own to return to.
  • Work that ran on NO queue goes onto the runtime's own reserved queue instead. That queue is not pasture's to configure: it carries no concurrency limit, and it polls at the runtime's fixed one-second cadence rather than the interval pasture sets for its own queues. Nothing in production reaches this case; it is reached by a caller that starts a workflow directly with RunWorkflow, which today is test code only.

All three shapes are pinned in internal/engine/subworkflows_test.go:

  • TestSliceQueue_RecoveryKeepsEachWorkflowOnItsOwnQueue: a slice returns to the slice queue.
  • TestRecovery_EpochControlWorkflowReturnsToItsOwnQueue: the production shape — the control workflow is enqueued as StartEpoch enqueues it, and recovery returns it to the control queue.
  • TestRecovery_OffQueueEpochWorkflowLandsOnTheReservedQueue: a workflow started off any queue lands on the runtime's reserved queue.

Index

Constants

View Source 
const (

	// EngineAgentName is the stable software-agent name the engine attributes
	// its phase-transition activities to. It is exported so callers that record
	// audit events attributed to the engine (e.g. the terminate handler) can
	// use the same stable name without duplicating the string. It is deliberately
	// NOT one of the well-known automaton agents, so adding it does not change
	// the well-known agent count or its registration tests.
	EngineAgentName = "pasture/automaton/epoch-engine"

	// ActivityKindPhaseTransition is the discriminator the engine passes to
	// protocol.DedupKey for a phase-transition activity. It is DELIBERATELY
	// distinct from the audit tier's event_type ("PhaseTransition"): an activity
	// is a different PROV-O entity (a unit of work owned by the engine's software
	// agent) than a system audit event, so they occupy independent id-spaces.
	// Both tiers use the SAME derivation mechanism (this one DedupKey encoder),
	// but the distinct kind makes the activity id differ from the audit dedup_key
	// for the same transition — id-equality across tiers would be a fragile
	// implicit join. Exactly-once still holds: each table is keyed on its own
	// kind. Exported so a cross-tier replay test derives the identical activity
	// id from the same const.
	ActivityKindPhaseTransition = "activity:phase-transition"
)
View Source 
const ControlQueueName = "pasture-control-queue"

ControlQueueName is the canonical DBOS queue name for epoch control workflows. CLI lifecycle commands enqueue onto this queue through a DBOS client; pastured hosts the registered workflow and dequeues it.

View Source 
const DefaultAppName = "pasture"

DefaultAppName is the pinned DBOS application name.

View Source 
const DefaultApplicationVersion = "1"

DefaultApplicationVersion is the pinned DBOS recovery COHORT marker.

Role: DBOS filters crash-recovery by (ExecutorID, ApplicationVersion) and otherwise defaults the version to a per-build binary hash. Pinning a stable, build-independent value here is precisely what lets a REBUILT binary still recover epochs that an earlier build left in flight — without it, every rebuild would start a new cohort and silently orphan the previous build's in-flight epochs.

Bump criteria: increment this ONLY on an incompatible change to the EpochWorkflow / EpochControlWorkflow shape that makes already-in-flight workflows non-resumable. A routine rebuild MUST NOT bump it (that would abandon in-flight epochs); after a deliberate bump, old in-flight workflows are resumed manually rather than auto-recovered.

Cross-binary invariant: every pasture process that opens the engine — the local CLI (epoch start) and the daemon — MUST pin this same value (together with DefaultExecutorID and DefaultAppName). If the CLI and the daemon use different values, each silently fails to recover the other's in-flight epochs.

View Source 
const DefaultExecutorID = "pasture"

DefaultExecutorID is the pinned DBOS executor id. DBOS filters crash-recovery by ExecutorID + ApplicationVersion; pinning the executor id keeps recovery attributable to "pasture" across restarts rather than a per-process default.

View Source 
const DefaultSliceQueueConcurrency = 8

DefaultSliceQueueConcurrency is the default per-executor concurrency limit for the slice queue. The value balances SQLite WAL write throughput against parallel-agent utilisation:

  • SQLite WAL serialises writers: only one write transaction commits at a time. Under 30+ unbounded writers the commit queue grows faster than it drains and busy_timeout errors accumulate.
  • K=8 allows up to 8 sub-workflows to hold a write transaction concurrently. This is a conservative default chosen to stay within the WAL commit throughput of a typical single-disk host; the right value depends on your storage — lower K (e.g. 4) on HDD or network-attached storage, higher K (e.g. 16) on NVMe-backed hosts with idle I/O headroom.
  • A benchmark validating a specific K for your setup is the authoritative guide; measure with your actual storage before changing this default.

Override via --slice-concurrency / PASTURE_SLICE_CONCURRENCY or the engine Config.SliceConcurrency field.

View Source 
const EpochControlWorkflowName = "pasture.epoch_control.v1"

EpochControlWorkflowName is the stable DBOS workflow name used by clients that enqueue epoch control work without linking to the engine implementation.

View Source 
const SequentialShutdownWaits = 7

SequentialShutdownWaits is how many waits one shutdown of THIS engine can spend the per-component budget on, one after another: the schedule reconciler, the queue runner, the work scheduler, in-flight work, and the database's notification listener, notifier and connection pool. The engine configures neither an administrative listener nor a remote-control connection, which would add two more.

Measured on the pinned runtime: a shutdown held up by work that is still running spends the budget ONCE (only the in-flight-work wait expires; every other wait returns at once). Seven is therefore the worst case an operator must budget for, not the normal cost.

View Source 
const SliceConcurrencyEnv = "PASTURE_SLICE_CONCURRENCY"

SliceConcurrencyEnv is the environment variable that overrides the per-executor concurrency limit for the slice queue. When set, its integer value is used instead of DefaultSliceQueueConcurrency.

View Source 
const SliceQueueName = "pasture-slice-queue"

SliceQueueName is the canonical DBOS queue name for slice and review sub-workflow dispatch. A single shared queue keeps the concurrency budget unified across both sub-workflow kinds — slices and reviews compete for the same K slots, so the total in-flight count across both is bounded by K.

Variables

This section is empty.

Functions

func DescribeDurableStartupFailure added in v0.0.8 

func DescribeDurableStartupFailure(where string, err error) error

DescribeDurableStartupFailure returns an actionable replacement for a durable-runtime start-up failure whose cause pasture can name, and returns err unchanged when it cannot name one. It returns nil for a nil error.

where is the caller's own location, used for the Where line: the engine and the epoch controller both build a durable runtime over the same shared handle, and an operator needs to know which of them refused to start.

Callers keep their own wrapping for the errors this function passes through.

func ReadProjection 

func ReadProjection(db *sql.DB, epochId string) (*protocol.EpochState, error)

ReadProjection returns the projected EpochState for epochId. It returns (nil, nil) when no projection exists yet (the epoch has not advanced), so callers can distinguish "unknown epoch" from a read error.

Callers that need available transitions recompute them from the returned state via protocol.NewEpochStateMachineFromState(...).AvailableTransitions(); the projection stores raw state, not derived views.

func RequireSupportedDurableSchema added in v0.0.8 

func RequireSupportedDurableSchema(ctx context.Context, where string, db *sql.DB, dbPath string) error

RequireSupportedDurableSchema refuses a pasture database whose durable schema a superseded runtime wrote, and reports every other preflight failure in the same actionable shape.

Call it on the exact *sql.DB that is about to become the durable runtime's system handle, before dbos.NewContext or dbos.NewClient. dbPath names the file for the operator. where is the caller's own location, because the engine and the epoch controller open the same file and an operator needs to know which of them refused to start.

The gate only reads. On refusal nothing was opened, created, or migrated, and the file is byte-for-byte as it was.

func ResolveSliceConcurrency 

func ResolveSliceConcurrency(flagVal int) (int, error)

ResolveSliceConcurrency resolves the effective per-executor concurrency limit K from the three override sources, highest-priority first:

  1. flagVal > 0: the caller-supplied CLI flag value (--slice-concurrency).
  2. $PASTURE_SLICE_CONCURRENCY env var (non-empty, parses as a positive int).
  3. DefaultSliceQueueConcurrency (8).

If the env var is set but not a valid positive integer, the function returns an actionable validation error (the caller should surface it and exit 1). A zero or negative flagVal is treated as "not set" (fall through to env/default).

This function is the single resolution rule shared by pastured and any other process that constructs an Engine; call it once at startup and pass the result to engine.Config.SliceConcurrency.

func WorstCaseShutdownDuration added in v0.0.8 

func WorstCaseShutdownDuration(perComponentTimeout time.Duration) time.Duration

WorstCaseShutdownDuration reports how long a Shutdown can take in the worst case for a given per-component budget. Callers that must fit inside an external stop deadline (a service manager, a container runtime) size the budget with this rather than with the raw value.

func WriteProjection 

func WriteProjection(ctx context.Context, db *sql.DB, state *protocol.EpochState, nowUnixNano int64) error

WriteProjection upserts the serialized EpochState for state.EpochId. It is idempotent (last-write-wins) — safe to re-run when a durable step replays, because the projection is a cache of the authoritative FSM state, not an append-only log. nowUnixNano timestamps the row's freshness.

Types

type ActivitySink 

type ActivitySink interface {
	// RegisterSoftwareAgent find-or-creates is the caller's concern; the engine
	// only registers its own stable agent once if absent.
	RegisterSoftwareAgent(namespace, name, version, source string) (provenance.SoftwareAgent, error)
	// StartActivityWithID records an activity under a caller-supplied id with
	// ON CONFLICT(id) DO NOTHING, so a replayed emission collapses to one row.
	StartActivityWithID(id provenance.ActivityID, agentID provenance.AgentID, phase provenance.Phase, stage provenance.Stage, notes string) (provenance.Activity, error)
}

ActivitySink is the narrow provenance surface the engine needs to record activities idempotently. protocol.TaskTracker satisfies it (via the embedded provenance.Tracker), as does provenance.Tracker directly.

type AdvanceStep 

type AdvanceStep struct {
	// ToPhase is the target phase for this transition.
	ToPhase protocol.PhaseId
	// TriggeredBy identifies who/what drove the transition (recorded as the
	// forensic row's role; defaults to the epoch role when empty).
	TriggeredBy string
	// ConditionMet describes the satisfied transition condition.
	ConditionMet string
	// Votes are recorded (in order) before the advance, to satisfy the
	// consensus gate at p4/p10.
	Votes []protocol.ReviewVoteSignal
	// BlockerDelta adjusts the blocker count before the advance: a positive
	// value records that many new blockers, a negative value resolves that
	// many. Used to exercise the p10 blocker gate.
	BlockerDelta int
}

AdvanceStep is one scripted transition in an epoch plan. It carries the votes and blocker delta to apply (deterministically, before the advance) so a single plan can exercise the consensus and blocker gates without an external signal source — the signal-driven control surface is a later slice.

type Config 

type Config struct {
	// DBPath is the unified pasture.db path. Required.
	DBPath string
	// ApplicationVersion is the pinned DBOS application version. REQUIRED:
	// DBOS recovery is filtered by it, and it defaults to a binary hash, so a
	// rebuilt binary would skip recovery of an in-flight epoch unless this is
	// pinned to a stable value across builds. New rejects an empty value.
	ApplicationVersion string
	// ExecutorID overrides DefaultExecutorID. Pinned across restarts.
	ExecutorID string
	// AppName overrides DefaultAppName.
	AppName string
	// Trail is the forensic sink for one audit row per transition. When nil,
	// New opens an owned SQLite trail on DBPath (also migrating the file to the
	// current schema, which creates the dedup_key column).
	Trail audit.Trail
	// SkipMigrations opens DBPath as a pre-migrated database when Trail is nil.
	// The audit layer still asserts the schema version. This is intended for
	// tests that copy a current golden database; production callers should leave
	// it false so the real migrator runs.
	SkipMigrations bool
	// Specs overrides the canonical phase transition table (for tests). nil →
	// protocol.PhaseSpecs.
	Specs map[protocol.PhaseId]protocol.PhaseSpec
	// Logger is the DBOS logger. nil → slog.Default().
	Logger *slog.Logger
	// OnTransition, when set, runs INSIDE the durable step for each successful
	// transition, AFTER the projection + forensic audit row are written and
	// BEFORE the step returns. It is the step-bracketing seam: idempotent
	// activity recording wires here (it shares the step's replay semantics, so
	// any external write it makes must be idempotent — e.g. a deterministic-id
	// ON CONFLICT insert). Returning an error fails the step (and so the
	// transition's durable commit).
	//
	// stepSeq is the deterministic per-transition step sequence (the same value
	// the audit dedup key is derived from). It is threaded in from the workflow
	// body because it cannot be recovered inside the hook: DBOS exposes it only
	// in the workflow body, and a replay re-runs only the crashed step, so a
	// hook-local counter would not be replay-stable. Hooks derive their own
	// deterministic keys from it via protocol.DedupKey.
	OnTransition func(ctx context.Context, epochId string, rec *protocol.TransitionRecord, stepSeq string) error
	// Tracker, when set, makes the engine record one PROV-O activity per
	// transition with a deterministic id (exactly-once across replay). nil ⇒
	// activities are not recorded and the engine behaves as it did without this
	// field. The engine resolves a stable software-agent id at New() so the
	// deterministic insert always references a present agent row.
	Tracker ActivitySink
	// SliceConcurrency is the per-executor concurrency limit K for the slice
	// queue. It bounds the number of slice and review sub-workflows that the
	// local executor runs concurrently, providing backpressure on the single
	// SQLite WAL writer bottleneck. <= 0 uses DefaultSliceQueueConcurrency.
	//
	// See DefaultSliceQueueConcurrency in internal/engine/queue.go for the
	// full trade-off rationale and tuning guidance.
	SliceConcurrency int
	// QueueBasePollingInterval overrides the DBOS queue base polling interval.
	// Zero keeps the DBOS production default. Tests may set a shorter interval
	// to keep bounded-concurrency assertions fast without changing production
	// queue cadence.
	QueueBasePollingInterval time.Duration
	// HooksMgr, when set, receives slice lifecycle events (SliceStarted,
	// SliceCompleted, SliceFailed) dispatched by slice sub-workflows. nil ⇒
	// hook dispatch is skipped (no observability events; the sub-workflow still
	// runs correctly).
	//
	// pastured wires HooksMgr when it hosts the engine. Callers that don't need
	// slice lifecycle observability (e.g. the local CLI, unit tests) may leave
	// this nil.
	HooksMgr *hooks.Manager
	Timeouts timeouts.Profile
}

Config configures an Engine.

type ControlInput 

type ControlInput struct {
	EpochId string
}

ControlInput is the EpochControlWorkflow input: the epoch id whose lifecycle this durable workflow drives. The workflow ID is set to the epoch ID by the caller, so senders address signals to the epoch by its own id.

type Engine 

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

Engine owns the shared modernc handle, the DBOS context, and the forensic trail. It registers and drives the EpochStateMachine over durable steps.

Lifecycle: New → Launch → (run workflows) → Shutdown.

func New 

func New(ctx context.Context, cfg Config) (*Engine, error)

New constructs an Engine: opens the shared handle with the WAL/busy-timeout DSN, ensures the projection table, opens (or adopts) the forensic trail, creates the DBOS context with the shared handle as SQLiteSystemDB and the pinned ExecutorID + ApplicationVersion, and registers EpochWorkflow.

The returned Engine is NOT yet launched; call Launch to run the recovery sweep and accept work. Always call Shutdown to release handles.

func (*Engine) ControlQueue 

func (e *Engine) ControlQueue() dbos.Queue

ControlQueue returns the DBOS queue used for epoch control workflows.

func (*Engine) DB 

func (e *Engine) DB() *sql.DB

DB returns the shared modernc handle (projection + DBOS tables live here).

func (*Engine) DBOS 

func (e *Engine) DBOS() dbos.Context

DBOS returns the underlying DBOS context so callers (and later slices) can RunWorkflow / Send / ListWorkflows against the engine's registered workflow.

func (*Engine) EnqueueReview 

func (e *Engine) EnqueueReview(in ReviewInput) (dbos.WorkflowHandle[ReviewResult], error)

EnqueueReview dispatches a ReviewSubWorkflow via the slice queue. The workflow id is derived from the epoch id, phase id, and review round so that each round of a review cycle runs a fresh sub-workflow rather than returning the memoized result of a prior round.

The caller should retain the returned handle to send submit_vote signals and to wait for the result. To compute the same workflow id for vote delivery, call protocol.ReviewWorkflowID(epochId, phaseId, round).

func (*Engine) EnqueueSlice 

func (e *Engine) EnqueueSlice(in SliceInput) (dbos.WorkflowHandle[SliceResult], error)

EnqueueSlice dispatches a SliceSubWorkflow via the slice queue, giving it the supplied workflow id (sliceId) so start_slice / complete_slice signals can address it. The caller supplies the epoch context needed for hook dispatch and parent progress signalling. The returned handle is live; callers may call GetResult to wait for the slice to complete.

func (*Engine) EpochControlWorkflow 

func (e *Engine) EpochControlWorkflow(ctx dbos.Context, in ControlInput) (protocol.EpochState, error)

EpochControlWorkflow is the signal-driven durable driver for one epoch.

Unlike the scripted EpochWorkflow (which replays a fixed plan), this workflow advances only in response to durable signals delivered by topic:

  • advance_phase drives one FSM transition.
  • submit_vote records a phase-scoped review vote (consumed before the gated advance that needs it).
  • register_session registers a session (idempotent by session id).
  • slice_progress appends a slice-progress event.

The slice-level start_slice / complete_slice topics are consumed by the slice sub-workflows, not this epoch loop.

Each loop blocks for the next advance_phase signal. When one arrives it first drains the three side-channel topics (non-blocking) — so any votes sent just before the advance are recorded and the consensus gate sees them — then applies the advance. On an idle timeout it drains the side channels anyway so sessions and slice progress reported between advances reach the projection. Every successful transition funnels through commitTransition, so the projection, the exactly-once forensic emit, and the activity hook are identical to the scripted driver. The loop ends when the FSM reaches the terminal phase.

func (*Engine) EpochWorkflow 

func (e *Engine) EpochWorkflow(ctx dbos.Context, in EpochInput) (protocol.EpochState, error)

EpochWorkflow is the durable workflow that drives the 12-phase epoch.

For each planned transition it (1) records votes and the blocker delta and runs EpochStateMachine.Advance in the workflow BODY — pure, deterministic, so the phase sequence replays identically — then (2) performs the I/O in ONE durable step: persist the EpochState projection and record exactly one forensic row keyed by the deterministic dedup key. One step per transition means one forensic emission per (kind, step), preserving the dedup invariant.

A failed advance (gate violation) is recorded as a failed transition and the plan continues; the durable step is skipped for that entry.

func (*Engine) Launch 

func (e *Engine) Launch() error

Launch runs the DBOS recovery sweep (resuming any in-flight epochs) and makes the engine ready to run new workflows. Call exactly once after New.

func (*Engine) ReadProjection 

func (e *Engine) ReadProjection(epochId string) (*protocol.EpochState, error)

ReadProjection returns the projected EpochState for epochId, or (nil, nil) if the epoch has not advanced yet. This is the read side of the projection that query and status surfaces consume.

func (*Engine) ReviewSubWorkflow 

func (e *Engine) ReviewSubWorkflow(ctx dbos.Context, in ReviewInput) (ReviewResult, error)

ReviewSubWorkflow is the DBOS sub-workflow for a single P4/P10 review phase.

Lifecycle:

  1. Dispatched via Engine.EnqueueReview to the slice queue; starts when a queue slot is free (bounded by K).
  2. Receives submit_vote signals (ReviewVoteSignal) via a polling Recv loop until all three ReviewAxis members have voted.
  3. Returns a ReviewResult with the collected per-axis vote map.

The submit_vote signals are addressed to this sub-workflow by the id assigned by Engine.EnqueueReview (protocol.ReviewWorkflowID(epochId, phaseId, round)).

Idempotency: if the same axis votes twice, the later vote overwrites the earlier one (last-writer-wins per ReviewAxis key).

func (*Engine) SetSliceConcurrency added in v0.0.8 

func (e *Engine) SetSliceConcurrency(k int) (int, error)

SetSliceConcurrency changes the per-executor concurrency limit K on the slice queue of a running engine, then reads the stored configuration back and returns the value that is actually in force.

The queue configuration is a row in the pasture database, not process-private state, so the change is durable and every process that serves the queue adopts it without a restart: each worker reloads the row on its next poll iteration and the queue supervisor republishes the set once a second (dbos/queue.go, queueRunner.run and queueRunner.runQueue). Work already running is not interrupted; the new limit governs the next dequeue.

The read-back is part of the operation, not a debugging aid: the write and the read are separate database round-trips, so another process can change the same row in between. The returned value is what the database holds after the change, and a disagreement with k is reported as an error that names both values.

This is the engine-side operator path. A command-line surface that calls it from a separate process — so an operator can change the limit of a daemon they are not inside — is still to be added, and is tracked with the rest of the queue behaviour at https://github.com/dayvidpham/pasture/issues/104.

k must be positive. Removing the limit entirely (unbounded dequeue) is deliberately not offered: the limit is the backpressure control for the single SQLite writer, and the trade-off is described on DefaultSliceQueueConcurrency.

func (*Engine) Shutdown 

func (e *Engine) Shutdown(perComponentTimeout time.Duration) error

Shutdown stops the durable runtime, then releases the handles the runtime does not own. Call it once.

perComponentTimeout is the budget for EACH wait inside the runtime, NOT a total for the whole shutdown. The runtime stops its parts one after another and gives every one of them the full value, so the worst case is SequentialShutdownWaits times this argument; use WorstCaseShutdownDuration to size it against an external stop deadline. Measured on the pinned runtime, a shutdown held up by work that is still running spends the budget once, because every other wait returns at once.

It returns nil when the runtime stopped inside its budget. Otherwise it returns an actionable error carrying a *ShutdownIncompleteError with the parts that were still running; the handles are released either way. The error is ALSO logged, because many callers (tests, probes, deferred cleanup) discard the return value and the failure must never go unreported.

The durable runtime owns the shared SQLite handle it was constructed with: its shutdown closes that handle unconditionally, on the timeout path too (dbos/internal/sysdb/dbq.go, sqlPoolAdapter.Close). The engine therefore cannot hold the handle open for a worker that outlives the timeout, and the close below is a harmless second close of an already-closed handle. internal/handlers/controller.go records the same ownership for the client; the two must stay in agreement.

The trail is closed either way: it is a separate handle that the runtime never writes through.

func (*Engine) SliceConcurrency 

func (e *Engine) SliceConcurrency() int

SliceConcurrency returns the per-executor concurrency limit K for the slice queue as this process last saw it. New stores the resolved start-up value (after applying the DefaultSliceQueueConcurrency fallback) rather than re-deriving it from the config, so the fallback logic exists in one place only.

SetSliceConcurrency then replaces the value with the one it READ BACK from storage, not the one it was asked for. That is deliberate: when another process wins the same row, the read-back value is the limit the queue really runs work at, and reporting the losing request instead would be a lie.

The value can still be out of date. This process reads the stored row when it starts and when it changes the limit, while any process may change that row at any moment. For the limit in force right now, read the row: dbos.RetrieveQueue on Engine.DBOS(), or from a terminal 

pasture queue concurrency get slice

func (*Engine) SliceQueue 

func (e *Engine) SliceQueue() dbos.Queue

SliceQueue returns the DBOS queue used for slice and review sub-workflow dispatch. Tests may inspect the queue name to verify wiring.

func (*Engine) SliceSubWorkflow 

func (e *Engine) SliceSubWorkflow(ctx dbos.Context, in SliceInput) (SliceResult, error)

SliceSubWorkflow is the DBOS sub-workflow for a single implementation slice.

Lifecycle:

  1. Dispatched via Engine.EnqueueSlice to the slice queue; starts when a queue slot is free (bounded by the configured concurrency limit K).
  2. Receives a start_slice signal (SliceStartSignal) via dbos.Recv before deciding the execution mode. If no signal arrives within the deadline the sub-workflow records an honest failure (Success=false) and returns; no completion hook fires and the parent projection receives Completed=false.
  3. Executes the slice in the chosen mode (mock / tmux / subprocess) inside a durable step.
  4. Receives an optional complete_slice signal (SliceCompleteSignal) that overrides the computed outcome.
  5. Dispatches hook events (SliceStarted / SliceCompleted / SliceFailed) through the engine's hook manager inside durable steps (memoized; not re-fired on crash recovery).
  6. Sends a slice_progress signal to the parent epoch workflow.

The start_slice and complete_slice signals are addressed to the sub-workflow by its sliceId (which is its DBOS workflow id).

func (*Engine) Timeouts added in v0.0.8 

func (e *Engine) Timeouts() timeouts.Profile

Timeouts returns the timeout profile in force for this engine. It is the profile Config.Timeouts asked for, or the production profile when the caller left that field zero. A caller that must wait for the engine reads its own ceiling from here instead of writing a second, separate number that can disagree with the engine.

It governs every wait the engine OWNS: the SQLite lock wait on the shared handle, the start_slice deadline in a slice sub-workflow, and the SQLite lock wait on the audit trail WHEN THE ENGINE OPENED THAT TRAIL ITSELF.

It does NOT reach a trail supplied through Config.Trail. Opening a trail also fixes its lock budget, so a caller that injects a trail has already chosen that budget and must open the trail with the same profile it passes here. cmd/pastured does exactly that, from one call site.

func (*Engine) Trail 

func (e *Engine) Trail() audit.Trail

Trail returns the forensic trail the engine records transitions into.

type EpochInput 

type EpochInput struct {
	EpochId  string
	Advances []AdvanceStep
}

EpochInput is the EpochWorkflow input: the epoch id and the ordered plan of transitions to drive.

type ReviewInput 

type ReviewInput struct {
	// EpochId is the parent epoch this review belongs to.
	EpochId string `json:"epochId"`
	// PhaseId identifies which review phase this is (e.g. "review" or "code-review").
	PhaseId string `json:"phaseId"`
	// Round is the review-cycle counter for this (epochId, phaseId) pair. It
	// starts at 1 and increments each time a review returns REVISE and the
	// protocol re-enters the review phase. Supplying the round ensures each
	// re-review runs a fresh DBOS sub-workflow (a different id) rather than
	// returning the memoized result of a prior round.
	//
	// The round value MUST come from a deterministic, replay-stable counter
	// tracked in workflow state, NOT from wall-clock time or a random value.
	// Default 0 is treated as round 1 by EnqueueReview for backwards
	// compatibility (existing callers that don't set Round still get the
	// correct first-round workflow id).
	Round int `json:"round,omitempty"`
}

ReviewInput is the input to a review sub-workflow.

type ReviewResult 

type ReviewResult struct {
	// PhaseId echoes the input for correlation on the parent side.
	PhaseId string `json:"phaseId"`
	// Success is true when all review axes received an ACCEPT vote.
	Success bool `json:"success"`
	// VoteResult is the per-axis vote map collected by the sub-workflow.
	VoteResult map[protocol.ReviewAxis]protocol.VoteType `json:"voteResult"`
}

ReviewResult is the output of a review sub-workflow.

type ShutdownComponent added in v0.0.8 

type ShutdownComponent string

ShutdownComponent names one part of the durable runtime that a shutdown waits for. The runtime stops its parts one after another and reports the ones that were still running when their wait expired, so an operator can tell "work was still in flight" apart from "a background loop would not stop".

The values are the names the runtime itself reports. They are typed so a caller matches on a constant instead of re-spelling a string. A runtime build that adds a part reports a name this list does not carry: such a name is passed through unchanged rather than dropped, and Meaning reports it as unrecognised. 

const (
	ShutdownComponentScheduleReconciler   ShutdownComponent = "schedule reconciler"
	ShutdownComponentQueueRunner          ShutdownComponent = "queue runner"
	ShutdownComponentWorkflowScheduler    ShutdownComponent = "workflow scheduler"
	ShutdownComponentAdminServer          ShutdownComponent = "admin server"
	ShutdownComponentWorkflows            ShutdownComponent = "workflows"
	ShutdownComponentConductor            ShutdownComponent = "conductor"
	ShutdownComponentNotificationListener ShutdownComponent = "system database notification listener"
	ShutdownComponentNotifier             ShutdownComponent = "system database notifier"
	ShutdownComponentConnectionPool       ShutdownComponent = "system database connection pool"
)

The parts a shutdown waits for, in the order the runtime waits for them. Two of them can never appear for this engine, and say so in Meaning: the engine configures neither an administrative listener nor a remote-control connection.

func (ShutdownComponent) Meaning added in v0.0.8 

func (c ShutdownComponent) Meaning() string

Meaning returns one plain sentence describing the part, for an operator who does not read this code. An unrecognised name (a newer runtime build) is reported as such instead of being hidden.

type ShutdownIncompleteError added in v0.0.8 

type ShutdownIncompleteError struct {
	// PerComponentTimeout is the budget each wait was given — not the total
	// the shutdown was allowed to take. See Engine.Shutdown.
	PerComponentTimeout time.Duration
	// Pending lists the parts still running when their wait expired, in the
	// order the runtime waited for them. Empty when the runtime reported a
	// failure in a shape this build could not read; Cause is then the only
	// account of it.
	Pending []ShutdownComponent
	// Cause is the runtime's own error, kept verbatim for diagnosis.
	Cause error
}

ShutdownIncompleteError reports a durable shutdown that ran out of time.

It is the typed detail behind the error Engine.Shutdown returns; reach it with errors.As when the caller must act on WHICH parts were still running (an operator report, a metric) rather than only on the fact of the failure.

func (*ShutdownIncompleteError) Error added in v0.0.8 

func (e *ShutdownIncompleteError) Error() string

Error reports which parts were still running and what each of them is.

func (*ShutdownIncompleteError) Unwrap added in v0.0.8 

func (e *ShutdownIncompleteError) Unwrap() error

Unwrap exposes the runtime's own error so errors.Is and errors.As reach it.

type SliceInput 

type SliceInput struct {
	// EpochId is the parent epoch this slice belongs to. Used for hook dispatch
	// and the parent progress signal.
	EpochId string `json:"epochId"`
	// SliceId is the unique identifier for this slice. It doubles as the
	// sub-workflow's id so start_slice / complete_slice signals address it.
	SliceId string `json:"sliceId"`
	// ParentWorkflowId is the id of the epoch control workflow that dispatched
	// this slice. When non-empty, the sub-workflow delivers a slice_progress
	// signal to it on completion.
	ParentWorkflowId string `json:"parentWorkflowId"`
}

SliceInput is the input to a slice sub-workflow.

type SliceResult 

type SliceResult struct {
	// SliceId echoes the input for correlation on the parent side.
	SliceId string `json:"sliceId"`
	// Success is true when the slice completed without error.
	Success bool `json:"success"`
	// Output holds a human-readable success message (non-empty when Success is true).
	Output string `json:"output,omitempty"`
	// Error holds the failure reason (non-empty when Success is false).
	Error *string `json:"error,omitempty"`
}

SliceResult is the output of a slice sub-workflow.

Source Files

View all Source files

Directories

Path Synopsis
Package enginetest builds durable-engine fixtures for tests in other packages.
 Package enginetest builds durable-engine fixtures for tests in other packages.

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.