Skip to content

Commit 5ffbad5

Browse files
Reflexclaude
Reflex
and
claude
committed
fix(h2-transport): guard enqueue after cancel; deduplicate body reads; avoid arrayBuffer copy
Three bugs fixed in the h2-transport added in 1.24.0: 1. **Uncaught exception on SSE early cancel** (`session.ts`): When a consumer breaks out of a `for await` loop over an SSE stream, the ReadableStream `cancel()` callback calls `stream.destroy()` — but Node.js h2 DATA frames already queued in the event loop can still fire `data` events. `controller.enqueue()` on a cancelled ReadableStream throws a TypeError that was unhandled, crashing the process. Fixed by wrapping `controller.enqueue()` and `controller.close()` in try/catch and calling `stream.destroy()` on error. 2. **Race condition in concurrent body reads** (`response.ts`): `_consumeBody()` set a boolean flag synchronously then did async reads, so two concurrent callers (e.g. SDK error-retry + user code) would have the second one throw "Body already consumed" before the first had finished. Fixed by replacing the flag with a shared Promise so concurrent callers all await the same read. 3. **Unnecessary ArrayBuffer copy in `arrayBuffer()`** (`response.ts`): `Buffer.concat()` always allocates a fresh buffer with `byteOffset=0` that owns its full backing `ArrayBuffer`, so the subsequent `.buffer.slice()` was copying the data a second time. Fixed by returning `buf.buffer` directly in that case. Also adds three loadtest scripts to exercise and guard these paths going forward: - `loadtest/large-response-test.ts` — 11 tests covering exec stdout/stderr (1 MB, 10 MB), writeFileContents + readFileContents (512 KB, 2 MB), downloadFile (5 MB, 50 MB), 5× concurrent downloads, SSE streaming (10k lines), and SSE early cancel. - `loadtest/h2-stress-test.ts` — configurable stress test (default: 10 × 20 MB concurrent downloads with md5 verification). - `loadtest/cancel-test.ts` — isolated repro for the SSE early-cancel bug. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
1 parent 572fa22 commit 5ffbad5

5 files changed

Lines changed: 745 additions & 20 deletions

File tree

‎loadtest/cancel-test.ts‎

Lines changed: 85 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,85 @@
1+
/**
2+
* Reproduce the early-cancel SSE bug with HTTP/2:
3+
* Break early from a large SSE stream and confirm no uncaught exceptions crash the process.
4+
*/
5+
import { Runloop } from '../src/sdk.ts';
6+
7+
const client = new Runloop({
8+
bearerToken: process.env['RUNLOOP_API_KEY'] ?? '',
9+
baseURL: process.env['RUNLOOP_BASE_URL'] ?? 'https://api.runloop.pro',
10+
timeout: 120_000,
11+
maxRetries: 0,
12+
http2: true,
13+
});
14+
15+
const uncaughtErrors: string[] = [];
16+
process.on('uncaughtException', (e) => {
17+
uncaughtErrors.push(e.message);
18+
console.error('UNCAUGHT:', e.message);
19+
});
20+
process.on('unhandledRejection', (e) => {
21+
uncaughtErrors.push(String(e));
22+
console.error('UNHANDLED REJECTION:', e);
23+
});
24+
25+
async function main() {
26+
console.log('Creating devbox...');
27+
const devbox = await client.devboxes.createAndAwaitRunning(
28+
{
29+
name: `cancel-test-${Date.now()}`,
30+
launch_parameters: { resource_size_request: 'X_SMALL', keep_alive_time_seconds: 300 },
31+
},
32+
{ longPoll: { timeoutMs: 5 * 60 * 1000 } },
33+
);
34+
console.log('Devbox:', devbox.id);
35+
36+
try {
37+
// Produce a large amount of output
38+
const exec = await client.devboxes.executions.executeAsync(devbox.id, {
39+
command: 'seq 1 50000',
40+
});
41+
await client.devboxes.executions.awaitCompleted(devbox.id, exec.execution_id, {
42+
longPoll: { timeoutMs: 30_000 },
43+
});
44+
45+
// Stream stdout but break immediately after the first event
46+
const stream = await client.devboxes.executions.streamStdoutUpdates(
47+
devbox.id,
48+
exec.execution_id,
49+
{},
50+
);
51+
let count = 0;
52+
for await (const _chunk of stream) {
53+
count++;
54+
break; // Cancel early while lots of data remains
55+
}
56+
console.log(`Broke out after ${count} chunk(s)`);
57+
58+
// Wait for any deferred errors to surface
59+
await new Promise((r) => setTimeout(r, 1000));
60+
console.log(`After 1s wait, uncaught errors so far: ${uncaughtErrors.length}`);
61+
62+
// Confirm the h2 pool is still healthy after the cancel
63+
const result = await client.devboxes.executeAndAwaitCompletion(devbox.id, {
64+
command: 'echo still-alive',
65+
});
66+
console.log('Pool health check after cancel:', result.stdout?.trim());
67+
68+
// Wait once more to be sure
69+
await new Promise((r) => setTimeout(r, 500));
70+
71+
if (uncaughtErrors.length > 0) {
72+
console.error('FAIL: got uncaught errors after SSE cancel:', uncaughtErrors);
73+
process.exitCode = 1;
74+
} else {
75+
console.log('PASS: no uncaught errors from early SSE cancel');
76+
}
77+
} finally {
78+
await client.devboxes.shutdown(devbox.id);
79+
}
80+
}
81+
82+
main().catch((err) => {
83+
console.error('Unhandled error in main:', err);
84+
process.exit(1);
85+
});

