Queueing: Guide¶

About this document

This guide explores the major features of the contemporary queueing system. For the initial introduction, see Overview and Quick Start.

Compatibility: This document focuses on the Upper Queue of CiviCRM v5.47+.

The Upper Queue was introduced by CiviCRM v5.47+. That version is the bare minimum for using this document.

Older versions only support Lower Queue. See also: Queueing: Comparison of Lower/Upper Queues.

The document also mentions some features added by 5.68, but they are not strictly required.

Key concepts¶

Concept	Description	Examples
Queue	A stored list of things to do.	"Pending upgrade steps", "Email outbox"
Queue type	The storage medium for reading/writing queues. (This will have a driver class.)	"Sql" (`CRM_Queue_Queue_Sql`)
Queue item	A record added to a queue. (This will track ownership, prioritization, locking, etc.)
Queue payload	The data being transported through the queue.	"RFC822 Email", "Executable Task"
Queue agent	Process that looks for things to do.	"Browser-based polling loop", "Server-based PHP daemon", "Server-based PHP cron"
Queue registration record	A stored configuration describing the type, payload, agent, etc.	MySQL table (`civicrm_queue`)

These concepts are put together to make a registration record, such as:

[
  'name' => 'upgrade-example',
  'type' => 'Sql',
  'payload' => 'task',
  'error' => 'abort',
  'agent' => 'browser',
]

After registering upgrade-example, you can get a PHP object and use it:

/** @var CRM_Queue_Queue $queue */
$queue = Civi::queue('upgrade-example');
$queue->createItem(...payload...);

To build a full application, you will need to make several decisions about how to register the queue. First, we consider the mechanism of registration, and then we explore the major properties. As you proceed through the page, drill-down on the options to learn more.

Examples¶

For this guide, we'll reference two examples:

System upgrade queue (upgrade-example): When upgrading a database (schema v1.0 => schema v1.1 => schema v1.2 => etc), there can be several data-intensive operations -- such as adding new columns, modifying indexes, and backfilling data. If there are many steps or many records, then this can take time. One may put these steps into a queue and run them incrementally.

All properties of upgrade-example

[
  'name' => 'upgrade-example',   // Logical name
  'type' => 'Sql',               // Store the queue in MySQL.
  'payload' => 'task',           // Enable support for executable tasks (PHP callbacks)
  'lease_time' => 300,           // Tasks should finish within 5 minutes. Otherwise, assume the agent is dead.
  'retry_limit' => 0,            // If a task fails, it may be tried a second time.
  'error' => 'abort',            // If any task encounters a final failure, then stop the entire queue.
  'agent' => 'browser',          // Use browser-based polling
]

Outbound email queue (email-example): Outbound email delivery requires communication with an external service -- which can be slow or unreliable. One may put these into a queue and let a background agent handle delivery. This way, email processing doesn't slow-down the web UI, and it can be retried.

All properties of email-example

[
  'name' => 'email-example',   // Logical name
  'type' => 'SqlParallel',     // Store the queue in MySQL.
  'payload' => 'mymail',       // We will put email messages directly in the queue.
  'batch_limit' => 5,          // Process up to 5 messages at a time.
  'lease_time' => 300,         // Tasks should finish within 5 minutes. Otherwise, assume the agent is dead.
  'retry_limit' => 5,          // If an email fails, it may be tried a few more times.
  'retry_interval' => 1200,    // If we encounter an ordinary exception, then politely reschedule for 20 minutes later.
  'error' => 'delete',         // If any task encounters a final failure, then stop the entire queue.
  'agent' => 'server',         // Use server-side polling
]

Register a queue¶

A registered queue is simply stored in the Queue entity (MySQL table civicrm_queue). There are a few ways to create a record:

Approach	Comments
`Civi::queue()`	Quick and dirty. Useful for rough drafts and automated tests.
`hook_managed`	Declarative. Useful for packaging one specific queue in an extension. Auto-update the registration.
APIv4	Programmable. Create, read, update, delete. Useful for administrative tools and dynamic queues.

Use Civi::queue() to register upgrade-example

The utility method Civi::queue() accepts an optional array $params.

Civi::queue(string $name, array $params = []): CRM_Queue_Queue

If given, the method will auto-initialize the queue.

