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

# Queues and concurrency

Queues control how many workflow instances can run simultaneously. Use queues to rate-limit execution, manage resource contention, and prevent overload.

## Why use queues?

Queues solve common concurrency challenges:

**Rate limiting:**

* Prevent overwhelming external APIs (OpenAI, Stripe, etc.)
* Control database connection usage
* Manage resource consumption

**Ordered execution:**

* Process per-user tasks sequentially
* Maintain ordering for critical operations
* Prevent race conditions

**Resource protection:**

* Limit concurrent access to shared resources
* Prevent thundering herd problems
* Control system load

## Default queue behavior

Every workflow automatically gets its own queue:

<CodeGroup>
  ```python Python theme={null}
  from polos import workflow, WorkflowContext

  @workflow
  async def process_data(ctx: WorkflowContext, input: dict):
      # Automatically assigned to queue named "process-data"
      # Default concurrency limit from environment
      result = await ctx.step.run("process", process, input)
      return result
  ```

  ```typescript TypeScript theme={null}
  import { defineWorkflow } from '@polos/sdk';

  const processData = defineWorkflow<dict, void, any>(
    { id: 'process-data' },
    async (ctx, input) => {
      // Automatically assigned to queue named "process-data"
      // Default concurrency limit from environment
      const result = await ctx.step.run('process', () => process(input));
      return result;
    }
  );
  ```
</CodeGroup>

**Default queue properties:**

* Queue name: Same as workflow ID
* Concurrency limit: From environment variable (`POLOS_DEFAULT_CONCURRENCY_LIMIT`)
* Behavior: Unlimited executions queue up, but only N run simultaneously

## Custom queue limits

Set a specific concurrency limit for a workflow

<CodeGroup>
  ```python Python theme={null}
  @workflow(queue={"concurrency_limit": 3})
  async def rate_limited_workflow(ctx: WorkflowContext, input: dict):
      # Only 3 instances run at once
      # Additional invocations queue until a slot opens
      await ctx.step.run("call_api", call_external_api, input)
      return {"status": "completed"}
  ```

  ```typescript TypeScript theme={null}
  import { defineWorkflow } from '@polos/sdk';

  const rateLimitedWorkflow = defineWorkflow<dict, void, { status: string }>(
    { id: 'rate-limited-workflow', queue: { concurrencyLimit: 3 } },
    async (ctx, input) => {
      // Only 3 instances run at once
      // Additional invocations queue until a slot opens
      await ctx.step.run('call_api', () => callExternalApi(input));
      return { status: 'completed' };
    }
  );
  ```
</CodeGroup>

## Shared queues

Multiple workflows can share the same queue

<CodeGroup>
  ```python Python theme={null}
  from polos import queue

  # Create a shared queue
  openai_queue = queue("openai-api", concurrency_limit=5)

  @workflow(queue=openai_queue)
  async def gpt_summarizer(ctx: WorkflowContext, input: SummarizerInput):
      summary = await ctx.step.run("summarize", call_openai_gpt, input.text)
      return summary

  @workflow(queue=openai_queue)
  async def gpt_analyzer(ctx: WorkflowContext, input: AnalyzerInput):
      analysis = await ctx.step.run("analyze", call_openai_gpt, input.data)
      return analysis

  # Both workflows share the "openai-api" queue
  # Combined max 5 concurrent executions across both workflows
  ```

  ```typescript TypeScript theme={null}
  import { Queue, defineWorkflow } from '@polos/sdk';

  // Create a shared queue
  const openaiQueue = new Queue('openai-api', { concurrencyLimit: 5 });

  const gptSummarizer = defineWorkflow<SummarizerInput, void, string>(
    { id: 'gpt-summarizer', queue: openaiQueue },
    async (ctx, input) => {
      const summary = await ctx.step.run('summarize', () => callOpenaiGpt(input.text));
      return summary;
    }
  );

  const gptAnalyzer = defineWorkflow<AnalyzerInput, void, string>(
    { id: 'gpt-analyzer', queue: openaiQueue },
    async (ctx, input) => {
      const analysis = await ctx.step.run('analyze', () => callOpenaiGpt(input.data));
      return analysis;
    }
  );

  // Both workflows share the "openai-api" queue
  // Combined max 5 concurrent executions across both workflows
  ```
</CodeGroup>

**Use cases for shared queues:**

* Rate-limit API calls across multiple workflows
* Control database connection pooling
* Manage resource contention

## Queue configuration options

### 1. Inline configuration

