Queues and messaging

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

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

Execution A → publishes to "/queue/scraping.tasks"
Execution B → publishes to "/queue/scraping.tasks"  ← 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

Every point-to-point queue name is prefixed with /queue/<execution_id>/:

Execution A (id=019ce6d0...) → publishes to "/queue/019ce6d0.../scraping.tasks"
Execution B (id=02af71b3...) → publishes to "/queue/02af71b3.../scraping.tasks"

Each processor subscribes to its own execution-prefixed queue. No cross-talk. Subscription names get the same treatment, suffixed with the first 8 chars of the execution id (web-scraper → web-scraper:019ce6d0).

Implementation lives in factflow_engine.scoped_queue.ExecutionScopedQueue. It wraps a QueueProtocol instance; every publish/subscribe call rewrites the queue name (QUEUE_PREFIX = "/queue/") before forwarding.

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

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)

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

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

Regardless of provider, the ack semantics are uniform: the delivery is settled when the handler returns, and what the handler returns decides how. There is no manual ack, and no fire-and-forget.

Return	Meaning	What the provider does
ACKNOWLEDGED	The work is done	Settles the delivery
FAILED	Transient — a retry could plausibly succeed	Re-attempts it a bounded number of times with a delay between attempts, then dead-letters it
REJECTED	Permanent — this delivery will never succeed	Moves it to the dead-letter destination, where it stays inspectable

Every provider bounds the retry and dead-letters what is left, so a message that always fails cannot circulate forever. How the delay is shaped differs, because each broker offers a different primitive:

Provider	Attempt counter	Delay between attempts	Where it ends up
Artemis	x-delivery-attempt header, re-published per attempt	Exponential, scheduled broker-side via _AMQ_SCHED_DELIVERY	<queue>.dlq
RabbitMQ	Same header, re-published per attempt	Exponential, via a TTL on a <queue>.retry holding queue that dead-letters back to the source	The dead-letter exchange
Pulsar	Broker-side redelivery count	Fixed, negative_ack_redelivery_delay_ms	<topic>-dlq, via Pulsar's own deadLetterPolicy

The retry budget itself means the same thing everywhere — it is counted against the port's own header rather than a broker-native counter, because all three brokers count redelivery differently and a budget keyed on any one of them would behave differently per broker.

Topics do not retry. A broadcast delivery has no per-subscriber redelivery, so re-publishing it would either reach nobody (an anycast copy on a multicast address) or reach everybody — one subscriber's transient failure redelivering to every other subscriber. A failed topic delivery is therefore dead-lettered for the subscription that failed, bounded at one attempt, leaving other subscribers untouched. The failure stays visible without being retried.

PENDING, DELIVERED and EXPIRED are internal lifecycle states. Never return them; a provider that receives one treats it as an incomplete handler and redelivers rather than risk losing the message.

The distinction between the two failures matters because without it every failure means "redeliver", so a message that can never succeed circulates until something else stops it — and nothing does. REJECTED is what lets a hopeless message leave the queue without being discarded.

Cancellation mid-handler causes redelivery, which is why adapter process() methods must be cancel-safe — they will be cancelled during shutdown.

Why retry is not delegated to the broker

On ActiveMQ Artemis over STOMP, a NACK means "remove this message", not "redeliver it". So a NACK'd transient failure was destroyed, and the broker's redelivery-delay and max-delivery-attempts never applied — the message was already gone before policy could act.

Retry is therefore expressed by republishing with an incremented attempt header and a broker-scheduled delivery time, and the attempt limit is owned by Factflow rather than the broker. The same reason applies to the attempt counter itself: a broker's own delivery count is unreliable for a message that was never acknowledged, so the count travels with the message in an x-delivery-attempt header.

STOMP offers no other route to it. Acknowledging inside a transaction and then aborting the transaction — the usual idiom for reject-and-requeue — was measured against Artemis 2.44 and also destroys the message: the acknowledgement is applied, the abort does not roll it back, and the message is neither redelivered to the live consumer nor recoverable by a later one. Leaving a delivery unsettled does preserve it, but it only comes back once the consumer goes away, so retrying one message would mean tearing down the subscription and stalling every other message in flight on that route.

Because the republish is what preserves the message, it is confirmed before the original is acknowledged. A SEND carrying a receipt header is answered with a RECEIPT once the broker has taken the message; without that, a publish is only a local socket write, and a silently dead connection would swallow the copy while the acknowledgement destroyed the last one. No receipt means the original stays in its queue.

The committed backend/config/artemis/broker.xml still matters, but as a bound on broker-driven redelivery after a connection loss — not as the retry mechanism.

Settling only what the broker still holds

Artemis tracks an outstanding delivery per TCP connection, keyed on (consumer id, message id), and removes it only when that same connection acknowledges it. Acknowledging one it has reissued is answered with an ERROR frame and a closed connection — which voids every other in-flight acknowledgement, causing the redelivery that produces more doomed acknowledgements. Left alone, the loop sustains itself and a run never converges.

Three conditions can invalidate a delivery, and each is checked before settling:

1. The connection changed — the frame was dispatched by a connection that is gone.
2. The consumer was replaced — a route restart swaps the consumer without necessarily losing the connection, which the connection check cannot see.
3. The delivery was reissued — STOMP gives a consumer no way to learn this, since a new message-id is minted each time, so the correlation id is used as the message's stable identity.

A settlement withheld for any of these is reported as superseded, distinct from a lost one: the delivery now in flight is the one that settles the message.

For what an operator should read, see Troubleshooting → Durability counters.

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. There is no CLI wrapper — call it directly:

curl -X POST http://localhost:8000/api/v1/system/mq-diagnostic

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

Related

• factflow-infra reference — provider implementations
• factflow-engine reference — ExecutionScopedQueue
• Orchestration — how the orchestrator uses scoped queues
• Queue diagnostics guide
• Reference: Queue payloads
• Troubleshooting — error classification, dead-letter destinations, durability counters