fix: Isolate outbox telemetry measurements

Keep optional database measurements from failing durable delivery. Cover broadcast age and report message duration only on terminal outcomes so Ruby and JavaScript emit matching samples.

Author: cardmagic

docs/observability.md

@@ -92,6 +92,9 @@ 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 limit applies to each combined category: one effect plus one broadcast with
+`limit: 1` returns `sampled: 1, truncated: true`, even when both source queries
+returned all their rows. The extra row proves that the category exceeds its cap.
 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,

lib/solid_objects/broadcast_executor.rb

@@ -41,6 +41,7 @@ def run_once
       broadcast = claim_next
       return false unless broadcast
 
+      Telemetry.outbox(broadcast)
       broadcast_adapter.call(broadcast)
       complete(broadcast)
       true

lib/solid_objects/effect_executor.rb

@@ -137,7 +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)
+      Telemetry.outbox(effect)
       return deliver_actor_message(effect) if effect.name == ACTOR_MESSAGE_EFFECT
 
       handler = SolidObjects.effect_registry.fetch(effect.name)

lib/solid_objects/executor.rb

@@ -161,7 +161,7 @@ def complete(result, observable_changes, state_after:, state_changed:)
       end
       SolidObjects.instrument_after_commit(:"recovery.completed", **instrumentation_payload) if recovery_message?
       report_large_state(state_after.byte_size)
-      SolidObjects.instrument_after_commit(:"message.completed", **instrumentation_payload, revision: message.sequence)
+      SolidObjects.instrument_after_commit(:"message.completed", **instrumentation_payload, revision: message.sequence, duration_milliseconds: elapsed_milliseconds)
       SolidObjects.wake_up.signal
     rescue CommittedTransactionError
       raise
@@ -424,6 +424,7 @@ def fail_message(error)
         :"message.failed",
         **instrumentation_payload,
         error_class: error.class.name,
+        duration_milliseconds: elapsed_milliseconds,
         dead:
       )
       SolidObjects.instrument_after_commit(:"recovery.failed", **instrumentation_payload) if dead && recovery_message?
@@ -463,7 +464,8 @@ def reject_message(rejection)
       SolidObjects.instrument_after_commit(
         :"message.rejected",
         **instrumentation_payload,
-        code: rejection.code
+        code: rejection.code,
+        duration_milliseconds: elapsed_milliseconds
       )
       SolidObjects.wake_up.signal
     end
@@ -530,6 +532,11 @@ def recovery_message?
       message.delivery_mode == "internal" && message.idempotency_key.to_s.start_with?("effect:") && message.idempotency_key.to_s.end_with?(":recovery")
     end
 
+    # @rbs () -> (Integer | Float)
+    def elapsed_milliseconds
+      ((::Process.clock_gettime(::Process::CLOCK_MONOTONIC) - @started_at) * 1000).round(3)
+    end
+
     # @rbs () -> Hash[Symbol, untyped]
     def instrumentation_payload
       {
@@ -539,8 +546,7 @@ def instrumentation_payload
         actor_id: message.actor_id,
         sequence: message.sequence,
         attempt: message.attempt_count,
-        request_id: message.request_id,
-        duration_milliseconds: ((::Process.clock_gettime(::Process::CLOCK_MONOTONIC) - @started_at) * 1000).round(3)
+        request_id: message.request_id
       }
     end
   end

lib/solid_objects/telemetry.rb

@@ -6,12 +6,31 @@ module Telemetry
       actorType actorId instanceId incarnation revision messageId requestId attempt sequence
       operation deliveryMode generation durationMilliseconds latenessMilliseconds ageMilliseconds
       depth count errorName outcome retryable status effectId effectName reminderId occurrence
-      truncated broadcastId code role reason processId processKind ownerId componentCount byteCount
+      outboxKind truncated broadcastId code role reason processId processKind ownerId componentCount byteCount
       thresholdBytes previousRunAt nextRunAt name commitAction payload
     ].freeze
     ALIASES = { "errorClass" => "errorName", "stateRevision" => "revision", "durationMs" => "durationMilliseconds" }.freeze
 
     class << self
+      # @rbs (Effect | Broadcast) -> void
+      def outbox(record)
+        return unless SolidObjects.configuration.instrumentation || ActiveSupport::Notifications.notifier.listening?("solid_objects.outbox.age")
+
+        instance = record.instance
+        SolidObjects.instrument(
+          :"outbox.age",
+          instance_id: record.instance_id,
+          actor_type: instance.actor_type,
+          actor_id: instance.actor_id,
+          message_id: record.message_id,
+          attempt: record.attempt_count,
+          outbox_kind: record.is_a?(Effect) ? "effect" : "broadcast",
+          age_milliseconds: [ ((SolidObjects.database_adapter.database_now - record.available_at) * 1000).round, 0 ].max
+        )
+      rescue
+        nil
+      end
+
       # @rbs (Symbol, Hash[Symbol, untyped]) -> void
       def emit(name, payload)
         observer = SolidObjects.configuration.instrumentation
