March Docs

Conduit.API

Functions

fncancel_croncancel_cron(storage : s, cron_id : String) : Result((), String)#

Cancel a cron schedule permanently (deletes the DB row).

fncancel_workflowcancel_workflow(workflow_id : String, reason : String, storage : s) : Result((), String)#

Cancel a running workflow with a reason.

fncluster_leadercluster_leader() : Option(String)#

Return the node_id of the current leader from this node's perspective.

The leader is the node that holds the Postgres advisory leader lock
(lock id 8473625). This value is updated in Vault by the local leader-election
background task — it reflects the most recent election result this node observed.
Returns None if the leader is unknown or this node has not yet run an election tick.
fncluster_nodescluster_nodes(storage : s) : Result(List(NodeInfo), String)#

Return all currently active cluster nodes.

Queries `conduit_workers` for nodes with status:'running'.
Useful for dashboards and monitoring.
fnenqueueenqueue(_job, job_type : String, payload : String, storage : s) : Result(String, String)#

Enqueue a job for immediate execution. Returns the job id on success.

`_job`     — the job module (used only to fetch config; conventionally passed
             as the module value, e.g. `SendWelcomeEmail`)
`job_type` — string name of the job module ("SendWelcomeEmail")
`payload`  — serialised args string (Phase 1: use `show(args)`)
`storage`  — storage backend

Phase 2 will provide a typed wrapper:
  Conduit.enqueue(SendWelcomeEmail, args)
fnenqueue_allenqueue_all(items : List((String, String, JobConfig)), storage : s) : List(Result(String, String))#

Enqueue multiple jobs in one call. Items: list of (job_type, payload, config). Returns a list of results in input order. Not atomic — a failure in one job does not affect the others.

fnenqueue_inenqueue_in(_job, job_type : String, payload : String, delay : Int, storage : s) : Result(String, String)#

Enqueue a job to run after delay. Returns the job id on success.

fnenqueue_in_withenqueue_in_with(job_type : String, payload : String, delay : Int, config : JobConfig, storage : s) : Result(String, String)#

Enqueue a job to run after delay with explicit config.

fnenqueue_withenqueue_with(job_type : String, payload : String, config : JobConfig, storage : s) : Result(String, String)#

Enqueue a job with an explicit Conduit.JobConfig. Use this when you need to override queue, priority, max_attempts, etc.

fnis_cluster_leaderis_cluster_leader() : Bool#

Return true if this node is currently the cluster leader.

Fast in-process check — reads from Vault, no DB round-trip.
fnpause_cronpause_cron(storage : s, cron_id : String) : Result((), String)#

Disable a cron schedule by id.

fnpause_queuepause_queue(queue : String) : ()#

Pause all workers polling queue on this node.

fnregisterregister(job_type : String, performer : String -> Result((), Conduit.ConduitError), config : JobConfig) : ()#

Register a job performer and its config in the in-process registry. Must be called before any worker can execute jobs of that type.

`job_type`  — string name of the job module (e.g. "SendWelcomeEmail")
`performer` — fn (payload_string) -> Result((), ConduitError)
`config`    — the job's JobConfig (used for backoff, DLQ, callbacks at runtime)

Call this at application startup for each job type:

  Conduit.register("SendWelcomeEmail",
    fn payload ->
      let args = todo("deserialise payload into SendWelcomeEmail.Args")
      SendWelcomeEmail.perform(SendWelcomeEmail, args)
    end,
    SendWelcomeEmail.config(SendWelcomeEmail)
  )
fnregister_cronregister_cron(cron_type : String, config : CronConfig, performer : String -> Result((), Conduit.ConduitError)) : ()#

Register a static cron at startup.

Static crons are defined in code; their schedule is upserted into the DB
each time `Conduit.start` is called, so they survive node restarts and
config changes.

`cron_type` — unique string name (e.g. "HourlyCleanupJob")
`config`    — CronConfig with schedule, timezone, queue, overlap
`performer` — fn(payload_string) -> Result((), ConduitError)

Example:

  Conduit.register_cron("HourlyCleanupJob",
    HourlyCleanupJob.config(HourlyCleanupJob),
    fn _payload -> HourlyCleanupJob.perform(HourlyCleanupJob) end
  )
fnregister_defaultregister_default(job_type : String, performer : String -> Result((), Conduit.ConduitError)) : ()#

Register a job performer using default config. Convenience wrapper for jobs that don't override fn config.

