diff --git a/CHANGELOG.md b/CHANGELOG.md index 3388ceb..1afb690 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,39 @@ # Changelog -## Unreleased +## 0.14.3 - 2026-08-29 + +- Cut one of the three full state copies a committed turn made. The executor + built the after image twice, once to answer whether the state changed and + once for the committed row, and a synchronous query built a third for the + mutation guard. One image now answers all three. Measured on SQLite, + committed throughput rises 4.8% at 13 KB of state, 3.1% at 116 KB, and 10.2% + at 1 MB. `benchmark/state_size.rb` is the scenario and `docs/benchmarks.md` + holds the numbers. The guard now compares the image taken after the + observables are read, so a query whose observable mutates state also fails + with `InvalidActor`. +- Stop building an encoded string that `Serialization.dump` discarded. The + method encoded every value to measure it, while most call sites pass no + `max_bytes`. It now encodes only when a limit applies. That encoding was also + the only check that a string held valid bytes, so `normalize` now checks the + encoding of every string and key it visits. A value it rejects raises + `InvalidPayload` at the call that staged it, as before, rather than a + `JSON::GeneratorError` from inside the commit transaction. +- Add `warn_state_bytes`, a soft threshold that defaults to 64 KB and must not + exceed `max_state_bytes`. A commit above it reports + `solid_objects.state.large` with the actor identity, the `byte_count`, and + the threshold. The event carries no application state, and it reports after + the commit. `max_state_bytes` keeps its 5 MB default, which measurement shows + is a limit rather than an operating point. The setting, the event, and its + payload match `warnStateBytes` in solid-objects-js, which defaults to 128 KB, + because the Node curve falls later than this one. +- Report a committed turn without letting a subscriber fail it. Every event + the executor emitted after its commit ran outside a rescue, so a subscriber + that raised turned a committed turn into a failed one: the runtime skipped + `message.completed`, tried to fail a message whose claim it had already + destroyed, and lost the worker pass. Those reports now go through + `instrument_after_commit`, which reports a raising subscriber as + `solid_objects.instrumentation.failed` and continues. This matches the + isolation `solid-objects-js` already applied to every event it emits. - Align the use-case claims with solid-objects-js. "Is it worth installing here?" listed long-lived workflows without a limit, while `docs/fit.md` diff --git a/Gemfile.lock b/Gemfile.lock index 0ec1750..1785899 100644 --- a/Gemfile.lock +++ b/Gemfile.lock @@ -1,7 +1,7 @@ PATH remote: . specs: - solid_objects (0.14.2) + solid_objects (0.14.3) 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.14.2) + solid_objects (0.14.3) 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/benchmark/state_size.rb b/benchmark/state_size.rb new file mode 100644 index 0000000..10a1689 --- /dev/null +++ b/benchmark/state_size.rb @@ -0,0 +1,5 @@ +# rbs_inline: enabled + +require_relative "support" + +SolidObjectsBenchmark.state_size diff --git a/benchmark/support.rb b/benchmark/support.rb index c2ae12a..2c15641 100644 --- a/benchmark/support.rb +++ b/benchmark/support.rb @@ -7,6 +7,25 @@ module SolidObjectsBenchmark DATABASE_PATH = File.expand_path("../tmp/solid_objects_benchmark.sqlite3", __dir__) + STATE_SIZES = [ 0, 16 * 1_024, 128 * 1_024, 1_024 * 1_024 ].freeze + STATE_ENTRY_BYTES = 26 + + class StateSizeActor < SolidObjects::Actor + actor_type "benchmark-state-size" + + attribute :count, default: 0 + attribute :filler, default: -> { {} } + + def fill(size:) + entries = size / STATE_ENTRY_BYTES + self.filler = Array.new(entries) { |index| [ "key-#{index}", "value-#{index}" ] }.to_h + filler.length + end + + def increment + self.count = count + 1 + end + end class CounterActor < SolidObjects::Actor actor_type "benchmark-counter" @@ -289,6 +308,12 @@ def component_delivery ) end + # @rbs () -> void + def state_size + warm_up_state_size + STATE_SIZES.each { |size| measure_state_size(size) } + end + # @rbs () -> void def query_count turn = message_turn_query_count @@ -433,6 +458,44 @@ def enqueue_round_robin count.times { |index| references[index % actor_count].async.increment } end + # @rbs () -> void + def warm_up_state_size + reference = StateSizeActor.ref("state-size-warm-up") + reference.fill(size: 0) + count.times { reference.async.increment } + worker = SolidObjects::Worker.new + drain(worker) + ensure + worker&.stop + end + + # @rbs (Integer) -> void + def measure_state_size(size) + silence_large_state_warning + actor_id = "state-size-#{size}" + reference = StateSizeActor.ref(actor_id) + reference.fill(size:) + state_bytes = committed_state_bytes(actor_id) + count.times { reference.async.increment } + worker = SolidObjects::Worker.new + measure("process #{count} messages with #{state_bytes} bytes of state") { drain(worker) } + ensure + worker&.stop + end + + # @rbs () -> void + def silence_large_state_warning + return unless SolidObjects.configuration.respond_to?(:warn_state_bytes=) + + SolidObjects.configuration.warn_state_bytes = SolidObjects.configuration.max_state_bytes + end + + # @rbs (String) -> Integer + def committed_state_bytes(actor_id) + instance = SolidObjects::Instance.find_by!(actor_type: "benchmark-state-size", actor_id:) + JSON.generate(instance.state).bytesize + end + # @rbs (SolidObjects::Worker) -> Integer def drain(worker) processed = 0 diff --git a/docs/benchmarks.md b/docs/benchmarks.md index 04d6898..e7262f4 100644 --- a/docs/benchmarks.md +++ b/docs/benchmarks.md @@ -109,6 +109,51 @@ result is why Solid Objects does not publish one latency promise. Network topology, adapter behavior, host schema, logging, callbacks, and contention all matter. +## State size and committed throughput + +A turn commits the whole state image. It copies the state, encodes it, and +writes the row, so every message pays for the size of the state its actor +keeps. Run the scenario with: + +```bash +COUNT=300 bundle exec ruby -Ilib benchmark/state_size.rb +``` + +Measured 2026-08-29 on an Apple M5 with 24 GB RAM, Ruby 4.0.5, Rails 8.1.3.1, +and SQLite 3.53.2. One hot actor received 300 messages at each state size. Each +figure is the median of five runs, and the two trees ran one after the other in +each round. The state holds many small entries, because a copy visits every +node, and one long string of the same length costs much less. The harness sets +`warn_state_bytes` to the hard limit, so neither tree pays for an event that +only one of them can emit. + +| Committed state | Before | After | Change | +| ---: | ---: | ---: | ---: | +| 23 bytes | 1,268.0 messages/s | 1,252.5 messages/s | -1.2% | +| 13,662 bytes | 624.1 messages/s | 654.1 messages/s | +4.8% | +| 118,786 bytes | 169.3 messages/s | 174.6 messages/s | +3.1% | +| 1,026,356 bytes | 21.6 messages/s | 23.8 messages/s | +10.2% | + +The "before" tree copied the whole state three times per committed turn and +encoded a string that it discarded whenever the caller gave no byte limit. The +"after" tree copies it twice and encodes only where a limit applies. It also +checks the encoding of every string it normalizes, which the discarded encoding +used to do, so part of the saving pays for that check. The gain grows with the +state, because the database write dominates a small turn. The empty-state row +sits inside run-to-run variance. + +The curve matters more than the change. Throughput falls about 28 times between +13 KB and 1 MB of state, and about 53 times between an empty state and 1 MB. +The `max_state_bytes` default of 5 MB is therefore a limit rather than an +operating point. `warn_state_bytes` defaults to 64 KB, and each commit above it +reports `solid_objects.state.large`. The Node package carries the same setting +as `warnStateBytes` and defaults it to 128 KB, because its measured curve falls +later: it keeps 98% of its empty-state throughput at 16 KB, where this gem +keeps 52% at 13 KB. + +These are developer-laptop numbers on one adapter. They show shape and ratio, +not a capacity guarantee. + ## Reactive delivery paths Measured 2026-08-09 on an Apple M5 with 200 iterations, for one actor mutation diff --git a/docs/operations.md b/docs/operations.md index 87b6f87..6249bdd 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -200,6 +200,7 @@ end | `max_messages_per_activation_pass` | 50 | | `max_activation_duration` | 5 seconds | | `max_mailbox_length` | 10,000 | +| `warn_state_bytes` | 64 KB | | `max_attempts` | 5 | | `process_heartbeat_interval` | 15 seconds | | `process_alive_threshold` | 60 seconds | @@ -217,6 +218,15 @@ Payload, state, and result byte limits; retry delay; table prefix; logging; wake-up; broadcast; database; and authorization adapters are also configurable. Invalid lease intervals, component counts, and size limits fail fast at boot. +`max_state_bytes` defaults to 5 MB, which is a limit rather than an operating +point. A turn copies the whole state, encodes it, and writes the row, so +committed throughput falls long before that limit: measured on SQLite, about 28 +times between 13 KB and 1 MB of state. See `docs/benchmarks.md` for the curve. +`warn_state_bytes` is the soft threshold, and it must not exceed +`max_state_bytes`. Each commit above it reports `solid_objects.state.large` and +nothing else changes, so an application that already keeps a large state keeps +working while its operator learns the cost. + Keep lease duration comfortably above renewal interval and expected database pause time. A handler can exceed the pass-duration budget because Ruby code is not safely preempted; alert on message duration and isolate untrusted work. @@ -308,6 +318,7 @@ Alert on: - ready and claimed membership counts; - mailbox-full rejections; - actor turn duration and failures; +- committed state above `warn_state_bytes`; - lost-activation rate; - dead-letter creation; - actor destruction rate; @@ -344,6 +355,23 @@ per queued item keeps only the last, and the earlier wake-up never happens. Watch this event if your actors schedule from a loop or from a handler that can run more than once. Rescheduling to the same time reports nothing. +`solid_objects.state.large` reports a committed turn whose state exceeded +`warn_state_bytes`. The payload carries the actor identity, the `byte_count` +the commit wrote, and the `threshold_bytes` it passed. It never carries the +state. The event reports after the commit, so a turn that rolled back reports +nothing. Every commit above the threshold reports, including a synchronous +query and a turn that changed nothing, so a hot actor reports once per message. +Aggregate by actor rather than alert on each event, and watch the reported size +rather than the event rate: a state that grows without a bound is what this +event exists to find. `solid-objects-js` emits the same event under the same +name and payload. + +`solid_objects.instrumentation.failed` reports a subscriber that raised while +the runtime reported a committed turn. The payload carries the +`instrumentation_event` that failed and the `error_class`. The turn itself is +unaffected, because it already committed. Watch this event to find a broken +subscriber, which would otherwise be silent. + `solid_objects.component.refreshed` covers every authorized component refresh request. Its payload carries the actor identity, `component_name`, `component_key`, declared `dependencies`, `refresh_method`, the rendered diff --git a/docs/roadmap.md b/docs/roadmap.md index 9811b11..cf0d540 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -134,7 +134,15 @@ loads them in every process, and a rejected subscription reports which condition caused it instead of closing the socket silently. - Backpressure: mailbox/payload/state/result caps and fair yields exist; - distributed per-actor rate limits and global admission control do not. + distributed per-actor rate limits and global admission control do not. The + state cap is a limit rather than an operating point. `max_state_bytes` + defaults to 5 MB, and committed throughput measured on SQLite falls about 53 + times between an empty state and 1 MB of state, which `docs/benchmarks.md` + records. A soft `warn_state_bytes` threshold, 64 KB by default, now + reports each commit above it as `solid_objects.state.large`. The hard default + stays at 5 MB, because lowering it would break an application whose actors + already exceed a lower value; a major release can lower it from the measured + curve. - Administration: `SolidObjects::Web` is a mountable Rack dashboard covering instances, mailbox, reminders, effects, broadcasts, dead letters, and processes, with actor-type and actor-id filtering, status filters, paging, a diff --git a/lib/solid_objects/configuration.rb b/lib/solid_objects/configuration.rb index 89f4c60..ab5bec4 100644 --- a/lib/solid_objects/configuration.rb +++ b/lib/solid_objects/configuration.rb @@ -15,6 +15,7 @@ class Configuration # @rbs @claim_scan_limit: Integer # @rbs @max_payload_bytes: Integer # @rbs @max_state_bytes: Integer + # @rbs @warn_state_bytes: Integer # @rbs @max_result_bytes: Integer # @rbs @max_attempts: Integer # @rbs @retry_delay: Proc @@ -63,6 +64,7 @@ class Configuration :claim_scan_limit, :max_payload_bytes, :max_state_bytes, + :warn_state_bytes, :max_result_bytes, :max_attempts, :retry_delay, @@ -116,6 +118,7 @@ def initialize @claim_scan_limit = 100 @max_payload_bytes = 1.megabyte @max_state_bytes = 5.megabytes + @warn_state_bytes = 64.kilobytes @max_result_bytes = 1.megabyte @max_attempts = 5 @retry_delay = ->(attempt) { [ 2**(attempt - 1), 60 ].min.to_f } @@ -223,6 +226,9 @@ def validate! positive_values.each do |name, value| raise ArgumentError, "#{name} must be positive" unless value.positive? end + if warn_state_bytes > max_state_bytes + raise ArgumentError, "warn_state_bytes must not exceed max_state_bytes" + end message_retention_by_actor_type.each do |actor_type, retention| raise ArgumentError, "actor type cannot be empty" if actor_type.to_s.empty? raise ArgumentError, "message retention must be positive" unless retention.positive? @@ -260,6 +266,7 @@ def positive_values claim_scan_limit:, max_payload_bytes:, max_state_bytes:, + warn_state_bytes:, max_result_bytes:, max_attempts:, process_heartbeat_interval:, diff --git a/lib/solid_objects/executor.rb b/lib/solid_objects/executor.rb index ccb3a18..4c6282c 100644 --- a/lib/solid_objects/executor.rb +++ b/lib/solid_objects/executor.rb @@ -25,9 +25,15 @@ def call SolidObjects.instrument(:"message.started", **instrumentation_payload) result = invoke_actor(message_context) - ensure_query_did_not_mutate_state!(state_before) observable_changes = changed_observables(observables_before, actor.observable_values) - complete(result, observable_changes, state_changed: actor.state.to_h != state_before) + state_after = actor.state.to_h + ensure_query_did_not_mutate_state!(state_before, state_after) + complete( + result, + observable_changes, + state_after:, + state_changed: state_after != state_before + ) true rescue LostActivation raise @@ -59,11 +65,11 @@ def invoke_actor(message_context) end end - # @rbs (Hash[String, untyped]) -> void - def ensure_query_did_not_mutate_state!(state_before) + # @rbs (Hash[String, untyped], Hash[String, untyped]) -> void + def ensure_query_did_not_mutate_state!(state_before, state_after) return unless message.delivery_mode == "sync" return unless actor.class.definition.queries.key?(message.operation.to_sym) - return if actor.state.to_h == state_before + return if state_after == state_before raise InvalidActor, "query #{message.operation.inspect} mutated actor state" end @@ -75,10 +81,10 @@ def changed_observables(before, after) end end - # @rbs (untyped, Hash[String, untyped], state_changed: bool) -> void - def complete(result, observable_changes, state_changed:) - serialized_state = Serialization.dump( - actor.state.to_h, + # @rbs (untyped, Hash[String, untyped], state_after: Hash[String, untyped], state_changed: bool) -> void + def complete(result, observable_changes, state_after:, state_changed:) + dumped_state = Serialization.dump_with_byte_size( + state_after, max_bytes: SolidObjects.configuration.max_state_bytes ) serialized_result = Serialization.dump( @@ -102,7 +108,7 @@ def complete(result, observable_changes, state_changed:) locked_message = Message.lock.find(message.id) execute_commit_actions(commit_action_intents) instance.update!( - state: serialized_state, + state: dumped_state.value, state_version: actor.class.state_version, state_revision: locked_message.sequence, last_used_at: SolidObjects.database_adapter.database_now @@ -129,17 +135,17 @@ def complete(result, observable_changes, state_changed:) end observable_changes.each_key do |observable_name| - SolidObjects.instrument( + SolidObjects.instrument_after_commit( :"broadcast.enqueued", **instrumentation_payload, observable_name: ) end moved_reminders.each do |moved| - SolidObjects.instrument(:"reminder.replaced", **moved) + SolidObjects.instrument_after_commit(:"reminder.replaced", **moved) end enqueued_effects.each do |effect| - SolidObjects.instrument( + SolidObjects.instrument_after_commit( :"effect.enqueued", effect_id: effect.effect_id, effect_name: effect.name, @@ -148,10 +154,25 @@ def complete(result, observable_changes, state_changed:) actor_id: message.actor_id ) end - SolidObjects.instrument(:"message.completed", **instrumentation_payload) + report_large_state(dumped_state.byte_size) + SolidObjects.instrument_after_commit(:"message.completed", **instrumentation_payload) SolidObjects.wake_up.signal end + # @rbs (Integer) -> void + def report_large_state(byte_count) + threshold = SolidObjects.configuration.warn_state_bytes + return if byte_count <= threshold + + SolidObjects.instrument_after_commit( + :"state.large", + actor_type: message.actor_type, + actor_id: message.actor_id, + byte_count:, + threshold_bytes: threshold + ) + end + # @rbs (Array[Actor::CommitActionIntent]) -> void def execute_commit_actions(intents) ensure_application_database_is_shared! if intents.any? @@ -352,7 +373,7 @@ def fail_message(error) end end - SolidObjects.instrument( + SolidObjects.instrument_after_commit( :"message.failed", **instrumentation_payload, error_class: error.class.name, @@ -387,7 +408,7 @@ def reject_message(rejection) claimed_message.destroy! end - SolidObjects.instrument( + SolidObjects.instrument_after_commit( :"message.rejected", **instrumentation_payload, code: rejection.code diff --git a/lib/solid_objects/instrumentation.rb b/lib/solid_objects/instrumentation.rb index c8b15dc..29fc57e 100644 --- a/lib/solid_objects/instrumentation.rb +++ b/lib/solid_objects/instrumentation.rb @@ -6,5 +6,38 @@ module Instrumentation def instrument(event, **payload, &block) ActiveSupport::Notifications.instrument("solid_objects.#{event}", payload, &block) end + + # @rbs (Symbol, **untyped) -> void + def instrument_after_commit(event, **payload) + instrument(event, **payload) + rescue => error + report_instrumentation_failure(event, error) + end + + private + + # @rbs (Symbol, Exception) -> void + def report_instrumentation_failure(event, error) + instrument( + :"instrumentation.failed", + instrumentation_event: "solid_objects.#{event}", + error_class: error.class.name + ) + rescue => failure + log_instrumentation_failure(event, failure) + end + + # @rbs (Symbol, Exception) -> void + def log_instrumentation_failure(event, error) + SolidObjects.configuration.logger.error( + { + event: "solid_objects.instrumentation.failed", + instrumentation_event: "solid_objects.#{event}", + error_class: error.class.name + } + ) + rescue + nil + end end end diff --git a/lib/solid_objects/serialization.rb b/lib/solid_objects/serialization.rb index ca9bfc1..8f82c32 100644 --- a/lib/solid_objects/serialization.rb +++ b/lib/solid_objects/serialization.rb @@ -4,17 +4,26 @@ module SolidObjects module Serialization MAX_NESTING = 100 + Dumped = Data.define(:value, :byte_size) + class << self # @rbs (untyped, ?max_bytes: Integer?) -> untyped def dump(value, max_bytes: nil) + return normalize(value) unless max_bytes + + dump_with_byte_size(value, max_bytes:).value + end + + # @rbs (untyped, ?max_bytes: Integer?) -> Dumped + def dump_with_byte_size(value, max_bytes: nil) normalized = normalize(value) - encoded = JSON.generate(normalized, max_nesting: MAX_NESTING) + byte_size = JSON.generate(normalized, max_nesting: MAX_NESTING).bytesize - if max_bytes && encoded.bytesize > max_bytes + if max_bytes && byte_size > max_bytes raise PayloadTooLarge, "serialized value exceeds #{max_bytes} bytes" end - normalized + Dumped.new(value: normalized, byte_size:) rescue JSON::GeneratorError, EncodingError => error raise InvalidPayload, error.message end @@ -60,7 +69,11 @@ def normalize(value, depth: 0) raise InvalidPayload, "serialized value is nested too deeply" if depth > MAX_NESTING case value - when nil, true, false, String, Integer + when nil, true, false, Integer + value + when String + raise InvalidPayload, "strings must hold valid #{value.encoding} bytes" unless value.valid_encoding? + value when Float raise InvalidPayload, "non-finite numbers are not supported" unless value.finite? @@ -89,7 +102,7 @@ def normalize_hash(value, depth:) # @rbs (untyped) -> String def normalize_key(key) - return key if key.is_a?(String) + return normalize(key) if key.is_a?(String) return key.to_s if key.is_a?(Symbol) raise InvalidPayload, "JSON object keys must be strings or symbols" diff --git a/lib/solid_objects/version.rb b/lib/solid_objects/version.rb index 80fca80..f4f4ed6 100644 --- a/lib/solid_objects/version.rb +++ b/lib/solid_objects/version.rb @@ -1,5 +1,5 @@ # rbs_inline: enabled module SolidObjects - VERSION = "0.14.2" + VERSION = "0.14.3" end diff --git a/sig/generated/lib/solid_objects/configuration.rbs b/sig/generated/lib/solid_objects/configuration.rbs index 05facca..c565f0f 100644 --- a/sig/generated/lib/solid_objects/configuration.rbs +++ b/sig/generated/lib/solid_objects/configuration.rbs @@ -2,7 +2,7 @@ module SolidObjects class Configuration - @table_name_prefix: String + @process_alive_threshold: Float @shutdown_timeout: Float @@ -60,6 +60,8 @@ module SolidObjects @transmission_actor_type_resolver: Proc + @table_name_prefix: String + @polling_interval: Float @idle_polling_interval: Float @@ -84,6 +86,8 @@ module SolidObjects @max_state_bytes: Integer + @warn_state_bytes: Integer + @max_result_bytes: Integer @max_attempts: Integer @@ -94,8 +98,6 @@ module SolidObjects @process_heartbeat_interval: Float - @process_alive_threshold: Float - attr_accessor table_name_prefix: untyped attr_accessor polling_interval: untyped @@ -122,6 +124,8 @@ module SolidObjects attr_accessor max_state_bytes: untyped + attr_accessor warn_state_bytes: untyped + attr_accessor max_result_bytes: untyped attr_accessor max_attempts: untyped diff --git a/sig/generated/lib/solid_objects/executor.rbs b/sig/generated/lib/solid_objects/executor.rbs index 8cb747c..558a6c6 100644 --- a/sig/generated/lib/solid_objects/executor.rbs +++ b/sig/generated/lib/solid_objects/executor.rbs @@ -24,14 +24,17 @@ module SolidObjects # @rbs (MessageContext) -> untyped def invoke_actor: (MessageContext) -> untyped - # @rbs (Hash[String, untyped]) -> void - def ensure_query_did_not_mutate_state!: (Hash[String, untyped]) -> void + # @rbs (Hash[String, untyped], Hash[String, untyped]) -> void + def ensure_query_did_not_mutate_state!: (Hash[String, untyped], Hash[String, untyped]) -> void # @rbs (Hash[String, untyped], Hash[String, untyped]) -> Hash[String, untyped] def changed_observables: (Hash[String, untyped], Hash[String, untyped]) -> Hash[String, untyped] - # @rbs (untyped, Hash[String, untyped], state_changed: bool) -> void - def complete: (untyped, Hash[String, untyped], state_changed: bool) -> void + # @rbs (untyped, Hash[String, untyped], state_after: Hash[String, untyped], state_changed: bool) -> void + def complete: (untyped, Hash[String, untyped], state_after: Hash[String, untyped], state_changed: bool) -> void + + # @rbs (Integer) -> void + def report_large_state: (Integer) -> void # @rbs (Array[Actor::CommitActionIntent]) -> void def execute_commit_actions: (Array[Actor::CommitActionIntent]) -> void diff --git a/sig/generated/lib/solid_objects/instrumentation.rbs b/sig/generated/lib/solid_objects/instrumentation.rbs index 03f69b3..02af2a5 100644 --- a/sig/generated/lib/solid_objects/instrumentation.rbs +++ b/sig/generated/lib/solid_objects/instrumentation.rbs @@ -4,5 +4,16 @@ module SolidObjects module Instrumentation # @rbs (Symbol, **untyped) { (Hash[Symbol, untyped]) -> untyped } -> untyped def instrument: (Symbol, **untyped) { (Hash[Symbol, untyped]) -> untyped } -> untyped + + # @rbs (Symbol, **untyped) -> void + def instrument_after_commit: (Symbol, **untyped) -> void + + private + + # @rbs (Symbol, Exception) -> void + def report_instrumentation_failure: (Symbol, Exception) -> void + + # @rbs (Symbol, Exception) -> void + def log_instrumentation_failure: (Symbol, Exception) -> void end end diff --git a/sig/generated/lib/solid_objects/serialization.rbs b/sig/generated/lib/solid_objects/serialization.rbs index e16215a..9bfc08b 100644 --- a/sig/generated/lib/solid_objects/serialization.rbs +++ b/sig/generated/lib/solid_objects/serialization.rbs @@ -4,9 +4,25 @@ module SolidObjects module Serialization MAX_NESTING: ::Integer + class Dumped < Data + attr_reader value(): untyped + + attr_reader byte_size(): untyped + + def self.new: (untyped value, untyped byte_size) -> instance + | (value: untyped, byte_size: untyped) -> instance + + def self.members: () -> [ :value, :byte_size ] + + def members: () -> [ :value, :byte_size ] + end + # @rbs (untyped, ?max_bytes: Integer?) -> untyped def self.dump: (untyped, ?max_bytes: Integer?) -> untyped + # @rbs (untyped, ?max_bytes: Integer?) -> Dumped + def self.dump_with_byte_size: (untyped, ?max_bytes: Integer?) -> Dumped + # @rbs (untyped) -> untyped def self.load: (untyped) -> untyped diff --git a/test/integration/state_commit_test.rb b/test/integration/state_commit_test.rb new file mode 100644 index 0000000..d57883f --- /dev/null +++ b/test/integration/state_commit_test.rb @@ -0,0 +1,269 @@ +# frozen_string_literal: true + +require "database_test_helper" + +class StateCommitTest < ActiveSupport::TestCase + class CountingActor < SolidObjects::Actor + actor_type "state-commit-counting" + + class << self + attr_accessor :state_copies + end + + attribute :count, default: 0 + + def increment + count_state_copies + self.count = count + 1 + end + + query :current do + count_state_copies + count + end + + query :mutating do + self.count = count + 1 + end + + private + + def count_state_copies + actor_class = self.class + state.define_singleton_method(:to_h) do |*arguments| + actor_class.state_copies += 1 + super(*arguments) + end + end + end + + class ObservableMutatingActor < SolidObjects::Actor + actor_type "state-commit-observable" + + class << self + attr_accessor :mutate_on_read + end + + attribute :count, default: 0 + + observable :reads do + self.count = count + 1 if self.class.mutate_on_read + count + end + + query :current do + self.class.mutate_on_read = true + count + end + end + + class LargeStateActor < SolidObjects::Actor + actor_type "state-commit-large" + + attribute :filler, default: "" + + def grow(size:) + self.filler = "x" * size + end + end + + class RollingBackActor < SolidObjects::Actor + actor_type "state-commit-rollback" + + attribute :filler, default: "" + + def grow(size:) + self.filler = "x" * size + commit_action :state_commit_failure + end + end + + class RaisingLogger + def error(*) + raise "logger failed" + end + + def method_missing(*) + nil + end + + def respond_to_missing?(*) + true + end + end + + setup do + CountingActor.state_copies = 0 + ObservableMutatingActor.mutate_on_read = false + end + + test "a committed message copies the state once after its handler runs" do + CountingActor.ref("message").async.increment + worker = SolidObjects::Worker.new + worker.run_until_idle + + assert_equal 1, CountingActor.state_copies + instance = SolidObjects::Instance.find_by!(actor_type: "state-commit-counting", actor_id: "message") + assert_equal({ "count" => 1 }, instance.state) + ensure + worker&.stop + end + + test "a committed message encodes the state once and the result once" do + CountingActor.ref("encodes").async.increment + worker = SolidObjects::Worker.new + + assert_equal 2, measured_dumps { worker.run_until_idle } + ensure + worker&.stop + end + + test "a committed query copies the state once after its handler runs" do + assert_equal 0, CountingActor.ref("query").sync.current + + assert_equal 1, CountingActor.state_copies + end + + test "a query that mutates state fails its message" do + SolidObjects.configuration.max_attempts = 1 + + error = assert_raises(SolidObjects::MessageFailed) do + CountingActor.ref("guard").sync.mutating + end + + assert_equal "SolidObjects::InvalidActor", error.details.fetch("class") + instance = SolidObjects::Instance.find_by!(actor_type: "state-commit-counting", actor_id: "guard") + assert_empty instance.state, "the rejected mutation must not reach the committed row" + end + + test "a query whose observable mutates state fails its message" do + SolidObjects.configuration.max_attempts = 1 + + error = assert_raises(SolidObjects::MessageFailed) do + ObservableMutatingActor.ref("observable").sync.current + end + + assert_equal "SolidObjects::InvalidActor", error.details.fetch("class") + instance = SolidObjects::Instance.find_by!(actor_type: "state-commit-observable", actor_id: "observable") + assert_empty instance.state, "the observable mutation must not reach the committed row" + end + + test "reports committed state above the soft threshold" do + SolidObjects.configuration.warn_state_bytes = 64 + events = [] + subscription = ActiveSupport::Notifications.subscribe("solid_objects.state.large") do |event| + events << event.payload + end + + LargeStateActor.ref("big").async.grow(size: 512) + worker = SolidObjects::Worker.new + worker.run_until_idle + + assert_equal 1, events.length + assert_equal %i[actor_type actor_id byte_count threshold_bytes].sort, events.sole.keys.sort + assert_equal "state-commit-large", events.sole.fetch(:actor_type) + assert_equal "big", events.sole.fetch(:actor_id) + assert_equal 64, events.sole.fetch(:threshold_bytes) + assert_operator events.sole.fetch(:byte_count), :>, 512 + ensure + ActiveSupport::Notifications.unsubscribe(subscription) if subscription + worker&.stop + end + + test "stays silent for committed state under the soft threshold" do + SolidObjects.configuration.warn_state_bytes = 4_096 + events = [] + subscription = ActiveSupport::Notifications.subscribe("solid_objects.state.large") do |event| + events << event.payload + end + + LargeStateActor.ref("small").async.grow(size: 512) + worker = SolidObjects::Worker.new + worker.run_until_idle + + assert_empty events + ensure + ActiveSupport::Notifications.unsubscribe(subscription) if subscription + worker&.stop + end + + test "stays silent for state a rolled back turn never wrote" do + SolidObjects.configuration.warn_state_bytes = 64 + SolidObjects.configuration.max_attempts = 1 + SolidObjects.register_commit_action(:state_commit_failure) { raise "commit action failed" } + events = [] + subscription = ActiveSupport::Notifications.subscribe("solid_objects.state.large") do |event| + events << event.payload + end + + RollingBackActor.ref("rolled-back").async.grow(size: 512) + worker = SolidObjects::Worker.new + worker.run_until_idle + + assert_empty events + instance = SolidObjects::Instance.find_by!(actor_type: "state-commit-rollback", actor_id: "rolled-back") + assert_empty instance.state + ensure + ActiveSupport::Notifications.unsubscribe(subscription) if subscription + worker&.stop + end + + test "a reporting failure does not disturb a turn that already committed" do + SolidObjects.configuration.warn_state_bytes = 64 + completions = [] + failures = [] + subscriptions = [ + ActiveSupport::Notifications.subscribe("solid_objects.state.large") { raise "subscriber failed" }, + ActiveSupport::Notifications.subscribe("solid_objects.message.completed") { completions << true }, + ActiveSupport::Notifications.subscribe("solid_objects.instrumentation.failed") { |event| failures << event.payload } + ] + + LargeStateActor.ref("noisy").async.grow(size: 512) + worker = SolidObjects::Worker.new + worker.run_until_idle + + message = SolidObjects::Message.find_by!(actor_type: "state-commit-large", actor_id: "noisy") + assert_predicate message, :completed? + assert_equal 1, completions.length, "a committed turn must still report completion" + assert_equal 1, failures.length + assert_equal "solid_objects.state.large", failures.sole.fetch(:instrumentation_event) + assert_empty SolidObjects::ReadyMessage.all, "a committed turn must not run again" + ensure + subscriptions&.each { |subscription| ActiveSupport::Notifications.unsubscribe(subscription) } + worker&.stop + end + + test "a failure that reports a failure does not disturb a committed turn" do + SolidObjects.configuration.warn_state_bytes = 64 + SolidObjects.configuration.logger = RaisingLogger.new + completions = [] + subscriptions = [ + ActiveSupport::Notifications.subscribe("solid_objects.state.large") { raise "subscriber failed" }, + ActiveSupport::Notifications.subscribe("solid_objects.instrumentation.failed") { raise "reporter failed" }, + ActiveSupport::Notifications.subscribe("solid_objects.message.completed") { completions << true } + ] + + LargeStateActor.ref("hostile").async.grow(size: 512) + worker = SolidObjects::Worker.new + worker.run_until_idle + + message = SolidObjects::Message.find_by!(actor_type: "state-commit-large", actor_id: "hostile") + assert_predicate message, :completed? + assert_equal 1, completions.length, "a committed turn must still report completion" + assert_empty SolidObjects::ReadyMessage.all, "a committed turn must not run again" + ensure + subscriptions&.each { |subscription| ActiveSupport::Notifications.unsubscribe(subscription) } + worker&.stop + end + + private + + def measured_dumps + calls = 0 + trace = TracePoint.new(:call) do |point| + calls += 1 if point.method_id == :dump_with_byte_size + end + trace.enable { yield } + calls + end +end diff --git a/test/unit/configuration_test.rb b/test/unit/configuration_test.rb index 623a8a8..f8a1bf1 100644 --- a/test/unit/configuration_test.rb +++ b/test/unit/configuration_test.rb @@ -49,6 +49,31 @@ class ConfigurationTest < ActiveSupport::TestCase assert_equal "instance retention must be positive", error.message end + test "warns about large state well below the hard state limit" do + configuration = SolidObjects::Configuration.new + + assert_equal 64.kilobytes, configuration.warn_state_bytes + assert_operator configuration.warn_state_bytes, :<, configuration.max_state_bytes + end + + test "rejects a soft state threshold above the hard state limit" do + configuration = SolidObjects::Configuration.new + configuration.warn_state_bytes = configuration.max_state_bytes + 1 + + error = assert_raises(ArgumentError) { configuration.validate! } + + assert_equal "warn_state_bytes must not exceed max_state_bytes", error.message + end + + test "rejects a non-positive state size warning threshold" do + configuration = SolidObjects::Configuration.new + configuration.warn_state_bytes = 0 + + error = assert_raises(ArgumentError) { configuration.validate! } + + assert_equal "warn_state_bytes must be positive", error.message + end + test "rejects a non-positive idle polling interval" do configuration = SolidObjects::Configuration.new configuration.idle_polling_interval = 0 diff --git a/test/unit/serialization_test.rb b/test/unit/serialization_test.rb index 52e6249..2d6de0f 100644 --- a/test/unit/serialization_test.rb +++ b/test/unit/serialization_test.rb @@ -50,6 +50,57 @@ def as_json end end + test "does not encode a value that has no byte limit" do + assert_equal 0, json_generate_calls { SolidObjects::Serialization.dump({ quantity: 1 }) } + end + + test "encodes once when it must enforce a byte limit" do + assert_equal 1, json_generate_calls { SolidObjects::Serialization.dump({ quantity: 1 }, max_bytes: 1_024) } + end + + test "rejects a value that is not JSON-compatible when it has no byte limit" do + assert_raises(SolidObjects::InvalidPayload) do + SolidObjects::Serialization.dump(Object.new) + end + end + + test "rejects a string that cannot be encoded when it has no byte limit" do + assert_raises(SolidObjects::InvalidPayload) do + SolidObjects::Serialization.dump(invalid_encoding_string) + end + end + + test "rejects a string that cannot be encoded inside a nested value" do + assert_raises(SolidObjects::InvalidPayload) do + SolidObjects::Serialization.dump({ "items" => [ { "note" => invalid_encoding_string } ] }) + end + end + + test "rejects a key that cannot be encoded" do + assert_raises(SolidObjects::InvalidPayload) do + SolidObjects::Serialization.dump({ invalid_encoding_string => "value" }) + end + end + + test "converts an encoding failure into an invalid payload" do + assert_raises(SolidObjects::InvalidPayload) do + SolidObjects::Serialization.dump(invalid_encoding_string, max_bytes: 1_024) + end + end + + test "reports the encoded byte size beside the normalized value" do + dumped = SolidObjects::Serialization.dump_with_byte_size({ quantity: 1 }, max_bytes: 1_024) + + assert_equal({ "quantity" => 1 }, dumped.value) + assert_equal 14, dumped.byte_size + end + + test "enforces the encoded byte limit while it reports the byte size" do + assert_raises(SolidObjects::PayloadTooLarge) do + SolidObjects::Serialization.dump_with_byte_size("four", max_bytes: 5) + end + end + test "returns an independent deep copy" do original = { "items" => [ { "quantity" => 1 } ] } copy = SolidObjects::Serialization.deep_copy(original) @@ -68,4 +119,19 @@ def as_json assert_predicate copy.fetch("items").first, :frozen? assert_raises(FrozenError) { copy.fetch("items").first["quantity"] = 2 } end + + private + + def json_generate_calls + calls = 0 + trace = TracePoint.new(:call, :c_call) do |point| + calls += 1 if point.method_id == :generate && point.self == JSON + end + trace.enable { yield } + calls + end + + def invalid_encoding_string + (+"\xC3").force_encoding(Encoding::UTF_8) + end end