sami/vdm
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

mockd: add --tasks N — a plausible large synthetic table for GUI's DoD

Browse Source 
GUI's M1 definition of done is "10 000 synthetic rows scroll at 60 fps with
flat memory over 10 minutes (mockd --tasks 10000)". This flag was missing from
the four unhappy-path flags that did land; the brief's own flag list omitted
it, which is corrected here too.

--tasks seeds a plausible population rather than N copies of one row: varied
state, size (log-uniform 50 KB - 20 GB), category, queue position and
description, drawn from the same category.list / queue.list fixtures the rest
of mockd already serves so a synthetic task can never name a category or
queue those methods don't also return. State distribution is roughly
55% complete / 8% failed / 4% cancelled / 6% paused / 2% retry_wait / 25%
queued, using the new TaskErrorCode taxonomy for failures.

"Progress advances across the whole set, not a handful of live rows" ruled
out the obvious cheap answer. A bounded, rotating pool of concurrently-active
downloads (--active-cap, default 24) is fed continuously from each queue's
FIFO — with the rest of that queue's queuePosition renumbered on every
promotion, as a real scheduler would — and a small fraction of active tasks
hit a transient failure and cycle through retry_wait before rejoining, so the
pool keeps rotating through new rows for the whole run instead of draining
once. Verified over a 10000-task, 60-second run: 61.5 MB RSS flat, and the
active pool's membership meaningfully different after 60s.

tick() only ever walks the active pool plus due retry-wait entries, never the
full task list, so its cost stays flat regardless of --tasks. A manual
download.add is still admitted immediately regardless of --active-cap — a
human driving the GUI by hand must never wait behind synthetic load.

Fixed a latent double-push while building this: any task 'connecting' at the
top of a tick was pushed to the progress batch once for the transition and
again at the loop's unconditional final push, inflating event.task.progress
payloads with a duplicate entry for that taskId. It predates this change (the
original tick() had the same shape) but only became visible once several
tasks are legitimately 'connecting' in the same tick, which --active-cap's
continuous promotion now does routinely.

--seed makes a run reproducible, which matters when a GUI bug only shows up
at a particular row.

