engine

package
v0.2.0 Latest Latest
Warning

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

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

Documentation

Overview

Package engine provides the core workflow execution engine for cleat.

It manages WASM module execution via pluggable backends (wasmtime, wazero), event-sourced workflow state, HTTP service integration, and plugin lifecycle. The App type orchestrates a runtime, store, service caller, and HTTP mux.

Key types:

  • App — top-level application orchestrator
  • Runtime — WASM runtime lifecycle manager
  • WasmBackend — interface for WASM execution backends
  • StoreFactory / WorkflowStore — event-sourced workflow persistence
  • HostHandler — host function wiring for WASM modules

Package host provides the core cleat workflow engine, including encryption at rest for sensitive event payloads.

Package host provides a wazero-based WASM runtime for executing cleat workflow modules produced by the `cleat build` command.

Architecture: 

Runtime — wraps wazero, registers host function imports, manages modules
Engine  — cleat execution with checkpoint/replay on top of Runtime
HostHandler — per-execution session interface (carried in context)

The host reads/writes strings in WASM linear memory using (ptr, len) pairs. All host function imports are registered on the "env" module. Per-execution state (replay history, step counter) is carried in context.Context.

Index

Constants

View Source 
const (
	EventCodeCall                  = 0
	EventCodeSleep                 = 1
	EventCodeAwaitSignals          = 2
	EventCodeSignalReceived        = 3
	EventCodeDefer                 = 4
	EventCodeChildWorkflow         = 5
	EventCodeAwaitChild            = 6
	EventCodeContinueAsNew         = 7
	EventCodeHeartbeat             = 8
	EventCodeAwaitAllChildren      = 9
	EventCodePluginCall            = 10
	EventCodeCreatePromise         = 11
	EventCodeAwaitPromise          = 12
	EventCodePromiseResolved       = 13
	EventCodePromiseRejected       = 14
	EventCodeUpdateHandler         = 15
	EventCodeStateMutation         = 16
	EventCodeRunDetached           = 17
	EventCodeAcquireLock           = 18
	EventCodeReleaseLock           = 19
	EventCodePluginCallStreamChunk = 20
	EventCodeSideEffect            = 21
	EventCodeScopeAcquired         = 22
	EventCodeFetch                 = 23
	EventCodeDurableLog            = 24
	EventCodeDurableSend           = 25
	EventCodeDurableScheduleInvoke = 26
	EventCodeScheduleCron          = 27
	EventCodeDeleteCron            = 28
	EventCodeListCrons             = 29
)

Event type codes for compact JSONB storage. Short int codes minimize storage size when a workflow has thousands of compacted events.

View Source 
const (
	// MisfireCatchUp delivers firings missed during an outage, one instant per
	// poll tick, up to the schedule's catch-up limit. The default, because the
	// engine promises at-least-once.
	MisfireCatchUp = "catch_up"

	// MisfireSkip resumes at the next future instant and delivers none of the
	// backlog. For schedules where a late firing is worse than no firing -- a
	// "send the 09:00 digest" job has nothing useful to say at 14:00.
	MisfireSkip = "skip"

	// OverlapAllow starts a run even if the previous one is still going. The
	// default only because it is what the scheduler has always done: changing
	// it would silently alter existing deployments. It is the wrong default for
	// most real schedules, since a job that occasionally overruns its interval
	// quietly becomes an unbounded fan-out.
	OverlapAllow = "allow"

	// OverlapSkip does not start a run while the previous one from this
	// schedule is still running or ready.
	OverlapSkip = "skip"

	// DefaultCatchUpLimit bounds catch_up when a schedule does not set its own.
	//
	// Generous for the schedules the bound is meant to protect -- hourly and
	// slower cannot reach it within any plausible outage -- and small enough
	// that a catch-up burst stays in the same order of magnitude as a normal
	// minute of work. It is a judgement call, not a measurement.
	DefaultCatchUpLimit = 60
)

Schedule policy values. These are stored as text and read by a background loop that has nobody to report a bad value to, so the database carries CHECK constraints as well -- a value the scheduler cannot interpret must be impossible to store rather than handled at 03:00.

View Source 
const (
	// DefaultMinVersionsToKeep is the minimum number of recent versions to
	// retain during GC, regardless of age or activity.
	DefaultMinVersionsToKeep = 3

	// DefaultMaxVersionAge is the maximum age of a deprecated version before
	// it becomes eligible for GC.
	DefaultMaxVersionAge = 30 * 24 * time.Hour // 30 days

	// DefaultGCOffset is the default age threshold for purging versions
	// that have zero active instances.
	DefaultGCOffset = 7 * 24 * time.Hour // 7 days
)
View Source 
const DefaultCompactionThreshold = 1000

DefaultCompactionThreshold is the default number of events before history compaction triggers. A workflow with more than this many events is eligible.

View Source 
const DefaultMaxCompactedEvents = 10000

DefaultMaxCompactedEvents caps the number of compacted events stored in a single compaction state JSONB. Beyond this limit, the oldest events are truncated into a summary to keep the compaction state size bounded.

View Source 
const DefaultMaxWasmStringLen = 1048576

DefaultMaxWasmStringLen is the default maximum WASM string length (1 MiB).

View Source 
const DefaultMemoryLimitPages = 512

DefaultMemoryLimitPages is the default max WASM linear memory (512 pages = 32 MB).

View Source 
const DefaultOutBufSize = 1048576

DefaultOutBufSize is the default WASM output buffer size (1 MiB).

View Source 
const DefaultScheduleTimezone = "UTC"

DefaultScheduleTimezone is the zone a schedule is evaluated in when it does not name one. UTC, because a schedule that means "02:00" should keep meaning the same instant regardless of which worker in the fleet happens to pick it up, and because UTC has no DST transitions to reason about.

View Source 
const DefaultTenantUUID = "00000000-0000-0000-0000-000000000000"

DefaultTenantUUID is the all-zeros UUID used when no tenant is specified.

View Source 
const DefaultWasmtimeExecutionTimeout = 30 * time.Second

DefaultWasmtimeExecutionTimeout bounds a single wasmtime invocation (one fresh execution or one replay pass) via epoch interruption when the caller has not configured a tighter one. This is a required safety net, not just a convenience default: wasmtime-go does not observe Go context cancellation while a WASM export call is in progress (see the comment on wasmtimeBackend.Execute in backend_wasmtime.go), so without a baked-in bound here an infinite loop in a workflow hangs the worker permanently regardless of any engine- or worker-level context timeout.

30s matches the wazero backend's hardcoded Runtime.callTimeout default (engine/runtime.go), so both backends behave the same way out of the box.

View Source 
const DefaultWasmtimeInstancesLimit = 256

DefaultWasmtimeInstancesLimit bounds how many module instances a single wasmtime store may create. Component bundles observed in this codebase (CPython runtime + adapters) use at most a few dozen instances; 256 leaves generous headroom while still bounding runaway instantiation.

View Source 
const DefaultWasmtimeMemoryLimitBytes = int64(DefaultMemoryLimitPages) * int64(wasmPageSize)

DefaultWasmtimeMemoryLimitBytes bounds linear memory per wasmtime store when the caller has not configured a tighter one via WithWasmtimeMemoryLimits. It matches DefaultMemoryLimitPages (32 MiB), the wazero backend's default (engine/runtime.go), so an operator sees the same memory ceiling regardless of which backend happens to run a given workflow.

View Source 
const DefaultWasmtimeTableElementsLimit = 8 * 1024 * 1024

DefaultWasmtimeTableElementsLimit bounds indirect-function-table growth per wasmtime store. Component-model bundles in this codebase size their largest table at 1,048,576 elements (see tblMinSize in wasmtimeBackend.ExecuteComponent); 8x that headroom keeps existing workflows working while still capping unbounded/attacker-controlled table growth.

View Source 
const MaxRetryAttempts = 100

MaxRetryAttempts is the worker-enforced ceiling for DurableCallWithRetry maxAttempts, preventing a misconfigured WASM module from retrying forever.

View Source 
const PendingSentinel = pendingSentinel

PendingSentinel is the exported form of pendingSentinel, provided so that external packages (notably the integrity test suite) can reference it without duplicating the sentinel value.

Variables

View Source 
var DebugTiming = os.Getenv("CLEAT_DEBUG_TIMING") == "1"

DebugTiming enables verbose per-step/per-execution timing output to stderr and structured logs. Set the CLEAT_DEBUG_TIMING=1 environment variable to enable it. Default off — timing I/O adds measurable overhead at high concurrency.

View Source 
var ErrAdminOpNotImplemented = errors.New("not implemented")

ErrAdminOpNotImplemented marks an admin operation the store genuinely does not implement, as opposed to one that failed.

The distinction is the whole reason it exists: cmd/cleat-worker mapped every error from these methods to 500, so "this endpoint was never built" and "the database is broken" were the same answer to a caller. AdminForceComplete and AdminForceFail are implemented in store_admin.go; AdminReReplay is not, and says so with 501 rather than pretending to have tried. What it needs is in IMPROVEMENT-PLAN.md 3.20.

View Source 
var ErrCrossTenantClaimUnsupported = errors.New("cross-tenant claim not supported by this store's topology")

ErrCrossTenantClaimUnsupported is returned by a store that implements CrossTenantClaimer but cannot honour it in the topology it finds itself in.

Implementing the interface and quietly returning one tenant's work would be worse than not implementing it: the caller would believe it was claiming for everyone. MySQL is the case that forced this -- MySQLStoreFactory gives each tenant its own physical database, so there is no predicate to drop; the other tenants' rows are not filtered out, they are in a different database. The same type is also used against a single shared database, where the claim IS meaningful, so the answer depends on how the store was built rather than on which type it is.

View Source 
var ErrFenceLost = errors.New("fence lost: workflow reassigned to another worker (generation mismatch)")

ErrFenceLost is returned by the generation-fenced workflow lifecycle methods (CompleteWorkflow, FailWorkflow, MoveToDeadLetterQueue, ContinueAsNew, FinalizeWorkflowSegment) when the fencing UPDATE affected zero rows -- i.e. the (workflow_id, worker_id, generation) tuple the caller presented no longer matches the row in the database. This happens when a worker stalls long enough to be reaped (ReapStaleInstances resets status/assigned_to and bumps generation) and is then reclaimed by another worker before the stalled worker's segment finishes and tries to persist its result. It is an expected, normal occurrence under reaping -- callers should treat it as "someone else now owns this workflow" and return cleanly rather than retrying or surfacing it as a failure.

View Source 
var ErrSuspended = fmt.Errorf("workflow suspended")

ErrSuspended is returned by CallExport when the workflow suspends.

View Source 
var ErrWasmtimeCGOUnavailable = errors.New("wasmtime backend requires CGO (binary built with CGO_ENABLED=0)")

ErrWasmtimeCGOUnavailable is the error NewWasmtimeBackend returns when the running binary was built with CGO disabled (CGO_ENABLED=0), so the cgo-only wasmtime backend was never compiled in (see backend_wasmtime_stub.go, guarded by "//go:build !cgo").

This is a deliberate, supported configuration (CLAUDE.md: "wazero — the CGO-less fallback and nothing else"), not a failure worth alarming on. It exists so callers can tell it apart from every other error NewWasmtimeBackend can return on a CGO-enabled build (guarded by "//go:build cgo", backend_wasmtime.go) — those indicate the backend of record failed to initialize despite being compiled in, which is a safety-relevant degradation: wazero cannot fence a compute-bound guest (see CLAUDE.md, "measured three ways, all failing"), so silently falling back to it should never be treated the same as an expected CGO-less build.

View Source 
var ErrWorkflowDefOwnedByAnotherTenant = errors.New("workflow definition is owned by another tenant")

ErrWorkflowDefOwnedByAnotherTenant is returned by DeployWorkflowDef when a definition of that name and version already exists and belongs to a different tenant. Callers can test for it with errors.Is.

View Source 
var MaxWasmStringLen uint32 = 1048576

MaxWasmStringLen is the maximum size of any string parameter read from WASM linear memory. This prevents a malicious or buggy WASM module from causing the host to allocate excessive memory via a single host function call. Set before creating any Runtime to configure. Default: 1 MiB.

View Source 
var OutBufSize uint32 = 1048576

OutBufSize is the output buffer size in bytes for WASM export calls. Set before creating any Runtime to configure. Default: 1 MiB.

View Source 
var WasmtimeLanguages = []string{"go", "assemblyscript", "java", "rust", "python"}

WasmtimeLanguages are the guest languages served by the wasmtime backend. Everything else falls back to the wazero Runtime.

This is the single source of truth, and it is single deliberately. There used to be two: cmd/cleat-worker registered "go" alone, and cleat/wasmtest registered go, assemblyscript, python and java. They disagreed in both directions -- the harness ran Python on a backend the worker never sends it to, and neither routed Rust -- so a test passing in the harness said nothing about the configuration the product actually runs.

Membership means "verified to load and execute on wasmtime", not "ought to work". Each entry here was run before it was added:

  • go: the primary path and the backend of record (CLAUDE.md).

  • assemblyscript, java: reached wasmtime for a long time by accident -- DetectLanguage could not identify them and defaulted to "go" -- and were confirmed to load and execute before being named explicitly. See 2.72.

  • rust: exercised by tests/cross-language, which builds the same wasm32-unknown-unknown cdylib that `cleat build --target rust` ships. All seven tests pass on wasmtime, including both cross-replay directions (execute under one runtime, replay the recorded history under the other), plus TestPluginCalls_Wasm_Rust in the plugin harness.

    Rust was previously excluded on the grounds that "wasmtime-go v44 still crashes on fn.Call for Rust cdylib core modules". That does not reproduce. The reason was true when written, as far as anyone can tell, and outlived its cause -- the same shape as the stale CGO_ENABLED=0 note in CLAUDE.md. Until 2026-08-04 it could not have been rechecked cheaply: tests/cross-language built wasm32-wasip1 rather than the shipped target, so the suite covered an artifact no user runs.

  • python: added 2026-08-05, and it is the entry whose history is worth knowing. Python is a Component Model guest, not a core module, so it takes backend_wasmtime.go's component branch rather than the ordinary instantiation path. There are two implementations of that branch: the native one in component_cgo.go, which hands the component to wasmtime's own Component Model runtime, and a hand-rolled decomposition path that re-implements shared-everything dynamic linking in Go.

    Only the second one ever ran. The native path sat behind the wasmtime_component_cgo build tag, which no build, CI job, Makefile or Dockerfile set, so every build got a stub that returned "not built" and fell through to decomposition -- where componentize-py output stops at `undefined element: out of bounds table access` instantiating instance 52 (module 8). Three sessions read that error as the state of Python-on-wasmtime. It was the state of the fallback.

    With the headers vendored (engine/wasmtimeinc) so the tag could be dropped, the same component runs: executes, records its durable call, returns. No change to the component path's logic was needed -- it was correct and uncompiled. Verified with a real HostHandler on a component componentize-py built fresh, not on the stale checked-in fixture, and the acceptance test named in IMPROVEMENT-PLAN 2.72, TestPythonWasmEndToEnd, is unskipped.

Absent, and why:

  • nothing. Every language cleat builds for is served by wasmtime. See IMPROVEMENT-PLAN 3.30 for what that leaves the wazero runtime doing: it is no longer the fallback for any language, and an unfenced backend that nothing routes to is a liability rather than a safety net.

See IMPROVEMENT-PLAN.md 2.72 and 1.5/2.28.

Functions

func CatchUpLimitOrDefault 

func CatchUpLimitOrDefault(n int) int

func CompactWorkflowHistory 

func CompactWorkflowHistory(ctx context.Context, store WorkflowStore, workflowID string, threshold int, metrics *prometheus.Metrics) error

CompactWorkflowHistory compacts the event history for a workflow. It loads all events, extracts a compaction checkpoint from events before the compaction point, deletes those events from the database, and stores the compaction state on the workflow_instances row.

func DeferralsFromHistory 

func DeferralsFromHistory(history []EventRecord) map[string]string

DeferralsFromHistory returns the defers registered during the execution that produced the given history. It scans the history for defer events.

func DurableCallIdempotencyKey 

func DurableCallIdempotencyKey(workflowID, runID string, step int) string

DurableCallIdempotencyKey derives the key for one durable call. 

key = base32(sha256(workflowID || 0x00 || runID || 0x00 || step))

Every input is deterministic on replay, so the key for a given logical step is identical on the original run and on every replay of it. That is the whole mechanism: after a crash, replay re-issues the call with the same key and a service that honours keys returns the original outcome rather than doing the work again.

runID is included so that ContinueAsNew — genuinely new work — gets fresh keys instead of colliding with the run it continues from.

The 0x00 separators matter. Without them the concatenation is ambiguous: workflow "ab" run "c" step 1 and workflow "a" run "bc" step 1 would hash identically, and two unrelated calls would silently deduplicate against each other. A NUL cannot appear in any of these identifiers, so the encoding is injective.

func ForceComplete 

func ForceComplete(ctx context.Context, store WorkflowStore, workflowID string, generation int64, operator string, result string) error

ForceComplete marks a workflow as done with the given result, bypassing worker ownership checks. The generation counter is checked to prevent stale writes. An audit event is written atomically with the status change.

func ForceFail 

func ForceFail(ctx context.Context, store WorkflowStore, workflowID string, generation int64, operator string, errorMsg, errorCode string) error

ForceFail marks a workflow as failed with the given error, bypassing worker ownership checks. The generation counter is checked to prevent stale writes. An audit event is written atomically with the status change.

func FormatRLSBypass 

func FormatRLSBypass(reasons []RLSBypassReason) string

FormatRLSBypass renders reasons as an operator-facing message, including what to do about it. Returns "" for no reasons.

func FreshCallCount 

func FreshCallCount() int64

FreshCallCount returns the total fresh DurableCall count from the atomic counter.

func FreshStepCount 

func FreshStepCount() int64

FreshStepCount returns the total fresh step count from the atomic counter.

func LoadRedactPatterns 

func LoadRedactPatterns(path string) error

LoadRedactPatterns reads a newline-separated list of sensitive-field substrings from path and appends them to the built-in patterns. Blank lines and lines starting with '#' are ignored. Returns an error if the file cannot be read.

func LoadScheduleLocation 

func LoadScheduleLocation(name string) (*time.Location, bool)

LoadScheduleLocation resolves a schedule's timezone name to a *time.Location, falling back to UTC when the name is empty or unloadable. The bool reports whether the fallback was taken, so a caller can log the difference between "this schedule is UTC" and "this schedule wanted a zone this process cannot load" -- which are very different operational situations and must not look the same in the logs.

func LogStaleAlerts 

func LogStaleAlerts(alerts []StaleVersionAlert)

LogStaleAlerts prints stale version alerts to the standard logger.

func MSSQLConnectionString 

func MSSQLConnectionString(host string, port int, user, password, database string) string

MSSQLConnectionString builds a SQL Server connection string. Format: sqlserver://user:pass@host:port?database=dbname&connection+timeout=30

func MisfirePolicyOrDefault 

func MisfirePolicyOrDefault(p string) string

MisfirePolicyOrDefault and friends normalise on the way into the store, so the column never holds an empty string that a reader has to know means something.

func NewWasmtimeBackend 

func NewWasmtimeBackend(ctx context.Context, opts ...WasmtimeOption) (*wasmtimeBackend, error)

NewWasmtimeBackend creates a new wasmtimeBackend with a fresh engine configured to bound WASM execution: epoch interruption is always enabled (see epochTickInterval / DefaultWasmtimeExecutionTimeout below) so a runaway workflow cannot hang the worker permanently, which is the bug this backend previously had with a bare wasmtime.NewEngine() and no Config at all. Fuel-based instruction metering is enabled additionally when WithWasmtimeInstructionLimit(n) is passed with n > 0.

func NextCronTime 

func NextCronTime(cronExpr string, from time.Time) time.Time

NextCronTime computes the next firing time at or after from for a five-field cron expression.

An expression that does not parse yields from+24h, preserving the behaviour callers have always had. That fallback is a poor answer and callers should prefer to reject the expression with ValidateCronExpr before it is ever stored -- but changing what this returns would silently re-time every already-stored schedule that does not parse, so the fallback stays and the validation goes in front of it.

The same value is returned for an expression that parses but can never match (`0 0 30 2 *` -- there is no 30th of February), after a four-year search. It evaluates in from's own location, which for a caller passing time.Now() is the host's local zone. That is exactly the ambiguity NextCronTimeIn exists to remove: two workers in different zones computed different firing times for the same schedule. Prefer NextCronTimeIn with the schedule's stored timezone.

func NextCronTimeIn 

func NextCronTimeIn(expr string, from time.Time, loc *time.Location) time.Time

NextCronTimeIn computes the next firing time strictly after from, evaluating the expression's wall-clock fields in loc.

It walks civil days rather than absolute minutes, because "07:00 every day" is a statement about a wall clock and a wall clock is not a fixed offset from UTC. A minute-by-minute scan over absolute time gets DST wrong in both directions: it fires twice in autumn and not at all in spring.

A nil loc is UTC.

func OverlapPolicyOrDefault 

func OverlapPolicyOrDefault(p string) string

func ReReplay 

func ReReplay(ctx context.Context, store WorkflowStore, workflowID string, generation int64, operator string) error

ReReplay resets a workflow to 'ready' state so the dispatcher picks it up for re-execution from its existing event history. The generation counter is checked to prevent stale writes. An audit event is written atomically with the status change.

func Redact 

func Redact(raw string) string

Redact processes a JSON string and returns a redacted version where sensitive fields have their values replaced with "[REDACTED]". It handles:

  • Nested objects recursively
  • String values that look like JWTs
  • Field names matching token, secret, password, credential, api_key, authorization, and similar patterns (case-insensitive)

If the input is not valid JSON, it is returned as-is.

func RedactMap 

func RedactMap(m map[string]any)

RedactMap is like Redact but operates on a map[string]any in-place, returning the same map. This is useful when the input is already parsed.

func RedactOnRead 

func RedactOnRead(data string) string

RedactOnRead applies retroactive redaction to a string that was loaded from persistent storage. It behaves identically to Redact but is named differently to clarify the caller's intent: this is read-path redaction for data that may have been stored before redaction was mandatory.

func RegisterVersionHandler 

func RegisterVersionHandler(mux *http.ServeMux, resolve VersionStoreResolver)

RegisterVersionHandler adds version management HTTP endpoints to the given ServeMux. All routes are prefixed with /api/versions. resolve is called once per request to obtain the tenant-scoped store; see VersionStoreResolver.

Endpoints: 

GET    /api/versions              — list all versions with metrics
GET    /api/versions/<name>       — list versions for a specific workflow
POST   /api/versions/<name>/<v>/deprecate     — mark version deprecated
POST   /api/versions/<name>/<v>/restore       — mark version active
POST   /api/versions/<name>/<v>/purge         — delete version permanently
GET    /api/versions/stale        — list stale version alerts
POST   /api/versions/gc           — run garbage collection

func ReplayStepCount 

func ReplayStepCount() int64

ReplayStepCount returns the total replay step count from the atomic counter.

func ResetPatterns 

func ResetPatterns()

ResetPatterns clears all custom patterns (for testing only).

func ResolveChildVersion 

func ResolveChildVersion(ctx context.Context, db *sql.DB,
	childName string, parentVersion int, opts ChildWorkflowOptions) (int, error)

ResolveChildVersion determines the version for a child workflow. Rules in priority order:

  1. Explicit pin: opts.Version > 0 → use that version.
  2. Parent's version: child uses same version as parent (default when opts.Version <= 0). This ensures tightly-coupled workflows stay on compatible versions.
  3. Latest compatible: if the parent's version does not exist for the child workflow name, fall back to: SELECT MAX(version) FROM workflow_defs WHERE name = $1 AND min_version <= parentVersion AND NOT deprecated

Args:

  • db: database connection (may be nil; if nil, rule 3 is skipped)
  • childName: the workflow definition name of the child
  • parentVersion: the version of the parent workflow
  • opts: version options from the caller

Returns the resolved version number, or an error if no version can be found.

func ResolvePlugins 

func ResolvePlugins(ctx context.Context, db *sql.DB, pluginDepsJSON string) (map[string]string, error)

ResolvePlugins resolves all plugin dependencies to specific versions. Takes the plugin_deps JSON from workflow_defs (format: {"llm": ">=1.2.0"}), queries plugin_defs, and returns pinned version strings.

Resolution strategy:

  1. For each plugin dependency, find all non-deprecated versions from plugin_defs ordered by created_at DESC (newest first).
  2. Return the newest version that satisfies the constraint.
  3. If no version satisfies the constraint, return an error.

func RunsOnWasmtime 

func RunsOnWasmtime(lang string) bool

RunsOnWasmtime reports whether a detected guest language is served by the wasmtime backend.

Callers that build an Engine should register backends from WasmtimeLanguages rather than consulting this; it exists for the worker, which also has to decide whether to construct a wazero Runtime at all. Those two decisions have to agree, and reading them from the same list is what makes them agree.

func UpdateNowMs 

func UpdateNowMs()

UpdateNowMs sets the global Now() seed to the current wall-clock time. Call this at worker startup and periodically (e.g., on each poll cycle) so that fresh workflow execution sessions see a reasonable wall-clock time. During replay, the seed is overridden from event history timestamps.

func ValidateCronExpr 

func ValidateCronExpr(expr string) error

ValidateCronExpr reports whether expr is a cron expression this engine can evaluate, returning nil if it is and a diagnostic naming the offending field if it is not.

This exists because the alternative was worse than useless. NextCronTime answers with a time.Time and has no way to say "that expression is nonsense", so it used to fall back to "24 hours from now" for anything it could not parse -- and its integer parser was fmt.Sscanf with the error discarded, so `abc` parsed as 0. `abc * * * *` therefore did not fail, and did not fall back either: it ran at minute zero of every hour, forever, having never told anyone. Rejecting up front is the only way a caller can report the problem to whoever typed it.

func ValidateMisfirePolicy 

func ValidateMisfirePolicy(p string) error

ValidateMisfirePolicy reports whether p is a misfire policy the scheduler understands. Empty is valid and means MisfireCatchUp.

func ValidateOverlapPolicy 

func ValidateOverlapPolicy(p string) error

ValidateOverlapPolicy reports whether p is an overlap policy the scheduler understands. Empty is valid and means OverlapAllow.

func ValidateTimezone 

func ValidateTimezone(name string) error

ValidateTimezone reports whether name is an IANA timezone this process can load, returning nil if it is.

Worth knowing where the answer comes from: time.LoadLocation reads the system zoneinfo database, so on a container image without tzdata installed every name except "UTC" and "Local" fails. cmd/cleat-worker imports _ "time/tzdata" so the worker carries its own copy and does not depend on the base image -- see the comment there.

func ValidateVersionCompatibility 

func ValidateVersionCompatibility(oldDef, newDef *WorkflowDef) error

ValidateVersionCompatibility checks whether a workflow instance running the old definition can be migrated to the new definition via ContinueAsNewWithVersion. Returns an error describing the first incompatibility found, or nil if the migration is safe.

Types

type AWSSecretsManagerProvider 

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

AWSSecretsManagerProvider resolves the connection string by calling the AWS CLI: 

aws secretsmanager get-secret-value --secret-id <name> --query SecretString --output text

func NewAWSSecretsManagerProvider 

func NewAWSSecretsManagerProvider(credentialPath string) *AWSSecretsManagerProvider

NewAWSSecretsManagerProvider creates an AWSSecretsManagerProvider. The credentialPath is the secret name or ARN in AWS Secrets Manager.

func (*AWSSecretsManagerProvider) GetConnectionString 

func (p *AWSSecretsManagerProvider) GetConnectionString(ctx context.Context) (string, error)

GetConnectionString runs `aws secretsmanager get-secret-value` and returns the connection string.

type AdaptiveFlusher 

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

AdaptiveFlusher tracks the recent step rate and automatically switches between direct (low-rate) and batched (high-rate) event persistence.

In batch mode, events accumulate in a mutex-protected slice. When the batch reaches maxBatch or maxWait elapses, the goroutine that triggers the flush does the DB work directly — no channels, no worker pool.

func NewAdaptiveFlusher 

func NewAdaptiveFlusher(db *sql.DB, tenantID string, maxWait time.Duration, maxBatch int, enterThreshold, exitThreshold float64, _ int) *AdaptiveFlusher

func (*AdaptiveFlusher) Flush 

func (af *AdaptiveFlusher) Flush(ctx context.Context, workflowID string, rec EventRecord, checksum string, workerID string, generation int64) (chan error, bool)

Flush is called from recordEvent. In direct mode it returns (nil, false) and the caller falls through to flushEvent. In batch mode it returns a done channel; the caller blocks on <-done until the batch is persisted.

workerID and generation are the claiming worker's identity, for fencing (B4) -- the same values Engine.flushEvent passes to Heartbeat, threaded through here because a single AdaptiveFlusher batches events from every workflow this worker process is running, each under its own generation. Pass workerID == "" to opt out, matching Engine.fencingEnabled() == false.

func (*AdaptiveFlusher) GetRate 

func (af *AdaptiveFlusher) GetRate() float64

func (*AdaptiveFlusher) InBatchMode 

func (af *AdaptiveFlusher) InBatchMode() bool

func (*AdaptiveFlusher) Run 

func (af *AdaptiveFlusher) Run(ctx context.Context)

Run is a no-op in this design — no background goroutines needed.

func (*AdaptiveFlusher) SetEncryption 

func (af *AdaptiveFlusher) SetEncryption(encrypt bool, enc *PayloadEncryption)

func (*AdaptiveFlusher) Stats 

func (af *AdaptiveFlusher) Stats() (int64, int64, int64)

type AdminActionEvent 

type AdminActionEvent struct {
	Action   string // "force_complete", "force_fail", "re_replay"
	Operator string // identity from auth context
	Reason   string // optional detail
	// contains filtered or unexported fields
}

AdminActionEvent records an administrative action performed on a workflow.

func (AdminActionEvent) Step 

func (e AdminActionEvent) Step() int

func (AdminActionEvent) Type 

func (e AdminActionEvent) Type() EventType

type AmbiguityResolver 

type AmbiguityResolver interface {
	// ResolveCall reports the outcome of the operation identified by
	// idempotencyKey, which is the key the original attempt sent.
	//
	// resolved=false means "cannot say" and is not an error: the service may
	// have no record, or no way to look one up. The engine reports the
	// ambiguity to the workflow, exactly as it does without a resolver.
	//
	// An error means the lookup itself failed. It is treated the same as
	// "cannot say" -- an unreachable resolver must not turn a recoverable
	// ambiguity into a different failure -- but it is logged, because a
	// resolver that always errors is indistinguishable from one that never
	// resolves anything.
	ResolveCall(ctx context.Context, service, operation, idempotencyKey string) (response string, resolved bool, err error)
}

AmbiguityResolver answers the question a crash leaves open: did the call actually happen, and what did it return?

Detection on its own converts a rare silent duplicate into a rare permanent failure, which for some workloads is worse. A workflow that learns its outcome is unknown, and has no way to find out, is stuck. This is the way out that costs nothing when it is not needed: most services that accept an idempotency key can also be asked what happened to one.

type App 

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

App manages the lifecycle of a cleat application and its plugins.

func NewApp 

func NewApp(ctx context.Context, config AppConfig) (*App, error)

NewApp creates a new cleat application, initializing plugins in order. Each plugin's Init is called with an Environment that includes the plugin's JSON configuration (from PluginConfigs) and the shared HTTP mux and logger. If a plugin implements HasHostFunctions, its host functions are registered automatically.

func (*App) Close 

func (a *App) Close() error

Close shuts down all plugins in reverse initialization order. Plugins that implement CloseablePlugin have their Close method called.

func (*App) Logger 

func (a *App) Logger() *slog.Logger

Logger returns the application logger.

func (*App) Mux 

func (a *App) Mux() *http.ServeMux

Mux returns the HTTP serve mux used by plugins.

func (*App) Registry 

func (a *App) Registry() *PluginRegistry

Registry returns the plugin function registry.

func (*App) StreamRegistry 

func (a *App) StreamRegistry() *PluginStreamRegistry

StreamRegistry returns the stream plugin function registry.

type AppConfig 

type AppConfig struct {
	// Runtime is the WASM runtime for executing workflows. Required.
	Runtime *Runtime

	// StoreFactory creates workflow stores for database access. Required.
	StoreFactory StoreFactory

	// ServiceCaller makes external API calls on behalf of workflows. Required.
	ServiceCaller ServiceCaller

	// Plugins is the list of plugins to load at startup.
	Plugins []plugin.Plugin

	// PluginConfigs maps plugin name to JSON configuration.
	// The config is passed to the plugin via Environment.Config.
	PluginConfigs map[string]json.RawMessage

	// Logger is the root logger for the application. If nil, a default
	// logger is created.
	Logger *slog.Logger

	// Mux is the HTTP serve mux for plugin route registration.
	Mux *http.ServeMux

	// Registry is the plugin function registry. If nil, a new one is created.
	Registry *PluginRegistry

	// StreamRegistry is the stream plugin function registry. If nil, a new one is created.
	StreamRegistry *PluginStreamRegistry
}

AppConfig configures a cleat application.

type AwaitAllChildrenEvent 

type AwaitAllChildrenEvent struct {
	RunIDsJSON   string
	OutcomesJSON string
	// contains filtered or unexported fields
}

AwaitAllChildrenEvent records the result of awaiting all child workflows.

func (AwaitAllChildrenEvent) Step 

func (e AwaitAllChildrenEvent) Step() int

func (AwaitAllChildrenEvent) Type 

func (e AwaitAllChildrenEvent) Type() EventType

type AwaitChildEvent 

type AwaitChildEvent struct {
	RunID    string
	Response string
	Err      string
	// contains filtered or unexported fields
}

AwaitChildEvent records the result of awaiting a single child workflow.

func (AwaitChildEvent) Step 

func (e AwaitChildEvent) Step() int

func (AwaitChildEvent) Type 

func (e AwaitChildEvent) Type() EventType

type AwaitPromiseEvent 

type AwaitPromiseEvent struct {
	PromiseID string
	// contains filtered or unexported fields
}

AwaitPromiseEvent records that a workflow began awaiting a promise.

func (AwaitPromiseEvent) Step 

func (e AwaitPromiseEvent) Step() int

func (AwaitPromiseEvent) Type 

func (e AwaitPromiseEvent) Type() EventType

type AwaitSignalsEvent 

type AwaitSignalsEvent struct {
	SignalNames string
	TimeoutMs   int64
	// contains filtered or unexported fields
}

AwaitSignalsEvent records the start of a signal-waiting period.

func (AwaitSignalsEvent) Step 

func (e AwaitSignalsEvent) Step() int

func (AwaitSignalsEvent) Type 

func (e AwaitSignalsEvent) Type() EventType

type CacheStats 

type CacheStats struct {
	Size    int   `json:"size"`
	MaxSize int   `json:"max_size"`
	Hits    int64 `json:"hits"`
	Misses  int64 `json:"misses"`
}

CacheStats exposes cache performance for observability.

type CallEvent 

type CallEvent struct {
	Service  string
	Op       string
	Request  string
	Response string
	Err      string
	// contains filtered or unexported fields
}

CallEvent records a durable service call (the most common event type).

func (CallEvent) Step 

func (e CallEvent) Step() int

func (CallEvent) Type 

func (e CallEvent) Type() EventType

type CallRecord 

type CallRecord = EventRecord

CallRecord is kept for backward compatibility in tests.

type CallSemantics 

type CallSemantics int

CallSemantics is the durability guarantee an operation asks for.

The right guarantee depends on the operation and only the workflow author knows which one applies: a GET is safe to repeat, a card charge is not, and a card charge with an idempotency key is. One global policy cannot be correct for all three, and the costs differ by an order of magnitude -- so this is declared per operation, with a default that is exactly today's behaviour. 

const (
	// AtLeastOnce is the default and costs nothing extra: dispatch, then
	// record. A crash in between re-executes the call on replay. This is what
	// docs/durable-calls.md has always documented.
	AtLeastOnce CallSemantics = iota

	// WriteAheadIntent commits a pending row before dispatching, so replay
	// after a crash can report the outcome as ambiguous rather than silently
	// repeating the call. Costs one extra synchronous round trip per call.
	WriteAheadIntent
)

type ChildWorkflowEvent 

type ChildWorkflowEvent struct {
	DefName          string
	Input            string
	ChildID          string
	ParentWorkflowID string
	// contains filtered or unexported fields
}

ChildWorkflowEvent records the start of a child workflow.

func (ChildWorkflowEvent) Step 

func (e ChildWorkflowEvent) Step() int

func (ChildWorkflowEvent) Type 

func (e ChildWorkflowEvent) Type() EventType

type ChildWorkflowOptions 

type ChildWorkflowOptions struct {
	// Version is the explicit child workflow version to use.
	//   > 0  : use this version explicitly
	//   <= 0 : use default resolution (parent's version, or latest compatible)
	Version int `json:"version"`
}

ChildWorkflowOptions carries version resolution configuration for spawning a child workflow. Passed through the WASM ABI from SDKs.

type ChildWorkflowStore 

type ChildWorkflowStore interface {
	// StartChildWorkflow creates a child workflow instance linked to a parent.
	// defVersion is the explicit workflow definition version to use, or 0 to use
	// default resolution (SELECT MAX(version)).
	StartChildWorkflow(ctx context.Context, parentID, defName, inputJSON string, defVersion int, parentClosePolicy string, priority int) (string, error)

	// StartChildWorkflowAtomic creates a child workflow and records the parent's
	// child_workflow event in a single database transaction, guaranteeing
	// exactly-once creation even if the worker crashes mid-execution.
	// The event is written to event_history atomically with the child row.
	// The caller should still append the event to the in-memory history for
	// same-execution replay. The later event flush will skip it via
	// ON CONFLICT (workflow_id, step) DO NOTHING.
	StartChildWorkflowAtomic(ctx context.Context, childID, parentID, defName, inputJSON string, defVersion int, parentClosePolicy string, event EventRecord, priority int) (runID string, err error)

	GetChildResult(ctx context.Context, runID string) (resultJSON string, completed bool, err error)

	// ResolveVersionByTag resolves a workflow version by tag name (e.g. "stable", "canary").
	// Returns the version number and nil error on success, or 0 and an error if the tag
	// is not found.
	ResolveVersionByTag(ctx context.Context, workflowName string, tag string) (int, error)
}

