Elixir Flow SDK

FerricStore Flow already exposes low-level embedded commands such as FerricStore.flow_create/2, FerricStore.flow_claim_due/2, and FerricStore.flow_complete/3.

The Elixir Flow SDK is the high-level embedded API for durable queues and state-machine workflows. It builds on FerricStore.flow_* commands. It does not add a replay engine, does not use the native TCP transport, and does not change Flow durability or atomicity: successful Flow writes are accepted through the quorum path and written to disk.

First Workflow

Start with one state that completes or retries:

defmodule EmailFlow do
  use FerricStore.Flow.Workflow,
    type: "email",
    partition_by: [:recipient_id],
    initial_state: :queued

  state :queued do
    mode :fifo
    lease_ms 30_000
    on_ok complete()
    on_error retry_or: :failed
  end

  state :failed do
    on_ok complete()
  end
end

Create and run one unit of work:

{:ok, _policy} = EmailFlow.install_policy()

{:ok, _flow} =
  EmailFlow.create(%{
    id: "email-1",
    recipient_id: "user-1",
    payload: "welcome:user-1"
  })

EmailFlow.run_once(:queued, worker: "worker-1", handler: fn job ->
  send_email(job.payload)
  {:ok, "sent"}
end)

{:ok, history} = EmailFlow.history("email-1")

The SDK handles command construction. Flow core still handles leases, fencing, durable state, and history.

How It Works Internally

Workflow SDK
  -> embedded FerricStore.flow_* functions
  -> Flow core
  -> Ra + Bitcask durable records
  -> hot Flow indexes
  -> async LMDB cold projections

Truth stays in Flow core. SDK modules are command builders and worker helpers.

Define A Workflow

defmodule BillingFlow do
  use FerricStore.Flow.Workflow,
    type: "billing",
    partition_by: [:tenant_id, :invoice_id],
    initial_state: :created

  state :created do
    lease_ms 60_000
    claim_payload true, max_bytes: 64_000

    retry max_retries: 8,
          backoff: [kind: :exponential, base_ms: 1_000, max_ms: :timer.hours(1)]

    on_ok :charged
    on_error retry_or: :failed
  end

  state :charged do
    lease_ms 30_000
    on_ok complete()
    on_error fail()
  end
end

This creates a normal Elixir module with functions such as:

BillingFlow.create(attrs)
BillingFlow.child(attrs)
BillingFlow.fanout(parent_job, children, opts)
BillingFlow.claim_due(:created, worker: "w1", limit: 100)
BillingFlow.run_once(:created, worker: "w1", handler: &handle/1)
BillingFlow.ok(job, result)
BillingFlow.error(job, reason)
BillingFlow.history(id)

Workflow Options

type is required.

use FerricStore.Flow.Workflow, type: "billing"

Every generated command uses this Flow type.

store selects the embedded API module. Default is FerricStore. Most applications should omit this option.

use FerricStore.Flow.Workflow,
  type: "billing",
  store: MyApp.Store

Use store when your app defines its own embedded instance:

defmodule MyApp.Store do
  use FerricStore, data_dir: "/data/ferric"
end

Then the SDK calls MyApp.Store.flow_create/2, MyApp.Store.flow_claim_due/2, etc.

partition_by builds partition_key from create attrs:

partition_by: [:tenant_id, :invoice_id]

Attrs:

%{tenant_id: "t1", invoice_id: "i9"}

become:

partition_key: "fpk:2:t12:i9"

Each component is encoded as its byte length, ":", and its bytes after the "fpk:" prefix. This makes component boundaries unambiguous even when values contain ":". Other SDKs must implement the same encoding when they derive partition keys. Explicit partition_key values are passed through unchanged.

Same partition goes to the same shard and keeps ordered Flow behavior.

initial_state sets the state used by create/2. If omitted, first declared state is used.

State DSL

state name do ... end declares SDK defaults and action rules.

`mode`

Durable execution mode for this state:

state :queued do
  mode :fifo
end

state :review do
  mode :parallel
end

Modes are installed by install_policy/1 through FLOW.POLICY.SET; they are not attached to individual create, transition, or claim commands. States with no declared mode are parallel. FIFO lanes are scoped by Flow type, state, and partition key.

`lease_ms`

Default lease for claims in this state:

state :created do
  lease_ms 60_000
end

Generated claim:

BillingFlow.claim_due(:created, worker: "w1")

calls:

FerricStore.flow_claim_due("billing",
  state: "created",
  worker: "w1",
  lease_ms: 60_000
)

`claim_payload`

Default payload hydration for claims:

claim_payload true, max_bytes: 64_000

This makes workers receive payload bytes by default up to the cap. If payload is larger than the cap, Flow returns payload reference/omitted metadata instead of materializing the large value.

Payload rule:

metadata/index/transition paths do not read payload bytes
payload read happens only when command requests payload hydration

`retry`

Per-state retry policy:

retry max_retries: 8,
      backoff: [kind: :fixed, base_ms: 1_000, max_ms: 30_000],
      exhausted_to: "failed"

This maps to the existing Flow retry policy. Limits and validation are enforced by Flow core.

