Stop calling the CRM inside node save: a Drupal queue worker in 60 lines

Last month a client site of mine started dropping webhooks. The pattern was familiar: a node save fired an HTTP call to a third-party CRM inside the request, the CRM had a bad afternoon, and editors waited eight seconds for a save that then failed anyway. The fix is as old as Drupal's Queue API, but the modern version is much terser than the tutorials suggest, so here is the version I now copy between projects.

Enqueue instead of calling

The save hook only records that work needs doing. Queue items should be small, serializable payloads, so I pass an entity ID and not the entity itself. Using the OOP hook style from Drupal 11:

<?php

declare(strict_types=1);

namespace Drupal\my_crm\Hook;

use Drupal\Core\Entity\EntityInterface;
use Drupal\Core\Hook\Attribute\Hook;
use Drupal\Core\Queue\QueueFactory;

final class NodeHooks {

  public function __construct(
    private readonly QueueFactory $queueFactory,
  ) {}

  #[Hook('entity_insert')]
  public function onInsert(EntityInterface $entity): void {
    if ($entity->getEntityTypeId() !== 'node' || $entity->bundle() !== 'article') {
      return;
    }
    $this->queueFactory->get('my_crm_webhook')->createItem([
      'nid' => (int) $entity->id(),
    ]);
  }

}

QueueFactory::get() returns a QueueInterface, and createItem() stores the data. The save now costs one database insert, whatever the CRM is doing.

The worker as a plugin attribute

Since Drupal 11.1 a queue worker is declared with the #[QueueWorker] attribute in Drupal\Core\Queue\Attribute instead of an annotation. It takes an id, an optional title, and an optional cron array whose only key is time: the seconds cron may spend on this worker. The docs are explicit that the plugin ID must match the queue's machine name, which is why mine is my_crm_webhook.

<?php

declare(strict_types=1);

namespace Drupal\my_crm\Plugin\QueueWorker;

use Drupal\Core\Plugin\ContainerFactoryPluginInterface;
use Drupal\Core\Queue\Attribute\QueueWorker;
use Drupal\Core\Queue\QueueWorkerBase;
use Drupal\Core\Queue\SuspendQueueException;
use Drupal\Core\StringTranslation\TranslatableMarkup;
use GuzzleHttp\ClientInterface;
use GuzzleHttp\Exception\ConnectException;
use GuzzleHttp\Exception\RequestException;
use Psr\Log\LoggerInterface;
use Symfony\Component\DependencyInjection\ContainerInterface;

#[QueueWorker(
  id: 'my_crm_webhook',
  title: new TranslatableMarkup('CRM webhook delivery'),
  cron: ['time' => 30],
)]
final class CrmWebhookWorker extends QueueWorkerBase implements ContainerFactoryPluginInterface {

  public function __construct(
    array $configuration,
    $plugin_id,
    $plugin_definition,
    private readonly ClientInterface $httpClient,
    private readonly LoggerInterface $logger,
  ) {
    parent::__construct($configuration, $plugin_id, $plugin_definition);
  }

  public static function create(ContainerInterface $container, array $configuration, $plugin_id, $plugin_definition): static {
    return new static(
      $configuration,
      $plugin_id,
      $plugin_definition,
      $container->get('http_client'),
      $container->get('logger.factory')->get('my_crm'),
    );
  }

  public function processItem($data): void {
    try {
      $this->httpClient->request('POST', 'https://crm.example.com/hooks/article', [
        'json' => ['nid' => $data['nid']],
        'timeout' => 5,
      ]);
    }
    catch (ConnectException $e) {
      // The CRM is unreachable: stop hammering it for this run.
      throw new SuspendQueueException($e->getMessage(), 0, $e);
    }
    catch (RequestException $e) {
      $status = $e->getResponse()?->getStatusCode() ?? 0;
      if ($status === 429 || $status >= 500) {
        throw new SuspendQueueException($e->getMessage(), 0, $e);
      }
      // A 4xx will never succeed on retry: log it and drop the item.
      $this->logger->error('CRM rejected node @nid with HTTP @status.', [
        '@nid' => $data['nid'],
        '@status' => $status,
      ]);
    }
  }

}

What each outcome does

The interface docs spell out the contract. If processItem() returns normally, the item is deleted. If it throws an ordinary exception, cron logs it and leaves the item in the queue. A SuspendQueueException behaves the same way but additionally stops processing the rest of that queue in this run, which is exactly what you want when the remote side is down: forty queued items shouldn't produce forty timeouts. RequeueException puts the item back to be retried straight away, and DelayedRequeueException exists for queues that implement DelayableQueueInterface. I haven't needed either here.

My split is deliberate. Transport trouble and 429/5xx mean the item is fine but the world isn't, so suspend. A 4xx means the item itself is bad, so log and return; otherwise a poison item blocks the queue forever.

Gotchas

Delivery is at-least-once: an item can be processed and still be re-run if the process dies before deletion, so the CRM endpoint should be idempotent (I send the node ID, not "create a record"). Also remember cron only drains the queue as often as cron runs; if you need it faster, run drush queue:run my_crm_webhook from a real scheduler. And pass IDs, not entities: by the time the worker runs, the entity may have changed or been deleted, and loading it fresh is the whole point. I haven't run this exact snippet, but I traced it against the API docs above, and it is the shape I use in production.