Change poller to pop messages one at a time so they don't get lost if something goes wrong during a large pop operation.

lib/sidekiq/scheduled.rb

@@ -28,17 +28,16 @@ module Sidekiq
             now = Time.now.to_f.to_s
             Sidekiq.redis do |conn|
               SETS.each do |sorted_set|
-                (messages, _) = conn.multi do
-                  conn.zrangebyscore(sorted_set, '-inf', now)
-                  conn.zremrangebyscore(sorted_set, '-inf', now)
-                end
-
-                messages.each do |message|
-                  logger.debug { "enqueued #{sorted_set}: #{message}" }
-                  msg = Sidekiq.load_json(message)
-                  conn.multi do
-                    conn.sadd('queues', msg['queue'])
-                    conn.rpush("queue:#{msg['queue']}", message)
+                while message = conn.zrangebyscore(sorted_set, '-inf', now, :limit => [0, 1]).first do
+                  if message
+                    msg = Sidekiq.load_json(message)
+                    if conn.zrem(sorted_set, message)
+                      conn.multi do
+                        conn.sadd('queues', msg['queue'])
+                        conn.rpush("queue:#{msg['queue']}", message)
+                      end
+                      logger.debug("enqueued #{sorted_set}: #{message}") if logger.debug?
+                    end
                   end
                 end
               end

test/test_retry.rb

@@ -199,26 +199,4 @@ class TestRetry < MiniTest::Unit::TestCase
     end
   end
 
-  describe 'poller' do
-    before do
-      @redis = MiniTest::Mock.new
-      Sidekiq.instance_variable_set(:@redis, @redis)
-
-      def @redis.with; yield self; end
-    end
-
-    it 'should poll like a bad mother...SHUT YO MOUTH' do
-      fake_msg = Sidekiq.dump_json({ 'class' => 'Bob', 'args' => [1,2], 'queue' => 'someq' })
-      @redis.expect :multi, [[fake_msg], 1], []
-      @redis.expect :multi, [[], nil], []
-      @redis.expect :multi, [[], nil], []
-      @redis.expect :multi, [[], nil], []
-
-      inst = Sidekiq::Scheduled::Poller.new
-      inst.poll
-
-      @redis.verify
-    end
-  end
-
 end

test/test_scheduled.rb

@@ -0,0 +1,48 @@
+require 'helper'
+require 'sidekiq/scheduled'
+
+class TestScheduled < MiniTest::Unit::TestCase
+  class ScheduledWorker
+    include Sidekiq::Worker
+    def perform(x)
+    end
+  end
+  
+  describe 'poller' do
+    before do
+      Sidekiq.redis = REDIS
+      Sidekiq.redis do |conn|
+        conn.flushdb
+      end
+    end
+
+    it 'should empty the retry and scheduled queues up to the current time' do
+      Sidekiq.redis do |conn|
+        error_1 = Sidekiq.dump_json('class' => ScheduledWorker.name, 'args' => ["error_1"], 'queue' => 'queue_1')
+        error_2 = Sidekiq.dump_json('class' => ScheduledWorker.name, 'args' => ["error_2"], 'queue' => 'queue_2')
+        error_3 = Sidekiq.dump_json('class' => ScheduledWorker.name, 'args' => ["error_3"], 'queue' => 'queue_3')
+        future_1 = Sidekiq.dump_json('class' => ScheduledWorker.name, 'args' => ["future_1"], 'queue' => 'queue_4')
+        future_2 = Sidekiq.dump_json('class' => ScheduledWorker.name, 'args' => ["future_2"], 'queue' => 'queue_5')
+        future_3 = Sidekiq.dump_json('class' => ScheduledWorker.name, 'args' => ["future_3"], 'queue' => 'queue_6')
+
+        conn.zadd("retry", (Time.now - 60).to_f.to_s, error_1)
+        conn.zadd("retry", (Time.now - 50).to_f.to_s, error_2)
+        conn.zadd("retry", (Time.now + 60).to_f.to_s, error_3)
+        conn.zadd("schedule", (Time.now - 60).to_f.to_s, future_1)
+        conn.zadd("schedule", (Time.now - 50).to_f.to_s, future_2)
+        conn.zadd("schedule", (Time.now + 60).to_f.to_s, future_3)
+
+        poller = Sidekiq::Scheduled::Poller.new
+        poller.poll
+        poller.terminate
+        
+        assert_equal [error_1], conn.lrange("queue:queue_1", 0, -1)
+        assert_equal [error_2], conn.lrange("queue:queue_2", 0, -1)
+        assert_equal [error_3], conn.zrange("retry", 0, -1)
+        assert_equal [future_1], conn.lrange("queue:queue_4", 0, -1)
+        assert_equal [future_2], conn.lrange("queue:queue_5", 0, -1)
+        assert_equal [future_3], conn.zrange("schedule", 0, -1)
+      end
+    end
+  end
+end