The pipeline spine

The spine is the load-bearing path every pipeline walks: a config is frozen into an execution, its routes subscribe to queues, adapters transform each message and write to storage, lineage records every hop, the execution completes when the spine drains, and replay lets you re-run any segment. Every arrow in the diagrams below is a real await point in the code.

This page walks the spine in order. The big-picture sequence comes first; then it zooms into the lifecycle of one message through one route — the most common case — before closing on completion, failure, and replay.

Config → execution: the start-up sequence

A run begins when you submit a config. ExecutionService.create freezes the config into a config_snapshot and OrchestratorManager.start spins up a PipelineOrchestrator.

Expand

sequenceDiagram
actor Op as Operator
participant CLI as factflow CLI
participant SRV as factflow-server
participant ES as ExecutionService
participant DB as PostgreSQL
participant Mgr as OrchestratorManager
participant Orch as PipelineOrchestrator
participant Proc as ReactiveRouteProcessor
participant Q as ExecutionScopedQueue
Op->>CLI: factflow config run CFG_ID
CLI->>SRV: POST /api/v1/executions
SRV->>ES: create(config_id)
ES->>DB: INSERT pipeline_execution (status=running, config_snapshot)
ES->>Mgr: start(execution)
Mgr->>Orch: new PipelineOrchestrator(snapshot)
Orch->>Proc: for each route in snapshot.routes
Proc->>Q: subscribe(exec:EXEC_ID:route.in)
Proc-->>Orch: processor ready
Orch->>Orch: wait all processors ready
Orch->>Q: publish init_message to first route
SRV-->>CLI: 201 execution_id + status=running

From `factflow config run` to first message published. Processors must subscribe BEFORE init_message publishes — messages published to unsubscribed queues are lost.

Invariants in this phase:

• config_snapshot is frozen. The row in PostgreSQL captures the full YAML at this moment. Subsequent config edits don't affect this execution — and replay later resolves routes from this snapshot, not the live config directory.
• Processors subscribe before init. Ordering matters: publish to a queue nobody is subscribed to yet and the message is dropped (topics) or buffered briefly (queues).
• Manager caps concurrency. OrchestratorManager enforces max_concurrent (default 10). Over-limit raises TooManyConcurrentExecutionsError. See Executions for how many spines run side by side.

The route: one processor, N adapters, M outbound queues

Once routes are subscribed and the init message is out, messages flow until every route drains. A route has one inbound queue, N adapters in sequence, and zero-or-more outbound queues, all driven by a single ReactiveRouteProcessor:

Expand

flowchart LR
Q["inbound<br/>/queue/route.in"] --> P[ReactiveRouteProcessor]
P --> A1[Adapter 1]
A1 --> A2[Adapter 2]
A2 --> A3[Adapter 3]
A3 --> OUT1["outbound<br/>/queue/next-route.in"]
A3 --> OUT2["outbound<br/>/queue/another-route.in"]
P -.reads from.-> Q
P -.writes to.-> OUT1
P -.writes to.-> OUT2

One processor per route. Adapters run in order on every message. A single adapter can emit multiple messages (fan-out).

See Adapters & routes for the contract a route declares and Queues & messaging for the scoped-queue mechanics behind every subscribe and publish.

Queue → processor: the inbound phase

When a message lands in the processor's inbound queue, three layers of concurrency gating are acquired in fixed order before any work begins:

Expand

sequenceDiagram
participant Broker
participant Proc as Processor
participant Sem1 as Global semaphore
participant Sem2 as Route semaphore
participant BP as Backpressure
Broker->>Proc: deliver message
Proc->>Sem1: acquire global slot
Sem1-->>Proc: granted
Proc->>Sem2: acquire route slot
Sem2-->>Proc: granted
Proc->>BP: check high-water mark
BP-->>Proc: OK / pause if saturated
Proc->>Proc: build PipelineContext (correlation_id, ...)

Three layers of concurrency gating, acquired global → route → backpressure. Global slot is per-execution, route slot is per-route, backpressure is the fan-out watermark.

Layer	Scope	Default
Global	All routes in one execution (max_handlers)	30
Per-route	One route (InboundConfig.concurrency)	1
Fan-out	Per-route publishing (max_fanout_pending)	500

Practical in-flight parallelism per route: min(prefetch, concurrency).

Adapter → storage: the work phase

Adapters run in the order declared in YAML. Each receives the previous adapter's AdapterResult.data as its next PipelineContext.message, and may write artefacts to storage along the way:

Expand

flowchart LR
CTX0[PipelineContext] --> A1[adapter 1 process]
A1 -->|AdapterResult.data| CTX1[PipelineContext]
CTX1 --> A2[adapter 2 process]
A2 -->|AdapterResult.data| CTX2[PipelineContext]
CTX2 --> A3[adapter 3 process]
A3 --> OUT[AdapterResult final]

Each adapter's AdapterResult.data becomes the next adapter's input. Returning data=None terminates the chain (no outbound publish).

For every adapter invocation:

1. Lineage row created. status=pending, with a hash of the input.
2. await adapter.process(ctx) — the adapter does its work, optionally writing artefacts via StorageProtocol.
3. Lineage row updated. status=completed or failed with an exception payload.
4. The result becomes the input to the next adapter.

If any adapter raises, the chain stops, lineage marks the failing row, and the processor returns FAILED. See Storage model for the sidecar pattern adapters write into and Adapters & routes for the process() contract.

Lineage: recorded on every hop

Lineage is woven through every step above, not bolted on afterward. Two properties make it reliable under failure:

