Conduit.Queue

Conduit.Queue — job enqueue logic and the in-process performer registry.

This module has two responsibilities:

Enqueue: build a JobRow, apply the unique-constraint guard, and persist

   via the storage backend.  Both immediate and delayed variants are here.

2. Registry: maintain a Vault-backed in-process map of job_type → (fn, config)
   so that workers can dispatch and look up runtime config without hitting
   the DB.

Workers call dispatch to execute a job and lookup_config to read the registered JobConfig (used for backoff, timeout, DLQ routing, and callbacks).

Functions

fndispatchdispatch(job_type : String, payload : String)#

Invoke the registered performer for job_type with payload.

Returns the performer's result directly:
  Ok(())                       — job succeeded.
  Err(ConduitError.Retry(msg)) — transient failure; worker will reschedule.
  Err(ConduitError.Discard(msg))— permanent failure; worker will dead-letter.
  Err(ConduitError.Snooze(ms)) — postpone; worker will snooze the job.
  Err(ConduitError.WorkerError(msg)) — infrastructure error; treated as Retry.

If no performer is registered for `job_type`, returns
`Err(ConduitError.Discard("No performer registered …"))` so the worker
dead-letters the job instead of retrying forever.

fnenqueueenqueue(storage : s, job_type : String, payload : String, config : Conduit.JobConfig) : Result(String, String)#

Enqueue a job for immediate execution and return its id.

Applies the unique-constraint guard before inserting:
  • If `config.unique_for` is None → insert unconditionally.
  • If a conflicting active job exists → apply `config.on_conflict`:
      Ignore  → return Ok(existing_id).  NOTE: existing_id is the string
                "existing", not the real job UUID; treat it as opaque.
      Raise   → return Err("unique constraint violated …").
      Replace → continue with insert (the caller is responsible for the old job).

The new row is inserted with status: "pending", attempt: 0, and
run_at: NOW().

fnenqueue_allenqueue_all(storage : s, items : List((String, String, Conduit.JobConfig))) : List(Result(String, String))#

Enqueue multiple jobs in one call. Returns results in the same order as input.

Each item is `(job_type, payload, config)`.  Jobs are inserted individually
using the existing `enqueue` logic (unique-key guard, telemetry, etc.).  This
is NOT a single atomic transaction — a failure on one job does not affect
the others.

Use this for fan-out patterns where you need to enqueue many jobs of
potentially different types in a single pass:

  Conduit.Queue.enqueue_all(storage,
    Cons(("EmailJob", payload1, email_cfg),
    Cons(("AuditJob", payload2, audit_cfg), Nil)))

fnenqueue_inenqueue_in(storage : s, job_type : String, payload : String, delay : Int, config : Conduit.JobConfig) : Result(String, String)#

Enqueue a job to run after delay milliseconds and return its id.

Identical to `enqueue` except run_at is set to NOW() + delay.  The unique
constraint guard is applied using the same key that `enqueue` would compute, so
a delayed job and an immediate job for the same type+payload are considered the
same unique job when `unique_for` is set.

fnlookup_configlookup_config(job_type : String) : Option(Conduit.JobConfig)#

Return the registered JobConfig for job_type, or None if not registered.

Workers call this after claiming a job to get the runtime config (backoff,
timeout, DLQ, callbacks) without re-querying the DB.  If the job type is not
registered, the worker falls back to `JobConfig.default()`.

fnregister_performerregister_performer(job_type : String, performer, config : Conduit.JobConfig) : ()#

Must be called at application startup (before workers start) for each job type
that this process will execute.  The registry is Vault-backed and is local to
the current OS process; every node in a cluster must register all performers it
needs to handle.

Calling `register_performer` a second time with the same `job_type` silently
overwrites the previous entry — useful for testing but avoid in production.