I tested my hypothesis above and it was correct. The queue isn't fully processed during a cron run, so there are stale node objects sitting in the queue
table in the database. The solution was to reload the node in the worker callback.
Original Code
/**
* Implements hook_cron().
*/
function ih_transcript_import_cron() {
// Load all Video nodes.
$videos = ih_transcript_import_load_video_nodes();
$queue = DrupalQueue::get('ih_video_transcripts');
foreach($videos as $video) {
$queue->createItem($video);
}
}
/**
* Implements hook_cron_queue_info().
*/
function ih_transcript_import_cron_queue_info() {
$queues['ih_video_transcripts'] = array(
'worker callback' => 'ih_transcript_import_import_transcript',
'time' => '60',
);
return $queues;
}
/**
* Implements callback_queue_worker().
*/
function ih_transcript_import_import_transcript($video_node) {
// THIS IS THE PROBLEMATIC PART, $video_node is stale since it is
// pulled from the queue table of the database and there may have been
// multiple cron runs since it was created (2 days in my case).
$video_node->field_transcript_text[$video_node->language][0]['value'] = "My value";
node_save($video_node);
}
The Fix
/**
* Implements callback_queue_worker().
*/
function ih_transcript_import_import_transcript($video_node) {
// THIS IS THE IMPORTANT FIX, reload the video node to ensure
// we are saving the most up to date data.
$video_node = node_load($video_node->nid);
$video_node->field_transcript_text[$video_node->language][0]['value'] = $transcript_text->data;
node_save($video_node);
}
Better Fix
Save ONLY the node Id in the queue rather than the entire node. Also, I added some code to ensure that there are no duplicates in the queue. My queue table had 64000 records!!
/**
* Implements hook_cron().
*/
function ih_transcript_import_cron() {
// Load all Video nodes.
$videos = ih_transcript_import_load_video_nodes();
$currentQueueItems = array();
$claimedItems = array();
$queue = DrupalQueue::get('ih_video_transcripts');
// We only want to add an item to a queue if it doesn't already exist in the
// queue to avoid the queue table becoming massive. Retrieve a list of
// node ids that already exist in the queue.
while($item = $queue->claimItem()) {
$currentQueueItems[] = $item->data;
// Store a copy of the claimed items, so we can release them again. We
// only claim them in order to grab the node Id, so need to release again
// to allow processing.
$claimedItems[] = $item;
}
// Release claimed queue items to allow them to be processed by other workers.
foreach($claimedItems as $claimedItem) {
$queue->releaseItem($claimedItem);
}
foreach($videos as $video) {
// Don't add videos that already exist in the queue.
if(!in_array($video->nid, $currentQueueItems)) {
$queue->createItem($video->nid);
}
}
}
/**
* Implements hook_cron_queue_info().
*/
function ih_transcript_import_cron_queue_info() {
$queues['ih_video_transcripts'] = array(
'worker callback' => 'ih_transcript_import_import_transcript',
'time' => '60',
);
return $queues;
}
/**
* Implements callback_queue_worker(). Can also be called independently of cron
* to import a single transcript.
*/
function ih_transcript_import_import_transcript($video_nid) {
$video_node = node_load($video_nid);
$video_node->field_transcript_text[$video_node->language][0]['value'] = $transcript_text->data;
node_save($video_node);
}
$node->revision = TRUE
must be set when using node_save. Turning on revisions is a recent thing and I didn't update my code with$node->revision = TRUE