$queue = Civi::queue('upgrade-example', [
  'type' => 'Sql',      // Store the queue in MySQL.
  'payload' => 'task',  // Enable support for executable tasks (PHP callbacks)
  'lease_time' => 300,  // Tasks should finish within 5 minutes. Otherwise, they are assumed to have failed.
  'retry_limit' => 0,   // If a task fails, it may be tried a second time.
  'error' => 'abort',   // If any task encounters a final failure, then stop the entire queue.
  'agent' => 'browser', // Use browser-based polling
]);

These properties only matter if the queue is new. If the queue already exists, then it will respect the pre-existing registration.

This method also returns a $queue object, so that you can immediately begin calling operations like createItem().

Use hook_managed to register email-example

function example_civicrm_managed(&$entities): void {
  $entities[] = [
    'module' => 'example',
    'name' => 'email-example',
    'entity' => 'Queue',
    'cleanup' => 'always',
    'update' => 'unmodified',
    'params' => [
      'version' => 4,
      'values' => [
        'name' => 'email-example',   // Logical name
        'type' => 'SqlParallel',     // Store the queue in MySQL.
        'payload' => 'mymail',       // We will put email messages directly in the queue.
        'batch_limit' => 5,          // Process up to 5 messages at a time.
        'lease_time' => 300,         // Tasks should finish within 5 minutes. Otherwise, they are assumed to have failed.
        'retry_limit' => 5,          // If a task fails, it may be tried a second time.
        'error' => 'delete',         // If any task encounters a final failure, then stop the entire queue.
        'agent' => 'server',         // Use server-side polling
      ],
    ],
  ];
}

Suppose you are redistributing an extension for use on many web-sites. You want to share a standard queue configuration. hook_managed is useful for declaring this standard configuration.

When the extension is installed, it will create the queue automatically. When the extension is upgraded, it will also upgrade the queue configuration. If the site-administrator customizes the configuration, then that will take precedence.

Use APIv4 to register dynamic queues

Alternatively, you can manage registered queues with APIv4. The advantage of this approach: it provides more actions for inspecting, updating, and deleting registrations. This allows administrators and developers to fine-tune the configuration over time, and it allows you to manage dynamic queues.

Civi\Api4\Queue supports the standard APIv4 operations, such as create(), get(), and update().

// Choose a dynamic name
$name = 'daily-tasks-' . date('Ymd');

// Create a new queue
Civi\Api4\Queue::create()
  ->setValues([
    'name' => $name,             // Logical name
    'type' => 'Sql',             // Store the queue in MySQL.
    'payload' => 'task',         // Enable support for executable tasks (PHP callbacks)
    'lease_time' => 300,         // Tasks should finish within 5 minutes. Otherwise, they are assumed to have failed.
    'retry_limit' => 1,          // If a task fails, it may be tried a second time.
    'error' => 'abort',          // If any task encounters a final failure, then stop the entire queue.
    'agent' => 'server',         // Use server-side polling
  ])
  ->execute();

// Read about an existing queue
$queueSpec = Civi\Api4\Queue::get()
  ->addWhere('name', 'LIKE', 'daily-tasks-%')
  ->execute()
  ->single();

// Update an existing queue
Civi\Api4\Queue::update()
  ->addWhere('name', '=', $name)
  ->setValues(['retry_limit' => 5])
  ->execute;

Note: Registration changes do not immediately affect live $queue objects.

You can use APIv4 to manage the registration records, but it does not modify a $queue object that is already live in PHP.

Generally, the queue-administration is separate from the queue-runtime, so this should not be a problem for many applications.

However, it may matter in some edge-cases -- perhaps during unit-testing or system-setup. If your PHP process needs to do both queue-administration and queue-runtime work, then consider forcing a reload:

Civi\Api4\Queue::update()->...->execute();  // Modify
CRM_Queue_Service::singleton(TRUE);         // Flush
$queue = Civi::queue('my-own-queue');       // Re-instantiate

Choose a type¶

The type of a queue indicates the storage-medium and the general data-flow. There are two main types built into CiviCRM:

Type	Storage	Dataflow	Description
`Sql`	MySQL	Linear	Data is stored in MySQL. Items execute in a strictly linear sequence (one-by-one). (If one task is running, then the next task is blocked.)
`SqlParallel`	MySQL	Parallel	Data is stored in MySQL. Multiple items may execute at the same time. (Items are generally ordered, but they are not strictly one-by-one.)

Expand the sections below to learn more about each type.