type CleatError 

type CleatError struct {
	Code       ErrorCode
	Op         string // operation that failed
	WorkflowID string
	Err        error // underlying error
}

CleatError is a typed error with classification for retry decisions.

func NewAmbiguousError 

func NewAmbiguousError(op, workflowID string, err error) *CleatError

NewAmbiguousError creates an ambiguous-outcome error — the call may have succeeded but the response was never persisted. The caller should check the external service before retrying.

func NewCancelledError 

func NewCancelledError(op, workflowID string, err error) *CleatError

NewCancelledError creates a cancellation error.

func NewPermanentError 

func NewPermanentError(op, workflowID string, err error) *CleatError

NewPermanentError creates a non-retryable error (invalid input, not found).

func NewRetriesExhaustedError 

func NewRetriesExhaustedError(op, workflowID string, err error) *CleatError

NewRetriesExhaustedError creates an error indicating retries were exhausted.

func NewTimeoutError 

func NewTimeoutError(op, workflowID string, err error) *CleatError

NewTimeoutError creates a timeout error.

func NewTransientError 

func NewTransientError(op, workflowID string, err error) *CleatError

NewTransientError creates a retryable error (DB connection, timeout).

func (*CleatError) Error 

func (e *CleatError) Error() string

func (*CleatError) Retryable 

func (e *CleatError) Retryable() bool

Retryable returns true if the error is transient and can be retried.

func (*CleatError) Unwrap 

func (e *CleatError) Unwrap() error

type CloseablePlugin 

type CloseablePlugin interface {
	Close() error
}

CloseablePlugin is an optional interface for plugins that need cleanup.

type CompactedChild 

type CompactedChild struct {
	RunID string `json:"run_id"`
	Name  string `json:"name"`
	Input string `json:"input"`
}

CompactedChild represents a child workflow started in the compacted portion that is still running (not yet completed).

type CompactedDefer 

type CompactedDefer struct {
	ID          string `json:"id"`
	Description string `json:"description"`
}

CompactedDefer represents a deferred cleanup callback registered in the compacted portion of history that has not yet been executed.

type CompactedEvent 

type CompactedEvent struct {
	Type          int    `json:"t"`
	Service       string `json:"svc,omitempty"`
	Op            string `json:"op,omitempty"`
	Request       string `json:"req,omitempty"`
	Response      string `json:"resp,omitempty"`
	Error         string `json:"err,omitempty"`
	DurationMs    int64  `json:"dur,omitempty"`
	SignalNames   string `json:"sigs,omitempty"`
	TimeoutMs     int64  `json:"to,omitempty"`
	SignalName    string `json:"sn,omitempty"`
	SignalPayload string `json:"sp,omitempty"`
	DeferID       string `json:"did,omitempty"`
	DeferDesc     string `json:"dd,omitempty"`
	ChildName     string `json:"cn,omitempty"`
	ChildInput    string `json:"ci,omitempty"`
	RunID         string `json:"rid,omitempty"`
	NewInput      string `json:"ni,omitempty"`
	PromiseName   string `json:"prom_name,omitempty"`
	PromiseID     string `json:"prom_id,omitempty"`
	PromiseResult string `json:"prom_res,omitempty"`
	PromiseError  string `json:"prom_err,omitempty"`

	// ErrNonRetryable mirrors EventRecord.ErrNonRetryable for a call event
	// (call or call_heartbeat -- both use EventTypeCall). Its omission here
	// was S4a: without it, a compacted-region call classified non-retryable
	// on first execution could only reconstruct as the Go zero value, false,
	// and replayed as an ordinary retryable failure. omitempty is safe for
	// this bool for the same reason it is on EventRecord itself: the zero
	// value already matches "not recorded" / "retryable", so a missing key
	// on decode reads back exactly as it always did pre-2.35.
	ErrNonRetryable bool `json:"nr,omitempty"`

	// TimestampMs mirrors EventRecord.TimestampMs and is set for every event
	// type, not just the ones with a dedicated case below -- Now() during
	// replay reads it off the *previous* history event (execSession.Now,
	// lifecycle.go), so any compacted event with TimestampMs left at zero
	// makes replay fall back to session-start or wall-clock time instead of
	// the recorded virtual clock. That is engine-introduced non-determinism
	// exactly like S4a, just triggered by Now() instead of Retryable().
	TimestampMs int64 `json:"ts,omitempty"`

	// NewVersion mirrors EventRecord.NewVersion for a continue_as_new event.
	// ContinueAsNewWithVersion reads rec.NewVersion straight off the replayed
	// event to decide which workflow version to restart as (lifecycle.go); a
	// missing value silently restarts version 0 (== "current version"),
	// which can run the wrong code for a versioned continue-as-new that fell
	// into the compacted region.
	NewVersion int `json:"nv,omitempty"`

	// StreamChunkIndex and StreamFinish mirror the same-named EventRecord
	// fields for a plugin_call_stream_chunk event. plugins.go reads them
	// back during replay to reconstruct stream position and completion;
	// without them every compacted stream chunk replays as chunk 0,
	// unfinished.
	StreamChunkIndex int  `json:"sci,omitempty"`
	StreamFinish     bool `json:"scf,omitempty"`

	// StateKeys mirrors EventRecord.StateKeys for a state_mutation event with
	// StateOp=="list" (ListState). lifecycle.go's ListState replay path reads
	// it back verbatim; the other state_mutation ops (set/increment/has)
	// never populate it.
	StateKeys string `json:"sks,omitempty"`

	// ParentWorkflowID and ParentClosePolicy mirror the same-named
	// EventRecord fields for a child_workflow event. Neither is read back
	// from history during replay (children.go's replay path only consumes
	// RunID); ParentClosePolicy is enforced against live store state when a
	// parent terminates (store_lifecycle.go's enforceParentClosePolicy).
	// Preserved anyway so the reconstructed event is a faithful copy rather
	// than a replay-sufficient-but-lossy one, and so the general round-trip
	// property test (TestEventRecordFieldsSurviveCompaction) does not need a
	// bespoke exemption for a field that is, in fact, cheap to carry through.
	ParentWorkflowID  string `json:"pwid,omitempty"`
	ParentClosePolicy string `json:"pcp,omitempty"`

	// Plugin call fields.
	PluginName   string `json:"pn,omitempty"`
	PluginFunc   string `json:"pf,omitempty"`
	PluginInput  string `json:"pi,omitempty"`
	PluginOutput string `json:"po,omitempty"`
	PluginError  string `json:"pe,omitempty"`
}

CompactedEvent is a sparse representation of a single event in the compacted portion of a workflow's history. Only fields relevant to the event type are populated; omitempty keeps JSONB storage compact. Short JSON tags minimize storage size for workflows with thousands of compacted events.

type CompactionState 

type CompactionState struct {
	Version       int               `json:"version"`
	CompactedStep int               `json:"compacted_step"`
	Events        []CompactedEvent  `json:"events"`
	PendingDefers []CompactedDefer  `json:"pending_defers,omitempty"`
	OpenChildren  []CompactedChild  `json:"open_children,omitempty"`
	QueryState    map[string]string `json:"query_state,omitempty"`

	// Summary is populated when Events exceeds DefaultMaxCompactedEvents and is
	// truncated. It records the truncation count for observability.
	Summary *TruncationSummary `json:"summary,omitempty"`
}

CompactionState holds the minimal state needed to reconstruct the compacted portion of a workflow's event history for deterministic replay.

type ConcurrencyKeyInfo 

type ConcurrencyKeyInfo struct {
	KeyHash    []byte    `json:"key_hash"`
	KeyText    string    `json:"key_text"`
	WorkflowID string    `json:"workflow_id"`
	AcquiredAt time.Time `json:"acquired_at"`
	ExpiresAt  time.Time `json:"expires_at"`
}

ConcurrencyKeyInfo holds the state of an acquired concurrency key.

type ConcurrencyKeyStore 

type ConcurrencyKeyStore interface {
	// AcquireConcurrencyKey tries to acquire a concurrency key for a workflow.
	// Returns true if acquired, false if already held by another workflow.
	// Automatically releases expired keys during acquisition.
	AcquireConcurrencyKey(ctx context.Context, key, workflowID string, ttl time.Duration) (acquired bool, err error)

	// ReleaseConcurrencyKey releases a specific concurrency key.
	ReleaseConcurrencyKey(ctx context.Context, key string) error
}

ConcurrencyKeyStore provides concurrency key acquisition and release for the engine.

type ContinueAsNewEvent 

type ContinueAsNewEvent struct {
	NewInput string
	// contains filtered or unexported fields
}

ContinueAsNewEvent records a continue-as-new instruction.

func (ContinueAsNewEvent) Step 

func (e ContinueAsNewEvent) Step() int

func (ContinueAsNewEvent) Type 

func (e ContinueAsNewEvent) Type() EventType

type CreatePromiseEvent 

type CreatePromiseEvent struct {
	PromiseName string
	PromiseID   string
	// contains filtered or unexported fields
}

CreatePromiseEvent records the creation of a durable promise.

func (CreatePromiseEvent) Step 

func (e CreatePromiseEvent) Step() int

func (CreatePromiseEvent) Type 

func (e CreatePromiseEvent) Type() EventType

type CrossSchemaChildStore 

type CrossSchemaChildStore interface {
	ChildWorkflowStore

	// StartChildWorkflowInSchema creates a child workflow in the given target schema.
	// The schema must be part of the engine's configured peerSchemas.
	StartChildWorkflowInSchema(ctx context.Context, targetSchema, parentID, defName, inputJSON string, defVersion int, parentClosePolicy string, priority int) (string, error)

	// GetChildResultInSchema polls a child workflow in the given target schema.
	GetChildResultInSchema(ctx context.Context, targetSchema, runID string) (resultJSON string, completed bool, err error)
}

CrossSchemaChildStore is an optional extension to ChildWorkflowStore for starting child workflows in a different PostgreSQL schema. This enables cross-instance workflow cooperation: an instance in schema A can start a child workflow in schema B, and the B worker pool picks it up.

type CrossTenantCapability 

type CrossTenantCapability struct {
	Claim           bool
	ClaimReason     string
	Schedules       bool
	SchedulesReason string
}

CrossTenantCapability reports whether a store can actually honour the cross-tenant paths, and why not when it cannot.

Both fields are answered independently because 023 and 024 are separate migrations with separate grants: a deployment can execute workflows for every tenant and still fire cron for only one.

type CrossTenantCapabilityChecker 

type CrossTenantCapabilityChecker interface {
	CheckCrossTenantCapability(ctx context.Context) CrossTenantCapability
}

CrossTenantCapabilityChecker is implemented by stores that can answer "would the cross-tenant paths work" WITHOUT exercising them.

Exercising them is not an option at startup. The claim is a write: running it to see whether it works would claim real workflows into a worker whose dispatch loop has not started, leaving them 'running' with no executor until the lease expires. So each dialect answers from its own catalog instead.

type CrossTenantClaimer 

type CrossTenantClaimer interface {
	ClaimWorkflowsAcrossTenants(ctx context.Context, workerID string, limit int) ([]*WorkflowInstance, error)
}

CrossTenantClaimer is implemented by stores that can claim runnable work for every tenant in a single query.

It is deliberately NOT part of WorkflowStore. A store that cannot do it -- because its dialect has no mechanism, or because the deployment has not granted one -- simply does not implement it, and the caller falls back to the ordinary tenant-scoped claim. Putting it on WorkflowStore would force every implementation and every test double to answer a question most of them have no business answering.

The returned instances carry TenantID, and the caller MUST re-scope to it before touching anything else. That is the whole bargain: one query sees across tenants so the dispatch loop does not have to poll each one, and everything downstream of it is scoped again immediately. See cmd/cleat-worker's storeForTenant.

type CrossTenantScheduleReader 

type CrossTenantScheduleReader interface {
	GetDueSchedulesAcrossTenants(ctx context.Context) ([]Schedule, error)
}

CrossTenantScheduleReader is implemented by stores that can read due schedules for every tenant in a single query.

Separate from CrossTenantClaimer rather than folded into it, because the two are provisioned separately and a deployment can legitimately have one and not the other: 023 grants the claim, 024 grants this, and a database that applied only the first should get a working dispatch loop and a loud warning about schedules rather than a worker that refuses both.

The returned schedules carry TenantID, and the caller MUST re-scope to it before doing anything else -- starting the run, and claiming the schedule. The widened view covers the read and nothing after it. See cmd/cleat-worker's scheduleLoop.

type DBCredentialProvider 

type DBCredentialProvider interface {
	// GetConnectionString returns the database connection string. If the
	// provider cannot resolve it the returned error explains why.
	GetConnectionString(ctx context.Context) (string, error)
}

DBCredentialProvider resolves a database connection string from a configured source. Implementations may read from environment variables, a secrets store, or a CLI tool.

func NewDBCredentialProvider 

func NewDBCredentialProvider(providerName, dbURL, credentialPath string) (DBCredentialProvider, error)

NewDBCredentialProvider creates the appropriate DBCredentialProvider based on providerName. Supported values: "env" (default), "vault", "aws-secrets-manager". When providerName is "env", the dbURL argument is passed to the env provider. For "vault" and "aws-secrets-manager", credentialPath must be non-empty.

type DBEventStream 

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

DBEventStream loads event history from the database on demand as the replay step counter advances. Instead of loading the entire history into memory, it fetches pages of events as needed.

The stream maintains a sliding window of loaded events. As the replay advances past the current page, the next page is fetched automatically.

func NewDBEventStream 

func NewDBEventStream(db *sql.DB, workflowID string, pageSize int) *DBEventStream

NewDBEventStream creates a new DBEventStream for the given workflow. The first page of events is loaded lazily on the first At() call.

func (*DBEventStream) Append 

func (s *DBEventStream) Append(rec EventRecord)

func (*DBEventStream) At 

func (s *DBEventStream) At(i int) *EventRecord

func (*DBEventStream) Close 

func (s *DBEventStream) Close() error

func (*DBEventStream) Len 

func (s *DBEventStream) Len() int

func (*DBEventStream) Slice 

func (s *DBEventStream) Slice(start, end int) []EventRecord

func (*DBEventStream) Total 

func (s *DBEventStream) Total() (int, error)

type DeferEvent 

type DeferEvent struct {
	Description string
	DeferID     string
	// contains filtered or unexported fields
}

DeferEvent records a registered defer callback.

func (DeferEvent) Step 

func (e DeferEvent) Step() int

func (DeferEvent) Type 

func (e DeferEvent) Type() EventType

type Dialect 

type Dialect string

Dialect identifies the SQL dialect of a database backend. 

const (
	DialectPostgres Dialect = "postgres"
	DialectMySQL    Dialect = "mysql"
	DialectMSSQL    Dialect = "mssql"
)

type Engine 

type Engine struct {
	Metrics *prometheus.Metrics
	// contains filtered or unexported fields
}

Engine provides cleat execution semantics (Execute/Replay) on top of a Runtime using a checkpoint/replay model.

func NewEngine 

func NewEngine(rt *Runtime, caller ServiceCaller, opts ...EngineOption) *Engine

NewEngine creates an Engine backed by the given Runtime and ServiceCaller.

func (*Engine) CallerHonoursIdempotencyKeys 

func (e *Engine) CallerHonoursIdempotencyKeys() bool

CallerHonoursIdempotencyKeys reports whether this engine's caller receives idempotency keys.

Exported for operators and tests: whether a deployment has exactly-once behaviour against a key-honouring service is otherwise invisible until a crash duplicates something in production.

func (*Engine) DB 

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

DB returns the tenant-scoped database connection.

func (*Engine) DispatchUpdate 

func (e *Engine) DispatchUpdate(ctx context.Context, name, payload string) (string, error)

DispatchUpdate dispatches an update to a workflow by invoking its registered handler. The handler receives the update name and payload JSON, and returns the result JSON. Returns an error if no update handler is configured on the engine.

func (*Engine) EncryptSensitivePayloads 

func (e *Engine) EncryptSensitivePayloads() bool

EncryptSensitivePayloads returns whether sensitive payload encryption is enabled.

func (*Engine) Encryption 

func (e *Engine) Encryption() *PayloadEncryption

Encryption returns the payload encryption instance.

func (*Engine) Execute 

func (e *Engine) Execute(ctx context.Context, wasmBytes []byte, entryPoint string, input json.RawMessage) (result string, history []EventRecord, suspended *SuspendResult, deferrals map[string]string, queryState map[string]string, err error)

Execute runs a fresh workflow execution. If the workflow suspends (sleep, await signals), it returns a nil result with non-nil SuspendResult. If the WASM binary uses the Component Model format, it decomposes it into constituent core modules following the component instance DAG.

func (*Engine) ExecuteCompiled 

func (e *Engine) ExecuteCompiled(ctx context.Context, compiled wazero.CompiledModule, entryPoint string, input json.RawMessage) (result string, history []EventRecord, suspended *SuspendResult, deferrals map[string]string, queryState map[string]string, err error)

ExecuteCompiled is like Execute but takes a pre-compiled module.

func (*Engine) Replay 

func (e *Engine) Replay(ctx context.Context, wasmBytes []byte, entryPoint string, input json.RawMessage, history []EventRecord) (result string, resultHistory []EventRecord, suspended *SuspendResult, deferrals map[string]string, queryState map[string]string, err error)

Replay replays a workflow from existing event history. Cached results are returned for matching steps; divergence triggers an error. queryState contains key-value state set via SetQueryState during execution.

func (*Engine) ReplayCompiled 

func (e *Engine) ReplayCompiled(ctx context.Context, compiled wazero.CompiledModule, entryPoint string, input json.RawMessage, history []EventRecord) (result string, resultHistory []EventRecord, suspended *SuspendResult, deferrals map[string]string, queryState map[string]string, err error)

ReplayCompiled is like Replay but takes a pre-compiled module. Use this when the module has already been compiled and cached by a WorkflowLoader, avoiding redundant compilation.

func (*Engine) RunDefer 

func (e *Engine) RunDefer(ctx context.Context, wasmBytes []byte, deferName string, input json.RawMessage) (string, error)

RunDefer invokes a defer cleanup function in the WASM module. This is called by the worker on workflow exit (after the main entry point returns) to run registered defer callbacks in LIFO order.

func (*Engine) RunDeferCompiled 

func (e *Engine) RunDeferCompiled(ctx context.Context, compiled wazero.CompiledModule, deferName string, input json.RawMessage) (string, error)

RunDeferCompiled is like RunDefer but takes a pre-compiled module.

func (*Engine) TenantID 

func (e *Engine) TenantID() string

TenantID returns the tenant identifier.

type EngineOption 

type EngineOption func(*Engine)

EngineOption configures an Engine.

func WithAllowVersionMismatch 

func WithAllowVersionMismatch(allow bool) EngineOption

WithAllowVersionMismatch allows replay despite version compatibility failures.

func WithAmbiguityResolver 

func WithAmbiguityResolver(r AmbiguityResolver) EngineOption

WithAmbiguityResolver sets the resolver consulted when replay finds a call that was dispatched but whose outcome was never recorded.

func WithBackend 

func WithBackend(language string, backend WasmBackend) EngineOption

WithBackend registers a WasmBackend for a language key.

func WithBackends 

func WithBackends(languages []string, backend WasmBackend) EngineOption

WithBackends registers one backend for several languages at once.

backendForWasm looks the detected language up in this map and returns nil when it is absent, which sends the workflow to the fallback runtime. So the set of languages registered here is the whole of the routing decision, and registering them one call at a time made it easy to leave one out silently.

func WithCancellationCheckInterval 

func WithCancellationCheckInterval(d time.Duration) EngineOption

WithCancellationCheckInterval sets the minimum wall-clock interval between PollCancellation DB queries. Zero (the default) checks on every durable step.

func WithChildBindingOverride 

func WithChildBindingOverride(override string) EngineOption

WithChildBindingOverride overrides the child binding policy for debugging. For example, "latest" forces resolution to the latest version regardless of the compiled-in policy. This is a worker-level, cross-tenant setting intended for development environments only.

func WithChildBindingPolicy 

func WithChildBindingPolicy(policy string) EngineOption

WithChildBindingPolicy sets the child binding policy from WASM metadata. The policy determines how child workflow versions are resolved at runtime:

  • "frozen" — use pinned ChildVersions from the metadata
  • "stable" — resolve to the version tagged "stable" at child creation time
  • "latest" — always resolve to MAX(version)
  • "" (empty) — will be inferred by EffectivePolicy: "frozen" if ChildVersions present, else "latest"

func WithChildWorkflowStore 

func WithChildWorkflowStore(cws ChildWorkflowStore) EngineOption

WithChildWorkflowStore sets the child workflow store.

func WithCompactionState 

func WithCompactionState(cs *CompactionState) EngineOption

WithCompactionState sets the compaction state.

func WithConcurrencyKeyStore 

func WithConcurrencyKeyStore(cks ConcurrencyKeyStore) EngineOption

func WithContinueAsNewHandler 

func WithContinueAsNewHandler(fn func(ctx context.Context, currentRunID, workerID string, generation int64, defName string, defVersion int, newInput string, newEvents []EventRecord, result string, queryState map[string]string, priority int) (newRunID string, err error)) EngineOption

WithContinueAsNewHandler sets a handler for atomic ContinueAsNew transitions.

func WithDB 

func WithDB(db *sql.DB) EngineOption

WithDB sets a tenant-scoped DB connection.

func WithDefName 

func WithDefName(name string) EngineOption

WithDefName sets the workflow definition name.

func WithDefVersion 

func WithDefVersion(v int) EngineOption

WithDefVersion sets the workflow definition version.

func WithDefaultWorkflowTimeout 

func WithDefaultWorkflowTimeout(d time.Duration) EngineOption

WithDefaultWorkflowTimeout sets the total workflow timeout.

func WithEncryption 

func WithEncryption(enc *PayloadEncryption, enabled bool) EngineOption

WithEncryption sets encryption at rest for sensitive event payloads.

func WithFetcher 

func WithFetcher(f Fetcher) EngineOption

WithFetcher sets the HTTP fetcher.

func WithFlusherRegistry 

func WithFlusherRegistry(r *TenantFlusherRegistry) EngineOption

WithFlusherRegistry sets the tenant-keyed adaptive batch flusher registry. When set, recordEvent uses the tenant-specific flusher to decide between direct per-step flushing and batched event persistence.

func WithGeneration 

func WithGeneration(generation int64) EngineOption

WithGeneration sets the generation this workerID claimed the workflow instance under (workflow_instances.generation at claim time).

Paired with WithWorkerID, this is what lets the per-step flush path (engine/flush.go, engine/adaptive_flush.go) and the write-ahead-intent path (engine/store_intent.go) fence their writes the same way CompleteWorkflow/FailWorkflow/FinalizeWorkflowSegment already fence theirs: a write only lands if the claim it was made under is still current. See engine.fencingEnabled.

Deliberately not folded into WithWorkerID or inferred from it: a claimed workflow's generation is never 0 (ClaimWorkflows always runs `generation = generation + 1` before handing a workflow to an engine), so leaving this unset (generation stays its zero value) keeps fencing off by construction for every existing caller that constructs an Engine without going through a claim -- tests, cleattest, embedded, and any Engine built with WithWorkerID for an unrelated reason (e.g. continueAsNewHandler's hand-off identity). Fencing only turns on when both the worker identity and a real, non-zero generation are supplied together.

func WithInitialEventCount 

func WithInitialEventCount(n int) EngineOption

WithInitialEventCount sets the starting event count.

func WithLogger 

func WithLogger(l *slog.Logger) EngineOption

WithLogger sets the structured logger (default: slog.Default()).

func WithMaxQuotaChildren 

func WithMaxQuotaChildren(n int) EngineOption

WithMaxQuotaChildren sets the max child workflows per workflow.

func WithMaxQuotaConcurrencyKeys 

func WithMaxQuotaConcurrencyKeys(n int) EngineOption

WithMaxQuotaConcurrencyKeys sets the max concurrency keys per workflow.

func WithMaxQuotaEvents 

func WithMaxQuotaEvents(n int) EngineOption

WithMaxQuotaEvents sets the max events before quota/auto-ContinueAsNew.

func WithMaxQuotaSchedules 

func WithMaxQuotaSchedules(n int) EngineOption

WithMaxQuotaSchedules sets the max cron schedules a tenant may hold.

Unlike the other three quotas this one is per tenant, not per workflow: a schedule outlives the run that created it, so counting them against that run would let a workflow create its limit, exit, and be started again. Zero means unlimited, matching the others.

func WithMaxRetryAttempts 

func WithMaxRetryAttempts(n int) EngineOption

WithMaxRetryAttempts sets a ceiling on retry attempts.

func WithNoPerStepFlush 

func WithNoPerStepFlush(v bool) EngineOption

WithNoPerStepFlush disables per-step flushEvent calls. Events are still accumulated in the session history and persisted atomically by FinalizeWorkflowSegment. This improves throughput at the cost of losing in-flight events on crash.

func WithPeerSchemas 

func WithPeerSchemas(schemas []string) EngineOption

WithPeerSchemas sets peer schemas for cross-instance operations.

func WithPluginCallGuard 

func WithPluginCallGuard(g *PluginCallGuard) EngineOption

WithPluginCallGuard sets the plugin call guard.

func WithPluginCallObserver 

func WithPluginCallObserver(o PluginCallObserver) EngineOption

WithPluginCallObserver sets a post-invocation observer.

func WithPluginRegistry 

func WithPluginRegistry(pr *PluginRegistry) EngineOption

WithPluginRegistry sets the plugin registry.

func WithPluginStreamRegistry 

func WithPluginStreamRegistry(psr *PluginStreamRegistry) EngineOption

WithPluginStreamRegistry sets the streaming plugin registry.

func WithPromiseStore 

func WithPromiseStore(ps PromiseStore) EngineOption

WithPromiseStore sets the promise store.

func WithReplayStepCallback 

func WithReplayStepCallback(cb ReplayStepCallback) EngineOption

WithReplayStepCallback sets a callback invoked after each replayed event.

func WithRequireSignalAuth 

func WithRequireSignalAuth(v bool) EngineOption

WithRequireSignalAuth enables signal authorization checks.

func WithSchema 

func WithSchema(schema string) EngineOption

WithSchema sets the PostgreSQL schema name.

func WithSignalAuthCheck 

func WithSignalAuthCheck(fn func(ctx context.Context, targetWorkflowID, callerDefName string) error) EngineOption

WithSignalAuthCheck sets signal authorization function.

func WithSignalStore 

func WithSignalStore(ss SignalStore) EngineOption

WithSignalStore sets the signal store.

func WithTenantID 

func WithTenantID(id string) EngineOption

WithTenantID sets the tenant ID.

func WithTraceID 

func WithTraceID(id string) EngineOption

WithTraceID sets the W3C Trace Context trace-id.

func WithUpdateHandler 

func WithUpdateHandler(fn func(name, payload string) (string, error)) EngineOption

WithUpdateHandler sets the update handler function.

func WithVersionValidation 

func WithVersionValidation(fn func() error) EngineOption

WithVersionValidation sets version compatibility validation.

func WithWASMInstanceTimeout 

func WithWASMInstanceTimeout(d time.Duration) EngineOption

WithWASMInstanceTimeout sets the per-execution WAT timeout.

func WithWasmCumulativeAllocationMax 

func WithWasmCumulativeAllocationMax(maxBytes int64, counter *atomic.Int64) EngineOption

WithWasmCumulativeAllocationMax sets the max cumulative WASM linear memory allocation in bytes (0 = unlimited) and the shared atomic counter used to track current cumulative allocation across all concurrent engines.

func WithWorkerID 

func WithWorkerID(id string) EngineOption

WithWorkerID sets the worker instance identifier.

func WithWorkflowEventVerifier 

func WithWorkflowEventVerifier(fn func(ctx context.Context, workflowID string) error, failOnMismatch bool) EngineOption

WithWorkflowEventVerifier sets checksum verification for replay.

func WithWorkflowID 

func WithWorkflowID(id string) EngineOption

WithWorkflowID sets the workflow instance ID.

func WithWorkflowState 

func WithWorkflowState(ws WorkflowState) EngineOption

WithWorkflowState sets the workflow state for version info.

func WithWorkflowStore 

func WithWorkflowStore(store WorkflowStore) EngineOption

WithWorkflowStore sets the workflow store.

func WithWriteAheadIntentOps 

func WithWriteAheadIntentOps(ops ...string) EngineOption

WithWriteAheadIntentOps declares operations that must use WriteAheadIntent, as "service.operation" strings.

Declared on the engine rather than at the call site because the guest-facing ABI has no room for a per-call argument: adding one means changing the host function signature and every SDK that binds it. The design doc's open question 9.1 prefers "both, with the call site winning"; this is the half that can be built without an ABI change, and nothing here forecloses the other half.

type EnvCredentialProvider 

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

EnvCredentialProvider resolves the connection string from the --db flag, then the DATABASE_URL env var, then the CLEAT_DATABASE_URL env var. This is the default provider used when no --db-credential-provider is set.

func NewEnvCredentialProvider 

func NewEnvCredentialProvider(dbURL string) *EnvCredentialProvider

NewEnvCredentialProvider creates an EnvCredentialProvider. The dbURL argument is the value of the --db flag (may be empty).

func (*EnvCredentialProvider) GetConnectionString 

func (p *EnvCredentialProvider) GetConnectionString(_ context.Context) (string, error)

GetConnectionString resolves the connection string by checking the --db flag first, then DATABASE_URL, then CLEAT_DATABASE_URL.

type ErrorCode 

type ErrorCode int

ErrorCode classifies errors for retry decisions. 

const (
	ErrUnknown          ErrorCode = iota
	ErrTransient                  // retryable (DB connection, timeout)
	ErrPermanent                  // non-retryable (invalid input, not found)
	ErrCancelled                  // workflow cancelled
	ErrTimeout                    // execution timeout
	ErrAmbiguous                  // call outcome unknown after crash (replay found pending intent)
	ErrRetriesExhausted           // retries exhausted
)

func (ErrorCode) String 

func (c ErrorCode) String() string

String returns a human-readable representation of the error code suitable for storage in the error_code column.

type Event 

type Event interface {
	Step() int
	Type() EventType
}

Event is the interface for all typed workflow events. Each concrete event type carries only the fields relevant to that event, eliminating the need to consult a 30-field god struct.

func EventFromRecord 

func EventFromRecord(r EventRecord) Event

EventFromRecord converts a flat EventRecord to the appropriate typed Event. It returns nil for unrecognised event types.

func EventsFromRecords 

func EventsFromRecords(records []EventRecord) []Event

EventsFromRecords converts a slice of EventRecords to a slice of Events.

type EventRecord 

type EventRecord struct {
	Step      int       `json:"step"`
	EventType EventType `json:"type"`

	// TimestampMs is the virtual time (ms since Unix epoch) that Now()
	// should return after this event completes. For non-sleep events it
	// is the wall-clock time when the event was recorded. For sleep
	// events it is the pre-sleep time plus the sleep duration, encoding
	// the post-sleep virtual time for deterministic replay.
	TimestampMs int64 `json:"timestamp_ms"`

	// CreatedAt is the wall-clock time when the event was recorded in the
	// database. Used for timeline visualization; may be zero for events
	// loaded without timestamps or for events created before this field
	// was added.
	CreatedAt time.Time `json:"created_at,omitempty"`

	// Call fields.
	Service  string `json:"service,omitempty"`
	Op       string `json:"op,omitempty"`
	Request  string `json:"request,omitempty"`
	Response string `json:"response,omitempty"`
	Err      string `json:"err,omitempty"`

	// ErrNonRetryable records that Err was classified as non-retryable when
	// the call was first made, so replay can reproduce that classification
	// instead of guessing at it from the message string.
	//
	// This is one bit rather than the full error class IMPROVEMENT-PLAN 2.35
	// describes, and deliberately so: it is the only part of a classification
	// the engine can actually populate today. ServiceCaller returns a bare
	// `error`, and the sole machine-readable signal any implementation can
	// send is the optional RetryableError interface, which
	// isDefinitelyNonRetryable already honours. Recording a richer taxonomy
	// would mean inventing values no caller supplies.
	//
	// The zero value is the pre-2.35 behaviour: an event recorded before this
	// field existed carries no such key, reads back as false, and replays as
	// callFailureCode exactly as it always did. That is why this is a bool and
	// not a code -- a code field's zero value would collide with
	// callErrorUnknown, which is a real classification.
	ErrNonRetryable bool `json:"err_non_retryable,omitempty"`

	// Pending records that this event is a write-ahead call intent whose
	// outcome was never written: the external call was dispatched and the
	// process died before the response came back. It is read from the row --
	// intent_at IS NOT NULL AND checksum IS NULL -- and never written as part
	// of one, which is why it carries no json tag. See store_intent.go.
	//
	// It is a derived read rather than a stored flag on purpose. The two
	// columns are set and cleared by the same statements that write the
	// outcome, so they cannot disagree with it; a third column recording
	// "pending" could.
	Pending bool `json:"-"`

	// Sleep fields.
	DurationMs int64 `json:"duration_ms,omitempty"`

	// AwaitSignals fields.
	SignalNames   string `json:"signal_names,omitempty"`
	TimeoutMs     int64  `json:"timeout_ms,omitempty"`
	SignalName    string `json:"signal_name,omitempty"`
	SignalPayload string `json:"signal_payload,omitempty"`

	// Defer fields.
	DeferDescription string `json:"defer_description,omitempty"`
	DeferID          string `json:"defer_id,omitempty"`

	// Promise fields.
	PromiseName   string `json:"promise_name,omitempty"`
	PromiseID     string `json:"promise_id,omitempty"`
	PromiseResult string `json:"promise_result,omitempty"`
	PromiseError  string `json:"promise_error,omitempty"`

	// Child workflow fields.
	ChildName         string `json:"child_name,omitempty"`
	ChildInput        string `json:"child_input,omitempty"`
	RunID             string `json:"run_id,omitempty"`
	ParentWorkflowID  string `json:"parent_workflow_id,omitempty"`
	ParentClosePolicy string `json:"parent_close_policy,omitempty"`

	// ContinueAsNew fields.
	NewInput   string `json:"new_input,omitempty"`
	NewVersion int    `json:"new_version,omitempty"` // for versioned continue_as_new

	// Plugin call fields.
	PluginName   string `json:"plugin_name,omitempty"`
	PluginFunc   string `json:"plugin_func,omitempty"`
	PluginInput  string `json:"plugin_input,omitempty"`
	PluginOutput string `json:"plugin_output,omitempty"`
	PluginError  string `json:"plugin_error,omitempty"`
	Idempotent   bool   `json:"idempotent,omitempty"`

	// Stream chunk fields.
	StreamChunkIndex int  `json:"stream_chunk_index,omitempty"`
	StreamFinish     bool `json:"stream_finish,omitempty"`

	// Update handler fields.
	UpdateHandlerName string `json:"update_handler_name,omitempty"`
	UpdatePayload     string `json:"update_payload,omitempty"`
	UpdateResponse    string `json:"update_response,omitempty"`
	UpdateError       string `json:"update_error,omitempty"`

	// State mutation fields.
	StateKey   string `json:"state_key,omitempty"`
	StateValue string `json:"state_value,omitempty"`
	StateDelta int64  `json:"state_delta,omitempty"`
	StateOp    string `json:"state_op,omitempty"`

	// State list fields.
	StateKeys string `json:"state_keys,omitempty"`

	// HTTP fetch fields.
	FetchMethod   string `json:"fetch_method,omitempty"`
	FetchURL      string `json:"fetch_url,omitempty"`
	FetchHeaders  string `json:"fetch_headers,omitempty"`
	FetchBody     string `json:"fetch_body,omitempty"`
	FetchResponse string `json:"fetch_response,omitempty"`

	// Cron schedule fields.
	CronWorkflowName string `json:"cron_workflow_name,omitempty"`
	CronExpr         string `json:"cron_expr,omitempty"`
	CronTimezone     string `json:"cron_timezone,omitempty"`
	CronInput        string `json:"cron_input,omitempty"`
	CronScheduleID   string `json:"cron_schedule_id,omitempty"`
	CronResult       string `json:"cron_result,omitempty"`

	// Detached workflow fields.
	DetachedName  string `json:"detached_name,omitempty"`
	DetachedInput string `json:"detached_input,omitempty"`
	DetachedRunID string `json:"detached_run_id,omitempty"`

	// Lock fields.
	LockKey      string `json:"lock_key,omitempty"`
	LockTTLMs    int64  `json:"lock_ttl_ms,omitempty"`
	LockAcquired int    `json:"lock_acquired,omitempty"`

	// SideEffect fields.
	SideEffectResult string `json:"side_effect_result,omitempty"`

	// Scope / virtual object fields.
	ScopeKey string `json:"scope_key,omitempty"`

	// Durable log fields.
	Message  string `json:"message,omitempty"`
	LogLevel string `json:"log_level,omitempty"`
	LogKV    string `json:"log_kv,omitempty"`
}

EventRecord is a single event in a workflow's execution history. It generalizes the previous CallRecord to support sleep, signal, and defer events.

func EventRecordFromEvent 

func EventRecordFromEvent(e Event) EventRecord

EventRecordFromEvent converts a typed Event to a flat EventRecord suitable for database persistence.

func RecordsFromEvents 

func RecordsFromEvents(events []Event) []EventRecord

RecordsFromEvents converts a slice of Events to a slice of EventRecords.

