Fix processing the same item twice for interrupted jobs (#13)

Fixes #11.
Fixes #12.

Author: fatkodima

CHANGELOG.md

@@ -1,5 +1,7 @@
 ## master (unreleased)
 
+- Fix processing the same item twice for interrupted jobs
+
 ## 0.4.0 (2024-05-10)
 
 - Support ordering using multiple directions for ActiveRecord enumerators

lib/sidekiq_iteration/active_record_enumerator.rb

@@ -78,7 +78,7 @@ def batches
       Enumerator.new(-> { records_size }) do |yielder|
         while (batch = next_batch(load: true))
           increment_iteration
-          yielder.yield(batch, cursor_value(batch.last))
+          yielder.yield(batch, cursor_value(batch.first))
         end
       end
     end

lib/sidekiq_iteration/iteration.rb

@@ -183,18 +183,19 @@ def iterate_with_enumerator(enumerator, arguments)
 
         enumerator.each do |object_from_enumerator, index|
           found_record = true
-          around_iteration do
-            each_iteration(object_from_enumerator, *arguments)
-          end
           @cursor_position = index
-          @current_run_iterations += 1
 
           throttle_condition = find_throttle_condition
           if throttle_condition
             @job_iteration_retry_backoff = throttle_condition.backoff
             @needs_reenqueue = true
             return false
           end
+
+          around_iteration do
+            each_iteration(object_from_enumerator, *arguments)
+          end
+          @current_run_iterations += 1
         end
 
         unless found_record

lib/sidekiq_iteration/nested_enumerator.rb

@@ -33,6 +33,9 @@ def iterate(current_items, current_cursor, index, &block)
           else
             iterate(current_items + [item], current_cursor + [cursor_value], index + 1, &block)
           end
+
+          # Nullify downstream enums cursors on each iteration.
+          @cursor.fill(nil, index + 1)
         end
       end
   end

test/active_record_enumerator_test.rb

@@ -40,14 +40,14 @@ class ActiveRecordEnumeratorTest < TestCase
       end
     end
 
-    test "#batches yields batches of records with the last record's cursor position" do
+    test "#batches yields batches of records with the first record's cursor position" do
       enum = build_enumerator.batches
       assert_equal(10, enum.size)
 
       products = Product.order(:id).take(4).each_slice(2).to_a
 
       enum.first(2).each_with_index do |(element, cursor), index|
-        assert_equal([products[index], products[index].last.id], [element, cursor])
+        assert_equal([products[index], products[index].first.id], [element, cursor])
       end
     end
 
@@ -101,7 +101,7 @@ class ActiveRecordEnumeratorTest < TestCase
       enum = build_enumerator(batch_size: 4).batches
       products = Product.order(:id).take(4)
 
-      assert_equal([products, products.last.id], enum.first)
+      assert_equal([products, products.first.id], enum.first)
     end
 
     test "paginates using integer conditionals for primary key when no columns are defined" do
@@ -116,14 +116,14 @@ class ActiveRecordEnumeratorTest < TestCase
       enum = build_enumerator(columns: [:id]).batches
       products = Product.order(:id).take(2)
 
-      assert_equal([products, products.last.id], enum.first)
+      assert_equal([products, products.first.id], enum.first)
     end
 
     test "single column as symbol" do
       enum = build_enumerator(columns: :id).batches
       products = Product.order(:id).take(2)
 
-      assert_equal([products, products.last.id], enum.first)
+      assert_equal([products, products.first.id], enum.first)
     end
 
     test "raises when columns does not include primary key" do
@@ -142,7 +142,7 @@ class ActiveRecordEnumeratorTest < TestCase
       enum = build_enumerator(columns: [:updated_at, :id]).batches
       products = Product.order(:updated_at, :id).take(2)
 
-      assert_equal([products, [products.last.updated_at.strftime(SQL_TIME_FORMAT), products.last.id]], enum.first)
+      assert_equal([products, [products.first.updated_at.strftime(SQL_TIME_FORMAT), products.first.id]], enum.first)
     end
 
     test "columns configured with primary key only queries primary key column once" do
@@ -163,7 +163,7 @@ class ActiveRecordEnumeratorTest < TestCase
 
     test ":order with single direction" do
       enum = build_enumerator(order: :desc).batches
-      product_batches = Product.order(id: :desc).take(4).in_groups_of(2).map { |products| [products, products.last.id] }
+      product_batches = Product.order(id: :desc).take(4).in_groups_of(2).map { |products| [products, products.first.id] }
 
       enum.first(2).each_with_index do |(batch, cursor), index|
         assert_equal(product_batches[index], [batch, cursor])
@@ -203,24 +203,22 @@ class ActiveRecordEnumeratorTest < TestCase
       enum = build_enumerator(cursor: one.id).batches
 
       queries = track_queries do
-        assert_equal([[one, two], two.id], enum.first)
+        assert_equal([[one, two], one.id], enum.first)
       end
       assert queries.any?(/"products"\."id" >= 1/)
 
-      assert_equal([[three, four], four.id], enum.first)
+      assert_equal([[three, four], three.id], enum.first)
     end
 
     test "can be resumed on multiple columns" do
       enum = build_enumerator(columns: [:created_at, :id]).batches
       products = Product.order(:created_at, :id).take(2)
 
-      cursor = [products.last.created_at.strftime(SQL_TIME_FORMAT), products.last.id]
+      cursor = [products.first.created_at.strftime(SQL_TIME_FORMAT), products.first.id]
       assert_equal([products, cursor], enum.first)
 
       enum = build_enumerator(columns: [:created_at, :id], cursor: cursor).batches
-      products = Product.order(:created_at, :id).offset(1).take(2)
-
-      cursor = [products.last.created_at.strftime(SQL_TIME_FORMAT), products.last.id]
+      cursor = [products.first.created_at.strftime(SQL_TIME_FORMAT), products.first.id]
       assert_equal([products, cursor], enum.first)
     end
 

test/integration/integration_test.rb

@@ -19,14 +19,14 @@ def setup
         start_worker_and_wait
 
         assert_equal(1, queue_size)
-        assert_equal(0, iteration_metadata["cursor_position"])
+        assert_equal(1, iteration_metadata["cursor_position"])
         assert_equal(1, iteration_metadata["times_interrupted"])
 
         conn.set("stop_after_num", 2)
         start_worker_and_wait
 
         assert_equal(1, queue_size)
-        assert_equal(2, iteration_metadata["cursor_position"])
+        assert_equal(3, iteration_metadata["cursor_position"])
         assert_equal(2, iteration_metadata["times_interrupted"])
 
         TerminateJob.perform_async
@@ -43,26 +43,27 @@ def setup
       start_worker_and_wait
 
       Sidekiq.redis do |conn|
-        assert_equal(2, iteration_metadata["cursor_position"])
+        assert_equal(3, iteration_metadata["cursor_position"])
         assert_equal(1, iteration_metadata["executions"])
         assert_equal(0, iteration_metadata["times_interrupted"])
-        assert_equal(3, conn.llen("records_performed"))
+        assert_equal(["0", "1", "2"], conn.lrange("records_performed", 0, -1))
         assert_equal(1, conn.get("on_start_called").to_i)
 
         start_worker_and_wait
 
-        assert_equal(4, iteration_metadata["cursor_position"])
+        assert_equal(6, iteration_metadata["cursor_position"])
         assert_equal(2, iteration_metadata["executions"])
         assert_equal(0, iteration_metadata["times_interrupted"])
-        assert_equal(6, conn.llen("records_performed"))
+        assert_equal(["0", "1", "2", "3", "4", "5"], conn.lrange("records_performed", 0, -1))
         assert_equal(1, conn.get("on_start_called").to_i)
         assert_equal(0, conn.get("on_complete_called").to_i)
-      end
 
-      # last attempt
-      start_worker_and_wait
+        # last attempt
+        start_worker_and_wait
 
-      assert_equal(0, queue_size)
+        assert_equal(1, Sidekiq::RetrySet.new.size)
+        assert_equal(["0", "1", "2", "3", "4", "5", "6", "7", "8"], conn.lrange("records_performed", 0, -1))
+      end
     end
 
     test "failing non iteration job" do

test/integration/jobs.rb

@@ -25,7 +25,7 @@ class FailingIterationJob
   include SidekiqIteration::Iteration
 
   sidekiq_options retry: 3
-  sidekiq_retry_in { 1 }
+  sidekiq_retry_in { 0 }
 
   def on_start
     Sidekiq.redis do |conn|

test/iteration_test.rb

@@ -140,7 +140,7 @@ def each_iteration(record)
       metadata = iteration_metadata(job)
       assert_equal(1, metadata["times_interrupted"])
       assert_equal(1, metadata["executions"])
-      assert_equal(Product.second.id, metadata["cursor_position"])
+      assert_equal(Product.third.id, metadata["cursor_position"])
 
       ActiveRecordIterationJob.perform_one
       assert_equal(4, ActiveRecordIterationJob.records_performed.size)
@@ -188,7 +188,7 @@ def each_iteration(records)
 
       job = peek_into_queue
       metadata = iteration_metadata(job)
-      assert_equal(processed_records[2], metadata["cursor_position"])
+      assert_equal(processed_records[3], metadata["cursor_position"])
       assert_equal(1, metadata["times_interrupted"])
       assert_equal(1, metadata["executions"])
 
@@ -203,14 +203,13 @@ def each_iteration(records)
       metadata = iteration_metadata(job)
       assert_equal(2, metadata["times_interrupted"])
       assert_equal(2, metadata["executions"])
-      assert_equal(processed_records[4], metadata["cursor_position"])
+      assert_equal(processed_records[6], metadata["cursor_position"])
       continue_iterating(BatchActiveRecordIterationJob)
 
       BatchActiveRecordIterationJob.perform_one
       assert_jobs_in_queue(0)
       assert_equal(4, BatchActiveRecordIterationJob.records_performed.size)
-      # 10 records + 2 times restarted and ran on the same records
-      assert_equal(12, BatchActiveRecordIterationJob.records_performed.flatten.size)
+      assert_equal(processed_records.size, BatchActiveRecordIterationJob.records_performed.flatten.size)
 
       assert_equal(1, BatchActiveRecordIterationJob.on_start_called)
       assert_equal(4, BatchActiveRecordIterationJob.around_iteration_called)
@@ -257,23 +256,27 @@ def each_iteration(record)
 
       MultipleColumnsActiveRecordIterationJob.perform_async
 
+      products = Product.all.order(:updated_at, :id).to_a
+
       1.upto(3) do |iter|
         MultipleColumnsActiveRecordIterationJob.perform_one
 
         job = peek_into_queue
         last_processed_record = MultipleColumnsActiveRecordIterationJob.records_performed.last
-        expected = [last_processed_record.updated_at.strftime("%Y-%m-%d %H:%M:%S.%6N"), last_processed_record.id]
+        last_processed_record_index = products.index(last_processed_record)
+        next_record = products[last_processed_record_index + 1]
+
+        expected = [next_record.updated_at.strftime("%Y-%m-%d %H:%M:%S.%6N"), next_record.id]
         metadata = iteration_metadata(job)
         assert_equal(expected, metadata["cursor_position"])
 
         assert_equal(iter * 3, MultipleColumnsActiveRecordIterationJob.records_performed.size)
       end
 
-      products = Product.all.order(:updated_at, :id).to_a
       expected = [
         products[0], products[1], products[2], # first iteration
-        products[2], products[3], products[4], # second iteration
-        products[4], products[5], products[6]  # second iteration
+        products[3], products[4], products[5], # second iteration
+        products[6], products[7], products[8]  # second iteration
       ]
       assert_equal(expected, MultipleColumnsActiveRecordIterationJob.records_performed)
     end
@@ -483,13 +486,13 @@ def each_iteration(*)
       iterate_exact_times(ArrayJob, 1) # starts from 1-st record
       ArrayJob.perform_inline
 
-      iterate_exact_times(ArrayJob, 3) # starts from 1-st record
+      iterate_exact_times(ArrayJob, 3) # starts from 2-nd record
       ArrayJob.perform_one
 
-      continue_iterating(ArrayJob) # starts from 3-rd record
+      continue_iterating(ArrayJob) # starts from 5-th record
       ArrayJob.perform_one
 
-      assert_equal([1, 3, 8], ArrayJob.current_run_iterations)
+      assert_equal([1, 3, 6], ArrayJob.current_run_iterations)
     end
 
     class IterateUntilMaxRuntimeJob < SimpleIterationJob

test/throttling_test.rb

@@ -53,7 +53,7 @@ def setup
       assert_equal(1, runs_metadata.fetch("cursor_position"))
       assert_equal(1, runs_metadata.fetch("times_interrupted"))
 
-      assert_equal([1, 2], ThrottleJob.iterations_performed)
+      assert_equal([1], ThrottleJob.iterations_performed)
     end
 
     test "pushes job back to the queue with the configured throttle backoff" do