feat(protocol): share standing-query evaluations across identical subscriptions by jsam · Pull Request #223 · inputlayer/inputlayer

jsam · 2026-10-03T23:00:34Z

Intent

Implement this work item fully and correctly. Always follow clean code principles and clean architecture. Millisecond query and update latency must not regress; no new locks on hot paths. Keep documentation current with the change.

The work item is E01, shared authorized standing evaluation, part of #133 (standing queries: share evaluations and debounce; the DD-backed deltas part of that issue is out of scope). Identical standing-query subscriptions must share one evaluation, owned by a worker rather than guarded by a global per-delta map mutex, keyed by the canonical query, knowledge graph, rules and visibility (authorization) state, with collision-safe keys (hash plus full equality) for both queries and result tuples. Keep the existing per-subscription dependency filtering and pending-change state machine (at most one evaluation in flight; changes seen meanwhile, including rule changes during a refresh, trigger a correct follow-up). Any coalescing delay must be bounded by an approved latency budget. Duplicate subscribers must cost less work (evaluation count and memory) than today, and a single subscriber must not get slower. Sharing must never leak rows across tenants or authorization scopes: two auth scopes, revoked access, unsubscribe and reuse of a subscription id, unrelated writes and hash collisions must all behave correctly, with one shared refresh where that is valid.

What Changed

Replaced the per-connection subscription registry with shared views. Identical subscriptions (same trimmed query text, same knowledge graph) now share one evaluation and one in-memory result across connections and users. Views are owned by a single SubscriptionHub worker task (hub, views, publication). Each view is identified by a collision-safe key, and result rows are kept in a hash-plus-equality ResultSet. Each view still filters changes by dependency and runs at most one evaluation at a time, with one follow-up for changes that land meanwhile. A rule change retires the affected view so later .subscribes get a fresh evaluation. A new subscriber waits for any refresh that is already running or due. When the last evaluation failed, a new subscriber triggers a single retry. A new views metric counts shared views.
Each Subscriber keeps its own read-access check, seq numbering and last delivered result. Access is checked when it subscribes, before its snapshot is sent and before every push. Revoked access now sends a subscription_error instead of a subscription_reset, and the subscription keeps its last result, so seq has no gap once access returns. A slow reader gets one combined delta instead of a queue. notifications_missed no longer affects subscription deltas.
Added http.rate_limit.subscription_coalesce_ms: a coalescing window, default 0 and capped at 100 ms. Updated the config example, the configuration and WebSocket API guides, the AsyncAPI spec and NoticeCode docs to match. Added view, subscriber and failure unit tests, a standing-query sharing test suite (tests/standing_query_tests/sharing.rs), updates to the differential oracle, and benchmarks for duplicate subscribers.

Risk Assessment

⚠️ Medium: The latest fix round correctly makes new subscribers wait for an in-flight, pending or due refresh and keeps the retry flag for concurrent waiters (traced through attach/on_complete/follow_up with in-flight, pending-relevant, pending-unrelated, due, and failed-retry sequences), but the overall change still replaces the standing-query subscription core with a new shared-view worker, which warrants medium caution.

Testing

I rebuilt the server at HEAD and ran a Node WebSocket driver against throwaway server instances (temp data dirs, env-var config, local ports) across five phases. Covered: sharing between auth scopes, isolation by KG and by relation, revocation, unsubscribing and reusing a subscription id, and rule changes. Also covered: result-cap errors on each commit, retries that don't send existing subscribers a duplicate error (including 3 concurrent waiters over 5 rounds), read-your-writes on the snapshot for an acknowledged write (200/200), and a 100 ms coalescing window (10 commits collapsed into 1 delta). Every live assertion passed. I could not observe the shared evaluation count or hash collisions from outside the server, so that scenario is untested live. The targeted standing_query integration tests and subscription unit tests cover it, and they all passed. The first read-your-writes run failed after 33 iterations. It had hit the server's default per-connection WebSocket limit of 100 messages per second, which was a problem with my test setup. I raised the limit in the throwaway config and reran: 200/200 passed. There is no UI, so the evidence is CLI and WebSocket transcripts.

Live validation: ✅ go - 10 of 11 scenarios driven live against the product

Scenario	Result	Live	Evidence
Admin and a viewer (two auth scopes) subscribe to the same ?a(X); both get snapshot [[1]] and the same seq-1 delta [2]	✅ pass	live	live_ws_transcript_sharing_auth.round3.txt S1
The same query on another KG gets an empty snapshot and no push for writes to the first KG	✅ pass	live	live_ws_transcript_sharing_auth.round3.txt S2
A write to an unrelated relation pushes nothing to subscribers	✅ pass	live	live_ws_transcript_sharing_auth.round3.txt S3
After revoking the viewer's ACL, the viewer gets an access-denied subscription_error and no rows, the owner still gets delta [3], and the revoked user cannot attach to the existing view	✅ pass	live	live_ws_transcript_sharing_auth.round3.txt S4
After unsubscribing, writes push nothing; resubscribing with the same id gives a fresh full snapshot and seq restarts at 1	✅ pass	live	live_ws_transcript_sharing_auth.round3.txt S5
After a rule is replaced, a new subscriber's snapshot reflects the new rule	✅ pass	live	live_ws_transcript_sharing_auth.round3.txt S6
Over max_result_rows, every relevant commit pushes subscription_error; new subscribers get the cap error without ringing the existing subscriber; the view recovers with a delta and full snapshot	✅ pass	live	live_ws_transcript_result_cap.round3.txt S7-S8
Read-your-writes: A is subscribed to ?order(X); B commits +order(N), gets the ack, then subscribes; B's snapshot contains N in 200/200 iterations, and A gets every row	✅ pass	live	live_ws_transcript_read_your_writes.round3.txt S9
Three new subscribers attach at the same time to a failed view (shared retry); all get the cap error, and the existing subscriber is never rung with a duplicate (5 rounds)	✅ pass	live	live_ws_transcript_concurrent_waiters.round3.txt S10
With subscription_coalesce_ms=100, a burst of 10 commits is delivered as 1 delta holding all 10 rows, within ~110ms of the first commit; a second subscriber shares the view	✅ pass	live	live_ws_transcript_coalesce.round3.txt S11
Duplicate subscribers share one evaluation per commit, and colliding hashes never merge views or rows	⏸️ untested	no	The prior payload did not establish a live result for this scenario. Evaluation counts are kept only in the in-process SubscriptionMetrics, which no HTTP/WS endpoint exposes. Forcing hash collisions n…

Evidence: Live WS driver (round 3)

// Live driver: real inputlayer-server over /ws. Prints a transcript.
const BASE = `ws://127.0.0.1:${process.env.PORT}/ws`;
const PHASE = process.env.PHASE;
const ADMIN_PW = 'adminpw12345';
let fails = 0;
const ok = (c, m) => { console.log((c ? 'PASS: ' : 'FAIL: ') + m); if (!c) fails++; };

class C {
  constructor(name) { this.name = name; this.q = []; this.waiters = []; }
  async open(kg, user, pw) {
    this.ws = new WebSocket(kg ? `${BASE}?kg=${kg}` : BASE);
    this.ws.onmessage = (e) => { const m = JSON.parse(e.data); const w = this.waiters.shift(); if (w) w(m); else this.q.push(m); };
    await new Promise((r, j) => { this.ws.onopen = r; this.ws.onerror = j; });
    this.ws.send(JSON.stringify({ type: 'login', username: user, password: pw }));
    const m = await this.next(5000); if (m?.type !== 'authenticated') throw new Error(JSON.stringify(m));
    return this;
  }
  next(ms) {
    if (this.q.length) return Promise.resolve(this.q.shift());
    return new Promise((r) => { const f = (m) => { clearTimeout(t); r(m); }; const t = setTimeout(() => { this.waiters = this.waiters.filter(x => x !== f); r(null); }, ms); this.waiters.push(f); });
  }
  async exec(p) {
    console.log(`[${this.name}] >> ${p}`);
    this.ws.send(JSON.stringify({ type: 'execute', program: p }));
    const held = [];
    for (;;) {
      const m = await this.next(10000);
      if (!m) throw new Error('timeout ' + p);
      if (m.type === 'result' || m.type === 'error') { this.q.unshift(...held); console.log(`[${this.name}] << ${JSON.stringify(m)}`); return m; }
      if (m.type?.startsWith('subscription_')) held.push(m);
    }
  }
  async push(sub, ms = 2000) {
    for (;;) {
      const m = await this.next(ms);
      if (!m) { console.log(`[${this.name}] push(${sub}) <none within ${ms}ms>`); return null; }
      if (m.type?.startsWith('subscription_')) { console.log(`[${this.name}] push(${sub}) ${JSON.stringify(m)}`); return m; }
    }
  }
  close() { this.ws.close(); }
}

const sortRows = (r) => JSON.stringify([...r].sort((a, b) => JSON.stringify(a) < JSON.stringify(b) ? -1 : 1));

