Implement new inline mode to ease tests for Mission Control

In this mode, the worker/scheduler runs inline (still with the thread pool)
until there's no more work, and then it stops.

app/models/solid_queue/ready_execution.rb

@@ -1,6 +1,5 @@
 module SolidQueue
   class ReadyExecution < Execution
-    scope :queued_as, ->(queues) { where(queue_name: queues) }
     scope :ordered, -> { order(priority: :asc) }
 
     before_create :assume_attributes_from_job
@@ -19,9 +18,13 @@ def claim(queues, limit)
         claimed_executions_for(candidate_job_ids)
       end
 
+      def queued_as(queues)
+        QueueParser.new(queues, self).scoped_relation
+      end
+
       private
         def query_candidates(queues, limit)
-          queue_scope(queues).ordered.limit(limit).lock("FOR UPDATE SKIP LOCKED").pluck(:job_id)
+          queued_as(queues).ordered.limit(limit).lock("FOR UPDATE SKIP LOCKED").pluck(:job_id)
         end
 
         def lock(job_ids)
@@ -35,10 +38,6 @@ def claimed_executions_for(job_ids)
 
           SolidQueue::ClaimedExecution.where(job_id: job_ids)
         end
-
-        def queue_scope(queues)
-          QueueParser.new(queues, self).scoped_relation
-        end
     end
 
     def claim

lib/solid_queue/runner.rb

@@ -14,13 +14,8 @@ module Runner
       attr_accessor :supervisor_pid
     end
 
-    def start(mode: :sync)
-      procline "starting in mode #{mode}"
-
-      @stopping = false
-      register_signal_handlers
-
-      SolidQueue.logger.info("[SolidQueue] Starting #{self}")
+    def start(mode: :supervised)
+      boot_in mode
 
       run_callbacks(:start) do
         if mode == :async
@@ -33,14 +28,27 @@ def start(mode: :sync)
 
     def stop
       @stopping = true
-      @thread.join if running_in_async_mode?
+      @thread&.join
     end
 
     def running?
       !stopping?
     end
 
     private
+      attr_reader :mode
+
+      def boot_in(mode)
+        @mode = mode.to_s.inquiry
+        @stopping = false
+
+        procline "starting in mode #{mode}"
+
+        register_signal_handlers
+
+        SolidQueue.logger.info("[SolidQueue] Starting #{self}")
+      end
+
       def register_signal_handlers
         %w[ INT TERM ].each do |signal|
           trap(signal) do
@@ -67,10 +75,11 @@ def start_loop
       end
 
       def shutting_down?
-        stopping? || supervisor_went_away?
+        stopping? || supervisor_went_away? || finished?
       end
 
       def run
+        raise NotImplementedError
       end
 
       def shutdown
@@ -81,19 +90,24 @@ def stopping?
         @stopping
       end
 
-      def shutdown_completed?
+      def finished?
+        running_inline? && all_work_completed?
       end
 
       def supervisor_went_away?
-        if running_in_async_mode?
-          false
-        else
-          supervisor_pid != ::Process.ppid
-        end
+        supervised? && supervisor_pid != ::Process.ppid
+      end
+
+      def supervised?
+        mode.supervised?
+      end
+
+      def all_work_completed?
+        false
       end
 
-      def running_in_async_mode?
-        @thread.present?
+      def running_inline?
+        mode.inline?
       end
 
       def hostname

lib/solid_queue/scheduler.rb

@@ -26,16 +26,6 @@ def run
       end
     end
 
-    def shutdown
-      super
-
-      @shutdown_completed = true
-    end
-
-    def shutdown_completed?
-      @shutdown_completed
-    end
-
     def metadata
       super.merge(batch_size: batch_size, polling_interval: polling_interval)
     end

lib/solid_queue/worker.rb

@@ -37,8 +37,8 @@ def shutdown
         pool.wait_for_termination(SolidQueue.shutdown_timeout)
       end
 
-      def shutdown_completed?
-        pool.shutdown?
+      def all_work_completed?
+        SolidQueue::ReadyExecution.queued_as(queues).empty?
       end
 
       def metadata

test/test_helper.rb

@@ -16,6 +16,10 @@
 end
 
 class ActiveSupport::TestCase
+  setup do
+    SolidQueue.logger = ActiveSupport::Logger.new(nil)
+  end
+
   teardown do
     JobBuffer.clear
     File.delete(SolidQueue.supervisor_pidfile) if File.exist?(SolidQueue.supervisor_pidfile)