diff --git a/src/__tests__/routes/parquetCleanup.test.ts b/src/__tests__/routes/parquetCleanup.test.ts new file mode 100644 index 000000000..d4ad58451 --- /dev/null +++ b/src/__tests__/routes/parquetCleanup.test.ts @@ -0,0 +1,132 @@ +import express, { NextFunction, Request, Response } from "express"; +import fs from "fs"; +import os from "os"; +import path from "path"; +import { finished } from "stream/promises"; +import request from "supertest"; + +jest.mock("../../db", () => ({ + prisma: { tokenTransfer: { findMany: jest.fn() } }, + toDisplayAmount: (amount: string) => amount, +})); +jest.mock("../../tokenCache", () => ({ getCachedTokenDecimals: () => 7 })); + +import { createExportsRouter } from "../../routes/exports"; +import { prisma } from "../../db"; + +const parquet = require("parquetjs-lite"); +const findMany = prisma.tokenTransfer.findMany as jest.Mock; +const createWriteStream = fs.createWriteStream.bind(fs); +let outputs: fs.WriteStream[]; +let receivedError: unknown; + +function app() { + const server = express(); + server.use(createExportsRouter()); + server.use((error: unknown, _req: Request, res: Response, _next: NextFunction) => { + receivedError = error; + res.status(500).json({ error: "export failed" }); + }); + return server; +} + +const transfer = { + id: 1, + contractId: "CEXAMPLE", + eventType: "transfer", + fromAddress: null, + toAddress: "GEXAMPLE", + amount: "10000000", + ledger: 123, + ledgerClosedAt: new Date("2026-01-01T00:00:00Z"), + txHash: "hash-1", + eventId: "event-1", + isSac: false, + createdAt: new Date("2026-01-01T00:00:00Z"), +}; + +beforeEach(() => { + findMany.mockReset(); + outputs = []; + receivedError = undefined; + jest.spyOn(fs, "createWriteStream").mockImplementation((file, options) => { + const stream = createWriteStream(file, options); + if (String(file).startsWith(path.join(os.tmpdir(), "transfers-"))) outputs.push(stream); + return stream; + }); +}); + +afterEach(async () => { + jest.restoreAllMocks(); + // Also release descriptors when a regression fails against the old handler. + for (const output of outputs) { + output.destroy(); + await finished(output).catch(() => {}); + await fs.promises.unlink(output.path).catch((error: NodeJS.ErrnoException) => { + if (error.code !== "ENOENT") throw error; + }); + } +}); + +describe("Parquet export file lifecycle", () => { + it("validates the request before opening a file", async () => { + const response = await request(app()).get("/transfers.parquet?maxRows=0"); + expect(response.status).toBe(400); + expect(outputs).toHaveLength(0); + expect(findMany).not.toHaveBeenCalled(); + }); + + it("does not open a writer when the truncation probe fails", async () => { + const error = new Error("probe failed"); + findMany.mockRejectedValueOnce(error); + + const response = await request(app()).get("/transfers.parquet"); + + expect(response.status).toBe(500); + expect(receivedError).toBe(error); + expect(outputs).toHaveLength(0); + }); + + it.each(["query", "append", "close"])( + "closes and removes the file after a %s failure, preserving the original error", + async (stage) => { + const error = new Error(`${stage} failed`); + findMany.mockResolvedValueOnce([]); + if (stage === "query") { + findMany.mockRejectedValueOnce(error); + } else { + findMany.mockResolvedValueOnce([transfer]); + const method = stage === "append" ? "appendRow" : "close"; + jest.spyOn(parquet.ParquetWriter.prototype, method).mockRejectedValueOnce(error); + } + + const response = await request(app()).get("/transfers.parquet?maxRows=1"); + + expect(response.status).toBe(500); + expect(receivedError).toBe(error); + expect(outputs).toHaveLength(1); + expect(outputs[0].closed).toBe(true); + expect(fs.existsSync(outputs[0].path)).toBe(false); + }, + ); + + it("still downloads a readable capped Parquet file", async () => { + findMany.mockResolvedValueOnce([{ id: 2 }]).mockResolvedValueOnce([transfer]); + + const response = await request(app()).get("/transfers.parquet?maxRows=1"); + + expect(response.status).toBe(200); + expect(response.headers["x-truncated"]).toBe("true"); + expect(response.headers["x-row-limit"]).toBe("1"); + const reader = await parquet.ParquetReader.openBuffer(response.body); + try { + const cursor = reader.getCursor(); + expect(await cursor.next()).toMatchObject({ amount: "10000000", eventId: "event-1" }); + expect(await cursor.next()).toBeNull(); + } finally { + await reader.close(); + } + expect(outputs).toHaveLength(1); + expect(outputs[0].closed).toBe(true); + }); +}); diff --git a/src/routes/exports.ts b/src/routes/exports.ts index a98134835..f21955801 100644 --- a/src/routes/exports.ts +++ b/src/routes/exports.ts @@ -4,6 +4,8 @@ import { prisma, toDisplayAmount } from "../db"; import os from "os"; import path from "path"; import fs from "fs"; +import { once } from "events"; +import { finished } from "stream/promises"; import { z } from "zod"; import { requestNetwork } from "../middleware/network"; import { parseOr400 } from "../openapi/validation"; @@ -195,6 +197,8 @@ async function handleParquetExport(req: Request, res: Response, next: NextFuncti os.tmpdir(), `transfers-${Date.now()}-${Math.random().toString(36).slice(2)}.parquet`, ); + let outputStream: fs.WriteStream | undefined; + let outputClosed: Promise | undefined; try { const parsed = parseOr400(exportQuerySchema, req.query, res); @@ -220,13 +224,21 @@ async function handleParquetExport(req: Request, res: Response, next: NextFuncti createdAt: { type: "UTF8" }, }); - const writer = await parquet.ParquetWriter.openFile(schema, tmpFile); - // Headers are set after the file is written here, so this could have been // detected mid-stream — but it goes through the same helper as CSV so the // two endpoints cannot answer differently about the same query. const truncated = await hasRowsBeyond(where, effectiveMax); + // Own the stream so a query, row, or footer failure can close it without + // calling close() a second time on a partially closed Parquet writer. + outputStream = fs.createWriteStream(tmpFile); + outputClosed = new Promise((resolve) => outputStream!.once("close", resolve)); + const outputFinished = finished(outputStream); + // Writer operations can fail before we reach the completion await below. + void outputFinished.catch(() => {}); + await once(outputStream, "open"); + const writer = await parquet.ParquetWriter.openStream(schema, outputStream); + for await (const row of streamTransfers(where, effectiveMax)) { await writer.appendRow({ id: row.id, @@ -246,6 +258,7 @@ async function handleParquetExport(req: Request, res: Response, next: NextFuncti } await writer.close(); + await outputFinished; res.setHeader("Content-Type", "application/octet-stream"); res.setHeader("Content-Disposition", "attachment; filename=\"transfers.parquet\""); @@ -262,7 +275,12 @@ async function handleParquetExport(req: Request, res: Response, next: NextFuncti next(err); }); } catch (err) { - fs.unlink(tmpFile, () => {}); + // Let the writer's initial header write finish before releasing the file. + if (outputStream && !outputStream.destroyed && !outputStream.writableEnded) { + outputStream.end(); + } + await outputClosed; + await fs.promises.unlink(tmpFile).catch(() => {}); next(err); } }