Function drain

apps/api/e2e/session-stress.ts:153–190

(sessions: Session[])

}

async function drain(sessions: Session[]) {
  scenario('drain: reap + flush until nothing remains');
  console.log(`   …waiting ${IDLE_WAIT_MS}ms for all sessions to idle out`);
  await sleep(IDLE_WAIT_MS);

  const ids = sessions.map((s) => s.sessionId);
  const target = ids.length;
  const deadline = Date.now() + DRAIN_TIMEOUT_MS;
  let ends = 0;
  let remaining = Number.POSITIVE_INFINITY;

  while (Date.now() < deadline) {
    await triggerReaper(); // enqueue session_end for idle sessions
    await sleep(1000); // let the worker process the close jobs
    await triggerCron('flushEvents'); // push session_start/end + events → CH
    await triggerCron('flushSessions'); // push sessions table rows → CH

    ends = await countByName(ids, 'session_end');
    remaining = await redis.zcard(wallclockKey);
    console.log(`   …session_end ${ends}/${target}, wallclock remaining ${remaining}`);
    if (ends >= target && remaining === 0) break;
    await sleep(1500);
  }

  check('clickhouse: one session_end per session', ends === target, `got ${ends}`);
  check('redis: wallclock index fully drained', remaining === 0, `${remaining} left`);

  const bufLen = await redis.llen(SESSION_BUFFER_LIST);
  check('redis: session buffer drained to ClickHouse', bufLen === 0, `${bufLen} rows pending`);

  // Spot-check a sample of devices: no session blob should linger after close.
  const sample = sessions.filter((_, k) => k % Math.ceil(sessions.length / 20) === 0);
  let leaked = 0;
  for (const s of sample) {
    if (await getBlob(s.deviceId)) leaked++;
  }
  check('redis: no session blobs leaked (sampled)', leaked === 0, `${leaked}/${sample.length} leaked`);
}

async function reconcile(sessions: Session[]) {
  scenario('reconcile: ClickHouse event counts');