> ## Documentation Index
> Fetch the complete documentation index at: https://docs.exulu.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Overview

> Background job processing with BullMQ for asynchronous task execution

## Overview

`ExuluQueues` manages background job queues using BullMQ and Redis. It enables asynchronous processing of long-running tasks like embeddings generation, document processing, scheduled data syncs, and agent workflows. Queues help keep your application responsive by offloading heavy work to background workers.

## Key features

<CardGroup cols={2}>
  <Card title="BullMQ integration" icon="bolt">
    Built on BullMQ, a robust Redis-based queue system
  </Card>

  <Card title="Concurrency control" icon="gauge">
    Configure worker and queue-level concurrency limits
  </Card>

  <Card title="Rate limiting" icon="clock">
    Control job processing rate to manage API limits
  </Card>

  <Card title="Retries & backoff" icon="rotate">
    Automatic retry with exponential or linear backoff
  </Card>

  <Card title="Timeouts" icon="hourglass">
    Set maximum execution time for jobs
  </Card>

  <Card title="Telemetry" icon="chart-line">
    OpenTelemetry integration for monitoring
  </Card>
</CardGroup>

## What is a queue?

A queue is a system for managing background jobs:

1. **Jobs are added** to the queue with data and options
2. **Workers process jobs** asynchronously in the background
3. **Results are tracked** with success/failure status
4. **Failed jobs retry** automatically based on configuration

Queues prevent blocking operations from slowing down your application and enable horizontal scaling by adding more workers.

## Why use queues?

<AccordionGroup>
  <Accordion title="Long-running operations">
    Tasks like embeddings generation, document processing, or data exports can take seconds or minutes. Queues let you return immediately to users while work continues in the background.
  </Accordion>

  <Accordion title="Rate limit management">
    External APIs often have rate limits. Queues control processing speed to stay within limits while maximizing throughput.
  </Accordion>

  <Accordion title="Reliability">
    Jobs persist in Redis. If a worker crashes, jobs aren't lost and will retry when workers restart.
  </Accordion>

  <Accordion title="Scalability">
    Add more worker processes to handle increased load without changing code.
  </Accordion>

  <Accordion title="Scheduled execution">
    Process jobs at specific times or intervals (e.g., daily data syncs, weekly reports).
  </Accordion>
</AccordionGroup>

## Quick start

```typescript theme={null}
import { ExuluQueues } from "@exulu/backend";

// Register a queue
const embeddingsQueue = ExuluQueues.register(
  "embeddings",           // Queue name
  {
    worker: 5,            // 5 concurrent jobs per worker
    queue: 10             // 10 global concurrent jobs
  },
  10,                     // Rate limit: 10 jobs/second
  180                     // Timeout: 180 seconds
);

// Use the queue
const queueConfig = await embeddingsQueue.use();

// Add a job to the queue
await queueConfig.queue.add("generate-embeddings", {
  contextId: "docs",
  itemId: "item-123"
});

// Create a worker to process jobs
import { Worker } from "bullmq";

const worker = new Worker(
  "embeddings",
  async (job) => {
    console.log("Processing:", job.data);

    // Do the work
    await generateEmbeddings(job.data.contextId, job.data.itemId);

    return { success: true };
  },
  {
    connection: redisServer,
    concurrency: queueConfig.concurrency.worker,
    limiter: {
      max: queueConfig.ratelimit,
      duration: 1000
    }
  }
);
```

## Architecture

### BullMQ and Redis

ExuluQueues wraps BullMQ, which uses Redis for:

* **Job storage** - Jobs persist in Redis
* **State management** - Track job status (waiting, active, completed, failed)
* **Locks** - Prevent duplicate processing
* **Pub/Sub** - Notify workers of new jobs

```typescript theme={null}
import { Queue } from "bullmq";

// ExuluQueues creates BullMQ Queue instances
const queue = new Queue("my-queue", {
  connection: {
    host: "localhost",
    port: 6379
  }
});
```

### Concurrency levels

ExuluQueues supports two concurrency levels:

<Tabs>
  <Tab title="Worker concurrency">
    Number of jobs a single worker process handles simultaneously:

    ```typescript theme={null}
    concurrency: {
      worker: 5  // Each worker processes 5 jobs at once
    }
    ```

    Higher values = more throughput per worker, but more memory/CPU usage.
  </Tab>

  <Tab title="Queue concurrency">
    Maximum jobs processing across all workers globally:

    ```typescript theme={null}
    concurrency: {
      queue: 10  // Max 10 jobs processing across all workers
    }
    ```

    Prevents overwhelming external services or databases.
  </Tab>
</Tabs>

**Example scenario:**

* Queue concurrency: 20
* Worker concurrency: 5
* 10 workers running

Each worker can handle 5 jobs, but only 20 jobs will run simultaneously across all workers.

### Rate limiting

Control job processing speed:

```typescript theme={null}
ExuluQueues.register(
  "api-calls",
  { worker: 1, queue: 5 },
  10  // Process max 10 jobs per second
);
```

Useful for:

* Staying within API rate limits
* Preventing database overload
* Controlling costs (API calls, LLM tokens)

## Common use cases