type EventStream 

type EventStream interface {
	// At returns the event at index i. The index must be >= 0.
	// For forward-only replay, i should be >= the last consumed index.
	// Returns nil if i is out of bounds.
	At(i int) *EventRecord

	// Len returns the number of events currently available.
	// For slice-backed streams this is the total length.
	// For DB-backed streams this is the number of events loaded so far
	// (may be less than total; call EnsureLoaded to fetch more).
	Len() int

	// Append adds a new event to the stream. This is used during fresh
	// execution to record new events.
	Append(rec EventRecord)

	// Slice returns a contiguous portion of the stream as a []EventRecord.
	// For DB-backed streams, this may trigger loading if the requested
	// range is not yet in memory. Returns all events from start to end-1.
	// If end <= 0, returns all events from start to the end of the stream.
	Slice(start, end int) []EventRecord

	// Total returns the total number of events in the stream (including
	// events not yet loaded). For slice-backed streams this equals Len().
	// For DB-backed streams this may require a COUNT query.
	Total() (int, error)

	// Close releases any resources held by the stream (e.g., database rows).
	Close() error
}

EventStream provides access to event history records without requiring the entire history to be loaded into memory at once. It supports both replay (forward-only consumption) and fresh execution (appending new events).

Implementations:

  • SliceEventStream: backed by []EventRecord (backward compatible, used when the history is already in memory).
  • DBEventStream: backed by a database cursor; loads events on demand as the replay step counter advances. Memory usage is proportional to the active working set, not the total history length.

All implementations must be safe for concurrent access ONLY when accessed from a single goroutine (the WASM execution goroutine).

func AsEventStream 

func AsEventStream(events []EventRecord) EventStream

AsEventStream is a helper to convert a []EventRecord to an EventStream. This is the main backward-compatibility adapter.

type EventType 

type EventType string

EventType classifies event history records. 

const (
	EventTypeCall                  EventType = "call"
	EventTypeAwaitSignals          EventType = "await_signals"
	EventTypeSignalReceived        EventType = "signal_received"
	EventTypeDefer                 EventType = "defer"
	EventTypeChildWorkflow         EventType = "child_workflow"
	EventTypeAwaitChild            EventType = "await_child"
	EventTypeContinueAsNew         EventType = "continue_as_new"
	EventTypeHeartbeat             EventType = "heartbeat"
	EventTypeAwaitAllChildren      EventType = "await_all_children"
	EventTypePluginCall            EventType = "plugin_call"
	EventTypeCreatePromise         EventType = "create_promise"
	EventTypeAwaitPromise          EventType = "await_promise"
	EventTypePromiseResolved       EventType = "promise_resolved"
	EventTypePromiseRejected       EventType = "promise_rejected"
	EventTypeUpdateHandler         EventType = "update_handler"
	EventTypeStateMutation         EventType = "state_mutation"
	EventTypeRunDetached           EventType = "run_detached"
	EventTypePluginCallStreamChunk EventType = "plugin_call_stream_chunk"
	EventTypeDurableLog            EventType = "durable_log"
	EventTypeAcquireLock           EventType = "acquire_lock"
	EventTypeReleaseLock           EventType = "release_lock"
	EventTypeSideEffect            EventType = "side_effect"
	EventTypeScopeAcquired         EventType = "scope_acquired"
	EventTypeDurableSend           EventType = "durable_send"
	EventTypeDurableScheduleInvoke EventType = "durable_schedule_invoke"
	EventTypeFetch                 EventType = "fetch"
	EventTypePollChild             EventType = "poll_child"
	EventTypeAwaitAnyChild         EventType = "await_any_child"
	EventTypeAdminAction           EventType = "admin_action"
	EventTypeScheduleCron          EventType = "schedule_cron"
	EventTypeDeleteCron            EventType = "delete_cron"
	EventTypeListCrons             EventType = "list_crons"
)

type ExecResult 

type ExecResult struct {
	Result    string // JSON result string
	Suspended bool   // true if workflow suspended
}

ExecResult holds the result of a WASM function call.

type ExecutionResult 

type ExecutionResult struct {
	Result     string
	History    []EventRecord
	Suspended  *SuspendResult
	Deferrals  map[string]string
	QueryState map[string]string
}

ExecutionResult holds the complete outcome of a workflow run.

type FaultInjector 

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

FaultInjector provides programmable fault injection for testing. It simulates infrastructure failures by manipulating database state, connection behavior, or timing.

func NewFaultInjector 

func NewFaultInjector(db *sql.DB) *FaultInjector

NewFaultInjector creates a new FaultInjector for the given database.

func (*FaultInjector) ActiveFaults 

func (fi *FaultInjector) ActiveFaults() []FaultType

ActiveFaults returns a slice of all currently active fault types.

func (*FaultInjector) Cleanup 

func (fi *FaultInjector) Cleanup()

Cleanup restores normal operation, clearing all active faults.

func (*FaultInjector) Context 

func (fi *FaultInjector) Context(ctx context.Context) context.Context

Context returns a context that is cancelled if a network partition fault is active. Use this to simulate connection failures in store operations.

func (*FaultInjector) InjectClockSkew 

func (fi *FaultInjector) InjectClockSkew(offset time.Duration)

InjectClockSkew simulates clock skew by modifying the database time offset. The offset is applied to all subsequent time-based operations. A positive offset simulates the database being ahead of the worker. A negative offset simulates the database being behind the worker.

func (*FaultInjector) InjectDiskLatency 

func (fi *FaultInjector) InjectDiskLatency(min, max time.Duration)

InjectDiskLatency configures the injector to simulate slow disk operations. Each subsequent database operation will be delayed by a random duration between min and max.

func (*FaultInjector) InjectNetworkPartition 

func (fi *FaultInjector) InjectNetworkPartition()

InjectNetworkPartition simulates a network partition by blocking all database operations. It cancels the current context and prevents new operations from succeeding. Call Cleanup() to restore connectivity.

func (*FaultInjector) InjectWorkerCrash 

func (fi *FaultInjector) InjectWorkerCrash(workerID string)

InjectWorkerCrash simulates a worker crash by releasing all workflows assigned to the given workerID back to the ready queue.

func (*FaultInjector) IsActive 

func (fi *FaultInjector) IsActive(ft FaultType) bool

IsActive reports whether a particular fault type is currently active.

func (*FaultInjector) Reset 

func (fi *FaultInjector) Reset()

Reset clears all active faults and restores the database to normal state.

type FaultType 

type FaultType int

FaultType represents the type of fault to inject. 

const (
	// FaultNetworkPartition simulates a network partition where the worker
	// cannot reach the database.
	FaultNetworkPartition FaultType = iota

	// FaultDiskFull simulates a full disk where write operations fail.
	FaultDiskFull

	// FaultDiskSlow simulates slow disk I/O.
	FaultDiskSlow

	// FaultClockSkew simulates clock skew between worker and database.
	FaultClockSkew

	// FaultWorkerCrash simulates a worker crashing mid-execution.
	FaultWorkerCrash
)

func (FaultType) String 

func (ft FaultType) String() string

String returns a human-readable name for the fault type.

type Fetcher 

type Fetcher interface {
	Fetch(ctx context.Context, method, url, headersJSON, body string) (responseJSON string, err error)
}

Fetcher makes HTTP requests on behalf of cleat workflows (Stream R).

type FlusherConfig 

type FlusherConfig struct {
	MaxWait        time.Duration
	MaxBatch       int
	EnterThreshold float64
	ExitThreshold  float64
}

FlusherConfig holds the configuration for creating AdaptiveFlusher instances.

type GCOptions 

type GCOptions struct {
	// MinVersionsToKeep retains at least this many recent versions for each
	// workflow, even if they are deprecated and have no active instances.
	MinVersionsToKeep int

	// MaxVersionAge is the maximum age of a deprecated version before it is
	// eligible for removal.
	MaxVersionAge time.Duration

	// DryRun logs what would be removed without actually deleting anything.
	DryRun bool

	// Now is the reference time for age calculations. Defaults to time.Now().
	Now time.Time
}

GCOptions controls the behavior of version garbage collection.

func DefaultGCOptions 

func DefaultGCOptions() GCOptions

DefaultGCOptions returns sensible defaults for version GC.

type GCResult 

type GCResult struct {
	// VersionsRemoved is the number of workflow definition versions deleted.
	VersionsRemoved int

	// VersionsSkipped is the number of deprecated versions that were
	// considered but skipped (e.g., because they still have active instances
	// or are protected by MinVersionsToKeep).
	VersionsSkipped int

	// Errors holds per-version errors that did not abort the entire GC run.
	Errors []error
}

GCResult summarizes the outcome of a garbage collection run.

func GarbageCollectVersions 

func GarbageCollectVersions(ctx context.Context, store WorkflowStore, opts GCOptions) (*GCResult, error)

GarbageCollectVersions removes deprecated workflow definition versions that are no longer needed. It considers age, active instance counts, and the configured minimum versions to keep.

func PurgeVersions 

func PurgeVersions(ctx context.Context, store WorkflowStore, workflowName string, olderThan time.Duration) (*GCResult, error)

PurgeVersions removes all deprecated versions for a specific workflow that have zero active instances and are older than the given offset. Unlike GarbageCollectVersions, this operates on a single named workflow.

type GuestCallErrorCode 

type GuestCallErrorCode struct {
	// Name is the constant's name with its CallError prefix stripped:
	// "Unknown" is Go's cleat.CallErrorUnknown, Rust's CallError::Unknown,
	// and the corresponding member in the Python, AssemblyScript and Java
	// SDKs.
	Name string
	// Code is the byte the engine packs into the classification field of a
	// durable-call result. This is wire ABI -- every guest SDK decodes it, so
	// a member can be added but an existing value can never be changed.
	Code byte
	// Retryable is what the guest reports for this code: CallError.Retryable()
	// in the Go SDK, and its equivalent elsewhere.
	Retryable bool
}

GuestCallErrorCode describes one member of the guest SDK's CallErrorCode enum as the engine understands it.

func GuestCallErrorCodes 

func GuestCallErrorCodes() []GuestCallErrorCode

GuestCallErrorCodes returns the engine's copy of the guest SDK's CallErrorCode enum, in value order.

A copy, so a caller cannot edit the table the contract test reads.

type HeartbeatEvent 

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

HeartbeatEvent records a heartbeat tick during a long-running call.

func (HeartbeatEvent) Step 

func (e HeartbeatEvent) Step() int

func (HeartbeatEvent) Type 

func (e HeartbeatEvent) Type() EventType

type HostHandler 

type HostHandler interface {
	DurableCall(ctx context.Context, m api.Module, service, operation, requestJSON string, responsePtr, responseMaxLen uint32) int64
	DurableSleep(ctx context.Context, m api.Module, durationMs int64) int64
	DurableAwaitSignals(ctx context.Context, m api.Module, signalNames string, timeoutMs int64, sigNamePtr, sigNameMaxLen, payloadPtr, payloadMaxLen uint32) int64
	DurableDefer(ctx context.Context, m api.Module, description string, deferIDPtr, deferIDMaxLen uint32) int64
	DurableLog(ctx context.Context, m api.Module, message string) int64
	PollCancellation(ctx context.Context, m api.Module, reasonPtr, reasonMaxLen uint32) int64
	PollSignal(ctx context.Context, m api.Module, signalName string, payloadPtr, payloadMaxLen uint32) int64
	ContinueAsNew(ctx context.Context, m api.Module, newInputJSON string) int64
	ContinueAsNewWithVersion(ctx context.Context, m api.Module, newInputJSON string, newVersion int) int64
	ChildWorkflow(ctx context.Context, m api.Module, name, inputJSON string, runIDPtr, runIDMaxLen uint32) int64
	ChildWorkflowWithOptions(ctx context.Context, m api.Module, name, inputJSON string, version int64, priority int64, parentClosePolicy string, runIDPtr, runIDMaxLen uint32) int64
	ChildWorkflowInSchema(ctx context.Context, m api.Module, targetSchema, name, inputJSON string, version int64, priority int64, parentClosePolicy string, runIDPtr, runIDMaxLen uint32) int64
	AwaitChild(ctx context.Context, m api.Module, runID string, resultPtr, resultMaxLen uint32) int64
	AwaitAllChildren(ctx context.Context, m api.Module, runIDsJSON string, resultsPtr, resultsMaxLen uint32) int64
	PollChild(ctx context.Context, m api.Module, runID string, resultPtr, resultMaxLen uint32) int64
	AwaitAnyChild(ctx context.Context, m api.Module, runIDsJSON string, resultPtr, resultMaxLen uint32) int64
	DurableCallWithRetry(ctx context.Context, m api.Module, service, operation, requestJSON string, maxAttempts, initialIntervalMs, backoffCoefficient100x, maxIntervalMs int64, nonRetryableErrorsJSON string, responsePtr, responseMaxLen uint32) int64
	DurableCallWithHeartbeat(ctx context.Context, m api.Module, service, operation, requestJSON string, heartbeatIntervalMs int64, responsePtr, responseMaxLen uint32) int64
	Version(ctx context.Context) int64
	MinVersion(ctx context.Context) int64
	SetQueryState(ctx context.Context, m api.Module, key, value string) int64
	Now(ctx context.Context) int64
	Random(ctx context.Context) int64
	CreatePromise(ctx context.Context, m api.Module, name string, promiseIDPtr, promiseIDMaxLen uint32) int64
	AwaitPromise(ctx context.Context, m api.Module, promiseID string, timeoutMs int64, resultPtr, resultMaxLen uint32) int64
	PluginCall(ctx context.Context, m api.Module, pluginName, functionName, inputJSON string, responsePtr, responseMaxLen uint32) int64
	PluginCallStreaming(ctx context.Context, m api.Module, pluginName, functionName, inputJSON string, responsePtr, responseMaxLen uint32) int64
	RegisterUpdateHandler(ctx context.Context, m api.Module, name string) int64

	// Signal correlation (ABI 2.23-2.25)
	SendSignalAndWait(ctx context.Context, m api.Module, targetRunID, signalName, payload string, timeoutMs int64, responsePtr, responseMaxLen uint32) int64
	ReplyToSignal(ctx context.Context, m api.Module, correlationID, response string) int64
	SignalWorkflow(ctx context.Context, m api.Module, targetRunID, signalName, payload string) int64

	// Scoped state / virtual objects (ABI 2.26-2.28)
	SetScope(ctx context.Context, m api.Module, objectType, instanceKey string, prevScopePtr, prevScopeMaxLen uint32) int64
	GetScope(ctx context.Context, m api.Module, objTypePtr, objTypeMaxLen, instKeyPtr, instKeyMaxLen uint32) int64
	UUID(ctx context.Context, m api.Module, seed string, uuidPtr, uuidMaxLen uint32) int64

	// Lock/concurrency key operations.
	AcquireLock(ctx context.Context, m api.Module, key string, ttlMs int64) int64
	ReleaseLock(ctx context.Context, m api.Module, key string) int64

	// SideEffect records non-deterministic computation result in event
	// history on first execution and returns cached result on replay.
	SideEffect(ctx context.Context, m api.Module, computedResult string, respPtr, respMaxLen uint32) int64

	// WorkflowID returns the current workflow's unique identifier.
	WorkflowID(ctx context.Context, m api.Module, idPtr, idMaxLen uint32) int64

	// RunID returns the current workflow run's unique identifier.
	RunID(ctx context.Context, m api.Module, idPtr, idMaxLen uint32) int64

	// ResolvePromise resolves a durable promise with a value.
	ResolvePromise(ctx context.Context, m api.Module, promiseID, value string) int64

	// RejectPromise rejects a durable promise with an error message.
	RejectPromise(ctx context.Context, m api.Module, promiseID, errMsg string) int64

	// DurableSend sends a fire-and-forget request to an external service.
	DurableSend(ctx context.Context, m api.Module, service, operation, requestJSON string) int64

	// DurableScheduleInvoke schedules a delayed one-shot invocation.
	DurableScheduleInvoke(ctx context.Context, m api.Module, service, operation, requestJSON string, delayMs int64) int64

	// RegisterQueryHandler records a query handler name for ABI compatibility
	// with already-compiled guests. It does NOT make the workflow externally
	// queryable: no worker code ever reads what this records back out and
	// dispatches a query to it. Kept as a no-op (rather than removed from the
	// ABI) only because removing the host import would break instantiation
	// of guests already compiled against it -- see
	// tests/plugin-harness/testdata/pythonworkflow/call_all_plugins.wasm.
	// Every SDK's public wrapper around this call was removed 2026-08-09
	// (see docs/determinism.md, "Why there is no RegisterQueryHandler"); do
	// not add a new one, and do not build anything that assumes this call
	// causes a query to be delivered later. Use SetQueryState/GetQueryState,
	// which is: GET /api/workflows/:id/query?key=... in cmd/cleat-worker.
	RegisterQueryHandler(ctx context.Context, m api.Module, name string) int64

	// State operations (Stream R)
	SetState(ctx context.Context, m api.Module, key, value string) int64
	GetState(ctx context.Context, m api.Module, key string, valuePtr, valueMaxLen uint32) int64
	DeleteState(ctx context.Context, m api.Module, key string) int64
	IncrState(ctx context.Context, m api.Module, key string, delta int64) int64
	HasState(ctx context.Context, m api.Module, key string) int64
	ListState(ctx context.Context, m api.Module, prefix string, keysPtr, keysMaxLen uint32) int64

	// Detached execution (Stream R)
	RunDetached(ctx context.Context, m api.Module, name, inputJSON string) int64

	// HTTP fetch (Stream R)
	Fetch(ctx context.Context, m api.Module, method, url, headersJSON, body string, responsePtr, responseMaxLen uint32) int64

	// JsonParse validates and normalizes a JSON string via the host's encoding/json.
	JsonParse(ctx context.Context, m api.Module, input string, outPtr, outMaxLen uint32) int64

	// JsonStringify validates and re-serializes a JSON string via the host's encoding/json.
	JsonStringify(ctx context.Context, m api.Module, input string, outPtr, outMaxLen uint32) int64

	// Cron schedules
	ScheduleCron(ctx context.Context, m api.Module, workflowName, cronExpr, timezone, inputJSON string, idPtr, idMaxLen uint32) int64
	DeleteCron(ctx context.Context, m api.Module, scheduleID string) int64
	ListCrons(ctx context.Context, m api.Module, outPtr, outMaxLen uint32) int64
}

HostHandler is the per-execution session interface. Each method corresponds to one host function import from //go:wasmimport env <name>.

type IdempotentCaller 

type IdempotentCaller interface {
	ServiceCaller

	// CallWithIdempotencyKey makes the call, passing a key that is stable
	// across every replay of the same logical step. Implementations should
	// forward it to the service — an `Idempotency-Key` header for HTTP, an
	// explicit argument for anything else — so that a call repeated after a
	// crash returns the original outcome instead of performing the work twice.
	CallWithIdempotencyKey(ctx context.Context, service, operation, requestJSON, idempotencyKey string) (responseJSON string, err error)
}

IdempotentCaller is an optional interface a ServiceCaller may implement to receive a per-call idempotency key.

IMPROVEMENT-PLAN §1.4 phase B, and docs/durable-call-intent-design.md §4. The design there prescribes adding the key as a parameter to ServiceCaller.Call itself, and names the cost honestly: "a breaking change for external callers and plugin authors. That is the main expense of this tier."

This is the same mechanism without that expense. A caller that can deduplicate implements this interface and gets the key; one that cannot is untouched, and no existing implementation stops compiling. The engine picks the richer method when it is available. The trade is that a caller which *could* honour keys but has not been updated goes on silently not honouring them — visible in a type switch rather than in a compile error. That is a worse failure mode than a breaking change in a codebase where nobody would notice; it is a better one here, because CallerHonoursIdempotencyKeys makes the distinction testable and tests/crash asserts on the observable outcome rather than on the wiring.

If the interface is ever collapsed into ServiceCaller, delete this and take the breaking change. Nothing here forecloses that.

type MSSQLStore 

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

MSSQLStore implements WorkflowStore using a Microsoft SQL Server database. Tenant isolation is enforced via SQL Server Row-Level Security (RLS). The connection pool's connector calls sp_set_session_context on every new connection, with per-transaction calls serving as defense-in-depth.

func NewMSSQLStore 

func NewMSSQLStore(db *sql.DB, taskQueues ...string) *MSSQLStore

NewMSSQLStore creates an MSSQLStore scoped to the given task queues. The taskQueues slice specifies which task queues this worker pool should poll (e.g., "default", "gpu", "high-memory"). Defaults to ["default"]. The tenantID defaults to the default tenant UUID from the tenant foundation migration.

func (*MSSQLStore) AcquireConcurrencyKey 

func (s *MSSQLStore) AcquireConcurrencyKey(ctx context.Context, key, workflowID string, ttl time.Duration) (bool, error)

func (*MSSQLStore) AdminForceComplete 

func (s *MSSQLStore) AdminForceComplete(ctx context.Context, workflowID string, generation int64, result string, operator string) error

AdminForceComplete marks a workflow as done, bypassing worker ownership.

The retry wrapper is the one CompleteWorkflow uses: it retries only errors SQL Server guarantees it rolled back, so a deadlock cannot apply the status change twice or leave it applied without its audit event. The generation bump makes the second attempt fail with a generation mismatch if the first one did in fact commit.

func (*MSSQLStore) AdminForceFail 

func (s *MSSQLStore) AdminForceFail(ctx context.Context, workflowID string, generation int64, errorMsg, errorCode string, operator string) error

AdminForceFail marks a workflow as failed, bypassing worker ownership.

func (*MSSQLStore) AdminReReplay 

func (s *MSSQLStore) AdminReReplay(ctx context.Context, workflowID string, generation int64, operator string) error

func (*MSSQLStore) AppendEventHistory 

func (s *MSSQLStore) AppendEventHistory(ctx context.Context, workflowID string, rec EventRecord) error

AppendEventHistory appends a single event to the history.

func (*MSSQLStore) AppendEventHistoryBatch 

func (s *MSSQLStore) AppendEventHistoryBatch(ctx context.Context, workflowID string, recs []EventRecord) error

AppendEventHistoryBatch appends multiple events to the history atomically.

func (*MSSQLStore) BatchHeartbeat 

func (s *MSSQLStore) BatchHeartbeat(ctx context.Context, workerID string) (int64, error)

BatchHeartbeat updates heartbeat_at for all workflows assigned to this worker with status 'running'. Uses a single UPDATE instead of N calls. NOTE: This intentionally does NOT check per-workflow generation because it operates on ALL running workflows for a worker, and generations differ per workflow. Individual generation-guarded operations (Heartbeat, CompleteWorkflow, FailWorkflow, etc.) prevent double-execution even if the batch heartbeat refreshes a stale workflow's heartbeat_at.

func (*MSSQLStore) CheckCancellation 

func (s *MSSQLStore) CheckCancellation(ctx context.Context, workflowID string) (bool, string, error)

CheckCancellation checks if a workflow has been cancelled.

func (*MSSQLStore) CheckCrossTenantCapability 

func (s *MSSQLStore) CheckCrossTenantCapability(ctx context.Context) CrossTenantCapability

CheckCrossTenantCapability answers from SQL Server's role membership.

One question covers both paths here, unlike PostgreSQL's two functions and two grants: dbo.fn_tenant_filter admits on IS_ROLEMEMBER(N'cleat_admin') and is bound to every tenant-scoped table, workflow_instances and workflow_schedules included. So a connection either sees across tenants for both or for neither.

There is no BYPASSRLS analogue to lose. The exemption lives in the predicate itself rather than in a role attribute, so the silent-degradation failure the PostgreSQL check exists for cannot arise the same way: dropping the role membership shows up here, and altering the predicate is a schema change.

func (*MSSQLStore) ClaimDueSchedule 

func (s *MSSQLStore) ClaimDueSchedule(ctx context.Context, name string, expectedNextRun, newNextRun time.Time, runID string) (bool, error)

ClaimDueSchedule advances a schedule's next_run_at, but only if it still holds expectedNextRun. See the interface doc for why this is a CAS.

func (*MSSQLStore) ClaimStickyWorkflows 

func (s *MSSQLStore) ClaimStickyWorkflows(ctx context.Context, workerID string, limit int) ([]*WorkflowInstance, error)

ClaimStickyWorkflows atomically claims up to limit runnable workflow instances that are sticky to this worker. Uses the sticky_worker_id filter for low-contention claiming. Returns fewer than limit if not enough sticky workflows are ready. Callers should fall back to ClaimWorkflows for remaining capacity. ClaimStickyWorkflows retries on errors SQL Server guarantees it rolled back -- a deadlock victim claimed nothing, so replaying the claim is sound. Errors that leave the outcome unknown are not retried; see withRollbackGuaranteedRetry (IMPROVEMENT-PLAN.md 2.26).

func (*MSSQLStore) ClaimWorkflow 

func (s *MSSQLStore) ClaimWorkflow(ctx context.Context, workerID string) (*WorkflowInstance, error)

ClaimWorkflow atomically claims a single runnable workflow instance.

func (*MSSQLStore) ClaimWorkflows 

func (s *MSSQLStore) ClaimWorkflows(ctx context.Context, workerID string, limit int) ([]*WorkflowInstance, error)

ClaimWorkflows atomically claims up to limit runnable workflow instances. Uses UPDATE...OUTPUT with READPAST/UPDLOCK hints (SQL Server's equivalent of FOR UPDATE SKIP LOCKED) wrapped in a transaction with RLS context. ClaimWorkflows retries on errors SQL Server guarantees it rolled back -- a deadlock victim claimed nothing, so replaying the claim is sound. Errors that leave the outcome unknown are not retried; see withRollbackGuaranteedRetry (IMPROVEMENT-PLAN.md 2.26).

func (*MSSQLStore) ClaimWorkflowsAcrossTenants 

func (s *MSSQLStore) ClaimWorkflowsAcrossTenants(ctx context.Context, workerID string, limit int) ([]*WorkflowInstance, error)

ClaimWorkflowsAcrossTenants claims runnable workflows for every tenant.

Unlike PostgresStore and MySQLStore, this is not a different query: SQL Server's isolation was never an application-level "AND tenant_id = ?" predicate to begin with, it is entirely dbo.fn_tenant_filter (the RLS filter predicate bound to workflow_instances and six other tables). That predicate already special-cases exactly this call: 

WHERE @tenant_id = CAST(SESSION_CONTEXT(N'tenant_id') AS UNIQUEIDENTIFIER)
   OR IS_ROLEMEMBER(N'cleat_admin') = 1

so a connection whose login is a member of dbo.cleat_admin sees every row regardless of SESSION_CONTEXT -- unset, stale, or set to some other tenant's ID, it does not matter, because the OR admits the row on role membership alone. migrations/mssql/012_admin_role.sql documents the mechanism and, deliberately, ships the role with no members: granting it is a deployment decision.

That is why claimWorkflowsAcrossTenantsOnce below runs the same SELECT FOR UPDATE / UPDATE / OUTPUT statement claimWorkflowsOnce does, with two differences. First, it never calls setSessionContext or beginTxWithContext -- setting SESSION_CONTEXT to any one tenant here would be misleading (this call is not scoped to a tenant) and is not even load-bearing for an admin connection, since the role check bypasses it either way. Second, its OUTPUT clause reads tenant_id through CONVERT(NVARCHAR(36), ...), which claimWorkflowsOnce and claimStickyWorkflowsOnce do not: they scan INSERTED.tenant_id (a UNIQUEIDENTIFIER) straight into a Go string, and go-mssqldb hands back that type's raw 16-byte storage rather than the hyphenated form -- the same hazard ResolveTenantFromAPIKey's doc comment already names ("MSSQL UNIQUEIDENTIFIER mixed-endian storage") and works around the same way. It went unnoticed there because nothing asserted on a claimed row's TenantID string content until this method's own integration test did. This call cannot inherit that: CrossTenantClaimer's entire contract is that the caller re-scopes on TenantID (cmd/cleat-worker's storeForTenant), so a mangled value here would silently misroute every claimed workflow rather than merely being an unused field. Whether the sibling methods should be fixed the same way is a separate question -- they don't currently promise TenantID to a caller that acts on it the way this one does -- and is left to whoever owns that call.

What this method cannot do is turn a non-admin connection into one: it cannot grant itself the role, and running the claim without checking first would silently return at most one tenant's work (whatever SESSION_CONTEXT happens to hold, or nothing at all if it was never set) -- which reads exactly like an idle queue to whoever is watching the dispatch loop, the specific failure mode CrossTenantClaimer exists to avoid. So this checks IS_ROLEMEMBER itself, before touching workflow_instances, and returns ErrCrossTenantClaimUnsupported naming the missing grant and the file that documents it, rather than an empty and misleading claim. The worker warns once and falls back to the per-tenant claim, the same answer MySQL's per-tenant-database topology gets: the operator is told, and dispatch keeps running on what this connection can legitimately see.

func (*MSSQLStore) CleanupMemorySamples 

func (s *MSSQLStore) CleanupMemorySamples(ctx context.Context, maxSamplesPerDef int) (int64, error)

func (*MSSQLStore) ClearStickyWorker 

func (s *MSSQLStore) ClearStickyWorker(ctx context.Context, workflowID string) error

func (*MSSQLStore) CompactHistory 

func (s *MSSQLStore) CompactHistory(ctx context.Context, workflowID string, compactionState []byte, compactionStep int, keepStep int) error

func (*MSSQLStore) CompleteCallIntent 

func (s *MSSQLStore) CompleteCallIntent(ctx context.Context, workflowID string, rec EventRecord, payload []byte, checksum string, workerID string, generation int64) error

func (*MSSQLStore) CompleteUpdateRequest 

func (s *MSSQLStore) CompleteUpdateRequest(ctx context.Context, workflowID, updateName, result, errMsg string) error

func (*MSSQLStore) CompleteWorkflow 

func (s *MSSQLStore) CompleteWorkflow(ctx context.Context, workflowID, workerID string, generation int64, result string, queryState map[string]string) error

CompleteWorkflow marks a workflow as completed with a result. CompleteWorkflow retries only on errors SQL Server guarantees it rolled back, so a deadlock no longer loses the terminal write. ErrFenceLost is returned before the commit and is not an mssql.Error, so the fence semantics are untouched by the retry. See withRollbackGuaranteedRetry.

func (*MSSQLStore) ContinueAsNew 

func (s *MSSQLStore) ContinueAsNew(ctx context.Context, currentRunID, workerID string, generation int64, defName string, defVersion int, newInput json.RawMessage, newEvents []EventRecord, result string, queryState map[string]string, priority int) (string, error)

ContinueAsNew atomically creates a new workflow run and completes the current one in a single database transaction. Returns the new run ID on success. ContinueAsNew retries only on errors SQL Server guarantees it rolled back. See withRollbackGuaranteedRetry.

func (*MSSQLStore) CountActiveInstances 

func (s *MSSQLStore) CountActiveInstances(ctx context.Context, name string, version int) (int, error)

CountActiveInstances returns the number of ready or running instances for a version.

func (*MSSQLStore) CountEventHistory 

func (s *MSSQLStore) CountEventHistory(ctx context.Context, workflowID string) (int, error)

CountEventHistory returns the total number of events for a workflow.

func (*MSSQLStore) CreatePromise 

func (s *MSSQLStore) CreatePromise(ctx context.Context, workflowID, promiseName, promiseID string) error

func (*MSSQLStore) CreateSchedule 

func (s *MSSQLStore) CreateSchedule(ctx context.Context, sch Schedule) error

func (*MSSQLStore) CreateUpdateRequest 

func (s *MSSQLStore) CreateUpdateRequest(ctx context.Context, workflowID, updateName, payload, promiseID string) error

func (*MSSQLStore) DeleteCompletedWorkflows 

func (s *MSSQLStore) DeleteCompletedWorkflows(ctx context.Context, olderThan time.Time) (int64, error)

DeleteCompletedWorkflows permanently deletes workflow_instances rows in a terminal, no-further-action status ('done', 'failed', 'terminated') whose completed_at is older than the cutoff. 'dead_lettered' is deliberately excluded -- see the interface doc (store_interface.go) and DeleteDeadLetteredWorkflows above.

No explicit event_history delete is needed here: migrations/mssql/001_schema.sql declares event_history's FK to workflow_instances ON DELETE CASCADE and SQL Server never dropped it (only PostgreSQL did, deliberately). Deleting the workflow_instances row below cascades event_history (and workflow_signals, workflow_promises, concurrency_keys, workflow_update_requests) automatically.

UNVERIFIED: no SQL Server instance was available to run this against; it is written to match DeleteDeadLetteredWorkflows immediately above exactly (same batching shape, same reliance on cascade), which was itself the verified reference for this dialect's FK graph.

func (*MSSQLStore) DeleteDeadLetteredWorkflows 

func (s *MSSQLStore) DeleteDeadLetteredWorkflows(ctx context.Context, olderThan time.Time) (int64, error)

func (*MSSQLStore) DeleteExpiredEvents 

func (s *MSSQLStore) DeleteExpiredEvents(ctx context.Context, olderThan time.Time) (int64, error)

func (*MSSQLStore) DeleteSchedule 

func (s *MSSQLStore) DeleteSchedule(ctx context.Context, name string) error

func (*MSSQLStore) DeliverSignal 

func (s *MSSQLStore) DeliverSignal(ctx context.Context, workflowID, signalName, payload string) error

func (*MSSQLStore) DeployWorkflowDef 

func (s *MSSQLStore) DeployWorkflowDef(ctx context.Context, def *WorkflowDef) error

DeployWorkflowDef inserts or updates a workflow definition.

func (*MSSQLStore) FailWorkflow 

func (s *MSSQLStore) FailWorkflow(ctx context.Context, workflowID, workerID string, generation int64, errorMsg, errorCode, errorOp string, queryState map[string]string) error

FailWorkflow marks a workflow as failed. FailWorkflow retries only on errors SQL Server guarantees it rolled back, so a deadlock no longer loses the terminal write. ErrFenceLost is returned before the commit and is not an mssql.Error, so the fence semantics are untouched by the retry. See withRollbackGuaranteedRetry.

func (*MSSQLStore) FinalizeWorkflowSegment 

func (s *MSSQLStore) FinalizeWorkflowSegment(ctx context.Context, runID, workerID string, generation int64, newEvents []EventRecord, finalStatus string, result string, errorCode string, errorOp string, queryState map[string]string, nextWakeAt time.Time) error

FinalizeWorkflowSegment atomically appends new events and updates the workflow status in a single database transaction. This eliminates the race between AppendEventHistoryBatch and the subsequent CompleteWorkflow / FailWorkflow / ReleaseWorkflow call.

finalStatus must be one of:

  • "done" — marks the workflow as completed with the given result
  • "failed" — marks the workflow as failed with the given error info
  • "ready" — returns the workflow to the ready queue (suspend)

Fields not relevant to the chosen status are ignored. FinalizeWorkflowSegment retries only on errors SQL Server guarantees it rolled back, so a deadlock no longer loses the terminal write. ErrFenceLost is returned before the commit and is not an mssql.Error, so the fence semantics are untouched by the retry. See withRollbackGuaranteedRetry.

func (*MSSQLStore) GetActiveInstanceCountsByVersion 

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

GetActiveInstanceCountsByVersion returns a map of "name:version" -> count.

func (*MSSQLStore) GetAllowedSignalCallers 

func (s *MSSQLStore) GetAllowedSignalCallers(ctx context.Context, workflowID string) ([]string, error)

func (*MSSQLStore) GetChildCount 

func (s *MSSQLStore) GetChildCount(ctx context.Context, parentWorkflowID string) (int, error)

func (*MSSQLStore) GetChildResult 

func (s *MSSQLStore) GetChildResult(ctx context.Context, runID string) (string, bool, error)

func (*MSSQLStore) GetCompactionCandidates 

func (s *MSSQLStore) GetCompactionCandidates(ctx context.Context, threshold int, limit int) ([]string, error)

func (*MSSQLStore) GetConcurrencyKeyCount 

func (s *MSSQLStore) GetConcurrencyKeyCount(ctx context.Context, workflowID string) (int, error)

func (*MSSQLStore) GetDueSchedules 

func (s *MSSQLStore) GetDueSchedules(ctx context.Context) ([]Schedule, error)

func (*MSSQLStore) GetDueSchedulesAcrossTenants 

func (s *MSSQLStore) GetDueSchedulesAcrossTenants(ctx context.Context) ([]Schedule, error)

GetDueSchedulesAcrossTenants returns every tenant's due schedules.

Same mechanism as ClaimWorkflowsAcrossTenants and the same reasoning: SQL Server's isolation here is dbo.fn_tenant_filter, and that predicate already admits a connection whose login is a member of dbo.cleat_admin regardless of SESSION_CONTEXT (migrations/mssql/012_admin_role.sql). So there is no second query to write -- there is the same query, on a connection the predicate lets through, and a check that this connection is actually one of those.

Unlike PostgreSQL, no migration 024 equivalent is needed: 012 already grants across every table fn_tenant_filter is bound to, and workflow_schedules is one of them. What a deployment must do is grant the role, which it must already have done for the cross-tenant claim.

func (*MSSQLStore) GetEventCount 

func (s *MSSQLStore) GetEventCount(ctx context.Context, workflowID string) (int, error)

func (*MSSQLStore) GetPendingUpdateRequests 

func (s *MSSQLStore) GetPendingUpdateRequests(ctx context.Context, workflowID string) ([]UpdateRequestInfo, error)

func (*MSSQLStore) GetPromise 

func (s *MSSQLStore) GetPromise(ctx context.Context, workflowID, promiseID string) (string, string, string, error)

func (*MSSQLStore) GetQueryState 

func (s *MSSQLStore) GetQueryState(ctx context.Context, workflowID, key string) (string, error)

func (*MSSQLStore) GetRoutingRules 

func (s *MSSQLStore) GetRoutingRules(ctx context.Context, workflowName string) ([]RoutingRule, error)

