Instrument pending external-event queue health - #2257
Conversation
Add an in-memory metrics collector for the pending external-event delivery queue, plus a delivery-terminal hook on ExternalEventStore so every delivered and terminal-failed transition is counted once from a single observation point. SpaceRuntime records enqueues (by source + target run/node state), flush attempts, claim conflicts, stale-session skips, and structured logs at the previously-silent skip seams, and exposes getQueueHealthSnapshot() merging cumulative counters with live gauges (depth, event-age p95, in-flight, digest backlog, persisted pending). Surface it via the space.externalEvents.queueHealth RPC. Service wires the shared metrics to the store hook.
Add a QueueHealthSummary panel to the external-events settings that fetches the daemon-wide queue-health snapshot via the new space.externalEvents.queueHealth RPC and renders cumulative counters (enqueue by source/target state, flush, delivered, failures by reason/category, skips) and live gauges (depth, age p95, in-flight, digest backlog). Add typed store helper + QueueHealthSnapshot types.
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 608e9dd745
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
lsm
left a comment
lsm
left a comment
There was a problem hiding this comment.
🤖 Review by glm-5.1 (GLM/Zhipu)
Model: glm-5.1 | Client: NeoKai | Provider: GLM (Zhipu)
Recommendation: REQUEST_CHANGES
Overall this is a well-built, well-tested change. The delivery-terminal hook on ExternalEventStore (counting terminal outcomes from a single changes > 0 point), the shared-metrics wiring between service/runtime/store, and the snapshot → RPC → UI plumbing all hang together correctly, and the unit + integration tests exercise real transitions (including the no-double-count and cap-eviction paths). The findings below are refinements to metric correctness — which is exactly the thing this task exists to make trustworthy. Detailed suggestions are in the anchored line comments.
P2 — Paused-space digest requeue isn't counted as a skip (metric-contract violation). staleSessionSkips' own docstring promises it counts deferrals where the target session is "no longer live (or its space was paused)," but space-runtime.ts only records the skip for sessionLoss. A digest firing while the space is paused requeues every item silently — so the panel reports zero stale-session skips in precisely the pause-induced backlog scenario the metric should surface.
P2 — Snapshot materializes every pending row + full-age sort on each refresh. listPendingDeliveries().length constructs every delivery record solely for .length, and getPendingDeliveryAges() pulls every age into JS for an in-memory p95 sort. There's no global pending cap, so under the pathological backlog this view exists to diagnose, the synchronous RPC can block the event loop and delay real delivery. Use SQL COUNT/MIN/MAX/AVG and a bounded/approximate p95.
P2 — Enqueue counter undercounts via preservePendingDigestItem. queueForPendingNode is instrumented, but preservePendingDigestItem (called from stop() to rehouse rate-limited digest items into the pending queue) pushes genuinely-new items without recordEnqueue. Those items were never counted (they bypassed queueForPendingNode into the digest buffer), so on shutdown with a pending digest the cumulative enqueue counter ends up lower than reality and disagrees with the live queueDepth gauge. Add a recordEnqueue after the dedup guard.
P3 — Failure-category mapping has gaps + one unreachable mapping. Several real terminal reasons reach the hook but land in other: auto_pr_subscription_cleared, node_execution_cancelled, run_interests_rebuilt, run_terminal_cleanup. Separately, blocked_run_gate_not_opened is categorized but is persisted via markEventFailed (event-level), which never fires the delivery-terminal hook — so that mapping is unreachable, and the unit test asserting it exercises a path that can't occur in production. Consider routing these to deliverability and dropping the dead mapping.
What's solid: single-source-of-truth hook with changes > 0 guard prevents double counting across retry/exhaustion/cleanup paths; the listPendingDeliveries(workflowRunId?) overload is used correctly for the daemon-wide snapshot; RPC registration follows the existing space.externalEvents.* pattern (no new auth surface); the integration test mirrors the service wiring and covers zero-state, enqueue attribution, cap eviction, and delivered paths.
Address PR #2257 review feedback: - Compute persisted-pending health via SQL aggregates (summarizePendingDeliveries: COUNT/MIN/MAX/AVG + LIMIT/OFFSET p95) instead of materializing every pending row + a full in-memory sort on each snapshot read. - Count paused-space deferrals as a separate pausedSpaceSkips counter (recorded at the digest-requeue, fresh-delivery, and retry-injection pause sites), so staleSessionSkips stays a true session-loss signal and the pause-induced backlog scenario is no longer reported as zero skips. - Instrument preservePendingDigestItem with recordEnqueue so rehoused digest items keep the cumulative enqueue counter consistent with queueDepth. - Fix failure-category mapping: drop unreachable blocked_run_gate_not_opened (event-level markEventFailed never fires the delivery hook) and route auto_pr_subscription_cleared / node_execution_cancelled / run_interests_rebuilt / run_terminal_cleanup to deliverability.
lsm
left a comment
lsm
left a comment
There was a problem hiding this comment.
🤖 Review by glm-5.1 (GLM/Zhipu)
Model: glm-5.1 | Client: NeoKai | Provider: GLM (Zhipu)
Recommendation: APPROVE
Re-reviewed commit 521ae4a06 from scratch. All four REQUEST_CHANGES findings are correctly resolved, with no regressions and no new issues:
- Materialization (P2 ✓):
summarizePendingDeliveriescomputes COUNT/MIN/MAX/AVG via SQL aggregates and p95 via a singleLIMIT 1 OFFSET klookup — no per-row JS allocation or in-memory sort on refresh. The two queries are synchronous (no event-loop yield between them), so count and p95 stay consistent. - Paused-space skip (P2 ✓): New
pausedSpaceSkipscounter recorded at all three pause-deferral sites (digest-requeue, fresh-delivery, retry-injection);staleSessionSkipsis now a pure session-loss signal with a fixed docstring. Wired into the web types and the operator panel. preservePendingDigestItemundercount (P2 ✓):recordEnqueueadded after the dedup guard, so rehoused digest items keep the cumulative counter consistent with the livequeueDepthgauge.- Failure-category mapping (P3 ✓): Unreachable
blocked_run_gate_not_openeddropped (annotated why);auto_pr_subscription_cleared/node_execution_cancelled/run_interests_rebuilt/run_terminal_cleanuprouted todeliverability.
Verified empirically: typecheck, knip, lint all exit 0; external-event-store (70), queue-health-metrics (22), space-runtime-external-events (176), and web QueueHealthSummary + Settings (48) tests pass. All 6 review threads resolved; PR open and mergeable.
Nice work — the metric semantics are now internally consistent (counters vs gauges agree, the skip signals match their contracts, and the failure taxonomy covers every real terminal reason).
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 521ae4a068
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| if (this.externalEventDeliveriesInFlight.has(deliveryKey)) { | ||
| this.queueHealthMetrics.recordClaimConflict(); | ||
| log.debug('SpaceRuntime: external event delivery already in flight; skipped dispatch', { |
There was a problem hiding this comment.
When a replay arrives while deliverToLongHorizonAgent is already in flight, the preceding long-horizon loop at lines 2284–2289 checks the same in-flight set but silently continues; only workflow targets reach this new counter call. Because the snapshot is explicitly daemon-wide and claimConflicts is target-agnostic, long-horizon delivery races are omitted from the metric. Record the conflict in the long-horizon dispatch and retry paths as well.
Useful? React with 👍 / 👎.
| const reason = event.reason ?? 'unknown'; | ||
| this.finalFailuresByReason.set(reason, (this.finalFailuresByReason.get(reason) ?? 0) + 1); |
There was a problem hiding this comment.
When delivery errors contain per-session data—for example, stale injection throws Sub-session not found: ${subSessionId}, which is persisted as a deliveryMode:... reason and becomes terminal after retry exhaustion—each worker session creates a distinct key here. This map lives for the process lifetime and every snapshot serializes all entries, so repeated stale-session failures cause daemon memory and the queue-health RPC/UI payload to grow without bound. Normalize dynamic reasons into stable labels or cap the retained raw-reason entries.
Useful? React with 👍 / 👎.
| const targetRun = this.config.workflowRunRepo.getRun(resolved.workflowRunId); | ||
| if (targetRun && this.pausedSpaceIds.has(targetRun.spaceId)) { | ||
| this.queueHealthMetrics.recordPausedSpaceSkip(); |
There was a problem hiding this comment.
When a workflow target retains a sessionId but isTargetSessionLive is false, the adjacent else if (resolved.sessionId) path attempts a deferred flush and, if that stale session cannot be rehydrated, requeues the delivery after injection fails without incrementing staleSessionSkips. Only the digest session-loss path currently records that counter, so ordinary single-event deferrals caused by crashed or superseded workers remain invisible in the queue-health panel. Record the stale-session skip when this non-live-session dispatch is actually deferred.
Useful? React with 👍 / 👎.
1 participant
Adds metrics, structured logs, and an operator/debug view for the pending external-event delivery queue.
ExternalEventQueueMetricsin-memory collector (enqueue by source + target run/node state, flush attempts, claim conflicts, stale-session skips, delivered, terminal failures by reason) + failure-category breakdown.ExternalEventStoregets a delivery-terminal hook (fires once per real terminal transition viachanges > 0) so delivered/failed outcomes are counted from a single point, plusgetPendingDeliveryAgesfor the age gauge.SpaceRuntimerecords enqueues/flushes/skips at the queue seams, emitslog.debugat the previously-silent skip points (claim conflict, stale session, deliverability guards), and exposesgetQueueHealthSnapshot()merging counters with live gauges (depth, event-age p95, in-flight, digest backlog, persisted pending).space.externalEvents.queueHealthRPC;SpaceRuntimeServicewires the shared metrics to the store hook.QueueHealthSummarypanel (under Space → Configure → Settings → Tools → external events) renders the daemon-wide snapshot.Counters are process-lifetime (reset on restart); gauges are computed at read time. Covered by unit tests for the metrics class, the store hook/age query, and a
SpaceRuntimesnapshot integration test, plus a web component test.