<Tabs>
  <Tab title="Embeddings generation">
    ```typescript theme={null}
    const embeddingsQueue = ExuluQueues.register(
      "embeddings",
      { worker: 5, queue: 10 },
      10,  // 10 jobs/sec
      300  // 5 min timeout
    );

    // Used by ExuluContext
    const context = new ExuluContext({
      id: "docs",
      embedder: new ExuluEmbedder({
        // ...
        queue: await embeddingsQueue.use()
      })
    });

    // Jobs are queued automatically
    await context.createItem(item, config);
    ```
  </Tab>

  <Tab title="Document processing">
    ```typescript theme={null}
    const processingQueue = ExuluQueues.register(
      "document-processing",
      { worker: 3, queue: 5 },
      5,   // 5 jobs/sec
      600  // 10 min timeout
    );

    // Used by processor
    const context = new ExuluContext({
      id: "docs",
      processor: {
        name: "PDF Processor",
        config: {
          queue: await processingQueue.use(),
          trigger: "onInsert"
        },
        execute: async ({ item }) => {
          const text = await extractPDFText(item.file_s3key);
          return { ...item, content: text };
        }
      }
    });
    ```
  </Tab>

  <Tab title="Scheduled data sync">
    ```typescript theme={null}
    const syncQueue = ExuluQueues.register(
      "data-sync",
      { worker: 1, queue: 1 },
      1,    // 1 job/sec
      1800  // 30 min timeout
    );

    // Used by ExuluContext source
    const context = new ExuluContext({
      id: "github-issues",
      sources: [{
        id: "github",
        name: "GitHub Sync",
        description: "Syncs issues from GitHub",
        config: {
          schedule: "0 */6 * * *",  // Every 6 hours
          queue: await syncQueue.use()
        },
        execute: async () => {
          const issues = await fetchGitHubIssues();
          return issues;
        }
      }]
    });
    ```
  </Tab>

  <Tab title="Agent workflows">
    ```typescript theme={null}
    const workflowQueue = ExuluQueues.register(
      "agent-workflows",
      { worker: 10, queue: 20 },
      20,  // 20 jobs/sec
      300  // 5 min timeout
    );

    // Used by ExuluAgent
    const agent = new ExuluAgent({
      id: "assistant",
      name: "Assistant",
      workflows: {
        enabled: true,
        queue: await workflowQueue.use()
      }
      // ...
    });
    ```
  </Tab>
</Tabs>

## Queue lifecycle

<Steps>
  <Step title="Register queue">
    Define queue configuration with `ExuluQueues.register()`

    ```typescript theme={null}
    const myQueue = ExuluQueues.register("my-queue", {...});
    ```
  </Step>

  <Step title="Initialize queue">
    Call `.use()` to create the BullMQ Queue instance

    ```typescript theme={null}
    const config = await myQueue.use();
    ```
  </Step>

  <Step title="Add jobs">
    Queue jobs for processing

    ```typescript theme={null}
    await config.queue.add("job-name", { data: "..." });
    ```
  </Step>

  <Step title="Create workers">
    Start worker processes to handle jobs

    ```typescript theme={null}
    const worker = new Worker("my-queue", processor, options);
    ```
  </Step>

  <Step title="Process jobs">
    Workers pick up and execute jobs asynchronously
  </Step>
</Steps>

## Integration with ExuluApp

ExuluApp automatically creates workers for registered queues:

```typescript theme={null}
const app = new ExuluApp();
await app.create({
  config: {
    workers: {
      enabled: true  // Creates workers for all queues
    }
  },
  contexts: {
    docs: docsContext  // Context with embedder queue
  },
  agents: {
    assistant: agent   // Agent with workflow queue
  }
});

// Workers are automatically created and started
```

When `workers.enabled: true`, ExuluApp:

1. Discovers all queues from contexts, agents, and embedders
2. Creates Worker instances for each queue
3. Configures concurrency and rate limits
4. Starts processing jobs

## Job states

Jobs move through these states:

```
waiting → active → completed
              ↓
           failed → waiting (retry)
```

* **waiting**: Job is queued, not yet picked up
* **active**: Worker is currently processing
* **completed**: Job finished successfully
* **failed**: Job threw an error

Failed jobs automatically retry based on configuration.

## Monitoring

### Built-in telemetry

ExuluQueues includes OpenTelemetry integration:

```typescript theme={null}
import { BullMQOtel } from "bullmq-otel";

const queue = new Queue("my-queue", {
  telemetry: new BullMQOtel("exulu-app")
});
```

This tracks:

* Job duration
* Success/failure rates
* Queue depth
* Processing latency

### Queue inspection

Check queue status programmatically:

```typescript theme={null}
const config = await myQueue.use();

// Get job counts
const counts = await config.queue.getJobCounts();
console.log(counts);
// { waiting: 5, active: 2, completed: 100, failed: 3 }

// Get specific jobs
const failedJobs = await config.queue.getFailed();
const activeJobs = await config.queue.getActive();

// Get queue metrics
const concurrency = await config.queue.getGlobalConcurrency();
console.log(`Global concurrency: ${concurrency}`);
```

## Best practices

<Tip>
  **Right-size concurrency**: Start with low concurrency and increase gradually. Monitor CPU/memory usage and adjust.
</Tip>

<Note>
  **Configure timeouts**: Set realistic timeouts based on expected job duration. Too short = premature failures, too long = stuck workers.
</Note>

<Warning>
  **Redis is required**: Queues require a Redis server. Ensure Redis is configured before using queues.
</Warning>

<Info>
  **Rate limits**: Set rate limits based on external service constraints (API limits, database capacity, LLM rate limits).
</Info>

## Redis configuration

ExuluQueues requires Redis connection info:

```bash theme={null}
# Environment variables
REDIS_HOST=localhost
REDIS_PORT=6379
REDIS_PASSWORD=optional-password
```

Or configure programmatically:

```typescript theme={null}
import { redisServer } from "@exulu/backend";

redisServer.host = "localhost";
redisServer.port = "6379";
redisServer.password = "optional-password";
```

## Next steps

<CardGroup cols={2}>
  <Card title="Configuration" icon="gear" href="/core/exulu-queues/configuration">
    Learn about queue configuration options
  </Card>

  <Card title="API reference" icon="code" href="/core/exulu-queues/api-reference">
    Explore methods and properties
  </Card>
</CardGroup>