diff --git a/app/src/main/assets/builtin-plugins/dsh-web-mobile/lib/compress.js b/app/src/main/assets/builtin-plugins/dsh-web-mobile/lib/compress.js index afb0b76d5..14937bccc 100644 --- a/app/src/main/assets/builtin-plugins/dsh-web-mobile/lib/compress.js +++ b/app/src/main/assets/builtin-plugins/dsh-web-mobile/lib/compress.js @@ -108,12 +108,17 @@ export function installResponseCompression({ matches = () => true } = {}) { } function flush(res) { const pending = take(res); - if (pending !== undefined && !res.destroyed) - origWriteHead.apply(res, pending.args); + if (pending === undefined || res.destroyed) return false; + origWriteHead.apply(res, pending.args); + return true; } function patchedWriteHead(...args) { - flush(this); + const flushed = flush(this); const raw = typeof args[1] === 'string' ? args[2] : args[1]; + // This call committed the deferred header write: a status-only re-entry is + // the middleware's implicit head write and must stay a no-op, matching + // Node's _implicitHeader once _header is set. + if (flushed && raw === undefined) return this; // Raw header arrays, including duplicate headers, keep Node's exact path. if (Array.isArray(raw) || (raw !== undefined && (raw === null || typeof raw !== 'object'))) return origWriteHead.apply(this, args); diff --git a/app/src/main/assets/builtin-plugins/dsh-web-mobile/lib/compress.js.map b/app/src/main/assets/builtin-plugins/dsh-web-mobile/lib/compress.js.map index c946405c8..1d3449d95 100644 --- a/app/src/main/assets/builtin-plugins/dsh-web-mobile/lib/compress.js.map +++ b/app/src/main/assets/builtin-plugins/dsh-web-mobile/lib/compress.js.map @@ -1 +1 @@ -{"version":3,"file":"compress.js","sourceRoot":"","sources":["dsha://tools/mobile-server/compress.js"],"sourcesContent":["/**\n * DSHA bounded compression for complete JSON responses.\n * Only a single end(data) can be compressed. Every write() is delegated at\n * once, with its original encoding, callback and backpressure. Headers alone\n * may wait one event-loop turn; no response body is ever accumulated here.\n * SSE, HEAD, already encoded bodies and oversized JSON remain native streams.\n */\nimport { brotliCompress, constants as zlibConstants, gzip } from 'node:zlib';\nimport { ServerResponse as NodeServerResponse } from 'node:http';\n\nconst MIN_JSON_BYTES = 4 * 1024;\nexport const MAX_JSON_BYTES = 1024 * 1024;\nexport const MAX_COMPRESSIONS = 2;\nlet inflight = 0;\nconst deferred = new WeakMap();\n\nexport function headerValue(headers, name) {\n const key = Object.keys(headers).find(key => key.toLowerCase() === name);\n return key === undefined ? undefined : String(headers[key]);\n}\nexport function isDeferrable(headers) {\n return headerValue(headers, 'content-encoding') === undefined\n && /(?:^|\\s|;)application\\/(?:[\\w.+-]+\\+)?json(?:\\s*;|$)/i.test(headerValue(headers, 'content-type') ?? '')\n && !/\\bno-transform\\b/i.test(headerValue(headers, 'cache-control') ?? '')\n && headerValue(headers, 'content-range') === undefined\n && headerValue(headers, 'transfer-encoding') === undefined\n && headerValue(headers, 'content-md5') === undefined\n && !/^\"/.test(headerValue(headers, 'etag') ?? '');\n}\nexport function varyWithAcceptEncoding(headers) {\n const key = Object.keys(headers).find(key => key.toLowerCase() === 'vary') ?? 'vary';\n const value = String(headers[key] ?? '');\n if (!value.split(',').some(token => /^(?:accept-encoding|\\*)$/i.test(token.trim())))\n headers[key] = value ? `${value}, Accept-Encoding` : 'Accept-Encoding';\n}\nfunction pickEncoding(res) {\n const values = new Map();\n for (const part of String(res.req?.headers['accept-encoding'] ?? '').split(',')) {\n const [name, ...params] = part.trim().toLowerCase().split(';');\n const q = params.find(value => /^\\s*q\\s*=/.test(value));\n const quality = q === undefined ? 1 : Number(q.split('=')[1]);\n values.set(name, Number.isFinite(quality) && quality >= 0 && quality <= 1 ? quality : 0);\n }\n const br = values.get('br') ?? values.get('*') ?? 0;\n const gzip = values.get('gzip') ?? values.get('*') ?? 0;\n return br > 0 && br >= gzip ? 'br' : gzip > 0 ? 'gzip' : null;\n}\nfunction hasBody(res, status = res.statusCode) {\n return res.req?.method !== 'HEAD' && status >= 200 && ![204, 205, 206, 304].includes(status);\n}\nfunction snapshotHeaders(headers) {\n return Object.fromEntries(Object.entries(headers).map(([key, value]) => [key, Array.isArray(value) ? value.slice() : value]));\n}\nfunction mergedHeaders(base, overlay) {\n const headers = { ...base };\n for (const [name, value] of Object.entries(overlay)) {\n for (const key of Object.keys(headers))\n if (key.toLowerCase() === name.toLowerCase()) delete headers[key];\n headers[name] = value;\n }\n return headers;\n}\n\nexport function installResponseCompression({ matches = () => true } = {}) {\n const proto = NodeServerResponse.prototype;\n const origWriteHead = proto.writeHead;\n const origWrite = proto.write;\n const origEnd = proto.end;\n const origFlushHeaders = proto.flushHeaders;\n const active = new Set();\n const nativeBody = new WeakSet();\n function delegate(method, res, args) {\n nativeBody.add(res);\n try { return method.apply(res, args); }\n finally { nativeBody.delete(res); }\n }\n function belongs(res) {\n try { return matches(res.req, res) === true; }\n catch { return false; }\n }\n // end() has accepted the complete body while libuv compresses it. Keep\n // Node's public end guard so a subsequent write/end follows its native\n // error path. Do not change any private stream/socket fields. The false\n // writableFinished view lasts only until the compressed native end.\n function reserveEnd(res) {\n const finished = Object.getOwnPropertyDescriptor(res, 'finished');\n if (!finished?.configurable || !finished.writable || finished.value !== false\n || Object.hasOwn(res, 'writableFinished')) return undefined;\n let value = finished.value;\n Object.defineProperty(res, 'finished', { configurable: true, enumerable: finished.enumerable,\n get: () => true, set: next => { value = next; } });\n Object.defineProperty(res, 'writableFinished', { configurable: true, get: () => false });\n return (closed = false) => {\n Object.defineProperty(res, 'finished', { ...finished, value: closed || value });\n delete res.writableFinished;\n };\n }\n\n function take(res) {\n const pending = deferred.get(res);\n if (pending !== undefined) {\n deferred.delete(res);\n active.delete(res);\n clearImmediate(pending.timer);\n res.off('close', pending.onClose);\n }\n return pending;\n }\n function flush(res) {\n const pending = take(res);\n if (pending !== undefined && !res.destroyed)\n origWriteHead.apply(res, pending.args);\n }\n function patchedWriteHead(...args) {\n flush(this);\n const raw = typeof args[1] === 'string' ? args[2] : args[1];\n // Raw header arrays, including duplicate headers, keep Node's exact path.\n if (Array.isArray(raw) || (raw !== undefined && (raw === null || typeof raw !== 'object')))\n return origWriteHead.apply(this, args);\n const headers = mergedHeaders(this.getHeaders(), raw ?? {});\n if (!belongs(this) || nativeBody.has(this) || this.headersSent || !hasBody(this, Number(args[0])) || !isDeferrable(headers) || pickEncoding(this) === null)\n return origWriteHead.apply(this, args);\n // Use Node itself to validate all overloads/status/header values now,\n // rather than deferring a native exception into an unrelated timer.\n const validation = new NodeServerResponse(this.req);\n for (const [name, value] of Object.entries(this.getHeaders())) validation.setHeader(name, value);\n const copied = args.slice();\n if (raw !== undefined) copied[typeof args[1] === 'string' ? 2 : 1] = snapshotHeaders(raw);\n origWriteHead.apply(validation, copied);\n this.statusCode = validation.statusCode;\n this.statusMessage = validation.statusMessage;\n const pending = { args: copied, headers: raw === undefined ? {} : copied[typeof args[1] === 'string' ? 2 : 1] };\n pending.onClose = () => take(this);\n pending.timer = setImmediate(() => flush(this));\n deferred.set(this, pending);\n active.add(this);\n this.once('close', pending.onClose);\n return this;\n }\n function patchedWrite(...args) {\n flush(this);\n return delegate(origWrite, this, args);\n }\n function patchedFlushHeaders(...args) {\n flush(this);\n return delegate(origFlushHeaders, this, args);\n }\n function patchedEnd(...args) {\n const pending = deferred.get(this);\n const headers = mergedHeaders(this.getHeaders(), pending?.headers ?? {});\n const encoding = pickEncoding(this);\n const chunk = args[0];\n const textEncoding = typeof args[1] === 'string' ? args[1] : undefined;\n let body;\n const validEncodingArgument = args[1] === undefined || typeof args[1] === 'string' || typeof args[1] === 'function';\n const validCallbackArgument = typeof args[1] === 'function' || args[2] === undefined || typeof args[2] === 'function';\n if (belongs(this) && inflight < MAX_COMPRESSIONS && !this.finished && !this.headersSent && !this.destroyed && !this.strictContentLength && validEncodingArgument && validCallbackArgument && hasBody(this) && isDeferrable(headers) && encoding !== null) {\n if (typeof chunk === 'string') {\n // Preserve invalid-encoding errors through the native end path.\n if (textEncoding === undefined || Buffer.isEncoding(textEncoding)) {\n const bytes = Buffer.byteLength(chunk, textEncoding);\n if (bytes >= MIN_JSON_BYTES && bytes <= MAX_JSON_BYTES) body = Buffer.from(chunk, textEncoding);\n }\n } else if (chunk instanceof Uint8Array && chunk.byteLength >= MIN_JSON_BYTES && chunk.byteLength <= MAX_JSON_BYTES) {\n body = Buffer.from(chunk);\n }\n }\n if (body === undefined) {\n flush(this);\n return delegate(origEnd, this, args);\n }\n const restoreEnd = reserveEnd(this);\n if (restoreEnd === undefined) {\n flush(this);\n return delegate(origEnd, this, args);\n }\n take(this);\n for (const key of Object.keys(headers))\n if (key.toLowerCase() === 'content-length') delete headers[key];\n headers['content-encoding'] = encoding;\n varyWithAcceptEncoding(headers);\n this.removeHeader('content-length');\n // Commit the selected representation now. Its size is not known until\n // the worker finishes, so Node supplies its native chunked boundary.\n // This also preserves synchronous header validation and headersSent.\n try { origWriteHead.call(this, this.statusCode, this.statusMessage, headers); }\n catch (error) { restoreEnd(); throw error; }\n const callback = typeof args[1] === 'function' ? args[1] : args[2];\n let called = false, complete = false;\n const done = error => {\n if (called) return;\n called = true;\n if (typeof callback === 'function') callback(error);\n };\n const finish = (error, compressed) => {\n if (complete) return;\n complete = true;\n this.off('close', onClose);\n restoreEnd(Boolean(error) || this.destroyed);\n if (error || this.destroyed) {\n const failure = error ?? Object.assign(new Error('response closed during compression'), { code: 'ERR_STREAM_PREMATURE_CLOSE' });\n this.destroy(failure);\n queueMicrotask(() => done(failure));\n return;\n }\n // The original encoding was consumed by Buffer.from; only end's\n // callback remains. Node still owns socket writes/backpressure.\n origEnd.call(this, compressed, done);\n };\n const onClose = () => finish();\n this.once('close', onClose);\n inflight++;\n const compressed = (error, bytes) => { inflight--; finish(error, bytes); };\n try {\n if (encoding === 'br') brotliCompress(body, { params: { [zlibConstants.BROTLI_PARAM_QUALITY]: 4 } }, compressed);\n else gzip(body, { level: 4 }, compressed);\n } catch (error) { inflight--; finish(error); }\n return this;\n }\n proto.writeHead = patchedWriteHead;\n proto.write = patchedWrite;\n proto.end = patchedEnd;\n proto.flushHeaders = patchedFlushHeaders;\n return () => {\n for (const res of active) flush(res);\n // Already accepted end bodies retain their worker and callback until\n // completion. Unload restores the hook immediately and never cancels\n // or resends a response that another route still owns.\n if (proto.writeHead === patchedWriteHead) proto.writeHead = origWriteHead;\n if (proto.write === patchedWrite) proto.write = origWrite;\n if (proto.end === patchedEnd) proto.end = origEnd;\n if (proto.flushHeaders === patchedFlushHeaders) proto.flushHeaders = origFlushHeaders;\n };\n}\n//# sourceMappingURL=compress.js.map\n"],"names":[],"mappings":"AAAA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA"} +{"version":3,"file":"compress.js","sourceRoot":"","sources":["dsha://tools/mobile-server/compress.js"],"sourcesContent":["/**\n * DSHA bounded compression for complete JSON responses.\n * Only a single end(data) can be compressed. Every write() is delegated at\n * once, with its original encoding, callback and backpressure. Headers alone\n * may wait one event-loop turn; no response body is ever accumulated here.\n * SSE, HEAD, already encoded bodies and oversized JSON remain native streams.\n */\nimport { brotliCompress, constants as zlibConstants, gzip } from 'node:zlib';\nimport { ServerResponse as NodeServerResponse } from 'node:http';\n\nconst MIN_JSON_BYTES = 4 * 1024;\nexport const MAX_JSON_BYTES = 1024 * 1024;\nexport const MAX_COMPRESSIONS = 2;\nlet inflight = 0;\nconst deferred = new WeakMap();\n\nexport function headerValue(headers, name) {\n const key = Object.keys(headers).find(key => key.toLowerCase() === name);\n return key === undefined ? undefined : String(headers[key]);\n}\nexport function isDeferrable(headers) {\n return headerValue(headers, 'content-encoding') === undefined\n && /(?:^|\\s|;)application\\/(?:[\\w.+-]+\\+)?json(?:\\s*;|$)/i.test(headerValue(headers, 'content-type') ?? '')\n && !/\\bno-transform\\b/i.test(headerValue(headers, 'cache-control') ?? '')\n && headerValue(headers, 'content-range') === undefined\n && headerValue(headers, 'transfer-encoding') === undefined\n && headerValue(headers, 'content-md5') === undefined\n && !/^\"/.test(headerValue(headers, 'etag') ?? '');\n}\nexport function varyWithAcceptEncoding(headers) {\n const key = Object.keys(headers).find(key => key.toLowerCase() === 'vary') ?? 'vary';\n const value = String(headers[key] ?? '');\n if (!value.split(',').some(token => /^(?:accept-encoding|\\*)$/i.test(token.trim())))\n headers[key] = value ? `${value}, Accept-Encoding` : 'Accept-Encoding';\n}\nfunction pickEncoding(res) {\n const values = new Map();\n for (const part of String(res.req?.headers['accept-encoding'] ?? '').split(',')) {\n const [name, ...params] = part.trim().toLowerCase().split(';');\n const q = params.find(value => /^\\s*q\\s*=/.test(value));\n const quality = q === undefined ? 1 : Number(q.split('=')[1]);\n values.set(name, Number.isFinite(quality) && quality >= 0 && quality <= 1 ? quality : 0);\n }\n const br = values.get('br') ?? values.get('*') ?? 0;\n const gzip = values.get('gzip') ?? values.get('*') ?? 0;\n return br > 0 && br >= gzip ? 'br' : gzip > 0 ? 'gzip' : null;\n}\nfunction hasBody(res, status = res.statusCode) {\n return res.req?.method !== 'HEAD' && status >= 200 && ![204, 205, 206, 304].includes(status);\n}\nfunction snapshotHeaders(headers) {\n return Object.fromEntries(Object.entries(headers).map(([key, value]) => [key, Array.isArray(value) ? value.slice() : value]));\n}\nfunction mergedHeaders(base, overlay) {\n const headers = { ...base };\n for (const [name, value] of Object.entries(overlay)) {\n for (const key of Object.keys(headers))\n if (key.toLowerCase() === name.toLowerCase()) delete headers[key];\n headers[name] = value;\n }\n return headers;\n}\n\nexport function installResponseCompression({ matches = () => true } = {}) {\n const proto = NodeServerResponse.prototype;\n const origWriteHead = proto.writeHead;\n const origWrite = proto.write;\n const origEnd = proto.end;\n const origFlushHeaders = proto.flushHeaders;\n const active = new Set();\n const nativeBody = new WeakSet();\n function delegate(method, res, args) {\n nativeBody.add(res);\n try { return method.apply(res, args); }\n finally { nativeBody.delete(res); }\n }\n function belongs(res) {\n try { return matches(res.req, res) === true; }\n catch { return false; }\n }\n // end() has accepted the complete body while libuv compresses it. Keep\n // Node's public end guard so a subsequent write/end follows its native\n // error path. Do not change any private stream/socket fields. The false\n // writableFinished view lasts only until the compressed native end.\n function reserveEnd(res) {\n const finished = Object.getOwnPropertyDescriptor(res, 'finished');\n if (!finished?.configurable || !finished.writable || finished.value !== false\n || Object.hasOwn(res, 'writableFinished')) return undefined;\n let value = finished.value;\n Object.defineProperty(res, 'finished', { configurable: true, enumerable: finished.enumerable,\n get: () => true, set: next => { value = next; } });\n Object.defineProperty(res, 'writableFinished', { configurable: true, get: () => false });\n return (closed = false) => {\n Object.defineProperty(res, 'finished', { ...finished, value: closed || value });\n delete res.writableFinished;\n };\n }\n\n function take(res) {\n const pending = deferred.get(res);\n if (pending !== undefined) {\n deferred.delete(res);\n active.delete(res);\n clearImmediate(pending.timer);\n res.off('close', pending.onClose);\n }\n return pending;\n }\n function flush(res) {\n const pending = take(res);\n if (pending === undefined || res.destroyed) return false;\n origWriteHead.apply(res, pending.args);\n return true;\n }\n function patchedWriteHead(...args) {\n const flushed = flush(this);\n const raw = typeof args[1] === 'string' ? args[2] : args[1];\n // This call committed the deferred header write: a status-only re-entry is\n // the middleware's implicit head write and must stay a no-op, matching\n // Node's _implicitHeader once _header is set.\n if (flushed && raw === undefined) return this;\n // Raw header arrays, including duplicate headers, keep Node's exact path.\n if (Array.isArray(raw) || (raw !== undefined && (raw === null || typeof raw !== 'object')))\n return origWriteHead.apply(this, args);\n const headers = mergedHeaders(this.getHeaders(), raw ?? {});\n if (!belongs(this) || nativeBody.has(this) || this.headersSent || !hasBody(this, Number(args[0])) || !isDeferrable(headers) || pickEncoding(this) === null)\n return origWriteHead.apply(this, args);\n // Use Node itself to validate all overloads/status/header values now,\n // rather than deferring a native exception into an unrelated timer.\n const validation = new NodeServerResponse(this.req);\n for (const [name, value] of Object.entries(this.getHeaders())) validation.setHeader(name, value);\n const copied = args.slice();\n if (raw !== undefined) copied[typeof args[1] === 'string' ? 2 : 1] = snapshotHeaders(raw);\n origWriteHead.apply(validation, copied);\n this.statusCode = validation.statusCode;\n this.statusMessage = validation.statusMessage;\n const pending = { args: copied, headers: raw === undefined ? {} : copied[typeof args[1] === 'string' ? 2 : 1] };\n pending.onClose = () => take(this);\n pending.timer = setImmediate(() => flush(this));\n deferred.set(this, pending);\n active.add(this);\n this.once('close', pending.onClose);\n return this;\n }\n function patchedWrite(...args) {\n flush(this);\n return delegate(origWrite, this, args);\n }\n function patchedFlushHeaders(...args) {\n flush(this);\n return delegate(origFlushHeaders, this, args);\n }\n function patchedEnd(...args) {\n const pending = deferred.get(this);\n const headers = mergedHeaders(this.getHeaders(), pending?.headers ?? {});\n const encoding = pickEncoding(this);\n const chunk = args[0];\n const textEncoding = typeof args[1] === 'string' ? args[1] : undefined;\n let body;\n const validEncodingArgument = args[1] === undefined || typeof args[1] === 'string' || typeof args[1] === 'function';\n const validCallbackArgument = typeof args[1] === 'function' || args[2] === undefined || typeof args[2] === 'function';\n if (belongs(this) && inflight < MAX_COMPRESSIONS && !this.finished && !this.headersSent && !this.destroyed && !this.strictContentLength && validEncodingArgument && validCallbackArgument && hasBody(this) && isDeferrable(headers) && encoding !== null) {\n if (typeof chunk === 'string') {\n // Preserve invalid-encoding errors through the native end path.\n if (textEncoding === undefined || Buffer.isEncoding(textEncoding)) {\n const bytes = Buffer.byteLength(chunk, textEncoding);\n if (bytes >= MIN_JSON_BYTES && bytes <= MAX_JSON_BYTES) body = Buffer.from(chunk, textEncoding);\n }\n } else if (chunk instanceof Uint8Array && chunk.byteLength >= MIN_JSON_BYTES && chunk.byteLength <= MAX_JSON_BYTES) {\n body = Buffer.from(chunk);\n }\n }\n if (body === undefined) {\n flush(this);\n return delegate(origEnd, this, args);\n }\n const restoreEnd = reserveEnd(this);\n if (restoreEnd === undefined) {\n flush(this);\n return delegate(origEnd, this, args);\n }\n take(this);\n for (const key of Object.keys(headers))\n if (key.toLowerCase() === 'content-length') delete headers[key];\n headers['content-encoding'] = encoding;\n varyWithAcceptEncoding(headers);\n this.removeHeader('content-length');\n // Commit the selected representation now. Its size is not known until\n // the worker finishes, so Node supplies its native chunked boundary.\n // This also preserves synchronous header validation and headersSent.\n try { origWriteHead.call(this, this.statusCode, this.statusMessage, headers); }\n catch (error) { restoreEnd(); throw error; }\n const callback = typeof args[1] === 'function' ? args[1] : args[2];\n let called = false, complete = false;\n const done = error => {\n if (called) return;\n called = true;\n if (typeof callback === 'function') callback(error);\n };\n const finish = (error, compressed) => {\n if (complete) return;\n complete = true;\n this.off('close', onClose);\n restoreEnd(Boolean(error) || this.destroyed);\n if (error || this.destroyed) {\n const failure = error ?? Object.assign(new Error('response closed during compression'), { code: 'ERR_STREAM_PREMATURE_CLOSE' });\n this.destroy(failure);\n queueMicrotask(() => done(failure));\n return;\n }\n // The original encoding was consumed by Buffer.from; only end's\n // callback remains. Node still owns socket writes/backpressure.\n origEnd.call(this, compressed, done);\n };\n const onClose = () => finish();\n this.once('close', onClose);\n inflight++;\n const compressed = (error, bytes) => { inflight--; finish(error, bytes); };\n try {\n if (encoding === 'br') brotliCompress(body, { params: { [zlibConstants.BROTLI_PARAM_QUALITY]: 4 } }, compressed);\n else gzip(body, { level: 4 }, compressed);\n } catch (error) { inflight--; finish(error); }\n return this;\n }\n proto.writeHead = patchedWriteHead;\n proto.write = patchedWrite;\n proto.end = patchedEnd;\n proto.flushHeaders = patchedFlushHeaders;\n return () => {\n for (const res of active) flush(res);\n // Already accepted end bodies retain their worker and callback until\n // completion. Unload restores the hook immediately and never cancels\n // or resends a response that another route still owns.\n if (proto.writeHead === patchedWriteHead) proto.writeHead = origWriteHead;\n if (proto.write === patchedWrite) proto.write = origWrite;\n if (proto.end === patchedEnd) proto.end = origEnd;\n if (proto.flushHeaders === patchedFlushHeaders) proto.flushHeaders = origFlushHeaders;\n };\n}\n//# sourceMappingURL=compress.js.map\n"],"names":[],"mappings":"AAAA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA;AACA"} diff --git a/tools/mobile-server/compress.js b/tools/mobile-server/compress.js index afb0b76d5..14937bccc 100644 --- a/tools/mobile-server/compress.js +++ b/tools/mobile-server/compress.js @@ -108,12 +108,17 @@ export function installResponseCompression({ matches = () => true } = {}) { } function flush(res) { const pending = take(res); - if (pending !== undefined && !res.destroyed) - origWriteHead.apply(res, pending.args); + if (pending === undefined || res.destroyed) return false; + origWriteHead.apply(res, pending.args); + return true; } function patchedWriteHead(...args) { - flush(this); + const flushed = flush(this); const raw = typeof args[1] === 'string' ? args[2] : args[1]; + // This call committed the deferred header write: a status-only re-entry is + // the middleware's implicit head write and must stay a no-op, matching + // Node's _implicitHeader once _header is set. + if (flushed && raw === undefined) return this; // Raw header arrays, including duplicate headers, keep Node's exact path. if (Array.isArray(raw) || (raw !== undefined && (raw === null || typeof raw !== 'object'))) return origWriteHead.apply(this, args); diff --git a/tools/test-mobile-server-streams.mjs b/tools/test-mobile-server-streams.mjs index 1f7b50ec4..124bed9d8 100644 --- a/tools/test-mobile-server-streams.mjs +++ b/tools/test-mobile-server-streams.mjs @@ -274,3 +274,44 @@ test('writeHead validates native status/header errors synchronously; callback er assert.equal(callbackError, 'ERR_STREAM_WRITE_AFTER_END'); assert.equal(responseError, callbackError); }); }); + +test('host gzip wrapper: implicit writeHead after deferred headers stays a no-op', async () => { + // 宿主 compression 的 on-headers 包装:把 writeHead 的头部并入响应头表,只把 + // statusCode 传给下一层;compression 的 end 代理在头未发出时补一次隐式 writeHead。 + // 小 JSON + Content-Length 低于宿主压缩阈值时,这条路径会走延迟分支。 + const body = JSON.stringify({ ok: true, note: 'y'.repeat(100) }); + let failure; + await fixture((_req, res) => { + const writeHead = res.writeHead; + res.writeHead = function (statusCode, reasonOrHeaders, headers) { + const raw = typeof reasonOrHeaders === 'string' ? headers : reasonOrHeaders; + if (raw !== undefined && raw !== null && typeof raw === 'object' && !Array.isArray(raw)) { + for (const [name, value] of Object.entries(raw)) this.setHeader(name, value); + } + return writeHead.call(this, statusCode, ...(typeof reasonOrHeaders === 'string' ? [reasonOrHeaders] : [])); + }; + const end = res.end; + res.end = function (...args) { + if (!this.headersSent) this.writeHead(this.statusCode); + return end.apply(this, args); + }; + try { + res.writeHead(200, { + 'Content-Type': 'application/json; charset=utf-8', + 'Content-Length': Buffer.byteLength(body), + 'Cache-Control': 'no-store', + }); + res.end(body); + } catch (error) { + failure = error; + res.destroy(); + } + }, async port => { + const result = await fetch(port).catch(error => ({ error })); + assert.equal(failure, undefined, failure && `${failure.code}: ${failure.message}`); + assert.ok(!('error' in result), 'the response must complete'); + assert.equal(result.body.toString(), body); + assert.equal(result.res.headers['content-length'], String(Buffer.byteLength(body))); + assert.equal(result.res.headers['content-encoding'], undefined); + }); +});