<CodeGroup>
  ```python Python theme={null}
  @workflow(queue={"concurrency_limit": 10})
  async def inline_config(ctx: WorkflowContext, input: dict):
      # Queue name = "inline-config" (workflow ID)
      # Concurrency limit = 10
      pass
  ```

  ```typescript TypeScript theme={null}
  import { defineWorkflow } from '@polos/sdk';

  const inlineConfig = defineWorkflow<dict, void, void>(
    { id: 'inline-config', queue: { concurrencyLimit: 10 } },
    async (ctx, input) => {
      // Queue name = "inline-config" (workflow ID)
      // Concurrency limit = 10
    }
  );
  ```
</CodeGroup>

### 2. Queue name only

<CodeGroup>
  ```python Python theme={null}
  @workflow(queue="processing-queue")
  async def named_queue(ctx: WorkflowContext, input: dict):
      # Queue name = "processing-queue"
      # Concurrency limit = default from environment
      pass
  ```

  ```typescript TypeScript theme={null}
  import { defineWorkflow } from '@polos/sdk';

  const namedQueue = defineWorkflow<dict, void, void>(
    { id: 'named-queue', queue: 'processing-queue' },
    async (ctx, input) => {
      // Queue name = "processing-queue"
      // Concurrency limit = default from environment
    }
  );
  ```
</CodeGroup>

### 3. Queue object

<CodeGroup>
  ```python Python theme={null}
  from polos import queue

  priority_queue = queue("priority", concurrency_limit=3)

  @workflow(queue=priority_queue)
  async def priority_workflow(ctx: WorkflowContext, input: dict):
      # Queue name = "priority"
      # Concurrency limit = 3
      pass
  ```

  ```typescript TypeScript theme={null}
  import { Queue, defineWorkflow } from '@polos/sdk';

  const priorityQueue = new Queue('priority', { concurrencyLimit: 3 });

  const priorityWorkflow = defineWorkflow<dict, void, void>(
    { id: 'priority-workflow', queue: priorityQueue },
    async (ctx, input) => {
      // Queue name = "priority"
      // Concurrency limit = 3
    }
  );
  ```
</CodeGroup>

## Per-entity queuing (Concurrency keys)

Use concurrency keys to create separate queues per user, tenant, or entity:

<CodeGroup>
  ```python Python theme={null}
  @workflow(queue="user-tasks")
  async def user_task(ctx: WorkflowContext, input: dict):
      # Process user-specific task
      await ctx.step.run("process", process_user_task, input)
      return {"status": "completed"}

  # Invoke with concurrency_key
  client = PolosClient()
  await user_task.invoke(
      client,
      payload={"task": "data"},
      concurrency_key=f"user:{user_id}"  # Separate queue per user
  )
  ```

  ```typescript TypeScript theme={null}
  import { PolosClient, defineWorkflow } from '@polos/sdk';

  const userTask = defineWorkflow<dict, void, { status: string }>(
    { id: 'user-task', queue: 'user-tasks' },
    async (ctx, input) => {
      // Process user-specific task
      await ctx.step.run('process', () => processUserTask(input));
      return { status: 'completed' };
    }
  );

  // Invoke with concurrencyKey
  const client = PolosClient.fromEnv();
  await userTask.invoke(
    client,
    { task: 'data' },
    { concurrencyKey: `user:${userId}` }  // Separate queue per user
  );
  ```
</CodeGroup>

**How it works:**

* Each unique `concurrency_key` gets its own virtual queue
* Concurrency limit applies per key (not globally)
* Perfect for multi-tenant systems

**Example: Per-user rate limiting**

<CodeGroup>
  ```python Python theme={null}
  # User A can have 5 concurrent tasks
  await user_task.invoke(client, payload={...}, concurrency_key="user:alice")
  await user_task.invoke(client, payload={...}, concurrency_key="user:alice")
  await user_task.invoke(client, payload={...}, concurrency_key="user:alice")

  # User B also gets 5 concurrent slots (independent from User A)
  await user_task.invoke(client, payload={...}, concurrency_key="user:bob")
  await user_task.invoke(client, payload={...}, concurrency_key="user:bob")
  ```

  ```typescript TypeScript theme={null}
  // User A can have 5 concurrent tasks
  await userTask.invoke(client, { /* ... */ }, { concurrencyKey: 'user:alice' });
  await userTask.invoke(client, { /* ... */ }, { concurrencyKey: 'user:alice' });
  await userTask.invoke(client, { /* ... */ }, { concurrencyKey: 'user:alice' });

  // User B also gets 5 concurrent slots (independent from User A)
  await userTask.invoke(client, { /* ... */ }, { concurrencyKey: 'user:bob' });
  await userTask.invoke(client, { /* ... */ }, { concurrencyKey: 'user:bob' });
  ```
