From dc2bdf36209c0fc461778f692206e2577344e760 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Mon, 10 Aug 2026 10:29:18 -0700 Subject: [PATCH 1/3] fix: give payload blocks an actor receiver and a context Three defects reported from a Rails integration, all in the payload path that had no end-to-end coverage. Payload blocks ran through `block.call(actor, context)` while every other block in the actor DSL runs through `instance_exec` on the actor. `self` was therefore the actor class, and an actor instance method called from a payload block raised NoMethodError against the class. They now instance_exec and still receive both arguments, so the documented signature is unchanged. Payloads are computed inside the channel rather than a controller, so they had no counterpart to `component_authorization_context`. The block and its `authorize_query` call received the raw Cable connection while a controller render passed an application object, leaving the application's hook to translate between them. `payload_authorization_context` resolves the connection once, defaults to the identity, and may also accept `payload_name:`. A raising payload block propagated out of the channel. On subscribe it rejected the subscription; on a broadcast it abandoned the remaining payload names. The browser saw only reactive updates that stopped arriving. Failures are now confined per payload name and reported as solid_objects.payload_broadcast_failed. The exception message is excluded on purpose: a payload block reads subscriber state, so its message is the one place that state could reach logs. The keyword detection behind both resolvers moves to CallableKeywords so the two cannot drift. --- CHANGELOG.md | 23 ++ .../solid_objects/components_controller.rb | 21 +- docs/realtime.md | 47 ++- docs/roadmap.md | 6 +- .../solid_objects/templates/solid_objects.rb | 8 + lib/solid_objects.rb | 1 + lib/solid_objects/actor_channel.rb | 48 ++- lib/solid_objects/callable_keywords.rb | 29 ++ lib/solid_objects/configuration.rb | 6 + lib/solid_objects/payload_broadcast.rb | 30 +- .../solid_objects/components_controller.rbs | 8 - .../lib/solid_objects/actor_channel.rbs | 15 + .../lib/solid_objects/callable_keywords.rbs | 16 + .../lib/solid_objects/configuration.rbs | 8 +- .../lib/solid_objects/payload_broadcast.rbs | 13 + test/integration/payload_delivery_test.rb | 379 ++++++++++++++++++ 16 files changed, 614 insertions(+), 44 deletions(-) create mode 100644 lib/solid_objects/callable_keywords.rb create mode 100644 sig/generated/lib/solid_objects/callable_keywords.rbs create mode 100644 test/integration/payload_delivery_test.rb diff --git a/CHANGELOG.md b/CHANGELOG.md index d13807e..efa47d9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,29 @@ ## 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. + - 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 diff --git a/app/controllers/solid_objects/components_controller.rb b/app/controllers/solid_objects/components_controller.rb index 36034da..751fa4f 100644 --- a/app/controllers/solid_objects/components_controller.rb +++ b/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] diff --git a/docs/realtime.md b/docs/realtime.md index d141623..3641095 100644 --- a/docs/realtime.md +++ b/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,15 @@ 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 payload the subscriber cannot query is +skipped rather than served partially, and that skip is silent by design. + ### mtg-playmat before and after Before, one mutation that touched three observables produced three refresh @@ -350,13 +371,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 diff --git a/docs/roadmap.md b/docs/roadmap.md index 3293881..e602567 100644 --- a/docs/roadmap.md +++ b/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 diff --git a/lib/generators/solid_objects/templates/solid_objects.rb b/lib/generators/solid_objects/templates/solid_objects.rb index 9b7dbe4..e002e46 100644 --- a/lib/generators/solid_objects/templates/solid_objects.rb +++ b/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: # diff --git a/lib/solid_objects.rb b/lib/solid_objects.rb index fd33505..f86a9e9 100644 --- a/lib/solid_objects.rb +++ b/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" diff --git a/lib/solid_objects/actor_channel.rb b/lib/solid_objects/actor_channel.rb index d91873f..35f6ece 100644 --- a/lib/solid_objects/actor_channel.rb +++ b/lib/solid_objects/actor_channel.rb @@ -78,19 +78,47 @@ 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 + payload_names.each { |name| transmit_state_payload(snapshot, name) } @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. + # @rbs (ActorSnapshot, String) -> void + def transmit_state_payload(snapshot, name) + payload = PayloadBroadcast.new( + snapshot:, + name:, + authorization_context: payload_authorization_context(name) + ).call + transmit TurboStreamRenderer.state_payload(payload) + rescue Unauthorized + nil + 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 + ) + 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 diff --git a/lib/solid_objects/callable_keywords.rb b/lib/solid_objects/callable_keywords.rb new file mode 100644 index 0000000..b929ae7 --- /dev/null +++ b/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 diff --git a/lib/solid_objects/configuration.rb b/lib/solid_objects/configuration.rb index 6cfb349..232c976 100644 --- a/lib/solid_objects/configuration.rb +++ b/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 diff --git a/lib/solid_objects/payload_broadcast.rb b/lib/solid_objects/payload_broadcast.rb index d62af38..747210e 100644 --- a/lib/solid_objects/payload_broadcast.rb +++ b/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( diff --git a/sig/generated/controllers/solid_objects/components_controller.rbs b/sig/generated/controllers/solid_objects/components_controller.rbs index a96add9..1705184 100644 --- a/sig/generated/controllers/solid_objects/components_controller.rbs +++ b/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] diff --git a/sig/generated/lib/solid_objects/actor_channel.rbs b/sig/generated/lib/solid_objects/actor_channel.rbs index 426e1cb..c24bfc2 100644 --- a/sig/generated/lib/solid_objects/actor_channel.rbs +++ b/sig/generated/lib/solid_objects/actor_channel.rbs @@ -21,6 +21,21 @@ module SolidObjects # @rbs (ActorSnapshot) -> void def transmit_state_payloads: (ActorSnapshot) -> void + # 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. + # @rbs (ActorSnapshot, String) -> void + def transmit_state_payload: (ActorSnapshot, String) -> void + + # 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: (String) -> untyped + # @rbs (ActorSnapshot) -> bool def newer_payload_revision?: (ActorSnapshot) -> bool diff --git a/sig/generated/lib/solid_objects/callable_keywords.rbs b/sig/generated/lib/solid_objects/callable_keywords.rbs new file mode 100644 index 0000000..d346922 --- /dev/null +++ b/sig/generated/lib/solid_objects/callable_keywords.rbs @@ -0,0 +1,16 @@ +# Generated from lib/solid_objects/callable_keywords.rb with RBS::Inline + +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 + # @rbs (untyped, Symbol) -> bool + def self.accepts?: (untyped, Symbol) -> bool + + # A lambda answers `parameters` directly; a callable object answers it + # through its `call` method. + # @rbs (untyped) -> Array[[ Symbol, Symbol ]] + private def self.parameters: (untyped) -> Array[[ Symbol, Symbol ]] + end +end diff --git a/sig/generated/lib/solid_objects/configuration.rbs b/sig/generated/lib/solid_objects/configuration.rbs index 0f5a91a..0d0cac6 100644 --- a/sig/generated/lib/solid_objects/configuration.rbs +++ b/sig/generated/lib/solid_objects/configuration.rbs @@ -2,8 +2,6 @@ module SolidObjects class Configuration - @table_name_prefix: String - @supervisor_monitor_interval: Float @retention_interval: Float @@ -42,6 +40,8 @@ module SolidObjects @component_authorization_context: Proc + @payload_authorization_context: Proc + @authorize_message: Proc @authorize_query: Proc @@ -52,6 +52,8 @@ module SolidObjects @authorize_administration: Proc + @table_name_prefix: String + @polling_interval: Float @sync_polling_interval: Float @@ -164,6 +166,8 @@ module SolidObjects attr_accessor component_authorization_context: untyped + attr_accessor payload_authorization_context: untyped + attr_accessor authorize_message: untyped attr_accessor authorize_query: untyped diff --git a/sig/generated/lib/solid_objects/payload_broadcast.rbs b/sig/generated/lib/solid_objects/payload_broadcast.rbs index d8c153a..4b9f44f 100644 --- a/sig/generated/lib/solid_objects/payload_broadcast.rbs +++ b/sig/generated/lib/solid_objects/payload_broadcast.rbs @@ -29,6 +29,19 @@ module SolidObjects # @rbs (ActorDefinition::Handler) -> untyped def rendered_payload: (ActorDefinition::Handler) -> untyped + # 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: (ActorDefinition::Handler) -> untyped + + # 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?: (NameError[untyped]) -> bool + # @rbs () -> void def authorize!: () -> void end diff --git a/test/integration/payload_delivery_test.rb b/test/integration/payload_delivery_test.rb new file mode 100644 index 0000000..29fa080 --- /dev/null +++ b/test/integration/payload_delivery_test.rb @@ -0,0 +1,379 @@ +# frozen_string_literal: true + +require "database_test_helper" +require "action_cable/test_helper" +require "action_cable/channel/test_case" + +ActionCable.server.config.cable = { "adapter" => "test" } + +# The payload path from a Cable subscription to a transmitted element had no +# coverage, which is how a raised payload block came to take the subscription +# down with it. +class PayloadDeliveryTest < ActionCable::Channel::TestCase + tests SolidObjects::ActorChannel + + class RoomActor < SolidObjects::Actor + actor_type "delivery-room" + + attribute :hands, default: -> { {} } + attribute :turn, default: 1 + + observable :turn + + broadcast_payload :room_state do |room, authorization| + { "turn" => room.turn, "hand" => room.hand_for(authorization.session_id) } + end + + broadcast_payload :spectator_state do |room, _authorization| + { "turn" => room.turn } + end + + # Reached through implicit self by the block below, which is what a payload + # block written like every other actor block expects to be able to do. + broadcast_payload :helper_state do + { "hand" => hand_for("alice") } + end + + broadcast_payload :broken_state do + raise "payload exploded with secret-hand and secret-token" + end + + def hand_for(session_id) = hands.fetch(session_id, []) + + def deal(session_id:, cards:) + self.hands = hands.merge(session_id => cards) + end + + def advance_turn + self.turn += 1 + end + end + + # Models an application authorization object: what a controller render would + # pass, and what the Cable path should be able to produce too. + class Authorization + attr_reader :session_id + + def initialize(session_id) + @session_id = session_id + end + end + + setup do + SolidObjects.reset! + RoomActor.ensure_registered! + SolidObjects.configuration.stream_signing_secret = "payload-delivery-secret" + SolidObjects.configuration.authorize_subscription = ->(**) { true } + SolidObjects.configuration.authorize_message = ->(**) { true } + SolidObjects.configuration.authorize_query = ->(**) { true } + SolidObjects.configuration.component_path_resolver = ->(view_context:) { "/solid_objects/components" } + ActionCable.server.config.logger = Logger.new(nil) + end + + test "the payload authorization context is resolved before authorization runs" do + reference = deal_to("alice") + contexts = [] + SolidObjects.configuration.payload_authorization_context = lambda do |connection:| + Authorization.new(connection.playmat_session_id) + end + SolidObjects.configuration.authorize_query = lambda do |**arguments| + contexts << arguments.fetch(:authorization_context) + true + end + stub_connection(playmat_session_id: "alice") + + subscribe token: payload_token(reference, %w[room_state]) + + assert contexts.any?, "the payload path should have authorized" + assert(contexts.all? { |context| context.is_a?(Authorization) }, + "authorize_query should receive the resolved context, not the raw connection") + end + + test "the payload block receives the resolved authorization context" do + reference = deal_to("alice", %w[Island Forest]) + SolidObjects.configuration.payload_authorization_context = lambda do |connection:| + Authorization.new(connection.playmat_session_id) + end + stub_connection(playmat_session_id: "alice") + + subscribe token: payload_token(reference, %w[room_state]) + + assert_equal %w[Island Forest], delivered_payload("room_state").fetch("hand") + end + + test "the default resolver passes the connection through unchanged" do + reference = deal_to("alice", %w[Island]) + received = [] + SolidObjects.configuration.authorize_query = lambda do |**arguments| + received << arguments.fetch(:authorization_context) + true + end + stub_connection(session_id: "alice") + + subscribe token: payload_token(reference, %w[spectator_state]) + + assert(received.any? { |context| context.respond_to?(:session_id) }, + "without a resolver the connection itself must still arrive") + end + + test "a resolver may also accept the payload name" do + reference = deal_to("alice") + names = [] + SolidObjects.configuration.payload_authorization_context = lambda do |connection:, payload_name:| + names << payload_name + Authorization.new(connection.playmat_session_id) + end + stub_connection(playmat_session_id: "alice") + + subscribe token: payload_token(reference, %w[room_state spectator_state]) + + assert_equal %w[room_state spectator_state], names.uniq.sort + end + + test "the payload block runs with the actor as its receiver" do + reference = deal_to("alice", %w[Island]) + stub_connection(session_id: "alice") + + subscribe token: payload_token(reference, %w[helper_state]) + + assert_equal %w[Island], delivered_payload("helper_state").fetch("hand"), + "an actor instance method should be reachable through implicit self" + end + + test "a raising payload block does not reject the subscription" do + reference = deal_to("alice") + stub_connection(session_id: "alice") + + subscribe token: payload_token(reference, %w[broken_state]) + + assert subscription.confirmed?, "one failed payload must not take the stream down" + assert_has_stream SolidObjects::StreamName.for(reference) + end + + test "a raising payload block is instrumented" do + reference = deal_to("alice") + events = capture_failures do + stub_connection(session_id: "alice") + subscribe token: payload_token(reference, %w[broken_state]) + end + + assert_equal 1, events.length + event = events.first + assert_equal "delivery-room", event[:actor_type] + assert_equal "broken_state", event[:payload_name] + assert_equal "RuntimeError", event[:error_class] + end + + test "the failure event carries no payload state" do + reference = deal_to("alice", %w[secret-hand]) + events = capture_failures do + stub_connection(session_id: "alice") + subscribe token: payload_token(reference, %w[broken_state]) + end + + serialized = events.first.inspect + refute_includes serialized, "secret-hand" + refute_includes serialized, "secret-token" + end + + test "a failing payload does not stop another payload from being delivered" do + reference = deal_to("alice") + stub_connection(session_id: "alice") + + subscribe token: payload_token(reference, %w[broken_state spectator_state]) + + refute_nil delivered_payload("spectator_state"), + "a later payload name must still be delivered" + end + + test "a failing payload does not stop component refreshes on the same connection" do + reference = deal_to("alice") + stub_connection(session_id: "alice") + subscribe( + token: payload_token(reference, %w[broken_state]), + components: JSON.generate([ component_token(reference, revision: 0) ]) + ) + transmissions.clear + + reference.advance_turn + receive_latest_broadcast + + assert component_refreshes.any?, + "component delivery must survive a failing payload on the same connection" + end + + test "an unauthorized payload is skipped rather than delivered" do + reference = deal_to("alice") + SolidObjects.configuration.authorize_query = lambda do |**arguments| + arguments.fetch(:message_name) != "room_state" + end + stub_connection(session_id: "alice") + + subscribe token: payload_token(reference, %w[room_state spectator_state]) + + assert_nil delivered_payload("room_state") + refute_nil delivered_payload("spectator_state") + end + + test "subscribing delivers the current payload" do + reference = deal_to("alice") + reference.advance_turn + stub_connection(session_id: "alice") + + subscribe token: payload_token(reference, %w[spectator_state]) + + assert_equal 2, delivered_payload("spectator_state").fetch("turn") + end + + test "a payload is not re-delivered at the same revision" do + reference = deal_to("alice") + stub_connection(session_id: "alice") + subscribe token: payload_token(reference, %w[spectator_state]) + transmissions.clear + + reference.advance_turn + receive_latest_broadcast + delivered = payload_elements("spectator_state").length + receive_latest_broadcast + + assert_equal delivered, payload_elements("spectator_state").length, + "a repeated revision must not transmit the payload again" + end + + # The one behaviour change, reported at the call site rather than as an + # unexplained NoMethodError. + test "a block relying on the old class receiver explains itself" do + class_scoped = Class.new(SolidObjects::Actor) do + actor_type "payload-class-receiver" + attribute :value, default: 1 + observable :value + def self.table_name = "legacy" + broadcast_payload(:legacy) { { "table" => table_name } } + end + class_scoped.ensure_registered! + reference = class_scoped.ref("one") + + error = assert_raises SolidObjects::InvalidPayloadBroadcast do + SolidObjects::PayloadBroadcast.new( + snapshot: SolidObjects::ActorSnapshot.new(reference), + name: "legacy", + authorization_context: Authorization.new("alice") + ).call + end + + assert_match(/actor instance/, error.message) + assert_match(/table_name/, error.message) + end + + test "an ordinary typo in a payload block is not blamed on the receiver" do + typo_actor = Class.new(SolidObjects::Actor) do + actor_type "payload-typo" + attribute :value, default: 1 + observable :value + broadcast_payload(:typo) { { "value" => valeu } } + end + typo_actor.ensure_registered! + reference = typo_actor.ref("one") + + error = assert_raises NameError do + SolidObjects::PayloadBroadcast.new( + snapshot: SolidObjects::ActorSnapshot.new(reference), + name: "typo", + authorization_context: Authorization.new("alice") + ).call + end + + refute_kind_of SolidObjects::InvalidPayloadBroadcast, error, + "a name the class cannot answer either is not the receiver change" + end + + test "a payload block keeping the documented signature is unaffected" do + reference = deal_to("alice", %w[Island Forest]) + stub_connection(session_id: "alice") + + subscribe token: payload_token(reference, %w[room_state]) + + assert_equal %w[Island Forest], delivered_payload("room_state").fetch("hand"), + "the two-argument form must keep working unchanged" + end + + test "rejects a payload authorization context that cannot be called" do + SolidObjects.configuration.payload_authorization_context = :not_callable + + assert_raises ArgumentError do + SolidObjects.configuration.validate! + end + end + + private + + def deal_to(session_id, cards = %w[Island]) + RoomActor.ref("table").tap do |reference| + reference.deal(session_id:, cards:) + end + end + + def payload_token(reference, payloads) + SolidObjects::StreamToken.generate(reference, observables: [], payloads:) + end + + def component_token(reference, revision:) + SolidObjects::ComponentToken.generate( + reference:, + component_name: "summary", + component_key: nil, + dependencies: %w[turn], + locals: {}, + refresh_method: "replace", + instance_id: 0, + revision:, + refresh_path: "/solid_objects/components" + ) + end + + def capture_failures + events = [] + subscription = ActiveSupport::Notifications.subscribe( + "solid_objects.payload_broadcast_failed" + ) { |event| events << event.payload } + yield + events + ensure + ActiveSupport::Notifications.unsubscribe(subscription) if subscription + end + + def payload_elements(name) + transmissions.select do |transmission| + transmission.to_s.include?(%(data-name="#{name}")) + end + end + + # The element body is HTML-escaped on the way out, exactly as the browser + # receives it. + def delivered_payload(name) + element = payload_elements(name).last + return nil unless element + + body = element[%r{]*>(.*?)}m, 1] + JSON.parse(CGI.unescapeHTML(body)) + end + + def component_refreshes + target = SolidObjects::DomIdentity.component(RoomActor.ref("table"), "summary") + transmissions.select do |transmission| + transmission.to_s.include?(%(target="#{target}")) + end + end + + def receive_latest_broadcast + worker = SolidObjects::Worker.new + worker.run_until_idle + broadcast = SolidObjects::Broadcast.order(:id).last + subscription.__send__( + :receive_broadcast, + SolidObjects::TurboStreamRenderer.observable(broadcast) + ) + ensure + worker&.stop + end +end From 36bb36db51424e1a0d2516a7fc7999dbfce5d0b5 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Mon, 10 Aug 2026 11:02:28 -0700 Subject: [PATCH 2/3] fix: retry a payload revision that failed Confining a payload failure was only half of it. The watermark that deduplicates payload delivery advanced whether or not the payload was delivered, so a transient error marked that revision done: every later attempt at it was deduplicated away, and the subscriber went without until the actor happened to mutate again. The watermark now advances only when every payload for the revision settled. An unauthorized payload counts as settled, since that decision is stable and retrying it would only re-deliver its authorized siblings. Every name is still attempted before the decision is made. --- CHANGELOG.md | 5 ++- docs/realtime.md | 9 ++++-- lib/solid_objects/actor_channel.rb | 18 +++++++++-- .../lib/solid_objects/actor_channel.rbs | 8 +++-- test/integration/payload_delivery_test.rb | 32 +++++++++++++++++++ 5 files changed, 64 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index efa47d9..f0f4600 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -23,7 +23,10 @@ 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. + 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 diff --git a/docs/realtime.md b/docs/realtime.md index 3641095..e00b2db 100644 --- a/docs/realtime.md +++ b/docs/realtime.md @@ -316,8 +316,13 @@ 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 payload the subscriber cannot query is -skipped rather than served partially, and that skip is silent by design. +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. 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 diff --git a/lib/solid_objects/actor_channel.rb b/lib/solid_objects/actor_channel.rb index 35f6ece..3eb6665 100644 --- a/lib/solid_objects/actor_channel.rb +++ b/lib/solid_objects/actor_channel.rb @@ -78,7 +78,13 @@ def transmit_state_payloads(snapshot) return if payload_names.nil? || payload_names.empty? return unless newer_payload_revision?(snapshot) - payload_names.each { |name| transmit_state_payload(snapshot, name) } + # 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 @@ -88,7 +94,11 @@ def transmit_state_payloads(snapshot) # 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. - # @rbs (ActorSnapshot, String) -> void + # + # 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:, @@ -96,8 +106,9 @@ def transmit_state_payload(snapshot, name) authorization_context: payload_authorization_context(name) ).call transmit TurboStreamRenderer.state_payload(payload) + true rescue Unauthorized - nil + true rescue => error SolidObjects.instrument( :payload_broadcast_failed, @@ -106,6 +117,7 @@ def transmit_state_payload(snapshot, name) payload_name: name, error_class: error.class.name ) + false end # Resolves the Cable connection to whatever the application uses as an diff --git a/sig/generated/lib/solid_objects/actor_channel.rbs b/sig/generated/lib/solid_objects/actor_channel.rbs index c24bfc2..db13c39 100644 --- a/sig/generated/lib/solid_objects/actor_channel.rbs +++ b/sig/generated/lib/solid_objects/actor_channel.rbs @@ -27,8 +27,12 @@ module SolidObjects # 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. - # @rbs (ActorSnapshot, String) -> void - def transmit_state_payload: (ActorSnapshot, String) -> void + # + # 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: (ActorSnapshot, String) -> bool # Resolves the Cable connection to whatever the application uses as an # authorization subject, so a payload block and `authorize_query` see the diff --git a/test/integration/payload_delivery_test.rb b/test/integration/payload_delivery_test.rb index 29fa080..d51686a 100644 --- a/test/integration/payload_delivery_test.rb +++ b/test/integration/payload_delivery_test.rb @@ -38,6 +38,21 @@ class RoomActor < SolidObjects::Actor raise "payload exploded with secret-hand and secret-token" end + # Fails its first attempt only, modelling a transient error such as a lock + # timeout while reading state. + broadcast_payload :flaky_state do + self.class.attempts += 1 + raise "transient" if self.class.attempts == 1 + + { "turn" => turn } + end + + class << self + attr_writer :attempts + + def attempts = @attempts ||= 0 + end + def hand_for(session_id) = hands.fetch(session_id, []) def deal(session_id:, cards:) @@ -61,6 +76,7 @@ def initialize(session_id) setup do SolidObjects.reset! + RoomActor.attempts = 0 RoomActor.ensure_registered! SolidObjects.configuration.stream_signing_secret = "payload-delivery-secret" SolidObjects.configuration.authorize_subscription = ->(**) { true } @@ -225,6 +241,22 @@ def initialize(session_id) assert_equal 2, delivered_payload("spectator_state").fetch("turn") end + # A failed payload must not advance the delivery watermark, or the subscriber + # never sees that revision: dedup would skip every later attempt at it, and + # the actor may not mutate again for a long time. + test "a payload that failed is retried at the same revision" do + reference = deal_to("alice") + reference.advance_turn + stub_connection(session_id: "alice") + subscribe token: payload_token(reference, %w[flaky_state]) + assert_nil delivered_payload("flaky_state"), "the first attempt should have failed" + + receive_latest_broadcast + + refute_nil delivered_payload("flaky_state"), + "a failed payload must be retried rather than marked delivered" + end + test "a payload is not re-delivered at the same revision" do reference = deal_to("alice") stub_connection(session_id: "alice") From d2151ec606de0feba0d3adc6196aa05be6b25a5f Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Mon, 10 Aug 2026 11:19:11 -0700 Subject: [PATCH 3/3] docs: say what a repeating payload failure means Holding the watermark open on failure means a persistently failing block retries once per actor mutation and reports every attempt, where a transient one reports once. An operator watching the event needs to know which of those they are looking at. --- docs/realtime.md | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/docs/realtime.md b/docs/realtime.md index e00b2db..77bd93b 100644 --- a/docs/realtime.md +++ b/docs/realtime.md @@ -320,9 +320,15 @@ 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. 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. +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