Skip to content

Commit 8609aff

Browse files
fix(resume): /live shows full batch on resume, not just post-resume tail
Root cause: the continuous loop's resume-intent branch called pipeline.runAudit(false, ...), which wipes the in-memory `results` array (results.length = 0) and sets totalNodes to the *remainder* only. /live's snapshot mirrors getResults(), so an already-tested batch vanished from the public live table on resume — logs kept streaming but the node table went empty. Admin (DB-backed) stayed complete, so the two surfaces desynced. Comprehensive fix (same-process AND cross-process/server-restart): - continuous.js: thread resume=true through _runOnePass into both runAudit and runSubPlanTest, so the pipeline preserves the results array and computes totalNodes = already-tested + remaining. - continuous.js: cross-process seed — after a restart the array is empty, so re-hydrate getResults() from persisted batch_results via new pipeline.seedResults() + _batchRowToResult mapper. Guarded to the real-runner path (test/injected runner persists nothing). - batch:start now carries snapshotSize = full batch (tested+remaining) and resumed:true. - server.js: sanitizeForPublic forwards resumed:true to /live. - live.html: on batch:start with resumed, call loadCurrentBatch() once so an already-open tab repaints the full batch (the bulk of nodes aren't re-emitted on resume). - pipeline.js: gate runSubPlanTest's runSpentUdvpn reset on !resume, matching runAudit — preserves cumulative run spend across resume (also fixes a latent subscription re-pay on resume). Tests: new test/resume-results-seed.test.js (13) covers seedResults append/dedup/guards + DB round-trip into getResults(); wired into test:integration.
1 parent a40d7ae commit 8609aff

6 files changed

Lines changed: 228 additions & 6 deletions

File tree

audit/continuous.js

Lines changed: 61 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -232,6 +232,28 @@ function _sanitizeBatchNodeResult(result, batchId) {
232232
};
233233
}
234234

235+
// Map a persisted batch_results row (snake_case) back into the camelCase result
236+
// shape the in-memory results array and /live snapshot expect. Used only by the
237+
// cross-process resume seed (see the resume-intent branch in _runLoop).
238+
function _batchRowToResult(row) {
239+
return {
240+
address: row.node_address,
241+
moniker: row.moniker || '',
242+
country: row.country || '',
243+
countryCode: row.country_code || '',
244+
city: row.city || '',
245+
type: row.type || null,
246+
actualMbps: row.actual_mbps,
247+
baselineAtTest: row.baseline_mbps,
248+
peers: row.peers,
249+
maxPeers: row.max_peers,
250+
error: row.error || null,
251+
errorCode: row.error_code || null,
252+
skipped: row.error_code === 'TEST_RUN_SKIP',
253+
testedAt: row.tested_at,
254+
};
255+
}
256+
235257
// ─── Core Loop ────────────────────────────────────────────────────────────────
236258