Compare: Strengths and weaknesses of "Linear" and "Parallel" dataflows

Linear queues are easier to understand and debug.
Parallel queues are more complex -- they have more moving parts which require more attention.
Linear queues are slower -- they execute 1-by-1.
Parallel queues achieve higher transaction rates -- they support batching, multiple CPUs, and multiple servers.
Linear queues make sense for dependent tasks -- they always execute tasks in predictable order.
Parallel queues make sense for independent tasks -- where tasks do not affect each other.

upgrade-example: Use a linear queue (Sql)

[
  'name' => 'upgrade-example',
  'type' => 'Sql',
  'retry_limit' => 0,
  'error' => abort',
]

In a system-upgrade queue, you have a series of steps like:

Apply schema changes for v1.1 (add a new column)
Apply schema changes for v1.2 (recompute a column)
Apply schema changes for v1.3 (add an index on the new column)

These steps implicitly depend on each other. You must add the new column before you can recompute its content or reindex it. If one step fails, then you need to stop -- don't try to run the other steps.

email-example: Use a parallel queue (SqlParallel)

[
  'name' => 'email-example',
  'type' => 'SqlParallel',
  'batch_limit' => 5,
  'retry_limit' => 5,
  'retry_interval' => 600,
  'error' => 'delete',
]

Suppose Alice makes a donation... and Bob also makes a donation. You need to send two email receipts, so you put two emails in the email-example queue.

These tasks are independent. You can run them side-by-side. It doesn't matter if the timing varies a few seconds. Errors are also independent: if you have trouble delivering email to Alice, then you don't want that to block the queue and prevent any email from going to Bob.

Choose a payload¶

The payload describes the content transported by the queue. Your choice of payload will lead to different trade-offs (flexibility, security, and performance). There are two main options:

Tasks: Add executable tasks to the queue. Each item in the queue is essentially a PHP callback.
Custom Payloads: Define a custom payload. You must choose a unique name. This approach requires more code, but it allows more tuning of performance and security.

Compare: Strengths and weaknesses of "tasks" and "custom payloads"

Tasks make it easy to use a wide range of functionality.
Custom payloads allow you to define your own data-model.
Tasks require less code.
Custom payloads require more code.
Tasks can be configured for multi-user queues (runAs).
Custom payloads can be optimized to do batch-operations.
Write-access to a task-queue must be carefully guarded.
Write-access to a generic-queue can be more relaxed (in theory).

upgrade-example: Use executable tasks

[
  'name' => 'upgrade-example',
  'payload' => 'task',
]

Recall that upgrade-example needs to run a series of distinct upgrade steps -- such as adding new tables, renaming columns, backfilling a new column, modifying an index, etc.

Every step is different, so it's convenient to use a different PHP callback for each step.

Additionally, the system-upgrade is planned by a trusted agent. You don't need to let end-users add items directly to this queue.

email-example: Use custom payload

[
  'name' => 'email-example',
  'payload' => 'mymail',
]

With the outbound email queue, every record added to the queue is the same kind of thing -- specifically, an email message. There is no need to support diverse callback functions.

Additionally, custom payloads support automatic batching. If you can dequeue messages as a batch, then you can relay them to the next email service as a batch.

At some point, you might wish to permit authorized users to enqueue outbound emails directly. If the queue uses callbacks, then it could be an open-ended security risk. A more restricted data-model could be opened up more safely.

Compatibility: CiviCRM 5.47-5.67 uses runner instead of payload.

CiviCRM 5.68 introduced the fields agent and payload. For CiviCRM 5.47-5.67, use the runner field instead.

If you want to maximize backward and forward compatibility, then set all of them:

[
  'agent' => 'server', // For 5.68+
  'payload' => 'task'  // For 5.68+
  'runner' => 'task',  // Deprecated. For 5.47-5.67.
]

Note that runner only supports server-based agents. It does not work with browser-based agents.

The later section "Add items to the queue" shows more detail about adding the payload.

Choose an error policy¶

Errors take several forms: unhandled exceptions, PHP fatals, power-outages, and more. After any of these errors, the built-in retry mechanism kicks in. The retry_limit and retry_interval determine how to retry.

If the item has been tried (and retried) as many times as permitted, then eventually the error is final. Final errors are reported to the log, and your log monitor should pick them up.

But the system must do something with a final error. There are two standard options:

error=>abort: Keep the failed item, but stop running the queue.
error=>delete: Delete the failed item, and continue running the queue.

upgrade-example: Every error should be immediately handled by a person

[
  'name' => 'upgrade-example',   // Logical name
  'lease_time' => 300,           // Tasks should finish within 5 minutes. Otherwise, they are assumed to have failed.
  'retry_limit' => 0,            // If a task fails, it may be tried a second time.
  'error' => 'abort',            // If any task encounters a final failure, then stop the entire queue.
]

A system-upgrade is a sensitive process -- a failed upgrade leaves the system in a confused state, andn it likely indicates a mistake in the upgrade planning. Proceeding with the upgrade may create cascaded errors. It requires human intervention to determine the next step.

`email-example: Retry several times before giving up

[
  'name' => 'email-example',   // Logical name
  'lease_time' => 300,         // Tasks should finish within 5 minutes. Otherwise, assume the agent is dead, and task can be retried.
  'retry_limit' => 5,          // If an email fails, it may be tried a few more times.
  'retry_interval' => 1200,    // If we encounter an ordinary exception, then politely reschedule for 20 minutes later.
  'error' => 'delete',         // If any task encounters a final failure, then stop the entire queue.
  'agent' => 'server',         // Use server-side polling
]

Additionally, you may customize the handling of errors with hook_queueTaskError.

Choose an agent¶

TODO

Add items to the queue¶

All items are added through the createItem() method:

Civi::queue(string $name)->createItem(mixed $payload, array $options = []): object;

Pay attention to the payload. As discussed before, we have two example queues with different payloads:

upgrade-example uses executable tasks (payload=>task) based on PHP callbacks.
email-example uses custom payload (payload=>mymail).

Expand the sections below for more detailed use-cases and example code.

upgrade-example: Add executable tasks with CRM_Queue_Task and PHP callbacks

Recall that we registered the queue by specifying:

[
  'name' => 'upgrade-example',
  'payload' => 'task',
]

Now, to add an executable task, use createItem() and CRM_Queue_Task.

$queue->createItem(
  new CRM_Queue_Task('task_add_column', ['civicrm_gizmo', 'status', 'varchar(32) DEFAULT NULL'])
);
$queue->createItem(
  new CRM_Queue_Task('task_add_column', ['civicrm_gizmo', 'mode', 'varchar(32) DEFAULT NULL'])
);
$queue->createItem(
  new CRM_Queue_Task('task_sql', ['UPDATE civicrm_gizmo SET status=COMPUTE(status_id), mode=COMPUTE(mode_id)'])
);
$queue->createItem(
  new CRM_Queue_Task('task_drop_column', ['civicrm_gizmo', 'status_id'])
);
$queue->createItem(
  new CRM_Queue_Task('task_drop_column', ['civicrm_gizmo', 'mode_id'])
);

Each instance of CRM_Queue_Task references a PHP function and its arguments. For this example, a quick implementation could be:

function task_add_column(CRM_Queue_TaskContext $ctx, string $table, string $column, string $props): bool {
  CRM_Core_DAO::executeQuery("ALTER TABLE $table ADD COLUMN $column $props");
  return TRUE;
}
function task_sql(CRM_Queue_TaskContext $ctx, string $sql): bool {
  CRM_Core_DAO::executeQuery($sql);
  return TRUE;
}
function task_drop_column(CRM_Queue_TaskContext $ctx, string $table, string $column): bool {
  CRM_Core_DAO::executeQuery("ALTER TABLE $table DROP COLUMN $column");
  return TRUE;
}

The callbacks have a few characteristics:

They are standalone functions or static-methods. (This makes them amenable to serialization.)
They receive some contextual information about the queue ($ctx) as well as open-ended arguments.
They return a boolean to indicate success.

For more details about CRM_Queue_Task, see Reference.

email-example: Add custom mail-type with array and EventSubscriberInterface

Recall that we registered the queue by specifying:

[
  'name' => 'email-example',
  'payload' => 'mymail',
]

Now we can call createItem() to add an email message.

We have total discretion for how to format this. For consistency, let's use the same fields as CRM_Utils_Mail::send().

Civi::queue('email-example')->createItem([
  'from' => 'info@example.org'
  'toName' => 'Alice',
  'toEmail' => 'alice@example.org',
  'subject' => 'Greetings',
  'html' => '<html><body>Hello Alice</body></html>',
]);

Finally, we need a subscriber that listens to hook_civicrm_queueRun_{PAYLOAD}.

/**
 * @service civi.queue.mymail
 */
class MyMail extends AutoService implements EventSubscriberInterface {

  use \CRM_Queue_BasicHandlerTrait;

  public static function getSubscribedEvents() {
    return [
      '&hook_civicrm_queueRun_mymail' => 'runBatch',
    ];
  }

  protected function validateItem($item): bool {
    return is_array($item->data)
      && isset($item->data['from'])
      && isset($item->data['subject']);
  }

  protected function runItem($item, \CRM_Queue_Queue $queue): void {
    fprintf(STDERR, "Send email from \"%s\" with subject \"%s\"\n",
      $item->data['from'],
      $item->data['subject']
    );
    \CRM_Utils_Mail::send($item->data);
  }

}

In this example, there are a few important things to note:

The registration (payload => mymail) matches the event name (hook_civicrm_queueRun_mymail).
$item is special record provided by the queue. $item->data is the payload that we designed.
To enable the service, you may need to call civix mixin --enable=scan-classes@1 and then cv flush.
The subscriber uses a helper, CRM_Queue_BasicHandlerTrait.

The BasicHandlerTrait (CiviCRM 5.68+) has a particular approach to batching, error-handling, and retrying. It defines extension-points such as validateItem() and runItem(). However, you may want more control over these policies, or you may need compatibility with an older version of CiviCRM. In that case, you can directly implement hook_civicrm_queueRun_{PAYLOAD}.

Tip: Fine-tune the ordering with weight and release_time

By default, new items are executed in the same order that they are inserted. For example, let's adapt the Quick Start code. Recall that qstart_fill() adds several tasks. It is basically doing this:

$queue = Civi::queue('qstart-greeter');
$queue->createItem(new CRM_Queue_Task('qstart_task_hello', ['Alice']));
$queue->createItem(new CRM_Queue_Task('qstart_task_hello', ['Bob']));
$queue->createItem(new CRM_Queue_Task('qstart_task_hello', ['Carol']));
...

That displays "Alice" then "Bob" then "Carol" (in the given order).

We can fine-tune this by passing queue-options, such as weight and release_time.

// Use `weight` to prioritize tasks. Lower values execute first.
$queue = Civi::queue('qstart-greeter');
$queue->createItem(new CRM_Queue_Task('qstart_task_hello', ['Alice']), ['weight' => 300]);
$queue->createItem(new CRM_Queue_Task('qstart_task_hello', ['Bob']), ['weight' => 200]);
$queue->createItem(new CRM_Queue_Task('qstart_task_hello', ['Carol']), ['weight' => 100]);
...

// Use `release_time' to delay execution. Separate items at 60 second intervals.
$queue = Civi::queue('qstart-greeter');
$queue->createItem(new CRM_Queue_Task('qstart_task_hello', ['Alice']), ['release_time' => time() + 60]);
$queue->createItem(new CRM_Queue_Task('qstart_task_hello', ['Bob']) ['release_time' => time() + 120]);
$queue->createItem(new CRM_Queue_Task('qstart_task_hello', ['Carol']) ['release_time' => time() + 180]);
...

