Adds the ability for an overseer to modify it's executor pool size.

When the executor_count is incorrect, it'll now add or remove executors
to correct the pool size.

Author: robacarp

spec/helpers/mock_executor.cr

@@ -1,6 +1,5 @@
 class MockExecutor < Mosquito::Runners::Executor
-  setter job_run : Mosquito::JobRun?
-  setter queue : Mosquito::Queue?
+  setter work_unit : Mosquito::WorkUnit?
 
   def state=(state : Mosquito::Runnable::State)
     super

spec/mosquito/runners/executor_spec.cr

@@ -16,8 +16,7 @@ describe "Mosquito::Runners::Executor" do
     register job_class
     job_class.reset_performance_counter!
     job_run = job_class.new.enqueue
-    executor.job_run = job_run
-    executor.queue = job_class.queue
+    executor.work_unit = WorkUnit.of(job_run, from: job_class.queue)
     executor.execute
   end
 
@@ -62,8 +61,7 @@ describe "Mosquito::Runners::Executor" do
         FailingJob.queue.enqueue job_run
 
         Timecop.freeze now do
-          executor.job_run = job_run
-          executor.queue = FailingJob.queue
+          executor.work_unit = WorkUnit.of(job_run, from: FailingJob.queue)
           executor.execute
         end
 
@@ -72,8 +70,7 @@ describe "Mosquito::Runners::Executor" do
 
         Timecop.freeze now + job.reschedule_interval(1) do
           coordinator.enqueue_delayed_jobs
-          executor.job_run = job_run
-          executor.queue = FailingJob.queue
+          executor.work_unit = WorkUnit.of(job_run, from: FailingJob.queue)
           executor.execute
         end
 
@@ -91,8 +88,7 @@ describe "Mosquito::Runners::Executor" do
         job_run.store
         NonReschedulableFailingJob.queue.enqueue job_run
 
-        executor.job_run = job_run
-        executor.queue = NonReschedulableFailingJob.queue
+        executor.work_unit = WorkUnit.of(job_run, from: NonReschedulableFailingJob.queue)
         executor.execute
 
         actual_ttl = backend.expires_in job_run.config_key
@@ -109,8 +105,7 @@ describe "Mosquito::Runners::Executor" do
         job_run.store
         QueuedTestJob.queue.enqueue job_run
 
-        executor.job_run = job_run
-        executor.queue = QueuedTestJob.queue
+        executor.work_unit = WorkUnit.of(job_run, from: QueuedTestJob.queue)
         executor.execute
 
         assert_logs_match "Success"
@@ -143,8 +138,7 @@ describe "Mosquito::Runners::Executor" do
       api
 
       spawn {
-        executor.job_run = job_run
-        executor.queue = SleepyJob.queue
+        executor.work_unit = WorkUnit.of(job_run, from: SleepyJob.queue)
         executor.execute
         job_finished.send true
       }

src/mosquito/runner.cr

@@ -71,6 +71,10 @@ module Mosquito
       end
     end
 
+    def self.overseer
+      instance.overseer
+    end
+
     private def self.instance : self
       @@instance ||= new
     end

src/mosquito/runners/executor.cr

@@ -32,8 +32,7 @@ module Mosquito::Runners
 
     # Where work is received from the overseer.
     getter job_pipeline : Channel(WorkUnit)
-    getter! job_run : JobRun
-    getter! queue : Queue
+    getter! work_unit : WorkUnit
 
     # Used to notify the overseer when this executor is idle.
     # Sends the {JobRun, Queue} tuple that was just finished, or nil
@@ -45,14 +44,30 @@ module Mosquito::Runners
       Observability::Executor.new self
     }
 
