diff --git a/services/graph-node/fly.toml b/services/graph-node/fly.toml index 1c595ab7..97dbee9d 100644 --- a/services/graph-node/fly.toml +++ b/services/graph-node/fly.toml @@ -7,6 +7,15 @@ primary_region = 'iad' [env] PORT = '3200' HOST = '0.0.0.0' + # Node caps its old-space heap at roughly half of a small machine's RAM, + # independent of what [[vm]] memory says. On this 1024mb machine V8 reported + # heap_size_limit = 493MB, so the 2026-05-06 bump from 256mb never reached + # the heap that was actually running out: on 2026-07-29 the resident record + # set (~112k records, ~450MB) filled that 493MB ceiling and any further + # allocation aborted the process with "Ineffective mark-compacts near heap + # limit". This makes the RAM the machine already has usable. Leaves ~200MB + # for non-heap RSS (node binary, buffers, archive replay stream). + NODE_OPTIONS = '--max-old-space-size=768' # Append-only archive of every successful /v1/ingest. Replayed on # cold-start so graph-node rebuilds its in-memory store after OOM / # deploy / fly machine reboot. Mounts on the [mounts] volume below. @@ -29,6 +38,13 @@ primary_region = 'iad' size = 'shared-cpu-1x' # Bumped from 256mb after observed OOM-kill on 2026-05-06 with ~1500 records # in-memory. graph-node holds all derived edges + per-record verification - # state in memory (no volume), so memory pressure scales with log size. - # 1024mb gives headroom for several thousand records before OOM. + # state in memory, so memory pressure scales with log size. Raising this + # alone is not enough: see NODE_OPTIONS above, V8 will not use the extra RAM + # unless told. Both settings move together. + # + # Headroom is finite and the log only grows (~4k records/day at 2026-07-29, + # ~4KB each in heap, so ~17MB/day). At a 768MB heap that is a few weeks + # before this becomes tight again. The durable fix is the disk-backed graph + # store already flagged in src/persistence.ts, which notes this in-memory + # shape is sustainable to ~10^5 records; the log passed that in July 2026. memory = '1024mb' diff --git a/services/graph-node/src/server.ts b/services/graph-node/src/server.ts index b8ea5459..ddaf5745 100644 --- a/services/graph-node/src/server.ts +++ b/services/graph-node/src/server.ts @@ -18,7 +18,7 @@ */ import { createServer, type IncomingMessage, type ServerResponse } from 'node:http' -import { validateSubmission, verifyRecord, canonicalRecord, sha256, hexEncode } from '@atrib/mcp' +import { validateSubmission, verifyRecord, sha256, hexEncode } from '@atrib/mcp' import type { AtribRecord } from '@atrib/mcp' import { buildGraph } from './graph-builder.js' import { createRecordStore, type RecordStore } from './store.js' @@ -825,19 +825,23 @@ async function handleTrace( const includeAnnotations = params.get('include_annotations') !== 'false' const includeRevisions = params.get('include_revisions') !== 'false' - // Index every record by its record_hash (canonicalRecord -> sha256 -> hex) - // so we can walk by hash. The store doesn't expose this index directly; - // build it here once per request. For a busy log this would move into the - // store, but at current scale (low thousands of records) the per-request - // build is fast enough. - const allRecords = store.getAllRecords() - const byHash = new Map() - for (const { record } of allRecords) { - const hashHex = hexEncode(sha256(canonicalRecord(record))) - byHash.set(hashHex, record) - } - - const startRecord = byHash.get(recordHashHex) + // Resolve records by record_hash through the store's own index. The store + // hashes each record exactly once at ingest (see createRecordStore's + // recordByHash map), so this is an O(1) lookup. + // + // This used to rebuild the whole index per request: getAllRecords() followed + // by canonicalRecord -> sha256 -> hexEncode over every record in the store. + // That was written when the log held "low thousands" of records. At 112k it + // allocated ~80MB of transient strings and burned >1s of CPU per request, + // against a V8 heap already ~450MB full of the resident record set. The + // allocation could not be satisfied, so V8 aborted the process + // ("Ineffective mark-compacts near heap limit") on every trace request, + // which Fly served as a 502. Keep hash resolution O(1); the walk below + // only ever does point lookups. + const recordByHash = (hashHex: string): AtribRecord | null => + store.getRecordByHash(hashHex)?.record ?? null + + const startRecord = recordByHash(recordHashHex) if (!startRecord) { return sendProblem( res, @@ -865,7 +869,7 @@ async function handleTrace( for (let hop = 0; hop < depth; hop++) { const next: string[] = [] for (const hash of frontier) { - const record = byHash.get(hash) + const record = recordByHash(hash) if (!record) continue // eslint-disable-next-line @typescript-eslint/no-explicit-any const r = record as any @@ -895,7 +899,7 @@ async function handleTrace( for (const ancestorHash of candidates) { if (visited.has(ancestorHash)) continue visited.add(ancestorHash) - const ancestor = byHash.get(ancestorHash) + const ancestor = recordByHash(ancestorHash) if (ancestor) { collected.push(ancestor) next.push(ancestorHash) @@ -961,16 +965,14 @@ async function handleChain( Math.min(Number.isNaN(rawDepth) ? TRACE_DEFAULT_DEPTH : rawDepth, TRACE_MAX_DEPTH), ) - // Build a hash index once per request so we can resolve chain_root references - // back to records. Mirrors handleTrace's approach. - const allRecords = store.getAllRecords() - const byHash = new Map() - for (const { record } of allRecords) { - const hashHex = hexEncode(sha256(canonicalRecord(record))) - byHash.set(hashHex, record) - } + // Resolve chain_root references back to records through the store's own + // hash index. Mirrors handleTrace: an O(1) point lookup, not a per-request + // rebuild of the whole index (see the note in handleTrace for why the + // rebuild was fatal at production log size). + const recordByHash = (hashHex: string): AtribRecord | null => + store.getRecordByHash(hashHex)?.record ?? null - const startRecord = byHash.get(recordHashHex) + const startRecord = recordByHash(recordHashHex) if (!startRecord) { return sendProblem( res, @@ -997,7 +999,7 @@ async function handleChain( if (visited.has(ancestorHash)) break visited.add(ancestorHash) - const ancestor = byHash.get(ancestorHash) + const ancestor = recordByHash(ancestorHash) if (!ancestor) break // chain_root references a record outside the store collected.push(ancestor) current = ancestor diff --git a/services/graph-node/test/trace-hash-lookup.test.ts b/services/graph-node/test/trace-hash-lookup.test.ts new file mode 100644 index 00000000..983f53d4 --- /dev/null +++ b/services/graph-node/test/trace-hash-lookup.test.ts @@ -0,0 +1,158 @@ +// SPDX-License-Identifier: Apache-2.0 + +/** + * Regression guard for the 2026-07-29 graph-node outage. + * + * /v1/trace and /v1/chain used to resolve a record_hash by rebuilding an index + * over the entire store on every request: getAllRecords() followed by + * canonicalRecord -> sha256 -> hexEncode per record. That was written for a log + * holding "low thousands" of records. At 112k it allocated ~80MB of transient + * strings per request against a V8 heap already ~450MB full of the resident + * record set, so V8 aborted the process ("Ineffective mark-compacts near heap + * limit") on every trace request and Fly served the result as a 502. It flapped + * the deploy-services graph canary and broke the explorer's trace view. + * + * The store already hashes every record once at ingest and exposes + * getRecordByHash(). These tests pin that hash resolution stays a point lookup: + * a full-store scan per request must not come back. + * + * The ยง1.9 revocation registry does legitimately scan globally, but it is + * memoised on record count by buildRegistryCached. So these tests warm the + * registry first and then assert an exact zero scans in steady state: that is + * what discriminates the fix from the defect. A "<= 1" assertion would not, + * because the defect's own count is 1 once the registry is already warm. + */ + +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, + canonicalRecord, + sha256, + hexEncode, +} from '@atrib/mcp' +import type { AtribRecord } from '@atrib/mcp' + +const TEST_KEY = new Uint8Array(32).fill(11) +const CONTEXT_ID = 'd'.repeat(32) + +function hashOf(record: AtribRecord): string { + return hexEncode(sha256(canonicalRecord(record))) +} + +async function makeRecord( + overrides: Partial<{ chain_root: string; informed_by: string[]; timestamp: number }> = {}, +): Promise { + const pk = await getPublicKey(TEST_KEY) + const record = { + spec_version: 'atrib/1.0' as const, + content_id: `sha256:${'c'.repeat(64)}`, + creator_key: base64urlEncode(pk), + chain_root: overrides.chain_root ?? genesisChainRoot(CONTEXT_ID), + event_type: 'https://atrib.dev/v1/types/tool_call', + context_id: CONTEXT_ID, + timestamp: overrides.timestamp ?? 1_700_000_000_000, + signature: '', + ...(overrides.informed_by ? { informed_by: overrides.informed_by } : {}), + } + return signRecord(record 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('trace/chain resolve record_hash without scanning the store', () => { + let url: string + let close: () => Promise + let counter: ReturnType + let genesisHash: string + let descendantHash: string + + beforeAll(async () => { + counter = countingStore() + const handle = await bindGraphServer(0, undefined, { store: counter.store }) + url = handle.url + close = handle.close + + // genesis <- descendant, linked both ways the two endpoints walk: + // informed_by (what /v1/trace follows) and chain_root (what /v1/chain follows). + const genesis = await makeRecord({ timestamp: 1000 }) + genesisHash = hashOf(genesis) + const descendant = await makeRecord({ + timestamp: 2000, + chain_root: `sha256:${genesisHash}`, + informed_by: [`sha256:${genesisHash}`], + }) + descendantHash = hashOf(descendant) + + for (const record of [genesis, descendant]) { + const res = await fetch(`${url}/v1/ingest`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify(record), + }) + expect(res.ok).toBe(true) + } + + // Warm the revocation registry so its one legitimate global scan is + // memoised and every scan assertion below can demand an exact zero. + await fetch(`${url}/v1/trace/${descendantHash}`) + }) + + afterAll(async () => { + await close() + }) + + it('/v1/trace still walks informed_by to the ancestor', async () => { + const res = await fetch(`${url}/v1/trace/${descendantHash}`) + expect(res.status).toBe(200) + const body = await res.json() + expect(body.start_record_hash).toBe(`sha256:${descendantHash}`) + const ids = body.graph.nodes.map((n: { id: string }) => n.id) + expect(ids).toContain(`sha256:${descendantHash}`) + expect(ids).toContain(`sha256:${genesisHash}`) + }) + + it('/v1/chain still walks chain_root to genesis', async () => { + const res = await fetch(`${url}/v1/chain/${descendantHash}`) + expect(res.status).toBe(200) + const body = await res.json() + expect(body.record_count).toBe(2) + }) + + it('/v1/trace does not rebuild a full-store hash index', async () => { + counter.reset() + const res = await fetch(`${url}/v1/trace/${descendantHash}`) + expect(res.status).toBe(200) + expect(counter.scans()).toBe(0) + }) + + it('/v1/chain does not rebuild a full-store hash index', async () => { + counter.reset() + const res = await fetch(`${url}/v1/chain/${descendantHash}`) + expect(res.status).toBe(200) + expect(counter.scans()).toBe(0) + }) + + it('an unknown record_hash still 404s without scanning the store', async () => { + counter.reset() + const res = await fetch(`${url}/v1/trace/${'0'.repeat(64)}`) + expect(res.status).toBe(404) + expect(counter.scans()).toBe(0) + }) +})