Fix BatchIterator skipping rows when subscribers' processed status is modified [MAILPOET-987]

lib/Tasks/Subscribers/BatchIterator.php

@@ -9,7 +9,8 @@ if(!defined('ABSPATH')) exit;
 class BatchIterator implements \Iterator, \Countable {
   private $task_id;
   private $batch_size;
-  private $offset = 0;
+  private $last_processed_id = 0;
+  private $batch_last_id;
 
   function __construct($task_id, $batch_size) {
     if($task_id <= 0) {
@@ -22,29 +23,29 @@ class BatchIterator implements \Iterator, \Countable {
   }
 
   function rewind() {
-    $this->offset = 0;
+    $this->last_processed_id = 0;
   }
 
   function current() {
     $subscribers = $this->getSubscribers()
       ->orderByAsc('subscriber_id')
       ->limit($this->batch_size)
-      ->offset($this->offset)
       ->findArray();
     $subscribers = Helpers::arrayColumn($subscribers, 'subscriber_id');
+    $this->batch_last_id = end($subscribers);
     return $subscribers;
   }
 
   function key() {
-    return $this->offset;
+    return null;
   }
 
   function next() {
-    $this->offset += $this->batch_size;
+    $this->last_processed_id = $this->batch_last_id;
   }
 
   function valid() {
-    return $this->offset < $this->count();
+    return $this->count() > 0;
   }
 
   function count() {
@@ -54,6 +55,7 @@ class BatchIterator implements \Iterator, \Countable {
   private function getSubscribers() {
     return ScheduledTaskSubscriber::select('subscriber_id')
       ->where('task_id', $this->task_id)
+      ->whereGt('subscriber_id', $this->last_processed_id)
       ->where('processed', ScheduledTaskSubscriber::STATUS_UNPROCESSED);
   }
 }

tests/unit/Tasks/Subscribers/BatchIteratorTest.php

@@ -6,9 +6,9 @@ use MailPoet\Tasks\Subscribers\BatchIterator;
 
 class BatchIteratorTest extends \MailPoetTest {
   function _before() {
-    $this->task_id = 123;
+    $this->task_id = 123; // random ID
     $this->batch_size = 2;
-    $this->subscriber_count = 5;
+    $this->subscriber_count = 10;
     for($i = 0; $i < $this->subscriber_count; $i++) {
       ScheduledTaskSubscriber::createOrUpdate(array(
         'task_id' => $this->task_id,
@@ -28,7 +28,7 @@ class BatchIteratorTest extends \MailPoetTest {
   }
 
   function testItConstructs() {
-    $iterator = new BatchIterator(123, 456);
+    $iterator = new BatchIterator(123, 456); // random IDs
     expect_that($iterator instanceof BatchIterator);
   }
 
@@ -37,6 +37,14 @@ class BatchIteratorTest extends \MailPoetTest {
     $i = 0;
     foreach($this->iterator as $batch) {
       $i++;
+
+      // process subscribers
+      ScheduledTaskSubscriber::where('task_id', $this->task_id)
+        ->whereIn('subscriber_id', $batch)
+        ->findResultSet()
+        ->set('processed', ScheduledTaskSubscriber::STATUS_PROCESSED)
+        ->save();
+
       if($i < $iterations) {
         expect(count($batch))->equals($this->batch_size);
       } else {