`on_ok`

Success action for BillingFlow.ok(job, result).

Transition to another state:

on_ok :charged

Complete terminally:

on_ok complete()

`on_error`

Error action for BillingFlow.error(job, reason).

Retry, then move to a state when retry budget is exhausted:

on_error retry_or: :failed

Fail terminally:

on_error fail()

Creating Flows

{:ok, flow} =
  BillingFlow.create(%{
    id: "invoice-123",
    tenant_id: "tenant-a",
    invoice_id: "invoice-123",
    payload: %{amount: 4200},
    correlation_id: "order-9"
  })

This calls flow_create/2 with:

type: "billing",
state: "created",
partition_key: "fpk:8:tenant-a11:invoice-123"

create_many/2 accepts a list of attrs:

BillingFlow.create_many([
  %{id: "invoice-1", tenant_id: "t1", invoice_id: "1", payload: p1},
  %{id: "invoice-2", tenant_id: "t2", invoice_id: "2", payload: p2}
])

The SDK passes per-item partition keys and lets Flow core group by shard. Each shard group keeps Flow batch atomicity.

Children and Fanout

Flow supports parent/child workflows through flow_spawn_children/3. The SDK adds child builders and fanout helpers so users do not have to hand-write the parent metadata and guard fields.

Build child specs from workflow modules:

email =
  EmailFlow.child(%{
    id: "email-123",
    tenant_id: "tenant-a",
    invoice_id: "invoice-123",
    payload: %{template: "paid"}
  })

audit =
  AuditFlow.child(%{
    id: "audit-123",
    tenant_id: "tenant-a",
    invoice_id: "invoice-123",
    payload: %{event: "invoice_paid"}
  })

child/2 adds the child workflow type, initial state, and derived partition_key. The parent link is still added by Flow core during spawn_children.

Fan out from a claimed parent job:

BillingFlow.fanout(parent_job, [email, audit],
  group_id: "notify-and-audit",
  wait: :all,
  on_all_ok: :children_done,
  on_any_error: :children_failed
)

This calls:

FerricStore.flow_spawn_children(parent_id, children,
  group_id: "notify-and-audit",
  wait: :all,
  wait_state: "waiting_children",
  success: "children_done",
  failure: "children_failed",
  partition_key: parent_job.partition_key,
  from_state: parent_job.state,
  lease_token: parent_job.lease_token,
  fencing_token: parent_job.fencing_token
)

fanout/3 is an alias for spawn_children/3. Both accept the same options.

Important options:

:group_id - idempotency key for this child group. Default is "fanout". Use a specific value when a parent can spawn more than one group.
:wait - :all, :any, or :none. Default is :all.
:wait_state - parent state while waiting. Default is "waiting_children" for :all and :any.
:on_all_ok / :on_success / :success - parent state when child group succeeds.
:on_any_error / :on_failure / :failure - parent state when child group fails.
:child_failure_policy / :on_child_failed - :fail_parent or :ignore. Default is :fail_parent.
:on_parent_closed - :cancel_children or :abandon_children. Default is :cancel_children.

Query children:

BillingFlow.children(parent_job, count: 100)
BillingFlow.waiting_children(parent_job, count: 100)

children/2 uses the parent lineage index. waiting_children/2 filters out terminal child states (completed, failed, cancelled) client-side.

Fanout correctness:

parent update and same-shard children are atomic in one Flow command
cross-shard children use Flow cross-shard coordination
group_id makes duplicate fanout calls idempotent
child completion updates parent summary through Flow core

Claiming Work

{:ok, jobs} =
  BillingFlow.claim_due(:created,
    worker: "payment-1",
    limit: 100
  )

Returned items are %FerricStore.Flow.Job{} structs. The raw Flow record is available as job.record.

Important fields:

job.id
job.state
job.partition_key
job.lease_token
job.fencing_token
job.payload
job.payload_ref

You can also claim any state:

BillingFlow.claim_due(:any, worker: "worker-1", limit: 100)

or selected states:

BillingFlow.claim_due([:created, :charged], worker: "worker-1")

Handling Jobs

Manual handling:

for job <- jobs do
  case charge(job.payload) do
    {:ok, receipt} ->
      BillingFlow.ok(job, receipt)

    {:error, reason} ->
      BillingFlow.error(job, reason)
  end
end

ok/3 and error/3 carry partition_key, lease_token, and fencing_token from the job. Core Flow still enforces lease/fencing correctness.

For one-shot polling without supervising a worker:

BillingFlow.run_once(:created,
  worker: "payment-1",
  limit: 100,
  handler: fn job ->
    case charge(job.payload) do
      {:ok, receipt} -> {:ok, receipt}
      {:error, reason} -> {:error, reason}
    end
  end
)

For applying handler result to an already claimed job:

BillingFlow.handle(job, fn job ->
  do_work(job)
end)

Handler return contract:

{:ok, result}    -> workflow.ok(job, result)
{:error, reason} -> workflow.error(job, reason)
:noreply         -> handler owns final Flow command
other            -> workflow.error(job, {:unexpected_worker_result, other})
exception        -> workflow.error(job, exception)

Explicit commands are also available:

