Skip to content

Lime Task Handler

Lime Task Handler is a wrapper around Celery that enables asynchronous task execution in Lime CRM. It allows you to offload long-running operations (bulk updates, reports, integrations, heavy processing) from HTTP requests to background workers, improving application responsiveness and reliability.

Run lime-task as a service

The lime-task-handler service accepts the same options as starting a Celery worker.

Basic Usage

lime-task-handler

When started without the --queues option, lime-task-handler will consume from all queues configured in your routing configuration. This differs from standard Celery behavior, which only consumes from the default queue.

Common Options

Queues: Specify which queue(s) the worker should consume from using the --queues option:

lime-task-handler --queues lime_task_queue_asap

Scheduled tasks: To run tasks on a schedule, start the Task Handler "on beat" using the --beat flag:

lime-task-handler --beat

Combining options: All options can be combined in one command:

lime-task-handler --queues lime_task_queue_asap --beat

Depending systems

The Task Handler uses RabbitMQ as a message broker and saves the results in either Elasticsearch or Redis, so make sure those services are running (docker service names: rabbitmq and elastic/redis).

Configuration

Task Routing and QoS Categories

Version Requirement

QoS-based task routing was introduced in lime-crm version 2.1187.0.

Lime Task uses a QoS (Quality of Service)-based routing system to direct tasks to appropriate queues. Tasks can be decorated with a qos parameter to categorize them by priority or execution requirements.

{
  "tasks_config": {
    "broker_connection_string": "amqp://",
    "backend_connection_string": "redis://localhost",
    "task_time_limit": null,
    "task_soft_time_limit": null,
    "task_queue_name": "lime_default_task_queue",
    "task_exchange_name": "lime_default_task_exchange",
    "task_routing_key_name": "lime_default_routing_key",
    "enable_scheduled_tasks": true,
    "result_expires": 2592000,
    "log_message_max_size": 0,
    "timezone": "UTC",
    "routing": {
      "qos": {
        "asap": "lime_task_queue_asap",
        "background": "lime_task_queue_background",
        "scheduled": "lime_task_queue_scheduled"
      },
      "tasks": {
        "my.special.task": "lime_task_queue_special"
      }
    }
  },
  "features_config": {
    "importer_with_taskhandler": false
  },
  "importer_config": {
    "connection_string": "amqp://guest@localhost//",
    "use_sql_server": false,
    "jobs_days_visible": 30,
    "sql_server_host": "localhost",
    "sql_server_database": "lime_crm_import",
    "sql_server_username": "",
    "sql_server_password": "",
    "use_s3": false,
    "s3_bucket": "",
    "s3_region": "",
    "s3_aws_access_key_id": null,
    "s3_aws_secret_access_key": null
  }
}

tasks:
    broker_connection_string: amqp://
    backend_connection_string: redis://localhost
    task_time_limit: None
    task_soft_time_limit: None
    task_queue_name: lime_default_task_queue
    task_exchange_name: lime_default_task_exchange
    task_routing_key_name: lime_default_routing_key
    enable_scheduled_tasks: True
    result_expires: 2592000  # 30 days in seconds
    log_message_max_size: 0
    timezone: UTC  # or Europe/Stockholm etc.
    routing:
        qos:
            asap: lime_task_queue_asap
            background: lime_task_queue_background
            scheduled: lime_task_queue_scheduled
        tasks:
            my.special.task: lime_task_queue_special

features: 
    importer_with_taskhandler: False

importer: 
    connection_string: amqp://guest@localhost//
    use_sql_server: False
    jobs_days_visible: 30
    sql_server_host: localhost
    sql_server_database: lime_crm_import
    sql_server_username: 
    sql_server_password: 
    use_s3: False
    s3_bucket: 
    s3_region: 
    s3_aws_access_key_id: None
    s3_aws_secret_access_key: None

Configuration Parameters

Key task configuration parameters:

  • result_expires: Time in seconds before task results are deleted from the backend (default: 2592000 = 30 days)
  • log_message_max_size: Maximum size of log messages in bytes, 0 means no limit (default: 0)
  • task_time_limit: Hard time limit for task execution in seconds (task will be killed if exceeded)
  • task_soft_time_limit: Soft time limit in seconds (raises exception but allows graceful cleanup)
  • enable_scheduled_tasks: Whether scheduled tasks are enabled for this application

Queue Name Prefix and Suffix

Queue names can be automatically prefixed and/or suffixed using environment variables. This is useful for isolating different application clusters or environments to prevent tasks from being sent across cluster boundaries.

Environment variables:

  • LIME_TASK_QUEUE_PREFIX: Prefix to add before all queue names
  • LIME_TASK_QUEUE_SUFFIX: Suffix to add after all queue names

Example:

With the configuration: 

tasks:
    task_queue_name: lime_default_task_queue
    routing:
        qos:
            asap: lime_task_queue_asap

And environment variables: 

LIME_TASK_QUEUE_PREFIX="prod_"
LIME_TASK_QUEUE_SUFFIX="_eu"

The actual queue names become:

  • prod_lime_default_task_queue_eu
  • prod_lime_task_queue_asap_eu

This transformation applies to:

  • Default queue (task_queue_name)
  • All QoS routing queues (routing.qos)
  • All task-specific routing queues (routing.tasks)

Cluster Isolation

When multiple application clusters share the same RabbitMQ instance, always set these environment variables to prevent accidental cross-cluster task routing when service configurations are copied between environments.

