diff --git a/devbox.lock b/devbox.lock index d834c1d..dde6860 100644 --- a/devbox.lock +++ b/devbox.lock @@ -103,7 +103,7 @@ }, "nodejs@20": { "last_modified": "2025-09-21T09:21:16Z", - "plugin_version": "0.0.2", + "plugin_version": "0.0.4", "resolved": "github:NixOS/nixpkgs/a1f79a1770d05af18111fbbe2a3ab2c42c0f6cd0#nodejs_20", "source": "devbox-search", "version": "20.19.5", diff --git a/integration_tests/src/Main.gren b/integration_tests/src/Main.gren index f3becf5..3c2a29f 100644 --- a/integration_tests/src/Main.gren +++ b/integration_tests/src/Main.gren @@ -5,6 +5,7 @@ module Main exposing ( .. ) import Test.Crypto as Crypto import Test.Task as Task +import Test.Stream as Stream import Test.Runner.Effectful exposing (concat) import Node @@ -18,6 +19,7 @@ main = (concat [ Crypto.tests , Task.tests + , Stream.tests ] ) ) diff --git a/integration_tests/src/Test/Stream.gren b/integration_tests/src/Test/Stream.gren new file mode 100644 index 0000000..b0af847 --- /dev/null +++ b/integration_tests/src/Test/Stream.gren @@ -0,0 +1,217 @@ +module Test.Stream exposing (tests) + +import Bytes exposing (Bytes) +import Expect +import Process +import Stream +import Task exposing (Task) +import Test.Runner.Effectful exposing (await, awaitError, concat, describe, test) + + +tests : Test.Runner.Effectful.Test +tests = + describe "Stream" + [ describe "concurrent operations on one writable stream" + [ await + (writableSink + |> Task.andThen + (\w -> + Task.concurrent + [ toUnit (Stream.write chunk w) + , toUnit (Stream.write chunk w) + ] + ) + ) + "two writes issued in the same tick" + (\_ -> + test "both writes succeed (no Locked)" + (\_ -> Expect.pass) + ) + , await + (writableSink + |> Task.andThen + (\w -> + Task.concurrent + [ toUnit (Stream.enqueue chunk w) + , toUnit (Stream.enqueue chunk w) + ] + ) + ) + "two enqueues issued in the same tick" + (\_ -> + test "both enqueues succeed (no Locked)" + (\_ -> Expect.pass) + ) + , await + (writableSink + |> Task.andThen + (\w -> + Task.concurrent + [ toUnit (Stream.write chunk w) + , toUnit (Stream.enqueue chunk w) + ] + ) + ) + "a write and an enqueue issued in the same tick" + (\_ -> + test "both succeed (no Locked)" + (\_ -> Expect.pass) + ) + , await + (writableSink + |> Task.andThen + (\w -> + Task.concurrent + [ toUnit (Stream.write chunk w) + , toUnit (Stream.write chunk w) + ] + |> Task.andThen + (\_ -> + Task.concurrent + [ toUnit (Stream.write chunk w) + , toUnit (Stream.write chunk w) + ] + ) + ) + ) + "two writes, then two more writes on the next tick" + (\_ -> + test "all four writes succeed (no Locked)" + (\_ -> Expect.pass) + ) + ] + , describe "write and close ordering" + [ await + (writableSink + |> Task.andThen + (\w -> + Stream.write chunk w + |> Task.andThen (\_ -> Stream.closeWritable w) + ) + ) + "write then close" + (\_ -> + test "both succeed" + (\_ -> Expect.pass) + ) + , awaitError + (writableSink + |> Task.andThen + (\w -> + Stream.closeWritable w + |> Task.andThen (\_ -> Stream.write chunk w) + ) + ) + "close then write" + (\err -> + test "the write fails with Cancelled once the stream is closed" + (\_ -> + when err is + Stream.Cancelled _ -> + Expect.pass + + _ -> + Expect.fail "expected the write to fail with Cancelled" + ) + ) + ] + , describe "writes to a pipe-locked stream report Locked" + [ awaitError + writeToPipeThroughLockedWritable + "write to a writable locked by an active pipeThrough" + (\err -> + test "fails with Locked" + (\_ -> Expect.equal Stream.Locked err) + ) + , awaitError + writeToPipeToLockedWritable + "write to a writable locked by an active pipeTo" + (\err -> + test "fails with Locked" + (\_ -> Expect.equal Stream.Locked err) + ) + ] + ] + + + +-- HELPERS + + +{-| A writable byte stream that accepts writes without applying indefinite +backpressure: `nullTransformation` discards input synchronously, yet it is a +real `WritableStream` whose `getWriter()` / `releaseLock()` locking is identical +to a file or stdout stream. That locking is what these tests exercise. +-} +writableSink : Task x (Stream.Writable Bytes) +writableSink = + Stream.nullTransformation Bytes.empty + |> Task.map Stream.writable + + +{-| Contents are irrelevant; only the locking behaviour matters. +-} +chunk : Bytes +chunk = + Bytes.fromString "x" + + +{-| Discard the result so heterogeneous stream operations (which return +different success types) can be combined in a single `Task.concurrent`. +-} +toUnit : Task x a -> Task x {} +toUnit task = + Task.map (\_ -> {}) task + + +{-| Establish a `pipeThrough` pipe, which locks the target transformation's +writable side for the lifetime of the pipe, then write to that writable. The +write is expected to fail with `Locked`. Closing the source writable lets the +background pipe drain so the process exits cleanly. +-} +writeToPipeThroughLockedWritable : Task Stream.Error (Stream.Writable Bytes) +writeToPipeThroughLockedWritable = + Stream.identityTransformation + |> Task.andThen + (\source -> + Stream.identityTransformation + |> Task.andThen + (\target -> + Stream.pipeThrough target (Stream.readable source) + |> Task.andThen (\_ -> Stream.write chunk (Stream.writable target)) + |> Task.onError + (\err -> + Stream.closeWritable (Stream.writable source) + |> Task.andThen (\_ -> Task.fail err) + ) + ) + ) + + +{-| Spawn a long-lived `pipeTo` (it resolves only once the source readable +closes) so it locks the writable in the background, then write to that +writable. The write is expected to fail with `Locked`. Closing the source +writable completes the pipe so the spawned process finishes and the process +exits cleanly. +-} +writeToPipeToLockedWritable : Task Stream.Error (Stream.Writable Bytes) +writeToPipeToLockedWritable = + Stream.identityTransformation + |> Task.andThen + (\source -> + Stream.nullTransformation Bytes.empty + |> Task.andThen + (\sink -> + Stream.pipeTo (Stream.writable sink) (Stream.readable source) + |> Process.spawn + |> Task.andThen + (\_ -> + Stream.write chunk (Stream.writable sink) + |> Task.onError + (\err -> + Stream.closeWritable (Stream.writable source) + |> Task.andThen (\_ -> Task.fail err) + ) + ) + ) + ) diff --git a/src/Gren/Kernel/Stream.js b/src/Gren/Kernel/Stream.js index 6efc3ba..a23e2fe 100644 --- a/src/Gren/Kernel/Stream.js +++ b/src/Gren/Kernel/Stream.js @@ -54,52 +54,104 @@ var _Stream_cancellationErrorString = function (err) { return "Unknown error"; }; -var _Stream_write = F2(function (value, stream) { - return __Scheduler_binding(function (callback) { - if (stream.locked) { - return callback(__Scheduler_fail(__Stream_Locked)); - } - - if (value instanceof DataView) { - value = new Uint8Array(value.buffer, value.byteOffset, value.byteLength); - } - - const writer = stream.getWriter(); - writer.ready - .then(() => { - const writePromise = writer.write(value); - writer.releaseLock(); - return writePromise; - }) - .then(() => { - callback(__Scheduler_succeed(stream)); - }) - .catch((err) => { +// One FIFO chain per WritableStream, shared by every writer-acquiring +// operation (write, enqueue, closeWritable). Concurrent operations on the same +// stream serialize through this chain instead of colliding on +// getWriter()/releaseLock(), which would otherwise surface a spurious `Locked` +// because the lock is acquired synchronously but released in a later microtask. +var _Stream_writeChains = new WeakMap(); + +function _Stream_writeNoop() {} + +// Schedule `work` (which must acquire and release its own writer) on the +// per-stream FIFO chain. Returns a promise that resolves with `work`'s outcome. +// The chain itself is kept alive on both success and failure +// (`run.then(noop, noop)`) so one failed write can't starve the queue. If +// `work` rejects with { __grenStreamLocked: true } the caller translates it to +// the `Locked` error. +function _Stream_runChained(stream, work) { + var prev = _Stream_writeChains.get(stream); + if (!prev) { + prev = Promise.resolve(); + } + var run = prev.then(work); + _Stream_writeChains.set( + stream, + run.then(_Stream_writeNoop, _Stream_writeNoop), + ); + return run; +} + +// Rejects the promise with the sentinel value that identifies an error of Locked. +function _Stream_rejectLocked() { + return Promise.reject({ __grenStreamLocked: true }); +} + +// Routes the settled `run` promise to a Scheduler callback, mapping the +// `__grenStreamLocked` sentinel to `Locked` and everything else to `Cancelled`. +function _Stream_reportRun(run, callback, onSuccess) { + run.then( + function () { + callback(onSuccess()); + }, + function (err) { + if (err && err.__grenStreamLocked) { + callback(__Scheduler_fail(__Stream_Locked)); + } else { callback( __Scheduler_fail( __Stream_Cancelled(_Stream_cancellationErrorString(err)), ), ); + } + }, + ); +} + +function _Stream_toUint8Array(value) { + if (value instanceof DataView) { + return new Uint8Array(value.buffer, value.byteOffset, value.byteLength); + } + return value; +} + +var _Stream_write = F2(function (value, stream) { + return __Scheduler_binding(function (callback) { + var bytes = _Stream_toUint8Array(value); + var run = _Stream_runChained(stream, function () { + if (stream.locked) { + return _Stream_rejectLocked(); + } + var writer = stream.getWriter(); + return writer.ready.then(function () { + var writePromise = writer.write(bytes); + writer.releaseLock(); + return writePromise; }); + }); + + _Stream_reportRun(run, callback, function () { + return __Scheduler_succeed(stream); + }); }); }); var _Stream_enqueue = F2(function (value, stream) { return __Scheduler_binding(function (callback) { - if (stream.locked) { - return callback(__Scheduler_fail(__Stream_Locked)); - } - - if (value instanceof DataView) { - value = new Uint8Array(value.buffer, value.byteOffset, value.byteLength); - } - - const writer = stream.getWriter(); - writer.ready.then(() => { - writer.write(value); - writer.releaseLock(); + var bytes = _Stream_toUint8Array(value); + var run = _Stream_runChained(stream, function () { + if (stream.locked) { + return _Stream_rejectLocked(); + } + var writer = stream.getWriter(); + return writer.ready.then(function () { + writer.write(bytes); + writer.releaseLock(); + }); + }); - callback(__Scheduler_succeed(stream)); + _Stream_reportRun(run, callback, function () { + return __Scheduler_succeed(stream); }); }); }); @@ -130,15 +182,25 @@ var _Stream_cancelWritable = F2(function (reason, stream) { var _Stream_closeWritable = function (stream) { return __Scheduler_binding(function (callback) { - if (stream.locked) { - return callback(__Scheduler_fail(__Stream_Locked)); - } - - const writer = stream.getWriter(); - writer.close(); - writer.releaseLock(); + var run = _Stream_runChained(stream, function () { + if (stream.locked) { + return _Stream_rejectLocked(); + } + var writer = stream.getWriter(); + return writer.close().then( + function () { + writer.releaseLock(); + }, + function (err) { + writer.releaseLock(); + throw err; + }, + ); + }); - callback(__Scheduler_succeed({})); + _Stream_reportRun(run, callback, function () { + return __Scheduler_succeed({}); + }); }); };