fix: Align observability event coverage

Include broadcast outbox age and activation message correlation. Keep the\ncombined diagnostics category cap explicit and remove unnecessary broad\ntype annotations from telemetry delivery.

Author: cardmagic

docs/observability.md

@@ -94,6 +94,9 @@ 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 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,

src/actor-runtime.ts

@@ -37,7 +37,10 @@ export interface ActorRuntime {
   ): Promise<ActorDiagnostics>
   observe(
     reference: ActorReferenceCore<Actor>,
-    options: { onEvent: EventObserver; authorizationContext?: unknown },
+    options: {
+      onEvent: EventObserver
+      authorizationContext?: SnapshotOptions["authorizationContext"]
+    },
   ): Promise<() => void>
   ref<ActorType extends Actor>(
     actorClass: ActorClass<ActorType>,

src/cloudflare/engine.ts

@@ -857,6 +857,7 @@ export class ActorEngine {
     try {
       this.emit("outbox.age", {
         messageId: outbox.messageId,
+        outboxKind: outbox.kind,
         attempt: outbox.attempt + 1,
         ageMilliseconds: Math.max(0, Date.now() - outbox.availableAt),
       })

src/cloudflare/runtime.ts

@@ -82,7 +82,10 @@ export class CloudflareRuntime implements ActorRuntime {
 
   async observe(
     _reference: ActorReferenceCore<Actor>,
-    _options: { onEvent: EventObserver; authorizationContext?: unknown },
+    _options: {
+      onEvent: EventObserver
+      authorizationContext?: SnapshotOptions["authorizationContext"]
+    },
   ): Promise<() => void> {
     return unsupported(
       "process-local observers; configure instrumentation on the actor Durable Object host",
@@ -100,7 +103,7 @@ export class CloudflareRuntime implements ActorRuntime {
       authorizationContext: normalizeJson(options.authorizationContext ?? null),
       payload: { action: "diagnostics", limit: options.limit ?? 100 },
     })
-    return readonlyCopy(value) as unknown as ActorDiagnostics
+    return readonlyCopy(value) as ActorDiagnostics & JsonObject
   }
 
   async invoke<Result = JsonValue>(options: {

src/reference.ts

@@ -244,14 +244,17 @@ export class ActorReferenceCore<ActorType extends Actor> {
 
   observe(options: {
     onEvent: EventObserver
-    authorizationContext?: unknown
+    authorizationContext?: SnapshotOptions["authorizationContext"]
   }): Promise<() => void> {
     return this.runtime.observe(this, options)
   }
 
   on(
     name: string,
-    options: { onEvent: EventObserver; authorizationContext?: unknown },
+    options: {
+      onEvent: EventObserver
+      authorizationContext?: SnapshotOptions["authorizationContext"]
+    },
   ): Promise<() => void> {
     return this.observe({
       ...options,

src/runtime.ts

@@ -1310,9 +1310,7 @@ export class SolidObjectsRuntime {
         await withActorContext({ actor, runtime: this }, () => actor.activate())
         activated = true
         this.emitInstrumentation("activation.completed", {
-          actorType: turn.message.actor_type,
-          actorId: turn.message.actor_id,
-          instanceId: turn.instance.id,
+          ...messageInstrumentation(turn.message),
           generation: String(turn.activationGeneration),
         })
       } catch (error) {
@@ -1454,11 +1452,10 @@ export class SolidObjectsRuntime {
       })
       if (outcome === "retrying")
         this.emitInstrumentation("message.retry", messageInstrumentation(turn.message))
-      if (outcome === "dead") {
-        if (recoveryMessage(turn.message))
-          this.emitInstrumentation("recovery.failed", messageInstrumentation(turn.message))
+      if (outcome === "dead" && recoveryMessage(turn.message))
+        this.emitInstrumentation("recovery.failed", messageInstrumentation(turn.message))
+      if (outcome === "dead")
         this.emitInstrumentation("dead_letter.created", messageInstrumentation(turn.message))
-      }
       return { actor, retainActivation: activated, activated }
     }
   }
@@ -1574,7 +1571,10 @@ export class SolidObjectsRuntime {
 
   async observe(
     reference: ActorReferenceCore<Actor>,
-    options: { onEvent: EventObserver; authorizationContext?: unknown },
+    options: {
+      onEvent: EventObserver
+      authorizationContext?: SnapshotOptions["authorizationContext"]
+    },
   ): Promise<() => void> {
     await this.authorizeAdministration({
       action: "observe",
@@ -1613,6 +1613,7 @@ export class SolidObjectsRuntime {
     try {
       this.emitInstrumentation("outbox.age", {
         ...effectInstrumentation(effect),
+        outboxKind: "effect",
         ageMilliseconds: Math.max(0, Date.now() - Number(effect.available_at_ms)),
       })
       const handler = this.effects.get(effect.name)
@@ -1676,6 +1677,11 @@ export class SolidObjectsRuntime {
   async executeBroadcast(broadcast: BroadcastRow): Promise<void> {
     const deliver = this.settings.broadcast
     try {
+      this.emitInstrumentation("outbox.age", {
+        ...broadcastInstrumentation(broadcast),
+        outboxKind: "broadcast",
+        ageMilliseconds: Math.max(0, Date.now() - Number(broadcast.available_at_ms)),
+      })
       const event = readonlyCopy({
         actorType: broadcast.actor_type,
         actorId: broadcast.actor_id,
@@ -2527,6 +2533,7 @@ function effectInstrumentation(effect: EffectRow): JsonObject {
 
 function broadcastInstrumentation(broadcast: BroadcastRow): JsonObject {
   return {
+    instanceId: broadcast.instance_id,
     broadcastId: broadcast.id,
     messageId: broadcast.message_id,
     actorType: broadcast.actor_type,

src/telemetry.ts

@@ -64,6 +64,7 @@ const fields = new Set([
   "nextRunAt",
   "name",
   "commitAction",
+  "outboxKind",
   "truncated",
   "payload",
   "waitingOn",
@@ -126,9 +127,9 @@ export function deliverTelemetry(options: {
   event: InstrumentationEvent
   logger: Logger
 }): void {
-  const failed = (error: unknown) => {
+  const failed = <ErrorValue>(error: ErrorValue) => {
     try {
-      const result: unknown = options.logger.error({
+      const result = options.logger.error({
         event: "solid_objects.instrumentation.failed",
         instrumentationEvent: options.event.name,
         errorName: error instanceof Error ? error.name : "Error",

src/wake-up-notification.ts

@@ -8,7 +8,7 @@ export function notifyWakeUp(options: {
 }): void {
   const logFailure = (errorName: string): void => {
     try {
-      const result: unknown = options.logger.error({
+      const result = options.logger.error({
         event: "solid_objects.wake_up.failed",
         role: options.role,
         errorName,

test/outboxes.test.ts

@@ -1,6 +1,10 @@
 import { afterEach, describe, expect, it } from "vitest"
 import { Actor, broadcastInvalidation, broadcastValue } from "../src/actor.js"
-import type { BroadcastEvent, SolidObjectsConfiguration } from "../src/configuration.js"
+import type {
+  BroadcastEvent,
+  InstrumentationEvent,
+  SolidObjectsConfiguration,
+} from "../src/configuration.js"
 import { NonRetryableError } from "../src/errors.js"
 import { configure, type SolidObjectsRuntime } from "../src/runtime.js"
 import { sqlite } from "../src/database/sqlite.js"
@@ -414,7 +418,11 @@ describe("reminders", () => {
 describe("observable broadcasts", () => {
   it("delivers values and invalidation-only observable names", async () => {
     const events: BroadcastEvent[] = []
+    const telemetry: InstrumentationEvent[] = []
     runtime = configuredRuntime({
+      instrumentation: (event) => {
+        telemetry.push(event)
+      },
       broadcast: async (event) => {
         events.push(event)
       },
@@ -424,6 +432,19 @@ describe("observable broadcasts", () => {
     await ObservableCounter.ref("counter").increment()
     expect(await runtime.broadcastWorker().runUntilIdle()).toBe(1)
 
+    expect(telemetry).toEqual(
+      expect.arrayContaining([
+        expect.objectContaining({
+          name: "solid_objects.outbox.age",
+          actorId: "counter",
+          incarnation: expect.any(String),
+          attributes: expect.objectContaining({
+            outboxKind: "broadcast",
+            ageMilliseconds: expect.any(Number),
+          }),
+        }),
+      ]),
+    )
     expect(events).toHaveLength(1)
     expect(events[0]).toMatchObject({
       actorType: "ObservableCounter",