fix: give payload blocks an actor receiver and a context

CHANGELOG.md

@@ -2,6 +2,32 @@
 
 ## Unreleased
 
+- Run payload broadcast blocks against the actor instance, like every other
+  block in the actor DSL. `self` was the actor class, so an actor instance
+  method called from a payload block raised
+  `NoMethodError: undefined method 'x' for class PlaymatRoom`. Blocks keep
+  receiving the actor and the authorization context as arguments, so the
+  documented signature is unaffected. A block that relied on the class receiver
+  now raises `InvalidPayloadBroadcast` naming the method and the change instead
+  of an unexplained `NameError`.
+- Add `payload_authorization_context`, the payload counterpart to
+  `component_authorization_context`. Payloads are computed inside the channel,
+  so without a resolver the payload block and its `authorize_query` call
+  received the raw Action Cable connection while a controller render passed an
+  application object, and the authorization hook had to tell them apart. The
+  resolver may also accept `payload_name:`. It defaults to returning the
+  connection unchanged.
+- Confine a failing payload to itself. A raising payload block propagated out of
+  the channel: on subscribe it rejected the subscription, and on a broadcast it
+  abandoned the remaining payload names, which showed up in the browser only as
+  reactive updates that stopped arriving. A failure is now reported as
+  `solid_objects.payload_broadcast_failed` with the actor type, actor id,
+  payload name, and exception class, and delivery continues. The exception
+  message is deliberately excluded so subscriber state cannot leak into logs. A
+  revision with a failed payload does not advance the delivery watermark, so a
+  transient failure is retried on the next broadcast instead of being recorded
+  as delivered and deduplicated away.
+
 - Run retention on the supervisor rather than leaving it configured but
   unscheduled. Every actor call writes a durable message row, so a policy that
   nothing invokes let history grow without bound until an application scheduled

app/controllers/solid_objects/components_controller.rb

@@ -138,26 +138,11 @@ def refresh(payload)
     # @rbs (Array[ComponentRegistration]) -> untyped
     def component_authorization_context(registrations)
       callable = SolidObjects.configuration.component_authorization_context
-      return callable.call(controller: self) unless accepts_registrations?(callable)
-
-      callable.call(controller: self, registrations:)
-    end
-
-    # A lambda answers `parameters` directly; a callable object answers it
-    # through its `call` method.
-    # @rbs (untyped) -> bool
-    def accepts_registrations?(callable)
-      callable_parameters(callable).any? do |type, name|
-        type == :keyrest || (%i[key keyreq].include?(type) && name == :registrations)
+      unless CallableKeywords.accepts?(callable, :registrations)
+        return callable.call(controller: self)
       end
-    end
-
-    # @rbs (untyped) -> Array[[ Symbol, Symbol ]]
-    def callable_parameters(callable)
-      return callable.parameters if callable.respond_to?(:parameters)
-      return callable.method(:call).parameters if callable.respond_to?(:call)
 
-      []
+      callable.call(controller: self, registrations:)
     end
 
     # @rbs (ComponentRegistration) -> Hash[Symbol, untyped]

docs/realtime.md

@@ -237,9 +237,11 @@ single actor mutation changes several components, an application pays several
 round trips for one logical update. A payload broadcast collapses that into one
 message on the stream the page already has open.
 
-Declare the payload on the actor. The block receives the actor and the
-subscriber's authorization context, and it runs **once per subscriber**, so two
-sessions watching the same actor never see each other's private state:
+Declare the payload on the actor. The block runs against the actor instance,
+like every other block in the actor DSL, and receives the actor and the
+subscriber's authorization context as arguments. It runs **once per
+subscriber**, so two sessions watching the same actor never see each other's
+private state:
 
 ```ruby
 class PlaymatRoom < SolidObjects::Actor
@@ -259,6 +261,16 @@ class PlaymatRoom < SolidObjects::Actor
 end
 ```
 
