Job queue – dealing with stuck records

Question

I have a cron job that gets stuck because I have a loop claiming items from the queue that ends on numberOfItems() == 0, which never happens because one of the queue items is stuck and cannot be claimed.

Clearly this is a flaw on my side, but I am trying to understand how to best address it. The claimItem() method returns FALSE every time in this case, so it should be easy to handle cases of bad queue processing, for example:

while ($item = $queue->claimItem()) {
    // process item
}

However, what if the first queue item is "faulty", while there are other queue items that are perfectly fine? I will never be able to retrieve them in this method. It also seems impossible to move on to the next item in the queue if claimItem() returns false (please correct me if I am wrong).

What is the correct way to handle such cases with Drupal's queue?

Answer 1

When looking at the code from the claimItem method it only returns FALSE if no items are available to claim.

So if $queue->claimItem() returns FALSE you can be sure that all available items are claimed (or failed to be claimed).

public function claimItem($lease_time = 30) {
  // Claim an item by updating its expire fields. If claim is not successful
  // another thread may have claimed the item in the meantime. Therefore loop
  // until an item is successfully claimed or we are reasonably sure there
  // are no unclaimed items left.
  while (TRUE) {
    try {
      $item = $this->connection->queryRange('SELECT data, created, item_id FROM {' . static::TABLE_NAME . '} q WHERE expire = 0 AND name = :name ORDER BY created, item_id ASC', 0, 1, [':name' => $this->name])->fetchObject();
    }
    catch (\Exception $e) {
      $this->catchException($e);
      // If the table does not exist there are no items currently available to
      // claim.
      return FALSE;
    }
    if ($item) {
      // Try to update the item. Only one thread can succeed in UPDATEing the
      // same row. We cannot rely on REQUEST_TIME because items might be
      // claimed by a single consumer which runs longer than 1 second. If we
      // continue to use REQUEST_TIME instead of the current time(), we steal
      // time from the lease, and will tend to reset items before the lease
      // should really expire.
      $update = $this->connection->update(static::TABLE_NAME)
        ->fields([
          'expire' => time() + $lease_time,
        ])
        ->condition('item_id', $item->item_id)
        ->condition('expire', 0);
      // If there are affected rows, this update succeeded.
      if ($update->execute()) {
        $item->data = unserialize($item->data);
        return $item;
      }
    }
    else {
      // No items currently available to claim.
      return FALSE;
    }
  }
}