import { describe, test } from 'node:test'; import assert from 'node:assert/strict'; import fs from 'node:fs'; import os from 'node:os'; import path from 'node:path'; import { CondensedGraph } from '../../src/core/CondensedGraph.js'; import { ShardedSnapshotBuilder } from '../../src/core/shards/ShardedSnapshotBuilder.js'; import { ShardedSnapshot } from '../../src/core/shards/ShardedSnapshot.js'; import { FileShardStorage } from '../../src/core/shards/FileShardStorage.js'; import { DeltaShardBinary } from '../../src/core/shards/DeltaShardBinary.js'; import { WaveletShardBinary } from '../../src/core/shards/WaveletShardBinary.js'; function buildSnapshot(bucketSize = 4) { const graph = new CondensedGraph(); const nodes = []; for (let i = 0; i < 6; i++) { nodes.push(graph._ensureNode(`node:${i}`)); } graph.addEdge(nodes[0], 'risk_score', nodes[1], 1.0, { value: 0.4 }); graph.addEdge(nodes[0], 'risk_score', nodes[2], 1.0, { value: 0.9 }); graph.addEdge(nodes[3], 'risk_score', nodes[4], 1.0, { value: 0.5 }); graph.addEdge(nodes[1], 'risk_limit', nodes[1], 1.0, { value: 0.6 }); graph.addEdge(nodes[2], 'risk_limit', nodes[2], 1.0, { value: 0.6 }); graph.addEdge(nodes[4], 'risk_limit', nodes[4], 1.0, { value: 0.6 }); graph.finalizePerfectHash(); graph.finalizeWaveletAdjacency({ dropAdjacencyList: true }); const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'sharded-delta-comp-')); const builder = new ShardedSnapshotBuilder({ bucketSize, includeDirections: ['out', 'in'] }); const manifest = builder.build(graph, dir); const storage = new FileShardStorage(dir); const snapshot = new ShardedSnapshot(manifest, storage, { cacheLimit: 8 }); snapshot.initializeSync(); return { snapshot, manifest, dir, nodes }; } function writeDeltaLayer(snapshot, dir, entries, mode = 'override') { fs.mkdirSync(dir, { recursive: true }); const merged = new Map(); for (const entry of entries) { const shardMeta = snapshot._selectShardMeta(entry.relationId, entry.direction, entry.srcId); assert.ok(shardMeta, 'Missing shard meta for delta entry'); const localSource = snapshot._localSource(entry.srcId, shardMeta); const key = shardMeta.cacheKey; let bucket = merged.get(key); if (!bucket) { bucket = { shardMeta, additions: [], removals: [] }; merged.set(key, bucket); } for (const add of entry.additions) { bucket.additions.push({ srcLocal: localSource, otherId: add.dstId, possBits: add.possBits, relBits: add.relBits }); } for (const rem of entry.removals) { bucket.removals.push({ srcLocal: localSource, otherId: rem.dstId }); } } const shards = []; for (const bucket of merged.values()) { const shardMeta = bucket.shardMeta; const buffer = DeltaShardBinary.serialize({ relationId: shardMeta.relationId, direction: shardMeta.direction, rangeStart: shardMeta.rangeStart, rangeEnd: shardMeta.rangeEnd, nodeCount: snapshot.nodeCount, additions: bucket.additions, removals: bucket.removals }); const shardKey = `delta-${shardMeta.key}`; fs.writeFileSync(path.join(dir, shardKey), new Uint8Array(buffer)); shards.push({ key: shardKey, relationId: shardMeta.relationId, direction: shardMeta.direction, rangeStart: shardMeta.rangeStart, rangeEnd: shardMeta.rangeEnd, cacheKey: shardMeta.cacheKey }); } return { shards, storage: new FileShardStorage(dir), mode }; } function compactLayer(snapshot, layer, manifest, outputDir, baseDir) { fs.rmSync(outputDir, { recursive: true, force: true }); fs.mkdirSync(outputDir, { recursive: true }); if (manifest.nodeTableKey) { fs.copyFileSync(path.join(baseDir, manifest.nodeTableKey), path.join(outputDir, manifest.nodeTableKey)); } if (manifest.componentKey) { fs.copyFileSync(path.join(baseDir, manifest.componentKey), path.join(outputDir, manifest.componentKey)); } for (const shard of manifest.shards || []) { const basePath = path.join(baseDir, shard.key); fs.copyFileSync(basePath, path.join(outputDir, shard.key)); } for (const shardMeta of layer.shards) { const base = snapshot._cacheIndex.get(shardMeta.cacheKey); if (!base) continue; const shard = snapshot._loadShardSync(base.relationId, base.direction, base.rangeStart); if (!shard) continue; const deltaBuffer = layer.storage.getSync(shardMeta.key); if (!deltaBuffer) continue; const deltaShard = DeltaShardBinary.deserialize(deltaBuffer); const rangeSize = shard.rangeEnd - shard.rangeStart; const sources = new Array(rangeSize); for (let localSource = 0; localSource < rangeSize; localSource++) { const range = snapshot._rangeForSource(shard, localSource); const list = []; if (range) { for (let pos = range.start; pos < range.end; pos++) { list.push({ otherId: shard.dstIds[pos], possBits: shard.possBits[pos], relBits: shard.relBits[pos] }); } } sources[localSource] = list; } for (const removal of deltaShard.removals) { const list = sources[removal.srcLocal]; if (!list) continue; const idx = list.findIndex((item) => item.otherId === removal.otherId); if (idx !== -1) list.splice(idx, 1); } for (const addition of deltaShard.additions) { const list = sources[addition.srcLocal] || (sources[addition.srcLocal] = []); list.push({ otherId: addition.otherId, possBits: addition.possBits, relBits: addition.relBits }); } const buffer = WaveletShardBinary.serialize({ relationId: shard.relationId, direction: shard.direction, rangeStart: shard.rangeStart, rangeEnd: shard.rangeEnd, nodeCount: snapshot.nodeCount, sources }); fs.writeFileSync(path.join(outputDir, base.key), new Uint8Array(buffer)); } } function evaluateRisk(snapshot, relScore, relLimit, userId, docId) { const scoreEdge = snapshot.findEdgeSync(userId, relScore, docId); if (!scoreEdge || scoreEdge.value === undefined) return false; const limitEdge = snapshot.findEdgeSync(docId, relLimit, docId); if (!limitEdge || limitEdge.value === undefined) return false; return scoreEdge.value <= limitEdge.value; } // ADR-003: sharded snapshots are stubs in src/core/shards/ — tests are the spec for when the subsystem is implemented. describe.skip('Sharded delta comparator equivalence', () => { test('overlay and compacted base agree on comparator outcomes', () => { const { snapshot, manifest, dir, nodes } = buildSnapshot(4); const relIdScore = snapshot.relationIdToName.indexOf('risk_score'); const relIdLimit = snapshot.relationIdToName.indexOf('risk_limit'); const deltaDir = path.join(dir, 'delta'); const layer = writeDeltaLayer(snapshot, deltaDir, [ { relationId: relIdScore, direction: 'out', srcId: nodes[0], additions: [{ dstId: nodes[3], possBits: 65535, relBits: 65535 }], removals: [] }, { relationId: relIdLimit, direction: 'out', srcId: nodes[3], additions: [{ dstId: nodes[3], possBits: 65535, relBits: 65535 }], removals: [] } ], 'override'); snapshot.setDeltaLayers([layer]); const compactDir = path.join(dir, 'compact'); compactLayer(snapshot, layer, manifest, compactDir, dir); const compactSnapshot = new ShardedSnapshot(manifest, new FileShardStorage(compactDir), { cacheLimit: 8 }); compactSnapshot.initializeSync(); const overlayResult = evaluateRisk(snapshot, relIdScore, relIdLimit, nodes[0], nodes[3]); const compactResult = evaluateRisk(compactSnapshot, relIdScore, relIdLimit, nodes[0], nodes[3]); assert.strictEqual(overlayResult, compactResult); fs.rmSync(dir, { recursive: true, force: true }); }); });