Skip to content
Factflow

Executions and orchestration

An execution is one run of one pipeline config. It owns a UUID v7 id, a frozen copy of the config it started from, and an isolated slice of every shared resource — queues, lineage rows, storage keys, metrics. The server process runs many executions at once; they share one broker connection but never see each other's messages.

What an execution is

Section titled “What an execution is”

When you start a pipeline, the server records an execution row keyed by a client-generated UUID v7. Storage paths use that id directly — there are no separate job ids.

The row carries a config snapshot: the parsed config frozen as JSONB at start time (config_snapshot). The execution runs against the snapshot, not the live config directory. Editing or deleting the config afterwards does not change a running execution, and replay resolves routes from the snapshot of the parent execution — not the global config. That isolation is what makes cross-pipeline replay deterministic.

Replays are themselves executions: they get their own id and snapshot, and link back via parent_execution_id and replay_source.

The runtime: three layers

Section titled “The runtime: three layers”

The orchestrator is the runtime that reads an execution's config snapshot and drives adapters to completion. It lives in factflow-engine; the execution records are owned by factflow-execution; the whole thing runs inside factflow-server.

OrchestratorManager       (one per server process — owns every running execution)
  └── PipelineOrchestrator   (one per execution — owns the routes)
        └── ReactiveRouteProcessor  (one per route — owns the queue subscription)
              └── PipelineAdapter    (many per processor — invoked in sequence per message)

OrchestratorManager

Section titled “OrchestratorManager”

Single-process, app-lifecycle service. Holds PipelineOrchestrator instances keyed by execution id, capped at max_concurrent (default 10) — over the cap, start_execution raises TooManyConcurrentExecutionsError. It wires each execution with an ExecutionScopedQueue and a per-execution adapter registry, exposes per-execution and global controls (pause/resume/cancel a route in one execution, or across all of them), and aggregates metrics. Stopping the manager gracefully stops every orchestrator. Replay publishers run as manager-owned asyncio tasks tied to the execution lifecycle.

PipelineOrchestrator

Section titled “PipelineOrchestrator”

One per execution. Builds a ReactiveRouteProcessor for each route in the config snapshot, wires them to the scoped queue, publishes the init message (unless skipped for replay), and signals completion when every route drains.

ReactiveRouteProcessor

Section titled “ReactiveRouteProcessor”

One per route. Subscribes to the route's inbound queue and, for each message, invokes the route's adapters in order, respecting the route's concurrency (max in-flight) and prefetch (broker-side buffer). Backpressure is handled by a BackpressureController.

Message lifecycle

Section titled “Message lifecycle”

Each message runs the route's adapters in order: a lineage row is recorded pending, await adapter.process(context) runs, the row is updated to completed/failed, and the returned AdapterResult becomes the next adapter's input (None ends the chain). The message is acked only after the whole chain returns — there is no explicit ack call, so a processor cancelled mid-chain yields redelivery, and adapters must be cancel-safe.

See The pipeline spine for the per-phase detail — concurrency gating, fan-out child pre-registration, stateful fan-in — and the end-to-end sequence.

Completion detection

Section titled “Completion detection”

An execution is complete when two conditions hold at the same check: every route is idle (no in-flight messages, no pending fan-out) and lineage has drained (no pending / initiated / processing rows). The orchestrator re-evaluates this gate on a short completion_check_interval (default 5 s, min 0.5 s). The two-condition form — not just "queues empty" — is what makes fan-out safe; see the pending-children pre-registration in The pipeline spine.

Detection is interval-checked, but the ExecutionWaiter that API and SSE clients block on is purely reactive: once the gate passes, the orchestrator fires an asyncio.Event and any waiter wakes immediately — no client-side polling.

Circuit breaker + backpressure

Section titled “Circuit breaker + backpressure”
  • Circuit breaker — per-adapter. Repeated failures open the breaker; subsequent messages short-circuit (CircuitOpenError) until it half-opens. Prevents one flaky adapter from grinding through a queue only to fail everything.
  • Backpressure — per-route. Adaptive concurrency based on in-flight count and recent latency. Prevents a fast upstream from overwhelming a slow downstream.

Both are tuned per route: circuit-breaker thresholds live on the adapter, and backpressure watermarks derive from inbound.prefetch (high_watermark = prefetch * 10). Defaults are sane.

Discovery and registration

Section titled “Discovery and registration”

On server startup, AdapterDiscovery scans every installed workflow package (factflow_webscraper, factflow_markdown, factflow_boost, …) for classes that implement the PipelineAdapter protocol — no decorator, no manual registry edits; presence in a workflow module is the registration. Each class supplies its type: name, and ValidatingAdapterRegistry verifies that the declared config matches its Pydantic config class. A config that references an unknown type: fails validation with error code E103.

Multi-execution: one broker, no cross-talk

Section titled “Multi-execution: one broker, no cross-talk”

Many executions of the same config reuse the same route names from the same snapshot. Without isolation, execution A's messages would land in execution B's processors. They don't, because every orchestrator talks to the broker through an ExecutionScopedQueue that rewrites point-to-point queue names to include the execution id (/queue/scraping.tasks → /queue/<exec_id>/scraping.tasks). Topics pass through unchanged — broadcast is shared by design.

flowchart TB
subgraph One["One server process"]
  Mgr[OrchestratorManager]
  Orch1[PipelineOrchestrator exec=A]
  Orch2[PipelineOrchestrator exec=B]
  Orch3[PipelineOrchestrator exec=C]
  Mgr --> Orch1
  Mgr --> Orch2
  Mgr --> Orch3
end
Orch1 -.scoped to A.-> QA[(/queue/A/...)]
Orch1 -.A's rows.-> LA[(A's lineage)]
Orch2 -.scoped to B.-> QB[(/queue/B/...)]
Orch2 -.B's rows.-> LB[(B's lineage)]
Orch3 -.scoped to C.-> QC[(/queue/C/...)]
Orch3 -.C's rows.-> LC[(C's lineage)]
OrchestratorManager owns one orchestrator per execution. Each talks to the broker through its own ExecutionScopedQueue and tracks idle/completion for its own execution only — checks never cross execution boundaries.

The shared completion topic (/topic/system.pipeline.progress) is the one place executions overlap on the wire: every orchestrator publishes and subscribes to it, then filters incoming events on execution_id to ignore other executions' signals.

For the full prefixing scheme, provider-specific name formats, and why topics stay shared, see Queues and messaging — the canonical home for the isolation mechanics.

Isolation guarantees

Section titled “Isolation guarantees”
  • Failure — a processor crash in execution A does not touch execution B; the orchestrators are independent. Provider-level failures (broker restart, DB disconnect) affect all executions equally and are handled by provider reconnect logic.
  • Metrics — GET /api/v1/executions/{id}/stats scopes to one execution; GET /api/v1/system/metrics aggregates across all of them.
  • Resources — queue-layer interference is impossible; CPU, memory, and LLM budget contention is managed by OS scheduling plus each adapter's rate limiting.

Why reactive, not scheduled?

Section titled “Why reactive, not scheduled?”

Traditional workflow engines poll a scheduler: "which tasks are ready? run them." Factflow doesn't poll for task readiness — the broker pushes messages to subscribed processors, which consume-and-ack. Less latency, no central bottleneck, natural backpressure. The one interval check is completion detection above — bounded and cheap.

Section titled “Related”