feat(inputlayer-js): subscribe(), watch() and on() with typed Change events - #241

HEAD 519074ba941281a5531398f2692bff900fca9e56

 RUN  v3.2.7 ~/.no-mistakes/worktrees/9eea230c3aae/01M43FP2QVZ84NFH1M5GPXHK5E/packages/inputlayer-js

(node:3419414) MaxListenersExceededWarning: Possible EventTarget memory leak detected. 11 disconnected listeners added to EventTarget. MaxListeners is 10. Use events.setMaxListeners() to increase limit
(Use `node --trace-warnings ...` to show where the warning was created)
(node:3419414) MaxListenersExceededWarning: Possible EventTarget memory leak detected. 11 disconnected listeners added to EventTarget. MaxListeners is 10. Use events.setMaxListeners() to increase limit
(node:3419414) MaxListenersExceededWarning: Possible EventTarget memory leak detected. 11 disconnected listeners added to EventTarget. MaxListeners is 10. Use events.setMaxListeners() to increase limit
(node:3419414) MaxListenersExceededWarning: Possible EventTarget memory leak detected. 11 disconnected listeners added to EventTarget. MaxListeners is 10. Use events.setMaxListeners() to increase limit
(node:3419414) MaxListenersExceededWarning: Possible EventTarget memory leak detected. 11 disconnected listeners added to EventTarget. MaxListeners is 10. Use events.setMaxListeners() to increase limit
(node:3419414) MaxListenersExceededWarning: Possible EventTarget memory leak detected. 11 disconnected listeners added to EventTarget. MaxListeners is 10. Use events.setMaxListeners() to increase limit
 ✓ tests/subscriptions.integration.test.ts > Live: subscriptions > snapshot plus deltas equals a fresh query at every checkpoint of random histories 7717ms
 ✓ tests/subscriptions.integration.test.ts > Live: subscriptions > projection_two_supporting_tuples: a projected row leaves with its last support 150ms
 ✓ tests/subscriptions.integration.test.ts > Live: subscriptions > assembles a streamed snapshot and streamed deltas 367ms
 ✓ tests/subscriptions.integration.test.ts > Live: subscriptions > a seq gap and a reset end in unverified, then an exact resync 103ms
 ✓ tests/subscriptions.integration.test.ts > Live: subscriptions > a slow consumer gets unverified, then a resync once it has read the queue 580ms
 ✓ tests/subscriptions.integration.test.ts > Live: subscriptions > reconnects: unverified at once, then a resync with only what changed meanwhile 137ms
 ✓ tests/subscriptions.integration.test.ts > Live: subscriptions > an ACL revoke ends the subscription: unverified, then SubscriptionRejectedError 127ms
 ✓ tests/subscriptions.integration.test.ts > Live: subscriptions > refuses what a standing query cannot track, before or at the engine 45ms
 ✓ tests/subscriptions.integration.test.ts > Live: subscriptions > watch yields the whole result; on calls back with each change 51ms

 Test Files  1 passed (1)
      Tests  9 passed (9)
   Start at  13:22:21
   Duration  10.33s (transform 207ms, setup 0ms, collect 260ms, tests 9.80s, environment 0ms, prepare 75ms)

HEAD 519074ba941281a5531398f2692bff900fca9e56

 RUN  v3.2.7 ~/.no-mistakes/worktrees/9eea230c3aae/01M43FP2QVZ84NFH1M5GPXHK5E/packages/inputlayer-js

(node:3414495) MaxListenersExceededWarning: Possible EventTarget memory leak detected. 11 disconnected listeners added to EventTarget. MaxListeners is 10. Use events.setMaxListeners() to increase limit
(Use `node --trace-warnings ...` to show where the warning was created)
 ✓ tests/connection.integration.test.ts (7 tests) 658ms
(node:3414495) MaxListenersExceededWarning: Possible EventTarget memory leak detected. 11 disconnected listeners added to EventTarget. MaxListeners is 10. Use events.setMaxListeners() to increase limit
 ✓ tests/subscription.test.ts (31 tests) 2621ms
   ✓ subscribe > a connection lost before the first snapshot reopens on the new connection  836ms
   ✓ subscribe > a connection lost during the session-rule check still yields the snapshot  706ms
   ✓ subscribe > a .subscribe reply lost past its deadline is retried after unsubscribing the id  808ms