This example uses tasks, but the same options apply to any items in a queue.

Tip: Define multi-user tasks with runAs

What if we have a shared task-queue, where it executes tasks on behalf of multiple users? We can set the runAs property.

function multiuser_fill(): void {
  foreach ([100, 200, 300] as $contactId) {
    $task = new CRM_Queue_Task('multiuser_task_example');
    $task->runAs = ['contactId' => $contactId, 'domainId' => 1];
    Civi::queue('my-queue')->createItem($task);
  }
}

function multiuser_task_example(): bool {
  $activities = Civi\Api4\Activity::get()->execute();
  ...
  return TRUE;
}

By setting runAs, we ensure that each task runs as a different user. When the task calls the Activity API, it will determine the permissions of the active user and filter the data accordingly.

Pay close attention to the chosen agent:

coworker has full support for multi-user queues. It sets the current user and spawns separate PHP processors for each user.
All agents validate that the task executes with the intended identity. If the validation fails, then the task fails.
Other agents may not have full support for runAs.

Run an agent¶

After you generate some tasks or work-items, you need an agent to claim and execute them. There are several ways to setup an agent, such as:

Agent	Description	Comments
Manual	Manually run a CLI command to execute items in a specific queue.	Good for debugging and development.
Browser	Start an AJAX client in the web-browser. It polls a specific queue and displays progress.	Compatible with all hosting environments.
Coworker	Start a background service. It monitors all queues and manages PHP processes.	Low latency. Always-on. One instance serves many queues and many users. Beta.
Custom Cron	Configure a custom set of cron jobs. Fine-tune performance characteristics of each queue.	No new code. Lots of configuration.

