diff --git a/CHANGELOG.md b/CHANGELOG.md index d13807e..f0f4600 100644 --- a/CHANGELOG.md +++ b/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 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..77bd93b 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,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 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..3eb6665 100644 --- a/lib/solid_objects/actor_channel.rb +++ b/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 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..db13c39 100644 --- a/sig/generated/lib/solid_objects/actor_channel.rbs +++ b/sig/generated/lib/solid_objects/actor_channel.rbs @@ -21,6 +21,25 @@ 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. + # + # 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 + # 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..d51686a --- /dev/null +++ b/test/integration/payload_delivery_test.rb @@ -0,0 +1,411 @@ +# 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 + + # 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:) + 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.attempts = 0 + 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 + + # 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") + 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