diff --git a/src/per_isolate/main.ts b/src/per_isolate/main.ts index 2645503bf24..64a1eb91dca 100644 --- a/src/per_isolate/main.ts +++ b/src/per_isolate/main.ts @@ -52,6 +52,8 @@ if (compatFlags['typescript_implemented_streams']) { FixedLengthStream, TextEncoderStream, TextDecoderStream, + CompressionStream, + DecompressionStream, ReadableStreamDrainingReader, } = require('webstreams/streams'); @@ -175,6 +177,20 @@ if (compatFlags['typescript_implemented_streams']) { writable: true, value: TextDecoderStream, }, + CompressionStream: { + __proto__: null, + configurable: true, + enumerable: false, + writable: true, + value: CompressionStream, + }, + DecompressionStream: { + __proto__: null, + configurable: true, + enumerable: false, + writable: true, + value: DecompressionStream, + }, }); // Bootstrap the cpp exports module diff --git a/src/per_isolate/per_isolate-env.d.ts b/src/per_isolate/per_isolate-env.d.ts index 8abb9300995..41fbc92f563 100644 --- a/src/per_isolate/per_isolate-env.d.ts +++ b/src/per_isolate/per_isolate-env.d.ts @@ -80,4 +80,7 @@ declare const utils: { isAnyArrayBuffer(value: unknown): value is ArrayBuffer | SharedArrayBuffer; markPromiseHandled(promise: Promise): void; getApiSymbol(name: string): symbol; + // The C++ compression codec factory (api/compression.h: + // newCompressionCodecCallback), consumed by webstreams/compression. + newCompressionCodec(mode: string, format: string): unknown; }; diff --git a/src/per_isolate/webstreams/AGENTS.md b/src/per_isolate/webstreams/AGENTS.md index 2b44a0ff0a4..cafadd15847 100644 --- a/src/per_isolate/webstreams/AGENTS.md +++ b/src/per_isolate/webstreams/AGENTS.md @@ -17,6 +17,7 @@ private-brand dispatch, no `instanceof`) apply here — see | `readable.ts` | Reader layer + queued controllers + the BACKEND-DISPATCH points (constructor, tee, chains, byte-capable gate, JS-to-C++ extraction) | | `writable.ts` / `transform.ts` / `strategies.ts` | WHATWG writable/transform/strategies | | `identity.ts` | IdentityTransformStream and FixedLengthStream (byte-capable identity transforms) | +| `compression.ts` | CompressionStream/DecompressionStream over the C++ codec handle (utils.newCompressionCodec) | | `encoding.ts` | TextEncoderStream and TextDecoderStream (pure JS codec transforms) | | `streams.ts` | Module aggregator (user-visible classes + the flag-gated DrainingReader) | | `types.d.ts` | TypeScript type definitions for the streams API | diff --git a/src/per_isolate/webstreams/compression.ts b/src/per_isolate/webstreams/compression.ts new file mode 100644 index 00000000000..8593854dbf0 --- /dev/null +++ b/src/per_isolate/webstreams/compression.ts @@ -0,0 +1,326 @@ +'use strict'; + +// CompressionStream and DecompressionStream — Compression Streams spec +// pairs implemented over the synchronous C++ codec handle produced by +// the flag-gated CompressionStream.newCodec static (captured below +// BEFORE main.ts replaces the global with the class defined here, so +// user code never observes it). +// +// ARCHITECTURE (see the compression design notes): the codec core is +// the C++ CodecStage (api/compression.h) — eager on push, buffering its +// own output. The pair is a JS writable sink feeding the handle plus a +// QUEUED byte-capable readable (BYOB served from the queue) that the +// sink's drains enqueue into. (The pipeline-optimization effort hosts +// this readable on the NATIVE backend for sink-end fusion; on this +// substrate it is a queued byte stream, per the design's E1 sequencing +// resolution.) +// +// SEMANTICS: +// - EAGER PUSH: write(chunk) runs the codec synchronously; corrupt +// input rejects the WRITE and a strict-mode incomplete stream +// rejects the CLOSE — the spec's transform()/flush() error timing. +// - LEGACY-PARITY SETTLEMENT: writes settle as soon as the codec +// consumed the chunk, without waiting for reads — matching the C++ +// implementation this replaces (which had no write backpressure), +// not the standard TransformStream's one-chunk lookahead. The +// divergence is deliberate and carried forward. +// - BYTE-CAPABLE READABLE: legacy parity — the C++ pair's readable +// accepts BYOB readers, so this one does too (WHATWG describes a +// default stream here). + +import type { + ReadableStream as ReadableStreamType, + WritableStream as WritableStreamType, +} from './types'; + +const { + DataViewPrototypeGetBuffer, + ObjectDefineProperties, + SymbolToStringTag, + TypeError, + TypedArrayPrototypeGetBuffer, + Uint8Array, + uncurryThis, +} = primordials; + +const { isArrayBuffer, isArrayBufferView, isSharedArrayBuffer, isDataView } = + utils; + +// Captured for primordials discipline — ToString coercion per spec. +const StringCoerce = String; + +const { + ReadableStream, + ReadableByteStreamController, +} = require('webstreams/readable'); +const { + WritableStream, + WritableStreamDefaultController, +} = require('webstreams/writable'); + +// --- Bootstrap captures --------------------------------------------------- + +const writableControllerError = uncurryThis( + WritableStreamDefaultController.prototype.error +) as (controller: object, reason: unknown) => void; + +const byteControllerEnqueue = uncurryThis( + ReadableByteStreamController.prototype.enqueue +) as (controller: object, chunk: ArrayBufferView) => void; +const byteControllerClose = uncurryThis( + ReadableByteStreamController.prototype.close +) as (controller: object) => void; +const byteControllerError = uncurryThis( + ReadableByteStreamController.prototype.error +) as (controller: object, reason: unknown) => void; + +// The synchronous codec handle produced by utils.newCompressionCodec: an +// internal JSG resource (CompressionCodec in api/compression.h). Its methods +// live on a per-isolate JSG prototype that user code can never reach — the +// handle instances are module-private and the type is registered as neither a +// global nor a nested type — so plain method calls are pollution-safe here +// (the same reachability argument as the #-brand internals). +interface CodecHandle { + push(chunk: ArrayBuffer | ArrayBufferView): void; + end(): void; + pullInto(view: ArrayBufferView): number; + available(): number; +} + +// The C++ codec factory, injected through the bootstrap's utils pseudo-global +// (never present on globalThis or any user-visible surface). +const newCodec = utils.newCompressionCodec as ( + mode: 'compress' | 'decompress', + format: string +) => CodecHandle; + +function isActualObject(value: unknown): boolean { + return value != null && typeof value === 'object'; +} + +// True for BufferSource chunks the codec accepts: ArrayBuffers and views, +// excluding anything SharedArrayBuffer-backed (per Web IDL, [AllowShared] is +// not granted here; WPT pins the rejection). Captured getters are used for +// the view's buffer — prototype accessors are user-patchable. +function isValidChunk(chunk: unknown): boolean { + if (isArrayBuffer(chunk)) return true; + if (isSharedArrayBuffer(chunk)) return false; + if (!isArrayBufferView(chunk)) return false; + const buffer = isDataView(chunk) + ? DataViewPrototypeGetBuffer(chunk) + : TypedArrayPrototypeGetBuffer(chunk); + return !isSharedArrayBuffer(buffer); +} + +interface CodecPair { + readable: ReadableStreamType; + writable: WritableStreamType; +} + +function createCodecPair( + mode: 'compress' | 'decompress', + format: unknown +): CodecPair { + // Spec: format is ToString-coerced, then validated — the handle + // factory performs the validation with the same TypeError message as + // the legacy constructor. + const formatString = StringCoerce(format); + const handle = newCodec(mode, formatString); + + let writableController: object | undefined; + let readableController: object; + + // Codec failure (corrupt input on write; strict end checks on close): + // error the readable side — the writable errors via the sink throw + // itself. Mirrors the legacy implementation's cancelInternal, which + // rejected pending reads and errored the state machine on any codec + // exception. + const failBoth = (reason: unknown): void => { + byteControllerError(readableController, reason); + }; + + // Drains all buffered stage output into the readable's queue. The + // enqueue is unconditional: every call site runs either right after a + // successful codec step (stream readable) or is unreachable once the + // pair has failed or been canceled (the errored/canceled writable + // rejects writes before the sink hooks run). + const drainStage = (): void => { + const available = handle.available(); + if (available <= 0) return; + const out = new Uint8Array(available); + handle.pullInto(out); + byteControllerEnqueue(readableController, out); + }; + + const writable = new WritableStream({ + start: (c: object): void => { + writableController = c; + }, + write: (chunk: unknown): void => { + if (!isValidChunk(chunk)) { + // An invalid chunk errors BOTH sides, matching the legacy + // implementation (any write failure errored the whole pair) — + // without this the readable side would hang on its pending + // pull. + const err = new TypeError( + 'The provided value is not of type (ArrayBuffer or ArrayBufferView)' + ); + failBoth(err); + throw err; + } + // EAGER: the codec consumes the chunk synchronously (the caller's + // buffer is never retained); a codec error throws HERE, rejecting + // the write — the spec's transform-time error timing. The throw + // errors the writable via the sink machinery; the readable is + // errored explicitly, mirroring the legacy cancelInternal path. + try { + handle.push(chunk as ArrayBuffer | ArrayBufferView); + } catch (e) { + // Deliver output the codec produced before the error point (e.g. the + // final valid bytes preceding trailing junk) to any pending read, then + // error. The WPT-pinned order: output first, error on later reads. + drainStage(); + failBoth(e); + throw e; + } + // Move any produced output to the readable immediately (writes + // never wait for reads — legacy-parity settlement; the queue + // buffers). + drainStage(); + }, + close: (): void => { + // Z_FINISH plus the strict-mode end checks; a throw rejects the + // close (the spec's flush-time error timing) with the same + // both-sides error propagation as write above. + try { + handle.end(); + } catch (e) { + drainStage(); + failBoth(e); + throw e; + } + // Deliver the flush tail, then close (buffered bytes are served + // to remaining reads before the close lands — queued byte-stream + // semantics). + drainStage(); + byteControllerClose(readableController); + }, + abort: (reason: unknown): void => { + byteControllerError(readableController, reason); + }, + }); + + // The readable half: a queued byte stream (BYOB-capable) whose queue + // the sink drains into. highWaterMark 0 documents that production is + // write-driven; the eager pushes enqueue regardless of desiredSize + // (unbounded buffering, exactly like the legacy pair). + const readable = new ReadableStream( + { + type: 'bytes', + start: (c: object): void => { + readableController = c; + }, + cancel: (reason: unknown): void => { + // Reader-side cancel tears down the write side, mirroring the + // legacy adapter's cancel → abortWrite path. Erroring a + // closed/errored writable is a spec no-op, so no state check is + // needed. + if (writableController !== undefined) { + writableControllerError(writableController, reason); + } + }, + }, + { highWaterMark: 0 } + ); + + return { + readable: readable as ReadableStreamType, + writable: writable as WritableStreamType, + }; +} + +let assertIsCompressionStream: (self: CompressionStream) => void; +let assertIsDecompressionStream: (self: DecompressionStream) => void; + +class CompressionStream { + #pair: CodecPair; + + static { + assertIsCompressionStream = function (self: CompressionStream) { + if (!isActualObject(self) || !(#pair in self)) + throw new TypeError('Illegal invocation'); + }; + } + + constructor(format: unknown) { + this.#pair = createCodecPair('compress', format); + } + + get readable(): ReadableStreamType { + assertIsCompressionStream(this); + return this.#pair.readable; + } + + get writable(): WritableStreamType { + assertIsCompressionStream(this); + return this.#pair.writable; + } +} + +class DecompressionStream { + #pair: CodecPair; + + static { + assertIsDecompressionStream = function (self: DecompressionStream) { + if (!isActualObject(self) || !(#pair in self)) + throw new TypeError('Illegal invocation'); + }; + } + + constructor(format: unknown) { + this.#pair = createCodecPair('decompress', format); + } + + get readable(): ReadableStreamType { + assertIsDecompressionStream(this); + return this.#pair.readable; + } + + get writable(): WritableStreamType { + assertIsDecompressionStream(this); + return this.#pair.writable; + } +} + +const kEnumerable = { __proto__: null, enumerable: true }; + +ObjectDefineProperties(CompressionStream.prototype, { + __proto__: null, + readable: kEnumerable, + writable: kEnumerable, + [SymbolToStringTag]: { + __proto__: null, + value: 'CompressionStream', + writable: false, + enumerable: false, + configurable: true, + }, +}); + +ObjectDefineProperties(DecompressionStream.prototype, { + __proto__: null, + readable: kEnumerable, + writable: kEnumerable, + [SymbolToStringTag]: { + __proto__: null, + value: 'DecompressionStream', + writable: false, + enumerable: false, + configurable: true, + }, +}); + +module.exports = { + CompressionStream, + DecompressionStream, +}; diff --git a/src/per_isolate/webstreams/streams.ts b/src/per_isolate/webstreams/streams.ts index 16b584dd234..da799705c77 100644 --- a/src/per_isolate/webstreams/streams.ts +++ b/src/per_isolate/webstreams/streams.ts @@ -33,6 +33,11 @@ const { const { TextEncoderStream, TextDecoderStream } = require('webstreams/encoding'); +const { + CompressionStream, + DecompressionStream, +} = require('webstreams/compression'); + module.exports = { ReadableStream, ReadableStreamDefaultReader, @@ -51,6 +56,8 @@ module.exports = { FixedLengthStream, TextEncoderStream, TextDecoderStream, + CompressionStream, + DecompressionStream, // Internal-only reader (the C++ bridge's bulk-read surface). Installed on // globalThis by main.ts ONLY under the internal-testing // expose_draining_reader flag, for exercising expectedLength pass-through diff --git a/src/workerd/api/BUILD.bazel b/src/workerd/api/BUILD.bazel index bbd222d02d7..a63724821f4 100644 --- a/src/workerd/api/BUILD.bazel +++ b/src/workerd/api/BUILD.bazel @@ -274,15 +274,33 @@ wd_cc_library( ], ) +# Shared compression machinery (allocator, and the common codec core): the common home for +# the primitives used by the web CompressionStream/DecompressionStream pair, node:zlib, and +# the TypeScript streams implementation's codec handle. +wd_cc_library( + name = "compression", + srcs = ["compression.c++"], + hdrs = ["compression.h"], + visibility = ["//visibility:public"], + deps = [ + "//src/workerd/io:features", + "//src/workerd/jsg", + "@capnp-cpp//src/kj/compat:kj-brotli", + "@capnp-cpp//src/kj/compat:kj-gzip", + "@nbytes", + "@zstd", + ], +) + wd_cc_library( name = "streams-compression", srcs = ["streams/compression.c++"], hdrs = ["streams/compression.h"], visibility = ["//visibility:public"], deps = [ + ":compression", "//src/workerd/io", "//src/workerd/util:state-machine", - "@nbytes", ], ) diff --git a/src/workerd/api/compression.c++ b/src/workerd/api/compression.c++ new file mode 100644 index 00000000000..bd47b3e643e --- /dev/null +++ b/src/workerd/api/compression.c++ @@ -0,0 +1,742 @@ +// Copyright (c) 2017-2022 Cloudflare, Inc. +// Licensed under the Apache 2.0 license found in the LICENSE file or at: +// https://opensource.org/licenses/Apache-2.0 + +#include "compression.h" + +#include +#include + +#include + +namespace workerd::api { + +CompressionAllocator::CompressionAllocator( + kj::Arc&& externalMemoryTarget) + : externalMemoryTarget(kj::mv(externalMemoryTarget)) {} + +void* CompressionAllocator::AllocForZlib(void* data, uInt items, uInt size) { + size_t real_size = + nbytes::MultiplyWithOverflowCheck(static_cast(items), static_cast(size)); + return AllocForBrotli(data, real_size); +} + +void* CompressionAllocator::AllocForBrotli(void* opaque, size_t size) { + auto* allocator = static_cast(opaque); + auto data = kj::heapArray(size); + auto begin = data.begin(); + + allocator->allocations.insert(begin, + {.data = kj::mv(data), + .memoryAdjustment = allocator->externalMemoryTarget->getAdjustment(size)}); + return begin; +} + +void CompressionAllocator::FreeForZlib(void* opaque, void* pointer) { + if (KJ_UNLIKELY(pointer == nullptr)) return; + auto* allocator = static_cast(opaque); + // No need to destroy memoryAdjustment here. + // Dropping the allocation from the hashmap will defer the adjustment + // until the isolate lock is held. + JSG_REQUIRE(allocator->allocations.erase(pointer), Error, "Zlib allocation should exist"_kj); +} + +// ======================================================================================= +// ZlibStream + +ZlibStream::ZlibStream(CompressionAllocator& allocator) { + stream.zalloc = CompressionAllocator::AllocForZlib; + stream.zfree = CompressionAllocator::FreeForZlib; + stream.opaque = &allocator; +} + +ZlibStream::~ZlibStream() noexcept(false) { + end(); +} + +kj::Maybe ZlibStream::init(Mode mode, Options options) { + KJ_ASSERT(!initialized, "ZlibStream::init() may only be called once"); + this->mode = mode; + int result = [&]() { + switch (mode) { + case Mode::COMPRESS: + return deflateInit2(&stream, options.level, Z_DEFLATED, options.windowBits, + options.memLevel, options.strategy); + case Mode::DECOMPRESS: + return inflateInit2(&stream, options.windowBits); + } + KJ_UNREACHABLE; + }(); + if (result != Z_OK) { + return result; + } + initialized = true; + return kj::none; +} + +kj::Maybe ZlibStream::reset() { + KJ_ASSERT(initialized, "ZlibStream::reset() requires an initialized stream"); + int result = [&]() { + switch (mode) { + case Mode::COMPRESS: + return deflateReset(&stream); + case Mode::DECOMPRESS: + return inflateReset(&stream); + } + KJ_UNREACHABLE; + }(); + if (result != Z_OK) { + return result; + } + return kj::none; +} + +int ZlibStream::end() { + if (!initialized || ended) { + return Z_OK; + } + ended = true; + switch (mode) { + case Mode::COMPRESS: + return deflateEnd(&stream); + case Mode::DECOMPRESS: + return inflateEnd(&stream); + } + KJ_UNREACHABLE; +} + +int ZlibStream::run(int flush) { + KJ_ASSERT(initialized && !ended, "ZlibStream::run() requires a live stream"); + switch (mode) { + case Mode::COMPRESS: + return deflate(&stream, flush); + case Mode::DECOMPRESS: + return inflate(&stream, flush); + } + KJ_UNREACHABLE; +} + +void ZlibStream::setInput(kj::ArrayPtr input) { + // zlib's next_in is non-const for historical reasons; deflate/inflate do not write + // through it. + stream.next_in = const_cast(input.begin()); + stream.avail_in = input.size(); +} + +void ZlibStream::setOutput(kj::ArrayPtr output) { + stream.next_out = output.begin(); + stream.avail_out = output.size(); +} + +size_t ZlibStream::availIn() const { + return stream.avail_in; +} + +size_t ZlibStream::availOut() const { + return stream.avail_out; +} + +kj::StringPtr ZlibStream::msg() const { + if (stream.msg == nullptr) return nullptr; + return kj::StringPtr(stream.msg); +} + +kj::StringPtr ZlibStream::errorCodeName(int code) { + switch (code) { + case Z_OK: + return "Z_OK"_kj; + case Z_STREAM_END: + return "Z_STREAM_END"_kj; + case Z_NEED_DICT: + return "Z_NEED_DICT"_kj; + case Z_ERRNO: + return "Z_ERRNO"_kj; + case Z_STREAM_ERROR: + return "Z_STREAM_ERROR"_kj; + case Z_DATA_ERROR: + return "Z_DATA_ERROR"_kj; + case Z_MEM_ERROR: + return "Z_MEM_ERROR"_kj; + case Z_BUF_ERROR: + return "Z_BUF_ERROR"_kj; + case Z_VERSION_ERROR: + return "Z_VERSION_ERROR"_kj; + default: + return "Z_UNKNOWN_ERROR"_kj; + } +} + +kj::Maybe ZlibStream::windowBitsForWebFormat(kj::StringPtr format) { + // 15 is the default value of the windowBits parameter for zlib; adding 16 selects the + // gzip wrapper, and negating selects a raw (headerless) stream. + if (format == "gzip"_kj) return 15 + 16; + if (format == "deflate"_kj) return 15; + if (format == "deflate-raw"_kj) return -15; + return kj::none; +} + +// ======================================================================================= +// CodecStage + +CodecStage::Context::Context(Mode mode, + kj::StringPtr format, + Flags flags, + kj::Arc&& externalMemoryTarget) + : allocator(kj::mv(externalMemoryTarget)), + stream(allocator), + strictCompression(flags) { + auto windowBits = KJ_ASSERT_NONNULL(ZlibStream::windowBitsForWebFormat(format)); + JSG_REQUIRE(stream.init(mode, ZlibStream::Options{.windowBits = windowBits}) == kj::none, Error, + "Failed to initialize compression context."_kj); +} + +void CodecStage::Context::setInput(const void* in, size_t size) { + stream.setInput(kj::arrayPtr(reinterpret_cast(in), size)); +} + +CodecStage::Context::Result CodecStage::Context::pumpOnce(int flush) { + stream.setOutput(kj::arrayPtr(buffer, sizeof(buffer))); + + int result = stream.run(flush); + + switch (stream.getMode()) { + case Mode::COMPRESS: + JSG_REQUIRE(result == Z_OK || result == Z_BUF_ERROR || result == Z_STREAM_END, TypeError, + "Compression failed."); + break; + case Mode::DECOMPRESS: + JSG_REQUIRE(result == Z_OK || result == Z_BUF_ERROR || result == Z_STREAM_END, TypeError, + "Decompression failed."); + break; + } + + return Result{ + .success = result == Z_OK, + .result = result, + .buffer = kj::arrayPtr(buffer, sizeof(buffer) - stream.availOut()), + }; +} + +void CodecStage::Context::enforceStrictChecks(int flush, const Result& result) { + if (stream.getMode() != Mode::DECOMPRESS || strictCompression != Flags::STRICT) { + return; + } + // The spec requires that a TypeError is produced if there is trailing data after the end + // of the compression stream. Called AFTER the caller has buffered the iteration's output: + // the final valid bytes (produced by the very pump step that observed the trailing junk) + // are still delivered to any read that consumes them before the error lands, which is the + // WPT-pinned observable order. + JSG_REQUIRE(!(result.result == Z_STREAM_END && stream.availIn() > 0), TypeError, + "Trailing bytes after end of compressed data"); + // Same applies to closing a stream before the complete decompressed data is available. + JSG_REQUIRE(!(flush == Z_FINISH && result.result == Z_BUF_ERROR && result.buffer.size() == 0), + TypeError, "Called close() on a decompression stream with incomplete data"); +} + +kj::ArrayPtr CodecStage::LazyBuffer::take(size_t readSize) { + KJ_ASSERT(readSize <= validSize); + kj::ArrayPtr chunk = kj::arrayPtr(&output[output.size() - validSize], readSize); + validSize -= readSize; + return chunk; +} + +void CodecStage::LazyBuffer::maybeShift() { + size_t unusedSpace = output.size() - validSize; + if (unusedSpace >= 1024 && unusedSpace >= (output.size() >> 3)) { + // Shifting buffer to erase data that has already been read. validSize remains the same. + memmove(output.begin(), output.begin() + unusedSpace, validSize); + output.truncate(validSize); + } +} + +void CodecStage::LazyBuffer::write(kj::ArrayPtr chunk) { + output.addAll(chunk); + validSize += chunk.size(); +} + +void CodecStage::LazyBuffer::clear() { + output.clear(); + validSize = 0; +} + +size_t CodecStage::LazyBuffer::size() { + return validSize; +} + +bool CodecStage::LazyBuffer::empty() { + return validSize == 0; +} + +CodecStage::CodecStage(Mode mode, + kj::StringPtr format, + Flags flags, + kj::Arc&& externalMemoryTarget) + : context(mode, format, flags, kj::mv(externalMemoryTarget)) {} + +void CodecStage::push(kj::ArrayPtr input) { + context.setInput(input.begin(), input.size()); + pump(Z_NO_FLUSH); +} + +void CodecStage::end() { + if (finished) return; + finished = true; + pump(Z_FINISH); +} + +size_t CodecStage::pull(kj::ArrayPtr dest) { + auto n = kj::min(dest.size(), output.size()); + if (n == 0) return 0; + dest.write(output.take(n)); + output.maybeShift(); + return n; +} + +size_t CodecStage::available() { + return output.size(); +} + +bool CodecStage::empty() { + return output.empty(); +} + +void CodecStage::clear() { + output.clear(); +} + +void CodecStage::pump(int flush) { + while (true) { + auto result = context.pumpOnce(flush); + // Buffer any produced output BEFORE the strict checks run: an iteration can both produce + // the stream's final bytes and observe the strict-mode error condition (e.g. trailing + // junk after the end of the compressed data), and the bytes must remain deliverable. + if (result.buffer.size() > 0) { + output.write(result.buffer); + } + context.enforceStrictChecks(flush, result); + if (result.buffer.size() == 0) { + if (result.success) { + // No output produced but input data has been processed based on the zlib return + // code; call pumpOnce again. + continue; + } + return; + } + } + KJ_UNREACHABLE; +} + +// ======================================================================================= +// CompressionCodec + +CompressionCodec::CompressionCodec(CodecStage::Mode mode, + kj::StringPtr format, + CodecStage::Flags flags, + kj::Arc&& externalMemoryTarget) + : stage(mode, format, flags, kj::mv(externalMemoryTarget)) {} + +void CompressionCodec::push(jsg::JsBufferSource chunk) { + stage.push(chunk.asArrayPtr()); +} + +void CompressionCodec::end() { + stage.end(); +} + +uint32_t CompressionCodec::pullInto(jsg::JsBufferSource view) { + return static_cast(stage.pull(view.asArrayPtr())); +} + +double CompressionCodec::available() { + return static_cast(stage.available()); +} + +void newCompressionCodecCallback(const v8::FunctionCallbackInfo& info) { + // liftKj converts thrown kj/jsg exceptions (e.g. the validation TypeErrors below) into JS + // exceptions (without it they would escape the raw callback and take down the process) and + // sets the returned value as the callback's return value. + jsg::liftKj(info, [&]() -> v8::Local { + auto& js = jsg::Lock::from(info.GetIsolate()); + + auto modeStr = JSG_REQUIRE_NONNULL(jsg::JsValue(info[0]).tryCast(), TypeError, + "newCompressionCodec() expects a string mode argument"); + auto formatStr = JSG_REQUIRE_NONNULL(jsg::JsValue(info[1]).tryCast(), TypeError, + "newCompressionCodec() expects a string format argument"); + auto mode = modeStr.toString(js); + auto format = formatStr.toString(js); + + JSG_REQUIRE(format == "deflate" || format == "gzip" || format == "deflate-raw", TypeError, + "The compression format must be either 'deflate', 'deflate-raw' or 'gzip'."); + CodecStage::Mode codecMode; + CodecStage::Flags codecFlags = CodecStage::Flags::NONE; + if (mode == "compress") { + codecMode = CodecStage::Mode::COMPRESS; + } else if (mode == "decompress") { + codecMode = CodecStage::Mode::DECOMPRESS; + if (FeatureFlags::get(js).getStrictCompression()) { + codecFlags = CodecStage::Flags::STRICT; + } + } else { + JSG_FAIL_REQUIRE(TypeError, "The codec mode must be either 'compress' or 'decompress'."); + } + + auto& handler = KJ_ASSERT_NONNULL(js.tryGetTypeHandler>()); + return handler.wrap(js, + js.alloc(codecMode, format, codecFlags, js.getExternalMemoryTarget())); + }); +} + +// ======================================================================================= +// Brotli / Zstd contexts + +void BrotliContext::setBuffers(kj::ArrayPtr input, kj::ArrayPtr output) { + nextIn = reinterpret_cast(input.begin()); + nextOut = output.begin(); + availIn = input.size(); + availOut = output.size(); +} + +void BrotliContext::setInputBuffer(kj::ArrayPtr input) { + nextIn = input.begin(); + availIn = input.size(); +} + +void BrotliContext::setOutputBuffer(kj::ArrayPtr output) { + nextOut = output.begin(); + availOut = output.size(); +} + +uint BrotliContext::getAvailOut() const { + return availOut; +} + +void BrotliContext::setFlush(int _flush) { + flush = static_cast(_flush); +} + +void BrotliContext::getAfterWriteResult(uint32_t* _availIn, uint32_t* _availOut) const { + *_availIn = availIn; + *_availOut = availOut; +} + +BrotliEncoderContext::BrotliEncoderContext(CompressionAllocator& allocator, ZlibMode _mode) + : BrotliContext(allocator, _mode) { + // NOTE: Ignores any returned errors. + // TODO(soon): It's possible that initialization doesn't need to happen until `initialize` is + // called elsewhere. I'm keeping it like this to avoid changing the existing behaviour. + auto _ = initialize(); +} + +void BrotliEncoderContext::work() { + JSG_REQUIRE(mode == ZlibMode::BROTLI_ENCODE, Error, "Mode should be BROTLI_ENCODE"_kj); + JSG_REQUIRE_NONNULL(state.get(), Error, "State should not be empty"_kj); + + const uint8_t* internalNext = nextIn; + lastResult = BrotliEncoderCompressStream( + state.get(), flush, &availIn, &internalNext, &availOut, &nextOut, nullptr); + nextIn += internalNext - nextIn; + + streamEnd = lastResult && BrotliEncoderIsFinished(state.get()); +} + +kj::Maybe BrotliEncoderContext::initialize() { + auto instance = BrotliEncoderCreateInstance( + CompressionAllocator::AllocForBrotli, CompressionAllocator::FreeForZlib, &allocator); + state = kj::disposeWith(kj::mv(instance)); + + if (state.get() == nullptr) { + return CompressionError( + "Could not initialize Brotli instance"_kj, "ERR_ZLIB_INITIALIZATION_FAILED"_kj, -1); + } + + return kj::none; +} + +kj::Maybe BrotliEncoderContext::resetStream() { + return initialize(); +} + +kj::Maybe BrotliEncoderContext::setParams(int key, uint32_t value) { + if (!BrotliEncoderSetParameter(state.get(), static_cast(key), value)) { + return CompressionError("Setting parameter failed", "ERR_BROTLI_PARAM_SET_FAILED", -1); + } + + return kj::none; +} + +kj::Maybe BrotliEncoderContext::getError() const { + if (!lastResult) { + return CompressionError("Compression failed", "ERR_BROTLI_COMPRESSION_FAILED", -1); + } + + return kj::none; +} + +bool BrotliEncoderContext::isStreamEnd() const { + return streamEnd; +} + +BrotliDecoderContext::BrotliDecoderContext(CompressionAllocator& allocator, ZlibMode _mode) + : BrotliContext(allocator, _mode) { + // NOTE: Ignores any returned errors. + // TODO(soon): It's possible that initialization doesn't need to happen until `initialize` is + // called elsewhere. I'm keeping it like this to avoid changing the existing behaviour. + auto _ = initialize(); +} + +kj::Maybe BrotliDecoderContext::initialize() { + auto instance = BrotliDecoderCreateInstance( + CompressionAllocator::AllocForBrotli, CompressionAllocator::FreeForZlib, &allocator); + state = kj::disposeWith(kj::mv(instance)); + + if (state.get() == nullptr) { + return CompressionError( + "Could not initialize Brotli instance", "ERR_ZLIB_INITIALIZATION_FAILED", -1); + } + + return kj::none; +} + +void BrotliDecoderContext::work() { + JSG_REQUIRE(mode == ZlibMode::BROTLI_DECODE, Error, "Mode should have been BROTLI_DECODE"_kj); + JSG_REQUIRE_NONNULL(state.get(), Error, "State should not be empty"_kj); + const uint8_t* internalNext = nextIn; + lastResult = BrotliDecoderDecompressStream( + state.get(), &availIn, &internalNext, &availOut, &nextOut, nullptr); + nextIn += internalNext - nextIn; + + if (lastResult == BROTLI_DECODER_RESULT_ERROR) { + error = BrotliDecoderGetErrorCode(state.get()); + errorString = kj::str("ERR_", BrotliDecoderErrorString(error)); + } +} + +kj::Maybe BrotliDecoderContext::resetStream() { + return initialize(); +} + +kj::Maybe BrotliDecoderContext::setParams(int key, uint32_t value) { + if (!BrotliDecoderSetParameter(state.get(), static_cast(key), value)) { + return CompressionError("Setting parameter failed", "ERR_BROTLI_PARAM_SET_FAILED", -1); + } + + return kj::none; +} + +kj::Maybe BrotliDecoderContext::getError() const { + if (error != BROTLI_DECODER_NO_ERROR) { + return CompressionError("Compression failed", errorString, -1); + } + + if (flush == BROTLI_OPERATION_FINISH && lastResult == BROTLI_DECODER_RESULT_NEEDS_MORE_INPUT) { + // Match zlib behavior, as brotli doesn't have its own code for this. + return CompressionError("Unexpected end of file", "Z_BUF_ERROR", Z_BUF_ERROR); + } + + return kj::none; +} + +bool BrotliDecoderContext::isStreamEnd() const { + return lastResult == BROTLI_DECODER_RESULT_SUCCESS; +} + +// ======================================================================================= +// Zstd Implementation + +void ZstdContext::setBuffers(kj::ArrayPtr input, kj::ArrayPtr output) { + setInputBuffer(input); + setOutputBuffer(output); +} + +void ZstdContext::setInputBuffer(kj::ArrayPtr input) { + input_.src = input.begin(); + input_.size = input.size(); + input_.pos = 0; +} + +void ZstdContext::setOutputBuffer(kj::ArrayPtr output) { + output_.dst = output.begin(); + output_.size = output.size(); + output_.pos = 0; +} + +void ZstdContext::setFlush(int flush) { + KJ_DASSERT(flush >= ZSTD_e_continue && flush <= ZSTD_e_end, + "flush must be a valid ZSTD_EndDirective value"); + flush_ = static_cast(flush); +} + +kj::uint ZstdContext::getAvailOut() const { + return output_.size - output_.pos; +} + +void ZstdContext::getAfterWriteResult(uint32_t* availIn, uint32_t* availOut) const { + *availIn = input_.size - input_.pos; + *availOut = output_.size - output_.pos; +} + +namespace { +// Helper to check ZSTD errors and return a CompressionError if present. +// Also sets the error code in the provided reference for later retrieval. +kj::Maybe zstdCheckError( + size_t result, ZSTD_ErrorCode& error, kj::StringPtr errorCode) { + if (ZSTD_isError(result)) { + error = ZSTD_getErrorCode(result); + return CompressionError(ZSTD_getErrorName(result), errorCode, -1); + } + return kj::none; +} + +// Wrappers for ZSTD free functions that return void (for use with kj::disposeWith). +void zstdFreeCCtx(ZSTD_CCtx* cctx) { + ZSTD_freeCCtx(cctx); +} +void zstdFreeDCtx(ZSTD_DCtx* dctx) { + ZSTD_freeDCtx(dctx); +} +} // namespace + +ZstdEncoderContext::ZstdEncoderContext(ZlibMode _mode) + : ZstdContext(_mode), + cctx_(kj::disposeWith(ZSTD_createCCtx())) {} + +kj::Maybe ZstdEncoderContext::initialize(uint64_t pledgedSrcSize) { + if (cctx_.get() == nullptr) { + return CompressionError( + "Could not initialize Zstd instance"_kj, "ERR_ZLIB_INITIALIZATION_FAILED"_kj, -1); + } + + if (pledgedSrcSize != ZSTD_CONTENTSIZE_UNKNOWN) { + size_t result = ZSTD_CCtx_setPledgedSrcSize(cctx_.get(), pledgedSrcSize); + KJ_IF_SOME(err, zstdCheckError(result, error_, "ERR_ZSTD_COMPRESSION_FAILED"_kj)) { + return kj::mv(err); + } + } + + return kj::none; +} + +void ZstdEncoderContext::work() { + JSG_REQUIRE(mode == ZlibMode::ZSTD_ENCODE, Error, "Mode should be ZSTD_ENCODE"_kj); + JSG_REQUIRE(cctx_.get() != nullptr, Error, "Zstd context should not be null"_kj); + + lastResult = ZSTD_compressStream2(cctx_.get(), &output_, &input_, flush_); + + if (ZSTD_isError(lastResult)) { + error_ = ZSTD_getErrorCode(lastResult); + } +} + +kj::Maybe ZstdEncoderContext::resetStream() { + if (cctx_.get() != nullptr) { + size_t result = ZSTD_CCtx_reset(cctx_.get(), ZSTD_reset_session_only); + KJ_IF_SOME(err, zstdCheckError(result, error_, "ERR_ZSTD_COMPRESSION_FAILED"_kj)) { + return kj::mv(err); + } + } + return kj::none; +} + +kj::Maybe ZstdEncoderContext::setParams(int key, int value) { + KJ_DASSERT(key >= ZSTD_c_compressionLevel, + "key must be a valid ZSTD_cParameter (first valid value is ZSTD_c_compressionLevel)"); + size_t result = ZSTD_CCtx_setParameter(cctx_.get(), static_cast(key), value); + if (ZSTD_isError(result)) { + return CompressionError(kj::str("Setting parameter failed: ", ZSTD_getErrorName(result)), + "ERR_ZSTD_PARAM_SET_FAILED"_kj, -1); + } + return kj::none; +} + +kj::Maybe ZstdEncoderContext::getError() const { + if (error_ != ZSTD_error_no_error) { + return CompressionError(kj::str("Zstd compression failed: ", ZSTD_getErrorString(error_)), + kj::str("ERR_ZSTD_COMPRESSION_FAILED"), -1); + } + + if (flush_ == ZSTD_e_end && lastResult != 0) { + // lastResult > 0 means more output is needed, which shouldn't happen at end + return CompressionError("Unexpected end of file"_kj, "Z_BUF_ERROR"_kj, Z_BUF_ERROR); + } + + return kj::none; +} + +bool ZstdEncoderContext::isStreamEnd() const { + // ZSTD_compressStream2 returns 0 when flush_ == ZSTD_e_end and the frame is fully flushed. + return !ZSTD_isError(lastResult) && lastResult == 0; +} + +ZstdDecoderContext::ZstdDecoderContext(ZlibMode _mode) + : ZstdContext(_mode), + dctx_(kj::disposeWith(ZSTD_createDCtx())) {} + +kj::Maybe ZstdDecoderContext::initialize() { + // dctx_ is created in the constructor. It can only be nullptr if ZSTD_createDCtx() + // failed due to memory allocation failure. + if (dctx_.get() == nullptr) { + return CompressionError( + "Could not initialize Zstd instance"_kj, "ERR_ZLIB_INITIALIZATION_FAILED"_kj, -1); + } + + return kj::none; +} + +void ZstdDecoderContext::work() { + JSG_REQUIRE(mode == ZlibMode::ZSTD_DECODE, Error, "Mode should be ZSTD_DECODE"_kj); + JSG_REQUIRE(dctx_.get() != nullptr, Error, "Zstd context should not be null"_kj); + + lastResult = ZSTD_decompressStream(dctx_.get(), &output_, &input_); + + if (ZSTD_isError(lastResult)) { + error_ = ZSTD_getErrorCode(lastResult); + } else if (input_.size > 0) { + // Track whether we're mid-frame: lastResult > 0 means more data needed, + // lastResult == 0 means frame is complete. + frameInProgress_ = (lastResult > 0); + } +} + +kj::Maybe ZstdDecoderContext::resetStream() { + if (dctx_.get() != nullptr) { + size_t result = ZSTD_DCtx_reset(dctx_.get(), ZSTD_reset_session_only); + KJ_IF_SOME(err, zstdCheckError(result, error_, "ERR_ZSTD_DECOMPRESSION_FAILED"_kj)) { + return kj::mv(err); + } + } + frameInProgress_ = false; + return kj::none; +} + +kj::Maybe ZstdDecoderContext::setParams(int key, int value) { + KJ_DASSERT(dctx_.get() != nullptr, "Zstd decompression context should not be null"); + size_t result = ZSTD_DCtx_setParameter(dctx_.get(), static_cast(key), value); + if (ZSTD_isError(result)) { + return CompressionError(kj::str("Setting parameter failed: ", ZSTD_getErrorName(result)), + "ERR_ZSTD_PARAM_SET_FAILED"_kj, -1); + } + return kj::none; +} + +kj::Maybe ZstdDecoderContext::getError() const { + if (error_ != ZSTD_error_no_error) { + return CompressionError(kj::str("Zstd decompression failed: ", ZSTD_getErrorString(error_)), + kj::str("ERR_ZSTD_DECOMPRESSION_FAILED"), -1); + } + + // If this is the final flush, we're mid-frame (frame was started but never + // completed), and the output buffer is not full (decoder had space but + // couldn't produce more output), the input was truncated. + if (flush_ == ZSTD_e_end && frameInProgress_ && output_.pos < output_.size) { + return CompressionError("unexpected end of file"_kj, "ERR_ZSTD_DECOMPRESSION_FAILED"_kj, -1); + } + + return kj::none; +} + +bool ZstdDecoderContext::isStreamEnd() const { + // ZSTD_decompressStream returns 0 when a frame is completely decoded and fully flushed. + return !ZSTD_isError(lastResult) && lastResult == 0; +} + +} // namespace workerd::api diff --git a/src/workerd/api/compression.h b/src/workerd/api/compression.h new file mode 100644 index 00000000000..97d6f928dfc --- /dev/null +++ b/src/workerd/api/compression.h @@ -0,0 +1,516 @@ +// Copyright (c) 2017-2022 Cloudflare, Inc. +// Licensed under the Apache 2.0 license found in the LICENSE file or at: +// https://opensource.org/licenses/Apache-2.0 + +#pragma once + +// Shared compression machinery. This is the common home for the compression primitives used +// by the three JavaScript-facing compression surfaces: +// +// - the web CompressionStream/DecompressionStream pair (api/streams/compression.{h,c++}), +// - the node:zlib module (api/node/zlib-util.{h,c++}), +// - the TypeScript streams implementation's pair (src/per_isolate/webstreams/), which +// drives the synchronous codec through a handle minted by the C++ side. +// +// The consumers keep their own semantics (spec-pinned TypeErrors vs. Node-fidelity error +// codes vs. TS orchestration); what lives here is mechanism. The fetch content-encoding path +// (system-streams.c++) wraps kj's async gzip/brotli streams directly and does not use this. + +#include + +#include +#include +#include +#include +#include + +namespace workerd::api { + +// A custom allocator to be used by the zlib and brotli libraries. +// The allocator should not and can not safely hold a reference to the jsg::Lock +// instance. Therefore, we lookup the current jsg::Lock instance from the +// isolate pointer and use that to get the external memory adjustment. +class CompressionAllocator final { + public: + CompressionAllocator(kj::Arc&& externalMemoryTarget); + + static void* AllocForZlib(void* data, uInt items, uInt size); + static void* AllocForBrotli(void* data, size_t size); + static void FreeForZlib(void* data, void* pointer); + + private: + struct Allocation { + kj::Array data; + kj::Maybe memoryAdjustment = kj::none; + }; + + kj::Arc externalMemoryTarget; + kj::HashMap allocations; +}; + +// The shared z_stream wrapper: owns the stream structure and its init/reset/end lifecycle, +// the input/output buffer plumbing, and the raw deflate()/inflate() step. Mechanism only -- +// consumers interpret the returned zlib codes according to their own policies (the web pair +// translates to its spec-pinned TypeErrors; node:zlib to Node-fidelity CompressionError +// codes), and node-specific zlib features (dictionaries, deflateParams) reach the structure +// through raw() until they grow shared consumers. +class ZlibStream final { + public: + enum class Mode { COMPRESS, DECOMPRESS }; + + struct Options { + // Final windowBits, already including any format adjustment (e.g. +16 for gzip, + // negated for raw); see windowBitsForWebFormat() and the node mode adjustments. + int windowBits; + // The remaining options apply to COMPRESS only. + int level = Z_DEFAULT_COMPRESSION; + int memLevel = 8; + int strategy = Z_DEFAULT_STRATEGY; + }; + + // The allocator must outlive this object. Construction only wires the allocation hooks; + // the underlying stream is not initialized until init() (node initializes lazily). + explicit ZlibStream(CompressionAllocator& allocator); + KJ_DISALLOW_COPY_AND_MOVE(ZlibStream); + + // Ends the stream if initialized, ignoring the result; consumers that need an + // error-checked shutdown (node) call end() explicitly first. + ~ZlibStream() noexcept(false); + + // deflateInit2/inflateInit2. Returns the zlib error code on failure, kj::none on success. + // May be called at most once. + kj::Maybe init(Mode mode, Options options); + + // deflateReset/inflateReset. Precondition: initialized. + kj::Maybe reset(); + + // deflateEnd/inflateEnd. Returns the zlib code (Z_OK if never initialized or already + // ended). Idempotent. + int end(); + + // One deflate()/inflate() step with the given flush mode, over the buffers established by + // setInput()/setOutput() (tracked in the stream's next/avail fields across calls). + // Precondition: initialized. + int run(int flush); + + void setInput(kj::ArrayPtr input); + void setOutput(kj::ArrayPtr output); + size_t availIn() const; + size_t availOut() const; + + // The stream's current error message (stream.msg), or empty if none. + kj::StringPtr msg() const; + + bool isInitialized() const { + return initialized; + } + Mode getMode() const { + return mode; + } + + // Escape hatch for consumer-specific zlib calls that take the z_stream directly + // (deflateSetDictionary, inflateSetDictionary, deflateParams, ...). Precondition: + // initialized (except for consumers wiring additional fields pre-init). + z_stream& raw() { + return stream; + } + + // The canonical name for a zlib return code (e.g. "Z_DATA_ERROR"); "Z_UNKNOWN_ERROR" for + // unrecognized codes. + static kj::StringPtr errorCodeName(int code); + + // Maps a Compression Streams spec format ("gzip" | "deflate" | "deflate-raw") to its + // windowBits value; kj::none for anything else. + static kj::Maybe windowBitsForWebFormat(kj::StringPtr format); + + private: + z_stream stream = {}; + Mode mode = Mode::COMPRESS; + bool initialized = false; + bool ended = false; +}; + +// The synchronous codec stage behind the web Compression Streams pairs (both the legacy C++ +// frontend in api/streams/compression.c++ and the TypeScript streams implementation's +// frontend): the Compression Streams spec policy — spec-pinned TypeErrors and the +// strict_compression_checks handling — over the shared ZlibStream core, plus the stage's own +// output buffer. Frontends are orchestration shells around one of these; ALL asynchrony is +// frontend-owned. +// +// PACING IS DELIBERATELY EAGER: push() runs the codec over the whole input chunk +// synchronously, accumulating ALL produced output in the stage buffer. The spec runs the +// codec inside the TransformStream transform()/flush() algorithms, so corrupt input MUST +// reject the write (and a strict-mode incomplete stream MUST reject the close) — that error +// timing is observable and WPT-pinned. Demand-paced production (pump only what a reader +// asked for) is therefore NOT valid on any frontend where write settlement is observable; it +// remains a possible future policy for fused native pipelines that own both ends. +class CodecStage final { + public: + using Mode = ZlibStream::Mode; + + enum class Flags { + NONE, + // The strict_compression_checks compat flag's decompression checks (trailing data after + // end of stream; close with incomplete data). + STRICT, + }; + + // `format` must be a valid web format ("gzip" | "deflate" | "deflate-raw"); the + // JS-visible format validation (with its spec-pinned TypeError) belongs to the frontends. + explicit CodecStage(Mode mode, + kj::StringPtr format, + Flags flags, + kj::Arc&& externalMemoryTarget); + KJ_DISALLOW_COPY_AND_MOVE(CodecStage); + + // Runs the codec over one input chunk to exhaustion, synchronously. The input is fully + // consumed before this returns (zlib copies what it needs into its own window), so the + // caller's buffer is not retained. Produced output accumulates in the stage buffer. + // Throws on codec error — the caller owns its own state/teardown response. + void push(kj::ArrayPtr input); + + // Finishes the codec (Z_FINISH), which also runs the strict-mode end checks for + // decompression (trailing data / incomplete stream). Idempotent: repeat calls are no-ops, + // preserving the historical allowance for multiple end() calls. + void end(); + + // Copies up to dest.size() buffered output bytes into dest, returning the count copied. + size_t pull(kj::ArrayPtr dest); + + size_t available(); + bool empty(); + + // Teardown (frontend cancel/abort path): drops all buffered output. + void clear(); + + private: + // The per-pump policy layer: one deflate()/inflate() step into the scratch buffer, with + // the spec's TypeErrors applied to the result. The strict-mode checks are a separate step + // (enforceStrictChecks) so the stage can buffer an erroring iteration's output BEFORE the + // strict error throws — the final valid bytes are still deliverable, per the WPT-pinned + // output-then-error order. + class Context { + public: + struct Result { + bool success = false; + int result = Z_OK; + kj::ArrayPtr buffer; + }; + + explicit Context(Mode mode, + kj::StringPtr format, + Flags flags, + kj::Arc&& externalMemoryTarget); + KJ_DISALLOW_COPY_AND_MOVE(Context); + + void setInput(const void* in, size_t size); + Result pumpOnce(int flush); + void enforceStrictChecks(int flush, const Result& result); + + private: + CompressionAllocator allocator; + ZlibStream stream; + kj::byte buffer[16384]; + + // For the eponymous compatibility flag + Flags strictCompression; + }; + + // Buffer class based on kj::Vector that erases data that has been read from it lazily to + // avoid excessive copying when reading a larger amount of buffered data in small chunks. + // validSize is used to track the amount of data that has not been read back yet. + class LazyBuffer { + public: + // Return a chunk of data and mark it as invalid. The returned chunk remains valid until + // data is shifted, cleared or destructor is called. maybeShift() should be called after + // the returned data has been processed. + kj::ArrayPtr take(size_t readSize); + + // Shift the output only if doing so results in reducing vector size by at least 1 KiB + // and 1/8 of its size to avoid copying for small reads. + void maybeShift(); + + void write(kj::ArrayPtr chunk); + void clear(); + + // The size of the valid data that has not been read back yet. This may be smaller than + // the size of the internal vector, which is not relevant to consumers. + size_t size(); + bool empty(); + + private: + kj::Vector output; + size_t validSize = 0; + }; + + void pump(int flush); + + Context context; + LazyBuffer output; + bool finished = false; +}; + +// The synchronous codec handle for the TypeScript streams implementation's +// CompressionStream/DecompressionStream pair (webstreams/compression.ts): a thin internal +// JSG resource over CodecStage. Minted exclusively by the bootstrap's +// utils.newCompressionCodec() (see newCompressionCodecCallback below and +// per-isolate-bootstrap.c++); never registered as a global or nested type, so instances are +// reachable only by the bootstrap module that created them — user code cannot obtain one, +// and the type is kept out of the generated TypeScript types. +// +// All methods are synchronous and IoContext-free: compression is pure CPU, so the +// TypeScript pair's waiting is entirely V8 promise machinery, and global-scope construction +// is legal. External-memory accounting rides the stage's CompressionAllocator. +class CompressionCodec final: public jsg::Object { + public: + CompressionCodec(CodecStage::Mode mode, + kj::StringPtr format, + CodecStage::Flags flags, + kj::Arc&& externalMemoryTarget); + + // Runs the codec over the chunk to exhaustion, synchronously and eagerly; the chunk is + // fully consumed and not retained. Throws on codec error (rejecting the TS pair's write — + // the spec's transform-time error timing). + void push(jsg::JsBufferSource chunk); + + // Z_FINISH plus the strict-mode end checks; throws reject the TS pair's close. Idempotent. + void end(); + + // Copies up to view.size() buffered output bytes into the view, returning the count. + uint32_t pullInto(jsg::JsBufferSource view); + + // double rather than uint32: a decompression stage buffer can in principle exceed uint32 + // range (the legacy implementation had the same unbounded buffering); JS numbers carry + // the full size exactly. + double available(); + + JSG_RESOURCE_TYPE(CompressionCodec) { + JSG_METHOD(push); + JSG_METHOD(end); + JSG_METHOD(pullInto); + JSG_METHOD(available); + + // Internal plumbing type: keep it out of the generated TypeScript types. + JSG_TS_OVERRIDE(type CompressionCodec = never); + } + + private: + CodecStage stage; +}; + +// The raw v8 callback behind the bootstrap's utils.newCompressionCodec(mode, format): +// validates the mode and format (with the spec TypeErrors the legacy constructors use), +// applies the strict_compression_checks compat flag for decompression, and allocates a +// CompressionCodec. Defined here (rather than in per-isolate-bootstrap.c++) so the +// compression knowledge stays with the machinery; the bootstrap only wires the utils member +// name to this callback. +void newCompressionCodecCallback(const v8::FunctionCallbackInfo& info); + +// ======================================================================================= +// Codec mode plumbing and the brotli/zstd context families. +// +// These follow Node.js' structure (node:zlib is their consumer today) but live here +// because they are compression machinery, not Node bindings: CompressionStream/ +// DecompressionStream are anticipated to grow brotli and zstd format support, at which +// point they gain web frontends over the same contexts. + +using ZlibModeValue = uint8_t; +enum class ZlibMode : ZlibModeValue { + NONE, + DEFLATE, + INFLATE, + GZIP, + GUNZIP, + DEFLATERAW, + INFLATERAW, + UNZIP, + BROTLI_DECODE, + BROTLI_ENCODE, + ZSTD_ENCODE, + ZSTD_DECODE +}; + +struct CompressionError { + CompressionError(kj::StringPtr _message, kj::StringPtr _code, int _err) + : message(kj::str(_message)), + code(kj::str(_code)), + err(_err) { + JSG_REQUIRE(message.size() != 0, Error, "Compression error message should not be null"); + } + + kj::String message; + kj::String code; + int err; +}; + +class BrotliContext { + public: + explicit BrotliContext(CompressionAllocator& allocator, ZlibMode _mode) + : allocator(allocator), + mode(_mode) {} + KJ_DISALLOW_COPY(BrotliContext); + void setBuffers(kj::ArrayPtr input, kj::ArrayPtr output); + void setInputBuffer(kj::ArrayPtr input); + void setOutputBuffer(kj::ArrayPtr output); + void setFlush(int flush); + kj::uint getAvailOut() const; + void getAfterWriteResult(uint32_t* availIn, uint32_t* availOut) const; + void setMode(ZlibMode _mode) { + mode = _mode; + } + + void clearBuffers() { + nextIn = nullptr; + nextOut = nullptr; + availIn = 0; + availOut = 0; + } + + struct Options { + jsg::Optional flush; + jsg::Optional finishFlush; + jsg::Optional chunkSize; + jsg::Optional> params; + jsg::Optional maxOutputLength; + JSG_STRUCT(flush, finishFlush, chunkSize, params, maxOutputLength); + }; + + protected: + CompressionAllocator& allocator; + ZlibMode mode; + const uint8_t* nextIn = nullptr; + uint8_t* nextOut = nullptr; + size_t availIn = 0; + size_t availOut = 0; + BrotliEncoderOperation flush = BROTLI_OPERATION_PROCESS; +}; + +class BrotliEncoderContext final: public BrotliContext { + public: + static const ZlibMode Mode = ZlibMode::BROTLI_ENCODE; + explicit BrotliEncoderContext(CompressionAllocator& allocator, ZlibMode _mode); + + KJ_DISALLOW_COPY_AND_MOVE(BrotliEncoderContext); + + // Equivalent to Node.js' `DoThreadPoolWork` implementation. + void work(); + kj::Maybe initialize(); + kj::Maybe resetStream(); + kj::Maybe setParams(int key, uint32_t value); + kj::Maybe getError() const; + bool isStreamEnd() const; + + private: + bool lastResult = false; + bool streamEnd = false; + kj::Own state; +}; + +class BrotliDecoderContext final: public BrotliContext { + public: + static const ZlibMode Mode = ZlibMode::BROTLI_DECODE; + explicit BrotliDecoderContext(CompressionAllocator& allocator, ZlibMode _mode); + + KJ_DISALLOW_COPY_AND_MOVE(BrotliDecoderContext); + + // Equivalent to Node.js' `DoThreadPoolWork` implementation. + void work(); + kj::Maybe initialize(); + kj::Maybe resetStream(); + kj::Maybe setParams(int key, uint32_t value); + kj::Maybe getError() const; + bool isStreamEnd() const; + + private: + BrotliDecoderResult lastResult = BROTLI_DECODER_RESULT_SUCCESS; + BrotliDecoderErrorCode error = BROTLI_DECODER_NO_ERROR; + kj::String errorString; + kj::Own state; +}; + +class ZstdContext { + public: + explicit ZstdContext(ZlibMode _mode): mode(_mode) {} + KJ_DISALLOW_COPY(ZstdContext); + + void setBuffers(kj::ArrayPtr input, kj::ArrayPtr output); + void setInputBuffer(kj::ArrayPtr input); + void setOutputBuffer(kj::ArrayPtr output); + void setFlush(int flush); + kj::uint getAvailOut() const; + void getAfterWriteResult(uint32_t* availIn, uint32_t* availOut) const; + void setMode(ZlibMode _mode) { + mode = _mode; + } + + void clearBuffers() { + input_ = {nullptr, 0, 0}; + output_ = {nullptr, 0, 0}; + } + + struct Options { + jsg::Optional flush; + jsg::Optional finishFlush; + jsg::Optional chunkSize; + jsg::Optional> params; + jsg::Optional maxOutputLength; + jsg::Optional pledgedSrcSize; + JSG_STRUCT(flush, finishFlush, chunkSize, params, maxOutputLength, pledgedSrcSize); + }; + + protected: + ZlibMode mode; + ZSTD_inBuffer input_{nullptr, 0, 0}; + ZSTD_outBuffer output_{nullptr, 0, 0}; + ZSTD_EndDirective flush_ = ZSTD_e_continue; +}; + +class ZstdEncoderContext final: public ZstdContext { + public: + static const ZlibMode Mode = ZlibMode::ZSTD_ENCODE; + explicit ZstdEncoderContext(ZlibMode _mode); + explicit ZstdEncoderContext(CompressionAllocator& _allocator, ZlibMode _mode) + : ZstdEncoderContext(_mode) {} + KJ_DISALLOW_COPY_AND_MOVE(ZstdEncoderContext); + + void work(); + kj::Maybe initialize(uint64_t pledgedSrcSize); + kj::Maybe resetStream(); + kj::Maybe setParams(int key, int value); + kj::Maybe getError() const; + bool isStreamEnd() const; + + private: + size_t lastResult = 0; + kj::Own cctx_; + ZSTD_ErrorCode error_ = ZSTD_error_no_error; +}; + +class ZstdDecoderContext final: public ZstdContext { + public: + static const ZlibMode Mode = ZlibMode::ZSTD_DECODE; + explicit ZstdDecoderContext(ZlibMode _mode); + explicit ZstdDecoderContext(CompressionAllocator& _allocator, ZlibMode _mode) + : ZstdDecoderContext(_mode) {} + KJ_DISALLOW_COPY_AND_MOVE(ZstdDecoderContext); + + void work(); + kj::Maybe initialize(); + kj::Maybe resetStream(); + kj::Maybe setParams(int key, int value); + kj::Maybe getError() const; + bool isStreamEnd() const; + + private: + size_t lastResult = 0; + kj::Own dctx_; + ZSTD_ErrorCode error_ = ZSTD_error_no_error; + bool frameInProgress_ = false; +}; + +} // namespace workerd::api + +KJ_DECLARE_NON_POLYMORPHIC(BrotliEncoderStateStruct) +KJ_DECLARE_NON_POLYMORPHIC(BrotliDecoderStateStruct) +KJ_DECLARE_NON_POLYMORPHIC(ZSTD_CCtx) +KJ_DECLARE_NON_POLYMORPHIC(ZSTD_DCtx) diff --git a/src/workerd/api/node/BUILD.bazel b/src/workerd/api/node/BUILD.bazel index cb5b8db9050..c3c62b480bd 100644 --- a/src/workerd/api/node/BUILD.bazel +++ b/src/workerd/api/node/BUILD.bazel @@ -66,7 +66,7 @@ wd_cc_library( ":node-core", "//src/node", "//src/rust/api", - "//src/workerd/api:streams-compression", + "//src/workerd/api:compression", "//src/workerd/io", "//src/workerd/util:autogate", "//src/workerd/util:mimetype", diff --git a/src/workerd/api/node/zlib-util.c++ b/src/workerd/api/node/zlib-util.c++ index 9cf394a66c9..a5d7dc60d42 100644 --- a/src/workerd/api/node/zlib-util.c++ +++ b/src/workerd/api/node/zlib-util.c++ @@ -178,7 +178,7 @@ kj::Maybe ZlibContext::getError() const { switch (err) { case Z_OK: case Z_BUF_ERROR: - if (stream.avail_out != 0 && flush == Z_FINISH) { + if (core.availOut() != 0 && flush == Z_FINISH) { return constructError("unexpected end of file"_kj); } break; @@ -209,10 +209,10 @@ kj::Maybe ZlibContext::setDictionary() { switch (mode) { case ZlibMode::DEFLATE: case ZlibMode::DEFLATERAW: - err = deflateSetDictionary(&stream, dictionary.begin(), dictionary.size()); + err = deflateSetDictionary(&core.raw(), dictionary.begin(), dictionary.size()); break; case ZlibMode::INFLATERAW: - err = inflateSetDictionary(&stream, dictionary.begin(), dictionary.size()); + err = inflateSetDictionary(&core.raw(), dictionary.begin(), dictionary.size()); break; default: break; @@ -226,40 +226,44 @@ kj::Maybe ZlibContext::setDictionary() { } bool ZlibContext::initializeZlib() { - if (initialized) { + if (core.isInitialized()) { return false; } - // zlib's manual states: "The application must initialize zalloc, zfree and opaque before calling - // the init function." - stream.zalloc = CompressionAllocator::AllocForZlib; - stream.zfree = CompressionAllocator::FreeForZlib; - stream.opaque = &allocator; - - switch (mode) { - case ZlibMode::DEFLATE: - case ZlibMode::GZIP: - case ZlibMode::DEFLATERAW: - err = deflateInit2(&stream, level, Z_DEFLATED, windowBits, memLevel, strategy); - break; - case ZlibMode::INFLATE: - case ZlibMode::GUNZIP: - case ZlibMode::INFLATERAW: - case ZlibMode::UNZIP: - err = inflateInit2(&stream, windowBits); - break; - default: - KJ_UNREACHABLE; - } + // The shared core wired the allocator hooks at construction; here we only pick the + // compression direction and hand over the (mode-adjusted) parameters. + auto direction = [&]() { + switch (mode) { + case ZlibMode::DEFLATE: + case ZlibMode::GZIP: + case ZlibMode::DEFLATERAW: + return ZlibStream::Mode::COMPRESS; + case ZlibMode::INFLATE: + case ZlibMode::GUNZIP: + case ZlibMode::INFLATERAW: + case ZlibMode::UNZIP: + return ZlibStream::Mode::DECOMPRESS; + default: + KJ_UNREACHABLE; + } + }(); - if (err != Z_OK) { + err = Z_OK; + KJ_IF_SOME(code, + core.init(direction, + ZlibStream::Options{ + .windowBits = windowBits, + .level = level, + .memLevel = memLevel, + .strategy = strategy, + })) { + err = code; dictionary.clear(); mode = ZlibMode::NONE; return true; } setDictionary(); - initialized = true; return true; } @@ -273,12 +277,12 @@ kj::Maybe ZlibContext::resetStream() { case ZlibMode::DEFLATE: case ZlibMode::DEFLATERAW: case ZlibMode::GZIP: - err = deflateReset(&stream); - break; case ZlibMode::INFLATE: case ZlibMode::INFLATERAW: case ZlibMode::GUNZIP: - err = inflateReset(&stream); + KJ_IF_SOME(code, core.reset()) { + err = code; + } break; default: break; @@ -298,6 +302,7 @@ void ZlibContext::work() { } const Bytef* next_expected_header_byte = nullptr; + auto& stream = core.raw(); // If the avail_out is left at 0, then it means that it ran out // of room. If there was avail_out left over, then it means @@ -306,7 +311,7 @@ void ZlibContext::work() { case ZlibMode::DEFLATE: case ZlibMode::GZIP: case ZlibMode::DEFLATERAW: - err = deflate(&stream, flush); + err = core.run(flush); break; case ZlibMode::UNZIP: if (stream.avail_in > 0) { @@ -356,16 +361,16 @@ void ZlibContext::work() { case ZlibMode::INFLATE: case ZlibMode::GUNZIP: case ZlibMode::INFLATERAW: - err = inflate(&stream, flush); + err = core.run(flush); // If data was encoded with dictionary (INFLATERAW will have it set in // SetDictionary, don't repeat that here) if (mode != ZlibMode::INFLATERAW && err == Z_NEED_DICT && !dictionary.empty()) { // Load it - err = inflateSetDictionary(&stream, dictionary.begin(), dictionary.size()); + err = inflateSetDictionary(&core.raw(), dictionary.begin(), dictionary.size()); if (err == Z_OK) { // And try to decode again - err = inflate(&stream, flush); + err = core.run(flush); } else if (err == Z_DATA_ERROR) { // Both inflateSetDictionary() and inflate() return Z_DATA_ERROR. // Make it possible for After() to tell a bad dictionary from bad @@ -382,7 +387,7 @@ void ZlibContext::work() { // used for padding. resetStream(); - err = inflate(&stream, flush); + err = core.run(flush); } break; default: @@ -400,7 +405,7 @@ kj::Maybe ZlibContext::setParams(int _level, int _strategy) { switch (mode) { case ZlibMode::DEFLATE: case ZlibMode::DEFLATERAW: - err = deflateParams(&stream, _level, _strategy); + err = deflateParams(&core.raw(), _level, _strategy); break; default: break; @@ -414,47 +419,30 @@ kj::Maybe ZlibContext::setParams(int _level, int _strategy) { } ZlibContext::~ZlibContext() noexcept(false) { - if (!initialized) { + if (!core.isInitialized()) { return; } - auto status = Z_OK; - switch (mode) { - case ZlibMode::DEFLATE: - case ZlibMode::DEFLATERAW: - case ZlibMode::GZIP: - status = deflateEnd(&stream); - break; - case ZlibMode::INFLATE: - case ZlibMode::INFLATERAW: - case ZlibMode::GUNZIP: - case ZlibMode::UNZIP: - status = inflateEnd(&stream); - break; - default: - break; - } + // Modes that never initialized a stream had nothing to end; for the rest, the shared + // core's end() dispatches deflateEnd/inflateEnd by direction, matching the historical + // per-mode switch. + auto status = core.end(); JSG_REQUIRE( status == Z_OK || status == Z_DATA_ERROR, Error, "Uncaught error on closing zlib stream"); } void ZlibContext::setBuffers(kj::ArrayPtr input, kj::ArrayPtr output) { - stream.avail_in = input.size(); - stream.next_in = input.begin(); - stream.avail_out = output.size(); - stream.next_out = output.begin(); + core.setInput(input); + core.setOutput(output); } void ZlibContext::setInputBuffer(kj::ArrayPtr input) { - // The define Z_CONST is not set, so zlib always takes mutable pointers - stream.next_in = const_cast(input.begin()); - stream.avail_in = input.size(); + core.setInput(input); } void ZlibContext::setOutputBuffer(kj::ArrayPtr output) { - stream.next_out = output.begin(); - stream.avail_out = output.size(); + core.setOutput(output); } template @@ -670,356 +658,6 @@ void ZlibUtil::ZlibStream::params(jsg::Lock& js, int _level, int _strategy) { } } -void BrotliContext::setBuffers(kj::ArrayPtr input, kj::ArrayPtr output) { - nextIn = reinterpret_cast(input.begin()); - nextOut = output.begin(); - availIn = input.size(); - availOut = output.size(); -} - -void BrotliContext::setInputBuffer(kj::ArrayPtr input) { - nextIn = input.begin(); - availIn = input.size(); -} - -void BrotliContext::setOutputBuffer(kj::ArrayPtr output) { - nextOut = output.begin(); - availOut = output.size(); -} - -uint BrotliContext::getAvailOut() const { - return availOut; -} - -void BrotliContext::setFlush(int _flush) { - flush = static_cast(_flush); -} - -void BrotliContext::getAfterWriteResult(uint32_t* _availIn, uint32_t* _availOut) const { - *_availIn = availIn; - *_availOut = availOut; -} - -BrotliEncoderContext::BrotliEncoderContext(CompressionAllocator& allocator, ZlibMode _mode) - : BrotliContext(allocator, _mode) { - // NOTE: Ignores any returned errors. - // TODO(soon): It's possible that initialization doesn't need to happen until `initialize` is - // called elsewhere. I'm keeping it like this to avoid changing the existing behaviour. - auto _ = initialize(); -} - -void BrotliEncoderContext::work() { - JSG_REQUIRE(mode == ZlibMode::BROTLI_ENCODE, Error, "Mode should be BROTLI_ENCODE"_kj); - JSG_REQUIRE_NONNULL(state.get(), Error, "State should not be empty"_kj); - - const uint8_t* internalNext = nextIn; - lastResult = BrotliEncoderCompressStream( - state.get(), flush, &availIn, &internalNext, &availOut, &nextOut, nullptr); - nextIn += internalNext - nextIn; - - streamEnd = lastResult && BrotliEncoderIsFinished(state.get()); -} - -kj::Maybe BrotliEncoderContext::initialize() { - auto instance = BrotliEncoderCreateInstance( - CompressionAllocator::AllocForBrotli, CompressionAllocator::FreeForZlib, &allocator); - state = kj::disposeWith(kj::mv(instance)); - - if (state.get() == nullptr) { - return CompressionError( - "Could not initialize Brotli instance"_kj, "ERR_ZLIB_INITIALIZATION_FAILED"_kj, -1); - } - - return kj::none; -} - -kj::Maybe BrotliEncoderContext::resetStream() { - return initialize(); -} - -kj::Maybe BrotliEncoderContext::setParams(int key, uint32_t value) { - if (!BrotliEncoderSetParameter(state.get(), static_cast(key), value)) { - return CompressionError("Setting parameter failed", "ERR_BROTLI_PARAM_SET_FAILED", -1); - } - - return kj::none; -} - -kj::Maybe BrotliEncoderContext::getError() const { - if (!lastResult) { - return CompressionError("Compression failed", "ERR_BROTLI_COMPRESSION_FAILED", -1); - } - - return kj::none; -} - -bool BrotliEncoderContext::isStreamEnd() const { - return streamEnd; -} - -BrotliDecoderContext::BrotliDecoderContext(CompressionAllocator& allocator, ZlibMode _mode) - : BrotliContext(allocator, _mode) { - // NOTE: Ignores any returned errors. - // TODO(soon): It's possible that initialization doesn't need to happen until `initialize` is - // called elsewhere. I'm keeping it like this to avoid changing the existing behaviour. - auto _ = initialize(); -} - -kj::Maybe BrotliDecoderContext::initialize() { - auto instance = BrotliDecoderCreateInstance( - CompressionAllocator::AllocForBrotli, CompressionAllocator::FreeForZlib, &allocator); - state = kj::disposeWith(kj::mv(instance)); - - if (state.get() == nullptr) { - return CompressionError( - "Could not initialize Brotli instance", "ERR_ZLIB_INITIALIZATION_FAILED", -1); - } - - return kj::none; -} - -void BrotliDecoderContext::work() { - JSG_REQUIRE(mode == ZlibMode::BROTLI_DECODE, Error, "Mode should have been BROTLI_DECODE"_kj); - JSG_REQUIRE_NONNULL(state.get(), Error, "State should not be empty"_kj); - const uint8_t* internalNext = nextIn; - lastResult = BrotliDecoderDecompressStream( - state.get(), &availIn, &internalNext, &availOut, &nextOut, nullptr); - nextIn += internalNext - nextIn; - - if (lastResult == BROTLI_DECODER_RESULT_ERROR) { - error = BrotliDecoderGetErrorCode(state.get()); - errorString = kj::str("ERR_", BrotliDecoderErrorString(error)); - } -} - -kj::Maybe BrotliDecoderContext::resetStream() { - return initialize(); -} - -kj::Maybe BrotliDecoderContext::setParams(int key, uint32_t value) { - if (!BrotliDecoderSetParameter(state.get(), static_cast(key), value)) { - return CompressionError("Setting parameter failed", "ERR_BROTLI_PARAM_SET_FAILED", -1); - } - - return kj::none; -} - -kj::Maybe BrotliDecoderContext::getError() const { - if (error != BROTLI_DECODER_NO_ERROR) { - return CompressionError("Compression failed", errorString, -1); - } - - if (flush == BROTLI_OPERATION_FINISH && lastResult == BROTLI_DECODER_RESULT_NEEDS_MORE_INPUT) { - // Match zlib behavior, as brotli doesn't have its own code for this. - return CompressionError("Unexpected end of file", "Z_BUF_ERROR", Z_BUF_ERROR); - } - - return kj::none; -} - -bool BrotliDecoderContext::isStreamEnd() const { - return lastResult == BROTLI_DECODER_RESULT_SUCCESS; -} - -// ======================================================================================= -// Zstd Implementation - -void ZstdContext::setBuffers(kj::ArrayPtr input, kj::ArrayPtr output) { - setInputBuffer(input); - setOutputBuffer(output); -} - -void ZstdContext::setInputBuffer(kj::ArrayPtr input) { - input_.src = input.begin(); - input_.size = input.size(); - input_.pos = 0; -} - -void ZstdContext::setOutputBuffer(kj::ArrayPtr output) { - output_.dst = output.begin(); - output_.size = output.size(); - output_.pos = 0; -} - -void ZstdContext::setFlush(int flush) { - KJ_DASSERT(flush >= ZSTD_e_continue && flush <= ZSTD_e_end, - "flush must be a valid ZSTD_EndDirective value"); - flush_ = static_cast(flush); -} - -kj::uint ZstdContext::getAvailOut() const { - return output_.size - output_.pos; -} - -void ZstdContext::getAfterWriteResult(uint32_t* availIn, uint32_t* availOut) const { - *availIn = input_.size - input_.pos; - *availOut = output_.size - output_.pos; -} - -namespace { -// Helper to check ZSTD errors and return a CompressionError if present. -// Also sets the error code in the provided reference for later retrieval. -kj::Maybe zstdCheckError( - size_t result, ZSTD_ErrorCode& error, kj::StringPtr errorCode) { - if (ZSTD_isError(result)) { - error = ZSTD_getErrorCode(result); - return CompressionError(ZSTD_getErrorName(result), errorCode, -1); - } - return kj::none; -} - -// Wrappers for ZSTD free functions that return void (for use with kj::disposeWith). -void zstdFreeCCtx(ZSTD_CCtx* cctx) { - ZSTD_freeCCtx(cctx); -} -void zstdFreeDCtx(ZSTD_DCtx* dctx) { - ZSTD_freeDCtx(dctx); -} -} // namespace - -ZstdEncoderContext::ZstdEncoderContext(ZlibMode _mode) - : ZstdContext(_mode), - cctx_(kj::disposeWith(ZSTD_createCCtx())) {} - -kj::Maybe ZstdEncoderContext::initialize(uint64_t pledgedSrcSize) { - if (cctx_.get() == nullptr) { - return CompressionError( - "Could not initialize Zstd instance"_kj, "ERR_ZLIB_INITIALIZATION_FAILED"_kj, -1); - } - - if (pledgedSrcSize != ZSTD_CONTENTSIZE_UNKNOWN) { - size_t result = ZSTD_CCtx_setPledgedSrcSize(cctx_.get(), pledgedSrcSize); - KJ_IF_SOME(err, zstdCheckError(result, error_, "ERR_ZSTD_COMPRESSION_FAILED"_kj)) { - return kj::mv(err); - } - } - - return kj::none; -} - -void ZstdEncoderContext::work() { - JSG_REQUIRE(mode == ZlibMode::ZSTD_ENCODE, Error, "Mode should be ZSTD_ENCODE"_kj); - JSG_REQUIRE(cctx_.get() != nullptr, Error, "Zstd context should not be null"_kj); - - lastResult = ZSTD_compressStream2(cctx_.get(), &output_, &input_, flush_); - - if (ZSTD_isError(lastResult)) { - error_ = ZSTD_getErrorCode(lastResult); - } -} - -kj::Maybe ZstdEncoderContext::resetStream() { - if (cctx_.get() != nullptr) { - size_t result = ZSTD_CCtx_reset(cctx_.get(), ZSTD_reset_session_only); - KJ_IF_SOME(err, zstdCheckError(result, error_, "ERR_ZSTD_COMPRESSION_FAILED"_kj)) { - return kj::mv(err); - } - } - return kj::none; -} - -kj::Maybe ZstdEncoderContext::setParams(int key, int value) { - KJ_DASSERT(key >= ZSTD_c_compressionLevel, - "key must be a valid ZSTD_cParameter (first valid value is ZSTD_c_compressionLevel)"); - size_t result = ZSTD_CCtx_setParameter(cctx_.get(), static_cast(key), value); - if (ZSTD_isError(result)) { - return CompressionError(kj::str("Setting parameter failed: ", ZSTD_getErrorName(result)), - "ERR_ZSTD_PARAM_SET_FAILED"_kj, -1); - } - return kj::none; -} - -kj::Maybe ZstdEncoderContext::getError() const { - if (error_ != ZSTD_error_no_error) { - return CompressionError(kj::str("Zstd compression failed: ", ZSTD_getErrorString(error_)), - kj::str("ERR_ZSTD_COMPRESSION_FAILED"), -1); - } - - if (flush_ == ZSTD_e_end && lastResult != 0) { - // lastResult > 0 means more output is needed, which shouldn't happen at end - return CompressionError("Unexpected end of file"_kj, "Z_BUF_ERROR"_kj, Z_BUF_ERROR); - } - - return kj::none; -} - -bool ZstdEncoderContext::isStreamEnd() const { - // ZSTD_compressStream2 returns 0 when flush_ == ZSTD_e_end and the frame is fully flushed. - return !ZSTD_isError(lastResult) && lastResult == 0; -} - -ZstdDecoderContext::ZstdDecoderContext(ZlibMode _mode) - : ZstdContext(_mode), - dctx_(kj::disposeWith(ZSTD_createDCtx())) {} - -kj::Maybe ZstdDecoderContext::initialize() { - // dctx_ is created in the constructor. It can only be nullptr if ZSTD_createDCtx() - // failed due to memory allocation failure. - if (dctx_.get() == nullptr) { - return CompressionError( - "Could not initialize Zstd instance"_kj, "ERR_ZLIB_INITIALIZATION_FAILED"_kj, -1); - } - - return kj::none; -} - -void ZstdDecoderContext::work() { - JSG_REQUIRE(mode == ZlibMode::ZSTD_DECODE, Error, "Mode should be ZSTD_DECODE"_kj); - JSG_REQUIRE(dctx_.get() != nullptr, Error, "Zstd context should not be null"_kj); - - lastResult = ZSTD_decompressStream(dctx_.get(), &output_, &input_); - - if (ZSTD_isError(lastResult)) { - error_ = ZSTD_getErrorCode(lastResult); - } else if (input_.size > 0) { - // Track whether we're mid-frame: lastResult > 0 means more data needed, - // lastResult == 0 means frame is complete. - frameInProgress_ = (lastResult > 0); - } -} - -kj::Maybe ZstdDecoderContext::resetStream() { - if (dctx_.get() != nullptr) { - size_t result = ZSTD_DCtx_reset(dctx_.get(), ZSTD_reset_session_only); - KJ_IF_SOME(err, zstdCheckError(result, error_, "ERR_ZSTD_DECOMPRESSION_FAILED"_kj)) { - return kj::mv(err); - } - } - frameInProgress_ = false; - return kj::none; -} - -kj::Maybe ZstdDecoderContext::setParams(int key, int value) { - KJ_DASSERT(dctx_.get() != nullptr, "Zstd decompression context should not be null"); - size_t result = ZSTD_DCtx_setParameter(dctx_.get(), static_cast(key), value); - if (ZSTD_isError(result)) { - return CompressionError(kj::str("Setting parameter failed: ", ZSTD_getErrorName(result)), - "ERR_ZSTD_PARAM_SET_FAILED"_kj, -1); - } - return kj::none; -} - -kj::Maybe ZstdDecoderContext::getError() const { - if (error_ != ZSTD_error_no_error) { - return CompressionError(kj::str("Zstd decompression failed: ", ZSTD_getErrorString(error_)), - kj::str("ERR_ZSTD_DECOMPRESSION_FAILED"), -1); - } - - // If this is the final flush, we're mid-frame (frame was started but never - // completed), and the output buffer is not full (decoder had space but - // couldn't produce more output), the input was truncated. - if (flush_ == ZSTD_e_end && frameInProgress_ && output_.pos < output_.size) { - return CompressionError("unexpected end of file"_kj, "ERR_ZSTD_DECOMPRESSION_FAILED"_kj, -1); - } - - return kj::none; -} - -bool ZstdDecoderContext::isStreamEnd() const { - // ZSTD_decompressStream returns 0 when a frame is completely decoded and fully flushed. - return !ZSTD_isError(lastResult) && lastResult == 0; -} - template jsg::Ref> ZlibUtil::ZstdCompressionStream< CompressionContext>::constructor(jsg::Lock& js, ZlibModeValue mode) { diff --git a/src/workerd/api/node/zlib-util.h b/src/workerd/api/node/zlib-util.h index 788aa2e510d..4e4a7bcd9d5 100644 --- a/src/workerd/api/node/zlib-util.h +++ b/src/workerd/api/node/zlib-util.h @@ -4,7 +4,7 @@ // Copyright Joyent and Node contributors. All rights reserved. MIT license. #pragma once -#include +#include #include #include @@ -23,27 +23,6 @@ // https://github.com/nodejs/node/blob/main/src/node_zlib.cc namespace workerd::api::node { -#ifndef ZLIB_ERROR_CODES -#define ZLIB_ERROR_CODES(V) \ - V(Z_OK) \ - V(Z_STREAM_END) \ - V(Z_NEED_DICT) \ - V(Z_ERRNO) \ - V(Z_STREAM_ERROR) \ - V(Z_DATA_ERROR) \ - V(Z_MEM_ERROR) \ - V(Z_BUF_ERROR) \ - V(Z_VERSION_ERROR) - -inline const char* ZlibStrerror(int err) { -#define V(code) \ - if (err == code) return #code; - ZLIB_ERROR_CODES(V) -#undef V - return "Z_UNKNOWN_ERROR"; -} -#endif // ZLIB_ERROR_CODES - // Certain zlib constants are defined by Node.js itself static constexpr auto Z_MIN_CHUNK = 64; static constexpr auto Z_MAX_CHUNK = 128 * 1024 * 1024; @@ -62,42 +41,13 @@ static constexpr auto Z_DEFAULT_WINDOWBITS = 15; static constexpr uint8_t GZIP_HEADER_ID1 = 0x1f; static constexpr uint8_t GZIP_HEADER_ID2 = 0x8b; -using ZlibModeValue = uint8_t; -enum class ZlibMode : ZlibModeValue { - NONE, - DEFLATE, - INFLATE, - GZIP, - GUNZIP, - DEFLATERAW, - INFLATERAW, - UNZIP, - BROTLI_DECODE, - BROTLI_ENCODE, - ZSTD_ENCODE, - ZSTD_DECODE -}; - // When possible, we intentionally override chunkSize to a value that is likely to perform better static constexpr auto ZLIB_PERFORMANT_CHUNK_SIZE = 40 * 1024; -struct CompressionError { - CompressionError(kj::StringPtr _message, kj::StringPtr _code, int _err) - : message(kj::str(_message)), - code(kj::str(_code)), - err(_err) { - JSG_REQUIRE(message.size() != 0, Error, "Compression error message should not be null"); - } - - kj::String message; - kj::String code; - int err; -}; - class ZlibContext final { public: explicit ZlibContext(CompressionAllocator& allocator, ZlibMode _mode) - : allocator(allocator), + : core(allocator), mode(_mode) {} ~ZlibContext() noexcept(false); @@ -117,6 +67,7 @@ class ZlibContext final { // when avail_out == 0, so we point next_out at a valid dummy byte instead. // With avail_out == 0, no data will actually be written to it. void clearBuffers() { + auto& stream = core.raw(); stream.next_in = nullptr; stream.avail_in = 0; stream.next_out = &dummyByte; @@ -132,8 +83,8 @@ class ZlibContext final { // Function signature is same as Node.js implementation. // Ref: https://github.com/nodejs/node/blob/9edf4a0856681a7665bd9dcf2ca7cac252784b98/src/node_zlib.cc#L880 void getAfterWriteResult(uint32_t* availIn, uint32_t* availOut) const { - *availIn = stream.avail_in; - *availOut = stream.avail_out; + *availIn = core.availIn(); + *availOut = core.availOut(); } void setMode(ZlibMode value) { mode = value; @@ -152,22 +103,18 @@ class ZlibContext final { } uint getAvailIn() const { - return stream.avail_in; + return core.availIn(); }; void setAvailIn(uint value) { - stream.avail_in = value; + core.raw().avail_in = value; }; uint getAvailOut() const { - return stream.avail_out; + return core.availOut(); } void setAvailOut(uint value) { - stream.avail_out = value; + core.raw().avail_out = value; }; - z_stream* getStream() { - return &stream; - } - // Zlib void initialize(int _level, int _windowBits, @@ -202,13 +149,18 @@ class ZlibContext final { kj::Maybe setDictionary(); CompressionError constructError(kj::StringPtr message) const { - if (stream.msg != nullptr) message = kj::StringPtr(stream.msg); + auto streamMsg = core.msg(); + if (streamMsg != nullptr) message = streamMsg; - return {kj::str(message), kj::str(ZlibStrerror(err)), err}; + return {kj::str(message), kj::str(ZlibStream::errorCodeName(err)), err}; }; - bool initialized = false; - CompressionAllocator& allocator; + // The shared z_stream core (api/compression.h) owns the stream structure and its + // lifecycle; the Node-specific machinery -- mode bookkeeping (including the UNZIP + // gzip-sniffing mode reassignment), dictionaries, deflateParams, and the Node-fidelity + // error surface -- lives here and reaches the structure through core.raw() where no + // core helper applies. + ZlibStream core; ZlibMode mode = ZlibMode::NONE; int flush = Z_NO_FLUSH; int windowBits = 0; @@ -219,7 +171,6 @@ class ZlibContext final { int err = Z_OK; unsigned int gzip_id_bytes_read = 0; - z_stream stream{}; // Dummy byte target for clearBuffers(). zlib's deflate() rejects // next_out == NULL even when avail_out == 0, so we need a valid address. Bytef dummyByte = 0; @@ -227,171 +178,6 @@ class ZlibContext final { using CompressionStreamErrorHandler = jsg::Function; -class BrotliContext { - public: - explicit BrotliContext(CompressionAllocator& allocator, ZlibMode _mode) - : allocator(allocator), - mode(_mode) {} - KJ_DISALLOW_COPY(BrotliContext); - void setBuffers(kj::ArrayPtr input, kj::ArrayPtr output); - void setInputBuffer(kj::ArrayPtr input); - void setOutputBuffer(kj::ArrayPtr output); - void setFlush(int flush); - kj::uint getAvailOut() const; - void getAfterWriteResult(uint32_t* availIn, uint32_t* availOut) const; - void setMode(ZlibMode _mode) { - mode = _mode; - } - - void clearBuffers() { - nextIn = nullptr; - nextOut = nullptr; - availIn = 0; - availOut = 0; - } - - struct Options { - jsg::Optional flush; - jsg::Optional finishFlush; - jsg::Optional chunkSize; - jsg::Optional> params; - jsg::Optional maxOutputLength; - JSG_STRUCT(flush, finishFlush, chunkSize, params, maxOutputLength); - }; - - protected: - CompressionAllocator& allocator; - ZlibMode mode; - const uint8_t* nextIn = nullptr; - uint8_t* nextOut = nullptr; - size_t availIn = 0; - size_t availOut = 0; - BrotliEncoderOperation flush = BROTLI_OPERATION_PROCESS; -}; - -class BrotliEncoderContext final: public BrotliContext { - public: - static const ZlibMode Mode = ZlibMode::BROTLI_ENCODE; - explicit BrotliEncoderContext(CompressionAllocator& allocator, ZlibMode _mode); - - KJ_DISALLOW_COPY_AND_MOVE(BrotliEncoderContext); - - // Equivalent to Node.js' `DoThreadPoolWork` implementation. - void work(); - kj::Maybe initialize(); - kj::Maybe resetStream(); - kj::Maybe setParams(int key, uint32_t value); - kj::Maybe getError() const; - bool isStreamEnd() const; - - private: - bool lastResult = false; - bool streamEnd = false; - kj::Own state; -}; - -class BrotliDecoderContext final: public BrotliContext { - public: - static const ZlibMode Mode = ZlibMode::BROTLI_DECODE; - explicit BrotliDecoderContext(CompressionAllocator& allocator, ZlibMode _mode); - - KJ_DISALLOW_COPY_AND_MOVE(BrotliDecoderContext); - - // Equivalent to Node.js' `DoThreadPoolWork` implementation. - void work(); - kj::Maybe initialize(); - kj::Maybe resetStream(); - kj::Maybe setParams(int key, uint32_t value); - kj::Maybe getError() const; - bool isStreamEnd() const; - - private: - BrotliDecoderResult lastResult = BROTLI_DECODER_RESULT_SUCCESS; - BrotliDecoderErrorCode error = BROTLI_DECODER_NO_ERROR; - kj::String errorString; - kj::Own state; -}; - -class ZstdContext { - public: - explicit ZstdContext(ZlibMode _mode): mode(_mode) {} - KJ_DISALLOW_COPY(ZstdContext); - - void setBuffers(kj::ArrayPtr input, kj::ArrayPtr output); - void setInputBuffer(kj::ArrayPtr input); - void setOutputBuffer(kj::ArrayPtr output); - void setFlush(int flush); - kj::uint getAvailOut() const; - void getAfterWriteResult(uint32_t* availIn, uint32_t* availOut) const; - void setMode(ZlibMode _mode) { - mode = _mode; - } - - void clearBuffers() { - input_ = {nullptr, 0, 0}; - output_ = {nullptr, 0, 0}; - } - - struct Options { - jsg::Optional flush; - jsg::Optional finishFlush; - jsg::Optional chunkSize; - jsg::Optional> params; - jsg::Optional maxOutputLength; - jsg::Optional pledgedSrcSize; - JSG_STRUCT(flush, finishFlush, chunkSize, params, maxOutputLength, pledgedSrcSize); - }; - - protected: - ZlibMode mode; - ZSTD_inBuffer input_{nullptr, 0, 0}; - ZSTD_outBuffer output_{nullptr, 0, 0}; - ZSTD_EndDirective flush_ = ZSTD_e_continue; -}; - -class ZstdEncoderContext final: public ZstdContext { - public: - static const ZlibMode Mode = ZlibMode::ZSTD_ENCODE; - explicit ZstdEncoderContext(ZlibMode _mode); - explicit ZstdEncoderContext(CompressionAllocator& _allocator, ZlibMode _mode) - : ZstdEncoderContext(_mode) {} - KJ_DISALLOW_COPY_AND_MOVE(ZstdEncoderContext); - - void work(); - kj::Maybe initialize(uint64_t pledgedSrcSize); - kj::Maybe resetStream(); - kj::Maybe setParams(int key, int value); - kj::Maybe getError() const; - bool isStreamEnd() const; - - private: - size_t lastResult = 0; - kj::Own cctx_; - ZSTD_ErrorCode error_ = ZSTD_error_no_error; -}; - -class ZstdDecoderContext final: public ZstdContext { - public: - static const ZlibMode Mode = ZlibMode::ZSTD_DECODE; - explicit ZstdDecoderContext(ZlibMode _mode); - explicit ZstdDecoderContext(CompressionAllocator& _allocator, ZlibMode _mode) - : ZstdDecoderContext(_mode) {} - KJ_DISALLOW_COPY_AND_MOVE(ZstdDecoderContext); - - void work(); - kj::Maybe initialize(); - kj::Maybe resetStream(); - kj::Maybe setParams(int key, int value); - kj::Maybe getError() const; - bool isStreamEnd() const; - - private: - size_t lastResult = 0; - kj::Own dctx_; - ZSTD_ErrorCode error_ = ZSTD_error_no_error; - bool frameInProgress_ = false; -}; - // Implements utilities in support of the Node.js Zlib class ZlibUtil final: public jsg::Object { public: @@ -859,21 +645,15 @@ class ZlibUtil final: public jsg::Object { #define EW_NODE_ZLIB_ISOLATE_TYPES \ api::node::ZlibUtil, api::node::ZlibUtil::ZlibStream, \ - api::node::ZlibUtil::BrotliCompressionStream, \ - api::node::ZlibUtil::BrotliCompressionStream, \ - api::node::ZlibUtil::ZstdCompressionStream, \ - api::node::ZlibUtil::ZstdCompressionStream, \ + api::node::ZlibUtil::BrotliCompressionStream, \ + api::node::ZlibUtil::BrotliCompressionStream, \ + api::node::ZlibUtil::ZstdCompressionStream, \ + api::node::ZlibUtil::ZstdCompressionStream, \ api::node::ZlibUtil::CompressionStream, \ - api::node::ZlibUtil::CompressionStream, \ - api::node::ZlibUtil::CompressionStream, \ - api::node::ZlibUtil::CompressionStream, \ - api::node::ZlibUtil::CompressionStream, \ - api::node::ZlibContext::Options, api::node::BrotliContext::Options, \ - api::node::ZstdContext::Options + api::node::ZlibUtil::CompressionStream, \ + api::node::ZlibUtil::CompressionStream, \ + api::node::ZlibUtil::CompressionStream, \ + api::node::ZlibUtil::CompressionStream, \ + api::node::ZlibContext::Options, api::BrotliContext::Options, api::ZstdContext::Options } // namespace workerd::api::node - -KJ_DECLARE_NON_POLYMORPHIC(BrotliEncoderStateStruct) -KJ_DECLARE_NON_POLYMORPHIC(BrotliDecoderStateStruct) -KJ_DECLARE_NON_POLYMORPHIC(ZSTD_CCtx) -KJ_DECLARE_NON_POLYMORPHIC(ZSTD_DCtx) diff --git a/src/workerd/api/streams.h b/src/workerd/api/streams.h index 917e5462e10..3654e5fa864 100644 --- a/src/workerd/api/streams.h +++ b/src/workerd/api/streams.h @@ -32,10 +32,10 @@ namespace workerd::api { api::IdentityTransformStream::QueuingStrategy, api::ReadableStream::ValuesOptions, \ api::ReadableStream::ReadableStreamAsyncIterator, \ api::ReadableStream::ReadableStreamAsyncIterator::Next, api::CompressionStream, \ - api::DecompressionStream, api::TextEncoderStream, api::TextDecoderStream, \ - api::TextDecoderStream::TextDecoderStreamInit, api::ByteLengthQueuingStrategy, \ - api::CountQueuingStrategy, api::QueuingStrategyInit, api::ReadableStreamNativeSource, \ - api::WritableStreamNativeSink + api::DecompressionStream, api::CompressionCodec, api::TextEncoderStream, \ + api::TextDecoderStream, api::TextDecoderStream::TextDecoderStreamInit, \ + api::ByteLengthQueuingStrategy, api::CountQueuingStrategy, api::QueuingStrategyInit, \ + api::ReadableStreamNativeSource, api::WritableStreamNativeSink // The list of streams.h types that are added to worker.c++'s JSG_DECLARE_ISOLATE_TYPE } // namespace workerd::api diff --git a/src/workerd/api/streams/compression.c++ b/src/workerd/api/streams/compression.c++ index 6d582d25101..36f5027e567 100644 --- a/src/workerd/api/streams/compression.c++ +++ b/src/workerd/api/streams/compression.c++ @@ -4,260 +4,40 @@ #include "compression.h" -#include "nbytes.h" - -#include #include -#include #include #include namespace workerd::api { -CompressionAllocator::CompressionAllocator( - kj::Arc&& externalMemoryTarget) - : externalMemoryTarget(kj::mv(externalMemoryTarget)) {} - -void* CompressionAllocator::AllocForZlib(void* data, uInt items, uInt size) { - size_t real_size = - nbytes::MultiplyWithOverflowCheck(static_cast(items), static_cast(size)); - return AllocForBrotli(data, real_size); -} - -void* CompressionAllocator::AllocForBrotli(void* opaque, size_t size) { - auto* allocator = static_cast(opaque); - auto data = kj::heapArray(size); - auto begin = data.begin(); - - allocator->allocations.insert(begin, - {.data = kj::mv(data), - .memoryAdjustment = allocator->externalMemoryTarget->getAdjustment(size)}); - return begin; -} - -void CompressionAllocator::FreeForZlib(void* opaque, void* pointer) { - if (KJ_UNLIKELY(pointer == nullptr)) return; - auto* allocator = static_cast(opaque); - // No need to destroy memoryAdjustment here. - // Dropping the allocation from the hashmap will defer the adjustment - // until the isolate lock is held. - JSG_REQUIRE(allocator->allocations.erase(pointer), Error, "Zlib allocation should exist"_kj); -} namespace { -class Context { +// The legacy async frontend: adapts the synchronous CodecStage (api/compression.h) to the +// promise-based AsyncInputStream + ExplicitEndOutputStream interfaces consumed by the +// internal streams machinery. Owns ALL of the asynchrony — the pending-read ring, the +// canceler, and the lifecycle state machine; the codec work itself lives entirely in the +// stage. +class CompressionStreamImpl final: public kj::Refcounted, + public kj::AsyncInputStream, + public capnp::ExplicitEndOutputStream { public: - enum class Mode { - COMPRESS, - DECOMPRESS, - }; - - enum class ContextFlags { - NONE, - STRICT, - }; - - struct Result { - bool success = false; - kj::ArrayPtr buffer; - }; - - explicit Context(Mode mode, - kj::StringPtr format, - ContextFlags flags, + explicit CompressionStreamImpl(CodecStage::Mode mode, + kj::String format, + CodecStage::Flags flags, kj::Arc&& externalMemoryTarget) - : allocator(kj::mv(externalMemoryTarget)), - mode(mode), - strictCompression(flags) - - { - // Configure allocator before any stream operations. - ctx.zalloc = CompressionAllocator::AllocForZlib; - ctx.zfree = CompressionAllocator::FreeForZlib; - ctx.opaque = &allocator; - - int result = Z_OK; - switch (mode) { - case Mode::COMPRESS: - result = deflateInit2(&ctx, Z_DEFAULT_COMPRESSION, Z_DEFLATED, getWindowBits(format), - 8, // memLevel = 8 is the default - Z_DEFAULT_STRATEGY); - break; - case Mode::DECOMPRESS: - result = inflateInit2(&ctx, getWindowBits(format)); - break; - default: - KJ_UNREACHABLE; - } - JSG_REQUIRE(result == Z_OK, Error, "Failed to initialize compression context."_kj); - } - - ~Context() noexcept(false) { - switch (mode) { - case Mode::COMPRESS: - deflateEnd(&ctx); - break; - case Mode::DECOMPRESS: - inflateEnd(&ctx); - break; - } - } - - KJ_DISALLOW_COPY_AND_MOVE(Context); - - void setInput(const void* in, size_t size) { - ctx.next_in = const_cast(reinterpret_cast(in)); - ctx.avail_in = size; - } - - Result pumpOnce(int flush) { - ctx.next_out = buffer; - ctx.avail_out = sizeof(buffer); - - int result = Z_OK; - - switch (mode) { - case Mode::COMPRESS: - result = deflate(&ctx, flush); - JSG_REQUIRE(result == Z_OK || result == Z_BUF_ERROR || result == Z_STREAM_END, TypeError, - "Compression failed."); - break; - case Mode::DECOMPRESS: - result = inflate(&ctx, flush); - JSG_REQUIRE(result == Z_OK || result == Z_BUF_ERROR || result == Z_STREAM_END, TypeError, - "Decompression failed."); - - if (strictCompression == ContextFlags::STRICT) { - // The spec requires that a TypeError is produced if there is trailing data after the end - // of the compression stream. - JSG_REQUIRE(!(result == Z_STREAM_END && ctx.avail_in > 0), TypeError, - "Trailing bytes after end of compressed data"); - // Same applies to closing a stream before the complete decompressed data is available. - JSG_REQUIRE( - !(flush == Z_FINISH && result == Z_BUF_ERROR && ctx.avail_out == sizeof(buffer)), - TypeError, "Called close() on a decompression stream with incomplete data"); - } - break; - default: - KJ_UNREACHABLE; - } - - return Result{ - .success = result == Z_OK, - .buffer = kj::arrayPtr(buffer, sizeof(buffer) - ctx.avail_out), - }; - } - - protected: - CompressionAllocator allocator; - - private: - static int getWindowBits(kj::StringPtr format) { - // We use a windowBits value of 15 combined with the magic value - // for the compression format type. For gzip, the magic value is - // 16, so the value returned is 15 + 16. For deflate, the magic - // value is 15. For raw deflate (i.e. deflate without a zlib header) - // the negative windowBits value is used, so -15. See the comments for - // deflateInit2() in zlib.h for details. - static constexpr auto GZIP = 16; - static constexpr auto DEFLATE = 15; - static constexpr auto DEFLATE_RAW = -15; - if (format == "gzip") - return DEFLATE + GZIP; - else if (format == "deflate") - return DEFLATE; - else if (format == "deflate-raw") - return DEFLATE_RAW; - KJ_UNREACHABLE; - } - - Mode mode; - z_stream ctx = {}; - kj::byte buffer[16384]; - - // For the eponymous compatibility flag - ContextFlags strictCompression; -}; - -// Buffer class based on std::vector that erases data that has been read from it lazily to avoid -// excessive copying when reading a larger amount of buffered data in small chunks. valid_size_ is -// used to track the amount of data that has not been read back yet. -class LazyBuffer { - public: - LazyBuffer(): valid_size_(0) {} - - // Return a chunk of data and mark it as invalid. The returned chunk remains valid until data is - // shifted, cleared or destructor is called. maybeShift() should be called after the returned data - // has been processed. - kj::ArrayPtr take(size_t read_size) { - KJ_ASSERT(read_size <= valid_size_); - kj::ArrayPtr chunk = kj::arrayPtr(&output[output.size() - valid_size_], read_size); - valid_size_ -= read_size; - return chunk; - } - - // Shift the output only if doing so results in reducing vector size by at least 1 KiB and 1/8 of - // its size to avoid copying for small reads. - void maybeShift() { - size_t unusedSpace = output.size() - valid_size_; - if (unusedSpace >= 1024 && unusedSpace >= (output.size() >> 3)) { - // Shifting buffer to erase data that has already been read. valid_size_ remains the same. - memmove(output.begin(), output.begin() + unusedSpace, valid_size_); - output.truncate(valid_size_); - } - } - - void write(kj::ArrayPtr chunk) { - output.addAll(chunk); - valid_size_ += chunk.size(); - } - - void clear() { - output.clear(); - valid_size_ = 0; - } - - // For convenience, provide the size of the valid data that has not been read back yet. This may - // be smaller than the size of the internal vector, which is not relevant for the stream - // implementation. - size_t size() { - return valid_size_; - } - - // As with size(), the buffer is considered empty if there is no valid data remaining. - size_t empty() { - return valid_size_ == 0; - } - - private: - kj::Vector output; - size_t valid_size_; -}; - -// Because we have to use an autogate to switch things over to the new state manager, we need -// to separate out a common base class for the compression stream internal state and separate -// two separate impls that differ only in how they manage state. Once the autogate is removed, -// we can delete the first impl class and merge everything back together. -template -class CompressionStreamBase: public kj::Refcounted, - public kj::AsyncInputStream, - public capnp::ExplicitEndOutputStream { - public: - explicit CompressionStreamBase(kj::String format, - Context::ContextFlags flags, - kj::Arc&& externalMemoryTarget) - : context(mode, format, flags, kj::mv(externalMemoryTarget)) {} + : stage(mode, format, flags, kj::mv(externalMemoryTarget)), + state(decltype(state)::create()) {} // WritableStreamSink implementation --------------------------------------------------- - kj::Promise write(kj::ArrayPtr buffer) override final { + kj::Promise write(kj::ArrayPtr buffer) override { requireActive("Write after close"); - context.setInput(buffer.begin(), buffer.size()); - writeInternal(Z_NO_FLUSH); + runCodec([&]() { stage.push(buffer); }); + maybeFulfillRead(); co_return; } - kj::Promise write(kj::ArrayPtr> pieces) override final { + kj::Promise write(kj::ArrayPtr> pieces) override { // We check state here so that we catch errors even if pieces is empty. requireActive("Write after close"); for (auto piece: pieces) { @@ -266,28 +46,29 @@ class CompressionStreamBase: public kj::Refcounted, co_return; } - kj::Promise end() override final { + kj::Promise end() override { transitionToEnded(); - writeInternal(Z_FINISH); + runCodec([&]() { stage.end(); }); + maybeFulfillRead(); co_return; } - kj::Promise whenWriteDisconnected() override final { + kj::Promise whenWriteDisconnected() override { return kj::NEVER_DONE; } - void abortWrite(kj::Exception&& reason) override final { + void abortWrite(kj::Exception&& reason) override { cancelInternal(kj::mv(reason)); } // AsyncInputStream implementation ----------------------------------------------------- - kj::Promise tryRead(void* buffer, size_t minBytes, size_t maxBytes) override final { + kj::Promise tryRead(void* buffer, size_t minBytes, size_t maxBytes) override { KJ_ASSERT(minBytes <= maxBytes); // Re-throw any stored exception throwIfException(); // If stream has ended normally and no buffered data, return EOF - if (isInTerminalState() && output.empty()) { + if (isInTerminalState() && stage.empty()) { co_return static_cast(0); } // Active or terminal with data remaining @@ -295,13 +76,6 @@ class CompressionStreamBase: public kj::Refcounted, kj::arrayPtr(reinterpret_cast(buffer), maxBytes), minBytes); } - protected: - virtual void requireActive(kj::StringPtr errorMessage) = 0; - virtual void transitionToEnded() = 0; - virtual void transitionToErrored(kj::Exception&& reason) = 0; - virtual void throwIfException() = 0; - virtual bool isInTerminalState() = 0; - private: struct PendingRead { kj::ArrayPtr buffer; @@ -310,8 +84,19 @@ class CompressionStreamBase: public kj::Refcounted, kj::Own> promise; }; + // Runs a stage operation, translating a codec exception into stream teardown (reject + // pending reads, error the state machine) before rethrowing — preserving the error path + // of the previously fused pump/state code. + template + void runCodec(Func&& func) { + KJ_IF_SOME(exception, kj::runCatchingExceptions(kj::fwd(func))) { + cancelInternal(exception.clone()); + kj::throwFatalException(kj::mv(exception)); + } + } + void cancelInternal(kj::Exception reason) { - output.clear(); + stage.clear(); while (!pendingReads.empty()) { auto pending = kj::mv(pendingReads.front()); @@ -326,20 +111,15 @@ class CompressionStreamBase: public kj::Refcounted, } kj::Promise tryReadInternal(kj::ArrayPtr dest, size_t minBytes) { - const auto copyIntoBuffer = [this](kj::ArrayPtr dest) { - auto maxBytesToCopy = kj::min(dest.size(), output.size()); - dest.write(output.take(maxBytesToCopy)); - output.maybeShift(); - return maxBytesToCopy; - }; - - // If the output currently contains >= minBytes, then we'll fulfill - // the read immediately, removing as many bytes as possible from the - // output queue. - // If we reached the end (terminal state), resolve the read immediately - // as well, since no new data is expected. - if (output.size() >= minBytes || isInTerminalState()) { - co_return copyIntoBuffer(dest); + // TODO(later): This does not yet implement any backpressure. A caller can keep calling + // write without reading, which will continue to fill the stage's internal buffer. + // + // If the stage currently buffers >= minBytes, then we'll fulfill the read immediately, + // removing as many bytes as possible from the output queue. + // If we reached the end (terminal state), resolve the read immediately as well, since + // no new data is expected. + if (stage.available() >= minBytes || isInTerminalState()) { + co_return stage.pull(dest); } // Otherwise, create a pending read. @@ -352,8 +132,8 @@ class CompressionStreamBase: public kj::Refcounted, }; // If there are any bytes queued, copy as much as possible into the buffer. - if (output.size() > 0) { - pendingRead.filled = copyIntoBuffer(dest); + if (stage.available() > 0) { + pendingRead.filled = stage.pull(dest); } pendingReads.push_back(kj::mv(pendingRead)); @@ -361,42 +141,11 @@ class CompressionStreamBase: public kj::Refcounted, co_return co_await canceler.wrap(kj::mv(promise.promise)); } - void writeInternal(int flush) { - // TODO(later): This does not yet implement any backpressure. A caller can keep calling - // write without reading, which will continue to fill the internal buffer. - KJ_ASSERT(flush == Z_FINISH || !isInTerminalState()); - Context::Result result; - - while (true) { - KJ_IF_SOME(exception, kj::runCatchingExceptions([this, flush, &result]() { - result = context.pumpOnce(flush); - })) { - cancelInternal(exception.clone()); - kj::throwFatalException(kj::mv(exception)); - } - - if (result.buffer.size() == 0) { - if (result.success) { - // No output produced but input data has been processed based on zlib return code, call - // pumpOnce again. - continue; - } - maybeFulfillRead(); - return; - } - - // Output has been produced, copy it to result buffer and continue loop to call pumpOnce - // again. - output.write(result.buffer); - } - KJ_UNREACHABLE; - } - - // Fulfill as many pending reads as we can from the output buffer. + // Fulfill as many pending reads as we can from the stage's buffered output. void maybeFulfillRead() { - // If there are pending reads and data to be read, we'll loop through - // the pending reads and fulfill them as much as possible. - while (!pendingReads.empty() && output.size() > 0) { + // If there are pending reads and data to be read, we'll loop through the pending reads + // and fulfill them as much as possible. + while (!pendingReads.empty() && stage.available() > 0) { auto& pending = pendingReads.front(); if (!pending.promise->isWaiting()) { @@ -418,15 +167,11 @@ class CompressionStreamBase: public kj::Refcounted, kj::throwFatalException(kj::mv(ex)); } - // The pending read is still viable so determine how much we can copy in. - auto amountToCopy = kj::min(pending.buffer.size() - pending.filled, output.size()); - kj::ArrayPtr chunk = output.take(amountToCopy); - pending.buffer.slice(pending.filled, pending.filled + amountToCopy).copyFrom(chunk); - pending.filled += amountToCopy; - output.maybeShift(); + // The pending read is still viable so copy in as much as we can. + pending.filled += stage.pull(pending.buffer.slice(pending.filled, pending.buffer.size())); - // If we've met the minimum bytes requirement for the pending read, fulfill - // the read promise. + // If we've met the minimum bytes requirement for the pending read, fulfill the read + // promise. if (pending.filled >= pending.minBytes) { auto p = kj::mv(pending); pendingReads.pop_front(); @@ -434,16 +179,16 @@ class CompressionStreamBase: public kj::Refcounted, continue; } - // If we reached this point in the loop, remaining must be 0 so that we - // don't keep iterating through on the same pending read. - KJ_ASSERT(output.empty()); + // If we reached this point in the loop, remaining must be 0 so that we don't keep + // iterating through on the same pending read. + KJ_ASSERT(stage.empty()); } if (isInTerminalState() && !pendingReads.empty()) { - // We are ended and we have pending reads. Because of the loop above, - // one of either pendingReads or output must be empty, so if we got this - // far, output.empty() must be true. Let's check. - KJ_ASSERT(output.empty()); + // We are ended and we have pending reads. Because of the loop above, one of either + // pendingReads or the stage buffer must be empty, so if we got this far, stage.empty() + // must be true. Let's check. + KJ_ASSERT(stage.empty()); // We need to flush any remaining reads. while (!pendingReads.empty()) { auto pending = kj::mv(pendingReads.front()); @@ -456,24 +201,9 @@ class CompressionStreamBase: public kj::Refcounted, } } - Context context; - - kj::Canceler canceler; - LazyBuffer output; - RingBuffer pendingReads; -}; - -template -class CompressionStreamImpl final: public CompressionStreamBase { - public: - explicit CompressionStreamImpl(kj::String format, - Context::ContextFlags flags, - kj::Arc&& externalMemoryTarget) - : CompressionStreamBase(kj::mv(format), flags, kj::mv(externalMemoryTarget)), - state(decltype(state)::template create()) {} + // Lifecycle ----------------------------------------------------------------------------- - protected: - void requireActive(kj::StringPtr errorMessage) override { + void requireActive(kj::StringPtr errorMessage) { KJ_IF_SOME(exception, state.tryGetErrorUnsafe()) { kj::throwFatalException(exception.clone()); } @@ -481,31 +211,30 @@ class CompressionStreamImpl final: public CompressionStreamBase { JSG_REQUIRE(state.isActive(), Error, errorMessage); } - void transitionToEnded() override { - // If already in a terminal state (Ended or Exception), this is a no-op. - // This matches the V1 behavior where calling end() multiple times was allowed. + void transitionToEnded() { + // If already in a terminal state (Ended or Exception), this is a no-op, preserving the + // historical allowance for multiple end() calls. if (state.isTerminal()) return; - auto result = state.template transitionFromTo(); + auto result = state.transitionFromTo(); KJ_REQUIRE(result != kj::none, "Stream already ended or errored"); } - void transitionToErrored(kj::Exception&& reason) override { - // Use forceTransitionTo because cancelInternal may be called when already - // in an error state (e.g., from writeInternal error handling). - state.template forceTransitionTo(kj::mv(reason)); + void transitionToErrored(kj::Exception&& reason) { + // Use forceTransitionTo because cancelInternal may be called when already in an error + // state (e.g., from the runCodec error handling). + state.forceTransitionTo(kj::mv(reason)); } - void throwIfException() override { + void throwIfException() { KJ_IF_SOME(exception, state.tryGetErrorUnsafe()) { kj::throwFatalException(exception.clone()); } } - virtual bool isInTerminalState() override { + bool isInTerminalState() { return state.isTerminal(); } - private: struct Ended { static constexpr kj::StringPtr NAME KJ_UNUSED = "ended"_kj; }; @@ -513,6 +242,8 @@ class CompressionStreamImpl final: public CompressionStreamBase { static constexpr kj::StringPtr NAME KJ_UNUSED = "open"_kj; }; + CodecStage stage; + // State machine for tracking compression stream lifecycle: // Open -> Ended (normal close via end()) // Open -> kj::Exception (error via abortWrite()) @@ -524,16 +255,23 @@ class CompressionStreamImpl final: public CompressionStreamBase { Ended, kj::Exception> state; + + kj::Canceler canceler; + RingBuffer pendingReads; }; // Adapter to bridge CompressionStreamImpl (which implements AsyncInputStream and // ExplicitEndOutputStream) to the ReadableStreamSource/WritableStreamSink interfaces. -template +// TODO(soon): This class is intended to be replaced by the new ReadableSource/WritableSink +// interfaces once fully implemented. We will need an adapter that knows how to handle both +// sides of the stream once fully implemented. The current implementation in +// system-streams.c++ implements separate adapters for each side that are not aware of each +// other, making it unsuitable for this specific case. class CompressionStreamAdapter final: public kj::Refcounted, public ReadableStreamSource, public WritableStreamSink { public: - explicit CompressionStreamAdapter(kj::Rc> impl) + explicit CompressionStreamAdapter(kj::Rc impl) : impl(kj::mv(impl)), ioContext(IoContext::current()) {} @@ -565,40 +303,23 @@ class CompressionStreamAdapter final: public kj::Refcounted, } private: - kj::Rc> impl; + kj::Rc impl; IoContext& ioContext; }; -kj::Rc> createCompressionStreamImpl( - kj::String format, - Context::ContextFlags flags, - kj::Arc&& externalMemoryTarget) { - return kj::rc>( - kj::mv(format), flags, kj::mv(externalMemoryTarget)); -} - -kj::Rc> createDecompressionStreamImpl( - kj::String format, - Context::ContextFlags flags, - kj::Arc&& externalMemoryTarget) { - return kj::rc>( - kj::mv(format), flags, kj::mv(externalMemoryTarget)); -} - } // namespace jsg::Ref CompressionStream::constructor(jsg::Lock& js, kj::String format) { JSG_REQUIRE(format == "deflate" || format == "gzip" || format == "deflate-raw", TypeError, "The compression format must be either 'deflate', 'deflate-raw' or 'gzip'."); - // TODO(cleanup): Once the autogate is removed, we can delete CompressionStreamImpl - kj::Rc> impl = createCompressionStreamImpl( - kj::mv(format), Context::ContextFlags::NONE, js.getExternalMemoryTarget()); + auto impl = kj::rc(CodecStage::Mode::COMPRESS, kj::mv(format), + CodecStage::Flags::NONE, js.getExternalMemoryTarget()); auto& ioContext = IoContext::current(); // Create a single adapter that implements both readable and writable sides - auto adapter = kj::refcounted>(kj::mv(impl)); + auto adapter = kj::refcounted(kj::mv(impl)); auto readableSide = kj::addRef(*adapter); auto writableSide = kj::mv(adapter); @@ -611,16 +332,15 @@ jsg::Ref DecompressionStream::constructor(jsg::Lock& js, kj JSG_REQUIRE(format == "deflate" || format == "gzip" || format == "deflate-raw", TypeError, "The compression format must be either 'deflate', 'deflate-raw' or 'gzip'."); - kj::Rc> impl = - createDecompressionStreamImpl(kj::mv(format), - FeatureFlags::get(js).getStrictCompression() ? Context::ContextFlags::STRICT - : Context::ContextFlags::NONE, - js.getExternalMemoryTarget()); + auto impl = kj::rc(CodecStage::Mode::DECOMPRESS, kj::mv(format), + FeatureFlags::get(js).getStrictCompression() ? CodecStage::Flags::STRICT + : CodecStage::Flags::NONE, + js.getExternalMemoryTarget()); auto& ioContext = IoContext::current(); // Create a single adapter that implements both readable and writable sides - auto adapter = kj::refcounted>(kj::mv(impl)); + auto adapter = kj::refcounted(kj::mv(impl)); auto readableSide = kj::addRef(*adapter); auto writableSide = kj::mv(adapter); @@ -629,4 +349,4 @@ jsg::Ref DecompressionStream::constructor(jsg::Lock& js, kj ioContext.getMetrics().tryCreateWritableByteStreamObserver())); } -} // namespace workerd::api +} // namespace workerd::api \ No newline at end of file diff --git a/src/workerd/api/streams/compression.h b/src/workerd/api/streams/compression.h index 6309179d385..a7d42ece594 100644 --- a/src/workerd/api/streams/compression.h +++ b/src/workerd/api/streams/compression.h @@ -4,6 +4,7 @@ #pragma once +#include #include #include @@ -11,28 +12,6 @@ namespace workerd::api { -// A custom allocator to be used by the zlib and brotli libraries. -// The allocator should not and can not safely hold a reference to the jsg::Lock -// instance. Therefore, we lookup the current jsg::Lock instance from the -// isolate pointer and use that to get the external memory adjustment. -class CompressionAllocator final { - public: - CompressionAllocator(kj::Arc&& externalMemoryTarget); - - static void* AllocForZlib(void* data, uInt items, uInt size); - static void* AllocForBrotli(void* data, size_t size); - static void FreeForZlib(void* data, void* pointer); - - private: - struct Allocation { - kj::Array data; - kj::Maybe memoryAdjustment = kj::none; - }; - - kj::Arc externalMemoryTarget; - kj::HashMap allocations; -}; - class CompressionStream: public TransformStream { public: using TransformStream::TransformStream; diff --git a/src/workerd/api/tests/BUILD.bazel b/src/workerd/api/tests/BUILD.bazel index fcc043c0e5d..3d82652f4ad 100644 --- a/src/workerd/api/tests/BUILD.bazel +++ b/src/workerd/api/tests/BUILD.bazel @@ -701,6 +701,12 @@ wd_test( data = ["persistent-stubs-test.js"], ) +wd_test( + src = "ts-compression-streams-test.wd-test", + args = ["--experimental"], + data = ["ts-compression-streams-test.js"], +) + wd_test( src = "compression-streams-test.wd-test", args = ["--experimental"], diff --git a/src/workerd/api/tests/ts-compression-streams-test.js b/src/workerd/api/tests/ts-compression-streams-test.js new file mode 100644 index 00000000000..23071c05ca4 --- /dev/null +++ b/src/workerd/api/tests/ts-compression-streams-test.js @@ -0,0 +1,283 @@ +// Copyright (c) 2026 Cloudflare, Inc. +// Licensed under the Apache 2.0 license found in the LICENSE file or at: +// https://opensource.org/licenses/Apache-2.0 + +// Tests for the TypeScript CompressionStream/DecompressionStream +// implementation (typescript_implemented_streams + the per-isolate +// bootstrap), backed by the synchronous C++ codec handle. Covers the +// deliberate semantics documented in webstreams/compression.ts: +// eager (transform-time) error timing, legacy-parity write settlement, +// and the byte-capable (BYOB) readable. +// +// The config enables strict_compression_checks, matching production +// defaults, so the strict flush-time checks are exercised too. + +import { strictEqual, deepStrictEqual, ok, rejects, throws } from 'node:assert'; + +const enc = new TextEncoder(); +const dec = new TextDecoder(); + +function userSource(...chunks) { + return new ReadableStream({ + start(controller) { + for (const chunk of chunks) { + controller.enqueue(chunk); + } + controller.close(); + }, + }); +} + +async function readAll(readable) { + const reader = readable.getReader(); + const chunks = []; + while (true) { + const { value, done } = await reader.read(); + if (done) break; + chunks.push(value); + } + return chunks; +} + +function toBytes(chunks) { + return chunks.flatMap((chunk) => [...chunk]); +} + +// Writes the chunks, closes, and drains the readable; returns the byte +// array. Writes settle without reads (legacy-parity settlement), so the +// sequential form is safe. +async function pump(pair, chunks) { + const writer = pair.writable.getWriter(); + for (const chunk of chunks) { + await writer.write(chunk); + } + await writer.close(); + return toBytes(await readAll(pair.readable)); +} + +// --- Core behavior --- + +export const writesSettleWithoutReads = { + async test() { + const cs = new CompressionStream('gzip'); + const writer = cs.writable.getWriter(); + // LEGACY-PARITY SETTLEMENT: writes settle as soon as the codec has + // consumed the chunk — no read required (unlike the rendezvous of + // IdentityTransformStream). + await writer.write(enc.encode('hello')); + await writer.close(); + const bytes = toBytes(await readAll(cs.readable)); + // gzip magic. + strictEqual(bytes[0], 0x1f); + strictEqual(bytes[1], 0x8b); + }, +}; + +export const roundTripAllFormats = { + async test() { + const payload = 'The quick brown fox jumps over the lazy dog. '.repeat(50); + for (const format of ['gzip', 'deflate', 'deflate-raw']) { + const data = enc.encode(payload); + const compressed = await pump(new CompressionStream(format), [data]); + // The codec actually compressed the repetitive payload. + strictEqual(compressed.length < data.length, true); + const restored = await pump(new DecompressionStream(format), [ + new Uint8Array(compressed), + ]); + strictEqual(dec.decode(new Uint8Array(restored)), payload); + } + }, +}; + +export const pipeThroughInterop = { + async test() { + const payload = enc.encode('compress me via pipes'); + const out = toBytes( + await readAll( + userSource(payload) + .pipeThrough(new CompressionStream('deflate')) + .pipeThrough(new DecompressionStream('deflate')) + ) + ); + deepStrictEqual(out, [...payload]); + }, +}; + +export const chunkedWritesRoundTrip = { + async test() { + // Multi-chunk writes exercise the stateful streaming path. + const parts = ['first ', 'second ', 'third']; + const compressed = await pump( + new CompressionStream('gzip'), + parts.map((p) => enc.encode(p)) + ); + // Split the compressed bytes into awkward chunks for decompression. + const half = compressed.length >> 1; + const restored = await pump(new DecompressionStream('gzip'), [ + new Uint8Array(compressed.slice(0, half)), + new Uint8Array(compressed.slice(half)), + ]); + strictEqual(dec.decode(new Uint8Array(restored)), parts.join('')); + }, +}; + +// --- Error timing --- + +export const corruptInputRejectsWrite = { + async test() { + const ds = new DecompressionStream('gzip'); + const writer = ds.writable.getWriter(); + // Transform-time error timing: the WRITE itself rejects, without + // any read on the readable side ever happening. + await rejects(writer.write(enc.encode('definitely not gzip')), TypeError); + // The readable side is errored too (both-sides propagation). + await rejects(readAll(ds.readable), TypeError); + }, +}; + +export const strictIncompleteCloseRejects = { + async test() { + // Build a valid gzip payload, then truncate the trailer so every + // WRITE succeeds but the stream is incomplete at close. + const compressed = await pump(new CompressionStream('gzip'), [ + enc.encode('hello world'), + ]); + const truncated = new Uint8Array( + compressed.slice(0, compressed.length - 4) + ); + const ds = new DecompressionStream('gzip'); + const writer = ds.writable.getWriter(); + await writer.write(truncated); + // Flush-time error timing: the strict incomplete-data check rejects + // the CLOSE (strict_compression_checks is enabled in this config). + await rejects(writer.close(), TypeError); + }, +}; + +export const invalidFormatThrows = { + test() { + throws(() => new CompressionStream('br'), /must be either/); + // Missing argument coerces to "undefined" — same TypeError. + throws(() => new DecompressionStream(), /must be either/); + }, +}; + +// --- Legacy-parity surface --- + +export const byobReadSupported = { + async test() { + const cs = new CompressionStream('gzip'); + const writer = cs.writable.getWriter(); + await writer.write(enc.encode('byob')); + await writer.close(); + // Legacy parity: the readable is byte-capable, so BYOB readers work + // (WHATWG describes a default stream here; the legacy C++ pair was + // BYOB-readable and this implementation preserves that). + const reader = cs.readable.getReader({ mode: 'byob' }); + const { value, done } = await reader.read(new Uint8Array(2), { min: 2 }); + strictEqual(done, false); + deepStrictEqual([...value], [0x1f, 0x8b]); + await reader.cancel(); + }, +}; + +export const classShape = { + test() { + strictEqual( + Object.prototype.toString.call(new CompressionStream('gzip')), + '[object CompressionStream]' + ); + strictEqual( + Object.prototype.toString.call(new DecompressionStream('gzip')), + '[object DecompressionStream]' + ); + // The internal C++ codec factory is not on the class: it is injected + // through the bootstrap's utils pseudo-global, never a JS-visible + // surface. + strictEqual('newCodec' in CompressionStream, false); + strictEqual('newCodec' in DecompressionStream, false); + // Brand checks: prototype getters reject foreign receivers. + const desc = Object.getOwnPropertyDescriptor( + CompressionStream.prototype, + 'readable' + ); + throws(() => desc.get.call({}), TypeError); + }, +}; + +export const abortErrorsBothSides = { + async test() { + const cs = new CompressionStream('gzip'); + const writer = cs.writable.getWriter(); + await writer.write(enc.encode('partial')); + const reason = new Error('abandon'); + await writer.abort(reason); + await rejects(readAll(cs.readable), /abandon/); + }, +}; + +export const cancelReadableErrorsWritable = { + async test() { + const cs = new CompressionStream('gzip'); + const reason = new Error('no more'); + await cs.readable.cancel(reason); + // Mirror of the identity pair: cancelling the readable errors the + // writable side. + const writer = cs.writable.getWriter(); + await rejects(writer.closed, /no more/); + }, +}; + +// Concatenates an array of Uint8Array chunks (byte-count-preserving, unlike +// toBytes' number-array form, which is convenient only for small payloads). +function concat(chunks) { + const total = chunks.reduce((sum, c) => sum + c.byteLength, 0); + const out = new Uint8Array(total); + let offset = 0; + for (const chunk of chunks) { + out.set(chunk, offset); + offset += chunk.byteLength; + } + return out; +} + +// Large multi-pump payload (many 16KiB codec pump iterations, chunked writes): +// pure in-worker coverage, no fetch involved. +export const largePayloadRoundtrip = { + async test() { + const size = 400 * 1024; + const original = new Uint8Array(size); + // Compressible but non-trivial content. + for (let i = 0; i < size; i++) + original[i] = (i * 31 + ((i / 512) | 0)) & 0xff; + + const cs = new CompressionStream('gzip'); + const writer = cs.writable.getWriter(); + const writes = (async () => { + // Chunked writes to exercise repeated eager pushes. + for (let off = 0; off < size; off += 64 * 1024) { + await writer.write( + original.subarray(off, Math.min(off + 64 * 1024, size)) + ); + } + await writer.close(); + })(); + const compressed = concat(await readAll(cs.readable)); + await writes; + ok( + compressed.byteLength < size, + `compressed output should be smaller: ${compressed.byteLength} vs ${size}` + ); + + const ds = new DecompressionStream('gzip'); + const writer2 = ds.writable.getWriter(); + const writes2 = (async () => { + await writer2.write(compressed); + await writer2.close(); + })(); + const roundtrip = concat(await readAll(ds.readable)); + await writes2; + strictEqual(roundtrip.byteLength, size); + deepStrictEqual(roundtrip, original); + }, +}; diff --git a/src/workerd/api/tests/ts-compression-streams-test.wd-test b/src/workerd/api/tests/ts-compression-streams-test.wd-test new file mode 100644 index 00000000000..bd9de1181a9 --- /dev/null +++ b/src/workerd/api/tests/ts-compression-streams-test.wd-test @@ -0,0 +1,22 @@ +using Workerd = import "/workerd/workerd.capnp"; + +const unitTests :Workerd.Config = ( + services = [ + ( name = "ts-compression-streams-test", + worker = ( + modules = [ + (name = "worker", esModule = embed "ts-compression-streams-test.js") + ], + compatibilityFlags = [ + "nodejs_compat", + "typescript_implemented_streams", + "strict_compression_checks", + "experimental", + ], + ) + ), + ], + autogates = [ + "workerd-autogate-per-isolate-javascript-bootstrap", + ], +); diff --git a/src/workerd/io/BUILD.bazel b/src/workerd/io/BUILD.bazel index c2bc380e492..e7cb65b1857 100644 --- a/src/workerd/io/BUILD.bazel +++ b/src/workerd/io/BUILD.bazel @@ -65,6 +65,7 @@ wd_cc_library( implementation_deps = [ "//src/per_isolate", "//src/rust/jsg", + "//src/workerd/api:compression", "//src/workerd/api:crypto-crc-impl", "//src/workerd/api:data-url", "//src/workerd/api/node:exceptions", diff --git a/src/workerd/io/per-isolate-bootstrap.c++ b/src/workerd/io/per-isolate-bootstrap.c++ index 121a8278c86..01820849640 100644 --- a/src/workerd/io/per-isolate-bootstrap.c++ +++ b/src/workerd/io/per-isolate-bootstrap.c++ @@ -4,6 +4,7 @@ #include "per-isolate-bootstrap.h" +#include #include #include #include @@ -133,10 +134,14 @@ static void MarkPromiseHandledFastApi(v8::Local unused, v8::Local& args) { - auto name = jsg::JsValue(args[0]); - auto str = JSG_REQUIRE_NONNULL( - name.tryCast(), TypeError, "getApiSymbol() expects a string argument"); - args.GetReturnValue().Set(v8::Symbol::ForApi(args.GetIsolate(), str)); + // liftKj converts a thrown kj/jsg exception (the validation TypeError below) into a JS + // exception; without it the raw callback would let it escape and take down the process. + jsg::liftKj(args, [&]() -> v8::Local { + auto name = jsg::JsValue(args[0]); + auto str = JSG_REQUIRE_NONNULL( + name.tryCast(), TypeError, "getApiSymbol() expects a string argument"); + return v8::Symbol::ForApi(args.GetIsolate(), str); + }); } static const v8::CFunction fast_mark_promise_handled_ = @@ -167,7 +172,10 @@ v8::Local getMethod(jsg::Lock& js, v8::FunctionCallback callback) { // Creates an object with methods for performing fast type checks on JS values. // Because we are not fully bootstrapped at this point, we don't want to rely // on jsg::Object and the type wrapper system, etc. Instead, just use a plain -// object with some properties set. +// object with some properties set. (Members implemented in the api layer, like +// newCompressionCodec, keep their knowledge there and are wired here by name; +// their callbacks may use the type wrapper at CALL time -- the isolate is fully +// set up by then -- through jsg::Lock's type-handler lookup.) jsg::JsRef createUtilsObject(jsg::Lock& js) { static constexpr std::string_view names[] = { #define V(Name) "is" #Name, @@ -177,6 +185,7 @@ jsg::JsRef createUtilsObject(jsg::Lock& js) { "isAnyArrayBuffer", "markPromiseHandled", "getApiSymbol", + "newCompressionCodec", }; auto tmpl = v8::DictionaryTemplate::New(js.v8Isolate, names); v8::MaybeLocal values[] = { @@ -186,6 +195,7 @@ jsg::JsRef createUtilsObject(jsg::Lock& js) { getFastMethodNoSideEffect(js, IsAnyArrayBuffer, &fast_is_any_array_buffer_), getFastMethod(js, MarkPromiseHandled, &fast_mark_promise_handled_), getMethod(js, GetApiSymbol), + getMethod(js, api::newCompressionCodecCallback), }; static_assert(kj::arrayPtr(names).size() == kj::arrayPtr(values).size()); diff --git a/src/wpt/BUILD.bazel b/src/wpt/BUILD.bazel index 71513028c0e..be3d161c3da 100644 --- a/src/wpt/BUILD.bazel +++ b/src/wpt/BUILD.bazel @@ -116,6 +116,23 @@ wpt_test( wpt_directory = "@wpt//:streams@module", ) +# TODO(streams-ts): registration pending an investigation: the suite's large-file +# fetches (/media/, ~400KB) fail with "Network connection lost" under the +# typescript_implemented_streams flag -- a fetch/serving-path issue, not a codec one +# (the same payload sizes round-trip in-worker in ts-compression-streams-test). +wpt_test( + name = "compression-ts", + autogates = ["workerd-autogate-per-isolate-javascript-bootstrap"], + compat_flags = ["typescript_implemented_streams"], + config = "compression-test-ts.ts", + start_server = True, + target_compatible_with = select({ + "@platforms//os:windows": ["@platforms//:incompatible"], + "//conditions:default": [], + }), + wpt_directory = "@wpt//:compression@module", +) + wpt_test( name = "streams-ts", size = "large", diff --git a/src/wpt/compression-test-ts.ts b/src/wpt/compression-test-ts.ts new file mode 100644 index 00000000000..903737f7261 --- /dev/null +++ b/src/wpt/compression-test-ts.ts @@ -0,0 +1,100 @@ +// Copyright (c) 2017-2022 Cloudflare, Inc. +// Licensed under the Apache 2.0 license found in the LICENSE file or at: +// https://opensource.org/licenses/Apache-2.0 + +import { type TestRunnerConfig } from 'harness/harness'; + +// The compression WPT suite against the TypeScript streams implementation's +// CompressionStream/DecompressionStream pair (webstreams/compression.ts over +// the shared C++ CodecStage). Expectations match the legacy configuration +// (compression-test.ts): the pair is behavior-matching by design. +export default { + 'compression-bad-chunks.any.js': { + comment: 'brotli compression is not supported', + expectedFailures: [/brotli/], + }, + 'compression-constructor-error.any.js': {}, + 'compression-including-empty-chunk.any.js': { + comment: 'brotli compression is not supported', + expectedFailures: [ + "the result of compressing [,Hello,Hello] with brotli should be 'HelloHello'", + "the result of compressing [Hello,,Hello] with brotli should be 'HelloHello'", + "the result of compressing [Hello,Hello,] with brotli should be 'HelloHello'", + ], + }, + 'compression-large-flush-output.any.js': { + comment: 'brotli compression is not supported', + expectedFailures: ['brotli compression with large flush output'], + }, + 'compression-multiple-chunks.any.js': { + comment: 'brotli compression is not supported', + expectedFailures: [/compressing \d+ chunks with brotli should work/], + }, + 'compression-output-length.any.js': { + comment: 'brotli compression is not supported', + expectedFailures: [ + 'the length of brotli data should be shorter than that of the original data', + ], + }, + 'compression-stream.any.js': { + comment: 'brotli compression is not supported', + expectedFailures: [ + /brotli .* data should be reinflated back to its origin/, + ], + }, + 'compression-with-detach.any.js': {}, + 'decompression-bad-chunks.any.js': { + comment: 'brotli compression is not supported', + expectedFailures: [/brotli/], + }, + 'decompression-buffersource.any.js': { + comment: 'brotli compression is not supported', + expectedFailures: [/brotli/], + }, + 'decompression-constructor-error.any.js': { + comment: + 'brotli compression is not supported - these pass because brotli throws', + }, + 'decompression-correct-input.any.js': { + comment: 'brotli compression is not supported', + expectedFailures: [/.*brotli.*/], + }, + 'decompression-corrupt-input.any.js': { + comment: 'brotli compression is not supported', + expectedFailures: [/brotli/], + }, + 'decompression-empty-input.any.js': { + comment: 'brotli compression is not supported', + expectedFailures: [/.*brotli.*/], + }, + 'decompression-extra-input.any.js': { + comment: 'brotli compression is not supported', + expectedFailures: [/brotli/], + }, + 'decompression-split-chunk.any.js': { + comment: 'brotli compression is not supported', + expectedFailures: [/.*brotli/], + }, + 'decompression-uint8array-output.any.js': { + comment: 'brotli compression is not supported', + expectedFailures: [ + 'decompressing brotli output should give Uint8Array chunks', + ], + }, + 'decompression-with-detach.any.js': { + comment: + 'Environmental, not a streams defect: compression-with-detach.any.js runs first in ' + + 'the same isolate and installs its Object.prototype.then trap without configurable, ' + + 'so this test\'s identical defineProperty throws "Cannot redefine property". Browsers ' + + 'give each .any.js file a fresh global; the shared-isolate harness cannot (the ' + + 'leftover is non-configurable, so it cannot even be deleted between files).', + expectedFailures: [ + 'data should be correctly decompressed even if input is detached partway', + ], + }, + // The interface-prototype subtests that fail against the legacy classes (see + // compression-test.ts) pass against the TypeScript pair: its prototype property + // attributes follow the IDL rules. + 'idlharness.https.any.js': {}, + 'third_party/pako/pako_inflate.min.js': {}, +} satisfies TestRunnerConfig; diff --git a/src/wpt/compression-test.ts b/src/wpt/compression-test.ts index e38b6c80f8a..713ffca8b13 100644 --- a/src/wpt/compression-test.ts +++ b/src/wpt/compression-test.ts @@ -84,7 +84,12 @@ export default { ], }, 'decompression-with-detach.any.js': { - comment: 'Detach test fails - needs investigation', + comment: + 'Environmental, not a streams defect: compression-with-detach.any.js runs first in ' + + 'the same isolate and installs its Object.prototype.then trap without configurable, ' + + 'so this test\'s identical defineProperty throws "Cannot redefine property". Browsers ' + + 'give each .any.js file a fresh global; the shared-isolate harness cannot (the ' + + 'leftover is non-configurable, so it cannot even be deleted between files).', expectedFailures: [ 'data should be correctly decompressed even if input is detached partway', ],