From 4a2d4c7fc7dd9a61e5bbcf70c23c39bd4468f30b Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Mon, 31 Aug 2026 17:56:39 -0600 Subject: [PATCH 1/6] Pin the SSE finite-generator completion contract with an integration regression anchor Promotes the QA-537 exploratory spec into the integration suite as integrationTests/server/sse-finite-generator.test.ts. It pins the shapes #1628 / PR #1632 turned on -- where a generator's terminal `done` step falls relative to its yielded values -- across N=0, N=1, N=5 and N=3000, plus a mid-stream-throw contrast arm, a liveness canary and an hdb.log uncaughtException sweep. The SSE suites in integrationTests/server/ had each grown their own copy of the AbortController-bounded SSE consumer, the /Probe/ reader and the hdb.log helpers. Rather than add a fourth, they move to integrationTests/utils/sseStream.ts and sse-throw-midstream.test.ts now imports them; the raw-socket capture used by stream-error-contract.test.ts is a deliberately different technique and stays put. Test-only: no product code changes. Refs #1628 Co-Authored-By: Claude Opus --- .../server/sse-finite-generator.test.ts | 258 ++++++++++++++++++ .../server/sse-finite-generator/config.yaml | 5 + .../server/sse-finite-generator/resources.js | 123 +++++++++ .../sse-finite-generator/schema.graphql | 9 + .../server/sse-throw-midstream.test.ts | 199 ++------------ integrationTests/utils/sseStream.ts | 209 ++++++++++++++ 6 files changed, 625 insertions(+), 178 deletions(-) create mode 100644 integrationTests/server/sse-finite-generator.test.ts create mode 100644 integrationTests/server/sse-finite-generator/config.yaml create mode 100644 integrationTests/server/sse-finite-generator/resources.js create mode 100644 integrationTests/server/sse-finite-generator/schema.graphql create mode 100644 integrationTests/utils/sseStream.ts diff --git a/integrationTests/server/sse-finite-generator.test.ts b/integrationTests/server/sse-finite-generator.test.ts new file mode 100644 index 0000000000..e754a1c649 --- /dev/null +++ b/integrationTests/server/sse-finite-generator.test.ts @@ -0,0 +1,258 @@ +/** + * QA-537 — regression verify for #1628 "SSE hang on a finite generator streamed to completion", + * fixed by PR #1632 (commit 69c8c89a9). + * + * Bug recap (see `server/serverHelpers/contentTypes.ts`, `transformIterable`): the SSE + * `serialize` transform was applied to the generator's terminal `{ value: undefined, done: true }` + * step as well as to its yielded values. `serialize()` dereferences `message.acknowledge` on its + * first line, so `undefined.acknowledge` threw a TypeError inside Readable.from's pull loop — an + * uncaughtException that left the HTTP response hanging, never closed, whenever a plain finite + * async generator was streamed to completion over `Accept: text/event-stream`. The fix passes the + * terminal `done` step through untransformed in both the sync and async branches. + * + * The shapes below vary where the terminal step falls relative to the yielded values, which is + * the axis the bug lived on: + * - EmptyGen (N=0) — terminal step is the FIRST step produced; the sharpest trigger + * - SingleGen (N=1) — minimal non-empty completing case + * - FiniteGen (N=5) — canonical completing case from the issue + * - LargeGen (N=3000) — same terminal-step path after a long stream (also a mild + * backpressure check) + * - ThrowGen (throws after 2 of 5) — contrast arm: a generator that rejects never reaches the + * terminal step at all, so it exercises the #1789 teardown path instead + * + * Related but distinct suites, none of which cover the completion shapes above: + * - `sse-throw-midstream.test.ts` (#1789 / QA-559) is the dedicated anchor for a generator that + * *throws*, over SSE on Node only. Its `CleanGen` control overlaps FiniteGen; ThrowGen here is + * the contrast arm, not the anchor. + * - `stream-error-contract.test.ts` (QA-890) pins the stream-*error* contract across SSE, NDJSON + * and iterable-REST on raw socket bytes. + * + * Every request is bounded by an AbortController, so a regressed hang fails this suite + * deterministically instead of wedging the run. After the streaming cases a plain `/Probe/` + * request confirms the worker is still healthy, and `hdb.log` is scanned for newly-appeared + * `uncaughtException` lines — the bug's other signature. + * + * Reproduction: + * npm run test:integration -- "integrationTests/server/sse-finite-generator.test.ts" + */ +import { suite, test, before, after } from 'node:test'; +import { ok, strictEqual } from 'node:assert'; +import { resolve } from 'node:path'; +import { setTimeout as sleep } from 'node:timers/promises'; +import { setupHarperWithFixture, teardownHarper, type ContextWithHarper } from '@harperfast/integration-testing'; +import { + consumeSse, + countUncaught, + getProbe, + readLogSafe, + resolveHdbLogPath, + uncaughtAfterSettle, + waitForProbe, +} from '../utils/sseStream.ts'; +// @ts-expect-error utils/client.mjs has no type declarations; runtime resolves fine +import { createApiClient } from '../apiTests/utils/client.mjs'; + +const FIXTURE_PATH = resolve(import.meta.dirname, 'sse-finite-generator'); +const skipSuite = process.platform === 'win32'; + +type Client = ReturnType; + +interface Lifecycle { + opened: number; + closed: number; +} + +interface ProbeSnap { + ok: boolean; + finite: Lifecycle; + empty: Lifecycle; + single: Lifecycle; + throwGen: Lifecycle; + large: Lifecycle; +} + +suite( + 'QA-537 SSE finite-generator completion regression verify (#1628 / PR #1632)', + { skip: skipSuite }, + (ctx: ContextWithHarper) => { + let client: Client; + let restBase = ''; + let authHeaders: Record = {}; + let logPath = ''; + + before(async () => { + await setupHarperWithFixture(ctx, FIXTURE_PATH, { + config: { threads: { count: 1 }, logging: { console: true, level: 'error' } }, + env: {}, + }); + client = createApiClient(ctx.harper); + restBase = client.restURL; + authHeaders = { Authorization: client.headers.Authorization as string }; + logPath = resolveHdbLogPath(ctx.harper as any); + + const deadline = Date.now() + 30_000; + let ready = false; + while (Date.now() < deadline) { + try { + const probe = await getProbe(restBase, authHeaders); + if (probe?.ok !== undefined) { + ready = true; + break; + } + } catch { + /* not ready */ + } + await sleep(250); + } + ok(ready, 'Probe route did not become ready within 30 seconds'); + }); + + after(async () => { + await teardownHarper(ctx); + }); + + async function assertCompletes(path: string, expectedEvents: number) { + const r = await consumeSse(`${restBase}${path}`, authHeaders, 15_000); + ok(r.status >= 200 && r.status < 300, `expected 2xx, got ${r.status}`); + ok( + !r.aborted, + `must not hit the AbortController timeout — a timeout here means the #1628 hang regressed. raw:\n${r.raw}` + ); + strictEqual( + r.events.length, + expectedEvents, + `expected ${expectedEvents} SSE data events, got ${r.events.length}` + ); + strictEqual( + r.terminatedBy, + 'end', + `a generator streamed to completion must close via a clean 'end', got terminatedBy=${r.terminatedBy} errored=${r.errored?.message ?? null}` + ); + return r; + } + + // ── 1: empty generator (N=0) — the terminal step is the very first step ────────────── + + test('1: EmptyGen (N=0) over SSE — zero events, response still closes cleanly', { timeout: 20_000 }, async () => { + const uncaughtBefore = countUncaught(readLogSafe(logPath)); + await assertCompletes('/EmptyGen/', 0); + strictEqual( + (await uncaughtAfterSettle(logPath)) - uncaughtBefore, + 0, + 'no NEW uncaughtException should be logged when the terminal step is the only step' + ); + }); + + // ── 2: single-event generator (N=1) ────────────────────────────────────────────────── + + test('2: SingleGen (N=1) over SSE — exactly 1 event, response closes cleanly', { timeout: 20_000 }, async () => { + const r = await assertCompletes('/SingleGen/', 1); + ok(r.events[0].includes('"n":0'), `expected the single event to contain n=0, got: ${r.events[0]}`); + }); + + // ── 3: canonical finite generator (N=5) ────────────────────────────────────────────── + + test( + '3: FiniteGen (N=5) over SSE — all 5 events arrive and the response closes cleanly', + { timeout: 20_000 }, + async () => { + const uncaughtBefore = countUncaught(readLogSafe(logPath)); + const r = await assertCompletes('/FiniteGen/', 5); + for (let i = 0; i < 5; i++) { + ok(r.events[i].includes(`"n":${i}`), `expected event ${i} to contain n=${i}, got: ${r.events[i]}`); + } + strictEqual( + (await uncaughtAfterSettle(logPath)) - uncaughtBefore, + 0, + 'no NEW uncaughtException should be logged for a completing finite SSE generator' + ); + } + ); + + // ── 4: large finite generator (N=3000) ─────────────────────────────────────────────── + + test( + '4: LargeGen (N=3000) over SSE — all 3000 events arrive and the response closes cleanly', + { timeout: 20_000 }, + async () => { + const r = await assertCompletes('/LargeGen/', 3000); + ok(r.events[0].includes('"n":0'), `expected first event n=0, got: ${r.events[0]}`); + ok(r.events[2999].includes('"n":2999'), `expected last event n=2999, got: ${r.events[2999]}`); + } + ); + + // ── 5: generator that throws partway (2 of 5) ──────────────────────────────────────── + + test( + '5: ThrowGen (throws after 2 of 5) over SSE — terminates in bounded time, no uncaughtException', + { timeout: 20_000 }, + async () => { + const uncaughtBefore = countUncaught(readLogSafe(logPath)); + + const r = await consumeSse(`${restBase}/ThrowGen/`, authHeaders, 15_000); + console.log( + `[QA-537][5] ThrowGen: status=${r.status} events=${r.events.length} terminatedBy=${r.terminatedBy} aborted=${r.aborted} errored=${r.errored?.message ?? null} elapsedMs=${r.elapsedMs}` + ); + + // A rejecting generator never reaches the terminal `done` step #1632 fixed; it exits via + // the #1789 pipeline() teardown, which destroys the response abruptly rather than ending + // it cleanly. Both are bounded — that is what is asserted here. The full throw contract + // lives in sse-throw-midstream.test.ts. + ok(!r.aborted, `must not hit the AbortController timeout — the response never terminated. raw:\n${r.raw}`); + ok(r.terminatedBy !== null, 'response must terminate via end/error/close, not hang indefinitely'); + ok( + r.events.length <= 2, + `expected at most the 2 events yielded before the throw, got ${r.events.length}. raw:\n${r.raw}` + ); + for (let i = 0; i < r.events.length; i++) { + ok(r.events[i].includes(`"n":${i}`), `expected event ${i} to contain n=${i}, got: ${r.events[i]}`); + } + strictEqual( + (await uncaughtAfterSettle(logPath)) - uncaughtBefore, + 0, + 'no NEW uncaughtException should be logged for a mid-stream throw' + ); + } + ); + + // ── Z: liveness canary + final uncaughtException sweep ────────────────────────────── + + test( + 'Z: liveness canary — worker survived every streaming shape above, no new uncaughtException', + { timeout: 30_000 }, + async () => { + const logBefore = readLogSafe(logPath); + const probe = await waitForProbe(restBase, authHeaders, (snapshot) => + [snapshot.finite, snapshot.empty, snapshot.single, snapshot.throwGen, snapshot.large].every( + (counter) => counter.opened >= 1 && counter.opened === counter.closed + ) + ); + console.log(`[QA-537][Z] liveness probe: ${probe ? JSON.stringify(probe) : 'DEAD'}`); + ok(probe !== null, 'Harper must still respond to Probe/ after all streaming cases'); + for (const [name, counter] of Object.entries({ + FiniteGen: probe!.finite, + EmptyGen: probe!.empty, + SingleGen: probe!.single, + ThrowGen: probe!.throwGen, + LargeGen: probe!.large, + })) { + ok(counter.opened >= 1, `${name} should have been opened`); + strictEqual(counter.closed, counter.opened, `${name} generator was not closed`); + } + + await sleep(1_000); // same non-event settle as the per-case sweeps above + const newLines = readLogSafe(logPath).slice(logBefore.length); + const newUncaught = countUncaught(newLines); + if (newUncaught > 0) { + console.log( + `[QA-537][Z] NEW uncaughtException lines found:\n${newLines + .split('\n') + .filter((line) => line.includes('uncaughtException')) + .join('\n')}` + ); + } + strictEqual(newUncaught, 0, 'no uncaughtException should appear while the worker is drained and probed'); + } + ); + } +); diff --git a/integrationTests/server/sse-finite-generator/config.yaml b/integrationTests/server/sse-finite-generator/config.yaml new file mode 100644 index 0000000000..efffc0833f --- /dev/null +++ b/integrationTests/server/sse-finite-generator/config.yaml @@ -0,0 +1,5 @@ +graphqlSchema: + files: '*.graphql' +jsResource: + files: resources.js +rest: true diff --git a/integrationTests/server/sse-finite-generator/resources.js b/integrationTests/server/sse-finite-generator/resources.js new file mode 100644 index 0000000000..d53d62496b --- /dev/null +++ b/integrationTests/server/sse-finite-generator/resources.js @@ -0,0 +1,123 @@ +// QA-537 — regression verify for #1628 ("Fix SSE hang on finite generator streamed to +// completion", PR #1632, commit 69c8c89a9). +// +// Bug recap: transformIterable (server/serverHelpers/contentTypes.ts) used to call the SSE +// `serialize` transform on the generator's TERMINAL `{ value: undefined, done: true }` step +// too. serialize()'s first line dereferences `message.acknowledge`, so `undefined.acknowledge` +// threw a TypeError inside Readable.from's pull loop — an uncaughtException that left the SSE +// HTTP response hanging (never closed) whenever a plain finite async generator was streamed to +// completion over `Accept: text/event-stream`. The fix guards transformIterable to pass the +// terminal `done` step through untransformed in both the sync and async branches. +// +// This fixture exercises multiple finite-generator shapes over SSE (dispatched via +// `resource.connect()`, which is what Harper's REST layer invokes for CONNECT/SSE requests — +// see server/REST.ts: `isSse` sets method to 'CONNECT', which calls `resource.connect(...)`): +// FiniteGen - canonical case: 5 events then a bare `return` (natural completion). +// EmptyGen - 0 events: generator returns immediately, hitting the terminal step first. +// SingleGen - exactly 1 event then completion. +// ThrowGen - yields 2 of an intended 5 events, then throws (partway failure, not the +// terminal-`done` code path; the full throw contract is anchored by the +// sse-throw-midstream fixture instead). +// LargeGen - 3000 events then completion (larger finite stream, same terminal-step path). +// Probe - readiness + per-resource open/close lifecycle counters, plain JSON. + +const G = (globalThis.__QA537__ ??= { + finite: { opened: 0, closed: 0 }, + empty: { opened: 0, closed: 0 }, + single: { opened: 0, closed: 0 }, + throwGen: { opened: 0, closed: 0 }, + large: { opened: 0, closed: 0 }, +}); + +function sleep(ms) { + return new Promise((r) => setTimeout(r, ms)); +} + +// GET /FiniteGen/ (Accept: text/event-stream) — canonical finite generator, N=5. +export class FiniteGen extends Resource { + static loadAsInstance = false; + static async *connect(_target, _incomingMessages, _request) { + G.finite.opened++; + try { + for (let i = 0; i < 5; i++) { + yield { n: i }; + await sleep(2); + } + } finally { + G.finite.closed++; + } + } +} + +// GET /EmptyGen/ (Accept: text/event-stream) — 0 events; the terminal `done` step is the +// very FIRST step produced, so this isolates the terminal-step handling from any mid-stream +// yields at all. +export class EmptyGen extends Resource { + static loadAsInstance = false; + // eslint-disable-next-line require-yield + static async *connect(_target, _incomingMessages, _request) { + G.empty.opened++; + try { + // intentionally yields nothing + } finally { + G.empty.closed++; + } + } +} + +// GET /SingleGen/ (Accept: text/event-stream) — exactly 1 event then completion. +export class SingleGen extends Resource { + static loadAsInstance = false; + static async *connect(_target, _incomingMessages, _request) { + G.single.opened++; + try { + yield { n: 0 }; + } finally { + G.single.closed++; + } + } +} + +// GET /ThrowGen/ (Accept: text/event-stream) — yields 2 of an intended 5, then throws. +// This does NOT hit the fixed terminal-`done` code path (the iterator never reaches a +// `done:true` step — it rejects instead); it is the bounded-termination contrast arm next to +// the natural-completion cases. +export class ThrowGen extends Resource { + static loadAsInstance = false; + static async *connect(_target, _incomingMessages, _request) { + G.throwGen.opened++; + try { + for (let i = 0; i < 5; i++) { + if (i === 2) throw new Error('QA537-intentional-throw-partway'); + yield { n: i }; + await sleep(2); + } + } finally { + G.throwGen.closed++; + } + } +} + +// GET /LargeGen/ (Accept: text/event-stream) — 3000 events then completion. Larger finite +// stream exercising the same terminal-step path at volume (also a mild backpressure check). +export class LargeGen extends Resource { + static loadAsInstance = false; + static async *connect(_target, _incomingMessages, _request) { + G.large.opened++; + try { + for (let i = 0; i < 3000; i++) { + yield { n: i }; + } + } finally { + G.large.closed++; + } + } +} + +// GET /Probe/ — readiness + lifecycle-counter snapshot (plain JSON, not SSE). +export class Probe extends Resource { + static loadAsInstance = false; + static async get() { + return { ok: true, ...G }; + } +} diff --git a/integrationTests/server/sse-finite-generator/schema.graphql b/integrationTests/server/sse-finite-generator/schema.graphql new file mode 100644 index 0000000000..41998e9365 --- /dev/null +++ b/integrationTests/server/sse-finite-generator/schema.graphql @@ -0,0 +1,9 @@ +# QA-537 — regression verify for #1628 ("Fix SSE hang on finite generator streamed to +# completion", merged via PR #1632 / commit 69c8c89a9). +# +# No table backing is needed for the generator Resources under test, but a jsResource +# fixture requires at least one @table/@export type to be a valid component, so this is a +# minimal placeholder. +type Placeholder @table @export { + id: ID @primaryKey +} diff --git a/integrationTests/server/sse-throw-midstream.test.ts b/integrationTests/server/sse-throw-midstream.test.ts index 7a7d9ec246..2ae5b7ad86 100644 --- a/integrationTests/server/sse-throw-midstream.test.ts +++ b/integrationTests/server/sse-throw-midstream.test.ts @@ -23,8 +23,8 @@ * test in contentTypes.test.js) in bounded time, receiving only the events flushed before the * throw, with NO uncaughtException logged and the worker still alive afterward. * - * This suite starts from the QA-537 harness/fixture pattern (qa-scratch/qa537-sse-finite- - * generator/) and its `ThrowGen` case, which *documented* (did not assert) the pre-fix hang. + * This suite starts from the QA-537 harness/fixture pattern (sse-finite-generator.test.ts, which + * anchors #1628) and its `ThrowGen` case, which *documented* (did not assert) the pre-fix hang. * Here the throw cases are promoted to hard assertions of clean termination: * ThrowFirst - throws on the very first step, before any bytes are yielded. * ThrowMid - yields 3 of an intended 6 events, then throws (genuine mid-stream failure). @@ -44,14 +44,19 @@ */ import { suite, test, before, after } from 'node:test'; import { ok, strictEqual } from 'node:assert'; -import { resolve, join } from 'node:path'; +import { resolve } from 'node:path'; import { readFileSync } from 'node:fs'; -import { StringDecoder } from 'node:string_decoder'; import { setTimeout as sleep } from 'node:timers/promises'; -import http from 'node:http'; -import https from 'node:https'; -import { URL } from 'node:url'; import { setupHarperWithFixture, teardownHarper, type ContextWithHarper } from '@harperfast/integration-testing'; +import { + consumeSse, + countUncaught, + getProbe, + readLogSafe, + resolveHdbLogPath, + uncaughtAfterSettle, + waitForProbe, +} from '../utils/sseStream.ts'; // @ts-expect-error utils/client.mjs has no type declarations; runtime resolves fine import { createApiClient } from '../apiTests/utils/client.mjs'; @@ -67,170 +72,6 @@ interface ProbeSnap { clean: { opened: number; closed: number }; } -async function getProbe(restBase: string, authHeaders: Record): Promise { - const url = new URL(`${restBase}/Probe/`); - const lib = url.protocol === 'https:' ? https : http; - return new Promise((resolvePromise, reject) => { - const req = lib.request( - url, - { - method: 'GET', - headers: { ...authHeaders, Accept: 'application/json' }, - rejectUnauthorized: false, - signal: AbortSignal.timeout(3_000), - } as any, - (res) => { - const chunks: Buffer[] = []; - res.on('data', (d: Buffer) => chunks.push(d)); - res.on('end', () => { - try { - resolvePromise(JSON.parse(Buffer.concat(chunks).toString('utf8'))); - } catch (e) { - reject(e); - } - }); - res.on('error', reject); - } - ); - req.on('error', reject); - req.end(); - }); -} - -async function waitForProbe( - restBase: string, - authHeaders: Record, - predicate: (probe: ProbeSnap) => boolean, - timeoutMs = 5_000 -): Promise { - const deadline = Date.now() + timeoutMs; - let probe: ProbeSnap | null = null; - while (Date.now() < deadline) { - probe = await getProbe(restBase, authHeaders).catch(() => null); - if (probe && predicate(probe)) return probe; - await sleep(50); - } - return probe; -} - -// ── AbortController-bounded SSE stream consumer ────────────────────────────────────────── -// Wraps the request in an AbortController with a generous but bounded timeout, so a genuine -// regression (hung response) shows up as a caught timeout/abort in the test, never a hung -// process. Resolves on whichever of 'end' / 'error' / 'close' fires first, since the fixed -// pipeline()-based teardown closes the response abruptly (not via a clean 'end') on a source -// error -- see PR #1789. - -interface SseResult { - status: number; - raw: string; - events: string[]; // parsed `data: ...` payloads, in order - terminatedBy: 'end' | 'error' | 'close' | null; // which event resolved the request - aborted: boolean; // did our own timeout abort fire (i.e. it hung)? - errored: Error | null; - elapsedMs: number; -} - -function consumeSse(urlStr: string, authHeaders: Record, timeoutMs = 12_000): Promise { - const url = new URL(urlStr); - const lib = url.protocol === 'https:' ? https : http; - const controller = new AbortController(); - let timedOut = false; - const start = Date.now(); - const timer = setTimeout(() => { - timedOut = true; - controller.abort(); - }, timeoutMs); - - return new Promise((resolvePromise) => { - const result: SseResult = { - status: 0, - raw: '', - events: [], - terminatedBy: null, - aborted: false, - errored: null, - elapsedMs: 0, - }; - let settled = false; - const finish = (terminatedBy: SseResult['terminatedBy'], err?: Error) => { - if (settled) return; - settled = true; - clearTimeout(timer); - result.terminatedBy = terminatedBy; - result.errored = err ?? null; - result.elapsedMs = Date.now() - start; - result.events = result.raw - .split('\n') - .filter((l) => l.startsWith('data: ')) - .map((l) => l.slice('data: '.length)); - resolvePromise(result); - }; - const req = lib.request( - url, - { - method: 'GET', - headers: { ...authHeaders, Accept: 'text/event-stream' }, - rejectUnauthorized: false, - signal: controller.signal, - } as any, - (res) => { - result.status = res.statusCode ?? 0; - // Decode incrementally: a bare d.toString('utf8') per chunk corrupts any multi-byte - // character that straddles a TCP chunk boundary into U+FFFD. - const decoder = new StringDecoder('utf8'); - res.on('data', (d: Buffer) => { - result.raw += decoder.write(d); - }); - res.on('end', () => { - result.raw += decoder.end(); - finish('end'); - }); - res.on('error', (e: Error) => finish('error', e)); - res.on('close', () => finish('close')); - } - ); - req.on('error', (e: any) => { - if (timedOut || e?.name === 'AbortError') { - if (settled) return; - settled = true; - clearTimeout(timer); - result.aborted = true; - result.elapsedMs = Date.now() - start; - result.events = result.raw - .split('\n') - .filter((l) => l.startsWith('data: ')) - .map((l) => l.slice('data: '.length)); - resolvePromise(result); - } else { - finish('error', e); - } - }); - req.end(); - }); -} - -function readLogSafe(logPath: string): string { - try { - return readFileSync(logPath, 'utf8'); - } catch { - return ''; - } -} - -function countUncaught(log: string): number { - return log.split('\n').filter((l) => l.includes('uncaughtException')).length; -} - -/** - * Asserting a NON-event: the worker logs an uncaughtException asynchronously, so reading hdb.log - * the instant the response closes can pass vacuously by simply outrunning the flush. Give the - * writer a bounded settle first — one of the cases AGENTS.md reserves a fixed sleep for. - */ -async function uncaughtAfterSettle(logPath: string): Promise { - await sleep(1_000); - return countUncaught(readLogSafe(logPath)); -} - // ── Suite ────────────────────────────────────────────────────────────────────────────────── suite( @@ -250,15 +91,13 @@ suite( client = createApiClient(ctx.harper); restBase = client.restURL; authHeaders = { Authorization: client.headers.Authorization as string }; - logPath = (ctx.harper as any).logDir - ? join((ctx.harper as any).logDir, 'hdb.log') - : join((ctx.harper as any).dataRootDir, 'log', 'hdb.log'); + logPath = resolveHdbLogPath(ctx.harper as any); const deadline = Date.now() + 30_000; let ready = false; while (Date.now() < deadline) { try { - const p = await getProbe(restBase, authHeaders); + const p = await getProbe(restBase, authHeaders); if (p?.ok !== undefined) { ready = true; break; @@ -301,7 +140,11 @@ suite( `expected 0 events (throw before any yield), got ${r.events.length}. raw:\n${r.raw}` ); - const probe = await waitForProbe(restBase, authHeaders, (snapshot) => snapshot.throwFirst.closed >= 1); + const probe = await waitForProbe( + restBase, + authHeaders, + (snapshot) => snapshot.throwFirst.closed >= 1 + ); ok(probe && probe.throwFirst.closed >= 1, 'ThrowFirst generator should have completed cleanup'); const uncaughtAfter = await uncaughtAfterSettle(logPath); strictEqual( @@ -339,7 +182,7 @@ suite( ok(r.events[i].includes(`"n":${i}`), `expected event ${i} to contain n=${i}, got: ${r.events[i]}`); } - const probe = await waitForProbe(restBase, authHeaders, (snapshot) => snapshot.throwMid.closed >= 1); + const probe = await waitForProbe(restBase, authHeaders, (snapshot) => snapshot.throwMid.closed >= 1); ok(probe && probe.throwMid.closed >= 1, 'ThrowMid generator should have completed cleanup'); const uncaughtAfter = await uncaughtAfterSettle(logPath); strictEqual( @@ -378,7 +221,7 @@ suite( { timeout: 30_000 }, async () => { const logBefore = readLogSafe(logPath); - const p = await waitForProbe( + const p = await waitForProbe( restBase, authHeaders, (snapshot) => snapshot.throwFirst.closed >= 1 && snapshot.throwMid.closed >= 1 && snapshot.clean.closed >= 1 diff --git a/integrationTests/utils/sseStream.ts b/integrationTests/utils/sseStream.ts new file mode 100644 index 0000000000..a1c7e3dc03 --- /dev/null +++ b/integrationTests/utils/sseStream.ts @@ -0,0 +1,209 @@ +/** + * Shared helpers for the SSE streaming regression suites in `integrationTests/server/` + * (`sse-finite-generator.test.ts`, `sse-throw-midstream.test.ts`). + * + * These suites all need the same three things, and each grew its own copy before this module + * existed: + * + * - an SSE consumer bounded by an AbortController, so a regressed hang fails the test in + * bounded time instead of wedging the whole run; + * - a `/Probe/` reader for the per-resource open/close lifecycle counters the fixtures keep, + * with a polling form for the counters that settle asynchronously after a response ends; + * - an `hdb.log` reader that counts `uncaughtException` lines, since the bugs these suites + * anchor (#1628, #1789) manifested as an uncaught throw inside the worker rather than as a + * bad response. + * + * Nothing here is Harper-specific beyond the `/Probe/` path convention and the log layout; the + * raw-socket capture used by `stream-error-contract.test.ts` is a deliberately different + * technique (it inspects chunk framing and the exact close mechanism) and is not shared. + */ +import { readFileSync } from 'node:fs'; +import { join } from 'node:path'; +import { StringDecoder } from 'node:string_decoder'; +import { setTimeout as sleep } from 'node:timers/promises'; +import http from 'node:http'; +import https from 'node:https'; +import { URL } from 'node:url'; + +export interface SseResult { + status: number; + raw: string; + /** parsed `data: ...` payloads, in order */ + events: string[]; + /** which response event resolved the request, or null if our own timeout fired first */ + terminatedBy: 'end' | 'error' | 'close' | null; + /** did our own timeout abort fire (i.e. the response hung)? */ + aborted: boolean; + errored: Error | null; + elapsedMs: number; +} + +function parseEvents(raw: string): string[] { + return raw + .split('\n') + .filter((line) => line.startsWith('data: ')) + .map((line) => line.slice('data: '.length)); +} + +/** + * Consume an SSE response, bounded by an AbortController. + * + * Resolves on whichever of 'end' / 'error' / 'close' fires first: the pipeline()-based teardown + * introduced by #1789 closes the response abruptly rather than via a clean 'end' when the source + * generator rejects, so waiting only for 'end' would read a correct abrupt close as a hang. + */ +export function consumeSse( + urlStr: string, + authHeaders: Record, + timeoutMs = 12_000 +): Promise { + const url = new URL(urlStr); + const lib = url.protocol === 'https:' ? https : http; + const controller = new AbortController(); + let timedOut = false; + const start = Date.now(); + const timer = setTimeout(() => { + timedOut = true; + controller.abort(); + }, timeoutMs); + + return new Promise((resolvePromise) => { + const result: SseResult = { + status: 0, + raw: '', + events: [], + terminatedBy: null, + aborted: false, + errored: null, + elapsedMs: 0, + }; + let settled = false; + const finish = (terminatedBy: SseResult['terminatedBy'], err?: Error) => { + if (settled) return; + settled = true; + clearTimeout(timer); + result.terminatedBy = terminatedBy; + result.errored = err ?? null; + result.elapsedMs = Date.now() - start; + result.events = parseEvents(result.raw); + resolvePromise(result); + }; + const req = lib.request( + url, + { + method: 'GET', + headers: { ...authHeaders, Accept: 'text/event-stream' }, + rejectUnauthorized: false, + signal: controller.signal, + } as any, + (res) => { + result.status = res.statusCode ?? 0; + // Decode incrementally: a bare d.toString('utf8') per chunk corrupts any multi-byte + // character that straddles a TCP chunk boundary into U+FFFD. + const decoder = new StringDecoder('utf8'); + res.on('data', (d: Buffer) => { + result.raw += decoder.write(d); + }); + res.on('end', () => { + result.raw += decoder.end(); + finish('end'); + }); + res.on('error', (e: Error) => finish('error', e)); + res.on('close', () => finish('close')); + } + ); + req.on('error', (e: any) => { + if (timedOut || e?.name === 'AbortError') { + if (settled) return; + settled = true; + clearTimeout(timer); + result.aborted = true; + result.elapsedMs = Date.now() - start; + result.events = parseEvents(result.raw); + resolvePromise(result); + } else { + finish('error', e); + } + }); + req.end(); + }); +} + +/** Read the fixture's `/Probe/` resource: readiness plus its open/close lifecycle counters. */ +export function getProbe(restBase: string, authHeaders: Record, timeoutMs = 3_000): Promise { + const url = new URL(`${restBase}/Probe/`); + const lib = url.protocol === 'https:' ? https : http; + return new Promise((resolvePromise, reject) => { + const req = lib.request( + url, + { + method: 'GET', + headers: { ...authHeaders, Accept: 'application/json' }, + rejectUnauthorized: false, + signal: AbortSignal.timeout(timeoutMs), + } as any, + (res) => { + const chunks: Buffer[] = []; + res.on('data', (d: Buffer) => chunks.push(d)); + res.on('end', () => { + try { + resolvePromise(JSON.parse(Buffer.concat(chunks).toString('utf8'))); + } catch (e) { + reject(e); + } + }); + res.on('error', reject); + } + ); + req.on('error', reject); + req.end(); + }); +} + +/** + * Poll `/Probe/` until `predicate` holds. A generator's `finally` block runs after the response + * has already terminated, so its close counter is not readable the instant the client resolves. + * Returns the last snapshot read (which may not satisfy the predicate) or null if none was. + */ +export async function waitForProbe( + restBase: string, + authHeaders: Record, + predicate: (probe: T) => boolean, + timeoutMs = 5_000 +): Promise { + const deadline = Date.now() + timeoutMs; + let probe: T | null = null; + while (Date.now() < deadline) { + probe = await getProbe(restBase, authHeaders).catch(() => null); + if (probe && predicate(probe)) return probe; + await sleep(50); + } + return probe; +} + +/** Where the test instance writes hdb.log, across framework versions that expose either field. */ +export function resolveHdbLogPath(harper: { logDir?: string; dataRootDir?: string }): string { + return harper.logDir ? join(harper.logDir, 'hdb.log') : join(harper.dataRootDir as string, 'log', 'hdb.log'); +} + +export function readLogSafe(logPath: string): string { + try { + return readFileSync(logPath, 'utf8'); + } catch { + return ''; + } +} + +export function countUncaught(log: string): number { + return log.split('\n').filter((line) => line.includes('uncaughtException')).length; +} + +/** + * Asserting a NON-event: the worker logs an uncaughtException asynchronously, so reading hdb.log + * the instant the response closes can pass vacuously by simply outrunning the flush. Give the + * writer a bounded settle first — one of the cases AGENTS.md reserves a fixed sleep for. + */ +export async function uncaughtAfterSettle(logPath: string): Promise { + await sleep(1_000); + return countUncaught(readLogSafe(logPath)); +} From 74cdbd5a8b9f3395340ef580ed4c74e60432a9bf Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Mon, 31 Aug 2026 18:11:36 -0600 Subject: [PATCH 2/6] Close the vacuous-pass gaps the pre-push review found in the SSE uncaughtException checks - awaitFixtureReady() now owns both suites' readiness poll and reads hdb.log with readFileSync, so an unreadable log fails setup instead of silently making every uncaughtException delta zero. - Every completion arm gets its own delta check, and both suites' final sweeps compare against the baseline taken in before() rather than a test-local slice, so a case whose own check was outrun by the log flush is still caught. - consumeSse() decides `aborted` from its own timer rather than from which event won the settle race: an abort destroys the response, so a hung stream also emits 'close', and attributing the settle to that event reported a genuine hang as a bounded termination. - waitForProbe() passes its remaining deadline down to each probe request. - Trimmed the narrating comments off the fixture, schema and helper module. Co-Authored-By: Claude Opus --- .../server/sse-finite-generator.test.ts | 115 ++++++------------ .../server/sse-finite-generator/resources.js | 42 ++----- .../sse-finite-generator/schema.graphql | 8 +- .../server/sse-throw-midstream.test.ts | 49 +++----- integrationTests/utils/sseStream.ts | 83 +++++++------ 5 files changed, 113 insertions(+), 184 deletions(-) diff --git a/integrationTests/server/sse-finite-generator.test.ts b/integrationTests/server/sse-finite-generator.test.ts index e754a1c649..6cde31f61b 100644 --- a/integrationTests/server/sse-finite-generator.test.ts +++ b/integrationTests/server/sse-finite-generator.test.ts @@ -1,36 +1,23 @@ /** * QA-537 — regression verify for #1628 "SSE hang on a finite generator streamed to completion", - * fixed by PR #1632 (commit 69c8c89a9). + * fixed by PR #1632. * - * Bug recap (see `server/serverHelpers/contentTypes.ts`, `transformIterable`): the SSE - * `serialize` transform was applied to the generator's terminal `{ value: undefined, done: true }` - * step as well as to its yielded values. `serialize()` dereferences `message.acknowledge` on its - * first line, so `undefined.acknowledge` threw a TypeError inside Readable.from's pull loop — an - * uncaughtException that left the HTTP response hanging, never closed, whenever a plain finite - * async generator was streamed to completion over `Accept: text/event-stream`. The fix passes the - * terminal `done` step through untransformed in both the sync and async branches. + * `transformIterable` (server/serverHelpers/contentTypes.ts) applied the SSE `serialize` transform + * to the generator's terminal `{ value: undefined, done: true }` step as well as to its yielded + * values. `serialize()` dereferences `message.acknowledge`, so the terminal step threw a TypeError + * inside Readable.from's pull loop — an uncaughtException that left the response hanging, never + * closed, whenever a finite async generator was streamed to completion over + * `Accept: text/event-stream`. The fix passes the terminal step through untransformed. * - * The shapes below vary where the terminal step falls relative to the yielded values, which is - * the axis the bug lived on: - * - EmptyGen (N=0) — terminal step is the FIRST step produced; the sharpest trigger - * - SingleGen (N=1) — minimal non-empty completing case - * - FiniteGen (N=5) — canonical completing case from the issue - * - LargeGen (N=3000) — same terminal-step path after a long stream (also a mild - * backpressure check) - * - ThrowGen (throws after 2 of 5) — contrast arm: a generator that rejects never reaches the - * terminal step at all, so it exercises the #1789 teardown path instead + * The shapes below vary where that terminal step falls relative to the yielded values, which is + * the axis the bug lived on: N=0 (terminal step is the first step produced, the sharpest trigger), + * N=1, N=5 (the issue's own case) and N=3000 (same path after a long stream). ThrowGen is the + * contrast arm — a generator that rejects never reaches the terminal step at all, so it exercises + * the #1789 teardown path instead. * - * Related but distinct suites, none of which cover the completion shapes above: - * - `sse-throw-midstream.test.ts` (#1789 / QA-559) is the dedicated anchor for a generator that - * *throws*, over SSE on Node only. Its `CleanGen` control overlaps FiniteGen; ThrowGen here is - * the contrast arm, not the anchor. - * - `stream-error-contract.test.ts` (QA-890) pins the stream-*error* contract across SSE, NDJSON - * and iterable-REST on raw socket bytes. - * - * Every request is bounded by an AbortController, so a regressed hang fails this suite - * deterministically instead of wedging the run. After the streaming cases a plain `/Probe/` - * request confirms the worker is still healthy, and `hdb.log` is scanned for newly-appeared - * `uncaughtException` lines — the bug's other signature. + * Neighbouring suites, neither of which covers those completion shapes: sse-throw-midstream.test.ts + * (#1789) is the dedicated anchor for a generator that *throws*; stream-error-contract.test.ts pins + * the stream-*error* contract across SSE, NDJSON and iterable-REST on raw socket bytes. * * Reproduction: * npm run test:integration -- "integrationTests/server/sse-finite-generator.test.ts" @@ -41,12 +28,12 @@ import { resolve } from 'node:path'; import { setTimeout as sleep } from 'node:timers/promises'; import { setupHarperWithFixture, teardownHarper, type ContextWithHarper } from '@harperfast/integration-testing'; import { + awaitFixtureReady, consumeSse, countUncaught, - getProbe, readLogSafe, - resolveHdbLogPath, uncaughtAfterSettle, + uncaughtLines, waitForProbe, } from '../utils/sseStream.ts'; // @ts-expect-error utils/client.mjs has no type declarations; runtime resolves fine @@ -79,6 +66,7 @@ suite( let restBase = ''; let authHeaders: Record = {}; let logPath = ''; + let uncaughtBaseline = 0; before(async () => { await setupHarperWithFixture(ctx, FIXTURE_PATH, { @@ -88,23 +76,7 @@ suite( client = createApiClient(ctx.harper); restBase = client.restURL; authHeaders = { Authorization: client.headers.Authorization as string }; - logPath = resolveHdbLogPath(ctx.harper as any); - - const deadline = Date.now() + 30_000; - let ready = false; - while (Date.now() < deadline) { - try { - const probe = await getProbe(restBase, authHeaders); - if (probe?.ok !== undefined) { - ready = true; - break; - } - } catch { - /* not ready */ - } - await sleep(250); - } - ok(ready, 'Probe route did not become ready within 30 seconds'); + ({ logPath, uncaughtBaseline } = await awaitFixtureReady(ctx.harper as any, restBase, authHeaders)); }); after(async () => { @@ -112,11 +84,12 @@ suite( }); async function assertCompletes(path: string, expectedEvents: number) { + const uncaughtBefore = countUncaught(readLogSafe(logPath)); const r = await consumeSse(`${restBase}${path}`, authHeaders, 15_000); ok(r.status >= 200 && r.status < 300, `expected 2xx, got ${r.status}`); ok( !r.aborted, - `must not hit the AbortController timeout — a timeout here means the #1628 hang regressed. raw:\n${r.raw}` + `${path} must not hit the AbortController timeout — a timeout here means the #1628 hang regressed. raw:\n${r.raw}` ); strictEqual( r.events.length, @@ -128,19 +101,18 @@ suite( 'end', `a generator streamed to completion must close via a clean 'end', got terminatedBy=${r.terminatedBy} errored=${r.errored?.message ?? null}` ); + strictEqual( + (await uncaughtAfterSettle(logPath)) - uncaughtBefore, + 0, + `no NEW uncaughtException should be logged for ${path}` + ); return r; } // ── 1: empty generator (N=0) — the terminal step is the very first step ────────────── test('1: EmptyGen (N=0) over SSE — zero events, response still closes cleanly', { timeout: 20_000 }, async () => { - const uncaughtBefore = countUncaught(readLogSafe(logPath)); await assertCompletes('/EmptyGen/', 0); - strictEqual( - (await uncaughtAfterSettle(logPath)) - uncaughtBefore, - 0, - 'no NEW uncaughtException should be logged when the terminal step is the only step' - ); }); // ── 2: single-event generator (N=1) ────────────────────────────────────────────────── @@ -156,16 +128,10 @@ suite( '3: FiniteGen (N=5) over SSE — all 5 events arrive and the response closes cleanly', { timeout: 20_000 }, async () => { - const uncaughtBefore = countUncaught(readLogSafe(logPath)); const r = await assertCompletes('/FiniteGen/', 5); for (let i = 0; i < 5; i++) { ok(r.events[i].includes(`"n":${i}`), `expected event ${i} to contain n=${i}, got: ${r.events[i]}`); } - strictEqual( - (await uncaughtAfterSettle(logPath)) - uncaughtBefore, - 0, - 'no NEW uncaughtException should be logged for a completing finite SSE generator' - ); } ); @@ -195,8 +161,8 @@ suite( ); // A rejecting generator never reaches the terminal `done` step #1632 fixed; it exits via - // the #1789 pipeline() teardown, which destroys the response abruptly rather than ending - // it cleanly. Both are bounded — that is what is asserted here. The full throw contract + // the #1789 pipeline() teardown, which destroys the response instead of ending it + // cleanly. Both are bounded — that is what is asserted here. The full throw contract // lives in sse-throw-midstream.test.ts. ok(!r.aborted, `must not hit the AbortController timeout — the response never terminated. raw:\n${r.raw}`); ok(r.terminatedBy !== null, 'response must terminate via end/error/close, not hang indefinitely'); @@ -215,13 +181,12 @@ suite( } ); - // ── Z: liveness canary + final uncaughtException sweep ────────────────────────────── + // ── Z: liveness canary + whole-suite uncaughtException sweep ──────────────────────── test( 'Z: liveness canary — worker survived every streaming shape above, no new uncaughtException', { timeout: 30_000 }, async () => { - const logBefore = readLogSafe(logPath); const probe = await waitForProbe(restBase, authHeaders, (snapshot) => [snapshot.finite, snapshot.empty, snapshot.single, snapshot.throwGen, snapshot.large].every( (counter) => counter.opened >= 1 && counter.opened === counter.closed @@ -240,18 +205,18 @@ suite( strictEqual(counter.closed, counter.opened, `${name} generator was not closed`); } - await sleep(1_000); // same non-event settle as the per-case sweeps above - const newLines = readLogSafe(logPath).slice(logBefore.length); - const newUncaught = countUncaught(newLines); - if (newUncaught > 0) { - console.log( - `[QA-537][Z] NEW uncaughtException lines found:\n${newLines - .split('\n') - .filter((line) => line.includes('uncaughtException')) - .join('\n')}` - ); + // Measured against the baseline taken in before(), so a case whose own delta check was + // outrun by the log flush is still caught here. + await sleep(1_000); + const offenders = uncaughtLines(readLogSafe(logPath)); + if (offenders.length > uncaughtBaseline) { + console.log(`[QA-537][Z] NEW uncaughtException lines:\n${offenders.slice(uncaughtBaseline).join('\n')}`); } - strictEqual(newUncaught, 0, 'no uncaughtException should appear while the worker is drained and probed'); + strictEqual( + offenders.length - uncaughtBaseline, + 0, + 'no uncaughtException should have appeared anywhere across the suite' + ); } ); } diff --git a/integrationTests/server/sse-finite-generator/resources.js b/integrationTests/server/sse-finite-generator/resources.js index d53d62496b..cdcda74bd9 100644 --- a/integrationTests/server/sse-finite-generator/resources.js +++ b/integrationTests/server/sse-finite-generator/resources.js @@ -1,25 +1,8 @@ -// QA-537 — regression verify for #1628 ("Fix SSE hang on finite generator streamed to -// completion", PR #1632, commit 69c8c89a9). +// Fixture for sse-finite-generator.test.ts (QA-537 / #1628) — finite async generators streamed to +// completion over SSE, plus a rejecting contrast arm. The test file carries the bug recap. // -// Bug recap: transformIterable (server/serverHelpers/contentTypes.ts) used to call the SSE -// `serialize` transform on the generator's TERMINAL `{ value: undefined, done: true }` step -// too. serialize()'s first line dereferences `message.acknowledge`, so `undefined.acknowledge` -// threw a TypeError inside Readable.from's pull loop — an uncaughtException that left the SSE -// HTTP response hanging (never closed) whenever a plain finite async generator was streamed to -// completion over `Accept: text/event-stream`. The fix guards transformIterable to pass the -// terminal `done` step through untransformed in both the sync and async branches. -// -// This fixture exercises multiple finite-generator shapes over SSE (dispatched via -// `resource.connect()`, which is what Harper's REST layer invokes for CONNECT/SSE requests — -// see server/REST.ts: `isSse` sets method to 'CONNECT', which calls `resource.connect(...)`): -// FiniteGen - canonical case: 5 events then a bare `return` (natural completion). -// EmptyGen - 0 events: generator returns immediately, hitting the terminal step first. -// SingleGen - exactly 1 event then completion. -// ThrowGen - yields 2 of an intended 5 events, then throws (partway failure, not the -// terminal-`done` code path; the full throw contract is anchored by the -// sse-throw-midstream fixture instead). -// LargeGen - 3000 events then completion (larger finite stream, same terminal-step path). -// Probe - readiness + per-resource open/close lifecycle counters, plain JSON. +// SSE requests reach these via `connect()`: server/REST.ts turns an `Accept: text/event-stream` +// GET into method CONNECT, which calls `resource.connect(...)`. const G = (globalThis.__QA537__ ??= { finite: { opened: 0, closed: 0 }, @@ -33,7 +16,7 @@ function sleep(ms) { return new Promise((r) => setTimeout(r, ms)); } -// GET /FiniteGen/ (Accept: text/event-stream) — canonical finite generator, N=5. +// GET /FiniteGen/ — canonical finite generator, N=5. export class FiniteGen extends Resource { static loadAsInstance = false; static async *connect(_target, _incomingMessages, _request) { @@ -49,9 +32,7 @@ export class FiniteGen extends Resource { } } -// GET /EmptyGen/ (Accept: text/event-stream) — 0 events; the terminal `done` step is the -// very FIRST step produced, so this isolates the terminal-step handling from any mid-stream -// yields at all. +// GET /EmptyGen/ — 0 events, so the terminal `done` step is the very FIRST step produced. export class EmptyGen extends Resource { static loadAsInstance = false; // eslint-disable-next-line require-yield @@ -65,7 +46,7 @@ export class EmptyGen extends Resource { } } -// GET /SingleGen/ (Accept: text/event-stream) — exactly 1 event then completion. +// GET /SingleGen/ — exactly 1 event then completion. export class SingleGen extends Resource { static loadAsInstance = false; static async *connect(_target, _incomingMessages, _request) { @@ -78,10 +59,8 @@ export class SingleGen extends Resource { } } -// GET /ThrowGen/ (Accept: text/event-stream) — yields 2 of an intended 5, then throws. -// This does NOT hit the fixed terminal-`done` code path (the iterator never reaches a -// `done:true` step — it rejects instead); it is the bounded-termination contrast arm next to -// the natural-completion cases. +// GET /ThrowGen/ — yields 2 of an intended 5, then throws. A rejecting iterator never reaches a +// `done:true` step, so this is the contrast arm rather than the fixed code path. export class ThrowGen extends Resource { static loadAsInstance = false; static async *connect(_target, _incomingMessages, _request) { @@ -98,8 +77,7 @@ export class ThrowGen extends Resource { } } -// GET /LargeGen/ (Accept: text/event-stream) — 3000 events then completion. Larger finite -// stream exercising the same terminal-step path at volume (also a mild backpressure check). +// GET /LargeGen/ — 3000 events then completion: the same terminal-step path at volume. export class LargeGen extends Resource { static loadAsInstance = false; static async *connect(_target, _incomingMessages, _request) { diff --git a/integrationTests/server/sse-finite-generator/schema.graphql b/integrationTests/server/sse-finite-generator/schema.graphql index 41998e9365..151a6c18ef 100644 --- a/integrationTests/server/sse-finite-generator/schema.graphql +++ b/integrationTests/server/sse-finite-generator/schema.graphql @@ -1,9 +1,5 @@ -# QA-537 — regression verify for #1628 ("Fix SSE hang on finite generator streamed to -# completion", merged via PR #1632 / commit 69c8c89a9). -# -# No table backing is needed for the generator Resources under test, but a jsResource -# fixture requires at least one @table/@export type to be a valid component, so this is a -# minimal placeholder. +# A jsResource fixture needs at least one @table/@export type to be a valid component; the +# generator Resources under test need no table backing. type Placeholder @table @export { id: ID @primaryKey } diff --git a/integrationTests/server/sse-throw-midstream.test.ts b/integrationTests/server/sse-throw-midstream.test.ts index 2ae5b7ad86..e79bb5eaf2 100644 --- a/integrationTests/server/sse-throw-midstream.test.ts +++ b/integrationTests/server/sse-throw-midstream.test.ts @@ -45,16 +45,15 @@ import { suite, test, before, after } from 'node:test'; import { ok, strictEqual } from 'node:assert'; import { resolve } from 'node:path'; -import { readFileSync } from 'node:fs'; import { setTimeout as sleep } from 'node:timers/promises'; import { setupHarperWithFixture, teardownHarper, type ContextWithHarper } from '@harperfast/integration-testing'; import { + awaitFixtureReady, consumeSse, countUncaught, - getProbe, readLogSafe, - resolveHdbLogPath, uncaughtAfterSettle, + uncaughtLines, waitForProbe, } from '../utils/sseStream.ts'; // @ts-expect-error utils/client.mjs has no type declarations; runtime resolves fine @@ -82,6 +81,7 @@ suite( let restBase = ''; let authHeaders: Record = {}; let logPath = ''; + let uncaughtBaseline = 0; before(async () => { await setupHarperWithFixture(ctx, FIXTURE_PATH, { @@ -91,24 +91,7 @@ suite( client = createApiClient(ctx.harper); restBase = client.restURL; authHeaders = { Authorization: client.headers.Authorization as string }; - logPath = resolveHdbLogPath(ctx.harper as any); - - const deadline = Date.now() + 30_000; - let ready = false; - while (Date.now() < deadline) { - try { - const p = await getProbe(restBase, authHeaders); - if (p?.ok !== undefined) { - ready = true; - break; - } - } catch { - /* not ready */ - } - await sleep(250); - } - ok(ready, 'Probe route did not become ready within 30 seconds'); - readFileSync(logPath, 'utf8'); + ({ logPath, uncaughtBaseline } = await awaitFixtureReady(ctx.harper as any, restBase, authHeaders)); }); after(async () => { @@ -220,7 +203,6 @@ suite( 'Z: liveness canary -- server survived all throw-mid-stream probes, no new uncaughtException', { timeout: 30_000 }, async () => { - const logBefore = readLogSafe(logPath); const p = await waitForProbe( restBase, authHeaders, @@ -241,19 +223,18 @@ suite( ); ok(p!.clean.opened >= 1 && p!.clean.closed >= 1, 'CleanGen should show a matched open/close pair'); - await sleep(1_000); // same non-event settle as the per-case sweeps above - const logAfter = readLogSafe(logPath); - const newLines = logAfter.slice(logBefore.length); - const newUncaught = newLines.split('\n').filter((l) => l.includes('uncaughtException')).length; - if (newUncaught > 0) { - console.log( - `[QA-559][Z] NEW uncaughtException lines found:\n${newLines - .split('\n') - .filter((l) => l.includes('uncaughtException')) - .join('\n')}` - ); + // Measured against the baseline taken in before(), so a case whose own delta check was + // outrun by the log flush is still caught here. + await sleep(1_000); + const offenders = uncaughtLines(readLogSafe(logPath)); + if (offenders.length > uncaughtBaseline) { + console.log(`[QA-559][Z] NEW uncaughtException lines:\n${offenders.slice(uncaughtBaseline).join('\n')}`); } - strictEqual(newUncaught, 0, 'no uncaughtException should have appeared anywhere across the whole suite'); + strictEqual( + offenders.length - uncaughtBaseline, + 0, + 'no uncaughtException should have appeared anywhere across the whole suite' + ); } ); } diff --git a/integrationTests/utils/sseStream.ts b/integrationTests/utils/sseStream.ts index a1c7e3dc03..e6c0773cd9 100644 --- a/integrationTests/utils/sseStream.ts +++ b/integrationTests/utils/sseStream.ts @@ -1,20 +1,11 @@ /** * Shared helpers for the SSE streaming regression suites in `integrationTests/server/` - * (`sse-finite-generator.test.ts`, `sse-throw-midstream.test.ts`). + * (`sse-finite-generator.test.ts`, `sse-throw-midstream.test.ts`): a bounded SSE consumer, a + * `/Probe/` reader for the lifecycle counters those fixtures keep, and an `hdb.log` + * uncaughtException counter — the bugs these suites anchor surfaced as an uncaught throw inside + * the worker rather than as a bad response. * - * These suites all need the same three things, and each grew its own copy before this module - * existed: - * - * - an SSE consumer bounded by an AbortController, so a regressed hang fails the test in - * bounded time instead of wedging the whole run; - * - a `/Probe/` reader for the per-resource open/close lifecycle counters the fixtures keep, - * with a polling form for the counters that settle asynchronously after a response ends; - * - an `hdb.log` reader that counts `uncaughtException` lines, since the bugs these suites - * anchor (#1628, #1789) manifested as an uncaught throw inside the worker rather than as a - * bad response. - * - * Nothing here is Harper-specific beyond the `/Probe/` path convention and the log layout; the - * raw-socket capture used by `stream-error-contract.test.ts` is a deliberately different + * The raw-socket capture in `stream-error-contract.test.ts` is a deliberately different * technique (it inspects chunk framing and the exact close mechanism) and is not shared. */ import { readFileSync } from 'node:fs'; @@ -28,11 +19,10 @@ import { URL } from 'node:url'; export interface SseResult { status: number; raw: string; - /** parsed `data: ...` payloads, in order */ events: string[]; - /** which response event resolved the request, or null if our own timeout fired first */ + /** which response event resolved the request; null when our own timeout fired instead */ terminatedBy: 'end' | 'error' | 'close' | null; - /** did our own timeout abort fire (i.e. the response hung)? */ + /** did our own timeout abort fire — i.e. the response hung? */ aborted: boolean; errored: Error | null; elapsedMs: number; @@ -51,6 +41,10 @@ function parseEvents(raw: string): string[] { * Resolves on whichever of 'end' / 'error' / 'close' fires first: the pipeline()-based teardown * introduced by #1789 closes the response abruptly rather than via a clean 'end' when the source * generator rejects, so waiting only for 'end' would read a correct abrupt close as a hang. + * + * `aborted` is decided by our own timer rather than by which event won the race to settle. An + * abort destroys the response, so a hung stream emits 'close' too — attributing the settle to + * that event would report a genuine hang as a bounded termination. */ export function consumeSse( urlStr: string, @@ -82,7 +76,8 @@ export function consumeSse( if (settled) return; settled = true; clearTimeout(timer); - result.terminatedBy = terminatedBy; + result.aborted = timedOut; + result.terminatedBy = timedOut ? null : terminatedBy; result.errored = err ?? null; result.elapsedMs = Date.now() - start; result.events = parseEvents(result.raw); @@ -98,8 +93,8 @@ export function consumeSse( } as any, (res) => { result.status = res.statusCode ?? 0; - // Decode incrementally: a bare d.toString('utf8') per chunk corrupts any multi-byte - // character that straddles a TCP chunk boundary into U+FFFD. + // A bare d.toString('utf8') per chunk corrupts any multi-byte character that straddles + // a TCP chunk boundary into U+FFFD. const decoder = new StringDecoder('utf8'); res.on('data', (d: Buffer) => { result.raw += decoder.write(d); @@ -113,23 +108,13 @@ export function consumeSse( } ); req.on('error', (e: any) => { - if (timedOut || e?.name === 'AbortError') { - if (settled) return; - settled = true; - clearTimeout(timer); - result.aborted = true; - result.elapsedMs = Date.now() - start; - result.events = parseEvents(result.raw); - resolvePromise(result); - } else { - finish('error', e); - } + if (timedOut || e?.name === 'AbortError') finish(null); + else finish('error', e); }); req.end(); }); } -/** Read the fixture's `/Probe/` resource: readiness plus its open/close lifecycle counters. */ export function getProbe(restBase: string, authHeaders: Record, timeoutMs = 3_000): Promise { const url = new URL(`${restBase}/Probe/`); const lib = url.protocol === 'https:' ? https : http; @@ -174,16 +159,36 @@ export async function waitForProbe( const deadline = Date.now() + timeoutMs; let probe: T | null = null; while (Date.now() < deadline) { - probe = await getProbe(restBase, authHeaders).catch(() => null); + probe = await getProbe(restBase, authHeaders, Math.max(1, deadline - Date.now())).catch(() => null); if (probe && predicate(probe)) return probe; await sleep(50); } return probe; } -/** Where the test instance writes hdb.log, across framework versions that expose either field. */ -export function resolveHdbLogPath(harper: { logDir?: string; dataRootDir?: string }): string { - return harper.logDir ? join(harper.logDir, 'hdb.log') : join(harper.dataRootDir as string, 'log', 'hdb.log'); +/** + * Wait for a freshly installed SSE fixture's `/Probe/` route, and take the suite's baseline + * uncaughtException count. + * + * The baseline is read with `readFileSync`, not `readLogSafe`: an hdb.log this suite cannot read + * (a changed harness layout, a log not yet created) would otherwise turn every uncaughtException + * assertion downstream into a vacuous pass, since a missing file counts zero both before and + * after. Failing setup is the only way that stays visible. + */ +export async function awaitFixtureReady( + harper: { logDir?: string; dataRootDir?: string }, + restBase: string, + authHeaders: Record, + timeoutMs = 30_000 +): Promise<{ logPath: string; uncaughtBaseline: number }> { + const logPath = harper.logDir ? join(harper.logDir, 'hdb.log') : join(harper.dataRootDir as string, 'log', 'hdb.log'); + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + const probe = await getProbe<{ ok?: boolean }>(restBase, authHeaders).catch(() => null); + if (probe?.ok !== undefined) return { logPath, uncaughtBaseline: countUncaught(readFileSync(logPath, 'utf8')) }; + await sleep(250); + } + throw new Error(`Probe route did not become ready within ${timeoutMs}ms at ${restBase}/Probe/`); } export function readLogSafe(logPath: string): string { @@ -194,8 +199,12 @@ export function readLogSafe(logPath: string): string { } } +export function uncaughtLines(log: string): string[] { + return log.split('\n').filter((line) => line.includes('uncaughtException')); +} + export function countUncaught(log: string): number { - return log.split('\n').filter((line) => line.includes('uncaughtException')).length; + return uncaughtLines(log).length; } /** From 2e46d3d1a6abadfbd912920ec551457a3e677750 Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Mon, 31 Aug 2026 18:20:57 -0600 Subject: [PATCH 3/6] Assert the delivered event sequence, not just its length The completion arms compared counts and endpoints, so a stream that reordered or duplicated interior frames -- or, for ThrowGen, delivered no pre-error prefix at all -- still passed. Each arm now compares the full `n` sequence; ThrowGen keeps a 1-2 upper/lower bound because the abrupt destroy can drop the last chunk before it drains. Co-Authored-By: Claude Opus --- .../server/sse-finite-generator.test.ts | 33 ++++++++++++------- 1 file changed, 21 insertions(+), 12 deletions(-) diff --git a/integrationTests/server/sse-finite-generator.test.ts b/integrationTests/server/sse-finite-generator.test.ts index 6cde31f61b..b354d3b66f 100644 --- a/integrationTests/server/sse-finite-generator.test.ts +++ b/integrationTests/server/sse-finite-generator.test.ts @@ -23,7 +23,7 @@ * npm run test:integration -- "integrationTests/server/sse-finite-generator.test.ts" */ import { suite, test, before, after } from 'node:test'; -import { ok, strictEqual } from 'node:assert'; +import { deepStrictEqual, ok, strictEqual } from 'node:assert'; import { resolve } from 'node:path'; import { setTimeout as sleep } from 'node:timers/promises'; import { setupHarperWithFixture, teardownHarper, type ContextWithHarper } from '@harperfast/integration-testing'; @@ -83,6 +83,11 @@ suite( await teardownHarper(ctx); }); + /** The `n` carried by each delivered event, in arrival order. */ + function eventNumbers(result: { events: string[] }): number[] { + return result.events.map((event) => JSON.parse(event).n); + } + async function assertCompletes(path: string, expectedEvents: number) { const uncaughtBefore = countUncaught(readLogSafe(logPath)); const r = await consumeSse(`${restBase}${path}`, authHeaders, 15_000); @@ -119,7 +124,7 @@ suite( test('2: SingleGen (N=1) over SSE — exactly 1 event, response closes cleanly', { timeout: 20_000 }, async () => { const r = await assertCompletes('/SingleGen/', 1); - ok(r.events[0].includes('"n":0'), `expected the single event to contain n=0, got: ${r.events[0]}`); + deepStrictEqual(eventNumbers(r), [0]); }); // ── 3: canonical finite generator (N=5) ────────────────────────────────────────────── @@ -129,9 +134,7 @@ suite( { timeout: 20_000 }, async () => { const r = await assertCompletes('/FiniteGen/', 5); - for (let i = 0; i < 5; i++) { - ok(r.events[i].includes(`"n":${i}`), `expected event ${i} to contain n=${i}, got: ${r.events[i]}`); - } + deepStrictEqual(eventNumbers(r), [0, 1, 2, 3, 4]); } ); @@ -142,8 +145,12 @@ suite( { timeout: 20_000 }, async () => { const r = await assertCompletes('/LargeGen/', 3000); - ok(r.events[0].includes('"n":0'), `expected first event n=0, got: ${r.events[0]}`); - ok(r.events[2999].includes('"n":2999'), `expected last event n=2999, got: ${r.events[2999]}`); + // Every index, not just the endpoints: a reordering or duplication in the middle of a long + // stream preserves both the count and the ends. + deepStrictEqual( + eventNumbers(r), + Array.from({ length: 3000 }, (_, i) => i) + ); } ); @@ -166,13 +173,15 @@ suite( // lives in sse-throw-midstream.test.ts. ok(!r.aborted, `must not hit the AbortController timeout — the response never terminated. raw:\n${r.raw}`); ok(r.terminatedBy !== null, 'response must terminate via end/error/close, not hang indefinitely'); + // The generator yields 2 before it throws. The upper bound is the contract; the lower bound + // is 1 rather than 2 because the abrupt destroy can drop the last chunk before it drains, + // and a prefix of zero events would mean the pre-error events were never delivered at all. + const delivered = eventNumbers(r); ok( - r.events.length <= 2, - `expected at most the 2 events yielded before the throw, got ${r.events.length}. raw:\n${r.raw}` + delivered.length >= 1 && delivered.length <= 2, + `expected a 1-2 event prefix before the throw, got ${delivered.length}. raw:\n${r.raw}` ); - for (let i = 0; i < r.events.length; i++) { - ok(r.events[i].includes(`"n":${i}`), `expected event ${i} to contain n=${i}, got: ${r.events[i]}`); - } + deepStrictEqual(delivered, [0, 1].slice(0, delivered.length)); strictEqual( (await uncaughtAfterSettle(logPath)) - uncaughtBefore, 0, From 587fe4fa344456686d00f92e16ef9b3f0c82cd15 Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Mon, 31 Aug 2026 18:25:52 -0600 Subject: [PATCH 4/6] Harden the shared SSE helpers against setup races and swallowed causes - awaitFixtureReady() polls for hdb.log instead of reading it once: the HTTP port can answer before the log writer has created the file, which would have turned the readability backstop itself into a deterministic ENOENT in before(). - Both final sweeps read the log strictly, so a log that disappears mid-run fails loudly rather than counting zero against a zero baseline. - consumeSse() only attributes a settle to its own timeout when that timer actually fired; any other AbortError now surfaces as an error with its cause instead of an empty result. - waitForProbe() treats a throwing predicate as "not satisfied yet", so a probe error payload fails on the caller's assertion rather than as a TypeError inside the poll. - The 2xx assertion reports the transport error alongside the status. Co-Authored-By: Claude Opus --- .../server/sse-finite-generator.test.ts | 5 +-- .../server/sse-throw-midstream.test.ts | 3 +- integrationTests/utils/sseStream.ts | 32 ++++++++++++++++--- 3 files changed, 33 insertions(+), 7 deletions(-) diff --git a/integrationTests/server/sse-finite-generator.test.ts b/integrationTests/server/sse-finite-generator.test.ts index b354d3b66f..420160fd3e 100644 --- a/integrationTests/server/sse-finite-generator.test.ts +++ b/integrationTests/server/sse-finite-generator.test.ts @@ -31,6 +31,7 @@ import { awaitFixtureReady, consumeSse, countUncaught, + readLogOrThrow, readLogSafe, uncaughtAfterSettle, uncaughtLines, @@ -91,7 +92,7 @@ suite( async function assertCompletes(path: string, expectedEvents: number) { const uncaughtBefore = countUncaught(readLogSafe(logPath)); const r = await consumeSse(`${restBase}${path}`, authHeaders, 15_000); - ok(r.status >= 200 && r.status < 300, `expected 2xx, got ${r.status}`); + ok(r.status >= 200 && r.status < 300, `expected 2xx, got ${r.status} (errored=${r.errored?.message ?? null})`); ok( !r.aborted, `${path} must not hit the AbortController timeout — a timeout here means the #1628 hang regressed. raw:\n${r.raw}` @@ -217,7 +218,7 @@ suite( // Measured against the baseline taken in before(), so a case whose own delta check was // outrun by the log flush is still caught here. await sleep(1_000); - const offenders = uncaughtLines(readLogSafe(logPath)); + const offenders = uncaughtLines(readLogOrThrow(logPath)); if (offenders.length > uncaughtBaseline) { console.log(`[QA-537][Z] NEW uncaughtException lines:\n${offenders.slice(uncaughtBaseline).join('\n')}`); } diff --git a/integrationTests/server/sse-throw-midstream.test.ts b/integrationTests/server/sse-throw-midstream.test.ts index e79bb5eaf2..7f30014057 100644 --- a/integrationTests/server/sse-throw-midstream.test.ts +++ b/integrationTests/server/sse-throw-midstream.test.ts @@ -51,6 +51,7 @@ import { awaitFixtureReady, consumeSse, countUncaught, + readLogOrThrow, readLogSafe, uncaughtAfterSettle, uncaughtLines, @@ -226,7 +227,7 @@ suite( // Measured against the baseline taken in before(), so a case whose own delta check was // outrun by the log flush is still caught here. await sleep(1_000); - const offenders = uncaughtLines(readLogSafe(logPath)); + const offenders = uncaughtLines(readLogOrThrow(logPath)); if (offenders.length > uncaughtBaseline) { console.log(`[QA-559][Z] NEW uncaughtException lines:\n${offenders.slice(uncaughtBaseline).join('\n')}`); } diff --git a/integrationTests/utils/sseStream.ts b/integrationTests/utils/sseStream.ts index e6c0773cd9..cf9fa2ebd8 100644 --- a/integrationTests/utils/sseStream.ts +++ b/integrationTests/utils/sseStream.ts @@ -108,7 +108,7 @@ export function consumeSse( } ); req.on('error', (e: any) => { - if (timedOut || e?.name === 'AbortError') finish(null); + if (timedOut) finish(null); else finish('error', e); }); req.end(); @@ -160,7 +160,13 @@ export async function waitForProbe( let probe: T | null = null; while (Date.now() < deadline) { probe = await getProbe(restBase, authHeaders, Math.max(1, deadline - Date.now())).catch(() => null); - if (probe && predicate(probe)) return probe; + // A predicate reading counter fields off an error payload throws; that is "not satisfied + // yet", and the caller asserts on the snapshot this returns. + try { + if (probe && predicate(probe)) return probe; + } catch { + /* not satisfied */ + } await sleep(50); } return probe; @@ -183,12 +189,30 @@ export async function awaitFixtureReady( ): Promise<{ logPath: string; uncaughtBaseline: number }> { const logPath = harper.logDir ? join(harper.logDir, 'hdb.log') : join(harper.dataRootDir as string, 'log', 'hdb.log'); const deadline = Date.now() + timeoutMs; + let ready = false; while (Date.now() < deadline) { const probe = await getProbe<{ ok?: boolean }>(restBase, authHeaders).catch(() => null); - if (probe?.ok !== undefined) return { logPath, uncaughtBaseline: countUncaught(readFileSync(logPath, 'utf8')) }; + if (probe?.ok !== undefined) { + ready = true; + break; + } await sleep(250); } - throw new Error(`Probe route did not become ready within ${timeoutMs}ms at ${restBase}/Probe/`); + if (!ready) throw new Error(`Probe route did not become ready within ${timeoutMs}ms at ${restBase}/Probe/`); + // Polled, not read once: the HTTP port can answer before the log writer has created the file. + while (Date.now() < deadline) { + try { + return { logPath, uncaughtBaseline: countUncaught(readFileSync(logPath, 'utf8')) }; + } catch { + await sleep(100); + } + } + throw new Error(`hdb.log never became readable at ${logPath} — uncaughtException checks would pass vacuously`); +} + +/** For the end-of-suite sweep: a log that vanished mid-run must fail, not silently count zero. */ +export function readLogOrThrow(logPath: string): string { + return readFileSync(logPath, 'utf8'); } export function readLogSafe(logPath: string): string { From 58ee739323e21ed4fb2f40a86dbb7af5fb4d95cd Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Mon, 31 Aug 2026 18:29:33 -0600 Subject: [PATCH 5/6] Flush the decoder on abrupt closes, guard the probe shape, trim narration - consumeSse() flushes the StringDecoder on every settle path, not just a clean 'end': an abrupt close would otherwise discard a multi-byte character the decoder was still holding. - Both liveness canaries assert the probe payload actually carries lifecycle counters, so a Harper error payload fails readably instead of as a TypeError on a missing field. - Condensed the file header and the fixture's per-route comments. Co-Authored-By: Claude Opus --- .../server/sse-finite-generator.test.ts | 37 +++++++++---------- .../server/sse-finite-generator/resources.js | 12 +++--- .../server/sse-throw-midstream.test.ts | 1 + integrationTests/utils/sseStream.ts | 10 +++-- 4 files changed, 29 insertions(+), 31 deletions(-) diff --git a/integrationTests/server/sse-finite-generator.test.ts b/integrationTests/server/sse-finite-generator.test.ts index 420160fd3e..fe6099be74 100644 --- a/integrationTests/server/sse-finite-generator.test.ts +++ b/integrationTests/server/sse-finite-generator.test.ts @@ -1,23 +1,17 @@ /** * QA-537 — regression verify for #1628 "SSE hang on a finite generator streamed to completion", - * fixed by PR #1632. + * fixed by PR #1632. `transformIterable` (server/serverHelpers/contentTypes.ts) applied the SSE + * `serialize` transform to a generator's terminal `{ value: undefined, done: true }` step, which + * threw inside Readable.from's pull loop and left the response hanging, never closed; the fix + * passes that step through untransformed. * - * `transformIterable` (server/serverHelpers/contentTypes.ts) applied the SSE `serialize` transform - * to the generator's terminal `{ value: undefined, done: true }` step as well as to its yielded - * values. `serialize()` dereferences `message.acknowledge`, so the terminal step threw a TypeError - * inside Readable.from's pull loop — an uncaughtException that left the response hanging, never - * closed, whenever a finite async generator was streamed to completion over - * `Accept: text/event-stream`. The fix passes the terminal step through untransformed. + * The arms vary where the terminal step falls relative to the yielded values — N=0 (it is the + * first step produced, the sharpest trigger), N=1, N=5, N=3000 — plus ThrowGen, whose rejection + * never reaches the terminal step and so exercises #1789's teardown instead. * - * The shapes below vary where that terminal step falls relative to the yielded values, which is - * the axis the bug lived on: N=0 (terminal step is the first step produced, the sharpest trigger), - * N=1, N=5 (the issue's own case) and N=3000 (same path after a long stream). ThrowGen is the - * contrast arm — a generator that rejects never reaches the terminal step at all, so it exercises - * the #1789 teardown path instead. - * - * Neighbouring suites, neither of which covers those completion shapes: sse-throw-midstream.test.ts - * (#1789) is the dedicated anchor for a generator that *throws*; stream-error-contract.test.ts pins - * the stream-*error* contract across SSE, NDJSON and iterable-REST on raw socket bytes. + * sse-throw-midstream.test.ts (#1789) anchors the throw path; stream-error-contract.test.ts pins + * the stream-error contract across SSE, NDJSON and iterable-REST on raw socket bytes. Neither + * covers the completion shapes above. * * Reproduction: * npm run test:integration -- "integrationTests/server/sse-finite-generator.test.ts" @@ -168,10 +162,9 @@ suite( `[QA-537][5] ThrowGen: status=${r.status} events=${r.events.length} terminatedBy=${r.terminatedBy} aborted=${r.aborted} errored=${r.errored?.message ?? null} elapsedMs=${r.elapsedMs}` ); - // A rejecting generator never reaches the terminal `done` step #1632 fixed; it exits via - // the #1789 pipeline() teardown, which destroys the response instead of ending it - // cleanly. Both are bounded — that is what is asserted here. The full throw contract - // lives in sse-throw-midstream.test.ts. + // A rejecting generator exits via #1789's pipeline() teardown, which destroys the response + // instead of ending it cleanly. Only boundedness is asserted here; the throw contract + // itself is sse-throw-midstream.test.ts's. ok(!r.aborted, `must not hit the AbortController timeout — the response never terminated. raw:\n${r.raw}`); ok(r.terminatedBy !== null, 'response must terminate via end/error/close, not hang indefinitely'); // The generator yields 2 before it throws. The upper bound is the contract; the lower bound @@ -204,6 +197,10 @@ suite( ); console.log(`[QA-537][Z] liveness probe: ${probe ? JSON.stringify(probe) : 'DEAD'}`); ok(probe !== null, 'Harper must still respond to Probe/ after all streaming cases'); + ok( + probe!.finite && probe!.empty && probe!.single && probe!.throwGen && probe!.large, + `Probe/ returned no lifecycle counters: ${JSON.stringify(probe)}` + ); for (const [name, counter] of Object.entries({ FiniteGen: probe!.finite, EmptyGen: probe!.empty, diff --git a/integrationTests/server/sse-finite-generator/resources.js b/integrationTests/server/sse-finite-generator/resources.js index cdcda74bd9..41271ec854 100644 --- a/integrationTests/server/sse-finite-generator/resources.js +++ b/integrationTests/server/sse-finite-generator/resources.js @@ -16,7 +16,6 @@ function sleep(ms) { return new Promise((r) => setTimeout(r, ms)); } -// GET /FiniteGen/ — canonical finite generator, N=5. export class FiniteGen extends Resource { static loadAsInstance = false; static async *connect(_target, _incomingMessages, _request) { @@ -32,7 +31,7 @@ export class FiniteGen extends Resource { } } -// GET /EmptyGen/ — 0 events, so the terminal `done` step is the very FIRST step produced. +// 0 events, so the terminal `done` step is the very first step produced. export class EmptyGen extends Resource { static loadAsInstance = false; // eslint-disable-next-line require-yield @@ -46,7 +45,6 @@ export class EmptyGen extends Resource { } } -// GET /SingleGen/ — exactly 1 event then completion. export class SingleGen extends Resource { static loadAsInstance = false; static async *connect(_target, _incomingMessages, _request) { @@ -59,8 +57,8 @@ export class SingleGen extends Resource { } } -// GET /ThrowGen/ — yields 2 of an intended 5, then throws. A rejecting iterator never reaches a -// `done:true` step, so this is the contrast arm rather than the fixed code path. +// A rejecting iterator never reaches a `done:true` step, so this is the contrast arm rather than +// the fixed code path. export class ThrowGen extends Resource { static loadAsInstance = false; static async *connect(_target, _incomingMessages, _request) { @@ -77,7 +75,7 @@ export class ThrowGen extends Resource { } } -// GET /LargeGen/ — 3000 events then completion: the same terminal-step path at volume. +// The same terminal-step path at volume. export class LargeGen extends Resource { static loadAsInstance = false; static async *connect(_target, _incomingMessages, _request) { @@ -92,7 +90,7 @@ export class LargeGen extends Resource { } } -// GET /Probe/ — readiness + lifecycle-counter snapshot (plain JSON, not SSE). +// Readiness + lifecycle-counter snapshot, served as plain JSON rather than SSE. export class Probe extends Resource { static loadAsInstance = false; static async get() { diff --git a/integrationTests/server/sse-throw-midstream.test.ts b/integrationTests/server/sse-throw-midstream.test.ts index 7f30014057..c0fb226a92 100644 --- a/integrationTests/server/sse-throw-midstream.test.ts +++ b/integrationTests/server/sse-throw-midstream.test.ts @@ -214,6 +214,7 @@ suite( p !== null, 'Harper must still respond to Probe/ after all throw-mid-stream cases (worker not crashed/wedged)' ); + ok(p!.throwFirst && p!.throwMid && p!.clean, `Probe/ returned no lifecycle counters: ${JSON.stringify(p)}`); ok( p!.throwFirst.opened >= 1 && p!.throwFirst.closed >= 1, 'ThrowFirst should show a matched open/close pair (generator finally ran)' diff --git a/integrationTests/utils/sseStream.ts b/integrationTests/utils/sseStream.ts index cf9fa2ebd8..f291ce0c73 100644 --- a/integrationTests/utils/sseStream.ts +++ b/integrationTests/utils/sseStream.ts @@ -72,10 +72,14 @@ export function consumeSse( elapsedMs: 0, }; let settled = false; + let flushDecoder: (() => string) | null = null; const finish = (terminatedBy: SseResult['terminatedBy'], err?: Error) => { if (settled) return; settled = true; clearTimeout(timer); + // An abrupt close settles through 'error'/'close', so flushing only on 'end' would drop a + // multi-byte character the decoder was still holding. + if (flushDecoder) result.raw += flushDecoder(); result.aborted = timedOut; result.terminatedBy = timedOut ? null : terminatedBy; result.errored = err ?? null; @@ -96,13 +100,11 @@ export function consumeSse( // A bare d.toString('utf8') per chunk corrupts any multi-byte character that straddles // a TCP chunk boundary into U+FFFD. const decoder = new StringDecoder('utf8'); + flushDecoder = () => decoder.end(); res.on('data', (d: Buffer) => { result.raw += decoder.write(d); }); - res.on('end', () => { - result.raw += decoder.end(); - finish('end'); - }); + res.on('end', () => finish('end')); res.on('error', (e: Error) => finish('error', e)); res.on('close', () => finish('close')); } From df25a3048ca2a9add059c27c53367865d71a4efa Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Mon, 31 Aug 2026 18:34:28 -0600 Subject: [PATCH 6/6] Fail readably when a probe payload carries no lifecycle counters The per-case cleanup checks dereferenced the polled snapshot directly, so a Harper error payload surfaced as a TypeError on a missing field rather than as the assertion that was actually failing. Co-Authored-By: Claude Opus --- integrationTests/server/sse-throw-midstream.test.ts | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/integrationTests/server/sse-throw-midstream.test.ts b/integrationTests/server/sse-throw-midstream.test.ts index c0fb226a92..a1af2b4d9e 100644 --- a/integrationTests/server/sse-throw-midstream.test.ts +++ b/integrationTests/server/sse-throw-midstream.test.ts @@ -129,7 +129,10 @@ suite( authHeaders, (snapshot) => snapshot.throwFirst.closed >= 1 ); - ok(probe && probe.throwFirst.closed >= 1, 'ThrowFirst generator should have completed cleanup'); + ok( + (probe?.throwFirst?.closed ?? 0) >= 1, + `ThrowFirst generator should have completed cleanup; probe: ${JSON.stringify(probe)}` + ); const uncaughtAfter = await uncaughtAfterSettle(logPath); strictEqual( uncaughtAfter - uncaughtBefore, @@ -167,7 +170,10 @@ suite( } const probe = await waitForProbe(restBase, authHeaders, (snapshot) => snapshot.throwMid.closed >= 1); - ok(probe && probe.throwMid.closed >= 1, 'ThrowMid generator should have completed cleanup'); + ok( + (probe?.throwMid?.closed ?? 0) >= 1, + `ThrowMid generator should have completed cleanup; probe: ${JSON.stringify(probe)}` + ); const uncaughtAfter = await uncaughtAfterSettle(logPath); strictEqual( uncaughtAfter - uncaughtBefore,