BillingFlow.transition(job, :charged, payload)
BillingFlow.complete(job, result)
BillingFlow.retry(job, reason)
BillingFlow.fail(job, reason)
BillingFlow.extend_lease(job, lease_ms: 60_000)

Reads and Queries

SDK read helpers map directly to Flow read APIs:

BillingFlow.get(id)
BillingFlow.get(id, payload: true)
BillingFlow.history(id)
BillingFlow.history(id, include_cold: true)
BillingFlow.list(:created, count: 100)
BillingFlow.by_parent(parent_id, count: 100)
BillingFlow.by_root(root_id, count: 100)
BillingFlow.by_correlation(correlation_id, count: 100)
BillingFlow.info()
BillingFlow.stuck()
BillingFlow.children(parent_job)
BillingFlow.waiting_children(parent_job)

By default, get/2 does not hydrate payload. Ask for payload explicitly:

BillingFlow.get(id, payload: true, payload_max_bytes: 64_000)

Installing Policy

The DSL can write state execution modes and retry policy defaults into Flow:

BillingFlow.install_policy()

This calls:

FerricStore.flow_policy_set("billing", states: ...)

BillingFlow.install_policy/1 treats the workflow module as the source of truth and replaces the stored policy by default, so removing a declaration also removes it from the installed policy. Pass replace: false only when the workflow declaration should patch an existing policy.

Direct FerricStore.flow_policy_set/2 calls patch by default: omitted type fields, states, and nested state settings are preserved. Successful set/get responses include the monotonic generation; pass expected_generation: generation when the caller must reject a stale update. Concurrent implicit patches are merged atomically by the policy allocator without losing either patch.

Command-local retry policy still wins over stored policy. This is useful when you want production defaults set once at boot.

Install a type-level maximum active runtime through install_policy/1, or override it on one create call:

BillingFlow.install_policy(max_active_ms: :timer.minutes(10))
BillingFlow.create(attrs, max_active_ms: :timer.minutes(2))
BillingFlow.create(attrs, max_active_ms: :infinity)

The deadline starts at Flow creation and includes queue wait, scheduled delay, running work, and retry backoff. It is not a per-state setting.

Optional Worker

You may use your own cron, Oban, Broadway, GenServer loop, or Kubernetes job. The SDK worker is only a convenience poller.

children = [
  {FerricStore.Flow.Worker,
   workflow: BillingFlow,
   state: :created,
   worker: "payment-#{node()}",
   limit: 100,
   interval_ms: 250,
   handler: &MyApp.PaymentWorker.handle/1}
]

Handler contract:

def handle(job) do
  case charge(job.payload) do
    {:ok, receipt} -> {:ok, receipt}
    {:error, reason} -> {:error, reason}
  end
end

Worker behavior:

{:ok, result}    -> workflow.ok(job, result)
{:error, reason} -> workflow.error(job, reason)
:noreply         -> handler owns final Flow command
exception        -> workflow.error(job, exception)

Reclaim

Expired running leases can be reclaimed through normal Flow claim/reclaim semantics:

BillingFlow.reclaim_once(:running,
  worker: "recovery-1",
  limit: 100
)

reclaim_once/2 returns %FerricStore.Flow.Job{} structs, same as claim_due/2. Use it for explicit recovery loops. Normal claim_due/2 can also mix due work with expired lease reclaim based on core Flow options.

Atomicity Boundary

Flow command atomicity stays the same as core Flow:

one Flow command = atomic through Ra/Bitcask
same-shard many command = atomic for that shard group
cross-shard many command = atomic per shard group

Handler-side reads/writes are not automatically part of the Flow transition.

Good:

customer = MyApp.Store.get!("customer:#{job.payload.customer_id}")
BillingFlow.ok(job, receipt)

If you need Flow transition plus other writes to be atomic, use an explicit transaction/cross-op API when available. Do not hide that inside SDK helpers.

Payload Guidance

Small payloads can be carried inline.

Large payloads should use Flow value refs or app-owned blob refs. The SDK does not put payload bytes into query indexes. Claim/get/history only hydrate payload when requested and bounded by payload_max_bytes.

Recommended defaults:

claim_payload true, max_bytes: 64_000

For larger worker inputs:

BillingFlow.claim_due(:created,
  worker: "w1",
  payload: true,
  payload_max_bytes: 2_000_000
)

Use large caps carefully. They increase response bytes and client memory use.

Testing

Use store: to inject a fake embedded API module:

defmodule TestBillingFlow do
  use FerricStore.Flow.Workflow,
    type: "billing",
    store: MyFakeStore,
    partition_by: [:tenant_id]

  state :created do
    on_ok complete()
    on_error retry_or: :failed
  end
end

This keeps SDK unit tests fast and separate from Ra/Bitcask integration tests.

What SDK Does Not Do

The SDK does not:

use a replay-driven workflow engine
hide Flow transitions inside normal Elixir code
make arbitrary KV writes atomic with Flow transitions
require native TCP transport or an external client
make LMDB part of command correctness
read payload bytes unless the command asks for payload hydration

← Previous Page Embedded Mode

Next Page → Commands Reference