Merge pull request #33 from cardmagic/agent/supervisor-component-registry

Let an extension run inside the supervisor

Author: cardmagic

CHANGELOG.md

@@ -1,5 +1,32 @@
 # Changelog
 
+## 0.11.0 - 2026-08-11
+
+- Add `SolidObjects.configuration.register_component`. An extension gem can now
+  register a long running component, and the supervisor runs it beside the
+  workers, the effect executors, the broadcast executors, and the reminder
+  schedulers. The component joins the same supervision, replacement, and
+  shutdown timeout. Without it, an extension has to ask an operator to run and
+  monitor a second process for work that belongs to the same runtime. A
+  registered component must answer `run`, `request_shutdown`, `stopped?`, and
+  `stop`. The supervisor checks that contract when it builds the component and
+  raises `ArgumentError` when a method is missing. Registration never calls the
+  block, so a component may need a database connection that the application
+  does not have while it boots.
+
+- Stop the components already built when a later one fails. The supervisor
+  builds its components one after another, so a factory that raised, or a
+  component that failed the contract check, left the earlier ones constructed
+  and unreachable while they still held whatever their constructors took. Each
+  one now receives `stop`, and a failure inside that cleanup never replaces the
+  failure that caused it.
+
+- Replace a crashed component through the builder that made it. The supervisor
+  called `component.class.new`, which discards every constructor argument, so a
+  component built with arguments returned with its defaults after a crash. Each
+  component now keeps its builder. The built in components take no constructor
+  arguments, so their behavior does not change.
+
 ## 0.10.3 - 2026-08-11
 
 - Delete every actor-owned row in `SolidObjects::TestHelper#reset_actors!`. It

Gemfile.lock

@@ -1,7 +1,7 @@
 PATH
   remote: .
   specs:
-    solid_objects (0.10.3)
+    solid_objects (0.11.0)
       actioncable (>= 8.0)
       actionpack (>= 8.0)
       actionview (>= 8.0)
@@ -383,7 +383,7 @@ CHECKSUMS
   rubocop-rails-omakase (1.1.0) sha256=2af73ac8ee5852de2919abbd2618af9c15c19b512c4cfc1f9a5d3b6ef009109d
   ruby-progressbar (1.13.0) sha256=80fc9c47a9b640d6834e0dc7b3c94c9df37f08cb072b7761e4a71e22cff29b33
   securerandom (0.4.1) sha256=cc5193d414a4341b6e225f0cb4446aceca8e50d5e1888743fac16987638ea0b1
-  solid_objects (0.10.3)
+  solid_objects (0.11.0)
   sqlite3 (2.9.5-aarch64-linux-gnu) sha256=78075b6337d3d182c6d2b4691049ed45cd220826160c9ea18946bf6a1de200dc
   sqlite3 (2.9.5-aarch64-linux-musl) sha256=18c801185deb4adc01ddb281e8f672a39e3d1729979ca91e39439cd3eac0402d
   sqlite3 (2.9.5-arm-linux-gnu) sha256=1bdfca0c7d63998c60b0f4a8e3c8df2d33800ccc4abd2d612eddbbbc92a4c48b

README.md

@@ -458,6 +458,39 @@ Deploy and monitor that process before enabling any feature marked as requiring
 a runtime role. A missing worker never makes a durable `async` message
 disappear, but it leaves the message pending indefinitely.
 
+### Running an extension in the same process
+
+An extension gem can register its own long-running component, and
+`solid_objects start` runs it beside the built-in roles. The component joins the
+same supervision, the same replacement after a crash, and the same shutdown
+timeout, so an operator deploys and monitors one process instead of two:
+
+```ruby
+SolidObjects.configure do |configuration|
+  configuration.register_component { MyExtension::FlushEngine.new }
+end
+```
+
+Pass `count:` for more than one instance. The block runs once for each instance,
+and again when the supervisor replaces a crashed one, so no two components share
+an object.
+
+A registered component answers four methods, the contract the built-in roles
+already keep:
+
+| Method | Purpose |
+| --- | --- |
+| `run` | Runs the loop. The supervisor calls it in its own thread |
+| `request_shutdown` | Asks the loop to finish. It must make `run` return |
+| `stopped?` | Reports whether the component already finished |
+| `stop` | Forces cleanup when the shutdown timeout expires first |
+
+The supervisor checks that contract when it builds the component, and a missing
+method raises `ArgumentError` as the supervisor starts, rather than hanging a
+shutdown later. Registration itself never calls the block, so a component is
+free to need a database connection that the application does not have while it
+boots.
+
 ## Defining an actor
 
 The Durable Object class becomes an ordinary Ruby class:

lib/solid_objects/configuration.rb

@@ -92,6 +92,9 @@ class Configuration
       :authorize_subscription,
       :authorize_administration
 
+    # @rbs @additional_components: Array[untyped]
+    attr_reader :additional_components
+
     # @rbs () -> void
     def initialize
       @table_name_prefix = "solid_objects_"
