test(stress): fix throughput-floor gate, throttle scan contamination, harden metrics

Applies review findings from PR #617:
- largeCatchup: the grace-adjusted deadline let a converged run post a
  measured rate below ENFORCED_FLOOR_MBPS and still pass, since the gate
  checked convergedAt !== null rather than the achieved rate itself. Gate on
  catchupMBps >= ENFORCED_FLOOR_MBPS directly, which also covers an outright
  timeout without a separate check.
- largeCatchup: the exact_count convergence poll forces a full value scan of
  B's table every 5s, and that scan grows with B's row count and competes
  with the replication replay it's trying to measure. Widen the poll interval
  (HARPER_STRESS_LARGE_CATCHUP_POLL_SECS, default 15s) to reduce how often the
  benchmark scans itself.
- stressShared: writeStressMetrics's fixed path can be read as fresh data
  after a crash that happens before it's called, on a self-hosted runner that
  reuses its workspace across runs. Add clearStressMetrics(), called at suite
  start, so an early failure uploads nothing rather than a stale prior run.
- stressShared: sampleHostCounters().window() pushed its on-demand reading
  into the same samples array the timer fills on a fixed cadence; two entries
  landing milliseconds apart produce a manufactured peak rate when
  summariseHostSamples() walks them pairwise. Stop pushing from window().

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Author: kriszyp

integrationTests/stress/largeCatchup.test.mjs