How To: Run queues manually

You can run a series of pending items using the Queue.run API:

cv api4 Queue.run queue=my-queue-name

This will continue working on the queue until either (a) it runs out of tasks or (b) it encounters a runtime limit.

The two runtime limits are maxRequests and maxDuration. You can pass these as extra parameters:

## Execute up to one request
cv api4 Queue.run queue=my-queue-name maxRequests=1

## Execute up to fifty requests or up to 5 minutes. (After the limit, shutdown gracefully.)
cv api4 Queue.run queue=my-queue-name maxRequests=50 maxDuration=600

How To: Run queues via web browser

TODO: Need to port the web-agent to use Queue.runItems

How To: Run queues via coworker (beta)

coworker is a process-manager that integrates with the CiviCRM queueing system. It monitors the list of queues. As necessary, it opens new connections to CiviCRM, executes tasks, and closes old connections. It accepts a number of options to throttle the workload.

coworker is included with civicrm-buildkit and civicrm-cli-tools. It can also be downloaded as a standalone program.

sudo wget 'https://download.civicrm.org/coworker/coworker.phar' -O '/usr/local/bin/coworker'
sudo chmod +x /usr/local/bin/coworker

When you have the coworker binary, then you can start monitoring the queues. Use a command like one of these:

## Run coworker. Connect to CiviCRM with the default method, `cv`.
cd /var/www/example.com
coworker run

## Alternatively, connect via drush
cd /var/www/example.com
coworker run --pipe='drush ev "civicrm_initialize(); Civi::pipe();"'

## Alternatively, connect via drush site-alias
coworker run --pipe='drush @example.com ev "civicrm_initialize(); Civi::pipe();"'

## Alternatively, connect via wp-cli
cd /var/www/example.com
coworker run --pipe='wp eval "civicrm_initialize(); Civi::pipe();"'

## Alternatively, connect via ssh+cv
coworker run --pipe='ssh webuser@backend.example.com cv pipe --cwd=/var/www/example.com'

At time of writing, coworker is in beta. As it matures, we will flesh out documentation in the "CiviCRM System Administrator Guide" (re: connection methods, systemd, etc). For now, more information is available at https://lab.civicrm.org/dev/coworker.

How To: Run queues via custom cron

In "System Administration Guide: Scheduled jobs", it presents a pattern for integrating cron and CiviCRM. The pattern is:

Identify an API (such as Job.execute API)
Choose a mechanism to connect to the API (such as cv, drush, wp-cli, wget, or curl)
Setup a crontab which calls this API.

You can apply the same pattern for running queues -- simply target Queue.run API instead.

This approach allows you to specify when (and how long) the system will work on each queue. For example, a high priority queue might run every 5 minutes (throughout the day). A low priority queue might be limited to offpeak hours.

*/5  *   * * * www-data ( cv api4 Queue.run queue=first  maxRequests=20 maxDuration=300 --cwd=/var/www/example.com/web )
*/15 *   * * * www-data ( cv api4 Queue.run queue=second maxRequests=20 maxDuration=300 --cwd=/var/www/example.com/web )
0/20 0-4 * * * www-data ( cv api4 Queue.run queue=third  maxRequests=50 maxDuration=1200 --cwd=/var/www/example.com/web )

This example shows 3 queues (first, second, third), and it executes via cv. For a custom cron, you need to fill-in your actual queues and choose the most appropriate mechanism (cv, drush, wp-cli, etc).

It should be emphasized that these mechanisms have different polling behavior. But the business logic of evaluating the task is the same (as per Creating tasks and work-items). They build on the same APIs, settings, and events.

They all invoke Civi\Api4\Queue::runItems to execute one batch of items.
They all check the registration record for options like batch_limit, lease_time, retry_limit, and error.
They all dispatch events like hook_civicrm_queueRun_{PAYLOAD} and hook_civicrm_queueTaskError.