diff --git a/CHANGELOG.md b/CHANGELOG.md index 3d1d0cb..7cc4a83 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,39 @@ # Changelog +## 0.14.0 - 2026-08-22 + +- Add `SolidObjects::Transmission.receive(envelope)`, the server ingest for + the browser transmit family in solid-objects-js. It validates a camelCase + transmit envelope, resolves the actor type through an optional + `resolve_actor_type:` proc, and enqueues one internal message with the + idempotency key `transmit:`, so a replayed envelope applies + once. Malformed envelopes raise the new + `SolidObjects::InvalidTransmission`. Internal delivery skips + `authorize_message`, so the host application must authenticate the + request before it calls `receive`; see `docs/transmission.md` for the + controller boundary. On a registry miss under Rails, `receive` loads the + application's actor classes once and retries, because a lazy-loading web + process has no other reason to have loaded the target class. Golden + fixtures in `compatibility/transmit-envelopes.json` pin the wire contract + shared with the JS runtime. +- Add `Actor#transmit` and `SolidObjects.register_transmit`, the staging + side of the transmit family. `transmit.increment(amount:)` stages a + `solid-objects.transmit` effect in the same commit as the state change; + `register_transmit` drains staged effects into camelCase envelopes and + hands each to the delivery block, which raises to retry. A claimed + transmit effect delivers every undelivered sibling for its actor up to + its own mailbox sequence, oldest first, so per-actor order survives a + failed delivery, and the receiving side dedups on `transmit:`. + A raw `emit "solid-objects.transmit"` with explicit `actorType` and + `actorId` targets a different actor, matching the JS staging surface. +- Mount `POST /solid_objects/transmit` in the engine, an ingest route + behind the new deny-by-default `authorize_transmission` policy. The + policy receives the parsed envelope and the controller, an unauthorized + envelope gets 403, and a permanently unappliable one gets 422, so a + sending outbox dead-letters it instead of retrying forever. The new + `transmission_actor_type_resolver` configuration maps diverged actor + type names for the engine route. + ## 0.13.3 - 2026-08-18 - Stop loading `ActiveRecord::Base` when the gem is required. The engine now diff --git a/Gemfile.lock b/Gemfile.lock index 937b83d..774973d 100644 --- a/Gemfile.lock +++ b/Gemfile.lock @@ -1,7 +1,7 @@ PATH remote: . specs: - solid_objects (0.13.3) + solid_objects (0.14.0) actioncable (>= 7.1) actionpack (>= 7.1) actionview (>= 7.1) @@ -384,7 +384,7 @@ CHECKSUMS rubocop-rails-omakase (1.1.0) sha256=2af73ac8ee5852de2919abbd2618af9c15c19b512c4cfc1f9a5d3b6ef009109d ruby-progressbar (1.13.0) sha256=80fc9c47a9b640d6834e0dc7b3c94c9df37f08cb072b7761e4a71e22cff29b33 securerandom (0.4.1) sha256=cc5193d414a4341b6e225f0cb4446aceca8e50d5e1888743fac16987638ea0b1 - solid_objects (0.13.3) + solid_objects (0.14.0) sqlite3 (2.9.5-aarch64-linux-gnu) sha256=78075b6337d3d182c6d2b4691049ed45cd220826160c9ea18946bf6a1de200dc sqlite3 (2.9.5-aarch64-linux-musl) sha256=18c801185deb4adc01ddb281e8f672a39e3d1729979ca91e39439cd3eac0402d sqlite3 (2.9.5-arm-linux-gnu) sha256=1bdfca0c7d63998c60b0f4a8e3c8df2d33800ccc4abd2d612eddbbbc92a4c48b diff --git a/app/controllers/solid_objects/transmissions_controller.rb b/app/controllers/solid_objects/transmissions_controller.rb new file mode 100644 index 0000000..08d4b76 --- /dev/null +++ b/app/controllers/solid_objects/transmissions_controller.rb @@ -0,0 +1,32 @@ +# rbs_inline: enabled + +require "action_controller/api" + +module SolidObjects + class TransmissionsController < ActionController::API + # @rbs () -> void + def create + envelope = JSON.parse(request.body.read) + return head :forbidden unless authorized_transmission?(envelope) + + Transmission.receive( + envelope, + resolve_actor_type: SolidObjects.configuration.transmission_actor_type_resolver + ) + head :ok + rescue JSON::ParserError, InvalidTransmission, UnknownActorType, UnknownMessage, + PayloadTooLarge, IdempotencyConflict + head :unprocessable_entity + end + + private + + # @rbs (untyped) -> bool + def authorized_transmission?(envelope) + SolidObjects.configuration.authorize_transmission.call( + envelope:, + authorization_context: self + ) + end + end +end diff --git a/compatibility/transmit-envelopes.json b/compatibility/transmit-envelopes.json new file mode 100644 index 0000000..93d3459 --- /dev/null +++ b/compatibility/transmit-envelopes.json @@ -0,0 +1,89 @@ +{ + "description": "Golden transmit envelopes shared by solid_objects and solid-objects-js. The JS bridge must produce and accept these; SolidObjects::Transmission.receive must accept the valid ones, apply the duplicate pair once, and reject the malformed ones.", + "valid": [ + { + "name": "increment with arguments", + "envelope": { + "effectId": "fixture-effect-0001", + "actorType": "transmit-counters", + "actorId": "fixture-counter", + "operation": "increment", + "arguments": { "amount": 2 } + }, + "idempotencyKey": "transmit:fixture-effect-0001" + }, + { + "name": "increment without arguments", + "envelope": { + "effectId": "fixture-effect-0002", + "actorType": "transmit-counters", + "actorId": "fixture-counter", + "operation": "increment" + }, + "idempotencyKey": "transmit:fixture-effect-0002" + } + ], + "duplicatePair": [ + { + "effectId": "fixture-effect-0003", + "actorType": "transmit-counters", + "actorId": "fixture-counter", + "operation": "increment", + "arguments": { "amount": 1 } + }, + { + "effectId": "fixture-effect-0003", + "actorType": "transmit-counters", + "actorId": "fixture-counter", + "operation": "increment", + "arguments": { "amount": 1 } + } + ], + "malformed": [ + { + "name": "missing effectId", + "envelope": { + "actorType": "transmit-counters", + "actorId": "fixture-counter", + "operation": "increment" + } + }, + { + "name": "empty actorId", + "envelope": { + "effectId": "fixture-effect-0004", + "actorType": "transmit-counters", + "actorId": "", + "operation": "increment" + } + }, + { + "name": "snake_case keys", + "envelope": { + "effect_id": "fixture-effect-0005", + "actor_type": "transmit-counters", + "actor_id": "fixture-counter", + "operation": "increment" + } + }, + { + "name": "non-string operation", + "envelope": { + "effectId": "fixture-effect-0006", + "actorType": "transmit-counters", + "actorId": "fixture-counter", + "operation": 7 + } + }, + { + "name": "arguments not an object", + "envelope": { + "effectId": "fixture-effect-0007", + "actorType": "transmit-counters", + "actorId": "fixture-counter", + "operation": "increment", + "arguments": [ 1 ] + } + } + ] +} diff --git a/config/routes.rb b/config/routes.rb index 225f1ff..a3eca86 100644 --- a/config/routes.rb +++ b/config/routes.rb @@ -1,6 +1,7 @@ # rbs_inline: enabled SolidObjects::Engine.routes.draw do + post :transmit, to: "transmissions#create" get :components, to: "components#show" get "components/batch", to: "components#batch" resources :instances, only: %i[index show] diff --git a/docs/roadmap.md b/docs/roadmap.md index 185f341..fe19564 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -72,6 +72,20 @@ Rails 7.1 and 7.2 is unmeasured against those servers. Rails 7.0 is out of range because its SQLite adapter requires `sqlite3 ~> 1.4`, and this gem needs the busy-handler control that arrived in `sqlite3` 2.x +- The transmit family, both sides. `SolidObjects::Transmission.receive` is + the ingest: envelope validation, actor type resolution with a per-call + `resolve_actor_type:` escape hatch, and an internal idempotent enqueue + keyed `transmit:`. `Actor#transmit` and + `SolidObjects.register_transmit` are the staging side: a transactional + `solid-objects.transmit` effect and a drain that delivers every + undelivered sibling for the actor up to the claimed effect's mailbox + sequence, oldest first, so per-actor order survives a failed delivery. + The wire contract is pinned by golden fixtures in + `compatibility/transmit-envelopes.json`. The engine mounts + `POST /solid_objects/transmit` behind a deny-by-default + `authorize_transmission` policy with a configurable actor type resolver. + Bidirectional replication as a declared surface, with echo suppression, + is not implemented - A JavaScript suite covering every browser module, run in CI with Node's test runner and jsdom, plus a browser suite running the same modules against real Chromium and a real Turbo build, with every GitHub Actions reference pinned to @@ -129,7 +143,10 @@ of who pressed what, and bulk-safe tools: retry is one dead letter at a time, because `DeadLetterManager` exposes no bulk operation. Pause is an operator brake and not a stop, since a pass already in flight finishes its turn and a - synchronous caller waiting on a paused instance times out. The page cost was + synchronous caller waiting on a paused instance times out. Retry also only + exists for message dead letters: a dead effect or broadcast has no retry + API, which matters for transmit effects because a dead one is a lost + replay until an operator returns its row to pending. The page cost was reasoned about rather than measured: the summary bar issues a fixed set of indexed aggregate queries per page, which is why `HEAD /` exists for uptime monitors, but no dashboard latency has been benchmarked against a large diff --git a/docs/transmission.md b/docs/transmission.md new file mode 100644 index 0000000..dd205fe --- /dev/null +++ b/docs/transmission.md @@ -0,0 +1,198 @@ +# The transmit family + +The transmit family replays one runtime's actor operations onto another +runtime over a shared wire contract. The Ruby gem holds both sides: + +- `Actor#transmit` and `SolidObjects.register_transmit` stage and deliver + envelopes. This is the sending side. +- `SolidObjects::Transmission.receive(envelope)` ingests envelopes. This is + the receiving side. + +[solid-objects-js](https://github.com/cardmagic/solid-objects-js) holds the +same two sides for the browser and Node. A browser actor stages a transmit +intent with `this.transmit().increment({ amount })` in the same transaction +as its state change; its effect worker drains that outbox with +at-least-once delivery, per-actor order, and retry backoff, and posts one +JSON envelope per effect to a route the host application owns. A Rails +actor does the same with `transmit.increment(amount:)`. Either ingest +accepts either sender, so Rails-to-Rails, Rails-to-Node, Node-to-Rails, +and browser-to-Rails replication all ride one contract. + +## The sending side + +```ruby +class Counter < SolidObjects::Actor + actor_type "counters" + + attribute :count, default: 0 + + def increment(amount: 1) + self.count += amount + transmit.increment(amount:) + end +end + +SolidObjects.register_transmit do |envelope| + DeliverToUpstream.call(envelope) +end +``` + +`transmit` returns the same fluent dispatcher `schedule` returns. It stages +one `solid-objects.transmit` effect in the same commit as the state change, +targeting the same operation on the same actor in the receiving runtime. For +a different target, stage the effect directly: + +```ruby +emit "solid-objects.transmit", + operation: "increment", + arguments: { amount: 2 }, + actorType: "other-counters", + actorId: "counter-1" +``` + +`SolidObjects.register_transmit(&deliver)` registers the drain handler for +that effect. The block receives one camelCase envelope per staged effect. +Raise inside the block while the upstream is unreachable; the effect +retries with backoff and dead-letters on exhaustion, like any other effect. + +The drain keeps per-actor order across failures: a claimed transmit effect +delivers every undelivered sibling for its actor up to its own mailbox +sequence, oldest first. The receiving side dedups on `transmit:`, +so a redelivered envelope applies once. + +Delivery is at-least-once by design, and the drain accepts redundant sends +as the price of ordering without cross-worker coordination. A drained +sibling's own effect row stays pending, because completing it would +require taking over another worker's claim; when its own claim runs, it +delivers again and the receiving side drops the replay. The same is true +when two workers claim effects for one actor concurrently. Both runtimes +share this behavior, and the Ruby suite pins it with a race test: order +holds, duplicates apply nothing, and every effect completes. + +## Retry budget and offline tolerance + +A raised delivery follows the effect retry policy: `max_attempts` (default +5) and `retry_delay` (default `2 ** (attempt - 1)` seconds, capped at 60). +The defaults give roughly fifteen seconds of offline tolerance before an +envelope dead-letters. An application that transmits across real outages +must raise both: + +```ruby +SolidObjects.configure do |configuration| + configuration.max_attempts = 30 + configuration.retry_delay = ->(attempt) { [ 2**(attempt - 1), 300 ].min.to_f } +end +``` + +These settings apply to every effect, not only transmits. A dead transmit +effect has no retry API; the dashboard lists it, and recovery means +returning its row to `pending` with a cleared `attempt_count`. Order +survives that recovery, because the drain orders by mailbox sequence, not +by retry time. + +## Wire contract + +The JS side owns the envelope format. The Ruby ingest accepts it verbatim. + +- Keys arrive camelCase: `effectId`, `actorType`, `actorId`, `operation`, + and an optional `arguments` object. There is no snake_case dialect. +- The idempotency key is `transmit:`, byte-identical to the JS + server ingest. A replayed envelope applies once. +- Delivery is at-least-once and per-actor ordered by the browser's drain. + The server preserves mailbox order and adds no ordering of its own. + +`compatibility/transmit-envelopes.json` pins the contract. Both runtimes +run a consuming test against the same fixture file. + +## What `receive` does + +1. It validates the envelope shape. A malformed envelope raises + `SolidObjects::InvalidTransmission`. +2. It resolves the actor type and looks it up in the registry. On a miss + under Rails it loads the application's actor classes once and retries, + because a lazy-loading process has no other reason to have loaded the + target class. A type that is still unknown raises + `SolidObjects::UnknownActorType`. An undeclared operation raises + `SolidObjects::UnknownMessage`. +3. It enqueues one internal message with the idempotency key + `transmit:`. Oversized arguments raise + `SolidObjects::PayloadTooLarge` before persistence. + +The enqueue uses `delivery_mode: "internal"`, the same mode the effect +executor uses. Internal delivery skips `authorize_message` by construction. +The host application must authenticate the request before it calls +`receive`. + +## The engine route + +The engine mounts `POST /solid_objects/transmit` (under wherever the host +mounts `SolidObjects::Engine`). It parses the body, authorizes it through +`authorize_transmission`, and passes it to `Transmission.receive` with the +configured `transmission_actor_type_resolver`. The policy denies by +default, because the ingest skips `authorize_message` by design; an +unauthorized envelope gets 403, and an envelope the server can never apply +gets 422: + +```ruby +SolidObjects.configure do |configuration| + configuration.authorize_transmission = lambda do |envelope:, authorization_context:| + ActiveSupport::SecurityUtils.secure_compare( + authorization_context.request.headers["Authorization"].to_s, + "Bearer #{Rails.application.credentials.transmit_token}" + ) + end +end +``` + +The policy receives the parsed envelope and the controller as +`authorization_context:`, so it can bind `actorType` and `actorId` to the +authenticated caller, not only check a shared token. Rate limits stay with +the host (Rack::Attack or the proxy), the same boundary the dashboard +draws. + +## A hand-rolled route + +An application that wants its own controller keeps the same shape: + +```ruby +class TransmitController < ApplicationController + skip_forgery_protection + + def create + head :forbidden and return unless authenticated_device? + + SolidObjects::Transmission.receive(JSON.parse(request.body.read)) + head :ok + rescue SolidObjects::InvalidTransmission, SolidObjects::UnknownActorType, + SolidObjects::UnknownMessage, SolidObjects::PayloadTooLarge, + SolidObjects::IdempotencyConflict, JSON::ParserError + head :unprocessable_entity + end +end +``` + +Return 422 for an envelope the server can never apply. The browser outbox +dead-letters that effect instead of retrying it forever. Return a 5xx for a +transient server fault, so the browser retries with backoff. +`SolidObjects::IdempotencyConflict` belongs in the 422 list: it means the +effect id was replayed with a different invocation, and no retry can ever +make that envelope apply. + +## Actor type mapping + +When both runtimes use the same actor type strings, no configuration is +needed. When the names diverge, pass `resolve_actor_type:` per call: + +```ruby +SolidObjects::Transmission.receive( + envelope, + resolve_actor_type: ->(actor_type) { actor_type.sub("browser-", "server-") } +) +``` + +## Scope + +Bidirectional replication as a first-class surface, where two runtimes +declare a replica pair and echo suppression keeps a replayed operation +from transmitting back, is a separate feature. The transmit family gives +it the mechanism; the declaration API does not exist yet. diff --git a/lib/generators/solid_objects/templates/solid_objects.rb b/lib/generators/solid_objects/templates/solid_objects.rb index 99aa8f0..1d97486 100644 --- a/lib/generators/solid_objects/templates/solid_objects.rb +++ b/lib/generators/solid_objects/templates/solid_objects.rb @@ -49,6 +49,22 @@ configuration.authorize_subscription = ->(**) { false } configuration.authorize_administration = ->(**) { false } + # The engine route POST /solid_objects/transmit ingests transmit envelopes + # from another Solid Objects runtime. It stays denied until its callers are + # authenticated, because the ingest skips authorize_message by design: + # + # configuration.authorize_transmission = lambda do |envelope:, authorization_context:| + # ActiveSupport::SecurityUtils.secure_compare( + # authorization_context.request.headers["Authorization"].to_s, + # "Bearer #{Rails.application.credentials.transmit_token}" + # ) + # end + # + # When the sending runtime names actor types differently, map them here: + # + # configuration.transmission_actor_type_resolver = ->(actor_type) { actor_type.sub("browser-", "server-") } + configuration.authorize_transmission = ->(**) { false } + # Configure component_authorization_context to return the authenticated # principal used for reactive component refreshes. diff --git a/lib/solid_objects.rb b/lib/solid_objects.rb index f491b2f..f256f6a 100644 --- a/lib/solid_objects.rb +++ b/lib/solid_objects.rb @@ -63,6 +63,8 @@ # was reachable only through the caller path, so requiring the gem was not # enough to run a role that uses it. require "solid_objects/mailbox" +require "solid_objects/application_actor_loader" +require "solid_objects/transmission" require "solid_objects/worker" require "solid_objects/effect_executor" require "solid_objects/reminder_scheduler" @@ -100,6 +102,20 @@ def register_effect(name, &handler) effect_registry.register(name, handler) end + # @rbs (?effect_name: String | Symbol) { (Hash[String, untyped]) -> untyped } -> Proc + def register_transmit(effect_name: Transmission::EFFECT_NAME, &deliver) + raise ArgumentError, "register_transmit requires a delivery block" unless deliver + + register_effect(effect_name) do |arguments, context| + Transmission.deliver_through( + effect_name: effect_name.to_s, + arguments:, + context:, + deliver: + ) + end + end + # @rbs () -> CommitActionRegistry def commit_action_registry @commit_action_registry ||= CommitActionRegistry.new diff --git a/lib/solid_objects/actor.rb b/lib/solid_objects/actor.rb index 8a1acc1..63f9e15 100644 --- a/lib/solid_objects/actor.rb +++ b/lib/solid_objects/actor.rb @@ -190,6 +190,17 @@ def emit(name, on_success: nil, on_failure: nil, **arguments) nil end + # @rbs () -> OperationDispatcher + def transmit + OperationDispatcher.new( + actor_type: self.class.actor_type, + handlers: self.class.definition.messages + ) do |operation, arguments| + emit(Transmission::EFFECT_NAME, operation: operation.to_s, arguments:) + nil + end + end + # @rbs (Symbol | String, **untyped) -> nil def commit_action(name, **arguments) CommitActionIntent.new( diff --git a/lib/solid_objects/configuration.rb b/lib/solid_objects/configuration.rb index e7b6678..89f4c60 100644 --- a/lib/solid_objects/configuration.rb +++ b/lib/solid_objects/configuration.rb @@ -47,6 +47,8 @@ class Configuration # @rbs @authorize_destroy: Proc # @rbs @authorize_subscription: Proc # @rbs @authorize_administration: Proc + # @rbs @authorize_transmission: Proc + # @rbs @transmission_actor_type_resolver: Proc attr_accessor :table_name_prefix, :polling_interval, @@ -92,7 +94,9 @@ class Configuration :authorize_query, :authorize_destroy, :authorize_subscription, - :authorize_administration + :authorize_administration, + :authorize_transmission, + :transmission_actor_type_resolver # @rbs @additional_components: Array[untyped] attr_reader :additional_components @@ -148,6 +152,8 @@ def initialize @authorize_destroy = ->(**) { false } @authorize_subscription = ->(**) { false } @authorize_administration = ->(**) { false } + @authorize_transmission = ->(**) { false } + @transmission_actor_type_resolver = ->(actor_type) { actor_type } @additional_components = [] end diff --git a/lib/solid_objects/errors.rb b/lib/solid_objects/errors.rb index 40a2330..fc2593b 100644 --- a/lib/solid_objects/errors.rb +++ b/lib/solid_objects/errors.rb @@ -19,6 +19,9 @@ class UnknownActorType < Error class UnknownMessage < Error end + class InvalidTransmission < Error + end + class InvalidPayload < Error end diff --git a/lib/solid_objects/transmission.rb b/lib/solid_objects/transmission.rb new file mode 100644 index 0000000..2b610f6 --- /dev/null +++ b/lib/solid_objects/transmission.rb @@ -0,0 +1,139 @@ +# rbs_inline: enabled + +module SolidObjects + module Transmission + REQUIRED_FIELDS = %w[effectId actorType actorId operation].freeze + IDEMPOTENCY_PREFIX = "transmit:" + EFFECT_NAME = "solid-objects.transmit" + UNDELIVERED_STATUSES = %w[pending processing].freeze + + class << self + # @rbs (untyped envelope, ?resolve_actor_type: ^(String) -> (String | Symbol), ?mailbox: Mailbox, ?actor_loader: ^() -> bool) -> MessageReference + def receive(envelope, resolve_actor_type: :itself.to_proc, mailbox: Mailbox.new, actor_loader: method(:load_application_actors)) + validate!(envelope) + + actor_type = resolve_actor_type.call(envelope["actorType"]).to_s + actor_class = fetch_actor_class(actor_type, actor_loader) + operation = envelope["operation"].to_sym + unless actor_class.definition.messages.key?(operation) + raise UnknownMessage, "unknown operation #{envelope["operation"].inspect}" + end + + mailbox.enqueue( + reference: Reference.new(actor_type:, actor_id: envelope["actorId"]), + operation:, + arguments: envelope.fetch("arguments", {}), + delivery_mode: "internal", + idempotency_key: "#{IDEMPOTENCY_PREFIX}#{envelope["effectId"]}" + ) + end + + # @rbs (effect_name: String, arguments: Hash[String, untyped], context: EffectContext, deliver: Proc) -> nil + def deliver_through(effect_name:, arguments:, context:, deliver:) + staged_envelope( + arguments, + effect_id: context.id, + actor_type: context.actor_type, + actor_id: context.actor_id + ) + undelivered_envelopes_through(effect_name:, context:).each do |envelope| + deliver.call(envelope) + end + nil + end + + private + + # @rbs (String, ^() -> bool) -> Class + def fetch_actor_class(actor_type, actor_loader) + SolidObjects.registry.fetch(actor_type) + rescue UnknownActorType + raise unless actor_loader.call + + SolidObjects.registry.fetch(actor_type) + end + + # @rbs () -> bool + def load_application_actors + return false unless defined?(Rails.application) && Rails.application + + ApplicationActorLoader.new.call + true + end + + # @rbs (Hash[String, untyped], effect_id: String, actor_type: String, actor_id: String) -> Hash[String, untyped] + def staged_envelope(arguments, effect_id:, actor_type:, actor_id:) + operation = arguments["operation"] + unless operation.is_a?(String) && !operation.empty? + raise InvalidTransmission, "transmit effect arguments require a non-empty operation" + end + + target_arguments = arguments["arguments"] + target_arguments = {} if target_arguments.nil? + unless target_arguments.is_a?(Hash) + raise InvalidTransmission, %(transmit effect arguments must hold a JSON object in "arguments") + end + + target_type = arguments["actorType"].nil? ? actor_type : arguments["actorType"] + target_id = arguments["actorId"].nil? ? actor_id : arguments["actorId"] + { "actorType" => target_type, "actorId" => target_id }.each do |field, value| + next if value.is_a?(String) && !value.empty? + + raise InvalidTransmission, "transmit effect #{field} must be a non-empty string" + end + + { + "effectId" => effect_id, + "actorType" => target_type, + "actorId" => target_id, + "operation" => operation, + "arguments" => target_arguments + } + end + + # @rbs (effect_name: String, context: EffectContext) -> Array[Hash[String, untyped]] + def undelivered_envelopes_through(effect_name:, context:) + source_sequence = Message.find(context.source_message_id).sequence + effects = Effect + .joins(:message) + .where(name: effect_name, status: UNDELIVERED_STATUSES) + .merge( + Message.where( + actor_type: context.actor_type, + actor_id: context.actor_id, + sequence: ..source_sequence + ) + ) + .order(Message.arel_table[:sequence].asc, :id) + + effects.filter_map do |effect| + staged_envelope( + effect.arguments, + effect_id: effect.effect_id, + actor_type: context.actor_type, + actor_id: context.actor_id + ) + rescue InvalidTransmission + nil + end + end + + # @rbs (untyped) -> void + def validate!(envelope) + raise InvalidTransmission, "envelope must be a JSON object" unless envelope.is_a?(Hash) + + REQUIRED_FIELDS.each do |field| + value = envelope[field] + next if value.is_a?(String) && !value.empty? + + raise InvalidTransmission, "envelope field #{field.inspect} must be a non-empty string" + end + + arguments = envelope.fetch("arguments", {}) + return if arguments.is_a?(Hash) + + raise InvalidTransmission, %(envelope field "arguments" must be a JSON object) + end + end + end +end diff --git a/lib/solid_objects/version.rb b/lib/solid_objects/version.rb index 0562095..3f78538 100644 --- a/lib/solid_objects/version.rb +++ b/lib/solid_objects/version.rb @@ -1,5 +1,5 @@ # rbs_inline: enabled module SolidObjects - VERSION = "0.13.3" + VERSION = "0.14.0" end diff --git a/sig/generated/controllers/solid_objects/transmissions_controller.rbs b/sig/generated/controllers/solid_objects/transmissions_controller.rbs new file mode 100644 index 0000000..7ac1240 --- /dev/null +++ b/sig/generated/controllers/solid_objects/transmissions_controller.rbs @@ -0,0 +1,13 @@ +# Generated from app/controllers/solid_objects/transmissions_controller.rb with RBS::Inline + +module SolidObjects + class TransmissionsController < ActionController::API + # @rbs () -> void + def create: () -> void + + private + + # @rbs (untyped) -> bool + def authorized_transmission?: (untyped) -> bool + end +end diff --git a/sig/generated/lib/solid_objects.rbs b/sig/generated/lib/solid_objects.rbs index d5e40a4..1eaf5bb 100644 --- a/sig/generated/lib/solid_objects.rbs +++ b/sig/generated/lib/solid_objects.rbs @@ -18,6 +18,9 @@ module SolidObjects # @rbs (String | Symbol) { (Hash[String, untyped], EffectContext) -> untyped } -> Proc def self.register_effect: (String | Symbol) { (Hash[String, untyped], EffectContext) -> untyped } -> Proc + # @rbs (?effect_name: String | Symbol) { (Hash[String, untyped]) -> untyped } -> Proc + def self.register_transmit: (?effect_name: String | Symbol) { (Hash[String, untyped]) -> untyped } -> Proc + # @rbs () -> CommitActionRegistry def self.commit_action_registry: () -> CommitActionRegistry diff --git a/sig/generated/lib/solid_objects/actor.rbs b/sig/generated/lib/solid_objects/actor.rbs index 59a243b..3a7bfdf 100644 --- a/sig/generated/lib/solid_objects/actor.rbs +++ b/sig/generated/lib/solid_objects/actor.rbs @@ -161,6 +161,9 @@ module SolidObjects # @rbs (Symbol | String, ?on_success: Symbol | String?, ?on_failure: Symbol | String?, **untyped) -> nil def emit: (Symbol | String, ?on_success: Symbol | String?, ?on_failure: Symbol | String?, **untyped) -> nil + # @rbs () -> OperationDispatcher + def transmit: () -> OperationDispatcher + # @rbs (Symbol | String, **untyped) -> nil def commit_action: (Symbol | String, **untyped) -> nil diff --git a/sig/generated/lib/solid_objects/configuration.rbs b/sig/generated/lib/solid_objects/configuration.rbs index 8432232..05facca 100644 --- a/sig/generated/lib/solid_objects/configuration.rbs +++ b/sig/generated/lib/solid_objects/configuration.rbs @@ -4,8 +4,6 @@ module SolidObjects class Configuration @table_name_prefix: String - @process_alive_threshold: Float - @shutdown_timeout: Float @supervisor_monitor_interval: Float @@ -58,6 +56,10 @@ module SolidObjects @authorize_administration: Proc + @authorize_transmission: Proc + + @transmission_actor_type_resolver: Proc + @polling_interval: Float @idle_polling_interval: Float @@ -92,6 +94,8 @@ module SolidObjects @process_heartbeat_interval: Float + @process_alive_threshold: Float + attr_accessor table_name_prefix: untyped attr_accessor polling_interval: untyped @@ -182,6 +186,10 @@ module SolidObjects attr_accessor authorize_administration: untyped + attr_accessor authorize_transmission: untyped + + attr_accessor transmission_actor_type_resolver: untyped + # @rbs @additional_components: Array[untyped] attr_reader additional_components: untyped diff --git a/sig/generated/lib/solid_objects/errors.rbs b/sig/generated/lib/solid_objects/errors.rbs index dcce53e..d858b25 100644 --- a/sig/generated/lib/solid_objects/errors.rbs +++ b/sig/generated/lib/solid_objects/errors.rbs @@ -19,6 +19,9 @@ module SolidObjects class UnknownMessage < Error end + class InvalidTransmission < Error + end + class InvalidPayload < Error end diff --git a/sig/generated/lib/solid_objects/transmission.rbs b/sig/generated/lib/solid_objects/transmission.rbs new file mode 100644 index 0000000..4bcb7ae --- /dev/null +++ b/sig/generated/lib/solid_objects/transmission.rbs @@ -0,0 +1,34 @@ +# Generated from lib/solid_objects/transmission.rb with RBS::Inline + +module SolidObjects + module Transmission + REQUIRED_FIELDS: untyped + + IDEMPOTENCY_PREFIX: ::String + + EFFECT_NAME: ::String + + UNDELIVERED_STATUSES: untyped + + # @rbs (untyped envelope, ?resolve_actor_type: ^(String) -> (String | Symbol), ?mailbox: Mailbox, ?actor_loader: ^() -> bool) -> MessageReference + def self.receive: (untyped envelope, ?resolve_actor_type: ^(String) -> (String | Symbol), ?mailbox: Mailbox, ?actor_loader: ^() -> bool) -> MessageReference + + # @rbs (effect_name: String, arguments: Hash[String, untyped], context: EffectContext, deliver: Proc) -> nil + def self.deliver_through: (effect_name: String, arguments: Hash[String, untyped], context: EffectContext, deliver: Proc) -> nil + + # @rbs (String, ^() -> bool) -> Class + private def self.fetch_actor_class: (String, ^() -> bool) -> Class + + # @rbs () -> bool + private def self.load_application_actors: () -> bool + + # @rbs (Hash[String, untyped], effect_id: String, actor_type: String, actor_id: String) -> Hash[String, untyped] + private def self.staged_envelope: (Hash[String, untyped], effect_id: String, actor_type: String, actor_id: String) -> Hash[String, untyped] + + # @rbs (effect_name: String, context: EffectContext) -> Array[Hash[String, untyped]] + private def self.undelivered_envelopes_through: (effect_name: String, context: EffectContext) -> Array[Hash[String, untyped]] + + # @rbs (untyped) -> void + private def self.validate!: (untyped) -> void + end +end diff --git a/sig/support/framework.rbs b/sig/support/framework.rbs index 65a06e5..7332a4e 100644 --- a/sig/support/framework.rbs +++ b/sig/support/framework.rbs @@ -21,6 +21,9 @@ end module ActionController class Base end + + class API + end end module Rails diff --git a/test/integration/install_generator_test.rb b/test/integration/install_generator_test.rb index bb97663..678246d 100644 --- a/test/integration/install_generator_test.rb +++ b/test/integration/install_generator_test.rb @@ -23,7 +23,8 @@ class InstallGeneratorTest < ActiveSupport::TestCase assert_includes initializer, "authorization_context[:source] == \"cli\"" assert_includes initializer, "component_authorization_context" refute_includes initializer, "component_authorization_context = lambda" - assert_equal 5, initializer.scan("= ->(**) { false }").length + assert_includes initializer, "configuration.authorize_transmission = ->(**) { false }" + assert_equal 6, initializer.scan("= ->(**) { false }").length end end end diff --git a/test/integration/load_contract_test.rb b/test/integration/load_contract_test.rb index 6327787..6e77d44 100644 --- a/test/integration/load_contract_test.rb +++ b/test/integration/load_contract_test.rb @@ -13,7 +13,6 @@ class LoadContractTest < ActiveSupport::TestCase # else that stops being loaded is a role waiting to fail in production, so # this list is the place to argue that a role never reaches it. DEFERRED = { - "application_actor_loader" => "used only by the CLI", "caller_process" => "the caller path, required by SolidObjects.caller_process", "cli" => "loaded by exe/solid_objects, and pulls in thor", "client" => "the caller path, required by SolidObjects.client", diff --git a/test/integration/polling_test.rb b/test/integration/polling_test.rb index 9d2eaa0..3e9bece 100644 --- a/test/integration/polling_test.rb +++ b/test/integration/polling_test.rb @@ -203,21 +203,26 @@ def signal test "processes local work promptly after reaching the idle ceiling" do SolidObjects.configuration.polling_interval = 0.025 SolidObjects.configuration.idle_polling_interval = 1.0 + reset_reasons = Queue.new + subscription = ActiveSupport::Notifications.subscribe( + "solid_objects.polling.interval_changed" + ) { |event| reset_reasons << event.payload.fetch(:reason) } worker = SolidObjects::Worker.new worker_thread = Thread.new { worker.run } Timeout.timeout(3) do sleep 0.005 until worker.current_polling_interval >= 1.0 end - started_at = Process.clock_gettime(Process::CLOCK_MONOTONIC) message = WakeLatencyActor.ref("local").async.increment - Timeout.timeout(0.4) do + Timeout.timeout(2) do sleep 0.005 until message.status == "completed" end - elapsed = Process.clock_gettime(Process::CLOCK_MONOTONIC) - started_at - assert_operator elapsed, :<, 0.4 + observed_reasons = [] + observed_reasons << reset_reasons.pop until reset_reasons.empty? + assert_includes observed_reasons, :wake_up ensure + ActiveSupport::Notifications.unsubscribe(subscription) if subscription worker&.request_shutdown worker_thread&.join(2) worker&.stop diff --git a/test/integration/transmission_test.rb b/test/integration/transmission_test.rb new file mode 100644 index 0000000..d81f221 --- /dev/null +++ b/test/integration/transmission_test.rb @@ -0,0 +1,207 @@ +# frozen_string_literal: true + +require "database_test_helper" + +class TransmissionTest < ActiveSupport::TestCase + class CounterActor < SolidObjects::Actor + actor_type "transmit-counters" + + attribute :value, default: 0 + + def increment(amount: 1) + self.value += amount + end + end + + FIXTURES = JSON.parse( + File.read(File.expand_path("../../compatibility/transmit-envelopes.json", __dir__)) + ) + + setup do + SolidObjects.register_actor(CounterActor.actor_type, CounterActor) + end + + def counter_state(actor_id) + SolidObjects::Instance.find_by!(actor_type: "transmit-counters", actor_id:).state + end + + def valid_envelope(overrides = {}) + { + "effectId" => "effect-1", + "actorType" => "transmit-counters", + "actorId" => "alice", + "operation" => "increment", + "arguments" => { "amount" => 2 } + }.merge(overrides) + end + + test "enqueues an internal message the actor applies" do + SolidObjects::Transmission.receive(valid_envelope) + + message = SolidObjects::Message.sole + assert_equal "internal", message.delivery_mode + assert_equal "increment", message.operation + assert_equal({ "amount" => 2 }, message.arguments) + assert_equal "transmit:effect-1", message.idempotency_key + + assert_equal 1, SolidObjects::Worker.new.run_until_idle + assert_equal({ "value" => 2 }, counter_state("alice")) + end + + test "applies a replayed envelope once" do + envelope = valid_envelope + first = SolidObjects::Transmission.receive(envelope) + second = SolidObjects::Transmission.receive(envelope) + + assert_equal first.id, second.id + assert_equal 1, SolidObjects::Message.count + + SolidObjects::Worker.new.run_until_idle + + assert_equal({ "value" => 2 }, counter_state("alice")) + end + + test "defaults absent arguments to an empty object" do + envelope = valid_envelope.except("arguments") + + SolidObjects::Transmission.receive(envelope) + + assert_equal({}, SolidObjects::Message.sole.arguments) + end + + test "skips message authorization for the internal enqueue" do + SolidObjects.configuration.authorize_message = ->(**) { false } + + SolidObjects::Transmission.receive(valid_envelope) + + assert_equal 1, SolidObjects::Message.count + end + + test "maps a diverged actor type through resolve_actor_type" do + envelope = valid_envelope("actorType" => "browser-counters") + resolver = ->(actor_type) { actor_type.sub("browser-", "transmit-") } + + SolidObjects::Transmission.receive(envelope, resolve_actor_type: resolver) + + assert_equal "transmit-counters", SolidObjects::Message.sole.actor_type + end + + test "loads application actors on a registry miss before failing" do + loaded = false + loader = -> do + loaded = true + SolidObjects.register_actor("lazy-counters", CounterActor) + true + end + + SolidObjects::Transmission.receive( + valid_envelope("actorType" => "lazy-counters"), + actor_loader: loader + ) + + assert loaded + assert_equal "lazy-counters", SolidObjects::Message.sole.actor_type + end + + test "rejects an unknown actor type" do + assert_raises(SolidObjects::UnknownActorType) do + SolidObjects::Transmission.receive(valid_envelope("actorType" => "missing")) + end + + assert_empty SolidObjects::Message.all + end + + test "rejects an unknown operation" do + assert_raises(SolidObjects::UnknownMessage) do + SolidObjects::Transmission.receive(valid_envelope("operation" => "erase")) + end + + assert_empty SolidObjects::Message.all + end + + test "enforces the payload byte cap" do + SolidObjects.configuration.max_payload_bytes = 64 + envelope = valid_envelope("arguments" => { "note" => "x" * 200 }) + + assert_raises(SolidObjects::PayloadTooLarge) do + SolidObjects::Transmission.receive(envelope) + end + + assert_empty SolidObjects::Message.all + end + + test "rejects a non-object envelope" do + [ nil, "envelope", [ valid_envelope ] ].each do |envelope| + assert_raises(SolidObjects::InvalidTransmission) do + SolidObjects::Transmission.receive(envelope) + end + end + + assert_empty SolidObjects::Message.all + end + + test "rejects an envelope with a missing, blank, or non-string field" do + [ + valid_envelope.except("effectId"), + valid_envelope("effectId" => ""), + valid_envelope("actorId" => ""), + valid_envelope("operation" => 7), + valid_envelope("actorType" => nil) + ].each do |envelope| + assert_raises(SolidObjects::InvalidTransmission) do + SolidObjects::Transmission.receive(envelope) + end + end + + assert_empty SolidObjects::Message.all + end + + test "rejects snake_case envelope keys" do + envelope = { + "effect_id" => "effect-1", + "actor_type" => "transmit-counters", + "actor_id" => "alice", + "operation" => "increment" + } + + assert_raises(SolidObjects::InvalidTransmission) do + SolidObjects::Transmission.receive(envelope) + end + end + + test "rejects non-object arguments" do + assert_raises(SolidObjects::InvalidTransmission) do + SolidObjects::Transmission.receive(valid_envelope("arguments" => [ 1 ])) + end + end + + test "accepts every valid golden fixture envelope" do + FIXTURES.fetch("valid").each do |fixture| + reference = SolidObjects::Transmission.receive(fixture.fetch("envelope")) + message = SolidObjects::Message.find(reference.id) + + assert_equal fixture.fetch("idempotencyKey"), message.idempotency_key, + "fixture #{fixture.fetch("name").inspect}" + end + end + + test "rejects every malformed golden fixture envelope" do + FIXTURES.fetch("malformed").each do |fixture| + assert_raises(SolidObjects::InvalidTransmission, "fixture #{fixture.fetch("name").inspect}") do + SolidObjects::Transmission.receive(fixture.fetch("envelope")) + end + end + + assert_empty SolidObjects::Message.all + end + + test "applies the duplicate golden fixture pair once" do + envelopes = FIXTURES.fetch("duplicatePair") + + references = envelopes.map do |envelope| + SolidObjects::Transmission.receive(envelope) + end + + assert_equal 1, references.map(&:id).uniq.length + end +end diff --git a/test/integration/transmissions_controller_test.rb b/test/integration/transmissions_controller_test.rb new file mode 100644 index 0000000..e1e9e42 --- /dev/null +++ b/test/integration/transmissions_controller_test.rb @@ -0,0 +1,147 @@ +# frozen_string_literal: true + +require "database_test_helper" +require "action_controller" +require "action_controller/test_case" +require_relative "../../app/controllers/solid_objects/transmissions_controller" + +class TransmissionsControllerTest < ActionController::TestCase + tests SolidObjects::TransmissionsController + + class CounterActor < SolidObjects::Actor + actor_type "engine-transmit-counters" + + attribute :count, default: 0 + + def increment(amount: 1) + self.count += amount + end + end + + setup do + @routes = ActionDispatch::Routing::RouteSet.new + @routes.draw do + post "transmit", to: "solid_objects/transmissions#create" + end + SolidObjects.register_actor(CounterActor.actor_type, CounterActor) + end + + def allow_transmissions + SolidObjects.configuration.authorize_transmission = ->(**) { true } + end + + def valid_envelope(overrides = {}) + { + "effectId" => "engine-effect-1", + "actorType" => "engine-transmit-counters", + "actorId" => "alice", + "operation" => "increment", + "arguments" => { "amount" => 2 } + }.merge(overrides) + end + + def post_envelope(envelope) + post :create, body: envelope.is_a?(String) ? envelope : JSON.generate(envelope) + end + + test "denies by default" do + post_envelope(valid_envelope) + + assert_response :forbidden + assert_empty SolidObjects::Message.all + end + + test "passes the envelope and controller to the policy" do + captured = nil + SolidObjects.configuration.authorize_transmission = lambda do |envelope:, authorization_context:| + captured = [ envelope, authorization_context ] + false + end + + post_envelope(valid_envelope) + + assert_response :forbidden + assert_equal valid_envelope, captured.fetch(0) + assert_kind_of SolidObjects::TransmissionsController, captured.fetch(1) + end + + test "ingests an authorized envelope" do + allow_transmissions + + post_envelope(valid_envelope) + + assert_response :ok + message = SolidObjects::Message.sole + assert_equal "internal", message.delivery_mode + assert_equal "transmit:engine-effect-1", message.idempotency_key + assert_equal({ "amount" => 2 }, message.arguments) + end + + test "applies a replayed envelope once" do + allow_transmissions + + post_envelope(valid_envelope) + post_envelope(valid_envelope) + + assert_response :ok + assert_equal 1, SolidObjects::Message.count + end + + test "maps actor types through the configured resolver" do + allow_transmissions + SolidObjects.configuration.transmission_actor_type_resolver = + ->(actor_type) { actor_type.sub("browser-", "engine-") } + + post_envelope(valid_envelope("actorType" => "browser-transmit-counters")) + + assert_response :ok + assert_equal "engine-transmit-counters", SolidObjects::Message.sole.actor_type + end + + test "rejects a malformed envelope" do + allow_transmissions + + post_envelope(valid_envelope.except("effectId")) + + assert_response :unprocessable_entity + assert_empty SolidObjects::Message.all + end + + test "rejects an unknown operation" do + allow_transmissions + + post_envelope(valid_envelope("operation" => "erase")) + + assert_response :unprocessable_entity + assert_empty SolidObjects::Message.all + end + + test "rejects a conflicting replay" do + allow_transmissions + + post_envelope(valid_envelope) + post_envelope(valid_envelope("arguments" => { "amount" => 999 })) + + assert_response :unprocessable_entity + assert_equal 1, SolidObjects::Message.count + end + + test "rejects a body that is not JSON" do + allow_transmissions + + post_envelope("not json") + + assert_response :unprocessable_entity + assert_empty SolidObjects::Message.all + end + + test "rejects an oversized envelope" do + allow_transmissions + SolidObjects.configuration.max_payload_bytes = 64 + + post_envelope(valid_envelope("arguments" => { "note" => "x" * 200 })) + + assert_response :unprocessable_entity + assert_empty SolidObjects::Message.all + end +end diff --git a/test/integration/transmit_staging_test.rb b/test/integration/transmit_staging_test.rb new file mode 100644 index 0000000..c5cdf66 --- /dev/null +++ b/test/integration/transmit_staging_test.rb @@ -0,0 +1,305 @@ +# frozen_string_literal: true + +require "database_test_helper" + +class TransmitStagingTest < ActiveSupport::TestCase + class CounterActor < SolidObjects::Actor + actor_type "staging-counters" + + attribute :count, default: 0 + + def increment(amount: 1) + self.count += amount + transmit.increment(amount:) + end + + def increment_mirror(amount:) + emit "solid-objects.transmit", + operation: "increment", + arguments: { amount: }, + actorType: "staging-mirrors", + actorId: "mirror-1" + end + + def stage_invalid + emit "solid-objects.transmit", arguments: { amount: 1 } + end + end + + class MirrorActor < SolidObjects::Actor + actor_type "staging-mirrors" + + attribute :count, default: 0 + attribute :applied, default: -> { [] } + + def increment(amount: 1) + self.count += amount + applied << amount + end + end + + class FixtureCounterActor < SolidObjects::Actor + attribute :value, default: 0 + + def increment(amount: 1) + self.value += amount + transmit.increment(amount:) + end + end + + FIXTURES = JSON.parse( + File.read(File.expand_path("../../compatibility/transmit-envelopes.json", __dir__)) + ) + + setup do + SolidObjects.configuration.retry_delay = ->(_attempt) { 0 } + SolidObjects.register_actor(CounterActor.actor_type, CounterActor) + SolidObjects.register_actor(MirrorActor.actor_type, MirrorActor) + SolidObjects.register_actor("transmit-counters", FixtureCounterActor) + end + + teardown do + @worker&.stop + @effect_executor&.stop + end + + def worker + @worker ||= SolidObjects::Worker.new + end + + def effect_executor + @effect_executor ||= SolidObjects::EffectExecutor.new + end + + def drain_effects(max_passes: 50) + max_passes.times do + break if SolidObjects::Effect.where(status: %w[pending processing]).none? + + effect_executor.run_once + end + end + + def mirror_state(actor_id = "alice") + SolidObjects::Instance.find_by!(actor_type: "staging-mirrors", actor_id:).state + end + + def deliver_to_mirror(envelope) + SolidObjects::Transmission.receive( + envelope, + resolve_actor_type: ->(_actor_type) { "staging-mirrors" } + ) + end + + test "stages the transmit effect in the actor commit" do + message_reference = CounterActor.ref("alice").async.increment(amount: 2) + + worker.run_until_idle + + effect = SolidObjects::Effect.find_by!(message_id: message_reference.id) + assert_equal "solid-objects.transmit", effect.name + assert_equal "pending", effect.status + assert_equal( + { "operation" => "increment", "arguments" => { "amount" => 2 } }, + effect.arguments + ) + assert_equal({ "count" => 2 }, effect.instance.state) + end + + test "delivers a camelCase envelope carrying the effect id" do + delivered = [] + SolidObjects.register_transmit { |envelope| delivered << envelope } + CounterActor.ref("alice").async.increment(amount: 2) + worker.run_until_idle + effect = SolidObjects::Effect.sole + + assert effect_executor.run_once + + assert_equal [ + { + "effectId" => effect.effect_id, + "actorType" => "staging-counters", + "actorId" => "alice", + "operation" => "increment", + "arguments" => { "amount" => 2 } + } + ], delivered + assert_equal "completed", effect.reload.status + end + + test "replays onto a server actor through Transmission.receive" do + SolidObjects.register_transmit { |envelope| deliver_to_mirror(envelope) } + CounterActor.ref("alice").async.increment(amount: 2) + + worker.run_until_idle + drain_effects + worker.run_until_idle + + assert_equal({ "count" => 2, "applied" => [ 2 ] }, mirror_state) + end + + test "keeps per-actor order when an early envelope fails first" do + failures_remaining = 1 + SolidObjects.register_transmit do |envelope| + if envelope.dig("arguments", "amount") == 1 && failures_remaining.positive? + failures_remaining -= 1 + raise "network down" + end + + deliver_to_mirror(envelope) + end + reference = CounterActor.ref("alice") + reference.async.increment(amount: 1) + reference.async.increment(amount: 2) + + worker.run_until_idle + drain_effects + worker.run_until_idle + + assert_equal({ "count" => 3, "applied" => [ 1, 2 ] }, mirror_state) + end + + test "a later claimed effect delivers an undelivered earlier sibling first" do + delivered = [] + SolidObjects.register_transmit do |envelope| + delivered << envelope.dig("arguments", "amount") + deliver_to_mirror(envelope) + end + reference = CounterActor.ref("alice") + reference.async.increment(amount: 1) + reference.async.increment(amount: 2) + worker.run_until_idle + earlier_effect = SolidObjects::Effect.order(:id).first + earlier_effect.update!(available_at: 1.hour.from_now) + + assert effect_executor.run_once + + assert_equal [ 1, 2 ], delivered + assert_equal "pending", earlier_effect.reload.status + worker.run_until_idle + assert_equal({ "count" => 3, "applied" => [ 1, 2 ] }, mirror_state) + end + + test "concurrent executors preserve order and dedup overlapping drains" do + first_delivery_started = Queue.new + release_first_delivery = Queue.new + stall_next_delivery = true + stall_mutex = Mutex.new + SolidObjects.register_transmit do |envelope| + should_stall = stall_mutex.synchronize do + stalling = stall_next_delivery + stall_next_delivery = false + stalling + end + if should_stall + first_delivery_started << true + release_first_delivery.pop + end + deliver_to_mirror(envelope) + end + reference = CounterActor.ref("alice") + reference.async.increment(amount: 1) + reference.async.increment(amount: 2) + worker.run_until_idle + executor_a = SolidObjects::EffectExecutor.new + executor_b = SolidObjects::EffectExecutor.new + + stalled_claim = Thread.new { executor_a.run_once } + Timeout.timeout(5) { first_delivery_started.pop } + assert executor_b.run_once + release_first_delivery << true + assert stalled_claim.value + + worker.run_until_idle + assert_equal({ "count" => 3, "applied" => [ 1, 2 ] }, mirror_state) + assert_equal %w[completed completed], SolidObjects::Effect.order(:id).pluck(:status) + ensure + release_first_delivery << true + stalled_claim&.join(2) + executor_a&.stop + executor_b&.stop + end + + test "recovers in order after an offline period" do + online = false + SolidObjects.register_transmit do |envelope| + raise "offline" unless online + + deliver_to_mirror(envelope) + end + reference = CounterActor.ref("alice") + reference.async.increment(amount: 1) + reference.async.increment(amount: 2) + reference.async.increment(amount: 3) + worker.run_until_idle + + 5.times { effect_executor.run_once } + worker.run_until_idle + assert_nil SolidObjects::Instance.find_by(actor_type: "staging-mirrors", actor_id: "alice") + + online = true + drain_effects + worker.run_until_idle + + assert_equal({ "count" => 6, "applied" => [ 1, 2, 3 ] }, mirror_state) + end + + test "routes an explicit raw emit target to a different actor" do + SolidObjects.register_transmit { |envelope| SolidObjects::Transmission.receive(envelope) } + CounterActor.ref("alice").async.increment_mirror(amount: 4) + + worker.run_until_idle + drain_effects + worker.run_until_idle + + assert_equal({ "count" => 4, "applied" => [ 4 ] }, mirror_state("mirror-1")) + end + + test "retries a raised delivery and dead-letters on exhaustion" do + SolidObjects.register_transmit { |_envelope| raise "always down" } + CounterActor.ref("alice").async.increment(amount: 2) + worker.run_until_idle + effect = SolidObjects::Effect.sole + + (effect.max_attempts + 2).times { effect_executor.run_once } + + effect.reload + assert_equal "dead", effect.status + assert_equal effect.max_attempts, effect.attempt_count + assert_equal "always down", effect.error.fetch("message") + end + + test "dead-letters a malformed staged effect and skips it in the drain" do + delivered = [] + SolidObjects.register_transmit { |envelope| delivered << envelope } + reference = CounterActor.ref("alice") + reference.async.stage_invalid + reference.async.increment(amount: 2) + worker.run_until_idle + invalid_effect = SolidObjects::Effect.order(:id).first + + (invalid_effect.max_attempts + 4).times { effect_executor.run_once } + + assert_equal "dead", invalid_effect.reload.status + assert_equal "SolidObjects::InvalidTransmission", invalid_effect.error.fetch("class") + assert_equal [ "increment" ], delivered.map { |envelope| envelope.fetch("operation") } + end + + test "stages an envelope that matches the golden fixture" do + fixture = FIXTURES.fetch("valid").first + delivered = [] + SolidObjects.register_transmit { |envelope| delivered << envelope } + SolidObjects::Reference.new(actor_type: "transmit-counters", actor_id: "fixture-counter") + .async.increment(amount: 2) + + worker.run_until_idle + drain_effects + + envelope = delivered.fetch(0) + assert_equal fixture.fetch("envelope").except("effectId"), envelope.except("effectId") + + message_reference = SolidObjects::Transmission.receive(envelope) + + assert_equal "transmit:#{envelope.fetch("effectId")}", + SolidObjects::Message.find(message_reference.id).idempotency_key + end +end