From 118c63df7abcd7917843ac5043927575753db0c3 Mon Sep 17 00:00:00 2001 From: Nader Helmy Date: Wed, 29 Jul 2026 03:19:40 -0500 Subject: [PATCH] feat: expose graph-node stats and health-check it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit graph-node had no health, stats, or metrics endpoint, and no Fly health check. So when the process entered a crash loop on 2026-07-29, Fly had no liveness signal at all: it kept routing to a dead instance and returning 502s, and the only thing that noticed was a post-deploy canary in a different job. The one number that would have predicted the outage, heap headroom, existed solely in Fly logs. Adds GET /v1/stats: record_count, context_count, heap used/limit/percent, uptime, and the last archive replay. Shape mirrors log-node's /v1/stats, which is likewise operational rather than normative, so no spec section: §3.4 is the graph query interface and this must never start returning graph facts. Kept O(1) and tested for it. The Fly check polls it on an interval, so if it ever walked the store it would become a periodic full scan of the thing it exists to protect, which is close to the original bug. Adds the Fly health check pointing at it. timeout is 10s rather than something near the ~0.15s a healthy response takes, because the failure worth catching is a wedged process, not a slow one: during the outage requests took 17-25s while the process was still listening and spending 65-75% of wall-clock in GC. grace_period is 60s because Fly clamps it there. `flyctl config validate` rejects more with "this will be lowered to 1 minute", so a larger value reads as protection that does not exist. That clamp bounds the design, not just the file: main.ts replays the whole archive before binding the port, and at the measured ~20.6k records/sec that gives about 1.24M records of runway. Machine sizing hits that before it hits a memory wall, so raising [[vm]] memory stops working past roughly the 4GB tier for startup reasons rather than heap reasons. Recorded in P059. Adds check-graph-smoke.mjs to the graph-canary job, before the canary because it has no dependencies and fails in seconds with a specific message where the canary would spend its full 90s deadline. It fails on contract violations, on a slow response, and on heap past 95%, that last one high enough not to block the deploy that fixes a memory problem. Runway is reported in MB always and in records only above 10k, since below that Node's fixed baseline dominates and per-record attribution reports nonsense. Kept separate from check-log-smoke.mjs on purpose: folding graph checks into it would re-couple a log-node deploy to graph-node health, which is the coupling the canary was moved out of deploy-log-node to remove. --- .github/scripts/check-graph-smoke.mjs | 214 ++++++++++++++++++++++++ .github/workflows/deploy-services.yml | 8 + DECISIONS.md | 20 +++ services/graph-node/fly.toml | 38 +++++ services/graph-node/src/main.ts | 23 ++- services/graph-node/src/server.ts | 72 +++++++- services/graph-node/src/store.ts | 9 + services/graph-node/test/server.test.ts | 1 + services/graph-node/test/stats.test.ts | 167 ++++++++++++++++++ 9 files changed, 544 insertions(+), 8 deletions(-) create mode 100644 .github/scripts/check-graph-smoke.mjs create mode 100644 services/graph-node/test/stats.test.ts diff --git a/.github/scripts/check-graph-smoke.mjs b/.github/scripts/check-graph-smoke.mjs new file mode 100644 index 000000000..8adc41ea2 --- /dev/null +++ b/.github/scripts/check-graph-smoke.mjs @@ -0,0 +1,214 @@ +// SPDX-License-Identifier: Apache-2.0 + +/* eslint-disable no-console -- + * This is a CLI check script: stdout IS its output surface, and the run log is + * where an operator reads the heap headroom. Disabled file-wide rather than + * raising the repo's `--max-warnings 809` ratchet, which exists to stop warning + * counts drifting up. + */ + +/** + * Post-deploy smoke check for graph-node's /v1/stats. + * + * Two things it is here to catch: + * + * 1. Contract regressions. /v1/stats is the Fly health-check target, so if its + * shape drifts the liveness signal silently degrades with it. + * + * 2. The runway. graph-node holds every record in memory, so heap headroom is + * how much room the current machine size has left. Before 2026-07-29 that + * number existed only in Fly logs; the live set reached the V8 ceiling and + * the process began aborting with nothing watching. See P059 in + * DECISIONS.md. + * + * Deliberately separate from check-log-smoke.mjs. That script runs in the + * log-node deploy job, and folding graph checks into it would re-couple a + * log-node deploy to graph-node's health, which is the coupling the graph + * canary was moved out of deploy-log-node to remove. + * + * Failure policy: contract violations always fail. Heap only fails past + * GRAPH_SMOKE_FAIL_HEAP_PCT (default 95), high enough that it will not block + * the very deploy that fixes a memory problem, while still refusing to report + * green on a service that is about to abort. Warn/error thresholds below mirror + * the in-process watchdog in services/graph-node/src/heap-watchdog.ts. + * + * Env: + * GRAPH_SMOKE_ORIGIN default https://graph.atrib.dev + * GRAPH_SMOKE_MAX_TOTAL_MS default 5000 + * GRAPH_SMOKE_FETCH_TIMEOUT_MS default 15000 + * GRAPH_SMOKE_ATTEMPTS default 3 + * GRAPH_SMOKE_RETRY_DELAY_MS default 2000 + * GRAPH_SMOKE_FAIL_HEAP_PCT default 95 + */ + +const origin = (process.env.GRAPH_SMOKE_ORIGIN ?? 'https://graph.atrib.dev').replace(/\/$/, '') +const maxTotalMs = readPositiveInt('GRAPH_SMOKE_MAX_TOTAL_MS', 5000) +const timeoutMs = readPositiveInt('GRAPH_SMOKE_FETCH_TIMEOUT_MS', 15000) +const maxAttempts = readPositiveInt('GRAPH_SMOKE_ATTEMPTS', 3) +const retryDelayMs = readPositiveInt('GRAPH_SMOKE_RETRY_DELAY_MS', 2000) +const failHeapPct = readPositiveInt('GRAPH_SMOKE_FAIL_HEAP_PCT', 95) + +// Mirror services/graph-node/src/heap-watchdog.ts. +const WARN_PCT = 70 +const ERROR_PCT = 85 + +function readPositiveInt(name, fallback) { + const raw = process.env[name] + if (raw === undefined || raw === '') return fallback + const parsed = Number(raw) + if (!Number.isInteger(parsed) || parsed <= 0) { + throw new Error(`${name} must be a positive integer, got ${JSON.stringify(raw)}`) + } + return parsed +} + +function sleep(ms) { + return new Promise((resolve) => setTimeout(resolve, ms)) +} + +async function fetchStatsOnce(attempt) { + const started = Date.now() + const controller = new AbortController() + const timer = setTimeout(() => controller.abort(), timeoutMs) + try { + const response = await fetch(`${origin}/v1/stats`, { signal: controller.signal }) + const body = await response.text() + const totalMs = Date.now() - started + if (!response.ok) { + throw new Error(`attempt ${attempt}: /v1/stats returned ${response.status}`) + } + return { body, totalMs } + } finally { + clearTimeout(timer) + } +} + +async function fetchStats() { + let lastError + for (let attempt = 1; attempt <= maxAttempts; attempt++) { + try { + return await fetchStatsOnce(attempt) + } catch (err) { + lastError = err + if (attempt < maxAttempts) await sleep(retryDelayMs) + } + } + throw lastError +} + +/** Contract checks. These are regressions, so they always fail the run. */ +function validateShape(parsed) { + const problems = [] + if (parsed.service !== 'atrib-graph-node') { + problems.push(`service was ${JSON.stringify(parsed.service)}`) + } + for (const field of ['record_count', 'context_count', 'uptime_s']) { + if (!Number.isInteger(parsed[field]) || parsed[field] < 0) { + problems.push(`${field} was ${JSON.stringify(parsed[field])}`) + } + } + const heap = parsed.heap + if (typeof heap !== 'object' || heap === null) { + problems.push('heap block missing') + return problems + } + for (const field of ['used_mb', 'limit_mb', 'used_pct']) { + if (typeof heap[field] !== 'number' || !Number.isFinite(heap[field])) { + problems.push(`heap.${field} was ${JSON.stringify(heap[field])}`) + } + } + if (typeof heap.used_mb === 'number' && typeof heap.limit_mb === 'number') { + if (heap.limit_mb <= 0) problems.push(`heap.limit_mb was ${heap.limit_mb}`) + if (heap.used_mb > heap.limit_mb) { + problems.push(`heap.used_mb ${heap.used_mb} exceeds limit_mb ${heap.limit_mb}`) + } + } + // A deployed graph-node with zero records means the archive did not replay, + // which is data loss rather than a fresh start. + if (parsed.record_count === 0) { + problems.push('record_count is 0; the archive did not replay') + } + return problems +} + +// Below this, per-record heap attribution is meaningless: Node's own baseline +// (~25MB) dominates a small store, so dividing total heap by record count +// reports tens of KB per record and a wildly wrong headroom. Headroom in MB is +// exact at any size, so that is always reported; the record-denominated view is +// gated on having enough records for the baseline to be noise. +const MIN_RECORDS_FOR_PER_RECORD_ESTIMATE = 10_000 + +function reportRunway(parsed) { + const { used_mb: used, limit_mb: limit, used_pct: pct } = parsed.heap + const records = parsed.record_count + const toMb = (targetPct) => Math.max(0, Math.round(limit * (targetPct / 100) - used)) + + console.log(`graph-node /v1/stats @ ${origin}`) + console.log(` records ${records.toLocaleString()} across ${parsed.context_count.toLocaleString()} contexts`) + console.log(` heap ${used}MB / ${limit}MB (${pct}%)`) + console.log(` headroom ${toMb(WARN_PCT)}MB to ${WARN_PCT}% warn, ${toMb(ERROR_PCT)}MB to ${ERROR_PCT}% error`) + + if (records >= MIN_RECORDS_FOR_PER_RECORD_ESTIMATE) { + // Attributes all heap to records, so it absorbs Node's fixed baseline and + // slightly overstates cost per record. That biases the headroom estimate + // low, which is the safe direction for a runway number. Divide by your own + // records/day to get days. + const kbPerRecord = (used * 1024) / records + const toRecords = (targetPct) => Math.max(0, Math.round((toMb(targetPct) * 1024) / kbPerRecord)) + console.log(` per record ~${kbPerRecord.toFixed(2)} KB (incl. fixed overhead, so conservative)`) + console.log(` runway ~${toRecords(WARN_PCT).toLocaleString()} records to warn, ~${toRecords(ERROR_PCT).toLocaleString()} to error`) + } else { + console.log(` runway not estimated below ${MIN_RECORDS_FOR_PER_RECORD_ESTIMATE.toLocaleString()} records (fixed overhead dominates)`) + } + + console.log(` uptime ${parsed.uptime_s}s`) + if (parsed.replay) { + console.log(` last replay ${parsed.replay.records.toLocaleString()} records in ${parsed.replay.ms}ms`) + } +} + +async function main() { + const { body, totalMs } = await fetchStats() + + let parsed + try { + parsed = JSON.parse(body) + } catch { + throw new Error(`/v1/stats did not return JSON: ${body.slice(0, 200)}`) + } + + const problems = validateShape(parsed) + if (problems.length > 0) { + throw new Error(`/v1/stats contract violations:\n - ${problems.join('\n - ')}`) + } + + reportRunway(parsed) + + // A slow /v1/stats is itself the signal. This endpoint is O(1); when the + // process is thrashing GC it stays up but stops answering promptly, which is + // exactly how the 2026-07-29 outage presented. + if (totalMs > maxTotalMs) { + throw new Error(`/v1/stats took ${totalMs}ms, over the ${maxTotalMs}ms budget`) + } + + const pct = parsed.heap.used_pct + if (pct >= failHeapPct) { + throw new Error( + `heap at ${pct}% is at or over the ${failHeapPct}% fail threshold. ` + + `Raise NODE_OPTIONS --max-old-space-size AND fly.toml [[vm]] memory together, ` + + `or act on P059. Override for a remediation deploy with GRAPH_SMOKE_FAIL_HEAP_PCT.`, + ) + } + if (pct >= ERROR_PCT) { + console.error(`::error::graph-node heap at ${pct}%, over the ${ERROR_PCT}% error threshold. See P059.`) + } else if (pct >= WARN_PCT) { + console.warn(`::warning::graph-node heap at ${pct}%, over the ${WARN_PCT}% warn threshold. See P059.`) + } + + console.log(`ok: /v1/stats healthy in ${totalMs}ms`) +} + +main().catch((err) => { + console.error(err instanceof Error ? err.message : String(err)) + process.exit(1) +}) diff --git a/.github/workflows/deploy-services.yml b/.github/workflows/deploy-services.yml index 9a98c5412..3a118e1df 100644 --- a/.github/workflows/deploy-services.yml +++ b/.github/workflows/deploy-services.yml @@ -224,6 +224,14 @@ jobs: node-version: '24' cache: pnpm + # Runs before the install: it has no dependencies, and it fails in + # seconds with a specific message when graph-node is wedged, where the + # canary below would spend its whole 90s deadline discovering the same + # thing less clearly. + - name: Smoke test graph-node stats and heap headroom + shell: bash + run: node .github/scripts/check-graph-smoke.mjs + - name: Install canary dependencies run: pnpm install --frozen-lockfile diff --git a/DECISIONS.md b/DECISIONS.md index 10c64e5b9..f0d69bf50 100644 --- a/DECISIONS.md +++ b/DECISIONS.md @@ -10345,6 +10345,26 @@ continuing to raise the ceiling with machine size. The first three change the service's operational shape; the fourth is bounded by machine cost and only ever defers the same failure. +**Raising machine size runs out before the heap does (2026-07-29).** Cold-start +archive replay is the binding constraint, not memory. main.ts replays the whole +archive before binding the port, and Fly clamps an `http_service` health check's +`grace_period` to 60 seconds. At the measured ~20.6k records/sec replay rate that +covers about 1.24M records. Sizing hits that first: + +| machine | heap | capacity @85% | replay | +| ------- | ---- | ------------- | ------ | +| 1GB | 768MB | 214k | ~10s | +| 2GB | 1536MB | 427k | ~21s | +| 4GB | 3072MB | 854k | ~41s | +| 8GB | 6144MB | 1.7M | ~83s, past the 60s cap | + +So the scaling answer expires somewhere past the 4GB tier for startup reasons +rather than heap reasons, and every doubling also doubles the 502-serving window +on every deploy and every restart. Two further properties that memory does not +buy: full mark-compact pauses grow with the live object graph (during the outage +the process spent 65-76% of wall-clock in GC), and `min_machines_running = 1` +means each restart is a whole-service outage regardless of machine size. + **Not urgent as a correctness matter, and not open-ended either.** Whatever is chosen must preserve two properties the current design gets for free: [§3.2.4](atrib-spec.md#324-edge-derivation-rules) edge derivation stays deterministic over the full record set, and the [§1.9](atrib-spec.md#19-key-rotation-and-revocation) diff --git a/services/graph-node/fly.toml b/services/graph-node/fly.toml index e82a03a30..5e39cada5 100644 --- a/services/graph-node/fly.toml +++ b/services/graph-node/fly.toml @@ -34,6 +34,44 @@ primary_region = 'iad' auto_start_machines = true min_machines_running = 1 +# Liveness. Until 2026-07-29 this service had no health check at all, so when +# the process entered a crash loop Fly had no signal: it kept routing to a dead +# instance and returning 502s, and the only thing that noticed was a post-deploy +# canary in another job. +# +# timeout is deliberately far above the ~0.15s a healthy /v1/stats takes. The +# failure mode worth catching is a wedged process, not a slow one: during the +# outage the event loop was spending ~65-75% of wall-clock in GC and requests +# took 17-25s while the process was still listening. 10s separates that from +# ordinary GC jitter without flapping. +# +# grace_period MUST exceed archive replay. main.ts replays the whole record +# archive BEFORE binding the port, so nothing is listening until it finishes: +# ~5s at 112k records, and replay time grows linearly with record count. +# +# 60s is the ceiling, not a preference. Fly silently clamps this value to one +# minute ("grace period greater than 1 minute; this will be lowered to 1 +# minute" from `flyctl config validate`), so asking for more is a no-op that +# reads as protection you do not have. +# +# That clamp is a real bound on this service's design, not just on this file. +# At the measured ~20.6k records/sec replay rate, 60s of grace covers about +# 1.24M records. Machine sizing hits it before it hits a memory wall: +# +# 1GB 214k records ~10s replay +# 2GB 427k records ~21s replay +# 4GB 854k records ~41s replay +# 8GB 1.7M records ~83s replay <- exceeds the 60s cap +# +# So "keep raising memory" stops working somewhere past the 4GB tier, for +# startup reasons rather than heap reasons. Recorded in P059. +[[http_service.checks]] + interval = '30s' + timeout = '10s' + grace_period = '60s' + method = 'GET' + path = '/v1/stats' + [[vm]] size = 'shared-cpu-1x' # Bumped from 256mb after observed OOM-kill on 2026-05-06 with ~1500 records diff --git a/services/graph-node/src/main.ts b/services/graph-node/src/main.ts index 0a88ee733..b383434b0 100644 --- a/services/graph-node/src/main.ts +++ b/services/graph-node/src/main.ts @@ -22,7 +22,7 @@ * local + automatic. */ -import { bindGraphServer } from './server.js' +import { bindGraphServer, type ServiceRuntimeInfo } from './server.js' import { createRecordStore } from './store.js' import { createArchiveAppender, replayArchive } from './persistence.js' import { statfs } from 'node:fs/promises' @@ -45,6 +45,12 @@ const store = createRecordStore() let appender: Awaited> | undefined +// Replay facts for /v1/stats. Replay happens here, before the server binds, so +// the server cannot measure it itself. Left undefined when no archive is set, +// which /v1/stats reports by omitting the `replay` block entirely. +let replayMs: number | undefined +let replayedRecords: number | undefined + if (archivePath) { // eslint-disable-next-line no-console console.log(`atrib-graph: archive enabled at ${archivePath}`) @@ -52,7 +58,8 @@ if (archivePath) { const result = await replayArchive(archivePath, (record, logIndex) => store.addRecord(record, logIndex), ) - const replayMs = Date.now() - replayStart + replayMs = Date.now() - replayStart + replayedRecords = result.ingested // eslint-disable-next-line no-console console.log( `atrib-graph: replayed ${result.ingested}/${result.total} records ` + @@ -64,12 +71,16 @@ if (archivePath) { appender = await createArchiveAppender(archivePath) } -// Conditionally include onRecordIngested, TypeScript strict -// (`exactOptionalPropertyTypes`) rejects setting an optional property +// Conditionally include onRecordIngested and the replay facts, TypeScript +// strict (`exactOptionalPropertyTypes`) rejects setting an optional property // to `undefined` directly. +const runtimeInfo: ServiceRuntimeInfo = + typeof replayMs === 'number' + ? { replay_ms: replayMs, replayed_records: replayedRecords ?? 0 } + : {} const bindOpts = appender - ? { store, onRecordIngested: (record: AtribRecord, logIndex: number | undefined) => appender!.append(record, logIndex) } - : { store } + ? { store, runtimeInfo, onRecordIngested: (record: AtribRecord, logIndex: number | undefined) => appender!.append(record, logIndex) } + : { store, runtimeInfo } const server = await bindGraphServer(port, host, bindOpts) // eslint-disable-next-line no-console diff --git a/services/graph-node/src/server.ts b/services/graph-node/src/server.ts index ddaf57459..bc4c228ad 100644 --- a/services/graph-node/src/server.ts +++ b/services/graph-node/src/server.ts @@ -3,7 +3,7 @@ /** * Graph query service HTTP server (section 3.4). * - * Serves 6 read endpoints + 1 ingestion endpoint: + * Serves 7 read endpoints + 1 ingestion endpoint: * GET /v1/graph/:context_id Full graph * GET /v1/graph/:context_id/nodes Nodes only * GET /v1/graph/:context_id/transaction Transaction node @@ -14,10 +14,14 @@ * GET /v1/trace/:record_hash Provenance trace: backward walk via INFORMED_BY + * PROVENANCE_OF (and optionally ANNOTATES + REVISES) from * the starting record, returning the ancestor subgraph + * GET /v1/stats Operational state: record/context counts, heap + * utilization, uptime, last archive replay. O(1); + * also the Fly health-check target. * POST /v1/ingest Record ingestion */ import { createServer, type IncomingMessage, type ServerResponse } from 'node:http' +import { getHeapStatistics } from 'node:v8' import { validateSubmission, verifyRecord, sha256, hexEncode } from '@atrib/mcp' import type { AtribRecord } from '@atrib/mcp' import { buildGraph } from './graph-builder.js' @@ -112,6 +116,20 @@ export interface BindOptions { * the producer-local mirror file remains the ultimate fallback (Layer 3). */ onRecordIngested?: (record: AtribRecord, logIndex: number | undefined) => void | Promise + + /** + * Facts about this boot that only the entry point knows, surfaced by + * /v1/stats. Archive replay happens in main.ts before the server binds, so + * the server cannot measure it itself. + */ + runtimeInfo?: ServiceRuntimeInfo +} + +export interface ServiceRuntimeInfo { + /** Wall-clock ms the archive replay took on this boot. */ + replay_ms?: number + /** Records restored from the archive on this boot. */ + replayed_records?: number } const CONTEXT_ID_RE = /^[0-9a-f]{32}$/ @@ -123,6 +141,7 @@ export async function bindGraphServer( ): Promise { const store = opts.store ?? createRecordStore() const onRecordIngested = opts.onRecordIngested + const runtimeInfo = opts.runtimeInfo ?? {} const server = createServer((req, res) => { // CORS for browser-based dashboards (D054). Read endpoints serve public data per spec §3; @@ -135,7 +154,7 @@ export async function bindGraphServer( res.end() return } - handleRequest(req, res, store, onRecordIngested).catch((err) => { + handleRequest(req, res, store, onRecordIngested, runtimeInfo).catch((err) => { console.error('graph-node: request handler crashed', err) if (!res.headersSent) { sendProblem(res, 500, 'internal-error', 'Internal server error', '') @@ -170,6 +189,7 @@ async function handleRequest( res: ServerResponse, store: RecordStore, onRecordIngested?: (record: AtribRecord, logIndex: number | undefined) => void | Promise, + runtimeInfo: ServiceRuntimeInfo = {}, ): Promise { const url = req.url ?? '' const method = req.method ?? '' @@ -199,6 +219,7 @@ async function handleRequest( creator_graph: `GET /${v}/creators//graph`, trace: `GET /${v}/trace/`, chain: `GET /${v}/chain/`, + stats: `GET /${v}/stats`, ingest: `POST /${v}/ingest`, }, explorer: 'https://explore.atrib.dev/', @@ -206,6 +227,53 @@ async function handleRequest( }) } + // GET /v1/stats: operational state, and the liveness probe Fly health checks + // hit. Mirrors log-node's /v1/stats so the two services report the same way. + // + // Deliberately NOT in the spec. §3.4 is the graph query interface; this is + // operational telemetry about one implementation, and log-node's /v1/stats is + // likewise absent from the spec. Adding a graph-fact endpoint here would need + // a §3.4 section and a conformance case; this does not, and must not start + // returning graph facts (§3.6 keeps the fact layer separate). + // + // Kept O(1) on purpose. It is polled on an interval by the Fly health check, + // so it must never walk the store. record_count and context_count are counter + // reads; heap comes from V8 directly. + // + // heap is here because graph-node holds every record in memory, so heap + // headroom divided by growth in records per day IS the service's runway. That + // number used to exist only in Fly logs; on 2026-07-29 the live set reached + // the V8 ceiling and the process began aborting with nothing watching. See + // P059 in DECISIONS.md. + if (method === 'GET' && urlPath === '/v1/stats') { + const heap = getHeapStatistics() + const usedMb = heap.used_heap_size / 1024 / 1024 + const limitMb = heap.heap_size_limit / 1024 / 1024 + return sendJson(res, 200, { + service: 'atrib-graph-node', + record_count: store.getRecordCount(), + context_count: store.getContextCount(), + heap: { + used_mb: Math.round(usedMb), + limit_mb: Math.round(limitMb), + // One decimal: the watchdog thresholds are whole percents, so this is + // precise enough to see which side of one you are on. + used_pct: limitMb > 0 ? Number(((usedMb / limitMb) * 100).toFixed(1)) : null, + }, + uptime_s: Math.round(process.uptime()), + // Absent when no archive is configured, rather than null, so a caller can + // distinguish "no archive" from "archive replayed zero records". + ...(typeof runtimeInfo.replay_ms === 'number' + ? { + replay: { + ms: runtimeInfo.replay_ms, + records: runtimeInfo.replayed_records ?? 0, + }, + } + : {}), + }) + } + // POST /v1/ingest if (method === 'POST' && url === '/v1/ingest') { return handleIngest(req, res, store, onRecordIngested) diff --git a/services/graph-node/src/store.ts b/services/graph-node/src/store.ts index 1eae39f54..c13d1eb7d 100644 --- a/services/graph-node/src/store.ts +++ b/services/graph-node/src/store.ts @@ -42,6 +42,11 @@ export interface RecordStore { getAllRecords(): { record: AtribRecord; log_index: number | null }[] /** Count of distinct records in the store. Used for cheap cache invalidation. */ getRecordCount(): number + /** + * Count of distinct context_ids in the store. O(1); reported by /v1/stats so + * operators can see session breadth alongside record volume. + */ + getContextCount(): number /** log_index for a specific record_hash, or null if unknown. */ getLogIndex(recordHashHex: string): number | null /** Stored record for a specific record_hash, or null if absent. */ @@ -191,6 +196,10 @@ export function createRecordStore(): RecordStore { return allRecords.length }, + getContextCount(): number { + return byContext.size + }, + getLogIndex(recordHashHex: string): number | null { return logIndexByHash.get(recordHashHex) ?? null }, diff --git a/services/graph-node/test/server.test.ts b/services/graph-node/test/server.test.ts index f4be53384..a74d7b251 100644 --- a/services/graph-node/test/server.test.ts +++ b/services/graph-node/test/server.test.ts @@ -466,6 +466,7 @@ describe('graph-node server (section 3.4)', () => { expect(body.current_version).toBe('v1') expect(body.endpoints.trace).toBe('GET /v1/trace/') expect(body.endpoints.chain).toBe('GET /v1/chain/') + expect(body.endpoints.stats).toBe('GET /v1/stats') expect(body.explorer).toBe('https://explore.atrib.dev/') }) diff --git a/services/graph-node/test/stats.test.ts b/services/graph-node/test/stats.test.ts new file mode 100644 index 000000000..776035873 --- /dev/null +++ b/services/graph-node/test/stats.test.ts @@ -0,0 +1,167 @@ +// SPDX-License-Identifier: Apache-2.0 + +/** + * GET /v1/stats. + * + * Two jobs, and the tests split along them: + * + * 1. Report the runway. graph-node holds every record in memory, so heap + * headroom divided by growth in records per day is how long the current + * machine size lasts. Before this endpoint that number lived only in Fly + * logs, and on 2026-07-29 the live set reached the V8 ceiling with nothing + * watching. See P059 in DECISIONS.md. + * + * 2. Be the Fly health-check target. That makes it a polled endpoint, so the + * O(1) test below is a real constraint and not a nicety: if this ever walks + * the store it becomes a periodic full scan of the thing it is meant to + * protect, which is close to what caused the outage in the first place. + */ + +import { describe, it, expect, beforeAll, afterAll } from 'vitest' +import { bindGraphServer } from '../src/server.js' +import { createRecordStore, type RecordStore } from '../src/store.js' +import { base64urlEncode, signRecord, getPublicKey, genesisChainRoot } from '@atrib/mcp' +import type { AtribRecord } from '@atrib/mcp' + +const TEST_KEY = new Uint8Array(32).fill(23) + +async function makeRecord(contextId: string, timestamp: number): Promise { + const pk = await getPublicKey(TEST_KEY) + return signRecord( + { + spec_version: 'atrib/1.0' as const, + content_id: `sha256:${'c'.repeat(64)}`, + creator_key: base64urlEncode(pk), + chain_root: genesisChainRoot(contextId), + event_type: 'https://atrib.dev/v1/types/tool_call', + context_id: contextId, + timestamp, + signature: '', + } as AtribRecord, + TEST_KEY, + ) +} + +/** Wraps a real store and counts full-store scans. */ +function countingStore(): { store: RecordStore; scans: () => number; reset: () => void } { + const inner = createRecordStore() + let scans = 0 + const store: RecordStore = { + ...inner, + getAllRecords() { + scans += 1 + return inner.getAllRecords() + }, + } + return { store, scans: () => scans, reset: () => { scans = 0 } } +} + +describe('GET /v1/stats', () => { + let url: string + let close: () => Promise + let counter: ReturnType + + const CTX_A = 'a'.repeat(32) + const CTX_B = 'b'.repeat(32) + + beforeAll(async () => { + counter = countingStore() + const handle = await bindGraphServer(0, undefined, { + store: counter.store, + runtimeInfo: { replay_ms: 4242, replayed_records: 3 }, + }) + url = handle.url + close = handle.close + + // 3 records across 2 distinct context_ids. + for (const [ctx, ts] of [ + [CTX_A, 1000], + [CTX_A, 2000], + [CTX_B, 3000], + ] as const) { + const res = await fetch(`${url}/v1/ingest`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify(await makeRecord(ctx, ts)), + }) + expect(res.ok).toBe(true) + } + }) + + afterAll(async () => { + await close() + }) + + it('reports record and context counts from the store', async () => { + const res = await fetch(`${url}/v1/stats`) + expect(res.status).toBe(200) + const body = await res.json() + expect(body.service).toBe('atrib-graph-node') + expect(body.record_count).toBe(3) + expect(body.context_count).toBe(2) + }) + + it('reports heap utilization with a usable percentage', async () => { + const body = await (await fetch(`${url}/v1/stats`)).json() + expect(body.heap.used_mb).toBeGreaterThan(0) + expect(body.heap.limit_mb).toBeGreaterThan(0) + expect(body.heap.used_mb).toBeLessThanOrEqual(body.heap.limit_mb) + expect(body.heap.used_pct).toBeGreaterThan(0) + expect(body.heap.used_pct).toBeLessThanOrEqual(100) + // The number an operator acts on, so it must survive JSON as a number and + // not a string. + expect(typeof body.heap.used_pct).toBe('number') + }) + + it('reports uptime and the last archive replay', async () => { + const body = await (await fetch(`${url}/v1/stats`)).json() + expect(Number.isInteger(body.uptime_s)).toBe(true) + expect(body.uptime_s).toBeGreaterThanOrEqual(0) + expect(body.replay).toEqual({ ms: 4242, records: 3 }) + }) + + it('omits replay entirely when no archive was replayed', async () => { + // Distinguishable from "archive replayed zero records", which reports + // { ms, records: 0 }. + const handle = await bindGraphServer(0) + try { + const body = await (await fetch(`${handle.url}/v1/stats`)).json() + expect('replay' in body).toBe(false) + expect(body.record_count).toBe(0) + expect(body.context_count).toBe(0) + } finally { + await handle.close() + } + }) + + it('does not scan the store, because the health check polls it', async () => { + counter.reset() + for (let i = 0; i < 5; i++) { + expect((await fetch(`${url}/v1/stats`)).status).toBe(200) + } + expect(counter.scans()).toBe(0) + }) + + it('sets CORS so the explorer can read it cross-origin (D054)', async () => { + const res = await fetch(`${url}/v1/stats`) + expect(res.headers.get('access-control-allow-origin')).toBe('*') + }) + + it('is listed in the service-info index', async () => { + const body = await (await fetch(`${url}/`)).json() + expect(body.endpoints.stats).toBe('GET /v1/stats') + }) + + it('tracks new ingests', async () => { + const before = (await (await fetch(`${url}/v1/stats`)).json()).record_count + const res = await fetch(`${url}/v1/ingest`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify(await makeRecord('f'.repeat(32), 9000)), + }) + expect(res.ok).toBe(true) + const after = await (await fetch(`${url}/v1/stats`)).json() + expect(after.record_count).toBe(before + 1) + expect(after.context_count).toBe(3) + }) +})