@@ -56,6 +56,7 @@ import {
 	summariseHostSamples,
 	formatHostCounters,
 	writeStressMetrics,
+	clearStressMetrics,
 	writeJobSummary,
 	fabricRocksConfig,
 	mb,
@@ -119,6 +120,13 @@ if (!stressEnabled()) {
 	// worker is blocked inside a RocksDB write stall — see harper-pro#603), so only a genuine
 	// wedge trips it.
 	const STALL_SECS = Number(process.env.HARPER_STRESS_LARGE_STALL_SECS ?? 300);
+	// The convergence poll's exact_count forces a full value scan of B's table (see the
+	// comment at the poll site), and that scan grows with B's row count and competes with
+	// replication replay for the same RocksDB I/O/CPU it's trying to measure — a 5s cadence
+	// means dozens of full-table scans self-contaminate the very throughput being gated.
+	// Widening the interval cuts scan overhead roughly proportionally; STALL_SECS (300s) and
+	// the metrics/log cadence stay coarse-poll-friendly at this interval.
+	const CATCHUP_POLL_SECS = Number(process.env.HARPER_STRESS_LARGE_CATCHUP_POLL_SECS ?? 15);
 	// Generous write budget: 5 min per GB plus 10 min fixed overhead.
 	const WRITE_BUDGET_SECS = TARGET_GB * 300 + 600;
 	const SUITE_TIMEOUT_MS = (WRITE_BUDGET_SECS + CATCHUP_BUDGET_SECS + 600) * 1000;
@@ -130,6 +138,7 @@ if (!stressEnabled()) {
 
 	suite(`Large catch-up — ${TARGET_GB} GB`, { timeout: SUITE_TIMEOUT_MS }, (ctx) => {
 		before(async () => {
+			clearStressMetrics('large-catchup');
 			const rocks = fabricRocksConfig();
 			// Catch-up throughput is gated by the WriteBufferManager budget: once B's memtables
 			// exhaust it, RocksDB stalls writers and replay collapses (see harper-pro#603). Log the
@@ -357,7 +366,7 @@ if (!stressEnabled()) {
 						`[large-catchup] catchup poll: B=${lastCount}/${targetCount} (${remaining}s remaining, ` +
 							`stalled ${stalledFor.toFixed(0)}s) ${formatHostCounters(hostSampler.window())}`
 					);
-					await delay(5_000);
+					await delay(CATCHUP_POLL_SECS * 1000);
 				}
 
 				const aSummary = summariseSamples(aSampler.stop());
@@ -444,13 +453,19 @@ if (!stressEnabled()) {
 				const uncaughtRe = /\[error\]: uncaughtException/g;
 				const [logA, logB] = await Promise.all([readLog(A), readLog(B)]);
 
-				// Missing the deadline means the run averaged below CATCHUP_FLOOR_MBPS, because the
-				// deadline is derived from that floor — so this reads as a throughput regression, not
-				// as "the clock ran out". Ordinary slow-mode nights (~5 MB/s at 10 GB) stay green.
+				// Gate on the measured rate itself, not on "did it converge before the deadline".
+				// The deadline is grace-padded (CATCHUP_GRACE_SECS added on top of the floor-implied
+				// time) so B has room to reconnect before the first batch lands, but a run that
+				// consumes that whole padded window to converge posts a catchupMBps below
+				// ENFORCED_FLOOR_MBPS despite convergedAt being non-null — convergedAt!==null alone
+				// let that pass. Checking the rate directly closes that gap and still fails an
+				// outright timeout for free (partial progress over the full budget is always < the
+				// rate needed to finish it).
 				ok(
-					convergedAt !== null,
+					catchupMBps >= ENFORCED_FLOOR_MBPS,
 					`B averaged ${catchupMBps.toFixed(1)} MB/s, under the ${ENFORCED_FLOOR_MBPS.toFixed(1)} MB/s catch-up floor ` +
-						`(${appliedRecords}/${targetCount - seedRecords.length} records in ${CATCHUP_BUDGET_SECS}s) — ` +
+						`(${appliedRecords}/${targetCount - seedRecords.length} records in ${catchupSecs.toFixed(1)}s` +
+						`${convergedAt ? '' : ', TIMED OUT'}) — ` +
 						`slow replay, not a wedge; a wedge would have tripped the ${STALL_SECS}s no-progress guard. ` +
 						`Host during catch-up: ${formatHostCounters(hostSummary)}`
 				);

integrationTests/stress/stressShared.mjs

@@ -9,7 +9,7 @@
 
 import { equal } from 'node:assert';
 import { readFile } from 'node:fs/promises';
-import { appendFileSync, mkdirSync, readFileSync, writeFileSync } from 'node:fs';
+import { appendFileSync, mkdirSync, readFileSync, rmSync, writeFileSync } from 'node:fs';
 import { join } from 'node:path';
 import { setTimeout as delay } from 'node:timers/promises';
 
@@ -145,6 +145,13 @@ export function hostCounterDelta(before, after) {
  * Harper node — it reads /proc directly — so it keeps running across node
  * restarts. `window()` returns the rates since the previous window() call (or
  * since start), which is what a poll loop wants to print alongside its progress.
+ *
+ * `window()` reads its own on-demand sample rather than pushing into `samples` —
+ * `samples` is the timer's uniformly-spaced series that summariseHostSamples()
+ * walks pairwise to find peaks; interleaving an on-demand reading (called from a
+ * poll loop running close to, but not synchronized with, the same interval) can
+ * land two entries milliseconds apart, and dividing a real counter delta by that
+ * near-zero elapsed time manufactures a spurious peak rate.
  */
 export function sampleHostCounters(opts = {}) {
 	const interval = opts.intervalMs ?? 5000;
@@ -162,7 +169,6 @@ export function sampleHostCounters(opts = {}) {
 		window() {
 			const now = readHostCounters();
 			if (!now) return null;
-			samples.push(now);
 			const delta = hostCounterDelta(windowStart, now);
 			windowStart = now;
 			return delta;
@@ -243,6 +249,25 @@ export function writeStressMetrics(name, metrics) {
 	}
 }
 
+/**
+ * Remove any previously-written metrics file for `name` before a run starts. Self-hosted
+ * runners reuse the same workspace across runs (unlike ephemeral GH-hosted ones), so the
+ * fixed path in writeStressMetrics() persists between them: a run that crashes before
+ * reaching writeStressMetrics — e.g. during setup, or an early OOM/wedge — would otherwise
+ * leave the *previous* run's metrics file in place, and CI uploads it as if it were this
+ * run's data. Call this at the start of a suite so an early failure uploads no file rather
+ * than a stale one that reads as a fresh success.
+ */
+export function clearStressMetrics(name) {
+	const dir = process.env.HARPER_INTEGRATION_TEST_LOG_DIR;
+	if (!dir) return;
+	try {
+		rmSync(join(dir, `${name}-metrics.json`), { force: true });
+	} catch {
+		// Best effort — a failure here must never fail the test.
+	}
+}
+
 /** Append a markdown block to the GitHub Actions job summary, when running under one. */
 export function writeJobSummary(markdown) {
 	const path = process.env.GITHUB_STEP_SUMMARY;