• Handler-return ack. There is no explicit ack()/nack() on QueueProtocol. The processor returns MessageStatus.ACKNOWLEDGED or FAILED; the queue provider acts on that.
• Lineage commits independently. Lineage writes happen on a separate connection. A lineage write failure does not fail the pipeline; a pipeline failure still records lineage.

Lineage covers what those rows look like and how to query them; Lineage debugging covers reading a failed run forensically.

Fan-out: pre-register children before publish

An adapter can emit multiple outbound messages via AdapterResult.emit_messages: list[...]. The completion gate depends on this happening in a specific order:

Expand

sequenceDiagram
participant A as Adapter
participant Lin as Lineage
participant Proc as Processor
participant Sem as Fanout semaphore
participant Q as Queue
A->>Lin: record_pending_children(count=N)
Note over Lin: CRITICAL: before publish
A->>Proc: AdapterResult(emit_messages=[m1, m2, ..., mN])
loop N messages
  Proc->>Sem: acquire fanout slot
  Sem-->>Proc: granted
  Proc->>Q: publish message
  Proc->>Sem: release (synchronous)
end
Proc-->>Broker: return ACKNOWLEDGED (parent done)

Fan-out: lineage child count pre-registered before any publish. The pre-registration is load-bearing — without it, the completion-detection gate races.

Why pre-register:

• The completion check asks "is pending + initiated + processing == 0?"
• Without pre-registration, the parent finishes → count drops to zero → completion fires.
• Then the children publish → count goes up again, but the orchestrator has already stopped.

With pre-registration, the parent's "done" is contingent on all N children being accounted for.

Outbound: handler-return acknowledgement

After all adapters succeed and fan-out publishes complete, the processor acks the original message. There is no explicit ack call. The handler returns a MessageStatus and the queue provider interprets it:

Expand

flowchart TB
A[Handler returns<br/>MessageStatus] --> B{Which?}
B -->|ACKNOWLEDGED| C[Broker removes from queue]
B -->|FAILED| D[Broker redelivers or DLQs]
C --> E[Release semaphores]
D --> E
E --> F[Processor ready for next]

Key consequence: if the processor is cancelled mid-chain (server shutdown, timeout), the message is redelivered by the broker. Adapter process() methods must be cancel-safe.

Stateful adapters (fan-in)

The above assumes a stateless adapter — every message processed independently. For adapters that batch across messages (StatefulAdapter), the shape shifts:

Expand

flowchart LR
M1[msg 1] --> A[StatefulAdapter]
M2[msg 2] --> A
M3[msg 3] --> A
A -->|batch not ready| K[continue_pipeline=False<br/>ACK messages, no emit]
A -->|batch ready| E[emit aggregated result]
T[Timeout trigger<br/>every 5s] --> A

StatefulAdapter accumulates across messages; emits only when the batch is full or the background checker fires on timeout.

Use case: a batch_embedding_generator that collects 100 segments before one LLM batch call.

Execution: completion detection

An execution is complete when two conditions are simultaneously true at the same polling moment:

Expand

flowchart LR
A["All routes idle<br/>(in_flight=0, fanout_pending=0)"] --> C{Both<br/>true?}
B["Lineage empty<br/>(pending + initiated + processing = 0)"] --> C
C -->|yes| D[Orchestrator.stop]
C -->|no| A
D --> E[ExecutionService.mark_completed]
E --> F[(PostgreSQL)]
E --> G[SSE event to clients]

Two-condition completion gate. Both must hold at the same moment — hence the pending-children pre-registration.

On completion the orchestrator runs a strict shutdown sequence:

1. Stop subscriptions first. Frees subscription names so a follow-up execution can reuse them without ActiveMQ warnings.
2. Then the DB callback. ExecutionService.mark_completed updates the row.
3. Then the SSE event. Any client on GET /executions/{id}/events receives status=completed.

The full happy path

Expand

flowchart LR
A[Op runs CLI] --> B[POST /executions]
B --> C[ExecutionService.create]
C --> D[Orchestrator.start]
D --> E[Processors subscribe]
E --> F[Publish init]
F --> G[Message loop]
G -->|all routes drain| H[Completion detection]
H --> I[Orchestrator.stop]
I --> J[Mark row completed]
J --> K[SSE event]
K --> L[CLI exits 0]

Failure paths

Three kinds of failure, three different flows:

Failure	What happens	Where to look
Adapter reports a transient failure	Message re-enqueued with backoff, up to its attempt limit, then dead-lettered	Lineage debugging
Adapter reports a permanent failure	Message moved to the dead-letter destination, lineage row → failed	Dead-letter inspection
Adapter sets metadata["fatal"]=True	Orchestrator aborts the execution, row → failed	lineage failure row + exec.error
Processor cancelled mid-message	Terminal DeliveryAbandoned lineage row; broker redelivers	lineage failure row

Transient versus permanent is the adapter's call, carried on retryable — as the field on AdapterResult or as metadata["retryable"]. fatal wins over both: redelivery cannot fix a bad credential. See Troubleshooting → Error classification for the full rules.

Replay: re-running a segment of the spine

Because the config_snapshot is frozen and lineage records every hop, any segment of a completed spine can be re-run. Replay resolves routes from the parent execution's snapshot — never the live config directory — so the same byte-for-byte topology is reconstructed. See Replay for from-storage and cross-pipeline replay flows.

Related

• Executions — multi-execution topology and operator-facing controls
• Adapters & routes — the route contract and process() shape
• Queues & messaging — the scoped-queue mechanics behind every publish
• Storage model — the sidecar pattern adapters write into
• Lineage — what lineage rows look like and how to query them
• Write an adapter — how to implement your own