Queue / Worker (Distributed)

Decouple requesting an agent run from executing it: producers enqueue jobs onto a Redis-backed queue, and a pool of workers dequeue and run them. This gives you horizontal scale, backpressure under load, retries, and fault isolation — a slow or failing worker doesn't block producers.

The example below is compiled code pulled from the paladin-doc-examples crate via mdBook {{#include}}. The Redis calls compile but are not executed by the check gate, so it stays in sync with the RedisQueueAdapter API without needing a live Redis.

Prerequisites: Run make dev (starts Redis) first, and enable the redis-queue feature on paladin-storage.

When to choose it

Choose it when you need scale-out across workers/hosts, backpressure for bursty load, automatic retries, or isolation between job execution and your request path.
Look elsewhere when load is low and in-process execution suffices (embedded library), or you only need synchronous request/response (HTTP service host).

Producer and worker

The producer enqueues a typed AgentJob; the worker dequeues it (as generic JSON), runs the agent through a PaladinExecutionService, and marks the item complete:

#![allow(unused)]
fn main() {
use std::sync::Arc;
use std::time::Duration;

use serde::{Deserialize, Serialize};

use paladin_core::base::entity::message::{Location, Message};
use paladin_core::platform::container::queue_item::QueueItem;
use paladin_ports::output::queue_port::QueuePort;
use paladin_storage::redis::{RedisQueueAdapter, RedisQueueConfig};

use paladin::application::services::paladin::paladin_execution_service::PaladinExecutionService;
use paladin::prelude::*; // Paladin, PaladinResult, ...

const QUEUE: &str = "agent-jobs";

/// The unit of work a producer enqueues and a worker executes.
#[derive(Clone, Serialize, Deserialize)]
struct AgentJob {
    agent: String,
    input: String,
}

/// **Producer** — connect to Redis and enqueue an agent job. Many producers can
/// enqueue concurrently; the queue absorbs bursts and applies backpressure.
pub async fn enqueue_job() -> Result<(), Box<dyn std::error::Error>> {
    let queue = RedisQueueAdapter::new(RedisQueueConfig::default(), None).await?;
    queue.create_queue(QUEUE.to_string(), None).await?;

    let job = AgentJob {
        agent: "summarizer".to_string(),
        input: "Summarise the Q3 earnings call.".to_string(),
    };
    let message = Message::new(
        Location::service("producer"),
        Location::service("worker"),
        job,
    );
    let item = QueueItem::new(QUEUE.to_string(), message, None);

    let id = queue.enqueue(QUEUE, item).await?;
    println!("enqueued job {id}");
    Ok(())
}

/// **Worker** — pull jobs off the queue and run them through a
/// `PaladinExecutionService`. Run many of these (in this process or across hosts)
/// to scale out. Each item is marked in-progress, then completed with its result.
pub async fn run_worker(
    queue: &RedisQueueAdapter,
    service: &PaladinExecutionService,
    agent: &Paladin,
) -> Result<(), Box<dyn std::error::Error>> {
    while let Some(item) = queue.dequeue(QUEUE).await? {
        let item_id = item.action.id;
        queue
            .start_processing(QUEUE, item_id, "worker-1".to_string())
            .await?;

        // The dequeued payload is generic JSON; read the agent input from it.
        let input = item.message.payload()["input"].as_str().unwrap_or_default();
        let result: PaladinResult = service.execute(agent, input).await?;

        queue
            .complete_processing(
                QUEUE,
                item_id,
                Some(serde_json::json!({ "output": result.output })),
            )
            .await?;
    }
    Ok(())
}
}

Run many workers — in this process via several tokio tasks, or as separate processes across hosts — all pulling from the same queue. start_processing / complete_processing / fail_processing track each item's lifecycle, and failures can retry up to the configured limit.

Configuring the queue

RedisQueueConfig is typically populated from config.yml:

queue:
  redis_host: "localhost"
  redis_port: 6379
  redis_db: 0
  connection_timeout: 30
  key_prefix: "paladin:queue"
  max_retries: 3

Run server: producer API + worker replicas

The Platform API (v0.10, docs/src/api-reference/platform-api.md) is this same queue/worker shape applied to POST /runs: paladin-server's API replicas are the producer — POST /runs performs one repository insert and one queue enqueue, with no engine work on the request path — and a separate paladin-worker Deployment is the consumer, dequeuing and driving WarEngine through RunWorkerPool. k8s/server/worker-deployment.yaml is a worked example: same image as the API Deployment, APP_RUN_STORE_BACKEND=postgres, APP_RUN_QUEUE_BACKEND=redis and APP_RUN_WORKER_CONCURRENCY=4 set so its pods only consume work — see k8s/README.md for the manifest and the Secret it needs.

A few properties of this split are worth stating plainly rather than assuming:

The API replicas and the worker replicas share the SAME run store and the SAME Redis queue. There is exactly one source of truth for a run's status (the store) and exactly one dispatch path onto a worker (the queue) — an API pod answering GET /runs/{id} and a worker pod driving that same run are reading/writing the identical row, never a per-pod copy.
Cancellation is cross-instance by construction. POST /runs/{id}/cancel writes a durable flag through the shared repository first; a worker on ANY instance — not necessarily the one that happens to be running that thread — observes the flag at the next superstep boundary via a CancellationProbe. Scaling worker replicas does not weaken cancellation.
The in-process auth token store is still single-replica-scoped (ADR-0041), and a worker/API split does not change that. paladin-worker pods never serve the /v1 auth-gated routes at all, so they add no new exposure to this limitation — it is entirely a property of however many paladin-server API replicas are running http.auth.bearer_token.enabled: true at once. Run ONE API replica if you need that credential path, or terminate auth upstream (a gateway/ingress issuing its own tokens) if you need more than one — the static-API-key path (http.auth.api_keys, the shipped default) has no such limitation, since the keys are byte-identical in every pod.

Paladin Framework

Queue / Worker (Distributed)

When to choose it

Producer and worker

Configuring the queue

Run server: producer API + worker replicas

See also