const admin = await new C('admin').open(null, 'admin', ADMIN_PW);
await admin.exec('.kg create subs');
await admin.exec('.kg create other');
await admin.exec('.user create mallory pw12345678 viewer');
await admin.exec('.kg acl grant subs mallory viewer');
await admin.exec('.kg use subs');
await admin.exec('+a(1)');

if (PHASE === 'share') {
console.log('\n=== S1 two auth scopes share one view; both get snapshot + delta ===');
const owner = await new C('owner').open('subs', 'admin', ADMIN_PW);
const mal = await new C('mallory').open('subs', 'mallory', 'pw12345678');
const so = await owner.exec('.subscribe o ?a(X)');
const sm = await mal.exec('.subscribe m ?a(X)');
ok(sortRows(so.rows) === '[[1]]' && sortRows(sm.rows) === '[[1]]', 'both snapshots are [[1]]');
await admin.exec('+a(2)');
const po = await owner.push('o'), pm = await mal.push('m');
ok(po?.seq === 1 && JSON.stringify(po.inserted) === '[[2]]' && pm?.seq === 1 && JSON.stringify(pm.inserted) === '[[2]]', 'both get seq 1 inserting [2]');

console.log('\n=== S2 same query on another KG is isolated ===');
const ok2 = await new C('other-kg').open('other', 'admin', ADMIN_PW);
const sx = await ok2.exec('.subscribe x ?a(X)');
ok(sortRows(sx.rows) === '[]', 'other KG snapshot empty (no rows leaked from subs)');
await admin.exec('+a(5)');
await owner.push('o'); await mal.push('m');
ok((await ok2.push('x', 1000)) === null, 'other-KG subscriber gets nothing for a write to subs');

console.log('\n=== S3 unrelated write pushes nothing ===');
await admin.exec('+zz(1)');
ok((await owner.push('o', 1000)) === null && (await mal.push('m', 500)) === null, 'no push for unrelated relation');

console.log('\n=== S4 revoked access: no rows to mallory, owner unaffected ===');
await admin.exec('.kg acl revoke subs mallory');
await admin.exec('+a(3)');
const po4 = await owner.push('o');
ok(JSON.stringify(po4?.inserted) === '[[3]]', 'owner gets delta [3]');
const pm4 = await mal.push('m');
ok(pm4 === null || (pm4.type === 'subscription_error' && !pm4.inserted), `mallory gets no rows (${pm4 ? pm4.type + ': ' + pm4.message : 'nothing'})`);
const m2 = await mal.exec('.subscribe m2 ?a(X)');
ok(m2.type === 'error', 'revoked mallory cannot attach to existing view');

console.log('\n=== S5 unsubscribe and reuse of subscription id ===');
await owner.exec('.unsubscribe o');
await admin.exec('+a(6)');
ok((await owner.push('o', 1000)) === null, 'no push after unsubscribe');
const so5 = await owner.exec('.subscribe o ?a(X)');
ok(sortRows(so5.rows) === '[[1],[2],[3],[5],[6]]', 'reused id gets fresh full snapshot: ' + sortRows(so5.rows));
await admin.exec('+a(7)');
const po5 = await owner.push('o');
ok(po5?.seq === 1 && JSON.stringify(po5.inserted) === '[[7]]', "reused id's seq restarts at 1 with only [7]");

console.log('\n=== S6 rule change: new subscriber right after rule reply sees new rules ===');
await admin.exec('+b(1)');
await admin.exec('+r(X) <- a(X), b(X)');
const r1 = await owner.exec('.subscribe r1 ?r(X)');
ok(sortRows(r1.rows) === '[[1]]', 'r under rule a&b = [1]');
await admin.exec('.rule drop r');
await admin.exec('+r(X) <- a(X)');
const r2 = await owner.exec('.subscribe r2 ?r(X)');
ok(sortRows(r2.rows) === '[[1],[2],[3],[5],[6],[7]]', 'new subscriber after rule change sees new rule rows: ' + sortRows(r2.rows));

}
if (PHASE === 'cap') {
console.log('\n=== S7 result cap (max_result_rows=3): every over-cap commit errors; retries do not re-ring; recovery ===');
await admin.exec('+c[(1),(2)]');
const capA = await new C('capA').open('subs', 'admin', ADMIN_PW);
const c0 = await capA.exec('.subscribe c ?c(X)');
ok(sortRows(c0.rows) === '[[1],[2]]', 'snapshot [[1],[2]]');
await admin.exec('+c[(3),(4)]');
let e1 = await capA.push('c');
ok(e1?.type === 'subscription_error' && /max_result_rows/.test(e1.message), 'crossing cap pushes subscription_error');
await admin.exec('+c(5)');
let e2 = await capA.push('c');
ok(e2?.type === 'subscription_error' && /max_result_rows/.test(e2.message), 'second over-cap commit pushes subscription_error (round-1 regression)');
await admin.exec('+c(6)');
let e3 = await capA.push('c');
ok(e3?.type === 'subscription_error', 'third over-cap commit pushes subscription_error');
const capB = await new C('capB').open('subs', 'mallory', 'pw12345678').catch(() => null);
const capB2 = await new C('capB').open('subs', 'admin', ADMIN_PW);
const b = await capB2.exec('.subscribe c ?c(X)');
ok(b.type === 'error' && /max_result_rows/.test(b.message), 'new subscriber to failed view gets a fresh cap error');
ok((await capA.push('c', 1000)) === null, "existing subscriber not rung with duplicate error by the new subscriber's retry");
const capC = await new C('capC').open('subs', 'admin', ADMIN_PW);
const cc = await capC.exec('.subscribe c ?c(X)');
ok(cc.type === 'error', 'a second retry also answers only its subscriber');
ok((await capA.push('c', 1000)) === null, 'existing subscriber still not rung');
await admin.exec('-c[(4),(5),(6)]');
const rec = await capA.push('c');
ok(rec?.type === 'subscription_delta' && JSON.stringify(rec.inserted) === '[[3]]', 'recovery delta (inserts [3]) once under cap');
const capD = await new C('capD').open('subs', 'admin', ADMIN_PW);
const cd = await capD.exec('.subscribe c ?c(X)');
ok(cd.type === 'result' && sortRows(cd.rows) === '[[1],[2],[3]]', 'new subscriber after recovery gets the full snapshot');

console.log('\n=== S8 failed view self-recovers for a new subscriber without a commit to its relation ===');
await admin.exec('+d[(1),(2),(3),(4)]');
const capE = await new C('capE').open('subs', 'admin', ADMIN_PW);
const de = await capE.exec('.subscribe d ?d(X)');
ok(de.type === 'error', 'over-cap first subscribe rejected');
await admin.exec('-d(4)');
const df = await capE.exec('.subscribe d ?d(X)');
ok(df.type === 'result' && sortRows(df.rows) === '[[1],[2],[3]]', 'subscribe after shrinking succeeds with 3 rows');

}

