Add ResourceGate for resource-aware dequeue gating

Introduces ResourceGate, an abstract base class that controls whether
queues should be dequeued based on external resource availability (GPU
utilization, CPU load, network bandwidth, etc.). The base class handles
TTL-based caching of check results; subclasses implement #check.

Built-in gates:
- OpenGate: always allows (default)
- ThresholdGate: samples a metric via callback, blocks when >= threshold

Gates are configured per-queue on QueueList via resource_gates hash.
The QueueList#queues method filters out gated queues on every access,
so dequeue adapters always see current resource state. Multiple queues
can share a gate instance to represent a shared resource boundary.

https://claude.ai/code/session_01PG9VitHQLVVWAJmMtFvQcA

Author: claude

spec/helpers/mock_queue_list.cr

@@ -1,7 +1,10 @@
 class MockQueueList < Mosquito::Runners::QueueList
-  getter queues
   setter state
 
+  def discovered_queues : Array(Mosquito::Queue)
+    @discovered_queues
+  end
+
   def stop(wait_group : WaitGroup = WaitGroup.new(1)) : WaitGroup
     self.state = Mosquito::Runnable::State::Stopping
     spawn do

spec/mosquito/dequeue_adapters/shuffle_dequeue_adapter_spec.cr

@@ -7,7 +7,7 @@ describe "Mosquito::ShuffleDequeueAdapter" do
 
   def register(job_class : Mosquito::Job.class)
     Mosquito::Base.register_job_mapping job_class.name.underscore, job_class
-    queue_list.queues << job_class.queue
+    queue_list.discovered_queues << job_class.queue
   end
 
   it "is the default adapter" do
@@ -61,7 +61,7 @@ describe "Mosquito::ShuffleDequeueAdapter" do
         overseer.dequeue_adapter = spy_adapter
 
         register QueuedTestJob
-        queue_list.queues << Mosquito::Queue.new("extra_queue")
+        queue_list.discovered_queues << Mosquito::Queue.new("extra_queue")
 
         overseer.dequeue_job?
 

spec/mosquito/dequeue_adapters/weighted_dequeue_adapter_spec.cr

@@ -6,7 +6,7 @@ describe "Mosquito::WeightedDequeueAdapter" do
 
   def register(job_class : Mosquito::Job.class)
     Mosquito::Base.register_job_mapping job_class.name.underscore, job_class
-    queue_list.queues << job_class.queue
+    queue_list.discovered_queues << job_class.queue
   end
 
   it "dequeues a job from a weighted queue" do

spec/mosquito/resource_gate_spec.cr

@@ -0,0 +1,67 @@
+require "../spec_helper"
+
+describe "Mosquito::OpenGate" do
+  it "always allows" do
+    gate = Mosquito::OpenGate.new
+    assert gate.allow?
+  end
+end
+
+describe "Mosquito::ThresholdGate" do
+  it "allows when metric is below threshold" do
+    gate = Mosquito::ThresholdGate.new(threshold: 80.0, sample_ttl: 0.seconds) { 50.0 }
+    assert gate.allow?
+  end
+
+  it "blocks when metric is at or above threshold" do
+    gate = Mosquito::ThresholdGate.new(threshold: 80.0, sample_ttl: 0.seconds) { 85.0 }
+    refute gate.allow?
+  end
+
+  it "blocks when metric equals threshold" do
+    gate = Mosquito::ThresholdGate.new(threshold: 80.0, sample_ttl: 0.seconds) { 80.0 }
+    refute gate.allow?
+  end
+end
+
+describe "Mosquito::ResourceGate caching" do
+  it "caches the check result within TTL" do
+    call_count = 0
+    gate = Mosquito::ThresholdGate.new(threshold: 80.0, sample_ttl: 5.seconds) do
+      call_count += 1
+      50.0
+    end
+
+    now = Time.utc
+    Timecop.freeze(now) do
+      gate.allow?
+      gate.allow?
+      gate.allow?
+      assert_equal 1, call_count
+    end
+  end
+
+  it "re-checks after TTL expires" do
+    call_count = 0
+    gate = Mosquito::ThresholdGate.new(threshold: 80.0, sample_ttl: 5.seconds) do
+      call_count += 1
+      50.0
+    end
+
+    now = Time.utc
+    Timecop.freeze(now) do
+      gate.allow?
+      assert_equal 1, call_count
+    end
+
+    Timecop.freeze(now + 3.seconds) do
+      gate.allow?
+      assert_equal 1, call_count, "Should still be cached at 3s"
+    end
+
+    Timecop.freeze(now + 6.seconds) do
+      gate.allow?
+      assert_equal 2, call_count, "Should re-check after 6s (past 5s TTL)"
+    end
+  end
+end

spec/mosquito/runners/coordinator_spec.cr

@@ -8,7 +8,7 @@ describe "Mosquito::Runners::Coordinator" do
   getter(enqueue_time) { Time.utc }
 
   def enqueue_job_run : JobRun