GetRoutingRules returns all routing rules for a workflow.

func (*MSSQLStore) GetWASMLength 

func (s *MSSQLStore) GetWASMLength(ctx context.Context, defName string, defVersion int) (int64, error)

GetWASMLength returns the byte length of the stored WASM binary.

func (*MSSQLStore) GetWorkflowByID 

func (s *MSSQLStore) GetWorkflowByID(ctx context.Context, id string) (*WorkflowInstance, error)

GetWorkflowByID returns a single workflow instance by ID.

func (*MSSQLStore) GetWorkflowDef 

func (s *MSSQLStore) GetWorkflowDef(ctx context.Context, name string, version int) (*WorkflowDef, error)

GetWorkflowDef returns a single workflow definition by name and version.

func (*MSSQLStore) GetWorkflowTag 

func (s *MSSQLStore) GetWorkflowTag(ctx context.Context, workflowName string, tag string) (int, error)

GetWorkflowTag returns the version for a given tag.

func (*MSSQLStore) GetWorkflowTags 

func (s *MSSQLStore) GetWorkflowTags(ctx context.Context, workflowName string) (map[string]int, error)

GetWorkflowTags returns all tag -> version mappings for a workflow.

func (*MSSQLStore) Heartbeat 

func (s *MSSQLStore) Heartbeat(ctx context.Context, workflowID, workerID string, generation int64) (bool, error)

Heartbeat updates the heartbeat timestamp. Returns false if the workflow is no longer assigned to this worker.

func (*MSSQLStore) ListPromises 

func (s *MSSQLStore) ListPromises(ctx context.Context, workflowID string) ([]PromiseInfo, error)

func (*MSSQLStore) ListSchedules 

func (s *MSSQLStore) ListSchedules(ctx context.Context) ([]Schedule, error)

func (*MSSQLStore) ListVersions 

func (s *MSSQLStore) ListVersions(ctx context.Context, defName string) ([]int, error)

ListVersions returns all deployed versions of a workflow.

func (*MSSQLStore) ListWorkflowDefs 

func (s *MSSQLStore) ListWorkflowDefs(ctx context.Context, name string) ([]WorkflowDef, error)

ListWorkflowDefs returns all versions of a workflow, ordered by version DESC.

func (*MSSQLStore) ListWorkflows 

func (s *MSSQLStore) ListWorkflows(ctx context.Context, filter WorkflowFilter) ([]WorkflowInstance, error)

ListWorkflows returns workflow instances filtered by the given filter parameters. Supported filters: Status, InputContains, ErrorContains, Search. Supports pagination via Offset and Limit (default 100, max 1000).

func (*MSSQLStore) LoadCompactionState 

func (s *MSSQLStore) LoadCompactionState(ctx context.Context, workflowID string) (*CompactionState, error)

func (*MSSQLStore) LoadDAGSpec 

func (s *MSSQLStore) LoadDAGSpec(ctx context.Context, defName string, defVersion int) (json.RawMessage, error)

LoadDAGSpec returns the dag_spec JSON for a workflow definition, or nil if none.

func (*MSSQLStore) LoadEventHistory 

func (s *MSSQLStore) LoadEventHistory(ctx context.Context, workflowID string) ([]EventRecord, error)

LoadEventHistory returns all event records for a workflow, ordered by step.

func (*MSSQLStore) LoadEventHistoryPaginated 

func (s *MSSQLStore) LoadEventHistoryPaginated(ctx context.Context, workflowID string, offset, limit int) ([]EventRecord, error)

LoadEventHistoryPaginated returns a page of event history for a workflow, with offset and limit support. Defaults limit to 1000 if limit <= 0, capped at 1000.

func (*MSSQLStore) LoadMemoryEstimates 

func (s *MSSQLStore) LoadMemoryEstimates(ctx context.Context) (map[string]float64, error)

func (*MSSQLStore) LoadMemoryStats 

func (s *MSSQLStore) LoadMemoryStats(ctx context.Context) ([]WorkflowMemoryStats, error)

func (*MSSQLStore) LoadWASM 

func (s *MSSQLStore) LoadWASM(ctx context.Context, defName string, defVersion int) ([]byte, error)

LoadWASM returns the compiled WASM bytes for a workflow definition.

func (*MSSQLStore) LoadWorkflowConfig 

func (s *MSSQLStore) LoadWorkflowConfig(ctx context.Context, defName string, defVersion int) (int, error)

LoadWorkflowConfig returns the max_history_length for a workflow definition.

func (*MSSQLStore) MarkVersionDeprecated 

func (s *MSSQLStore) MarkVersionDeprecated(ctx context.Context, name string, version int, deprecated bool) error

MarkVersionDeprecated sets the deprecated flag on a workflow version.

func (*MSSQLStore) MoveToDeadLetterQueue 

func (s *MSSQLStore) MoveToDeadLetterQueue(ctx context.Context, workflowID, workerID string, generation int64, errMsg, errorCode, errorOp string) error

MoveToDeadLetterQueue marks a workflow as dead_lettered because it failed after exhausting all retry attempts. MoveToDeadLetterQueue retries only on errors SQL Server guarantees it rolled back, so a deadlock no longer loses the terminal write. ErrFenceLost is returned before the commit and is not an mssql.Error, so the fence semantics are untouched by the retry. See withRollbackGuaranteedRetry.

func (*MSSQLStore) PickVersionByRouting 

func (s *MSSQLStore) PickVersionByRouting(ctx context.Context, workflowName string) (int, error)

PickVersionByRouting performs weighted random version selection. Returns 0 if no routing rules exist.

func (*MSSQLStore) PollAndClaimSignal 

func (s *MSSQLStore) PollAndClaimSignal(ctx context.Context, workflowID, signalName string) (string, bool, error)

func (*MSSQLStore) PollCancellation 

func (s *MSSQLStore) PollCancellation(ctx context.Context, workflowID string) (bool, string, error)

func (*MSSQLStore) PollSignal 

func (s *MSSQLStore) PollSignal(ctx context.Context, workflowID, signalName string) (string, bool, error)

func (*MSSQLStore) PurgeWorkflowDef 

func (s *MSSQLStore) PurgeWorkflowDef(ctx context.Context, name string, version int) error

PurgeWorkflowDef permanently deletes a workflow definition.

func (*MSSQLStore) QueueDepth 

func (s *MSSQLStore) QueueDepth(ctx context.Context) (int64, error)

func (*MSSQLStore) ReapExpiredConcurrencyKeys 

func (s *MSSQLStore) ReapExpiredConcurrencyKeys(ctx context.Context) (int64, error)

func (*MSSQLStore) ReapStaleInstances 

func (s *MSSQLStore) ReapStaleInstances(ctx context.Context, timeout time.Duration) (int, error)

func (*MSSQLStore) RecordWorkflowMemorySample 

func (s *MSSQLStore) RecordWorkflowMemorySample(ctx context.Context, defName string, sampleBytes int64) error

func (*MSSQLStore) RejectPromise 

func (s *MSSQLStore) RejectPromise(ctx context.Context, workflowID, promiseID, errMsg string) error

func (*MSSQLStore) ReleaseConcurrencyKey 

func (s *MSSQLStore) ReleaseConcurrencyKey(ctx context.Context, key string) error

func (*MSSQLStore) ReleaseWorkflow 

func (s *MSSQLStore) ReleaseWorkflow(ctx context.Context, workflowID, workerID string, generation int64, nextWakeAt time.Time) error

ReleaseWorkflow returns a workflow to the ready queue with a next wake time.

func (*MSSQLStore) ReleaseWorkflowConcurrencyKeys 

func (s *MSSQLStore) ReleaseWorkflowConcurrencyKeys(ctx context.Context, workflowID string) error

func (*MSSQLStore) RemoveRoutingRule 

func (s *MSSQLStore) RemoveRoutingRule(ctx context.Context, ruleID string) error

RemoveRoutingRule deletes a routing rule by ID.

func (*MSSQLStore) RemoveWorkflowTag 

func (s *MSSQLStore) RemoveWorkflowTag(ctx context.Context, workflowName string, tag string) error

RemoveWorkflowTag deletes a tag assignment.

func (*MSSQLStore) RequestCancellation 

func (s *MSSQLStore) RequestCancellation(ctx context.Context, workflowID, reason string) error

RequestCancellation sets the cancellation flag on a workflow.

func (*MSSQLStore) ResolveCallIntent 

func (s *MSSQLStore) ResolveCallIntent(ctx context.Context, workflowID string, rec EventRecord, payload []byte, workerID string, generation int64) error

func (*MSSQLStore) ResolveLatestVersion 

func (s *MSSQLStore) ResolveLatestVersion(ctx context.Context, defName string) (int, error)

ResolveLatestVersion resolves the latest version for a named definition.

func (*MSSQLStore) ResolvePromise 

func (s *MSSQLStore) ResolvePromise(ctx context.Context, workflowID, promiseID, result string) error

func (*MSSQLStore) ResolveTenantFromAPIKey 

func (s *MSSQLStore) ResolveTenantFromAPIKey(ctx context.Context, keyHash []byte) (uuid.UUID, error)

ResolveTenantFromAPIKey looks up a tenant UUID by API key hash. Uses CONVERT(NVARCHAR(36), tenant_id) to avoid byte-swapping issues with MSSQL UNIQUEIDENTIFIER mixed-endian storage. admin.tenant_api_keys, qualified. This read was unqualified, which resolves against the connecting principal's default schema and so landed on dbo.tenant_api_keys -- a table migrations/mssql/001_schema.sql creates and no production code ever writes. The only writer is auth.TenantStore, which writes admin.tenant_api_keys, so on SQL Server this lookup queried a table that was always empty and API-key tenant resolution could not succeed.

PostgreSQL names it admin.tenant_api_keys (engine/store_deployment.go:90); MySQL's unqualified name is correct there, because MySQL puts each tenant in its own database rather than a schema.

The dbo.tenants / dbo.tenant_api_keys pair is duplicate schema and should be dropped in a migration -- not done here, because a DROP needs to know what an existing deployment has put in them.

func (*MSSQLStore) ResolveVersionByTag 

func (s *MSSQLStore) ResolveVersionByTag(ctx context.Context, workflowName string, tag string) (int, error)

ResolveVersionByTag resolves a tag to a version number. If tag is "latest", returns the highest non-deprecated version.

func (*MSSQLStore) RetryWorkflow 

func (s *MSSQLStore) RetryWorkflow(ctx context.Context, workflowID string) error

RetryWorkflow moves a dead_lettered workflow back to a runnable state. Resets status to 'ready', clears the worker assignment and error fields, and sets next_wake_at to now so the workflow is re-queued immediately.

func (*MSSQLStore) SetRoutingRule 

func (s *MSSQLStore) SetRoutingRule(ctx context.Context, workflowName string, targetVersion int, weight float64) error

SetRoutingRule creates a routing rule for a workflow version.

func (*MSSQLStore) SetScheduleEnabled 

func (s *MSSQLStore) SetScheduleEnabled(ctx context.Context, name string, enabled bool) error

func (*MSSQLStore) SetWorkflowTag 

func (s *MSSQLStore) SetWorkflowTag(ctx context.Context, workflowName string, version int, tag string) error

SetWorkflowTag assigns a tag to a specific version. Uses MERGE so reassigning a tag updates in place.

func (*MSSQLStore) StartChildWorkflow 

func (s *MSSQLStore) StartChildWorkflow(ctx context.Context, parentID, defName, inputJSON string, defVersion int, parentClosePolicy string, priority int) (string, error)

func (*MSSQLStore) StartChildWorkflowAtomic 

func (s *MSSQLStore) StartChildWorkflowAtomic(ctx context.Context, childID, parentID, defName, inputJSON string, defVersion int, parentClosePolicy string, event EventRecord, priority int) (string, error)

func (*MSSQLStore) StartNewRun 

func (s *MSSQLStore) StartNewRun(ctx context.Context, runID, defName string, defVersion int, input json.RawMessage, idempotencyKey string, tenantID string, priority int) (string, bool, error)

StartNewRun creates a new workflow instance. If idempotencyKey is non-empty, provides exactly-once semantics.

func (*MSSQLStore) StreamEventHistory 

func (s *MSSQLStore) StreamEventHistory(ctx context.Context, workflowID string, pageSize int) (<-chan EventRecord, <-chan error)

StreamEventHistory loads event history for a workflow in pages, returning events through a channel. Events are fetched in pages of pageSize as the caller reads from the channel. The channel is closed when all events have been sent.

func (*MSSQLStore) TerminateWorkflow 

func (s *MSSQLStore) TerminateWorkflow(ctx context.Context, workflowID, reason string) error

func (*MSSQLStore) TraceWorkflow 

func (s *MSSQLStore) TraceWorkflow(ctx context.Context, workflowID, traceID string) error

TraceWorkflow sets the W3C trace_id on a workflow instance.

func (*MSSQLStore) UpdateScheduleNextRun 

func (s *MSSQLStore) UpdateScheduleNextRun(ctx context.Context, name string, nextRun time.Time) error

func (*MSSQLStore) UpdateStickyWorker 

func (s *MSSQLStore) UpdateStickyWorker(ctx context.Context, workflowID, workerID string) error

func (*MSSQLStore) ValidateVersion 

func (s *MSSQLStore) ValidateVersion(ctx context.Context, defName string, defVersion int) (bool, error)

ValidateVersion checks whether the given version is valid.

func (*MSSQLStore) VerifyWorkflowEvents 

func (s *MSSQLStore) VerifyWorkflowEvents(ctx context.Context, workflowID string) error

VerifyWorkflowEvents loads all events for a workflow, recomputes their SHA-256 checksums, and verifies integrity against stored checksums.

func (*MSSQLStore) WithEncryption 

func (s *MSSQLStore) WithEncryption(enc *PayloadEncryption, enabled bool) *MSSQLStore

WithEncryption returns a copy of the store with encryption at rest enabled. NOTE: Encryption at rest is not yet supported on MSSQL backends. This method is present for forward compatibility so that StreamEventHistory can contain the same decryption guard as the Postgres variant.

func (*MSSQLStore) WithIdempotencyKeyTTL 

func (s *MSSQLStore) WithIdempotencyKeyTTL(ttl time.Duration) *MSSQLStore

WithIdempotencyKeyTTL returns a copy of the store with the given idempotency key TTL.

func (*MSSQLStore) WithLogger 

func (s *MSSQLStore) WithLogger(l *slog.Logger) *MSSQLStore

WithLogger returns a copy of the store with the given structured logger.

func (*MSSQLStore) WithReadRedactionDisabled 

func (s *MSSQLStore) WithReadRedactionDisabled(disabled bool) *MSSQLStore

WithReadRedactionDisabled returns a copy of the store with redaction on the read path disabled. Used during replay to avoid overhead.

func (*MSSQLStore) WithTenant 

func (s *MSSQLStore) WithTenant(tenantID string) *MSSQLStore

WithTenant returns a copy of the store scoped to the given tenant ID. This is used in the dispatch loop to set the correct tenant context before executing a workflow. The returned store's methods will set the RLS session variable via sp_set_session_context.

func (*MSSQLStore) WriteCallIntent 

func (s *MSSQLStore) WriteCallIntent(ctx context.Context, workflowID string, rec EventRecord, workerID string, generation int64) error

type MSSQLStoreFactory 

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

MSSQLStoreFactory implements StoreFactory for Microsoft SQL Server. It manages per-tenant connection pools with sp_set_session_context baked into the connector, enforcing RLS at the connection level.

func NewMSSQLStoreFactory 

func NewMSSQLStoreFactory(connStr string, idempotencyKeyTTL ...time.Duration) *MSSQLStoreFactory

NewMSSQLStoreFactory creates an MSSQLStoreFactory. connStr is the SQL Server connection string used to open per-tenant pools.

func (*MSSQLStoreFactory) Close 

func (f *MSSQLStoreFactory) Close() error

Close closes all tenant connection pools.

func (*MSSQLStoreFactory) Dialect 

func (f *MSSQLStoreFactory) Dialect() Dialect

Dialect returns DialectMSSQL.

func (*MSSQLStoreFactory) DriverName 

func (f *MSSQLStoreFactory) DriverName() string

DriverName returns "mssql".

func (*MSSQLStoreFactory) OpenStore 

func (f *MSSQLStoreFactory) OpenStore(ctx context.Context, tenantID string, taskQueues ...string) (WorkflowStore, io.Closer, error)

OpenStore creates an MSSQLStore scoped to the given tenant. Each tenant gets a dedicated connection pool with RLS session context baked into every connection.

NOTE: Encryption at rest (--encrypt-sensitive-payloads) is not yet supported on MSSQL backends. See PostgresStore.WithEncryption for the reference implementation.

func (*MSSQLStoreFactory) WithLogger 

func (f *MSSQLStoreFactory) WithLogger(l *slog.Logger) *MSSQLStoreFactory

WithLogger sets the structured logger on the factory. Stores created by OpenStore will inherit it.

func (*MSSQLStoreFactory) WithTenantPoolMaxConns 

func (f *MSSQLStoreFactory) WithTenantPoolMaxConns(n int) *MSSQLStoreFactory

WithTenantPoolMaxConns sets the max open connections per tenant pool.

type MySQLStore 

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

MySQLStore implements WorkflowStore using a MySQL 8.0+ or MariaDB 10.6+ database. MySQL has no row-level security, so tenant isolation is enforced at the database level — each tenant gets its own database (cleat_<tenant_id>). The store's connection pool is scoped to the tenant's database, making cross-tenant data access impossible at the connection level. WHERE tenant_id = ? clauses are retained as defense-in-depth.

Because there is no RLS, those WHERE clauses are the only thing scoping a query to a tenant once a connection is open — PostgreSQL's seven policies have no MySQL equivalent (IMPROVEMENT-PLAN.md 1.7). A tenantID of "" is therefore not a harmless empty filter: it is a query running with no identity and no database-level backstop. See requireTenant below.

func NewMySQLStore 

func NewMySQLStore(db *sql.DB, taskQueues ...string) *MySQLStore

NewMySQLStore creates a MySQLStore scoped to the given task queues. The taskQueues slice specifies which task queues this worker pool should poll (e.g., "default", "gpu", "high-memory"). Defaults to ["default"]. The tenantID defaults to the default tenant UUID.

func (*MySQLStore) AcquireConcurrencyKey 

func (s *MySQLStore) AcquireConcurrencyKey(ctx context.Context, key, workflowID string, ttl time.Duration) (bool, error)

AcquireConcurrencyKey tries to acquire a concurrency key for a workflow. Returns true if acquired, false if already held by another workflow.

func (*MySQLStore) AdminForceComplete 

func (s *MySQLStore) AdminForceComplete(ctx context.Context, workflowID string, generation int64, result string, operator string) error

AdminForceComplete marks a workflow as done, bypassing worker ownership.

func (*MySQLStore) AdminForceFail 

func (s *MySQLStore) AdminForceFail(ctx context.Context, workflowID string, generation int64, errorMsg, errorCode string, operator string) error

AdminForceFail marks a workflow as failed, bypassing worker ownership.

func (*MySQLStore) AdminReReplay 

func (s *MySQLStore) AdminReReplay(ctx context.Context, workflowID string, generation int64, operator string) error

func (*MySQLStore) AppendEventHistory 

func (s *MySQLStore) AppendEventHistory(ctx context.Context, workflowID string, rec EventRecord) error

AppendEventHistory appends a single event to the history.

func (*MySQLStore) AppendEventHistoryBatch 

func (s *MySQLStore) AppendEventHistoryBatch(ctx context.Context, workflowID string, recs []EventRecord) error

AppendEventHistoryBatch appends multiple events atomically.

func (*MySQLStore) BatchHeartbeat 

func (s *MySQLStore) BatchHeartbeat(ctx context.Context, workerID string) (int64, error)

BatchHeartbeat updates heartbeat_at for all workflows assigned to this worker with status 'running'. Uses a single UPDATE instead of N calls. NOTE: This intentionally does NOT check per-workflow generation because it operates on ALL running workflows for a worker, and generations differ per workflow. Individual generation-guarded operations (Heartbeat, CompleteWorkflow, FailWorkflow, etc.) prevent double-execution even if the batch heartbeat refreshes a stale workflow's heartbeat_at.

func (*MySQLStore) CheckCancellation 

func (s *MySQLStore) CheckCancellation(ctx context.Context, workflowID string) (bool, string, error)

CheckCancellation checks if a workflow has been cancelled.

func (*MySQLStore) CheckCrossTenantCapability 

func (s *MySQLStore) CheckCrossTenantCapability(ctx context.Context) CrossTenantCapability

CheckCrossTenantCapability answers from the topology this store was built for, because on MySQL that is the whole question -- there is no grant to check and no policy to be exempt from.

func (*MySQLStore) ClaimDueSchedule 

func (s *MySQLStore) ClaimDueSchedule(ctx context.Context, name string, expectedNextRun, newNextRun time.Time, runID string) (bool, error)

ClaimDueSchedule advances a schedule's next_run_at, but only if it still holds expectedNextRun. See the interface doc for why this is a CAS.

func (*MySQLStore) ClaimStickyWorkflows 

func (s *MySQLStore) ClaimStickyWorkflows(ctx context.Context, workerID string, limit int) ([]*WorkflowInstance, error)

ClaimStickyWorkflows atomically claims up to limit runnable workflow instances that are sticky to this worker. Uses SELECT ... FOR UPDATE SKIP LOCKED with sticky_worker_id filtering for low-contention claiming.

func (*MySQLStore) ClaimWorkflow 

func (s *MySQLStore) ClaimWorkflow(ctx context.Context, workerID string) (*WorkflowInstance, error)

ClaimWorkflow atomically dequeues a runnable workflow instance. Uses SELECT ... FOR UPDATE SKIP LOCKED. Delegates to ClaimWorkflows.

func (*MySQLStore) ClaimWorkflows 

func (s *MySQLStore) ClaimWorkflows(ctx context.Context, workerID string, limit int) ([]*WorkflowInstance, error)

ClaimWorkflows atomically claims up to limit runnable workflow instances. Uses SELECT ... FOR UPDATE SKIP LOCKED to avoid contention. MySQL does not support UPDATE ... RETURNING, so we use a three-step process inside a transaction: SELECT FOR UPDATE, UPDATE, SELECT.

func (*MySQLStore) ClaimWorkflowsAcrossTenants 

func (s *MySQLStore) ClaimWorkflowsAcrossTenants(ctx context.Context, workerID string, limit int) ([]*WorkflowInstance, error)

ClaimWorkflowsAcrossTenants claims runnable workflows for every tenant.

MySQL has no row-level security (IMPROVEMENT-PLAN.md 1.7): every method on this store scopes itself with its own "AND tenant_id = ?" predicate rather than relying on anything the database enforces. So unlike Postgres, which borrows a role's RLS exemption (admin.claim_workflows, migrations/postgres/023_cross_tenant_claim.sql), or SQL Server, which relies on a role its filter predicate already special-cases (dbo.cleat_admin, migrations/mssql/012_admin_role.sql), there is no separate mechanism to call into here. This is the exact same three-step claim as ClaimWorkflows above with that one predicate removed from the SELECT.

The result columns and order match scanClaimedWorkflows (id, def_name, def_version, status, input, assigned_to, next_wake_at, tenant_id, created_at, error_code, error_op, generation, priority, trace_id) exactly, so this reuses it rather than the dialect.scanWorkflowInstanceExtra helper ClaimWorkflows uses -- that helper's MySQL branch does nothing different from scanClaimedWorkflows, it exists only for the MSSQL input-as-string quirk in the same function.

