Skip to content

Commit 316d558

Browse files
jrvb-rlReflexclaude
authored
fix(h2-transport): guard enqueue after cancel; deduplicate body reads; avoid arrayBuffer copy (#808)
Co-authored-by: Reflex <agent@reflex.dev> Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
1 parent 572fa22 commit 316d558

5 files changed

Lines changed: 734 additions & 20 deletions

File tree

loadtest/cancel-test.ts

Lines changed: 81 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,81 @@
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(devbox.id, exec.execution_id, {});
47+
let count = 0;
48+
for await (const _chunk of stream) {
49+
count++;
50+
break; // Cancel early while lots of data remains
51+
}
52+
console.log(`Broke out after ${count} chunk(s)`);
53+
54+
// Wait for any deferred errors to surface
55+
await new Promise((r) => setTimeout(r, 1000));
56+
console.log(`After 1s wait, uncaught errors so far: ${uncaughtErrors.length}`);
57+
58+
// Confirm the h2 pool is still healthy after the cancel
59+
const result = await client.devboxes.executeAndAwaitCompletion(devbox.id, {
60+
command: 'echo still-alive',
61+
});
62+
console.log('Pool health check after cancel:', result.stdout?.trim());
63+
64+
// Wait once more to be sure
65+
await new Promise((r) => setTimeout(r, 500));
66+
67+
if (uncaughtErrors.length > 0) {
68+
console.error('FAIL: got uncaught errors after SSE cancel:', uncaughtErrors);
69+
process.exitCode = 1;
70+
} else {
71+
console.log('PASS: no uncaught errors from early SSE cancel');
72+
}
73+
} finally {
74+
await client.devboxes.shutdown(devbox.id);
75+
}
76+
}
77+
78+
main().catch((err) => {
79+
console.error('Unhandled error in main:', err);
80+
process.exit(1);
81+
});

loadtest/h2-stress-test.ts

Lines changed: 146 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,146 @@
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(` ✗ File ${n}: size mismatch — expected ${expectedSize}, got ${buf.length}`);
107+
allOk = false;
108+
continue;
109+
}
110+
111+
const localHash = md5(buf);
112+
if (localHash !== remoteHash) {
113+
console.error(` ✗ File ${n}: md5 mismatch — remote=${remoteHash} local=${localHash} DATA CORRUPTED`);
114+
allOk = false;
115+
} else {
116+
console.log(` ✓ File ${n}: ${FILE_SIZE_MB} MB, md5 OK`);
117+
}
118+
}
119+
120+
const totalMB = CONCURRENCY * FILE_SIZE_MB;
121+
const throughput = (totalMB / elapsed).toFixed(1);
122+
console.log(
123+
`\n${CONCURRENCY} downloads completed in ${elapsed.toFixed(1)}s (${throughput} MB/s effective throughput)`,
124+
);
125+
126+
if (allOk) {
127+
console.log(`\n=== PASS: all data intact (transport: ${transport}) ===`);
128+
process.exit(0);
129+
} else {
130+
console.error(`\n=== FAIL: data corruption detected (transport: ${transport}) ===`);
131+
process.exit(1);
132+
}
133+
} finally {
134+
console.log('\nShutting down devbox...');
135+
try {
136+
await client.devboxes.shutdown(id);
137+
} catch {
138+
// ignore
139+
}
140+
}
141+
}
142+
143+
main().catch((err) => {
144+
console.error('Unhandled error:', err);
145+
process.exit(1);
146+
});

0 commit comments

Comments
 (0)