-    queue_list.queues << queue
+    queue_list.discovered_queues << queue
 
     job_run = JobRun.new "blah"
 

spec/mosquito/runners/executor_spec.cr

@@ -9,7 +9,7 @@ describe "Mosquito::Runners::Executor" do
 
   def register(job_class : Mosquito::Job.class)
     Mosquito::Base.register_job_mapping job_class.name.underscore, job_class
-    queue_list.queues << job_class.queue
+    queue_list.discovered_queues << job_class.queue
   end
 
   def run_job(job_class : Mosquito::Job.class)

spec/mosquito/runners/overseer_spec.cr

@@ -8,7 +8,7 @@ describe "Mosquito::Runners::Overseer" do
 
   def register(job_class : Mosquito::Job.class)
     Mosquito::Base.register_job_mapping job_class.name.underscore, job_class
-    queue_list.queues << job_class.queue
+    queue_list.discovered_queues << job_class.queue
   end
 
   def run_job(job_class : Mosquito::Job.class)

spec/mosquito/runners/queue_list_spec.cr

@@ -95,4 +95,84 @@ describe "Mosquito::Runners::QueueList" do
       end
     end
   end
+
+  describe "resource gate filtering" do
+    it "excludes queues whose gate blocks" do
+      clean_slate do
+        enqueue_jobs
+        queue_list.each_run
+
+        gate = Mosquito::ThresholdGate.new(threshold: 80.0, sample_ttl: 0.seconds) { 90.0 }
+        queue_list.resource_gates = {"passing_job" => gate.as(Mosquito::ResourceGate)}
+
+        refute_includes queue_list.queues.map(&.name), "passing_job"
+        assert_includes queue_list.queues.map(&.name), "failing_job"
+        assert_includes queue_list.queues.map(&.name), "io_queue"
+      end
+    end
+
+    it "includes queues whose gate allows" do
+      clean_slate do
+        enqueue_jobs
+        queue_list.each_run
+
+        gate = Mosquito::ThresholdGate.new(threshold: 80.0, sample_ttl: 0.seconds) { 50.0 }
+        queue_list.resource_gates = {"passing_job" => gate.as(Mosquito::ResourceGate)}
+
+        assert_includes queue_list.queues.map(&.name), "passing_job"
+      end
+    end
+
+    it "ungated queues are always included" do
+      clean_slate do
+        enqueue_jobs
+        queue_list.each_run
+
+        gate = Mosquito::ThresholdGate.new(threshold: 80.0, sample_ttl: 0.seconds) { 90.0 }
+        queue_list.resource_gates = {"passing_job" => gate.as(Mosquito::ResourceGate)}
+
+        assert_equal 2, queue_list.queues.size
+      end
+    end
+
+    it "multiple queues can share a gate" do
+      clean_slate do
+        enqueue_jobs
+        queue_list.each_run
+
+        gate = Mosquito::ThresholdGate.new(threshold: 80.0, sample_ttl: 0.seconds) { 90.0 }
+        queue_list.resource_gates = {
+          "passing_job" => gate.as(Mosquito::ResourceGate),
+          "failing_job" => gate.as(Mosquito::ResourceGate),
+        }
+
+        assert_equal ["io_queue"], queue_list.queues.map(&.name)
+      end
+    end
+
+    it "gate state is evaluated on each access" do
+      clean_slate do
+        enqueue_jobs
+        queue_list.each_run
+
+        value = 90.0
+        gate = Mosquito::ThresholdGate.new(threshold: 80.0, sample_ttl: 0.seconds) { value }
+        queue_list.resource_gates = {"passing_job" => gate.as(Mosquito::ResourceGate)}
+
+        refute_includes queue_list.queues.map(&.name), "passing_job"
+
+        value = 50.0
+        assert_includes queue_list.queues.map(&.name), "passing_job"
+      end
+    end
+
+    it "returns all queues when no gates are configured" do
+      clean_slate do
+        enqueue_jobs
+        queue_list.each_run
+
+        assert_equal 3, queue_list.queues.size
+      end
+    end
+  end
 end

src/mosquito/configuration.cr

@@ -20,6 +20,10 @@ module Mosquito
 
     property dequeue_adapter : Mosquito::DequeueAdapter = Mosquito::ShuffleDequeueAdapter.new
 
+    # Maps queue names to resource gates. Queues whose gate returns
+    # `false` from `#allow?` are excluded from dequeuing.
+    property resource_gates : Hash(String, Mosquito::ResourceGate) = {} of String => Mosquito::ResourceGate
+
     property publish_metrics : Bool = false
 
     # How often a mosquito runner should emit a heartbeat metric.

src/mosquito/gates/open_gate.cr

@@ -0,0 +1,15 @@
+require "../resource_gate"
+
+module Mosquito
+  # A gate that always allows dequeuing. This is the default when no
+  # resource constraint is configured.
+  class OpenGate < ResourceGate
+    def initialize
+      super(sample_ttl: 0.seconds)
+    end
+
+    protected def check : Bool
+      true
+    end
+  end
+end