+    getter? decommissioned : Bool = false
+    @stop_channel = Channel(Nil).new(1)
+
+    # Marks this executor for graceful shutdown. It will stop after
+    # completing its current job (if any).
+    def decommission!
+      return if @decommissioned
+      @decommissioned = true
+      @stop_channel.send(nil)
+    end
+
+    private def job_run : JobRun
+      work_unit.job_run
+    end
+
+    private def queue : Queue
+      work_unit.queue
+    end
+
     private def state=(state : State)
       # Send a message to the overseer that this executor is idle,
       # including the job that was just finished (if any).
       if state == State::Idle
-        finished = if (jr = @job_run) && (q = @queue)
-          WorkUnit.of(jr, from: q)
-        end
-        spawn { finished_bell.send finished }
+        spawn { finished_bell.send @work_unit }
       end
 
       super
@@ -76,19 +91,44 @@ module Mosquito::Runners
       self.state = State::Idle
     end
 
+    def stop(wait_group : WaitGroup = WaitGroup.new(1)) : WaitGroup
+      decommission!
+      super
+    end
+
     # :nodoc:
     def each_run : Nil
-      dequeue = job_pipeline.receive?
-      return if dequeue.nil?
+      if @decommissioned
+        self.state = State::Stopping
+        return
+      end
+
+      dequeue : WorkUnit? = nil
+      begin
+        select
+        when dequeue = job_pipeline.receive
+        when @stop_channel.receive
+          self.state = State::Stopping
+          return
+        end
+      rescue Channel::ClosedError
+        return
+      end
+
+      return unless dequeue
 
       self.state = State::Working
-      @job_run = dequeue.job_run
-      @queue = dequeue.queue
+      @work_unit = dequeue
       log.trace { "Dequeued #{job_run} from #{queue.name}" }
       execute
       log.trace { "Finished #{job_run} from #{queue.name}" }
-      self.state = State::Idle
 
+      if @decommissioned
+        self.state = State::Stopping
+        return
+      end
+
+      self.state = State::Idle
       observer.heartbeat!
     end
 

src/mosquito/runners/overseer.cr

@@ -34,6 +34,10 @@ module Mosquito::Runners
     # The number of executors to start.
     getter executor_count : Int32
 
+    def executor_count=(count : Int32)
+      @executor_count = Math.max(count, 1)
+    end
+
     getter idle_wait : Time::Span
 
     def initialize
@@ -165,7 +169,7 @@ module Mosquito::Runners
         spawn { @finished_notifier.send nil }
       end
 
-      check_for_deceased_runners
+      adjust_executor_pool
 
       run_at_most every: Mosquito.configuration.heartbeat_interval, label: :heartbeat do
         observer.heartbeat
@@ -195,18 +199,26 @@ module Mosquito::Runners
     #
     # When a dead executor is found, any job it was working on has its
     # failure counter incremented and follows the standard retry logic.
-    def check_for_deceased_runners : Nil
+    def adjust_executor_pool : Nil
+      # Remove dead/crashed executors and recover their jobs.
       executors.select {|executor| executor.dead? || executor.state.crashed? }
         .each do |dead_executor|
           observer.executor_died dead_executor
           recover_job_from dead_executor
           executors.delete dead_executor
         end
 
+      # Scale up: spawn new executors to reach the target count.
       (executor_count - executors.size).times do
         executors << build_executor.tap(&.run)
       end
 
+      # Scale down: decommission excess executors and remove them from the pool.
+      # They will finish their current job (if any) and then stop.
+      while executors.size > executor_count
+        executors.pop.decommission!
+      end
+
       observer.update_executor_list executors
 
       if queue_list.dead?
@@ -274,10 +286,10 @@ module Mosquito::Runners
     # If a dead executor was working on a job, increment its failure
     # counter and follow the standard retry logic.
     private def recover_job_from(dead_executor : Executor) : Nil
-      return unless job_run = dead_executor.job_run?
+      return unless work_unit = dead_executor.work_unit?
 
-      observer.recovered_job_from_executor job_run, dead_executor
-      job_run.retry_or_banish dead_executor.queue
+      observer.recovered_job_from_executor work_unit.job_run, dead_executor
+      work_unit.job_run.retry_or_banish work_unit.queue
     end
 
   end