It is NOT equivalent to the Postgres and SQL Server cross-tenant claims on the topology cmd/cleat-worker actually builds (NewMySQLStoreFactory in main.go). That factory gives each tenant its own physical database (cleat_<tenant_id>; see MySQLStoreFactory's doc comment), and this store's db field is a connection pool opened against exactly one of them. Dropping the tenant_id predicate does not widen what that connection can see -- the other tenants' rows are not filtered out, they are not IN this database. On that topology this method would return exactly what ClaimWorkflows returns while reporting that it had swept every tenant, so it refuses instead: perTenantDatabase is set by OpenStore, and the refusal is ErrCrossTenantClaimUnsupported, which cmd/cleat-worker's claimGeneral handles by logging once and falling back to the per-tenant claim. Same work gets done either way; the difference is whether the operator is told. Real cross-tenant claiming on that topology would mean a different mechanism entirely -- enumerating cleat_* databases via the master connection and unioning across them -- and that is not implemented.

Where this method does do what its name says: a deployment that points NewMySQLStore at one shared database and relies on the tenant_id column alone, with no MySQLStoreFactory in the picture. That is the shape every MySQL test in this package uses, including the one this is tested against, and it is a real (if less common) deployment shape -- nothing in MySQLStore requires the factory's per-tenant-database topology.

func (*MySQLStore) CleanupMemorySamples 

func (s *MySQLStore) CleanupMemorySamples(ctx context.Context, maxSamplesPerDef int) (int64, error)

CleanupMemorySamples deletes samples beyond maxSamplesPerDef per def_name.

func (*MySQLStore) ClearStickyWorker 

func (s *MySQLStore) ClearStickyWorker(ctx context.Context, workflowID string) error

ClearStickyWorker removes the sticky worker assignment.

func (*MySQLStore) CompactHistory 

func (s *MySQLStore) CompactHistory(ctx context.Context, workflowID string, compactionState []byte, compactionStep int, keepStep int) error

CompactHistory deletes old events and persists the compaction checkpoint for a workflow. compactionStep records the step up to which events were compacted; keepStep controls which events are deleted (step < keepStep).

func (*MySQLStore) CompleteCallIntent 

func (s *MySQLStore) CompleteCallIntent(ctx context.Context, workflowID string, rec EventRecord, payload []byte, checksum string, workerID string, generation int64) error

func (*MySQLStore) CompleteUpdateRequest 

func (s *MySQLStore) CompleteUpdateRequest(ctx context.Context, workflowID, updateName, result, errMsg string) error

CompleteUpdateRequest marks an update request as completed with a result or error.

func (*MySQLStore) CompleteWorkflow 

func (s *MySQLStore) CompleteWorkflow(ctx context.Context, workflowID, workerID string, generation int64, result string, queryState map[string]string) error

CompleteWorkflow marks a workflow as completed with a result.

func (*MySQLStore) ContinueAsNew 

func (s *MySQLStore) ContinueAsNew(ctx context.Context, currentRunID, workerID string, generation int64, defName string, defVersion int, newInput json.RawMessage, newEvents []EventRecord, result string, queryState map[string]string, priority int) (string, error)

ContinueAsNew atomically creates a new workflow run AND completes the current one in a single database transaction. If the transaction fails neither operation takes effect. Returns the new run ID on success.

func (*MySQLStore) CountActiveInstances 

func (s *MySQLStore) CountActiveInstances(ctx context.Context, name string, version int) (int, error)

CountActiveInstances returns the number of running/ready instances for a version.

func (*MySQLStore) CountEventHistory 

func (s *MySQLStore) CountEventHistory(ctx context.Context, workflowID string) (int, error)

CountEventHistory returns the total number of events for a workflow.

func (*MySQLStore) CreatePromise 

func (s *MySQLStore) CreatePromise(ctx context.Context, workflowID, promiseName, promiseID string) error

CreatePromise creates a new promise for a workflow.

func (*MySQLStore) CreateSchedule 

func (s *MySQLStore) CreateSchedule(ctx context.Context, sch Schedule) error

CreateSchedule inserts a new cron schedule.

func (*MySQLStore) CreateUpdateRequest 

func (s *MySQLStore) CreateUpdateRequest(ctx context.Context, workflowID, updateName, payload, promiseID string) error

CreateUpdateRequest registers an incoming update request for a workflow.

func (*MySQLStore) DeleteCompletedWorkflows 

func (s *MySQLStore) DeleteCompletedWorkflows(ctx context.Context, olderThan time.Time) (int64, error)

DeleteCompletedWorkflows permanently deletes workflow_instances rows in a terminal, no-further-action status ('done', 'failed', 'terminated') whose completed_at is older than the cutoff. 'dead_lettered' is deliberately excluded -- see the interface doc (store_interface.go) and DeleteDeadLetteredWorkflows above.

Unlike PostgresStore's DeleteCompletedWorkflows, no explicit event_history delete is needed here: migrations/mysql/001_schema.sql declares event_history's FK to workflow_instances ON DELETE CASCADE and MySQL never dropped it (only PostgreSQL did, deliberately, in migrations/postgres/003_procedures.sql), so deleting the workflow_instances row below cascades event_history (and workflow_signals, workflow_promises, concurrency_keys, workflow_update_requests) automatically.

func (*MySQLStore) DeleteDeadLetteredWorkflows 

func (s *MySQLStore) DeleteDeadLetteredWorkflows(ctx context.Context, olderThan time.Time) (int64, error)

DeleteDeadLetteredWorkflows permanently deletes dead-lettered workflow instances whose completed_at is older than the cutoff. Child rows (event_history, signals, promises, concurrency_keys, update_requests) are automatically deleted via ON DELETE CASCADE.

func (*MySQLStore) DeleteExpiredEvents 

func (s *MySQLStore) DeleteExpiredEvents(ctx context.Context, olderThan time.Time) (int64, error)

DeleteExpiredEvents deletes event history rows for workflows that are in a terminal state (completed/failed) and whose last update is older than the cutoff time. It also cleans up associated compaction states. Returns the number of event rows deleted.

func (*MySQLStore) DeleteSchedule 

func (s *MySQLStore) DeleteSchedule(ctx context.Context, name string) error

DeleteSchedule removes a schedule by name.

func (*MySQLStore) DeliverSignal 

func (s *MySQLStore) DeliverSignal(ctx context.Context, workflowID, signalName, payload string) error

DeliverSignal stores a signal for a workflow. Uses ON DUPLICATE KEY UPDATE so that re-delivering the same signal name replaces the payload.

func (*MySQLStore) DeployWorkflowDef 

func (s *MySQLStore) DeployWorkflowDef(ctx context.Context, def *WorkflowDef) error

DeployWorkflowDef inserts or updates a workflow definition.

func (*MySQLStore) FailWorkflow 

func (s *MySQLStore) FailWorkflow(ctx context.Context, workflowID, workerID string, generation int64, errorMsg, errorCode, errorOp string, queryState map[string]string) error

FailWorkflow marks a workflow as failed.

func (*MySQLStore) FinalizeWorkflowSegment 

func (s *MySQLStore) FinalizeWorkflowSegment(ctx context.Context, runID, workerID string, generation int64, newEvents []EventRecord, finalStatus string, result string, errorCode string, errorOp string, queryState map[string]string, nextWakeAt time.Time) error

FinalizeWorkflowSegment atomically appends new events and updates the workflow status in a single database transaction. This eliminates the race between AppendEventHistoryBatch and the subsequent CompleteWorkflow / FailWorkflow / ReleaseWorkflow call.

finalStatus must be one of:

  • "done" — marks the workflow as completed with the given result
  • "failed" — marks the workflow as failed with the given error info
  • "ready" — returns the workflow to the ready queue (suspend)

Fields not relevant to the chosen status are ignored.

func (*MySQLStore) GetActiveInstanceCountsByVersion 

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

GetActiveInstanceCountsByVersion returns a map of "name:version" -> count for all workflow definitions that have active instances.

func (*MySQLStore) GetAllowedSignalCallers 

func (s *MySQLStore) GetAllowedSignalCallers(ctx context.Context, workflowID string) ([]string, error)

GetAllowedSignalCallers returns the allowed_signals list for a workflow.

func (*MySQLStore) GetChildCount 

func (s *MySQLStore) GetChildCount(ctx context.Context, parentWorkflowID string) (int, error)

GetChildCount returns the number of active (non-terminal) child workflows for the given parent workflow. Terminal statuses are excluded.

func (*MySQLStore) GetChildResult 

func (s *MySQLStore) GetChildResult(ctx context.Context, runID string) (string, bool, error)

GetChildResult checks whether a child workflow has completed and returns its result.

func (*MySQLStore) GetCompactionCandidates 

func (s *MySQLStore) GetCompactionCandidates(ctx context.Context, threshold int, limit int) ([]string, error)

GetCompactionCandidates returns up to limit workflow IDs whose event history exceeds the threshold and could benefit from compaction.

func (*MySQLStore) GetConcurrencyKeyCount 

func (s *MySQLStore) GetConcurrencyKeyCount(ctx context.Context, workflowID string) (int, error)

GetConcurrencyKeyCount returns the number of non-expired concurrency keys held by the given workflow.

func (*MySQLStore) GetDueSchedules 

func (s *MySQLStore) GetDueSchedules(ctx context.Context) ([]Schedule, error)

GetDueSchedules returns enabled schedules whose next_run_at <= NOW(6).

func (*MySQLStore) GetDueSchedulesAcrossTenants 

func (s *MySQLStore) GetDueSchedulesAcrossTenants(ctx context.Context) ([]Schedule, error)

GetDueSchedulesAcrossTenants returns every tenant's due schedules.

Same story as ClaimWorkflowsAcrossTenants, and the same refusal. MySQL has no row-level security, so this is the tenant-scoped query with its predicate dropped -- which widens nothing on the topology cmd/cleat-worker actually builds, because MySQLStoreFactory gives each tenant its OWN physical database (cleat_<tenant_id>). The other tenants' schedules are not filtered out, they are in another database, so the query would return one tenant's schedules and report that it had swept them all.

So it refuses, and the worker warns once and falls back to the per-tenant read. Against a single shared database -- NewMySQLStore with no factory, which is what every MySQL test in this package uses and a real if less common deployment -- it does what its name says.

func (*MySQLStore) GetEventCount 

func (s *MySQLStore) GetEventCount(ctx context.Context, workflowID string) (int, error)

GetEventCount returns the event_count for a workflow instance.

func (*MySQLStore) GetPendingUpdateRequests 

func (s *MySQLStore) GetPendingUpdateRequests(ctx context.Context, workflowID string) ([]UpdateRequestInfo, error)

GetPendingUpdateRequests returns all pending (not yet dispatched) update requests.

func (*MySQLStore) GetPromise 

func (s *MySQLStore) GetPromise(ctx context.Context, workflowID, promiseID string) (status string, result string, errMsg string, err error)

GetPromise returns the current status and result of a promise.

func (*MySQLStore) GetQueryState 

func (s *MySQLStore) GetQueryState(ctx context.Context, workflowID, key string) (string, error)

GetQueryState returns the query state for a workflow instance key.

func (*MySQLStore) GetRoutingRules 

func (s *MySQLStore) GetRoutingRules(ctx context.Context, workflowName string) ([]RoutingRule, error)

GetRoutingRules returns all routing rules for a workflow.

func (*MySQLStore) GetWASMLength 

func (s *MySQLStore) GetWASMLength(ctx context.Context, defName string, defVersion int) (int64, error)

GetWASMLength returns the byte length of the stored WASM binary.

func (*MySQLStore) GetWorkflowByID 

func (s *MySQLStore) GetWorkflowByID(ctx context.Context, id string) (*WorkflowInstance, error)

GetWorkflowByID returns a single workflow instance by ID.

func (*MySQLStore) GetWorkflowDef 

func (s *MySQLStore) GetWorkflowDef(ctx context.Context, name string, version int) (*WorkflowDef, error)

GetWorkflowDef returns a single workflow definition by name and version.

func (*MySQLStore) GetWorkflowTag 

func (s *MySQLStore) GetWorkflowTag(ctx context.Context, workflowName string, tag string) (int, error)

GetWorkflowTag returns the version for a given tag.

func (*MySQLStore) GetWorkflowTags 

func (s *MySQLStore) GetWorkflowTags(ctx context.Context, workflowName string) (map[string]int, error)

GetWorkflowTags returns all tag -> version mappings for a workflow.

func (*MySQLStore) Heartbeat 

func (s *MySQLStore) Heartbeat(ctx context.Context, workflowID, workerID string, generation int64) (bool, error)

Heartbeat updates the heartbeat timestamp to prevent timeout. Returns false if the workflow is no longer assigned to this worker.

func (*MySQLStore) ListPromises 

func (s *MySQLStore) ListPromises(ctx context.Context, workflowID string) ([]PromiseInfo, error)

ListPromises returns all promises for a workflow ordered by creation time.

func (*MySQLStore) ListSchedules 

func (s *MySQLStore) ListSchedules(ctx context.Context) ([]Schedule, error)

ListSchedules returns all registered schedules for the current tenant.

func (*MySQLStore) ListVersions 

func (s *MySQLStore) ListVersions(ctx context.Context, defName string) ([]int, error)

ListVersions returns all deployed versions of a workflow.

func (*MySQLStore) ListWorkflowDefs 

func (s *MySQLStore) ListWorkflowDefs(ctx context.Context, name string) ([]WorkflowDef, error)

ListWorkflowDefs returns all versions of a workflow, ordered by version DESC. If name is empty, returns all workflow definitions across all workflows.

func (*MySQLStore) ListWorkflows 

func (s *MySQLStore) ListWorkflows(ctx context.Context, filter WorkflowFilter) ([]WorkflowInstance, error)

ListWorkflows returns workflow instances filtered by the given filter parameters.

func (*MySQLStore) LoadCompactionState 

func (s *MySQLStore) LoadCompactionState(ctx context.Context, workflowID string) (*CompactionState, error)

LoadCompactionState returns the compaction state for a workflow, or nil if the workflow has not been compacted.

func (*MySQLStore) LoadDAGSpec 

func (s *MySQLStore) LoadDAGSpec(ctx context.Context, defName string, defVersion int) (json.RawMessage, error)

LoadDAGSpec returns the dag_spec JSON for a workflow definition, or nil if none.

func (*MySQLStore) LoadEventHistory 

func (s *MySQLStore) LoadEventHistory(ctx context.Context, workflowID string) ([]EventRecord, error)

LoadEventHistory returns the full event history for a workflow, ordered by step.

func (*MySQLStore) LoadEventHistoryPaginated 

func (s *MySQLStore) LoadEventHistoryPaginated(ctx context.Context, workflowID string, offset, limit int) ([]EventRecord, error)

LoadEventHistoryPaginated returns a page of event history for a workflow, with offset and limit support. Defaults limit to 1000 if limit <= 0, capped at 1000.

func (*MySQLStore) LoadMemoryEstimates 

func (s *MySQLStore) LoadMemoryEstimates(ctx context.Context) (map[string]float64, error)

LoadMemoryEstimates returns EWMA mean bytes for all def_names.

func (*MySQLStore) LoadMemoryStats 

func (s *MySQLStore) LoadMemoryStats(ctx context.Context) ([]WorkflowMemoryStats, error)

LoadMemoryStats returns full distribution statistics for all def_names. Percentiles are computed in Go since MySQL lacks PERCENTILE_CONT.

func (*MySQLStore) LoadWASM 

func (s *MySQLStore) LoadWASM(ctx context.Context, defName string, defVersion int) ([]byte, error)

LoadWASM returns the compiled WASM bytes for a workflow definition.

func (*MySQLStore) LoadWorkflowConfig 

func (s *MySQLStore) LoadWorkflowConfig(ctx context.Context, defName string, defVersion int) (int, error)

LoadWorkflowConfig returns the max_history_length for a workflow definition.

func (*MySQLStore) MarkVersionDeprecated 

func (s *MySQLStore) MarkVersionDeprecated(ctx context.Context, name string, version int, deprecated bool) error

MarkVersionDeprecated sets the deprecated flag on a workflow version.

func (*MySQLStore) MoveToDeadLetterQueue 

func (s *MySQLStore) MoveToDeadLetterQueue(ctx context.Context, workflowID, workerID string, generation int64, errMsg, errorCode, errorOp string) error

MoveToDeadLetterQueue marks a workflow as dead_lettered because it failed after exhausting all retry attempts. This is a terminal status similar to 'failed' but indicates the workflow was retried without success.

func (*MySQLStore) PickVersionByRouting 

func (s *MySQLStore) PickVersionByRouting(ctx context.Context, workflowName string) (int, error)

PickVersionByRouting performs weighted random version selection. Returns 0 if no routing rules exist.

func (*MySQLStore) PollAndClaimSignal 

func (s *MySQLStore) PollAndClaimSignal(ctx context.Context, workflowID, signalName string) (string, bool, error)

PollAndClaimSignal atomically checks for and claims a pending signal. Uses SELECT ... FOR UPDATE followed by DELETE in a transaction to emulate PostgreSQL's DELETE ... RETURNING.

func (*MySQLStore) PollCancellation 

func (s *MySQLStore) PollCancellation(ctx context.Context, workflowID string) (bool, string, error)

PollCancellation checks whether the workflow has been cancelled.

func (*MySQLStore) PollSignal 

func (s *MySQLStore) PollSignal(ctx context.Context, workflowID, signalName string) (string, bool, error)

PollSignal checks for a delivered signal without consuming it. This is non-destructive — the signal remains available after polling.

func (*MySQLStore) PurgeWorkflowDef 

func (s *MySQLStore) PurgeWorkflowDef(ctx context.Context, name string, version int) error

PurgeWorkflowDef permanently deletes a workflow definition (WASM bytes and all).

func (*MySQLStore) QueueDepth 

func (s *MySQLStore) QueueDepth(ctx context.Context) (int64, error)

QueueDepth returns the count of ready workflows in the store's task queues.

func (*MySQLStore) ReapExpiredConcurrencyKeys 

func (s *MySQLStore) ReapExpiredConcurrencyKeys(ctx context.Context) (int64, error)

ReapExpiredConcurrencyKeys deletes all expired concurrency keys for the current tenant. Returns the number of keys deleted.

func (*MySQLStore) ReapStaleInstances 

func (s *MySQLStore) ReapStaleInstances(ctx context.Context, timeout time.Duration) (int, error)

ReapStaleInstances reclaims workflow instances that have been running but whose heartbeat has not been updated within the given timeout. Returns the number of instances reclaimed.

func (*MySQLStore) RecordWorkflowMemorySample 

func (s *MySQLStore) RecordWorkflowMemorySample(ctx context.Context, defName string, sampleBytes int64) error

RecordWorkflowMemorySample inserts a new memory sample and updates the EWMA summary.

func (*MySQLStore) RejectPromise 

func (s *MySQLStore) RejectPromise(ctx context.Context, workflowID, promiseID, errMsg string) error

RejectPromise marks a promise as rejected with the given error message. Also wakes the workflow instance so it can pick up the rejected promise on the next poll cycle instead of waiting for the original timeout.

func (*MySQLStore) ReleaseConcurrencyKey 

func (s *MySQLStore) ReleaseConcurrencyKey(ctx context.Context, key string) error

ReleaseConcurrencyKey releases a specific concurrency key.

func (*MySQLStore) ReleaseWorkflow 

func (s *MySQLStore) ReleaseWorkflow(ctx context.Context, workflowID, workerID string, generation int64, nextWakeAt time.Time) error

ReleaseWorkflow returns a workflow to the ready queue with a next wake time. Used when a workflow suspends (sleep/await signals).

func (*MySQLStore) ReleaseWorkflowConcurrencyKeys 

func (s *MySQLStore) ReleaseWorkflowConcurrencyKeys(ctx context.Context, workflowID string) error

ReleaseWorkflowConcurrencyKeys releases all concurrency keys held by a workflow. Runs as a best-effort operation.

func (*MySQLStore) RemoveRoutingRule 

func (s *MySQLStore) RemoveRoutingRule(ctx context.Context, ruleID string) error

RemoveRoutingRule deletes a routing rule by ID.

func (*MySQLStore) RemoveWorkflowTag 

func (s *MySQLStore) RemoveWorkflowTag(ctx context.Context, workflowName string, tag string) error

RemoveWorkflowTag deletes a tag assignment.

func (*MySQLStore) RequestCancellation 

func (s *MySQLStore) RequestCancellation(ctx context.Context, workflowID, reason string) error

RequestCancellation sets the cancellation flag on a workflow.

func (*MySQLStore) ResolveCallIntent 

func (s *MySQLStore) ResolveCallIntent(ctx context.Context, workflowID string, rec EventRecord, payload []byte, workerID string, generation int64) error

func (*MySQLStore) ResolveLatestVersion 

func (s *MySQLStore) ResolveLatestVersion(ctx context.Context, defName string) (int, error)

ResolveLatestVersion resolves the latest (highest) version for a named definition.

func (*MySQLStore) ResolvePromise 

func (s *MySQLStore) ResolvePromise(ctx context.Context, workflowID, promiseID, result string) error

ResolvePromise marks a promise as resolved with the given result. Also wakes the workflow instance so it can pick up the resolved promise on the next poll cycle instead of waiting for the original timeout.

func (*MySQLStore) ResolveTenantFromAPIKey 

func (s *MySQLStore) ResolveTenantFromAPIKey(ctx context.Context, keyHash []byte) (uuid.UUID, error)

ResolveTenantFromAPIKey looks up a tenant UUID by API key hash.

func (*MySQLStore) ResolveVersionByTag 

func (s *MySQLStore) ResolveVersionByTag(ctx context.Context, workflowName string, tag string) (int, error)

ResolveVersionByTag resolves a tag to a version number. If tag is "latest", returns the highest non-deprecated version.

func (*MySQLStore) RetryWorkflow 

func (s *MySQLStore) RetryWorkflow(ctx context.Context, workflowID string) error

RetryWorkflow moves a dead_lettered workflow back to a runnable state.

func (*MySQLStore) SetRoutingRule 

func (s *MySQLStore) SetRoutingRule(ctx context.Context, workflowName string, targetVersion int, weight float64) error

SetRoutingRule creates a routing rule for a workflow version.

func (*MySQLStore) SetScheduleEnabled 

func (s *MySQLStore) SetScheduleEnabled(ctx context.Context, name string, enabled bool) error

SetScheduleEnabled enables or disables a schedule.

func (*MySQLStore) SetWorkflowTag 

func (s *MySQLStore) SetWorkflowTag(ctx context.Context, workflowName string, version int, tag string) error

SetWorkflowTag assigns a tag to a specific version. Uses INSERT ... ON DUPLICATE KEY UPDATE so reassigning a tag updates in place.

func (*MySQLStore) StartChildWorkflow 

func (s *MySQLStore) StartChildWorkflow(ctx context.Context, parentID, defName, inputJSON string, defVersion int, parentClosePolicy string, priority int) (string, error)

StartChildWorkflow creates a child workflow instance linked to a parent. defVersion is the explicit workflow definition version to use, or 0 to use default resolution (SELECT MAX(version)).

func (*MySQLStore) StartChildWorkflowAtomic 

func (s *MySQLStore) StartChildWorkflowAtomic(ctx context.Context, childID, parentID, defName, inputJSON string, defVersion int, parentClosePolicy string, event EventRecord, priority int) (string, error)

StartChildWorkflowAtomic creates a child workflow and records the parent's child_workflow event in a single transaction, guaranteeing exactly-once creation.

func (*MySQLStore) StartNewRun 

func (s *MySQLStore) StartNewRun(ctx context.Context, runID, defName string, defVersion int, input json.RawMessage, idempotencyKey string, tenantID string, priority int) (string, bool, error)

StartNewRun creates a new workflow instance. If idempotencyKey is non-empty, provides exactly-once semantics: a subsequent call with the same key returns the existing workflow ID without creating a duplicate. Returns the workflow ID, whether it already existed, and any error.

func (*MySQLStore) StreamEventHistory 

func (s *MySQLStore) StreamEventHistory(ctx context.Context, workflowID string, pageSize int) (<-chan EventRecord, <-chan error)

StreamEventHistory loads event history for a workflow in pages, returning events through a channel. Events are fetched in pages of pageSize as the caller reads from the channel. The channel is closed when all events have been sent.

func (*MySQLStore) TerminateWorkflow 

func (s *MySQLStore) TerminateWorkflow(ctx context.Context, workflowID, reason string) error

TerminateWorkflow force-terminates a workflow, setting status to 'terminated'.

func (*MySQLStore) TraceWorkflow 

func (s *MySQLStore) TraceWorkflow(ctx context.Context, workflowID, traceID string) error

TraceWorkflow sets the W3C trace_id on a workflow instance.

func (*MySQLStore) UpdateScheduleNextRun 

func (s *MySQLStore) UpdateScheduleNextRun(ctx context.Context, name string, nextRun time.Time) error

UpdateScheduleNextRun updates a schedule's next_run_at after firing.

func (*MySQLStore) UpdateStickyWorker 

func (s *MySQLStore) UpdateStickyWorker(ctx context.Context, workflowID, workerID string) error

UpdateStickyWorker sets the sticky worker for a workflow.

func (*MySQLStore) ValidateVersion 

func (s *MySQLStore) ValidateVersion(ctx context.Context, defName string, defVersion int) (bool, error)

ValidateVersion checks whether the given version is valid (exists and not deprecated).

func (*MySQLStore) VerifyWorkflowEvents 

func (s *MySQLStore) VerifyWorkflowEvents(ctx context.Context, workflowID string) error

VerifyWorkflowEvents loads all events for a workflow and verifies their integrity by recomputing SHA-256 checksums and comparing them against the stored checksums. Before the checksum column migration, it loads and computes checksums silently and returns nil.

func (*MySQLStore) WithEncryption 

func (s *MySQLStore) WithEncryption(enc *PayloadEncryption, enabled bool) *MySQLStore

WithEncryption returns a copy of the store with encryption at rest enabled. NOTE: Encryption at rest is not yet supported on MySQL backends. This method is present for forward compatibility so that StreamEventHistory can contain the same decryption guard as the Postgres variant.

func (*MySQLStore) WithIdempotencyKeyTTL 

func (s *MySQLStore) WithIdempotencyKeyTTL(ttl time.Duration) *MySQLStore

WithIdempotencyKeyTTL returns a copy of the store with the given idempotency key TTL.

func (*MySQLStore) WithLogger 

func (s *MySQLStore) WithLogger(l *slog.Logger) *MySQLStore

WithLogger returns a copy of the store with the given structured logger.

func (*MySQLStore) WithReadRedactionDisabled 

func (s *MySQLStore) WithReadRedactionDisabled(disabled bool) *MySQLStore

WithReadRedactionDisabled returns a copy of the store with redaction on the read path disabled. This is used during replay to avoid the overhead of retroactive redaction on every event load.

func (*MySQLStore) WithTenant 

func (s *MySQLStore) WithTenant(tenantID string) *MySQLStore

WithTenant returns a copy of the store scoped to the given tenant ID. This is used in the dispatch loop to set the correct tenant context before executing a workflow. The returned store's methods will add WHERE tenant_id = ? to every tenant-scoped query.

func (*MySQLStore) WriteCallIntent 

func (s *MySQLStore) WriteCallIntent(ctx context.Context, workflowID string, rec EventRecord, workerID string, generation int64) error

type MySQLStoreFactory 

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

MySQLStoreFactory implements StoreFactory for MySQL/MariaDB with per-tenant database isolation. Each tenant gets its own MySQL database (cleat_<tenant_id>), and the connection pool is scoped to that database.

func NewMySQLStoreFactory 

func NewMySQLStoreFactory(masterDB *sql.DB, baseDSN string, idempotencyKeyTTL ...time.Duration) *MySQLStoreFactory

NewMySQLStoreFactory creates a MySQLStoreFactory. masterDB is a *sql.DB connected without a default database (used for administrative operations like CREATE DATABASE). baseDSN is a DSN template with connection parameters but without a database name — the per-tenant database name is appended for each tenant's connection.

func (*MySQLStoreFactory) Close 

func (f *MySQLStoreFactory) Close() error

Close closes all tenant connection pools.

func (*MySQLStoreFactory) CreateTenantDatabase 

func (f *MySQLStoreFactory) CreateTenantDatabase(ctx context.Context, tenantID string) (*sql.DB, error)

CreateTenantDatabase creates a new database for the given tenant and returns a connection pool scoped to that database. It is idempotent — if the database already exists, it just opens a new pool to it.

func (*MySQLStoreFactory) Dialect 

func (f *MySQLStoreFactory) Dialect() Dialect

Dialect returns DialectMySQL.

func (*MySQLStoreFactory) DriverName 

func (f *MySQLStoreFactory) DriverName() string

DriverName returns "mysql".

func (*MySQLStoreFactory) DropTenantDatabase 

func (f *MySQLStoreFactory) DropTenantDatabase(tenantID string) error

DropTenantDatabase removes a tenant database and closes its connection pool.

func (*MySQLStoreFactory) OpenStore 

func (f *MySQLStoreFactory) OpenStore(ctx context.Context, tenantID string, taskQueues ...string) (WorkflowStore, io.Closer, error)

OpenStore creates a MySQLStore scoped to the given tenant and task queues. The store's connection pool is scoped to the tenant's database.

NOTE: Encryption at rest (--encrypt-sensitive-payloads) is not yet supported on MySQL backends. See PostgresStore.WithEncryption for the reference implementation.

func (*MySQLStoreFactory) TenantDB 

func (f *MySQLStoreFactory) TenantDB(ctx context.Context, tenantID string) (*sql.DB, error)

TenantDB returns a *sql.DB connection pool for the given tenant, creating a new per-tenant database and connection pool if one does not already exist.

func (*MySQLStoreFactory) WithLogger 

func (f *MySQLStoreFactory) WithLogger(l *slog.Logger) *MySQLStoreFactory

WithLogger sets the structured logger on the factory. Stores created by OpenStore will inherit it.

func (*MySQLStoreFactory) WithTenantPoolMaxConns 

func (f *MySQLStoreFactory) WithTenantPoolMaxConns(n int) *MySQLStoreFactory

WithTenantPoolMaxConns sets the max open connections per tenant pool.

type PayloadEncryption 

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

PayloadEncryption provides AES-256-GCM encryption and decryption for sensitive event payload fields. The key must be exactly 32 bytes after base64 decoding. The wire format is: 

base64(nonce || ciphertext)

where nonce is 12 random bytes and ciphertext includes the 16-byte GCM authentication tag.

func NewPayloadEncryption 

func NewPayloadEncryption(keyBase64 string) (*PayloadEncryption, error)

NewPayloadEncryption creates a PayloadEncryption from a base64-encoded key string. The decoded key must be exactly 32 bytes (AES-256).

func (*PayloadEncryption) Decrypt 

func (pe *PayloadEncryption) Decrypt(data []byte) ([]byte, error)

Decrypt decrypts data produced by Encrypt (nonce || ciphertext).

func (*PayloadEncryption) DecryptBase64 

func (pe *PayloadEncryption) DecryptBase64(encoded string) ([]byte, error)

DecryptBase64 base64-decodes the input then decrypts the result.

func (*PayloadEncryption) DecryptJSON 

func (pe *PayloadEncryption) DecryptJSON(jsonValue []byte) ([]byte, error)

DecryptJSON parses a JSON string literal containing base64-encoded ciphertext and decrypts it, returning the original JSON bytes.

func (*PayloadEncryption) DecryptString 

func (pe *PayloadEncryption) DecryptString(encoded string) (string, error)

DecryptString decrypts a base64-encoded ciphertext string.

func (*PayloadEncryption) Encrypt 

func (pe *PayloadEncryption) Encrypt(plaintext []byte) ([]byte, error)

Encrypt encrypts plaintext using AES-256-GCM and returns nonce || ciphertext.

NOTE: In a multi-tenant deployment, the tenant_id should be passed as additional authenticated data (AAD) to bind ciphertexts to their tenant and prevent cross-tenant ciphertext substitution. This defense-in-depth improvement is reserved for future work; currently AAD is nil.

func (*PayloadEncryption) EncryptJSON 

func (pe *PayloadEncryption) EncryptJSON(jsonBytes []byte) ([]byte, error)

EncryptJSON encrypts JSON bytes and returns them as a JSON string literal (i.e., a quoted base64 string that is valid JSON). This allows encrypted payloads to be stored in JSONB columns.

func (*PayloadEncryption) EncryptString 

func (pe *PayloadEncryption) EncryptString(plaintext string) (string, error)

EncryptString encrypts a plaintext string and returns base64-encoded output.

type PluginCallEvent 

type PluginCallEvent struct {
	PluginName string
	FuncName   string
	Input      string
	Output     string
	Err        string
	Idempotent bool
	// contains filtered or unexported fields
}

PluginCallEvent records a plugin host function invocation.

func (PluginCallEvent) Step 

func (e PluginCallEvent) Step() int

func (PluginCallEvent) Type 

func (e PluginCallEvent) Type() EventType

type PluginCallGuard 

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

PluginCallGuard enforces call_plugin capability restrictions. It restricts which WASM plugins can call which other plugins' functions.

func NewPluginCallGuard 

func NewPluginCallGuard() *PluginCallGuard

NewPluginCallGuard creates a new PluginCallGuard with no restrictions.

func (*PluginCallGuard) Allow 

func (g *PluginCallGuard) Allow(callerName string, targets []string)

Allow sets which plugins the given caller can call. targets of ["*"] means allow all.

func (*PluginCallGuard) Check 

func (g *PluginCallGuard) Check(callerName, targetName string) error

Check verifies that caller is allowed to call target. Returns nil if allowed, or an error if denied. If caller is not in the guard (e.g., workflow code, Go compile-time plugins), the call is always allowed.

type PluginCallObserver 

type PluginCallObserver func(pluginName, functionName string, d time.Duration, err error)

PluginCallObserver is called after every plugin function invocation with the call details and duration. Set via WithPluginCallObserver to record metrics.

type PluginCallStreamChunkEvent 

type PluginCallStreamChunkEvent struct {
	PluginName string
	FuncName   string
	Input      string
	Output     string
	ChunkIndex int
	Finish     bool
	// contains filtered or unexported fields
}

PluginCallStreamChunkEvent records one chunk from a streaming plugin call.

func (PluginCallStreamChunkEvent) Step 

func (e PluginCallStreamChunkEvent) Step() int

func (PluginCallStreamChunkEvent) Type 

func (e PluginCallStreamChunkEvent) Type() EventType

type PluginConstraint 

type PluginConstraint struct {
	Name       string `json:"name"`
	Constraint string `json:"constraint"` // semver: ">=1.2.0", "~1.2.0", "^1.2.0", "=1.2.0"
}

PluginConstraint represents a version constraint for a plugin dependency.

type PluginDef 

type PluginDef struct {
	Name       string          `json:"name"`
	Version    string          `json:"version"` // semver string, e.g. "1.2.3"
	WASMBytes  []byte          `json:"wasm_bytes,omitempty"`
	Config     json.RawMessage `json:"config"`
	CreatedAt  time.Time       `json:"created_at"`
	Deprecated bool            `json:"deprecated"`
}

PluginDef is a row from the plugin_defs table.

type PluginLoader 

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

PluginLoader loads plugin WASM modules from the plugin_defs table and resolves semver constraints to find the best matching plugin version.

Plugin versioning uses semver because plugins are consumed as libraries with compatibility ranges. This is distinct from workflow versioning which uses monotonic integers for discrete business process versions.

func NewPluginLoader 

func NewPluginLoader(db *sql.DB, rt *Runtime, maxSize ...int) *PluginLoader

NewPluginLoader creates a PluginLoader backed by the given database connection and wazero runtime. maxSize is the maximum number of compiled plugin modules to cache (defaults to 50 if <= 0).

func (*PluginLoader) DeployPlugin 

func (l *PluginLoader) DeployPlugin(ctx context.Context, name string, version string, wasmBytes []byte, config map[string]any) error

DeployPlugin inserts a new plugin definition into the database. If the definition already exists, it is updated (upsert semantics). 

sql: INSERT INTO plugin_defs (name, version, wasm_bytes, config)
     VALUES ($1, $2, $3, $4)
     ON CONFLICT (name, version) DO UPDATE SET
       wasm_bytes = $3, config = $4, deprecated = false, created_at = now()

func (*PluginLoader) DeployPluginWithCapabilities 

func (l *PluginLoader) DeployPluginWithCapabilities(ctx context.Context, name string, version string, wasmBytes []byte, config map[string]any, declared plugin.Capabilities) error

DeployPluginWithCapabilities is like DeployPlugin but additionally validates the declared capabilities against configured limits before deploying. If the capabilities violate the limits, the deployment is refused.

func (*PluginLoader) DeprecatePlugin 

func (l *PluginLoader) DeprecatePlugin(ctx context.Context, name string, version string) error

DeprecatePlugin marks a plugin version as deprecated. 

sql: UPDATE plugin_defs SET deprecated = true WHERE name = $1 AND version = $2

func (*PluginLoader) ListPluginVersions 

func (l *PluginLoader) ListPluginVersions(ctx context.Context, name string) ([]PluginDef, error)

ListPluginVersions returns all deployed versions of a plugin, ordered by semver descending. 

sql: SELECT name, version, wasm_bytes, config, created_at, deprecated
     FROM plugin_defs WHERE name = $1 ORDER BY version DESC

func (*PluginLoader) LoadPlugin 

func (l *PluginLoader) LoadPlugin(ctx context.Context, name string, version string) (wazero.CompiledModule, error)

LoadPlugin loads a compiled WASM module for a specific plugin version. Results are cached in an LRU cache keyed by (name, version). 

sql: SELECT wasm_bytes FROM plugin_defs
     WHERE name = $1 AND version = $2 AND NOT deprecated

func (*PluginLoader) ResolvePlugin 

func (l *PluginLoader) ResolvePlugin(ctx context.Context, name string, constraint string) (string, *PluginDef, error)

ResolvePlugin finds the best matching plugin version for the given semver constraint. It queries all non-deprecated versions of the named plugin and returns the highest version that satisfies the constraint. 

sql: SELECT name, version, wasm_bytes, config, created_at, deprecated
     FROM plugin_defs
     WHERE name = $1 AND NOT deprecated
     ORDER BY version

Resolution is done in Go using semver comparison after filtering by the constraint range.

func (*PluginLoader) SetLimits 

func (l *PluginLoader) SetLimits(limits plugin.CapabilityLimits)

SetLimits configures the maximum capabilities allowed for WASM plugins loaded through this loader. If limits is the zero value, no capability restrictions are enforced.

type PluginRegistry 

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

PluginRegistry maps plugin function names to implementations. It also tracks plugin health: if a plugin function panics, the entire plugin is marked unhealthy and all its functions return an error without being invoked.

func NewPluginRegistry 

func NewPluginRegistry() *PluginRegistry

func (*PluginRegistry) Has 

func (pr *PluginRegistry) Has(pluginName, funcName string) bool

Has reports whether a plugin function is registered.

func (*PluginRegistry) IsPluginHealthy 

func (pr *PluginRegistry) IsPluginHealthy(pluginName string) bool

IsPluginHealthy reports whether the given plugin has not panicked.

func (*PluginRegistry) Lookup 

func (pr *PluginRegistry) Lookup(pluginName, funcName string) (plugin.PluginFunc, bool, bool)

func (*PluginRegistry) MarkPluginUnhealthy 

func (pr *PluginRegistry) MarkPluginUnhealthy(pluginName string, err error)

MarkPluginUnhealthy marks a plugin as unhealthy with the given error. All future invocations of the plugin's host functions are blocked.

func (*PluginRegistry) PluginHealthStatus 

func (pr *PluginRegistry) PluginHealthStatus() []plugin.HealthStatus

PluginHealthStatus returns the current health status of all plugins that have been marked unhealthy. Healthy plugins are not included.

func (*PluginRegistry) Register 

func (pr *PluginRegistry) Register(pluginName, funcName string, fn plugin.PluginFunc) error

Register adds a plugin function. Returns an error if the function name is already registered for this plugin. The function is wrapped with panic recovery so a plugin crash does not take down the worker.

func (*PluginRegistry) RegisterIdempotent 

func (pr *PluginRegistry) RegisterIdempotent(pluginName, funcName string, fn plugin.PluginFunc) error

RegisterIdempotent registers a plugin function that is safe to re-invoke during replay (e.g., read-only S3 GET operations). The function is wrapped with panic recovery.

func (*PluginRegistry) SetHealthTracker 

func (pr *PluginRegistry) SetHealthTracker(t *plugin.PluginHealthTracker)

SetHealthTracker replaces the default health tracker with a shared one. Used to share a single tracker between PluginRegistry and PluginStreamRegistry so a panic in any function marks the plugin unhealthy across both registries.

func (*PluginRegistry) UnhealthyError 

func (pr *PluginRegistry) UnhealthyError(pluginName string) error

UnhealthyError returns the error that caused the plugin to be marked unhealthy, or nil if the plugin is healthy.

type PluginStreamRegistry 

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

PluginStreamRegistry maps plugin function names to streaming implementations.

func NewPluginStreamRegistry 

func NewPluginStreamRegistry() *PluginStreamRegistry

func (*PluginStreamRegistry) Has 

func (psr *PluginStreamRegistry) Has(pluginName, funcName string) bool

Has reports whether a streaming plugin function is registered.

func (*PluginStreamRegistry) IsPluginHealthy 

func (psr *PluginStreamRegistry) IsPluginHealthy(pluginName string) bool

IsPluginHealthy reports whether the given streaming plugin has not panicked.

func (*PluginStreamRegistry) Lookup 

func (psr *PluginStreamRegistry) Lookup(pluginName, funcName string) (plugin.PluginStreamFunc, bool)

func (*PluginStreamRegistry) MarkPluginUnhealthy 

func (psr *PluginStreamRegistry) MarkPluginUnhealthy(pluginName string, err error)

MarkPluginUnhealthy marks a streaming plugin as unhealthy with the given error.

func (*PluginStreamRegistry) PluginHealthStatus 

func (psr *PluginStreamRegistry) PluginHealthStatus() []plugin.HealthStatus

PluginHealthStatus returns the current health status of all streaming plugins that have been marked unhealthy. Healthy plugins are not included.

func (*PluginStreamRegistry) Register 

func (psr *PluginStreamRegistry) Register(pluginName, funcName string, fn plugin.PluginStreamFunc) error

func (*PluginStreamRegistry) RegisterStream 

func (psr *PluginStreamRegistry) RegisterStream(pluginName string, opts plugin.FuncOptions, fn plugin.PluginStreamFunc) error

RegisterStream implements plugin.StreamFuncRegistry.

func (*PluginStreamRegistry) SetHealthTracker 

func (psr *PluginStreamRegistry) SetHealthTracker(t *plugin.PluginHealthTracker)

SetHealthTracker replaces the default health tracker with a shared one. Used to share a single tracker between PluginRegistry and PluginStreamRegistry.

func (*PluginStreamRegistry) UnhealthyError 

func (psr *PluginStreamRegistry) UnhealthyError(pluginName string) error

UnhealthyError returns the error that caused the streaming plugin to be marked unhealthy, or nil if the plugin is healthy.

type PostgresStore 

type PostgresStore struct {
	Metrics *prometheus.Metrics
	// contains filtered or unexported fields
}

PostgresStore implements WorkflowStore using a PostgreSQL database.

func NewPostgresStore 

func NewPostgresStore(db *sql.DB, taskQueues ...string) *PostgresStore

NewPostgresStore creates a PostgresStore scoped to the given task queues. The taskQueues slice specifies which task queues this worker pool should poll (e.g., "default", "gpu", "high-memory"). Defaults to ["default"]. The tenantID defaults to the default tenant UUID from the tenant foundation migration.

func (*PostgresStore) AcquireConcurrencyKey 

func (s *PostgresStore) AcquireConcurrencyKey(ctx context.Context, key, workflowID string, ttl time.Duration) (bool, error)

CreatePromise creates a new promise for a workflow instance.

func (*PostgresStore) AdminForceComplete 

func (s *PostgresStore) AdminForceComplete(ctx context.Context, workflowID string, generation int64, result string, operator string) error

AdminForceComplete marks a workflow as done, bypassing worker ownership.

func (*PostgresStore) AdminForceFail 

func (s *PostgresStore) AdminForceFail(ctx context.Context, workflowID string, generation int64, errorMsg, errorCode string, operator string) error

AdminForceFail marks a workflow as failed, bypassing worker ownership.

func (*PostgresStore) AdminReReplay 

func (s *PostgresStore) AdminReReplay(ctx context.Context, workflowID string, generation int64, operator string) error

AdminReReplay replays a workflow's event history for debugging.

func (*PostgresStore) AppendEventHistory 

func (s *PostgresStore) AppendEventHistory(ctx context.Context, workflowID string, rec EventRecord) error

func (*PostgresStore) AppendEventHistoryBatch 

func (s *PostgresStore) AppendEventHistoryBatch(ctx context.Context, workflowID string, recs []EventRecord) error

func (*PostgresStore) BatchHeartbeat 

func (s *PostgresStore) BatchHeartbeat(ctx context.Context, workerID string) (int64, error)

BatchHeartbeat updates heartbeat_at for all workflows assigned to this worker. NOTE: This intentionally does NOT check per-workflow generation because it operates on ALL running workflows for a worker, and generations differ per workflow. Individual generation-guarded operations (Heartbeat, CompleteWorkflow, FailWorkflow, etc.) prevent double-execution even if the batch heartbeat refreshes a stale workflow's heartbeat_at.

func (*PostgresStore) CheckCancellation 

func (s *PostgresStore) CheckCancellation(ctx context.Context, workflowID string) (bool, string, error)

func (*PostgresStore) CheckCrossTenantCapability 

func (s *PostgresStore) CheckCrossTenantCapability(ctx context.Context) CrossTenantCapability

CheckCrossTenantCapability answers from the PostgreSQL catalog.

It checks three things per function, and the third is the reason this exists at all: 

does it exist        migration not applied -- would raise 42883 at runtime
may this role EXECUTE  grant not made      -- would raise 42501 at runtime
does its OWNER have BYPASSRLS

The first two are detected at runtime too, as 42883 and 42501, and answered by falling back to the tenant-scoped path with a warning.

The third is different, and it is the reason this check earns its place. A function whose owner has lost BYPASSRLS is subject to the same policies as its caller, and these functions are called outside beginTxWithRLS so no tenant GUC is set on that connection. 001_schema.sql's policies are fail-closed -- cleat.assert_tenant_set() RAISES on an unset GUC rather than COALESCE-ing to a default -- so every call fails with 

cleat.tenant_id is not set -- tenant context required for RLS-scoped query

That is loud, which is good, and it was worth measuring rather than assuming: an earlier version of this comment (and of 023's header) claimed the call would quietly return fewer rows instead. It does not, and the difference is the fail-closed policy choice rather than luck.

What it is NOT is useful. P0001 names neither the function nor the attribute, it does not map to the provisioning-gap sentinel, so it propagates as a hard error on every tick, and an operator reading it has no path from "tenant context required" to "a role lost a privilege". This check names the cause, before the first tick.

023 and 024 set the attribute in the same migration that creates the role so it cannot drift on install, but a role is a database-wide object an operator can ALTER afterwards.

func (*PostgresStore) ClaimDueSchedule 

func (s *PostgresStore) ClaimDueSchedule(ctx context.Context, name string, expectedNextRun, newNextRun time.Time, runID string) (bool, error)

ClaimDueSchedule advances a schedule's next_run_at, but only if it still holds expectedNextRun. See the interface doc for why this is a CAS.

func (*PostgresStore) ClaimStickyWorkflows 

func (s *PostgresStore) ClaimStickyWorkflows(ctx context.Context, workerID string, limit int) ([]*WorkflowInstance, error)

func (*PostgresStore) ClaimWorkflow 

func (s *PostgresStore) ClaimWorkflow(ctx context.Context, workerID string) (*WorkflowInstance, error)

func (*PostgresStore) ClaimWorkflows 

func (s *PostgresStore) ClaimWorkflows(ctx context.Context, workerID string, limit int) ([]*WorkflowInstance, error)

func (*PostgresStore) ClaimWorkflowsAcrossTenants 

func (s *PostgresStore) ClaimWorkflowsAcrossTenants(ctx context.Context, workerID string, limit int) ([]*WorkflowInstance, error)

ClaimWorkflowsAcrossTenants claims runnable workflows for every tenant.

It calls admin.claim_workflows (migrations/postgres/023_cross_tenant_claim.sql) rather than issuing the claim directly, because the exemption that lets it see across tenants belongs to that function's owner and nowhere else. The statement inside is the same one ClaimWorkflows runs; the column list here is the contract with it.

No beginTxWithRLS. That helper exists to set the tenant GUC the policies read, and this call must not be scoped to a tenant -- setting one would filter the very rows it exists to find. The function performs its own UPDATE, so a single statement is already atomic.

func (*PostgresStore) CleanupMemorySamples 

func (s *PostgresStore) CleanupMemorySamples(ctx context.Context, maxSamplesPerDef int) (int64, error)

CleanupMemorySamples deletes samples beyond maxSamplesPerDef per def_name.

func (*PostgresStore) ClearStickyWorker 

func (s *PostgresStore) ClearStickyWorker(ctx context.Context, workflowID string) error

ClearStickyWorker removes the sticky worker assignment.

func (*PostgresStore) CompactHistory 

func (s *PostgresStore) CompactHistory(ctx context.Context, workflowID string, compactionState []byte, compactionStep int, keepStep int) error

CompactHistory deletes old events and saves compaction state for a workflow.

func (*PostgresStore) CompleteCallIntent 

func (s *PostgresStore) CompleteCallIntent(ctx context.Context, workflowID string, rec EventRecord, payload []byte, checksum string, workerID string, generation int64) error

func (*PostgresStore) CompleteUpdateRequest 

func (s *PostgresStore) CompleteUpdateRequest(ctx context.Context, workflowID, updateName, result, errMsg string) error

func (*PostgresStore) CompleteWorkflow 

func (s *PostgresStore) CompleteWorkflow(ctx context.Context, workflowID, workerID string, generation int64, result string, queryState map[string]string) error

func (*PostgresStore) ContinueAsNew 

func (s *PostgresStore) ContinueAsNew(ctx context.Context, currentRunID, workerID string, generation int64, defName string, defVersion int, newInput json.RawMessage, newEvents []EventRecord, result string, queryState map[string]string, priority int) (string, error)

func (*PostgresStore) CountActiveInstances 

func (s *PostgresStore) CountActiveInstances(ctx context.Context, name string, version int) (int, error)

func (*PostgresStore) CountEventHistory 

func (s *PostgresStore) CountEventHistory(ctx context.Context, workflowID string) (int, error)

func (*PostgresStore) CountEventHistoryTotal 

func (s *PostgresStore) CountEventHistoryTotal(ctx context.Context) (int, error)

CountEventHistoryTotal returns total rows in event_history.

func (*PostgresStore) CountStalledWorkflows 

func (s *PostgresStore) CountStalledWorkflows(ctx context.Context, threshold time.Duration) (int, error)

CountStalledWorkflows counts running workflows without recent progress.

func (*PostgresStore) CreatePromise 

func (s *PostgresStore) CreatePromise(ctx context.Context, workflowID, promiseName, promiseID string) error

func (*PostgresStore) CreateSchedule 

func (s *PostgresStore) CreateSchedule(ctx context.Context, sch Schedule) error

func (*PostgresStore) CreateUpdateRequest 

func (s *PostgresStore) CreateUpdateRequest(ctx context.Context, workflowID, updateName, payload, promiseID string) error

func (*PostgresStore) DeleteCompletedWorkflows 

func (s *PostgresStore) DeleteCompletedWorkflows(ctx context.Context, olderThan time.Time) (int64, error)

DeleteCompletedWorkflows permanently deletes workflow_instances rows in a terminal, no-further-action status ('done', 'failed', 'terminated') whose completed_at is older than the cutoff. See the interface doc (store_interface.go) for why 'dead_lettered' is deliberately excluded -- it has its own lifecycle and its own deletion path.

This is Finding S2: DeleteExpiredEvents deletes event_history for 'done'/'failed' workflows but never touches the workflow_instances row itself, so that table was unbounded by anything but lifetime workflow count. This is the method that actually reclaims it.

Follows deleteDeadLetteredWorkflowsBatch's pattern exactly, including the same FK-graph fix: event_history has no FK back to workflow_instances on PostgreSQL (dropped deliberately by migrations/postgres/003_procedures.sql because finalize_workflow_status deletes a 'done'/'failed' workflow's events itself) so it must be deleted explicitly here rather than assumed to cascade. That assumption is also wrong for 'terminated' workflows on this dialect specifically: TerminateWorkflow does not call finalize_workflow_status, so a force-terminated workflow's events are never deleted by any other path either.

func (*PostgresStore) DeleteDeadLetteredWorkflows 

func (s *PostgresStore) DeleteDeadLetteredWorkflows(ctx context.Context, olderThan time.Time) (int64, error)

DeleteDeadLetteredWorkflows permanently deletes dead-lettered workflow instances whose completed_at is older than the cutoff.

Two bugs were found and fixed here together (both discovered verifying Stream I / Finding S3's tenant-deletion work, which shares this function's FK-graph question):

  1. The previous version ran its DELETE on s.db directly -- the plain pool, with no RLS context set. workflow_instances carries `FORCE ROW LEVEL SECURITY` with a fail-closed policy (cleat.assert_tenant_set()), so under a real RLS-enforcing connection (any role that is not a superuser and does not own the table -- e.g. cleat_app in production) that statement does not silently do nothing: it raises "cleat.tenant_id is not set" and the whole call errors. Verified directly against a real cleat_rls_test_role connection: the old query, run as that role with the tenant_id predicate satisfied but no set_config call preceding it, fails with exactly that error. Fixed by running inside beginTxWithRLS, which calls setRLSOnTx before any query -- the same pattern every other tenant-scoped method here uses.
  2. The doc comment claimed child rows -- "event_history, signals, promises, concurrency_keys, update_requests" -- are "automatically deleted via ON DELETE CASCADE". True for four of the five, but migrations/postgres/003_procedures.sql deliberately DROPs the FK from event_history to workflow_instances ("no longer needed; events are deleted on terminal") because finalize_workflow_status() deletes a workflow's events itself when it reaches 'done' or 'failed'. MoveToDeadLetterQueue does not call finalize_workflow_status -- it does a plain UPDATE ... SET status = 'dead_lettered' -- so a dead-lettered workflow's event_history rows are never deleted there either. Verified directly: seeding a dead_lettered workflow with one event_history row and running the old DELETE FROM workflow_instances query left that event_history row in place, orphaned (no workflow_instances row it can still join to) and undeletable by any later call to this function, since it only ever looks at workflow_instances.status. Fixed by deleting event_history explicitly, by the same batch of IDs, in the same transaction.

func (*PostgresStore) DeleteExpiredEvents 

func (s *PostgresStore) DeleteExpiredEvents(ctx context.Context, olderThan time.Time) (int64, error)

DeleteExpiredEvents deletes event history rows for completed/failed workflows whose completed_at is older than the cutoff. It uses batching to avoid locking the event_history table when there are millions of rows to delete.

func (*PostgresStore) DeleteSchedule 

func (s *PostgresStore) DeleteSchedule(ctx context.Context, name string) error

func (*PostgresStore) DeliverSignal 

func (s *PostgresStore) DeliverSignal(ctx context.Context, workflowID, signalName, payload string) error

func (*PostgresStore) DeployWorkflowDef 

func (s *PostgresStore) DeployWorkflowDef(ctx context.Context, def *WorkflowDef) error

func (*PostgresStore) EstimateEventHistorySize 

func (s *PostgresStore) EstimateEventHistorySize(ctx context.Context) (int64, error)

EstimateEventHistorySize returns the estimated size of event_history in bytes.

func (*PostgresStore) FailWorkflow 

func (s *PostgresStore) FailWorkflow(ctx context.Context, workflowID, workerID string, generation int64, errorMsg, errorCode, errorOp string, queryState map[string]string) error

func (*PostgresStore) FinalizeWorkflowSegment 

func (s *PostgresStore) FinalizeWorkflowSegment(ctx context.Context, runID, workerID string, generation int64, newEvents []EventRecord, finalStatus string, result string, errorCode string, errorOp string, queryState map[string]string, nextWakeAt time.Time) error

func (*PostgresStore) GetActiveInstanceCountsByVersion 

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

func (*PostgresStore) GetAllowedSignalCallers 

func (s *PostgresStore) GetAllowedSignalCallers(ctx context.Context, workflowID string) ([]string, error)

func (*PostgresStore) GetChildCount 

func (s *PostgresStore) GetChildCount(ctx context.Context, parentWorkflowID string) (int, error)

func (*PostgresStore) GetChildResult 

func (s *PostgresStore) GetChildResult(ctx context.Context, runID string) (string, bool, error)

func (*PostgresStore) GetChildResultInSchema 

func (s *PostgresStore) GetChildResultInSchema(ctx context.Context, targetSchema, runID string) (string, bool, error)

func (*PostgresStore) GetCompactionCandidates 

func (s *PostgresStore) GetCompactionCandidates(ctx context.Context, threshold int, limit int) ([]string, error)

GetCompactionCandidates returns workflow IDs that need compaction.

func (*PostgresStore) GetConcurrencyKeyCount 

func (s *PostgresStore) GetConcurrencyKeyCount(ctx context.Context, workflowID string) (int, error)

GetConcurrencyKeyCount returns the number of non-expired concurrency keys held by the given workflow.

func (*PostgresStore) GetDueSchedules 

func (s *PostgresStore) GetDueSchedules(ctx context.Context) ([]Schedule, error)

func (*PostgresStore) GetDueSchedulesAcrossTenants 

func (s *PostgresStore) GetDueSchedulesAcrossTenants(ctx context.Context) ([]Schedule, error)

GetDueSchedulesAcrossTenants returns every tenant's due schedules.

It calls admin.get_due_schedules (migrations/postgres/024_cross_tenant_schedules.sql) for the same reason ClaimWorkflowsAcrossTenants calls admin.claim_workflows: the exemption that lets it see across tenants belongs to that function's owner and nowhere else.

No beginTxWithRLS. That helper sets the tenant GUC the policies read, and this call must not be scoped to a tenant -- setting one would filter the very rows it exists to find.

func (*PostgresStore) GetEventCount 

func (s *PostgresStore) GetEventCount(ctx context.Context, workflowID string) (int, error)

GetEventCount returns the event_count for a workflow instance.

func (*PostgresStore) GetPendingUpdateRequests 

func (s *PostgresStore) GetPendingUpdateRequests(ctx context.Context, workflowID string) ([]UpdateRequestInfo, error)

func (*PostgresStore) GetPromise 

func (s *PostgresStore) GetPromise(ctx context.Context, workflowID, promiseID string) (status string, result string, errMsg string, err error)

func (*PostgresStore) GetQueryState 

func (s *PostgresStore) GetQueryState(ctx context.Context, workflowID, key string) (string, error)

CompleteWorkflow marks a workflow as done.

func (*PostgresStore) GetRoutingRules 

func (s *PostgresStore) GetRoutingRules(ctx context.Context, workflowName string) ([]RoutingRule, error)

func (*PostgresStore) GetWASMLength 

func (s *PostgresStore) GetWASMLength(ctx context.Context, defName string, defVersion int) (int64, error)

func (*PostgresStore) GetWorkflowByID 

func (s *PostgresStore) GetWorkflowByID(ctx context.Context, id string) (*WorkflowInstance, error)

GetWorkflowByID returns a single workflow instance by ID.

func (*PostgresStore) GetWorkflowDef 

func (s *PostgresStore) GetWorkflowDef(ctx context.Context, name string, version int) (*WorkflowDef, error)

func (*PostgresStore) GetWorkflowTag 

func (s *PostgresStore) GetWorkflowTag(ctx context.Context, workflowName string, tag string) (int, error)

func (*PostgresStore) GetWorkflowTags 

func (s *PostgresStore) GetWorkflowTags(ctx context.Context, workflowName string) (map[string]int, error)

func (*PostgresStore) Heartbeat 

func (s *PostgresStore) Heartbeat(ctx context.Context, workflowID, workerID string, generation int64) (bool, error)

Heartbeat renews this worker's lease on one workflow instance, fenced on (assigned_to, generation), and reports whether the lease still held.

The decision: wire it, not delete it (B4)

This was the one per-workflow generation-checked heartbeat in the store and nothing called it: cmd/cleat-worker calls only BatchHeartbeat, which by its own doc comment does not check generation because it refreshes every workflow this worker holds in one statement. A generation-checked function nothing calls is a trap for the next reader -- it reads like a safety net that is actually just dead code.

Where it is actually used now

Not as the primary fencing mechanism for the two hot paths B4 found unfenced. Postgres's per-step flush (engine/flush.go's insertEventSQL) and all three dialects' write-ahead-intent statements (engine/store_intent.go) fold the (assigned_to, generation) check directly into the INSERT/UPDATE's own WHERE clause instead -- see insertEventSQL's doc for why: an earlier version of this fix called Heartbeat as a separate statement before every write, and a round-trip-counting measurement against the real test database found that cost exactly double the round trips per event (8 vs 4, tenanted path) for a guarantee that was still only an argument about timing, not an atomic fact. Heartbeat's own SQL did not need to change for that rewrite; the callers did.

