sqlstore

package
v0.2.0 Latest Latest
Warning

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

 Go to latest Published: Oct 16, 2026 License: MIT, MIT Imports: 21 Imported by: 0

Documentation

Overview

Package sqlstore is a SQLite-backed implementation of go-workflow's runtime.StateStore and every companion runtime store interface (waits, memoization, recovery, compensation, scheduler resources, and the rest; see assertions.go for the full list).

It is stdlib database/sql plus go-workflow only: the caller opens the database with the SQLite driver of its choice, applies the schema with Migrate (or applies Schema through its own migration runner), and hands the *sql.DB to New. The package never imports a driver outside its tests. 

db, _ := sql.Open("sqlite", "workflow.db") // any database/sql SQLite driver
db.SetMaxOpenConns(1)
_ = sqlstore.Migrate(ctx, db)
store, _ := sqlstore.New(db)
var _ runtime.StateStore = store

The SQL is SQLite dialect (BEGIN IMMEDIATE, json_extract, ? placeholders). A different database is a second implementation qualified by the same go-workflow conformance suite, not a dialect switch here.

Hosts that keep their own tables in the same database can run statements in the store's write transaction with Store.WriteTx and the DBTX it supplies.

Index

Constants

This section is empty.

Variables

View Source 
var DefaultRunColumns = RunColumns{
	Table: "workflow_runs", ID: "run_id", Status: "status",
	Generation: "generation", CreatedAt: "created_at",
}

DefaultRunColumns is the RunColumns the store uses when WithRunColumns is not given.

Functions

func Migrate 

func Migrate(ctx context.Context, db *sql.DB) error

Migrate creates the store's tables, indexes and triggers when they are missing and records the applied schema version in its own table. It is safe to call repeatedly and on a database that already contains the tables. It does not detect or repair drift in existing tables; hosts with their own migration history should apply Schema through their runner instead.

func Schema 

func Schema() fs.FS

Schema returns the embedded final-shape SQL (tables, indexes and triggers) as a file system of numbered .sql files, for hosts that apply schema through their own migration runner. Every statement is create-if-missing. The files are the same ones Migrate applies.

Types

type DBTX 

type DBTX interface {
	ExecContext(context.Context, string, ...any) (sql.Result, error)
	QueryContext(context.Context, string, ...any) (*sql.Rows, error)
	QueryRowContext(context.Context, string, ...any) *sql.Row
}

DBTX is the statement surface handed to a WriteTx callback. Every read or write inside the callback must use it, never Store.DB, or a single-connection database deadlocks.

type Hooks 

type Hooks interface {
	// AfterRunWritten runs after a run row is inserted or updated. The
	// snapshot is the run as now stored.
	AfterRunWritten(ctx context.Context, tx DBTX, run workflowruntime.RunSnapshot) error
	// AfterNodeWritten runs after a node invocation row (and its lease) is
	// inserted or updated. The snapshot is the node as now stored.
	AfterNodeWritten(ctx context.Context, tx DBTX, node workflowruntime.NodeInvocationSnapshot) error
}

Hooks lets a host mirror canonical run and node writes into its own tables. Each hook runs inside the write transaction, after the canonical statement that produced the snapshot succeeded; tx is the transaction's DBTX and must be the only handle the hook uses. A returned error rolls the whole write back, including the canonical statement, and is returned to the caller (wrap it in runtime.ErrInvalidRecord when the caller should treat it as a rejected record).

type Option 

type Option func(*Store) error

Option configures New.

func WithHooks 

func WithHooks(hooks Hooks) Option

WithHooks installs write hooks; see Hooks. A nil value clears them.

func WithRunColumns 

func WithRunColumns(columns RunColumns) Option

WithRunColumns overrides the run table and column names; see RunColumns.

type RunColumns 

type RunColumns struct {
	Table, ID, Status, Generation, CreatedAt string
}

RunColumns names the run table and the four columns of it the store reads and writes. It exists for hosts whose runs table is shared with product data under different names. The zero value of every field selects the default, which reproduces the schema Migrate creates: 

