fix(exports): close failed Parquet output streams

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);
+  });
+});

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<void> | 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<void>((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);
   }
 }