From 45e559bf3f8f410570d4593412b21b68cac94d91 Mon Sep 17 00:00:00 2001 From: mintaka Date: Fri, 28 Aug 2026 00:40:48 -0400 Subject: [PATCH] fix(compass-agent): isolate transport gauge tests from the shared metric registry (RIG-2656) The three transport LEVEL gauges (trace_queue_depth, priority_retry_depth, no_progress_depth) were asserted as absolute values read off Effect's process-global in-memory metric registry, which every ManagedRuntime and Effect.runSync shares. A gauge is last-writer-wins, and bun runs test files concurrently in one process, so a sibling test file constructing its own spine/control-source moved the shared gauge between a test's Metric.set and its synchronous read -- the assertion flaked ~2/3 under the concurrent suite while passing 100% in isolation. Counters were unaffected: they are read as a before/after delta, which concurrent movement cannot corrupt. Per-runtime registry isolation is impossible in Effect 3.x -- the registry is a hard process-global globalValue singleton with no FiberRef override. Instead the gauges are built through a namespace-prefix factory: production passes no prefix (the exact frozen metric names, byte-identical behavior), and each affected test passes a unique prefix so its gauge read hits a private registry entry immune to concurrent writers. The registry keys structurally on the metric name, so two same-name gauge instances collide to one entry and a unique-named gauge is private by construction. createPublishSpine takes an optional metricNamespace positional (internal seam, already effect-typed); createSocketControlSource takes it via its options bag as a plain string, never an effect type, so the re-exported factory's public .d.ts stays free of the effect package (export-surface guard). A new discriminating test reproduces the race deterministically: after the spine sets its private gauge, a hostile writer clobbers the shared key; the namespaced read is unaffected while the shared-key read returns the clobber. Collapsing the namespace to the shared key reddens exactly that test, proving the isolation is load-bearing. Every gauge assertion is mutation-proven non-vacuous. Full suite 635 pass/0 fail across repeated concurrent runs. Co-authored-by: Matt Wilkinson --- .../src/transport/control-metrics.test.ts | 26 ++++-- .../src/transport/control-source.ts | 16 +++- .../src/transport/otel-metrics.test.ts | 89 +++++++++++++++---- .../src/transport/otel-metrics.ts | 49 ++++++---- .../src/transport/publish-spine.ts | 16 +++- 5 files changed, 152 insertions(+), 44 deletions(-) diff --git a/packages/compass-agent/src/transport/control-metrics.test.ts b/packages/compass-agent/src/transport/control-metrics.test.ts index 726fe5fd6..61bf0181b 100644 --- a/packages/compass-agent/src/transport/control-metrics.test.ts +++ b/packages/compass-agent/src/transport/control-metrics.test.ts @@ -61,7 +61,7 @@ import { createUnixSocketTransport, type RunnerTransport } from "./index"; import { controlUnmapped, flapResets, - noProgressDepth, + noProgressDepthGauge, reconnects, } from "./otel-metrics"; import type { PublishSpine } from "./publish-spine"; @@ -314,12 +314,15 @@ test("no_progress_depth tracks the consecutive-no-progress level", async () => { // so `noProgress` climbs 1→2→3 and the gauge is set each drop; then a clean // close (no drop, no set) leaves the gauge at its peak. // - // The gauge is a process-global shared across tests, so seed it to a sentinel - // (99) this scenario cannot produce immediately before driving. That makes the - // site self-diagnostic: removing `Metric.set(noProgressDepth, noProgress)` - // leaves the gauge stuck at 99 (not 3) → red. (The reset-to-0 test below is the - // sibling guard for the same site on the progress path; the coupling is - // intentional, not accidental cross-test residue.) + // A UNIQUE metric namespace gives the gauge a private registry key, immune to + // the cross-file gauge race: a concurrent sibling test file can no + // longer move this absolute level between the source's set and the read. The + // sentinel (99) seeded on that SAME private key before driving keeps the site + // self-diagnostic: removing `Metric.set(noProgressDepth, noProgress)` leaves + // the gauge stuck at 99 (not 3) → red. (The reset-to-0 test below is the + // sibling guard for the same site on the progress path.) + const namespace = `${crypto.randomUUID()}.`; + const noProgressDepth = noProgressDepthGauge(namespace); const rec = emptyRecorder(); let t = 0; const socketPath = await serve(rec, { @@ -333,7 +336,7 @@ test("no_progress_depth tracks the consecutive-no-progress level", async () => { t += 6000; }), noopImmediate, - { onUnmapped: () => {}, now: () => t }, + { onUnmapped: () => {}, now: () => t, metricNamespace: namespace }, ); Effect.runSync(Metric.set(noProgressDepth, 99)); const outcome = await drive(source); @@ -348,6 +351,11 @@ test("a progress-making reconnect resets no_progress_depth to 0", async () => { // progress and zeroes it; a clean close leaves the gauge at that reset. The // two drops (reconnects delta 2) prove the source did climb-then-reset rather // than never leaving 0. Mirrors O2's priority_retry_depth reset test. + // Unique namespace: the source sets this private gauge to 1 then resets it to + // 0; both writes land on the private key, so the mutation check stays + // non-vacuous and the read is immune to the cross-file race. + const namespace = `${crypto.randomUUID()}.`; + const noProgressDepth = noProgressDepthGauge(namespace); const rec = emptyRecorder(); const gate = ackGate(); let t = 0; @@ -370,7 +378,7 @@ test("a progress-making reconnect resets no_progress_depth to 0", async () => { t += 6000; }), noopImmediate, - { onUnmapped: () => {}, now: () => t }, + { onUnmapped: () => {}, now: () => t, metricNamespace: namespace }, ); const outcome = await drive(source); expect(outcome.ended).toBe("cleanly"); diff --git a/packages/compass-agent/src/transport/control-source.ts b/packages/compass-agent/src/transport/control-source.ts index 8d832a59c..25c05d88d 100644 --- a/packages/compass-agent/src/transport/control-source.ts +++ b/packages/compass-agent/src/transport/control-source.ts @@ -73,7 +73,7 @@ import type { RunnerTransport } from "./index"; import { controlUnmapped, flapResets, - noProgressDepth, + noProgressDepthGauge, reconnects, } from "./otel-metrics"; import { getTransportRuntime } from "./runtime-channel"; @@ -251,6 +251,19 @@ export interface SocketControlSourceOptions { * floor. */ readonly now?: () => number; + /** + * Namespace prefix for the `no_progress_depth` LEVEL gauge this source sets. + * A plain string (never an `effect` type) so this exported options bag stays + * free of the `effect` package — `createSocketControlSource` is re-exported + * from the package entry, and the export-surface guard forbids an `effect` + * type on the public `.d.ts`. Defaults to "" — production yields the exact + * frozen metric name. A test passes a unique prefix so its gauge read hits a + * private registry entry, immune to the cross-file gauge race: the + * shared process-global registry keys structurally on the metric name, and a + * bare gauge would be moved by a concurrent sibling test file between this + * source's Metric.set and the test's synchronous read. + */ + readonly metricNamespace?: string; } const defaultOnUnmapped = (u: UnmappedEvent): void => @@ -284,6 +297,7 @@ export function createSocketControlSource( ): ControlSource { const onUnmapped = options.onUnmapped ?? defaultOnUnmapped; const now = options.now ?? (() => performance.now()); + const noProgressDepth = noProgressDepthGauge(options.metricNamespace ?? ""); const spine = transport.publishSpine(); const acks = new AckCursor(spine); const buffer = new AsyncBuffer(); diff --git a/packages/compass-agent/src/transport/otel-metrics.test.ts b/packages/compass-agent/src/transport/otel-metrics.test.ts index f1dba5e55..504a8f93d 100644 --- a/packages/compass-agent/src/transport/otel-metrics.test.ts +++ b/packages/compass-agent/src/transport/otel-metrics.test.ts @@ -33,10 +33,10 @@ import { durableGiveUps, priorityBatchRetries, priorityFramesLost, - priorityRetryDepth, + priorityRetryDepthGauge, traceFramesLostFailedBatch, traceFramesLostOverflow, - traceQueueDepth, + traceQueueDepthGauge, } from "./otel-metrics"; import { createPublishSpine, @@ -140,13 +140,21 @@ test("trace_queue_depth samples the backlog at the take, before draining it", as // A resolving publish so the pump drains and drain() joins it. A backlog that // fits one batch (< PUBLISH_BATCH_MAX) is fully queued before the deferred // first takeBatch runs, so the gauge — sampled BEFORE the take — reads the - // whole backlog, not the post-drain residual. Read synchronously right after - // drain() so no other fiber moves the shared gauge between set and read. + // whole backlog, not the post-drain residual. A UNIQUE metric namespace gives + // this gauge a private registry key, so a concurrent sibling test file writing + // the shared key cannot move the absolute value between set and read (the + // cross-file gauge race). The race-reproduction test below proves this + // isolation is load-bearing, not decorative. + const namespace = `${crypto.randomUUID()}.`; const backlog = 100; - const spine = createPublishSpine(() => Promise.resolve(undefined)); + const spine = createPublishSpine( + () => Promise.resolve(undefined), + undefined, + namespace, + ); for (let i = 0; i < backlog; i++) spine.enqueueTrace(traceFrame()); await spine.drain(); - expect(gaugeValue(traceQueueDepth)).toBe(backlog); + expect(gaugeValue(traceQueueDepthGauge(namespace))).toBe(backlog); // Mutation check: removing `Metric.set(traceQueueDepth, traceSize())` from // takeBatch leaves the gauge at its prior value, not `backlog` → reddens. }); @@ -156,8 +164,14 @@ test("a priority give-up increments priority_frames_lost, additive to failedPrio // ladder is exhausted the frame is a definitive never-drop loss. const before = counterCount(priorityFramesLost); const retriesBefore = counterCount(priorityBatchRetries); - const spine = createPublishSpine(() => - Promise.reject(new Error("dead socket")), + // A unique namespace isolates the absolute retry-depth read from the shared + // registry key: only the gauge is namespaced; the counters stay on + // the shared key and are read as deltas, which are race-safe. + const namespace = `${crypto.randomUUID()}.`; + const spine = createPublishSpine( + () => Promise.reject(new Error("dead socket")), + undefined, + namespace, ); spine.enqueuePriority(traceFrame()); await spine.drain(); @@ -172,30 +186,73 @@ test("a priority give-up increments priority_frames_lost, additive to failedPrio expect(counterCount(priorityBatchRetries) - retriesBefore).toBe( PRIORITY_BATCH_RETRY_MS.length, ); - expect(gaugeValue(priorityRetryDepth)).toBe(PRIORITY_BATCH_RETRY_MS.length); + expect(gaugeValue(priorityRetryDepthGauge(namespace))).toBe( + PRIORITY_BATCH_RETRY_MS.length, + ); }); test("a delivered priority batch resets priority_retry_depth to 0 after its retries", async () => { // Fail once, then deliver: one bounded retry, then a successful send resets the // pump-scoped depth level. Distinct from the give-up path — no frame is lost. const retriesBefore = counterCount(priorityBatchRetries); + // Unique namespace: the retry path first sets the gauge to 1, then the + // successful send resets it to 0 — both writes land on this private key, so the + // mutation check (drop the reset → gauge stuck at 1) stays non-vacuous while + // the read is immune to the cross-file race. + const namespace = `${crypto.randomUUID()}.`; let attempt = 0; - const spine = createPublishSpine(() => { - attempt++; - return attempt === 1 - ? Promise.reject(new Error("transient blip")) - : Promise.resolve(undefined); - }); + const spine = createPublishSpine( + () => { + attempt++; + return attempt === 1 + ? Promise.reject(new Error("transient blip")) + : Promise.resolve(undefined); + }, + undefined, + namespace, + ); spine.enqueuePriority(traceFrame()); await spine.drain(); expect(counterCount(priorityBatchRetries) - retriesBefore).toBe(1); // Reset to 0 on the successful send; read synchronously after drain. - expect(gaugeValue(priorityRetryDepth)).toBe(0); + expect(gaugeValue(priorityRetryDepthGauge(namespace))).toBe(0); expect(spine.failedPriorityCount()).toBe(0); // Mutation check: removing `Metric.set(priorityRetryDepth, 0)` on the success // arm leaves the gauge at 1 → the depth assertion reddens. }); +test("a namespaced gauge read survives a concurrent writer clobbering the shared key", async () => { + // The root cause of the gauge flake, reproduced deterministically. Under the concurrent + // full suite a sibling test file constructs its own spine and writes the SAME + // shared gauge key between this test's Metric.set and its synchronous read; a + // gauge is an absolute last-writer-wins level, so the read saw the wrong value. + // Here that hostile concurrent writer is made explicit: after the spine sets + // its depth gauge, we clobber the SHARED (un-namespaced) key with a wrong + // value, then read back. The namespaced read is unaffected — it hits a private + // registry entry — while a read of the shared key would return the clobbered + // value. This is the discriminating assertion the fix turns green: point the + // spine at the empty namespace (the pre-fix shared key) and the two reads + // collapse onto the same clobbered entry, reddening the inequality. + const namespace = `${crypto.randomUUID()}.`; + const spine = createPublishSpine( + () => Promise.reject(new Error("dead socket")), + undefined, + namespace, + ); + spine.enqueuePriority(traceFrame()); + await spine.drain(); + // The spine set its private gauge to the exhausted ladder length. + const isolated = gaugeValue(priorityRetryDepthGauge(namespace)); + expect(isolated).toBe(PRIORITY_BATCH_RETRY_MS.length); + // A hostile concurrent writer clobbers the SHARED key with a value the spine + // never wrote (the exact interleaving a sibling test file causes in CI). + const clobbered = PRIORITY_BATCH_RETRY_MS.length + 999; + Effect.runSync(Metric.set(priorityRetryDepthGauge(""), clobbered)); + // The namespaced read is immune; a shared-key read now returns the clobber. + expect(gaugeValue(priorityRetryDepthGauge(namespace))).toBe(isolated); + expect(gaugeValue(priorityRetryDepthGauge(""))).toBe(clobbered); +}); + test("a durable send counts one attempt per try and one give-up when the retry budget is exhausted", async () => { // onDurable always throws → the send exhausts DURABLE_RETRY_BACKOFF_MS and // gives up. attempts = BACKOFF.length + 1 (initial + one per delay); give-ups diff --git a/packages/compass-agent/src/transport/otel-metrics.ts b/packages/compass-agent/src/transport/otel-metrics.ts index 5141dea85..2966dc9f4 100644 --- a/packages/compass-agent/src/transport/otel-metrics.ts +++ b/packages/compass-agent/src/transport/otel-metrics.ts @@ -48,16 +48,29 @@ export const priorityBatchRetries = Metric.counter( { incremental: true }, ); -// The pump-scoped consecutive retry budget as a LEVEL: set to the new retry -// depth on each retry, reset to 0 on a successful send. -export const priorityRetryDepth = Metric.gauge( - "compass_agent.transport.publish.priority_retry_depth", -); +// The two publish-spine LEVEL gauges, built through a namespace-prefix factory. +// A gauge is an absolute last-writer-wins level, so a test reading one back must +// not share its registry key with a concurrent writer in a sibling test file: +// the shared process-global registry keys structurally on the metric NAME, and +// bun runs test files concurrently in one process. The builders take an optional +// namespace prefix — production passes none, yielding the exact frozen name; a +// test passes a unique prefix, yielding a private registry entry immune to the +// cross-file gauge race. Counters are unaffected: they are read as a +// before/after DELTA, which concurrent movement cannot corrupt. +export const priorityRetryDepthGauge = ( + namespace = "", +): Metric.Metric.Gauge => + // The pump-scoped consecutive retry budget as a LEVEL: set to the new retry + // depth on each retry, reset to 0 on a successful send. + Metric.gauge( + `${namespace}compass_agent.transport.publish.priority_retry_depth`, + ); -// Trace queue depth, sampled at each batch take. -export const traceQueueDepth = Metric.gauge( - "compass_agent.transport.publish.trace_queue_depth", -); +export const traceQueueDepthGauge = ( + namespace = "", +): Metric.Metric.Gauge => + // Trace queue depth, sampled at each batch take. + Metric.gauge(`${namespace}compass_agent.transport.publish.trace_queue_depth`); // Every durable send attempt (initial + each retry) on the frame sink. export const durableAttempts = Metric.counter( @@ -85,13 +98,17 @@ export const reconnects = Metric.counter( { incremental: true }, ); -// The consecutive-no-progress level as a LEVEL: set to `noProgress` after each -// drop's progress check (against CONTROL_RECONNECT_NO_PROGRESS_MAX), reset to 0 -// when a reconnect makes progress — a level, not a count, exactly like the -// publish spine's priority_retry_depth gauge above. -export const noProgressDepth = Metric.gauge( - "compass_agent.transport.control.no_progress_depth", -); +// The consecutive-no-progress LEVEL gauge, built through the same namespace +// factory as the publish-spine gauges above and for the same reason (the +// cross-file gauge race). Set to `noProgress` after each drop's +// progress check (against CONTROL_RECONNECT_NO_PROGRESS_MAX), reset to 0 when a +// reconnect makes progress — a level, not a count, exactly like the publish +// spine's priority_retry_depth gauge. Production passes no namespace (frozen +// name); a test passes a unique prefix for a private registry entry. +export const noProgressDepthGauge = ( + namespace = "", +): Metric.Metric.Gauge => + Metric.gauge(`${namespace}compass_agent.transport.control.no_progress_depth`); // Every min-uptime flap reset of the backoff ladder — the reset-on-open // flap-detector zeroing `attempt` after a past-floor connection dropped diff --git a/packages/compass-agent/src/transport/publish-spine.ts b/packages/compass-agent/src/transport/publish-spine.ts index 11f3a53aa..32ede6c04 100644 --- a/packages/compass-agent/src/transport/publish-spine.ts +++ b/packages/compass-agent/src/transport/publish-spine.ts @@ -49,10 +49,10 @@ import type { PublishFrameRequest } from "../gen/compass/v1/agent_gateway_pb"; import { priorityBatchRetries, priorityFramesLost, - priorityRetryDepth, + priorityRetryDepthGauge, traceFramesLostFailedBatch, traceFramesLostOverflow, - traceQueueDepth, + traceQueueDepthGauge, } from "./otel-metrics"; import type { TransportRuntime } from "./runtime-channel"; @@ -112,10 +112,22 @@ export interface PublishSpine { // publish driver), the spine falls back to its OWN default runtime and disposes // it at the end of drain(). A borrowed runtime is NEVER disposed here — the // transport's close() owns that. +// +// A `metricNamespace` prefixes the two LEVEL gauges this spine sets +// (trace_queue_depth, priority_retry_depth). It defaults to "" — production +// yields the exact frozen metric names. A test passes a unique prefix so its +// gauge reads hit a private registry entry, immune to the cross-file gauge race +// the shared process-global registry keys structurally on the metric +// name, so a bare gauge would be moved by a concurrent sibling test file between +// this spine's Metric.set and the test's synchronous read. Counters take no +// namespace — they are read as a before/after delta, robust to that movement. export function createPublishSpine( publish: (stream: AsyncIterable) => Promise, borrowedRuntime?: TransportRuntime, + metricNamespace = "", ): PublishSpine { + const traceQueueDepth = traceQueueDepthGauge(metricNamespace); + const priorityRetryDepth = priorityRetryDepthGauge(metricNamespace); // Effect is confined module-private behind the spine: it backs the sliding // trace queue, the wake latch, and the forked pump fiber. The default logger // is removed on the fallback runtime so a handled pump-send failure does not