warren.runtime

Full pipeline composition: exchange, workers + completion criteria.

A pipeline uses a single exchange. Multi-exchange deployments (a worker publishing to several exchanges at once) are deferred — see warren/docs/routing.md.

One worker type: how it is wired into the pipeline's exchange.

Raises:

• ValueError: If both binding_key and binding_keys are set.

How a worker publishes its result to the pipeline's exchange.

The exchange's type decides how the routing key is set: fanout ignores it (leave route/route_func unset); direct/topic require exactly one of route (a static key) or route_func (computed per message, e.g. MessageFieldRouter).

Async factory callable: receives a context, returns a worker.

Async because factories may need to construct stores whose setup() is async (index creation, schema validation). Simple factories that don't need async work can still be async def with no await inside.

Bundle of resources the runner hands to each worker factory.

Workers need different resources: some need a document fetcher, some need direct document storage, and a factory can build more stores (like BinaryResultsStore) from the raw clients. A single context keeps factory signatures stable as more concerns get added.

Concrete runner that wires the pubsub backend and the stores for one worker.

All worker-specific decisions are driven by the WorkerSpec — no branching on worker type name.

Creates infrastructure from RuntimeConfig and builds default components in setup(). Inject custom components to override defaults (e.g. for testing).

Build worker instances, plus the status and retry runners.

The one place that knows which registry store goes into which runner parameter. Every runner shares infra (and with it the broker) and draws its stores from stores, so a stage reads what the previous stage wrote. The retry runner is included when config.retry.enabled.

Raises:

• ValueError: If instances names a worker type that is not in the pipeline, or gives a count that is not an integer of 1 or more.

Set up and run several runners in this event loop until one stops.

Every runner is set up before any of them runs, so every queue is bound before anything publishes (the in-process broker, like RabbitMQ, drops a message nobody is bound for). The run ends when the first runner returns or raises, or when until returns. Everything is then cancelled and torn down, and the first failure is re-raised.

WorkerRunnerBase.run() installs SIGINT/SIGTERM handlers per runner, so with several runners only the last one's handler survives. That is enough: on Ctrl-C that one runner returns, which ends the run for all. The handlers in place before the run are restored afterwards.

Top-level runtime configuration.

RabbitMQ connection + consumer settings (environment-specific).

Exchange definitions (topology) live on the PipelineSpec, not here. Config holds only what varies per deployment: where the broker is and how to consume.

Kafka settings composed from framework config models.

The default topic is jobs — the Kafka counterpart of the RMQ jobs fanout exchange. topic.create_if_missing defaults to False, so setup fails when the topic does not exist. Create the topic out-of-band, or set topic.create_if_missing.

Expiry for cached payloads, in seconds.

Fetched payload cache expiry and local-file resolution.

MongoDB job record expiry, in seconds.

Retry worker toggle plus the retry policy every consumer manager applies.

Connection attempts and bounded exponential backoff at startup.

Content namespace settings (see warren/docs/scoping.md).

All infrastructure connections for a worker process.

On backend: memory there is no MongoDB or Redis, so both clients are None and stores are injected into the runners instead.

Create and initialize all connections from runtime config.

The caller is responsible for closing via close_runtime_infrastructure.

The function tries config.startup.attempts times to connect to the pubsub backend, MongoDB and Redis. After a failed attempt, it closes the connections of that attempt and waits (see StartupConfig). On backend: memory, it creates only the in-process connection manager and returns RuntimeInfra(None, None, manager).

Raises:

• Exception: The connection error of the last attempt, re-raised unchanged.

Close all connections. Best-effort — logs and continues on errors.

Fail clearly when the memory backend would need a default store.

Default stores are built from the Mongo and Redis clients. The memory backend has neither, so every store a runner needs must be injected. Without this check the failure is an AttributeError on None deep inside a store constructor.

Raises:

• WarrenError: On the memory backend when any value is None.

Create the (unset-up) pubsub connection manager for the backend.

The caller is responsible for setup() / teardown().

Raises:

• OptionalDependencyError: if the selected backend's transport extra is not installed.

Create a publisher onto exchange for the configured backend.

Raises:

• WarrenError: if exchange is not fanout on Kafka.
• OptionalDependencyError: if the selected backend's transport extra is not installed.

Create the consumer manager for worker_type on the backend.

Owns the queue/group naming convention for all call sites:

• RabbitMQ: queue f"{exchange.name}.{worker_type}", bound with binding_keys.
• Memory: queue f"{exchange.name}.{worker_type}", same as RabbitMQ.
• Kafka: consumer group config.kafka.consumer.group_id or f"\{topic.name\}.\{worker_type\}" (pre-resolved here); binding keys are ignored on fanout.

The publisher split (data / control / observer — see warren/docs/routing.md) is forwarded verbatim to the backend consumer manager; both backends share the same constructor shape.

Raises:

• ValueError: if both binding_key and binding_keys are set.
• WarrenError: if exchange is not fanout on Kafka.
• OptionalDependencyError: if the selected backend's transport extra is not installed.

Build the default document resolver registry.

path only when path_base_dir is given, confined to it; cloud (GCS, S3) and url when their optional extras are installed. A module function so callers that are not a runner instance can build a document fetcher too.

Validate a pipeline's exchange wiring; raise on any error.

Raises:

• PipelineValidationError: with every problem found, not just the first.

Validate a job's RoutingPlan against deployed worker capabilities.

Run at job-submission time, before any message is published.

Checks: every node maps to a deployed worker; every edge u -> v is nominally type-compatible (produces[u] in accepts[v]); and — if given — every entry node accepts entry_data_type.

Raises:

• RoutingPlanValidationError: with every problem found.

Map each worker type to its declared (accepts, produces) capability.

A PipelineSpec is internally inconsistent. Carries all findings.

A job's RoutingPlan is invalid for the deployed pipeline.