Parallelize federation tasks with worker pool (#419)

* Parallelize federation work with worker pool

* Handle worker pool shutdown fallback during federation announcements

web/lib/potato_mesh/application.rb

@@ -43,6 +43,7 @@ require_relative "application/errors"
 require_relative "application/database"
 require_relative "application/networking"
 require_relative "application/identity"
+require_relative "application/worker_pool"
 require_relative "application/federation"
 require_relative "application/prometheus"
 require_relative "application/queries"
@@ -130,6 +131,7 @@ module PotatoMesh
       set :public_folder, File.expand_path("../../public", __dir__)
       set :views, File.expand_path("../../views", __dir__)
       set :federation_thread, nil
+      set :federation_worker_pool, nil
       set :port, resolve_port
       set :bind, DEFAULT_BIND_ADDRESS
 
@@ -153,6 +155,12 @@ module PotatoMesh
       ensure_self_instance_record!
       update_all_prometheus_metrics_from_nodes
 
+      if federation_enabled?
+        ensure_federation_worker_pool!
+      else
+        shutdown_federation_worker_pool!
+      end
+
       if federation_announcements_active?
         start_initial_federation_announcement!
         start_federation_announcer!

web/lib/potato_mesh/application/federation.rb

@@ -126,6 +126,61 @@ module PotatoMesh
         db&.close
       end
 
+      # Retrieve or initialize the worker pool servicing federation jobs.
+      #
+      # @return [PotatoMesh::App::WorkerPool, nil] active worker pool or nil when disabled.
+      def federation_worker_pool
+        ensure_federation_worker_pool!
+      end
+
+      # Ensure the federation worker pool exists when federation remains enabled.
+      #
+      # @return [PotatoMesh::App::WorkerPool, nil] active worker pool if created.
+      def ensure_federation_worker_pool!
+        return nil unless federation_enabled?
+
+        existing = settings.respond_to?(:federation_worker_pool) ? settings.federation_worker_pool : nil
+        return existing if existing&.alive?
+
+        pool = PotatoMesh::App::WorkerPool.new(
+          size: PotatoMesh::Config.federation_worker_pool_size,
+          max_queue: PotatoMesh::Config.federation_worker_queue_capacity,
+          name: "potato-mesh-fed",
+        )
+
+        at_exit do
+          begin
+            pool.shutdown(timeout: PotatoMesh::Config.federation_task_timeout_seconds)
+          rescue StandardError
+            # Suppress shutdown errors during interpreter teardown.
+          end
+        end
+
+        set(:federation_worker_pool, pool) if respond_to?(:set)
+        pool
+      end
+
+      # Shutdown and clear the federation worker pool if present.
+      #
+      # @return [void]
+      def shutdown_federation_worker_pool!
+        existing = settings.respond_to?(:federation_worker_pool) ? settings.federation_worker_pool : nil
+        return unless existing
+
+        begin
+          existing.shutdown(timeout: PotatoMesh::Config.federation_task_timeout_seconds)
+        rescue StandardError => e
+          warn_log(
+            "Failed to shut down federation worker pool",
+            context: "federation",
+            error_class: e.class.name,
+            error_message: e.message,
+          )
+        ensure
+          set(:federation_worker_pool, nil) if respond_to?(:set)
+        end
+      end
+
       def federation_target_domains(self_domain)
         normalized_self = sanitize_instance_domain(self_domain)&.downcase
         ordered = []
@@ -258,9 +313,39 @@ module PotatoMesh
         attributes, signature = ensure_self_instance_record!
         payload_json = JSON.generate(instance_announcement_payload(attributes, signature))
         domains = federation_target_domains(attributes[:domain])
+        pool = federation_worker_pool
+        scheduled = []
+
         domains.each do |domain|
+          if pool
+            begin
+              task = pool.schedule do
+                announce_instance_to_domain(domain, payload_json)
+              end
+              scheduled << [domain, task]
+              next
+            rescue PotatoMesh::App::WorkerPool::QueueFullError
+              warn_log(
+                "Skipped asynchronous federation announcement",
+                context: "federation.announce",
+                domain: domain,
+                reason: "worker queue saturated",
+              )
+            rescue PotatoMesh::App::WorkerPool::ShutdownError
+              warn_log(
+                "Worker pool unavailable, falling back to synchronous announcement",
+                context: "federation.announce",
+                domain: domain,
+              )
+              pool = nil
+            end
+          end
+
           announce_instance_to_domain(domain, payload_json)
         end
+
+        wait_for_federation_tasks(scheduled)
+
         unless domains.empty?
           debug_log(
             "Federation announcement cycle complete",
@@ -270,6 +355,38 @@ module PotatoMesh
         end
       end
 
+      # Wait for scheduled federation tasks to complete while logging failures.
+      #
+      # @param scheduled [Array<(String, PotatoMesh::App::WorkerPool::Task)>] pairs of domains and tasks.
+      # @return [void]
+      def wait_for_federation_tasks(scheduled)
+        return if scheduled.empty?
+
+        timeout = PotatoMesh::Config.federation_task_timeout_seconds
+        scheduled.each do |domain, task|
+          begin
+            task.wait(timeout: timeout)
+          rescue PotatoMesh::App::WorkerPool::TaskTimeoutError => e
+            warn_log(
+              "Federation announcement task timed out",
+              context: "federation.announce",
+              domain: domain,
+              timeout: timeout,
+              error_class: e.class.name,
+              error_message: e.message,
+            )
+          rescue StandardError => e
+            warn_log(
+              "Federation announcement task failed",
+              context: "federation.announce",
+              domain: domain,
+              error_class: e.class.name,
+              error_message: e.message,
+            )
+          end
+        end
+      end
+
       def start_federation_announcer!
         # Federation broadcasts must not execute when federation support is disabled.
         return nil unless federation_enabled?
@@ -499,6 +616,58 @@ module PotatoMesh
         [nil, nil, e.message]
       end
 
+      # Enqueue a federation crawl for the supplied domain using the worker pool.
+      #
+      # @param domain [String] sanitized remote domain to crawl.
+      # @param per_response_limit [Integer, nil] maximum entries processed per response.
+      # @param overall_limit [Integer, nil] maximum unique domains visited.
+      # @return [Boolean] true when the crawl was scheduled successfully.
+      def enqueue_federation_crawl(domain, per_response_limit:, overall_limit:)
+        pool = federation_worker_pool
+        unless pool
+          debug_log(
+            "Skipped remote instance crawl",
+            context: "federation.instances",
+            domain: domain,
+            reason: "federation disabled",
+          )
+          return false
+        end
+
+        application = is_a?(Class) ? self : self.class
+        pool.schedule do
+          db = application.open_database
+          begin
+            application.ingest_known_instances_from!(
+              db,
+              domain,
+              per_response_limit: per_response_limit,
+              overall_limit: overall_limit,
+            )
+          ensure
+            db&.close
+          end
+        end
+
+        true
+      rescue PotatoMesh::App::WorkerPool::QueueFullError
+        warn_log(
+          "Skipped remote instance crawl",
+          context: "federation.instances",
+          domain: domain,
+          reason: "worker queue saturated",
+        )
+        false
+      rescue PotatoMesh::App::WorkerPool::ShutdownError
+        warn_log(
+          "Skipped remote instance crawl",
+          context: "federation.instances",
+          domain: domain,
+          reason: "worker pool shut down",
+        )
+        false
+      end
+
       # Recursively ingest federation records exposed by the supplied domain.
       #
       # @param db [SQLite3::Database] open database connection used for writes.

web/lib/potato_mesh/application/routes/ingest.rb

@@ -240,8 +240,7 @@ module PotatoMesh
 
             db = open_database
             upsert_instance_record(db, attributes, signature)
-            ingest_known_instances_from!(
-              db,
+            enqueued = enqueue_federation_crawl(
               attributes[:domain],
               per_response_limit: PotatoMesh::Config.federation_max_instances_per_response,
               overall_limit: PotatoMesh::Config.federation_max_domains_per_crawl,
@@ -251,6 +250,7 @@ module PotatoMesh
               context: "ingest.register",
               domain: attributes[:domain],
               instance_id: attributes[:id],
+              crawl_enqueued: enqueued,
             )
             status 201
             { status: "registered" }.to_json

web/lib/potato_mesh/application/worker_pool.rb

@@ -0,0 +1,212 @@
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# frozen_string_literal: true
+
+module PotatoMesh
+  module App
+    # WorkerPool executes submitted blocks using a bounded set of Ruby threads.
+    #
+    # The pool enforces an upper bound on queued tasks, surfaces errors raised
+    # by jobs, and supports graceful shutdown during application teardown.
+    class WorkerPool
+      # Raised when the worker pool queue has reached its configured capacity.
+      class QueueFullError < StandardError; end
+
+      # Raised when a task fails to complete before the requested timeout.
+      class TaskTimeoutError < StandardError; end
+
+      # Raised when scheduling occurs after the pool has been shut down.
+      class ShutdownError < StandardError; end
+
+      # Internal structure responsible for coordinating task completion.
+      class Task
+        # @return [Object, nil] value produced by the task block when available.
+        attr_reader :value
+
+        # @return [StandardError, nil] error raised by the task block when set.
+        attr_reader :error
+
+        def initialize
+          @mutex = Mutex.new
+          @condition = ConditionVariable.new
+          @complete = false
+          @value = nil
+          @error = nil
+        end
+
+        # Mark the task as completed successfully.
+        #
+        # @param result [Object] value produced by the job.
+        # @return [void]
+        def fulfill(result)
+          @mutex.synchronize do
+            return if @complete
+
+            @complete = true
+            @value = result
+            @condition.broadcast
+          end
+        end
+
+        # Mark the task as failed with the provided error.
+        #
+        # @param failure [StandardError] exception raised while executing the job.
+        # @return [void]
+        def reject(failure)
+          @mutex.synchronize do
+            return if @complete
+
+            @complete = true
+            @error = failure
+            @condition.broadcast
+          end
+        end
+
+        # Wait for the task to complete, raising any stored failure.
+        #
+        # @param timeout [Numeric, nil] optional timeout in seconds.
+        # @return [Object] the value produced by the job when successful.
+        # @raise [TaskTimeoutError] when the timeout elapses prior to completion.
+        # @raise [StandardError] when the job raised an exception.
+        def wait(timeout: nil)
+          deadline = timeout && monotonic_now + timeout
+
+          @mutex.synchronize do
+            until @complete
+              if deadline
+                remaining = deadline - monotonic_now
+                raise TaskTimeoutError, "task deadline exceeded" if remaining <= 0
+
+                @condition.wait(@mutex, remaining)
+              else
+                @condition.wait(@mutex)
+              end
+            end
+
+            raise @error if @error
+
+            @value
+          end
+        end
+
+        # Check whether the task has finished executing.
+        #
+        # @return [Boolean] true when the task is complete.
+        def complete?
+          @mutex.synchronize { @complete }
+        end
+
+        private
+
+        def monotonic_now
+          Process.clock_gettime(Process::CLOCK_MONOTONIC)
+        end
+      end
+
+      STOP_SIGNAL = Object.new
+
+      # @return [Array<Thread>] threads created to service the pool.
+      attr_reader :threads
+
+      # Initialize a worker pool using the supplied configuration.
+      #
+      # @param size [Integer] number of worker threads to spawn.
+      # @param max_queue [Integer, nil] optional upper bound on queued jobs.
+      # @param name [String] prefix assigned to worker thread names.
+      def initialize(size:, max_queue: nil, name: "worker-pool")
+        raise ArgumentError, "size must be positive" unless size.is_a?(Integer) && size.positive?
+
+        @name = name
+        @queue = max_queue ? SizedQueue.new(max_queue) : Queue.new
+        @threads = []
+        @stopped = false
+        @mutex = Mutex.new
+        spawn_workers(size)
+      end
+
+      # Determine whether the worker pool is still accepting work.
+      #
+      # @return [Boolean] true when the pool remains active.
+      def alive?
+        @mutex.synchronize { !@stopped }
+      end
+
+      # Submit a block of work for asynchronous execution.
+      #
+      # @yieldreturn [Object] result produced by the job block.
+      # @return [Task] task tracking the asynchronous execution.
+      # @raise [QueueFullError] when the queue cannot accept additional work.
+      # @raise [ShutdownError] when the pool is no longer active.
+      def schedule(&block)
+        raise ArgumentError, "block required" unless block
+
+        task = Task.new
+
+        @mutex.synchronize do
+          raise ShutdownError, "worker pool has been shut down" if @stopped
+
+          begin
+            @queue.push([task, block], true)
+          rescue ThreadError => e
+            raise QueueFullError, e.message
+          end
+        end
+
+        task
+      end
+
+      # Stop accepting work and wait for the worker threads to finish.
+      #
+      # @param timeout [Numeric, nil] seconds to wait for each worker to exit.
+      # @return [void]
+      def shutdown(timeout: nil)
+        threads = nil
+
+        @mutex.synchronize do
+          return if @stopped
+
+          @stopped = true
+          threads = @threads.dup
+        end
+
+        threads.each { @queue << STOP_SIGNAL }
+        threads.each { |thread| thread.join(timeout) }
+      end
+
+      private
+
+      def spawn_workers(size)
+        size.times do |index|
+          worker = Thread.new do
+            Thread.current.name = "#{@name}-#{index}" if Thread.current.respond_to?(:name=)
+            Thread.current.report_on_exception = false if Thread.current.respond_to?(:report_on_exception=)
+
+            loop do
+              task, block = @queue.pop
+              break if task.equal?(STOP_SIGNAL)
+
+              begin
+                result = block.call
+                task.fulfill(result)
+              rescue StandardError => e
+                task.reject(e)
+              end
+            end
+          end
+
+          @threads << worker
+        end
+      end
+    end
+  end
+end

web/lib/potato_mesh/config.rb

@@ -36,6 +36,9 @@ module PotatoMesh
     DEFAULT_REMOTE_INSTANCE_READ_TIMEOUT = 60
     DEFAULT_FEDERATION_MAX_INSTANCES_PER_RESPONSE = 64
     DEFAULT_FEDERATION_MAX_DOMAINS_PER_CRAWL = 256
+    DEFAULT_FEDERATION_WORKER_POOL_SIZE = 4
+    DEFAULT_FEDERATION_WORKER_QUEUE_CAPACITY = 128
+    DEFAULT_FEDERATION_TASK_TIMEOUT_SECONDS = 120
     DEFAULT_INITIAL_FEDERATION_DELAY_SECONDS = 2
 
     # Retrieve the configured API token used for authenticated requests.
@@ -356,6 +359,36 @@ module PotatoMesh
       )
     end
 
+    # Determine the worker pool size used for federation tasks.
+    #
+    # @return [Integer] number of worker threads dedicated to federation jobs.
+    def federation_worker_pool_size
+      fetch_positive_integer(
+        "FEDERATION_WORKERS",
+        DEFAULT_FEDERATION_WORKER_POOL_SIZE,
+      )
+    end
+
+    # Determine the queue capacity for pending federation jobs.
+    #
+    # @return [Integer] maximum number of queued tasks before rejecting work.
+    def federation_worker_queue_capacity
+      fetch_positive_integer(
+        "FEDERATION_WORK_QUEUE",
+        DEFAULT_FEDERATION_WORKER_QUEUE_CAPACITY,
+      )
+    end
+
+    # Determine the timeout applied when awaiting federation worker tasks.
+    #
+    # @return [Integer] seconds to wait for asynchronous jobs to complete.
+    def federation_task_timeout_seconds
+      fetch_positive_integer(
+        "FEDERATION_TASK_TIMEOUT",
+        DEFAULT_FEDERATION_TASK_TIMEOUT_SECONDS,
+      )
+    end
+
     # Maximum acceptable age for remote node data.
     #
     # @return [Integer] seconds before remote nodes are considered stale.

web/spec/app_spec.rb

@@ -1318,7 +1318,7 @@ RSpec.describe "Potato Mesh Sinatra app" do
     end
 
     before do
-      allow_any_instance_of(Sinatra::Application).to receive(:fetch_instance_json) do |_instance, host, path|
+      fetch_stub = lambda do |host, path|
         case path
         when "/.well-known/potato-mesh"
           [well_known_document, URI("https://#{host}#{path}")]
@@ -1328,6 +1328,29 @@ RSpec.describe "Potato Mesh Sinatra app" do
           [nil, []]
         end
       end
+
+      allow_any_instance_of(Sinatra::Application).to receive(:fetch_instance_json) do |_instance, host, path|
+        fetch_stub.call(host, path)
+      end
+
+      allow(PotatoMesh::Application).to receive(:fetch_instance_json) do |host, path|
+        fetch_stub.call(host, path)
+      end
+
+      allow_any_instance_of(Sinatra::Application).to receive(:enqueue_federation_crawl) do |instance, domain, per_response_limit:, overall_limit:|
+        db = instance.open_database
+        begin
+          instance.ingest_known_instances_from!(
+            db,
+            domain,
+            per_response_limit: per_response_limit,
+            overall_limit: overall_limit,
+          )
+        ensure
+          db&.close
+        end
+        true
+      end
     end
 
     it "stores a federated instance when validation succeeds" do

web/spec/config_spec.rb

@@ -285,6 +285,78 @@ RSpec.describe PotatoMesh::Config do
     end
   end
 
+  describe ".federation_worker_pool_size" do
+    it "returns the baked-in pool size when unset" do
+      within_env("FEDERATION_WORKERS" => nil) do
+        expect(described_class.federation_worker_pool_size).to eq(
+          PotatoMesh::Config::DEFAULT_FEDERATION_WORKER_POOL_SIZE,
+        )
+      end
+    end
+
+    it "accepts positive overrides" do
+      within_env("FEDERATION_WORKERS" => "9") do
+        expect(described_class.federation_worker_pool_size).to eq(9)
+      end
+    end
+
+    it "rejects invalid overrides" do
+      within_env("FEDERATION_WORKERS" => "0") do
+        expect(described_class.federation_worker_pool_size).to eq(
+          PotatoMesh::Config::DEFAULT_FEDERATION_WORKER_POOL_SIZE,
+        )
+      end
+    end
+  end
+
+  describe ".federation_worker_queue_capacity" do
+    it "returns the baked-in queue capacity when unset" do
+      within_env("FEDERATION_WORK_QUEUE" => nil) do
+        expect(described_class.federation_worker_queue_capacity).to eq(
+          PotatoMesh::Config::DEFAULT_FEDERATION_WORKER_QUEUE_CAPACITY,
+        )
+      end
+    end
+
+    it "accepts positive overrides" do
+      within_env("FEDERATION_WORK_QUEUE" => "33") do
+        expect(described_class.federation_worker_queue_capacity).to eq(33)
+      end
+    end
+
+    it "rejects invalid overrides" do
+      within_env("FEDERATION_WORK_QUEUE" => "-1") do
+        expect(described_class.federation_worker_queue_capacity).to eq(
+          PotatoMesh::Config::DEFAULT_FEDERATION_WORKER_QUEUE_CAPACITY,
+        )
+      end
+    end
+  end
+
+  describe ".federation_task_timeout_seconds" do
+    it "returns the baked-in timeout when unset" do
+      within_env("FEDERATION_TASK_TIMEOUT" => nil) do
+        expect(described_class.federation_task_timeout_seconds).to eq(
+          PotatoMesh::Config::DEFAULT_FEDERATION_TASK_TIMEOUT_SECONDS,
+        )
+      end
+    end
+
+    it "accepts positive overrides" do
+      within_env("FEDERATION_TASK_TIMEOUT" => "47") do
+        expect(described_class.federation_task_timeout_seconds).to eq(47)
+      end
+    end
+
+    it "rejects invalid overrides" do
+      within_env("FEDERATION_TASK_TIMEOUT" => "-7") do
+        expect(described_class.federation_task_timeout_seconds).to eq(
+          PotatoMesh::Config::DEFAULT_FEDERATION_TASK_TIMEOUT_SECONDS,
+        )
+      end
+    end
+  end
+
   describe ".db_path" do
     it "returns the default path inside the data directory" do
       expect(described_class.db_path).to eq(described_class.default_db_path)

web/spec/federation_spec.rb

@@ -48,6 +48,23 @@ RSpec.describe PotatoMesh::App::Federation do
         def reset_warn_messages
           @warn_messages = []
         end
+
+        def settings
+          @settings ||= Struct.new(
+            :federation_thread,
+            :initial_federation_thread,
+            :federation_worker_pool,
+          ).new
+        end
+
+        def set(key, value)
+          writer = "#{key}="
+          if settings.respond_to?(writer)
+            settings.public_send(writer, value)
+          else
+            raise ArgumentError, "unsupported setting #{key}"
+          end
+        end
       end
     end
   end
@@ -57,6 +74,11 @@ RSpec.describe PotatoMesh::App::Federation do
     federation_helpers.instance_variable_set(:@remote_instance_verify_callback, nil)
     federation_helpers.reset_debug_messages
     federation_helpers.reset_warn_messages
+    federation_helpers.shutdown_federation_worker_pool!
+  end
+
+  after do
+    federation_helpers.shutdown_federation_worker_pool!
   end
 
   describe ".remote_instance_cert_store" do
@@ -476,4 +498,199 @@ RSpec.describe PotatoMesh::App::Federation do
       expect(captured_request["User-Agent"]).to eq(federation_helpers.send(:federation_user_agent_header))
     end
   end
+
+  describe ".ensure_federation_worker_pool!" do
+    before do
+      allow(PotatoMesh::Config).to receive(:federation_worker_pool_size).and_return(1)
+      allow(PotatoMesh::Config).to receive(:federation_worker_queue_capacity).and_return(1)
+      allow(PotatoMesh::Config).to receive(:federation_task_timeout_seconds).and_return(0.05)
+    end
+
+    it "returns nil when federation is disabled" do
+      allow(federation_helpers).to receive(:federation_enabled?).and_return(false)
+
+      expect(federation_helpers.ensure_federation_worker_pool!).to be_nil
+    end
+
+    it "creates and memoizes the worker pool" do
+      allow(federation_helpers).to receive(:federation_enabled?).and_return(true)
+
+      pool = federation_helpers.ensure_federation_worker_pool!
+      expect(pool).to be_a(PotatoMesh::App::WorkerPool)
+      expect(federation_helpers.ensure_federation_worker_pool!).to equal(pool)
+    ensure
+      pool&.shutdown(timeout: 0.05)
+      federation_helpers.set(:federation_worker_pool, nil)
+    end
+  end
+
+  describe ".shutdown_federation_worker_pool!" do
+    it "logs an error when shutdown fails" do
+      pool = instance_double(PotatoMesh::App::WorkerPool)
+      allow(pool).to receive(:shutdown).and_raise(StandardError, "boom")
+
+      federation_helpers.set(:federation_worker_pool, pool)
+      federation_helpers.shutdown_federation_worker_pool!
+
+      expect(federation_helpers.warn_messages.last).to include("Failed to shut down federation worker pool")
+      expect(federation_helpers.send(:settings).federation_worker_pool).to be_nil
+    end
+  end
+
+  describe ".enqueue_federation_crawl" do
+    let(:pool) { instance_double(PotatoMesh::App::WorkerPool) }
+
+    it "returns false and logs when the pool is unavailable" do
+      allow(federation_helpers).to receive(:federation_worker_pool).and_return(nil)
+
+      result = federation_helpers.enqueue_federation_crawl(
+        "remote.mesh",
+        per_response_limit: 5,
+        overall_limit: 9,
+      )
+
+      expect(result).to be(false)
+      expect(federation_helpers.debug_messages.last).to include("Skipped remote instance crawl")
+    end
+
+    it "schedules ingestion work on the pool" do
+      allow(federation_helpers).to receive(:federation_worker_pool).and_return(pool)
+      db = instance_double(SQLite3::Database)
+      allow(db).to receive(:close)
+
+      expect(federation_helpers).to receive(:open_database).and_return(db)
+      expect(federation_helpers).to receive(:ingest_known_instances_from!).with(
+        db,
+        "remote.mesh",
+        per_response_limit: 5,
+        overall_limit: 9,
+      )
+
+      task = instance_double(PotatoMesh::App::WorkerPool::Task)
+      expect(pool).to receive(:schedule) do |&block|
+        block.call
+        task
+      end
+
+      result = federation_helpers.enqueue_federation_crawl(
+        "remote.mesh",
+        per_response_limit: 5,
+        overall_limit: 9,
+      )
+
+      expect(result).to be(true)
+      expect(db).to have_received(:close)
+    end
+
+    it "logs when the worker queue is saturated" do
+      allow(federation_helpers).to receive(:federation_worker_pool).and_return(pool)
+      allow(pool).to receive(:schedule).and_raise(PotatoMesh::App::WorkerPool::QueueFullError, "full")
+      expect(federation_helpers).to receive(:warn_log).with(
+        "Skipped remote instance crawl",
+        hash_including(
+          context: "federation.instances",
+          domain: "remote.mesh",
+          reason: "worker queue saturated",
+        ),
+      ).and_call_original
+
+      result = federation_helpers.enqueue_federation_crawl(
+        "remote.mesh",
+        per_response_limit: 1,
+        overall_limit: 2,
+      )
+
+      expect(result).to be(false)
+    end
+
+    it "logs when the worker pool is shutting down" do
+      allow(federation_helpers).to receive(:federation_worker_pool).and_return(pool)
+      allow(pool).to receive(:schedule).and_raise(PotatoMesh::App::WorkerPool::ShutdownError, "closed")
+      expect(federation_helpers).to receive(:warn_log).with(
+        "Skipped remote instance crawl",
+        hash_including(
+          context: "federation.instances",
+          domain: "remote.mesh",
+          reason: "worker pool shut down",
+        ),
+      ).and_call_original
+
+      result = federation_helpers.enqueue_federation_crawl(
+        "remote.mesh",
+        per_response_limit: 1,
+        overall_limit: 2,
+      )
+
+      expect(result).to be(false)
+    end
+  end
+
+  describe ".wait_for_federation_tasks" do
+    it "does nothing for empty input" do
+      federation_helpers.wait_for_federation_tasks([])
+      expect(federation_helpers.warn_messages).to be_empty
+    end
+
+    it "logs timeouts" do
+      task = instance_double(PotatoMesh::App::WorkerPool::Task)
+      allow(task).to receive(:wait).and_raise(PotatoMesh::App::WorkerPool::TaskTimeoutError, "late")
+      allow(PotatoMesh::Config).to receive(:federation_task_timeout_seconds).and_return(0.01)
+
+      federation_helpers.wait_for_federation_tasks([["remote.mesh", task]])
+
+      expect(federation_helpers.warn_messages.last).to include("task timed out")
+    end
+
+    it "logs unexpected failures" do
+      task = instance_double(PotatoMesh::App::WorkerPool::Task)
+      allow(task).to receive(:wait).and_raise(RuntimeError, "boom")
+      allow(PotatoMesh::Config).to receive(:federation_task_timeout_seconds).and_return(0.01)
+
+      federation_helpers.wait_for_federation_tasks([["remote.mesh", task]])
+
+      expect(federation_helpers.warn_messages.last).to include("task failed")
+    end
+  end
+
+  describe ".announce_instance_to_all_domains" do
+    let(:pool) { instance_double(PotatoMesh::App::WorkerPool) }
+
+    before do
+      allow(federation_helpers).to receive(:federation_enabled?).and_return(true)
+      allow(federation_helpers).to receive(:ensure_self_instance_record!).and_return([
+        { domain: "self.mesh" },
+        "signature",
+      ])
+      allow(federation_helpers).to receive(:instance_announcement_payload).and_return({})
+      allow(JSON).to receive(:generate).and_return("payload-json")
+      allow(federation_helpers).to receive(:federation_target_domains).and_return(%w[alpha.mesh beta.mesh])
+    end
+
+    it "schedules announcements on the worker pool" do
+      task = instance_double(PotatoMesh::App::WorkerPool::Task)
+      allow(federation_helpers).to receive(:wait_for_federation_tasks)
+      allow(federation_helpers).to receive(:federation_worker_pool).and_return(pool)
+      expect(pool).to receive(:schedule).twice.and_return(task)
+
+      federation_helpers.announce_instance_to_all_domains
+    end
+
+    it "falls back to synchronous announcements when the queue is saturated" do
+      allow(federation_helpers).to receive(:federation_worker_pool).and_return(pool)
+      allow(pool).to receive(:schedule).and_raise(PotatoMesh::App::WorkerPool::QueueFullError, "full")
+      allow(federation_helpers).to receive(:wait_for_federation_tasks)
+      expect(federation_helpers).to receive(:announce_instance_to_domain).with("alpha.mesh", "payload-json").once
+      expect(federation_helpers).to receive(:announce_instance_to_domain).with("beta.mesh", "payload-json").once
+
+      federation_helpers.announce_instance_to_all_domains
+    end
+
+    it "runs synchronously when the worker pool is unavailable" do
+      allow(federation_helpers).to receive(:federation_worker_pool).and_return(nil)
+      expect(federation_helpers).to receive(:announce_instance_to_domain).with("alpha.mesh", "payload-json")
+      expect(federation_helpers).to receive(:announce_instance_to_domain).with("beta.mesh", "payload-json")
+
+      federation_helpers.announce_instance_to_all_domains
+    end
+  end
 end

web/spec/worker_pool_spec.rb

@@ -0,0 +1,97 @@
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# frozen_string_literal: true
+
+require "spec_helper"
+require "timeout"
+
+RSpec.describe PotatoMesh::App::WorkerPool do
+  def with_pool(size: 2, queue: 2)
+    pool = PotatoMesh::App::WorkerPool.new(size: size, max_queue: queue, name: "spec-pool")
+    yield pool
+  ensure
+    pool&.shutdown(timeout: 0.5)
+  end
+
+  describe "#schedule" do
+    it "executes jobs asynchronously and exposes their return values" do
+      with_pool do |pool|
+        task = pool.schedule { 21 + 21 }
+        expect(task.wait(timeout: 1)).to eq(42)
+      end
+    end
+
+    it "propagates exceptions raised by the job block" do
+      with_pool do |pool|
+        task = pool.schedule { raise ArgumentError, "boom" }
+        expect { task.wait(timeout: 1) }.to raise_error(ArgumentError, "boom")
+      end
+    end
+
+    it "raises an error when the queue is saturated" do
+      with_pool(size: 1, queue: 1) do |pool|
+        gate = Queue.new
+        first_task = pool.schedule { gate.pop; :first }
+
+        Timeout.timeout(1) do
+          sleep 0.01 until gate.num_waiting.positive?
+        end
+
+        second_task = pool.schedule { gate.pop; :second }
+
+        expect do
+          pool.schedule { :third }
+        end.to raise_error(described_class::QueueFullError)
+
+        gate << nil
+        gate << nil
+        expect(first_task.wait(timeout: 1)).to eq(:first)
+        expect(second_task.wait(timeout: 1)).to eq(:second)
+      end
+    end
+  end
+
+  describe "#shutdown" do
+    it "prevents new work from being scheduled" do
+      pool = described_class.new(size: 1, max_queue: 1, name: "spec-pool")
+      pool.shutdown(timeout: 0.5)
+
+      expect do
+        pool.schedule { :after_shutdown }
+      end.to raise_error(described_class::ShutdownError)
+    ensure
+      pool.shutdown(timeout: 0.5)
+    end
+  end
+
+  describe PotatoMesh::App::WorkerPool::Task do
+    it "raises a timeout when the job exceeds the provided deadline" do
+      with_pool do |pool|
+        task = pool.schedule { sleep 0.1; :done }
+        expect do
+          task.wait(timeout: 0.01)
+        end.to raise_error(PotatoMesh::App::WorkerPool::TaskTimeoutError)
+        expect(task.wait(timeout: 1)).to eq(:done)
+      end
+    end
+
+    it "reports completion status" do
+      with_pool do |pool|
+        task = pool.schedule { :result }
+        expect(task.complete?).to be(false)
+        expect(task.wait(timeout: 1)).to eq(:result)
+        expect(task.complete?).to be(true)
+      end
+    end
+  end
+end