Continuum.Runtime.Journal behaviour (continuum v0.8.1)

Copy Markdown View Source

Behaviour for the event-history journal.

Two adapters ship with v0.1:

All append operations carry a lease_token (or nil for unleased in-memory / pre-dispatch execution). The Postgres adapter rejects writes whose token does not match the run row's fencing token; nil only writes to unleased rows.

Summary

Callbacks

append!(instance, run_id, event, lease_token)
await_child_terminal!(instance, parent_run_id, child_run_id, command_id, seq, lease_token)

Journal the parent's view of a child's terminal state, or report :pending.

complete!(instance, run_id, result, lease_token)
deliver_signal!(instance, run_id, name, payload)
deliver_signal!(instance, run_id, name, payload, opts)
fail!(instance, run_id, error, lease_token)
get_run(instance, run_id)

Look up the run record. Returns nil if no such run, or a map with at least :state, :result, :error keys (atoms / terms — already decoded).

load(instance, run_id)
load_with_snapshot(instance, run_id, lease_token)
start_child!(instance, parent_run_id, child, lease_token)

Start a child run and journal child.started_event on the parent.

start_run(instance, run_id, workflow, input)
start_run(instance, run_id, workflow, input, opts)
suspend!(instance, run_id, lease_token)
take_snapshot!(instance, snapshot)

Functions

default()

Returns the configured default journal adapter.

Callbacks

append!(instance, run_id, event, lease_token)

@callback append!(
  instance :: Continuum.Runtime.Instance.t(),
  run_id :: binary(),
  event :: map(),
  lease_token :: integer() | nil
) :: :ok

await_child_terminal!(instance, parent_run_id, child_run_id, command_id, seq, lease_token)

(since 0.8.0) (optional) 
@callback await_child_terminal!(
  instance :: Continuum.Runtime.Instance.t(),
  parent_run_id :: binary(),
  child_run_id :: binary(),
  command_id :: term(),
  seq :: non_neg_integer(),
  lease_token :: integer() | nil
) ::
  {:completed, term(), map()}
  | {:failed, term(), map()}
  | {:cancelled, map()}
  | :pending

Journal the parent's view of a child's terminal state, or report :pending.

Returns the winner event alongside the outcome so the parent can advance its cursor over exactly the event that was appended.

complete!(instance, run_id, result, lease_token)

@callback complete!(
  instance :: Continuum.Runtime.Instance.t(),
  run_id :: binary(),
  result :: term(),
  lease_token :: integer() | nil
) :: :ok

deliver_signal!(instance, run_id, name, payload)

(optional) 
@callback deliver_signal!(
  instance :: Continuum.Runtime.Instance.t(),
  run_id :: binary(),
  name :: atom(),
  payload :: term()
) :: :ok | {:ok, binary()} | {:error, term()}

deliver_signal!(instance, run_id, name, payload, opts)

(optional) 
@callback deliver_signal!(
  instance :: Continuum.Runtime.Instance.t(),
  run_id :: binary(),
  name :: atom(),
  payload :: term(),
  opts :: keyword()
) :: {:ok, binary(), :delivered | :duplicate} | {:error, term()}

fail!(instance, run_id, error, lease_token)

@callback fail!(
  instance :: Continuum.Runtime.Instance.t(),
  run_id :: binary(),
  error :: term(),
  lease_token :: integer() | nil
) :: :ok

get_run(instance, run_id)

@callback get_run(instance :: Continuum.Runtime.Instance.t(), run_id :: binary()) ::
  nil | map()

Look up the run record. Returns nil if no such run, or a map with at least :state, :result, :error keys (atoms / terms — already decoded).

load(instance, run_id)

@callback load(instance :: Continuum.Runtime.Instance.t(), run_id :: binary()) :: [map()]

load_with_snapshot(instance, run_id, lease_token)

@callback load_with_snapshot(
  instance :: Continuum.Runtime.Instance.t(),
  run_id :: binary(),
  lease_token :: integer() | nil
) :: {Continuum.Snapshot.t() | nil, [map()]}

start_child!(instance, parent_run_id, child, lease_token)

(since 0.8.0) (optional) 
@callback start_child!(
  instance :: Continuum.Runtime.Instance.t(),
  parent_run_id :: binary(),
  child :: map(),
  lease_token :: integer() | nil
) :: :ok

Start a child run and journal child.started_event on the parent.

Optional: a journal that does not export it makes await child and start_child/3 unavailable, and the effect raises with that explanation rather than silently doing nothing.

child carries :child_run_id, :workflow, :input, :parent_command_id, :trace_context, and :started_event.

start_run(instance, run_id, workflow, input)

@callback start_run(
  instance :: Continuum.Runtime.Instance.t(),
  run_id :: binary(),
  workflow :: module(),
  input :: term()
) :: :ok | {:error, term()}

start_run(instance, run_id, workflow, input, opts)

(optional) 
@callback start_run(
  instance :: Continuum.Runtime.Instance.t(),
  run_id :: binary(),
  workflow :: module(),
  input :: term(),
  opts :: keyword()
) :: :ok | {:ok, Continuum.Runtime.Lease.t()} | {:error, term()}

suspend!(instance, run_id, lease_token)

@callback suspend!(
  instance :: Continuum.Runtime.Instance.t(),
  run_id :: binary(),
  lease_token :: integer() | nil
) :: :ok

take_snapshot!(instance, snapshot)

@callback take_snapshot!(
  instance :: Continuum.Runtime.Instance.t(),
  snapshot :: Continuum.Snapshot.t()
) :: :ok

Functions

default()

Returns the configured default journal adapter.