path: root/packages/meshbay-hub/tests/harness/search_pool_harness.mjs

// Run the SHIPPED `ConnectionPool`, `fetchGroupIndex` and `fetchAllIndexes`
// against a fake clock and fake nodes, and count offers.
//
// The question this answers is how many connections Search negotiates, how many
// at once, and whether one it is reading from can be closed under it — the
// things that decided whether a phone ran into the hub's per-account ceilings.
// The pool, the index fetch and the sweep are lifted out of search-page.js as
// text; what is modelled is the clock and the nodes (`connectToGroup`, which
// is where an offer is made).
//
// Usage: node search_pool_harness.mjs <path to search-page.js> <json config>
import { readFileSync } from 'fs';

const src = readFileSync(process.argv[2], 'utf8');
const cfg = JSON.parse(process.argv[3] || '{}');

const {
  scenario = 'sweep',
  groups = 5,
  // Group indexes whose node never answers; `connectToGroup` fails after deadMs.
  dead = [],
  connectMs = 2000,       // a phone on 4G
  deadMs = 10000,
  indexMs = 300,
  runUntil = 600000,
} = cfg;

// ── The clock ────────────────────────────────────────────────────────────────

let now = 0;
let nextId = 1;
let timers = [];
globalThis.setTimeout = (fn, ms) => {
  const t = { at: now + (ms || 0), fn, id: nextId++ };
  timers.push(t);
  return t.id;
};
globalThis.clearTimeout = (id) => { timers = timers.filter((t) => t.id !== id); };
// The pool orders its connections by `Date.now()`, so that is the fake clock too.
Date.now = () => now;
const flush = async () => { for (let i = 0; i < 50; i++) await Promise.resolve(); };
async function run(until) {
  await flush();
  while (timers.length) {
    const due = timers.reduce((a, b) => (b.at < a.at ? b : a));
    if (due.at > until) break;
    timers = timers.filter((t) => t !== due);
    now = due.at;
    due.fn();
    await flush();
  }
  now = until;
}
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

// ── The nodes ────────────────────────────────────────────────────────────────

let offers = 0;
let negotiating = 0;
let peakNegotiating = 0;
let readsOnClosed = 0;
let pings = 0;
const transports = [];
// Groups whose held connection has died quietly, as a phone's does in its sleep.
const silentlyDead = new Set();

class FakeTransport {
  constructor(groupId) {
    this.groupId = groupId;
    this.closedFlag = false;
    transports.push(this);
  }
  get connected() { return !this.closedFlag; }
  close() { this.closedFlag = true; }
  async ping(timeoutMs) {
    pings += 1;
    if (silentlyDead.has(this.groupId)) {
      await sleep(timeoutMs);
      throw new Error('Response timeout');
    }
    await sleep(100);
  }
  async fetchIndex() {
    // Indexes differ in size, so reads end at different times and overlap
    // other groups' negotiations — which uniform timings never do.
    const n = Number(String(this.groupId).split('-')[1]);
    await sleep(indexMs * (1 + (n % 4)));
    if (this.closedFlag) { readsOnClosed += 1; throw new Error('Transport closed'); }
    return { entries: [{ id: `e-${this.groupId}` }] };
  }
}

async function connectToGroup(_hub, groupId) {
  offers += 1;
  negotiating += 1;
  peakNegotiating = Math.max(peakNegotiating, negotiating);
  try {
    const n = Number(String(groupId).split('-')[1]);
    if (dead.includes(n)) {
      await sleep(deadMs);
      throw new Error('Connection timeout');
    }
    await sleep(connectMs);
    return { transport: new FakeTransport(groupId), ack: {} };
  } finally {
    negotiating -= 1;
  }
}

globalThis.window = {};
let remembered = null;
globalThis.localStorage = { getItem: () => null, setItem: (_k, v) => { remembered = v; } };
const session = { bundleKey: 'k' };
const _loadBundleKey = async () => 'k';
const cacheGroupIndex = () => {};

// ── The code under test, as text ─────────────────────────────────────────────

