Runic.Runner.Store behaviour (Runic v0.1.0-alpha.9)

Behaviour for workflow persistence adapters.

Adapters handle saving and loading workflow event logs for durability across process restarts.

Stream Semantics (Event-Sourced)

The preferred interface uses append/3 and stream/2 for incremental event persistence. Events are appended after each execution cycle and streamed on recovery to rebuild workflow state via Workflow.from_events/1.

Stores that implement append/3 and stream/2 get automatic event-sourced checkpointing and recovery from the Worker.

stream/2 must always return the full event stream for a workflow. Stores that support cursor-aware or windowed replay can additionally implement stream/3 with options.

Supported stream/3 options:

:after_cursor — exclusive lower bound. after_cursor: 10 returns events with sequence/cursor greater than 10.
:limit — optional maximum number of events to return. Adapters may ignore this when their backing stream does not support bounded reads.
:batch_size — optional page-size hint for stores that fetch event rows in batches.

stream/3 should be a superset of stream/2: calling it with an empty option list should return the full stream. Adapters should ignore unknown options they do not support.

Legacy Semantics (Snapshot)

The save/3 and load/2 callbacks persist the full workflow log as a snapshot. These remain the required baseline interface for backward compatibility. Stores that only implement save/load continue to work unchanged.

Optional Capabilities

Snapshots (save_snapshot/4, load_snapshot/2): Point-in-time workflow snapshots for faster recovery (replay from snapshot + events after cursor instead of full replay).
Fact storage (save_fact/3, load_fact/2): Content-addressed fact value storage for hybrid rehydration without loading all values into memory.

Summary

Types

cursor()

event()

log()

state()

stream_opts()

workflow_id()

Callbacks

append(workflow_id, events, state)

checkpoint(workflow_id, log, state)

delete(workflow_id, state)

exists?(workflow_id, state)

init_store(opts)

list(state)

load(workflow_id, state)

load_fact(fact_hash, state)

load_snapshot(workflow_id, state)

save(workflow_id, log, state)

save_fact(fact_hash, value, state)

save_snapshot(workflow_id, cursor, snapshot, state)

stream(workflow_id, state)

stream(workflow_id, state, stream_opts)

Functions

supports_snapshots?(store_mod)

Returns true if the store module supports snapshot save/load semantics.

supports_stream?(store_mod)

Returns true if the store module supports event-sourced stream semantics.

supports_stream_options?(store_mod)

Returns true if the store module supports option-aware stream replay.

Types

cursor()

@type cursor() :: non_neg_integer()

event()

@type event() :: struct()

log()

@type log() :: [struct()]

state()

@type state() :: term()

stream_opts()

@type stream_opts() :: [
  after_cursor: cursor(),
  limit: pos_integer(),
  batch_size: pos_integer()
]

workflow_id()

@type workflow_id() :: term()

Callbacks

append(workflow_id, events, state)

(optional)

@callback append(workflow_id(), events :: [event()], state()) ::
  {:ok, cursor()} | {:error, term()}

checkpoint(workflow_id, log, state)

(optional)

@callback checkpoint(workflow_id(), log(), state()) :: :ok | {:error, term()}

delete(workflow_id, state)

(optional)

@callback delete(workflow_id(), state()) :: :ok | {:error, term()}

exists?(workflow_id, state)

(optional)

@callback exists?(workflow_id(), state()) :: boolean()

init_store(opts)

@callback init_store(opts :: keyword()) :: {:ok, state()} | {:error, term()}

list(state)

(optional)

@callback list(state()) :: {:ok, [workflow_id()]} | {:error, term()}

load(workflow_id, state)

@callback load(workflow_id(), state()) :: {:ok, log()} | {:error, :not_found | term()}

load_fact(fact_hash, state)

(optional)

@callback load_fact(fact_hash :: term(), state()) ::
  {:ok, term()} | {:error, :not_found | term()}

load_snapshot(workflow_id, state)

(optional)

@callback load_snapshot(workflow_id(), state()) ::
  {:ok, {cursor(), binary()}} | {:error, :not_found | term()}

save(workflow_id, log, state)

@callback save(workflow_id(), log(), state()) :: :ok | {:error, term()}

save_fact(fact_hash, value, state)

(optional)

@callback save_fact(fact_hash :: term(), value :: term(), state()) ::
  :ok | {:error, term()}

save_snapshot(workflow_id, cursor, snapshot, state)

(optional)

@callback save_snapshot(workflow_id(), cursor(), snapshot :: binary(), state()) ::
  :ok | {:error, term()}

stream(workflow_id, state)

(optional)

@callback stream(workflow_id(), state()) ::
  {:ok, Enumerable.t()} | {:error, :not_found | term()}

stream(workflow_id, state, stream_opts)

(optional)

@callback stream(workflow_id(), state(), stream_opts()) ::
  {:ok, Enumerable.t()} | {:error, :not_found | term()}

Functions

supports_snapshots?(store_mod)

@spec supports_snapshots?(module()) :: boolean()

Returns true if the store module supports snapshot save/load semantics.

supports_stream?(store_mod)

@spec supports_stream?(module()) :: boolean()

Returns true if the store module supports event-sourced stream semantics.

supports_stream_options?(store_mod)

@spec supports_stream_options?(module()) :: boolean()

Returns true if the store module supports option-aware stream replay.