Index: modules/aggregator/aggregator.install
===================================================================
RCS file: /cvs/drupal/drupal/modules/aggregator/aggregator.install,v
retrieving revision 1.28
diff -u -p -r1.28 aggregator.install
--- modules/aggregator/aggregator.install	4 Dec 2009 16:49:45 -0000	1.28
+++ modules/aggregator/aggregator.install	30 Jan 2010 17:29:41 -0000
@@ -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 flag.
+ */
+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'));
+}
\ No newline at end of file
Index: modules/aggregator/aggregator.module
===================================================================
RCS file: /cvs/drupal/drupal/modules/aggregator/aggregator.module,v
retrieving revision 1.434
diff -u -p -r1.434 aggregator.module
--- modules/aggregator/aggregator.module	13 Jan 2010 14:35:41 -0000	1.434
+++ modules/aggregator/aggregator.module	30 Jan 2010 17:29:42 -0000
@@ -318,14 +318,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)) {
+      // Flag feed as queued to avoid queueing it more than once.
+      db_update('aggregator_feed')
+        ->fields(array('queued' => REQUEST_TIME))
+        ->condition('fid', $feed->fid)
+        ->execute();
+    }
   }
+
+  // Unflag queued items after 6 hours assuming that the update has failed.
+  db_update('aggregator_feed')
+    ->fields(array('queued' => 0))
+    ->condition('queued', REQUEST_TIME - (3600 * 6), '<')
+    ->execute();
 }
 
 /**
@@ -621,6 +633,7 @@ function aggregator_refresh($feed) {
           'hash' => $md5,
           'etag' => empty($feed->etag) ? '' : $feed->etag,
           'modified' => empty($feed->modified) ? 0 : $feed->modified,
+          'queued' => 0,
         ))
         ->execute();
 
@@ -642,7 +655,7 @@ function aggregator_refresh($feed) {
   }
   else {
     db_update('aggregator_feed')
-      ->fields(array('checked' => REQUEST_TIME))
+      ->fields(array('checked' => REQUEST_TIME, 'queued' => 0))
       ->condition('fid', $feed->fid)
       ->execute();