fnregister_workflowregister_workflow(#

Register a workflow at startup.

`workflow_type` — unique string name (e.g. "OnboardUserWorkflow")
`runner`        — curried fn: input_json -> ctx -> Result(output_json, WorkflowError)
`config`        — WorkflowConfig (execution_mode, timeout, tags)

Example:

  Conduit.register_workflow(
    "OnboardUserWorkflow",
    fn input -> fn ctx -> OnboardUserWorkflow.run(OnboardUserWorkflow, input, ctx) end end,
    OnboardUserWorkflow.config(OnboardUserWorkflow)
  )
fnreplay_workflowreplay_workflow(workflow_id : String, storage : s) : Result(String, String)#

Replay a completed or failed workflow using its recorded event history.

Loads all events from the event store, sets up deterministic replay mode
in Vault, then re-runs the workflow function. Checkpoints whose events
are already recorded return the recorded value without calling the fn;
if the workflow has new checkpoints beyond the recorded events, those
execute normally and append new events.

Returns the workflow output or error.
fnreplay_workflow_toreplay_workflow_to(workflow_id : String, max_sequence : Int, storage : s) : Result(String, String)#

Replay a workflow up to a specific event sequence (point-in-time replay).

Only events with sequence <= max_sequence are loaded. The workflow
re-executes from the start: recorded events are replayed, and the first
checkpoint beyond max_sequence will execute its fn normally.

Useful for debugging: "show me what the workflow saw at step N".
fnresume_cronresume_cron(storage : s, cron_id : String) : Result((), String)#

Re-enable a paused cron schedule by id.

fnresume_queueresume_queue(queue : String) : ()#

Resume all workers polling queue on this node.

fnschedule_cronschedule_cron(storage : s, sched : Conduit.CronSchedule) : Result((), String)#

Insert or update a dynamic cron schedule in the DB at runtime.

Unlike static crons, dynamic crons are not registered in code; they are
created, modified, or removed via this API at runtime.

`storage` — storage backend
`sched`   — full CronSchedule row to upsert
fnsignal_workflowsignal_workflow(workflow_id : String, signal_type : String, payload_json : String, storage : s) : Result(Int, String)#

Send a signal to a running workflow.

The signal is stored in conduit_workflow_signals immediately. If the workflow
is blocked in wait_for_signal, it will pick it up on the next poll cycle.

`workflow_id`  — the workflow to signal
`signal_type`  — must match the signal type the workflow is waiting for
`payload_json` — JSON payload carried with the signal
`storage`      — storage backend
fnstartstart(config : Conduit.Config, storage : s) : List(Task(()))#

Start the Conduit worker pool, cron scheduler, and workflow runner.

`config`  — application config (queues, worker count, poll interval, cron tick)
`storage` — storage backend (e.g. Conduit.Storage.Postgres.Config)

Spawns `workers_per_queue` worker tasks per queue and one cron scheduler task.
Also registers the built-in `conduit.workflow_runner` performer so workers
can execute workflow jobs.

Static crons registered via `Conduit.register_cron` are upserted into the DB
during startup so they survive node restarts.

Returns a list of all task handles. In production, wrap this in a supervisor
for auto-restart.

Example:

  let storage = { repo: MyApp.Repo }
  Conduit.start(Conduit.Config.default(), storage)
fnstart_workflowstart_workflow(workflow_type : String, input_json : String, storage : s) : Result(WorkflowHandle, String)#

Start a workflow and return a handle immediately.

Inserts a row in conduit_workflows, enqueues a conduit.workflow_runner job,
and returns a WorkflowHandle with the workflow id.

`workflow_type` — must match a registered workflow
`input_json`    — JSON-serialized input
`storage`       — storage backend
fnstopstop(storage : s, config : Conduit.Config) : ()#

Gracefully stop this Conduit node.

  1. Sets the in-process drain flag so all workers stop fetching new jobs.
  2. Waits up to config.drain_timeout_ms for in-flight jobs to finish.
  3. Deregisters the node from conduit_workers and releases the leader
   advisory lock.

Call this during application shutdown before the process exits.  The drain
timeout is a best-effort window; jobs that are still running after it will
be reclaimed by another node's stale-rescue loop.
fnworkflow_eventsworkflow_events(workflow_id : String, storage : s) : Result(List(WorkflowEvent), String)#

Query the event history for a workflow. Returns all events in sequence order.

fnworkflow_statusworkflow_status(workflow_id : String, storage : s) : Result(Option(String), String)#

Query the current status of a workflow.

Returns the workflow status string (running | completed | failed | cancelled | timed_out)
or None if the workflow does not exist.