store.Store: state persistence interface reference

The store.Store interface is the single source of truth for all Aether state. It aggregates four sub-interfaces — WorkflowRunStore, TaskRunStore, SchemaStore, and CronWorkflowStore — into one type that is injected into the engine via WithStore(). Implementations define the persistence strategy: in-memory, SQL, Redis, or any other backend.

The engine uses a Token-based optimistic concurrency model. Every UpdateWorkflowRun and UpdateTaskRun call passes a Token opaque uint64. Implementations that enforce concurrency control compare this token against the stored value and return ErrTokenMismatch if they differ. Single-threaded stores may ignore the token entirely.

Store (aggregate interface)

type Store interface {
    WorkflowRunStore
    TaskRunStore
    SchemaStore
    CronWorkflowStore
    Close() error
}

Close() releases any resources held by the store (connections, goroutines, etc.).

WorkflowRunStore

Manages the lifecycle of workflow-level run records.

type WorkflowRunStore interface {
    CreateWorkflowRun(ctx context.Context, run *WorkflowRun) error
    GetWorkflowRun(ctx context.Context, runID string) (*WorkflowRun, error)
    UpdateWorkflowRun(ctx context.Context, run *WorkflowRun) (*WorkflowRun, error)
    ListActiveWorkflowRuns(ctx context.Context) ([]*WorkflowRun, error)
    DeleteWorkflowRun(ctx context.Context, runID string) error
    ListWorkflowRunsByCronID(ctx context.Context, cronWorkflowID string) ([]*WorkflowRun, error)
}

CreateWorkflowRun

Stores the initial workflow run record. Called once at submit time with immutable data.

*WorkflowRun

required

The initial run record. RunID, Workflow (raw JSON), and CreatedAt are set by the engine.

GetWorkflowRun

Retrieves a workflow run by ID. Returns ErrNotFound if it does not exist.

string

required

The workflow run ID returned by Engine.Submit.

UpdateWorkflowRun

Persists mutable fields. Only non-nil pointer fields in the run argument are written; nil fields are left unchanged. Returns the post-update state on success.Uses Token-based optimistic locking — return ErrTokenMismatch on conflict.

ListActiveWorkflowRuns

Returns all non-terminal workflow runs that have a Deadline set. Used by the timeout watchdog.

DeleteWorkflowRun

Removes a workflow run and all its associated TaskRun records. Returns ErrNotFound if the run does not exist.

ListWorkflowRunsByCronID

Returns all WorkflowRun records created by a given CronWorkflow. Used for concurrency policy checks and history cleanup.

TaskRunStore

Manages task run records within a workflow.

type TaskRunStore interface {
    CreateTaskRun(ctx context.Context, run *TaskRun) error
    GetTaskRun(ctx context.Context, taskRunID string) (*TaskRun, error)
    UpdateTaskRun(ctx context.Context, run *TaskRun) (*TaskRun, error)
    ListTaskRuns(ctx context.Context, workflowRunID string) ([]*TaskRun, error)
    ListTaskRunsByParent(ctx context.Context, workflowRunID, parentRunID string) ([]*TaskRun, error)
    ListActiveTaskRuns(ctx context.Context) ([]*TaskRun, error)
}

CreateTaskRun must be idempotent by the composite key (workflowRunID, parentRunID, scope, taskName). If a record with the same key already exists, implementations MUST return nil — not an error, not a duplicate. The engine relies on this for concurrent-safe scheduling.

CreateTaskRun

Persists a new task run. Idempotent by composite key — repeated calls with the same key are a no-op. This is critical for loop scheduling where iterations share the same taskName but have unique Scope values.

GetTaskRun

Retrieves a single task run by its unique RunID.

UpdateTaskRun

Persists mutable fields using the nil-field partial update convention. Immutable fields (RunID, WorkflowRunID, ParentRunID, Scope, TaskName, etc.) are always ignored. Returns the post-update state on success.

ListTaskRuns

Returns all task runs for a given workflow run, in creation order.

ListTaskRunsByParent

