#578676 follow-up by alex_b: Fix endless adding of non-processed aggregator feeds to queue

modules/aggregator/aggregator.install

@@ -143,6 +143,12 @@ function aggregator_schema() {
         'default' => 0,
         'description' => 'Last time feed was checked for new items, as Unix timestamp.',
       ),
+      'queued' => array(
+        'type' => 'int',
+        'not null' => TRUE,
+        'default' => 0,
+        'description' => 'Time when this feed was queued for refresh, 0 if not queued.',
+      ),
       'link' => array(
         'type' => 'varchar',
         'length' => 255,
@@ -195,6 +201,9 @@ function aggregator_schema() {
       'url'  => array('url'),
       'title' => array('title'),
     ),
+    'indexes' => array(
+      'queued' => array('queued'),
+    ),
   );
 
   $schema['aggregator_item'] = array(
@@ -275,3 +284,16 @@ function aggregator_update_7000() {
 function aggregator_update_7001() {
   variable_set('aggregator_teaser_length', variable_get('teaser_length'));
 }
+
+/**
+ * Add queued timestamp.
+ */
+function aggregator_update_7002() {
+  db_add_field('aggregator_feed', 'queued', array(
+    'type' => 'int',
+    'not null' => TRUE,
+    'default' => 0,
+    'description' => 'Time when this feed was queued for refresh, 0 if not queued.',
+  ));
+  db_add_index('aggregator_feed', 'queued', array('queued'));
+}

modules/aggregator/aggregator.module

@@ -319,14 +319,26 @@ function aggregator_permission() {
  * Queues news feeds for updates once their refresh interval has elapsed.
  */
 function aggregator_cron() {
-  $result = db_query('SELECT * FROM {aggregator_feed} WHERE checked + refresh < :time AND refresh != :never', array(
+  $result = db_query('SELECT * FROM {aggregator_feed} WHERE queued = 0 AND checked + refresh < :time AND refresh != :never', array(
     ':time' => REQUEST_TIME,
     ':never' => AGGREGATOR_CLEAR_NEVER
   ));
   $queue = DrupalQueue::get('aggregator_feeds');
   foreach ($result as $feed) {
-    $queue->createItem($feed);
+    if ($queue->createItem($feed)) {
+      // Add timestamp to avoid queueing item more than once.
+      db_update('aggregator_feed')
+        ->fields(array('queued' => REQUEST_TIME))
+        ->condition('fid', $feed->fid)
+        ->execute();
+    }
   }
+
+  // Remove queued timestamp after 6 hours assuming the update has failed.
+  db_update('aggregator_feed')
+    ->fields(array('queued' => 0))
+    ->condition('queued', REQUEST_TIME - (3600 * 6), '<')
+    ->execute();
 }
 
 /**
@@ -615,7 +627,6 @@ function aggregator_refresh($feed) {
         ->key(array('fid' => $feed->fid))
         ->fields(array(
           'url' => $feed->url,
-          'checked' => REQUEST_TIME,
           'link' => empty($feed->link) ? $feed->url : $feed->link,
           'description' => empty($feed->description) ? '' : $feed->description,
           'image' => empty($feed->image) ? '' : $feed->image,
@@ -642,14 +653,15 @@ function aggregator_refresh($feed) {
     }
   }
   else {
-    db_update('aggregator_feed')
-      ->fields(array('checked' => REQUEST_TIME))
-      ->condition('fid', $feed->fid)
-      ->execute();
-
     drupal_set_message(t('There is no new syndicated content from %site.', array('%site' => $feed->title)));
   }
 
+  // Regardless of successful or not, indicate that this feed has been checked.
+  db_update('aggregator_feed')
+    ->fields(array('checked' => REQUEST_TIME, 'queued' => 0))
+    ->condition('fid', $feed->fid)
+    ->execute();
+
   // Expire old feed items.
   if (function_exists('aggregator_expire')) {
     aggregator_expire($feed);

modules/aggregator/aggregator.test

@@ -638,3 +638,49 @@ class ImportOPMLTestCase extends AggregatorTestCase {
     $this->submitImportForm();
   }
 }
+
+class AggregatorCronTestCase extends AggregatorTestCase {
+  public static function getInfo() {
+    return array(
+      'name' => 'Update on cron functionality',
+      'description' => 'Update feeds on cron.',
+      'group' => 'Aggregator'
+    );
+  }
+
+  /**
+   * Add feeds update them on cron.
+   */
+  public function testCron() {
+    // Create feed and test basic updating on cron.
+    global $base_url;
+    $key = variable_get('cron_key', 'drupal');
+    $this->createSampleNodes();
+    $feed = $this->createFeed();
+    $this->drupalGet($base_url . '/cron.php', array('external' => TRUE, 'query' => array('cron_key' => $key)));
+    $this->assertEqual(5, db_query('SELECT COUNT(*) FROM {aggregator_item} WHERE fid = :fid', array(':fid' => $feed->fid))->fetchField(), 'Expected number of items in database.');
+    $this->removeFeedItems($feed);
+    $this->assertEqual(0, db_query('SELECT COUNT(*) FROM {aggregator_item} WHERE fid = :fid', array(':fid' => $feed->fid))->fetchField(), 'Expected number of items in database.');
+    $this->drupalGet($base_url . '/cron.php', array('external' => TRUE, 'query' => array('cron_key' => $key)));
+    $this->assertEqual(5, db_query('SELECT COUNT(*) FROM {aggregator_item} WHERE fid = :fid', array(':fid' => $feed->fid))->fetchField(), 'Expected number of items in database.');
+
+    // Test feed locking when queued for update.
+    $this->removeFeedItems($feed);
+    db_update('aggregator_feed')
+      ->condition('fid', $feed->fid)
+      ->fields(array(
+        'queued' => REQUEST_TIME,
+      ))
+      ->execute();
+    $this->drupalGet($base_url . '/cron.php', array('external' => TRUE, 'query' => array('cron_key' => $key)));
+    $this->assertEqual(0, db_query('SELECT COUNT(*) FROM {aggregator_item} WHERE fid = :fid', array(':fid' => $feed->fid))->fetchField(), 'Expected number of items in database.');
+    db_update('aggregator_feed')
+      ->condition('fid', $feed->fid)
+      ->fields(array(
+        'queued' => 0,
+      ))
+      ->execute();
+    $this->drupalGet($base_url . '/cron.php', array('external' => TRUE, 'query' => array('cron_key' => $key)));
+    $this->assertEqual(5, db_query('SELECT COUNT(*) FROM {aggregator_item} WHERE fid = :fid', array(':fid' => $feed->fid))->fetchField(), 'Expected number of items in database.');
+  }
+}