@@ -142,6 +145,45 @@ def initialize
       @authorize_destroy = ->(**) { false }
       @authorize_subscription = ->(**) { false }
       @authorize_administration = ->(**) { false }
+      @additional_components = []
+    end
+
+    # Registers a long running component that the supervisor runs beside its own
+    # workers. An extension gem uses this to share one process, rather than ask
+    # an operator to run and monitor a second one.
+    #
+    # The block must return an object that answers `run`, `request_shutdown`,
+    # `stopped?`, and `stop`, which is the contract the built in components
+    # already keep. The supervisor calls the block once for each supervisor it
+    # builds, and again when it replaces a crashed component, so two
+    # supervisors never share one component instance.
+    #
+    # @rbs (?count: Integer) { () -> untyped } -> void
+    def register_component(count: 1, &factory)
+      raise ArgumentError, "register_component requires a block" unless factory
+      raise ArgumentError, "count must be positive" unless count.positive?
+
+      count.times { @additional_components << factory }
+    end
+
+    # The supervisor checks the contract here rather than at registration,
+    # because a component often needs a database connection to exist, and
+    # registration happens while the application boots.
+    # @rbs () -> Array[untyped]
+    def build_additional_components
+      additional_components.map { |factory| factory.call.tap { |component| validate_component!(component) } }
+    end
+
+    # A component that misses part of the contract would hang the supervisor at
+    # shutdown, or crash the moment it starts. The build fails instead, where
+    # the caller can read the reason.
+    # @rbs (untyped) -> void
+    def validate_component!(component)
+      %i[run request_shutdown stopped? stop].each do |method_name|
+        next if component.respond_to?(method_name)
+
+        raise ArgumentError, "a registered component must respond to #{method_name}"
+      end
     end
 
     # @rbs () -> self

lib/solid_objects/supervisor.rb

@@ -19,12 +19,13 @@ def initialize(
       broadcast_worker_count: SolidObjects.configuration.broadcast_worker_count,
       reminder_scheduler_count: SolidObjects.configuration.reminder_scheduler_count
     )