Returns task runs sharing the same parent container. parentRunID="" returns top-level tasks. This is the core query for scoped DAG scheduling — advanceScope uses it to find sibling tasks.

ListActiveTaskRuns

Returns all non-terminal task runs that have a Deadline set. Used by the timeout watchdog.

SchemaStore

Persists executor schemas for recovery after a master restart.

type SchemaStore interface {
    UpsertSchema(ctx context.Context, workerID string, schema model.ExecutorSchema) error
    ListSchemas(ctx context.Context) ([]SchemaRecord, error)
    DeleteSchema(ctx context.Context, execType, workerID string) error
}

workerID="" in UpsertSchema marks a schema as an orphan (loaded during recovery). DeleteSchema accepts optional filters — empty string means no restriction on that dimension.

CronWorkflowStore

Manages CronWorkflow persistence.

type CronWorkflowStore interface {
    CreateCronWorkflow(ctx context.Context, record *CronWorkflowRecord) error
    GetCronWorkflow(ctx context.Context, cronID string) (*CronWorkflowRecord, error)
    UpdateCronWorkflow(ctx context.Context, record *CronWorkflowRecord) error
    DeleteCronWorkflow(ctx context.Context, cronID string) error
    ListCronWorkflows(ctx context.Context) ([]*CronWorkflowRecord, error)
}

DeleteCronWorkflow does NOT cascade-delete associated WorkflowRun records. ListCronWorkflows is called during Engine.Start() to re-register schedules after a crash.

Persistent model types

WorkflowRun

type WorkflowRun struct {
    // Immutable (set at creation)
    RunID          string
    Workflow       json.RawMessage // raw workflow JSON snapshot
    CreatedAt      time.Time
    CronWorkflowID string          // non-empty when created by a CronWorkflow trigger

    // Mutable (nil = do not modify in UpdateWorkflowRun)
    Status   *model.Phase
    Message  *string
    Outputs  *model.Outputs
    Metrics  *model.Metrics
    Deadline *time.Time  // absolute timeout; nil = no workflow-level deadline

    // Internal
    Token     uint64
    UpdatedAt time.Time
}

TaskRun

type TaskRun struct {
    // Immutable (set at creation)
    RunID         string
    WorkflowRunID string
    ParentRunID   string   // parent container TaskRun ID; "" = top-level scope
    Depth         int      // tree depth; 0 = top-level
    Scope         string   // path segment, e.g. "main-pipeline/"
    TaskName      string   // task name within current scope
    TemplateName  string   // referenced template name
    TemplateType  string   // "dag" / "task" / "loop"
    CreatedAt     time.Time

    // Mutable (nil = do not modify in UpdateTaskRun)
    Inputs     *model.Inputs
    Status     *model.Phase
    Message    *string
    Outputs    *model.Outputs
    Metrics    *model.Metrics
    RetryCount *int       // retries consumed; 0 = first attempt
    Deadline   *time.Time // absolute timeout; nil = no task-level deadline

    // Internal
    Token     uint64
    UpdatedAt time.Time
}

Error sentinels

error

Returned when a requested resource does not exist.

error

Returned when an Update call’s token does not match the stored token. Signals normal optimistic-lock contention — not an infrastructure failure. The engine treats this as a no-op for concurrent updates.

Inject with WithStore

engine, err := aether.New(
    aether.WithStore(myStore), // required
    // ...
)

Protocol

Go API

store.Store: state persistence interface reference

Store (aggregate interface)

WorkflowRunStore

TaskRunStore

SchemaStore

CronWorkflowStore

Persistent model types

WorkflowRun

TaskRun

Error sentinels

Inject with WithStore

​Store (aggregate interface)

​WorkflowRunStore

​TaskRunStore

​SchemaStore

​CronWorkflowStore

​Persistent model types

​WorkflowRun

​TaskRun

​Error sentinels

​Inject with WithStore

Store (aggregate interface)

WorkflowRunStore

TaskRunStore

SchemaStore

CronWorkflowStore

Persistent model types

WorkflowRun

TaskRun

Error sentinels

Inject with WithStore