if (PHASE === 'ryw') {
console.log('\n=== S9 read-your-writes: B commits +order(42), gets ack, then subscribes; snapshot must contain 42 (A already subscribed) ===');
const A = await new C('A').open('subs', 'admin', ADMIN_PW);
await A.exec('+order(0)');
const sa = await A.exec('.subscribe a ?order(X)');
ok(sortRows(sa.rows) === '[[0]]', 'A snapshot [[0]]');
const B = await new C('B').open('subs', 'admin', ADMIN_PW);
let bad = 0;
const N = Number(process.env.N || 40);
for (let i = 1; i <= N; i++) {
  const v = 42 * 1000 + i;
  B.ws.send(JSON.stringify({ type: 'execute', program: `+order(${v})` }));
  let ack; for (;;) { const m = await B.next(10000); if (m.type === 'result' || m.type === 'error') { ack = m; break; } }
  B.ws.send(JSON.stringify({ type: 'execute', program: `.subscribe b${i} ?order(X)` }));
  let snap; for (;;) { const m = await B.next(10000); if (m.type === 'result' || m.type === 'error') { snap = m; break; } }
  const has = snap.type === 'result' && snap.rows.some(r => r[0] === v);
  if (!has) { bad++; console.log(`  iter ${i}: snapshot MISSING ${v}: ${JSON.stringify(snap).slice(0,300)}`); }
  B.ws.send(JSON.stringify({ type: 'execute', program: `.unsubscribe b${i}` }));
  for (;;) { const m = await B.next(10000); if (m.type === 'result' || m.type === 'error') break; }
}
ok(bad === 0, `B's post-ack snapshot contained its own write in ${N - bad}/${N} iterations`);
// A received a delta for each write (drain)
let deltas = 0, ins = new Set();
for (;;) { const m = await A.next(1500); if (!m) break; if (m.type === 'subscription_delta') { deltas++; m.inserted.forEach(r => ins.add(r[0])); } }
ok(ins.size === N, `A (existing subscriber) received all ${N} inserted rows across ${deltas} deltas`);

}
if (PHASE === 'waiters') {
console.log('\n=== S10 two concurrent new subscribers to a failed view share one retry; existing subscriber not rung ===');
await admin.exec('+c[(1),(2)]');
const capA = await new C('capA').open('subs', 'admin', ADMIN_PW);
await capA.exec('.subscribe c ?c(X)');
await admin.exec('+c[(3),(4)]');
const e1 = await capA.push('c');
ok(e1?.type === 'subscription_error', 'existing subscriber got the crossing-cap error');
for (let round = 1; round <= 5; round++) {
  const s = await Promise.all([1,2,3].map(k => new C('w'+k).open('subs', 'admin', ADMIN_PW)));
  const res = await Promise.all(s.map((c, k) => c.exec(`.subscribe c ?c(X)`)));
  ok(res.every(r => r.type === 'error' && /max_result_rows/.test(r.message)), `round ${round}: all 3 concurrent new subscribers rejected with the cap error`);
  ok((await capA.push('c', 800)) === null, `round ${round}: existing subscriber not rung with duplicate error`);
  s.forEach(c => c.close());
}
}
if (PHASE === 'coalesce') {
console.log('\n=== S11 coalescing window (subscription_coalesce_ms=100): a burst of commits yields fewer deltas, all rows arrive, bounded latency ===');
const A = await new C('A').open('subs', 'admin', ADMIN_PW);
await A.exec('.subscribe a ?e(X)');
const t0 = Date.now();
for (let i = 1; i <= 10; i++) admin.ws.send(JSON.stringify({ type: 'execute', program: `+e(${i})` }));
let acks = 0; while (acks < 10) { const m = await admin.next(5000); if (m?.type === 'result') acks++; }
const tAck = Date.now();
let deltas = 0, ins = new Set(), last = 0;
for (;;) { const m = await A.next(1500); if (!m) break; if (m.type === 'subscription_delta') { deltas++; last = Date.now(); m.inserted.forEach(r => ins.add(r[0])); console.log(`[A] delta seq=${m.seq} inserted=${JSON.stringify(m.inserted)} at +${last - t0}ms`); } }
ok(ins.size === 10, `all 10 rows delivered (${ins.size})`);
ok(deltas < 10, `burst coalesced into ${deltas} deltas (< 10 commits)`);
ok(last - tAck < 1000, `last delta within ${last - tAck}ms of the final ack`);
const B = await new C('B').open('subs', 'admin', ADMIN_PW);
const sb = await B.exec('.subscribe b ?e(X)');
ok(sb.rows.length === 10, 'second subscriber shares view; snapshot has 10 rows');
}
console.log(`\nOVERALL: ${fails ? 'SOME FAIL (' + fails + ')' : 'ALL PASS'}`);
process.exit(fails ? 1 : 0);

Evidence: Server launcher for disposable instances

#!/bin/bash
# Usage: run_phase.round3.sh PHASE PORT [extra env...]
PHASE=$1; PORT=$2; shift 2
W=~/.no-mistakes/worktrees/9eea230c3aae/01M41XN8JSRPNN3WGVSDF886ZF
D=$(mktemp -d /tmp/il-live-XXXX); : > $D/c.toml
env -i PATH=$PATH HOME=$D INPUTLAYER_ADMIN_PASSWORD=adminpw12345 INPUTLAYER_STORAGE__DATA_DIR=$D/data \
  INPUTLAYER_HTTP__RATE_LIMIT__PER_IP_MAX_RPS=0 "$@" \
  $W/target/debug/inputlayer-server --host 127.0.0.1 --port $PORT --config $D/c.toml >$D/server.log 2>&1 &
SP=$!
for i in $(seq 100); do curl -s -o /dev/null http://127.0.0.1:$PORT/health && break; sleep 0.2; done
PORT=$PORT PHASE=$PHASE node ~/.no-mistakes/evidence/01M41XN8JSRPNN3WGVSDF886ZF/live_ws_driver.round3.mjs; RC=$?
kill $SP; wait $SP 2>/dev/null; tail -5 $D/server.log >&2; rm -rf $D; exit $RC

Evidence: Sharing/auth/unsubscribe/rule-change transcript

[admin] >> .kg create subs
[admin] << {"type":"result","columns":["message"],"rows":[["Knowledge graph 'subs' created."],["Switched to knowledge graph: subs"]],"row_count":2,"total_count":2,"truncated":false,"execution_time_ms":3,"row_provenance":["unknown","unknown"],"switched_kg":"subs","errors":[]}
[admin] >> .kg create other
[admin] << {"type":"result","columns":["message"],"rows":[["Knowledge graph 'other' created."],["Switched to knowledge graph: other"]],"row_count":2,"total_count":2,"truncated":false,"execution_time_ms":2,"row_provenance":["unknown","unknown"],"switched_kg":"other","errors":[]}
[admin] >> .user create mallory pw12345678 viewer
[admin] << {"type":"result","columns":["message"],"rows":[["User 'mallory' created with role 'viewer'."]],"row_count":1,"total_count":1,"truncated":false,"execution_time_ms":0,"row_provenance":["unknown"],"errors":[]}
[admin] >> .kg acl grant subs mallory viewer
[admin] << {"type":"result","columns":["message"],"rows":[["Granted 'viewer' access on 'subs' to 'mallory'."]],"row_count":1,"total_count":1,"truncated":false,"execution_time_ms":0,"row_provenance":["unknown"],"errors":[]}
[admin] >> .kg use subs
[admin] << {"type":"result","columns":["message"],"rows":[["Switched to knowledge graph: subs"]],"row_count":1,"total_count":1,"truncated":false,"execution_time_ms":0,"row_provenance":["unknown"],"switched_kg":"subs","errors":[]}
[admin] >> +a(1)
[admin] << {"type":"result","columns":["message"],"rows":[["Inserted 1 fact(s) into 'a'."]],"row_count":1,"total_count":1,"truncated":false,"execution_time_ms":4,"row_provenance":["unknown"],"errors":[]}

=== S1 two auth scopes share one view; both get snapshot + delta ===
[owner] >> .subscribe o ?a(X)
[owner] << {"type":"result","columns":["X"],"rows":[[1]],"row_count":1,"total_count":1,"truncated":false,"execution_time_ms":3,"errors":[],"subscribed":{"subscription":"o","generation":1,"revision":10}}
[mallory] >> .subscribe m ?a(X)
[mallory] << {"type":"result","columns":["X"],"rows":[[1]],"row_count":1,"total_count":1,"truncated":false,"execution_time_ms":0,"errors":[],"subscribed":{"subscription":"m","generation":1,"revision":10}}
PASS: both snapshots are [[1]]
[admin] >> +a(2)
[admin] << {"type":"result","columns":["message"],"rows":[["Inserted 1 fact(s) into 'a'."]],"row_count":1,"total_count":1,"truncated":false,"execution_time_ms":1,"row_provenance":["unknown"],"errors":[]}
[owner] push(o) {"type":"subscription_delta","subscription":"o","generation":1,"knowledge_graph":"subs","seq":1,"revision":11,"columns":["X"],"inserted":[[2]],"retracted":[]}
[mallory] push(m) {"type":"subscription_delta","subscription":"m","generation":1,"knowledge_graph":"subs","seq":1,"revision":11,"columns":["X"],"inserted":[[2]],"retracted":[]}
PASS: both get seq 1 inserting [2]

=== S2 same query on another KG is isolated ===
[other-kg] >> .subscribe x ?a(X)
[other-kg] << {"type":"result","columns":[],"rows":[],"row_count":0,"total_count":0,"truncated":false,"execution_time_ms":2,"errors":[],"subscribed":{"subscription":"x","generation":1,"revision":7}}
PASS: other KG snapshot empty (no rows leaked from subs)
[admin] >> +a(5)
[admin] << {"type":"result","columns":["message"],"rows":[["Inserted 1 fact(s) into 'a'."]],"row_count":1,"total_count":1,"truncated":false,"execution_time_ms":1,"row_provenance":["unknown"],"errors":[]}
[owner] push(o) {"type":"subscription_delta","subscription":"o","generation":1,"knowledge_graph":"subs","seq":2,"revision":12,"columns":["X"],"inserted":[[5]],"retracted":[]}
[mallory] push(m) {"type":"subscription_delta","subscription":"m","generation":1,"knowledge_graph":"subs","seq":2,"revision":12,"columns":["X"],"inserted":[[5]],"retracted":[]}
[other-kg] push(x) <none within 1000ms>
PASS: other-KG subscriber gets nothing for a write to subs

=== S3 unrelated write pushes nothing ===
[admin] >> +zz(1)
[admin] << {"type":"result","columns":["message"],"rows":[["Inserted 1 fact(s) into 'zz'."]],"row_count":1,"total_count":1,"truncated":false,"execution_time_ms":4,"row_provenance":["unknown"],"errors":[]}
[owner] push(o) <none within 1000ms>
[mallory] push(m) <none within 500ms>
PASS: no push for unrelated relation