-      @components = build_components(
+      @builders = component_builders(
         worker_count:,
         effect_worker_count:,
         broadcast_worker_count:,
         reminder_scheduler_count:
       )
+      @components = build_all(@builders)
       @threads = []
       @monitor = nil
       @started = false
@@ -77,7 +78,7 @@ def stop
 
     private
 
-    attr_reader :components, :threads
+    attr_reader :components, :threads, :builders
 
     # A role that raises leaves its thread dead. Without replacement the
     # process keeps running while quietly doing less work, so the supervisor
@@ -116,7 +117,11 @@ def replace_dead_roles
         replaced = @lifecycle.synchronize do
           next false unless @started
 
-          replacement = component.class.new
+          # A component built by this supervisor has a builder, which carries
+          # whatever the constructor was given. A component put in place by
+          # other means has none, so the class is the only thing left to go on.
+          builder = builders[index] || -> { component.class.new }
+          replacement = builder.call
           components[index] = replacement
           threads[index] = supervise(replacement)
           replacement
@@ -250,17 +255,53 @@ def release_wake_up
       nil
     end
 
-    # @rbs (worker_count: Integer, effect_worker_count: Integer, broadcast_worker_count: Integer, reminder_scheduler_count: Integer) -> Array[Worker | EffectExecutor | ReminderScheduler | BroadcastExecutor]
-    def build_components(
+    # A constructor can take a resource, and a later builder can raise. Without
+    # this, the components built first would be dropped while still holding
+    # whatever they took, and nothing would ever give it back.
+    # @rbs (Array[^() -> untyped]) -> Array[untyped]
+    def build_all(builders)
+      built = []
+      builders.each do |builder|
+        # The component joins the list before the contract check, so a
+        # component that fails the check is stopped along with the rest.
+        built << (component = builder.call)
+        SolidObjects.configuration.validate_component!(component)
+      end
+      built
+    rescue Exception # rubocop:disable Lint/RescueException
+      built.each { |component| stop_after_failed_build(component) }
+      raise
+    end
+
+    # The failure that stopped the build is the one worth reporting, so a
+    # failure inside the cleanup never replaces it.
+    # @rbs (untyped) -> void
+    def stop_after_failed_build(component)
+      component.stop if component.respond_to?(:stop)
+    rescue Exception => error # rubocop:disable Lint/RescueException
+      SolidObjects.instrument(
+        :"supervisor.component_cleanup_failed",
+        role: component.class.name,
+        error_class: error.class.name
+      )
+    end
+
+    # Each component keeps the builder that made it, so a replacement after a
+    # crash is built the same way as the original. Components registered
+    # through the configuration run beside the built in ones, under the same
+    # supervision, restart, and shutdown timeout.
+    # @rbs (worker_count: Integer, effect_worker_count: Integer, broadcast_worker_count: Integer, reminder_scheduler_count: Integer) -> Array[^() -> untyped]
+    def component_builders(
       worker_count:,
       effect_worker_count:,
       broadcast_worker_count:,
       reminder_scheduler_count:
     )
-      Array.new(worker_count) { Worker.new } +
-        Array.new(effect_worker_count) { EffectExecutor.new } +
-        Array.new(broadcast_worker_count) { BroadcastExecutor.new } +
-        Array.new(reminder_scheduler_count) { ReminderScheduler.new }
+      Array.new(worker_count) { -> { Worker.new } } +
+        Array.new(effect_worker_count) { -> { EffectExecutor.new } } +
+        Array.new(broadcast_worker_count) { -> { BroadcastExecutor.new } } +
+        Array.new(reminder_scheduler_count) { -> { ReminderScheduler.new } } +
+        SolidObjects.configuration.additional_components
     end
 
     # @rbs () -> void

lib/solid_objects/version.rb

@@ -1,5 +1,5 @@
 # rbs_inline: enabled
 
 module SolidObjects
-  VERSION = "0.10.3"
+  VERSION = "0.11.0"
 end

sig/generated/lib/solid_objects/configuration.rbs

@@ -2,6 +2,10 @@
 
 module SolidObjects
   class Configuration
+    @process_alive_threshold: Float
+
+    @shutdown_timeout: Float
+
     @supervisor_monitor_interval: Float
 
     @retention_interval: Float
@@ -86,10 +90,6 @@ module SolidObjects
 
     @process_heartbeat_interval: Float
 
-    @process_alive_threshold: Float
-
-    @shutdown_timeout: Float
-
     attr_accessor table_name_prefix: untyped
 
     attr_accessor polling_interval: untyped
@@ -178,9 +178,37 @@ module SolidObjects
 
     attr_accessor authorize_administration: untyped
 
+    # @rbs @additional_components: Array[untyped]
+    attr_reader additional_components: untyped
+
     # @rbs () -> void
     def initialize: () -> void
 
+    # Registers a long running component that the supervisor runs beside its own
+    # workers. An extension gem uses this to share one process, rather than ask
+    # an operator to run and monitor a second one.
+    #
+    # The block must return an object that answers `run`, `request_shutdown`,
+    # `stopped?`, and `stop`, which is the contract the built in components
+    # already keep. The supervisor calls the block once for each supervisor it
+    # builds, and again when it replaces a crashed component, so two
+    # supervisors never share one component instance.
+    #
+    # @rbs (?count: Integer) { () -> untyped } -> void
+    def register_component: (?count: Integer) { () -> untyped } -> void
+
+    # The supervisor checks the contract here rather than at registration,
+    # because a component often needs a database connection to exist, and
+    # registration happens while the application boots.
+    # @rbs () -> Array[untyped]
+    def build_additional_components: () -> Array[untyped]
+
+    # A component that misses part of the contract would hang the supervisor at
+    # shutdown, or crash the moment it starts. The build fails instead, where
+    # the caller can read the reason.
+    # @rbs (untyped) -> void
+    def validate_component!: (untyped) -> void
+
     # @rbs () -> self
     def validate!: () -> self
 

sig/generated/lib/solid_objects/supervisor.rbs

@@ -36,6 +36,8 @@ module SolidObjects
 
     attr_reader threads: untyped
 
+    attr_reader builders: untyped
+
     # A role that raises leaves its thread dead. Without replacement the
     # process keeps running while quietly doing less work, so the supervisor
     # watches its threads and restarts any that stopped before shutdown.
@@ -101,8 +103,23 @@ module SolidObjects
     # @rbs () -> void
     def release_wake_up: () -> void
 
-    # @rbs (worker_count: Integer, effect_worker_count: Integer, broadcast_worker_count: Integer, reminder_scheduler_count: Integer) -> Array[Worker | EffectExecutor | ReminderScheduler | BroadcastExecutor]
-    def build_components: (worker_count: Integer, effect_worker_count: Integer, broadcast_worker_count: Integer, reminder_scheduler_count: Integer) -> Array[Worker | EffectExecutor | ReminderScheduler | BroadcastExecutor]
+    # A constructor can take a resource, and a later builder can raise. Without
+    # this, the components built first would be dropped while still holding
+    # whatever they took, and nothing would ever give it back.
+    # @rbs (Array[^() -> untyped]) -> Array[untyped]
+    def build_all: (Array[^() -> untyped]) -> Array[untyped]
+
+    # The failure that stopped the build is the one worth reporting, so a
+    # failure inside the cleanup never replaces it.
+    # @rbs (untyped) -> void
+    def stop_after_failed_build: (untyped) -> void
+
+    # Each component keeps the builder that made it, so a replacement after a
+    # crash is built the same way as the original. Components registered
+    # through the configuration run beside the built in ones, under the same
+    # supervision, restart, and shutdown timeout.
+    # @rbs (worker_count: Integer, effect_worker_count: Integer, broadcast_worker_count: Integer, reminder_scheduler_count: Integer) -> Array[^() -> untyped]
+    def component_builders: (worker_count: Integer, effect_worker_count: Integer, broadcast_worker_count: Integer, reminder_scheduler_count: Integer) -> Array[^() -> untyped]
 
     # @rbs () -> void
     def join_until_timeout: () -> void