(node:3414495) MaxListenersExceededWarning: Possible EventTarget memory leak detected. 11 disconnected listeners added to EventTarget. MaxListeners is 10. Use events.setMaxListeners() to increase limit
(node:3414495) MaxListenersExceededWarning: Possible EventTarget memory leak detected. 11 disconnected listeners added to EventTarget. MaxListeners is 10. Use events.setMaxListeners() to increase limit
(node:3414495) MaxListenersExceededWarning: Possible EventTarget memory leak detected. 11 disconnected listeners added to EventTarget. MaxListeners is 10. Use events.setMaxListeners() to increase limit
(node:3414495) MaxListenersExceededWarning: Possible EventTarget memory leak detected. 11 disconnected listeners added to EventTarget. MaxListeners is 10. Use events.setMaxListeners() to increase limit
 ✓ tests/subscriptions.integration.test.ts (9 tests) 10098ms
   ✓ Live: subscriptions > snapshot plus deltas equals a fresh query at every checkpoint of random histories  7831ms
   ✓ Live: subscriptions > assembles a streamed snapshot and streamed deltas  452ms
   ✓ Live: subscriptions > a slow consumer gets unverified, then a resync once it has read the queue  584ms

 Test Files  3 passed (3)
      Tests  47 passed (47)
   Start at  13:22:02
   Duration  10.63s (transform 323ms, setup 0ms, collect 799ms, tests 13.38s, environment 1ms, prepare 228ms)

HEAD 519074ba941281a5531398f2692bff900fca9e56
[13:22:37.046] server up pid 3421358
[13:22:37.187] subscribe(): snapshot rev=7 verified=true +[{"id":1,"owner":10},{"id":2,"owner":10}] -[]
[13:22:37.187] on(owner): snapshot rev=7 verified=true +[{"Owner":10}] -[]
[13:22:37.187] watch(): rows=[{"id":1,"owner":10},{"id":2,"owner":10}] rev=7 verified=true
[13:22:37.189] PASS first event is the exact snapshot (2 rows)
[13:22:37.189] PASS projected snapshot has owner 10 once
[13:22:37.189] --- write: +task(3,20), -task(1,10)
[13:22:37.192] subscribe(): delta rev=8 verified=true +[{"id":3,"owner":20}] -[]
[13:22:37.192] on(owner): delta rev=8 verified=true +[{"Owner":20}] -[]
[13:22:37.192] watch(): rows=[{"id":1,"owner":10},{"id":2,"owner":10},{"id":3,"owner":20}] rev=8 verified=true
[13:22:37.193] subscribe(): delta rev=9 verified=true +[] -[{"id":1,"owner":10}]
[13:22:37.193] watch(): rows=[{"id":2,"owner":10},{"id":3,"owner":20}] rev=9 verified=true
[13:22:37.213] PASS writes arrive as deltas
[13:22:37.214] PASS projection: owner 10 not retracted while task 2 still supports it
[13:22:37.214] --- real outage: stopping server process
[13:22:37.251] subscribe(): unverified(connection_lost) rev=9 verified=false +[] -[]
[13:22:37.251] on(owner): unverified(connection_lost) rev=9 verified=false +[] -[]
[13:22:37.251] watch(): rows=[{"id":2,"owner":10},{"id":3,"owner":20}] rev=9 verified=false
[13:22:37.254] server stopped
[13:22:37.254] PASS subscribe(): unverified(connection_lost) at once
[13:22:37.254] PASS watch(): verified=false during outage
[13:22:38.754] --- restarting server on same data dir
[13:22:38.856] subscribe(): resync rev=1 verified=true +[] -[]
[13:22:38.856] on(owner): resync rev=1 verified=true +[] -[]
[13:22:38.856] watch(): rows=[{"id":2,"owner":10},{"id":3,"owner":20}] rev=1 verified=true
[13:22:38.857] late subscribe(): snapshot rev=1 verified=true +[{"id":2,"owner":10},{"id":3,"owner":20}] -[]
[13:22:39.009] server up pid 3421810
[13:22:39.009] PASS subscription opened during outage yields its snapshot after restart
[13:22:39.009] PASS resync after restart replays nothing unchanged (empty diff)
[13:22:39.009] --- write after reconnect: +task(4,30), -task(2,10), -task(3,20)
[13:22:39.012] subscribe(): delta rev=4 verified=true +[{"id":4,"owner":30}] -[]
[13:22:39.013] on(owner): delta rev=4 verified=true +[{"Owner":30}] -[]
[13:22:39.013] late subscribe(): delta rev=4 verified=true +[{"id":4,"owner":30}] -[]
[13:22:39.013] watch(): rows=[{"id":2,"owner":10},{"id":3,"owner":20},{"id":4,"owner":30}] rev=4 verified=true
[13:22:39.014] subscribe(): delta rev=5 verified=true +[] -[{"id":2,"owner":10}]
[13:22:39.014] on(owner): delta rev=5 verified=true +[] -[{"Owner":10}]
[13:22:39.014] late subscribe(): delta rev=5 verified=true +[] -[{"id":2,"owner":10}]
[13:22:39.014] watch(): rows=[{"id":3,"owner":20},{"id":4,"owner":30}] rev=5 verified=true
[13:22:39.015] subscribe(): delta rev=6 verified=true +[] -[{"id":3,"owner":20}]
[13:22:39.015] on(owner): delta rev=6 verified=true +[] -[{"Owner":20}]
[13:22:39.015] late subscribe(): delta rev=6 verified=true +[] -[{"id":3,"owner":20}]
[13:22:39.015] watch(): rows=[{"id":4,"owner":30}] rev=6 verified=true
[13:22:39.036] PASS subscriber set equals fresh query ["[4,30]"]
[13:22:39.036] PASS watch(): whole set, verified again
[13:22:39.036] PASS projection: owner 10 retracted when its last support left
[13:22:39.037] closed iterator ended cleanly
[13:22:39.071] server stopped
[13:22:39.073] ALL CHECKS PASSED