+Because the block runs against the actor, an actor instance method is reachable
+without a receiver, so shared logic does not have to be duplicated into the
+block:
+
+```ruby
+broadcast_payload :playmat_state do |_room, authorization|
+  { "turn" => turn, "hand" => hand_for(authorization.session_id) }
+end
+```
+
 Subscribe the scope to it:
 
 ```erb
@@ -298,6 +310,26 @@ Payload blocks read committed actor state through the same snapshot components
 use. They cannot write application records, and the return value must be a JSON
 object or array so the wire format stays inspectable.
 
+A payload is one subscriber's view of one name, so a failure is confined to it.
+A raising block does not reject the subscription, stop the other payload names,
+or stop component refreshes on the same connection. The failure is reported as
+`solid_objects.payload_broadcast_failed` carrying the actor type, actor id,
+payload name, and exception class. The exception message is deliberately not
+included: a payload block reads subscriber state, so its message is the one
+place that state could leak into logs.
+
+A revision with a failed payload does not advance the delivery watermark, so a
+transient failure is retried on the next broadcast rather than being recorded as
+delivered. Retries are driven by broadcasts rather than a timer, so a payload
+that fails persistently retries once per actor mutation and reports each
+attempt. A repeating stream of `payload_broadcast_failed` for one `payload_name`
+therefore means a persistent fault in that block, not a one-off; a single event
+that does not recur was transient and has already been recovered.
+
+A payload the subscriber cannot query is skipped rather than served partially;
+that decision is stable, so it settles the revision and the skip is silent by
+design.
+
 ### mtg-playmat before and after
 
 Before, one mutation that touched three observables produced three refresh
@@ -350,13 +382,31 @@ end
 Callbacks that accept only `controller:` continue to work; the extra keyword is
 passed only to callables that declare it.
 
-The three contexts are intentionally different:
+Payloads have the same resolver, because they are computed inside the channel
+rather than in a controller. Without one, a payload block and its
+`authorize_query` call receive the Cable connection while a controller render
+passes an application object, and the authorization hook has to tell them
+apart. Resolve both to the same type and it does not:
+
+```ruby
+configuration.component_authorization_context = ->(controller:) { controller.current_account }
+configuration.payload_authorization_context = ->(connection:) { connection.current_account }
+```
+
+The resolved value is what the payload block receives as its second argument and
+what `authorize_query` receives as `authorization_context`. A resolver may also
+accept `payload_name:` when the subject depends on which payload was requested.
+The default returns the connection unchanged, so an application that has not
+configured one is unaffected.
+
+The contexts are intentionally different:
 
 | Boundary | Authorization context |
 | --- | --- |
 | Initial Action View render | Explicit `authorization_context:` passed to `solid_object` |
 | Action Cable subscription | The authenticated Cable connection |
 | Component refresh | Value returned by `component_authorization_context` for the engine controller request |
+| State payload | Value returned by `payload_authorization_context` for the Cable connection |
 
 Do not substitute a signed token for any of them. Keys and locals are visible
 to the browser and signed for integrity, not encrypted or authorized. Never

docs/roadmap.md

@@ -79,7 +79,11 @@
   though it costs one browser request. Reconnect convergence previously
   bypassed batching entirely, issuing one request per stale component at the
   moment a restart reconnects every client at once; it now shares the batching
-  the live invalidation path uses.
+  the live invalidation path uses. Payload delivery over Action Cable was
+  untested end to end, which is how a raising payload block came to reject the
+  subscription; it is now covered and confined, and the payload authorization
+  context is resolved through `payload_authorization_context` rather than
+  handing the block a raw Cable connection.
 - Backpressure: mailbox/payload/state/result caps and fair yields exist;
   distributed per-actor rate limits and global admission control do not.
 - Administration: actor and dead-letter views plus policy hooks exist; richer

lib/generators/solid_objects/templates/solid_objects.rb

@@ -52,6 +52,14 @@
   # Configure component_authorization_context to return the authenticated
   # principal used for reactive component refreshes.
 
+  # Payloads are delivered over Action Cable, so without a resolver the payload
+  # block and authorize_query receive the Cable connection while a controller
+  # render passes an application object. Resolve both to the same type and the
+  # authorization hook stops having to tell them apart:
+  #
+  # configuration.component_authorization_context = ->(controller:) { controller.current_account }
+  # configuration.payload_authorization_context = ->(connection:) { connection.current_account }
+
   # On hosts where shell access is already an authenticated administrative
   # boundary, this enables only gem commands that pass the CLI context:
   #

lib/solid_objects.rb

@@ -13,6 +13,7 @@
 require "solid_objects/version"
 require "solid_objects/errors"
 require "solid_objects/sync_deadline"
+require "solid_objects/callable_keywords"
 require "solid_objects/configuration"
 require "solid_objects/instrumentation"
 require "solid_objects/log_subscriber"

lib/solid_objects/actor_channel.rb

@@ -78,19 +78,59 @@ def transmit_state_payloads(snapshot)
       return if payload_names.nil? || payload_names.empty?
       return unless newer_payload_revision?(snapshot)
 
-      payload_names.each do |name|
-        payload = PayloadBroadcast.new(
-          snapshot:,
-          name:,
-          authorization_context: connection
-        ).call
-        transmit TurboStreamRenderer.state_payload(payload)
-      rescue Unauthorized
-        next
-      end
+      # The watermark records what the subscriber has, so a revision with a
+      # failed payload must not advance it: dedup would skip every later
+      # attempt at that revision and the actor may not mutate again for a long
+      # time. Every name is still attempted before the decision is made.
+      attempts = payload_names.map { |name| transmit_state_payload(snapshot, name) }
+      return if attempts.any?(false)
+
       @payload_revision = [ snapshot.instance_id, snapshot.revision ]
     end
 
+    # A payload is one subscriber's view of one name. Letting it raise through
+    # here would reject the subscription or abandon the rest of a broadcast, so
+    # a failure is confined to the payload that caused it and reported. The
+    # exception message is deliberately not instrumented: a payload block reads
+    # actor state, so its message is the one place subscriber state could leak
+    # into logs.
+    #
+    # Returns whether this revision was settled for the name. An unauthorized
+    # payload is settled: the decision is stable, so retrying it would only
+    # re-deliver its authorized siblings.
+    # @rbs (ActorSnapshot, String) -> bool
+    def transmit_state_payload(snapshot, name)
+      payload = PayloadBroadcast.new(
+        snapshot:,
+        name:,
+        authorization_context: payload_authorization_context(name)
+      ).call
+      transmit TurboStreamRenderer.state_payload(payload)
+      true
+    rescue Unauthorized
+      true
+    rescue => error
+      SolidObjects.instrument(
+        :payload_broadcast_failed,
+        actor_type: reference.actor_type,
+        actor_id: reference.actor_id,
+        payload_name: name,
+        error_class: error.class.name
+      )
+      false
+    end
+
+    # Resolves the Cable connection to whatever the application uses as an
+    # authorization subject, so a payload block and `authorize_query` see the
+    # same object a controller render would pass.
+    # @rbs (String) -> untyped
+    def payload_authorization_context(name)
+      callable = SolidObjects.configuration.payload_authorization_context
+      return callable.call(connection:) unless CallableKeywords.accepts?(callable, :payload_name)
+
+      callable.call(connection:, payload_name: name)
+    end
+
     # @rbs (ActorSnapshot) -> bool
     def newer_payload_revision?(snapshot)
       current = @payload_revision

lib/solid_objects/callable_keywords.rb

@@ -0,0 +1,29 @@
+# rbs_inline: enabled
+
+module SolidObjects
+  # Authorization context resolvers gained keywords after applications had
+  # already written them, so a resolver is called with what it declared it
+  # accepts rather than with everything the caller could offer.
+  module CallableKeywords
+    class << self
+      # @rbs (untyped, Symbol) -> bool
+      def accepts?(callable, keyword)
+        parameters(callable).any? do |type, name|
+          type == :keyrest || (%i[key keyreq].include?(type) && name == keyword)
+        end
+      end
+
+      private
+
+      # A lambda answers `parameters` directly; a callable object answers it
+      # through its `call` method.
+      # @rbs (untyped) -> Array[[ Symbol, Symbol ]]
+      def parameters(callable)
+        return callable.parameters if callable.respond_to?(:parameters)
+        return callable.method(:call).parameters if callable.respond_to?(:call)
+
+        []
+      end
+    end
+  end
+end

lib/solid_objects/configuration.rb

@@ -40,6 +40,7 @@ class Configuration
     # @rbs @wake_up_adapter: untyped
     # @rbs @component_path_resolver: Proc?
     # @rbs @component_authorization_context: Proc
+    # @rbs @payload_authorization_context: Proc
     # @rbs @authorize_message: Proc
     # @rbs @authorize_query: Proc
     # @rbs @authorize_destroy: Proc
@@ -84,6 +85,7 @@ class Configuration
       :wake_up_adapter,
       :component_path_resolver,
       :component_authorization_context,
+      :payload_authorization_context,
       :authorize_message,
       :authorize_query,
       :authorize_destroy,
@@ -129,6 +131,7 @@ def initialize
       @wake_up_adapter = nil
       @component_path_resolver = nil
       @component_authorization_context = ->(controller:) { controller }
+      @payload_authorization_context = ->(connection:) { connection }
       @logger = if defined?(Rails) && Rails.respond_to?(:logger) && Rails.logger
         Rails.logger
       else
@@ -183,6 +186,9 @@ def validate!
       unless component_authorization_context.respond_to?(:call)
         raise ArgumentError, "component_authorization_context must respond to call"
       end
+      unless payload_authorization_context.respond_to?(:call)
+        raise ArgumentError, "payload_authorization_context must respond to call"
+      end
 
       self
     end

lib/solid_objects/payload_broadcast.rb

@@ -41,7 +41,7 @@ def call
     # @rbs (ActorDefinition::Handler) -> untyped
     def rendered_payload(handler)
       payload = Serialization.dump(
-        handler.block.call(snapshot.actor, authorization_context),
+        evaluated_payload(handler),
         max_bytes: MAXIMUM_PAYLOAD_BYTES
       )
       return payload if payload.is_a?(Hash) || payload.is_a?(Array)
@@ -50,6 +50,34 @@ def rendered_payload(handler)
         "payload broadcast #{name.inspect} must return a JSON object or array"
     end
 
+    # The block runs against the actor instance, like every other block in the
+    # actor DSL, and still receives the actor and the resolved authorization
+    # context as arguments, so blocks written to the documented signature are
+    # unaffected.
+    # @rbs (ActorDefinition::Handler) -> untyped
+    def evaluated_payload(handler)
+      actor = snapshot.actor
+      actor.instance_exec(actor, authorization_context, &handler.block)
+    rescue NameError => error
+      raise unless class_level_receiver?(error)
+
+      raise InvalidPayloadBroadcast,
+        "payload broadcast #{name.inspect} called #{error.name.inspect} on the " \
+          "actor class. Payload blocks now run against the actor instance, like " \
+          "every other actor block. Call it on the class explicitly."
+    end
+
+    # Distinguishes a block that relied on the old class-level receiver from an
+    # ordinary typo, so the one behaviour change reports itself instead of
+    # surfacing as an unexplained NameError.
+    # @rbs (NameError[untyped]) -> bool
+    def class_level_receiver?(error)
+      error.receiver.equal?(snapshot.actor) &&
+        snapshot.actor_class.respond_to?(error.name)
+    rescue ArgumentError, NameError
+      false
+    end
+
     # @rbs () -> void
     def authorize!
       authorized = SolidObjects.configuration.authorize_query.call(

sig/generated/controllers/solid_objects/components_controller.rbs

@@ -27,14 +27,6 @@ module SolidObjects
     # @rbs (Array[ComponentRegistration]) -> untyped
     def component_authorization_context: (Array[ComponentRegistration]) -> untyped
 
-    # A lambda answers `parameters` directly; a callable object answers it
-    # through its `call` method.
-    # @rbs (untyped) -> bool
-    def accepts_registrations?: (untyped) -> bool
-
-    # @rbs (untyped) -> Array[[ Symbol, Symbol ]]
-    def callable_parameters: (untyped) -> Array[[ Symbol, Symbol ]]
-
     # @rbs (ComponentRegistration) -> Hash[Symbol, untyped]
     def registration_payload: (ComponentRegistration) -> Hash[Symbol, untyped]