Heartbeat is still called from three places, all of them the case where folding the check into the write statement was not available or not worth it:

  • flush.go's afterFencedInsert and store_intent.go's intentFenceOrNotPending call it as a *disambiguation* step, and only on the rare path where a fenced write's own statement reported zero rows affected -- distinguishing "the fence failed" from "the row was already terminal / not pending", which are both legitimate zero-row outcomes for different reasons. The common case (a row was actually written) never reaches this call.
  • engine/flush.go calls it once, upfront, before dispatching to MySQLStore's/MSSQLStore's flushEventForStep (flush_dialect.go). Those two dialects' per-step insert goes through appendEventsInTxOpts, a function also used for genuinely unfenced multi-event batch writes (FinalizeWorkflowSegment, AppendEventHistoryBatch), so folding a fence predicate into its SQL would fence those other callers too; a Heartbeat-before-write check, scoped to the one caller that needs it, was the trade made instead. See flush_dialect.go's perStepEventFlusher doc.
  • adaptive_flush.go's partitionFencedBatch does not call this method directly -- it runs the equivalent renewal for every distinct claim in a batch in one query -- but exists for the same reason: a single Heartbeat call cannot fence a batch spanning many workflow instances at once.

For the two call sites that still do a Heartbeat-before-write (unlike the disambiguation callers, these do incur it on every call, not just the rare path): a successful Heartbeat does not just check the lease, it renews it -- heartbeat_at = now(), unconditionally, for the row that matched. Since ReapStaleInstances only reclaims a workflow whose heartbeat_at predates the reap timeout (tens of seconds in every deployment config this repo ships), the window that matters is "between the renewal and the timeout elapsing", not "between the renewal and the write milliseconds later" -- which is why this is safe despite being two statements rather than one, for the two places it is still used that way.

func (*PostgresStore) ListPromises 

func (s *PostgresStore) ListPromises(ctx context.Context, workflowID string) ([]PromiseInfo, error)

func (*PostgresStore) ListSchedules 

func (s *PostgresStore) ListSchedules(ctx context.Context) ([]Schedule, error)

func (*PostgresStore) ListVersions 

func (s *PostgresStore) ListVersions(ctx context.Context, defName string) ([]int, error)

func (*PostgresStore) ListWorkflowDefs 

func (s *PostgresStore) ListWorkflowDefs(ctx context.Context, name string) ([]WorkflowDef, error)

func (*PostgresStore) ListWorkflows 

func (s *PostgresStore) ListWorkflows(ctx context.Context, filter WorkflowFilter) ([]WorkflowInstance, error)

ListWorkflows returns workflow instances filtered by the given filter parameters, ordered by creation time DESC. Supports search by input content, error message, and combined full-text search, as well as pagination via Offset/Limit.

func (*PostgresStore) LoadCompactionState 

func (s *PostgresStore) LoadCompactionState(ctx context.Context, workflowID string) (*CompactionState, error)

LoadCompactionState loads the compaction state JSON for a workflow instance.

func (*PostgresStore) LoadDAGSpec 

func (s *PostgresStore) LoadDAGSpec(ctx context.Context, defName string, defVersion int) (json.RawMessage, error)

func (*PostgresStore) LoadEventHistory 

func (s *PostgresStore) LoadEventHistory(ctx context.Context, workflowID string) ([]EventRecord, error)

func (*PostgresStore) LoadEventHistoryPaginated 

func (s *PostgresStore) LoadEventHistoryPaginated(ctx context.Context, workflowID string, offset, limit int) ([]EventRecord, error)

func (*PostgresStore) LoadMemoryEstimates 

func (s *PostgresStore) LoadMemoryEstimates(ctx context.Context) (map[string]float64, error)

LoadMemoryEstimates returns EWMA mean bytes for all def_names.

func (*PostgresStore) LoadMemoryStats 

func (s *PostgresStore) LoadMemoryStats(ctx context.Context) ([]WorkflowMemoryStats, error)

LoadMemoryStats returns full distribution statistics for all def_names.

func (*PostgresStore) LoadWASM 

func (s *PostgresStore) LoadWASM(ctx context.Context, defName string, defVersion int) ([]byte, error)

func (*PostgresStore) LoadWorkflowConfig 

func (s *PostgresStore) LoadWorkflowConfig(ctx context.Context, defName string, defVersion int) (int, error)

func (*PostgresStore) MarkVersionDeprecated 

func (s *PostgresStore) MarkVersionDeprecated(ctx context.Context, name string, version int, deprecated bool) error

func (*PostgresStore) MoveToDeadLetterQueue 

func (s *PostgresStore) MoveToDeadLetterQueue(ctx context.Context, workflowID, workerID string, generation int64, errMsg, errorCode, errorOp string) error

func (*PostgresStore) PickVersionByRouting 

func (s *PostgresStore) PickVersionByRouting(ctx context.Context, workflowName string) (int, error)

func (*PostgresStore) PollAndClaimSignal 

func (s *PostgresStore) PollAndClaimSignal(ctx context.Context, workflowID, signalName string) (string, bool, error)

func (*PostgresStore) PollCancellation 

func (s *PostgresStore) PollCancellation(ctx context.Context, workflowID string) (bool, string, error)

func (*PostgresStore) PollSignal 

func (s *PostgresStore) PollSignal(ctx context.Context, workflowID, signalName string) (string, bool, error)

PollSignal satisfies the SignalStore interface by checking for a delivered signal, without consuming it. This must be a plain read: it used to delegate straight to PollAndClaimSignal, whose name and doc comment both say it "atomically checks for AND CLAIMS" a signal (i.e. DELETEs the row) -- the opposite of what SignalStore's own doc comment promises for PollSignal ("checks for a delivered signal", no mention of consuming it). A second PollSignal call for the same signal would find nothing, having silently deleted it on the first call.

func (*PostgresStore) PurgeWorkflowDef 

func (s *PostgresStore) PurgeWorkflowDef(ctx context.Context, name string, version int) error

func (*PostgresStore) QueueDepth 

func (s *PostgresStore) QueueDepth(ctx context.Context) (int64, error)

QueueDepth returns the count of ready workflows in the store's task queues.

func (*PostgresStore) ReapExpiredConcurrencyKeys 

func (s *PostgresStore) ReapExpiredConcurrencyKeys(ctx context.Context) (int64, error)

ReapExpiredConcurrencyKeys deletes all expired concurrency keys for the current tenant. Returns the number of keys deleted.

func (*PostgresStore) ReapStaleInstances 

func (s *PostgresStore) ReapStaleInstances(ctx context.Context, timeout time.Duration) (int, error)

func (*PostgresStore) RecordWorkflowMemorySample 

func (s *PostgresStore) RecordWorkflowMemorySample(ctx context.Context, defName string, sampleBytes int64) error

CreateUpdateRequest registers an incoming update request for a workflow.

func (*PostgresStore) RejectPromise 

func (s *PostgresStore) RejectPromise(ctx context.Context, workflowID, promiseID, errMsg string) error

func (*PostgresStore) ReleaseConcurrencyKey 

func (s *PostgresStore) ReleaseConcurrencyKey(ctx context.Context, key string) error

ReleaseConcurrencyKey releases a specific concurrency key.

func (*PostgresStore) ReleaseWorkflow 

func (s *PostgresStore) ReleaseWorkflow(ctx context.Context, workflowID, workerID string, generation int64, nextWakeAt time.Time) error

func (*PostgresStore) ReleaseWorkflowConcurrencyKeys 

func (s *PostgresStore) ReleaseWorkflowConcurrencyKeys(ctx context.Context, workflowID string) error

ReleaseWorkflowConcurrencyKeys releases all concurrency keys held by a workflow.

func (*PostgresStore) RemoveRoutingRule 

func (s *PostgresStore) RemoveRoutingRule(ctx context.Context, ruleID string) error

func (*PostgresStore) RemoveWorkflowTag 

func (s *PostgresStore) RemoveWorkflowTag(ctx context.Context, workflowName string, tag string) error

func (*PostgresStore) RequestCancellation 

func (s *PostgresStore) RequestCancellation(ctx context.Context, workflowID, reason string) error

func (*PostgresStore) ResolveCallIntent 

func (s *PostgresStore) ResolveCallIntent(ctx context.Context, workflowID string, rec EventRecord, payload []byte, workerID string, generation int64) error

func (*PostgresStore) ResolveLatestVersion 

func (s *PostgresStore) ResolveLatestVersion(ctx context.Context, defName string) (int, error)

func (*PostgresStore) ResolvePromise 

func (s *PostgresStore) ResolvePromise(ctx context.Context, workflowID, promiseID, result string) error

func (*PostgresStore) ResolveTenantFromAPIKey 

func (s *PostgresStore) ResolveTenantFromAPIKey(ctx context.Context, keyHash []byte) (uuid.UUID, error)

func (*PostgresStore) ResolveVersionByTag 

func (s *PostgresStore) ResolveVersionByTag(ctx context.Context, workflowName string, tag string) (int, error)

func (*PostgresStore) RetryWorkflow 

func (s *PostgresStore) RetryWorkflow(ctx context.Context, workflowID string) error

func (*PostgresStore) SetRoutingRule 

func (s *PostgresStore) SetRoutingRule(ctx context.Context, workflowName string, targetVersion int, weight float64) error

func (*PostgresStore) SetScheduleEnabled 

func (s *PostgresStore) SetScheduleEnabled(ctx context.Context, name string, enabled bool) error

func (*PostgresStore) SetSyncCommitOff 

func (s *PostgresStore) SetSyncCommitOff(v bool)

SetSyncCommitOff sets synchronous_commit = off for finalize transactions.

func (*PostgresStore) SetWorkflowTag 

func (s *PostgresStore) SetWorkflowTag(ctx context.Context, workflowName string, version int, tag string) error

func (*PostgresStore) StartChildWorkflow 

func (s *PostgresStore) StartChildWorkflow(ctx context.Context, parentID, defName, inputJSON string, defVersion int, parentClosePolicy string, priority int) (string, error)

func (*PostgresStore) StartChildWorkflowAtomic 

func (s *PostgresStore) StartChildWorkflowAtomic(ctx context.Context, childID, parentID, defName, inputJSON string, defVersion int, parentClosePolicy string, event EventRecord, priority int) (string, error)

func (*PostgresStore) StartChildWorkflowInSchema 

func (s *PostgresStore) StartChildWorkflowInSchema(ctx context.Context, targetSchema, parentID, defName, inputJSON string, defVersion int, parentClosePolicy string, priority int) (string, error)

StartChildWorkflowInSchema creates a child workflow in the given target schema. Implements CrossSchemaChildStore for cross-instance workflow cooperation.

Tenant attribution: the child belongs to the target schema's tenant, because the target schema is a different microservice and the child runs as part of it. Where that tenant is recoverable from the schema name (the convention admin.create_tenant_role establishes), this sets both the RLS context and the tenant_id column to it, so the row is attributed to the destination.

Where it is not recoverable -- an operator-chosen peer schema name like "svc_billing" -- the engine genuinely does not know which tenant owns the destination, so it writes neither, and the destination table's own DEFAULT applies. Writing the *parent's* tenant would be worse than writing nothing: it would silently file one service's workflow under another service's tenant. If the destination enforces RLS, that insert will be refused, which is the correct outcome for "we cannot say who this belongs to" -- see IMPROVEMENT-PLAN §2.23.

func (*PostgresStore) StartNewRun 

func (s *PostgresStore) StartNewRun(ctx context.Context, runID, defName string, defVersion int, input json.RawMessage, idempotencyKey string, tenantID string, priority int) (string, bool, error)

func (*PostgresStore) StreamEventHistory 

func (s *PostgresStore) StreamEventHistory(ctx context.Context, workflowID string, pageSize int) (<-chan EventRecord, <-chan error)

func (*PostgresStore) TerminateWorkflow 

func (s *PostgresStore) TerminateWorkflow(ctx context.Context, workflowID, reason string) error

TerminateWorkflow force-terminates a workflow, setting status to 'terminated'. Unlike FailWorkflow, this does not require the worker to own the workflow.

func (*PostgresStore) TraceWorkflow 

func (s *PostgresStore) TraceWorkflow(ctx context.Context, workflowID, traceID string) error

func (*PostgresStore) UpdateScheduleNextRun 

func (s *PostgresStore) UpdateScheduleNextRun(ctx context.Context, name string, nextRun time.Time) error

func (*PostgresStore) UpdateStickyWorker 

func (s *PostgresStore) UpdateStickyWorker(ctx context.Context, workflowID, workerID string) error

UpdateStickyWorker sets the sticky worker for a workflow.

func (*PostgresStore) ValidateVersion 

func (s *PostgresStore) ValidateVersion(ctx context.Context, defName string, defVersion int) (bool, error)

func (*PostgresStore) VerifyWorkflowEvents 

func (s *PostgresStore) VerifyWorkflowEvents(ctx context.Context, workflowID string) error

func (*PostgresStore) WithEncryption 

func (s *PostgresStore) WithEncryption(enc *PayloadEncryption, enabled bool) *PostgresStore

WithEncryption returns a copy of the store with encryption at rest enabled.

func (*PostgresStore) WithIdempotencyKeyTTL 

func (s *PostgresStore) WithIdempotencyKeyTTL(ttl time.Duration) *PostgresStore

WithIdempotencyKeyTTL returns a copy of the store with the given idempotency key TTL.

func (*PostgresStore) WithLogger 

func (s *PostgresStore) WithLogger(l *slog.Logger) *PostgresStore

WithLogger returns a copy of the store with the given structured logger.

func (*PostgresStore) WithNotifyChannel 

func (s *PostgresStore) WithNotifyChannel(channel string) *PostgresStore

WithNotifyChannel returns a copy of the store that sends PostgreSQL NOTIFY on dispatchable state changes (new workflows, released timers, signals, promises).

func (*PostgresStore) WithReadRedactionDisabled 

func (s *PostgresStore) WithReadRedactionDisabled(disabled bool) *PostgresStore

WithReadRedactionDisabled returns a copy of the store with redaction on the read path disabled. Used during replay to avoid overhead.

func (*PostgresStore) WithTenant 

func (s *PostgresStore) WithTenant(tenantID string) *PostgresStore

WithTenant returns a copy of the store scoped to the given tenant ID. This is used in the dispatch loop to set the correct tenant context before executing a workflow. The returned store's methods will set the RLS session variable via set_config.

func (*PostgresStore) WriteCallIntent 

func (s *PostgresStore) WriteCallIntent(ctx context.Context, workflowID string, rec EventRecord, workerID string, generation int64) error

type PostgresStoreFactory 

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

PostgresStoreFactory implements StoreFactory for PostgreSQL.

func NewPostgresStoreFactory 

func NewPostgresStoreFactory(db *sql.DB, schemaName string, idempotencyKeyTTL ...time.Duration) *PostgresStoreFactory

NewPostgresStoreFactory creates a PostgresStoreFactory. The db connection must already be open. schemaName is the PostgreSQL schema for cleat tables (defaults to "public").

func (*PostgresStoreFactory) Dialect 

func (f *PostgresStoreFactory) Dialect() Dialect

Dialect returns DialectPostgres.

func (*PostgresStoreFactory) DriverName 

func (f *PostgresStoreFactory) DriverName() string

DriverName returns "postgres".

func (*PostgresStoreFactory) OpenStore 

func (f *PostgresStoreFactory) OpenStore(ctx context.Context, tenantID string, taskQueues ...string) (WorkflowStore, io.Closer, error)

OpenStore creates a PostgresStore scoped to the given tenant and task queues.

func (*PostgresStoreFactory) WithEncryption 

func (f *PostgresStoreFactory) WithEncryption(enc *PayloadEncryption, enabled bool) *PostgresStoreFactory

WithEncryption sets encryption at rest on the factory. When enabled, sensitive payload fields are encrypted before being written to the database.

func (*PostgresStoreFactory) WithLogger 

func (f *PostgresStoreFactory) WithLogger(l *slog.Logger) *PostgresStoreFactory

WithLogger sets the structured logger on the factory. Stores created by OpenStore will inherit it.

func (*PostgresStoreFactory) WithMetrics 

func (f *PostgresStoreFactory) WithMetrics(m *prometheus.Metrics) *PostgresStoreFactory

WithMetrics sets the metrics instance on the factory. Stores created by OpenStore will inherit it.

func (*PostgresStoreFactory) WithNotifyChannel 

func (f *PostgresStoreFactory) WithNotifyChannel(channel string) *PostgresStoreFactory

WithNotifyChannel sets the PostgreSQL NOTIFY channel for dispatch wake-up. When non-empty, OpenStore configures the returned PostgresStore to send pg_notify on dispatchable state changes.

func (*PostgresStoreFactory) WithSyncCommitOff 

func (f *PostgresStoreFactory) WithSyncCommitOff(v bool) *PostgresStoreFactory

WithSyncCommitOff sets synchronous_commit = off for finalize transactions.

type PromiseInfo 

type PromiseInfo struct {
	PromiseID   string     `json:"promise_id"`
	PromiseName string     `json:"promise_name"`
	Status      string     `json:"status"`
	Result      string     `json:"result,omitempty"`
	ErrorMsg    string     `json:"error_msg,omitempty"`
	CreatedAt   time.Time  `json:"created_at"`
	ResolvedAt  *time.Time `json:"resolved_at,omitempty"`
}

PromiseInfo holds the state of a cleat promise.

type PromiseRejectedEvent 

type PromiseRejectedEvent struct {
	PromiseID string
	Err       string
	// contains filtered or unexported fields
}

PromiseRejectedEvent records that a promise was rejected.

func (PromiseRejectedEvent) Step 

func (e PromiseRejectedEvent) Step() int

func (PromiseRejectedEvent) Type 

func (e PromiseRejectedEvent) Type() EventType

type PromiseResolvedEvent 

type PromiseResolvedEvent struct {
	PromiseID string
	Result    string
	// contains filtered or unexported fields
}

PromiseResolvedEvent records that a promise resolved successfully.

func (PromiseResolvedEvent) Step 

func (e PromiseResolvedEvent) Step() int

func (PromiseResolvedEvent) Type 

func (e PromiseResolvedEvent) Type() EventType

type PromiseStore 

type PromiseStore interface {
	CreatePromise(ctx context.Context, workflowID, promiseName, promiseID string) error
	ResolvePromise(ctx context.Context, workflowID, promiseID, result string) error
	RejectPromise(ctx context.Context, workflowID, promiseID, errMsg string) error
	GetPromise(ctx context.Context, workflowID, promiseID string) (status string, result string, errMsg string, err error)
}

PromiseStore provides promise resolution capabilities for running workflows.

type QueryBuilder 

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

QueryBuilder accumulates a SQL query with automatic dialect-correct placeholder numbering. Use for methods with optional WHERE filters (ListWorkflows, ListWorkflowDefs, etc.).

func NewQueryBuilder 

func NewQueryBuilder(d Dialect, baseSQL string) *QueryBuilder

NewQueryBuilder returns a QueryBuilder initialized with a base SQL fragment. The base should end at a point where WHERE conditions can be appended (e.g. "SELECT ... FROM t WHERE 1=1").

func (*QueryBuilder) AddArgs 

func (qb *QueryBuilder) AddArgs(args ...any)

AddArgs appends arguments directly (for use with AddRaw when you manually wrote placeholders). Increments nextPos by len(args).

func (*QueryBuilder) AddCondition 

func (qb *QueryBuilder) AddCondition(condFmt string, arg any)

AddCondition appends " AND <cond>" with one auto-numbered placeholder. condFmt must contain exactly one "%s" verb for the placeholder.

func (*QueryBuilder) AddLikeCondition 

func (qb *QueryBuilder) AddLikeCondition(column string, pattern string, caseInsensitive bool)

AddLikeCondition appends " AND <column> LIKE/ILIKE <placeholder>" with the given pattern. Dialect-aware: Postgres uses ILIKE for case-insensitive matching; MySQL and MSSQL use LIKE (their default collations are case-insensitive).

func (*QueryBuilder) AddRaw 

func (qb *QueryBuilder) AddRaw(sql string)

AddRaw appends a raw SQL fragment (no placeholders auto-managed). The fragment is appended after the current builder content.

func (*QueryBuilder) NextPos 

func (qb *QueryBuilder) NextPos() int

NextPos returns the next placeholder index (for callers that need to write raw SQL with placeholders and then sync the counter).

func (*QueryBuilder) SQL 

func (qb *QueryBuilder) SQL() (string, []any)

SQL returns the built query string and argument slice.

type RLSBypassReason 

type RLSBypassReason struct {
	// Kind is a short machine-readable label: "superuser", "bypassrls",
	// "rls_disabled", "owner_not_forced", or "no_policies".
	Kind string
	// Detail is the human-readable explanation, including the object it
	// applies to where there is one.
	Detail string
}

RLSBypassReason describes one way the current connection escapes Row-Level Security. A connection with no reasons is genuinely subject to the policies.

func CheckRLSEnforced 

func CheckRLSEnforced(ctx context.Context, db *sql.DB) ([]RLSBypassReason, error)

CheckRLSEnforced reports every reason the connection behind db would not have Row-Level Security applied to it.

This matters more here than the phrase "defence in depth" would suggest. GetWorkflowByID and ListWorkflows carry no application-level tenant_id filter at all, so for those paths the RLS policies are not one layer of isolation among several -- they are the only one. A connection that bypasses them sees every tenant's workflows, and nothing in the Go code will stop it.

PostgreSQL exempts a role from RLS in two ways, and neither can be closed from inside the schema:

  • Superuser, and BYPASSRLS. Unconditional. There is no FORCE that applies to a superuser; it is documented behaviour, not an oversight.
  • Table ownership, unless the table has FORCE ROW LEVEL SECURITY. 001_schema.sql sets FORCE on every tenant-scoped table, so this one is closed as long as the migrations are applied -- which is exactly why it is checked rather than assumed.

Every configuration cleat shipped connected as a superuser (docker-compose.cluster.yml uses POSTGRES_USER=cleat; CI and local development use `postgres`), so the policies were present, correct, tested, and bypassed in practice by every connection that ever ran against them. migrations/postgres/005_app_role.sql adds the role to connect as instead.

A nil, empty slice means RLS is enforced. Errors are returned only for failures to interrogate the database.

type ReadOnlyDB 

type ReadOnlyDB struct {
	Inner *sql.DB
}

ReadOnlyDB wraps *sql.DB and implements plugin.PluginDB by enforcing read-only access. Write operations return an error.

func (*ReadOnlyDB) Begin 

func (r *ReadOnlyDB) Begin(ctx context.Context) (plugin.PluginTx, error)

func (*ReadOnlyDB) Exec 

func (r *ReadOnlyDB) Exec(ctx context.Context, query string, args ...any) (int64, error)

func (*ReadOnlyDB) Ping 

func (r *ReadOnlyDB) Ping(ctx context.Context) error

func (*ReadOnlyDB) Query 

func (r *ReadOnlyDB) Query(ctx context.Context, query string, args ...any) (plugin.Rows, error)

func (*ReadOnlyDB) QueryRow 

func (r *ReadOnlyDB) QueryRow(ctx context.Context, query string, args ...any) plugin.RowScanner

type ReplayStepAction 

type ReplayStepAction int

ReplayStepAction is the return value from a ReplayStepCallback. 

const (
	// ReplayNext continues replay to the next event.
	ReplayNext ReplayStepAction = iota
	// ReplayQuit aborts the replay immediately.
	ReplayQuit
)

type ReplayStepCallback 

type ReplayStepCallback func(step int, event *EventRecord, queryState map[string]string) ReplayStepAction

