diff --git a/reference-implementation/server/index.ts b/reference-implementation/server/index.ts index d58edd54..714b1fca 100644 --- a/reference-implementation/server/index.ts +++ b/reference-implementation/server/index.ts @@ -777,6 +777,8 @@ interface ServerOpts { nekoProxyAllowedHosts?: readonly string[] | null; nekoProxyAutoLogin?: boolean; nekoWindowSettleProbe?: ((url: string) => Promise) | null; + /** Receives each scheduled validation promise before it starts. */ + onManualUploadValidationTask?: (task: Promise) => void; onScheduleMutation?: (() => void) | null; ownerAuthForceSecureCookies?: boolean; ownerAuthPassword?: string; @@ -6228,6 +6230,7 @@ export function buildAsApp(opts: ServerOpts = {}) { getOwnerSubjectId, handleError, importBaseDir, + onManualUploadValidationTask: opts.onManualUploadValidationTask, pdppError, requireOwnerSession: ownerAuth.requireOwnerSession, resolveRegisteredConnectorManifest, @@ -8397,6 +8400,7 @@ export async function startServer(opts: ServerOpts = {}) { nekoProxyAllowedHosts: opts.nekoProxyAllowedHosts, nekoProxyAutoLogin: opts.nekoProxyAutoLogin, nekoWindowSettleProbe: opts.nekoWindowSettleProbe, + onManualUploadValidationTask: opts.onManualUploadValidationTask, ownerAuthForceSecureCookies: opts.ownerAuthForceSecureCookies, ownerAuthPassword: opts.ownerAuthPassword, ownerAuthSameSite: opts.ownerAuthSameSite, diff --git a/reference-implementation/server/routes/ref-manual-upload-draft-connection.ts b/reference-implementation/server/routes/ref-manual-upload-draft-connection.ts index b6cd1086..6684e70e 100644 --- a/reference-implementation/server/routes/ref-manual-upload-draft-connection.ts +++ b/reference-implementation/server/routes/ref-manual-upload-draft-connection.ts @@ -246,6 +246,7 @@ export interface MountRefManualUploadDraftConnectionContext { handleError: (res: unknown, err: unknown) => void; importBaseDir: string; now?: () => string; + onManualUploadValidationTask?: (task: Promise) => void; pdppError: PdppErrorFn; requireOwnerSession: MiddlewareHandler; resolveRegisteredConnectorManifest: (connectorId: string) => Promise; @@ -911,6 +912,7 @@ async function validateAndStageArtifact( // three success/failed/duplicate branches above. force:true makes this // a safe no-op if rename() already moved the file out before the error. await removeStagingArtifact(artifact.stagingPath); + throw err; } } @@ -1525,18 +1527,24 @@ function mountPostStagedArtifact(app: AppLike, ctx: MountRefManualUploadDraftCon status: "uploaded", }); - setImmediate(() => { - validateAndStageArtifact(ctx, { - artifactId, - connectorId, - displayNameRaw: rawDisplayName, - fileName, - manifest, - ownerSubjectId: ownerSubjectId as string, - setup, - targetConnectionId, - }).catch(() => undefined); + const validationTask = new Promise((resolve, reject) => { + setImmediate(() => { + validateAndStageArtifact(ctx, { + artifactId, + connectorId, + displayNameRaw: rawDisplayName, + fileName, + manifest, + ownerSubjectId: ownerSubjectId as string, + setup, + targetConnectionId, + }).then(resolve, reject); + }); }); + // Keep the original promise available to lifecycle owners, including tests. + // The observer can report rejection even though the HTTP response is already 202. + validationTask.catch(() => undefined); + ctx.onManualUploadValidationTask?.(validationTask); await emitManualUploadAudit(ctx, req, res, { // biome-ignore lint/suspicious/noUnnecessaryConditions: TypeScript boundary permits nullish input; this guard preserves runtime behavior. diff --git a/reference-implementation/test/manual-upload-draft-connection-route.test.ts b/reference-implementation/test/manual-upload-draft-connection-route.test.ts index 3893cdb8..e22858ce 100644 --- a/reference-implementation/test/manual-upload-draft-connection-route.test.ts +++ b/reference-implementation/test/manual-upload-draft-connection-route.test.ts @@ -3,6 +3,8 @@ import assert from "node:assert/strict"; import { existsSync, mkdtempSync, readdirSync, readFileSync, rmSync } from "node:fs"; +import fsPromises from "node:fs/promises"; +import { syncBuiltinESMExports } from "node:module"; import { tmpdir } from "node:os"; import { join } from "node:path"; import test from "node:test"; @@ -40,20 +42,38 @@ async function closeServer(server: StartedServer): Promise { ]); } -async function withServer(fn: (ctx: { asUrl: string; tmp: string }) => Promise): Promise { +async function withServer( + fn: (ctx: { asUrl: string; tmp: string; server: StartedServer }) => Promise +): Promise { + const validationTasks: Promise[] = []; const tmp = mkdtempSync(join(tmpdir(), "pdpp-manual-upload-")); const server = (await startServer({ asPort: 0, autoEnrollEligibleSchedules: false, dbPath: join(tmp, "pdpp.sqlite"), + onManualUploadValidationTask: (task) => { + // Observe immediately, retain the original rejection, and await every task at teardown. + task.catch(() => undefined); + validationTasks.push(task); + }, ownerAuthPassword: OWNER_PASSWORD, ownerAuthSubjectId: OWNER_SUBJECT_ID, quiet: true, rsPort: 0, })) as StartedServer; const asUrl = `http://localhost:${server.asPort}`; + const bodyResults = await Promise.allSettled([fn({ asUrl, server, tmp })]); try { - await fn({ asUrl, tmp }); + // A terminal row can precede filesystem cleanup. Only settled task handles + // establish that this test can release its database and staging directory. + const taskResults = await Promise.allSettled(validationTasks); + const failures = [...bodyResults, ...taskResults].filter((result) => result.status === "rejected"); + if (failures.length > 0) { + throw new AggregateError( + failures.map((result) => result.reason), + "Manual upload test or validation failed" + ); + } } finally { await closeServer(server); rmSync(tmp, { force: true, recursive: true }); @@ -302,16 +322,20 @@ interface ManualUploadBody { validation_expectations?: string[]; } +// A wall-clock budget bounds status assertions; attempt-count budgets stretch +// with slower HTTP requests. Exhaustion fails the assertion, while withServer +// separately awaits actual validation completion before teardown. async function waitForArtifact( asUrl: string, cookie: string, artifactId: string, expectedStatuses: readonly string[], - maxAttempts = 30 -): Promise { + budgetMs = 30_000 +): Promise { const statuses = new Set(expectedStatuses); + const deadline = Date.now() + budgetMs; let latest: JsonResult | null = null; - for (let attempt = 0; attempt < maxAttempts; attempt += 1) { + while (Date.now() < deadline) { // biome-ignore lint/performance/noAwaitInLoops: localized test assertion preserves its explicit contract. latest = await getArtifact(asUrl, cookie, artifactId); const body = latest.body as ArtifactBody; @@ -320,7 +344,10 @@ async function waitForArtifact( } await new Promise((resolve) => setTimeout(resolve, 25)); } - return latest; + const lastStatus = (latest?.body as ArtifactBody | undefined)?.status ?? ""; + throw new Error( + `artifact ${artifactId} did not reach [${expectedStatuses.join(", ")}] within ${budgetMs}ms; last status=${lastStatus}` + ); } interface ZipEntry { @@ -627,31 +654,116 @@ test("a successfully staged upload leaves no orphaned _staging directory behind" }); }); -test("a WhatsApp .txt upload well past the old 1 GiB cap streams to disk and validates successfully", async () => { - // Proves the production HTTP route end-to-end for an artifact well beyond - // the old hardcoded 1 GiB WhatsApp cap this task raised — this upload - // would have been REJECTED outright before this change, purely on size, - // before any streaming/memory question even arose. The upload body is - // generated and streamed via a ReadableStream (never materialized as one - // Buffer/string in THIS test process). - // - // This test does NOT assert a memory-growth bound. Measured directly - // (see this task's report): even with the raw-file-buffering bug fixed, - // parsing a large-message-COUNT .txt export still holds the full parsed - // message array in memory before any record is emitted, and that array's - // total object/string overhead measurably EXCEEDS the raw file size for - // realistic prose-length messages (confirmed ~1.6x at 1.9 GiB) — a - // separate, disclosed residual from the whole-file-buffer bug this task - // fixed. The deterministic proof that the RAW FILE is never buffered - // whole lives in manual-upload-whatsapp-no-whole-file-read.test.ts (call- - // interception via mock.module, not an RSS heuristic); this test's job is - // functional correctness (the upload is accepted and validates) at a size - // that would have been rejected by the old cap, not a memory assertion. +test("teardown waits for validation cleanup after the artifact reaches failed status", async (t) => { + const originalRm = fsPromises.rm; + let releaseCleanup!: () => void; + const cleanupGate = new Promise((resolve) => { + releaseCleanup = resolve; + }); + let cleanupStarted!: () => void; + const cleanupEntered = new Promise((resolve) => { + cleanupStarted = resolve; + }); + let bodyFinished!: () => void; + const bodyDone = new Promise((resolve) => { + bodyFinished = resolve; + }); + let tmpPath = ""; + let finished = false; + let closeStarted = false; + const rmMock = t.mock.method(fsPromises, "rm", async (...[path, options]: Parameters) => { + if (tmpPath && String(path).startsWith(join(tmpPath, "imports", "_staging"))) { + cleanupStarted(); + await cleanupGate; + } + return originalRm(path, options); + }); + syncBuiltinESMExports(); + const run = withServer(async ({ asUrl, tmp, server }) => { + tmpPath = tmp; + const originalClose = server.asServer.close; + t.mock.method(server.asServer, "close", function (this: CloseableServer, callback?: (err?: Error) => void) { + closeStarted = true; + return originalClose.call(this, callback); + }); + await registerConnector(asUrl, "google_maps"); + const cookie = await login(asUrl); + const staged = await stageUpload(asUrl, cookie, "google-maps", "Timeline.json", '{"not":"timeline"}'); + assert.equal(staged.status, 202, staged.text); + const artifactId = asBody(staged.body).artifact_id; + assert.ok(artifactId); + await waitForArtifact(asUrl, cookie, artifactId, ["failed"]); + bodyFinished(); + }); + const observed = run.finally(() => { + finished = true; + }); + // Attach now so even a premature teardown failure cannot become unhandled. + observed.catch(() => undefined); + try { + await Promise.race([cleanupEntered, observed]); + await Promise.race([bodyDone, observed]); + // Flush the callback continuation so teardown reaches its task wait. + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(closeStarted, false, "server closure must wait for the actual task"); + assert.equal(finished, false, "terminal status must not release an unfinished validation task"); + assert.equal(existsSync(tmpPath), true, "the task still owns its directory"); + releaseCleanup(); + await observed; + assert.equal(existsSync(tmpPath), false); + } finally { + releaseCleanup(); + await observed.catch(() => undefined); + rmMock.mock.restore(); + syncBuiltinESMExports(); + } +}); + +test("a validation filesystem failure rejects teardown after cleanup", async (t) => { + const originalRename = fsPromises.rename; + const failure = new Error("injected validation rename failure"); + let tmpPath = ""; + const renameMock = t.mock.method( + fsPromises, + "rename", + async (...[from, to]: Parameters) => { + if (tmpPath && String(from).startsWith(join(tmpPath, "imports", "_staging"))) { + throw failure; + } + return await originalRename(from, to); + } + ); + syncBuiltinESMExports(); + try { + await assert.rejects( + withServer(async ({ asUrl, tmp }) => { + tmpPath = tmp; + await registerConnector(asUrl, "google_maps"); + const cookie = await login(asUrl); + const staged = await stageUpload(asUrl, cookie, "google-maps", "Timeline.json", VALID_TIMELINE_BODY); + assert.equal(staged.status, 202, staged.text); + const artifactId = asBody(staged.body).artifact_id; + assert.ok(artifactId); + await waitForArtifact(asUrl, cookie, artifactId, ["failed"]); + }), + (error: unknown) => error instanceof AggregateError && error.errors.includes(failure) + ); + assert.equal(existsSync(tmpPath), false, "rejection must still release the test directory"); + } finally { + renameMock.mock.restore(); + syncBuiltinESMExports(); + } +}); + +test("a roughly 200 MiB WhatsApp .txt upload streams to disk and validates successfully", async () => { + // Exercise the real HTTP upload and validation path with a generated stream. + // This fixture does not exceed 1 GiB or assert a memory-growth bound; the + // no-whole-file-read test separately checks file buffering by interception. await withServer(async ({ asUrl }) => { await registerConnector(asUrl, "whatsapp"); const cookie = await login(asUrl); - const targetBytes = 200 * 1024 * 1024; // 200 MiB — well past the old 1 GiB cap's REASON for being tested (proves the cap is gone), far below multi-GB OOM/tmpfs risk + const targetBytes = 200 * 1024 * 1024; // Exercise a large streamed body without a multi-GiB fixture. const oneMessage = "[6/5/24, 9:15:22 AM] Alice: Hello there, this is a realistically-sized conversational message for the test.\n"; const chunkText = oneMessage.repeat(1000); // ~120 KB per chunk @@ -684,7 +796,8 @@ test("a WhatsApp .txt upload well past the old 1 GiB cap streams to disk and val assert.equal(resp.status, 202, JSON.stringify(stagedBody)); assert.ok(stagedBody.artifact_id, "expected an artifact_id"); - const done = await waitForArtifact(asUrl, cookie, stagedBody.artifact_id, ["staged", "failed"], 400); + // Validating the roughly 200 MiB fixture can be slow on a loaded host. + const done = await waitForArtifact(asUrl, cookie, stagedBody.artifact_id, ["staged", "failed"], 240_000); const doneBody = asBody(done?.body); assert.equal(doneBody.status, "staged", JSON.stringify(doneBody)); assert.equal(doneBody.validation?.status, "valid");