diff --git a/web/lib/potato_mesh/application.rb b/web/lib/potato_mesh/application.rb index b593b9f..8ed64bc 100644 --- a/web/lib/potato_mesh/application.rb +++ b/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! diff --git a/web/lib/potato_mesh/application/federation.rb b/web/lib/potato_mesh/application/federation.rb index d8da4bd..a28d4ad 100644 --- a/web/lib/potato_mesh/application/federation.rb +++ b/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. diff --git a/web/lib/potato_mesh/application/routes/ingest.rb b/web/lib/potato_mesh/application/routes/ingest.rb index abb80ea..60ea2ba 100644 --- a/web/lib/potato_mesh/application/routes/ingest.rb +++ b/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 diff --git a/web/lib/potato_mesh/application/worker_pool.rb b/web/lib/potato_mesh/application/worker_pool.rb new file mode 100644 index 0000000..4e20d33 --- /dev/null +++ b/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] 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 diff --git a/web/lib/potato_mesh/config.rb b/web/lib/potato_mesh/config.rb index fc6f15e..75cfec3 100644 --- a/web/lib/potato_mesh/config.rb +++ b/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. diff --git a/web/spec/app_spec.rb b/web/spec/app_spec.rb index 2768798..8c93b86 100644 --- a/web/spec/app_spec.rb +++ b/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 diff --git a/web/spec/config_spec.rb b/web/spec/config_spec.rb index b034364..1dd636e 100644 --- a/web/spec/config_spec.rb +++ b/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) diff --git a/web/spec/federation_spec.rb b/web/spec/federation_spec.rb index c7723f7..83f88d9 100644 --- a/web/spec/federation_spec.rb +++ b/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 diff --git a/web/spec/worker_pool_spec.rb b/web/spec/worker_pool_spec.rb new file mode 100644 index 0000000..9ed2a53 --- /dev/null +++ b/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