const constant = (name) => {
  const m = new RegExp(`^const ${name} = (\\d+);`, 'm').exec(src);
  if (!m) throw new Error(`${name} is gone from search-page.js`);
  return `const ${name} = ${m[1]};`;
};
const lift = (signature) => {
  const start = src.indexOf(signature);
  if (start < 0) throw new Error(`${signature} is gone from search-page.js`);
  return src.slice(start, src.indexOf('\n}\n', start) + 3);
};

const make = new Function(
  'connectToGroup', 'window', 'session', '_loadBundleKey', 'cacheGroupIndex',
  'localStorage',
  `${constant('MAX_IN_FLIGHT')}
   ${constant('MAX_POOL_SIZE')}
   ${constant('REUSE_PING_MS')}
   const DOWN_KEY = 'harness';
   ${lift('class ConnectionPool {')}
   ${lift('async function fetchGroupIndex(')}
   ${lift('function lastKnownDown(')}
   ${lift('function rememberDown(')}
   ${lift('async function inFlight(')}
   ${lift('async function fetchAllIndexes(')}
   return { ConnectionPool, fetchAllIndexes, MAX_IN_FLIGHT, MAX_POOL_SIZE };`,
);
const code = make(connectToGroup, globalThis.window, session, _loadBundleKey,
                  cacheGroupIndex, globalThis.localStorage);

// ── The scenarios ────────────────────────────────────────────────────────────

const list = Array.from({ length: groups }, (_, i) => ({ id: `g-${i}`, name: `G${i}` }));
let evicted = 0;
const pool = new code.ConnectionPool('https://hub.example', () => { evicted += 1; });
const out = { maxInFlight: code.MAX_IN_FLIGHT, maxPool: code.MAX_POOL_SIZE };

// Groups whose index has landed — the only ones with tiles on screen.
const shown = new Set();
const sweep = async () => {
  const r = await code.fetchAllIndexes(pool, list, 'tok', 'alice', 'u1', () => {},
    (results) => { for (const id of results.keys()) shown.add(id); });
  return { found: r.results.size, unreachable: r.unreachable.length, at: now };
};

if (scenario === 'sweep') {
  sweep().then((r) => { Object.assign(out, r); });
  await run(runUntil);
} else if (scenario === 'refresh') {
  // A sweep, then the Search page's own refresh button: the second must reuse.
  sweep().then(async () => {
    out.offersFirst = offers;
    const again = await sweep();
    out.offersSecond = offers - out.offersFirst;
    Object.assign(out, again);
  });
  await run(runUntil);
} else if (scenario === 'slept') {
  // A sweep, then the phone sleeps and two connections die without saying so.
  sweep().then(async () => {
    out.offersFirst = offers;
    silentlyDead.add('g-0');
    silentlyDead.add('g-1');
    const again = await sweep();
    out.offersSecond = offers - out.offersFirst;
    Object.assign(out, again);
  });
  await run(runUntil);
} else if (scenario === 'tiles') {
  // A tile of every group asks for its connection while the sweep is on.
  sweep().then((r) => { Object.assign(out, r); });
  for (const g of list) pool.connect(g.id, 'tok', 'k', 'alice', 'u1').catch(() => {});
  await run(runUntil);
} else if (scenario === 'scroll') {
  // Tiles of groups already on screen keep using their connections while the
  // sweep is still reading other groups' indexes — which makes a connection an
  // index is being read from the least recently used one in the pool.
  let sweeping = true;
  sweep().then((r) => { sweeping = false; Object.assign(out, r); });
  const touch = async () => {
    while (sweeping) {
      await sleep(250);
      for (const id of shown) {
        if (pool.has(id)) pool.connect(id, 'tok', 'k', 'alice', 'u1').catch(() => {});
      }
    }
  };
  touch();
  await run(runUntil);
} else if (scenario === 'unmount') {
  // The page goes away while connections are still being negotiated.
  sweep().catch(() => {});
  await run(connectMs / 2);
  pool.closeAll();
  await run(runUntil);
  out.openAfter = transports.filter((t) => !t.closedFlag).length;
}

Object.assign(out, {
  offers, peakNegotiating, readsOnClosed, pings, evicted, poolSize: pool.size,
  remembered: remembered === null ? null : JSON.parse(remembered),
  openTransports: transports.filter((t) => !t.closedFlag).length,
});
console.log(JSON.stringify(out));