=== S4 revoked access: no rows to mallory, owner unaffected ===
[admin] >> .kg acl revoke subs mallory
[admin] << {"type":"result","columns":["message"],"rows":[["Revoked access on 'subs' from 'mallory'."]],"row_count":1,"total_count":1,"truncated":false,"execution_time_ms":0,"row_provenance":["unknown"],"errors":[]}
[admin] >> +a(3)
[admin] << {"type":"result","columns":["message"],"rows":[["Inserted 1 fact(s) into 'a'."]],"row_count":1,"total_count":1,"truncated":false,"execution_time_ms":2,"row_provenance":["unknown"],"errors":[]}
[owner] push(o) {"type":"subscription_delta","subscription":"o","generation":1,"knowledge_graph":"subs","seq":3,"revision":15,"columns":["X"],"inserted":[[3]],"retracted":[]}
PASS: owner gets delta [3]
[mallory] push(m) {"type":"subscription_error","subscription":"m","generation":1,"message":"Access denied to knowledge graph 'subs'. The subscription keeps its last delivered result; once access is restored, its next delta is relative to it."}
PASS: mallory gets no rows (subscription_error: Access denied to knowledge graph 'subs'. The subscription keeps its last delivered result; once access is restored, its next delta is relative to it.)
[mallory] >> .subscribe m2 ?a(X)
[mallory] << {"type":"error","message":"Access denied"}
PASS: revoked mallory cannot attach to existing view

=== S5 unsubscribe and reuse of subscription id ===
[owner] >> .unsubscribe o
[owner] << {"type":"result","columns":["message"],"rows":[["Unsubscribed 'o'."]],"row_count":1,"total_count":1,"truncated":false,"execution_time_ms":0,"errors":[]}
[admin] >> +a(6)
[admin] << {"type":"result","columns":["message"],"rows":[["Inserted 1 fact(s) into 'a'."]],"row_count":1,"total_count":1,"truncated":false,"execution_time_ms":1,"row_provenance":["unknown"],"errors":[]}
[owner] push(o) <none within 1000ms>
PASS: no push after unsubscribe
[owner] >> .subscribe o ?a(X)
[owner] << {"type":"result","columns":["X"],"rows":[[1],[2],[3],[5],[6]],"row_count":5,"total_count":5,"truncated":false,"execution_time_ms":0,"errors":[],"subscribed":{"subscription":"o","generation":2,"revision":16}}
PASS: reused id gets fresh full snapshot: [[1],[2],[3],[5],[6]]
[admin] >> +a(7)
[admin] << {"type":"result","columns":["message"],"rows":[["Inserted 1 fact(s) into 'a'."]],"row_count":1,"total_count":1,"truncated":false,"execution_time_ms":2,"row_provenance":["unknown"],"errors":[]}
[owner] push(o) {"type":"subscription_delta","subscription":"o","generation":2,"knowledge_graph":"subs","seq":1,"revision":17,"columns":["X"],"inserted":[[7]],"retracted":[]}
PASS: reused id's seq restarts at 1 with only [7]

=== S6 rule change: new subscriber right after rule reply sees new rules ===
[admin] >> +b(1)
[admin] << {"type":"result","columns":["message"],"rows":[["Inserted 1 fact(s) into 'b'."]],"row_count":1,"total_count":1,"truncated":false,"execution_time_ms":3,"row_provenance":["unknown"],"errors":[]}
[admin] >> +r(X) <- a(X), b(X)
[admin] << {"type":"result","columns":["message"],"rows":[["Rule 'r' registered."]],"row_count":1,"total_count":1,"truncated":false,"execution_time_ms":5,"row_provenance":["unknown"],"errors":[]}
[owner] >> .subscribe r1 ?r(X)
[owner] << {"type":"result","columns":["X"],"rows":[[1]],"row_count":1,"total_count":1,"truncated":false,"execution_time_ms":4,"errors":[],"subscribed":{"subscription":"r1","generation":3,"revision":19}}
PASS: r under rule a&b = [1]
[admin] >> .rule drop r
[admin] << {"type":"result","columns":["message"],"rows":[["Rule 'r' dropped."]],"row_count":1,"total_count":1,"truncated":false,"execution_time_ms":5,"row_provenance":["unknown"],"errors":[]}
[admin] >> +r(X) <- a(X)
[admin] << {"type":"result","columns":["message"],"rows":[["Rule 'r' registered."]],"row_count":1,"total_count":1,"truncated":false,"execution_time_ms":5,"row_provenance":["unknown"],"errors":[]}
[owner] >> .subscribe r2 ?r(X)
[owner] << {"type":"result","columns":["X"],"rows":[[1],[2],[3],[5],[6],[7]],"row_count":6,"total_count":6,"truncated":false,"execution_time_ms":3,"errors":[],"subscribed":{"subscription":"r2","generation":4,"revision":21}}
PASS: new subscriber after rule change sees new rule rows: [[1],[2],[3],[5],[6],[7]]

OVERALL: ALL PASS

Evidence: Result cap / retry / recovery transcript

[admin] >> .kg create subs
[admin] << {"type":"result","columns":["message"],"rows":[["Knowledge graph 'subs' created."],["Switched to knowledge graph: subs"]],"row_count":2,"total_count":2,"truncated":false,"execution_time_ms":3,"row_provenance":["unknown","unknown"],"switched_kg":"subs","errors":[]}
[admin] >> .kg create other
[admin] << {"type":"result","columns":["message"],"rows":[["Knowledge graph 'other' created."],["Switched to knowledge graph: other"]],"row_count":2,"total_count":2,"truncated":false,"execution_time_ms":2,"row_provenance":["unknown","unknown"],"switched_kg":"other","errors":[]}
[admin] >> .user create mallory pw12345678 viewer
[admin] << {"type":"result","columns":["message"],"rows":[["User 'mallory' created with role 'viewer'."]],"row_count":1,"total_count":1,"truncated":false,"execution_time_ms":0,"row_provenance":["unknown"],"errors":[]}
[admin] >> .kg acl grant subs mallory viewer
[admin] << {"type":"result","columns":["message"],"rows":[["Granted 'viewer' access on 'subs' to 'mallory'."]],"row_count":1,"total_count":1,"truncated":false,"execution_time_ms":0,"row_provenance":["unknown"],"errors":[]}
[admin] >> .kg use subs
[admin] << {"type":"result","columns":["message"],"rows":[["Switched to knowledge graph: subs"]],"row_count":1,"total_count":1,"truncated":false,"execution_time_ms":0,"row_provenance":["unknown"],"switched_kg":"subs","errors":[]}
[admin] >> +a(1)
[admin] << {"type":"result","columns":["message"],"rows":[["Inserted 1 fact(s) into 'a'."]],"row_count":1,"total_count":1,"truncated":false,"execution_time_ms":3,"row_provenance":["unknown"],"errors":[]}