237259
/**
@@ -244,7 +266,7 @@ function _sanitizeBatchNodeResult(result, batchId) {
244266
* @param {Array|null} frozenNodes - Pre-resolved node snapshot; p2p runAudit uses it instead of re-querying
245267
* @returns {Promise<{ passed: number, failed: number }>}
246268
*/
247-
async function _runOnePass(loopState, batchId, frozenNodes = null) {
269+
async function _runOnePass(loopState, batchId, frozenNodes = null, resume = false) {
248270
// Build a broadcast intercept that captures 'result' events for batch tracking.
249271
// All other broadcast types (log, state) are silently dropped (noop) —
250272
// the continuous loop emits its own high-level events instead.
@@ -307,6 +329,10 @@ async function _runOnePass(loopState, batchId, frozenNodes = null) {
307329
_ctrl.subscriptionGranter,
308330
st,
309331
batchBroadcast,
332+
// On resume, opts.resume=true keeps the sub-plan run from resetting
333+
// counters/startedAt and makes its totalNodes math include the already-
334+
// tested nodes instead of just the remainder.
335+
{ resume },
310336
)
311337
: (st) => {
312338
// The continuous loop persists per-node to batch_results separately;
@@ -316,7 +342,11 @@ async function _runOnePass(loopState, batchId, frozenNodes = null) {
316342
// no restore needed (nothing else reads these fields off loopState).
317343
st.activeDbRunId = null;
318344
st.activeRunDir = null;
319-
return pipeline.runAudit(false, st, batchBroadcast, frozenNodes, {
345+
// resume=true on the resume-intent pass: pipeline preserves the
346+
// in-memory results array (no wipe) and sets totalNodes =
347+
// results.length + remaining, so getResults() — which /live's snapshot
348+
// mirrors — reflects the FULL batch, not just the post-resume tail.
349+
return pipeline.runAudit(resume, st, batchBroadcast, frozenNodes, {
320350
testRun: !!_ctrl.testRun,
321351
pricingMode: _ctrl.pricingMode || null,
322352
});
@@ -414,17 +444,44 @@ async function _runLoop() {
414444
const viableNodes = allNodes.filter(n => n?.address && !testedSet.has(n.address));
415445
frozenNodes = viableNodes;
416446

447+
// Cross-process resume: after a server restart the in-memory results
448+
// array is empty, but the reused batch already has its tested nodes in
449+
// batch_results. Re-seed them so getResults() — the sole source for
450+
// /live's snapshot — and the pipeline's resume totalNodes math reflect
451+
// the FULL batch, not just the remainder we're about to test. No-op in
452+
// the same-process case (the array still holds them) and under an
453+
// injected runner (test path persists nothing).
454+
if (!_runnerFn) {
455+
try {
456+
const pipeline = await import('./pipeline.js');
457+
if (pipeline.getResults().length === 0 && currentBatchId > 0) {
458+
const { getBatchResults } = await import('../core/db.js');
459+
const { results: priorRows } = getBatchResults(currentBatchId, { limit: 100000 }, 'real');
460+
const restored = pipeline.seedResults((priorRows || []).map(_batchRowToResult));
461+
if (restored > 0) {
462+
_emitScoped('log', { msg: `↩ Resume: restored ${restored} already-tested node(s) from batch ${currentBatchId}` });
463+
}
464+
}
465+
} catch (err) {
466+
console.error(`[continuous] resume seed failed (batch ${currentBatchId}): ${err.message}`);
467+
}
468+
}
469+
417470
_emitScoped('batch:start', {
418471
batchId: currentBatchId,
419-
snapshotSize: viableNodes.length,
472+
// Full batch size (tested + remaining), not just the remainder — so
473+
// /live's "X / Y" denominator shows the whole batch on resume.
474+
snapshotSize: allNodes.length || viableNodes.length,
420475
mode: _ctrl.mode,
421476
startedAt: iterStart,
422477
iteration: _ctrl.iteration,
423478
resumed: true,
424479
});
425480

426481
try {
427-
({ passed, failed } = await _runOnePass(loopState, currentBatchId, frozenNodes));
482+
// resume=true: preserve the in-memory results (seeded above) and make
483+
// the pipeline's totalNodes include the already-tested nodes.
484+
({ passed, failed } = await _runOnePass(loopState, currentBatchId, frozenNodes, true));
428485
} catch (err) {
429486
iterErr = err;
430487
_ctrl.lastError = err.message || String(err);

audit/pipeline.js

Lines changed: 22 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -163,6 +163,27 @@ if (existsSync(RESULTS_FILE)) {
163163

164164
export function getResults() { return results; }
165165

166+
/**
167+
* Seed the in-memory results array with already-tested rows (deduped by address,
168+
* same as upsertResult). Used by the continuous resume path after a server
169+
* restart, when the reused batch already has tested nodes persisted in the DB
170+
* but the in-memory `results` array is empty — without this, getResults() (the
171+
* sole source for /live's snapshot and the resume `totalNodes` math) would only
172+
* ever reflect the post-resume remainder. No-op for addresses already present,
173+
* so it is safe to call on a partially-populated array. Returns count appended.
174+
*/
175+
export function seedResults(rows) {
176+
if (!Array.isArray(rows)) return 0;
177+
let added = 0;
178+
for (const r of rows) {
179+
if (!r || !r.address) continue;
180+
const idx = results.findIndex(x => x.address === r.address);
181+
if (idx !== -1) { results[idx] = r; }
182+
else { results.push(r); added++; }
183+
}
184+
return added;
185+
}
186+
166187
// ─── Crash-Safe Results Persistence ────────────────────────────────────────
167188
// Write to temp file then rename (atomic on most filesystems).
168189
// Also continuously save to the active run directory so a kill never loses data.
@@ -1846,7 +1867,7 @@ export async function runSubPlanTest(planId, subscriptionId, granterAddr, state,
18461867
state.balanceUdvpn = parseInt(balRes?.amount || '0', 10);
18471868
state.balance = `${(state.balanceUdvpn / 1_000_000).toFixed(4)} P2P`;
18481869
state.spentUdvpn = 0;
1849-
state.runSpentUdvpn = 0; // fresh pass — cumulative run spend
1870+
if (!resume) state.runSpentUdvpn = 0; // preserve cumulative run spend across resume (parity with runAudit:641)
18501871
broadcast('state', { state });
18511872

18521873
// Pre-broadcast fee grant verification — abort with structured error if missing/expired.

live.html

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2158,6 +2158,13 @@ <h1 class="live-page-title">
21582158
setStatus('running');
21592159
showPaused(false);
21602160
cbApplyBatchStart(d);
2161+
// RESUME: the bulk of this batch's nodes were tested before this SSE
2162+
// connection's event window and are NOT re-emitted — only the
2163+
// remaining nodes stream. An already-open /live tab would otherwise
2164+
// show just the post-resume tail. Pull the full current batch once so
2165+
// the table repaints completely (getResults() is the complete batch
2166+
// server-side after the resume seed/preserve fix).
2167+
if (d.resumed) { loadCurrentBatch(); }
21612168
appendLog({
21622169
tag: 'ROUND', cat: 'sys', status: 'ok',
21632170
msg: `Round ${d.iteration != null ? d.iteration : (d.batchId || '?')} — testing ${d.snapshotSize || d.snapshot_size || d.total || '?'} nodes`,

package.json

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -40,7 +40,7 @@
4040
"audit": "node server.js",
4141
"test": "node test/smoke.test.js && node test/db.smoke.test.js && node test/continuous.smoke.test.js && node test/security.test.js && node test/failure-log-ux.test.js && node test/public-no-action-buttons.test.js && node test/onchain-decode.test.js",
4242
"test:ui": "node test/ui.smoke.test.js",
43-
"test:integration": "node test/continuous.audit-integration.test.js && node test/continuous.live-db-write.test.js && node test/node-detail.smoke.test.js",
43+
"test:integration": "node test/continuous.audit-integration.test.js && node test/continuous.live-db-write.test.js && node test/node-detail.smoke.test.js && node test/resume-results-seed.test.js",
4444
"verify:prod-db": "node scripts/verify-prod-db.mjs",
4545
"cleanup": "node scripts/cleanup.mjs"
4646
},

server.js

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1940,6 +1940,9 @@ function sanitizeForPublic(evt) {
19401940
if (evt.batchId != null) safe.batchId = evt.batchId;
19411941
if (evt.snapshotSize != null) safe.snapshotSize = evt.snapshotSize;
19421942
if (evt.startedAt != null) safe.startedAt = evt.startedAt;
1943+
// batch:start carries resumed:true on the resume-intent pass. /live uses it to
1944+
// re-hydrate the full batch (already-tested nodes aren't re-emitted on resume).
1945+
if (evt.resumed === true) safe.resumed = true;
19431946
if (evt.gapMs != null) safe.gapMs = evt.gapMs;
19441947
if (evt.nextBatchAt != null) safe.nextBatchAt = evt.nextBatchAt;
19451948
// batch:node:result public-safe fields.

test/resume-results-seed.test.js

Lines changed: 134 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,134 @@
1+
/**
2+
* Resume — In-Memory Results Restoration (root-cause regression test)
3+
*
4+
* Bug: on resume, the continuous loop called pipeline.runAudit(resume=false),
5+
* which wipes the in-memory `results` array (pipeline.js: `results.length = 0`)
6+
* and sets totalNodes to the *remainder* only. Since /live's snapshot mirrors
7+
* getResults(), an already-tested batch vanished from /live on resume — logs
8+
* kept streaming but the node table went empty. Admin (DB-backed) stayed
9+
* complete, so the two desynced.
10+
*
11+
* Fix: continuous resume now (a) threads resume=true so the pipeline preserves
12+
* the results array + computes totalNodes = already-tested + remaining, and
13+
* (b) for the cross-process case (server restart → empty array) re-seeds the
14+
* in-memory results from the persisted batch_results via pipeline.seedResults().
15+
*
16+
* This test covers the seed primitive + dedup semantics directly (no chain/
17+
* wallet needed) and proves a DB-backed restore round-trips into getResults().
18+
*
19+
* Run: node test/resume-results-seed.test.js
20+
*/
21+
22+
import Database from 'better-sqlite3';
23+
24+
const out = { pass: 0, fail: 0, errors: [] };
25+
function ok(cond, name) {
26+
if (cond) { out.pass++; console.log(` PASS ${name}`); }
27+
else { out.fail++; out.errors.push(name); console.log(` FAIL ${name}`); }
28+
}
29+
30+
async function main() {
31+
process.env.NODE_ENV = 'test';
32+
33+
console.log('\nResume — In-Memory Results Restoration\n');
34+
35+
// ─── 1. pipeline.seedResults — append + dedup by address ──────────────
36+
console.log('[1] pipeline.seedResults appends and dedups by address');
37+
const pipeline = await import('../audit/pipeline.js');
38+
// Start from a clean in-memory results array.
39+
pipeline.getResults().length = 0;
40+
41+
const added1 = pipeline.seedResults([
42+
{ address: 'sentnode1aaa', actualMbps: 12.5, errorCode: null },
43+
{ address: 'sentnode1bbb', actualMbps: null, errorCode: 'HANDSHAKE_FAIL' },
44+
]);
45+
ok(added1 === 2, `first seed appends 2 (got ${added1})`);
46+
ok(pipeline.getResults().length === 2, `getResults() has 2 rows (got ${pipeline.getResults().length})`);
47+
48+
// Re-seeding the same address must NOT duplicate — it replaces in place.
49+
const added2 = pipeline.seedResults([
50+
{ address: 'sentnode1aaa', actualMbps: 99.9, errorCode: null },
51+
{ address: 'sentnode1ccc', actualMbps: 5.0, errorCode: null },
52+
]);
53+
ok(added2 === 1, `second seed appends only the new address (got ${added2})`);
54+
ok(pipeline.getResults().length === 3, `getResults() has 3 distinct rows (got ${pipeline.getResults().length})`);
55+
const aaa = pipeline.getResults().find(r => r.address === 'sentnode1aaa');
56+
ok(aaa && aaa.actualMbps === 99.9, `re-seeded address is replaced, not duplicated (mbps=${aaa?.actualMbps})`);
57+
58+
// Guards: non-array → 0, rows without address skipped.
59+
ok(pipeline.seedResults(null) === 0, 'seedResults(null) → 0');
60+
ok(pipeline.seedResults([{ moniker: 'no-addr' }]) === 0, 'rows without address are skipped');
61+
ok(pipeline.getResults().length === 3, 'guarded calls do not grow the array');
62+
63+
// ─── 2. DB round-trip: batch_results → getBatchResults → seedResults ──
64+
console.log('[2] persisted batch_results restore into getResults()');
65+
const { useDb, getDb, insertBatch, insertBatchResult, getBatchResults } =
66+
await import('../core/db.js');
67+
useDb(getDb(':memory:'));
68+
69+
const batchId = insertBatch({
70+
started_at: Date.now(),
71+
snapshot_size: 3,
72+
mode: 'p2p',
73+
snapshot_addresses: ['sentnode1xxx', 'sentnode1yyy', 'sentnode1zzz'],
74+
}, 'real');
75+
ok(batchId > 0, `insertBatch returned id (got ${batchId})`);
76+
77+
// Two nodes already tested before the (simulated) restart.
78+
insertBatchResult(batchId, {
79+
address: 'sentnode1xxx', type: 'wireguard', moniker: 'Alpha',
80+
country: 'US', countryCode: 'US', city: 'NYC',
81+
actualMbps: 22.4, peers: 3, maxPeers: 10,
82+
error: null, errorCode: null, baselineMbps: 18.0, testedAt: Date.now(),
83+
}, 'real');
84+
insertBatchResult(batchId, {
85+
address: 'sentnode1yyy', type: 'v2ray', moniker: 'Bravo',
86+
country: 'DE', countryCode: 'DE', city: 'Berlin',
87+
actualMbps: null, peers: null, maxPeers: null,
88+
error: 'tunnel timeout', errorCode: 'TUNNEL_TIMEOUT', baselineMbps: 18.0, testedAt: Date.now(),
89+
}, 'real');
90+
91+
const { results: rows } = getBatchResults(batchId, { limit: 1000 }, 'real');
92+
ok(rows.length === 2, `getBatchResults returns the 2 persisted rows (got ${rows.length})`);
93+
94+
// Map snake_case DB rows → result shape (mirrors continuous._batchRowToResult)
95+
const mapped = rows.map(row => ({
96+
address: row.node_address,
97+
moniker: row.moniker || '',
98+
country: row.country || '',
99+
countryCode: row.country_code || '',
100+
city: row.city || '',
101+
type: row.type || null,
102+
actualMbps: row.actual_mbps,
103+
baselineAtTest: row.baseline_mbps,
104+
peers: row.peers,
105+
maxPeers: row.max_peers,
106+
error: row.error || null,
107+
errorCode: row.error_code || null,
108+
skipped: row.error_code === 'TEST_RUN_SKIP',
109+
testedAt: row.tested_at,
110+
}));
111+
112+
// Fresh in-memory array (simulates the empty array after a server restart).
113+
pipeline.getResults().length = 0;
114+
const restored = pipeline.seedResults(mapped);
115+
ok(restored === 2, `restored 2 already-tested nodes into getResults() (got ${restored})`);
116+
117+
const res = pipeline.getResults();
118+
const xxx = res.find(r => r.address === 'sentnode1xxx');
119+
const yyy = res.find(r => r.address === 'sentnode1yyy');
120+
ok(xxx && xxx.actualMbps === 22.4 && xxx.type === 'wireguard',
121+
'restored passing node carries mbps + transport');
122+
ok(yyy && yyy.errorCode === 'TUNNEL_TIMEOUT' && yyy.actualMbps == null,
123+
'restored failed node carries error code + null mbps');
124+
125+
// Cleanup so we don't leak rows into any later in-process consumer.
126+
pipeline.getResults().length = 0;
127+
128+
console.log(`\n${'='.repeat(60)}\nRESULTS: ${out.pass} passed, ${out.fail} failed (${out.pass + out.fail} total)`);
129+
if (out.errors.length) for (const e of out.errors) console.log(` FAIL: ${e}`);
130+
console.log('='.repeat(60));
131+
process.exit(out.fail ? 1 : 0);
132+
}
133+
134+
main().catch(e => { console.error('FATAL:', e.stack || e); process.exit(1); });

0 commit comments

Comments
 (0)