Table workflow_runs, ID run_id, Status status,
Generation generation, CreatedAt created_at

The table must also carry the store's other run columns (plan_digest, inputs_ref_json, outputs_ref_json, updated_at) under their default names, and any additional column it has must be nullable or defaulted, because the store's INSERT names only the columns it owns. Foreign keys from the other store tables must target the ID column.

type Store 

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

Store is a SQLite-backed runtime.StateStore. It owns the workflow_* tables created by Migrate (or by Schema applied through a host's own runner).

func New 

func New(db *sql.DB, opts ...Option) (*Store, error)

New wraps an open database. The database must already contain the store's tables (see Migrate and Schema); New does not migrate. The returned store shares the database's lifetime and must not outlive it.

The database should be opened with a busy timeout and, for a single writer process, SetMaxOpenConns(1); WriteTx holds one connection for the duration of each write transaction.

func (*Store) ActivateCompensationEntry 

func (s *Store) ActivateCompensationEntry(ctx context.Context, r workflowruntime.ActivateCompensationEntryRequest) (workflowruntime.ActivateCompensationEntryResult, error)

func (*Store) ActivateNodeRetry 

func (s *Store) ActivateNodeRetry(ctx context.Context, request workflowruntime.ActivateNodeRetryRequest) (workflowruntime.ActivateNodeRetryResult, error)

func (*Store) AdmitNode 

func (s *Store) AdmitNode(ctx context.Context, request workflowruntime.AdmitNodeRequest) (workflowruntime.AdmitNodeResult, error)

func (*Store) AppendEvent 

func (s *Store) AppendEvent(ctx context.Context, request workflowruntime.AppendEventRequest) (workflowruntime.Event, error)

AppendEvent implements runtime.StateStore with an atomic per-run sequence.

func (*Store) ApplyExternalOperation 

func (s *Store) ApplyExternalOperation(ctx context.Context, request workflowruntime.ApplyExternalOperationRequest) (workflowruntime.ApplyExternalOperationResult, error)

func (*Store) ApplyRunFailurePolicy 

func (s *Store) ApplyRunFailurePolicy(ctx context.Context, request workflowruntime.ApplyRunFailurePolicyRequest) (workflowruntime.ApplyRunFailurePolicyResult, error)

func (*Store) ApplyServiceHeartbeat 

func (s *Store) ApplyServiceHeartbeat(ctx context.Context, request workflowruntime.ApplyServiceHeartbeatRequest) (workflowruntime.ServiceSnapshot, error)

func (*Store) ApplyServiceReady 

func (s *Store) ApplyServiceReady(ctx context.Context, request workflowruntime.ApplyServiceReadyRequest) (workflowruntime.ApplyServiceReadyResult, error)

func (*Store) ApplyServiceStop 

func (s *Store) ApplyServiceStop(ctx context.Context, request workflowruntime.ApplyServiceStopRequest) (workflowruntime.ApplyServiceStopResult, error)

func (*Store) BeginManualCompensation 

func (s *Store) BeginManualCompensation(ctx context.Context, r workflowruntime.BeginManualCompensationRequest) (workflowruntime.FreezeCompensationResult, error)

func (*Store) BeginReactorContinuation 

func (s *Store) BeginReactorContinuation(ctx context.Context, request workflowruntime.ReactorContinuationRequest) (workflowruntime.ReactorSnapshot, workflowruntime.ReactorContinuationSnapshot, workflowruntime.IdempotencyOutcome, error)

func (*Store) BeginReactorDelivery 

func (s *Store) BeginReactorDelivery(ctx context.Context, request workflowruntime.BeginReactorDeliveryRequest) (workflowruntime.ReactorSnapshot, workflowruntime.ReactorDeliverySnapshot, workflowruntime.IdempotencyOutcome, error)

func (*Store) BeginReplay 

func (s *Store) BeginReplay(ctx context.Context, request workflowruntime.BeginReplayRequest) (workflowruntime.BeginReplayResult, error)

func (*Store) BeginRunUpdate 

func (s *Store) BeginRunUpdate(ctx context.Context, request workflowruntime.BeginRunUpdateRequest) (workflowruntime.RunUpdateSnapshot, workflowruntime.IdempotencyOutcome, error)

func (*Store) BeginTerminalIntent 

func (s *Store) BeginTerminalIntent(ctx context.Context, request workflowruntime.BeginTerminalIntentRequest) (workflowruntime.BeginTerminalIntentResult, error)

func (*Store) BindNodeInputs 

func (s *Store) BindNodeInputs(ctx context.Context, request workflowruntime.BindNodeInputsRequest) (workflowruntime.BindNodeInputsResult, error)

func (*Store) BindPin 

func (s *Store) BindPin(ctx context.Context, request workflowruntime.BindPinRequest) (workflowruntime.BindPinResult, error)

func (*Store) CancelCompensation 

func (s *Store) CancelCompensation(ctx context.Context, r workflowruntime.CancelCompensationRequest) (workflowruntime.CompensationLedgerSnapshot, error)

func (*Store) ClaimNode 

func (s *Store) ClaimNode(ctx context.Context, request workflowruntime.ClaimNodeRequest) (workflowruntime.ClaimResult, error)

ClaimNode implements runtime.StateStore. Only ready nodes are eligible; non-ready nodes and ready nodes with a live lease return an idempotently persisted Acquired=false result after claim-generation CAS succeeds.

func (*Store) ClaimReactorDelivery 

func (s *Store) ClaimReactorDelivery(ctx context.Context, request workflowruntime.ClaimReactorDeliveryRequest) (workflowruntime.ReactorDeliverySnapshot, error)

func (*Store) CompleteFanOut 

func (s *Store) CompleteFanOut(ctx context.Context, request workflowruntime.CompleteFanOutRequest) (workflowruntime.CompleteFanOutResult, error)

func (*Store) CompleteReactorContinuation 

func (s *Store) CompleteReactorContinuation(ctx context.Context, key string, expected uint64, at time.Time) (workflowruntime.ReactorSnapshot, workflowruntime.ReactorContinuationSnapshot, error)

func (*Store) CompleteReactorDelivery 

func (s *Store) CompleteReactorDelivery(ctx context.Context, request workflowruntime.CompleteReactorDeliveryRequest) (workflowruntime.ReactorSnapshot, workflowruntime.ReactorDeliverySnapshot, error)

func (*Store) CompleteRunUpdate 

func (s *Store) CompleteRunUpdate(ctx context.Context, request workflowruntime.CompleteRunUpdateRequest) (workflowruntime.RunUpdateSnapshot, error)

func (*Store) CompleteTerminalIntent 

func (s *Store) CompleteTerminalIntent(ctx context.Context, request workflowruntime.CompleteTerminalIntentRequest) (workflowruntime.CompleteTerminalIntentResult, error)

func (*Store) CreateNodeInvocation 

func (s *Store) CreateNodeInvocation(ctx context.Context, request workflowruntime.CreateNodeInvocationRequest) (workflowruntime.NodeInvocationSnapshot, error)

CreateNodeInvocation implements runtime.StateStore.

func (*Store) CreateRun 

func (s *Store) CreateRun(ctx context.Context, request workflowruntime.CreateRunRequest) (workflowruntime.RunSnapshot, workflowruntime.IdempotencyOutcome, error)

CreateRun implements runtime.StateStore and records its plan reference and start idempotency outcome in the same transaction.

func (*Store) DB 

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

DB returns the database the store was constructed with.

func (*Store) ExpandFanOut 

func (s *Store) ExpandFanOut(ctx context.Context, request workflowruntime.ExpandFanOutRequest) (workflowruntime.ExpandFanOutResult, error)

func (*Store) FailCompensationEntry 

func (s *Store) FailCompensationEntry(ctx context.Context, r workflowruntime.FailCompensationEntryRequest) (workflowruntime.SealCompensationEntryResult, error)

func (*Store) FailReactor 

func (s *Store) FailReactor(ctx context.Context, request workflowruntime.FailReactorRequest) (workflowruntime.ReactorSnapshot, error)

func (*Store) FindOpenSignalWait 

func (s *Store) FindOpenSignalWait(ctx context.Context, selector workflowruntime.SignalSelector) (workflowruntime.WaitSnapshot, error)

func (*Store) FindSignalWait 

func (s *Store) FindSignalWait(ctx context.Context, selector workflowruntime.SignalSelector, idempotencyKey string) (workflowruntime.WaitSnapshot, error)

func (*Store) FinishCompensableAttempt 

func (s *Store) FinishCompensableAttempt(ctx context.Context, request workflowruntime.FinishCompensableAttemptRequest) (workflowruntime.FinishCompensableAttemptResult, error)

func (*Store) FinishNodeAttempt 

func (s *Store) FinishNodeAttempt(ctx context.Context, request workflowruntime.FinishNodeAttemptRequest) (workflowruntime.FinishNodeAttemptResult, error)

FinishNodeAttempt implements runtime.StateStore with one transaction for the running attempt, aggregate node, lease release, and lifecycle event.

func (*Store) FreezeCompensation 

func (s *Store) FreezeCompensation(ctx context.Context, r workflowruntime.FreezeCompensationRequest) (workflowruntime.FreezeCompensationResult, error)

FreezeCompensation and the remaining transitions use snapshot_json as the semantic source and exact generation predicates as their CAS boundary.

func (*Store) GetCacheEntry 

func (s *Store) GetCacheEntry(ctx context.Context, key string, now time.Time) (workflowruntime.CacheEntry, bool, error)

GetCacheEntry implements runtime.StateStore. Expired rows are retained for later maintenance but are never returned as cache hits.

func (*Store) GetPinnedValue 

func (s *Store) GetPinnedValue(ctx context.Context, key string, now time.Time) (workflowruntime.PinnedValue, bool, error)

GetPinnedValue implements runtime.StateStore and excludes expired pins.

func (*Store) InspectSchedulerResources 

func (s *Store) InspectSchedulerResources(ctx context.Context, request workflowruntime.SchedulerResourceQuery) (workflowruntime.SchedulerResourceState, error)

func (*Store) ListAttempts 

func (s *Store) ListAttempts(ctx context.Context, id workflowruntime.NodeInvocationID) ([]workflowruntime.AttemptSnapshot, error)

ListAttempts implements runtime.StateStore in attempt-number order.

func (*Store) ListChildRuns 

func (s *Store) ListChildRuns(ctx context.Context, parent workflowruntime.RunID) ([]workflowruntime.ChildRunLink, error)

func (*Store) ListCompensationEntries 

func (s *Store) ListCompensationEntries(ctx context.Context, runID workflowruntime.RunID) ([]workflowruntime.CompensationEntrySnapshot, error)

func (*Store) ListEvents 

func (s *Store) ListEvents(ctx context.Context, query workflowruntime.EventQuery) ([]workflowruntime.Event, error)

ListEvents implements runtime.StateStore in ascending per-run sequence.

func (*Store) ListPinnedValues 

func (s *Store) ListPinnedValues(ctx context.Context, now time.Time) ([]workflowruntime.PinnedValue, error)

ListPinnedValues implements runtime.StateStore in stable key order.

func (*Store) ListRunInvocations 

func (s *Store) ListRunInvocations(ctx context.Context, runID workflowruntime.RunID) ([]workflowruntime.NodeInvocationSnapshot, error)

func (*Store) LoadAttempt 

func (s *Store) LoadAttempt(ctx context.Context, id workflowruntime.AttemptID) (workflowruntime.AttemptSnapshot, error)

LoadAttempt implements runtime.StateStore.

func (*Store) LoadCompensationEntryByHandler 

func (s *Store) LoadCompensationEntryByHandler(ctx context.Context, id workflowruntime.NodeInvocationID) (workflowruntime.CompensationEntrySnapshot, error)

func (*Store) LoadCompensationLedger 

func (s *Store) LoadCompensationLedger(ctx context.Context, runID workflowruntime.RunID) (workflowruntime.CompensationLedgerSnapshot, error)

func (*Store) LoadControlDecision 

func (s *Store) LoadControlDecision(ctx context.Context, id workflowruntime.ControlDecisionID) (workflowruntime.ControlDecisionSnapshot, error)

func (*Store) LoadExternalOperation 

func (s *Store) LoadExternalOperation(ctx context.Context, id workflowruntime.AttemptID) (workflowruntime.ExternalOperationSnapshot, error)

func (*Store) LoadFanOut 

func (s *Store) LoadFanOut(ctx context.Context, parent workflowruntime.NodeInvocationID) (workflowruntime.FanOutSnapshot, error)

func (*Store) LoadFanOutItemResults 

func (s *Store) LoadFanOutItemResults(ctx context.Context, parent workflowruntime.NodeInvocationID) ([]workflowruntime.FanOutItemResult, error)

func (*Store) LoadMemoEntry 

func (s *Store) LoadMemoEntry(ctx context.Context, key string) (workflowruntime.MemoEntry, error)

func (*Store) LoadNodeInvocation 

func (s *Store) LoadNodeInvocation(ctx context.Context, id workflowruntime.NodeInvocationID) (workflowruntime.NodeInvocationSnapshot, error)

LoadNodeInvocation implements runtime.StateStore.

func (*Store) LoadPin 

func (s *Store) LoadPin(ctx context.Context, id workflowruntime.NodeInvocationID) (workflowruntime.PinBinding, error)

func (*Store) LoadPlan 

func (s *Store) LoadPlan(ctx context.Context, digest string) (workflowruntime.PlanRef, error)

LoadPlan implements runtime.StateStore.

func (*Store) LoadReactor 

func (s *Store) LoadReactor(ctx context.Context, id string) (workflowruntime.ReactorSnapshot, error)

func (*Store) LoadReactorDelivery 

func (s *Store) LoadReactorDelivery(ctx context.Context, reactorID, key string) (workflowruntime.ReactorDeliverySnapshot, error)

func (*Store) LoadReplayProvenance 

func (s *Store) LoadReplayProvenance(ctx context.Context, runID workflowruntime.RunID) (workflowruntime.ReplayProvenance, error)

func (*Store) LoadRetryActivation 

func (s *Store) LoadRetryActivation(ctx context.Context, id string) (workflowruntime.RetryActivationSnapshot, error)

func (*Store) LoadRun 

func (s *Store) LoadRun(ctx context.Context, id workflowruntime.RunID) (workflowruntime.RunSnapshot, error)

LoadRun implements runtime.StateStore.

func (*Store) LoadRunPolicyDecision 

func (s *Store) LoadRunPolicyDecision(ctx context.Context, runID workflowruntime.RunID) (workflowruntime.RunPolicyDecisionSnapshot, error)

func (*Store) LoadRunUpdate 

func (s *Store) LoadRunUpdate(ctx context.Context, key string) (workflowruntime.RunUpdateSnapshot, error)

func (*Store) LoadService 

func (s *Store) LoadService(ctx context.Context, start workflowruntime.NodeInvocationID) (workflowruntime.ServiceSnapshot, error)

func (*Store) LoadTerminalIntent 

func (s *Store) LoadTerminalIntent(ctx context.Context, runID workflowruntime.RunID) (workflowruntime.TerminalIntentSnapshot, error)

func (*Store) LoadValueRecord 

func (s *Store) LoadValueRecord(ctx context.Context, ref values.ValueSetRef) (workflowruntime.ValueRecord, error)

func (*Store) LoadValues 

func (s *Store) LoadValues(ctx context.Context, ref values.ValueSetRef) (values.ValueSet, error)

LoadValues implements runtime.StateStore and verifies both the caller's digest and the immutable row's content digest before returning decoded data.

func (*Store) LoadWait 

func (s *Store) LoadWait(ctx context.Context, id workflowruntime.WaitID) (workflowruntime.WaitSnapshot, error)

LoadWait implements runtime.StateStore.

func (*Store) LoadWaitContinuation 

func (s *Store) LoadWaitContinuation(ctx context.Context, id workflowruntime.AttemptID) (workflowruntime.WaitSnapshot, error)

LoadWaitContinuation returns the single resumed wait bound to exactly id. Open, timed-out, canceled, and safely-unbound legacy rows never qualify.

func (*Store) MarkReactorWaiting 

func (s *Store) MarkReactorWaiting(ctx context.Context, id string, expected uint64, at time.Time) (workflowruntime.ReactorSnapshot, error)

func (*Store) PrepareServiceStart 

func (s *Store) PrepareServiceStart(ctx context.Context, request workflowruntime.PrepareServiceStartRequest) (workflowruntime.ServiceSnapshot, error)

func (*Store) PutCacheEntry 

func (s *Store) PutCacheEntry(ctx context.Context, entry workflowruntime.CacheEntry) error

PutCacheEntry implements runtime.StateStore. A cache key is a mutable index to immutable output values, so a later valid entry replaces the prior row.

func (*Store) PutPinnedValue 

func (s *Store) PutPinnedValue(ctx context.Context, pin workflowruntime.PinnedValue) error

PutPinnedValue implements runtime.StateStore. Pin keys are mutable stable names, while their target value-set references remain immutable.

func (*Store) QueryRunState 

func (s *Store) QueryRunState(ctx context.Context, request workflowruntime.RunStateQuery) (workflowruntime.RunStateView, error)

func (*Store) ReconcileCrashedAttempt 

func (s *Store) ReconcileCrashedAttempt(ctx context.Context, request workflowruntime.ReconcileCrashedAttemptRequest) (workflowruntime.ReconcileCrashedAttemptResult, error)

func (*Store) RecordChildRun 

func (s *Store) RecordChildRun(ctx context.Context, link workflowruntime.ChildRunLink) error

func (*Store) RecordControlDecision 

func (s *Store) RecordControlDecision(ctx context.Context, request workflowruntime.RecordControlDecisionRequest) (workflowruntime.RecordControlDecisionResult, error)

func (*Store) RecordExternalActivation 

func (s *Store) RecordExternalActivation(ctx context.Context, request workflowruntime.ExternalActivationRequest) (workflowruntime.ExternalActivationSnapshot, workflowruntime.IdempotencyOutcome, error)

RecordExternalActivation implements runtime.StateStore. One activation declaration may fire repeatedly; idempotency_key, not activation_id, is the immutable firing identity.

func (*Store) RecordMemoEntry 

func (s *Store) RecordMemoEntry(ctx context.Context, entry workflowruntime.MemoEntry) (workflowruntime.MemoEntry, workflowruntime.IdempotencyOutcome, error)

func (*Store) RecordPlan 

func (s *Store) RecordPlan(ctx context.Context, plan workflowruntime.PlanRef) error

RecordPlan implements runtime.StateStore.

func (*Store) RecoverCancellationIntents 

func (s *Store) RecoverCancellationIntents(ctx context.Context, query workflowruntime.CancellationIntentQuery) ([]workflowruntime.CancellationIntentSnapshot, error)

func (*Store) RecoverChildTerminalWaits 

func (s *Store) RecoverChildTerminalWaits(ctx context.Context, limit int) ([]workflowruntime.ChildTerminalWait, error)

func (*Store) RecoverCompensation 

func (s *Store) RecoverCompensation(ctx context.Context, limit int) ([]workflowruntime.CompensationLedgerSnapshot, error)

func (*Store) RecoverExternalOperations 

func (s *Store) RecoverExternalOperations(ctx context.Context, query workflowruntime.ExternalOperationQuery) ([]workflowruntime.ExternalOperationSnapshot, error)

func (*Store) RecoverOpenWaits 

func (s *Store) RecoverOpenWaits(ctx context.Context, query workflowruntime.OpenWaitQuery) ([]workflowruntime.WaitSnapshot, error)

func (*Store) RecoverReactorContinuations 

func (s *Store) RecoverReactorContinuations(ctx context.Context, limit int) ([]workflowruntime.ReactorContinuationSnapshot, error)

func (*Store) RecoverReactorDeliveries 

func (s *Store) RecoverReactorDeliveries(ctx context.Context, limit int) ([]workflowruntime.ReactorDeliverySnapshot, error)

func (*Store) RecoverReactors 

func (s *Store) RecoverReactors(ctx context.Context, limit int) ([]workflowruntime.ReactorSnapshot, error)

func (*Store) RecoverRetryActivations 

func (s *Store) RecoverRetryActivations(ctx context.Context, query workflowruntime.RetryActivationQuery) ([]workflowruntime.RetryActivationSnapshot, error)

func (*Store) RecoverRunUpdates 

func (s *Store) RecoverRunUpdates(ctx context.Context, limit int) ([]workflowruntime.RunUpdateSnapshot, error)

func (*Store) RecoverServiceStart 

func (s *Store) RecoverServiceStart(ctx context.Context, request workflowruntime.RecoverServiceStartRequest) (workflowruntime.SuspendServiceStartResult, error)

func (*Store) RecoverServices 

func (s *Store) RecoverServices(ctx context.Context, query workflowruntime.ServiceQuery) ([]workflowruntime.ServiceSnapshot, error)

func (*Store) Recovery 

func (s *Store) Recovery(ctx context.Context, query workflowruntime.RecoveryQuery) (workflowruntime.RecoverySnapshot, error)

Recovery implements runtime.StateStore with deterministic, independently limited categories. Categories are persisted facts and may overlap.

func (*Store) ReleaseNodeClaim 

func (s *Store) ReleaseNodeClaim(ctx context.Context, request workflowruntime.ReleaseClaimRequest) error

ReleaseNodeClaim implements runtime.StateStore with lease fencing and a node record generation CAS in one transaction.

func (*Store) ReleaseReactorDelivery 

func (s *Store) ReleaseReactorDelivery(ctx context.Context, request workflowruntime.ReleaseReactorDeliveryRequest) (workflowruntime.ReactorDeliverySnapshot, error)

func (*Store) RenewNodeLease 

func (s *Store) RenewNodeLease(ctx context.Context, request workflowruntime.RenewLeaseRequest) (workflowruntime.ClaimLease, error)

RenewNodeLease implements runtime.StateStore with lease fencing and an extension-only expiry update in one transaction. Renewal is deliberately a lease-only mutation: it must not advance the semantic node generation or updated_at and invalidate a dispatcher's in-flight lifecycle CAS.

func (*Store) RequestExternalOperationCancel 

func (s *Store) RequestExternalOperationCancel(ctx context.Context, request workflowruntime.RequestExternalOperationCancelRequest) (workflowruntime.RequestExternalOperationCancelResult, error)

func (*Store) RequestRunCancellation 

func (s *Store) RequestRunCancellation(ctx context.Context, request workflowruntime.RequestRunCancellationRequest) (workflowruntime.RequestRunCancellationResult, error)

func (*Store) RequestRunCancellationWithFinalizers 

func (s *Store) RequestRunCancellationWithFinalizers(ctx context.Context, request workflowruntime.RequestRunCancellationWithFinalizersRequest) (workflowruntime.RequestRunCancellationWithFinalizersResult, error)

func (*Store) ResolveCancellationIntent 

func (s *Store) ResolveCancellationIntent(ctx context.Context, request workflowruntime.ResolveCancellationIntentRequest) (workflowruntime.ResolveCancellationIntentResult, error)

func (*Store) ResumeNodeWait 

func (s *Store) ResumeNodeWait(ctx context.Context, request workflowruntime.ResumeNodeWaitRequest) (workflowruntime.ResumeWaitResult, error)

func (*Store) RetryCompensation 

func (s *Store) RetryCompensation(ctx context.Context, r workflowruntime.RetryCompensationRequest) (workflowruntime.CompensationLedgerSnapshot, error)

func (*Store) ReuseNodeOutputs 

func (s *Store) ReuseNodeOutputs(ctx context.Context, request workflowruntime.ReuseNodeOutputsRequest) (workflowruntime.ReuseNodeOutputsResult, error)

func (*Store) SaveNodeInvocation 

func (s *Store) SaveNodeInvocation(ctx context.Context, request workflowruntime.SaveNodeInvocationRequest) (workflowruntime.NodeInvocationSnapshot, error)

SaveNodeInvocation implements runtime.StateStore and rejects lifecycle, attempt, value, and claim bypasses before its SQL generation CAS.

func (*Store) SaveRun 

func (s *Store) SaveRun(ctx context.Context, request workflowruntime.SaveRunRequest) (workflowruntime.RunSnapshot, error)

SaveRun implements runtime.StateStore and rejects lifecycle bypasses before executing its SQL generation CAS.

func (*Store) SaveValues 

func (s *Store) SaveValues(ctx context.Context, request workflowruntime.SaveValuesRequest) (values.ValueSetRef, error)

SaveValues implements runtime.StateStore with an immutable, digest-bound value-set row. SQLite assigns the opaque reference identity.

func (*Store) ScheduleNodeRetry 

func (s *Store) ScheduleNodeRetry(ctx context.Context, request workflowruntime.ScheduleNodeRetryRequest) (workflowruntime.ScheduleNodeRetryResult, error)

func (*Store) SealCompensationEntry 

func (s *Store) SealCompensationEntry(ctx context.Context, r workflowruntime.SealCompensationEntryRequest) (workflowruntime.SealCompensationEntryResult, error)

func (*Store) StartNodeAttempt 

func (s *Store) StartNodeAttempt(ctx context.Context, request workflowruntime.StartNodeAttemptRequest) (workflowruntime.StartNodeAttemptResult, error)

StartNodeAttempt implements runtime.StateStore with one transaction for the node transition, attempt creation, and lifecycle event.

func (*Store) SuspendExternalOperation 

func (s *Store) SuspendExternalOperation(ctx context.Context, request workflowruntime.SuspendExternalOperationRequest) (workflowruntime.SuspendExternalOperationResult, error)

func (*Store) SuspendNodeWait 

func (s *Store) SuspendNodeWait(ctx context.Context, request workflowruntime.SuspendNodeWaitRequest) (workflowruntime.SuspendWaitResult, error)

func (*Store) SuspendServiceStart 

func (s *Store) SuspendServiceStart(ctx context.Context, request workflowruntime.SuspendServiceStartRequest) (workflowruntime.SuspendServiceStartResult, error)

func (*Store) SuspendServiceTeardown 

func (s *Store) SuspendServiceTeardown(ctx context.Context, request workflowruntime.SuspendServiceTeardownRequest) (workflowruntime.SuspendServiceTeardownResult, error)

func (*Store) TimeoutWait 

func (s *Store) TimeoutWait(ctx context.Context, request workflowruntime.TimeoutWaitRequest) (workflowruntime.WaitTimeoutResult, error)

func (*Store) TransitionNode 

func (s *Store) TransitionNode(ctx context.Context, request workflowruntime.NodeTransitionRequest) (workflowruntime.NodeTransitionResult, error)

TransitionNode implements runtime.StateStore with atomic node, lease, and lifecycle-event persistence.

func (*Store) TransitionRun 

func (s *Store) TransitionRun(ctx context.Context, request workflowruntime.RunTransitionRequest) (workflowruntime.RunTransitionResult, error)

TransitionRun implements runtime.StateStore with atomic state/event writes.

func (*Store) WriteTx 

func (s *Store) WriteTx(ctx context.Context, operation string, fn func(DBTX) error) error

WriteTx runs fn inside one BEGIN IMMEDIATE transaction on a dedicated connection. A non-nil error from fn rolls the transaction back and is returned unchanged; otherwise the transaction commits. operation labels errors raised by the transaction machinery itself.

Source Files

View all Source files

Jump to

Keyboard shortcuts

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