=== S7 result cap (max_result_rows=3): every over-cap commit errors; retries do not re-ring; recovery ===
[admin] >> +c[(1),(2)]
[admin] << {"type":"result","columns":["message"],"rows":[["Inserted 2 fact(s) into 'c'."]],"row_count":1,"total_count":1,"truncated":false,"execution_time_ms":3,"row_provenance":["unknown"],"errors":[]}
[capA] >> .subscribe c ?c(X)
[capA] << {"type":"result","columns":["X"],"rows":[[1],[2]],"row_count":2,"total_count":2,"truncated":false,"execution_time_ms":3,"errors":[],"subscribed":{"subscription":"c","generation":1,"revision":11}}
PASS: snapshot [[1],[2]]
[admin] >> +c[(3),(4)]
[admin] << {"type":"result","columns":["message"],"rows":[["Inserted 2 fact(s) into 'c'."]],"row_count":1,"total_count":1,"truncated":false,"execution_time_ms":1,"row_provenance":["unknown"],"errors":[]}
[capA] push(c) {"type":"subscription_error","subscription":"c","generation":1,"message":"Subscription result exceeds storage.performance.max_result_rows (3); no complete result to deliver. Narrow the query."}
PASS: crossing cap pushes subscription_error
[admin] >> +c(5)
[admin] << {"type":"result","columns":["message"],"rows":[["Inserted 1 fact(s) into 'c'."]],"row_count":1,"total_count":1,"truncated":false,"execution_time_ms":2,"row_provenance":["unknown"],"errors":[]}
[capA] push(c) {"type":"subscription_error","subscription":"c","generation":1,"message":"Subscription result exceeds storage.performance.max_result_rows (3); no complete result to deliver. Narrow the query."}
PASS: second over-cap commit pushes subscription_error (round-1 regression)
[admin] >> +c(6)
[admin] << {"type":"result","columns":["message"],"rows":[["Inserted 1 fact(s) into 'c'."]],"row_count":1,"total_count":1,"truncated":false,"execution_time_ms":1,"row_provenance":["unknown"],"errors":[]}
[capA] push(c) {"type":"subscription_error","subscription":"c","generation":1,"message":"Subscription result exceeds storage.performance.max_result_rows (3); no complete result to deliver. Narrow the query."}
PASS: third over-cap commit pushes subscription_error
[capB] >> .subscribe c ?c(X)
[capB] << {"type":"error","message":"Subscription result exceeds storage.performance.max_result_rows (3); no complete result to deliver. Narrow the query."}
PASS: new subscriber to failed view gets a fresh cap error
[capA] push(c) <none within 1000ms>
PASS: existing subscriber not rung with duplicate error by the new subscriber's retry
[capC] >> .subscribe c ?c(X)
[capC] << {"type":"error","message":"Subscription result exceeds storage.performance.max_result_rows (3); no complete result to deliver. Narrow the query."}
PASS: a second retry also answers only its subscriber
[capA] push(c) <none within 1000ms>
PASS: existing subscriber still not rung
[admin] >> -c[(4),(5),(6)]
[admin] << {"type":"result","columns":["message"],"rows":[["Deleted 3 fact(s) from 'c'."]],"row_count":1,"total_count":1,"truncated":false,"execution_time_ms":2,"row_provenance":["unknown"],"errors":[]}
[capA] push(c) {"type":"subscription_delta","subscription":"c","generation":1,"knowledge_graph":"subs","seq":1,"revision":15,"columns":["X"],"inserted":[[3]],"retracted":[]}
PASS: recovery delta (inserts [3]) once under cap
[capD] >> .subscribe c ?c(X)
[capD] << {"type":"result","columns":["X"],"rows":[[1],[2],[3]],"row_count":3,"total_count":3,"truncated":false,"execution_time_ms":0,"errors":[],"subscribed":{"subscription":"c","generation":1,"revision":15}}
PASS: new subscriber after recovery gets the full snapshot

=== S8 failed view self-recovers for a new subscriber without a commit to its relation ===
[admin] >> +d[(1),(2),(3),(4)]
[admin] << {"type":"result","columns":["message"],"rows":[["Inserted 4 fact(s) into 'd'."]],"row_count":1,"total_count":1,"truncated":false,"execution_time_ms":4,"row_provenance":["unknown"],"errors":[]}
[capE] >> .subscribe d ?d(X)
[capE] << {"type":"error","message":"Subscription result exceeds storage.performance.max_result_rows (3); no complete result to deliver. Narrow the query."}
PASS: over-cap first subscribe rejected
[admin] >> -d(4)
[admin] << {"type":"result","columns":["message"],"rows":[["Deleted 1 facts from 'd'."]],"row_count":1,"total_count":1,"truncated":false,"execution_time_ms":2,"row_provenance":["unknown"],"errors":[]}
[capE] >> .subscribe d ?d(X)
[capE] << {"type":"result","columns":["X"],"rows":[[1],[2],[3]],"row_count":3,"total_count":3,"truncated":false,"execution_time_ms":1,"errors":[],"subscribed":{"subscription":"d","generation":1,"revision":17}}
PASS: subscribe after shrinking succeeds with 3 rows

OVERALL: ALL PASS

Evidence: Read-your-writes transcript (200 iterations)

[admin] >> .kg create subs
[admin] << {"type":"result","columns":["message"],"rows":[["Knowledge graph 'subs' created."],["Switched to knowledge graph: subs"]],"row_count":2,"total_count":2,"truncated":false,"execution_time_ms":3,"row_provenance":["unknown","unknown"],"switched_kg":"subs","errors":[]}
[admin] >> .kg create other
[admin] << {"type":"result","columns":["message"],"rows":[["Knowledge graph 'other' created."],["Switched to knowledge graph: other"]],"row_count":2,"total_count":2,"truncated":false,"execution_time_ms":3,"row_provenance":["unknown","unknown"],"switched_kg":"other","errors":[]}
[admin] >> .user create mallory pw12345678 viewer
[admin] << {"type":"result","columns":["message"],"rows":[["User 'mallory' created with role 'viewer'."]],"row_count":1,"total_count":1,"truncated":false,"execution_time_ms":0,"row_provenance":["unknown"],"errors":[]}
[admin] >> .kg acl grant subs mallory viewer
[admin] << {"type":"result","columns":["message"],"rows":[["Granted 'viewer' access on 'subs' to 'mallory'."]],"row_count":1,"total_count":1,"truncated":false,"execution_time_ms":0,"row_provenance":["unknown"],"errors":[]}
[admin] >> .kg use subs
[admin] << {"type":"result","columns":["message"],"rows":[["Switched to knowledge graph: subs"]],"row_count":1,"total_count":1,"truncated":false,"execution_time_ms":0,"row_provenance":["unknown"],"switched_kg":"subs","errors":[]}
[admin] >> +a(1)
[admin] << {"type":"result","columns":["message"],"rows":[["Inserted 1 fact(s) into 'a'."]],"row_count":1,"total_count":1,"truncated":false,"execution_time_ms":3,"row_provenance":["unknown"],"errors":[]}

=== S9 read-your-writes: B commits +order(42), gets ack, then subscribes; snapshot must contain 42 (A already subscribed) ===
[A] >> +order(0)
[A] << {"type":"result","columns":["message"],"rows":[["Inserted 1 fact(s) into 'order'."]],"row_count":1,"total_count":1,"truncated":false,"execution_time_ms":9,"row_provenance":["unknown"],"errors":[]}
[A] >> .subscribe a ?order(X)
[A] << {"type":"result","columns":["X"],"rows":[[0]],"row_count":1,"total_count":1,"truncated":false,"execution_time_ms":3,"errors":[],"subscribed":{"subscription":"a","generation":1,"revision":11}}
PASS: A snapshot [[0]]
PASS: B's post-ack snapshot contained its own write in 200/200 iterations
PASS: A (existing subscriber) received all 200 inserted rows across 200 deltas

OVERALL: ALL PASS

Evidence: Concurrent waiters on failed view transcript (local file: ~/.no-mistakes/evidence/01M41XN8JSRPNN3WGVSDF886ZF/live_ws_transcript_concurrent_waiters.round3.txt)

Evidence: Coalescing window transcript

[admin] >> .kg create subs
[admin] << {"type":"result","columns":["message"],"rows":[["Knowledge graph 'subs' created."],["Switched to knowledge graph: subs"]],"row_count":2,"total_count":2,"truncated":false,"execution_time_ms":4,"row_provenance":["unknown","unknown"],"switched_kg":"subs","errors":[]}
[admin] >> .kg create other
[admin] << {"type":"result","columns":["message"],"rows":[["Knowledge graph 'other' created."],["Switched to knowledge graph: other"]],"row_count":2,"total_count":2,"truncated":false,"execution_time_ms":5,"row_provenance":["unknown","unknown"],"switched_kg":"other","errors":[]}
[admin] >> .user create mallory pw12345678 viewer
[admin] << {"type":"result","columns":["message"],"rows":[["User 'mallory' created with role 'viewer'."]],"row_count":1,"total_count":1,"truncated":false,"execution_time_ms":0,"row_provenance":["unknown"],"errors":[]}
[admin] >> .kg acl grant subs mallory viewer
[admin] << {"type":"result","columns":["message"],"rows":[["Granted 'viewer' access on 'subs' to 'mallory'."]],"row_count":1,"total_count":1,"truncated":false,"execution_time_ms":0,"row_provenance":["unknown"],"errors":[]}
[admin] >> .kg use subs
[admin] << {"type":"result","columns":["message"],"rows":[["Switched to knowledge graph: subs"]],"row_count":1,"total_count":1,"truncated":false,"execution_time_ms":0,"row_provenance":["unknown"],"switched_kg":"subs","errors":[]}
[admin] >> +a(1)
[admin] << {"type":"result","columns":["message"],"rows":[["Inserted 1 fact(s) into 'a'."]],"row_count":1,"total_count":1,"truncated":false,"execution_time_ms":11,"row_provenance":["unknown"],"errors":[]}

=== S11 coalescing window (subscription_coalesce_ms=100): a burst of commits yields fewer deltas, all rows arrive, bounded latency ===
[A] >> .subscribe a ?e(X)
[A] << {"type":"result","columns":[],"rows":[],"row_count":0,"total_count":0,"truncated":false,"execution_time_ms":3,"errors":[],"subscribed":{"subscription":"a","generation":1,"revision":10}}
[A] delta seq=1 inserted=[[10],[1],[2],[3],[4],[5],[6],[7],[8],[9]] at +110ms
PASS: all 10 rows delivered (10)
PASS: burst coalesced into 1 deltas (< 10 commits)
PASS: last delta within 89ms of the final ack
[B] >> .subscribe b ?e(X)
[B] << {"type":"result","columns":["X"],"rows":[[10],[1],[2],[3],[4],[5],[6],[7],[8],[9]],"row_count":10,"total_count":10,"truncated":false,"execution_time_ms":0,"errors":[],"subscribed":{"subscription":"b","generation":1,"revision":20}}
PASS: second subscriber shares view; snapshot has 10 rows

