From f47542c59206046cefba5a88ecbbff51bbca19bb Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Fri, 31 Jul 2026 14:17:38 -0700 Subject: [PATCH] fix(gateway): centralize streamed media descriptor ownership (#117000) Co-authored-by: Peter Steinberger --- src/gateway/control-ui.ts | 63 +++--------- src/gateway/http-byte-range.test.ts | 99 ++++++++++++++++++- src/gateway/http-byte-range.ts | 58 +++++++++++ src/gateway/managed-image-attachments.test.ts | 59 ++++++++++- src/gateway/managed-image-attachments.ts | 56 +++-------- 5 files changed, 243 insertions(+), 92 deletions(-) diff --git a/src/gateway/control-ui.ts b/src/gateway/control-ui.ts index 9f24511380be..25f90d7d4ced 100644 --- a/src/gateway/control-ui.ts +++ b/src/gateway/control-ui.ts @@ -81,7 +81,11 @@ import { sendControlUiHtmlBody, serveControlUiAsset, } from "./control-ui-static.js"; -import { resolveByteResponse, writeByteHeaders } from "./http-byte-range.js"; +import { + createGatewayByteStream, + resolveByteResponse, + writeByteHeaders, +} from "./http-byte-range.js"; import { buildMissingScopeForbiddenBody, sendGatewayAuthFailure } from "./http-common.js"; import { getBearerToken, @@ -667,21 +671,13 @@ export async function handleControlUiAssistantMediaRequest( return true; } - let opened: Awaited> | null = null; - let localPath; - let handleClosed = false; - const closeOpenedHandle = async () => { - if (!opened || handleClosed) { - return; - } - handleClosed = true; - await opened.handle.close().catch(() => {}); - }; + let byteStream: ReturnType | undefined; try { const resolvedReference = await resolveMediaReferenceLocalPathInfo(source); - localPath = resolvedReference.path; + const localPath = resolvedReference.path; await assertLocalMediaAllowed(localPath, localRoots); - opened = await openLocalFileSafely({ filePath: localPath }); + let opened = await openLocalFileSafely({ filePath: localPath }); + byteStream = createGatewayByteStream(res, opened.handle, () => respondControlUiNotFound(res)); const sniffLength = Math.min(opened.stat.size, 8192); const sniffBuffer = sniffLength > 0 ? Buffer.allocUnsafe(sniffLength) : undefined; const bytesRead = @@ -709,16 +705,18 @@ export async function handleControlUiAssistantMediaRequest( kind: mediaKind, }); if (playback.kind === "preparing") { - await closeOpenedHandle(); + await byteStream.close(); sendJson(res, 202, { status: "preparing" }); return true; } if (playback.kind === "transcoded") { const transcoded = await openLocalFileSafely({ filePath: playback.path }).catch(() => null); if (transcoded) { - await closeOpenedHandle(); + await byteStream.close(); opened = transcoded; - handleClosed = false; + byteStream = createGatewayByteStream(res, opened.handle, () => + respondControlUiNotFound(res), + ); contentType = playback.contentType; filename = replacePlaybackFileExtension(filename, playback.extension); } @@ -738,39 +736,10 @@ export async function handleControlUiAssistantMediaRequest( ifNoneMatchHeader: req.headers["if-none-match"], }); writeByteHeaders(res, byteResponse); - if ( - req.method === "HEAD" || - byteResponse.kind === "not-modified" || - byteResponse.kind === "unsatisfiable" || - opened.stat.size === 0 - ) { - await closeOpenedHandle(); - res.end(); - return true; - } - const stream = opened.handle.createReadStream({ - start: byteResponse.kind === "partial" ? byteResponse.range.start : 0, - end: byteResponse.kind === "partial" ? byteResponse.range.end : opened.stat.size - 1, - autoClose: false, - }); - const finishClose = () => { - void closeOpenedHandle(); - }; - stream.once("end", finishClose); - stream.once("close", finishClose); - stream.once("error", () => { - void closeOpenedHandle(); - if (!res.headersSent) { - respondControlUiNotFound(res); - } else { - res.destroy(); - } - }); - res.once("close", finishClose); - stream.pipe(res); + await byteStream.pipe(byteResponse, req.method); return true; } catch { - await closeOpenedHandle(); + await byteStream?.close(); respondControlUiNotFound(res); return true; } diff --git a/src/gateway/http-byte-range.test.ts b/src/gateway/http-byte-range.test.ts index 08ad69fd490d..3ab4dac9da3b 100644 --- a/src/gateway/http-byte-range.test.ts +++ b/src/gateway/http-byte-range.test.ts @@ -1,6 +1,13 @@ -import type { ServerResponse } from "node:http"; +import fs from "node:fs/promises"; +import http, { type ServerResponse } from "node:http"; +import os from "node:os"; +import path from "node:path"; import { describe, expect, it, vi } from "vitest"; -import { resolveByteResponse, writeByteHeaders } from "./http-byte-range.js"; +import { + createGatewayByteStream, + resolveByteResponse, + writeByteHeaders, +} from "./http-byte-range.js"; const FILE = { size: 10, mtimeMs: 1_752_000_000_123.5 }; @@ -154,3 +161,91 @@ describe("byte ETag generation", () => { expect(etag).toMatch(/^"[A-Za-z0-9_-]+"$/); }); }); + +describe("Gateway byte response descriptor lifecycle", () => { + it("destroys the real file stream and closes its descriptor once when its HTTP client disconnects", async () => { + const directory = await fs.mkdtemp(path.join(os.tmpdir(), "gateway-byte-stream-")); + const filePath = path.join(directory, "media.bin"); + const body = Buffer.alloc(8 * 1024 * 1024, 7); + await fs.writeFile(filePath, body); + const handle = await fs.open(filePath, "r"); + const closeHandle = vi.spyOn(handle, "close"); + const createReadStream = vi.spyOn(handle, "createReadStream"); + let resolveResponseClose!: () => void; + const responseClosed = new Promise((resolve) => { + resolveResponseClose = resolve; + }); + const server = http.createServer((_request, response) => { + const owner = createGatewayByteStream(response, handle, () => { + response.statusCode = 404; + response.end("not found"); + }); + const byteResponse = resolveByteResponse({ + file: { size: body.byteLength, mtimeMs: 1 }, + method: "GET", + }); + writeByteHeaders(response, byteResponse); + void owner.pipe(byteResponse, "GET"); + response.once("close", resolveResponseClose); + }); + await new Promise((resolve) => { + server.listen(0, "127.0.0.1", resolve); + }); + const address = server.address(); + if (!address || typeof address === "string") { + throw new Error("expected test HTTP server to bind to a TCP port"); + } + + try { + await new Promise((resolve, reject) => { + const request = http.get({ host: "127.0.0.1", port: address.port }, (response) => { + response.once("data", () => { + response.destroy(); + resolve(); + }); + }); + request.once("error", reject); + }); + await responseClosed; + await vi.waitFor(() => { + expect(closeHandle).toHaveBeenCalledOnce(); + expect(handle.fd).toBe(-1); + }); + + const streamedFile = createReadStream.mock.results[0]?.value; + expect(streamedFile?.destroyed).toBe(true); + expect(streamedFile?.readableEnded).toBe(false); + } finally { + server.closeAllConnections(); + await new Promise((resolve) => { + server.close(() => resolve()); + }); + await fs.rm(directory, { recursive: true, force: true }); + } + }); + + it("closes a newly opened descriptor when its response ended before streaming began", async () => { + const directory = await fs.mkdtemp(path.join(os.tmpdir(), "gateway-byte-ended-")); + const filePath = path.join(directory, "media.bin"); + await fs.writeFile(filePath, "media"); + const handle = await fs.open(filePath, "r"); + const closeHandle = vi.spyOn(handle, "close"); + const response = new http.ServerResponse({ method: "GET" } as http.IncomingMessage); + response.end(); + const owner = createGatewayByteStream(response, handle, () => {}); + + try { + await owner.pipe( + resolveByteResponse({ file: { size: 5, mtimeMs: 1 }, method: "GET" }), + "GET", + ); + expect(closeHandle).toHaveBeenCalledOnce(); + expect(handle.fd).toBe(-1); + } finally { + if (handle.fd >= 0) { + await handle.close(); + } + await fs.rm(directory, { recursive: true, force: true }); + } + }); +}); diff --git a/src/gateway/http-byte-range.ts b/src/gateway/http-byte-range.ts index 4845dd547931..d687e7907511 100644 --- a/src/gateway/http-byte-range.ts +++ b/src/gateway/http-byte-range.ts @@ -1,4 +1,5 @@ import { createHash } from "node:crypto"; +import type { FileHandle } from "node:fs/promises"; import type { ServerResponse } from "node:http"; import { matchesHttpIfNoneMatch } from "./http-conditional.js"; @@ -146,3 +147,60 @@ export function writeByteHeaders(res: ServerResponse, plan: ByteResponsePlan): v res.setHeader("Content-Range", `bytes */${plan.size}`); } } + +export function createGatewayByteStream( + res: ServerResponse, + handle: Pick, + onReadError: () => void, +) { + let stream: ReturnType | undefined; + let closed = false; + const close = async () => { + if (closed) { + return; + } + closed = true; + if (stream) { + stream.destroy(); + return; + } + await handle.close().catch(() => {}); + }; + const release = () => { + void close(); + }; + // The ReadStream owns the FileHandle after creation; destroying it closes the descriptor once. + res.once("close", release); + + return { + close, + async pipe(plan: ByteResponsePlan, method: string | undefined) { + if (method === "HEAD" || !("contentLength" in plan) || plan.contentLength === 0) { + await close(); + res.end(); + return; + } + if (closed || res.destroyed || res.writableEnded) { + await close(); + return; + } + stream = handle.createReadStream({ + start: plan.kind === "partial" ? plan.range.start : 0, + end: plan.kind === "partial" ? plan.range.end : plan.contentLength - 1, + autoClose: true, + }); + stream.once("end", release).once("close", release); + stream.once("error", () => { + release(); + if (!res.destroyed && !res.writableEnded) { + if (res.headersSent) { + res.destroy(); + } else { + onReadError(); + } + } + }); + stream.pipe(res); + }, + }; +} diff --git a/src/gateway/managed-image-attachments.test.ts b/src/gateway/managed-image-attachments.test.ts index 68d82bde8a76..e808b1a12331 100644 --- a/src/gateway/managed-image-attachments.test.ts +++ b/src/gateway/managed-image-attachments.test.ts @@ -6,7 +6,7 @@ import type { AddressInfo } from "node:net"; import path from "node:path"; import { pathToFileURL } from "node:url"; import { maxBytesForKind } from "@openclaw/media-core/constants"; -import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { afterEach, beforeEach, describe, expect, it, vi, type MockInstance } from "vitest"; import { createNoisyPngBuffer as createNoisyPngFixtureBuffer, createSolidPngBuffer, @@ -21,6 +21,7 @@ import { MANAGED_OUTGOING_ORIGINALS_SUBDIR, readManagedImageRecord, } from "./managed-image-record-store.js"; +import { makeMockHttpResponse } from "./test-http-response.js"; type PlaybackTranscodeResolution = Awaited< ReturnType<(typeof import("../media/playback-transcode.js"))["resolvePlaybackTranscode"]> @@ -535,6 +536,62 @@ describe("handleManagedOutgoingImageHttpRequest", () => { expect(result.body).toEqual(body); }); + it("closes the opened managed-media descriptor when playback resolution rejects", async () => { + const { attachmentId, sessionKey, originalPath } = await createFixture(stateDir, { + filename: "voice.caf", + contentType: "audio/x-caf", + body: Buffer.from("caff-original"), + }); + authorizeGatewayHttpRequestOrReplyMock.mockResolvedValue({ ok: true, authMethod: "token" }); + resolveOpenAiCompatibleHttpOperatorScopesMock.mockReturnValue(["operator.read"]); + resolveOpenAiCompatibleHttpSenderIsOwnerMock.mockReturnValue(true); + loadSessionEntryMock.mockReturnValue({ + storePath: path.join(stateDir, "gateway-sessions.json"), + entry: { sessionId: "sess-1", sessionFile: "session.jsonl" }, + }); + resolveSessionHistoryTranscriptPathMock.mockResolvedValue("session.jsonl"); + readSessionMessagesMock.mockResolvedValue([ + { + role: "assistant", + content: [ + { + type: "audio", + url: `/api/chat/media/outgoing/${encodeURIComponent(sessionKey)}/${attachmentId}/full`, + }, + ], + __openclaw: { id: "msg-1" }, + }, + ]); + resolvePlaybackTranscodeMock.mockRejectedValueOnce(new Error("playback inspection failed")); + const originalOpen = fs.open; + let closeOpenedHandle: MockInstance<() => Promise> | undefined; + const openSpy = vi.spyOn(fs, "open").mockImplementation(async (...args) => { + const handle = await originalOpen(...args); + if (String(args[0]) === originalPath) { + closeOpenedHandle = vi.spyOn(handle, "close"); + } + return handle; + }); + const { res } = makeMockHttpResponse(); + + try { + await expect( + handleManagedOutgoingImageHttpRequest( + { + url: `/api/chat/media/outgoing/${encodeURIComponent(sessionKey)}/${attachmentId}/full?playback=1`, + method: "GET", + headers: {}, + } as http.IncomingMessage, + res, + { auth: { mode: "test" } as never, stateDir }, + ), + ).rejects.toThrow("playback inspection failed"); + expect(closeOpenedHandle).toHaveBeenCalledOnce(); + } finally { + openSpy.mockRestore(); + } + }); + it("passes native managed playback bytes through unchanged", async () => { const body = Buffer.from("ID3-native-audio"); const { attachmentId, sessionKey } = await createFixture(stateDir, { diff --git a/src/gateway/managed-image-attachments.ts b/src/gateway/managed-image-attachments.ts index 10fcac9de336..8a44497c0665 100644 --- a/src/gateway/managed-image-attachments.ts +++ b/src/gateway/managed-image-attachments.ts @@ -34,7 +34,11 @@ import { safeEqualSecret } from "../security/secret-equal.js"; import { buildAssistantMediaContentDisposition } from "./assistant-media-content-disposition.js"; import type { AuthRateLimiter } from "./auth-rate-limit.js"; import type { ResolvedGatewayAuth } from "./auth.js"; -import { resolveByteResponse, writeByteHeaders } from "./http-byte-range.js"; +import { + createGatewayByteStream, + resolveByteResponse, + writeByteHeaders, +} from "./http-byte-range.js"; import { sendJson, sendMethodNotAllowed, sendMissingScopeForbidden } from "./http-common.js"; import { authorizeGatewayHttpRequestOrReply, @@ -1459,6 +1463,8 @@ export async function handleManagedOutgoingMediaHttpRequest( sendStatus(res, 404, "not found"); return true; } + const respondNotFound = () => sendStatus(res, 404, "not found"); + let byteStream = createGatewayByteStream(res, opened.handle, respondNotFound); let responseContentType = record.original.contentType || "application/octet-stream"; let responseFilename = record.original.filename; @@ -1472,17 +1478,21 @@ export async function handleManagedOutgoingMediaHttpRequest( sourceStat: opened.stat, mimeType: responseContentType, kind: mediaKind, + }).catch(async (error: unknown) => { + await byteStream.close(); + throw error; }); if (playback.kind === "preparing") { - await opened.handle.close().catch(() => {}); + await byteStream.close(); sendJson(res, 202, { status: "preparing" }); return true; } if (playback.kind === "transcoded") { const transcoded = await openLocalFileSafely({ filePath: playback.path }).catch(() => null); if (transcoded) { - await opened.handle.close().catch(() => {}); + await byteStream.close(); opened = transcoded; + byteStream = createGatewayByteStream(res, opened.handle, respondNotFound); responseContentType = playback.contentType; responseFilename = replacePlaybackFileExtension( responseFilename ?? "generated-media", @@ -1492,14 +1502,6 @@ export async function handleManagedOutgoingMediaHttpRequest( } } - let handleClosed = false; - const closeOpenedHandle = async () => { - if (handleClosed) { - return; - } - handleClosed = true; - await opened.handle.close().catch(() => {}); - }; res.setHeader("content-type", responseContentType); res.setHeader("x-content-type-options", "nosniff"); res.setHeader("referrer-policy", "no-referrer"); @@ -1521,38 +1523,8 @@ export async function handleManagedOutgoingMediaHttpRequest( ifNoneMatchHeader: req.headers["if-none-match"], }); writeByteHeaders(res, byteResponse); - if ( - req.method === "HEAD" || - byteResponse.kind === "not-modified" || - byteResponse.kind === "unsatisfiable" || - opened.stat.size === 0 - ) { - await closeOpenedHandle(); - res.end(); - return true; - } - // Stream from the verified descriptor so a path swap cannot bypass fs-safe after validation. - const stream = opened.handle.createReadStream({ - start: byteResponse.kind === "partial" ? byteResponse.range.start : 0, - end: byteResponse.kind === "partial" ? byteResponse.range.end : opened.stat.size - 1, - autoClose: false, - }); - const finishClose = () => { - void closeOpenedHandle(); - }; - stream.once("end", finishClose); - stream.once("close", finishClose); - stream.once("error", () => { - void closeOpenedHandle(); - if (!res.headersSent) { - sendStatus(res, 404, "not found"); - } else { - res.destroy(); - } - }); - res.once("close", finishClose); - stream.pipe(res); + await byteStream.pipe(byteResponse, req.method); return true; } /* oxlint-disable max-lines -- TODO: split this grandfathered oversized file. */