diff --git a/feeds.info b/feeds.info
index 8041cbf..118b2ac 100644
--- a/feeds.info
+++ b/feeds.info
@@ -46,6 +46,7 @@ files[] = tests/feeds_processor_user.test
 files[] = tests/feeds_scheduler.test
 files[] = tests/feeds_mapper_link.test
 files[] = tests/feeds_mapper_taxonomy.test
+files[] = tests/feeds_node_expiry.test
 files[] = tests/parser_csv.test
 
 ; Views integration
diff --git a/feeds.install b/feeds.install
index 2c42c00..728dc31 100644
--- a/feeds.install
+++ b/feeds.install
@@ -665,3 +665,14 @@ function feeds_update_7208(&$sandbox) {
     $sandbox['#finished'] = 1;
   }
 }
+
+/**
+ * Reschedules feeds jobs.
+ */
+function feeds_update_7209() {
+  variable_set('feeds_reschedule', TRUE);
+
+  db_delete('job_schedule')
+    ->condition('name', 'feeds_source_expire')
+    ->execute();
+}
diff --git a/feeds.module b/feeds.module
index b402823..456f076 100644
--- a/feeds.module
+++ b/feeds.module
@@ -42,12 +42,14 @@ function feeds_hook_info() {
  */
 function feeds_cron() {
   if ($importers = feeds_reschedule()) {
-    foreach ($importers as $id) {
-      feeds_importer($id)->schedule();
-      $rows = db_query("SELECT feed_nid FROM {feeds_source} WHERE id = :id", array(':id' => $id));
-      foreach ($rows as $row) {
-        feeds_source($id, $row->feed_nid)->schedule();
-      }
+    if ($importers === TRUE) {
+      $sources = db_query("SELECT feed_nid, id FROM {feeds_source}");
+    }
+    else {
+      $sources = db_query("SELECT feed_nid, id FROM {feeds_source} WHERE id IN (:ids)", array(':ids' => $importers));
+    }
+    foreach ($sources as $source) {
+      feeds_source($source->id, $source->feed_nid)->schedule();
     }
     feeds_reschedule(FALSE);
   }
@@ -70,8 +72,8 @@ function feeds_cron_job_scheduler_info() {
   $info['feeds_source_clear'] = array(
     'queue name' => 'feeds_source_clear',
   );
-  $info['feeds_importer_expire'] = array(
-    'queue name' => 'feeds_importer_expire',
+  $info['feeds_source_expire'] = array(
+    'queue name' => 'feeds_source_expire',
   );
   $info['feeds_push_unsubscribe'] = array(
     'queue name' => 'feeds_push_unsubscribe',
@@ -92,8 +94,8 @@ function feeds_cron_queue_info() {
     'worker callback' => 'feeds_source_clear',
     'time' => 15,
   );
-  $queues['feeds_importer_expire'] = array(
-    'worker callback' => 'feeds_importer_expire',
+  $queues['feeds_source_expire'] = array(
+    'worker callback' => 'feeds_source_expire',
     'time' => 15,
   );
   $queues['feeds_push_unsubscribe'] = array(
@@ -140,18 +142,18 @@ function feeds_source_clear($job) {
 /**
  * Scheduler callback for expiring content.
  */
-function feeds_importer_expire($job) {
-  $importer = feeds_importer($job['type']);
+function feeds_source_expire($job) {
+  $source = feeds_source($job['type'], $job['id']);
   try {
-    $importer->existing()->expire();
+    $source->existing()->expire();
   }
   catch (FeedsNotExistingException $e) {
     // Do nothing.
   }
   catch (Exception $e) {
-    $importer->log('expire', $e->getMessage(), array(), WATCHDOG_ERROR);
+    $source->log('expire', $e->getMessage(), array(), WATCHDOG_ERROR);
   }
-  $importer->scheduleExpire();
+  $source->scheduleExpire();
 }
 
 /**
@@ -612,9 +614,8 @@ function feeds_node_insert($node) {
     if (feeds_importer($importer_id)->config['import_on_create'] && !isset($node->feeds['suppress_import'])) {
       $source->startImport();
     }
-    // Schedule source and importer.
+    // Schedule the source.
     $source->schedule();
-    feeds_importer($importer_id)->schedule();
   }
 }
 
diff --git a/feeds.pages.inc b/feeds.pages.inc
index f976d9c..7e1a164 100644
--- a/feeds.pages.inc
+++ b/feeds.pages.inc
@@ -111,7 +111,6 @@ function feeds_import_form_submit($form, &$form_state) {
 
   // Add to schedule, make sure importer is scheduled, too.
   $source->schedule();
-  $source->importer->schedule();
 }
 
 /**
diff --git a/includes/FeedsImporter.inc b/includes/FeedsImporter.inc
index 011a7de..31be6b2 100644
--- a/includes/FeedsImporter.inc
+++ b/includes/FeedsImporter.inc
@@ -55,49 +55,6 @@ class FeedsImporter extends FeedsConfigurable {
   }
 
   /**
-   * Remove items older than $time.
-   *
-   * @param $time
-   *   All items older than REQUEST_TIME - $time will be deleted. If not
-   *   given, internal processor settings will be used.
-   *
-   * @return
-   *   FEEDS_BATCH_COMPLETE if the expiry process finished. A decimal between
-   *   0.0 and 0.9 periodic if expiry is still in progress.
-   *
-   * @throws
-   *   Throws Exception if an error occurs when expiring items.
-   */
-  public function expire($time = NULL) {
-    return $this->processor->expire($time);
-  }
-
-  /**
-   * Schedule all periodic tasks for this importer.
-   */
-  public function schedule() {
-    $this->scheduleExpire();
-  }
-
-  /**
-   * Schedule expiry of items.
-   */
-  public function scheduleExpire() {
-    $job = array(
-      'type' => $this->id,
-      'period' => 0,
-      'periodic' => TRUE,
-    );
-    if (FEEDS_EXPIRE_NEVER != $this->processor->expiryTime()) {
-      $job['period'] = 3600;
-      JobScheduler::get('feeds_importer_expire')->set($job);
-    }
-    else {
-      JobScheduler::get('feeds_importer_expire')->remove($job);
-    }
-  }
-
-  /**
    * Report how many items *should* be created on one page load by this
    * importer.
    *
@@ -155,23 +112,17 @@ class FeedsImporter extends FeedsConfigurable {
   }
 
   /**
-   * Delete configuration. Removes configuration information
-   * from database, does not delete configuration itself.
+   * Deletes configuration.
+   *
+   * Removes configuration information from database, does not delete
+   * configuration itself.
    */
   public function delete() {
     db_delete('feeds_importer')
       ->condition('id', $this->id)
       ->execute();
-    $job = array(
-      'type' => $this->id,
-      'id' => 0,
-    );
-    if ($this->export_type & EXPORT_IN_CODE) {
-      feeds_reschedule($this->id);
-    }
-    else {
-      JobScheduler::get('feeds_importer_expire')->remove($job);
-    }
+
+    feeds_reschedule($this->id);
   }
 
   /**
diff --git a/includes/FeedsSource.inc b/includes/FeedsSource.inc
index 98cd0f1..2158074 100644
--- a/includes/FeedsSource.inc
+++ b/includes/FeedsSource.inc
@@ -18,6 +18,7 @@ define('FEEDS_FETCH', 'fetch');
 define('FEEDS_PARSE', 'parse');
 define('FEEDS_PROCESS', 'process');
 define('FEEDS_PROCESS_CLEAR', 'process_clear');
+define('FEEDS_PROCESS_EXPIRE', 'process_expire');
 
 /**
  * Declares an interface for a class that defines default values and form
@@ -274,6 +275,7 @@ class FeedsSource extends FeedsConfigurable {
    */
   public function schedule() {
     $this->scheduleImport();
+    $this->scheduleExpire();
   }
 
   /**
@@ -303,6 +305,27 @@ class FeedsSource extends FeedsConfigurable {
   }
 
   /**
+   * Schedule background expire tasks.
+   */
+  public function scheduleExpire() {
+    // Schedule as soon as possible if a batch is active.
+    $period = $this->progressExpiring() === FEEDS_BATCH_COMPLETE ? 3600 : 0;
+
+    $job = array(
+      'type' => $this->id,
+      'id' => $this->feed_nid,
+      'period' => $period,
+      'periodic' => TRUE,
+    );
+    if ($this->importer->processor->expiryTime() == FEEDS_EXPIRE_NEVER) {
+      JobScheduler::get('feeds_source_expire')->remove($job);
+    }
+    else {
+      JobScheduler::get('feeds_source_expire')->set($job);
+    }
+  }
+
+  /**
    * Schedule background clearing tasks.
    */
   public function scheduleClear() {
@@ -417,6 +440,26 @@ class FeedsSource extends FeedsConfigurable {
   }
 
   /**
+   * Removes all expired items from a feed.
+   */
+  public function expire() {
+    $this->acquireLock();
+    try {
+      $result = $this->importer->processor->expire($this);
+    }
+    catch (Exception $e) {
+      // Will throw after the lock is released.
+    }
+    $this->releaseLock();
+
+    if (isset($e)) {
+      throw $e;
+    }
+
+    return $result;
+  }
+
+  /**
    * Report progress as float between 0 and 1. 1 = FEEDS_BATCH_COMPLETE.
    */
   public function progressParsing() {
@@ -451,6 +494,13 @@ class FeedsSource extends FeedsConfigurable {
   }
 
   /**
+   * Report progress on expiry.
+   */
+  public function progressExpiring() {
+    return $this->state(FEEDS_PROCESS_EXPIRE)->progress;
+  }
+
+  /**
    * Return a state object for a given stage. Lazy instantiates new states.
    *
    * @todo Rename getConfigFor() accordingly to config().
@@ -556,6 +606,7 @@ class FeedsSource extends FeedsConfigurable {
       'id' => $this->feed_nid,
     );
     JobScheduler::get('feeds_source_import')->remove($job);
+    JobScheduler::get('feeds_source_expire')->remove($job);
   }
 
   /**
@@ -726,4 +777,5 @@ class FeedsSource extends FeedsConfigurable {
   protected function releaseLock() {
     lock_release("feeds_source_{$this->id}_{$this->feed_nid}");
   }
+
 }
diff --git a/plugins/FeedsNodeProcessor.inc b/plugins/FeedsNodeProcessor.inc
index f2db596..915b893 100644
--- a/plugins/FeedsNodeProcessor.inc
+++ b/plugins/FeedsNodeProcessor.inc
@@ -134,28 +134,12 @@ class FeedsNodeProcessor extends FeedsProcessor {
   }
 
   /**
-   * Implement expire().
-   *
-   * @todo: move to processor stage?
+   * Overrides parent::expiryQuery().
    */
-  public function expire($time = NULL) {
-    if ($time === NULL) {
-      $time = $this->expiryTime();
-    }
-    if ($time == FEEDS_EXPIRE_NEVER) {
-      return;
-    }
-    $count = $this->getLimit();
-    $nodes = db_query_range("SELECT n.nid FROM {node} n JOIN {feeds_item} fi ON fi.entity_type = 'node' AND n.nid = fi.entity_id WHERE fi.id = :id AND n.created < :created", 0, $count, array(':id' => $this->id, ':created' => REQUEST_TIME - $time));
-    $nids = array();
-    foreach ($nodes as $node) {
-      $nids[$node->nid] = $node->nid;
-    }
-    $this->entityDeleteMultiple($nids);
-    if (db_query_range("SELECT 1 FROM {node} n JOIN {feeds_item} fi ON fi.entity_type = 'node' AND n.nid = fi.entity_id WHERE fi.id = :id AND n.created < :created", 0, 1, array(':id' => $this->id, ':created' => REQUEST_TIME - $time))->fetchField()) {
-      return FEEDS_BATCH_ACTIVE;
-    }
-    return FEEDS_BATCH_COMPLETE;
+  protected function expiryQuery(FeedsSource $source, $time) {
+    $select = parent::expiryQuery($source, $time);
+    $select->condition('e.created', REQUEST_TIME - $time, '<');
+    return $select;
   }
 
   /**
@@ -380,4 +364,5 @@ class FeedsNodeProcessor extends FeedsProcessor {
     }
     return 0;
   }
+
 }
diff --git a/plugins/FeedsProcessor.inc b/plugins/FeedsProcessor.inc
index bf8c3fb..e495e65 100755
--- a/plugins/FeedsProcessor.inc
+++ b/plugins/FeedsProcessor.inc
@@ -397,23 +397,87 @@ abstract class FeedsProcessor extends FeedsPlugin {
   }
 
   /**
-   * Delete feed items younger than now - $time. Do not invoke expire on a
-   * processor directly, but use FeedsImporter::expire() instead.
+   * Deletes feed items older than REQUEST_TIME - $time.
    *
-   * @see FeedsImporter::expire().
-   * @see FeedsDataProcessor::expire().
+   * Do not invoke expire on a processor directly, but use
+   * FeedsSource::expire() instead.
+   *
+   * @param FeedsSource $source
+   *   The source to expire entities for.
    *
    * @param $time
-   *   If implemented, all items produced by this configuration that are older
-   *   than REQUEST_TIME - $time should be deleted.
-   *   If $time === NULL processor should use internal configuration.
+   *   (optional) All items produced by this configuration that are older than
+   *   REQUEST_TIME - $time should be deleted. If NULL, processor should use
+   *   internal configuration. Defaults to NULL.
    *
-   * @return
+   * @return float
    *   FEEDS_BATCH_COMPLETE if all items have been processed, a float between 0
    *   and 0.99* indicating progress otherwise.
+   *
+   * @see FeedsSource::expire()
    */
-  public function expire($time = NULL) {
-    return FEEDS_BATCH_COMPLETE;
+  public function expire(FeedsSource $source, $time = NULL) {
+    $state = $source->state(FEEDS_PROCESS_EXPIRE);
+
+    if ($time === NULL) {
+      $time = $this->expiryTime();
+    }
+    if ($time == FEEDS_EXPIRE_NEVER) {
+      return;
+    }
+
+    $select = $this->expiryQuery($source, $time);
+
+    // If there is no total, query it.
+    if (!$state->total) {
+      $state->total = $select->countQuery()->execute()->fetchField();
+    }
+
+    // Delete a batch of entities.
+    $entity_ids = $select->range(0, $this->getLimit())->execute()->fetchCol();
+    if ($entity_ids) {
+      $this->entityDeleteMultiple($entity_ids);
+      $state->deleted += count($entity_ids);
+      $state->progress($state->total, $state->deleted);
+    }
+    else {
+      $state->progress($state->total, $state->total);
+    }
+  }
+
+  /**
+   * Returns a database query used to select entities to expire.
+   *
+   * Processor classes should override this method to set the age portion of the
+   * query.
+   *
+   * @param FeedsSource $source
+   *   The feed source.
+   * @param int $time
+   *   Delete entities older than this.
+   *
+   * @return SelectQuery
+   *   A select query to execute.
+   *
+   * @see FeedsNodeProcessor::expiryQuery()
+   */
+  protected function expiryQuery(FeedsSource $source, $time) {
+    // Build base select statement.
+    $info = $this->entityInfo();
+    $id_key = db_escape_field($info['entity keys']['id']);
+
+    $select = db_select($info['base table'], 'e');
+    $select->addField('e', $info['entity keys']['id'], 'entity_id');
+    $select->join(
+      'feeds_item',
+      'fi',
+      "e.$id_key = fi.entity_id AND fi.entity_type = :entity_type", array(
+        ':entity_type' => $this->entityType(),
+    ));
+    $select->condition('fi.id', $this->id);
+    $select->condition('fi.feed_nid', $source->feed_nid);
+
+    return $select;
   }
 
   /**
diff --git a/tests/feeds.test b/tests/feeds.test
index a2bac29..1a98190 100644
--- a/tests/feeds.test
+++ b/tests/feeds.test
@@ -346,8 +346,13 @@ class FeedsWebTestCase extends DrupalWebTestCase {
 
     // Check whether feed got properly added to scheduler.
     $this->assertEqual(1, db_query("SELECT COUNT(*) FROM {job_schedule} WHERE type = :id AND id = 0 AND name = 'feeds_source_import' AND last <> 0 AND scheduled = 0", array(':id' => $id))->fetchField());
-    // There must be only one entry for callback 'expire' - no matter what the feed_nid is.
-    $this->assertEqual(0, db_query("SELECT COUNT(*) FROM {job_schedule} WHERE type = :id AND name = 'feeds_importer_expire' AND last <> 0 AND scheduled = 0", array(':id' => $id))->fetchField());
+    // Check expire scheduler.
+    if (feeds_importer($id)->processor->expiryTime() == FEEDS_EXPIRE_NEVER) {
+      $this->assertEqual(0, db_query("SELECT COUNT(*) FROM {job_schedule} WHERE type = :id AND id = 0 AND name = 'feeds_source_expire'", array(':id' => $id))->fetchField());
+    }
+    else {
+      $this->assertEqual(1, db_query("SELECT COUNT(*) FROM {job_schedule} WHERE type = :id AND id = 0 AND name = 'feeds_source_expire'", array(':id' => $id))->fetchField());
+    }
   }
 
   /**
diff --git a/tests/feeds_processor_node.test b/tests/feeds_processor_node.test
index a61df78..e5d4ecd 100644
--- a/tests/feeds_processor_node.test
+++ b/tests/feeds_processor_node.test
@@ -455,4 +455,48 @@ class FeedsRSStoNodesTest extends FeedsWebTestCase {
     $node_count = db_query("SELECT COUNT(*) FROM {node}")->fetchField();
     $this->assertEqual($node_count, 11, t('Correct number of nodes in the database.'));
   }
+
+  /**
+   * Tests expiring nodes.
+   */
+  public function testExpiry() {
+    // Create importer configuration.
+    $this->setSettings('syndication', NULL, array('content_type' => ''));
+    $this->setSettings('syndication', 'FeedsNodeProcessor', array(
+      'expire' => 2592000,
+    ));
+
+    // Create importer.
+    $this->importURL('syndication');
+
+    // Set date of a few nodes to current date so they don't expire.
+    $edit = array(
+      'date' => date('Y-m-d'),
+    );
+    $this->drupalPost('node/2/edit', $edit, 'Save');
+    $this->assertText(date('m/d/Y'), 'Found correct date.');
+    $this->drupalPost('node/5/edit', $edit, 'Save');
+    $this->assertText(date('m/d/Y'), 'Found correct date.');
+
+    // Run cron to schedule jobs.
+    $this->cronRun();
+
+    // Set feeds source expire to run immediately.
+    db_update('job_schedule')
+      ->fields(array(
+        'next' => 0,
+      ))
+      ->condition('name', 'feeds_source_expire')
+      ->execute();
+
+    // Run cron to execute scheduled jobs.
+    $this->cronRun();
+
+    // Query the feeds_items table and count the number of entries.
+    $row_count = db_query('SELECT COUNT(*) FROM {feeds_item}')->fetchField();
+
+    // Check that number of feeds items is equal to the expected items.
+    $this->assertEqual($row_count, 2, 'Nodes expired.');
+  }
+
 }
diff --git a/tests/feeds_scheduler.test b/tests/feeds_scheduler.test
index 861e8c9..afab3db 100644
--- a/tests/feeds_scheduler.test
+++ b/tests/feeds_scheduler.test
@@ -151,14 +151,14 @@ class FeedsSchedulerTestCase extends FeedsWebTestCase {
     // Set expire settings, check rescheduling.
     $max_last = db_query("SELECT MAX(last) FROM {job_schedule} WHERE type = 'syndication' AND name = 'feeds_source_import' AND period = 0")->fetchField();
     $min_last = db_query("SELECT MIN(last) FROM {job_schedule} WHERE type = 'syndication' AND name = 'feeds_source_import' AND period = 0")->fetchField();
-    $this->assertEqual(0, db_query("SELECT COUNT(*) FROM {job_schedule} WHERE type = 'syndication' AND name = 'feeds_importer_expire' AND last <> 0 AND scheduled = 0")->fetchField());
+    $this->assertEqual(0, db_query("SELECT COUNT(*) FROM {job_schedule} WHERE type = 'syndication' AND name = 'feeds_source_expire'")->fetchField());
     $this->drupalLogin($this->admin_user);
     $this->setSettings('syndication', 'FeedsNodeProcessor', array('expire' => 86400));
     $this->drupalLogout();
     sleep(1);
     $this->cronRun();
-    // There should be a feeds_importer_expire job now, and all last fields should be reset.
-    $this->assertEqual(1, db_query("SELECT COUNT(*) FROM {job_schedule} WHERE type = 'syndication' AND name = 'feeds_importer_expire' AND last <> 0 AND scheduled = 0 AND period = 3600")->fetchField());
+    // There should be 20 feeds_source_expire jobs now, and all last fields should be reset.
+    $this->assertEqual(count($nids), db_query("SELECT COUNT(*) FROM {job_schedule} WHERE type = 'syndication' AND name = 'feeds_source_expire' AND last <> 0 AND scheduled = 0 AND period = 3600")->fetchField());
     $new_max_last = db_query("SELECT MAX(last) FROM {job_schedule} WHERE type = 'syndication' AND name = 'feeds_source_import' AND period = 0")->fetchField();
     $new_min_last = db_query("SELECT MIN(last) FROM {job_schedule} WHERE type = 'syndication' AND name = 'feeds_source_import' AND period = 0")->fetchField();
     $this->assertNotEqual($new_max_last, $max_last);
@@ -179,18 +179,19 @@ class FeedsSchedulerTestCase extends FeedsWebTestCase {
     $this->assertNotEqual($new_min_last, $min_last);
     $this->assertEqual($new_max_last, $new_min_last);
     $this->assertEqual(0, db_query("SELECT COUNT(*) FROM {job_schedule} WHERE type = 'syndication' AND name = 'feeds_source_import' AND period <> 3600")->fetchField());
-    $this->assertEqual(1, db_query("SELECT COUNT(*) FROM {job_schedule} WHERE type = 'syndication' AND name = 'feeds_importer_expire' AND period = 3600 AND last = :last", array(':last' => $new_min_last))->fetchField());
+    $this->assertEqual(count($nids), db_query("SELECT COUNT(*) FROM {job_schedule} WHERE type = 'syndication' AND name = 'feeds_source_expire' AND period = 3600 AND last = :last", array(':last' => $new_min_last))->fetchField());
 
     // Delete source, delete importer, check schedule.
     $this->drupalLogin($this->admin_user);
     $nid = array_shift($nids);
     $this->drupalPost("node/$nid/delete", array(), t('Delete'));
     $this->assertEqual(0, db_query("SELECT COUNT(*) FROM {job_schedule} WHERE type = 'syndication' AND name = 'feeds_source_import' AND id = :nid", array(':nid' => $nid))->fetchField());
+    $this->assertEqual(0, db_query("SELECT COUNT(*) FROM {job_schedule} WHERE type = 'syndication' AND name = 'feeds_source_expire' AND id = :nid", array(':nid' => $nid))->fetchField());
     $this->assertEqual(count($nids), db_query("SELECT COUNT(*) FROM {job_schedule} WHERE type = 'syndication' AND name = 'feeds_source_import'")->fetchField());
-    $this->assertEqual(1, db_query("SELECT COUNT(*) FROM {job_schedule} WHERE type = 'syndication' AND name = 'feeds_importer_expire' AND id = 0")->fetchField());
+    $this->assertEqual(count($nids), db_query("SELECT COUNT(*) FROM {job_schedule} WHERE type = 'syndication' AND name = 'feeds_source_expire'")->fetchField());
 
     $this->drupalPost('admin/structure/feeds/syndication/delete', array(), t('Delete'));
-    $this->assertEqual(0, db_query("SELECT COUNT(*) FROM {job_schedule} WHERE type = 'syndication' AND name = 'feeds_importer_expire' AND id = 0")->fetchField());
+    $this->assertEqual(count($nids), db_query("SELECT COUNT(*) FROM {job_schedule} WHERE type = 'syndication' AND name = 'feeds_source_expire'")->fetchField());
     $this->assertEqual(count($nids), db_query("SELECT COUNT(*) FROM {job_schedule} WHERE type = 'syndication' AND name = 'feeds_source_import'")->fetchField());
   }