‎loadtest/h2-stress-test.ts‎

Lines changed: 150 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,150 @@
1+
/**
2+
* H2 stress test: 10 concurrent 20 MB downloads over HTTP/2.
3+
* Verifies the h2-transport pool handles high concurrency without data corruption.
4+
*
5+
* Usage:
6+
* source ~/env && SMOKE_HTTP2=1 npx tsx loadtest/h2-stress-test.ts
7+
*/
8+
9+
import { createHash } from 'node:crypto';
10+
import { Runloop } from '../src/sdk.ts';
11+
12+
const API_KEY = process.env['RUNLOOP_API_KEY'] ?? '';
13+
const BASE_URL = process.env['RUNLOOP_BASE_URL'] ?? 'https://api.runloop.pro';
14+
const USE_HTTP2 = ['1', 'true'].includes((process.env['SMOKE_HTTP2'] ?? '').toLowerCase());
15+
const CONCURRENCY = parseInt(process.env['CONCURRENCY'] ?? '10', 10);
16+
const FILE_SIZE_MB = parseInt(process.env['FILE_SIZE_MB'] ?? '20', 10);
17+
18+
if (!API_KEY) {
19+
console.error('RUNLOOP_API_KEY is required');
20+
process.exit(1);
21+
}
22+
23+
const client = new Runloop({
24+
bearerToken: API_KEY,
25+
baseURL: BASE_URL,
26+
timeout: 300_000,
27+
maxRetries: 1,
28+
http2: USE_HTTP2,
29+
});
30+
31+
function md5(data: Buffer): string {
32+
return createHash('md5').update(data).digest('hex');
33+
}
34+
35+
async function main() {
36+
const transport = USE_HTTP2 ? 'HTTP/2 (h2-transport)' : 'HTTP/1.1 (node-fetch)';
37+
console.log(`\n=== H2 Stress Test ===`);
38+
console.log(`Transport: ${transport}`);
39+
console.log(`Concurrency: ${CONCURRENCY} simultaneous downloads`);
40+
console.log(`File size: ${FILE_SIZE_MB} MB each`);
41+
console.log(`Total data: ${CONCURRENCY * FILE_SIZE_MB} MB\n`);
42+
43+
console.log('Creating devbox...');
44+
const devbox = await client.devboxes.createAndAwaitRunning(
45+
{
46+
name: `h2-stress-${Date.now()}`,
47+
launch_parameters: {
48+
resource_size_request: 'SMALL',
49+
keep_alive_time_seconds: 60 * 15,
50+
},
51+
},
52+
{ longPoll: { timeoutMs: 10 * 60 * 1000 } },
53+
);
54+
const id = devbox.id;
55+
console.log(`Devbox: ${id}\n`);
56+
57+
try {
58+
// Create test files
59+
console.log(`Creating ${CONCURRENCY} × ${FILE_SIZE_MB} MB files on devbox...`);
60+
await client.devboxes.executeAndAwaitCompletion(id, {
61+
command: `for i in $(seq 1 ${CONCURRENCY}); do dd if=/dev/urandom bs=1M count=${FILE_SIZE_MB} of=/tmp/stress-$i.dat 2>/dev/null; done`,
62+
});
63+
64+
const hashResult = await client.devboxes.executeAndAwaitCompletion(id, {
65+
command: 'md5sum /tmp/stress-*.dat',
66+
last_n: String(CONCURRENCY + 5),
67+
});
68+
69+
const remoteHashes: Record<string, string> = {};
70+
for (const line of (hashResult.stdout ?? '').trim().split('\n')) {
71+
const [hash, path] = line.split(/\s+/);
72+
const name = path?.split('/').pop();
73+
if (hash && name) remoteHashes[name] = hash;
74+
}
75+
76+
if (Object.keys(remoteHashes).length !== CONCURRENCY) {
77+
throw new Error(
78+
`Expected ${CONCURRENCY} hashes, got ${Object.keys(remoteHashes).length}:\n${hashResult.stdout}`,
79+
);
80+
}
81+
console.log(`Files created. Remote hashes verified for ${Object.keys(remoteHashes).length} files.\n`);
82+
83+
// Concurrent downloads
84+
console.log(`Downloading ${CONCURRENCY} files concurrently...`);
85+
const t0 = Date.now();
86+
87+
const downloads = await Promise.all(
88+
Array.from({ length: CONCURRENCY }, (_, i) => i + 1).map((n) =>
89+
client.devboxes
90+
.downloadFile(id, { path: `/tmp/stress-${n}.dat` })
91+
.then((r) => r.arrayBuffer())
92+
.then((ab) => ({ n, buf: Buffer.from(ab) })),
93+
),
94+
);
95+
96+
const elapsed = (Date.now() - t0) / 1000;
97+
98+
// Verify each download
99+
let allOk = true;
100+
for (const { n, buf } of downloads) {
101+
const fname = `stress-${n}.dat`;
102+
const expectedSize = FILE_SIZE_MB * 1024 * 1024;
103+
const remoteHash = remoteHashes[fname];
104+
105+
if (buf.length !== expectedSize) {
106+
console.error(
107+
` ✗ File ${n}: size mismatch — expected ${expectedSize}, got ${buf.length}`,
108+
);
109+
allOk = false;
110+
continue;
111+
}
112+
113+
const localHash = md5(buf);
114+
if (localHash !== remoteHash) {
115+
console.error(
116+
` ✗ File ${n}: md5 mismatch — remote=${remoteHash} local=${localHash} DATA CORRUPTED`,
117+
);
118+
allOk = false;
119+
} else {
120+
console.log(` ✓ File ${n}: ${FILE_SIZE_MB} MB, md5 OK`);
121+
}
122+
}
123+
124+
const totalMB = CONCURRENCY * FILE_SIZE_MB;
125+
const throughput = (totalMB / elapsed).toFixed(1);
126+
console.log(
127+
`\n${CONCURRENCY} downloads completed in ${elapsed.toFixed(1)}s (${throughput} MB/s effective throughput)`,
128+
);
129+
130+
if (allOk) {
131+
console.log(`\n=== PASS: all data intact (transport: ${transport}) ===`);
132+
process.exit(0);
133+
} else {
134+
console.error(`\n=== FAIL: data corruption detected (transport: ${transport}) ===`);
135+
process.exit(1);
136+
}
137+
} finally {
138+
console.log('\nShutting down devbox...');
139+
try {
140+
await client.devboxes.shutdown(id);
141+
} catch {
142+
// ignore
143+
}
144+
}
145+
}
146+
147+
main().catch((err) => {
148+
console.error('Unhandled error:', err);
149+
process.exit(1);
150+
});

0 commit comments

Comments
 (0)