Approval Gates

Pause a workflow run until a human or external system approves or rejects it.

An approval gate pauses a workflow run until something external — a human reviewer or another system — resolves it. No task runs at the gate: the run stops there until it's approved, rejected, or a timeout fires, and only then do its successors run or get skipped.

Adding a gate

python

from flexiq import Queue
from flexiq.workflows import Workflow

queue = Queue(db_path="tasks.db")

@queue.task()
def build_task() -> str: ...

@queue.task()
def deploy_task() -> None: ...

wf = Workflow(name="release_pipeline")
wf.step("build", build_task)
wf.gate(
    "approve_deploy",
    after="build",
    message="Review build artifacts before deploying to production.",
    timeout=86400,       # 24 hours
    on_timeout="reject",
)
wf.step("deploy", deploy_task, after="approve_deploy")

run = queue.submit_workflow(wf)

import { Queue } from "@byteveda/flexiq";

const queue = new Queue({ dbPath: "tasks.db" });

const handle = queue.workflows
  .define("release-pipeline")
  .step("build", "buildTask")
  .gate("approve-deploy", {
    after: "build",
    message: "Review build artifacts before deploying to production.",
    timeoutMs: 24 * 60 * 60 * 1000, // 24 h
    onTimeout: "reject",
  })
  .step("deploy", "deployTask", { after: "approve-deploy" })
  .submit();

queue.runWorker();

java

Workflow release = Workflow.named("release-pipeline")
        .step("build", buildTask, artifact)
        .gate("approve-deploy",
                GateConfig.timeout(
                        Duration.ofHours(24),
                        GateAction.REJECT,
                        "Review build artifacts before deploying to production."),
                "build")
        .step("deploy", deployTask, artifact, "approve-deploy");

WorkflowRun run = queue.submitWorkflow(release);

try (Worker worker = queue.worker()
        .handle(buildTask, p -> p)
        .handle(deployTask, p -> p)
        .trackWorkflows(release) // the tracker holds deploy's payload
        .start()) {
    // ...
}

There is no gate in Rust. Workflows are the opt-in flexiq-workflows crate (flexiq = { version = "2.0.0", features = ["workflows"] }), and a gate there is one field — an opaque string that submission stores and never reads:

rust

use flexiq::workflows::StepMetadata;

