From 5534dd0296dc5a3ecfa5da2e33a9f3e4cbc85f91 Mon Sep 17 00:00:00 2001 From: Jurij Skornik Date: Mon, 3 Aug 2026 00:39:10 +0200 Subject: [PATCH 01/16] =?UTF-8?q?feat(sync):=20W1=20measurement=20contract?= =?UTF-8?q?=20=E2=80=94=20source-attributed=20sync=20instruments=20(I1?= =?UTF-8?q?=E2=80=93I9)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Attribute sync cost to its trigger. Today no exported metric can answer "which lane is consuming the store": `operation` (which encodes the admission source) never reaches an instrument, per-operation bytes are folded into an in-process sum that is never exported, the changelog lane contributes zero bytes, and `outcome` is inert for `sync-global`. #2003 solved source-attributed pressure on the instantaneous snapshot only. Adds nine instruments across both request lanes, attributed to the closed `SYNC_ADMISSION_SOURCES` vocabulary, plus generator-emitted query artifacts validated by a real PromQL parser. Also fixes a real shutdown defect found during implementation. Shipped code calls `stopTelemetry()` before `server.close()` and `agent.stop()`, so the providers are torn down and `rebuildMetrics()` rebinds `getMetrics()` to a no-op meter while parent-side sync is still running — terminal catch-up records export while the attempts, bytes and active time belonging to them are silently dropped. Terminating the catch-up worker does not quiesce that work: `handleInvoke` awaits agent methods with no signal and no cancel hook. Both A24 ordering tests fail against the shipped sequence, so this is a regression proof, not a synthetic mutant. Key decisions: - `source` reaches the record sites via AsyncLocalStorage, not a threaded parameter. Threading would have put `source` in the same scope as `syncPageFetchCoalescingKey`, so the mutant guarding the highest-severity constraint would have guarded a hazard the implementation created. Ambient context makes it structural, and the changelog lane's `runResync` fallback inherits the correct label instead of reporting `unspecified`. Follows the existing `chain/src/rpc-usage.ts` idiom. - `runSyncSingleFlight` takes an explicit source at the three generic scopes: they coalesce above the admission boundary, so an ambient read there would label every generic join `unspecified` and stop I6's cross-family check firing. - No `timeout` in the attempt-outcome vocabulary: the router and pool emit at least seven incompatible deadline shapes and the only classifiers are `.message.includes(...)`. Any pre-response rejection that is not caller cancellation is `transport_error`. - Validation rejection is marked with a non-enumerable tag, never a replacement error — `makeLegacySyncBusyError`'s message is matched by `isSyncBackoffWorthyError`, so replacing it would silently change backoff and `failedPhases` accounting. - The terminal flush is bounded per leg. `MetricReader.forceFlush()` applies no timeout without `timeoutMillis` and leaves the trailing exporter flush unwrapped; `BasicTracerProvider.forceFlush()` takes no arguments and must be raced. Per-leg rather than an outer race, so each bound stays independently observable. - Both subscribe mint sites return 503 + Retry-After before any id is generated during shutdown; dedupe and replay are unaffected. Verification: agent packet 9 files/185 tests, CLI 6/6 files/149 tests, node-ui 19/19, packet reachability gate 16/16, four observability commands green on Windows and in the Linux CI shape, promtool SUCCESS 66 rules via a pinned multi-arch digest. Instrumentation overhead 0.005-0.006 ms/page against a 1 ms ceiling. A17 measured against a pristine pre-W1 referent: 154 + 31 = 185 exactly. Refs #2018, #2006 Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01SaXdkmteqStxGBpXcUPGd1 --- .github/workflows/observability-artifacts.yml | 45 + .../agent/scripts/bench-sync-telemetry.mjs | 451 ++++++++++ packages/agent/src/dkg-agent-lifecycle.ts | 222 ++++- packages/agent/src/dkg-agent-swm-host.ts | 6 + packages/agent/src/index.ts | 2 + packages/agent/src/p2p/sync-transport.ts | 119 ++- packages/agent/src/sync/attempt-telemetry.ts | 359 ++++++++ packages/agent/src/sync/catchup-policy.ts | 52 +- packages/agent/src/sync/error-tags.ts | 24 +- .../agent/src/sync/requester/page-fetch.ts | 7 + packages/agent/test/_helpers/w1-metrics.ts | 131 +++ packages/agent/test/core-fills-gap.test.ts | 7 + .../agent/test/sync-attempt-telemetry.test.ts | 310 +++++++ .../test/sync-operation-telemetry.test.ts | 503 +++++++++++ packages/agent/vitest.unit.config.ts | 3 + packages/cli/src/daemon/catchup-telemetry.ts | 235 +++++ packages/cli/src/daemon/lifecycle.ts | 71 +- .../cli/src/daemon/routes/context-graph.ts | 104 ++- packages/cli/src/daemon/state.ts | 21 + packages/cli/src/daemon/teardown.ts | 148 +++ .../daemon-catchup-telemetry-shutdown.test.ts | 844 ++++++++++++++++++ packages/cli/vitest.unit.config.ts | 4 + packages/core/src/telemetry-api.ts | 95 ++ packages/node-ui/src/index.ts | 3 + packages/node-ui/src/telemetry.ts | 108 ++- packages/node-ui/test/telemetry.test.ts | 337 ++++++- scripts/verify-w1-packet.mjs | 163 ++++ .../observability/generate-observability.mjs | 35 +- tools/observability/lib/w1.mjs | 487 ++++++++++ tools/observability/verify-check-mode.mjs | 118 +++ tools/observability/verify-w1-render.mjs | 325 +++++++ tools/observability/w1/w1-queries.md | 667 ++++++++++++++ tools/observability/w1/w1-rules.yaml | 211 +++++ 33 files changed, 6136 insertions(+), 81 deletions(-) create mode 100644 packages/agent/scripts/bench-sync-telemetry.mjs create mode 100644 packages/agent/src/sync/attempt-telemetry.ts create mode 100644 packages/agent/test/_helpers/w1-metrics.ts create mode 100644 packages/agent/test/sync-attempt-telemetry.test.ts create mode 100644 packages/agent/test/sync-operation-telemetry.test.ts create mode 100644 packages/cli/src/daemon/catchup-telemetry.ts create mode 100644 packages/cli/src/daemon/teardown.ts create mode 100644 packages/cli/test/daemon-catchup-telemetry-shutdown.test.ts create mode 100644 scripts/verify-w1-packet.mjs create mode 100644 tools/observability/lib/w1.mjs create mode 100644 tools/observability/verify-check-mode.mjs create mode 100644 tools/observability/verify-w1-render.mjs create mode 100644 tools/observability/w1/w1-queries.md create mode 100644 tools/observability/w1/w1-rules.yaml diff --git a/.github/workflows/observability-artifacts.yml b/.github/workflows/observability-artifacts.yml index 32a73f8550..c4831a5a3d 100644 --- a/.github/workflows/observability-artifacts.yml +++ b/.github/workflows/observability-artifacts.yml @@ -9,6 +9,16 @@ name: Observability artifacts # smoke-tests the operator-facing render mode (concrete datasource UIDs + # node-label profile), so a regression there can't hide behind the # committed-defaults check. +# +# The W1 sync-measurement artifacts (tools/observability/w1/) get two further +# steps, because "generated" is not "correct" and neither one is "valid +# PromQL": verify-w1-render.mjs asserts the semantic contract (instrument +# inventory, dual native/translated metric spellings, the source-family +# mapping, the all-source denominators, a node filter on every selector, both +# observation windows, and report/fixture expression identity), and a PINNED +# promtool container really PARSES the emitted rule fixture. Without those the +# path filter below would trigger on a W1 edit while validating none of its +# contents. on: pull_request: paths: @@ -42,6 +52,17 @@ jobs: - name: Verify committed artifacts match the generator run: node tools/observability/generate-observability.mjs --check + # The step above normalizes CRLF/LF before comparing, so a Windows + # checkout (core.autocrlf=true rewrites every artifact to CRLF — the + # metrics dashboard alone holds ~660 pairs) is not permanently red. But a + # comparison that answered "equal" to EVERYTHING would also pass that + # step, forever. This one builds its own CRLF and LF artifact trees plus + # four corrupted ones and runs the real command against each, pinning + # both directions — which also exercises the CRLF path here on Linux, + # where git only ever produces LF. + - name: Verify check mode is line-ending agnostic but still catches drift + run: node tools/observability/verify-check-mode.mjs + # verify-profile-render.mjs PARSES the rendered JSON and asserts the # whole profile-sensitive surface (every metrics target, the $node # variable, per-node alert exprs + summaries, notification group-bys, @@ -57,3 +78,27 @@ jobs: --vm-uid test-vm-uid --loki-uid test-loki-uid --prom-node-label service_instance_id node tools/observability/verify-profile-render.mjs /tmp/render \ --prom-node-label service_instance_id --vm-uid test-vm-uid --loki-uid test-loki-uid + + # W1 decision queries: the semantic contract behind the numbers, checked + # per selector rather than by sampling. Run against BOTH renders so the + # W1 queries are proven to follow --prom-node-label like every other + # profile-sensitive surface (the /tmp/render tree is produced by the + # step above). + - name: Verify the W1 sync-measurement artifacts + run: | + set -euo pipefail + node tools/observability/verify-w1-render.mjs tools/observability + node tools/observability/verify-w1-render.mjs /tmp/render --prom-node-label service_instance_id + + # Real PromQL validation. `promtool` is not installed on the developer + # host, so the container IS the reproducible path on both Windows and + # Linux; the tag is readability, the DIGEST is the reproducibility + # contract. `--entrypoint promtool` is REQUIRED: the image entrypoint is + # ["/bin/prometheus"], so the un-overridden form fails with + # `prometheus: error: unexpected promtool`. + - name: Parse the W1 rule fixture (pinned promtool) + run: | + set -euo pipefail + docker run --rm -v "${PWD}/tools/observability:/w" --entrypoint promtool \ + prom/prometheus@sha256:6559acbd5d770b15bb3c954629ce190ac3cbbdb2b7f1c30f0385c4e05104e218 \ + check rules /w/w1/w1-rules.yaml diff --git a/packages/agent/scripts/bench-sync-telemetry.mjs b/packages/agent/scripts/bench-sync-telemetry.mjs new file mode 100644 index 0000000000..4f0db4cf89 --- /dev/null +++ b/packages/agent/scripts/bench-sync-telemetry.mjs @@ -0,0 +1,451 @@ +#!/usr/bin/env node +// SPDX-License-Identifier: Apache-2.0 +/** + * W1 §8.5 — hot-path overhead of the sync telemetry record sites (A18). + * + * node packages/agent/scripts/bench-sync-telemetry.mjs --pages 200 --warmup 50 --json + * + * ## Why this file exists at all + * + * v4 of the plan gave a budget with no script. An implementer could then time a + * mock that never reaches a record site and still report that the budget passed. + * So the one property this benchmark must never lose is that it drives the REAL + * helpers. There is deliberately **no stub fallback**: if the real module cannot + * be loaded, this exits non-zero rather than measuring something else and + * reporting a number that looks like an answer. + * + * ## What is actually measured (do not overclaim from these numbers) + * + * Per page — the real per-attempt path, exactly as the send bracket calls it: + * `syncAttemptAttributes()` → `recordSyncAttemptRequestBytes()` (I2) + * → `recordSyncAttempt()` (I1) → `recordSyncAttemptResponseBytes()` (I3) + * + * Per operation — the real I4 boundary instrumentation: + * `withSyncAdmissionSource()` (the AsyncLocalStorage scope that makes `source` + * ambient) wrapping a `monotonicNowMs()` bracket → `recordSyncOperationDuration()` + * + * It does NOT drive `runContextGraphSyncWithBackpressure` itself, because that + * needs a live `DKGAgent`, peers and a store — whose cost would dominate and + * whose variance would swamp the signal. The claim this benchmark supports is + * therefore precisely: *"the instrumentation added to the attempt path and to + * the I4 boundary costs <= X ms per page"*, not *"a sync page costs X ms"*. + * `--print-seams` lists the resolved functions so a reviewer can confirm the + * real ones were loaded. + * + * ## Arms + * + * `noop` — no global MeterProvider registered, so `getMetrics()` binds to the + * OpenTelemetry API's no-op meter. This is a node with telemetry off. + * `sdk` — a real `MeterProvider` + `PeriodicExportingMetricReader` over a stub + * exporter, with the export interval pushed past the run so the timer + * never fires: we are measuring the RECORD path, which is what sits on + * the hot path, not the exporter's I/O, which does not. + * + * Arms are **interleaved** round by round and reduced by median. Running one arm + * to completion and then the other lets JIT warm-up, GC scheduling and CPU + * frequency drift land entirely on one side and be reported as instrumentation + * cost; alternating cancels monotonic drift, and the median discards the + * scheduler outliers a shared machine produces. + */ + +import { createRequire } from 'node:module'; +import fs from 'node:fs'; +import path from 'node:path'; +import { fileURLToPath, pathToFileURL } from 'node:url'; + +const require = createRequire(import.meta.url); +const scriptDir = path.dirname(fileURLToPath(import.meta.url)); +const pkgRoot = path.resolve(scriptDir, '..'); + +// ─────────────────────────────────────────────────────────────── args ──────── + +function parseArgs(argv) { + const opts = { + pages: 200, + warmup: 50, + rounds: 15, + pagesPerOperation: 20, + requestBytes: 512, + responseBytes: 64 * 1024, + json: false, + ci: false, + printSeams: false, + }; + for (let i = 0; i < argv.length; i += 1) { + const arg = argv[i]; + const num = (name) => { + const raw = argv[i + 1]; + const value = Number(raw); + if (!Number.isFinite(value) || value <= 0) { + fail(`--${name} requires a positive number (got ${JSON.stringify(raw)})`); + } + i += 1; + return value; + }; + switch (arg) { + case '--pages': opts.pages = num('pages'); break; + case '--warmup': opts.warmup = num('warmup'); break; + case '--rounds': opts.rounds = num('rounds'); break; + case '--pages-per-operation': opts.pagesPerOperation = num('pages-per-operation'); break; + case '--request-bytes': opts.requestBytes = num('request-bytes'); break; + case '--response-bytes': opts.responseBytes = num('response-bytes'); break; + case '--json': opts.json = true; break; + // Host variance on a shared runner is expected, so CI records the numbers + // without gating (§8.5). The bound is still evaluated and reported. + case '--ci': opts.ci = true; break; + case '--print-seams': opts.printSeams = true; break; + case '--help': case '-h': usage(); process.exit(0); break; + default: fail(`unknown argument: ${arg}`); + } + } + return opts; +} + +function usage() { + console.log( + 'usage: node packages/agent/scripts/bench-sync-telemetry.mjs ' + + '[--pages N] [--warmup N] [--rounds N] [--pages-per-operation N] ' + + '[--request-bytes N] [--response-bytes N] [--json] [--ci] [--print-seams]', + ); +} + +function fail(message) { + console.error(`bench-sync-telemetry: ${message}`); + process.exit(2); +} + +// ─────────────────────────────────────────────── load the REAL seam ────────── + +/** + * Resolve the real record helpers out of the built package. + * + * Cross-package imports load `dist`, so §8.1's build must have run. Two guards, + * both hard failures rather than fallbacks: + * + * 1. the module must exist and export every helper we intend to time; + * 2. the built file must not be older than its source — benchmarking a stale + * `dist` measures code that is not the code under review, which is the same + * class of mistake as running a mutation against an unrebuilt dependency. + */ +async function loadRealSeam() { + const distFile = path.join(pkgRoot, 'dist/sync/attempt-telemetry.js'); + const srcFile = path.join(pkgRoot, 'src/sync/attempt-telemetry.ts'); + const buildHint = + 'build the closure first (§8.1):\n' + + ' pnpm --filter @origintrail-official/dkg-core build\n' + + ' pnpm --filter @origintrail-official/dkg-agent build'; + + if (!fs.existsSync(distFile)) { + fail(`real record helper not built: ${distFile} is missing.\n${buildHint}`); + } + if (fs.existsSync(srcFile)) { + const distMtime = fs.statSync(distFile).mtimeMs; + const srcMtime = fs.statSync(srcFile).mtimeMs; + if (srcMtime > distMtime) { + fail( + `dist is STALE: src/sync/attempt-telemetry.ts is newer than its build ` + + `output, so this run would measure code that is not on disk.\n${buildHint}`, + ); + } + } + + const mod = await import(pathToFileURL(distFile).href); + const required = [ + 'syncAttemptAttributes', + 'recordSyncAttempt', + 'recordSyncAttemptRequestBytes', + 'recordSyncAttemptResponseBytes', + 'recordSyncOperationDuration', + 'withSyncAdmissionSource', + 'monotonicNowMs', + ]; + const absent = required.filter((name) => typeof mod[name] !== 'function' && name !== 'monotonicNowMs'); + if (typeof mod.monotonicNowMs !== 'function') absent.push('monotonicNowMs'); + if (absent.length) { + fail( + `the real record helper is missing exports: ${absent.join(', ')}.\n` + + 'This benchmark deliberately has no stub fallback — a stub would let a ' + + 'budget "pass" without ever reaching a record site.', + ); + } + return { mod, distFile }; +} + +// ───────────────────────────────────────────────────── metric arms ────────── + +function loadOtel() { + try { + return { + api: require('@opentelemetry/api'), + sdk: require('@opentelemetry/sdk-metrics'), + core: require('@origintrail-official/dkg-core'), + }; + } catch (error) { + fail( + `could not load the OpenTelemetry SDK / dkg-core from ${pkgRoot}: ` + + `${error instanceof Error ? error.message : String(error)}`, + ); + } +} + +/** + * Swallows batches. `PeriodicExportingMetricReader` requires the temporality + * selector; everything else resolves immediately so teardown cannot hang. + */ +function createStubExporter(sdk) { + let batches = 0; + return { + batches: () => batches, + exporter: { + export(_resourceMetrics, resultCallback) { + batches += 1; + resultCallback({ code: sdk.ExportResultCode?.SUCCESS ?? 0 }); + }, + forceFlush: async () => {}, + shutdown: async () => {}, + selectAggregationTemporality: () => sdk.AggregationTemporality.CUMULATIVE, + }, + }; +} + +function installNoopArm({ api, core }) { + // No provider registered ⇒ the API hands back a no-op meter. `disable()` clears + // any provider a previous arm registered; `rebuildMetrics()` then rebinds the + // cached instruments, which is what every record site reads through. + api.metrics.disable(); + core.rebuildMetrics(); + return { teardown: async () => {} }; +} + +function installSdkArm({ api, sdk, core }, runtimeMs) { + const stub = createStubExporter(sdk); + const provider = new sdk.MeterProvider({ + readers: [ + new sdk.PeriodicExportingMetricReader({ + exporter: stub.exporter, + // Past the end of the run: we are timing the record path, which is on + // the hot path. The exporter's timer is not. + exportIntervalMillis: Math.max(600_000, runtimeMs * 10), + }), + ], + }); + api.metrics.disable(); + api.metrics.setGlobalMeterProvider(provider); + core.rebuildMetrics(); + return { + exportedBatches: stub.batches, + teardown: async () => { + await provider.shutdown().catch(() => {}); + api.metrics.disable(); + core.rebuildMetrics(); + }, + }; +} + +// ───────────────────────────────────────────────────── workload ───────────── + +/** + * One round = `pages` attempt records plus the I4 boundary work for each + * completed operation. Returns milliseconds for the whole round; the caller + * divides by `pages` so the unit is ms/page, which is what A18 bounds. + * + * Attribute values are fixed, so both arms allocate identically and the SDK arm + * accumulates into a bounded number of series — the same shape a real node with + * closed vocabularies produces. + */ +function runRound(seam, opts) { + const { + syncAttemptAttributes, + recordSyncAttempt, + recordSyncAttemptRequestBytes, + recordSyncAttemptResponseBytes, + recordSyncOperationDuration, + withSyncAdmissionSource, + monotonicNowMs, + } = seam; + + const operations = Math.max(1, Math.ceil(opts.pages / opts.pagesPerOperation)); + const pagesPerOperation = Math.ceil(opts.pages / operations); + + const startedAt = monotonicNowMs(); + for (let op = 0; op < operations; op += 1) { + // The real I4 boundary: the ambient-source scope wrapping a monotonic + // bracket, ending in the duration record. + withSyncAdmissionSource('catchup-foreground', () => { + const operationStartedAt = monotonicNowMs(); + for (let page = 0; page < pagesPerOperation; page += 1) { + // Exactly the per-attempt sequence the send bracket performs. + const attributes = syncAttemptAttributes({ + transport: 'legacy', + plane: 'durable', + phase: 'data', + }); + recordSyncAttemptRequestBytes(attributes, opts.requestBytes); + recordSyncAttempt(attributes, 'response'); + recordSyncAttemptResponseBytes(attributes, opts.responseBytes, 'response'); + } + recordSyncOperationDuration({ + lane: 'durable', + source: 'catchup-foreground', + outcome: 'resolved', + durationMs: monotonicNowMs() - operationStartedAt, + }); + }); + } + return monotonicNowMs() - startedAt; +} + +// ───────────────────────────────────────────────────── statistics ─────────── + +function median(values) { + const sorted = [...values].sort((a, b) => a - b); + const mid = sorted.length >> 1; + return sorted.length % 2 ? sorted[mid] : (sorted[mid - 1] + sorted[mid]) / 2; +} + +function percentile(values, p) { + const sorted = [...values].sort((a, b) => a - b); + const idx = Math.min(sorted.length - 1, Math.max(0, Math.ceil((p / 100) * sorted.length) - 1)); + return sorted[idx]; +} + +// ───────────────────────────────────────────────────────── main ───────────── + +async function main() { + const opts = parseArgs(process.argv.slice(2)); + const { mod: seam, distFile } = await loadRealSeam(); + const otel = loadOtel(); + + if (opts.printSeams) { + console.log(`resolved real seam: ${distFile}`); + for (const name of Object.keys(seam).sort()) { + if (typeof seam[name] === 'function') console.log(` - ${name}`); + } + } + + // Warm-up runs against BOTH arms so neither pays first-call JIT during + // measurement. The SDK arm additionally lazily creates its series on first + // record, which would otherwise show up as instrumentation cost. + const warmupOpts = { ...opts, pages: opts.warmup }; + for (const install of [installNoopArm, installSdkArm]) { + const arm = install(otel, 1_000); + runRound(seam, warmupOpts); + runRound(seam, warmupOpts); + await arm.teardown(); + } + + const samples = { noop: [], sdk: [] }; + let exportedBatches = 0; + + for (let round = 0; round < opts.rounds; round += 1) { + // Interleaved, and the order flips every round so a systematic + // first-in-round advantage cannot accrue to one arm. + const order = round % 2 === 0 ? ['noop', 'sdk'] : ['sdk', 'noop']; + for (const armName of order) { + const arm = armName === 'noop' + ? installNoopArm(otel) + : installSdkArm(otel, opts.rounds * 100); + const elapsedMs = runRound(seam, opts); + samples[armName].push(elapsedMs / opts.pages); + if (armName === 'sdk' && arm.exportedBatches) exportedBatches += arm.exportedBatches(); + await arm.teardown(); + } + } + + const noopMedian = median(samples.noop); + const sdkMedian = median(samples.sdk); + const absoluteDeltaMs = sdkMedian - noopMedian; + const relativePct = noopMedian > 0 ? (absoluteDeltaMs / noopMedian) * 100 : Infinity; + + // A18: <= 2 % relative AND <= 1 ms absolute per page — except that when the + // no-op baseline is under 5 ms/page the percentage is noise and the ABSOLUTE + // bound governs. A record-site-only benchmark is always in that regime, so the + // relative number is reported for the record and does not decide the verdict. + const ABSOLUTE_BUDGET_MS = 1; + const RELATIVE_BUDGET_PCT = 2; + const NOISE_FLOOR_MS_PER_PAGE = 5; + const absoluteGoverns = noopMedian < NOISE_FLOOR_MS_PER_PAGE; + const withinAbsolute = absoluteDeltaMs <= ABSOLUTE_BUDGET_MS; + const withinRelative = relativePct <= RELATIVE_BUDGET_PCT; + const pass = absoluteGoverns ? withinAbsolute : withinAbsolute && withinRelative; + + const result = { + acceptance: 'A18', + governingBound: absoluteGoverns ? 'absolute' : 'absolute+relative', + pass, + gated: !opts.ci, + config: { + pages: opts.pages, + warmup: opts.warmup, + rounds: opts.rounds, + pagesPerOperation: opts.pagesPerOperation, + requestBytes: opts.requestBytes, + responseBytes: opts.responseBytes, + }, + seam: { + module: path.relative(path.resolve(pkgRoot, '../..'), distFile).replace(/\\/g, '/'), + perAttempt: [ + 'syncAttemptAttributes', + 'recordSyncAttemptRequestBytes', + 'recordSyncAttempt', + 'recordSyncAttemptResponseBytes', + ], + i4Boundary: ['withSyncAdmissionSource', 'monotonicNowMs', 'recordSyncOperationDuration'], + stubbed: [], + }, + msPerPage: { + noopMedian, + sdkMedian, + noopP95: percentile(samples.noop, 95), + sdkP95: percentile(samples.sdk, 95), + absoluteDeltaMs, + relativePct, + }, + budget: { + absoluteMs: ABSOLUTE_BUDGET_MS, + relativePct: RELATIVE_BUDGET_PCT, + noiseFloorMsPerPage: NOISE_FLOOR_MS_PER_PAGE, + withinAbsolute, + withinRelative, + }, + environment: { + node: process.version, + platform: `${process.platform}-${process.arch}`, + cpus: require('node:os').cpus()?.[0]?.model ?? 'unknown', + }, + // A non-zero count would mean the exporter timer fired mid-measurement and + // its I/O is inside the numbers. It must stay 0. + exportedBatchesDuringMeasurement: exportedBatches, + samples, + }; + + if (opts.json) { + console.log(JSON.stringify(result, null, 2)); + } else { + const us = (ms) => `${(ms * 1000).toFixed(3)} µs`; + console.log(''); + console.log('W1 §8.5 sync-telemetry overhead (A18)'); + console.log(` seam ${result.seam.module}`); + console.log(` config pages=${opts.pages} warmup=${opts.warmup} rounds=${opts.rounds}`); + console.log(` no-op median ${us(noopMedian)}/page (p95 ${us(result.msPerPage.noopP95)})`); + console.log(` sdk median ${us(sdkMedian)}/page (p95 ${us(result.msPerPage.sdkP95)})`); + console.log(` delta ${us(absoluteDeltaMs)}/page (${relativePct.toFixed(1)} %)`); + console.log(` governing bound ${result.governingBound} (no-op baseline ` + + `${absoluteGoverns ? 'below' : 'above'} the ${NOISE_FLOOR_MS_PER_PAGE} ms/page noise floor)`); + console.log(` verdict ${pass ? 'PASS' : 'FAIL'}` + (opts.ci ? ' (recorded, not gated)' : '')); + console.log(''); + } + + if (exportedBatches > 0) { + console.error( + `warning: the stub exporter received ${exportedBatches} batch(es) during ` + + 'measurement; export I/O may be inside the numbers.', + ); + } + if (!pass && !opts.ci) process.exit(1); +} + +main().catch((error) => { + console.error(error); + process.exit(2); +}); diff --git a/packages/agent/src/dkg-agent-lifecycle.ts b/packages/agent/src/dkg-agent-lifecycle.ts index 51f29f893c..0d44a7c13d 100644 --- a/packages/agent/src/dkg-agent-lifecycle.ts +++ b/packages/agent/src/dkg-agent-lifecycle.ts @@ -284,6 +284,7 @@ import { runSyncOnConnect, SyncOnConnectPostSyncError, type SyncOnConnectOutcome import { mapWithConcurrency } from './map-with-concurrency.js'; import { CATCHUP_MAX_CONCURRENT_PEER_SYNCS } from './sync/catchup-concurrency.js'; import { + catchupAdmissionSource, runCatchupPlanesWithPolicy, type CatchupMode, } from './sync/catchup-policy.js'; @@ -319,6 +320,24 @@ import { type SyncAdmissionSource, type SyncSchedulerLane, } from './sync/policy.js'; +import { + activeSyncAdmissionSource, + isSyncOperationCancellation, + monotonicNowMs, + recordSyncAttempt, + recordSyncAttemptRequestBytes, + recordSyncAttemptResponseBytes, + recordSyncOperationDuration, + recordSyncOperationRejected, + recordSyncSingleFlightJoin, + syncAttemptAttributes, + syncOperationRejectionReason, + syncPlaneFor, + withSyncAdmissionSource, + type SyncAttemptOutcome, + type SyncOperationOutcome, + type SyncSingleFlightScope, +} from './sync/attempt-telemetry.js'; import { generateCustodialAgent, registerSelfSovereignAgent, agentFromPrivateKey, ensureWorkspaceEncryptionKey, @@ -519,11 +538,23 @@ type InFlightSyncPageFetch = { promise: Promise; controller: AbortController; waiters: number; + /** + * Admission source of the fetch's OWNER — metadata stored BESIDE the shared + * promise, never part of the coalescing key. Putting it in the key would fork + * one physical page fetch into one per trigger, which is the opposite of what + * coalescing is for. I6 records the resulting attribution ambiguity instead. + */ + ownerSource: SyncAdmissionSource; +}; +type InFlightSyncSingleFlight = { + promise: Promise; + /** Same contract as {@link InFlightSyncPageFetch.ownerSource}. */ + ownerSource: SyncAdmissionSource; }; type ContextGraphCatchupResult = Awaited>; const inFlightSyncPageFetchesByAgent = new WeakMap>(); -const inFlightSyncSingleFlightsByAgent = new WeakMap>>(); +const inFlightSyncSingleFlightsByAgent = new WeakMap>(); const alreadyMemberDelegationRefreshChains = new WeakMap>>(); const durableContextGraphSyncChains = new WeakMap>>(); @@ -667,27 +698,57 @@ function inFlightSyncPageFetchesFor(agent: DKGAgent): Map( agent: DKGAgent, key: string, factory: () => Promise, + meta: { scope: SyncSingleFlightScope; source?: SyncAdmissionSource }, ): Promise { let inFlight = inFlightSyncSingleFlightsByAgent.get(agent); if (!inFlight) { inFlight = new Map(); inFlightSyncSingleFlightsByAgent.set(agent, inFlight); } + const { scope } = meta; + const joinerSource = normalizeSyncAdmissionSource(meta.source ?? activeSyncAdmissionSource()); const existing = inFlight.get(key); - if (existing) return existing as Promise; + if (existing) { + // Recorded at MAP-HIT time, before any bytes move: a join is a decision to + // share work, and by the time the shared promise settles there is nothing + // left to attribute. + recordSyncSingleFlightJoin({ + scope, + ownerSource: existing.ownerSource, + joinerSource, + }); + return existing.promise as Promise; + } + // Mirrors the page-fetch map's `let entry!` idiom below: the cleanup closure + // must compare the ENTRY it created, not the promise, so a later generation + // for the same key cannot be evicted by an earlier one's `finally`. + let entry!: InFlightSyncSingleFlight; const promise = Promise.resolve() .then(factory) .finally(() => { - if (inFlight.get(key) === promise) { + if (inFlight.get(key) === entry) { inFlight.delete(key); } }); - inFlight.set(key, promise); + entry = { promise, ownerSource: joinerSource }; + inFlight.set(key, entry); return promise; } @@ -920,6 +981,25 @@ export interface ContextGraphCatchupOptions { * retries. Background mode remains best-effort and never waits for capacity. */ mode?: CatchupMode; + /** + * Bounded, METADATA-ONLY admission source, replacing the one `mode` would + * imply. It exists because `source` is trigger attribution on some routes and + * execution MODE on others: `catchupSourceForMode('background')` always emits + * `catchup-background`, so VM-recovery traffic — which enters through the same + * default-background path — was reported as ordinary background catch-up and + * silently undercounted against the already-defined `vm-recovery` label. + * + * It changes NO scheduling decision (priority still follows `mode`) and it + * MUST NOT enter any coalescing or single-flight key: two catch-ups that + * differ only by this value are the same physical work, and forking the key + * on it would turn a label into duplicated network traffic. Clamped to the + * closed source set wherever it is applied. + * + * Post-approval curator/broadcast catch-up deliberately does NOT set it: it + * keeps `catchup-background`, now defined as "post-approval and background + * catch-up, after VM recovery receives its override". + */ + sourceOverride?: SyncAdmissionSource; } export type DurableSyncOptions = { @@ -1226,6 +1306,43 @@ export class LifecycleSyncMethods extends DKGAgentBase { this.node.stopSignal ?? undefined, operationSignal, ); + // W1 §6.4 — time the INNER closure, not the outer call. `withGlobalSyncBackpressure` + // invokes `work()` only after `await admission.release`, so this boundary + // excludes admission queue wait while still covering the disabled-policy + // fast path (which emits nothing today). It encloses peer auth/fetch, + // decode, verification and the atomic store commit, and includes nested + // store-scheduler queueing; it excludes the per-(peer, CG) serialization + // wait, which happens outside admission. + // + // `started` discriminates work that ran from work that never began: an + // abort while queued and an abort during work both surface as AbortError, + // and a never-started call must go to I5 rather than enter the duration + // histogram as a 0 ms sample. + let started = false; + const timedWork = async (): Promise => { + started = true; + const startedAt = monotonicNowMs(); + let outcome: SyncOperationOutcome = 'resolved'; + try { + // The admission source becomes ambient for the whole operation here — + // this is the single choke point every admission passes through, and it + // has already normalized the value. Everything below (both request + // lanes, and the changelog lane's legacy `runResync` fallback) reports + // its attempts and bytes under this label without carrying it as a + // parameter, so it can never reach a coalescing key. + return await withSyncAdmissionSource(source, work); + } catch (error) { + outcome = isSyncOperationCancellation(error) ? 'cancelled' : 'error'; + throw error; + } finally { + recordSyncOperationDuration({ + lane, + source, + outcome, + durationMs: monotonicNowMs() - startedAt, + }); + } + }; try { return await withGlobalSyncBackpressure( { @@ -1240,8 +1357,19 @@ export class LifecycleSyncMethods extends DKGAgentBase { signal: admissionBoundary.signal, logInfo: (opCtx, message) => this.log.info(opCtx, message), }, - work, + timedWork, ); + } catch (error) { + // Rejected BEFORE the work closure ever ran: queue overflow, priority + // displacement, or cancellation while queued. Never a 0 ms I4 sample. + if (!started) { + recordSyncOperationRejected({ + lane, + source, + reason: syncOperationRejectionReason(error), + }); + } + throw error; } finally { admissionBoundary.dispose(); } @@ -4584,7 +4712,10 @@ export class LifecycleSyncMethods extends DKGAgentBase { } }; return singleFlightKey - ? runSyncSingleFlight(this, singleFlightKey, runWithinBoundary) + ? runSyncSingleFlight(this, singleFlightKey, runWithinBoundary, { + scope: 'durable', + source: options?.source, + }) : runWithinBoundary(); } @@ -4907,10 +5038,51 @@ export class LifecycleSyncMethods extends DKGAgentBase { return c ? { era: c.era, seq: c.seq } : undefined; }, setCursor: (era, seq) => this.changelogCursors.set(remotePeerId, contextGraphId, era, seq), - send: (bytes) => this.messenger.sendToPeer(remotePeerId, PROTOCOL_SYNC_CHANGELOG, bytes, { - timeoutMs: SYNC_PAGE_TIMEOUT_MS, - signal: this.node.stopSignal ?? undefined, - }), + // W1 §6.3 — the changelog lane's physical send, instrumented through the + // SAME helpers as the legacy lane but deliberately NOT routed through + // `sendSyncRequest`: that would drag `withRetry`, single-use payload + // semantics and the legacy busy validator into a lane that has none of + // them. Byte accounting is added here because this lane reported ZERO + // bytes before W1, and its fallbacks re-enter the legacy lane — omitting + // it would leave the denominator silently partial and uncorrectable + // after collection. + // + // `plane: 'durable'` is exact rather than approximate: `runChangelogLane` + // admits only public Context Graphs, and `isForeignGraph` rejects the + // `/_shared_memory` and `/_private` planes, so everything this lane + // transfers is durable. Shared-memory content defers to `runResync`, + // which is separately instrumented as `transport=legacy`. + send: async (bytes) => { + const attributes = syncAttemptAttributes({ + transport: 'changelog', + plane: 'durable', + phase: 'delta', + }); + recordSyncAttemptRequestBytes(attributes, bytes.byteLength); + let outcome: SyncAttemptOutcome = 'transport_error'; + let responseByteLength: number | undefined; + try { + const response = await this.messenger.sendToPeer(remotePeerId, PROTOCOL_SYNC_CHANGELOG, bytes, { + timeoutMs: SYNC_PAGE_TIMEOUT_MS, + signal: this.node.stopSignal ?? undefined, + }); + responseByteLength = response.byteLength; + outcome = 'response'; + return response; + } catch (error) { + // Terminal state, not message text — the same reasoning as the legacy + // bracket. This lane has no in-transport validator, so it can never + // produce `validation_rejected`: a `denied` changelog response is a + // successfully DELIVERED response that the decoder classifies later. + outcome = this.node.stopSignal?.aborted === true ? 'cancelled' : 'transport_error'; + throw error; + } finally { + recordSyncAttempt(attributes, outcome); + if (responseByteLength !== undefined) { + recordSyncAttemptResponseBytes(attributes, responseByteLength, outcome); + } + } + }, // Resync = the legacy verified lane for just this CG (no re-entry into the changelog // branch). Fold its result in, and report completeness so the driver only advances // the cursor to headSeq when the resync verifiably fetched everything below it. @@ -5108,9 +5280,20 @@ export class LifecycleSyncMethods extends DKGAgentBase { assetUals, }); const inFlight = inFlightSyncPageFetchesFor(this); + // Read once, here: this fetch runs inside the admitted operation, so the + // ambient source is the trigger that both a join and the shared fetch's + // own attempts belong to. + const pageFetchSource = activeSyncAdmissionSource(); const existing = coalescingKey ? inFlight.get(coalescingKey) : undefined; if (existing) { if (!existing.controller.signal.aborted) { + // At map-hit time, before any bytes move. An aborted entry below is NOT + // a join: it is evicted and this caller starts its own fetch. + recordSyncSingleFlightJoin({ + scope: 'page', + ownerSource: existing.ownerSource, + joinerSource: pageFetchSource, + }); return waitForSyncPageFetch(existing, signal); } if (coalescingKey) inFlight.delete(coalescingKey); @@ -5200,7 +5383,7 @@ export class LifecycleSyncMethods extends DKGAgentBase { // Keep that from becoming unhandled while still propagating failures to // active waiters through the original promise. }); - entry = { promise: sharedFetch, controller, waiters: 0 }; + entry = { promise: sharedFetch, controller, waiters: 0, ownerSource: pageFetchSource }; if (coalescingKey) inFlight.set(coalescingKey, entry); return waitForSyncPageFetch(entry, signal); } @@ -5534,7 +5717,10 @@ export class LifecycleSyncMethods extends DKGAgentBase { }); }; - return runSyncSingleFlight(this, singleFlightKey, runSync); + return runSyncSingleFlight(this, singleFlightKey, runSync, { + scope: 'shared-memory', + source: options?.source, + }); } /** @@ -5658,9 +5844,14 @@ export class LifecycleSyncMethods extends DKGAgentBase { const ctx = createOperationContext('sync'); const includeSharedMemory = options?.includeSharedMemory ?? false; const mode = options?.mode ?? 'background'; + const sourceOverride = options?.sourceOverride; this.trackSyncContextGraph(contextGraphId); + // `sourceOverride` is deliberately ABSENT from this key: it is attribution + // metadata, not work identity. Adding it would fork one physical catch-up + // into one per trigger — real duplicated network traffic bought with a + // label. The resulting attribution ambiguity is what I6 measures. const singleFlightKey = contextGraphCatchupSingleFlightKey({ contextGraphId, includeSharedMemory, @@ -5717,7 +5908,11 @@ export class LifecycleSyncMethods extends DKGAgentBase { return this.runCatchupOverPeers(contextGraphId, includeSharedMemory, peers, { totalPeers: orderedPeers.length, mode, + sourceOverride, }); + }, { + scope: 'context-graph', + source: catchupAdmissionSource(mode, sourceOverride), }); } @@ -5807,7 +6002,7 @@ export class LifecycleSyncMethods extends DKGAgentBase { contextGraphId: string, includeSharedMemory: boolean, peers: Array<{ toString(): string }>, - stats?: { totalPeers?: number; mode?: CatchupMode }, + stats?: { totalPeers?: number; mode?: CatchupMode; sourceOverride?: SyncAdmissionSource }, ): Promise<{ /** Ordered connected peers before optional caller windowing. */ connectedPeers: number; @@ -5946,6 +6141,7 @@ export class LifecycleSyncMethods extends DKGAgentBase { const mode = stats?.mode ?? 'background'; return runCatchupPlanesWithPolicy({ mode, + sourceOverride: stats?.sourceOverride, includeSharedMemory, syncDurable: ({ priority, source }) => this.syncFromPeerDetailed( remotePeerId, diff --git a/packages/agent/src/dkg-agent-swm-host.ts b/packages/agent/src/dkg-agent-swm-host.ts index bfbcac3441..70d4a19fb2 100644 --- a/packages/agent/src/dkg-agent-swm-host.ts +++ b/packages/agent/src/dkg-agent-swm-host.ts @@ -4013,6 +4013,12 @@ export class SwmHostModeMethods extends DKGAgentBase { includeSharedMemory: true, maxPeers: 1, peerRotationKey: localCgId, + // This is recovery, not routine background catch-up. Without the + // override it would enter through the default-background path and + // be reported as `catchup-background`, merging repair traffic into + // the background lane. Attribution only — mode, priority, peer + // selection and the coalescing key are all unchanged. + sourceOverride: 'vm-recovery', }); if (fixedMaxAttempts === undefined) { maxAttempts = Math.max( diff --git a/packages/agent/src/index.ts b/packages/agent/src/index.ts index f9724c8f74..52b73ad5a5 100644 --- a/packages/agent/src/index.ts +++ b/packages/agent/src/index.ts @@ -323,6 +323,7 @@ export { export { CATCHUP_BACKPRESSURE_MAX_WAIT_MS, FOREGROUND_CATCHUP_SYNC_PRIORITY, + catchupAdmissionSource, catchupPriorityForMode, catchupSourceForMode, runCatchupPlaneWithPolicy, @@ -334,6 +335,7 @@ export { type CatchupPlanePolicyClock, type CatchupPlanePolicyOptions, type CatchupPlanePolicyResult, + type CatchupPlaneSourceOverride, type CatchupPlaneResult, } from './sync/catchup-policy.js'; // Which peer may let one answer stand for a WHOLE Context Graph is the load- diff --git a/packages/agent/src/p2p/sync-transport.ts b/packages/agent/src/p2p/sync-transport.ts index 718272a350..a5ea3edd4f 100644 --- a/packages/agent/src/p2p/sync-transport.ts +++ b/packages/agent/src/p2p/sync-transport.ts @@ -1,6 +1,19 @@ import { randomUUID } from 'node:crypto'; import { withRetry, withSpan, getMetrics } from '@origintrail-official/dkg-core'; -import { markSyncTransportFailure } from '../sync/error-tags.js'; +import { + markSyncTransportFailure, + markSyncValidationRejection, + isSyncValidationRejection, +} from '../sync/error-tags.js'; +import { + recordSyncAttempt, + recordSyncAttemptRequestBytes, + recordSyncAttemptResponseBytes, + syncAttemptAttributes, + type SyncAttemptOutcome, + type SyncAttemptPhase, + type SyncAttemptPlane, +} from '../sync/attempt-telemetry.js'; import type { Messenger } from './messenger.js'; /** @@ -105,6 +118,14 @@ interface SyncSendParams { validateResponse?: (responseBytes: Uint8Array) => void | Promise; protocolId: string; onRetry: (attempt: number, delay: number, err: unknown) => void; + /** + * W1 attempt labels. Both vary per call and are always locally known at the + * page loop, so they are ordinary parameters — unlike the admission source, + * which is a property of the enclosing operation and is read from the ambient + * context (see `sync/attempt-telemetry.ts`). Never a Context Graph or peer id. + */ + plane: SyncAttemptPlane; + phase: SyncAttemptPhase; } export async function sendSyncRequest(params: SyncSendParams): Promise { @@ -114,28 +135,88 @@ export async function sendSyncRequest(params: SyncSendParams): Promise { - throwIfAborted(params.signal); - const requestBytes = await params.requestFactory(); - throwIfAborted(params.signal); - const messageId = randomUUID(); - let responseBytes: Uint8Array; + // Resolved once per attempt so all three W1 points describe the same + // send, and so the ambient source is read once rather than three times. + const attributes = syncAttemptAttributes({ + transport: 'legacy', + plane: params.plane, + phase: params.phase, + }); + // `sendStarted` is what makes this an ATTEMPT counter rather than a + // closure counter. A `finally` on the whole retry closure would fire on + // five distinct states, three of which move zero bytes (abort before the + // factory, a factory/signing throw, abort after the factory) — so a + // signing failure would be reported as a network attempt. + let sendStarted = false; + let responded = false; + let responseByteLength = 0; + let outcome: SyncAttemptOutcome | undefined; try { - responseBytes = await params.send( - params.remotePeerId, - params.protocolId, - requestBytes, - params.timeoutMs, - messageId, - params.signal, - ); + throwIfAborted(params.signal); + const requestBytes = await params.requestFactory(); + throwIfAborted(params.signal); + const messageId = randomUUID(); + let responseBytes: Uint8Array; + sendStarted = true; + recordSyncAttemptRequestBytes(attributes, requestBytes.byteLength); + try { + responseBytes = await params.send( + params.remotePeerId, + params.protocolId, + requestBytes, + params.timeoutMs, + messageId, + params.signal, + ); + } catch (error) { + markSyncTransportFailure(error); + // Classified by TERMINAL STATE, never by message text: a deadline + // `TimeoutError` and a caller cancel reach here as the same + // `AbortError` shape, and `PooledStreamResetError('request timeout')` + // is the same class as 'pool closed'. The caller's own signal is the + // only non-textual evidence of caller cancellation that exists. + outcome = params.signal?.aborted === true ? 'cancelled' : 'transport_error'; + throw error; + } + responded = true; + responseByteLength = responseBytes.byteLength; + // Cancellation requested AFTER receipt is still a delivered response: + // the bytes crossed the wire and cost exactly what a used page costs. + throwIfAborted(params.signal); + try { + await params.validateResponse?.(responseBytes); + } catch (error) { + // Tag, never replace — the original error's message drives peer + // backoff and `failedPhases` accounting downstream. + markSyncValidationRejection(error); + throw error; + } + throwIfAborted(params.signal); + outcome = 'response'; + return responseBytes; } catch (error) { - markSyncTransportFailure(error); + if (outcome === undefined && responded) { + // The send resolved, so this is a post-receipt failure. Only the + // validator's own rejection is `validation_rejected`; everything else + // (a post-receipt abort, a caller-side throw) is still `response`. + // No message fallback: an untagged error is never guessed from text. + outcome = isSyncValidationRejection(error) ? 'validation_rejected' : 'response'; + } throw error; + } finally { + // I1 is finalized HERE, in the surrounding per-attempt `finally`, after + // validation and cancellation classification. An outcome fixed at send + // resolution could never later become `validation_rejected`. + if (sendStarted) { + const terminal = outcome ?? 'transport_error'; + recordSyncAttempt(attributes, terminal); + // I3 exists only if the send RESOLVED — including a response the + // validator rejected, whose bytes were still received and paid for. + if (responded) { + recordSyncAttemptResponseBytes(attributes, responseByteLength, terminal); + } + } } - throwIfAborted(params.signal); - await params.validateResponse?.(responseBytes); - throwIfAborted(params.signal); - return responseBytes; }, { maxAttempts: params.retryAttempts, diff --git a/packages/agent/src/sync/attempt-telemetry.ts b/packages/agent/src/sync/attempt-telemetry.ts new file mode 100644 index 0000000000..7aeed8ba73 --- /dev/null +++ b/packages/agent/src/sync/attempt-telemetry.ts @@ -0,0 +1,359 @@ +// SPDX-License-Identifier: Apache-2.0 +/** + * W1 sync measurement — the record sites for I1–I6. + * + * Both request lanes report through the helpers here (never through each + * other's transport): the legacy page lane calls them from the send-only + * bracket in `p2p/sync-transport.ts`, and the OT-RFC-59 changelog lane calls + * them from its own `sendToPeer` closure. Routing changelog through + * `sendSyncRequest` to share the instrumentation would drag `withRetry`, + * single-use payload semantics and the legacy busy validator along with it. + * + * ## Two rules this module exists to enforce + * + * 1. **Instrumentation never throws on the hot path.** Every emission is + * guarded, and every label is clamped to a closed vocabulary — an unknown + * value becomes `unspecified` rather than widening the label space or + * raising. The normalizers deliberately mirror `normalizeSyncAdmissionSource` + * (`sync/policy.ts`) rather than inventing a new abstraction. + * 2. **No Context Graph id and no peer id, ever.** Nothing here accepts one. + * + * ## Why the admission source is ambient + * + * `source` is trigger attribution: it is decided once, at the single admission + * choke point (`runContextGraphSyncWithBackpressure`), and is then a property of + * the whole operation rather than of any individual send. The physical send sits + * ~8 call frames below that boundary, behind several injected `fetchSyncPages` + * dependency signatures, and one of those paths — the changelog lane's + * `runResync` fallback — re-enters the legacy lane, where the source is not in + * scope at all. Passing it down as a parameter would leave that fallback's bytes + * attributed to `unspecified`, i.e. exactly the silently partial denominator the + * two-lane requirement exists to prevent. + * + * Carrying it in an `AsyncLocalStorage` follows the established idiom in + * `packages/chain/src/rpc-usage.ts` (`withRpcUsageConsumer` / + * `activeRpcUsageConsumer`, same "clamp to a closed set, never throw" contract), + * and it makes one hard constraint STRUCTURAL rather than merely tested: the + * source is never a value any coalescing/single-flight key builder can see, so + * it cannot leak into a key. `plane` and `phase` stay explicit parameters — + * those vary per attempt and are always locally available. + * + * Coalesced work is attributed to the fetch's OWNER, because the shared promise + * is created inside the owner's context. That ambiguity is deliberate and is + * owned by I6: a cross-family join invalidates the observation window. + */ + +import { AsyncLocalStorage } from 'node:async_hooks'; +import { getMetrics } from '@origintrail-official/dkg-core'; +import { normalizeSyncAdmissionSource, type SyncAdmissionSource } from './policy.js'; +import { getSyncBackpressureBusyError } from './backpressure.js'; + +/** The clamp target for every vocabulary here, including `source`. */ +const UNSPECIFIED = 'unspecified'; + +export type SyncAttemptTransport = 'legacy' | 'changelog'; +export type SyncAttemptPlane = 'durable' | 'shared-memory'; +/** `delta` is the changelog lane's only phase; the rest mirror `SyncPhase`. */ +export type SyncAttemptPhase = 'data' | 'meta' | 'snapshot' | 'catalog' | 'delta'; +/** + * Terminal state of one physically invoked send. + * + * There is deliberately no `timeout`: the router and pool emit at least seven + * mutually incompatible deadline-ish shapes, a deadline `TimeoutError` and a + * caller cancel land on the same `AbortError` differing only in message text, + * and the only classifiers available are `.message.includes(...)`. Any + * pre-response rejection that is not caller cancellation is `transport_error`. + */ +export type SyncAttemptOutcome = + | 'response' + | 'validation_rejected' + | 'cancelled' + | 'transport_error'; + +/** + * Requester lanes accepted at the I4/I5 seam. Narrower than `SyncSchedulerLane` + * on purpose: that union also carries responder and pre-authorization lanes, + * which are not logical sync operations and must not enter the denominator. + */ +export type SyncOperationLane = 'durable' | 'changelog' | 'shared_memory' | 'swm_recovery'; +/** + * `resolved`, not `success`: some orchestration paths deliberately catch a + * per-Context-Graph failure and resolve with fallback diagnostics, so `success` + * would overclaim domain success. + */ +export type SyncOperationOutcome = 'resolved' | 'error' | 'cancelled'; +export type SyncOperationRejectionReason = 'queue_full' | 'displaced' | 'aborted_before_start'; +/** One-to-one with the instrumented coalescing maps. */ +export type SyncSingleFlightScope = 'context-graph' | 'durable' | 'shared-memory' | 'page'; + +const TRANSPORTS: ReadonlySet = new Set(['legacy', 'changelog']); +const PLANES: ReadonlySet = new Set(['durable', 'shared-memory']); +const PHASES: ReadonlySet = new Set([ + 'data', 'meta', 'snapshot', 'catalog', 'delta', +]); +const ATTEMPT_OUTCOMES: ReadonlySet = new Set([ + 'response', 'validation_rejected', 'cancelled', 'transport_error', +]); +const OPERATION_LANES: ReadonlySet = new Set([ + 'durable', 'changelog', 'shared_memory', 'swm_recovery', +]); +const OPERATION_OUTCOMES: ReadonlySet = new Set([ + 'resolved', 'error', 'cancelled', +]); +const REJECTION_REASONS: ReadonlySet = new Set([ + 'queue_full', 'displaced', 'aborted_before_start', +]); +const SINGLE_FLIGHT_SCOPES: ReadonlySet = new Set([ + 'context-graph', 'durable', 'shared-memory', 'page', +]); + +/** + * Clamp a value to its closed vocabulary. "Closed vocabulary" means CLAMP: a + * value that crossed a worker/RPC boundary, arrived through a cast, or came + * from a future member nobody updated here must degrade to `unspecified`, never + * throw and never widen the label space. + */ +function clampLabel(allowed: ReadonlySet, value: string | undefined): string { + return typeof value === 'string' && allowed.has(value) ? value : UNSPECIFIED; +} + +export function normalizeSyncAttemptTransport(value: string | undefined): string { + return clampLabel(TRANSPORTS, value); +} + +export function normalizeSyncAttemptPlane(value: string | undefined): string { + return clampLabel(PLANES, value); +} + +export function normalizeSyncAttemptPhase(value: string | undefined): string { + return clampLabel(PHASES, value); +} + +export function normalizeSyncAttemptOutcome(value: string | undefined): string { + return clampLabel(ATTEMPT_OUTCOMES, value); +} + +export function normalizeSyncOperationLane(value: string | undefined): string { + return clampLabel(OPERATION_LANES, value); +} + +export function normalizeSyncOperationOutcome(value: string | undefined): string { + return clampLabel(OPERATION_OUTCOMES, value); +} + +export function normalizeSyncOperationRejectionReason(value: string | undefined): string { + return clampLabel(REJECTION_REASONS, value); +} + +export function normalizeSyncSingleFlightScope(value: string | undefined): string { + return clampLabel(SINGLE_FLIGHT_SCOPES, value); +} + +/** `plane` from the boolean every requester seam already carries. */ +export function syncPlaneFor(includeSharedMemory: boolean): SyncAttemptPlane { + return includeSharedMemory ? 'shared-memory' : 'durable'; +} + +const syncAdmissionSourceContext = new AsyncLocalStorage(); + +/** + * Establish the admission source for one logical sync operation. Called ONLY + * from `runContextGraphSyncWithBackpressure`, which is the single choke point + * every admission passes through and which already normalizes the value. + */ +export function withSyncAdmissionSource(source: SyncAdmissionSource, fn: () => T): T { + return syncAdmissionSourceContext.run(normalizeSyncAdmissionSource(source), fn); +} + +/** + * The admission source of the operation this send belongs to. Work that runs + * outside any admitted operation reports `unspecified` — honestly unattributed + * rather than guessed. §7.3 treats an `unspecified` sample as a signal that the + * window is not classifiable, so it must never silently vanish. + */ +export function activeSyncAdmissionSource(): SyncAdmissionSource { + return syncAdmissionSourceContext.getStore() ?? UNSPECIFIED; +} + +/** + * The `{transport, plane, phase, source}` labels shared by I1–I3 for one + * attempt. Resolved once so all three points of a single attempt agree, and so + * the ambient read happens exactly once per attempt rather than three times. + */ +export interface SyncAttemptAttributes { + readonly transport: string; + readonly plane: string; + readonly phase: string; + readonly source: string; +} + +export function syncAttemptAttributes(input: { + transport: SyncAttemptTransport; + plane: SyncAttemptPlane; + phase: SyncAttemptPhase | string; + /** Omitted ⇒ taken from the ambient admission context. */ + source?: SyncAdmissionSource; +}): SyncAttemptAttributes { + return { + transport: normalizeSyncAttemptTransport(input.transport), + plane: normalizeSyncAttemptPlane(input.plane), + phase: normalizeSyncAttemptPhase(input.phase), + source: normalizeSyncAdmissionSource(input.source ?? activeSyncAdmissionSource()), + }; +} + +/** + * Elapsed-time source for I4. `performance.now()` is monotonic; `Date.now()` + * follows wall-clock corrections, and an NTP step during a long sync would + * otherwise produce a negative or wildly inflated occupancy sample. + */ +export const monotonicNowMs: () => number = typeof performance?.now === 'function' + ? () => performance.now() + : Date.now; + +/** + * Did the caller cancel, rather than the operation fail? Both an abort while + * queued and an abort mid-work surface as an `AbortError`; libp2p's transport + * layer additionally raises `DOMException`s carrying `code: 'ABORT_ERR'` + * (existing precedent: `dkg-agent-registry.ts`'s dial classifier). + */ +export function isSyncOperationCancellation(error: unknown): boolean { + if (!error || typeof error !== 'object') return false; + const candidate = error as { name?: unknown; code?: unknown }; + return candidate.name === 'AbortError' || candidate.code === 'ABORT_ERR'; +} + +/** + * Why an operation never started. Derived from the admission error's own TYPE + * (`SyncBackpressureBusyError.reason`, reached through the standard + * `Error.cause` chain), never from its message. Anything else that prevented + * the work closure from running is cancellation before start. + */ +export function syncOperationRejectionReason(error: unknown): SyncOperationRejectionReason { + return getSyncBackpressureBusyError(error)?.reason ?? 'aborted_before_start'; +} + +/** Bytes must be a finite, non-negative number before they reach a counter. */ +function usableByteLength(byteLength: number): number | undefined { + return Number.isFinite(byteLength) && byteLength >= 0 ? byteLength : undefined; +} + +/** + * I2 — encoded request payload bytes at the router boundary. Recorded ONCE, + * immediately before the send is invoked, so an attempt that never receives a + * response still contributes its request bytes. + */ +export function recordSyncAttemptRequestBytes( + attributes: SyncAttemptAttributes, + byteLength: number, +): void { + const bytes = usableByteLength(byteLength); + if (bytes === undefined) return; + try { + getMetrics().syncAttemptRequestBytes.add(bytes, { ...attributes }); + } catch { + /* instrumentation must never break a sync send */ + } +} + +/** I1 — exactly one terminal point per physically invoked send. */ +export function recordSyncAttempt( + attributes: SyncAttemptAttributes, + outcome: SyncAttemptOutcome | string, +): void { + try { + getMetrics().syncAttemptTotal.add(1, { + ...attributes, + outcome: normalizeSyncAttemptOutcome(outcome), + }); + } catch { + /* instrumentation must never break a sync send */ + } +} + +/** + * I3 — encoded response payload bytes. Emitted only when the send RESOLVED, + * including a response the in-transport validator then rejected. + */ +export function recordSyncAttemptResponseBytes( + attributes: SyncAttemptAttributes, + byteLength: number, + outcome: SyncAttemptOutcome | string, +): void { + const bytes = usableByteLength(byteLength); + if (bytes === undefined) return; + try { + getMetrics().syncAttemptResponseBytes.add(bytes, { + ...attributes, + outcome: normalizeSyncAttemptOutcome(outcome), + }); + } catch { + /* instrumentation must never break a sync send */ + } +} + +/** + * I4 — active occupancy of ONE completed logical sync operation, excluding + * admission queue wait. Its `count` is the operation denominator, so a call + * that never started must not appear here at all — see + * {@link recordSyncOperationRejected}. + */ +export function recordSyncOperationDuration(input: { + lane: SyncOperationLane | string; + source: SyncAdmissionSource | string; + outcome: SyncOperationOutcome; + durationMs: number; +}): void { + if (!Number.isFinite(input.durationMs) || input.durationMs < 0) return; + try { + getMetrics().syncOperationDurationMs.record(input.durationMs, { + lane: normalizeSyncOperationLane(input.lane), + source: normalizeSyncAdmissionSource(input.source), + outcome: normalizeSyncOperationOutcome(input.outcome), + }); + } catch { + /* instrumentation must never break a sync operation */ + } +} + +/** + * I5 — an operation rejected BEFORE starting. Kept separate from I4 so + * never-started work can never enter the duration histogram as a 0 ms sample, + * which would deflate the mean and inflate the denominator at once. + */ +export function recordSyncOperationRejected(input: { + lane: SyncOperationLane | string; + source: SyncAdmissionSource | string; + reason: SyncOperationRejectionReason | string; +}): void { + try { + getMetrics().syncOperationRejectedTotal.add(1, { + lane: normalizeSyncOperationLane(input.lane), + source: normalizeSyncAdmissionSource(input.source), + reason: normalizeSyncOperationRejectionReason(input.reason), + }); + } catch { + /* instrumentation must never break a sync operation */ + } +} + +/** + * I6 — a coalescing/single-flight join, recorded at MAP-HIT time (before any + * bytes move, and before the joined work can finish). `owner_source` is stored + * as metadata beside the shared promise; it is never part of the key. + */ +export function recordSyncSingleFlightJoin(input: { + scope: SyncSingleFlightScope | string; + ownerSource: SyncAdmissionSource | string; + joinerSource: SyncAdmissionSource | string; +}): void { + try { + getMetrics().syncSingleflightJoinsTotal.add(1, { + scope: normalizeSyncSingleFlightScope(input.scope), + owner_source: normalizeSyncAdmissionSource(input.ownerSource), + joiner_source: normalizeSyncAdmissionSource(input.joinerSource), + }); + } catch { + /* instrumentation must never break a sync operation */ + } +} diff --git a/packages/agent/src/sync/catchup-policy.ts b/packages/agent/src/sync/catchup-policy.ts index d3cd44a159..8247ef6a50 100644 --- a/packages/agent/src/sync/catchup-policy.ts +++ b/packages/agent/src/sync/catchup-policy.ts @@ -1,4 +1,4 @@ -import type { SyncAdmissionSource } from './policy.js'; +import { normalizeSyncAdmissionSource, type SyncAdmissionSource } from './policy.js'; export type CatchupMode = 'background' | 'foreground'; @@ -75,7 +75,47 @@ export interface CatchupPlaneResult { export interface CatchupPlaneContext { priority?: number; - source?: CatchupAdmissionSource; + /** + * Widened from {@link CatchupAdmissionSource} because a caller may override + * the mode-derived value (see {@link CatchupPlaneSourceOverride}), and + * `vm-recovery` is not expressible in the two-member catch-up subset. Both + * consumers already accept the full admission union, and the scheduler + * re-clamps regardless, so the label space is unchanged. + */ + source?: SyncAdmissionSource; +} + +export interface CatchupPlaneSourceOverride { + /** + * Replace the source this catch-up would otherwise be attributed to. + * + * Kept separate from `mode` on purpose: `mode` decides scheduling (priority + * and whether refusals are retried) while this decides only attribution. VM + * recovery runs as ordinary background catch-up — it should keep background's + * scheduling — but reporting it as `catchup-background` merges recovery + * traffic into the background lane and undercounts it. + * + * Metadata only. It reaches no key, no wire envelope and no scheduling + * decision, and is clamped to the closed source set before use. + */ + sourceOverride?: SyncAdmissionSource; +} + +/** + * The single point of truth for "what source does this catch-up admit under". + * + * Shared by the admission seam and by the single-flight join site, because + * those two must agree: if the join recorded a mode-derived label while the + * admission recorded an overridden one, I6's owner/joiner families and I4's + * source would describe the same work differently. + */ +export function catchupAdmissionSource( + mode: CatchupMode, + sourceOverride?: SyncAdmissionSource, +): SyncAdmissionSource { + return sourceOverride === undefined + ? catchupSourceForMode(mode) + : normalizeSyncAdmissionSource(sourceOverride); } export interface CatchupBackpressureRetryPolicy { @@ -112,7 +152,7 @@ export interface CatchupPlanePolicyClock { export interface CatchupPlanePolicyOptions< TDurable extends CatchupPlaneResult, TShared extends CatchupPlaneResult, -> extends CatchupPlanePolicyClock { +> extends CatchupPlanePolicyClock, CatchupPlaneSourceOverride { mode: CatchupMode; includeSharedMemory: boolean; syncDurable: (context: CatchupPlaneContext) => Promise; @@ -194,7 +234,7 @@ export function nextCatchupBackpressureDelayMs(input: { export async function runCatchupPlaneWithPolicy( mode: CatchupMode, run: (context: CatchupPlaneContext) => Promise, - options: CatchupPlanePolicyClock = {}, + options: CatchupPlanePolicyClock & CatchupPlaneSourceOverride = {}, ): Promise { // `retryDelaysMs` configured the fixed [100, 250, 500] ladder that #2006 replaced // with a wall-clock budget. Retaining it as `?: never` makes a TypeScript caller @@ -211,9 +251,11 @@ export async function runCatchupPlaneWithPolicy( ); } + // Priority still follows `mode` alone — the override is attribution, not + // scheduling, so an overridden plane queues exactly as it did before. const context: CatchupPlaneContext = { priority: catchupPriorityForMode(mode), - source: catchupSourceForMode(mode), + source: catchupAdmissionSource(mode, options.sourceOverride), }; if (mode !== 'foreground') return run(context); diff --git a/packages/agent/src/sync/error-tags.ts b/packages/agent/src/sync/error-tags.ts index 170c2ae1b1..d3d62d6b85 100644 --- a/packages/agent/src/sync/error-tags.ts +++ b/packages/agent/src/sync/error-tags.ts @@ -1,7 +1,7 @@ import { isOversizedRdfLiteralError } from '@origintrail-official/dkg-core'; import { isChainRpcTransportError } from '@origintrail-official/dkg-chain'; -type SyncErrorTag = 'syncPeerResponded' | 'syncTransportFailure'; +type SyncErrorTag = 'syncPeerResponded' | 'syncTransportFailure' | 'syncValidationRejected'; function markSyncError(error: unknown, tag: SyncErrorTag): void { if (!error || (typeof error !== 'object' && typeof error !== 'function')) return; @@ -28,6 +28,28 @@ export function markSyncTransportFailure(error: unknown): void { markSyncError(error, 'syncTransportFailure'); } +/** + * The peer's response ARRIVED and the in-transport validator then rejected it + * (W1 attempt outcome `validation_rejected`, whose received bytes still count). + * + * Tagging, never replacing: `makeLegacySyncBusyError`'s message is matched by + * {@link isSyncBackoffWorthyError}, so minting a substitute error would silently + * change peer backoff, the durable-data verifiable-prefix return and + * `failedPhases` accounting — a behaviour change dressed as telemetry. The tag + * is non-enumerable and best-effort, exactly like the two above. + * + * There is no message fallback for this marker. A rejection that reaches the + * record site untagged is classified by its terminal state, never guessed from + * text: the deadline/cancel/reset surfaces are indistinguishable by message. + */ +export function markSyncValidationRejection(error: unknown): void { + markSyncError(error, 'syncValidationRejected'); +} + +export function isSyncValidationRejection(error: unknown): boolean { + return Boolean(error && typeof error === 'object' && (error as { syncValidationRejected?: boolean }).syncValidationRejected); +} + export function didSyncPeerRespond(error: unknown): boolean { return Boolean(error && typeof error === 'object' && ( (error as { syncPeerResponded?: boolean }).syncPeerResponded || diff --git a/packages/agent/src/sync/requester/page-fetch.ts b/packages/agent/src/sync/requester/page-fetch.ts index f2c52a48e5..0004b4a6c9 100644 --- a/packages/agent/src/sync/requester/page-fetch.ts +++ b/packages/agent/src/sync/requester/page-fetch.ts @@ -5,6 +5,7 @@ import { type SingleUseSyncSender, } from '../../p2p/sync-transport.js'; import { isSyncBackoffWorthyError, markSyncPeerResponded } from '../error-tags.js'; +import { syncPlaneFor } from '../attempt-telemetry.js'; import { appendInPlace } from '../append-in-place.js'; import type { SyncPhase } from '../auth/request-build.js'; import { exactAssetFilterKey } from '../exact-assets.js'; @@ -348,6 +349,12 @@ export async function fetchSyncPages(params: FetchSyncPagesParams): Promise; + value: number; +} + +export interface HistogramPoint { + attributes: Record; + count: number; + sum: number; + max: number; +} + +export class W1MetricsHarness { + private provider: MeterProvider | null = null; + private exporter: InMemoryMetricExporter | null = null; + + install(): void { + this.exporter = new InMemoryMetricExporter(AggregationTemporality.CUMULATIVE); + this.provider = new MeterProvider({ + readers: [new PeriodicExportingMetricReader({ + exporter: this.exporter, + // Long enough that only the explicit flushes below produce a batch, so + // a background tick can never add a second CUMULATIVE copy. + exportIntervalMillis: 600_000, + })], + }); + metrics.setGlobalMeterProvider(this.provider); + rebuildMetrics(); + } + + /** + * Reset BEFORE flushing, every time. + * + * The exporter appends each batch to one array while the temporality is + * CUMULATIVE, so a test that queries twice would otherwise read the union of + * two full snapshots and see every count doubled — and a helper that flushed + * only once would answer a later query from a STALE snapshot, silently + * reporting zero for work done after the first read. Both failure modes are + * "the assertion cannot fail for the right reason", so neither is acceptable. + */ + private async rawPoints(name: string): Promise; value: unknown }>> { + this.exporter!.reset(); + await this.provider!.forceFlush(); + const out: Array<{ attributes: Record; value: unknown }> = []; + for (const resourceMetrics of this.exporter!.getMetrics()) { + for (const scopeMetrics of resourceMetrics.scopeMetrics) { + for (const metric of scopeMetrics.metrics) { + if (metric.descriptor.name !== name) continue; + for (const dataPoint of metric.dataPoints) { + out.push({ + attributes: dataPoint.attributes as Record, + value: dataPoint.value, + }); + } + } + } + } + return out; + } + + async counter(name: string): Promise { + const points = await this.rawPoints(name); + return points.map((point) => ({ attributes: point.attributes, value: point.value as number })); + } + + async histogram(name: string): Promise { + const points = await this.rawPoints(name); + return points.map((point) => { + const value = point.value as { count: number; sum?: number; max?: number }; + return { + attributes: point.attributes, + count: value.count, + sum: value.sum ?? 0, + max: value.max ?? 0, + }; + }); + } + + /** Total across every attribute combination — the "how many points" question. */ + async total(name: string): Promise { + const points = await this.counter(name); + return points.reduce((sum, point) => sum + point.value, 0); + } + + /** Points whose attributes include every entry of `match`. */ + async matching(name: string, match: Record): Promise { + const points = await this.counter(name); + return points.filter((point) => + Object.entries(match).every(([key, value]) => point.attributes[key] === value)); + } + + /** The distinct attribute-key set of an instrument — the cardinality contract. */ + async attributeKeys(name: string): Promise { + const points = await this.rawPoints(name); + return [...new Set(points.flatMap((point) => Object.keys(point.attributes)))].sort(); + } + + async dispose(): Promise { + if (this.provider) { + await this.provider.shutdown().catch(() => {}); + this.provider = null; + } + this.exporter = null; + metrics.disable(); + rebuildMetrics(); + } +} diff --git a/packages/agent/test/core-fills-gap.test.ts b/packages/agent/test/core-fills-gap.test.ts index 495da486df..ba2e658d74 100644 --- a/packages/agent/test/core-fills-gap.test.ts +++ b/packages/agent/test/core-fills-gap.test.ts @@ -1599,15 +1599,22 @@ describe('Phase D - VM reconcile damping', () => { await expect(internals.reconcileChainOrdinal('47', onChainCgId, 0, undefined)).resolves.toEqual({ status: 'pending' }); expect(fetch.calls).toHaveLength(2); + // W1 §5.5 — `sourceOverride` is attribution only: it changes no peer + // selection, no rotation and no admission priority, which is why this + // options object is otherwise unchanged. Pinned as an EXACT literal + // because "not catchup-background" would also be satisfied by dropping + // the override entirely and landing on some other excluded source. expect(fetch.calls[0]).toEqual(['47', { includeSharedMemory: true, maxPeers: 1, peerRotationKey: '47', + sourceOverride: 'vm-recovery', }]); expect(fetch.calls[1]).toEqual(['47', { includeSharedMemory: true, maxPeers: 1, peerRotationKey: '47', + sourceOverride: 'vm-recovery', }]); }); diff --git a/packages/agent/test/sync-attempt-telemetry.test.ts b/packages/agent/test/sync-attempt-telemetry.test.ts new file mode 100644 index 0000000000..6a98352ac3 --- /dev/null +++ b/packages/agent/test/sync-attempt-telemetry.test.ts @@ -0,0 +1,310 @@ +// SPDX-License-Identifier: Apache-2.0 +/** + * W1 §6.2 / §10 — the per-ATTEMPT contract (I1–I3), driven through the REAL + * `sendSyncRequest`, plus the clamping and never-throws guarantees of the + * shared record helpers. + * + * Acceptance covered here: A2, A3, A7 (attempt instruments), A11, A14, A19. + * Mutants these assertions are written to kill: M1, M2, M3. + */ +import { describe, it, expect, afterEach } from 'vitest'; +import { getMetrics } from '@origintrail-official/dkg-core'; +import { sendSyncRequest } from '../src/p2p/sync-transport.js'; +import { + activeSyncAdmissionSource, + normalizeSyncAttemptOutcome, + normalizeSyncAttemptPhase, + normalizeSyncAttemptPlane, + normalizeSyncAttemptTransport, + normalizeSyncOperationLane, + normalizeSyncOperationOutcome, + normalizeSyncOperationRejectionReason, + normalizeSyncSingleFlightScope, + recordSyncAttempt, + recordSyncAttemptRequestBytes, + recordSyncAttemptResponseBytes, + recordSyncOperationDuration, + recordSyncOperationRejected, + recordSyncSingleFlightJoin, + syncAttemptAttributes, + syncOperationRejectionReason, + syncPlaneFor, + withSyncAdmissionSource, +} from '../src/sync/attempt-telemetry.js'; +import { SyncBackpressureBusyError } from '../src/sync/backpressure.js'; +import { W1MetricsHarness, I1, I2, I3 } from './_helpers/w1-metrics.js'; + +const PROTO = '/dkg/10.0.2/sync'; +const REQUEST_BYTES = new Uint8Array([1, 2, 3, 4]); +const RESPONSE_BYTES = new Uint8Array([9, 9]); + +const harness = new W1MetricsHarness(); + +afterEach(async () => { + await harness.dispose(); +}); + +function attemptParams(overrides: Record = {}): any { + return { + remotePeerId: 'remote-peer', + timeoutMs: 1000, + retryAttempts: 1, + contextGraphId: 'cg-1', + offset: 0, + protocolId: PROTO, + plane: 'durable', + phase: 'data', + requestFactory: async () => REQUEST_BYTES, + send: async () => RESPONSE_BYTES, + onRetry: () => {}, + ...overrides, + }; +} + +describe('W1 I1–I3 — per-attempt accounting through the real sendSyncRequest', () => { + it('A2/M1: three physical sends produce three attempt points, not one per call', async () => { + harness.install(); + let sends = 0; + await sendSyncRequest(attemptParams({ + retryAttempts: 3, + send: async () => { + sends += 1; + if (sends < 3) throw new Error('stream reset'); + return RESPONSE_BYTES; + }, + })); + + expect(sends).toBe(3); + // Recording once per CALL (M1) yields 1; recording once per SEND yields 3. + expect(await harness.total(I1)).toBe(3); + expect((await harness.matching(I1, { outcome: 'transport_error' })) + .reduce((sum, point) => sum + point.value, 0)).toBe(2); + expect((await harness.matching(I1, { outcome: 'response' })) + .reduce((sum, point) => sum + point.value, 0)).toBe(1); + // Every send paid for its request bytes; only the one that came back has a response. + expect(await harness.total(I2)).toBe(3 * REQUEST_BYTES.byteLength); + expect(await harness.total(I3)).toBe(RESPONSE_BYTES.byteLength); + }, 20_000); + + it('A3/M2: an attempt that receives NO response still counts its request bytes', async () => { + harness.install(); + await expect(sendSyncRequest(attemptParams({ + send: async () => { throw new Error('all multiaddr dials failed'); }, + }))).rejects.toThrow(/multiaddr/); + + // M2 ("request bytes only on success") makes this zero. + expect(await harness.total(I2)).toBe(REQUEST_BYTES.byteLength); + expect(await harness.total(I1)).toBe(1); + expect(await harness.matching(I1, { outcome: 'transport_error' })).toHaveLength(1); + // I3 exists only when the send RESOLVED. + expect(await harness.counter(I3)).toEqual([]); + }); + + it('A14/M3: a rejected response is validation_rejected, and its bytes are still counted', async () => { + harness.install(); + await expect(sendSyncRequest(attemptParams({ + validateResponse: () => { throw new Error('Legacy sync responder busy at peer for "cg-1" (data)'); }, + }))).rejects.toThrow(/Legacy sync responder busy/); + + const attempts = await harness.matching(I1, { outcome: 'validation_rejected' }); + expect(attempts).toHaveLength(1); + expect(attempts[0]!.value).toBe(1); + // M3 classifies the same attempt as `response`; asserting the exact label + // AND the absence of `response` kills it from both sides. + expect(await harness.matching(I1, { outcome: 'response' })).toEqual([]); + + const responseBytes = await harness.matching(I3, { outcome: 'validation_rejected' }); + expect(responseBytes).toHaveLength(1); + expect(responseBytes[0]!.value).toBe(RESPONSE_BYTES.byteLength); + }); + + it('does not replace the validator\'s error, so downstream classifiers still see it', async () => { + harness.install(); + // `isSyncBackoffWorthyError` matches this message; a minted replacement + // would silently change peer backoff and failedPhases accounting. + const message = 'Legacy sync responder busy at peer for "cg-1" (data)'; + const original = new Error(message); + await expect(sendSyncRequest(attemptParams({ + validateResponse: () => { throw original; }, + }))).rejects.toBe(original); + expect(original.message).toBe(message); + }); + + it('A19: a requestFactory failure mints ZERO I1/I2/I3 points', async () => { + harness.install(); + let sends = 0; + await expect(sendSyncRequest(attemptParams({ + requestFactory: async () => { throw new Error('signing key unavailable'); }, + send: async () => { sends += 1; return RESPONSE_BYTES; }, + }))).rejects.toThrow(/signing key/); + + expect(sends).toBe(0); + // A closure-level `finally` would report this signing failure as a network attempt. + expect(await harness.counter(I1)).toEqual([]); + expect(await harness.counter(I2)).toEqual([]); + expect(await harness.counter(I3)).toEqual([]); + }); + + it('an abort BEFORE the factory mints zero points; an abort during the send is `cancelled`', async () => { + harness.install(); + const preAborted = AbortSignal.abort(new Error('caller gave up')); + await expect(sendSyncRequest(attemptParams({ signal: preAborted }))) + .rejects.toBeTruthy(); + expect(await harness.counter(I1)).toEqual([]); + + const controller = new AbortController(); + await expect(sendSyncRequest(attemptParams({ + signal: controller.signal, + send: async () => { + controller.abort(new Error('caller gave up')); + throw new Error('peer-closed-stream'); + }, + }))).rejects.toBeTruthy(); + + expect(await harness.matching(I1, { outcome: 'cancelled' })).toHaveLength(1); + // Request bytes were already committed before the send was invoked. + expect(await harness.total(I2)).toBe(REQUEST_BYTES.byteLength); + expect(await harness.counter(I3)).toEqual([]); + }); + + it('cancellation requested AFTER receipt is still a delivered `response`', async () => { + harness.install(); + const controller = new AbortController(); + await expect(sendSyncRequest(attemptParams({ + signal: controller.signal, + send: async () => { + // The bytes crossed the wire; only then does the caller give up. + controller.abort(new Error('caller gave up')); + return RESPONSE_BYTES; + }, + }))).rejects.toBeTruthy(); + + expect(await harness.matching(I1, { outcome: 'response' })).toHaveLength(1); + expect(await harness.matching(I1, { outcome: 'cancelled' })).toEqual([]); + expect(await harness.total(I3)).toBe(RESPONSE_BYTES.byteLength); + }); + + it('A7: attempt points carry only the bounded label set — no CG id, no peer id', async () => { + harness.install(); + await sendSyncRequest(attemptParams()); + + expect(await harness.attributeKeys(I1)).toEqual(['outcome', 'phase', 'plane', 'source', 'transport']); + expect(await harness.attributeKeys(I2)).toEqual(['phase', 'plane', 'source', 'transport']); + expect(await harness.attributeKeys(I3)).toEqual(['outcome', 'phase', 'plane', 'source', 'transport']); + + const points = await harness.counter(I1); + const values = points.flatMap((point) => Object.values(point.attributes).map(String)); + expect(values.some((value) => value.includes('cg-1'))).toBe(false); + expect(values.some((value) => value.includes('remote-peer'))).toBe(false); + }); + + it('labels the attempt with the ambient admission source, and `unspecified` without one', async () => { + harness.install(); + await withSyncAdmissionSource('catchup-foreground', () => sendSyncRequest(attemptParams())); + await sendSyncRequest(attemptParams({ phase: 'meta' })); + + expect(await harness.matching(I1, { source: 'catchup-foreground', phase: 'data' })).toHaveLength(1); + expect(await harness.matching(I1, { source: 'unspecified', phase: 'meta' })).toHaveLength(1); + }); + + it('propagates the ambient source across awaits and restores it afterwards', async () => { + expect(activeSyncAdmissionSource()).toBe('unspecified'); + await withSyncAdmissionSource('reconcile', async () => { + await Promise.resolve(); + await new Promise((resolve) => setTimeout(resolve, 1)); + expect(activeSyncAdmissionSource()).toBe('reconcile'); + }); + expect(activeSyncAdmissionSource()).toBe('unspecified'); + }); +}); + +describe('W1 A11 — closed vocabularies clamp, and instrumentation never throws', () => { + it('clamps every unknown label to `unspecified` instead of widening or throwing', () => { + expect(normalizeSyncAttemptTransport('quic')).toBe('unspecified'); + expect(normalizeSyncAttemptTransport(undefined)).toBe('unspecified'); + expect(normalizeSyncAttemptTransport('changelog')).toBe('changelog'); + expect(normalizeSyncAttemptPlane('private')).toBe('unspecified'); + expect(normalizeSyncAttemptPlane('shared-memory')).toBe('shared-memory'); + expect(normalizeSyncAttemptPhase('delta')).toBe('delta'); + // `timeout` is deliberately NOT in the outcome vocabulary — no causal + // deadline predicate exists, so it must clamp rather than be accepted. + expect(normalizeSyncAttemptOutcome('timeout')).toBe('unspecified'); + expect(normalizeSyncAttemptOutcome('validation_rejected')).toBe('validation_rejected'); + // Responder and pre-authorization lanes are not logical sync operations. + expect(normalizeSyncOperationLane('responder')).toBe('unspecified'); + expect(normalizeSyncOperationLane('pre_authorization')).toBe('unspecified'); + expect(normalizeSyncOperationLane('swm_recovery')).toBe('swm_recovery'); + expect(normalizeSyncOperationOutcome('success')).toBe('unspecified'); + expect(normalizeSyncOperationOutcome('resolved')).toBe('resolved'); + expect(normalizeSyncOperationRejectionReason('rejected')).toBe('unspecified'); + expect(normalizeSyncSingleFlightScope('cg')).toBe('unspecified'); + expect(normalizeSyncSingleFlightScope('context-graph')).toBe('context-graph'); + }); + + it('clamps an out-of-set source that reached the attribute builder through a cast', () => { + const attributes = syncAttemptAttributes({ + transport: 'legacy', + plane: 'durable', + phase: 'data', + source: 'catchup-urgent' as never, + }); + expect(attributes.source).toBe('unspecified'); + }); + + it('maps the plane boolean both ways', () => { + expect(syncPlaneFor(false)).toBe('durable'); + expect(syncPlaneFor(true)).toBe('shared-memory'); + }); + + it('derives the I5 reason from the admission error TYPE, through the cause chain', () => { + expect(syncOperationRejectionReason(new SyncBackpressureBusyError('full'))).toBe('queue_full'); + expect(syncOperationRejectionReason(new SyncBackpressureBusyError('bumped', 'displaced'))).toBe('displaced'); + expect(syncOperationRejectionReason( + new Error('wrapped', { cause: new SyncBackpressureBusyError('full') }), + )).toBe('queue_full'); + // A message that merely LOOKS like backpressure is not backpressure. + expect(syncOperationRejectionReason(new Error('Sync backpressure rejected durable:cg'))) + .toBe('aborted_before_start'); + expect(syncOperationRejectionReason(undefined)).toBe('aborted_before_start'); + }); + + it('swallows a throwing meter at every record site, and the send still succeeds', async () => { + harness.install(); + const exploding = { add() { throw new Error('meter exploded'); }, record() { throw new Error('meter exploded'); } }; + const instruments = getMetrics() as unknown as Record; + for (const name of [ + 'syncAttemptTotal', 'syncAttemptRequestBytes', 'syncAttemptResponseBytes', + 'syncOperationDurationMs', 'syncOperationRejectedTotal', 'syncSingleflightJoinsTotal', + ]) { + instruments[name] = exploding; + } + + const attributes = syncAttemptAttributes({ transport: 'legacy', plane: 'durable', phase: 'data' }); + expect(() => recordSyncAttempt(attributes, 'response')).not.toThrow(); + expect(() => recordSyncAttemptRequestBytes(attributes, 10)).not.toThrow(); + expect(() => recordSyncAttemptResponseBytes(attributes, 10, 'response')).not.toThrow(); + expect(() => recordSyncOperationDuration({ + lane: 'durable', source: 'reconcile', outcome: 'resolved', durationMs: 5, + })).not.toThrow(); + expect(() => recordSyncOperationRejected({ + lane: 'durable', source: 'reconcile', reason: 'queue_full', + })).not.toThrow(); + expect(() => recordSyncSingleFlightJoin({ + scope: 'page', ownerSource: 'on-connect', joinerSource: 'reconcile', + })).not.toThrow(); + + // The hot path itself must be unaffected. + const result = await sendSyncRequest(attemptParams()); + expect(Array.from(result)).toEqual(Array.from(RESPONSE_BYTES)); + }); + + it('drops non-finite byte counts rather than poisoning a counter', async () => { + harness.install(); + const attributes = syncAttemptAttributes({ transport: 'legacy', plane: 'durable', phase: 'data' }); + recordSyncAttemptRequestBytes(attributes, Number.NaN); + recordSyncAttemptResponseBytes(attributes, -1, 'response'); + recordSyncOperationDuration({ lane: 'durable', source: 'reconcile', outcome: 'resolved', durationMs: Number.NaN }); + expect(await harness.counter(I2)).toEqual([]); + expect(await harness.counter(I3)).toEqual([]); + }); +}); diff --git a/packages/agent/test/sync-operation-telemetry.test.ts b/packages/agent/test/sync-operation-telemetry.test.ts new file mode 100644 index 0000000000..609fae3b4b --- /dev/null +++ b/packages/agent/test/sync-operation-telemetry.test.ts @@ -0,0 +1,503 @@ +// SPDX-License-Identifier: Apache-2.0 +/** + * W1 §6.3 / §6.4 / §6.5 — the OPERATION-level contract, driven through a real + * `DKGAgent`: the I4/I5 admission seam, the I6 join sites, the changelog lane's + * own attempts and bytes, and the `vm-recovery` source override. + * + * Acceptance covered here: A6, A8, A12, A15. + * Mutants these assertions are written to kill: M4, M5, M6, M10, M11, M13. + */ +import { describe, it, expect, afterEach } from 'vitest'; +import { createOperationContext } from '@origintrail-official/dkg-core'; +import { MockChainAdapter } from '@origintrail-official/dkg-chain'; + +import { DKGAgent, runCatchupPlaneWithPolicy } from '../src/index.js'; +import { getSyncBackpressureSnapshot } from '../src/sync/backpressure.js'; +import { withSyncAdmissionSource } from '../src/sync/attempt-telemetry.js'; +import { encodeChangelogResponse } from '../src/sync/changelog/wire.js'; +import type { SyncPageResult } from '../src/sync/requester/page-fetch.js'; +import { stubLifecycleFetch } from './_helpers/sync-fetch-coalescing.js'; +import { W1MetricsHarness, I1, I2, I3, I4, I5, I6 } from './_helpers/w1-metrics.js'; + +const PEER_A = '12D3KooWSmU3owJvB9sFw8uApDgKrv2VBMecsGGvgAc4Gq6hB57M'; +const DEFAULT_DEADLINE = Date.UTC(2100, 0, 1); +const CG = 'w1-cg'; + +const harness = new W1MetricsHarness(); +const liveAgents: DKGAgent[] = []; + +afterEach(async () => { + for (const agent of liveAgents.splice(0)) await agent.stop().catch(() => {}); + await harness.dispose(); +}); + +function deferred() { + let resolve!: (value: T) => void; + let reject!: (error: unknown) => void; + const promise = new Promise((res, rej) => { resolve = res; reject = rej; }); + return { promise, resolve, reject }; +} + +async function flushMicrotasks(): Promise { + for (let i = 0; i < 5; i += 1) await Promise.resolve(); +} + +async function waitFor(condition: () => boolean, timeoutMs = 2_000): Promise { + const deadline = Date.now() + timeoutMs; + while (!condition()) { + if (Date.now() > deadline) throw new Error('condition was not met before timeout'); + await new Promise((resolve) => setTimeout(resolve, 1)); + } +} + +function emptySyncPage(phase: string): SyncPageResult { + return { + quads: [], + bytesReceived: 0, + resumedFromOffset: 0, + nextOffset: 0, + checkpointKey: `checkpoint:${phase}`, + completed: true, + timedOut: false, + }; +} + +async function createAgent(options: { + sendToPeer?: (...args: unknown[]) => Promise; + syncGlobalMaxInflight?: number; + syncGlobalQueueLimit?: number; +} = {}): Promise { + const agent = await DKGAgent.create({ + name: 'W1OperationTelemetry', + listenHost: '127.0.0.1', + chainAdapter: new MockChainAdapter(), + syncGlobalMaxInflight: options.syncGlobalMaxInflight ?? 2, + syncGlobalQueueLimit: options.syncGlobalQueueLimit ?? 2, + }); + liveAgents.push(agent); + (agent as any).messenger = { sendToPeer: options.sendToPeer ?? (async () => new Uint8Array(0)) }; + (agent as any).buildSyncRequest = async () => new Uint8Array([1, 2, 3]); + return agent; +} + +/** Reduce a durable page fetch + verification to a no-op so the sync completes fast. */ +function stubDurableSyncBody(agent: DKGAgent, onFetch?: () => void): void { + stubLifecycleFetch(agent, async ({ phase }) => { + onFetch?.(); + return emptySyncPage(phase); + }); + (agent as any).processDurableBatchInWorker = async () => ({ + verifiedData: [], + verifiedMeta: [], + totalFetchedDataQuads: 0, + totalFetchedMetaQuads: 0, + rejectedKcs: 0, + emptyResponses: 1, + metaOnlyResponses: 0, + dataRejectedMissingMeta: 0, + }); + // Keep every Context Graph on the legacy lane: this peer advertises no + // changelog protocol, exactly as a pre-RFC-59 peer would. + (agent as any).getPeerProtocols = async () => []; +} + +function admit( + agent: DKGAgent, + options: { lane?: string; source?: string; label?: string; operationSignal?: AbortSignal }, + work: () => Promise, +): Promise { + return (agent as any).runContextGraphSyncWithBackpressure( + createOperationContext('sync'), + CG, + options.lane ?? 'durable', + options.label ?? 'w1-op', + work, + { source: options.source, operationSignal: options.operationSignal }, + ); +} + +describe('W1 I4/I5 — the operation denominator and its rejections', () => { + it('A12/M11: the duration sample excludes admission queue wait', async () => { + harness.install(); + const agent = await createAgent({ syncGlobalMaxInflight: 1, syncGlobalQueueLimit: 2 }); + const blocker = deferred(); + + const blocking = admit(agent, { source: 'on-connect', label: 'blocker' }, () => blocker.promise); + await waitFor(() => getSyncBackpressureSnapshot().inflight === 1); + + // The queued operation's WORK is a no-op; all of its wall-clock cost is + // queue wait. M11 (timing the outer call) folds that wait into the sample. + const queued = admit(agent, { source: 'reconcile', label: 'queued' }, async () => 'done'); + await waitFor(() => getSyncBackpressureSnapshot().queued === 1); + const QUEUE_WAIT_MS = 400; + await new Promise((resolve) => setTimeout(resolve, QUEUE_WAIT_MS)); + + blocker.resolve(); + await blocking; + expect(await queued).toBe('done'); + + const samples = await harness.histogram(I4); + const queuedSample = samples.find((point) => point.attributes.source === 'reconcile'); + expect(queuedSample).toBeDefined(); + expect(queuedSample!.count).toBe(1); + expect(queuedSample!.attributes).toMatchObject({ lane: 'durable', outcome: 'resolved' }); + expect(queuedSample!.max).toBeLessThan(QUEUE_WAIT_MS / 2); + + // Positive control: the blocker's own occupancy WAS measured, so a sample + // near zero above means "queue wait excluded", not "the clock is broken". + const blockerSample = samples.find((point) => point.attributes.source === 'on-connect'); + expect(blockerSample!.count).toBe(1); + expect(blockerSample!.max).toBeGreaterThanOrEqual(QUEUE_WAIT_MS * 0.8); + }); + + it('A12/M10: an operation rejected before starting goes to I5, never a 0 ms I4 sample', async () => { + harness.install(); + const agent = await createAgent({ syncGlobalMaxInflight: 1, syncGlobalQueueLimit: 0 }); + const blocker = deferred(); + + const blocking = admit(agent, { source: 'on-connect', label: 'blocker' }, () => blocker.promise); + await waitFor(() => getSyncBackpressureSnapshot().inflight === 1); + + let started = false; + await expect(admit(agent, { source: 'reconcile', label: 'rejected' }, async () => { + started = true; + })).rejects.toThrow(/backpressure/i); + expect(started).toBe(false); + + const rejections = await harness.matching(I5, { + lane: 'durable', source: 'reconcile', reason: 'queue_full', + }); + expect(rejections).toHaveLength(1); + expect(rejections[0]!.value).toBe(1); + // M10 emits a 0 ms duration sample here; the denominator must not grow. + const samples = await harness.histogram(I4); + expect(samples.filter((point) => point.attributes.source === 'reconcile')).toEqual([]); + + blocker.resolve(); + await blocking; + }); + + it('records `aborted_before_start` when a queued operation is cancelled', async () => { + harness.install(); + const agent = await createAgent({ syncGlobalMaxInflight: 1, syncGlobalQueueLimit: 2 }); + const blocker = deferred(); + const controller = new AbortController(); + + const blocking = admit(agent, { source: 'on-connect', label: 'blocker' }, () => blocker.promise); + await waitFor(() => getSyncBackpressureSnapshot().inflight === 1); + + let started = false; + const cancelled = admit( + agent, + { source: 'catchup-foreground', label: 'cancelled', operationSignal: controller.signal }, + async () => { started = true; }, + ); + await waitFor(() => getSyncBackpressureSnapshot().queued === 1); + controller.abort(); + await expect(cancelled).rejects.toBeTruthy(); + expect(started).toBe(false); + + expect(await harness.matching(I5, { + lane: 'durable', source: 'catchup-foreground', reason: 'aborted_before_start', + })).toHaveLength(1); + + blocker.resolve(); + await blocking; + }); + + it('separates a failed operation from a cancelled one, and clamps a non-requester lane', async () => { + harness.install(); + const agent = await createAgent(); + + await expect(admit(agent, { source: 'reconcile' }, async () => { + throw new Error('store commit failed'); + })).rejects.toThrow(/store commit/); + + const abortError = new Error('caller gave up'); + abortError.name = 'AbortError'; + await expect(admit(agent, { source: 'on-connect' }, async () => { + throw abortError; + })).rejects.toBe(abortError); + + // `responder` is a real SyncSchedulerLane member but not a requester lane; + // it must clamp rather than widen the I4 label space. + await admit(agent, { source: 'reconcile', lane: 'responder' }, async () => undefined); + + const samples = await harness.histogram(I4); + expect(samples.find((p) => p.attributes.source === 'reconcile' && p.attributes.lane === 'durable')! + .attributes.outcome).toBe('error'); + expect(samples.find((p) => p.attributes.source === 'on-connect')!.attributes.outcome).toBe('cancelled'); + expect(samples.some((p) => p.attributes.lane === 'unspecified')).toBe(true); + expect(samples.some((p) => p.attributes.lane === 'responder')).toBe(false); + }); + + it('A7: operation points carry only bounded labels — no CG id, no peer id', async () => { + harness.install(); + const agent = await createAgent({ syncGlobalMaxInflight: 1, syncGlobalQueueLimit: 0 }); + const blocker = deferred(); + const blocking = admit(agent, { source: 'on-connect' }, () => blocker.promise); + await waitFor(() => getSyncBackpressureSnapshot().inflight === 1); + await expect(admit(agent, { source: 'reconcile' }, async () => undefined)).rejects.toBeTruthy(); + blocker.resolve(); + await blocking; + + expect(await harness.attributeKeys(I4)).toEqual(['lane', 'outcome', 'source']); + expect(await harness.attributeKeys(I5)).toEqual(['lane', 'reason', 'source']); + }); +}); + +describe('W1 A8/M5 — `source` is in no coalescing key, at every scope', () => { + it('page scope: two differently-sourced identical fetches share ONE physical send', async () => { + harness.install(); + const response = deferred(); + let sends = 0; + const agent = await createAgent({ + sendToPeer: async () => { sends += 1; return response.promise; }, + }); + + const fetchPage = () => (agent as any).fetchSyncPages( + createOperationContext('sync'), PEER_A, CG, false, 'data', + `did:dkg:context-graph:${CG}`, DEFAULT_DEADLINE, + ); + const first = withSyncAdmissionSource('on-connect', fetchPage); + await flushMicrotasks(); + const second = withSyncAdmissionSource('reconcile', fetchPage); + await flushMicrotasks(); + + // M5 (source in the page key) forks this into two sends. + expect(sends).toBe(1); + response.resolve(new Uint8Array(0)); + const [a, b] = await Promise.all([first, second]); + expect(a).toBe(b); + + // M4 (drop/overwrite the owner source) collapses these two labels into one. + const joins = await harness.matching(I6, { + scope: 'page', owner_source: 'on-connect', joiner_source: 'reconcile', + }); + expect(joins).toHaveLength(1); + expect(joins[0]!.value).toBe(1); + expect(await harness.attributeKeys(I6)).toEqual(['joiner_source', 'owner_source', 'scope']); + }); + + it('context-graph scope: the source OVERRIDE does not fork the generic single-flight key', async () => { + harness.install(); + const agent = await createAgent(); + // The single-flight map is the subject here; the peer-discovery preamble in + // front of it needs a started libp2p node, which this unit agent has not. + (agent as any).isPrivateContextGraph = async () => false; + (agent as any).resolvePreferredSyncPeerId = async () => undefined; + (agent as any).primeCatchupConnections = async () => undefined; + (agent as any).node = { ...(agent as any).node, libp2p: { getConnections: () => [] } }; + let catchupRuns = 0; + const release = deferred(); + (agent as any).runCatchupOverPeers = async () => { + catchupRuns += 1; + await release.promise; + return { connectedPeers: 0, totalPeers: 0, selectedPeers: 0, syncCapablePeers: 0 }; + }; + + const first = (agent as any).syncContextGraphFromConnectedPeers(CG, { sourceOverride: 'vm-recovery' }); + await waitFor(() => catchupRuns === 1); + const second = (agent as any).syncContextGraphFromConnectedPeers(CG, {}); + await flushMicrotasks(); + + // M5 at the GENERIC scope: the override must not be part of the key. + expect(catchupRuns).toBe(1); + release.resolve(); + const [a, b] = await Promise.all([first, second]); + expect(a).toBe(b); + + expect(await harness.matching(I6, { + scope: 'context-graph', owner_source: 'vm-recovery', joiner_source: 'catchup-background', + })).toHaveLength(1); + }); + + it('durable scope: two differently-sourced durable syncs share one run', async () => { + harness.install(); + const agent = await createAgent(); + let fetchCalls = 0; + stubDurableSyncBody(agent, () => { fetchCalls += 1; }); + + const first = (agent as any).syncFromPeerDetailed( + PEER_A, [CG], undefined, undefined, undefined, { source: 'on-connect' }, + ); + const second = (agent as any).syncFromPeerDetailed( + PEER_A, [CG], undefined, undefined, undefined, { source: 'reconcile' }, + ); + const [a, b] = await Promise.all([first, second]); + + expect(a).toBe(b); + const singleRunFetches = fetchCalls; + expect(singleRunFetches).toBeGreaterThan(0); + + expect(await harness.matching(I6, { + scope: 'durable', owner_source: 'on-connect', joiner_source: 'reconcile', + })).toHaveLength(1); + // One admitted operation, not two — the coalesced pair shares a denominator entry. + const samples = await harness.histogram(I4); + expect(samples.filter((point) => point.attributes.lane === 'durable') + .reduce((sum, point) => sum + point.count, 0)).toBe(1); + }); + + it('shared-memory scope: two differently-sourced SWM syncs share one run', async () => { + harness.install(); + const agent = await createAgent(); + stubDurableSyncBody(agent); + // A precomputed plan removes the only `await` before the single-flight map, + // so the join is deterministic rather than a microtask-ordering race. + const sharedMemorySyncPlan = { + publicContextGraphIds: [CG], + privateRecoverFromCurator: [], + eligibleContextGraphIds: [CG], + }; + + const first = (agent as any).syncSharedMemoryFromPeerDetailed( + PEER_A, [CG], { source: 'on-connect', sharedMemorySyncPlan }, + ); + const second = (agent as any).syncSharedMemoryFromPeerDetailed( + PEER_A, [CG], { source: 'catchup-foreground', sharedMemorySyncPlan }, + ); + const [a, b] = await Promise.all([first, second]); + expect(a).toBe(b); + + expect(await harness.matching(I6, { + scope: 'shared-memory', owner_source: 'on-connect', joiner_source: 'catchup-foreground', + })).toHaveLength(1); + }); +}); + +describe('W1 A6/M6 — the changelog lane reports its own attempts AND bytes', () => { + it('records transport=changelog, phase=delta with both byte legs', async () => { + harness.install(); + const denial = encodeChangelogResponse({ kind: 'denied' }); + let requestByteLength = 0; + const agent = await createAgent({ + sendToPeer: async (..._args: unknown[]) => { + requestByteLength = (_args[2] as Uint8Array).byteLength; + return denial; + }, + }); + (agent as any).getOrCreateSyncVerifyWorker = () => ({ + parseAndFilter: async () => ({ quads: [], totalQuads: 0 }), + }); + + const result = await (agent as any).runChangelogSyncForCg( + createOperationContext('sync'), PEER_A, CG, + ); + expect(result.deniedPhases).toBe(1); + + const changelogLabels = { transport: 'changelog', plane: 'durable', phase: 'delta' }; + // M6 deletes this record entirely; the denominator then silently omits a + // whole lane, and the omission is uncorrectable after collection. + const attempts = await harness.matching(I1, { ...changelogLabels, outcome: 'response' }); + expect(attempts).toHaveLength(1); + expect(attempts[0]!.value).toBe(1); + + const requestBytes = await harness.matching(I2, changelogLabels); + expect(requestBytes).toHaveLength(1); + expect(requestBytes[0]!.value).toBe(requestByteLength); + expect(requestByteLength).toBeGreaterThan(0); + + // Byte accounting is the part this lane had none of before W1. + const responseBytes = await harness.matching(I3, { ...changelogLabels, outcome: 'response' }); + expect(responseBytes).toHaveLength(1); + expect(responseBytes[0]!.value).toBe(denial.byteLength); + }); + + it('labels a failed changelog send transport_error, keeping its request bytes', async () => { + harness.install(); + const agent = await createAgent({ + sendToPeer: async () => { throw new Error('all multiaddr dials failed'); }, + }); + (agent as any).getOrCreateSyncVerifyWorker = () => ({ + parseAndFilter: async () => ({ quads: [], totalQuads: 0 }), + }); + + await expect((agent as any).runChangelogSyncForCg( + createOperationContext('sync'), PEER_A, CG, + )).rejects.toThrow(/multiaddr/); + + expect(await harness.matching(I1, { + transport: 'changelog', phase: 'delta', outcome: 'transport_error', + })).toHaveLength(1); + expect(await harness.total(I2)).toBeGreaterThan(0); + expect(await harness.counter(I3)).toEqual([]); + }); + + it('inherits the ambient admission source, so both lanes share one denominator', async () => { + harness.install(); + const agent = await createAgent({ + sendToPeer: async () => encodeChangelogResponse({ kind: 'denied' }), + }); + (agent as any).getOrCreateSyncVerifyWorker = () => ({ + parseAndFilter: async () => ({ quads: [], totalQuads: 0 }), + }); + + await admit(agent, { source: 'catchup-foreground', lane: 'changelog' }, () => + (agent as any).runChangelogSyncForCg(createOperationContext('sync'), PEER_A, CG)); + + expect(await harness.matching(I1, { + transport: 'changelog', source: 'catchup-foreground', + })).toHaveLength(1); + }); +}); + +describe('W1 A15/M13 — VM recovery is labelled `vm-recovery`, not `catchup-background`', () => { + it('the override replaces the mode-derived source and leaves priority alone', async () => { + const seen: Array<{ priority?: number; source?: string }> = []; + const run = async (context: { priority?: number; source?: string }) => { + seen.push({ ...context }); + return {}; + }; + + await runCatchupPlaneWithPolicy('background', run); + await runCatchupPlaneWithPolicy('background', run, { sourceOverride: 'vm-recovery' }); + await runCatchupPlaneWithPolicy('foreground', run, { sourceOverride: 'vm-recovery' }); + // An out-of-set override is clamped, never smuggled through. + await runCatchupPlaneWithPolicy('background', run, { sourceOverride: 'vm-repair' as never }); + + expect(seen[0]).toEqual({ priority: undefined, source: 'catchup-background' }); + expect(seen[1]).toEqual({ priority: undefined, source: 'vm-recovery' }); + expect(seen[2]).toEqual({ priority: 2_000, source: 'vm-recovery' }); + expect(seen[3]).toEqual({ priority: undefined, source: 'unspecified' }); + }); + + it('M13: catch-up samples carry the EXACT label `vm-recovery` end to end', async () => { + harness.install(); + const agent = await createAgent(); + stubDurableSyncBody(agent); + (agent as any).waitForSyncProtocol = async () => true; + + await (agent as any).runCatchupOverPeers(CG, false, [{ toString: () => PEER_A }], { + totalPeers: 1, + mode: 'background', + sourceOverride: 'vm-recovery', + }); + + const samples = await harness.histogram(I4); + const sources = samples.map((point) => point.attributes.source); + // Asserting "not recurring" would NOT kill M13 — `catchup-background` is + // excluded from the recurring family too, so the exact label is the test. + expect(sources).toContain('vm-recovery'); + expect(sources).not.toContain('catchup-background'); + expect(samples.find((point) => point.attributes.source === 'vm-recovery')!.count) + .toBeGreaterThanOrEqual(1); + expect(samples.every((point) => point.attributes.source === 'vm-recovery')).toBe(true); + }); + + it('without the override the same path stays `catchup-background`', async () => { + harness.install(); + const agent = await createAgent(); + stubDurableSyncBody(agent); + (agent as any).waitForSyncProtocol = async () => true; + + await (agent as any).runCatchupOverPeers(CG, false, [{ toString: () => PEER_A }], { + totalPeers: 1, + mode: 'background', + }); + + const sources = (await harness.histogram(I4)).map((point) => point.attributes.source); + expect(sources).toContain('catchup-background'); + expect(sources).not.toContain('vm-recovery'); + }); +}); diff --git a/packages/agent/vitest.unit.config.ts b/packages/agent/vitest.unit.config.ts index 51776d5c88..b21f1c24e5 100644 --- a/packages/agent/vitest.unit.config.ts +++ b/packages/agent/vitest.unit.config.ts @@ -76,6 +76,9 @@ export default defineConfig({ "test/sync-append-in-place.test.ts", "test/sync-memory-metrics.test.ts", "test/sync-responder-metrics.test.ts", + "test/sync-transport-metrics.test.ts", + "test/sync-attempt-telemetry.test.ts", + "test/sync-operation-telemetry.test.ts", "test/sync-fetch-coalescing.test.ts", "test/sync-fetch-coalescing-durable.test.ts", "test/sync-backpressure.test.ts", diff --git a/packages/cli/src/daemon/catchup-telemetry.ts b/packages/cli/src/daemon/catchup-telemetry.ts new file mode 100644 index 0000000000..e4fc9c832b --- /dev/null +++ b/packages/cli/src/daemon/catchup-telemetry.ts @@ -0,0 +1,235 @@ +// daemon/catchup-telemetry.ts +// +// Catch-up request/job accounting (W1 instruments I7–I9) plus the job ledger +// the graceful-shutdown path drains. +// +// Lives in its own module — rather than on `CatchupTracker` — because two +// modules that never import each other need the SAME object: the subscribe +// route in `routes/context-graph.ts` mints jobs, and `lifecycle.ts` drains +// them during shutdown. That is the same reason `state.ts` exists, and it +// keeps `CatchupTracker` a plain data bag that route tests can build by hand. +// +// Two invariants govern everything here: +// +// 1. **Requests are not jobs.** They are N:1 — a dedupe returns a running +// job, an already-ready replay returns a completed one, and `400`/`403` +// plus both shutdown `503`s mint nothing at all. I7 counts route returns; +// I8 counts job identities. +// 2. **Instrumentation never throws.** Every record site is synchronous and +// swallowed. The walk job's terminal point is emitted from a detached, +// `void`ed continuation, where an escaping throw would become an +// unhandled rejection rather than a failed metric. + +import { getMetrics } from '@origintrail-official/dkg-core'; +import type { CatchupJob, CatchupJobState } from './types.js'; + +/** + * Grace period the shutdown drain gives in-flight walks to settle normally. + * + * Sized against `SHUTDOWN_HARD_TIMEOUT_MS` (15 s): 5 s here plus the 2 s + * terminal-flush reserve leaves ~8 s for runner close, `agent.stop()`, the + * final telemetry shutdown and the database close. + */ +export const CATCHUP_SHUTDOWN_DRAIN_BUDGET_MS = 5_000; + +/** Which of the two mint sites produced a job. */ +export type CatchupAdmission = 'walk' | 'synthetic'; + +/** + * Closed `result` vocabulary for I7 — one value per subscribe-route return. + * `shutting_down` covers BOTH admission guards; they are distinguishable by + * `admission` on the job side only when a job exists, which for a 503 it never + * does, and that is the point. + */ +export type CatchupRequestResult = + | 'bad_request' + | 'forbidden' + | 'deduped' + | 'ready_replay' + | 'ready_synthetic' + | 'queued' + | 'shutting_down'; + +/** + * A job whose terminal record is still owed, plus the continuation that will + * produce it. + * + * `terminalRecorded` lives on THIS object rather than being derived from + * `catchupTracker.jobs`, and the difference is load-bearing: that map prunes + * to 100 entries by oldest `queuedAt` regardless of status, so a long-running + * job can be evicted while still in flight. Keying idempotency off it would + * let a job be counted twice, or not at all. + */ +export interface CatchupJobLedgerEntry { + readonly job: CatchupJob; + readonly admission: CatchupAdmission; + /** + * Monotonic start for I9. `Date.now()` deltas can go negative across an NTP + * step, and catch-up walks are long enough (measured runs of 305 s and + * 382 s) for that to be a real sample. + */ + readonly startedAtMono: number; + /** Retained continuation; the shutdown drain awaits it. Never rejects. */ + task?: Promise; + terminalRecorded: boolean; +} + +/** + * Live walk jobs, keyed by jobId. Synthetic jobs are born terminal and never + * enter it — there is nothing to drain and their record is emitted at once. + */ +const ledger = new Map(); + +/** Terminal states a job can settle in; `queued`/`running` are not terminal. */ +const TERMINAL_STATES: ReadonlySet = new Set([ + 'done', + 'failed', + 'denied', + 'deferred', + 'unreachable', +]); + +/** + * A job that never reached a terminal state is reported as `failed`. + * + * Reached when the shutdown drain expires with a walk still in flight. It is + * the truthful label: the daemon is about to terminate the worker, whose exit + * handler rejects every pending run, so the job's own `finally` — if it ever + * runs — would have written `failed` anyway. Clamping here keeps `status` a + * closed vocabulary instead of leaking `running` into the series. + */ +function terminalStatusFor(job: CatchupJob): CatchupJobState { + return TERMINAL_STATES.has(job.status) ? job.status : 'failed'; +} + +/** I7 — one point per subscribe-route return. Never throws. */ +export function recordCatchupRequest( + result: CatchupRequestResult, + includeSharedMemory: boolean, +): void { + try { + getMetrics().contextGraphCatchupRequestsTotal.add(1, { + result, + include_shared_memory: includeSharedMemory, + }); + } catch { + /* instrumentation must never break the route */ + } +} + +/** + * I8 (+ I9 for walks) — at most one point per jobId, ever. + * + * Synchronous and non-throwing by construction, because the walk call site is + * a detached task's `finally`. The idempotency bit is set BEFORE the record so + * that even a throw inside the metric API cannot produce a second point. + */ +export function recordTerminalOnce(entry: CatchupJobLedgerEntry): void { + if (entry.terminalRecorded) return; + entry.terminalRecorded = true; + try { + const metrics = getMetrics(); + metrics.contextGraphCatchupJobsTotal.add(1, { + status: terminalStatusFor(entry.job), + admission: entry.admission, + }); + // I9 is walk-only. A synthetic job has queuedAt = startedAt = finishedAt + // and never ran anything, so a 0 ms sample would only dilute the histogram + // the catch-up buckets exist to resolve. + if (entry.admission === 'walk') { + metrics.contextGraphCatchupJobDurationMs.record( + Math.max(0, performance.now() - entry.startedAtMono), + { admission: entry.admission }, + ); + } + } catch { + /* instrumentation must never break shutdown or a catch-up job */ + } +} + +/** + * Register a walk job and return its ledger entry. The caller assigns + * `entry.task` once the continuation exists, and must pass the ENTRY (not the + * jobId) into that closure, so releasing the map slot cannot erase the + * idempotency bit before a late `finally` runs. + */ +export function beginWalkCatchupJob(job: CatchupJob): CatchupJobLedgerEntry { + const entry: CatchupJobLedgerEntry = { + job, + admission: 'walk', + startedAtMono: performance.now(), + terminalRecorded: false, + }; + ledger.set(job.jobId, entry); + return entry; +} + +/** + * Drop a settled walk job from the drain set. Deletes only if the slot still + * holds THIS entry, so a re-used id can never evict a newer job. + */ +export function releaseCatchupJob(entry: CatchupJobLedgerEntry): void { + if (ledger.get(entry.job.jobId) === entry) ledger.delete(entry.job.jobId); +} + +/** I8 for the already-ready mint: born terminal, recorded immediately. */ +export function recordSyntheticCatchupJob(job: CatchupJob): void { + recordTerminalOnce({ + job, + admission: 'synthetic', + startedAtMono: performance.now(), + terminalRecorded: false, + }); +} + +/** + * Grace drain for graceful shutdown: let in-flight walks settle normally, + * then record a terminal point for every job still owed one. + * + * MUST run while the catch-up runner's worker is still alive. `close()` IS + * `worker.terminate()`, and the constructor-registered `exit` handler rejects + * every pending run — so draining after termination grants no grace at all, it + * merely observes every job being forced onto its `failed` path. + * + * Bounded by `budgetMs`: on expiry the still-running jobs are recorded as + * `failed` and their late `finally` becomes a no-op, so a hung walk cannot eat + * the shutdown deadline. + */ +export async function drainCatchupJobs( + budgetMs: number, + log: (message: string) => void = () => {}, +): Promise<{ drained: number; expired: boolean }> { + const entries = [...ledger.values()]; + const tasks = entries.flatMap((entry) => (entry.task ? [entry.task] : [])); + let expired = false; + if (tasks.length > 0) { + let timer: ReturnType | undefined; + const budget = new Promise<'expired'>((resolve) => { + timer = setTimeout(() => resolve('expired'), budgetMs); + }); + try { + expired = (await Promise.race([Promise.allSettled(tasks), budget])) === 'expired'; + } finally { + if (timer) clearTimeout(timer); + } + if (expired) { + log( + `[catchup-drain] ${tasks.length} catch-up job(s) did not settle within ${budgetMs}ms; ` + + 'recording them as failed and continuing shutdown.', + ); + } + } + for (const entry of entries) recordTerminalOnce(entry); + ledger.clear(); + return { drained: entries.length, expired }; +} + +/** Test seam: the ledger is process-global, like `daemonState`. */ +export function resetCatchupJobLedger(): void { + ledger.clear(); +} + +/** Test seam: current in-flight walk count. */ +export function catchupLedgerSize(): number { + return ledger.size; +} diff --git a/packages/cli/src/daemon/lifecycle.ts b/packages/cli/src/daemon/lifecycle.ts index b0fae05696..85755ed302 100644 --- a/packages/cli/src/daemon/lifecycle.ts +++ b/packages/cli/src/daemon/lifecycle.ts @@ -90,6 +90,7 @@ import { OtlpLogWorker, initTelemetry, shutdownTelemetry, + flushTelemetry, LlmClient, SqliteMessageIdempotencyStore, SqliteProtocolOutboxStore, @@ -213,6 +214,11 @@ import { type CatchupTracker, toCatchupStatusResponse, } from './types.js'; +import { drainCatchupJobs } from './catchup-telemetry.js'; +import { + buildProducerQuiescentTeardownSteps, + runProducerQuiescentTeardown, +} from './teardown.js'; import { type MarkItDownTarget, manifestRepoRoot, @@ -3738,6 +3744,12 @@ export async function runDaemonInner( async function shutdown(exitCode = 0) { if (shuttingDown) return; shuttingDown = true; + // FIRST statement, ahead of every await below: the subscribe route has no + // other way to see that shutdown started (`shuttingDown` is a closure-local + // `let` in this function's scope), and a job minted after this point would + // be queued against a runner whose worker is about to be terminated — its + // exit handler rejects every pending run. Both mint sites now 503 instead. + daemonState.catchupAcceptingJobs = false; log("Shutting down..."); // Tell the supervisor's liveness watcher (PR #664) that this is a graceful // shutdown before any slow cleanup runs. The watcher reads `api.port`'s @@ -3770,28 +3782,47 @@ export async function runDaemonInner( rpcUsageTelemetry.stop(); rateLimiter.destroy(); metricsCollector?.stop(); - // Stops log exporters AND flushes + shuts down the OTel SDK. - await stopTelemetry(); natStatusWatcherStop?.(); resetNatStatus(); - await publisherState.runtime - ?.stop() - .catch((err: any) => - log(`Publisher runtime stop error: ${err?.message ?? String(err)}`), - ); - // Drain the async-promote worker before closing the agent — once - // `agent.stop()` runs the queue's underlying triple store goes - // away. We let in-flight promotes complete (or hit - // `shutdownTimeoutMs`); RFC §6.2 forbids marking `running → - // queued` here so the next boot's `recoverOnStartup()` decides. - await promoteWorkerLifecycle?.stop(shuttingDown ? 'daemon shutting down' : null); - await daemonState.catchupRunner - ?.close() - .catch((err: any) => - log(`Catch-up runner stop error: ${err?.message ?? String(err)}`), - ); - server.close(); - await agent.stop(); + + // ── Producer-quiescent teardown ──────────────────────────────────── + // Both the ORDER and the WIRING live in `./teardown.ts` — the order in + // `runProducerQuiescentTeardown`, the slot assignment in + // `buildProducerQuiescentTeardownSteps` — so a test can execute each + // and fail on either a reorder or a mis-wiring. This call site only + // names the daemon's own resources. + await runProducerQuiescentTeardown( + buildProducerQuiescentTeardownSteps({ + server, + drainCatchupJobs, + flushTelemetry, + stopPublisherRuntime: async () => { + await publisherState.runtime + ?.stop() + .catch((err: any) => + log(`Publisher runtime stop error: ${err?.message ?? String(err)}`), + ); + }, + // We let in-flight promotes complete (or hit `shutdownTimeoutMs`); + // RFC §6.2 forbids marking `running → queued` here so the next + // boot's `recoverOnStartup()` decides. + stopPromoteWorker: async () => { + await promoteWorkerLifecycle?.stop(shuttingDown ? 'daemon shutting down' : null); + }, + closeCatchupRunner: async () => { + await daemonState.catchupRunner + ?.close() + .catch((err: any) => + log(`Catch-up runner stop error: ${err?.message ?? String(err)}`), + ); + }, + stopAgent: () => agent.stop(), + // Stops log exporters AND flushes + shuts down the OTel SDK. + stopTelemetry, + log, + }), + ); + // Stop the managed Oxigraph child AFTER the agent has stopped // issuing store queries, so an in-flight SPARQL request never // races the killed server. No-op when not using oxigraph-server. diff --git a/packages/cli/src/daemon/routes/context-graph.ts b/packages/cli/src/daemon/routes/context-graph.ts index 947e27e15e..c2360d6666 100644 --- a/packages/cli/src/daemon/routes/context-graph.ts +++ b/packages/cli/src/daemon/routes/context-graph.ts @@ -170,6 +170,13 @@ import { type CatchupTracker, toCatchupStatusResponse, } from '../types.js'; +import { + beginWalkCatchupJob, + recordCatchupRequest, + recordSyntheticCatchupJob, + recordTerminalOnce, + releaseCatchupJob, +} from '../catchup-telemetry.js'; import { type MarkItDownTarget, manifestRepoRoot, @@ -435,6 +442,31 @@ function respondReconcileError(res: ServerResponse, err: unknown): void { return jsonResponse(res, 500, { error: message }); } +/** + * Refuse to mint a new catch-up job because the daemon is shutting down. + * + * Shaped after `respondIfApiQueryStoreBusy` — retryable 503 plus `Retry-After` + * — because that is what this is: the request is fine, the node just cannot + * take on new work it will never drain. Returned from BOTH mint sites, which + * is why I7's `result` vocabulary needed a seventh value; a 503 that clamped + * to `unspecified` would hide the one route outcome shutdown introduces. + */ +function catchupShuttingDownResponse(res: ServerResponse, includeSharedMemory: boolean): void { + recordCatchupRequest('shutting_down', includeSharedMemory); + return jsonResponse( + res, + 503, + { + error: + 'Node is shutting down and is no longer accepting catch-up jobs; retry once it is back up.', + code: 'CATCHUP_SHUTTING_DOWN', + retryable: true, + }, + undefined, + { 'Retry-After': '5' }, + ); +} + async function handleReconcileContextGraphRoute( ctx: Pick, isNodeAdminCaller: boolean, @@ -1685,12 +1717,18 @@ export async function handleContextGraphRoutes(ctx: RequestContext): Promise a.toLowerCase() === callerAddr.toLowerCase())) { + recordCatchupRequest('forbidden', shouldSyncSharedMemory); return jsonResponse(res, 403, { error: `Your agent (${callerAddr}) is not on the allowlist for this curated project. Ask the curator to invite you first.`, }); @@ -1719,9 +1758,6 @@ export async function handleContextGraphRoutes(ctx: RequestContext): Promise