</CodeGroup>

## Serial execution

Force workflows to run one at a time:

<CodeGroup>
  ```python Python theme={null}
  @workflow(queue={"concurrency_limit": 1})
  async def serial_workflow(ctx: WorkflowContext, input: dict):
      # Only 1 instance runs at a time
      # Perfect for operations that must not overlap
      await ctx.step.run("critical_section", critical_operation, input)
      return {"status": "completed"}
  ```

  ```typescript TypeScript theme={null}
  import { defineWorkflow } from '@polos/sdk';

  const serialWorkflow = defineWorkflow<dict, void, { status: string }>(
    { id: 'serial-workflow', queue: { concurrencyLimit: 1 } },
    async (ctx, input) => {
      // Only 1 instance runs at a time
      // Perfect for operations that must not overlap
      await ctx.step.run('critical_section', () => criticalOperation(input));
      return { status: 'completed' };
    }
  );
  ```
</CodeGroup>

**Use cases:**

* Database migrations
* File system operations
* Operations with global side effects

## Queue behavior

### Queuing and execution

<CodeGroup>
  ```python Python theme={null}
  @workflow(queue={"concurrency_limit": 2})
  async def limited_workflow(ctx: WorkflowContext, input: dict):
      await ctx.step.run("work", do_work, input)
      return {"done": True}

  # Invoke 5 times
  client = PolosClient()
  for i in range(5):
      await limited_workflow.invoke(client, {"id": i})
  ```

  ```typescript TypeScript theme={null}
  import { PolosClient, defineWorkflow } from '@polos/sdk';

  const limitedWorkflow = defineWorkflow<dict, void, { done: boolean }>(
    { id: 'limited-workflow', queue: { concurrencyLimit: 2 } },
    async (ctx, input) => {
      await ctx.step.run('work', () => doWork(input));
      return { done: true };
    }
  );

  // Invoke 5 times
  const client = PolosClient.fromEnv();
  for (let i = 0; i < 5; i++) {
    await limitedWorkflow.invoke(client, { id: i });
  }
  ```
</CodeGroup>

```
# Execution timeline:
# Time 0: Workflow 1 starts, Workflow 2 starts (limit = 2)
# Time 0: Workflow 3, 4, 5 queued
# Time 5: Workflow 1 completes → Workflow 3 starts
# Time 7: Workflow 2 completes → Workflow 4 starts
# Time 10: Workflow 3 completes → Workflow 5 starts
```

**Key points:**

* Queued workflows wait until a slot opens
* FIFO order (first in, first out)
* No compute consumed while queued

### Dynamic queue assignment

Override queue at invocation time:

<CodeGroup>
  ```python Python theme={null}
  @workflow(queue="default-queue")
  async def flexible_workflow(ctx: WorkflowContext, input: dict):
      await ctx.step.run("process", process_data, input)
      return {"status": "completed"}

  # Use default queue
  client = PolosClient()
  await flexible_workflow.invoke(client, {"data": "..."})

  # Override with different queue
  await flexible_workflow.invoke(
      client,
      {"data": "..."},
      queue="priority-queue"
  )
  ```

  ```typescript TypeScript theme={null}
  import { PolosClient, defineWorkflow } from '@polos/sdk';

  const flexibleWorkflow = defineWorkflow<dict, void, { status: string }>(
    { id: 'flexible-workflow', queue: 'default-queue' },
    async (ctx, input) => {
      await ctx.step.run('process', () => processData(input));
      return { status: 'completed' };
    }
  );

  // Use default queue
  const client = PolosClient.fromEnv();
  await flexibleWorkflow.invoke(client, { data: '...' });

  // Override with different queue
  await flexibleWorkflow.invoke(
    client,
    { data: '...' },
    { queue: 'priority-queue' }
  );
  ```
</CodeGroup>

## Queue restriction

Scheduled workflows cannot specify queues.

## Key takeaways

* **Default queue per workflow** - Named after workflow ID
* **Concurrency limits** control max simultaneous executions
* **Shared queues** rate-limit across multiple workflows
* **Concurrency keys** create per-entity virtual queues
* **Scheduled workflows** cannot customize queues