@@ -27,7 +46,7 @@ def event(name, payload)
         attributes = safe_attributes(payload)
         adapter = DatabaseAdapter.family(Record.connection).to_s
         event_name = "solid_objects.#{name}"
-        labels = { "event" => event_name, "adapter" => adapter, "actorType" => attributes.fetch("actorType", "") }
+        labels = { "event" => event_name, "adapter" => adapter, "actorType" => attributes["actorType"].to_s }
         metrics = [ { "name" => "solid_objects.events", "kind" => "counter", "unit" => "1", "value" => 1, "labels" => labels } ]
         [
           [ "durationMilliseconds", "solid_objects.duration", "histogram", "ms" ],

sig/generated/lib/solid_objects/executor.rbs

@@ -108,6 +108,9 @@ module SolidObjects
     # @rbs () -> bool
     def recovery_message?: () -> bool
 
+    # @rbs () -> (Integer | Float)
+    def elapsed_milliseconds: () -> (Integer | Float)
+
     # @rbs () -> Hash[Symbol, untyped]
     def instrumentation_payload: () -> Hash[Symbol, untyped]
   end

sig/generated/lib/solid_objects/telemetry.rbs

@@ -6,6 +6,9 @@ module SolidObjects
 
     ALIASES: untyped
 
+    # @rbs (Effect | Broadcast) -> void
+    def self.outbox: (Effect | Broadcast) -> void
+
     # @rbs (Symbol, Hash[Symbol, untyped]) -> void
     def self.emit: (Symbol, Hash[Symbol, untyped]) -> void
 

test/integration/telemetry_test.rb

@@ -6,6 +6,7 @@ class TelemetryTest < ActiveSupport::TestCase
   class Counter < SolidObjects::Actor
     actor_type "telemetry-counter"
     attribute :count, default: 0
+    observable :count, broadcast: :value
 
     def arrange
       emit :telemetry_effect, secret: "private effect"
@@ -40,6 +41,8 @@ def increment
     assert_equal 1, event.fetch("attempt")
     assert event.fetch("incarnation")
     refute_includes event.fetch("metrics").to_json, "actorId"
+    started = events.find { |entry| entry.fetch("name") == "solid_objects.message.started" }
+    refute_includes started.fetch("metrics").map { |metric| metric.fetch("name") }, "solid_objects.duration"
   end
   test "diagnostics and observers require authorization and are bounded" do
     reference = Counter.ref("diagnostics")
@@ -118,4 +121,44 @@ def increment
   ensure
     worker&.stop
   end
+  test "outbox measurement failure cannot fail delivery" do
+    SolidObjects.configuration.instrumentation = ->(_) { true }
+    Counter.ref("measurement").arrange
+    calls = []
+    SolidObjects.register_effect(:telemetry_effect) {
+      calls << :delivered
+      "result"
+    }
+    executor = SolidObjects::EffectExecutor.new
+    executor.define_singleton_method(:claim_next) do
+      super().tap do |effect|
+        effect.define_singleton_method(:available_at) { raise "measurement failed" }
+      end
+    end
+    assert executor.run_once
+    assert_equal [ :delivered ], calls
+    assert_equal "completed", SolidObjects::Effect.first.status
+  ensure
+    executor&.stop
+  end
+
+  test "samples broadcast age and caps the combined outbox category" do
+    events = []
+    SolidObjects.configuration.instrumentation = ->(event) { events << event }
+    SolidObjects.configuration.authorize_administration = ->(**) { true }
+    SolidObjects.configuration.broadcast_adapter = ->(_) { true }
+    reference = Counter.ref("broadcast")
+    reference.increment
+    reference.arrange
+    summary = reference.diagnostics(limit: 1)
+    assert_equal 1, summary.fetch("outbox").fetch("sampled")
+    assert summary.fetch("outbox").fetch("truncated")
+    executor = SolidObjects::BroadcastExecutor.new
+    assert executor.run_once
+    event = events.find { |entry| entry.fetch("name") == "solid_objects.outbox.age" && entry.fetch("attributes")["outboxKind"] == "broadcast" }
+    assert event
+    assert_operator event.fetch("metrics").last.fetch("value"), :>=, 0
+  ensure
+    executor&.stop
+  end
 end