value-graph 0.1.0: JSON-free, bigint-free binary snapshot layer
- src/snapshot.js: compact portable wire (VGST store / VGGP graph), binary maps for params/pattern/objects (no JSON), buffer values (no JS bigint), CRC-32 guarded. - createFileStore: binary per-key files (encodeEntryBytes/decodeEntryBytes). - ValueGraph snapshot/snapshotFile/restore/loadFile; keys base64url(params-bytes). - 60 tests (node:test + rigor incl. 'flip ANY byte never decodes silently').
This commit is contained in:
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,778 @@
|
||||
/**
|
||||
* @arbiter/value-graph tests — compile → run separation for BOTH queries and
|
||||
* invalidations/eager updates, pull-on-duplex-push queries, TTL + version
|
||||
* staleness, dependent-DAG cascade, all DSL value types, swappable stores.
|
||||
*
|
||||
* Resolver contract: CALLBACK ORIENTED (no async/await/promises). Resolvers may
|
||||
* call `cb(err, result)` asynchronously (returning `undefined`) or return the
|
||||
* result synchronously. A returned Promise is rejected.
|
||||
*/
|
||||
import { describe, it } from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { mkdtempSync, readdirSync, existsSync, rmSync, readFileSync } from 'node:fs';
|
||||
import { tmpdir } from 'node:os';
|
||||
import { join } from 'node:path';
|
||||
import {
|
||||
ValueGraph, createMapStore, createFileStore,
|
||||
encodeSnapshot, decodeSnapshot, encodeGraphSnapshot, decodeGraphSnapshot,
|
||||
encodeParams, decodeParams
|
||||
} from '../src/index.js';
|
||||
import { createSink } from '@push-stream-std/push-stream-base';
|
||||
|
||||
// Promise-free library, promise-ful test harness: adapt the callback API.
|
||||
function get(vg, subject, rel, params = {}) {
|
||||
return new Promise((resolve, reject) => {
|
||||
vg.get(subject, rel, params, (err, result) => err ? reject(err) : resolve(result));
|
||||
});
|
||||
}
|
||||
function run(vg, plan) {
|
||||
return new Promise((resolve, reject) => {
|
||||
vg.run(plan, (err, result) => err ? reject(err) : resolve(result));
|
||||
});
|
||||
}
|
||||
|
||||
function collect(stream, { until, timeout = 4000 } = {}) {
|
||||
return new Promise((resolve, reject) => {
|
||||
const items = [];
|
||||
const timer = setTimeout(() => reject(new Error(`collect timeout: ${JSON.stringify(items)}`)), timeout);
|
||||
stream.pipe(createSink(
|
||||
(data) => {
|
||||
items.push(data);
|
||||
if (until && until(items)) { clearTimeout(timer); resolve(items); }
|
||||
},
|
||||
(err) => { clearTimeout(timer); err ? reject(err) : resolve(items); }
|
||||
));
|
||||
});
|
||||
}
|
||||
|
||||
describe('ValueGraph compile → run (queries)', () => {
|
||||
it('prunes cached subtrees into a single trivial leaf, then runs it', async () => {
|
||||
let calls = 0;
|
||||
const vg = new ValueGraph();
|
||||
vg.compute('double_balance', (s, p, { deps }) => { calls++; return { value: deps.base_balance * 2, unit: 'usd_cents' }; }, { dependsOn: ['base_balance'] });
|
||||
vg.compute('base_balance', (s, p, ctx, cb) => cb(null, 21));
|
||||
|
||||
const p1 = vg.plan('t:1', 'double_balance', {});
|
||||
assert.equal(p1.size, 2);
|
||||
assert.deepEqual(p1.nodes.get('double_balance').parents, ['base_balance']);
|
||||
assert.equal((await run(vg, p1)).value, 42);
|
||||
|
||||
const p2 = vg.plan('t:1', 'double_balance', {});
|
||||
assert.equal(p2.nodes.get('double_balance').trivial, true);
|
||||
assert.deepEqual(p2.nodes.get('double_balance').parents, []);
|
||||
assert.equal(p2.size, 1, 'cached subtree pruned');
|
||||
assert.equal(calls, 1, 'no recompute');
|
||||
assert.equal((await run(vg, p2)).value, 42);
|
||||
});
|
||||
|
||||
it('accepts synchronous resolver returns (pure compute)', async () => {
|
||||
const vg = new ValueGraph();
|
||||
vg.compute('double_balance', (s, p, { deps }) => deps.base_balance * 2, { dependsOn: ['base_balance'] });
|
||||
vg.compute('base_balance', () => 21);
|
||||
assert.equal((await get(vg, 't:1', 'double_balance', {})).value, 42);
|
||||
});
|
||||
|
||||
it('rejects a Promise-returning resolver (promise-free contract)', async () => {
|
||||
const vg = new ValueGraph();
|
||||
vg.compute('bad', async () => 42);
|
||||
await assert.rejects(() => get(vg, 'u:1', 'bad', {}), /returned a Promise/);
|
||||
});
|
||||
|
||||
it('recomputes after TTL expiry (plan marks the node non-trivial)', async () => {
|
||||
let t = 0;
|
||||
let value = 10;
|
||||
const vg = new ValueGraph({ clock: () => t, defaultTTL: 1000 });
|
||||
vg.compute('quota', (s, p, ctx, cb) => cb(null, value));
|
||||
|
||||
assert.equal((await get(vg, 't:1', 'quota', {})).value, 10);
|
||||
assert.equal(vg.plan('t:1', 'quota', {}).nodes.get('quota').trivial, true);
|
||||
t = 1500;
|
||||
value = 20;
|
||||
assert.equal(vg.plan('t:1', 'quota', {}).nodes.get('quota').trivial, false, 'expired → non-trivial');
|
||||
assert.equal((await get(vg, 't:1', 'quota', {})).value, 20);
|
||||
});
|
||||
|
||||
it('resolves a query duplex with repeated get pushes (pull on duplex push)', async () => {
|
||||
let calls = 0;
|
||||
const vg = new ValueGraph();
|
||||
vg.compute('balance', (s, p, ctx, cb) => { calls++; cb(null, { value: 42, unit: 'usd_cents' }); });
|
||||
|
||||
const q = vg.query('user:alice', 'balance', {});
|
||||
const events = collect(q.source, { until: (items) => items.filter(i => i.value !== undefined).length >= 2 });
|
||||
q.sink.write({ get: true });
|
||||
q.sink.write({ get: true });
|
||||
const items = await events;
|
||||
|
||||
const values = items.filter(i => i.value !== undefined).map(i => i.value);
|
||||
assert.deepEqual(values, [42, 42]);
|
||||
assert.equal(calls, 1, 'second get served from the cached trivial plan');
|
||||
});
|
||||
|
||||
it('rejects a dependency cycle at compile time', async () => {
|
||||
const vg = new ValueGraph();
|
||||
vg.compute('a', () => 1, { dependsOn: ['b'] });
|
||||
vg.compute('b', () => 2, { dependsOn: ['a'] });
|
||||
await assert.rejects(() => get(vg, 'x', 'a', {}), /cycle/);
|
||||
});
|
||||
});
|
||||
|
||||
describe('ValueGraph compile → run (invalidation / eager updates)', () => {
|
||||
it('invalidate compiles the dependent DAG and recomputes downstream', async () => {
|
||||
let base = 10;
|
||||
const vg = new ValueGraph();
|
||||
vg.compute('base_balance', () => base);
|
||||
vg.compute('double_balance', (s, p, { deps }) => ({ value: deps.base_balance * 2, unit: 'usd_cents' }), { dependsOn: ['base_balance'] });
|
||||
vg.compute('quad_balance', (s, p, { deps }) => deps.double_balance * 2, { dependsOn: ['double_balance'] });
|
||||
|
||||
assert.equal((await get(vg, 't:1', 'quad_balance', {})).value, 40);
|
||||
assert.equal(vg.plan('t:1', 'quad_balance', {}).size, 1, 'cached: single leaf');
|
||||
|
||||
base = 30;
|
||||
vg.invalidate('t:1', 'base_balance', {}); // affects base + double + quad transitively
|
||||
|
||||
const p = vg.plan('t:1', 'quad_balance', {});
|
||||
assert.equal(p.size, 3, 'invalidation expanded the plan to the full dependent DAG');
|
||||
assert.equal((await run(vg, p)).value, 120, 'recomputed down the chain');
|
||||
});
|
||||
|
||||
it('pushes a stale trigger down on invalidation and recomputes', async () => {
|
||||
let value = 100;
|
||||
const vg = new ValueGraph();
|
||||
vg.compute('available', () => value);
|
||||
|
||||
const q = vg.query('tenant:acme', 'available', { feature: 'tokens_in:gpt-4' });
|
||||
const events = collect(q.source, {
|
||||
until: (items) => items.some(i => i.stale) && items.filter(i => i.value !== undefined).length >= 2
|
||||
});
|
||||
q.sink.write({ get: true });
|
||||
value = 50;
|
||||
vg.invalidate('tenant:acme', 'available', { feature: 'tokens_in:gpt-4' });
|
||||
q.sink.write({ get: true });
|
||||
const items = await events;
|
||||
|
||||
assert.ok(items.some(i => i.stale === true), 'stale trigger pushed down');
|
||||
assert.ok(items.filter(i => i.value !== undefined).some(i => i.value === 50), 'recomputed to 50');
|
||||
});
|
||||
|
||||
it('set compiles the dependent DAG: writes the source, invalidates dependents', async () => {
|
||||
let calls = 0;
|
||||
const vg = new ValueGraph();
|
||||
vg.compute('base_balance', () => 10);
|
||||
vg.compute('double_balance', (s, p, { deps }) => { calls++; return deps.base_balance * 2; }, { dependsOn: ['base_balance'] });
|
||||
|
||||
assert.equal((await get(vg, 't:1', 'double_balance', {})).value, 20);
|
||||
// Eager update from outside (e.g. a ledger posting).
|
||||
vg.set('t:1', 'base_balance', {}, { value: 50, unit: 'usd_cents' });
|
||||
|
||||
// double_balance is no longer cached — the set invalidated it.
|
||||
assert.equal(vg.plan('t:1', 'double_balance', {}).nodes.get('double_balance').trivial, false);
|
||||
const got = await get(vg, 't:1', 'double_balance', {});
|
||||
assert.equal(got.value, 100, 'dependent recomputed after eager set');
|
||||
});
|
||||
|
||||
it('cascades a stale trigger to dependent queries on set', async () => {
|
||||
const vg = new ValueGraph();
|
||||
vg.compute('base_balance', () => 10);
|
||||
vg.compute('double_balance', (s, p, { deps }) => deps.base_balance * 2, { dependsOn: ['base_balance'] });
|
||||
|
||||
const q = vg.query('t:1', 'double_balance', {});
|
||||
const events = collect(q.source, { until: (items) => items.some(i => i.stale) });
|
||||
q.sink.write({ get: true });
|
||||
vg.set('t:1', 'base_balance', {}, { value: 99, unit: 'usd_cents' });
|
||||
const items = await events;
|
||||
assert.ok(items.some(i => i.stale === true && i.relation === 'double_balance' && i.via === 'base_balance'));
|
||||
});
|
||||
});
|
||||
|
||||
describe('ValueGraph value types', () => {
|
||||
it('supports number, string, boolean, timestamp, and arrays', async () => {
|
||||
const vg = new ValueGraph();
|
||||
vg.compute('reputation', () => 7.5);
|
||||
vg.compute('role', () => 'admin');
|
||||
vg.compute('active', () => true);
|
||||
vg.compute('last_login', () => 1700000000000);
|
||||
vg.compute('permissions', () => ['read', 'write']);
|
||||
|
||||
assert.equal((await get(vg, 'u:1', 'reputation', {})).value, 7.5);
|
||||
assert.equal((await get(vg, 'u:1', 'role', {})).value, 'admin');
|
||||
assert.equal((await get(vg, 'u:1', 'active', {})).value, true);
|
||||
assert.equal((await get(vg, 'u:1', 'last_login', {})).value, 1700000000000);
|
||||
assert.deepEqual((await get(vg, 'u:1', 'permissions', {})).value, ['read', 'write']);
|
||||
});
|
||||
|
||||
it('supports interval values ({ lower, upper }) for possibilistic estimates', async () => {
|
||||
const vg = new ValueGraph();
|
||||
vg.compute('estimated_tokens', () => ({ lower: 800, upper: 1400 }));
|
||||
const entry = await get(vg, 't:1', 'estimated_tokens', {});
|
||||
assert.deepEqual(entry.value, { lower: 800, upper: 1400 });
|
||||
});
|
||||
|
||||
it('rejects an unregistered value type', async () => {
|
||||
const vg = new ValueGraph();
|
||||
vg.compute('bad', () => ({ value: { nested: true } }));
|
||||
await assert.rejects(() => get(vg, 'u:1', 'bad', {}), /must return a value/);
|
||||
});
|
||||
});
|
||||
|
||||
describe('ValueGraph typed values (DSL-declared returnType)', () => {
|
||||
it('exposes the declared spec via relationSpec', () => {
|
||||
const vg = new ValueGraph();
|
||||
vg.define('balance', { operator: 'source', returnType: 'number', params: [{ name: 'tenant', type: 'Employee', isArray: false }], fn: () => 0 });
|
||||
const spec = vg.relationSpec('balance');
|
||||
assert.equal(spec.returnType, 'number');
|
||||
assert.equal(spec.params[0].name, 'tenant');
|
||||
assert.equal(spec.operator, 'source');
|
||||
assert.equal(vg.relationSpec('nope'), null);
|
||||
});
|
||||
|
||||
it('set enforces the declared return type', () => {
|
||||
const vg = new ValueGraph();
|
||||
vg.define('role', { operator: 'source', returnType: 'string', fn: () => null });
|
||||
vg.set('t:1', 'role', {}, { value: 'admin' });
|
||||
assert.throws(() => vg.set('t:1', 'role', {}, { value: 42 }), /must match declared type 'string'/);
|
||||
});
|
||||
|
||||
it('resolver results are validated against the declared return type', async () => {
|
||||
const vg = new ValueGraph();
|
||||
vg.define('role', { operator: 'source', returnType: 'string', fn: () => 42 });
|
||||
await assert.rejects(() => get(vg, 't:1', 'role', {}), /must match declared type 'string'/);
|
||||
});
|
||||
|
||||
it('accepts interval values via set (consistent with the resolver path)', async () => {
|
||||
const vg = new ValueGraph();
|
||||
vg.define('estimate', { operator: 'source', returnType: 'interval', fn: () => null });
|
||||
vg.set('t:1', 'estimate', {}, { value: { lower: 100, upper: 200 } });
|
||||
const e = await get(vg, 't:1', 'estimate', {});
|
||||
assert.deepEqual(e.value, { lower: 100, upper: 200 });
|
||||
});
|
||||
|
||||
it('entity-typed values accept keys (non-primitive type names)', () => {
|
||||
const vg = new ValueGraph();
|
||||
vg.define('owner', { operator: 'source', returnType: 'Employee', fn: () => null });
|
||||
vg.set('t:1', 'owner', {}, { value: 'user:alice' }); // key accepted
|
||||
assert.throws(() => vg.set('t:1', 'owner', {}, { value: { id: 1 } }), /requires a value/); // object not a valid key
|
||||
});
|
||||
});
|
||||
|
||||
describe('ValueGraph invalidation guards', () => {
|
||||
it('invalidate(subject) without a relation throws instead of clearing the whole store', () => {
|
||||
const vg = new ValueGraph();
|
||||
vg.compute('a', () => 1);
|
||||
assert.throws(() => vg.invalidate('t:1'), /requires a valueRelation/);
|
||||
});
|
||||
|
||||
it('invalidate() with no arguments still clears the whole store', async () => {
|
||||
const vg = new ValueGraph();
|
||||
vg.compute('a', () => 1);
|
||||
await get(vg, 't:1', 'a', {});
|
||||
assert.equal(vg.plan('t:1', 'a', {}).nodes.get('a').trivial, true);
|
||||
vg.invalidate();
|
||||
assert.equal(vg.plan('t:1', 'a', {}).nodes.get('a').trivial, false, 'cleared');
|
||||
});
|
||||
});
|
||||
|
||||
describe('ValueGraph backing store', () => {
|
||||
it('uses a swappable backing store', async () => {
|
||||
const store = createMapStore();
|
||||
let put = 0;
|
||||
const tracked = { ...store, set(key, value) { put++; store.set(key, value); } };
|
||||
const vg = new ValueGraph({ store: tracked });
|
||||
vg.compute('score', () => 7);
|
||||
assert.equal((await get(vg, 'u:1', 'score', {})).value, 7);
|
||||
assert.equal(put, 1);
|
||||
});
|
||||
|
||||
it('returns null when no compute is registered and nothing is cached', async () => {
|
||||
const vg = new ValueGraph();
|
||||
assert.equal(await get(vg, 'u:1', 'nope', {}), null);
|
||||
});
|
||||
});
|
||||
|
||||
describe('ValueGraph larger-than-memory (disk-backed store)', () => {
|
||||
function tmpDir() {
|
||||
return mkdtempSync(join(tmpdir(), 'value-graph-'));
|
||||
}
|
||||
|
||||
it('persists entries to disk and reads them back into a fresh graph instance', async () => {
|
||||
const dir = tmpDir();
|
||||
try {
|
||||
// First process: compute and cache against the disk store.
|
||||
let calls = 0;
|
||||
const vg1 = new ValueGraph({ store: createFileStore(dir) });
|
||||
vg1.compute('double_balance', (s, p, { deps }) => deps.base_balance * 2, { dependsOn: ['base_balance'] });
|
||||
vg1.compute('base_balance', (s, p, ctx, cb) => { calls++; cb(null, 21); });
|
||||
assert.equal((await get(vg1, 't:1', 'double_balance', {})).value, 42);
|
||||
assert.equal(calls, 1);
|
||||
assert.ok(readdirSync(dir).length > 0, 'entries written to disk');
|
||||
|
||||
// Second process: a brand-new graph over the SAME directory.
|
||||
const vg2 = new ValueGraph({ store: createFileStore(dir) });
|
||||
vg2.compute('double_balance', (s, p, { deps }) => deps.base_balance * 2, { dependsOn: ['base_balance'] });
|
||||
vg2.compute('base_balance', () => { throw new Error('should not recompute from disk'); });
|
||||
const plan = vg2.plan('t:1', 'double_balance', {});
|
||||
assert.equal(plan.nodes.get('double_balance').trivial, true, 'cached leaf restored from disk');
|
||||
assert.equal((await run(vg2, plan)).value, 42, 'value read from disk, no recompute');
|
||||
} finally {
|
||||
rmSync(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it('keeps only a bounded number of entries resident (hot-page cache)', async () => {
|
||||
const dir = tmpDir();
|
||||
try {
|
||||
const vg = new ValueGraph({ store: createFileStore(dir, { maxResident: 2 }) });
|
||||
for (let i = 0; i < 5; i++) {
|
||||
vg.compute(`v${i}`, () => i * 10);
|
||||
}
|
||||
for (let i = 0; i < 5; i++) {
|
||||
assert.equal((await get(vg, 't', `v${i}`, {})).value, i * 10);
|
||||
}
|
||||
// All five entries persist on disk even though only 2 are resident.
|
||||
assert.equal(readdirSync(dir).filter((f) => f.endsWith('.bin')).length, 5);
|
||||
const again = await get(vg, 't', 'v0', {}); // evicted from cache → reloaded from disk
|
||||
assert.equal(again.value, 0);
|
||||
} finally {
|
||||
rmSync(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it('invalidate/set delete the persisted files', async () => {
|
||||
const dir = tmpDir();
|
||||
try {
|
||||
const vg = new ValueGraph({ store: createFileStore(dir) });
|
||||
vg.compute('balance', () => 100);
|
||||
assert.equal((await get(vg, 't:1', 'balance', {})).value, 100);
|
||||
assert.equal(readdirSync(dir).filter((f) => f.endsWith('.bin')).length, 1);
|
||||
|
||||
vg.invalidate('t:1', 'balance', {});
|
||||
assert.equal(readdirSync(dir).filter((f) => f.endsWith('.bin')).length, 0, 'invalidate removed the file');
|
||||
|
||||
vg.set('t:1', 'balance', {}, { value: 42, unit: 'usd_cents' });
|
||||
assert.equal(readdirSync(dir).filter((f) => f.endsWith('.bin')).length, 1, 'set persisted the file');
|
||||
assert.equal((await get(vg, 't:1', 'balance', {})).value, 42);
|
||||
} finally {
|
||||
rmSync(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it('honors the store contract for a query duplex on a disk-backed graph', async () => {
|
||||
const dir = tmpDir();
|
||||
try {
|
||||
const vg = new ValueGraph({ store: createFileStore(dir) });
|
||||
vg.compute('available', () => 88);
|
||||
const q = vg.query('tenant:acme', 'available', { feature: 'gpt-4' });
|
||||
const events = collect(q.source, { until: (items) => items.some((i) => i.value === 88) });
|
||||
q.sink.write({ get: true });
|
||||
await events;
|
||||
assert.ok(readdirSync(dir).some((f) => f.endsWith('.bin')), 'query result persisted to disk');
|
||||
} finally {
|
||||
rmSync(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
describe('ValueGraph operator model (ADR-000 OWA)', () => {
|
||||
function make(operators, parents) {
|
||||
const vg = new ValueGraph();
|
||||
for (const [rel, value] of Object.entries(parents)) {
|
||||
vg.define(rel, { operator: 'source', fn: () => value });
|
||||
}
|
||||
for (const [rel, op] of Object.entries(operators)) {
|
||||
vg.define(rel, { operator: `fusion:${op}`, parents: Object.keys(parents) });
|
||||
}
|
||||
return vg;
|
||||
}
|
||||
|
||||
it('applies the OWA default weight distributions', async () => {
|
||||
// values 10, 20, 30
|
||||
const vg = make({}, { a: 10, b: 20, c: 30 });
|
||||
vg.define('mx', { operator: 'fusion:max', parents: ['a', 'b', 'c'] });
|
||||
vg.define('mn', { operator: 'fusion:min', parents: ['a', 'b', 'c'] });
|
||||
vg.define('av', { operator: 'fusion:average', parents: ['a', 'b', 'c'] });
|
||||
vg.define('sm', { operator: 'fusion:sum_unbounded', parents: ['a', 'b', 'c'] });
|
||||
vg.define('md', { operator: 'fusion:median', parents: ['a', 'b', 'c'] });
|
||||
|
||||
assert.equal((await get(vg, 't', 'mx', {})).value, 30);
|
||||
assert.equal((await get(vg, 't', 'mn', {})).value, 10);
|
||||
assert.ok(Math.abs((await get(vg, 't', 'av', {})).value - 20) < 1e-9, 'average ≈ 20');
|
||||
assert.equal((await get(vg, 't', 'sm', {})).value, 60);
|
||||
assert.equal((await get(vg, 't', 'md', {})).value, 20);
|
||||
});
|
||||
|
||||
it('applies custom OWA weights', async () => {
|
||||
const vg = make({}, { a: 100, b: 0 });
|
||||
vg.define('custom', { operator: 'fusion:custom', parents: ['a', 'b'], weights: [0.75, 0.25] });
|
||||
// sorted [100, 0]; 0.75*100 + 0.25*0 = 75
|
||||
assert.equal((await get(vg, 't', 'custom', {})).value, 75);
|
||||
});
|
||||
|
||||
it('supports optimistic / pessimistic / top2', async () => {
|
||||
const vg = make({}, { a: 10, b: 20, c: 30 });
|
||||
vg.define('opt', { operator: 'fusion:optimistic', parents: ['a', 'b', 'c'] });
|
||||
vg.define('pes', { operator: 'fusion:pessimistic', parents: ['a', 'b', 'c'] });
|
||||
vg.define('t2', { operator: 'fusion:top2', parents: ['a', 'b', 'c'] });
|
||||
// optimistic [0.5,0.25,0.125]→ normalized; pes reversed; top2 = 0.5*30+0.5*20 = 25
|
||||
assert.ok((await get(vg, 't', 'opt', {})).value > 20, 'optimistic favors high values');
|
||||
assert.ok((await get(vg, 't', 'pes', {})).value < 20, 'pessimistic favors low values');
|
||||
assert.equal((await get(vg, 't', 't2', {})).value, 25);
|
||||
});
|
||||
|
||||
it('supports non-numeric max/min/majority (strings)', async () => {
|
||||
const vg = make({}, { a: 'low', b: 'high', c: 'high' });
|
||||
vg.define('mx', { operator: 'fusion:max', parents: ['a', 'b', 'c'] });
|
||||
vg.define('maj', { operator: 'fusion:majority', parents: ['a', 'b', 'c'] });
|
||||
assert.equal((await get(vg, 't', 'mx', {})).value, 'low' > 'high' ? 'low' : 'high');
|
||||
assert.equal((await get(vg, 't', 'maj', {})).value, 'high');
|
||||
});
|
||||
|
||||
it('computes via an operator node with parents (deps passed via ctx)', async () => {
|
||||
const vg = new ValueGraph();
|
||||
vg.define('base', { operator: 'source', fn: (s, p, ctx, cb) => cb(null, 7) });
|
||||
vg.define('twice', { operator: 'compute', parents: ['base'], fn: (subject, params, { deps }) => deps.base * 2 });
|
||||
assert.equal((await get(vg, 't', 'twice', {})).value, 14);
|
||||
});
|
||||
|
||||
it('reads an attribute via the resolveAttribute callback hook', async () => {
|
||||
const graph = new Map([['user:1', { balance: 500 }]]);
|
||||
const vg = new ValueGraph({
|
||||
resolveAttribute: (nodeKey, path, params, ctx, cb) => {
|
||||
const node = graph.get(nodeKey);
|
||||
cb(null, node ? node[path] : undefined);
|
||||
}
|
||||
});
|
||||
vg.define('balance', { operator: 'attribute', attribute: 'balance' });
|
||||
assert.equal((await get(vg, 'user:1', 'balance', {})).value, 500);
|
||||
});
|
||||
|
||||
it('resolves a pattern measure via the resolvePattern callback hook, lazily and cached', async () => {
|
||||
let calls = 0;
|
||||
const edges = [['u:1', 'r', 'x'], ['u:1', 'r', 'y'], ['u:1', 'r', 'z']];
|
||||
const vg = new ValueGraph({
|
||||
resolvePattern: (pattern, subject, params, ctx, cb) => {
|
||||
calls++;
|
||||
cb(null, edges.filter(e => e[0] === subject && e[1] === pattern.relation).length);
|
||||
}
|
||||
});
|
||||
vg.define('degree', { operator: 'pattern', pattern: { relation: 'r' } });
|
||||
assert.equal((await get(vg, 'u:1', 'degree', {})).value, 3);
|
||||
assert.equal((await get(vg, 'u:1', 'degree', {})).value, 3);
|
||||
assert.equal(calls, 1, 'pattern cached');
|
||||
});
|
||||
|
||||
it('invalidating a parent recomputes a fusion node', async () => {
|
||||
let b = 10;
|
||||
const vg = new ValueGraph();
|
||||
vg.define('a', { operator: 'source', fn: () => 5 });
|
||||
vg.define('base', { operator: 'source', fn: () => b });
|
||||
vg.define('sum', { operator: 'fusion:sum_unbounded', parents: ['a', 'base'] });
|
||||
assert.equal((await get(vg, 't', 'sum', {})).value, 15);
|
||||
b = 100;
|
||||
vg.invalidate('t', 'base', {});
|
||||
assert.equal((await get(vg, 't', 'sum', {})).value, 105, 'fusion recomputed after parent invalidation');
|
||||
});
|
||||
});
|
||||
|
||||
describe('ValueGraph blackbox node (function of its dependencies)', () => {
|
||||
it('receives parent values (ctx.deps) and full entries (ctx.entries)', async () => {
|
||||
const vg = new ValueGraph();
|
||||
vg.define('base', { operator: 'source', fn: (s, p, ctx, cb) => cb(null, { value: 21, unit: 'usd_cents' }) });
|
||||
let seen = null;
|
||||
vg.define('total', {
|
||||
operator: 'blackbox',
|
||||
parents: ['base'],
|
||||
fn: (subject, params, ctx, cb) => {
|
||||
seen = ctx;
|
||||
cb(null, ctx.deps.base * 2);
|
||||
}
|
||||
});
|
||||
|
||||
const entry = await get(vg, 't:1', 'total', {});
|
||||
assert.equal(entry.value, 42);
|
||||
assert.equal(seen.deps.base, 21, 'ctx.deps exposes parent VALUES');
|
||||
assert.equal(seen.entries.base.value, 21, 'ctx.entries exposes the full parent result');
|
||||
assert.equal(seen.entries.base.unit, 'usd_cents');
|
||||
assert.equal(seen.entries.base.source, 'source');
|
||||
assert.equal(seen.entries.base.fresh, true);
|
||||
});
|
||||
|
||||
it('queries an external service keyed on parent edges (callback style)', async () => {
|
||||
const vg = new ValueGraph();
|
||||
const edges = [['u:1', 'r', 'x'], ['u:1', 'r', 'y']];
|
||||
vg.define('related', {
|
||||
operator: 'pattern',
|
||||
pattern: { relation: 'r' },
|
||||
fn: (s, p, ctx, cb) => cb(null, edges.filter(e => e[0] === s))
|
||||
});
|
||||
// Fake Overlay: given the related edges, score the subject asynchronously.
|
||||
const overlayQuery = (edgeList, cb) => {
|
||||
setTimeout(() => cb(null, { value: edgeList.length * 10, unit: 'score', source: 'overlay:evals' }), 5);
|
||||
};
|
||||
vg.define('score', {
|
||||
operator: 'blackbox',
|
||||
parents: ['related'],
|
||||
fn: (subject, params, ctx, cb) => overlayQuery(ctx.deps.related, cb)
|
||||
});
|
||||
|
||||
const entry = await get(vg, 'u:1', 'score', {});
|
||||
assert.equal(entry.value, 20);
|
||||
assert.equal(entry.source, 'overlay:evals');
|
||||
});
|
||||
|
||||
it('recomputes a blackbox node when a parent edge set changes', async () => {
|
||||
const edges = [['u:1', 'r', 'x']];
|
||||
const vg = new ValueGraph();
|
||||
vg.define('related', {
|
||||
operator: 'pattern',
|
||||
pattern: { relation: 'r' },
|
||||
fn: (s, p, ctx, cb) => cb(null, edges.filter(e => e[0] === s))
|
||||
});
|
||||
vg.define('count', {
|
||||
operator: 'blackbox',
|
||||
parents: ['related'],
|
||||
fn: (s, p, { deps }) => deps.related.length
|
||||
});
|
||||
|
||||
assert.equal((await get(vg, 'u:1', 'count', {})).value, 1);
|
||||
edges.push(['u:1', 'r', 'y'], ['u:1', 'r', 'z']);
|
||||
vg.invalidate('u:1', 'related', {});
|
||||
assert.equal((await get(vg, 'u:1', 'count', {})).value, 3, 'blackbox re-run against the new edges');
|
||||
});
|
||||
|
||||
it('requires an fn', async () => {
|
||||
const vg = new ValueGraph();
|
||||
vg.define('broken', { operator: 'blackbox', parents: [] });
|
||||
await assert.rejects(() => get(vg, 't', 'broken', {}), /blackbox node 'broken' needs an fn/);
|
||||
});
|
||||
});
|
||||
|
||||
describe('ValueGraph compact snapshots (persistence & transport)', () => {
|
||||
function tmpDir() {
|
||||
return mkdtempSync(join(tmpdir(), 'value-graph-snap-'));
|
||||
}
|
||||
|
||||
// A graph whose entries exercise every value type: number, string,
|
||||
// boolean, buffer (Uint8Array), array (with nested object + buffer), interval.
|
||||
// NOTE: no JS bigint — large integers are Buffers (the wire is JSON-free).
|
||||
function richGraph({ store, defaultTTL = 0 } = {}) {
|
||||
const vg = new ValueGraph({ store, defaultTTL });
|
||||
vg.define('num', { operator: 'source', fn: () => 42 });
|
||||
vg.define('str', { operator: 'source', fn: () => 'hello' });
|
||||
vg.define('bool', { operator: 'source', fn: () => true });
|
||||
vg.define('buf', { operator: 'source', fn: () => new Uint8Array([1, 2, 3, 254, 255]) });
|
||||
vg.define('arr', { operator: 'source', fn: () => [1, 'two', new Uint8Array([3, 4]), { nested: true }] });
|
||||
vg.define('iv', { operator: 'source', fn: () => ({ lower: 0.5, upper: 1.5 }) });
|
||||
vg.define('fusion_risk', {
|
||||
operator: 'fusion:custom', parents: ['num', 'buf'], weights: [0.25, 0.75],
|
||||
capSum: true, returnType: 'number'
|
||||
});
|
||||
return vg;
|
||||
}
|
||||
|
||||
it('buffer (Uint8Array) values round-trip byte-exactly through the wire', async () => {
|
||||
const vg = richGraph();
|
||||
await get(vg, 't', 'buf', {});
|
||||
const buf = vg.snapshot();
|
||||
const snap = decodeGraphSnapshot(buf);
|
||||
const value = snap.records.find((r) => r.relation === 'buf').entry.value;
|
||||
assert.deepEqual([...value], [1, 2, 3, 254, 255]);
|
||||
});
|
||||
|
||||
it('value-level round-trip: encodeSnapshot/decodeSnapshot is exact for all value types', async () => {
|
||||
const vg = richGraph();
|
||||
// num with a nested-params key (params round-trip through _parseKey, JSON-free)
|
||||
await get(vg, 't', 'num', { x: [1, { y: 2 }] });
|
||||
await get(vg, 't', 'str', {});
|
||||
await get(vg, 't', 'bool', {});
|
||||
await get(vg, 't', 'buf', {});
|
||||
await get(vg, 't', 'arr', {});
|
||||
await get(vg, 't', 'iv', {});
|
||||
|
||||
const buf = vg.snapshot();
|
||||
assert.ok(buf.length > 0);
|
||||
assert.equal(Buffer.from(buf.subarray(0, 4)).toString('ascii'), 'VGGP', 'graph magic');
|
||||
const snap = decodeGraphSnapshot(buf);
|
||||
assert.equal(snap.records.length, 6);
|
||||
assert.equal(snap.schema.length, 7);
|
||||
const byRel = new Map(snap.records.map((r) => [r.relation, r.entry]));
|
||||
assert.equal(byRel.get('num').value, 42);
|
||||
assert.equal(byRel.get('str').value, 'hello');
|
||||
assert.equal(byRel.get('bool').value, true);
|
||||
assert.deepEqual([...byRel.get('buf').value], [1, 2, 3, 254, 255]);
|
||||
assert.deepEqual(byRel.get('arr').value[0], 1);
|
||||
assert.equal(byRel.get('arr').value[1], 'two');
|
||||
assert.deepEqual([...byRel.get('arr').value[2]], [3, 4]);
|
||||
assert.deepEqual(byRel.get('arr').value[3], { nested: true });
|
||||
assert.deepEqual(byRel.get('iv').value, { lower: 0.5, upper: 1.5 });
|
||||
assert.deepEqual(snap.records.find((r) => r.relation === 'num').params, { x: [1, { y: 2 }] }, 'nested params round-trip');
|
||||
});
|
||||
|
||||
it('graph snapshot restores into a fresh graph: values served WITHOUT recompute', async () => {
|
||||
let calls = 0;
|
||||
const vg = richGraph();
|
||||
await get(vg, 't', 'num', {});
|
||||
await get(vg, 't', 'buf', {});
|
||||
const buf = vg.snapshot();
|
||||
|
||||
const restored = ValueGraph.restore(buf, {
|
||||
resolvers: { num: () => { calls++; return -1; } }
|
||||
});
|
||||
assert.equal(calls, 0);
|
||||
const e = await get(restored, 't', 'num', {});
|
||||
assert.equal(e.value, 42);
|
||||
assert.equal(e.fresh, true, 'restored entry treated as fresh (version re-stamped)');
|
||||
assert.deepEqual([...(await get(restored, 't', 'buf', {})).value], [1, 2, 3, 254, 255]);
|
||||
assert.equal(restored.relationSpec('fusion_risk').operator, 'fusion:custom');
|
||||
assert.deepEqual(restored.relationSpec('fusion_risk').parents, ['num', 'buf']);
|
||||
assert.equal(restored.relationSpec('fusion_risk').returnType, 'number');
|
||||
});
|
||||
|
||||
it('schema metadata round-trips: operator, parents, weights, capSum, returnType', async () => {
|
||||
const vg = new ValueGraph({ defaultTTL: 0 });
|
||||
vg.define('s', { operator: 'source', fn: () => 1 });
|
||||
vg.define('f', {
|
||||
operator: 'fusion:priority', parents: ['s'], priorities: [0.4],
|
||||
ttl: 5000, returnType: 'number', params: [{ name: 'p', type: 'number' }]
|
||||
});
|
||||
await get(vg, 't', 's', {});
|
||||
const restored = ValueGraph.restore(vg.snapshot(), { resolvers: { s: () => 1 } });
|
||||
const spec = restored.relationSpec('f');
|
||||
assert.equal(spec.operator, 'fusion:priority');
|
||||
assert.deepEqual(spec.parents, ['s']);
|
||||
assert.equal(spec.ttl, 5000);
|
||||
assert.equal(spec.returnType, 'number');
|
||||
assert.deepEqual(spec.params, [{ name: 'p', type: 'number' }]);
|
||||
});
|
||||
|
||||
it('loadFile/snapshotFile: portable single-file transport between "processes"', async () => {
|
||||
const dir = tmpDir();
|
||||
try {
|
||||
// Process 1: disk-backed graph computes values, writes ONE portable file.
|
||||
const vg1 = new ValueGraph({ store: createFileStore(dir), defaultTTL: 0 });
|
||||
vg1.compute('double_balance', (s, p, { deps }) => deps.base_balance * 2, { dependsOn: ['base_balance'] });
|
||||
vg1.compute('base_balance', (s, p, ctx, cb) => cb(null, 21));
|
||||
assert.equal((await get(vg1, 't:1', 'double_balance', {})).value, 42);
|
||||
const snapPath = join(dir, 'graph.snap');
|
||||
vg1.snapshotFile(snapPath);
|
||||
assert.ok(existsSync(snapPath));
|
||||
|
||||
// Process 2: brand-new graph, EMPTY store, reads the portable file.
|
||||
// Resolvers re-registered; the resolver would throw if recomputed.
|
||||
const loaded = ValueGraph.loadFile(snapPath, {
|
||||
resolvers: { base_balance: () => { throw new Error('recompute!'); } }
|
||||
});
|
||||
assert.equal(loaded.relations.size, 2, 'schema restored');
|
||||
const e = await get(loaded, 't:1', 'double_balance', {});
|
||||
assert.equal(e.value, 42);
|
||||
assert.equal(e.fresh, true, 'no recompute — cache restored from disk');
|
||||
} finally {
|
||||
rmSync(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it('store-level snapshot: opaque key→entry dump round-trips through any store', async () => {
|
||||
const vg = new ValueGraph({ defaultTTL: 0 });
|
||||
vg.compute('balance', (s, p, { deps }) => deps.rate * deps.units, { dependsOn: ['rate', 'units'] });
|
||||
vg.compute('rate', () => 3);
|
||||
vg.compute('units', () => 4);
|
||||
assert.equal((await get(vg, 'a', 'balance', {})).value, 12);
|
||||
|
||||
const storeBuf = vg.snapshot({ includeSchema: false });
|
||||
const snap = decodeSnapshot(storeBuf);
|
||||
assert.equal(snap.entries.length, 3);
|
||||
assert.equal(Buffer.from(storeBuf.subarray(0, 4)).toString('ascii'), 'VGST', 'store magic');
|
||||
|
||||
// Feed the raw dump into a fresh map store, then serve from it.
|
||||
const fresh = new ValueGraph({ store: createMapStore(), defaultTTL: 0 });
|
||||
for (const [key, entry] of snap.entries) fresh.store.set(key, entry);
|
||||
assert.equal((await get(fresh, 'a', 'balance', {})).value, 12, 'opaque dump replays into a new store');
|
||||
});
|
||||
|
||||
it('createFileStore loadFrom: a store constructed already loaded from a snapshot', async () => {
|
||||
const dirA = tmpDir();
|
||||
const dirB = tmpDir();
|
||||
try {
|
||||
const vg = new ValueGraph({ store: createFileStore(dirA), defaultTTL: 0 });
|
||||
vg.compute('balance', () => 99);
|
||||
assert.equal((await get(vg, 'u:1', 'balance', {})).value, 99);
|
||||
const snapPath = join(dirA, 'store.snap');
|
||||
vg.store.snapshotFile(snapPath);
|
||||
|
||||
// New store in a DIFFERENT directory, constructed with loadFrom.
|
||||
const store = createFileStore(dirB, { loadFrom: snapPath });
|
||||
assert.ok(store.get(vg._key('u:1', 'balance', {})) !== undefined, 'snapshot preloaded');
|
||||
const vg2 = new ValueGraph({ store, defaultTTL: 0 });
|
||||
vg2.compute('balance', () => { throw new Error('recompute!'); });
|
||||
assert.equal((await get(vg2, 'u:1', 'balance', {})).value, 99, 'served from the loaded snapshot');
|
||||
assert.ok(readdirSync(dirB).filter((f) => f.endsWith('.bin')).length >= 1, 'materialized into working dir');
|
||||
} finally {
|
||||
rmSync(dirA, { recursive: true, force: true });
|
||||
rmSync(dirB, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it('corrupt or truncated snapshots are rejected (CRC guard), never silently mis-read', async () => {
|
||||
const vg = new ValueGraph({ defaultTTL: 0 });
|
||||
vg.compute('balance', () => 7);
|
||||
await get(vg, 'u', 'balance', {});
|
||||
const buf = vg.snapshot();
|
||||
assert.doesNotThrow(() => decodeGraphSnapshot(buf));
|
||||
|
||||
// Flip a byte inside createdAt: parse succeeds, CRC must reject it.
|
||||
const flipped = Buffer.from(buf);
|
||||
flipped[9] ^= 0xff;
|
||||
assert.throws(() => decodeGraphSnapshot(flipped), /CRC mismatch/);
|
||||
|
||||
// Flip a byte inside a record payload: the parser may desync — the result
|
||||
// must be a coherent corruption error, never a silent wrong answer.
|
||||
const payloadFlip = Buffer.from(buf);
|
||||
payloadFlip[payloadFlip.length - 8] ^= 0xff;
|
||||
assert.throws(() => decodeGraphSnapshot(payloadFlip), /CRC mismatch|corrupt/);
|
||||
|
||||
const truncated = Buffer.from(buf.subarray(0, buf.length - 4)); // missing CRC
|
||||
assert.throws(() => decodeGraphSnapshot(truncated), /truncated/);
|
||||
|
||||
const badMagic = Buffer.from(buf);
|
||||
badMagic[0] = 0x00;
|
||||
assert.throws(() => decodeGraphSnapshot(badMagic), /not a value-graph snapshot/);
|
||||
|
||||
// Store flavour gets the same guards.
|
||||
const storeBuf = vg.snapshot({ includeSchema: false });
|
||||
assert.doesNotThrow(() => decodeSnapshot(storeBuf));
|
||||
assert.throws(() => decodeSnapshot(Buffer.from(storeBuf.subarray(0, storeBuf.length - 5))), /truncated|CRC|corrupt/);
|
||||
});
|
||||
|
||||
it('instance restore merges a snapshot into a live graph (schema declared if absent)', async () => {
|
||||
const src = new ValueGraph({ defaultTTL: 0 });
|
||||
src.compute('rate', () => 5);
|
||||
assert.equal((await get(src, 'u', 'rate', {})).value, 5);
|
||||
|
||||
const dst = new ValueGraph({ defaultTTL: 0 });
|
||||
dst.compute('quota', () => 100);
|
||||
await get(dst, 'u', 'quota', {});
|
||||
dst.restore(src.snapshot());
|
||||
|
||||
assert.equal(dst.relations.has('rate'), true, 'foreign schema declared');
|
||||
assert.equal((await get(dst, 'u', 'rate', {})).value, 5, 'foreign entries loaded');
|
||||
assert.equal((await get(dst, 'u', 'quota', {})).value, 100, 'local entries untouched');
|
||||
});
|
||||
|
||||
it('disk-backed store snapshot is byte-identical to the in-memory encode', async () => {
|
||||
const dir = tmpDir();
|
||||
try {
|
||||
const vg = new ValueGraph({ store: createFileStore(dir), defaultTTL: 0 });
|
||||
vg.compute('balance', () => 11);
|
||||
vg.compute('tag', () => 'x');
|
||||
await get(vg, 'u:1', 'balance', {});
|
||||
await get(vg, 'u:1', 'tag', {});
|
||||
|
||||
const snapPath = join(dir, 'g.snap');
|
||||
vg.snapshotFile(snapPath, { createdAt: 12345 });
|
||||
const fromFile = readFileSync(snapPath);
|
||||
const fromBuffer = vg.snapshot({ createdAt: 12345 });
|
||||
// Same schema+records, deterministic createdAt + CRC → identical bytes.
|
||||
assert.deepEqual([...fromFile], [...fromBuffer], 'file snapshot == buffer snapshot');
|
||||
} finally {
|
||||
rmSync(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user