From 3462b43bf773119f124f97ec78e30b2318811221 Mon Sep 17 00:00:00 2001 From: Esity Date: Tue, 31 Mar 2026 18:28:27 -0500 Subject: [PATCH] add cluster control broadcast exchange (#2) --- CHANGELOG.md | 12 ++ .../extensions/node/actors/cluster_control.rb | 35 +++++ lib/legion/extensions/node/runners/node.rb | 24 ++++ lib/legion/extensions/node/transport.rb | 1 + .../transport/exchanges/cluster_control.rb | 17 +++ .../transport/messages/cluster_killswitch.rb | 45 +++++++ .../transport/messages/cluster_settings.rb | 43 +++++++ .../node/transport/queues/cluster_control.rb | 22 ++++ lib/legion/extensions/node/version.rb | 2 +- .../node/actors/cluster_control_spec.rb | 88 +++++++++++++ .../extensions/node/runners/node_spec.rb | 121 ++++++++++++++++++ .../exchanges/cluster_control_spec.rb | 24 ++++ .../messages/cluster_killswitch_spec.rb | 116 +++++++++++++++++ .../messages/cluster_settings_spec.rb | 120 +++++++++++++++++ .../transport/queues/cluster_control_spec.rb | 48 +++++++ 15 files changed, 717 insertions(+), 1 deletion(-) create mode 100644 lib/legion/extensions/node/actors/cluster_control.rb create mode 100644 lib/legion/extensions/node/transport/exchanges/cluster_control.rb create mode 100644 lib/legion/extensions/node/transport/messages/cluster_killswitch.rb create mode 100644 lib/legion/extensions/node/transport/messages/cluster_settings.rb create mode 100644 lib/legion/extensions/node/transport/queues/cluster_control.rb create mode 100644 spec/legion/extensions/node/actors/cluster_control_spec.rb create mode 100644 spec/legion/extensions/node/transport/exchanges/cluster_control_spec.rb create mode 100644 spec/legion/extensions/node/transport/messages/cluster_killswitch_spec.rb create mode 100644 spec/legion/extensions/node/transport/messages/cluster_settings_spec.rb create mode 100644 spec/legion/extensions/node/transport/queues/cluster_control_spec.rb diff --git a/CHANGELOG.md b/CHANGELOG.md index b711670..b7e4723 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,17 @@ # Changelog +## [0.3.6] - 2026-03-31 + +### Added +- `legion.cluster.control` topic exchange for cluster-wide broadcast (durable, topic type) +- Per-node auto-delete queue `legion.cluster.control.` bound to the exchange on boot +- `ClusterSettings` message: publishes settings changes with configurable routing key to the cluster control exchange +- `ClusterKillswitch` message: publishes emergency extension block via routing key `settings.extensions.blocked` +- `ClusterControl` subscription actor: subscribes to the per-node cluster control queue, routes to existing runners by routing key +- `Runners::Node#broadcast_settings(settings:, routing_key: 'settings', restart: false)`: sends settings to all nodes via the cluster control exchange +- `Runners::Node#killswitch(extension:)`: blocks an extension cluster-wide with immediate reload +- 51 new specs covering all new components (162 total, 0 failures) + ## [0.3.5] - 2026-03-30 ### Changed diff --git a/lib/legion/extensions/node/actors/cluster_control.rb b/lib/legion/extensions/node/actors/cluster_control.rb new file mode 100644 index 0000000..ce1b044 --- /dev/null +++ b/lib/legion/extensions/node/actors/cluster_control.rb @@ -0,0 +1,35 @@ +# frozen_string_literal: true + +module Legion + module Extensions + module Node + module Actor + class ClusterControl < Legion::Extensions::Actors::Subscription + def runner_class + Legion::Extensions::Node::Runners::Node + end + + def queue_class + Legion::Extensions::Node::Transport::Queues::ClusterControl + end + + def disabled? + false + end + + def use_runner? + true + end + + def check_subtask? + false + end + + def generate_task? + false + end + end + end + end + end +end diff --git a/lib/legion/extensions/node/runners/node.rb b/lib/legion/extensions/node/runners/node.rb index 05ff246..a899996 100644 --- a/lib/legion/extensions/node/runners/node.rb +++ b/lib/legion/extensions/node/runners/node.rb @@ -120,6 +120,30 @@ def receive_vault_token(message:, routing_key:, public_key:, **) public_key: public_key) end + def broadcast_settings(settings:, routing_key: 'settings', restart: false, **_opts) + log.debug "broadcast_settings: routing_key=#{routing_key} keys=#{settings.keys.join(', ')}" + Legion::Extensions::Node::Transport::Messages::ClusterSettings.new( + settings: settings, + routing_key: routing_key, + restart: restart + ).publish + {} + rescue StandardError => e + log.error "broadcast_settings failed: #{e.message}" + {} + end + + def killswitch(extension:, **_opts) + log.debug "killswitch: blocking extension #{extension} cluster-wide" + Legion::Extensions::Node::Transport::Messages::ClusterKillswitch.new( + extension: extension.to_s.delete_prefix('lex-') + ).publish + {} + rescue StandardError => e + log.error "killswitch failed: #{e.message}" + {} + end + private def publish_update_result(action:, status:, detail: nil, error: nil) diff --git a/lib/legion/extensions/node/transport.rb b/lib/legion/extensions/node/transport.rb index 983c9c3..48abf4a 100644 --- a/lib/legion/extensions/node/transport.rb +++ b/lib/legion/extensions/node/transport.rb @@ -13,6 +13,7 @@ def self.additional_e_to_q array.push(from: 'node', to: 'node', routing_key: 'node.data.#') if Legion::Settings[:data]&.[](:connected) || false array.push(from: 'node', to: 'node', routing_key: 'node.cache.#') if Legion::Settings[:cache]&.[](:connected) || false array.push(from: 'node', to: 'node', routing_key: 'node.crypt.#') + array.push(from: 'cluster_control', to: 'cluster_control', routing_key: '#') array end end diff --git a/lib/legion/extensions/node/transport/exchanges/cluster_control.rb b/lib/legion/extensions/node/transport/exchanges/cluster_control.rb new file mode 100644 index 0000000..8027909 --- /dev/null +++ b/lib/legion/extensions/node/transport/exchanges/cluster_control.rb @@ -0,0 +1,17 @@ +# frozen_string_literal: true + +module Legion + module Extensions + module Node + module Transport + module Exchanges + class ClusterControl + EXCHANGE_NAME = 'legion.cluster.control' + EXCHANGE_TYPE = :topic + EXCHANGE_OPTIONS = { durable: true }.freeze + end + end + end + end + end +end diff --git a/lib/legion/extensions/node/transport/messages/cluster_killswitch.rb b/lib/legion/extensions/node/transport/messages/cluster_killswitch.rb new file mode 100644 index 0000000..bb03006 --- /dev/null +++ b/lib/legion/extensions/node/transport/messages/cluster_killswitch.rb @@ -0,0 +1,45 @@ +# frozen_string_literal: true + +module Legion + module Extensions + module Node + module Transport + module Messages + class ClusterKillswitch < Legion::Transport::Message + ROUTING_KEY = 'settings.extensions.blocked' + + def routing_key + ROUTING_KEY + end + + def exchange + Legion::Extensions::Node::Transport::Exchanges::ClusterControl + end + + def type + 'task' + end + + def encrypt? + false + end + + def message + { + function: 'update_settings', + settings: { extensions: { blocked: [@options[:extension]] } }, + restart: true + } + end + + def validate + raise 'extension must be a String' unless @options[:extension].is_a?(String) + + @valid = true + end + end + end + end + end + end +end diff --git a/lib/legion/extensions/node/transport/messages/cluster_settings.rb b/lib/legion/extensions/node/transport/messages/cluster_settings.rb new file mode 100644 index 0000000..7728e34 --- /dev/null +++ b/lib/legion/extensions/node/transport/messages/cluster_settings.rb @@ -0,0 +1,43 @@ +# frozen_string_literal: true + +module Legion + module Extensions + module Node + module Transport + module Messages + class ClusterSettings < Legion::Transport::Message + def routing_key + @options.fetch(:routing_key, 'settings') + end + + def exchange + Legion::Extensions::Node::Transport::Exchanges::ClusterControl + end + + def type + 'task' + end + + def encrypt? + false + end + + def message + { + function: 'update_settings', + settings: @options[:settings], + restart: @options.fetch(:restart, false) + } + end + + def validate + raise 'settings must be a Hash' unless @options[:settings].is_a?(Hash) + + @valid = true + end + end + end + end + end + end +end diff --git a/lib/legion/extensions/node/transport/queues/cluster_control.rb b/lib/legion/extensions/node/transport/queues/cluster_control.rb new file mode 100644 index 0000000..4f39b09 --- /dev/null +++ b/lib/legion/extensions/node/transport/queues/cluster_control.rb @@ -0,0 +1,22 @@ +# frozen_string_literal: true + +module Legion + module Extensions + module Node + module Transport + module Queues + class ClusterControl < Legion::Transport::Queue + def queue_name + "legion.cluster.control.#{Legion::Settings[:client][:name]}" + end + + def queue_options + { durable: false, exclusive: false, auto_delete: true, + arguments: { 'x-queue-type': 'classic' } } + end + end + end + end + end + end +end diff --git a/lib/legion/extensions/node/version.rb b/lib/legion/extensions/node/version.rb index 0551d74..f15c2e8 100644 --- a/lib/legion/extensions/node/version.rb +++ b/lib/legion/extensions/node/version.rb @@ -3,7 +3,7 @@ module Legion module Extensions module Node - VERSION = '0.3.5' + VERSION = '0.3.6' end end end diff --git a/spec/legion/extensions/node/actors/cluster_control_spec.rb b/spec/legion/extensions/node/actors/cluster_control_spec.rb new file mode 100644 index 0000000..a63926a --- /dev/null +++ b/spec/legion/extensions/node/actors/cluster_control_spec.rb @@ -0,0 +1,88 @@ +# frozen_string_literal: true + +require 'spec_helper' + +unless defined?(Legion::Extensions::Actors::Subscription) + module Legion + module Extensions + module Actors + class Subscription + def initialize(**); end + end + end + end + end +end + +unless defined?(Legion::Extensions::Node::Runners::Node) + module Legion + module Extensions + module Node + module Runners + module Node + end + end + end + end + end +end + +unless defined?(Legion::Extensions::Node::Transport::Queues::ClusterControl) + module Legion + module Extensions + module Node + module Transport + module Queues + ClusterControl = Class.new + end + end + end + end + end +end + +require 'legion/extensions/node/actors/cluster_control' + +RSpec.describe Legion::Extensions::Node::Actor::ClusterControl do + subject(:instance) { described_class.allocate } + + def call(method_name) + described_class.instance_method(method_name).bind_call(instance) + end + + describe '#runner_class' do + it 'returns Legion::Extensions::Node::Runners::Node' do + expect(call(:runner_class)).to eq(Legion::Extensions::Node::Runners::Node) + end + end + + describe '#queue_class' do + it 'returns Legion::Extensions::Node::Transport::Queues::ClusterControl' do + expect(call(:queue_class)).to eq(Legion::Extensions::Node::Transport::Queues::ClusterControl) + end + end + + describe '#disabled?' do + it 'returns false' do + expect(call(:disabled?)).to be false + end + end + + describe '#use_runner?' do + it 'returns true' do + expect(call(:use_runner?)).to be true + end + end + + describe '#check_subtask?' do + it 'returns false' do + expect(call(:check_subtask?)).to be false + end + end + + describe '#generate_task?' do + it 'returns false' do + expect(call(:generate_task?)).to be false + end + end +end diff --git a/spec/legion/extensions/node/runners/node_spec.rb b/spec/legion/extensions/node/runners/node_spec.rb index c705c8c..08dec1a 100644 --- a/spec/legion/extensions/node/runners/node_spec.rb +++ b/spec/legion/extensions/node/runners/node_spec.rb @@ -45,6 +45,20 @@ def initialize(**); end def publish; end end end + + unless defined?(Legion::Extensions::Node::Transport::Messages::ClusterSettings) + class ClusterSettings + def initialize(**); end + def publish; end + end + end + + unless defined?(Legion::Extensions::Node::Transport::Messages::ClusterKillswitch) + class ClusterKillswitch + def initialize(**); end + def publish; end + end + end end end @@ -407,4 +421,111 @@ def error(*); end end end end + + describe '#broadcast_settings' do + before do + @settings_store[:client] = { name: 'test-node' } + allow(Legion::Extensions::Node::Transport::Messages::ClusterSettings).to receive(:new).and_return( + instance_double(Legion::Extensions::Node::Transport::Messages::ClusterSettings, publish: nil) + ) + end + + it 'publishes a ClusterSettings message' do + expect(Legion::Extensions::Node::Transport::Messages::ClusterSettings).to receive(:new) + runner.broadcast_settings(settings: { feature: true }) + end + + it 'returns an empty hash' do + result = runner.broadcast_settings(settings: { feature: true }) + expect(result).to eq({}) + end + + it 'defaults routing_key to "settings"' do + expect(Legion::Extensions::Node::Transport::Messages::ClusterSettings).to receive(:new).with( + hash_including(routing_key: 'settings') + ).and_return(instance_double(Legion::Extensions::Node::Transport::Messages::ClusterSettings, publish: nil)) + runner.broadcast_settings(settings: { feature: true }) + end + + it 'passes a custom routing_key through' do + expect(Legion::Extensions::Node::Transport::Messages::ClusterSettings).to receive(:new).with( + hash_including(routing_key: 'flags.beta') + ).and_return(instance_double(Legion::Extensions::Node::Transport::Messages::ClusterSettings, publish: nil)) + runner.broadcast_settings(settings: { feature: true }, routing_key: 'flags.beta') + end + + it 'passes restart: true when specified' do + expect(Legion::Extensions::Node::Transport::Messages::ClusterSettings).to receive(:new).with( + hash_including(restart: true) + ).and_return(instance_double(Legion::Extensions::Node::Transport::Messages::ClusterSettings, publish: nil)) + runner.broadcast_settings(settings: { feature: true }, restart: true) + end + + context 'when publish raises an error' do + before do + allow(Legion::Extensions::Node::Transport::Messages::ClusterSettings).to receive(:new).and_raise( + StandardError, 'publish failed' + ) + end + + it 'does not crash' do + expect { runner.broadcast_settings(settings: { feature: true }) }.not_to raise_error + end + + it 'returns an empty hash' do + result = runner.broadcast_settings(settings: { feature: true }) + expect(result).to eq({}) + end + end + end + + describe '#killswitch' do + before do + @settings_store[:client] = { name: 'test-node' } + allow(Legion::Extensions::Node::Transport::Messages::ClusterKillswitch).to receive(:new).and_return( + instance_double(Legion::Extensions::Node::Transport::Messages::ClusterKillswitch, publish: nil) + ) + end + + it 'publishes a ClusterKillswitch message' do + expect(Legion::Extensions::Node::Transport::Messages::ClusterKillswitch).to receive(:new) + runner.killswitch(extension: 'bad-ext') + end + + it 'returns an empty hash' do + result = runner.killswitch(extension: 'bad-ext') + expect(result).to eq({}) + end + + it 'strips the lex- prefix from the extension name' do + expect(Legion::Extensions::Node::Transport::Messages::ClusterKillswitch).to receive(:new).with( + extension: 'bad-ext' + ).and_return(instance_double(Legion::Extensions::Node::Transport::Messages::ClusterKillswitch, publish: nil)) + runner.killswitch(extension: 'lex-bad-ext') + end + + it 'passes extension without prefix unchanged' do + expect(Legion::Extensions::Node::Transport::Messages::ClusterKillswitch).to receive(:new).with( + extension: 'bad-ext' + ).and_return(instance_double(Legion::Extensions::Node::Transport::Messages::ClusterKillswitch, publish: nil)) + runner.killswitch(extension: 'bad-ext') + end + + context 'when publish raises an error' do + before do + allow(Legion::Extensions::Node::Transport::Messages::ClusterKillswitch).to receive(:new).and_raise( + StandardError, 'killswitch failed' + ) + end + + it 'does not crash' do + expect { runner.killswitch(extension: 'bad-ext') }.not_to raise_error + end + + it 'returns an empty hash' do + result = runner.killswitch(extension: 'bad-ext') + expect(result).to eq({}) + end + end + end end diff --git a/spec/legion/extensions/node/transport/exchanges/cluster_control_spec.rb b/spec/legion/extensions/node/transport/exchanges/cluster_control_spec.rb new file mode 100644 index 0000000..744bfbc --- /dev/null +++ b/spec/legion/extensions/node/transport/exchanges/cluster_control_spec.rb @@ -0,0 +1,24 @@ +# frozen_string_literal: true + +require 'spec_helper' +require 'legion/extensions/node/transport/exchanges/cluster_control' + +RSpec.describe Legion::Extensions::Node::Transport::Exchanges::ClusterControl do + describe 'constants' do + it 'has the correct exchange name' do + expect(described_class::EXCHANGE_NAME).to eq('legion.cluster.control') + end + + it 'uses topic exchange type' do + expect(described_class::EXCHANGE_TYPE).to eq(:topic) + end + + it 'is durable' do + expect(described_class::EXCHANGE_OPTIONS).to include(durable: true) + end + + it 'options are frozen' do + expect(described_class::EXCHANGE_OPTIONS).to be_frozen + end + end +end diff --git a/spec/legion/extensions/node/transport/messages/cluster_killswitch_spec.rb b/spec/legion/extensions/node/transport/messages/cluster_killswitch_spec.rb new file mode 100644 index 0000000..1c6c0c8 --- /dev/null +++ b/spec/legion/extensions/node/transport/messages/cluster_killswitch_spec.rb @@ -0,0 +1,116 @@ +# frozen_string_literal: true + +require 'spec_helper' +require 'legion/extensions/node/transport/exchanges/cluster_control' + +CLUSTER_KILLSWITCH_ROUTING_KEY = 'settings.extensions.blocked' + +# Mirror ClusterKillswitch message logic in a proxy class to avoid superclass mismatch. +CLUSTER_KILLSWITCH_TEST_CLASS = Class.new do + def initialize(**opts) + @options = opts + end + + def routing_key + CLUSTER_KILLSWITCH_ROUTING_KEY + end + + def exchange + Legion::Extensions::Node::Transport::Exchanges::ClusterControl + end + + def type + 'task' + end + + def encrypt? + false + end + + def message + { + function: 'update_settings', + settings: { extensions: { blocked: [@options[:extension]] } }, + restart: true + } + end + + def validate + raise 'extension must be a String' unless @options[:extension].is_a?(String) + + @valid = true + end +end + +RSpec.describe 'ClusterKillswitch message' do + subject(:msg) { CLUSTER_KILLSWITCH_TEST_CLASS.new(extension: 'my-ext') } + + describe 'routing key constant' do + it 'is settings.extensions.blocked' do + expect(CLUSTER_KILLSWITCH_ROUTING_KEY).to eq('settings.extensions.blocked') + end + end + + describe '#routing_key' do + it 'returns settings.extensions.blocked' do + expect(msg.routing_key).to eq('settings.extensions.blocked') + end + end + + describe '#exchange' do + it 'targets the ClusterControl exchange' do + expect(msg.exchange).to eq(Legion::Extensions::Node::Transport::Exchanges::ClusterControl) + end + end + + describe '#type' do + it 'returns "task"' do + expect(msg.type).to eq('task') + end + end + + describe '#encrypt?' do + it 'returns false' do + expect(msg.encrypt?).to be false + end + end + + describe '#message' do + it 'includes function update_settings' do + expect(msg.message[:function]).to eq('update_settings') + end + + it 'sets restart to true for immediate effect' do + expect(msg.message[:restart]).to be true + end + + it 'wraps extension in a blocked array' do + expect(msg.message[:settings][:extensions][:blocked]).to eq(['my-ext']) + end + end + + describe '#validate' do + context 'when extension is a String' do + it 'sets @valid to true without raising' do + expect { msg.validate }.not_to raise_error + expect(msg.instance_variable_get(:@valid)).to be true + end + end + + context 'when extension is not a String' do + subject(:bad_msg) { CLUSTER_KILLSWITCH_TEST_CLASS.new(extension: :symbol_ext) } + + it 'raises an error' do + expect { bad_msg.validate }.to raise_error(RuntimeError, 'extension must be a String') + end + end + + context 'when extension is missing' do + subject(:missing_msg) { CLUSTER_KILLSWITCH_TEST_CLASS.new } + + it 'raises an error' do + expect { missing_msg.validate }.to raise_error(RuntimeError, 'extension must be a String') + end + end + end +end diff --git a/spec/legion/extensions/node/transport/messages/cluster_settings_spec.rb b/spec/legion/extensions/node/transport/messages/cluster_settings_spec.rb new file mode 100644 index 0000000..2e84130 --- /dev/null +++ b/spec/legion/extensions/node/transport/messages/cluster_settings_spec.rb @@ -0,0 +1,120 @@ +# frozen_string_literal: true + +require 'spec_helper' +require 'legion/extensions/node/transport/exchanges/cluster_control' + +# Mirror ClusterSettings message logic in a proxy class to avoid superclass mismatch. +CLUSTER_SETTINGS_TEST_CLASS = Class.new do + def initialize(**opts) + @options = opts + end + + def routing_key + @options.fetch(:routing_key, 'settings') + end + + def exchange + Legion::Extensions::Node::Transport::Exchanges::ClusterControl + end + + def type + 'task' + end + + def encrypt? + false + end + + def message + { + function: 'update_settings', + settings: @options[:settings], + restart: @options.fetch(:restart, false) + } + end + + def validate + raise 'settings must be a Hash' unless @options[:settings].is_a?(Hash) + + @valid = true + end +end + +RSpec.describe 'ClusterSettings message' do + let(:settings_hash) { { feature_flags: { new_feature: true } } } + + subject(:msg) { CLUSTER_SETTINGS_TEST_CLASS.new(settings: settings_hash) } + + describe '#routing_key' do + it 'defaults to "settings"' do + expect(msg.routing_key).to eq('settings') + end + + it 'uses provided routing_key' do + m = CLUSTER_SETTINGS_TEST_CLASS.new(settings: settings_hash, routing_key: 'flags.beta') + expect(m.routing_key).to eq('flags.beta') + end + end + + describe '#exchange' do + it 'targets the ClusterControl exchange' do + expect(msg.exchange).to eq(Legion::Extensions::Node::Transport::Exchanges::ClusterControl) + end + end + + describe '#type' do + it 'returns "task"' do + expect(msg.type).to eq('task') + end + end + + describe '#encrypt?' do + it 'returns false' do + expect(msg.encrypt?).to be false + end + end + + describe '#message' do + it 'includes function update_settings' do + expect(msg.message[:function]).to eq('update_settings') + end + + it 'includes the settings hash' do + expect(msg.message[:settings]).to eq(settings_hash) + end + + it 'defaults restart to false' do + expect(msg.message[:restart]).to be false + end + + it 'passes restart: true when specified' do + m = CLUSTER_SETTINGS_TEST_CLASS.new(settings: settings_hash, restart: true) + expect(m.message[:restart]).to be true + end + end + + describe '#validate' do + context 'when settings is a Hash' do + it 'sets @valid to true without raising' do + expect { msg.validate }.not_to raise_error + expect(msg.instance_variable_get(:@valid)).to be true + end + end + + context 'when settings is not a Hash' do + subject(:bad_msg) { CLUSTER_SETTINGS_TEST_CLASS.new(settings: 'not_a_hash') } + + it 'raises an error' do + expect { bad_msg.validate }.to raise_error(RuntimeError, 'settings must be a Hash') + end + end + + context 'when settings is missing' do + subject(:missing_msg) { CLUSTER_SETTINGS_TEST_CLASS.new } + + it 'raises an error' do + expect { missing_msg.validate }.to raise_error(RuntimeError, 'settings must be a Hash') + end + end + end +end diff --git a/spec/legion/extensions/node/transport/queues/cluster_control_spec.rb b/spec/legion/extensions/node/transport/queues/cluster_control_spec.rb new file mode 100644 index 0000000..0fb102b --- /dev/null +++ b/spec/legion/extensions/node/transport/queues/cluster_control_spec.rb @@ -0,0 +1,48 @@ +# frozen_string_literal: true + +require 'spec_helper' + +# Mirror ClusterControl queue logic in a proxy class to avoid superclass mismatch +# with Legion::Transport::Queue which is not available in the test environment. +CLUSTER_CONTROL_QUEUE_TEST_CLASS = Class.new do + def queue_name + "legion.cluster.control.#{Legion::Settings[:client][:name]}" + end + + def queue_options + { durable: false, exclusive: false, auto_delete: true, + arguments: { 'x-queue-type': 'classic' } } + end +end + +RSpec.describe 'ClusterControl queue' do + subject(:queue) { CLUSTER_CONTROL_QUEUE_TEST_CLASS.new } + + before do + allow(Legion::Settings).to receive(:[]).with(:client).and_return({ name: 'test-node' }) + end + + describe '#queue_name' do + it 'includes the node name' do + expect(queue.queue_name).to eq('legion.cluster.control.test-node') + end + end + + describe '#queue_options' do + it 'is not durable' do + expect(queue.queue_options[:durable]).to be false + end + + it 'is not exclusive' do + expect(queue.queue_options[:exclusive]).to be false + end + + it 'auto-deletes when node disconnects' do + expect(queue.queue_options[:auto_delete]).to be true + end + + it 'uses classic queue type' do + expect(queue.queue_options[:arguments][:'x-queue-type']).to eq('classic') + end + end +end