add broadcast exchange for cluster-wide settings push (#2)

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.<name>` 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

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

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)

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

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

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

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

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

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

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