Skip to content

Workflow Queue Setup - Dedicated Workers

Overview

Workflow tasks (trigger_agentic_workflow) are now routed to a dedicated workflows queue with dedicated workers, preventing them from being blocked by long-running source check tasks.

Changes Made

1. Celery Configuration (src/worker/celeryconfig.py)

Added workflow queue routing:

Python
task_routes = {
    # ... existing routes
    'src.worker.celery_app.trigger_agentic_workflow': {'queue': 'workflows'},
}

task_queues = {
    # ... existing queues
    'workflows': {
        'exchange': 'workflows',
        'routing_key': 'workflows',
    },
}

2. Docker Compose (docker-compose.yml)

Updated main worker to exclude workflows queue (note: collection_immediate is listed first so user-initiated "Collect Now" tasks are consumed before scheduled collection):

YAML
worker:
  command: celery -A src.worker.celery_app worker --loglevel=${CELERY_LOG_LEVEL:-info} -Q collection_immediate,default,source_checks,maintenance,reports,connectivity,collection --concurrency=${WORKER_CONCURRENCY:-2} --without-gossip --without-mingle --without-heartbeat

Added dedicated workflow worker:

YAML
workflow_worker:
  container_name: cti_workflow_worker
  command: celery -A src.worker.celery_app worker --loglevel=${CELERY_LOG_LEVEL:-info} -Q workflows --concurrency=${WORKFLOW_WORKER_CONCURRENCY:-2} --without-gossip --without-mingle --without-heartbeat
  # ... same environment and volumes as main worker

Benefits

  1. Isolation: Workflow tasks are never blocked by source checks.
  2. Performance: Dedicated workers run workflow execution without contention from other queues.
  3. Scalability: Workflow workers scale independently of the main worker (see Scaling).
  4. Reliability: Workflow failures don't affect other task types.

Restart Recovery

If a worker restart interrupts a workflow, its execution may remain marked running. On the next trigger attempt for that article, an execution with no updated_at change for 30 minutes is marked failed with a recovery message, which permits a forced retry. Recent executions are left running; this recovers the trigger path only and does not resume interrupted work.

Deployment

Restart Services

Bash
docker-compose up -d workflow_worker
docker-compose restart worker

Verify Setup

Check workflow worker is running:

Bash
docker ps | grep workflow_worker

Check queue routing:

Bash
docker exec cti_workflow_worker celery -A src.worker.celery_app inspect registered | grep trigger_agentic_workflow

Check worker is consuming workflows queue:

Bash
docker exec cti_workflow_worker celery -A src.worker.celery_app inspect active_queues

Monitor workflow tasks:

Bash
docker logs cti_workflow_worker --tail 50 -f

Queue Architecture

Text Only
┌─────────────────┐
│  Main Worker    │  → Processes: collection_immediate, default, source_checks,
│  (cti_worker)   │              maintenance, reports, connectivity, collection
└─────────────────┘

┌─────────────────┐
│ Workflow Worker │  → Processes: workflows (trigger_agentic_workflow)
│(workflow_worker)│
└─────────────────┘

Monitoring

Check Queue Depth

Bash
docker exec cti_redis redis-cli LLEN workflows

Check Overall Worker Health

For a broader view (not just the workflows queue), see Agent Orientation -- Workers and schedules and the AGENTS.md "Runtime" section (repository root) for the restart requirement on cti_worker / cti_workflow_worker after .py changes.

Check Workflow Worker Stats

Bash
docker exec cti_workflow_worker celery -A src.worker.celery_app inspect stats

Check Active Tasks

Bash
docker exec cti_workflow_worker celery -A src.worker.celery_app inspect active

Check Reserved Tasks

Bash
docker exec cti_workflow_worker celery -A src.worker.celery_app inspect reserved

Troubleshooting

Workflow tasks still pending

  1. Check workflow worker is running:

    Bash
    docker ps | grep workflow_worker

  2. Check worker is consuming workflows queue:

    Bash
    docker exec cti_workflow_worker celery -A src.worker.celery_app inspect active_queues
    Should show workflows queue.

  3. Check for errors:

    Bash
    docker logs cti_workflow_worker --tail 100

  4. Verify task routing:

    Bash
    docker exec cti_workflow_worker celery -A src.worker.celery_app inspect registered | grep trigger_agentic_workflow

Worker not starting

  1. Check Redis connectivity:

    Bash
    docker exec cti_workflow_worker celery -A src.worker.celery_app inspect ping

  2. Check database connectivity:

    Bash
    docker exec cti_workflow_worker python3 -c "from src.database.manager import DatabaseManager; db = DatabaseManager(); print('DB OK')"

Scaling

To scale workflow workers (e.g., for high evaluation load):

Bash
docker-compose up -d --scale workflow_worker=3

Or update docker-compose.yml:

YAML
workflow_worker:
  deploy:
    replicas: 3

  • src/worker/celeryconfig.py - Queue and routing configuration
  • docker-compose.yml - Worker service definitions
  • src/worker/celery_app.py - trigger_agentic_workflow task

Last updated: 2026-07-03 Last reviewed: 2026-09-01