feat: Add portable actor observability

Match JavaScript telemetry and actor diagnostics using native Active Support notifications and Active Record scopes. Isolate subscriber failures while preserving application exceptions, and keep private error text out of default logs.

Refs cardmagic/solid-objects-js#42

Author: cardmagic

CHANGELOG.md

@@ -1,5 +1,10 @@
 # Changelog
 
+## Unreleased
+
+- Add portable telemetry, isolated observer hooks, metric definitions, and bounded authorized actor diagnostics matching JavaScript.
+
+
 ## 0.16.0 - 2026-09-23
 
 - Find a message whose reference a caller lost.

README.md

@@ -162,6 +162,7 @@ Exactly once is not hiding in a more advanced configuration. Read the
 - [Five-minute Rails guide](https://solidobjects.dev/5min/rails)
 - [Choosing Solid Objects](docs/fit.md)
 - [Operations and recovery](docs/operations.md)
+- [Observability and diagnostics](docs/observability.md)
 - [Reminders](docs/reminders.md)
 - [Reactive ERB](docs/realtime.md)
 - [Detailed architecture](docs/architecture.md)

docs/observability.md

@@ -0,0 +1,123 @@
+# Portable observability
+
+Configure `instrumentation(event)` to receive structured events. No exporter SDK
+is required. The same JSON envelope is emitted by Ruby, SQLite, PostgreSQL, MySQL,
+and the Durable Objects host. Existing JavaScript `name`, `occurredAt`, and
+`attributes` fields remain available. Ruby's Active Support notifications remain
+available with their existing snake_case payloads.
+
+## Schema version 1
+
+Every event has `schemaVersion`, `name` (prefixed with `solid_objects.`),
+`occurredAt` (UTC ISO 8601), `adapter`, `actorType`, `actorId`, `incarnation`,
+`revision`, `messageId`, `attempt`, `attributes`, and `metrics`.
+Unavailable identifiers are null; `attempt` is zero outside a message attempt.
+An incarnation identifies a persisted actor instance, independently of its lease
+generation. Revisions and IDs are strings. Process-wide events have null actor
+identity. A message ID or revision correlates actor work where applicable.
+
+Only known scalar metadata fields enter the portable envelope. Arguments, state,
+results, credentials, backtraces, exception text, nested provider data, and unknown
+attributes are excluded. Actor IDs remain correlation data: applications should
+use opaque actor identifiers and apply their own retention policy to event logs.
+Events and metric samples are immutable. Throwing observers, rejected observer
+promises, and a failing instrumentation error logger cannot change a turn's
+result. Delivery is best effort and synchronous callbacks should be short;
+JavaScript does not await exporters. Telemetry is not a durable audit trail.
+
+| Event | Meaning |
+| --- | --- |
+| `activation.started/completed/failed` | Local actor activation hook lifecycle |
+| `message.started/completed/failed/rejected` | One attempt's execution outcome |
+| `message.retry` | Failed attempt durably queued for another attempt |
+| `dead_letter.created` | Message exhausted retries or failed permanently |
+| `mailbox.depth` | On-demand diagnostic sample; exact depth only if not truncated |
+| `reminder.enqueued` | Due reminder dispatch; lateness is measured from due time |
+| `outbox.age` | Delivery observation; age is time since the item's current availability time |
+| `recovery.reclaimed` | A previously claimed, interrupted message begins another attempt |
+| `recovery.completed/failed` | Durable effect recovery callback commits or enters the dead-letter queue |
+| `snapshot.read` | Authorized snapshot constructed without exposing its contents |
+| `realtime.connected/disconnected` | Actor subscription added or removed |
+
+Events describe local observations. Concurrent deletion, crashes, and failed
+exporters can omit events. Never infer exactly-once delivery from event counts.
+Additional existing runtime events retain their names.
+
+## Metrics and tracing
+
+Metrics are sample descriptions. Exporting them is opt-in: the runtime does not
+register meters, allocate per-actor metric series, or install a vendor SDK.
+
+| Name | Kind | Unit | Aggregation |
+| --- | --- | --- | --- |
+| `solid_objects.events` | counter | `1` | Sum one per event |
+| `solid_objects.duration` | histogram | `ms` | Distribution of observed attempt duration |
+| `solid_objects.reminder.lateness` | histogram | `ms` | Distribution of reminder dispatch delay |
+| `solid_objects.outbox.age` | histogram | `ms` | Distribution of delivery delay since availability |
+| `solid_objects.mailbox.depth` | gauge | `1` | Last exact sampled actor depth; omit truncated samples |
+
+Labels contain only event name, adapter family, and declared actor type. Keep the
+actor type registry finite. Never add actor ID, incarnation, message ID, operation
+arguments, request IDs, or error text to metric labels. A gauge without an actor
+label represents the most recently observed actor; it is not total fleet backlog.
+For tracing, correlate start/outcome events using `adapter`, `incarnation`,
+`messageId`, and `attempt`, and close or expire spans when no outcome arrives.
+
+## Actor observers and diagnostics
+
+```ruby
+cart = ShoppingCart.ref("demo-cart")
+stop = cart.observe(authorization_context: operator) do |event|
+  logger.info(event.to_json)
+end
+summary = cart.diagnostics(authorization_context: operator, limit: 50)
+stop.call
+```
+
+Ruby uses `reference.observe(authorization_context:) { |event| ... }` and
+`reference.diagnostics(authorization_context:, limit: 50)`. Stop observing by
+calling the returned proc. `on` filters one event name, such as `message.retry`.
+Observers receive only this actor's events in the current runtime/process; they
+are not subscriptions to workers on other hosts. Dispose them when the caller's
+session ends or authorization is revoked. JS limits local observers to 1,000.
+For remote Durable Objects, configure `instrumentation` on the actor host and
+filter by actor identity there; process-local reference observers raise
+`UnsupportedCapability`. Remote `reference.diagnostics` is supported.
+
+Both APIs default to denied. Set `authorizeAdministration` / `authorize_administration`
+to allow action `observe` or `inspect`, resource `actor_diagnostics`, and resource ID
+`JSON.stringify([actorType, actorId])`. Ruby receives a symbol action. Authorization
+runs before reading summaries or registering observers; possessing an actor ID
+confers no permission.
+
+Diagnostics read at most `limit + 1` rows per queue source, with a hard limit of
+100. Each category returns `sampled`, `truncated`, and `oldestAgeMilliseconds`.
+The last value measures nonnegative time since availability (or terminal failure
+for recovery callbacks); future reminders have zero age. No payloads or row
+identifiers are returned. Samples are observations across several queries,
+not an atomic fleet snapshot. Large queues can still require database scanning;
+the bound limits materialized rows and response size, not query execution time.
+
+Categories are mailbox (ready and claimed), outbox (pending and processing effects
+and broadcasts), reminders (scheduled and paused), retries (failed messages still
+eligible to run), and recoveryFailures (dead internal effect recovery
+callback messages). Durable Objects does not implement process-heartbeat effect recovery;
+its recoveryFailures category is empty. Recovery failures are durable records,
+not a history of transient database or exporter exceptions.
+
+## A query that works across adapters
+
+Write one envelope per line to `events.jsonl`, using the same instrumentation hook
+with SQLite and PostgreSQL. This query reports completed attempts by adapter:
+
+```sh
+jq -s 'map(select(.name == "solid_objects.message.completed"))
+  | group_by(.adapter)
+  | map({adapter: .[0].adapter, completed: length,
+         mean_ms: (map(.attributes.durationMilliseconds) | add / length)})' events.jsonl
+```
+
+A dashboard can chart the event counter by adapter and outcome and the duration,
+reminder lateness, and outbox age distributions using exactly the same fields.
+Compare workloads with the same actor types and sampling policy. Counts measure
+observations and cannot replace a database query for authoritative queue state.

docs/roadmap.md

@@ -2,6 +2,9 @@
 
 ## Implemented and tested
 
+- Portable telemetry with a shared JSON schema, metric samples, isolated observer
+  callbacks, and bounded authorized actor diagnostics. See [observability](observability.md).
+
 - Rails engine, install generator, migration, and CLI
 - Explicit actor registry, references, JSON state, and state migrations
 - Fluent direct synchronous RPC, configured `sync`, and durable `async`

lib/solid_objects.rb

@@ -16,6 +16,8 @@
 require "solid_objects/callable_keywords"
 require "solid_objects/configuration"
 require "solid_objects/instrumentation"
+require "solid_objects/telemetry"
+require "solid_objects/diagnostics"
 require "solid_objects/log_subscriber"
 require "solid_objects/serialization"
 require "solid_objects/context"

lib/solid_objects/activation.rb

@@ -24,15 +24,19 @@ def initialize(lease:)
       @actor = build_actor(instance)
       @last_used_at = monotonic_now
       @pass_exhausted = false
+      SolidObjects.instrument(:"activation.started", instance_id: instance.id, actor_type: instance.actor_type, actor_id: instance.actor_id, generation: lease.generation)
       actor.activate
       SolidObjects.instrument(
-        :"activation.started",
+        :"activation.completed",
         instance_id: instance.id,
         actor_type: instance.actor_type,
         actor_id: instance.actor_id,
         owner_id: lease.owner_id,
         generation: lease.generation
       )
+    rescue => error
+      SolidObjects.instrument(:"activation.failed", instance_id: lease.instance_id, actor_type: instance&.actor_type, actor_id: instance&.actor_id, error_class: error.class.name)
+      raise
     end
 
     # @rbs () -> Integer
@@ -104,8 +108,7 @@ def deactivate
         actor_id: actor.actor_id,
         owner_id: lease.owner_id,
         generation: lease.generation,
-        error_class: error.class.name,
-        error_message: error.message
+        error_class: error.class.name
       )
       SolidObjects.configuration.logger.error(
         "SolidObjects activation deactivation failed " \

lib/solid_objects/actor_channel.rb

@@ -49,6 +49,7 @@ def subscribed
       end
       refresh_outdated_components(snapshot)
       transmit_state_payloads(snapshot)
+      SolidObjects.instrument(:"realtime.connected", actor_type: reference.actor_type, actor_id: reference.actor_id, instance_id: snapshot.instance_id, revision: snapshot.revision)
     rescue KeyError,
       JSON::ParserError,
       InvalidStreamToken,
@@ -57,6 +58,13 @@ def subscribed
       reject_and_report(reject_reason(error), actor_type:, actor_id:, error:)
     end
 
+    # @rbs () -> void
+    def unsubscribed
+      return unless reference
+
+      SolidObjects.instrument(:"realtime.disconnected", actor_type: reference.actor_type, actor_id: reference.actor_id)
+    end
+
     private
 
     attr_reader :reference,

lib/solid_objects/configuration.rb

@@ -40,6 +40,7 @@ class Configuration
     # @rbs @broadcast_worker_count: Integer
     # @rbs @reminder_scheduler_count: Integer
     # @rbs @connects_to: Hash[Symbol, untyped]?
+    # @rbs @instrumentation: Proc?
     # @rbs @logger: untyped
     # @rbs @stream_signing_secret: String?
     # @rbs @broadcast_adapter: Proc?
@@ -95,6 +96,7 @@ class Configuration
       :reminder_scheduler_count,
       :connects_to,
       :logger,
+      :instrumentation,
       :stream_signing_secret,
       :broadcast_adapter,
       :wake_up_adapter,
@@ -159,6 +161,7 @@ def initialize
       @component_path_resolver = nil
       @component_authorization_context = ->(controller:) { controller }
       @payload_authorization_context = ->(connection:) { connection }
+      @instrumentation = nil
       @logger = if defined?(Rails) && Rails.respond_to?(:logger) && Rails.logger
         Rails.logger
       else

lib/solid_objects/diagnostics.rb

@@ -0,0 +1,86 @@
+# rbs_inline: enabled
+
+module SolidObjects
+  class Diagnostics
+    # @rbs @reference: Reference
+    # @rbs (Reference) -> void
+    def initialize(reference)
+      @reference = reference
+    end
+
+    # @rbs (?limit: Integer, ?authorization_context: untyped) -> Hash[String, untyped]
+    def summary(limit: 100, authorization_context: nil)
+      authorize!(:inspect, authorization_context:)
+      raise ArgumentError, "diagnostic limit must be an integer between 1 and 100" unless limit.is_a?(Integer) && limit.between?(1, 100)
+
+      instance = Instance.find_by(actor_type: reference.actor_type, actor_id: reference.actor_id)
+      now = SolidObjects.database_adapter.database_now
+      instance_id = instance&.id
+      mailbox = ReadyMessage.where(instance_id:).order(:available_at).limit(limit + 1).pluck(:available_at) +
+        ClaimedMessage.where(instance_id:).order(:claimed_at).limit(limit + 1).pluck(:claimed_at)
+      outbox = Effect.where(instance_id:, status: %w[pending processing]).order(:available_at).limit(limit + 1).pluck(:available_at) +
+        Broadcast.where(instance_id:, status: %w[pending processing]).order(:available_at).limit(limit + 1).pluck(:available_at)
+      reminders = Reminder.where(instance_id:, status: %w[scheduled paused]).order(:next_run_at).limit(limit + 1).pluck(:next_run_at)
+      retries = ReadyMessage.joins(:message).where(instance_id:).where.not(Message.table_name => { error: nil }).order(:available_at).limit(limit + 1).pluck(:available_at)
+      recovery_messages = Message.where(instance_id:, delivery_mode: "internal").where("idempotency_key LIKE ?", "effect:%:recovery").select(:id)
+      recovery_failures = DeadLetter.where(instance_id:, message_id: recovery_messages).order(:last_failed_at).limit(limit + 1).pluck(:last_failed_at)
+      result = {
+        "actorType" => reference.actor_type,
+        "actorId" => reference.actor_id,
+        "incarnation" => instance_id&.to_s,
+        "revision" => instance&.state_revision&.to_s,
+        "adapter" => DatabaseAdapter.family(Record.connection).to_s,
+        "occurredAt" => now.utc.iso8601(3),
+        "limit" => limit,
+        "mailbox" => summarize(mailbox, now:, limit:),
+        "outbox" => summarize(outbox, now:, limit:),
+        "reminders" => summarize(reminders, now:, limit:),
+        "retries" => summarize(retries, now:, limit:),
+        "recoveryFailures" => summarize(recovery_failures, now:, limit:)
+      }
+      SolidObjects.instrument(:"mailbox.depth", actor_type: reference.actor_type, actor_id: reference.actor_id, instance_id:, count: result.fetch("mailbox").fetch("sampled"), truncated: result.fetch("mailbox").fetch("truncated"), depth: result.fetch("mailbox").fetch("truncated") ? nil : result.fetch("mailbox").fetch("sampled"))
+      Serialization.readonly_copy(result)
+    end
+
+    # @rbs (?authorization_context: untyped) { (Hash[String, untyped]) -> untyped } -> Proc
+    def observe(authorization_context: nil, &block)
+      authorize!(:observe, authorization_context:)
+      subscription = ActiveSupport::Notifications.subscribe(/\Asolid_objects\./) do |notification|
+        payload = notification.payload
+        next unless payload[:actor_type] == reference.actor_type && payload[:actor_id] == reference.actor_id
+
+        begin
+          block.call(Telemetry.event(notification.name.delete_prefix("solid_objects.").to_sym, payload))
+        rescue
+          nil
+        end
+      end
+      -> { ActiveSupport::Notifications.unsubscribe(subscription) }
+    end
+
+    private
+
+    attr_reader :reference
+
+    # @rbs (Symbol, ?authorization_context: untyped) -> void
+    def authorize!(action, authorization_context: nil)
+      allowed = SolidObjects.configuration.authorize_administration.call(
+        action:,
+        resource: "actor_diagnostics",
+        resource_id: [ reference.actor_type, reference.actor_id ].to_json,
+        authorization_context:
+      )
+      raise Unauthorized, "actor diagnostics are not authorized" unless allowed
+    end
+
+    # @rbs (Array[Time?], now: Time, limit: Integer) -> Hash[String, untyped]
+    def summarize(timestamps, now:, limit:)
+      oldest = timestamps.compact.min
+      {
+        "sampled" => [ timestamps.length, limit ].min,
+        "truncated" => timestamps.length > limit,
+        "oldestAgeMilliseconds" => oldest ? [ ((now - oldest) * 1000).round, 0 ].max : nil
+      }
+    end
+  end
+end

lib/solid_objects/effect_executor.rb

@@ -137,6 +137,7 @@ def claim_next
 
     # @rbs (Effect) -> untyped
     def deliver(effect)
+      SolidObjects.instrument(:"outbox.age", instance_id: effect.instance_id, actor_type: effect.instance.actor_type, actor_id: effect.instance.actor_id, message_id: effect.message_id, attempt: effect.attempt_count, age_milliseconds: [ ((database_adapter.database_now - effect.available_at) * 1000).round, 0 ].max)
       return deliver_actor_message(effect) if effect.name == ACTOR_MESSAGE_EFFECT
 
       handler = SolidObjects.effect_registry.fetch(effect.name)
@@ -208,6 +209,9 @@ def complete(effect, result)
       Mailbox.new.announce(result_message) if result_message
       SolidObjects.instrument(
         :"effect.completed",
+        instance_id: effect.instance_id,
+        actor_type: effect.instance.actor_type,
+        actor_id: effect.instance.actor_id,
         effect_id: effect.effect_id,
         effect_name: effect.name,
         message_id: effect.message_id,