Skip to content
Factflow

Queue isolation

Factflow runs multiple pipeline executions concurrently in one server process. Every execution publishes and subscribes to queues named after routes from the same shared pipeline configs. Without isolation, messages from execution A would land in execution B’s processors.

ExecutionScopedQueue prevents that.

The problem

Section titled “The problem”

Two executions, same config, same route names. The naive mapping:

Execution A → publishes to "web_scraper_queue"
Execution B → publishes to "web_scraper_queue"  ← SAME QUEUE

Processors subscribe to the route name. A’s messages get consumed by B’s processors and vice versa. Data goes to the wrong storage keys, lineage records point at the wrong execution, chaos.

The solution

Section titled “The solution”

Every queue name is prefixed with the execution id:

Execution A (id=abc123) → publishes to "exec:abc123:web_scraper_queue"
Execution B (id=def456) → publishes to "exec:def456:web_scraper_queue"

Each processor subscribes to its own execution-prefixed queue. No cross-talk.

Implementation lives in factflow_execution.scoped_queue.ExecutionScopedQueue. It wraps a QueueProtocol instance; every publish/subscribe/ack call rewrites the queue name before forwarding.

Topics are intentionally shared

Section titled “Topics are intentionally shared”

Point-to-point queues (one consumer per message) are execution-scoped. Topics (broadcast — every subscriber gets a copy) are deliberately not scoped. Cross-execution signalling (e.g., “config directory reloaded”) uses topics.

Knowing which construct is which: the queue naming strategies (AmqpNamingStrategy, StompNamingStrategy, PulsarNamingStrategy) handle the provider-specific mapping of these abstract concepts to AMQP exchanges / STOMP destinations / Pulsar topics.

Provider-specific quirks

Section titled “Provider-specific quirks”

Artemis (STOMP)

Section titled “Artemis (STOMP)”

Default provider. Queue names are STOMP destinations (/queue/...). Subscription headers include timestamps — see the queue-provider-conventions.md rule for the exact format (strict ISO-8601 with timezone; a mismatch causes silent broker rejection).

RabbitMQ (AMQP 0-9-1)

Section titled “RabbitMQ (AMQP 0-9-1)”

Exchange + queue topology. Topics map to fanout exchanges. Queue names must match the AMQP naming rules ([a-zA-Z0-9_\-.:]+). AmqpNamingStrategy validates.

Pulsar

Section titled “Pulsar”

Topic-based; point-to-point is emulated with shared subscriptions. Pulsar has different guarantees around ordering — relevant only if your adapter cares about strict FIFO within a queue.

Handler-return ack

Section titled “Handler-return ack”

Regardless of provider, the ack semantics are uniform: the message is acked when the handler returns. No manual ack. No fire-and-forget. Cancellation mid-handler causes redelivery.

This is why adapter process() methods must be cancel-safe — they will be cancelled during shutdown.

Diagnostic endpoint

Section titled “Diagnostic endpoint”

POST /api/v1/system/mq-diagnostic runs a quick publish + subscribe round-trip against every configured queue to verify the broker connection is healthy. CLI wrapper:

Terminal window
factflow system mq-diagnostic

Expected output: one line per route confirming publish + consume succeeded.

Section titled “Related”