/**
 * End-user drive of kg.subscribe()/watch()/on() against a real inputlayer-server,
 * including a real server outage (process stopped and restarted on the same data dir)
 * and a subscription opened while the server is down.
 * Run from packages/inputlayer-js: npx vite-node <this file>
 */
import { spawn, type ChildProcess } from 'node:child_process';
import { mkdtempSync, rmSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { InputLayer, relation, type Change } from '~/.no-mistakes/worktrees/9eea230c3aae/01M43FP2QVZ84NFH1M5GPXHK5E/packages/inputlayer-js/src/index';

const ROOT = '~/.no-mistakes/worktrees/9eea230c3aae/01M43FP2QVZ84NFH1M5GPXHK5E';
const PORT = 8097;
const URL = `ws://127.0.0.1:${PORT}/ws`;
const dataDir = mkdtempSync(join(tmpdir(), 'il-sub-demo-'));
let server: ChildProcess | undefined;
const log = (...a: unknown[]) => console.log(`[${new Date().toISOString().slice(11, 23)}]`, ...a);
const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
let failures = 0;
const check = (ok: boolean, what: string) => {
  log(ok ? 'PASS' : 'FAIL', what);
  if (!ok) failures++;
};

async function startServer() {
  server = spawn(`/tmp/il-p1g-MKcd/inputlayer-server`, ['--host', '127.0.0.1', '--data-dir', dataDir, '--port', String(PORT)], {
    env: { ...process.env, INPUTLAYER_ADMIN_PASSWORD: 'admin' },
    stdio: 'ignore',
  });
  for (let i = 0; i < 80; i++) {
    try {
      if ((await fetch(`http://127.0.0.1:${PORT}/health`)).ok) return log('server up pid', server.pid);
    } catch {}
    await sleep(250);
  }
  throw new Error('server did not start');
}
async function stopServer() {
  const s = server!;
  const done = new Promise((r) => s.once('exit', r));
  s.kill('SIGTERM');
  await done;
  log('server stopped');
}
async function waitFor(p: () => boolean, ms = 15000) {
  const end = Date.now() + ms;
  while (!p()) {
    if (Date.now() > end) throw new Error('timeout');
    await sleep(20);
  }
}
const fmt = (c: Change) =>
  `${c.kind}${c.kind === 'unverified' ? `(${c.reason})` : ''} rev=${c.revision} verified=${c.verified} +${JSON.stringify(c.inserted)} -${JSON.stringify(c.retracted)}`;
const key = (r: object) => JSON.stringify(Object.values(r));

async function main() {
  await startServer();
  const opts = { url: URL, username: 'admin', password: 'admin', reconnectDelay: 0.05, maxReconnectDelay: 0.5, maxReconnectAttempts: 200 };
  const writerClient = new InputLayer(opts);
  await writerClient.connect();
  const writer = writerClient.knowledgeGraph('sub_demo');
  const Task = relation('Task', { id: 'int', owner: 'int' });
  await writer.define(Task);
  await writer.insert(Task, [{ id: 1, owner: 10 }, { id: 2, owner: 10 }]);

  const listenerClient = new InputLayer(opts);
  await listenerClient.connect();
  const kg = listenerClient.knowledgeGraph('sub_demo');

  // 1. async iterator
  const held = new Map<string, object>();
  const iterEvents: Change[] = [];
  const sub = kg.subscribe(Task);
  const iterDone = (async () => {
    for await (const c of sub) {
      iterEvents.push(c);
      log('subscribe():', fmt(c));
      for (const r of c.retracted) held.delete(key(r));
      for (const r of c.inserted) held.set(key(r), r);
    }
  })().catch((e) => log('subscribe() ended with', e));
  // 2. watch
  const watchSeen: Array<{ rows: unknown[]; verified: boolean; revision: number }> = [];
  const watcher = kg.watch(Task);
  const watchDone = (async () => {
    for await (const live of watcher) {
      watchSeen.push(live as never);
      log('watch():', `rows=${JSON.stringify(live.rows)} rev=${live.revision} verified=${live.verified}`);
    }
  })().catch((e) => log('watch() ended with', e));
  // 3. on (projected: owner only — two tasks share owner 10)
  const onEvents: Change[] = [];
  const handle = kg.on({ select: [Task.col('owner').toAst()], join: [Task] }, (c) => {
    onEvents.push(c);
    log('on(owner):', fmt(c));
  }, { onError: (e) => log('on() error', e) });

  await waitFor(() => iterEvents.length >= 1 && watchSeen.length >= 1 && onEvents.length >= 1);
  check(iterEvents[0].kind === 'snapshot' && iterEvents[0].inserted.length === 2, 'first event is the exact snapshot (2 rows)');
  check(onEvents[0].kind === 'snapshot' && onEvents[0].inserted.length === 1, 'projected snapshot has owner 10 once');

  log('--- write: +task(3,20), -task(1,10)');
  await writer.insert(Task, { id: 3, owner: 20 });
  await writer.delete(Task, { id: 1, owner: 10 });
  await waitFor(() => held.size === 2 && held.has('[3,20]') && !held.has('[1,10]'));
  check(iterEvents.slice(1).every((c) => c.kind === 'delta'), 'writes arrive as deltas');
  const ownerRetracts = onEvents.flatMap((c) => c.retracted);
  check(ownerRetracts.length === 0, 'projection: owner 10 not retracted while task 2 still supports it');

  log('--- real outage: stopping server process');
  await stopServer();
  await waitFor(() => iterEvents.some((c) => c.kind === 'unverified') && onEvents.some((c) => c.kind === 'unverified') && watchSeen.some((w) => !w.verified));
  check(iterEvents.at(-1)!.kind === 'unverified' && iterEvents.at(-1)!.reason === 'connection_lost', 'subscribe(): unverified(connection_lost) at once');
  check(watchSeen.at(-1)!.verified === false, 'watch(): verified=false during outage');

  // 4. a subscription opened while the server is down (first open must survive the outage)
  const lateEvents: Change[] = [];
  const late = kg.subscribe(Task);
  const lateDone = (async () => {
    for await (const c of late) {
      lateEvents.push(c);
      log('late subscribe():', fmt(c));
    }
  })().catch((e) => {
    log('late subscribe() ended with', e);
    lateEvents.push({ kind: 'ERROR' } as never);
  });
  await sleep(1500);

  log('--- restarting server on same data dir');
  await startServer();
  await waitFor(() => iterEvents.at(-1)!.kind === 'resync' && onEvents.at(-1)!.kind === 'resync' && lateEvents.length >= 1, 20000);
  check(lateEvents[0].kind === 'snapshot' && lateEvents[0].inserted.length === 2, 'subscription opened during outage yields its snapshot after restart');
  const resync = iterEvents.at(-1)!;
  check(resync.inserted.length === 0 && resync.retracted.length === 0, 'resync after restart replays nothing unchanged (empty diff)');

  log('--- write after reconnect: +task(4,30), -task(2,10), -task(3,20)');
  await writer.insert(Task, { id: 4, owner: 30 });
  await writer.delete(Task, { id: 2, owner: 10 });
  await writer.delete(Task, { id: 3, owner: 20 });
  await waitFor(() => held.size === 1 && held.has('[4,30]'));
  await waitFor(() => onEvents.flatMap((c) => c.retracted).some((r) => key(r) === '[10]'));
  const fresh = (await writer.execute('?task(I, O)')).toTuples().map((t) => JSON.stringify(t)).sort();
  check(JSON.stringify([...held.keys()].sort()) === JSON.stringify(fresh), `subscriber set equals fresh query ${JSON.stringify(fresh)}`);
  check(watchSeen.at(-1)!.verified && JSON.stringify(watchSeen.at(-1)!.rows) === JSON.stringify([{ id: 4, owner: 30 }]), 'watch(): whole set, verified again');
  check(onEvents.flatMap((c) => c.retracted).some((r) => key(r) === '[10]'), 'projection: owner 10 retracted when its last support left');

  await sub.close();
  await watcher.return(undefined);
  await handle.close();
  await late.close();
  await Promise.all([iterDone, watchDone, lateDone]);
  log('closed iterator ended cleanly');
  await writerClient.dropKnowledgeGraph('sub_demo').catch(() => undefined);
  await listenerClient.close();
  await writerClient.close();
}

main()
  .catch((e) => {
    failures++;
    log('ERROR', e);
  })
  .finally(async () => {
    if (server && server.exitCode === null) await stopServer();
    rmSync(dataDir, { recursive: true, force: true });
    log(failures === 0 ? 'ALL CHECKS PASSED' : `${failures} CHECK(S) FAILED`);
    process.exit(failures === 0 ? 0 : 1);
  });

first {"rows":[],"revision":225,"verified":true}
return resolved; pending next: {"done":true}