Co-Authored-By: Claude Sonnet 5 <[email protected]>
Claude-Session: https://claude.ai/code/session_012fgjnqFCS5h5L7gZTZo3rV
This commit is contained in:
sami
2026-09-10 00:01:44 +04:00
co-authored by Claude Sonnet 5
parent fdacf732fa
commit f60070c420
5 changed files with 551 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/mockd/src/state.ts
+162 -14
View File
@@ -7,17 +7,39 @@
* answered straight from a golden file.
*
* This is deliberately not a download engine. Bytes advance on a clock, not from a socket.
*
* At scale (`--tasks 10000`) two things matter that don't at a handful of tasks:
*
* 1. `tick()` must not become O(total tasks) every 250 ms. It only ever touches the
* bounded *active* set (`activeCap`, default 24) plus whatever retry-wait entries just
* came due — never the thousands of tasks sitting in a terminal or queued state.
* 2. Progress has to reach a broad slice of the table over the life of a run, not the same
* couple of rows forever. As an active task finishes, the next one is promoted from its
* queue's FIFO (with the rest of that queue's `queuePosition` renumbered, as a real
* scheduler would), and a slice of "finished" active tasks fail instead and go through
* `retry_wait` before being promoted again — so the active pool keeps rotating through
* new rows for the whole run rather than draining once and going static.
*/
import { randomUUID } from 'node:crypto';
import type { TaskState, TaskSummary } from '../../../extension/src/shared/protocol/types.js';
import { randomTaskErrorCode, taskErrorFor, type SynthResult } from './synth.js';
export interface MockOptions {
readonly progressHz: number;
readonly speedBps: number;
/** Ceiling on concurrently-"downloading" tasks. Bulk-seeded tasks rotate through this;
* a manual download.add is admitted immediately regardless (see add()) — a human
* driving the GUI by hand should never be told to wait behind synthetic load. */
readonly activeCap: number;
}
const ACTIVE: readonly TaskState[] = ['connecting', 'downloading'];
// Chance per active task per tick that a healthy transfer hits a transient failure instead
// of completing normally — keeps the active pool cycling through retry_wait for the whole
// life of a long run rather than just draining the initial 'queued' population once.
const TRANSIENT_FAILURE_CHANCE = 0.0015;
const RETRY_DELAY_MS: readonly [number, number] = [5_000, 45_000];
export class MockState {
readonly tasks = new Map<string, TaskSummary>();
@@ -25,15 +47,47 @@ export class MockState {
settings: Record<string, unknown> = {};
private readonly opts: MockOptions;
/** taskIds currently connecting/downloading. Only this set is walked every tick. */
private readonly active = new Set<string>();
/** Per-queue FIFO of taskIds waiting to be promoted, front = next admitted. */
private readonly pendingByQueue = new Map<string, string[]>();
/** taskIds with no queueId, or whose queue is unknown to this process (e.g. added via
* download.add with startMode 'queue' and a queueId mockd has no fixture-derived queue
* for) — a fallback FIFO so nothing is silently unpromotable. */
private readonly pendingUnqueued: string[] = [];
private readonly retryWaiting: Array<{ taskId: string; dueAt: number }> = [];
private readonly speedFactor = new Map<string, number>();
constructor(seed: readonly TaskSummary[], opts: MockOptions) {
this.opts = opts;
for (const t of seed) this.tasks.set(t.taskId, { ...t });
for (const t of seed) this.registerNew({ ...t });
}
/** Bulk-seed a large synthetic population without re-deriving the pool bookkeeping the
* generator already computed — the one place a --tasks 10000 startup does real O(n)
* work, and it happens once, not per tick. */
seedSynthetic(result: SynthResult): void {
for (const task of result.tasks) this.tasks.set(task.taskId, task);
for (const taskId of result.initiallyActive) this.active.add(taskId);
for (const [queueId, ids] of result.queuedOrder) this.pendingByQueue.set(queueId, [...ids]);
for (const [taskId, factor] of result.speedFactor) this.speedFactor.set(taskId, factor);
}
list(): TaskSummary[] {
return [...this.tasks.values()];
}
private registerNew(task: TaskSummary): void {
this.tasks.set(task.taskId, task);
if (ACTIVE.includes(task.state)) this.active.add(task.taskId);
}
/**
* A manual add (download.add, capture.offer) is admitted immediately regardless of the
* active pool — a human testing the GUI by hand must never be told to wait behind
* synthetic --tasks load. It participates in tick()'s active set from the moment it is
* created, same as any other active task.
*/
add(partial: Partial<TaskSummary> & { url: string }): TaskSummary {
const now = new Date().toISOString();
const task: TaskSummary = {
@@ -58,7 +112,7 @@ export class MockState {
completedAt: null,
error: null,
};
this.tasks.set(task.taskId, task);
this.registerNew(task);
return task;
}
@@ -66,6 +120,7 @@ export class MockState {
const task = this.tasks.get(taskId);
if (!task) return null;
const changed = task.state !== state;
const wasActive = ACTIVE.includes(task.state);
task.state = state;
task.speedBps = ACTIVE.includes(state) ? this.opts.speedBps : 0;
if (!ACTIVE.includes(state)) task.etaSeconds = null;
@@ -73,29 +128,116 @@ export class MockState {
task.downloadedBytes = task.sizeBytes ?? task.downloadedBytes;
task.completedAt = new Date().toISOString();
}
const isActive = ACTIVE.includes(state);
if (isActive && !wasActive) this.active.add(taskId);
else if (!isActive && wasActive) this.active.delete(taskId);
return { changed, task };
}
remove(taskId: string): boolean {
this.active.delete(taskId);
return this.tasks.delete(taskId);
}
private popNextQueued(): string | undefined {
for (const [, ids] of this.pendingByQueue) {
const taskId = ids.shift();
if (taskId !== undefined) {
for (let i = 0; i < ids.length; i += 1) {
const t = this.tasks.get(ids[i]!);
if (t) t.queuePosition = i + 1;
}
return taskId;
}
}
return this.pendingUnqueued.shift();
}
private admit(taskId: string): void {
const task = this.tasks.get(taskId);
if (!task) return; // removed while queued — nothing to admit
task.state = 'connecting';
task.speedBps = 0;
task.queuePosition = null;
task.lastTryAt = new Date().toISOString();
this.active.add(taskId);
}
private fillActiveFromQueues(): void {
while (this.active.size < this.opts.activeCap) {
const taskId = this.popNextQueued();
if (taskId === undefined) return;
this.admit(taskId);
}
}
/**
* Advance one progress tick. Returns the tasks that moved and any that just finished,
* so the caller can emit event.task.progress and event.task.state from one place.
* Advance one progress tick. Returns the tasks that moved and any that just finished
* (completed, or newly failed), so the caller can emit event.task.progress and
* event.task.state from one place.
*/
tick(): { moved: TaskSummary[]; completed: TaskSummary[] } {
tick(): { moved: TaskSummary[]; completed: TaskSummary[]; failed: TaskSummary[] } {
const moved: TaskSummary[] = [];
const completed: TaskSummary[] = [];
const perTick = Math.floor(this.opts.speedBps / this.opts.progressHz);
const failed: TaskSummary[] = [];
const now = Date.now();
for (const task of this.tasks.values()) {
// Retry-wait entries whose backoff elapsed rejoin the active pool directly (they have
// already waited their turn once; they don't go back through a queue's FIFO).
while (this.retryWaiting.length > 0 && this.retryWaiting[0]!.dueAt <= now) {
const { taskId } = this.retryWaiting.shift()!;
if (this.active.size < this.opts.activeCap) {
this.admit(taskId);
const task = this.tasks.get(taskId);
// Don't push to `moved` here: admit() leaves the task 'connecting', and the main
// loop below visits every active taskId (this one included, since admit() just
// added it) and pushes exactly once when it flips to 'downloading'. Pushing here
// too would duplicate this task in the same tick's event.task.progress batch.
if (task) task.error = null;
} else {
// Pool is full: park it at the front of the fallback FIFO rather than dropping it.
this.pendingUnqueued.unshift(taskId);
}
}
this.fillActiveFromQueues();
const perTickBase = Math.floor(this.opts.speedBps / this.opts.progressHz);
for (const taskId of [...this.active]) {
const task = this.tasks.get(taskId);
if (!task) {
this.active.delete(taskId); // stale — removed mid-transfer
continue;
}
if (task.state === 'connecting') {
// One settle tick before bytes start moving — also fixes a latent double-push:
// falling through to the unconditional moved.push() below would otherwise queue
// this task twice in the same event.task.progress batch (once here, once there).
task.state = 'downloading';
moved.push(task);
continue;
}
if (task.state !== 'downloading') continue;
if (task.downloadedBytes > 0 && Math.random() < TRANSIENT_FAILURE_CHANCE) {
const retryable = Math.random() < 0.7;
const code = randomTaskErrorCode(Math.random, retryable);
task.error = taskErrorFor(Math.random, code, retryable);
task.state = retryable ? 'retry_wait' : 'failed';
task.speedBps = 0;
task.etaSeconds = null;
this.active.delete(taskId);
failed.push(task);
moved.push(task);
if (retryable) {
const [lo, hi] = RETRY_DELAY_MS;
this.retryWaiting.push({ taskId, dueAt: now + lo + Math.random() * (hi - lo) });
}
continue;
}
const factor = this.speedFactor.get(taskId) ?? 1;
const perTick = Math.max(1, Math.round(perTickBase * factor));
task.speedBps = jitter(perTick * this.opts.progressHz);
task.downloadedBytes += perTick;
const size = task.sizeBytes;
@@ -105,27 +247,33 @@ export class MockState {
task.speedBps = 0;
task.etaSeconds = null;
task.completedAt = new Date().toISOString();
this.active.delete(taskId);
completed.push(task);
} else if (size !== null && size !== undefined && task.speedBps > 0) {
task.etaSeconds = Math.ceil((size - task.downloadedBytes) / task.speedBps);
}
moved.push(task);
}
return { moved, completed };
// Backfill whatever just finished, so the pool is at capacity again for the next tick
// rather than idling until someone happens to call fillActiveFromQueues().
this.fillActiveFromQueues();
return { moved, completed, failed };
}
globalSpeed(): { downBps: number; activeCount: number; queuedCount: number } {
let downBps = 0;
let activeCount = 0;
let queuedCount = 0;
for (const t of this.tasks.values()) {
if (ACTIVE.includes(t.state)) {
downBps += t.speedBps;
for (const taskId of this.active) {
const task = this.tasks.get(taskId);
if (task) {
downBps += task.speedBps;
activeCount += 1;
} else if (t.state === 'queued') {
queuedCount += 1;
}
}
let queuedCount = this.pendingUnqueued.length + this.retryWaiting.length;
for (const [, ids] of this.pendingByQueue) queuedCount += ids.length;
return { downBps, activeCount, queuedCount };
}
}