Accept notified jobs for workers waiting on an empty buffer

lib/que/job_buffer.rb

@@ -77,7 +77,7 @@ def accept?(metajobs)
       sync do
         return [] if _stopping?
 
-        start_index = _buffer_space
+        start_index = _free_capacity(metajobs)
         final_index = metajobs.length - 1
 
         return metajobs if start_index > final_index
@@ -157,6 +157,41 @@ def _buffer_space
       maximum_size - _size
     end
 
+    # How many of these metajobs can be taken on without evicting anything
+    # that's already buffered. That's the free buffer space, except when there
+    # is none at all, in which case it's the number of workers currently blocked
+    # waiting for a job they'd accept - push hands jobs straight to waiting
+    # workers before it trims the buffer to maximum_size, so a job claimed by an
+    # idle worker never occupies buffer space. Without falling back to them a
+    # buffer with maximum_size: 0 rejects every job offered, even when every
+    # worker is idle, leaving them to be found by polling.
+    #
+    # metajobs is sorted, and a worker's priority threshold admits every job at
+    # least as important as the ones it rejects, so the eligible worker count
+    # only falls as we advance and the first job we can't be sure of placing
+    # ends the run.
+    def _free_capacity(metajobs)
+      space = _buffer_space
+      return space if space > 0
+
+      thresholds = _waiting_priorities
+      index = 0
+
+      while index < metajobs.length && index < _sufficient_count(thresholds, metajobs[index])
+        index += 1
+      end
+
+      index
+    end
+
+    def _waiting_priorities
+      priority_queues.flat_map { |priority, pq| Array.new(pq.waiting_count, priority) }
+    end
+
+    def _sufficient_count(thresholds, metajob)
+      thresholds.count { |threshold| metajob.priority_sufficient?(threshold) }
+    end
+
     def pop(count)
       @array.pop(count)
     end

spec/que/job_buffer_spec.rb

@@ -271,6 +271,98 @@ def new_metajob(key)
 
       assert_equal [], job_buffer.accept?(job_array)
     end
+
+    describe "when the buffer has no spare space" do
+      let(:maximum_size) { 0 }
+
+      it "should not accept a job when no worker is waiting for one" do
+        assert_equal [], job_buffer.accept?([job_array[0]])
+      end
+
+      it "should accept no more jobs than there are workers waiting" do
+        job_buffer # Pre-initialize to avoid race conditions.
+
+        2.times { Thread.new { job_buffer.shift } }
+        sleep_until_equal(2) { job_buffer.waiting_count }
+
+        assert_equal job_array[0..1], job_buffer.accept?(job_array[0..2])
+      end
+
+      it "should pass an accepted job to the waiting worker without buffering it" do
+        job_buffer # Pre-initialize to avoid race conditions.
+
+        t = Thread.new { Thread.current[:job] = job_buffer.shift }
+        sleep_until_equal(1) { job_buffer.waiting_count }
+
+        assert_equal [job_array[0]], job_buffer.accept?([job_array[0]])
+        assert_nil job_buffer.push(job_array[0])
+        sleep_until_equal(false) { t.status }
+
+        assert_equal job_array[0], t[:job]
+        assert_equal [], job_buffer.to_a
+      end
+    end
+
+    describe "when the buffer has spare space" do
+      let(:maximum_size) { 2 }
+
+      it "should not accept more jobs than the buffer has space for, whatever workers are waiting" do
+        job_buffer # Pre-initialize to avoid race conditions.
+
+        4.times { Thread.new { job_buffer.shift } }
+        sleep_until_equal(4) { job_buffer.waiting_count }
+
+        assert_equal job_array[0..1], job_buffer.accept?(job_array[0..3])
+      end
+    end
+
+    describe "when the buffer is full" do
+      before do
+        job_buffer.push(*8.times.map { |i| new_metajob(priority: 30, run_at: old, id: i + 1) })
+
+        @waiter = Thread.new { job_buffer.shift(10) }
+        sleep_until_equal(1) { job_buffer.waiting_count }
+      end
+
+      after do
+        job_buffer.stop
+        @waiter.join
+      end
+
+      it "should accept a job for the waiting worker on top of the ones it can displace" do
+        jobs = 10.times.map { |i| new_metajob(priority: 5, run_at: now, id: i + 10) }
+
+        assert_equal 9, job_buffer.accept?(jobs).length
+      end
+    end
+
+    describe "when the only worker waiting has a priority threshold" do
+      let :job_buffer do
+        Que::JobBuffer.new(maximum_size: 0, priorities: [10])
+      end
+
+      before do
+        @waiter = Thread.new { job_buffer.shift(10) }
+        sleep_until_equal(1) { job_buffer.waiting_count }
+      end
+
+      after do
+        job_buffer.stop
+        @waiter.join
+      end
+
+      it "should accept a job the worker's priority admits" do
+        job = new_metajob(priority: 5, run_at: now, id: 1)
+
+        assert_equal [job], job_buffer.accept?([job])
+      end
+
+      it "should not accept a job the worker's priority does not admit" do
+        job = new_metajob(priority: 50, run_at: now, id: 1)
+
+        assert_equal [], job_buffer.accept?([job])
+      end
+    end
   end
 
   describe "buffer_space" do