Enqueue batched delayed jobs outside the transaction

README.md

@@ -135,7 +135,7 @@ environment variables:
 * `BACKGROUND` - [Run in the background](#running-in-the-background) if
 non-empty (via `Process.daemon`, if supported) (default `false`)
 * `DELAYED_REQUEUE_BATCH_SIZE` - Set the delayed job batch size if enabled (default `100`). If `<= 1`, this disables batching.
-* `DISABLE_DELAYED_REQUEUE_BATCHES` - Disable batched delayed job queuing (default `false`) - [See section below on consequences](#batched-delayed-job-and-resque-enqueue-hooks)
+* `DISABLE_DELAYED_REQUEUE_BATCHES` - Disable batched delayed job queuing (default `false`)
 * `DYNAMIC_SCHEDULE` - Enables [dynamic scheduling](#dynamic-schedules)
 if non-empty (default `false`)
 * `RAILS_ENV` - Environment to use in procline (`$0`) (default empty)
@@ -784,35 +784,6 @@ This table explains the version requirements for rufus-scheduler
 | `~> 4.0`         | `~> 3.0`        |
 | `< 4.0`          | `~> 2.0`        |
 
-##### Batched delayed job and resque enqueue hooks
-
-Batching delayed job queuing can speed up when per-second job counts grows,
-avoiding situations that may cause delayed enqueues to fall behind. This
-batching wraps enqueues in a `multi` pipeline, making far fewer roundtrips to
-the server.
-
-However, in `redis` gem `>= 4.0`, any operations to Redis within the `multi`
-block must use the multi handle so that the actions are captured. Resque's hooks
-do not currently have a way to pass this around, and so compatibility with other
-Resque plugins or hooks which access Redis at enqueue time is impacted with
-batch mode (A main class of plugins this affects are single runner locks which
-use redis for their locking mechanism). In these cases, you should consider
-disabling the batching by setting the `DISABLE_DELAYED_REQUEUE_BATCHES`
-environment variable to `true`.
-
-Detecting when this occurs can be tricky, you must watch for logs
-emitted by your `resque-scheduler` process such as:
-
-```text
-Redis::CommandError: ERR
-MULTI calls can not be nested
-NoMethodError: undefined method nil? for <Redis::Future
-NoMethodError: undefined method `to_i' for <Redis::Future
-```
-
-If you're in this situation, you will also see delayed jobs to not be enqueued
-(they will get dropped). Disabling batching is the workaround.
-
 ### Contributing
 
 See [CONTRIBUTING.md](CONTRIBUTING.md)

lib/resque/scheduler.rb

@@ -239,7 +239,7 @@ def enqueue_delayed_items_for_timestamp(timestamp)
             if am_master
               if batch_delayed_items?
                 actual_batch_size = enqueue_items_in_batch_for_timestamp(timestamp, batch_size)
-                log "queued batch of #{actual_batch_size} jobs" if actual_batch_size != -1
+                log "queued batch of #{actual_batch_size} jobs"
               else
                 item = enqueue_next_item(timestamp)
                 actual_batch_size = item.nil? ? 0 : 1
@@ -251,8 +251,6 @@ def enqueue_delayed_items_for_timestamp(timestamp)
 
           # continue processing until there are no more items in this
           # timestamp. If we don't have a full batch, this is the last one.
-          # This also breaks us in the event of a redis transaction failure
-          # i.e. enqueue_items_in_batch_for_timestamp returned -1
           break if actual_batch_size < batch_size
         end
 
@@ -266,44 +264,25 @@ def timestamp_key(timestamp)
       def enqueue_items_in_batch_for_timestamp(timestamp, batch_size)
         timestamp_bucket_key = timestamp_key(timestamp)
 
-        encoded_jobs_to_requeue = Resque.redis.lrange(timestamp_bucket_key, 0, batch_size - 1)
-
-        # Watch is used to ensure that the timestamp bucket we are operating on
-        # is not altered by any other clients between the watch call and when we call exec
-        # (to execute the multi block). We should error catch on the redis.exec return value
-        # as that will indicate if the entire transaction was aborted or not. Though we should
-        # be safe as our ltrim is inside the multi block and therefore also would have been
-        # aborted. So nothing would have been queued, but also nothing lost from the bucket.
-        watch_result = Resque.redis.watch(timestamp_bucket_key) do
-          Resque.redis.multi do |pipeline|
-            encoded_jobs_to_requeue.each do |encoded_job|
-              pipeline.srem("timestamps:#{encoded_job}", timestamp_bucket_key)
-
-              decoded_job = Resque.decode(encoded_job)
-              enqueue(decoded_job, :delayed)
-            end
+        encoded_jobs_to_requeue, = Resque.redis.multi do |transaction|
+          transaction.lrange(timestamp_bucket_key, 0, batch_size - 1)
+          transaction.ltrim(timestamp_bucket_key, batch_size, -1)
+        end
 
-            pipeline.ltrim(timestamp_bucket_key, batch_size, -1)
+        Resque.redis.pipelined do |pipeline|
+          encoded_jobs_to_requeue.each do |encoded_job|
+            pipeline.srem("timestamps:#{encoded_job}", [timestamp_bucket_key])
           end
         end
 
-        # Did the multi block successfully remove from this timestamp and enqueue the jobs?
-        success = !watch_result.nil?
-
-        # If this was the last batch in this timestamp bucket, clean up
-        if success && encoded_jobs_to_requeue.count < batch_size
-          Resque.clean_up_timestamp(timestamp_bucket_key, timestamp)
+        encoded_jobs_to_requeue.each do |encoded_job|
+          enqueue(Resque.decode(encoded_job), :delayed)
         end
 
-        unless success
-          # Our batched transaction failed in Redis due to the timestamp_bucket_key value
-          # being modified while we built our multi block. We return -1 to ensure we break
-          # out of the loop iterating on this timestamp so it can be re-processed via the
-          # loop in handle_delayed_items.
-          return -1
+        if encoded_jobs_to_requeue.count < batch_size
+          Resque.clean_up_timestamp(timestamp_bucket_key, timestamp)
         end
 
-        # will return 0 if none were left to batch
         encoded_jobs_to_requeue.count
       end
 

test/delayed_queue_test.rb

@@ -1104,6 +1104,19 @@
   end
 end
 
+class JobWithEnqueueHook
+  @queue = :hooked
+
+  class << self
+    attr_accessor :hook
+  end
+
+  def self.before_enqueue_hook(*args)
+    hook.call(*args)
+    true
+  end
+end
+
 context 'DelayedQueue batch delayed item queue' do
   batch_disabled = Resque::Scheduler.disable_delayed_requeue_batches
   batch_size = Resque::Scheduler.delayed_requeue_batch_size
@@ -1153,19 +1166,22 @@
     assert_equal(90, Resque.size(Resque.queue_from_class(SomeIvarJob)))
   end
 
-  # test to make sure the timestamp is cleaned up
-  test 'enqueue_delayed_items_for_timestamp handles a watch failure' do
+  test 'enqueue hooks can use redis and touch the timestamp without duplicating the batch' do
     t = Time.now + 60
+    3.times { |i| Resque.enqueue_at(t, JobWithEnqueueHook, i) }
 
-    # create 100 jobs
-    100.times { Resque.enqueue_at(t, SomeIvarJob) }
-    assert_equal(100, Resque.delayed_timestamp_size(t))
-
-    Resque.redis.stubs(:watch).returns(nil)
+    hook_results = []
+    JobWithEnqueueHook.hook = lambda do |*|
+      hook_results << Resque.redis.incr('hook_calls')
+      Resque.delayed_push(t, class: 'SomeIvarJob', args: []) if hook_results.one?
+    end
 
-    Resque.expects(:clean_up_timestamp).never
+    Resque::Scheduler.handle_delayed_items(t)
 
-    Resque::Scheduler.enqueue_delayed_items_for_timestamp(t)
+    assert_equal([1, 2, 3], hook_results)
+    assert_equal(3, Resque.size(:hooked))
+    assert_equal(1, Resque.size(:ivar))
+    assert_equal(0, Resque.delayed_queue_schedule_size)
   end
 
   test 'enqueue_delayed_items_for_timestamp cleans up a timestamp' do