OVERALL: ALL PASS

Evidence: Standing query integration tests (local file: ~/.no-mistakes/evidence/01M41XN8JSRPNN3WGVSDF886ZF/standing_query_tests.round3.log)
Evidence: Subscription unit + collision tests (local file: ~/.no-mistakes/evidence/01M41XN8JSRPNN3WGVSDF886ZF/subscription_unit_tests.round3.log)
Outcome: 🔧 2 issues found → auto-fixed ✅ across 3 runs (20m7s)

Pipeline

Updates from git push no-mistakes

... (3 earlier update rounds omitted to keep the PR body within GitHub's 65536-char limit; full history is in the run log.)

🔧 **Review** - 4 issues found → auto-fixed (3) ✅

🔧 Fix applied.
4 warnings still open:

⚠️ src/protocol/subscription/views.rs:195 - After a view's latest refresh fails, every new .subscribe to that query is rejected with the stale error, and nothing triggers a retry. The view re-evaluates only when a commit touches its dependencies. Transient failures are real here: request_control(None) applies query_timeout_ms, and that deadline also covers the wait for the query semaphore. So one timeout under load blocks the query for every new subscriber, across users and connections, until an unrelated relevant write arrives. Before this change, each .subscribe ran a fresh evaluation and recovered by itself. The docs describe this rejection as intended (websocket-api.mdx), so this needs your decision. Proposed fix: when someone attaches to a Failed view that is idle, dispatch a refresh and put the subscriber in waiting, the way the first evaluation is handled, instead of returning Attach::Rejected.
⚠️ src/protocol/subscription/views.rs:34 - Intent conformance. The intent says identical subscriptions must share one evaluation "keyed by the canonical query, knowledge graph, rules and visibility (authorization) state". ViewKey holds only knowledge_graph and query: String (the trimmed text). The docs say a query spelled differently, for example with other variable names, gets its own view. The key carries no rules or visibility component. Instead the evaluation runs with auth: None (reevaluate.rs:76), and read access is checked per subscriber at subscribe and at delivery. Authorization today is KG-level only (authorize_program/get_kg_role_for_user), so the rows look identical for every authorized principal. Persistent rules belong to the graph's snapshot, so this design looks safe. It still departs from the literal key the intent requires, and it uses trimmed text where the intent says canonical query. Please confirm the substitution is acceptable, or require a canonicalized query plus a visibility or rules component in the key.
⚠️ src/protocol/subscription/hub.rs:130 - Round 1 fixed view-key-vs-intent by retiring view keys when a rule changes, but a stale evaluation can still reach a new subscriber. The fix does key retirement only when the worker processes the RuleChange/KgChange notification (hub.rs:141-142). The worker's select! is biased toward commands (hub.rs:129-130). Rule changes are published to the broadcast synchronously during the write (write_run.rs:156 -> send_notification), before the client gets its reply. Failing sequence: agent A runs .rule add r(X) :- a(X), b(X). Agent B (or A itself) sees the reply and sends .subscribe s ?r(X). The worker has a broadcast backlog, for example from a burst of writes, or both events are ready in the same poll. The worker handles the Attach command first. ViewRegistry::attach finds the old key, which is not retired yet, and returns Attach::Attached with the publication computed under the old rules. B's snapshot reflects rules that were already replaced when it subscribed. A delta follows only after the notification is drained. This contradicts the recorded operator decision "so a rule change can never serve an old evaluation to a new subscriber". The same ordering lets a new subscriber's snapshot predate a write that was already acknowledged. Smallest fix at the shared boundary: in Worker::on_command, before handling Command::Attach, drain self.notifications.try_recv() until it is empty, applying each item through the same on_rule_change/on_change/on_unknown_changes logic as the select arm. Then every commit published before the attach command was sent is applied before key lookup. Detach stays command-first, as the module doc requires. Add a test: queue a RuleChange on the broadcast, then send the attach, and assert the subscriber gets a fresh view.
⚠️ src/protocol/subscription/views.rs:478 - This comes from round 1's failed-view-rejects-new-subscribers fix. Now every new .subscribe to a failed view triggers a retry (views.rs:207-213). When the retry fails again, publish creates another Outcome::Failed publication and rings every existing subscriber (views.rs:478-494). Each existing subscriber then gets another identical subscription_error push, even though nothing it watches changed. Concrete case: a query's result exceeds max_result_rows, so the view keeps its last complete result and is Failed. Each .subscribe attempt from other users or connections now sends a duplicate cap error to every current subscriber. The noise is driven by other principals' actions and grows with subscribe attempts. Fix: in publish's Err arm, when latest.outcome is already Failed with the same message, skip creating a new publication and ringing subscribers. The waiters are still answered with the error through release_waiting, because failure is computed separately in on_complete.

🔧 Fix applied.
✅ Re-checked - no issues remain.

⚠️ src/protocol/subscription/views.rs:207 - This defect is in code that fix rounds 2/3 introduced for retry-failure-republishes-to-existing, and it leaves that fix incomplete. attach sets view.retry = view.due.is_none() && view.query.is_some() unconditionally, so a second subscriber attaching while the retry is in flight (query is None) resets retry to false. Sequence: the view is Failed('capped'), with subscriber 1 attached. Subscriber 2 attaches, so retry=true and the retry is dispatched. Subscriber 3 attaches while it runs, so retry=false and 3 joins waiting. The retry fails again with 'capped'. publish (views.rs:477) no longer matches the retry guard, so it creates a new Failed publication and rings subscriber 1 with a duplicate identical subscription_error. That contradicts the round-2 operator instruction ('a retry triggered by a new subscriber that fails again must answer only the waiting subscribers') and the docs (websocket-api.mdx: 'existing subscribers get no duplicate subscription_error'). The concurrent case is common because the retry is single-flight precisely so that several subscribers arriving together share it. The test a_refresh_keeps_the_result_and_a_new_subscriber_reevaluates_once (failures.rs:45-58) covers two waiters only on the success path, and a_retry_failing_the_same_way_answers_only_its_waiters covers one waiter, so neither catches this. Fix: update the flag only when this attach actually dispatches, e.g. if view.query.is_some() { view.retry = view.due.is_none(); }, and leave an in-flight retry's flag unchanged. Add a test: two waiters, the retry fails with the same error, and existing subscriber 1 is not rung.
⚠️ src/protocol/subscription/views.rs:200 - A .subscribe sent after an acknowledged data write can get a snapshot that does not include that write. Before this change, every .subscribe ran a fresh evaluation. Now attach returns live.latest whenever the view's latest publication is not Failed. This happens even when the drain in Worker::on_command (hub.rs:162) has just seen a relevant commit and dispatched or scheduled a refresh (the view is in flight, has pending, or has due set). Concrete sequence: agent A holds a subscription to ?order(X). Agent B runs +order(42), gets the reply, then sends .subscribe s ?order(X). The drain dispatches the view's refresh, then attach returns Attached with the pre-write publication. B's snapshot lacks 42, which arrives only in a later delta. The snapshot is labelled with its older revision, so a client following the protocol converges, but read-your-writes on the snapshot is lost. Fix round 2 closed this gap for rule changes only (rule-change-attach-race; the docs promise a fresh evaluation only after a rule change), and its finding text noted that the data-write case stays open. Whether a subscriber may attach to a result older than writes acknowledged before its .subscribe is a product decision. Closing it means making such subscribers wait for the refresh after their pending change, not just the one in flight (views.rs:205-212 waiting, plus follow_up release). That trades a little subscribe latency for consistency. Please decide whether to accept the current behavior and document it in websocket-api.mdx, or require the wait.

🔧 Fix applied.
✅ Re-checked - no issues remain.

🔧 **Test** - 2 issues found → auto-fixed ✅

🚨 src/protocol/subscription/views.rs:477 - Commit 9730d39 added Err(e) if matches!(&latest.outcome, Outcome::Failed(f) if *f == e) => return Vec::new() to publish. It is meant to stop a retry triggered by a new subscriber from ringing existing subscribers again. But it also drops errors from commit-triggered refreshes. While a view stays over max_result_rows, each later relevant commit now pushes nothing to existing subscribers. Before this change they got another subscription_error each time. The integration test test_crossing_result_cap_keeps_last_complete_result (tests/standing_query_tests/main.rs:397) expects an error push for each of +a(4), +a(5), -a(1). It times out deterministically: 2 of 2 runs failed. A live run against the binary shows the same thing: the second over-cap commit gets no push within 2s. The operator's round-2 decision only covered a retry started by a new subscriber that fails again. Suggested fix: skip the publication only when the refresh was a subscriber-triggered retry, for example by tracking a retry flag on the refresh. Keep publishing commit-triggered failures. Alternatively, if suppressing repeat errors is now the intended product behavior, update the test and the docs. The docs (websocket-api.mdx:422) still say an over-cap re-evaluation pushes a subscription_error.
🚨 live validation verdict: no-go (8 of 9 scenarios were driven live against the product); failed: While a subscription stays over max_result_rows, each further relevant commit pushes a subscription_error to the existing subscriber
Live validation: ❌ no-go - 8 of 9 scenarios driven live against the product

Scenario	Result	Live	Evidence
Two users in different auth scopes subscribe to the same query and both get the same snapshot and seq-1 delta from one shared evaluation	✅ pass	live	live_ws_transcript_sharing_auth.txt S1; sharing::two_auth_scopes_share_one_evaluation_and_only_the_readable_one_gets_rows (evaluations()==1 then 2)
The same query on another knowledge graph is its own view and receives no rows or pushes from writes to the first KG	✅ pass	live	live_ws_transcript_sharing_auth.txt S2
A write to an unrelated relation pushes nothing to subscribers	✅ pass	live	live_ws_transcript_sharing_auth.txt S3; test_unrelated_write_triggers_no_evaluation
After access is revoked, the shared refresh gives the owner a delta and the revoked user an Access denied error with no rows; a new attach is refused; after re-grant the user's delta is relative to it…	✅ pass	live	live_ws_transcript_sharing_auth.txt S4
After unsubscribe, no pushes arrive; reusing the same subscription id gives a fresh snapshot, generation 2, and seq restarting at 1	✅ pass	live	live_ws_transcript_sharing_auth.txt S5
A .subscribe sent after a rule change's reply gets an evaluation under the new rules, not the old shared view	✅ pass	live	live_ws_transcript_sharing_auth.txt S6; hub::tests::an_attach_after_an_announced_rule_change_starts_a_new_view
Adversarial: colliding view-key hashes and row hashes never share a view or confuse rows	⏸️ untested	no	The prior payload did not establish a live result: hash collisions cannot be forced through the real server's public interface, so this was only covered by unit tests (colliding_keys_never_share_a_vie…
A new subscriber to a failed (over-cap) view triggers a fresh evaluation and gets the error; existing subscribers get no duplicate error; the recovery delta follows the last complete result	✅ pass	live	live_ws_transcript_result_cap.txt S7 (capB error, capA quiet, recovery delta inserts [3])
While a subscription stays over max_result_rows, each further relevant commit pushes a subscription_error to the existing subscriber	❌ fail	live	live_ws_transcript_result_cap.txt S7 ('second over-cap commit' got no push in 2s); test_crossing_result_cap_keeps_last_complete_result times out

cargo test --test standing_query_tests (end-to-end over real ws against an in-process server; 19 pass, 1 fails: test_crossing_result_cap_keeps_last_complete_result)
cargo test --test standing_query_tests crossing (re-run, fails again: deterministic)
cargo build --bin inputlayer-server and launched with a disposable temp config/data dir on 127.0.0.1:18931
Live Node WebSocket client (drive.mjs) S1-S6 against the server with max_result_rows=100000: shared snapshot/deltas for admin+viewer, other-KG isolation, unrelated write, revoke/regrant, unsubscribe + id reuse, rule change before subscribe
Live Node WebSocket client S7 against the server with max_result_rows=3: crossing the cap, a repeated over-cap commit, a new subscriber retrying a failed view, recovery delta
cargo test --lib protocol::subscription (45 pass, including colliding_keys_never_share_a_view, colliding_hashes_never_confuse_rows, an_attach_after_an_announced_rule_change_starts_a_new_view, a_retry_failing_the_same_way_answers_only_its_waiters)

🔧 Fix applied.
✅ Re-checked - no issues remain.

Live validation: ✅ go - 10 of 10 scenarios driven live against the product

Scenario	Result	Live	Evidence
A subscription that stays over max_result_rows gets a subscription_error push on every later relevant commit (the round-1 regression)	✅ pass	live	live_ws_transcript_result_cap.round2.txt (S7: errors pushed after +c[(3),(4)], +c(5) and +c(6)); also standing_query_tests test_crossing_result_cap_keeps_last_complete_result
A new subscriber to a failed view gets a fresh error, and existing subscribers get no duplicate push (checked twice)	✅ pass	live	live_ws_transcript_result_cap.round2.txt (S7: capB and capC are rejected; capA gets nothing within 1s each time)
After the result drops back under the cap, the existing subscriber gets a recovery delta and a new subscriber gets the full snapshot	✅ pass	live	live_ws_transcript_result_cap.round2.txt (S7: delta inserts [3]; capD snapshot is [[1],[2],[3]])
A failed view recovers for a later subscriber once the data shrinks	✅ pass	live	live_ws_transcript_result_cap.round2.txt (S8)
Two auth scopes (admin, viewer) share one view; both get the same snapshot and delta	✅ pass	live	live_ws_transcript_sharing_auth.round2.txt (S1)
The same query on another KG is a separate view: no rows and no pushes leak across	✅ pass	live	live_ws_transcript_sharing_auth.round2.txt (S2)
A write to an unrelated relation pushes nothing	✅ pass	live	live_ws_transcript_sharing_auth.round2.txt (S3)
After revoke, the revoked user gets an access-denied error with no rows and cannot attach; the owner still gets deltas	✅ pass	live	live_ws_transcript_sharing_auth.round2.txt (S4)
Unsubscribe stops pushes, and a reused subscription id gets a fresh snapshot with seq restarting at 1	✅ pass	live	live_ws_transcript_sharing_auth.round2.txt (S5)
A subscriber who subscribes right after a rule change gets results under the new rule, not the old shared view	✅ pass	live	live_ws_transcript_sharing_auth.round2.txt (S6)

cargo build --bin inputlayer-server
cargo test --test standing_query_tests (20/20 pass, including test_crossing_result_cap_keeps_last_complete_result, which failed in round 1)
cargo test --lib protocol::subscription (46 pass, including views::tests::failures::a_retry_failing_the_same_way_answers_only_its_waiters and a_commit_failing_the_same_way_still_reaches_every_subscriber)
Live: ran two isolated inputlayer-server instances (temp data dirs, ports 18731 with max_result_rows=3 and 18732 at the default) and drove them over /ws with a Node WebSocket driver (PHASE=share and PHASE=cap node drive.mjs)

✅ No issues found.

Live validation: ✅ go - 10 of 11 scenarios driven live against the product

Scenario	Result	Live	Evidence
Admin and a viewer (two auth scopes) subscribe to the same ?a(X); both get snapshot [[1]] and the same seq-1 delta [2]	✅ pass	live	live_ws_transcript_sharing_auth.round3.txt S1
The same query on another KG gets an empty snapshot and no push for writes to the first KG	✅ pass	live	live_ws_transcript_sharing_auth.round3.txt S2
A write to an unrelated relation pushes nothing to subscribers	✅ pass	live	live_ws_transcript_sharing_auth.round3.txt S3
After revoking the viewer's ACL, the viewer gets an access-denied subscription_error and no rows, the owner still gets delta [3], and the revoked user cannot attach to the existing view	✅ pass	live	live_ws_transcript_sharing_auth.round3.txt S4
After unsubscribing, writes push nothing; resubscribing with the same id gives a fresh full snapshot and seq restarts at 1	✅ pass	live	live_ws_transcript_sharing_auth.round3.txt S5
After a rule is replaced, a new subscriber's snapshot reflects the new rule	✅ pass	live	live_ws_transcript_sharing_auth.round3.txt S6
Over max_result_rows, every relevant commit pushes subscription_error; new subscribers get the cap error without ringing the existing subscriber; the view recovers with a delta and full snapshot	✅ pass	live	live_ws_transcript_result_cap.round3.txt S7-S8
Read-your-writes: A is subscribed to ?order(X); B commits +order(N), gets the ack, then subscribes; B's snapshot contains N in 200/200 iterations, and A gets every row	✅ pass	live	live_ws_transcript_read_your_writes.round3.txt S9
Three new subscribers attach at the same time to a failed view (shared retry); all get the cap error, and the existing subscriber is never rung with a duplicate (5 rounds)	✅ pass	live	live_ws_transcript_concurrent_waiters.round3.txt S10
With subscription_coalesce_ms=100, a burst of 10 commits is delivered as 1 delta holding all 10 rows, within ~110ms of the first commit; a second subscriber shares the view	✅ pass	live	live_ws_transcript_coalesce.round3.txt S11
Duplicate subscribers share one evaluation per commit, and colliding hashes never merge views or rows	⏸️ untested	no	The prior payload did not establish a live result for this scenario. Evaluation counts are kept only in the in-process SubscriptionMetrics, which no HTTP/WS endpoint exposes. Forcing hash collisions n…

cargo build --bin inputlayer-server (rebuilt at HEAD 0cc9f79f)
run_phase.round3.sh share 18471 (live WS driver: S1-S6 sharing/auth/unsubscribe/rule change)
run_phase.round3.sh cap 18472 INPUTLAYER_STORAGE__PERFORMANCE__MAX_RESULT_ROWS=3 (live S7-S8 result cap, retry, recovery)
N=200 run_phase.round3.sh ryw 18476 INPUTLAYER_HTTP__RATE_LIMIT__WS_MAX_MESSAGES_PER_SEC=100000 (live S9 read-your-writes, agent A/B order(42) pattern x200)
run_phase.round3.sh waiters 18474 INPUTLAYER_STORAGE__PERFORMANCE__MAX_RESULT_ROWS=3 (live S10 three concurrent new subscribers on a failed view, 5 rounds)
run_phase.round3.sh coalesce 18475 INPUTLAYER_HTTP__RATE_LIMIT__SUBSCRIPTION_COALESCE_MS=100 (live S11 coalescing window)
cargo test --test standing_query_tests (20 integration tests incl. sharing::*)
cargo test --lib protocol::subscription (50 unit tests incl. shared-evaluation counts and retry waiters)
cargo test --lib colliding (query-key and result-tuple hash collisions)

🔧 **Document** - 1 issue found ✅

ℹ️ gui/lib/docs-bundle.ts:3572 - The committed generated bundle gui/lib/docs-bundle.ts still holds the old websocket-api.mdx text (for example 'at most one evaluation runs per subscription', 'its seq is skipped', 'standing queries re-evaluate'). It is built from docs/content by gui/scripts/bundle-docs.mjs, and the gui prebuild/build/dev scripts run that step, so built artifacts are correct. The committed copy had already drifted before this change: running the script now also pulls in the backup guide, API key expiry and rotation, and other earlier docs. I left the copy unchanged to stay in scope. Suggested follow-up: regenerate it in its own commit, and either add a drift check in CI or stop committing the generated file.

✅ No issues found.

✅ **Lint** - passed

✅ No issues found.

✅ **Push** - passed

✅ No issues found.

Identical standing queries (same knowledge graph, same query text) now share one view: one evaluation per relevant commit and one result in memory, across connections and users. - views: shared-view state machine keyed by the full (KG, query) key, keeping the dependency and pending-change machine per view, with an optional bounded coalescing window (http.rate_limit.subscription_coalesce_ms, default 0, at most 100 ms). - hub: one worker task owns the views, its own notification receiver and the evaluations; connections attach and detach over a channel. - publication: lock-free hand-off (ArcSwap cell plus one queued wake-up per subscriber), so a slow connection coalesces instead of queueing. - subscriber: per-subscriber seq and last delivered result; read access is checked before a delta is built, so a denied push keeps the last result and leaves no seq gap. - result_set: rows keyed by structural hash plus exact equality instead of canonical JSON strings. - Read-access cache is keyed by role as well as ACL generation, so a demotion is seen at once. Evaluation no longer re-authorizes per subscriber; authorization happens at subscribe and before each push.

…s on rule change

… repeated failures

…triggered retries

…try flag

…ry request (#249) The release-checkpoint performance gate fails on main: cheap_query serial p50 is 1.110x the approved baseline e732cd5 on the 32-vCPU runner (115 -> 126 us). Concurrent latency and throughput did not move. Bisected to #214 (per-connection request pipelining), made worse by #222. Cause: the pipeline runs requests in a `FuturesUnordered`. Its `poll_next` wakes its own task and returns `Pending` after any poll that polled every future in the set. With one request in flight, which is every serial client, that happens on each request's first poll. The connection task wakes itself, tokio schedules that as a yield, and a yield wakes an idle worker to take the task. Each request therefore costs one more context switch (4.0 -> 5.0 per query) and a hand-off of the connection task to another core. #222 added no work per query but made the hand-off dearer: user cycles per query rose 15% at the same instruction count. Fix: keep one never-woken future in the set. After the pipeline's first poll it is never polled again, so no poll reaches every future and the forced yield no longer fires. The yield after futures that wake themselves still applies. Pipelining, ordering, deadlines, cancellation and the credential fence are unchanged. Nothing is added per request: no lock, atomic or allocation. Each connection gets one more set node. Bisect (cheap_query, `?edge(1, Y)`, serial). Server counters per query from perf stat. Ratios are against e732cd5, from an interleaved probe with one pinned server per commit, 40 rounds: | commit | PR | user instr/query | ctx switches/query | p50 ratio | |---|---|---|---|---| | e732cd5 | baseline | 1.282M | 4.0 | 1.000 | | d75ce6e | #208 atomic programs | 1.279M | 4.0 | 0.993 | | 4578fe6 | #215 notification stream | 1.271M | 4.0 | 0.991 | | 71a4d24 | #216 online backup | 1.275M | 4.0 | 1.016 | | 2f621c5 | #200 first-update latency | 1.272M | 4.0 | 1.009 | | e96df5c | #207 | 1.279M | 4.0 | 0.998 | | 4dd8fe9 | #209 shard index | 1.272M | 4.0 | 0.987 | | 277c7f0 | #206 | 1.274M | 4.0 | 1.008 | | 3554c98 | #214 pipelining, deadlines | 1.296M | 5.0 | 1.020 | | 3492a59 | #213 | 1.294M | 5.0 | 1.030 | | 2ba9a5a | #212 provenance | 1.297M | 5.0 | 1.029 | | 9ed61b4 | #204 key expiry | 1.297M | 5.0 | 1.027 | | be35ee3 | #221 | 1.297M | 5.0 | 1.013 | | c98a590 | #222 bounded streaming | 1.298M | 5.0 | 1.047 | | a6fd628 | #219 wildcard fix | 1.297M | 5.0 | 1.061 | | b7f7ad6 | #220 WAL durability | 1.300M | 5.0 | 1.056 | | de54807 | #223 shared standing eval | 1.299M | 5.0 | 1.071 | | 55ec34d | #225 checkpoint-1 fixes | 1.300M | 5.0 | 1.045 | | 23712b8 | main | 1.300M | 5.0 | 1.064 | | this change | | 1.300M | 4.0 | 1.033 | Confirming A/Bs with the gate's own fixture (fresh pinned servers each round, 40 rounds, ratio [95% interval]): - #214 vs its parent 277c7f0: 1.039 [1.027..1.049] - #222 vs its parent be35ee3: 1.025 [1.006..1.044] - A/A 3554c98: 1.010 [1.002..1.025] - e732cd5 -> main: 1.048 [1.028..1.066] - main -> this change: 0.978 [0.958..0.989] - e732cd5 -> this change: 1.021 [1.009..1.044], PASS (budget 1.05) User cycles per query (perf stat, same instruction count): e732cd5 958K, be35ee3 1053K, c98a590 1206K, this change 1006K. Full gate, all fixtures, 20 rounds, servers pinned to CPUs 2-15: - main -> this change: no metric worse. Every required metric is PASS, or INCONCLUSIVE with an interval that includes 1.0. cheap_query p50 0.979 [0.953..1.007]. - e732cd5 -> this change: cheap_query p50 1.029 [1.019..1.072], INCONCLUSIVE on a busy host (load average up to 22); bound_query p50 1.022, PASS. The insert and delta metrics improved since the baseline. The shared host is about 4x slower per query than the CI runner, so the same hand-off is a smaller share of each query here than there. The remaining ~1.7% of user instructions per query since #214 is the pipeline's own per-request bookkeeping: the boxed request future (one 4-8 KiB allocation), its set node, the request control and the in-flight maps.

jsam added 6 commits October 3, 2026 23:02

no-mistakes(review): Retry failed views, fence snapshots, retire view…

95a07da

…s on rule change

no-mistakes(review): Apply announced rule changes before attach; skip…

e9af311

… repeated failures

no-mistakes(test): Suppress repeated view errors only for subscriber-…

409e85f

…triggered retries

no-mistakes(document): Format subscription failure test assertion

b69f0f1

no-mistakes(review): Make new subscribers wait for refreshes; keep re…

0cc9f79

…try flag

jsam force-pushed the fm/il-e01 branch from 283a310 to 0cc9f79 Compare October 3, 2026 23:27

jsam merged commit de54807 into main Oct 3, 2026
14 checks passed

jsam deleted the fm/il-e01 branch October 3, 2026 23:35

Navigation Menu

Repository navigation

feat(protocol): share standing-query evaluations across identical subscriptions - #223

feat(protocol): share standing-query evaluations across identical subscriptions#223
jsam merged 6 commits into
mainfrom
fm/il-e01

jsam commented Oct 3, 2026 •

edited

Loading

Uh oh!

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

Conversation

jsam commented Oct 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Intent

What Changed

Risk Assessment

Testing

Pipeline

Uh oh!

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

jsam commented Oct 3, 2026 •

edited

Loading