ReplayStepCallback is called after each event is consumed during replay. step is the 0-based index within the replay history. event is a pointer to the EventRecord that was just consumed (may be nil for inline paths that don't construct a full record). queryState is a snapshot (cloned copy) of the current key-value state. Return ReplayQuit to abort the replay immediately (cancels the execution context).

type RetryableError 

type RetryableError interface {
	Retryable() bool
}

RetryableError is optionally implemented by errors to indicate retryability.

type RoutingRule 

type RoutingRule struct {
	ID            string
	WorkflowName  string
	TargetVersion int
	Weight        float64
}

RoutingRule represents a traffic-splitting rule for A/B testing.

type RunDetachedEvent 

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

RunDetachedEvent records a run-detached operation.

func (RunDetachedEvent) Step 

func (e RunDetachedEvent) Step() int

func (RunDetachedEvent) Type 

func (e RunDetachedEvent) Type() EventType

type Runtime 

type Runtime struct {
	MemoryLimitPages uint32 // max WASM linear memory in pages (64KB each)

	Metrics *prometheus.Metrics
	// contains filtered or unexported fields
}

Runtime wraps a wazero runtime with pre-registered host function imports.

func NewRuntime 

func NewRuntime(ctx context.Context, memoryLimitPages uint32, instructionLimit uint64) (*Runtime, error)

NewRuntime creates a Runtime with all cleat_* host functions and the plugin_call host function registered on the "env" module. WASI preview1 is also instantiated for Go wasip1 support. Plugin host functions are registered via the Engine's PluginRegistry — not through NewRuntime.

Floating-point determinism architecture:

WASM floating-point (f32/f64) operations follow IEEE 754-2019, which guarantees bit-identical results for the same operations on the same inputs across all compliant hardware. wazero's interpreter mode (the default for cleat workflows) implements strict IEEE 754 semantics without any "fast math" optimizations that could break determinism.

However, there are important gotchas:

  1. NaN payloads: IEEE 754 allows multiple bit patterns for NaN. WASM f32/f64 operations that produce NaN may return different NaN payloads across CPU architectures or wazero versions. This is only a problem if NaN payloads affect control flow (e.g., comparing NaN values).
  2. Denormal numbers: Some CPUs implement "flush-to-zero" for denormals, while others preserve them. wazero's interpreter preserves denormals.
  3. Compiler optimizations: The host Go compiler may apply FMA (fused multiply-add) or other optimizations that change the exact order of floating-point operations. wazero's WASM interpreter does not apply such optimizations to WASM code.

Best practice: avoid floating-point in workflow control flow conditions. Use math.Float64bits()/math.Float32bits() for exact bitwise comparison, or use integer arithmetic. See docs/determinism.md for more details.

func (*Runtime) CallExport 

func (r *Runtime) CallExport(ctx context.Context, mod api.Module, exportName string, inputJSON []byte) (string, error)

CallExport invokes an exported WASM function with JSON input. It writes inputJSON into the module's linear memory, calls the export, and decodes the int64 result per the exports.go convention. If the export returns the suspend sentinel, it returns ("", nil, ErrSuspended).

func (*Runtime) CallExportWithSuspend 

func (r *Runtime) CallExportWithSuspend(ctx context.Context, mod api.Module, exportName string, inputJSON []byte) (result string, suspended bool, err error)

CallExportWithSuspend invokes an exported WASM function and detects suspension.

func (*Runtime) Close 

func (r *Runtime) Close(ctx context.Context) error

Close releases all resources held by the runtime.

func (*Runtime) CompileModule 

func (r *Runtime) CompileModule(ctx context.Context, wasmBytes []byte) (wazero.CompiledModule, error)

CompileModule pre-compiles a WASM binary for repeated instantiation.

func (*Runtime) InitModule 

func (r *Runtime) InitModule(ctx context.Context, mod api.Module) error

InitModule starts the Go wasip1 runtime by calling _start in a background goroutine. _start initializes WASI and calls main() which blocks to keep the module alive. After yielding until the runtime is ready, the module can accept export calls.

Readiness detection uses exponential backoff after _start has been dispatched. Most Go wasip1 binaries complete runtime initialization in under 1ms (stack setup, GC init, scheduler start). Without a shared-memory readiness flag (which would require SDK changes), we check that the module is responsive by verifying its memory remains accessible — a live WASM module with initialized memory is a reliable indicator that _start succeeded.

For non-Go modules (e.g., Rust, C) that don't have _start, this is a no-op.

func (*Runtime) InstantiateAndInit 

func (r *Runtime) InstantiateAndInit(ctx context.Context, wasmBytes []byte) (api.Module, error)

InstantiateAndInit compiles, instantiates, and initialises a WASM module. Convenience wrapper used by tests and the worker.

func (*Runtime) InstantiateModule 

func (r *Runtime) InstantiateModule(ctx context.Context, compiled wazero.CompiledModule) (api.Module, error)

InstantiateModule creates a new module instance without running _start. Use InitModule to start the Go runtime afterwards.

func (*Runtime) InstantiateModuleNamed 

func (r *Runtime) InstantiateModuleNamed(ctx context.Context, compiled wazero.CompiledModule, name string) (api.Module, error)

InstantiateModuleNamed instantiates a compiled module with the given name. Named modules can be imported by other modules via wazero's module linking. This is used when instantiating plugin modules alongside the workflow module so that the workflow can import from named plugin modules. If name is empty, the module gets wazero's default unnamed module config.

func (*Runtime) Stderr 

func (r *Runtime) Stderr() string

Stderr returns captured stderr output from the most recent module.

func (*Runtime) Stdout 

func (r *Runtime) Stdout() string

Stdout returns captured stdout output from the most recent module.

type SQLDBAdapter 

type SQLDBAdapter struct {
	DB *sql.DB
}

SQLDBAdapter wraps *sql.DB and implements plugin.PluginDB with full read-write access. Used when a plugin declares DatabaseAccessReadWrite.

func (*SQLDBAdapter) Begin 

func (a *SQLDBAdapter) Begin(ctx context.Context) (plugin.PluginTx, error)

func (*SQLDBAdapter) Exec 

func (a *SQLDBAdapter) Exec(ctx context.Context, query string, args ...any) (int64, error)

func (*SQLDBAdapter) Ping 

func (a *SQLDBAdapter) Ping(ctx context.Context) error

func (*SQLDBAdapter) Query 

func (a *SQLDBAdapter) Query(ctx context.Context, query string, args ...any) (plugin.Rows, error)

func (*SQLDBAdapter) QueryRow 

func (a *SQLDBAdapter) QueryRow(ctx context.Context, query string, args ...any) plugin.RowScanner

type Schedule 

type Schedule struct {
	Name           string          `json:"name"`
	DefName        string          `json:"def_name"`
	EntryPoint     string          `json:"entry_point"`
	CronExpression string          `json:"cron_expression"`
	Input          json.RawMessage `json:"input"`
	Enabled        bool            `json:"enabled"`
	NextRunAt      time.Time       `json:"next_run_at"`
	LastRunAt      *time.Time      `json:"last_run_at,omitempty"`

	// Timezone is the IANA zone the cron expression's wall-clock fields are
	// evaluated in (see engine.NextCronTimeIn). Empty means
	// DefaultScheduleTimezone; the stores write 'UTC' rather than ” so the
	// column is never ambiguous about whether a zone was chosen.
	Timezone string `json:"timezone"`

	// MisfirePolicy decides what a firing missed during an outage means:
	// "catch_up" (default) delivers the backlog one instant per tick up to
	// CatchUpLimit, "skip" resumes at the next future instant. Empty means
	// the default.
	MisfirePolicy string `json:"misfire_policy"`

	// CatchUpLimit bounds how many owed firings catch_up will work through
	// before giving up and resuming in the future. Zero means the default;
	// see engine.DefaultCatchUpLimit.
	CatchUpLimit int `json:"catch_up_limit"`

	// OverlapPolicy decides what happens when an instant arrives and the run
	// this schedule started last has not finished: "allow" (default, and what
	// the scheduler has always done) or "skip". Empty means the default.
	OverlapPolicy string `json:"overlap_policy"`

	// LastRunID is the run this schedule started most recently. It is what
	// makes OverlapPolicy "skip" answerable at all -- without it there is no
	// way to tell a run this schedule started from any other run of the same
	// definition.
	LastRunID string `json:"last_run_id,omitempty"`

	// TenantID owns this schedule, and is the tenant the runs it starts belong
	// to.
	//
	// Populated on read. It is NOT read from the caller on CreateSchedule --
	// the stores write their own s.tenantID there, so a caller cannot create a
	// schedule for a tenant it is not scoped to.
	TenantID string `json:"tenant_id"`
}

Schedule is a row from workflow_schedules.

type ServiceCaller 

type ServiceCaller interface {
	Call(ctx context.Context, service, operation, requestJSON string) (responseJSON string, err error)
}

ServiceCaller makes actual external API calls on behalf of cleat workflows.

type Shard 

type Shard struct {
	Config ShardConfig
	Store  WorkflowStore
	Close  func() error
}

Shard wraps a WorkflowStore for a single database shard.

type ShardConfig 

type ShardConfig struct {
	Name    string   `json:"name"`
	ConnStr string   `json:"conn_str"`
	Schema  string   `json:"schema,omitempty"` // PostgreSQL schema name; "public" if empty
	Tenants []string `json:"tenants,omitempty"`
}

ShardConfig is a single database shard configuration loaded from JSON.

type ShardedStore 

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

ShardedStore implements WorkflowStore across multiple PostgreSQL shards.

Each shard hosts a full copy of the schema but owns a subset of the total workflow data. ClaimWorkflow polls every shard (to discover runnable work across the fleet). Most other operations use a consistent hash of the workflow ID to route to the owning shard. Global operations (schedules, reaping, listing) fan out to every shard and merge results.

func NewShardedStore 

func NewShardedStore(configs []ShardConfig, stores []WorkflowStore, closers []func() error) (*ShardedStore, error)

NewShardedStore creates a ShardedStore from pre-constructed WorkflowStore instances (one per shard). Each store is already initialized and migrated. The stores slice must be non-empty and have the same length as configs. closers are called when the ShardedStore is closed; one per store.

func (*ShardedStore) AcquireConcurrencyKey 

func (s *ShardedStore) AcquireConcurrencyKey(ctx context.Context, key, workflowID string, ttl time.Duration) (bool, error)

AcquireConcurrencyKey routes by key text hash for consistent sharding.

func (*ShardedStore) AdminForceComplete 

func (s *ShardedStore) AdminForceComplete(ctx context.Context, workflowID string, generation int64, result string, operator string) error

AdminForceComplete marks a workflow as done, bypassing worker ownership.

func (*ShardedStore) AdminForceFail 

func (s *ShardedStore) AdminForceFail(ctx context.Context, workflowID string, generation int64, errorMsg, errorCode string, operator string) error

AdminForceFail marks a workflow as failed, bypassing worker ownership.

func (*ShardedStore) AdminReReplay 

func (s *ShardedStore) AdminReReplay(ctx context.Context, workflowID string, generation int64, operator string) error

AdminReReplay replays a workflow's event history for debugging.

func (*ShardedStore) AppendEventHistory 

func (s *ShardedStore) AppendEventHistory(ctx context.Context, workflowID string, rec EventRecord) error

AppendEventHistory routes by workflow ID.

func (*ShardedStore) AppendEventHistoryBatch 

func (s *ShardedStore) AppendEventHistoryBatch(ctx context.Context, workflowID string, recs []EventRecord) error

AppendEventHistoryBatch routes by workflow ID.

func (*ShardedStore) BatchHeartbeat 

func (s *ShardedStore) BatchHeartbeat(ctx context.Context, workerID string) (int64, error)

BatchHeartbeat fans out to all shards, aggregating the total count.

func (*ShardedStore) CheckCancellation 

func (s *ShardedStore) CheckCancellation(ctx context.Context, workflowID string) (bool, string, error)

CheckCancellation routes by workflow ID.

func (*ShardedStore) ClaimDueSchedule 

func (s *ShardedStore) ClaimDueSchedule(ctx context.Context, name string, expectedNextRun, newNextRun time.Time, runID string) (bool, error)

ClaimDueSchedule claims on the shard that holds the schedule.

Unlike UpdateScheduleNextRun, this deliberately does NOT fan out to every shard. The CAS is what decides who owns a firing instant, and a fan-out would report "claimed" if any shard's row matched -- turning a single-winner election into a poll. Schedules are replicated across shards, so the first shard whose row still holds expectedNextRun is the winner and the rest are already-advanced copies.

func (*ShardedStore) ClaimStickyWorkflows 

func (s *ShardedStore) ClaimStickyWorkflows(ctx context.Context, workerID string, limit int) ([]*WorkflowInstance, error)

ClaimStickyWorkflows claims up to limit sticky workflow instances across all shards. Sticky workflows use idx_instances_sticky for low-contention claiming. Iterates through shards collecting workflows until limit is reached or shards exhausted.

func (*ShardedStore) ClaimWorkflow 

func (s *ShardedStore) ClaimWorkflow(ctx context.Context, workerID string) (*WorkflowInstance, error)

ClaimWorkflow polls every shard and returns the first runnable workflow found. This is the primary dispatch path so we fan-out across all shards.

func (*ShardedStore) ClaimWorkflows 

func (s *ShardedStore) ClaimWorkflows(ctx context.Context, workerID string, limit int) ([]*WorkflowInstance, error)

ClaimWorkflows claims up to limit runnable workflows across all shards. Iterates through shards collecting workflows until limit is reached or shards exhausted.

func (*ShardedStore) CleanupMemorySamples 

func (s *ShardedStore) CleanupMemorySamples(ctx context.Context, maxSamplesPerDef int) (int64, error)

CleanupMemorySamples fans out to all shards and sums deleted counts.

func (*ShardedStore) ClearStickyWorker 

func (s *ShardedStore) ClearStickyWorker(ctx context.Context, workflowID string) error

ClearStickyWorker routes by workflow ID.

func (*ShardedStore) Close 

func (s *ShardedStore) Close()

Close closes all shard stores.

func (*ShardedStore) CompactHistory 

func (s *ShardedStore) CompactHistory(ctx context.Context, workflowID string, compactionState []byte, compactionStep int, keepStep int) error

CompactHistory routes by workflow ID.

func (*ShardedStore) CompleteUpdateRequest 

func (s *ShardedStore) CompleteUpdateRequest(ctx context.Context, workflowID, updateName, result, errMsg string) error

CompleteUpdateRequest routes by workflow ID.

func (*ShardedStore) CompleteWorkflow 

func (s *ShardedStore) CompleteWorkflow(ctx context.Context, workflowID, workerID string, generation int64, result string, queryState map[string]string) error

CompleteWorkflow routes by workflow ID.

func (*ShardedStore) ContinueAsNew 

func (s *ShardedStore) ContinueAsNew(ctx context.Context, currentRunID, workerID string, generation int64, defName string, defVersion int, newInput json.RawMessage, newEvents []EventRecord, result string, queryState map[string]string, priority int) (string, error)

ContinueAsNew routes by current run ID so that both the new-run insert and the old-run completion land on the same shard.

func (*ShardedStore) CountActiveConcurrencyKeys 

func (s *ShardedStore) CountActiveConcurrencyKeys(ctx context.Context) (int, error)

CountActiveConcurrencyKeys returns the total active concurrency keys across all shards.

func (*ShardedStore) CountActiveInstances 

func (s *ShardedStore) CountActiveInstances(ctx context.Context, name string, version int) (int, error)

CountActiveInstances delegates to the shard determined by the workflow name.

func (*ShardedStore) CountEventHistory 

func (s *ShardedStore) CountEventHistory(ctx context.Context, workflowID string) (int, error)

CountEventHistory routes by workflow ID.

func (*ShardedStore) CountEventHistoryTotal 

func (s *ShardedStore) CountEventHistoryTotal(ctx context.Context) (int, error)

CountEventHistoryTotal returns the total row count across all shards.

func (*ShardedStore) CountStalledWorkflows 

func (s *ShardedStore) CountStalledWorkflows(ctx context.Context, threshold time.Duration) (int, error)

CountStalledWorkflows returns the maximum stalled count across all shards. Using max rather than sum because stalled workflows on different shards are independent — the max captures the worst-case shard, which is the most actionable signal for operator attention.

func (*ShardedStore) CreatePromise 

func (s *ShardedStore) CreatePromise(ctx context.Context, workflowID, promiseName, promiseID string) error

CreatePromise routes by workflow ID.

func (*ShardedStore) CreateSchedule 

func (s *ShardedStore) CreateSchedule(ctx context.Context, sch Schedule) error

CreateSchedule registers a schedule on every shard.

func (*ShardedStore) CreateUpdateRequest 

func (s *ShardedStore) CreateUpdateRequest(ctx context.Context, workflowID, updateName, payload, promiseID string) error

CreateUpdateRequest routes by workflow ID.

func (*ShardedStore) DeleteCompletedWorkflows 

func (s *ShardedStore) DeleteCompletedWorkflows(ctx context.Context, olderThan time.Time) (int64, error)

DeleteCompletedWorkflows fans out to all shards and sums the deleted counts.

func (*ShardedStore) DeleteDeadLetteredWorkflows 

func (s *ShardedStore) DeleteDeadLetteredWorkflows(ctx context.Context, olderThan time.Time) (int64, error)

DeleteDeadLetteredWorkflows fans out to all shards and sums the deleted counts.

func (*ShardedStore) DeleteExpiredEvents 

func (s *ShardedStore) DeleteExpiredEvents(ctx context.Context, olderThan time.Time) (int64, error)

DeleteExpiredEvents fans out to all shards and sums the deleted counts. Errors from individual shards are collected and returned as a single multi-error; remaining shards are still processed.

func (*ShardedStore) DeleteSchedule 

func (s *ShardedStore) DeleteSchedule(ctx context.Context, name string) error

DeleteSchedule removes a schedule from every shard.

func (*ShardedStore) DeliverSignal 

func (s *ShardedStore) DeliverSignal(ctx context.Context, workflowID, signalName, payload string) error

DeliverSignal routes by workflow ID.

func (*ShardedStore) DeployWorkflowDef 

func (s *ShardedStore) DeployWorkflowDef(ctx context.Context, def *WorkflowDef) error

DeployWorkflowDef delegates to the shard determined by the workflow name.

func (*ShardedStore) EstimateEventHistorySize 

func (s *ShardedStore) EstimateEventHistorySize(ctx context.Context) (int64, error)

EstimateEventHistorySize returns the estimated total size across all shards.

func (*ShardedStore) FailWorkflow 

func (s *ShardedStore) FailWorkflow(ctx context.Context, workflowID, workerID string, generation int64, errorMsg, errorCode, errorOp string, queryState map[string]string) error

FailWorkflow routes by workflow ID.

func (*ShardedStore) FinalizeWorkflowSegment 

func (s *ShardedStore) FinalizeWorkflowSegment(ctx context.Context, runID, workerID string, generation int64, newEvents []EventRecord, finalStatus string, result string, errorCode string, errorOp string, queryState map[string]string, nextWakeAt time.Time) error

FinalizeWorkflowSegment routes by workflow ID.

func (*ShardedStore) GetActiveInstanceCountsByVersion 

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

GetActiveInstanceCountsByVersion queries each shard and aggregates results.

func (*ShardedStore) GetAllowedSignalCallers 

func (s *ShardedStore) GetAllowedSignalCallers(ctx context.Context, workflowID string) ([]string, error)

GetAllowedSignalCallers routes by workflow ID.

func (*ShardedStore) GetChildCount 

func (s *ShardedStore) GetChildCount(ctx context.Context, parentWorkflowID string) (int, error)

GetChildCount routes by parent workflow ID.

func (*ShardedStore) GetChildResult 

func (s *ShardedStore) GetChildResult(ctx context.Context, runID string) (string, bool, error)

GetChildResult routes by child run ID.

func (*ShardedStore) GetCompactionCandidates 

func (s *ShardedStore) GetCompactionCandidates(ctx context.Context, threshold int, limit int) ([]string, error)

GetCompactionCandidates runs on every shard and merges results.

func (*ShardedStore) GetConcurrencyKeyCount 

func (s *ShardedStore) GetConcurrencyKeyCount(ctx context.Context, workflowID string) (int, error)

GetConcurrencyKeyCount routes by workflow ID.

func (*ShardedStore) GetDueSchedules 

func (s *ShardedStore) GetDueSchedules(ctx context.Context) ([]Schedule, error)

GetDueSchedules collects due schedules from every shard (deduped by name).

func (*ShardedStore) GetEventCount 

func (s *ShardedStore) GetEventCount(ctx context.Context, workflowID string) (int, error)

GetEventCount returns the event_count for a workflow instance.

func (*ShardedStore) GetPendingUpdateRequests 

func (s *ShardedStore) GetPendingUpdateRequests(ctx context.Context, workflowID string) ([]UpdateRequestInfo, error)

GetPendingUpdateRequests routes by workflow ID.

func (*ShardedStore) GetPromise 

func (s *ShardedStore) GetPromise(ctx context.Context, workflowID, promiseID string) (string, string, string, error)

GetPromise routes by workflow ID.

func (*ShardedStore) GetQueryState 

func (s *ShardedStore) GetQueryState(ctx context.Context, workflowID, key string) (string, error)

GetQueryState routes by workflow ID.

func (*ShardedStore) GetRoutingRules 

func (s *ShardedStore) GetRoutingRules(ctx context.Context, workflowName string) ([]RoutingRule, error)

GetRoutingRules returns all routing rules for a workflow. Delegates to the shard determined by the workflow name.

func (*ShardedStore) GetWASMLength 

func (s *ShardedStore) GetWASMLength(ctx context.Context, defName string, defVersion int) (int64, error)

GetWASMLength returns the byte length of the stored WASM binary.

func (*ShardedStore) GetWorkflowByID 

func (s *ShardedStore) GetWorkflowByID(ctx context.Context, id string) (*WorkflowInstance, error)

GetWorkflowByID tries each shard (workflow could be on any shard).

func (*ShardedStore) GetWorkflowDef 

func (s *ShardedStore) GetWorkflowDef(ctx context.Context, name string, version int) (*WorkflowDef, error)

GetWorkflowDef delegates to the shard determined by the workflow name.

func (*ShardedStore) GetWorkflowTag 

func (s *ShardedStore) GetWorkflowTag(ctx context.Context, workflowName string, tag string) (int, error)

GetWorkflowTag returns the version for a given tag. Delegates to the shard determined by the workflow name.

func (*ShardedStore) GetWorkflowTags 

func (s *ShardedStore) GetWorkflowTags(ctx context.Context, workflowName string) (map[string]int, error)

GetWorkflowTags returns all tag -> version mappings for a workflow. Delegates to the shard determined by the workflow name.

func (*ShardedStore) Heartbeat 

func (s *ShardedStore) Heartbeat(ctx context.Context, workflowID, workerID string, generation int64) (bool, error)

Heartbeat routes by workflow ID.

func (*ShardedStore) ListPromises 

func (s *ShardedStore) ListPromises(ctx context.Context, workflowID string) ([]PromiseInfo, error)

ListPromises routes by workflow ID.

func (*ShardedStore) ListSchedules 

func (s *ShardedStore) ListSchedules(ctx context.Context) ([]Schedule, error)

ListSchedules merges schedules from all shards (deduped by name).

func (*ShardedStore) ListVersions 

func (s *ShardedStore) ListVersions(ctx context.Context, defName string) ([]int, error)

ListVersions tries each shard (definitions are replicated across shards).

func (*ShardedStore) ListWorkflowDefs 

func (s *ShardedStore) ListWorkflowDefs(ctx context.Context, name string) ([]WorkflowDef, error)

ListWorkflowDefs queries each shard and aggregates results.

func (*ShardedStore) ListWorkflows 

func (s *ShardedStore) ListWorkflows(ctx context.Context, filter WorkflowFilter) ([]WorkflowInstance, error)

ListWorkflows merges results from all shards.

func (*ShardedStore) LoadCompactionState 

func (s *ShardedStore) LoadCompactionState(ctx context.Context, workflowID string) (*CompactionState, error)

LoadCompactionState routes by workflow ID.

func (*ShardedStore) LoadDAGSpec 

func (s *ShardedStore) LoadDAGSpec(ctx context.Context, defName string, defVersion int) (json.RawMessage, error)

LoadDAGSpec tries each shard (defs are replicated across shards).

func (*ShardedStore) LoadEventHistory 

func (s *ShardedStore) LoadEventHistory(ctx context.Context, workflowID string) ([]EventRecord, error)

LoadEventHistory routes by workflow ID.

func (*ShardedStore) LoadEventHistoryBatch 

func (s *ShardedStore) LoadEventHistoryBatch(ctx context.Context, workflowIDs []string) (map[string][]EventRecord, error)

LoadEventHistoryBatch returns event histories for multiple workflow IDs by dispatching per-ID to the appropriate shard.

func (*ShardedStore) LoadEventHistoryPaginated 

func (s *ShardedStore) LoadEventHistoryPaginated(ctx context.Context, workflowID string, offset, limit int) ([]EventRecord, error)

LoadEventHistoryPaginated routes by workflow ID.

func (*ShardedStore) LoadMemoryEstimates 

func (s *ShardedStore) LoadMemoryEstimates(ctx context.Context) (map[string]float64, error)

LoadMemoryEstimates fans out to all shards and merges results.

func (*ShardedStore) LoadMemoryStats 

func (s *ShardedStore) LoadMemoryStats(ctx context.Context) ([]WorkflowMemoryStats, error)

LoadMemoryStats fans out to all shards and appends results.

func (*ShardedStore) LoadWASM 

func (s *ShardedStore) LoadWASM(ctx context.Context, defName string, defVersion int) ([]byte, error)

LoadWASM tries each shard (WASM definitions are replicated across all shards).

func (*ShardedStore) LoadWorkflowConfig 

func (s *ShardedStore) LoadWorkflowConfig(ctx context.Context, defName string, defVersion int) (int, error)

LoadWorkflowConfig tries each shard (defs are replicated across shards).

func (*ShardedStore) MarkVersionDeprecated 

func (s *ShardedStore) MarkVersionDeprecated(ctx context.Context, name string, version int, deprecated bool) error

MarkVersionDeprecated delegates to the shard determined by the workflow name.

func (*ShardedStore) MoveToDeadLetterQueue 

func (s *ShardedStore) MoveToDeadLetterQueue(ctx context.Context, workflowID, workerID string, generation int64, errMsg, errorCode, errorOp string) error

MoveToDeadLetterQueue routes by workflow ID.

func (*ShardedStore) PickVersionByRouting 

func (s *ShardedStore) PickVersionByRouting(ctx context.Context, workflowName string) (int, error)

PickVersionByRouting checks A/B routing rules for the given workflow name. Delegates to the shard determined by the workflow name.

func (*ShardedStore) PollAndClaimSignal 

func (s *ShardedStore) PollAndClaimSignal(ctx context.Context, workflowID, signalName string) (string, bool, error)

PollAndClaimSignal routes by workflow ID.

func (*ShardedStore) PollCancellation 

func (s *ShardedStore) PollCancellation(ctx context.Context, workflowID string) (bool, string, error)

PollCancellation satisfies the SignalStore interface. It routes by workflow ID.

func (*ShardedStore) PollSignal 

func (s *ShardedStore) PollSignal(ctx context.Context, workflowID, signalName string) (string, bool, error)

PollSignal satisfies the SignalStore interface. It routes by workflow ID.

func (*ShardedStore) PurgeWorkflowDef 

func (s *ShardedStore) PurgeWorkflowDef(ctx context.Context, name string, version int) error

PurgeWorkflowDef delegates to the shard determined by the workflow name.

func (*ShardedStore) QueueDepth 

func (s *ShardedStore) QueueDepth(ctx context.Context) (int64, error)

QueueDepth fans out to all shards and sums the counts.

func (*ShardedStore) ReapExpiredConcurrencyKeys 

func (s *ShardedStore) ReapExpiredConcurrencyKeys(ctx context.Context) (int64, error)

ReapExpiredConcurrencyKeys runs on every shard and returns the total count.

func (*ShardedStore) ReapStaleInstances 

func (s *ShardedStore) ReapStaleInstances(ctx context.Context, timeout time.Duration) (int, error)

ReapStaleInstances runs on every shard and returns the total reclaimed count.

func (*ShardedStore) RecordWorkflowMemorySample 

func (s *ShardedStore) RecordWorkflowMemorySample(ctx context.Context, defName string, sampleBytes int64) error

RecordWorkflowMemorySample routes by defName hash for consistent shard affinity.

func (*ShardedStore) RejectPromise 

func (s *ShardedStore) RejectPromise(ctx context.Context, workflowID, promiseID, errMsg string) error

RejectPromise routes by workflow ID.

func (*ShardedStore) ReleaseConcurrencyKey 

func (s *ShardedStore) ReleaseConcurrencyKey(ctx context.Context, key string) error

ReleaseConcurrencyKey routes by key text hash.

func (*ShardedStore) ReleaseWorkflow 

func (s *ShardedStore) ReleaseWorkflow(ctx context.Context, workflowID, workerID string, generation int64, nextWakeAt time.Time) error

ReleaseWorkflow routes by workflow ID.

func (*ShardedStore) ReleaseWorkflowConcurrencyKeys 

func (s *ShardedStore) ReleaseWorkflowConcurrencyKeys(ctx context.Context, workflowID string) error

ReleaseWorkflowConcurrencyKeys routes by workflow ID.

func (*ShardedStore) RemoveRoutingRule 

func (s *ShardedStore) RemoveRoutingRule(ctx context.Context, ruleID string) error

RemoveRoutingRule deletes a routing rule by ID. Delegates to the shard determined by the rule ID.

func (*ShardedStore) RemoveWorkflowTag 

func (s *ShardedStore) RemoveWorkflowTag(ctx context.Context, workflowName string, tag string) error

RemoveWorkflowTag deletes a tag assignment. Delegates to the shard determined by the workflow name.

func (*ShardedStore) RequestCancellation 

func (s *ShardedStore) RequestCancellation(ctx context.Context, workflowID, reason string) error

RequestCancellation routes by workflow ID.

func (*ShardedStore) ResolveLatestVersion 

func (s *ShardedStore) ResolveLatestVersion(ctx context.Context, defName string) (int, error)

ResolveLatestVersion delegates to the shard determined by the workflow name.

func (*ShardedStore) ResolvePromise 

func (s *ShardedStore) ResolvePromise(ctx context.Context, workflowID, promiseID, result string) error

ResolvePromise routes by workflow ID.

func (*ShardedStore) ResolveTenantFromAPIKey 

func (s *ShardedStore) ResolveTenantFromAPIKey(ctx context.Context, keyHash []byte) (uuid.UUID, error)

ResolveTenantFromAPIKey looks up a tenant UUID by API key hash across all shards.

func (*ShardedStore) ResolveVersionByTag 

func (s *ShardedStore) ResolveVersionByTag(ctx context.Context, workflowName string, tag string) (int, error)

ResolveVersionByTag resolves a tag to a workflow definition version. Delegates to the shard determined by the workflow name.

func (*ShardedStore) RetryWorkflow 

func (s *ShardedStore) RetryWorkflow(ctx context.Context, workflowID string) error

RetryWorkflow routes by workflow ID.

func (*ShardedStore) SetRoutingRule 

func (s *ShardedStore) SetRoutingRule(ctx context.Context, workflowName string, targetVersion int, weight float64) error

SetRoutingRule creates a routing rule for a workflow version. Delegates to the shard determined by the workflow name.

func (*ShardedStore) SetScheduleEnabled 

func (s *ShardedStore) SetScheduleEnabled(ctx context.Context, name string, enabled bool) error

SetScheduleEnabled updates a schedule on every shard.

func (*ShardedStore) SetWorkflowTag 

func (s *ShardedStore) SetWorkflowTag(ctx context.Context, workflowName string, version int, tag string) error

SetWorkflowTag assigns a tag to a specific version. Delegates to the shard determined by the workflow name.

func (*ShardedStore) Shards 

func (s *ShardedStore) Shards() []*Shard

Shards returns the underlying shard list (for inspection / metrics).

func (*ShardedStore) StartChildWorkflow 

func (s *ShardedStore) StartChildWorkflow(ctx context.Context, parentID, defName, inputJSON string, defVersion int, parentClosePolicy string, priority int) (string, error)

StartChildWorkflow places the child on the same shard as the parent. defVersion is passed through to the underlying store for version resolution.

func (*ShardedStore) StartChildWorkflowAtomic 

func (s *ShardedStore) StartChildWorkflowAtomic(ctx context.Context, childID, parentID, defName, inputJSON string, defVersion int, parentClosePolicy string, event EventRecord, priority int) (string, error)

StartChildWorkflowAtomic routes by the root ancestor UUID (stripping .c suffix) so the child lands on the same shard as its family. Generates a shard-aligned ".c{step}.{rand}" child ID if childID is empty. The random suffix guarantees uniqueness across generations (a parent and its child can both be at step 5).

func (*ShardedStore) StartNewRun 

func (s *ShardedStore) StartNewRun(ctx context.Context, runID, defName string, defVersion int, input json.RawMessage, idempotencyKey string, tenantID string, priority int) (string, bool, error)

StartNewRun generates a UUID and routes by it, so the workflow lands on a shard determined by its own ID (not its definition name). This ensures all subsequent operations (LoadEventHistory, child workflows, signals, etc.) route to the same shard.

func (*ShardedStore) StreamEventHistory 

func (s *ShardedStore) StreamEventHistory(ctx context.Context, workflowID string, pageSize int) (<-chan EventRecord, <-chan error)

StreamEventHistory routes by workflow ID.

func (*ShardedStore) TerminateWorkflow 

func (s *ShardedStore) TerminateWorkflow(ctx context.Context, workflowID, reason string) error

TerminateWorkflow routes by workflow ID.

func (*ShardedStore) TraceWorkflow 

func (s *ShardedStore) TraceWorkflow(ctx context.Context, workflowID, traceID string) error

TraceWorkflow routes by workflow ID.

func (*ShardedStore) UpdateScheduleNextRun 

func (s *ShardedStore) UpdateScheduleNextRun(ctx context.Context, name string, nextRun time.Time) error

UpdateScheduleNextRun updates a schedule on every shard.

func (*ShardedStore) UpdateStickyWorker 

func (s *ShardedStore) UpdateStickyWorker(ctx context.Context, workflowID, workerID string) error

UpdateStickyWorker routes by workflow ID.

func (*ShardedStore) ValidateVersion 

func (s *ShardedStore) ValidateVersion(ctx context.Context, defName string, defVersion int) (bool, error)

ValidateVersion delegates to the shard determined by the workflow name.

func (*ShardedStore) VerifyWorkflowEvents 

func (s *ShardedStore) VerifyWorkflowEvents(ctx context.Context, workflowID string) error

VerifyWorkflowEvents routes by workflow ID.

type SignalReceivedEvent 

type SignalReceivedEvent struct {
	SignalName    string
	SignalPayload string
	// contains filtered or unexported fields
}

SignalReceivedEvent records that a signal was delivered to the workflow.

func (SignalReceivedEvent) Step 

func (e SignalReceivedEvent) Step() int

func (SignalReceivedEvent) Type 

func (e SignalReceivedEvent) Type() EventType

type SignalStore 

type SignalStore interface {
	// DeliverSignal stores a signal for a workflow.
	DeliverSignal(ctx context.Context, workflowID, signalName, payload string) error
	// PollSignal checks for a delivered signal.
	PollSignal(ctx context.Context, workflowID, signalName string) (payload string, found bool, err error)
	// PollCancellation checks whether the workflow has been cancelled.
	PollCancellation(ctx context.Context, workflowID string) (cancelled bool, reason string, err error)
}

SignalStore provides signal delivery capabilities for running workflows.

type SliceEventStream 

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

SliceEventStream wraps a []EventRecord as an EventStream. This is the default implementation used when the history has already been loaded into memory (e.g., from the worker's loadEventHistory call).

func NewSliceEventStream 

func NewSliceEventStream(events []EventRecord) *SliceEventStream

NewSliceEventStream creates a SliceEventStream from an existing slice. The caller retains ownership of the underlying slice.

func (*SliceEventStream) Append 

func (s *SliceEventStream) Append(rec EventRecord)

func (*SliceEventStream) At 

func (s *SliceEventStream) At(i int) *EventRecord

func (*SliceEventStream) Close 

func (s *SliceEventStream) Close() error

func (*SliceEventStream) Len 

func (s *SliceEventStream) Len() int

func (*SliceEventStream) Slice 

func (s *SliceEventStream) Slice(start, end int) []EventRecord

func (*SliceEventStream) Total 

func (s *SliceEventStream) Total() (int, error)

type StaleVersionAlert 

type StaleVersionAlert struct {
	Name             string `json:"name"`
	Version          int    `json:"version"`
	Deprecated       bool   `json:"deprecated"`
	ActiveInstances  int    `json:"active_instances"`
	DaysSinceCreated int    `json:"days_since_created"`
	Message          string `json:"message"`
}

StaleVersionAlert describes a workflow version that may need attention.

func CheckStaleVersions 

func CheckStaleVersions(ctx context.Context, store WorkflowStore, staleThreshold, purgeThreshold time.Duration) ([]StaleVersionAlert, error)

CheckStaleVersions scans for versions that may need attention:

  • Non-deprecated versions older than staleThreshold that still have active instances (potential migration candidates).
  • Deprecated versions older than purgeThreshold with zero active instances (GC candidates).

type StateMutationEvent 

type StateMutationEvent struct {
	Key   string
	Value string
	Delta int64
	Op    string
	// contains filtered or unexported fields
}

StateMutationEvent records a state mutation operation.

func (StateMutationEvent) Step 

func (e StateMutationEvent) Step() int

func (StateMutationEvent) Type 

func (e StateMutationEvent) Type() EventType

type StoreFactory 

type StoreFactory interface {
	// OpenStore creates or connects to a WorkflowStore scoped to the given tenant.
	// The tenantID identifies which tenant the store should operate on.
	// The taskQueues slice specifies which queues this store should poll.
	OpenStore(ctx context.Context, tenantID string, taskQueues ...string) (WorkflowStore, io.Closer, error)

	// DriverName returns the database/sql driver name for health checks.
	DriverName() string

	// Dialect returns the SQL dialect of this factory's backend.
	Dialect() Dialect
}

StoreFactory creates WorkflowStore instances. Each database backend implements one. The factory encapsulates connection management, schema setup, and backend-specific configuration — callers never need to know whether the store is backed by PostgreSQL, MySQL, or SQLite.

type SuspendError 

type SuspendError struct {
	Reason     string
	Until      time.Time // if non-zero, the workflow should wake at this time
	NewInput   string    // for continue_as_new: the new input payload
	NewVersion int       // for continue_as_new with version: the new workflow version
}

SuspendError signals that the workflow should be suspended.

func (*SuspendError) Error 

func (e *SuspendError) Error() string

type SuspendResult 

type SuspendResult struct {
	History      []EventRecord
	SuspendUntil time.Time
	Reason       string
	NewInput     string            // for continue_as_new: the new input payload
	NewVersion   int               // for continue_as_new with version: the new workflow version
	Deferrals    map[string]string // registered defers (deferID -> description)

	// ContinueAsNewHandled is true when the engine has already persisted the
	// ContinueAsNew transition (events + new run + old run completion) as part
	// of the suspend path. The worker should NOT call store.ContinueAsNew again.
	ContinueAsNewHandled bool

	// NewRunID is the new workflow run ID when ContinueAsNew has been handled
	// by the engine. Empty if not handled or if the suspend is for another reason.
	NewRunID string
}

SuspendResult holds the outcome of a suspended workflow execution.

type TenantFlusherRegistry 

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

TenantFlusherRegistry creates and caches per-tenant AdaptiveFlusher instances. Each tenant gets its own rate tracking and batch accumulator, preventing cross-tenant interference and ensuring batch payloads carry a single tenant_id.

func NewTenantFlusherRegistry 

func NewTenantFlusherRegistry(db *sql.DB, config FlusherConfig) *TenantFlusherRegistry

NewTenantFlusherRegistry creates a registry that lazily provisions per-tenant AdaptiveFlusher instances with the given configuration.

func (*TenantFlusherRegistry) For 

func (r *TenantFlusherRegistry) For(tenantID string) *AdaptiveFlusher

For returns the AdaptiveFlusher for the given tenant, creating one if it does not already exist. Safe for concurrent use.

TODO: Add TTL-based eviction or LRU bound for transient tenants.

func (*TenantFlusherRegistry) Remove 

func (r *TenantFlusherRegistry) Remove(tenantID string)

Remove cleans up a tenant flusher that is no longer needed.

func (*TenantFlusherRegistry) SetEncryption 

func (r *TenantFlusherRegistry) SetEncryption(encrypt bool, enc *PayloadEncryption)

SetEncryption propagates encryption settings to the registry and all existing per-tenant flusher instances.

func (*TenantFlusherRegistry) Shutdown 

func (r *TenantFlusherRegistry) Shutdown()

Shutdown drains all per-tenant accumulators. Call before the worker exits.

type TruncationSummary 

type TruncationSummary struct {
	TruncatedCount int `json:"truncated_count"`
}

TruncationSummary records how many compacted events were truncated to keep the compaction state JSONB size bounded.

type UpdateHandlerEvent 

type UpdateHandlerEvent struct {
	HandlerName string
	// contains filtered or unexported fields
}

UpdateHandlerEvent records an update handler registration.

func (UpdateHandlerEvent) Step 

func (e UpdateHandlerEvent) Step() int

func (UpdateHandlerEvent) Type 

func (e UpdateHandlerEvent) Type() EventType

type UpdateRequestInfo 

type UpdateRequestInfo struct {
	WorkflowID string    `json:"workflow_id"`
	UpdateName string    `json:"update_name"`
	Payload    string    `json:"payload"`
	PromiseID  string    `json:"promise_id,omitempty"`
	Status     string    `json:"status"`
	Result     string    `json:"result,omitempty"`
	ErrorMsg   string    `json:"error_msg,omitempty"`
	CreatedAt  time.Time `json:"created_at"`
}

UpdateRequestInfo holds the state of an incoming update request.

type VaultCredentialProvider 

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

VaultCredentialProvider resolves the connection string by calling the HashiCorp Vault CLI: 

vault kv get -field=connection_string <path>

func NewVaultCredentialProvider 

func NewVaultCredentialProvider(credentialPath string) *VaultCredentialProvider

NewVaultCredentialProvider creates a VaultCredentialProvider. The credentialPath is the Vault path to read (e.g., "secret/cleat/db").

func (*VaultCredentialProvider) GetConnectionString 

func (p *VaultCredentialProvider) GetConnectionString(ctx context.Context) (string, error)

GetConnectionString runs `vault kv get -field=connection_string <path>` and returns the connection string.

type VersionMetrics 

type VersionMetrics struct {
	Name            string    `json:"name"`
	Version         int       `json:"version"`
	Deprecated      bool      `json:"deprecated"`
	CreatedAt       time.Time `json:"created_at"`
	Age             string    `json:"age"` // human-readable age
	ActiveInstances int       `json:"active_instances"`
	ABIVersion      int       `json:"abi_version"`
	MinVersion      int       `json:"min_version"`
}

VersionMetrics holds operational metrics for a single workflow version.

type VersionMetricsSummary 

type VersionMetricsSummary struct {
	TotalVersions        int              `json:"total_versions"`
	ActiveVersions       int              `json:"active_versions"`
	Deprecated           int              `json:"deprecated"`
	TotalActiveInstances int              `json:"total_active_instances"`
	Workflows            []VersionMetrics `json:"workflows"`
}

VersionMetricsSummary is a complete summary of version metrics across all workflow definitions.

func CollectVersionMetrics 

func CollectVersionMetrics(ctx context.Context, store WorkflowStore) (*VersionMetricsSummary, error)

CollectVersionMetrics gathers metrics for all deployed workflow versions.

type VersionStoreResolver 

type VersionStoreResolver func(w http.ResponseWriter, r *http.Request) (store WorkflowStore, ok bool)

VersionStoreResolver resolves the WorkflowStore scoped to an HTTP request's authenticated tenant. Implementations write the error response themselves and return ok=false when no store can be resolved (no authenticated tenant, or a tenant whose store cannot be opened), mirroring cmd/cleat-worker's apiServer.scopedStore -- which is exactly what production wiring passes here (see the call in cmd/cleat-worker/main.go).

This type exists because RegisterVersionHandler used to take a single process-wide WorkflowStore, opened once at boot against the default tenant. Every version endpoint -- including POST /api/versions/<name>/<v>/purge, which permanently deletes workflow definitions -- then served every caller from that one tenant's data regardless of who authenticated, and any caller could purge the default tenant's definitions. Routing each request through a resolver is the same fix cmd/cleat-worker/server.go applies to every other handler; this package cannot import cmd/cleat-worker's apiServer or the auth package directly (auth already imports engine), so the resolver is the seam between them.

func StaticVersionStore 

func StaticVersionStore(store WorkflowStore) VersionStoreResolver

StaticVersionStore returns a VersionStoreResolver that always resolves to store, regardless of the request. It exists for callers that have no per-request tenant to scope to -- tests, and an embedded/local-dev setup that only ever has one tenant. Production HTTP wiring must not use this: it recreates exactly the defect VersionStoreResolver exists to close.

type WASMCache 

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

WASMCache is a size-bounded LRU cache for compiled WASM binaries. It is safe for concurrent use.

func NewWASMCache 

func NewWASMCache(maxEntries int, maxBytes int64) *WASMCache

NewWASMCache creates a new WASM cache with the given limits.

func (*WASMCache) Get 

func (c *WASMCache) Get(key string) ([]byte, bool)

Get returns cached bytes for the given key, updating LRU order. Returns nil and false if the key is not in the cache.

func (*WASMCache) Len 

func (c *WASMCache) Len() int

Len returns the number of entries in the cache.

func (*WASMCache) Put 

func (c *WASMCache) Put(key string, data []byte)

Put inserts data into the cache, evicting LRU entries if over limits.

func (*WASMCache) TotalBytes 

func (c *WASMCache) TotalBytes() int64

TotalBytes returns the total cached byte count.

type WasmBackend 

type WasmBackend interface {
	// Execute runs a WASM module. The session provides the HostHandler for
	// all host function implementations. The backend handles compilation,
	// instantiation, memory management, and export calling.
	Execute(ctx context.Context, wasmBytes []byte, entryPoint string, input json.RawMessage, session HostHandler) (*ExecResult, error)

	// Close releases all backend resources.
	Close(ctx context.Context) error

	// Name returns a human-readable backend name for diagnostics.
	Name() string

	// PerExecution returns a new backend instance that shares the underlying
	// compilation engine but has its own per-execution mutable state (handler,
	// work data). This is required by the engine to prevent data races when
	// Execute is called concurrently from multiple goroutines.
	PerExecution() WasmBackend
}

WasmBackend executes compiled WASM modules. Each backend owns compilation, instantiation, and host function wiring.

type WasmDiskCache 

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

WasmDiskCache provides a disk-backed cache for raw WASM module bytes. Cache entries are content-addressed by sha256(wasm_bytes), so version changes automatically invalidate stale entries.

This cache stores raw WASM bytes on disk, avoiding database round-trips on worker restart. An index file maps (defName, defVersion) to the sha256 content hash so lookups can be performed without knowing the content.

Compiled module serialization is not supported by the current wazero version, so compilation still occurs on each worker start.

func NewWasmDiskCache 

func NewWasmDiskCache(cacheDir string, maxLen int) *WasmDiskCache

NewWasmDiskCache creates a WasmDiskCache rooted at cacheDir. Returns nil if cacheDir is empty (caching disabled). maxLen is the maximum number of cached files to keep on disk (default 100).

func (*WasmDiskCache) LookupByKey 

func (c *WasmDiskCache) LookupByKey(key string) []byte

LookupByKey retrieves raw WASM bytes by their sha256 cache key. Returns nil if the entry does not exist.

func (*WasmDiskCache) LookupBytes 

func (c *WasmDiskCache) LookupBytes(wasmBytes []byte) []byte

LookupBytes attempts to load raw WASM bytes from the disk cache by content. Returns the bytes if found, nil otherwise.

func (*WasmDiskCache) LookupDef 

func (c *WasmDiskCache) LookupDef(name string, version int) []byte

LookupDef retrieves raw WASM bytes for a workflow definition by name and version. Returns nil on cache miss.

func (*WasmDiskCache) StoreDef 

func (c *WasmDiskCache) StoreDef(name string, version int, wasmBytes []byte)

StoreDef writes raw WASM bytes to the disk cache indexed by (name, version) and content-addressed by sha256. This is a no-op if the entry already exists.

type WasmtimeOption 

type WasmtimeOption func(*wasmtimeLimits)

WasmtimeOption configures resource limits for a wasmtimeBackend, applied at construction time (NewWasmtimeBackend) and enforced on every store it creates thereafter (see wasmtimeBackend.configureStore).

func WithWasmtimeExecutionTimeout 

func WithWasmtimeExecutionTimeout(d time.Duration) WasmtimeOption

WithWasmtimeExecutionTimeout bounds a single wasmtime invocation via epoch interruption. d <= 0 keeps DefaultWasmtimeExecutionTimeout. A per-call context deadline (e.g. from engine.WithWASMInstanceTimeout or engine.WithDefaultWorkflowTimeout), when tighter than this, still wins — see wasmtimeBackend.configureStore.

func WithWasmtimeInstructionLimit 

func WithWasmtimeInstructionLimit(n uint64) WasmtimeOption

WithWasmtimeInstructionLimit bounds fuel (roughly one unit per WASM instruction executed) consumed per invocation. 0 disables fuel metering — the wasmtime analogue of the wazero-only --wasm-instruction-limit flag defaulting to "no limit".

Note for anyone tempted to raise this as the primary defense: fuel exhaustion mid-replay could in principle make replay diverge from the original execution if the configured limit changes between the two runs (e.g. a flag change), or on hardware where the same WASM instructions consume different amounts of fuel due to a wasmtime version skew. Epoch interruption (WithWasmtimeExecutionTimeout) does not have this property any more than wall-clock time already does for CPU-bound workflows, which is why it is the primary, always-on bound and fuel is an optional, opt-in secondary one.

func WithWasmtimeMemoryLimits 

func WithWasmtimeMemoryLimits(memoryBytes, tableElements, instances int64) WasmtimeOption

WithWasmtimeMemoryLimits bounds linear memory, table elements, and instance counts per store (wasmtime's StoreLimits / ResourceLimiter). Values <= 0 keep the backend's built-in default for that dimension (see the Default* constants above).

type WorkflowDef 

type WorkflowDef struct {
	Name       string            `json:"name"`
	Version    int               `json:"version"`
	WASMBytes  []byte            `json:"wasm_bytes,omitempty"`
	ABIVersion int               `json:"abi_version"`
	MinVersion int               `json:"min_version"`
	PluginDeps map[string]string `json:"plugin_deps,omitempty"`
	CreatedAt  time.Time         `json:"created_at"`
	Deprecated bool              `json:"deprecated"`
}

type WorkflowFilter 

type WorkflowFilter struct {
	Status        string
	InputContains string
	ErrorContains string
	Search        string
	Offset        int
	Limit         int
}

WorkflowFilter contains optional filter parameters for listing workflow instances. Empty/zero values mean "no filter" for that parameter.

type WorkflowInstance 

type WorkflowInstance struct {
	ID         string          `json:"id"`
	DefName    string          `json:"def_name"`
	DefVersion int             `json:"def_version"`
	MinVersion int             `json:"min_version"`
	Status     string          `json:"status"`
	Input      json.RawMessage `json:"input"`
	Result     string          `json:"result,omitempty"`
	Error      string          `json:"error,omitempty"`
	ErrorCode  string          `json:"error_code,omitempty"`
	ErrorOp    string          `json:"error_op,omitempty"`
	AssignedTo string          `json:"assigned_to"`
	NextWakeAt time.Time       `json:"next_wake_at"`
	TenantID   string          `json:"tenant_id,omitempty"`
	CreatedAt  time.Time       `json:"created_at,omitempty"`
	Generation int64           `json:"generation"`
	Priority   int             `json:"priority"`
	TraceID    string          `json:"trace_id,omitempty"`
}

WorkflowInstance is a row from workflow_instances.

type WorkflowLoader 

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

WorkflowLoader loads and caches compiled WASM modules for workflow definitions. It queries the workflow_defs table by (name, version), compiles the WASM bytes via the wazero runtime, and caches the compiled modules in an LRU cache keyed by (name, version).

The loader also supports a disk-backed cache (WasmDiskCache) that survives worker restarts. Cache entries are content-addressed by sha256(wasm_bytes) so version changes automatically invalidate stale entries.

Thread-safety: the LRU cache is protected by a mutex. Compiled modules themselves are safe for concurrent instantiation per wazero guarantees.

func NewWorkflowLoader 

func NewWorkflowLoader(db *sql.DB, rt *Runtime, diskCache *WasmDiskCache, maxSize ...int) *WorkflowLoader

NewWorkflowLoader creates a WorkflowLoader backed by the given database connection and wazero runtime. maxSize is the maximum number of compiled modules to keep in the LRU cache (defaults to 100 if <= 0). diskCache is an optional disk-backed cache for persistence across restarts (nil disables).

func (*WorkflowLoader) ActiveVersions 

func (l *WorkflowLoader) ActiveVersions(ctx context.Context) (map[string][]int, error)

ActiveVersions returns the set of workflow definition versions that have active (ready or running) instances. Used to determine which versions are safe to garbage-collect or deprecate.

SQL: SELECT def_name, def_version FROM workflow_instances 

WHERE status IN ('ready', 'running')
GROUP BY def_name, def_version

func (*WorkflowLoader) CacheStats 

func (l *WorkflowLoader) CacheStats() CacheStats

CacheStats returns current cache statistics for observability.

func (*WorkflowLoader) Deploy 

func (l *WorkflowLoader) Deploy(ctx context.Context, name string, version int, wasmBytes []byte, pluginDeps map[string]string, minVersion int) error

Deploy inserts a new workflow definition into the database. If the definition already exists, it is updated (upsert semantics).

SQL: INSERT INTO workflow_defs (name, version, wasm_bytes, abi_version, plugin_deps, min_version) 

VALUES ($1, $2, $3, $4, $5, $6)
ON CONFLICT (name, version) DO UPDATE SET
  wasm_bytes = $3, abi_version = $4, plugin_deps = $5, min_version = $6,
  deprecated = false, created_at = now()

func (*WorkflowLoader) Deprecate 

func (l *WorkflowLoader) Deprecate(ctx context.Context, name string, version int) error

Deprecate marks a workflow definition as deprecated. Active instances will continue to run, but new instances will not be created with this version (unless explicitly requested).

SQL: UPDATE workflow_defs SET deprecated = true WHERE name = $1 AND version = $2

func (*WorkflowLoader) ListVersions 

func (l *WorkflowLoader) ListVersions(ctx context.Context, name string) ([]WorkflowDef, error)

ListVersions returns all deployed versions of a workflow definition, ordered by version descending.

SQL: SELECT name, version, wasm_bytes, abi_version, plugin_deps, min_version, created_at, deprecated 

FROM workflow_defs WHERE name = $1 ORDER BY version DESC

func (*WorkflowLoader) Load 

func (l *WorkflowLoader) Load(ctx context.Context, name string, version int) (wazero.CompiledModule, error)

Load returns a compiled WASM module for the given workflow definition. It first checks the LRU cache; on a miss it queries the database, compiles, and caches the result. Returns an error if the definition is not found or is deprecated.

The disk cache (if configured) is checked after the LRU cache and before the database: if raw WASM bytes are found on disk, the database query is skipped. Compilation still occurs since the current wazero version does not support compiled module serialization.

SQL: SELECT wasm_bytes, abi_version, plugin_deps, min_version 

FROM workflow_defs
WHERE name = $1 AND version = $2 AND NOT deprecated

func (*WorkflowLoader) ResolveLatestVersion 

func (l *WorkflowLoader) ResolveLatestVersion(ctx context.Context, name string) (int, error)

ResolveLatestVersion returns the highest non-deprecated version for a workflow definition. Returns 0 with no error if no non-deprecated version exists.

SQL: SELECT COALESCE(MAX(version), 0) FROM workflow_defs 

WHERE name = $1 AND NOT deprecated

type WorkflowMemoryStats 

type WorkflowMemoryStats struct {
	DefName     string  `json:"def_name"`
	MinBytes    int64   `json:"min_bytes"`
	AvgBytes    float64 `json:"avg_bytes"`
	MaxBytes    int64   `json:"max_bytes"`
	P10         int64   `json:"p10"`
	P25         int64   `json:"p25"`
	P50         int64   `json:"p50"`
	P75         int64   `json:"p75"`
	P90         int64   `json:"p90"`
	P99         int64   `json:"p99"`
	SampleCount int     `json:"sample_count"`
}

WorkflowMemoryStats holds distribution statistics for per-definition memory usage.

type WorkflowState 

type WorkflowState interface {
	// Version returns the workflow definition version for this instance.
	Version() int
	// MinVersion returns the minimum version this code supports.
	MinVersion() int
	// ChildVersion returns the pinned version for a child workflow name
	// from compile-time WASM metadata. Returns (0, false) if no pin exists.
	ChildVersion(name string) (int, bool)
	// Priority returns the scheduling priority of this workflow instance.
	Priority() int
}

WorkflowState provides access to workflow instance state.

type WorkflowStore 

type WorkflowStore interface {
	// ClaimWorkflow atomically dequeues a runnable workflow instance.
	// Uses SELECT ... FOR UPDATE SKIP LOCKED.
	ClaimWorkflow(ctx context.Context, workerID string) (*WorkflowInstance, error)

	// ClaimWorkflows atomically claims up to limit runnable workflow instances.
	// Like ClaimWorkflow but batches multiple claims into one query.
	ClaimWorkflows(ctx context.Context, workerID string, limit int) ([]*WorkflowInstance, error)

	// ClaimStickyWorkflows atomically claims up to limit runnable workflow instances
	// that are sticky to this worker. Uses idx_instances_sticky for low-contention
	// claiming. Returns fewer than limit if not enough sticky workflows are ready.
	// Callers should fall back to ClaimWorkflows for remaining capacity.
	ClaimStickyWorkflows(ctx context.Context, workerID string, limit int) ([]*WorkflowInstance, error)

	// LoadEventHistory returns the full event history for a workflow.
	LoadEventHistory(ctx context.Context, workflowID string) ([]EventRecord, error)

	// LoadEventHistoryPaginated returns a page of event history for a workflow.
	// offset is the number of events to skip (0-based), limit caps the page size
	// (defaults to 1000 if limit <= 0, capped at 1000).
	LoadEventHistoryPaginated(ctx context.Context, workflowID string, offset, limit int) ([]EventRecord, error)

	// CountEventHistory returns the total number of events for a workflow.
	CountEventHistory(ctx context.Context, workflowID string) (int, error)

	// AppendEventHistory appends a single event to the history.
	// Uses ON CONFLICT (workflow_id, step) DO NOTHING for idempotency.
	AppendEventHistory(ctx context.Context, workflowID string, rec EventRecord) error

	// AppendEventHistoryBatch appends multiple events atomically.
	AppendEventHistoryBatch(ctx context.Context, workflowID string, recs []EventRecord) error

	// VerifyWorkflowEvents loads all events for a workflow and verifies their
	// integrity by recomputing SHA-256 checksums and comparing them against the
	// stored checksums (once the event_history.checksum column is migrated).
	// Before the migration, it loads and computes checksums silently and returns
	// nil. Returns an error if any checksum mismatch is detected.
	VerifyWorkflowEvents(ctx context.Context, workflowID string) error

	// LoadWASM returns the compiled WASM bytes for a workflow definition.
	LoadWASM(ctx context.Context, defName string, defVersion int) ([]byte, error)

	// GetWASMLength returns the byte length of the stored WASM binary.
	GetWASMLength(ctx context.Context, defName string, defVersion int) (int64, error)

	// ListVersions returns all deployed versions of a workflow.
	ListVersions(ctx context.Context, defName string) ([]int, error)

	// Heartbeat updates the heartbeat timestamp to prevent timeout.
	// Returns false if the workflow is no longer assigned to this worker
	// or if the generation does not match (workflow was reaped).
	Heartbeat(ctx context.Context, workflowID, workerID string, generation int64) (bool, error)

	// BatchHeartbeat updates heartbeat_at for all workflows assigned to this
	// worker with status 'running'. Uses a single UPDATE instead of N calls.
	// NOTE: This intentionally does NOT check per-workflow generation because
	// it operates on ALL workflows for a worker, and generations differ per
	// workflow. Individual generation-guarded operations (Heartbeat,
	// CompleteWorkflow, FailWorkflow, etc.) prevent double-execution even if
	// the batch heartbeat refreshes a stale workflow's heartbeat_at.
	BatchHeartbeat(ctx context.Context, workerID string) (int64, error)

	// CompleteWorkflow marks a workflow as completed with a result.
	CompleteWorkflow(ctx context.Context, workflowID, workerID string, generation int64, result string, queryState map[string]string) error

	// FailWorkflow marks a workflow as failed.
	FailWorkflow(ctx context.Context, workflowID, workerID string, generation int64, errorMsg, errorCode, errorOp string, queryState map[string]string) error

	// MoveToDeadLetterQueue marks a workflow as dead_lettered because it failed
	// after exhausting all retry attempts. This is a terminal status similar to
	// 'failed' but indicates the workflow was retried without success.
	MoveToDeadLetterQueue(ctx context.Context, workflowID, workerID string, generation int64, errMsg, errorCode, errorOp string) error

	// RetryWorkflow moves a dead_lettered workflow back to a runnable state.
	// Resets status to 'ready', clears the assigned worker and all error fields,
	// and sets next_wake_at to now so the workflow is picked up immediately.
	RetryWorkflow(ctx context.Context, workflowID string) error

	// ReleaseWorkflow returns a workflow to the ready queue.
	// Used when a workflow suspends (sleep/await signals).
	ReleaseWorkflow(ctx context.Context, workflowID, workerID string, generation int64, nextWakeAt time.Time) error

	// ContinueAsNew atomically creates a new workflow run AND completes the
	// current one in a single database transaction.  If the transaction fails
	// neither operation takes effect.  Returns the new run ID on success.
	ContinueAsNew(ctx context.Context, currentRunID, workerID string, generation int64, defName string, defVersion int, newInput json.RawMessage, newEvents []EventRecord, result string, queryState map[string]string, priority int) (newRunID string, err error)

	// FinalizeWorkflowSegment atomically appends new events and updates the
	// workflow status in a single database transaction.  finalStatus is one of
	// "done", "failed" or "ready" (suspend).  Fields not relevant to the chosen
	// status are ignored.  If the transaction fails neither events nor status
	// are written.
	FinalizeWorkflowSegment(ctx context.Context, runID, workerID string, generation int64, newEvents []EventRecord, finalStatus string, result string, errorCode string, errorOp string, queryState map[string]string, nextWakeAt time.Time) error

	// RequestCancellation sets the cancellation flag on a workflow.
	RequestCancellation(ctx context.Context, workflowID, reason string) error

	// CheckCancellation checks if a workflow has been cancelled.
	CheckCancellation(ctx context.Context, workflowID string) (cancelled bool, reason string, err error)

	// DeliverSignal stores a signal for a workflow.
	DeliverSignal(ctx context.Context, workflowID, signalName, payload string) error

	// PollSignal checks for a delivered signal.
	PollSignal(ctx context.Context, workflowID, signalName string) (payload string, found bool, err error)

	// PollCancellation checks whether the workflow has been cancelled.
	PollCancellation(ctx context.Context, workflowID string) (cancelled bool, reason string, err error)

	// PollAndClaimSignal atomically checks for and claims a pending signal.
	PollAndClaimSignal(ctx context.Context, workflowID, signalName string) (payload string, found bool, err error)

	// StartNewRun creates a new workflow instance.
	// If idempotencyKey is non-empty, provides exactly-once semantics: a
	// subsequent call with the same key returns the existing workflow ID
	// without creating a duplicate.
	// tenantID must be a valid UUID; the all-zeros default is accepted
	// for single-tenant installations without RLS.
	StartNewRun(ctx context.Context, runID, defName string, defVersion int, input json.RawMessage, idempotencyKey string, tenantID string, priority int) (string, bool, error)

	// StartChildWorkflow creates a child workflow instance linked to a parent.
	// defVersion is the explicit workflow definition version to use, or 0 to use
	// default resolution (SELECT MAX(version)).
	StartChildWorkflow(ctx context.Context, parentID, defName, inputJSON string, defVersion int, parentClosePolicy string, priority int) (runID string, err error)

	// StartChildWorkflowAtomic creates a child workflow and records the parent's
	// child_workflow event in a single transaction, guaranteeing exactly-once creation.
	StartChildWorkflowAtomic(ctx context.Context, childID, parentID, defName, inputJSON string, defVersion int, parentClosePolicy string, event EventRecord, priority int) (runID string, err error)

	// GetChildResult checks whether a child workflow has completed and returns its result.
	GetChildResult(ctx context.Context, runID string) (resultJSON string, completed bool, err error)

	// ReapStaleInstances reclaims workflow instances that have been running
	// but whose heartbeat has not been updated within the given timeout.
	// Returns the number of instances reclaimed.
	ReapStaleInstances(ctx context.Context, timeout time.Duration) (int, error)

	// GetQueryState returns the query state for a workflow instance key.
	GetQueryState(ctx context.Context, workflowID, key string) (string, error)

	// ListWorkflows returns workflow instances filtered by the given filter parameters.
	// Supported filters: Status, InputContains, ErrorContains, Search.
	// Supports pagination via Offset and Limit (default 100, max 1000).
	ListWorkflows(ctx context.Context, filter WorkflowFilter) ([]WorkflowInstance, error)

	// GetWorkflowByID returns a single workflow instance by ID.
	GetWorkflowByID(ctx context.Context, id string) (*WorkflowInstance, error)

	// CreateSchedule inserts a new cron schedule.
	CreateSchedule(ctx context.Context, s Schedule) error

	// ListSchedules returns all registered schedules.
	ListSchedules(ctx context.Context) ([]Schedule, error)

	// DeleteSchedule removes a schedule by name.
	DeleteSchedule(ctx context.Context, name string) error

	// SetScheduleEnabled enables or disables a schedule.
	SetScheduleEnabled(ctx context.Context, name string, enabled bool) error

	// GetDueSchedules returns enabled schedules whose next_run_at <= now().
	GetDueSchedules(ctx context.Context) ([]Schedule, error)

	// UpdateScheduleNextRun updates a schedule's next_run_at after firing.
	UpdateScheduleNextRun(ctx context.Context, name string, nextRun time.Time) error

	// ClaimDueSchedule advances a schedule from one firing instant to the next,
	// but only if it is still sitting on the instant the caller saw. It reports
	// whether this caller was the one that moved it.
	//
	// This is a compare-and-swap, and it is what makes a firing happen a
	// bounded number of times in a fleet. GetDueSchedules takes row locks, but
	// releases them when the read's transaction ends -- before the caller has
	// started anything -- so two workers polling a few milliseconds apart both
	// see the same row as due. Whoever wins this CAS owns that instant;
	// everybody else gets false and does nothing.
	//
	// expectedNextRun must be the NextRunAt the caller read. A mismatch means
	// another worker has already advanced the schedule.
	//
	// runID is recorded as the schedule's last_run_id, which is what makes
	// OverlapPolicy "skip" answerable: without it there is no way to tell a run
	// this schedule started from any other run of the same definition. Pass ""
	// to leave it unchanged.
	ClaimDueSchedule(ctx context.Context, name string, expectedNextRun, newNextRun time.Time, runID string) (claimed bool, err error)

	// LoadWorkflowConfig returns the max_history_length for a workflow definition.
	LoadWorkflowConfig(ctx context.Context, defName string, defVersion int) (maxHistoryLength int, err error)

	// LoadDAGSpec returns the dag_spec JSON for a workflow definition, or nil if none.
	LoadDAGSpec(ctx context.Context, defName string, defVersion int) (json.RawMessage, error)

	// TraceWorkflow sets the W3C trace_id on a workflow instance.
	TraceWorkflow(ctx context.Context, workflowID, traceID string) error

	// GetCompactionCandidates returns up to limit workflow IDs whose event
	// history exceeds the threshold and could benefit from compaction.
	GetCompactionCandidates(ctx context.Context, threshold int, limit int) ([]string, error)

	// LoadCompactionState returns the compaction state for a workflow, or nil
	// if the workflow has not been compacted.
	LoadCompactionState(ctx context.Context, workflowID string) (*CompactionState, error)

	// CompactHistory deletes old events and persists the compaction checkpoint
	// for a workflow. compactionStep records the step up to which events were
	// compacted; keepStep controls which events are deleted (step < keepStep).
	CompactHistory(ctx context.Context, workflowID string, compactionState []byte, compactionStep int, keepStep int) error

	// CreatePromise creates a new promise for a workflow.
	CreatePromise(ctx context.Context, workflowID, promiseName, promiseID string) error

	// ResolvePromise marks a promise as resolved with the given result.
	ResolvePromise(ctx context.Context, workflowID, promiseID, result string) error

	// RejectPromise marks a promise as rejected with the given error message.
	RejectPromise(ctx context.Context, workflowID, promiseID, errMsg string) error

	// GetPromise returns the current status and result of a promise.
	GetPromise(ctx context.Context, workflowID, promiseID string) (status string, result string, errMsg string, err error)

	// ListPromises returns all promises for a workflow ordered by creation time.
	ListPromises(ctx context.Context, workflowID string) ([]PromiseInfo, error)

	// CreateUpdateRequest registers an incoming update request for a workflow.
	// The update will be dispatched to the workflow's registered handler.
	CreateUpdateRequest(ctx context.Context, workflowID, updateName, payload, promiseID string) error

	// GetPendingUpdateRequests returns all pending (not yet dispatched) update
	// requests for a workflow.
	GetPendingUpdateRequests(ctx context.Context, workflowID string) ([]UpdateRequestInfo, error)

	// CompleteUpdateRequest marks an update request as completed with a result or error.
	CompleteUpdateRequest(ctx context.Context, workflowID, updateName, result, errMsg string) error

	// AcquireConcurrencyKey tries to acquire a concurrency key for a workflow.
	// Returns true if acquired, false if already held by another workflow.
	// Automatically releases expired keys during acquisition.
	AcquireConcurrencyKey(ctx context.Context, key, workflowID string, ttl time.Duration) (acquired bool, err error)

	// ReleaseConcurrencyKey releases a specific concurrency key.
	ReleaseConcurrencyKey(ctx context.Context, key string) error

	// ReleaseWorkflowConcurrencyKeys releases all concurrency keys held by a workflow.
	ReleaseWorkflowConcurrencyKeys(ctx context.Context, workflowID string) error

	// ReapExpiredConcurrencyKeys deletes all expired concurrency keys.
	// Returns the number of keys deleted.
	ReapExpiredConcurrencyKeys(ctx context.Context) (int64, error)

	// UpdateStickyWorker sets the sticky worker for a workflow.
	UpdateStickyWorker(ctx context.Context, workflowID, workerID string) error

	// ClearStickyWorker removes the sticky worker assignment.
	ClearStickyWorker(ctx context.Context, workflowID string) error

	// DeployWorkflowDef inserts or updates a workflow definition.
	DeployWorkflowDef(ctx context.Context, def *WorkflowDef) error

	// ListWorkflowDefs returns all versions of a workflow, ordered by version DESC.
	// If name is empty, returns all workflow definitions across all workflows.
	ListWorkflowDefs(ctx context.Context, name string) ([]WorkflowDef, error)

	// GetWorkflowDef returns a single workflow definition by name and version.
	GetWorkflowDef(ctx context.Context, name string, version int) (*WorkflowDef, error)

	// MarkVersionDeprecated sets the deprecated flag on a workflow version.
	MarkVersionDeprecated(ctx context.Context, name string, version int, deprecated bool) error

	// PurgeWorkflowDef permanently deletes a workflow definition (WASM bytes and all).
	PurgeWorkflowDef(ctx context.Context, name string, version int) error

	// CountActiveInstances returns the number of running/ready instances for a version.
	CountActiveInstances(ctx context.Context, name string, version int) (int, error)

	// ResolveLatestVersion resolves the latest version for a named definition.
	ResolveLatestVersion(ctx context.Context, defName string) (int, error)

	// ValidateVersion checks whether the given version is valid (exists and not deprecated).
	ValidateVersion(ctx context.Context, defName string, defVersion int) (bool, error)

	// GetActiveInstanceCountsByVersion returns a map of "name:version" -> count for
	// all workflow definitions that have active instances.
	GetActiveInstanceCountsByVersion(ctx context.Context) (map[string]int, error)

	// RecordWorkflowMemorySample inserts a new memory sample and updates the EWMA summary.
	RecordWorkflowMemorySample(ctx context.Context, defName string, sampleBytes int64) error

	// LoadMemoryEstimates returns EWMA mean bytes for all def_names.
	LoadMemoryEstimates(ctx context.Context) (map[string]float64, error)

	// LoadMemoryStats returns full distribution statistics for all def_names.
	LoadMemoryStats(ctx context.Context) ([]WorkflowMemoryStats, error)

	// QueueDepth returns the count of ready workflows in the store's task queues.
	QueueDepth(ctx context.Context) (int64, error)

	// CleanupMemorySamples deletes samples beyond maxSamplesPerDef per def_name.
	CleanupMemorySamples(ctx context.Context, maxSamplesPerDef int) (int64, error)

	// DeleteExpiredEvents deletes event history rows for workflows that are in a
	// terminal state (completed/failed) and whose last update is older than the
	// cutoff time.  It also deletes associated compaction states.
	// Returns the number of event rows deleted.
	DeleteExpiredEvents(ctx context.Context, olderThan time.Time) (int64, error)

	// ResolveTenantFromAPIKey looks up a tenant UUID by API key hash.
	// Returns uuid.Nil if the key is not found or revoked.
	ResolveTenantFromAPIKey(ctx context.Context, keyHash []byte) (uuid.UUID, error)

	// TerminateWorkflow force-terminates a workflow, setting status to
	// 'terminated'. Unlike FailWorkflow, this does not require the worker
	// to own the workflow. Use sparingly — it leaves the workflow in an
	// indeterminate state and should only be used when a workflow is truly stuck.
	TerminateWorkflow(ctx context.Context, workflowID, reason string) error

	// DeleteDeadLetteredWorkflows permanently deletes workflow instances that are
	// in the dead_lettered state and whose completed_at is older than the cutoff.
	// Child rows (event_history, signals, promises, concurrency_keys, update_requests)
	// are deleted automatically via ON DELETE CASCADE. Returns the number of
	// workflow instances deleted.
	DeleteDeadLetteredWorkflows(ctx context.Context, olderThan time.Time) (int64, error)

	// DeleteCompletedWorkflows permanently deletes workflow_instances rows that
	// reached a terminal, no-further-action status ('done', 'failed',
	// 'terminated' -- NOT 'dead_lettered', which has its own lifecycle and its
	// own deletion path, DeleteDeadLetteredWorkflows) and whose completed_at is
	// older than the cutoff.
	//
	// Finding S2: nothing else in this store ever deletes a
	// workflow_instances row. DeleteExpiredEvents deletes event_history for
	// terminal workflows but leaves the workflow_instances row itself; this
	// method is what actually bounds that table's size by active/recent
	// workflow count rather than lifetime workflow count.
	//
	// Child rows: same FK asymmetry DeleteDeadLetteredWorkflows documents.
	// PostgreSQL dropped the FK from event_history to workflow_instances
	// (migrations/postgres/003_procedures.sql) deliberately, because
	// finalize_workflow_status() deletes a 'done'/'failed' workflow's events
	// itself -- but MoveToDeadLetterQueue does not, and neither does
	// TerminateWorkflow, so a dead-lettered or force-terminated workflow's
	// events are not guaranteed to already be gone when this runs. MySQL and
	// SQL Server never dropped that FK and declare it ON DELETE CASCADE.
	// Implementations must delete event_history explicitly (not rely on
	// cascade) wherever the dialect does not cascade it, or those rows are
	// orphaned the instant the workflow_instances row is gone. The four
	// other child tables (workflow_signals, workflow_promises,
	// concurrency_keys, workflow_update_requests) cascade on every dialect.
	//
	// Implementations must run tenant-scoped (this store's own tenant) and,
	// on PostgreSQL, inside the RLS transaction context (beginTxWithRLS):
	// workflow_instances carries FORCE ROW LEVEL SECURITY with a fail-closed
	// policy, so a plain-pool DELETE issued by a non-superuser,
	// non-table-owner role (the shape cleat_app has in production) does not
	// silently affect zero rows -- it raises "cleat.tenant_id is not set".
	//
	// Batched like DeleteDeadLetteredWorkflows: large deletes run as
	// repeated bounded transactions rather than one unbounded one, so this
	// does not hold a long lock or a long-running transaction against a
	// table millions of rows deep. Returns the total number of
	// workflow_instances rows deleted.
	DeleteCompletedWorkflows(ctx context.Context, olderThan time.Time) (int64, error)

	// StreamEventHistory loads event history for a workflow in pages, returning
	// events through a channel. Events are fetched in pages of pageSize as the
	// caller reads from the channel. The channel is closed when all events have
	// been sent. The context can be used to cancel the stream mid-way.
	StreamEventHistory(ctx context.Context, workflowID string, pageSize int) (<-chan EventRecord, <-chan error)

	// GetChildCount returns the number of active (non-terminal) child workflows
	// for the given parent workflow. This is used for per-workflow child quota
	// enforcement. Terminal statuses ('done', 'failed', 'dead_lettered')
	// are excluded from the count.
	GetChildCount(ctx context.Context, parentWorkflowID string) (int, error)

	// GetConcurrencyKeyCount returns the number of non-expired concurrency keys
	// held by the given workflow. This is used for per-workflow concurrency key
	// quota enforcement. Keys whose expires_at is in the past are excluded.
	GetConcurrencyKeyCount(ctx context.Context, workflowID string) (int, error)

	// GetEventCount returns the event_count for a workflow instance.
	// Used by the engine for auto-ContinueAsNew when the event cap is hit.
	GetEventCount(ctx context.Context, workflowID string) (int, error)

	// GetAllowedSignalCallers returns the allowed_signals list for a workflow.
	// Returns nil when allowed_signals is NULL or empty (deny-all semantics).
	GetAllowedSignalCallers(ctx context.Context, workflowID string) ([]string, error)

	// SetWorkflowTag assigns a tag (e.g., "stable", "canary") to a specific version.
	SetWorkflowTag(ctx context.Context, workflowName string, version int, tag string) error

	// RemoveWorkflowTag deletes a tag assignment.
	RemoveWorkflowTag(ctx context.Context, workflowName string, tag string) error

	// GetWorkflowTag returns the version for the given tag.
	GetWorkflowTag(ctx context.Context, workflowName string, tag string) (int, error)

	// GetWorkflowTags returns all tag -> version mappings for a workflow.
	GetWorkflowTags(ctx context.Context, workflowName string) (map[string]int, error)

	// SetRoutingRule creates a traffic-splitting rule for a workflow version.
	SetRoutingRule(ctx context.Context, workflowName string, targetVersion int, weight float64) error

	// RemoveRoutingRule deletes a routing rule by its ID.
	RemoveRoutingRule(ctx context.Context, ruleID string) error

	// GetRoutingRules returns all routing rules for a workflow.
	GetRoutingRules(ctx context.Context, workflowName string) ([]RoutingRule, error)

	// PickVersionByRouting performs weighted random version selection.
	// Returns 0 if no routing rules exist (caller should use default resolution).
	PickVersionByRouting(ctx context.Context, workflowName string) (int, error)

	// ResolveVersionByTag resolves a version tag to a version number.
	// Special case: tag "latest" returns MAX(version) WHERE NOT deprecated.
	ResolveVersionByTag(ctx context.Context, workflowName string, tag string) (int, error)

	// AdminForceComplete marks a workflow as done, bypassing worker ownership.
	AdminForceComplete(ctx context.Context, workflowID string, generation int64, result string, operator string) error

	// AdminForceFail marks a workflow as failed, bypassing worker ownership.
	AdminForceFail(ctx context.Context, workflowID string, generation int64, errorMsg, errorCode string, operator string) error

	// AdminReReplay replays a workflow's event history for debugging.
	AdminReReplay(ctx context.Context, workflowID string, generation int64, operator string) error
}

Source Files

View all Source files

Directories

Path Synopsis

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.