js-rigor: batch cache staleness, tx-batch crash, TTL bypass; value freshness campaigns
Engine fixes: - RelationUpdates.updateRelationsBatch: invalidate arbiter-level caches (rule result cache, ChainRule caches, direct-check cache) per affected relation — batch updates bypassed Arbiter.addRelation and served stale decisions after batch modify/swap with warm caches - updateRelationsBatchTransactional rollback: new Map(Set) crashed with 'Iterator value is not an entry object' — fixed to new Set - RelationalComparatorRule: value extraction (direct-list and cached direct paths) now gates on valueManager._isValueExpired — TTL-expired values no longer feed comparator decisions Campaigns: - value-freshness-parity.test.js: batch modify/swap/tx rollback freshness with comparator mirror (batch MODIFY of a missing relation is a silent no-op — pinned) - ttl-expiry-parity.test.js: injected-clock TTL expiry through the comparator path (exact parity with caching off; bounded staleness with caching on), faithful changed_last_at mirror semantics
This commit is contained in:
John Dvorak
@@ -387,6 +387,7 @@ export class RelationalComparatorRule extends BaseRule {
|
||||
}
|
||||
const rel = this.arbiter.relationManager.getDirectRelation(srcId, relName, dstId, options);
|
||||
if (!rel || typeof rel.value !== 'number') continue;
|
||||
if (this.arbiter.valueManager && this.arbiter.valueManager._isValueExpired(rel)) continue;
|
||||
|
||||
valueResults.push({
|
||||
value: rel.value,
|
||||
@@ -780,6 +781,9 @@ export class RelationalComparatorRule extends BaseRule {
|
||||
}
|
||||
|
||||
_getCachedDirectValue(relation, ttl) {
|
||||
if (this.arbiter.valueManager && this.arbiter.valueManager._isValueExpired(relation)) {
|
||||
return null;
|
||||
}
|
||||
const cacheKey = `${relation.src}|${relation.rel}|${relation.dst}`;
|
||||
const cached = this._directValueCache.get(cacheKey);
|
||||
if (cached && cached.stateId === relation.stateId) {
|
||||
|
||||
@@ -409,7 +409,23 @@ export class RelationUpdates {
|
||||
this._processBatchRemovals(removeOps, chunkSize);
|
||||
this._processBatchModifications(modifyOps, chunkSize);
|
||||
this._processBatchAdditions(addOps, chunkSize);
|
||||
|
||||
|
||||
// The batch path bypasses Arbiter.addRelation/removeRelation, so the
|
||||
// arbiter-level rule caches (rule result cache, ChainRule caches,
|
||||
// direct-check cache) are never invalidated there. A query that warmed
|
||||
// those caches before the batch would keep serving stale decisions.
|
||||
const affectedRelations = new Set();
|
||||
for (const op of updates) {
|
||||
if (op && op.relation !== undefined) affectedRelations.add(op.relation);
|
||||
}
|
||||
for (const relation of affectedRelations) {
|
||||
this.manager.arbiter._invalidateDirectCheckCache(null, relation, null);
|
||||
this.manager.arbiter.invalidateRuleResultCacheByRelation(relation);
|
||||
if (this.manager.arbiter.authChecker) {
|
||||
this.manager.arbiter.authChecker.invalidateRuleCaches(relation);
|
||||
}
|
||||
}
|
||||
|
||||
return this.manager;
|
||||
}
|
||||
|
||||
@@ -596,7 +612,7 @@ export class RelationUpdates {
|
||||
outgoingEdges: new Map(this.manager.arbiter.indices.outgoingEdges),
|
||||
incomingEdges: new Map(this.manager.arbiter.indices.incomingEdges)
|
||||
};
|
||||
const originalRelationKeys = new Map(this.manager._relationKeys);
|
||||
const originalRelationKeys = new Set(this.manager._relationKeys);
|
||||
const originalRelationKeyToIndex = new Map(this.manager._relationKeyToIndex);
|
||||
|
||||
try {
|
||||
|
||||
@@ -0,0 +1,157 @@
|
||||
/**
|
||||
* rigor/ttl-expiry-parity.test.js — value-TTL expiry through the full
|
||||
* comparator check path with an injected clock.
|
||||
*
|
||||
* Contracts pinned:
|
||||
* - values written with setTTL(rel, ttlMs) expire after ttlMs of engine
|
||||
* time; expired operands no longer participate in relational
|
||||
* comparisons (the comparator's value extraction must consult the
|
||||
* value manager's TTL gate, not raw relation values).
|
||||
* - with the decision cache enabled, a decision cached before expiry may
|
||||
* linger up to the rule-result-cache TTL, but MUST become fresh once
|
||||
* the cache entry itself expires (bounded staleness).
|
||||
*
|
||||
* The mirror tracks per-relation write timestamps under the same injected
|
||||
* clock and computes the expected comparison result.
|
||||
*/
|
||||
import { describe, it } from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { rigor } from '@rigor/core';
|
||||
import { Arbiter } from '../../src/index.js';
|
||||
|
||||
const TTL = 5000;
|
||||
const DOCS = 2;
|
||||
const docKey = (i) => `doc:${i}`;
|
||||
|
||||
const realNow = Date.now;
|
||||
let engineNow = 1_000_000_000_000;
|
||||
Date.now = () => engineNow;
|
||||
|
||||
function makeWrapper(disableCaching) {
|
||||
const arbiter = new Arbiter(disableCaching ? { disableCaching: true } : {});
|
||||
arbiter.addNode('user:alice', 'user');
|
||||
for (let i = 0; i < DOCS; i++) arbiter.addNode(docKey(i), 'doc');
|
||||
arbiter.setRelationConfig('has_balance', { type: 'direct' });
|
||||
arbiter.setRelationConfig('has_price', { type: 'direct' });
|
||||
arbiter.setRelationConfig('premium', {
|
||||
type: 'relational_comparator',
|
||||
comparator: '>',
|
||||
left: { rule: { type: 'direct', relation: 'has_balance' }, extractValue: true },
|
||||
right: { evaluateFrom: 'object', rule: { type: 'direct', relation: 'has_price' }, extractValue: true }
|
||||
});
|
||||
arbiter.valueManager.setTTL('has_balance', TTL);
|
||||
arbiter.valueManager.setTTL('has_price', TTL);
|
||||
|
||||
const balance = new Map();
|
||||
const price = new Map();
|
||||
const ops = [];
|
||||
|
||||
const wrapper = {
|
||||
engine: arbiter,
|
||||
balance,
|
||||
price,
|
||||
setValue(doc, kind, value) {
|
||||
const rel = kind === 'balance' ? 'has_balance' : 'has_price';
|
||||
const src = kind === 'balance' ? 'user:alice' : docKey(doc);
|
||||
// Pin the write timestamp: clone-replay re-writes relations with the
|
||||
// CURRENT Date.now(), which would silently un-expire old values; the
|
||||
// engine honors the changed_last_at override.
|
||||
arbiter.addRelation(src, rel, docKey(doc), { value, possibility: 1.0, changed_last_at: engineNow });
|
||||
// The engine only refreshes changed_last_at when the value actually
|
||||
// changes; the mirror must mirror that.
|
||||
const target = kind === 'balance' ? balance : price;
|
||||
const existing = target.get(doc);
|
||||
if (!existing || existing.value !== value) target.set(doc, { value, ts: engineNow });
|
||||
return { ok: true };
|
||||
},
|
||||
advanceTime(ms) {
|
||||
engineNow += ms;
|
||||
return { ok: true, now: engineNow };
|
||||
},
|
||||
check(doc) {
|
||||
const result = arbiter.check('user:alice', 'premium', docKey(doc));
|
||||
const b = balance.get(doc);
|
||||
const p = price.get(doc);
|
||||
const bFresh = b !== undefined && engineNow - b.ts <= TTL;
|
||||
const pFresh = p !== undefined && engineNow - p.ts <= TTL;
|
||||
const expected = bFresh && pFresh && b.value > p.value ? 1 : 0;
|
||||
return { engine: result.possibility, expected, reason: result.reason, bFresh, pFresh };
|
||||
},
|
||||
clone() {
|
||||
const fresh = makeWrapper(disableCaching);
|
||||
for (const op of ops) {
|
||||
const [name, ...args] = op;
|
||||
fresh[name](...args);
|
||||
}
|
||||
return fresh;
|
||||
}
|
||||
};
|
||||
|
||||
const record = (name, fn) => (...args) => {
|
||||
const res = fn(...args);
|
||||
ops.push([name, ...args]);
|
||||
return res;
|
||||
};
|
||||
wrapper.setValue = record('setValue', wrapper.setValue);
|
||||
wrapper.advanceTime = record('advanceTime', wrapper.advanceTime);
|
||||
wrapper.check = record('check', wrapper.check);
|
||||
return wrapper;
|
||||
}
|
||||
|
||||
describe('Value TTL expiry through the comparator path (rigor)', () => {
|
||||
it('FIXED MATRIX: expired operands flip the decision; cache staleness is bounded', () => {
|
||||
{
|
||||
const w = makeWrapper(true);
|
||||
w.setValue(0, 'balance', 100);
|
||||
w.setValue(0, 'price', 50);
|
||||
assert.equal(w.check(0).engine, 1, 'fresh: 100 > 50');
|
||||
|
||||
w.advanceTime(TTL + 1000);
|
||||
assert.equal(w.check(0).engine, 0, 'after TTL both operands expired -> deny');
|
||||
|
||||
w.setValue(0, 'price', 200);
|
||||
assert.equal(w.check(0).engine, 0, 'balance expired, price fresh -> deny');
|
||||
|
||||
w.setValue(0, 'balance', 300);
|
||||
assert.equal(w.check(0).engine, 1, 'both refreshed -> allow');
|
||||
|
||||
const cw = makeWrapper(false);
|
||||
cw.setValue(0, 'balance', 100);
|
||||
cw.setValue(0, 'price', 50);
|
||||
assert.equal(cw.check(0).engine, 1, 'cached arbiter: allow cached');
|
||||
cw.advanceTime(TTL + 1000);
|
||||
const staleWindow = cw.check(0);
|
||||
cw.advanceTime(70000); // past ruleResultCacheTTL (60s)
|
||||
assert.equal(cw.check(0).engine, 0, 'bounded staleness: fresh deny after cache TTL');
|
||||
assert.ok(staleWindow.engine >= 0, 'stale window decision is well-formed');
|
||||
}
|
||||
});
|
||||
|
||||
it('PROPERTY CAMPAIGN: parity holds across time advances and refreshes (no decision cache)', async () => {
|
||||
const result = await rigor.campaign(
|
||||
[rigor.object('graph', makeWrapper.bind(null, true), [
|
||||
rigor.method('setValue', function (d, k, v) { return this.setValue(d, k, v); },
|
||||
rigor.args(rigor.gen.int(0, DOCS - 1), rigor.gen.enum(['balance', 'price']), rigor.gen.oneOf([5, 50, 100, 200]))),
|
||||
rigor.method('advanceTime', function (ms) { return this.advanceTime(ms); },
|
||||
rigor.args(rigor.gen.oneOf([1000, 4000, 6000, 30000, 70000]))),
|
||||
rigor.method('check', function (d) { return this.check(d); },
|
||||
rigor.args(rigor.gen.int(0, DOCS - 1)))
|
||||
])],
|
||||
rigor.crucible([
|
||||
rigor.invariant('comparator parity under time', (ctx) => {
|
||||
if (ctx.action !== 'graph.check' || ctx.error !== null) return true;
|
||||
return ctx.actual.engine === ctx.actual.expected;
|
||||
}),
|
||||
rigor.invariant('no action errors', (ctx) => ctx.error === null)
|
||||
])
|
||||
).run({ effort: 400, seed: 'ttl-expiry-parity', maxTraceLength: 30 });
|
||||
|
||||
const inv = result.crucibleVerdict;
|
||||
assert.equal(inv.passed, true, [
|
||||
`TTL parity violated in ${inv.failureCount} cases:`,
|
||||
...result.failures.slice(0, 3).map((f) =>
|
||||
` [${f.invariant}] action=${f.action} args=${JSON.stringify(f.args)} actual=${JSON.stringify(f.actual)} error=${f.error}`
|
||||
)
|
||||
].join('\n'));
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,203 @@
|
||||
/**
|
||||
* rigor/value-freshness-parity.test.js — value-layer and batch-path cache
|
||||
* freshness.
|
||||
*
|
||||
* The needle class: queries that warm the rule-result / decision caches,
|
||||
* followed by mutations applied through the BATCH path (which bypasses
|
||||
* Arbiter.addRelation/removeRelation and their cache invalidation), must
|
||||
* still return fresh answers:
|
||||
* - comparator decisions reflect the latest values after batch modify
|
||||
* and batch remove+add;
|
||||
* - direct checks reflect batch adds/removes with warm caches;
|
||||
* - transactional batches either fully apply (success) or fully roll
|
||||
* back (poisoned op -> state unchanged, checks unchanged).
|
||||
*
|
||||
* The mirror tracks balance/price per document; comparator expectation is
|
||||
* balance > price (engine returns 1/0 with reasons
|
||||
* allow_rule_matched / values_compared_comparison_false).
|
||||
*/
|
||||
import { describe, it } from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { rigor } from '@rigor/core';
|
||||
import { Arbiter } from '../../src/index.js';
|
||||
|
||||
const DOCS = 3;
|
||||
const docKey = (i) => `doc:${i}`;
|
||||
|
||||
function makeWrapper() {
|
||||
const arbiter = new Arbiter();
|
||||
arbiter.addNode('user:alice', 'user');
|
||||
for (let i = 0; i < DOCS; i++) arbiter.addNode(docKey(i), 'doc');
|
||||
arbiter.setRelationConfig('has_balance', { type: 'direct' });
|
||||
arbiter.setRelationConfig('has_price', { type: 'direct' });
|
||||
arbiter.setRelationConfig('can_read', { type: 'direct', relation: 'owner' });
|
||||
arbiter.setRelationConfig('premium', {
|
||||
type: 'relational_comparator',
|
||||
comparator: '>',
|
||||
left: { rule: { type: 'direct', relation: 'has_balance' }, extractValue: true },
|
||||
right: { evaluateFrom: 'object', rule: { type: 'direct', relation: 'has_price' }, extractValue: true }
|
||||
});
|
||||
|
||||
const balance = new Map();
|
||||
const price = new Map();
|
||||
const ops = [];
|
||||
|
||||
const wrapper = {
|
||||
engine: arbiter,
|
||||
balance,
|
||||
price,
|
||||
setValue(doc, kind, value) {
|
||||
const rel = kind === 'balance' ? 'has_balance' : 'has_price';
|
||||
const src = kind === 'balance' ? 'user:alice' : docKey(doc);
|
||||
arbiter.addRelation(src, rel, docKey(doc), { value, possibility: 1.0 });
|
||||
(kind === 'balance' ? balance : price).set(doc, value);
|
||||
return { ok: true };
|
||||
},
|
||||
batchSetValue(doc, kind, value) {
|
||||
const rel = kind === 'balance' ? 'has_balance' : 'has_price';
|
||||
const src = kind === 'balance' ? 'user:alice' : docKey(doc);
|
||||
const target = kind === 'balance' ? balance : price;
|
||||
const batch = [{ operation: 'modify', srcKey: src, relation: rel, dstKey: docKey(doc), options: { value, possibility: 1.0 } }];
|
||||
arbiter.relationManager.updateRelationsBatch(batch);
|
||||
// A batch MODIFY of a relation that does not exist is a silent no-op
|
||||
// in the engine (mirrors _modifyRelation), so the mirror must only
|
||||
// record when the relation was already present.
|
||||
if (target.has(doc)) target.set(doc, value);
|
||||
return { ok: true };
|
||||
},
|
||||
batchSwap(doc, kind, value) {
|
||||
const rel = kind === 'balance' ? 'has_balance' : 'has_price';
|
||||
const src = kind === 'balance' ? 'user:alice' : docKey(doc);
|
||||
const batch = [
|
||||
{ operation: 'remove', srcKey: src, relation: rel, dstKey: docKey(doc), options: {} },
|
||||
{ operation: 'add', srcKey: src, relation: rel, dstKey: docKey(doc), options: { value, possibility: 1.0 } }
|
||||
];
|
||||
arbiter.relationManager.updateRelationsBatch(batch);
|
||||
(kind === 'balance' ? balance : price).set(doc, value);
|
||||
return { ok: true };
|
||||
},
|
||||
txBatch(doc, kind, value, poison) {
|
||||
const rel = kind === 'balance' ? 'has_balance' : 'has_price';
|
||||
const src = kind === 'balance' ? 'user:alice' : docKey(doc);
|
||||
const batch = [{
|
||||
operation: 'modify',
|
||||
srcKey: src,
|
||||
relation: rel,
|
||||
dstKey: docKey(doc),
|
||||
options: poison ? { value, possibility: 1.5 } : { value, possibility: 1.0 }
|
||||
}];
|
||||
const result = arbiter.relationManager.updateRelationsBatchTransactional(batch);
|
||||
if (result.success) {
|
||||
const target = kind === 'balance' ? balance : price;
|
||||
if (target.has(doc)) target.set(doc, value);
|
||||
}
|
||||
return { success: result.success, error: result.error };
|
||||
},
|
||||
check(doc) {
|
||||
const result = arbiter.check('user:alice', 'premium', docKey(doc));
|
||||
const b = balance.get(doc);
|
||||
const p = price.get(doc);
|
||||
const expected = b !== undefined && p !== undefined && b > p ? 1 : 0;
|
||||
return {
|
||||
engine: result.possibility,
|
||||
expected,
|
||||
reason: result.reason,
|
||||
balance: b,
|
||||
price: p
|
||||
};
|
||||
},
|
||||
checkRead(doc) {
|
||||
const result = arbiter.check('user:alice', 'can_read', docKey(doc));
|
||||
return { engine: result.possibility, reason: result.reason };
|
||||
},
|
||||
clone() {
|
||||
const fresh = makeWrapper();
|
||||
for (const op of ops) {
|
||||
const [name, ...args] = op;
|
||||
fresh[name](...args);
|
||||
}
|
||||
return fresh;
|
||||
}
|
||||
};
|
||||
|
||||
const record = (name, fn) => (...args) => {
|
||||
const res = fn(...args);
|
||||
ops.push([name, ...args]);
|
||||
return res;
|
||||
};
|
||||
wrapper.setValue = record('setValue', wrapper.setValue);
|
||||
wrapper.batchSetValue = record('batchSetValue', wrapper.batchSetValue);
|
||||
wrapper.batchSwap = record('batchSwap', wrapper.batchSwap);
|
||||
wrapper.txBatch = record('txBatch', wrapper.txBatch);
|
||||
wrapper.check = record('check', wrapper.check);
|
||||
wrapper.checkRead = record('checkRead', wrapper.checkRead);
|
||||
return wrapper;
|
||||
}
|
||||
|
||||
const docArg = rigor.gen.int(0, DOCS - 1);
|
||||
const kindArg = rigor.gen.enum(['balance', 'price']);
|
||||
const valueArg = rigor.gen.oneOf([5, 10, 50, 100, 200]);
|
||||
|
||||
describe('Value-layer and batch-path freshness (rigor)', () => {
|
||||
it('FIXED MATRIX: batch modify, batch swap, and transactional rollback stay fresh', () => {
|
||||
const w = makeWrapper();
|
||||
w.setValue(0, 'balance', 100);
|
||||
w.setValue(0, 'price', 50);
|
||||
assert.equal(w.check(0).engine, 1, 'v1: 100 > 50');
|
||||
|
||||
w.batchSetValue(0, 'balance', 10);
|
||||
assert.equal(w.check(0).engine, 0, 'batch modify must be fresh: 10 > 50 is false');
|
||||
|
||||
w.batchSwap(0, 'balance', 200);
|
||||
assert.equal(w.check(0).engine, 1, 'batch remove+add must be fresh: 200 > 50');
|
||||
|
||||
const tx = w.txBatch(0, 'balance', 1, true);
|
||||
assert.equal(tx.success, false, 'poisoned tx must report failure');
|
||||
assert.equal(w.check(0).engine, 1, 'tx rollback must leave state unchanged: 200 > 50');
|
||||
|
||||
const txOk = w.txBatch(0, 'balance', 5, false);
|
||||
assert.equal(txOk.success, true, 'clean tx must succeed');
|
||||
assert.equal(w.check(0).engine, 0, 'clean tx must apply: 5 > 50 is false');
|
||||
|
||||
w.engine.addRelation('user:alice', 'owner', 'doc:0', { possibility: 0.9 });
|
||||
assert.equal(w.checkRead(0).engine, 0.9, 'direct check before batch');
|
||||
w.engine.relationManager.updateRelationsBatch([
|
||||
{ operation: 'remove', srcKey: 'user:alice', relation: 'owner', dstKey: 'doc:0', options: {} }
|
||||
]);
|
||||
assert.equal(w.checkRead(0).engine, 0, 'batch remove must invalidate warm direct caches');
|
||||
});
|
||||
|
||||
it('PROPERTY CAMPAIGN: interleaved single/batch/tx mutations keep comparator parity', async () => {
|
||||
const result = await rigor.campaign(
|
||||
[rigor.object('graph', makeWrapper, [
|
||||
rigor.method('setValue', function (d, k, v) { return this.setValue(d, k, v); }, rigor.args(docArg, kindArg, valueArg)),
|
||||
rigor.method('batchSetValue', function (d, k, v) { return this.batchSetValue(d, k, v); }, rigor.args(docArg, kindArg, valueArg)),
|
||||
rigor.method('batchSwap', function (d, k, v) { return this.batchSwap(d, k, v); }, rigor.args(docArg, kindArg, valueArg)),
|
||||
rigor.method('txBatch', function (d, k, v, poison) { return this.txBatch(d, k, v, poison); }, rigor.args(docArg, kindArg, valueArg, rigor.gen.boolean())),
|
||||
rigor.method('check', function (d) { return this.check(d); }, rigor.args(docArg))
|
||||
])],
|
||||
rigor.crucible([
|
||||
rigor.invariant('comparator parity after every mutation', (ctx) => {
|
||||
if (ctx.action !== 'graph.check') return true;
|
||||
if (ctx.error !== null) return true;
|
||||
return ctx.actual.engine === ctx.actual.expected;
|
||||
}),
|
||||
rigor.invariant('no action errors', (ctx) => {
|
||||
if (ctx.action === 'graph.txBatch') return true;
|
||||
return ctx.error === null;
|
||||
}),
|
||||
rigor.after('graph.txBatch', ({ actual }) => {
|
||||
return typeof actual.success === 'boolean';
|
||||
})
|
||||
])
|
||||
).run({ effort: 400, seed: 'value-freshness-parity', maxTraceLength: 30 });
|
||||
|
||||
const inv = result.crucibleVerdict;
|
||||
assert.equal(inv.passed, true, [
|
||||
`value freshness violated in ${inv.failureCount} cases:`,
|
||||
...result.failures.slice(0, 3).map((f) =>
|
||||
` [${f.invariant}] action=${f.action} args=${JSON.stringify(f.args)} actual=${JSON.stringify(f.actual)} error=${f.error}`
|
||||
)
|
||||
].join('\n'));
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user