StepMetadata {
    task_name: "approve_deploy".into(),
    // Recorded verbatim. No Rust code parses this, times it out, or waits on it.
    gate: Some(r#"{"timeoutMs":86400000,"onTimeout":"reject"}"#.into()),
    ..Default::default()
};

Two consequences worth knowing before you reach for it. A gate node that is not listed in deferred_node_names is enqueued as an ordinary job under its task_name — it does not pause anything. And a deferred node is excluded from its successors' depends_on, so deferring only the gate lets deploy start immediately. A pause you can rely on means deferring the gate and everything downstream of it, then enqueueing those steps yourself once you decide the gate has been resolved.

A gated workflow must be registered on the tracking worker with trackWorkflows(workflow). The gate's downstream steps are deferred nodes — their jobs are only created when the gate resolves, and the tracker supplies their payloads from the registered definition.

The deferral set is what does the pausing, so it goes in the submission:

rust

use std::collections::HashSet;

SubmitStaticWorkflowRequest {
    // The gate AND everything downstream of it. A node left out of this set
    // gets a job at submission time and runs regardless.
    deferred_node_names: HashSet::from([
        "approve_deploy".to_string(),
        "deploy".to_string(),
    ]),
    ..request
};

A deferred node gets a WorkflowNode row at Pending and no job. It is also filtered out of its successors' depends_on — the successor's predecessor has no job id to point at — which is exactly why deferring the gate alone does not hold deploy back.

So the payloads have to survive until you enqueue those steps. StepMetadata has args_template and kwargs_template for that: base64 of the serialized arguments, stored on the definition, so a node whose job is created later can be built without you keeping anything in memory.

Gate node status

While a gate is pending, its node status is waiting-for-approval; the run itself keeps running. Downstream steps don't start until the gate resolves.

python

run.node_status("approve_deploy")  # NodeStatus.WAITING_APPROVAL
run.status().state                 # WorkflowState.RUNNING — the run keeps going

const nodes = handle.nodes();
const gate = nodes.find(n => n.nodeName === "approve-deploy");
console.log(gate?.status); // "waiting_approval"

java

WorkflowStatus status = run.status().orElseThrow();
status.node("approve-deploy").orElseThrow().status; // WAITING_APPROVAL

rust

use flexiq::workflows::{WorkflowNodeStatus, WorkflowStorage};

let node = wf.get_workflow_node(run_id, "approve_deploy")?.expect("gate node");
assert_eq!(node.status, WorkflowNodeStatus::WaitingApproval);

The WaitingApproval variant exists and serialises as "waiting_approval", which is what makes a Rust-written row readable everywhere else. Nothing in Rust ever puts a node into it, though — you set it yourself with update_workflow_node_status when your own code decides the run has reached the gate.

Resolving a gate

Resolve a gate with the run ID and the gate's node name — approve to let its successors proceed, or reject (optionally with a reason) to fail the gate and skip them.

python

# Call this on the same `queue` instance that is running the worker for
# this run — see the note below.
queue.approve_gate(run.id, "approve_deploy")

# Reject, with an optional reason recorded as the gate's error.
queue.reject_gate(run.id, "approve_deploy", error="Artifacts failed QA review.")

// Approve — downstream steps are enqueued.
queue.workflows.approveGate(runId, "approve-deploy");

// Reject — downstream steps are skipped, run transitions to "failed".
queue.workflows.rejectGate(runId, "approve-deploy");

// Reject with a reason recorded in storage.
queue.workflows.rejectGate(runId, "approve-deploy", "Artifacts failed QA review.");

// Unified helper — pass a boolean.
queue.workflows.resolveGate(runId, "approve-deploy", true);   // approve
queue.workflows.resolveGate(runId, "approve-deploy", false, "Reason."); // reject

java

// Approve — the gate completes and downstream steps are enqueued.
worker.approveGate(run.runId(), "approve-deploy");

// Reject with a reason — downstream steps are skipped, run transitions to FAILED.
worker.rejectGate(run.runId(), "approve-deploy", "Artifacts failed QA review.");

There is no approve or reject call. WorkflowStorage is the entire surface, and resolving means writing the node row you want:

rust

use flexiq::now_millis;
use flexiq::workflows::WorkflowStorage;

// Approve — one write; this setter also moves the status to `Completed`.
wf.set_workflow_node_completed(run_id, "approve_deploy", now_millis(), None)?;

// Reject — records the reason and moves the status to `Failed`.
wf.set_workflow_node_error(run_id, "approve_deploy", "Artifacts failed QA review.")?;

Both writes are fenced to the handle's namespace and fail if the node does not exist, so a typo is an error rather than a silent no-op. Nothing reacts to them: enqueueing the steps the gate was holding back, or marking them skipped, is the same job as writing the row.

Resolve from the process running the worker

approve_gate/reject_gate only advance the run when called on the same Queue instance that is running the worker for it — the tracker keeps each run's plan and deferred-node payloads in memory, not in storage. Calling either method on an unrelated Queue() still flips the gate's stored status, but its successors won't be created until a call reaches the tracker that owns the run. If you need to resolve gates from application code (e.g. a request handler), run the worker in a background thread in that same process — see Building Workflows.

Resolution works from any process that opens the same storage — the resolver does not need to be the worker process.

Resolution goes through the Worker that is tracking workflows — calling either method on a worker built without trackWorkflows() throws a WorkflowException.

Resolving is a storage write, so it works from any process that opens the same WorkflowStorageBackend — there is no tracker to be co-located with, because there is no tracker. The other half of that trade is that the write is all that happens. Approving looks like this end to end:

rust

// 1. The gate itself.
wf.set_workflow_node_completed(run_id, "approve_deploy", now_millis(), None)?;

// 2. The step it was holding back — deferred, so it has a row and no job.
let job = queue.enqueue(deploy::call(artifact))?;
wf.set_workflow_node_job(run_id, "deploy", &job.id)?;

set_workflow_node_job is the setter for exactly this case: a node whose job is created after the run was submitted. Skip it and the row never learns which job it became, so nothing reading the run can tell you.

Rejecting is the mirror image — set_workflow_node_error on the gate, then update_workflow_node_status(run_id, "deploy", WorkflowNodeStatus::Skipped) for each step you are not going to enqueue. Nothing walks the DAG marking them for you.

Every one of these setters answers WorkflowError::NodeNotFound when it matches no row, so a mistyped node name is an error rather than a silent no-op. None of them check the node's current status, though: resolving an already-resolved gate overwrites it. Guard that yourself if the endpoint is reachable twice.

Timeout behaviour

A gate with no timeout waits forever:

python

wf.gate("approve_deploy", after="build")  # timeout=None (default) — waits forever

.gate("approve-deploy", { after: "build" }) // no timeoutMs — waits forever

java

GateConfig.manual();                                          // waits forever
GateConfig.timeout(Duration.ofHours(24), GateAction.REJECT);  // auto-reject after 24 h
GateConfig.timeout(Duration.ofHours(24), GateAction.APPROVE, "Auto-ships unless stopped.");

timeoutMs and onTimeout are two keys inside the opaque gate string, so neither of these StepMetadata fields changes what happens — both are stored text:

rust

gate: Some(r#"{"timeoutMs":86400000,"onTimeout":"approve"}"#.into()),
gate: Some("{}".into()), // no timeout — nothing was counting either way

No timer ships in Rust. If a gate should auto-resolve, run the clock yourself and call set_workflow_node_completed or set_workflow_node_error when it fires — the same two writes a manual resolution makes.

Once the timeout elapses, the gate auto-resolves:

`on_timeout`	Effect
`"reject"` (default)	Same as calling `reject_gate` — downstream skipped, run `FAILED`
`"approve"`	Same as calling `approve_gate` — downstream proceeds

`onTimeout`	Effect
`"reject"` (default)	Same as calling `rejectGate` — downstream skipped, run `"failed"`
`"approve"`	Same as calling `approveGate` — downstream proceeds

Setting timeoutMs without onTimeout defaults to "reject".

`GateAction`	Effect
`REJECT` (default)	Same as `rejectGate` — downstream skipped, run `FAILED`
`APPROVE`	Same as `approveGate` — downstream proceeds

The timeout must be positive; the timer is driven by the tracking worker and is cancelled if the gate is resolved manually first.

`onTimeout` in the `gate` string	Effect
`"reject"`	None. Nothing reads the key.
`"approve"`	None. Nothing reads the key.
omitted	None — and no default is applied, because no code path has one.

The value is worth writing anyway: it travels in the definition, so a reader in any other runtime sees the intent you recorded. It just is not a promise this crate keeps.

Driving it yourself is a scan, not a timer per gate — a single periodic job that looks for gates past their deadline scales better than a task sleeping on each one:

rust

for node in wf.get_workflow_nodes(run_id)? {
    if node.status != WorkflowNodeStatus::WaitingApproval {
        continue;
    }
    // `started_at` is when your own code parked the node; a gate that was
    // never started has no deadline to be past.
    let Some(started) = node.started_at else { continue };
    if now_millis() - started > timeout_ms {
        wf.set_workflow_node_error(run_id, &node.node_name, "gate timed out")?;
    }
}

Because the deadline is yours, so is the race: a gate resolved by hand one tick before the sweep reaches it will be overwritten, since neither setter checks the status it is replacing.

Gate options

Parameter	Type	Default	Description
`after`	`str \| list[str]`	`None`	Predecessor step name(s)
`timeout`	`float \| None`	`None`	Seconds until auto-resolve; `None` waits forever
`on_timeout`	`str`	`"reject"`	Resolution on timeout: `"approve"` or `"reject"`
`message`	`str \| None`	`None`	Human-readable message included in the `WORKFLOW_GATE_REACHED` event payload

Option	Type	Default	Description
`after`	`string \| string[]`	—	Predecessor step name(s)
`message`	`string`	—	Human-readable description shown in the dashboard
`timeoutMs`	`number`	—	Auto-resolve after this many milliseconds
`onTimeout`	`"approve" \| "reject"`	`"reject"`	Resolution on timeout

`GateConfig` component	Type	Default	Description
`timeout`	`Duration`	`null` (wait forever)	Auto-resolve after this long
`onTimeout`	`GateAction`	`REJECT`	Resolution taken when the timeout elapses
`message`	`String`	`null`	Human-readable description shown to the approver

A gate is not a struct with options. It is one StepMetadata field holding a JSON string, plus whatever you do around it:

`StepMetadata` field	Type	What it does here
`gate`	`Option<String>`	Marks the node as a gate for anything that reads the definition. Never parsed by this crate.
`task_name`	`String`	Required even on a gate. If the node is not deferred, this is the task that gets enqueued.
`args_template` / `kwargs_template`	`Option<String>`	Base64 of serialized args, so a deferred node's job can be built later.
`condition`	`Option<String>`	`on_success`, `on_failure` or `always`. Stored, not evaluated — see below.

And the keys conventionally inside the gate string:

Key	Read by
`timeoutMs`	Your own sweep
`onTimeout`	Your own sweep
`message`	Whatever renders the approval

SubmitStaticWorkflowRequest is where the rest lives — deferred_node_names for the pause, default_timeout_ms, default_priority and default_max_retries for any node whose StepMetadata names none, and queue_default for the queue.

Gate conditions

Gates respect the same condition DSL as regular steps, evaluated against the gate's predecessors before it's entered:

python

wf.step("test", run_tests)
wf.gate("approve", after="test", condition="on_success")
wf.step("deploy", deploy_task, after="approve")

If test fails, the gate is skipped (its condition isn't met) — and deploy, the gate's successor, is skipped too.

Node: gate conditions are Python-only

A Node gate node has no entry condition — .gate() options cover only the predecessor(s), timeoutMs, onTimeout, and message. To make a gate conditional, put the condition on the step(s) around it instead.

Java: gate conditions are Python-only

A Java gate node has no entry condition — GateConfig covers only timeout, onTimeout, and message. To make a gate conditional, put the condition on the step(s) around it instead.

StepMetadata::condition takes on_success, on_failure or always, and no code in the crate reads it — not for a gate node and not for an ordinary one. get_ready_workflow_nodes answers a narrower question: which nodes are Pending with every DAG predecessor Completed. A failed predecessor makes a node not ready; it never makes it skipped.

So a conditional gate is a branch in whatever parks it:

rust

let test = wf.get_workflow_node(run_id, "test")?.expect("test node");

match test.status {
    WorkflowNodeStatus::Completed => {
        wf.update_workflow_node_status(run_id, "approve", WorkflowNodeStatus::WaitingApproval)?;
    }
    // The gate's condition did not hold, so nobody should be asked.
    _ => {
        wf.update_workflow_node_status(run_id, "approve", WorkflowNodeStatus::Skipped)?;
        wf.update_workflow_node_status(run_id, "deploy", WorkflowNodeStatus::Skipped)?;
    }
}

Note the second Skipped. Successors are not skipped transitively — they are deferred nodes with no job, so leaving them alone leaves them Pending forever, which reads as "still waiting" rather than "never happening".

Rejection behaviour

Rejecting a gate marks the gate node failed and skips its default (on-success) successors; the run ends in a failed state. Any on-failure or always successors still run — use those (see conditions) for a rejection branch.

Calling approveGate or rejectGate on a gate that has already been resolved (approved, rejected, or timed out) is a no-op — the tracker ignores stale resolutions.

Python has no equivalent guard: calling approve_gate/reject_gate again on an already-resolved gate re-runs the resolution instead of no-op'ing.

Marking the gate node Failed does not end the run. The run row has its own state column and nothing moves it, so a rejection is three writes, not one:

rust

use flexiq::workflows::WorkflowState;

wf.set_workflow_node_error(run_id, "approve_deploy", "Artifacts failed QA review.")?;
wf.update_workflow_node_status(run_id, "deploy", WorkflowNodeStatus::Skipped)?;

wf.update_workflow_run_state(run_id, WorkflowState::Failed, Some("gate rejected"))?;
wf.set_workflow_run_completed(run_id, now_millis())?;

The two run-level writes are deliberately separate: set_workflow_run_completed stamps completed_at only, because which terminal state accompanies it depends on the nodes.

update_workflow_run_state does not validate the transition — it writes what you give it. WorkflowState::can_transition_to is the check, and calling it first is on you:

rust

let run = wf.get_workflow_run(run_id)?.expect("run");
if !run.state.can_transition_to(WorkflowState::Failed) {
    return Ok(()); // already terminal — a second rejection has nothing to do
}

That is also the guard against a stale resolution. There is no tracker to ignore one for you.

An on-failure or always branch is not available either: condition is stored and never evaluated, so a rejection branch is steps you enqueue in the else arm of your own handler.

Events

When a gate enters WAITING_APPROVAL, a WORKFLOW_GATE_REACHED event fires in the process that reached it — the same process whose tracker owns the run (see the note under "Resolving a gate"). Subscribe with queue.on_event() (see the events reference for the full list of event types):

python

from flexiq.events import EventType

def notify_team(event_type, payload):
    send_slack(f"Workflow {payload['run_id']} needs approval at {payload['node_name']}")

queue.on_event(EventType.WORKFLOW_GATE_REACHED, notify_team)

The payload carries run_id, node_name, and message (the gate's message, or None if it wasn't set).

Node: no gate events

There is no gate/workflow event in Node — queue.on(...) delivers only job-outcome events (job.completed, job.retrying, job.dead, job.cancelled). To react to a gate being reached, notify from the resolver side (the endpoint that calls approveGate / rejectGate), or poll node status for waiting_approval (see "Resolving a gate").

Java: no gate events

There is no gate/workflow event in Java — the worker's event listeners deliver only job outcomes (SUCCESS, RETRY, DEAD, CANCELLED). To react to a gate being reached, notify from the resolver side, or poll node status for WAITING_APPROVAL (see "Resolving a gate").

There is no event surface at all — no gate event, and no job-outcome event either. WorkerBuilder exposes no hook, and the nearest seam, flexiq_core::Worker::on_outcome, is not reachable through it.

That costs less here than it might, because nothing reaches a gate on its own. The code that parks the node is yours, so the notification goes on the same line:

rust

wf.update_workflow_node_status(run_id, "approve_deploy", WorkflowNodeStatus::WaitingApproval)?;
notify_reviewers(run_id, "approve_deploy", gate_message)?;  // your own transport

Send after the write, not before, so a failed write cannot produce a request for approval of something nobody is waiting on.

A separate process that has to find gates instead of being told about them polls for them — get_workflow_nodes(run_id) filtered on WaitingApproval, or list_workflow_runs narrowed to WorkflowState::Running first if you do not already have a run id.

Webhooks and external triggers

A common pattern is an HTTP endpoint that receives a webhook and resolves the gate — any external system can drive a gate this way:

python

@app.post("/hooks/deploy-approval")
def deploy_approval():
    # Fail closed — resolving a gate is a privileged state transition.
    verify_webhook_signature(request)      # abort(401) on a bad signature
    body = request.get_json()
    authorize_run(request, body["run_id"])  # abort(403) unless the caller may resolve this run
    if body["approved"]:
        queue.approve_gate(body["run_id"], "approve_deploy")
    else:
        queue.reject_gate(body["run_id"], "approve_deploy", error=body.get("reason"))
    return "", 204

app.post("/hooks/deploy-approval", async (req, res) => {
  // Fail closed — resolving a gate is a privileged state transition.
  verifyWebhookSignature(req);       // throw 401 on a bad signature
  const { runId, approved, reason } = req.body;
  await authorizeRun(req, runId);    // throw 403 unless the caller may resolve this run
  queue.workflows.resolveGate(runId, "approve-deploy", approved, reason);
  res.sendStatus(204);
});

java

// Fail closed — resolving a gate is a privileged state transition.
void onDeployApproval(HttpRequest request) {
    verifyWebhookSignature(request);        // 401 on a bad signature
    ApprovalPayload payload = parse(request);
    authorizeRun(request, payload.runId()); // 403 unless the caller may resolve this run
    if (payload.approved()) {
        worker.approveGate(payload.runId(), "approve-deploy");
    } else {
        worker.rejectGate(payload.runId(), "approve-deploy", payload.reason());
    }
}

The crate serves no HTTP, so the endpoint is your framework's. Give the handler a WorkflowStorageBackend — it is Clone, so one per process is enough — and resolve with the two writes above:

rust

use flexiq::now_millis;
use flexiq::workflows::{WorkflowStorage, WorkflowStorageBackend};

// `Request`, `Approval` and the two guards are your application's.
fn on_deploy_approval(wf: &WorkflowStorageBackend, request: &Request) -> Result<(), ApiError> {
    // Fail closed — resolving a gate is a privileged state transition.
    verify_webhook_signature(request)?;         // 401 on a bad signature
    let approval: Approval = request.json()?;
    authorize_run(request, &approval.run_id)?;  // 403 unless the caller may resolve this run

    if approval.approved {
        wf.set_workflow_node_completed(&approval.run_id, "approve_deploy", now_millis(), None)?;
    } else {
        let reason = approval.reason.as_deref().unwrap_or("rejected");
        wf.set_workflow_node_error(&approval.run_id, "approve_deploy", reason)?;
    }
    Ok(())
}

Any process that opens the same storage can do this — there is no tracker to be co-located with. What that process does not get for free is the run moving on afterwards; enqueue the downstream steps here too.

This endpoint must run in the same process as the worker for the constraint described under "Resolving a gate" — queue here is the same Queue instance passed to run_worker().

Coming from BullMQ? FlowProducer has no pause/resume primitive — a human-in-the-loop approval step is something you'd build yourself (e.g. a job that polls a database flag). flexiq's .gate() is a first-class step kind: the run genuinely pauses, and approveGate/rejectGate resolve it from any process.