gui: fix RpcClient double-free on stop() and the 1000-row list cap

Both found live while building gui/tests/dod's DoD harness, not from reading the code — see that commit for how. RpcClient::stop() left conn_ dangling after joining the worker thread: the thread's own finish() flushes the DeferredDelete stop()'s connect(&thread_, &QThread::finished, conn_, &QObject::deleteLater) already posted, so conn_ is gone by the time stop() returns, but nothing cleared the pointer. Any caller that calls stop() and later lets the client destruct (the harness's own client.stop() at shutdown; also plain, correct API usage) hit a double-free in the destructor's leftover `delete conn_`. Caught by ASan on the very first run that actually exercised the stop-then-destroy path. requestInitialList() also called download.list with a hardcoded `{"limit": 1000}`, silently capping the table at 1000 rows no matter how many the daemon actually has — download.list.schema.json's own description says "the GUI pages", not "the GUI takes it all in one call". The scroll-60fps DoD gate refused to run against mockd --tasks 10000 rather than "pass" against a 1000-row table, which is what surfaced it. requestInitialList() now pages (5000 per call, the schema's own max) until `total` is satisfied, then resets the model once with everything. Separately: RpcConnection's session.subscribe list never included event.settings.changed or event.grabber.progress, even though RpcClient has carried signals for both since the Options/Grabber work — session.subscribe "replaces the previous selection" and "nothing is delivered until this is called", so both events were being silently dropped by any real daemon that enforces the subscription (mockd does; verified live with a second subscribed client actually receiving event.settings.changed after this fix, round-tripped through a real veloxd's settings.set). GrabberWizard's 5 s poll fallback is exactly why this went unnoticed until now — it covered for the missing push the whole time. Co-Authored-By: Claude Sonnet 5 <[email protected]> Claude-Session: https://claude.ai/code/session_01NSCdCWFXBTSBBK3MzWtJiC
2026-09-12 21:27:28 +04:00
5 changed files with 49 additions and 427 deletions
@@ -19,12 +19,7 @@ import {
 } from './context-menus.js';
 import { MediaBridge, notifyTab } from './media-bridge.js';
 import { createTransport, transportStorage, type TransportStatus, type VeloxTransport } from './transport/index.js';
-import {
+import type { CaptureOfferParams, CaptureRules, DownloadSpec } from '../shared/protocol/index.js';
  SESSION_SUBSCRIBE_PARAMS_EVENTS_ITEM_VALUES,
  type CaptureOfferParams,
  type CaptureRules,
  type DownloadSpec,
 } from '../shared/protocol/index.js';
 let transport: VeloxTransport | undefined;
 let rules: CaptureRules = DEFAULT_CAPTURE_RULES;
@@ -80,31 +75,10 @@ async function refreshRules(): Promise<void> {
  }
 }
 /**
 * "Nothing is delivered until this is called" (session.subscribe's own description) —
 * without it, event.task.progress et al. never reach this connection at all, no matter
 * how many listeners bridge.ts registers locally. Requests the whole set every time
 * because any popup/options document could open at any moment and none of them narrow
 * per-tab; a fresh connection (first connect, or after a drop) starts with nothing
 * subscribed until this runs again.
 */
 async function subscribeToEvents(): Promise<void> {
  try {
    await mustTransport().call('session.subscribe', {
      events: [...SESSION_SUBSCRIBE_PARAMS_EVENTS_ITEM_VALUES],
    });
  } catch {
    // Best-effort; a reconnect (or the next event.settings.changed-driven refresh) retries.
  }
 }
 function onTransportState(status: TransportStatus): void {
  const detail = status.fatal ?? (status.needsPairing ? 'needs pairing' : '');
  console.debug(`[velox] transport ${status.state}${detail ? ` — ${detail}` : ''}`);
-  if (status.state === 'connected') {
+  if (status.state === 'connected') void refreshRules();
    void refreshRules();
    void subscribeToEvents();
  }
 }
 async function setOverride(override: 'auto' | 'ws' | 'uds'): Promise<void> {
@@ -1,389 +0,0 @@
 // Runs the transport, the capture hook, and the popup event path against a REAL veloxd
 // — not FakeDaemon. Everything else in this suite is faithful to the documented wire
 // protocol, but "faithful" isn't "real"; this is what actually proves it.
 //
 // Requires VELOXD_BIN (path to a built veloxd) in the environment. Skips itself with a
 // clear message otherwise, so `npm test` and CI (no daemon binary lying around) are
 // unaffected. Run it like:
 //
 //   VELOXD_BIN=/path/to/build/dev/bin/veloxd npx vitest run tests/live
 //
 // Each veloxd instance gets its own scratch XDG_RUNTIME_DIR/XDG_DATA_HOME/
 // XDG_CONFIG_HOME/HOME (main.cpp's single-instance lock is keyed to the runtime dir, so
 // this can run alongside another developer's or CI's own veloxd on the same machine).
 // VELOX_PAIR_AUTO=1 stands in for the GUI's Allow-prompt approver during dev/test
 // (rpc/pairing.hpp's EnvAutoApprover) — pairing itself is exercised for real, only the
 // human click is stubbed.
 import { spawn, type ChildProcessWithoutNullStreams } from 'node:child_process';
 import { mkdtempSync, mkdirSync, rmSync } from 'node:fs';
 import { readFile } from 'node:fs/promises';
 import { tmpdir } from 'node:os';
 import { dirname, join, resolve } from 'node:path';
 import { fileURLToPath } from 'node:url';
 import { afterAll, beforeAll, describe, expect, it } from 'vitest';
 import { WebSocket as WsClient } from 'ws';
 import { RpcError, TransportClosedError } from '../../src/background/transport/types.js';
 import { WebSocketTransport, type WebSocketCtor, type WebSocketTransportDeps } from '../../src/background/transport/websocket.js';
 import { CaptureHook } from '../../src/background/capture/index.js';
 import type { OnHeadersReceivedDetails } from '../../src/background/capture/index.js';
 import type { CaptureRules, DownloadSpec, TaskProgressEvent, TaskStateEvent } from '../../src/shared/protocol/index.js';
 const VELOXD_BIN = process.env.VELOXD_BIN;
 const REPO_ROOT = resolve(dirname(fileURLToPath(import.meta.url)), '../../..');
 const TESTSERVER_PY = join(REPO_ROOT, 'tools/testserver/testserver.py');
 // A fixed moz-extension origin, used both as session.pair's extensionId and as the WS
 // upgrade's Origin header — real Firefox sets the latter itself; ws's client needs it
 // spelled out (docs/05 §4: the daemon refuses the upgrade without a moz-extension:// Origin).
 const EXTENSION_ID = '11111111-2222-3333-4444-555555555555';
 const ORIGIN = `moz-extension://${EXTENSION_ID}`;
 class OriginWebSocket extends WsClient {
  constructor(url: string) {
    super(url, { origin: ORIGIN });
  }
 }
 const CTOR = OriginWebSocket as unknown as WebSocketCtor;
 function sleep(ms: number): Promise<void> {
  return new Promise((r) => setTimeout(r, ms));
 }
 async function waitFor(cond: () => Promise<boolean> | boolean, timeoutMs: number, what: string): Promise<void> {
  const deadline = Date.now() + timeoutMs;
  for (;;) {
    if (await cond()) return;
    if (Date.now() > deadline) throw new Error(`timed out waiting for ${what}`);
    await sleep(50);
  }
 }
 interface VeloxdInstance {
  proc: ChildProcessWithoutNullStreams;
  scratch: string;
  wsPort: number;
  /** True once the process has actually exited, by signal or otherwise. Node only sets
   *  `proc.exitCode` for a normal exit — a signal-killed process reports its death via
   *  `signalCode` and an `exit` event instead, never a non-null `exitCode`. */
  hasExited(): boolean;
  kill(signal?: NodeJS.Signals): void;
 }
 async function startVeloxd(bin: string): Promise<VeloxdInstance> {
  const scratch = mkdtempSync(join(tmpdir(), 'velox-live-'));
  const runtime = join(scratch, 'rt');
  const data = join(scratch, 'data');
  const config = join(scratch, 'cfg');
  const home = join(scratch, 'home');
  mkdirSync(runtime, { mode: 0o700 });
  mkdirSync(data, { recursive: true });
  mkdirSync(config, { recursive: true });
  mkdirSync(join(home, 'Downloads'), { recursive: true });
  const proc = spawn(bin, [], {
    env: {
      ...process.env,
      VELOX_PAIR_AUTO: '1',
      XDG_RUNTIME_DIR: runtime,
      XDG_DATA_HOME: data,
      XDG_CONFIG_HOME: config,
      HOME: home,
    },
  });
  let exited = false;
  proc.on('exit', () => {
    exited = true;
  });
  let stderr = '';
  proc.stderr.on('data', (d) => {
    stderr += String(d);
  });
  const portFile = join(runtime, 'velox', 'ws.port');
  try {
    await waitFor(async () => {
      if (exited) throw new Error(`veloxd exited early (code ${proc.exitCode}, signal ${proc.signalCode}): ${stderr}`);
      try {
        await readFile(portFile);
        return true;
      } catch {
        return false;
      }
    }, 10_000, 'veloxd to write ws.port');
  } catch (e) {
    proc.kill('SIGKILL');
    rmSync(scratch, { recursive: true, force: true });
    throw e;
  }
  const wsPort = Number((await readFile(portFile, 'utf8')).trim());
  return {
    proc,
    scratch,
    wsPort,
    hasExited: () => exited,
    kill(signal: NodeJS.Signals = 'SIGTERM') {
      proc.kill(signal);
    },
  };
 }
 interface TestServerInstance {
  proc: ChildProcessWithoutNullStreams;
  baseUrl: string;
 }
 async function startTestServer(): Promise<TestServerInstance> {
  const proc = spawn('python3', [TESTSERVER_PY, '--port', '0'], {});
  let stdout = '';
  let port: number | null = null;
  proc.stdout.on('data', (d) => {
    stdout += String(d);
    const m = /^(\d+)\s*$/m.exec(stdout);
    if (m) port = Number(m[1]);
  });
  await waitFor(() => port !== null, 5_000, 'testserver to print its port');
  const baseUrl = `http://127.0.0.1:${port}`;
  await waitFor(async () => {
    try {
      const res = await fetch(`${baseUrl}/__health`);
      return res.ok;
    } catch {
      return false;
    }
  }, 5_000, 'testserver /__health');
  return { proc, baseUrl };
 }
 function memDeps(init: { token?: string | null } = {}): { deps: WebSocketTransportDeps; store: { token: string | null } } {
  const store = { token: init.token ?? null };
  return {
    store,
    deps: {
      getToken: async () => store.token,
      setToken: async (t) => {
        store.token = t;
      },
      getCachedPort: async () => null,
      setCachedPort: async () => undefined,
      extensionId: EXTENSION_ID,
    },
  };
 }
 function makeTransport(port: number, deps: WebSocketTransportDeps, extra: Partial<WebSocketTransportDeps> = {}): WebSocketTransport {
  return new WebSocketTransport({
    ...deps,
    ...extra,
    webSocketCtor: CTOR,
    portRange: { start: port, end: port },
    openTimeoutMs: 2000,
  });
 }
 const maybeDescribe = VELOXD_BIN ? describe : describe.skip;
 if (!VELOXD_BIN) {
  console.warn('tests/live/real-veloxd.test.ts: VELOXD_BIN not set — skipping (see file header).');
 }
 maybeDescribe('WebSocketTransport against a real veloxd', () => {
  let daemon: VeloxdInstance;
  let testserver: TestServerInstance;
  // The mid-test fail-open case kills `daemon` and a later test starts a replacement —
  // every scratch dir that ever existed gets cleaned up here, not just the last one.
  const allScratchDirs: string[] = [];
  async function freshVeloxd(): Promise<VeloxdInstance> {
    const d = await startVeloxd(VELOXD_BIN!);
    allScratchDirs.push(d.scratch);
    return d;
  }
  beforeAll(async () => {
    daemon = await freshVeloxd();
    testserver = await startTestServer();
  }, 20_000);
  afterAll(() => {
    daemon?.kill('SIGKILL');
    testserver?.proc.kill('SIGKILL');
    for (const dir of allScratchDirs) rmSync(dir, { recursive: true, force: true });
  });
  it('session.hello without a token surfaces NotPaired / needsPairing', async () => {
    const { deps } = memDeps();
    const t = makeTransport(daemon.wsPort, deps, { autoPair: false });
    await expect(t.connect()).rejects.toBeInstanceOf(RpcError);
    expect(t.status.needsPairing).toBe(true);
    t.disconnect();
  });
  let pairedToken: string;
  it('pairs (VELOX_PAIR_AUTO=1 stands in for the human Allow click) and hellos with the issued token', async () => {
    const { deps, store } = memDeps();
    const t = makeTransport(daemon.wsPort, deps); // autoPair: true (default)
    await t.connect();
    expect(t.state).toBe('connected');
    expect(t.status.daemonVersion).toBeTruthy();
    expect(store.token).toBeTruthy();
    pairedToken = store.token!;
    t.disconnect();
  });
  it('the pairing token survives a reconnect: a fresh transport reuses it with no fresh pairing', async () => {
    const { deps } = memDeps({ token: pairedToken });
    // autoPair: false — if this succeeds at all, it can only be because the stored
    // token from the previous test was accepted outright, not because this transport
    // silently re-paired.
    const t = makeTransport(daemon.wsPort, deps, { autoPair: false });
    await t.connect();
    expect(t.state).toBe('connected');
    t.disconnect();
  });
  it('a wrong token is rejected, and repeating it rate-limits the next pairing attempt', async () => {
    // Five failed session.hello attempts from this origin (ws_server.cpp records a
    // rate-limiter failure on every not-paired hello, not only on a failed session.pair)
    // exhausts the window; the sixth thing this origin tries — a pairing attempt — gets
    // RateLimited rather than a fresh token.
    for (let i = 0; i < 5; i += 1) {
      const { deps } = memDeps({ token: 'not-the-real-token' });
      const t = makeTransport(daemon.wsPort, deps, { autoPair: false });
      await expect(t.connect()).rejects.toBeInstanceOf(RpcError);
      t.disconnect();
    }
    const { deps } = memDeps(); // no token -> autoPair kicks in -> session.pair
    const t = makeTransport(daemon.wsPort, deps);
    await expect(t.connect()).rejects.toBeInstanceOf(RpcError);
    expect(t.status.needsPairing).toBe(true);
    expect(t.status.retryAfterSec).toBeGreaterThan(0);
    t.disconnect();
  });
  it('download.add creates a real task the engine picks up', async () => {
    const { deps } = memDeps({ token: pairedToken });
    const t = makeTransport(daemon.wsPort, deps, { autoPair: false });
    await t.connect();
    try {
      const spec: DownloadSpec = { url: `${testserver.baseUrl}/plain/file/64K`, filename: 'plain-download.bin' };
      const added = await t.call('download.add', spec);
      expect(added.taskId).toBeTruthy();
      await waitFor(async () => {
        const detail = await t.call('download.get', { taskId: added.taskId });
        const state = detail.summary.state;
        return state === 'complete' || state === 'downloading' || state === 'verifying';
      }, 10_000, 'the task to leave the queued state');
    } finally {
      t.disconnect();
    }
  }, 15_000);
  it('capture.offer end to end: the real capture path takes a monitored download, ignores its own duplicate, and fails open when the daemon dies mid-offer', async () => {
    const { deps } = memDeps({ token: pairedToken });
    const t = makeTransport(daemon.wsPort, deps, { autoPair: false });
    await t.connect();
    const rules: CaptureRules = await t.call('capture.getRules', {});
    expect(rules.monitoredExtensions).toContain('zip'); // seeded default (0001_initial.sql)
    const hook = new CaptureHook({
      offer: (params, opts) => t.call('capture.offer', params, opts),
      stash: { take: () => undefined, peek: () => undefined },
      getCookies: async () => [],
      getRules: () => rules,
      origin: ORIGIN,
    });
    // A throttled URL so the task the first offer creates is still active (not yet
    // complete) when the dedupe offer for the same URL follows immediately after.
    const url = `${testserver.baseUrl}/throttled/file/512K`;
    const details: OnHeadersReceivedDetails = {
      requestId: 'live-1',
      url,
      method: 'GET',
      type: 'other',
      statusCode: 200,
      tabId: 1,
      responseHeaders: [{ name: 'content-disposition', value: 'attachment; filename="live-capture.zip"' }],
    };
    const first = await hook.handle(details);
    expect(first).toEqual({ cancel: true }); // the daemon took it — Firefox never starts its own download
    const list = await t.call('download.list', { filter: { query: 'live-capture' } });
    expect(list.items.length).toBeGreaterThan(0);
    const task = list.items[0]!;
    expect(task.categoryId).toBe('programs'); // "zip" routes to the built-in Programs category
    expect(task.saveDir).toContain('Downloads/Programs');
    // Same URL again, task still active: the daemon's own dedupe (has_active_duplicate)
    // says Ignore, so the hook proceeds instead of cancelling a second time.
    const dup = await hook.handle({ ...details, requestId: 'live-2' });
    expect(dup).toEqual({});
    // Now kill the daemon mid-offer and prove fail-open holds against the REAL binary,
    // not just FakeDaemon: the hook must still resolve to {} (Firefox downloads
    // normally) well inside its own 750 ms budget.
    daemon.kill('SIGKILL');
    await waitFor(() => daemon.hasExited(), 5_000, 'veloxd to actually die');
    const started = Date.now();
    const afterDeath = await hook.handle({ ...details, requestId: 'live-3', url: `${url}?after-death=1` });
    const elapsedMs = Date.now() - started;
    expect(afterDeath).toEqual({}); // fail open — never {cancel: true} with a dead daemon
    expect(elapsedMs).toBeLessThan(900); // budget is 750ms; the hook's own timer bounds this
    t.disconnect();
  }, 20_000);
  it('event.task.progress reaches a subscribed client (the popup\'s own path)', async () => {
    if (daemon.hasExited()) {
      // The previous test kills the daemon on purpose to prove fail-open; start a fresh
      // one so this test still exercises the real event path end to end.
      daemon = await freshVeloxd();
    }
    const { deps } = memDeps();
    const t = makeTransport(daemon.wsPort, deps); // fresh daemon instance -> fresh pairing
    await t.connect();
    try {
      await t.call('session.subscribe', {
        events: ['event.task.added', 'event.task.state', 'event.task.progress'],
      });
      const progressEvents: TaskProgressEvent[] = [];
      const stateEvents: TaskStateEvent[] = [];
      t.on('event.task.progress', (p) => progressEvents.push(p as TaskProgressEvent));
      t.on('event.task.state', (p) => stateEvents.push(p as TaskStateEvent));
      const spec: DownloadSpec = { url: `${testserver.baseUrl}/throttled/file/1M`, filename: 'progress-check.bin' };
      const added = await t.call('download.add', spec);
      // /throttled defaults to 1 MiB/s, so a 1 MiB file takes ~1s — long enough that at
      // least one 4 Hz progress tick (event.task.progress's documented cap) lands before
      // it completes, exactly the path popup/store.ts consumes in the real extension.
      await waitFor(
        () => progressEvents.some((e) => e.tasks.some((row) => row.taskId === added.taskId)),
        8_000,
        'a live event.task.progress tick for our task',
      );
      expect(stateEvents.some((e) => e.taskId === added.taskId)).toBe(true);
    } finally {
      t.disconnect();
    }
  }, 15_000);
  it('fail-open also holds through the transport itself: a call against a dead socket rejects, never hangs past its deadline', async () => {
    const { deps } = memDeps();
    const t = makeTransport(daemon.wsPort, deps);
    await t.connect();
    t.disconnect(); // closes the socket without telling the daemon anything is wrong
    await expect(t.call('capture.offer', { url: 'https://example.com/x.zip', method: 'GET', tabUrl: '' }, { timeoutMs: 200 })).rejects.toBeInstanceOf(
      TransportClosedError,
    );
  });
 });
@@ -56,6 +56,13 @@ void RpcClient::stop() {
    QMetaObject::invokeMethod(conn_, "stop", Qt::QueuedConnection);
    thread_.quit();
    thread_.wait();
    // thread_.wait() does not return until thread_'s own finish() has already flushed the
    // DeferredDelete this class's own connect(&thread_, &QThread::finished, conn_,
    // &QObject::deleteLater) posted — conn_ is gone by now. Null it out so a later call
    // (stop() is a public slot; a caller stopping and then destroying the client is normal
    // use, and the destructor's own `delete conn_` for the never-started case must not
    // run a second time against memory this path already freed).
    conn_ = nullptr;
 }
 void RpcClient::call(const QString &methodName, const QJsonObject &params,
@@ -76,17 +83,44 @@ void RpcClient::onConnectionState(int state) {
 }
 void RpcClient::requestInitialList() {
-    call(QString::fromLatin1(method::kDownloadList), QJsonObject{{"limit", 1000}},
+    fetchListPage(0, {});
-         [this](const RpcReply &reply) {
+}
 // download.list.schema.json: "Filtering, sorting and paging all happen in the daemon so
 // the GUI never materializes 100k rows to show 40" — limit maxes out at 5000, so one call
 // cannot ever return everything for a table the DoD's own gate says can hold 10 000 rows.
 // A single fixed-limit call here silently truncated the table below that (caught by
 // gui/tests/dod's scroll-60fps gate refusing to run against a 1000-row table when mockd
 // seeded 10000). Page until `total` is satisfied, then reset the model exactly once.
 void RpcClient::fetchListPage(int offset, QJsonArray accumulated) {
    constexpr int kPageSize = 5000;  // download.list's own maximum
    constexpr int kMaxPages = 100;   // 500 000 rows — a safety cap, not an expected ceiling
    call(QString::fromLatin1(method::kDownloadList),
         QJsonObject{{"offset", offset}, {"limit", kPageSize}},
         [this, offset, accumulated](const RpcReply &reply) mutable {
             if (!reply.ok()) {
                 qCWarning(lcRpc, "download.list failed: %d %s", reply.error.code,
                           qUtf8Printable(reply.error.message));
                 if (!accumulated.isEmpty()) {
                     emit taskListReset(accumulated);  // show what we got rather than nothing
                 }
                 return;
             }
-             const QJsonArray items = reply.result.toObject().value("items").toArray();
+             const QJsonObject result = reply.result.toObject();
-             qCInfo(lcRpc, "initial download.list: %lld row(s)",
+             const QJsonArray page = result.value("items").toArray();
-                    static_cast<long long>(items.size()));
+             const qint64 total = static_cast<qint64>(result.value("total").toDouble());
-             emit taskListReset(items);
+             for (const QJsonValue &item : page) {
                 accumulated.append(item);
             }
             const bool morePages =
                 !page.isEmpty() && accumulated.size() < total && (offset / kPageSize) < kMaxPages;
             if (morePages) {
                 fetchListPage(offset + static_cast<int>(page.size()), accumulated);
                 return;
             }
             qCInfo(lcRpc, "initial download.list: %lld of %lld row(s)",
                    static_cast<long long>(accumulated.size()), static_cast<long long>(total));
             emit taskListReset(accumulated);
         });
 }
@@ -67,6 +67,7 @@ class RpcClient : public QObject {
  private:
    void requestInitialList();
    void fetchListPage(int offset, QJsonArray accumulated);
    QThread thread_;
    RpcConnection *conn_ = nullptr;  // owned by thread_ affinity, deleted on thread finish
@@ -154,10 +154,12 @@ void RpcConnection::dispatchFrame(const QJsonObject &frame) {
                socket_->abort();  // version mismatch or refused — bounce and retry
                return;
            }
-            sendRaw(kSubscribeId, QString::fromLatin1(method::kSessionSubscribe),
+            sendRaw(
-                    QJsonObject{{"events", QJsonArray{event::kTaskAdded, event::kTaskRemoved,
+                kSubscribeId, QString::fromLatin1(method::kSessionSubscribe),
-                                                      event::kTaskState, event::kTaskProgress,
+                QJsonObject{
-                                                      event::kSpeedGlobal, event::kNotify}}});
+                    {"events", QJsonArray{event::kTaskAdded, event::kTaskRemoved, event::kTaskState,
                                          event::kTaskProgress, event::kSpeedGlobal, event::kNotify,
                                          event::kSettingsChanged, event::kGrabberProgress}}});
            return;
        }
        if (id == kSubscribeId) {