Continuum.Replay (continuum v0.8.1)

Replay a journaled history against workflow code, without touching anything.

This is the replay loop with the runtime removed: no lease is taken, no engine is started, no row is written. The context is handed its history and snapshot up front and carries Continuum.Runtime.Journal.ReadOnly, so the two paths that used to mutate during "read-only" replay — the in-memory adapter running an activity body inline past the journaled tail, and the Postgres adapter consuming a pending signal row to resolve a tail await — both refuse instead.

run/4 replays a history you already hold. of_run/2 loads one out of Postgres first, resolving the run's journaled (workflow, version_hash) pair through Continuum.VersionRegistry exactly the way a resuming engine would, and reports what the workflow would do next.

mix continuum.replay is the operator-facing front end for of_run/2.

Outcomes

Replay ends in one of four ways:

{:ok, result} — the history carried the workflow to a return value.
{:suspended, reason} — the workflow stopped at a pending effect. A {:history_exhausted, _} reason means the history ran out before the workflow was done, which under a live journal is where the next effect would have been performed.
{:continued, next_run_id} — the workflow tail-called continue_as_new/1.
{:error, reason} — most usefully {:error, {:error, %Continuum.ReplayDriftError{}, _}}, which names the cursor and command id where code and history disagree.

Summary

Types

report()

What of_run/2 reports back about a durable run.

result()

Functions

of_run(run_id, opts \\ [])

Load a durable run and replay it read-only.

run(workflow_module, input, history, opts \\ [])

Replay history against workflow_module and return the outcome.

Types

report()

(since 0.8.0)

@type report() :: %{
  run_id: binary(),
  workflow: String.t(),
  version_hash: binary(),
  entrypoint: module(),
  namespace: String.t(),
  stored_state: atom(),
  stored_result: term(),
  event_count: non_neg_integer(),
  snapshot: %{through_seq: integer(), format_version: integer()} | nil,
  outcome: :completed | :suspended | :continued | :drift | :error,
  detail: term(),
  agrees_with_stored_result?: boolean() | nil
}

What of_run/2 reports back about a durable run.

result()

(since 0.8.0)

@type result() ::
  {:ok, term()}
  | {:suspended, term()}
  | {:continued, binary()}
  | {:error, term()}

Functions

of_run(run_id, opts \\ [])

(since 0.8.0)

@spec of_run(
  binary(),
  keyword()
) :: {:ok, report()} | {:error, term()}

Load a durable run and replay it read-only.

Resolves the run's journaled (workflow, version_hash) through Continuum.VersionRegistry. A version this node cannot load is reported as {:error, {:unknown_version, _}} rather than replayed against whatever code happens to be loaded — replaying the wrong version reports drift that is an artifact of the deploy, not of the run.

Options:

:instance / :repo — where to read from.
:snapshot — false to ignore stored snapshots and replay from events alone. Defaults to true.
:against — replay against this module instead of the journaled entrypoint, to see what a code change would do to a run in flight.
:redactor — applied to the reported result and suspend reason. Defaults to the :observer_redactor application setting.

run(workflow_module, input, history, opts \\ [])

(since 0.8.0)

@spec run(module(), term(), [map()], keyword()) :: result()

Replay history against workflow_module and return the outcome.

Options:

:run_id — the id reported in drift errors. Defaults to "continuum-replay".
:snapshot — a Continuum.Snapshot to replay the compacted prefix from. Ignored when its version hash does not match the module's.
:instance — the instance whose name appears in the context.
:journal — defaults to Continuum.Runtime.Journal.ReadOnly. Overriding it gives up the read-only guarantee; Continuum.Test does so deliberately for adapter-specific tests.
:lease_token — only meaningful alongside a writable :journal.
:follow_journaled_entrypoint — when replay drifts on the very first event because the history names a generated V_<hash> entrypoint, retry through that entrypoint. Defaults to true, which is what makes replay(MyFlow, ...) work against a durable history. of_run/2 sets it to false: it has already resolved the entrypoint, and following the journaled one would silently override :against.