QoS Categories

The routing.qos configuration maps QoS categories to queue names. Lime CRM uses these categories for core tasks:

  1. asap: High-priority tasks that should be executed as soon as possible with lowest latency
  2. background: Lower-priority tasks with no latency requirement, used to avoid blocking asap tasks
  3. scheduled: Scheduled tasks that should run "on time" - similar to background but with priority for time-sensitive operations

Packages can define their own QoS categories (e.g., "email" or "documentsignings") to either prioritize tasks even more than "asap" or isolate specific workloads from the general "background" queue.

Note

QoS categories are an indirection mechanism to avoid hardcoded queue names in code. Task implementors should think in terms of QoS categories, not queues. CRM operations teams can then map these categories to actual queues based on infrastructure needs.

Task Routing Priority

The TaskRouter determines the queue for a task using the following priority order:

  1. Task name mapping (routing.tasks): Override for specific tasks
  2. Queue header (__queue): Sender can override the queue via headers
  3. QoS category header (__qos_category): Sender can override the QoS category via headers
  4. Decorator QoS category: The @task(qos=...) decorator parameter
  5. Default queue: Falls back to task_queue_name from config

Task-Specific Routing

The routing.tasks configuration allows you to route specific tasks to dedicated queues, overriding their QoS-based routing. This is useful for isolating problematic tasks or providing special handling for specific operations.

Task names use exact matching (wildcards are not supported). You can route a task to any queue, including existing QoS queues or dedicated queues.

Multiple Workers Strategy

You have two strategies for running task handler workers:

Strategy 1: All-in-One Workers

Run one or more lime-task-handler instances without the --queues option. Each worker will consume from all configured queues:

# Worker consuming from all configured queues
lime-task-handler

This approach is simpler to manage but means all tasks share the same worker resources.

Strategy 2: Dedicated Workers per Queue

Run separate lime-task-handler instances for different queues using the --queues option. This provides better isolation and allows you to scale specific workloads independently:

# Terminal 1: ASAP queue worker
lime-task-handler --queues lime_task_queue_asap

# Terminal 2: Background queue worker
lime-task-handler --queues lime_task_queue_background

# Terminal 3: Scheduled tasks worker with beat
lime-task-handler --queues lime_task_queue_scheduled --beat

# Terminal 4: Default queue worker
lime-task-handler --queues lime_default_task_queue

When to use dedicated workers:

  1. Isolate slow or problematic tasks from time-sensitive operations
  2. Scale specific workloads independently (e.g., more workers for background tasks)
  3. Ensure scheduled tasks run on time without being blocked by long-running background jobs
  4. Provide different concurrency settings or resource limits per queue type

Monitoring Task Routing

Use limefu tasks routes to view the configured task routing for your application:

limefu tasks routes -a <application_name>

This command shows:

  1. QoS Category Routes: Mapping of QoS categories to queue names
  2. Specific Task Name Routes: Task-specific routing overrides
  3. Registered Tasks: All registered tasks with their routing information

The output includes the routing type for each task (e.g., task_qos, task_route, default_queue) to help you understand how tasks are being routed.

Configuration vs Actual Routing

limefu tasks routes shows the configured routing rules, but the actual routing for a specific task invocation may differ if the sender overrides the queue using the __queue or __qos_category headers. To see the actual routing decisions, use the structured log output described below.

Troubleshooting with Structured Logs

The Task Handler logs include structured fields that help you monitor and troubleshoot task routing and execution. These fields are output as JSON attributes in the logs and are designed for use with log aggregators like ELK (Elasticsearch, Logstash, Kibana).

Task Routing Logs

When a task is routed, the following fields are logged showing the actual routing decision (including any header overrides):

  • task_route_task: The name of the task being routed
  • task_route_type: How the task was routed (task_route, header_queue, header_qos, task_qos, or default_queue)
  • task_queue: The queue the task was actually routed to

These logs appear in all application stack service logs when tasks are sent (not just taskhandler logs).

Task Execution Logs

When a task completes execution, the following fields are logged:

  • task_id: Unique identifier for the task execution
  • task_name: The name of the task that was executed
  • task_queue: The queue the task was consumed from
  • task_user_id: User ID associated with the task
  • task_latency_ms: Time (in milliseconds) the task waited in the queue before execution
  • task_duration_ms: Time (in milliseconds) the task took to execute
  • task_outcome: Result of the task execution (SUCCESS or FAILURE)
  • application: Application identifier

These logs appear only in taskhandler logs when tasks are executed.

Example Log Queries

Find which queue a specific task was routed to: 

task_route_task:"my.package.tasks.process_report" AND task_queue:*

Find tasks that were routed using header overrides: 

task_route_type:"header_queue" OR task_route_type:"header_qos"

Find slow tasks with high latency: 

task_latency_ms:>5000

Find failed tasks: 

task_outcome:"FAILURE"

Kibana Dashboards for Local Development

Lime Project includes pre-built Kibana dashboards for visualizing task routing and execution metrics. To enable logging to Kibana in your local development environment, configure the logging settings in the .env file generated by lime-project. The dashboards will automatically visualize task routing patterns, queue latencies, and failure rates.

RabbitMQ Monitoring

Use the RabbitMQ management UI to monitor queue activity:

  1. Navigate to the "Queues and Streams" page
  2. Add the "Consumer count" column to see which queues have active task handlers
  3. Monitor message rates, queue depths, and consumer utilization