diff --git a/SPEC.md b/SPEC.md index 92c01dba5..1df1fe66d 100644 --- a/SPEC.md +++ b/SPEC.md @@ -79,7 +79,9 @@ record frame { } ``` -It is assumed that streams using this framing protocol can communicate "closing" to peers using some out-of-band mechanism. +An empty frame, i.e. a frame with empty `data`, closes its `path`: the sender MUST NOT send any more frames on that `path`, including another empty frame, and a receiver MUST treat a frame received on a closed `path` as an error. Closing a `path` does not close the paths nested in it. A sender MAY leave a `path` open, unless stated otherwise (see [Futures](#futures)). + +Frames do not communicate the end of transmit, i.e. that no more frames follow. Each transport MUST communicate it using its own mechanism, for example, TCP shutdown or finishing a QUIC stream. The end of transmit closes every `path`. ## Transport specifications @@ -106,6 +108,8 @@ For example: If `v` is pending, encoded `bool` value is sent on index `0` (corresponding to first parameter) +A `future` without a payload type has an empty value encoding, so once such a pending future resolves, the sender MUST close its index, either with an empty frame or by ending transmit (see [Framed stream specification](#framed-stream-specification)), and the receiver MUST consider it resolved once the index is closed. + ### Streams `stream` values are encoded as `variant stream { pending, ready(list) }`. diff --git a/crates/transport/src/frame/conn/client.rs b/crates/transport/src/frame/conn/client.rs index 35afff0bc..b3b86b924 100644 --- a/crates/transport/src/frame/conn/client.rs +++ b/crates/transport/src/frame/conn/client.rs @@ -11,7 +11,7 @@ use crate::frame::conn::{Incoming, Outgoing}; use crate::frame::{ConnHandler, PROTOCOL}; /// Encodes the header of an invocation, which is followed by `params_len` bytes of -/// parameters, to a `BytesMut` +/// parameters, to a `BytesMut`. If `params_len` is 0, only the header is encoded. /// /// This is low-level API, most users should use [`invoke`]. #[instrument(level = "trace", skip_all)] @@ -31,8 +31,10 @@ pub fn encode_invocation( buf.put_u8(PROTOCOL); CoreNameEncoder.encode(instance, buf)?; CoreNameEncoder.encode(func, buf)?; - buf.put_u8(0); - Leb128Encoder.encode(params_len, buf)?; + if params_len > 0 { + buf.put_u8(0); + Leb128Encoder.encode(params_len, buf)?; + } Ok(()) } diff --git a/crates/transport/src/frame/conn/mod.rs b/crates/transport/src/frame/conn/mod.rs index 379760b1b..ab3c458c2 100644 --- a/crates/transport/src/frame/conn/mod.rs +++ b/crates/transport/src/frame/conn/mod.rs @@ -94,23 +94,30 @@ impl Header { } } +/// Sender of a subscribed path +#[derive(Debug)] +enum TxSlot { + Open(mpsc::Sender>), + Closed, +} + /// Index trie containing async stream subscriptions #[derive(Debug, Default)] enum IndexTrie { #[default] Empty, Leaf { - tx: Option>>, + tx: Option, rx: Option>>, }, IndexNode { - tx: Option>>, + tx: Option, rx: Option>>, nested: Vec>, }, // TODO: Add partially-indexed `WildcardIndexNode` WildcardNode { - tx: Option>>, + tx: Option, rx: Option>>, nested: Option>, }, @@ -131,7 +138,10 @@ impl<'a> ), ) -> Self { match path { - [] => Self::Leaf { tx: Some(tx), rx }, + [] => Self::Leaf { + tx: Some(TxSlot::Open(tx)), + rx, + }, [None, path @ ..] => Self::WildcardNode { tx: None, rx: None, @@ -190,13 +200,23 @@ impl]>> FromIterator

for IndexTrie { } impl IndexTrie { + /// Builds a trie of `paths`, which sends frames received on the root path to `tx` + fn new]>>( + paths: impl IntoIterator, + tx: mpsc::Sender>, + ) -> Self { + let mut index: Self = paths + .into_iter() + .filter(|path| !path.as_ref().is_empty()) + .collect(); + index.insert(&[], tx, None); + index + } + /// Calls `f` with the sender and receiver of each node in the trie fn for_each( &mut self, - f: &mut impl FnMut( - &mut Option>>, - &mut Option>>, - ), + f: &mut impl FnMut(&mut Option, &mut Option>>), ) { match self { Self::Empty => {} @@ -261,14 +281,15 @@ impl IndexTrie { } } - /// Gets a sender - #[instrument(level = "trace", skip(self), ret(level = "trace"))] - fn get_tx(&mut self, path: &[usize]) -> Option>> { + /// Gets the sender slot of a subscription + #[instrument(level = "trace", skip(self))] + fn get_tx_mut(&mut self, path: &[usize]) -> Option<&mut TxSlot> { let Some((i, path)) = path.split_first() else { return match self { Self::Empty => None, - Self::Leaf { tx, .. } => tx.clone(), - Self::IndexNode { tx, .. } | Self::WildcardNode { tx, .. } => tx.clone(), + Self::Leaf { tx, .. } + | Self::IndexNode { tx, .. } + | Self::WildcardNode { tx, .. } => tx.as_mut(), }; }; match self { @@ -276,7 +297,7 @@ impl IndexTrie { Self::IndexNode { nested, .. } => { let nested = nested.get_mut(*i)?; let nested = nested.as_mut()?; - nested.get_tx(path) + nested.get_tx_mut(path) } // TODO: Demux the subscription //Self::WildcardNode { ref mut nested, .. } => { // nested.as_mut().and_then(|nested| nested.take(path)) @@ -284,11 +305,13 @@ impl IndexTrie { } } - /// Closes all senders in the trie + /// Closes all open senders in the trie #[instrument(level = "trace", skip(self), ret(level = "trace"))] fn close_tx(&mut self) { self.for_each(&mut |tx, _| { - tx.take(); + if let Some(TxSlot::Open(..)) = tx { + tx.take(); + } }); } @@ -330,7 +353,7 @@ impl IndexTrie { } Self::IndexNode { tx, rx, nested } => match (&tx, &rx, path) { (None, None, []) => { - *tx = Some(sender); + *tx = Some(TxSlot::Open(sender)); *rx = receiver; true } @@ -351,7 +374,7 @@ impl IndexTrie { }, Self::WildcardNode { tx, rx, nested } => match (&tx, &rx, path) { (None, None, []) => { - *tx = Some(sender); + *tx = Some(TxSlot::Open(sender)); *rx = receiver; true } @@ -413,7 +436,7 @@ impl Incoming { { let (rx_tx, rx_rx) = mpsc::channel(128); let state = Arc::new(IngressState { - index: std::sync::Mutex::new(paths.into_iter().collect()), + index: std::sync::Mutex::new(IndexTrie::new(paths, rx_tx)), err: std::sync::OnceLock::default(), unprocessed: AtomicBool::default(), }); @@ -424,11 +447,10 @@ impl Incoming { async move { let res = async { let mut io = pin!(async { - let res = ingress(&mut rx, &state.index, &rx_tx).await; + let res = ingress(&mut rx, &state.index).await; if let Err(err) = &res { _ = state.err.set(copy_io_error(err)); } - drop(rx_tx); trace!("shutting down index trie"); state .index @@ -534,7 +556,9 @@ impl AsyncRead for Incoming { trace!(buf = ?buf.filled(), "read buffer"); if buf.filled().is_empty() { self.rx.take(); - if let Some(err) = self.state.err.get() { + if let Some(err) = self.state.err.get() + && !self.state.is_closed(&self.path) + { return Poll::Ready(Err(copy_io_error(err))); } } @@ -548,6 +572,14 @@ struct IngressState { unprocessed: AtomicBool, } +impl IngressState { + /// Returns whether an empty frame closed the `path` + fn is_closed(&self, path: &[usize]) -> bool { + let mut index = self.index.lock().unwrap_or_else(PoisonError::into_inner); + matches!(index.get_tx_mut(path), Some(TxSlot::Closed)) + } +} + enum IngressGuard { Abort { _task: AbortOnDropHandle<()> }, Drain { _dropped: oneshot::Sender<()> }, @@ -637,6 +669,7 @@ impl Outgoing { } /// Writes `buf` to the stream as a single frame, without copying it. + /// Empty `buf` is not written, since an empty frame closes the stream, see [`Self::close`]. pub async fn write_bytes(&mut self, buf: Bytes) -> std::io::Result<()> { if buf.is_empty() { return Ok(()); @@ -644,6 +677,12 @@ impl Outgoing { self.reserve().await?; self.send_reserved(buf) } + + /// Closes the stream by writing an empty frame. No more data may be written to it afterwards. + pub async fn close(&mut self) -> std::io::Result<()> { + self.reserve().await?; + self.send_reserved(Bytes::default()) + } } impl AsyncWrite for Outgoing { @@ -677,7 +716,6 @@ impl AsyncWrite for Outgoing { async fn ingress( mut rx: impl AsyncRead + Unpin, index: &std::sync::Mutex, - param_tx: &mpsc::Sender>, ) -> std::io::Result<()> { loop { trace!("reading path length"); @@ -692,8 +730,8 @@ async fn ingress( let n = usize::try_from(n) .map_err(|err| std::io::Error::new(std::io::ErrorKind::InvalidInput, err))?; trace!(n, "read path length"); - let tx = if n == 0 { - param_tx + let path = if n == 0 { + Vec::default() } else { let mut path = Vec::with_capacity(n.min(MAX_INITIAL_PATH_CAPACITY)); for i in 0..n { @@ -704,21 +742,34 @@ async fn ingress( path.push(p); } trace!(?path, "read path"); - + path + }; + trace!("reading data length"); + let n = rx.read_u32_leb128().await?; + trace!(n, "read data length"); + let tx = { trace!("locking index trie"); - let mut index = index + let mut trie = index .lock() .map_err(|err| std::io::Error::other(err.to_string()))?; - &index.get_tx(&path).ok_or_else(|| { + let tx = trie.get_tx_mut(&path).ok_or_else(|| { std::io::Error::new( std::io::ErrorKind::NotFound, format!("`{path:?}` subscription not found"), ) - })? + })?; + if n == 0 { + trace!(?path, "closing path"); + if let TxSlot::Closed = mem::replace(tx, TxSlot::Closed) { + return Err(error_closed(&path)); + } + continue; + } + let TxSlot::Open(tx) = tx else { + return Err(error_closed(&path)); + }; + tx.clone() }; - trace!("reading data length"); - let n = rx.read_u32_leb128().await?; - trace!(n, "read data length"); let mut data = (&mut rx).take(n.into()); while data.limit() > 0 { let k = usize::try_from(data.limit()) @@ -750,6 +801,14 @@ fn close_rx(state: &IngressState) -> std::io::Result<()> { Ok(()) } +/// Returns the error for a frame received on a closed `path` +fn error_closed(path: &[usize]) -> std::io::Error { + std::io::Error::new( + std::io::ErrorKind::InvalidData, + format!("received a frame on closed path `{path:?}`"), + ) +} + /// Returns the error for received data, which was not processed fn error_unprocessed() -> std::io::Error { std::io::Error::new( @@ -892,6 +951,21 @@ mod tests { done(done_rx).await.unwrap(); } + #[test_log::test(tokio::test)] + async fn incoming_empty_frame_closes_path() { + let (mut tx, mut rx, done_rx) = incoming(true); + let mut sub = rx.index(&[0]).unwrap(); + tx.write_all(&[0x00, 0x00]).await.unwrap(); + tx.write_all(&FRAME).await.unwrap(); + tx.write_all(&[0x01, 0x00, 0x00]).await.unwrap(); + tx.write_all(&[0x01, 0x01, 0x01, 0x2a]).await.unwrap(); + assert_eq!(failed(done_rx).await, ErrorKind::NotFound); + let mut buf = vec![]; + sub.read_to_end(&mut buf).await.unwrap(); + assert_eq!(buf, [0x2a]); + assert_eq!(rx.read_to_end(&mut Vec::default()).await.unwrap(), 0); + } + #[test_log::test(tokio::test)] async fn incoming_drop_rejects_data() { let (mut tx, rx, done_rx) = incoming(true); @@ -961,42 +1035,49 @@ mod tests { } #[test_log::test(tokio::test)] - async fn ingress_truncated_path() { - let index = std::sync::Mutex::new(IndexTrie::Empty); - let (tx, _rx) = mpsc::channel(1); - let err = ingress([0x02, 0x00].as_slice(), &index, &tx) - .await - .unwrap_err(); - assert_eq!(err.kind(), std::io::ErrorKind::UnexpectedEof); + async fn ingress_rejects_invalid_frames() { + for (frames, kind) in [ + ([0x02, 0x00].as_slice(), ErrorKind::UnexpectedEof), + ( + &[0x00, 0xff, 0xff, 0xff, 0xff, 0x0f, 0x42], + ErrorKind::UnexpectedEof, + ), + (&[0x00, 0x00, 0x00, 0x01, 0x2a], ErrorKind::InvalidData), + (&[0x00, 0x00, 0x00, 0x00], ErrorKind::InvalidData), + ( + &[0x01, 0x00, 0x00, 0x01, 0x00, 0x01, 0x2a], + ErrorKind::InvalidData, + ), + ( + &[0x01, 0x00, 0x00, 0x01, 0x00, 0x00], + ErrorKind::InvalidData, + ), + (&[0x01, 0x01, 0x00], ErrorKind::NotFound), + (&[0x01, 0x01, 0x01, 0x2a], ErrorKind::NotFound), + ] { + let (tx, _rx) = mpsc::channel(1); + let index = std::sync::Mutex::new(IndexTrie::new( + [[Some(0)].as_slice(), &[Some(1), Some(0)]], + tx, + )); + let err = ingress(frames, &index).await.unwrap_err(); + assert_eq!(err.kind(), kind, "{frames:02x?}"); + } } #[test_log::test(tokio::test)] async fn ingress_chunked_data() { - let index = std::sync::Mutex::new(IndexTrie::Empty); let (tx, mut rx) = mpsc::channel(4); + let index = std::sync::Mutex::new(IndexTrie::new::<[_; 0]>([], tx)); let n = DEFAULT_MAX_INITIAL_CAPACITY + 1; let mut frame = vec![0x00, 0x81, 0x80, 0x40]; frame.resize(frame.len() + n, 0x42); - ingress(frame.as_slice(), &index, &tx).await.unwrap(); - drop(tx); + ingress(frame.as_slice(), &index).await.unwrap(); + drop(index); let mut chunks = vec![]; while let Some(chunk) = rx.recv().await { chunks.push(chunk.unwrap().len()); } assert_eq!(chunks, [DEFAULT_MAX_INITIAL_CAPACITY, 1]); } - - #[test_log::test(tokio::test)] - async fn ingress_truncated_data() { - let index = std::sync::Mutex::new(IndexTrie::Empty); - let (tx, _rx) = mpsc::channel(1); - let err = ingress( - [0x00, 0xff, 0xff, 0xff, 0xff, 0x0f, 0x42].as_slice(), - &index, - &tx, - ) - .await - .unwrap_err(); - assert_eq!(err.kind(), std::io::ErrorKind::UnexpectedEof); - } } diff --git a/crates/transport/src/value.rs b/crates/transport/src/value.rs index ed29b5eae..ba844aaed 100644 --- a/crates/transport/src/value.rs +++ b/crates/transport/src/value.rs @@ -444,7 +444,11 @@ impl Sink for Outgoing { let mut enc = T::Encoder::default(); let mut buf = BytesMut::default(); enc.encode(item, &mut buf)?; - self.write_bytes(buf.freeze()).await?; + if buf.is_empty() { + self.close().await?; + } else { + self.write_bytes(buf.freeze()).await?; + } match enc.take_deferred() { Some(f) => f(self, Vec::default()).await, _ => Ok(()), @@ -1923,17 +1927,19 @@ where return Err(std::io::ErrorKind::UnexpectedEof.into()); }; let item = item?; + let rx = dec.decoder_mut().take_deferred(); + if rx.is_none() { + ensure_processed(&dec)?; + } if tx.send(item).is_err() { return Err(error_receiver_closed()); } - if let Some(rx) = dec.decoder_mut().take_deferred() { + if let Some(rx) = rx { let mut buf = mem::take(dec.read_buffer_mut()); let mut r = dec.into_inner(); buf.unsplit(mem::take(&mut r.buffer)); r.buffer = buf; rx(r, Vec::default()).await?; - } else { - ensure_processed(&dec)?; } Ok(()) } @@ -2626,6 +2632,11 @@ mod tests { let (_fut, f) = pending::(); let err = f(buffered(b"\x03foo\x2a"), Vec::default()).await; assert_eq!(err.unwrap_err().kind(), ErrorKind::InvalidData); + + let (fut, f) = pending:: + Send>>>(); + let err = f(buffered(b"\x2a"), Vec::default()).await; + assert_eq!(err.unwrap_err().kind(), ErrorKind::InvalidData); + assert!(fut.now_or_never().is_none()); } #[test_log::test(tokio::test)] @@ -2639,6 +2650,13 @@ mod tests { assert_eq!(err.unwrap_err().kind(), ErrorKind::InvalidData); } + #[test_log::test(tokio::test)] + async fn future_unit_resolves_on_end() { + let (fut, f) = pending:: + Send>>>(); + f(buffered(b""), Vec::default()).await.unwrap(); + fut.await; + } + #[test_log::test(tokio::test)] async fn future_rejects_dropped_receiver() { let (fut, f) = pending::(); diff --git a/go/frame_writer.go b/go/frame_writer.go index 5047c11c8..90684c36c 100644 --- a/go/frame_writer.go +++ b/go/frame_writer.go @@ -111,6 +111,9 @@ func sendFrame(ctx context.Context, ch chan<- Frame, frame Frame) error { } func (w *FrameStreamWriter) Write(p []byte) (int, error) { + if len(p) == 0 { + return 0, nil + } data := make([]byte, len(p)) copy(data, p) if err := sendFrame(w.ctx, w.ch, Frame{ diff --git a/go/frame_writer_test.go b/go/frame_writer_test.go index accf14284..94637cf49 100644 --- a/go/frame_writer_test.go +++ b/go/frame_writer_test.go @@ -71,6 +71,7 @@ func TestFrameStreamWriter(t *testing.T) { var dst sink w := wrpc.NewFrameStreamWriter(ctx, &dst) + assertWriteString(t, w, "") assertWriteString(t, w, "hello") assertClose(t, w) diff --git a/js/README.md b/js/README.md index f85c2d955..354dad07d 100644 --- a/js/README.md +++ b/js/README.md @@ -60,11 +60,12 @@ that I/O fails, the invocation fails and the transport is aborted. ### Streams and futures -`t.stream(elem)` and `t.future(some)` describe asynchronous values. A `stream` -parameter is given as an (async-)iterable of chunks (a `Uint8Array` for -`stream`, an array of elements otherwise); a `future` as a value, a promise, -or a thunk. Decoded `stream` results are async iterables of chunks and `future` -results are promises: +`t.stream(elem)` and `t.future(some)` describe asynchronous values, and +`t.future()` a bare `future` without a payload. A `stream` parameter is given as +an (async-)iterable of chunks (a `Uint8Array` for `stream`, an array of +elements otherwise); a `future` as a value, a promise, or a thunk. Decoded +`stream` results are async iterables of chunks and `future` results are +promises, resolving to `undefined` for a bare `future`: ```js import { t, invoke, fromWebStreams } from "@bytecodealliance/wrpc"; @@ -98,13 +99,19 @@ consumed, as in the Rust implementation. So consume streams concurrently with each other, with awaiting futures, and with `done`: awaiting a future while an earlier stream is left unread can wait forever, since the future's data may be queued behind the stream's on the transport. Leave a `for await` loop early (or call -`cancel()`) to discard the rest of a stream you no longer need, including the -streams and futures nested in the chunks you did not receive. Those nested in -the chunks you did receive stay live, so consume or cancel them too. +`cancel()`) to discard the rest of a stream you no longer need. + +As in the Rust implementation, received `list` and `stream` elements may not +carry pending `stream` or `future` values: receiving one fails the invocation, as +does sending one in a `stream` element. Data the peer sends for a sub-stream that the decoded values do not carry — for a `stream` in an `option` that is `none`, say — fails the invocation. +Received values never fail: if receiving a `future` fails, its promise never +settles, and if receiving a `stream` fails, it ends. The failure is reported by +`done`, as in the Rust implementation. + ### Serving `accept` is the server-side counterpart: read an invocation's header off a diff --git a/js/package-lock.json b/js/package-lock.json index 2179cd4b8..51acb0f18 100644 --- a/js/package-lock.json +++ b/js/package-lock.json @@ -1,12 +1,12 @@ { "name": "@bytecodealliance/wrpc", - "version": "0.2.1", + "version": "0.3.0", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "@bytecodealliance/wrpc", - "version": "0.2.1", + "version": "0.3.0", "license": "Apache-2.0 WITH LLVM-exception", "devDependencies": { "@eslint/js": "^10.0.1", diff --git a/js/package.json b/js/package.json index faa0df32f..a2267df87 100644 --- a/js/package.json +++ b/js/package.json @@ -1,6 +1,6 @@ { "name": "@bytecodealliance/wrpc", - "version": "0.2.1", + "version": "0.3.0", "description": "Transport-agnostic, component-native wRPC client and server for JavaScript", "type": "module", "license": "Apache-2.0 WITH LLVM-exception", diff --git a/js/src/index.d.ts b/js/src/index.d.ts index 4d1027462..6b45b22a2 100644 --- a/js/src/index.d.ts +++ b/js/src/index.d.ts @@ -26,7 +26,7 @@ export type Type = | { kind: "variant"; cases: { name: string; ty: Type | null }[] } | { kind: "enum"; names: string[] } | { kind: "flags"; names: string[] } - | { kind: "future"; some: Type } + | { kind: "future"; some: Type | null } | { kind: "stream"; elem: Type } | { kind: "own" | "borrow"; name: string }; @@ -46,20 +46,16 @@ export type Value = /** * A descriptor recorded by {@link decodeValue} for a pending `stream`/`future` - * value, naming the relative sub-stream `path` and the placeholder to feed, or - * for a `stream` sent `inline` as a single ready chunk whose elements carry - * pending values. + * value, naming the relative sub-stream `path` and the placeholder to feed. */ export type DecodeDeferred = | { path: number[]; kind: "future"; - ty: Type; + ty: Type | null; resolve: (value: Value) => void; - reject: (err: unknown) => void; } - | { path: number[]; kind: "stream"; ty: Type; queue: Chan } - | { path: number[]; kind: "inline"; queue: Chan }; + | { path: number[]; kind: "stream"; ty: Type; queue: Chan }; // ----- types.js ------------------------------------------------------------- @@ -79,7 +75,7 @@ export namespace types { export const string: Type; export function list(elem: Type): Type; export function option(some: Type): Type; - export function future(some: Type): Type; + export function future(some?: Type | null): Type; export function stream(elem: Type): Type; export function result(ok?: Type | null, err?: Type | null): Type; export function tuple(...elems: Type[]): Type; @@ -138,8 +134,6 @@ export class Chan implements AsyncIterable { readonly capacity: number; readonly closed: boolean; readonly cancelled: boolean; - /** The number of values handed to the consumer. */ - readonly taken: number; /** * Call `f` once the consumer cancels the channel, or now if it already has. * Hooks are dropped once the consumer has received the end of the channel. @@ -181,7 +175,7 @@ export function encodeValue( w: Writer, ty: Type, v: Value, - sink?: { path: number[]; kind: "stream" | "future"; ty: Type; value: Value }[], + sink?: { path: number[]; kind: "stream" | "future"; ty: Type | null; value: Value }[], path?: number[], ): void; export function decodeValue(r: Reader, ty: Type, sink?: DecodeDeferred[], path?: number[]): Value; @@ -223,6 +217,7 @@ export class Outgoing { constructor(mux: Mux, path: number[]); path: number[]; write(data: Uint8Array): Promise; + close(): Promise; index(rel: number[]): Outgoing; } @@ -239,8 +234,7 @@ export class Mux { incoming(path: number[]): AsyncReader; /** * Declare the sub-streams of the pending values in the value decoded at - * `path` (the root tuple, a future's value, or a stream's element, in - * order), with paths relative to it. + * `path` (the root tuple, or a future's value), with paths relative to it. */ seal(path: number[], expected: { path: number[]; nested: boolean }[]): void; /** @@ -249,10 +243,8 @@ export class Mux { * fails the invocation. */ finish(path: number[], reader?: AsyncReader): void; - /** Discard the sub-stream at `path` and those nested in its children from index `from` (default 0) on. */ - discard(path: number[], from?: number): void; - /** Call `f` once the sub-stream at `path` is discarded. */ - ondiscard(path: number[], f: () => void): void; + /** Discard the sub-stream at `path`. */ + discard(path: number[]): void; outgoing(path: number[]): Outgoing; abort(err?: unknown): void; closeWrite(): Promise; diff --git a/js/src/io.js b/js/src/io.js index 552eb132a..e47756fac 100644 --- a/js/src/io.js +++ b/js/src/io.js @@ -269,8 +269,6 @@ export class Chan { this.senders = []; this.closed = false; this.cancelled = false; - /** The number of values handed to the consumer. */ - this.taken = 0; /** @type {unknown} */ this.error = undefined; } @@ -297,7 +295,6 @@ export class Chan { this.queue.push(value); return; } - this.taken++; w.resolve({ value, done: false }); } @@ -352,7 +349,6 @@ export class Chan { next() { if (this.queue.length) { const value = /** @type {T} */ (this.queue.shift()); - this.taken++; this.senders.shift()?.(); return Promise.resolve({ value, done: false }); } diff --git a/js/src/mux.js b/js/src/mux.js index 7e4b93283..3b7c09842 100644 --- a/js/src/mux.js +++ b/js/src/mux.js @@ -34,6 +34,7 @@ export const DISCARDED = new Error("sub-stream discarded"); * @property {number[]} path * @property {Chan} chan * @property {boolean} empty whether only empty frames arrived for it + * @property {boolean} closed whether an empty frame closed it */ /** @@ -49,13 +50,9 @@ export const DISCARDED = new Error("sub-stream discarded"); * @property {Chan | undefined} chan the data of the sub-stream at * this path, if one is expected here * @property {boolean} sealed whether every sub-stream nested in it is known - * @property {number} decided the number of stream elements known, whose - * children are all in the trie * @property {boolean} done whether all its own data has been read - * @property {Set | undefined} pruned the indices of the children - * forgotten once finished, beyond those of the `decided` stream elements - * @property {number} discardFrom `Infinity`, unless the node is discarded: then - * its own data and that of its children from this index on is dropped + * @property {boolean} closed whether an empty frame closed it + * @property {boolean} discarded whether its data is dropped * @property {Map> | undefined} pending the data * buffered for the paths through each missing child, by path key */ @@ -65,10 +62,9 @@ const node = (sealed) => ({ children: new Map(), chan: undefined, sealed, - decided: 0, done: false, - pruned: undefined, - discardFrom: Infinity, + closed: false, + discarded: false, pending: undefined, }); @@ -100,7 +96,7 @@ function* readFrameHeader(r) { } /** @param {number[]} path */ -const unexpectedError = (path) => new Error(`unexpected data for sub-stream [${path}]`); +export const unexpectedError = (path) => new Error(`unexpected data for sub-stream [${path}]`); /** @param {number[]} path @returns {never} */ function unexpected(path) { @@ -150,11 +146,25 @@ export class Outgoing { this.path = path; } - /** @param {Uint8Array} data @returns {Promise} */ + /** + * Write `data` as a frame, unless it is empty, since an empty frame closes + * the sub-stream. + * @param {Uint8Array} data @returns {Promise} + */ write(data) { + if (data.length === 0) return Promise.resolve(); return this.mux._writeFrame(this.path, data); } + /** + * Close the sub-stream by writing an empty frame. No more data may be + * written to it afterwards. + * @returns {Promise} + */ + close() { + return this.mux._writeFrame(this.path, new Uint8Array(0)); + } + /** @param {number[]} rel @returns {Outgoing} */ index(rel) { return new Outgoing(this.mux, [...this.path, ...rel]); @@ -170,7 +180,9 @@ export class Outgoing { * the root sub-stream `[]`, and those declared with {@link Mux.seal} as each * value carrying pending `stream`/`future` values is decoded, until they are * {@link Mux.finish finished}. Data for a path that may still be declared is - * buffered until it is; data for any other path fails the invocation. + * buffered until it is; data for any other path fails the invocation. Pending + * values nested in `list` or `stream` elements are not supported, so the + * expected sub-streams are bounded by the types of the decoded values. */ export class Mux { /** @param {Transport} transport */ @@ -216,8 +228,8 @@ export class Mux { if (!(await m.fill(1))) break; const [path, n] = await m.decode(readFrameHeader); if (n === 0) { - // Peers may send empty frames, even for sub-streams that are finished. - this._route(path, true); + // An empty frame closes its sub-stream, which may already be finished. + this._route(path, true)?.close(); continue; } const chan = this._route(path); @@ -234,44 +246,43 @@ export class Mux { } /** - * The channel of the sub-stream at `path`, `null` if its data is discarded, - * or where the lookup stopped if it is not yet known whether it is expected. - * With `empty`, for an empty frame, a finished sub-stream also gives `null`. + * The node of the expected sub-stream at `path`, or where the lookup + * stopped if it is not yet known whether it is expected. With `empty`, for + * an empty frame, a finished sub-stream is found too. * @param {number[]} path * @param {boolean} [empty] - * @returns {Chan | null | Undecided} + * @returns {Node | Undecided} */ _find(path, empty = false) { let n = this.root; for (const i of path) { - if (i >= n.discardFrom) return null; const child = n.children.get(i); - if (!child) { - const finished = i < n.decided || !!n.pruned?.has(i); - if (empty && finished) return null; - return n.sealed || finished ? unexpected(path) : { at: n, i }; - } + if (!child) return n.sealed ? unexpected(path) : { at: n, i }; n = child; } - if (n.done) return empty ? null : unexpected(path); - if (n.discardFrom !== Infinity) return null; - return n.chan ?? unexpected(path); + if (!n.chan || n.closed || (n.done && !empty)) unexpected(path); + return n; } /** * The channel to route the data of the sub-stream at `path` to, or - * `undefined` to drop it. With `empty`, for an empty frame, the data of a - * path that may still be expected is buffered as if it were finished. + * `undefined` to drop it. With `empty`, for an empty frame, the sub-stream + * is closed, and the data of a path that may still be expected is buffered + * as if it were finished. * @param {number[]} path * @param {boolean} [empty] */ _route(path, empty = false) { const found = this._find(path, empty); - if (found === null) return undefined; - if (found instanceof Chan) return found; + if (!("at" in found)) { + found.closed ||= empty; + return found.done || found.discarded ? undefined : found.chan; + } let pending = found.at.pending?.get(found.i)?.get(key(path)); - if (!pending) attach(found, (pending = { path, chan: new Chan(), empty })); + if (pending?.closed) unexpected(path); + if (!pending) attach(found, (pending = { path, chan: new Chan(), empty, closed: false })); pending.empty &&= empty; + pending.closed ||= empty; return pending.chan; } @@ -284,7 +295,6 @@ export class Mux { _takePending(path) { let n = this.root; for (const i of path) { - if (i >= n.discardFrom) return undefined; const child = n.children.get(i); if (!child) { const group = n.pending?.get(i); @@ -299,15 +309,14 @@ export class Mux { } /** - * The node at `path`, `undefined` if there is none, or `null` if it is - * discarded by an ancestor. With `create`, missing nodes are created. + * The node at `path`, `undefined` if there is none. With `create`, missing + * nodes are created. * @param {number[]} path * @param {boolean} [create] */ _get(path, create = false) { let n = this.root; for (const i of path) { - if (i >= n.discardFrom) return null; let child = n.children.get(i); if (!child) { if (!create) return undefined; @@ -321,39 +330,27 @@ export class Mux { /** * Declare the sub-streams nested in the value at `path` once it is decoded: - * the root tuple at `[]`, a future's value at its own path, or element `k` - * of a stream at `[...stream, k]`, decoded in order. `expected` lists the - * paths, relative to `path`, of the value's pending `stream`/`future` - * values, and whether each may nest sub-streams of its own. Data for any - * other path nested in the value fails the invocation. + * the root tuple at `[]`, or a future's value at its own path. `expected` + * lists the paths, relative to `path`, of the value's pending + * `stream`/`future` values, and whether each may nest sub-streams of its + * own. Data for any other path nested in the value fails the invocation. * @param {number[]} path * @param {{ path: number[], nested: boolean }[]} expected */ seal(path, expected) { /** @type {Undecided[]} */ const settle = []; - if (path.length) { - const parent = this._get(path.slice(0, -1)); - if (parent && !parent.sealed) { - const from = parent.decided; - parent.decided = Math.max(from, path[path.length - 1] + 1); - for (let i = from; i < parent.decided; i++) settle.push({ at: parent, i }); - } - } const taken = expected.map((e) => this._takePending([...path, ...e.path])); const scope = this._get(path, expected.length > 0); if (scope) scope.sealed = true; expected.forEach((e, j) => { const t = taken[j]; if (t) settle.push(t); - const n = this._get([...path, ...e.path], true); - if (!n) { - t?.pending?.chan.cancel(DISCARDED); - return; - } + const n = /** @type {Node} */ (this._get([...path, ...e.path], true)); n.sealed = !e.nested; if (n.chan) return; n.chan = t?.pending?.chan ?? new Chan(); + n.closed = t?.pending?.closed ?? false; n.chan.capacity = CAPACITY; if (this.inboundClosed) n.chan.close(this.inboundError); }); @@ -363,8 +360,7 @@ export class Mux { /** * Declare that the sub-stream at `path` has no more data: its stream ended, - * or its future's value was decoded. Once the sub-streams nested in it - * finish too, it is forgotten, and any further data for it fails the + * or its future's value was decoded. Any further data for it fails the * invocation, as does any data of it left unread in `reader`. * @param {number[]} path * @param {AsyncReader} [reader] the reader it was read over @@ -379,68 +375,23 @@ export class Mux { n.done = true; n.chan?.close(); for (const i of [...(n.pending?.keys() ?? [])]) this._settle({ at: n, i }); - this._prune(path); } /** - * Forget the node at `path` and its ancestors as long as they are finished, - * without children, and not discarded. + * Discard the sub-stream at `path`: drop its buffered and future data, and + * fail its reader with {@link DISCARDED}. * @param {number[]} path */ - _prune(path) { - const nodes = [this.root]; - for (const i of path) { - const child = nodes[nodes.length - 1].children.get(i); - if (!child) return; - nodes.push(child); - } - for (let d = path.length; d > 0; d--) { - const n = nodes[d]; - if (n.children.size || n.discardFrom !== Infinity || (n.chan && !n.done)) return; - const parent = nodes[d - 1]; - const i = path[d - 1]; - parent.children.delete(i); - if (i >= parent.decided) (parent.pruned ??= new Set()).add(i); - } - } - - /** - * Discard the sub-stream at `path` and those nested in its children from - * index `from` on: drop their buffered and future data, and fail their - * readers with {@link DISCARDED}. - * @param {number[]} path - * @param {number} [from] - */ - discard(path, from = 0) { + discard(path) { const n = this._get(path); - if (!n || (n.done && ![...n.children.keys()].some((i) => i >= from))) return; - n.discardFrom = Math.min(n.discardFrom, from); + if (!n || n.done) return; + n.discarded = true; n.chan?.cancel(DISCARDED); - for (const [i, group] of n.pending ?? []) { - if (i < from) continue; - for (const { chan } of group.values()) chan.cancel(DISCARDED); - n.pending?.delete(i); - } - for (const [i, child] of n.children) { - if (i < from) continue; - eachChan(child, (chan) => chan.cancel(DISCARDED)); - n.children.delete(i); - } - } - - /** - * Call `f` once the sub-stream at `path` is discarded. - * @param {number[]} path @param {() => void} f - */ - ondiscard(path, f) { - const chan = this._get(path)?.chan; - if (chan) chan.onCancel(f); - else f(); } /** * Look up the paths buffered in the group at `at` again: drop or reject - * those decided since, and keep the rest where their lookup stops now. + * those declared since, and keep the rest where their lookup stops now. * @param {Undecided} at */ _settle(at) { @@ -455,8 +406,8 @@ export class Mux { this.abort(err); return; } - if (found === null || found instanceof Chan) pending.chan.cancel(DISCARDED); - else attach(found, pending); + if ("at" in found) attach(found, pending); + else pending.chan.cancel(DISCARDED); } } @@ -469,17 +420,14 @@ export class Mux { } /** - * The reader for the expected sub-stream at `path` (the root is `[]`). The - * reader of a discarded sub-stream fails with {@link DISCARDED}. + * The reader for the expected sub-stream at `path` (the root is `[]`). * @param {number[]} path * @returns {AsyncReader} */ incoming(path) { - const n = this._get(path); - if (n !== null && !n?.chan) throw new Error(`no sub-stream is expected at [${path}]`); - const chan = n?.chan; + const chan = this._get(path)?.chan; + if (!chan) throw new Error(`no sub-stream is expected at [${path}]`); return new AsyncReader(async () => { - if (!chan) throw DISCARDED; const { value, done } = await chan.next(); return done ? null : value; }); diff --git a/js/src/session.js b/js/src/session.js index 4d56fa395..e8749069f 100644 --- a/js/src/session.js +++ b/js/src/session.js @@ -10,7 +10,7 @@ // writer task per sub-stream. import { Writer, atomic, readVarU32 } from "./io.js"; -import { DISCARDED, Mux, PROTOCOL } from "./mux.js"; +import { DISCARDED, Mux, PROTOCOL, unexpectedError } from "./mux.js"; import { decodeSteps, encodeValue, hasAsync, readChunk, toBytes } from "./value.js"; /** @typedef {import("./index.d.ts").Type} Type */ @@ -63,22 +63,20 @@ function writeDeferred(out, entry) { : writeStream(out, entry.ty, entry.value); } -/** @param {Outgoing} out @param {Type} ty @param {any} value */ +/** @param {Outgoing} out @param {Type | null} ty @param {any} value */ async function writeFuture(out, ty, value) { const v = await resolveFuture(value); const w = new Writer(); /** @type {any[]} */ const sink = []; - encodeValue(w, ty, v, sink, []); - await out.write(w.finish()); + if (ty) encodeValue(w, ty, v, sink, []); + const bytes = w.finish(); + await (bytes.length > 0 ? out.write(bytes) : out.close()); await Promise.all(sink.map((e) => writeDeferred(out.index(e.path), e))); } /** @param {Outgoing} out @param {Type} elem @param {any} src */ async function writeStream(out, elem, src) { - let base = 0; - /** @type {Promise[]} */ - const tasks = []; for await (const chunk of streamChunks(src, elem)) { const w = new Writer(); if (elem.kind === "u8") { @@ -87,29 +85,21 @@ async function writeStream(out, elem, src) { if (bytes.length === 0) continue; w.varU(bytes.length); w.bytes(bytes); - base += bytes.length; } else { if (!Array.isArray(chunk)) throw new TypeError("expected an array for a `stream` chunk"); if (chunk.length === 0) continue; w.varU(chunk.length); - for (let k = 0; k < chunk.length; k++) { - /** @type {any[]} */ - const sink = []; - encodeValue(w, elem, chunk[k], sink, []); - for (const e of sink) { - const task = writeDeferred(out.index([base + k, ...e.path]), e); - task.catch(() => {}); - tasks.push(task); - } - } - base += chunk.length; + /** @type {any[]} */ + const sink = []; + for (const v of chunk) encodeValue(w, elem, v, sink, []); + if (sink.length) + throw new Error("pending values in `list` or `stream` elements are not supported"); } await out.write(w.finish()); } const end = new Writer(); end.varU(0); // a stream is terminated by an empty chunk await out.write(end.finish()); - await Promise.all(tasks); } // ----- decoding side -------------------------------------------------------- @@ -167,7 +157,7 @@ async function receiveTuple(mux, root, types) { function expect(mux, path, sink) { mux.seal( path, - sink.flatMap((e) => (e.kind === "inline" ? [] : [{ path: e.path, nested: hasAsync(e.ty) }])), + sink.map((e) => ({ path: e.path, nested: e.kind === "future" && !!e.ty && hasAsync(e.ty) })), ); } @@ -191,14 +181,6 @@ function readNested(mux, path, sink, ar) { */ async function readDeferred(mux, path, entry, ar) { switch (entry.kind) { - case "inline": { - // Discard the sub-streams nested in the chunk unless the consumer received it. - const queue = entry.queue; - queue.onCancel(() => { - if (queue.taken === 0) mux.discard(path); - }); - return; - } case "future": return readFuture(mux, path, entry, ar ?? mux.incoming(path)); case "stream": @@ -213,16 +195,17 @@ async function readDeferred(mux, path, entry, ar) { async function readFuture(mux, path, entry, ar) { let decoded; try { - decoded = await decodeOne(ar, entry.ty); + if (entry.ty) decoded = await decodeOne(ar, entry.ty); + else if (await ar.fill(1)) throw unexpectedError(path); + else decoded = { value: undefined, sink: [] }; } catch (err) { - entry.reject(err); if (err === DISCARDED) return; throw err; } const { value, sink } = decoded; expect(mux, path, sink); // A value nested at the future's own path continues reading it. - if (!sink.some((e) => e.kind !== "inline" && e.path.length === 0)) mux.finish(path, ar); + if (!sink.some((e) => e.path.length === 0)) mux.finish(path, ar); const nested = readNested(mux, path, sink, ar); entry.resolve(value); await nested; @@ -230,70 +213,28 @@ async function readFuture(mux, path, entry, ar) { /** * Receive a stream's chunks over `ar` into `queue`, waiting while it is full. - * If the consumer cancels the queue, the rest of the stream is discarded, - * along with the sub-streams nested in the elements it did not receive. + * If the consumer cancels the queue, the rest of the stream is discarded. * @param {Mux} mux @param {number[]} path @param {Type} elem * @param {import("./index.d.ts").Chan} queue @param {AsyncReader} ar */ async function readStream(mux, path, elem, queue, ar) { - const nested = hasAsync(elem); - /** @type {number[]} the index of the first element of each chunk not yet taken */ - const starts = []; - let taken = 0; - let base = 0; - queue.onCancel(() => mux.discard(path, starts[queue.taken - taken] ?? base)); - // Nobody receives the rest of a discarded stream. - mux.ondiscard(path, () => queue.cancel()); - /** @type {Promise[]} */ - const tasks = []; + queue.onCancel(() => mux.discard(path)); try { for (;;) { - // The peer may close the connection instead of sending the empty terminator. - if (!(await ar.fill(1))) { - mux.finish(path); - break; - } const len = await ar.decode((r) => atomic(r, readVarU32)); if (len === 0) { mux.finish(path, ar); break; } - if (nested) { - starts.splice(0, queue.taken - taken); - taken = queue.taken; - starts.push(base); - } - let chunk; - if (elem.kind === "u8") { - chunk = await ar.take(len); - } else if (!nested) { - chunk = await ar.decode((r) => readChunk(r, elem, len)); - } else { - chunk = []; - for (let k = 0; k < len; k++) { - const { value, sink } = await decodeOne(ar, elem); - chunk.push(value); - const at = [...path, base + k]; - expect(mux, at, sink); - if (!sink.length) continue; - const task = readNested(mux, at, sink); - task.catch(() => {}); - tasks.push(task); - } - } - base += len; + const chunk = + elem.kind === "u8" ? await ar.take(len) : await ar.decode((r) => readChunk(r, elem, len)); if (!(await queue.send(chunk))) break; } } catch (err) { - if (!queue.cancelled && err !== DISCARDED) { - // Surface a transport/decode failure to whoever is consuming the stream, - // rather than leaving the consumer hanging on a queue that never closes. - queue.close(err); - throw err; - } + if (!queue.cancelled && err !== DISCARDED) throw err; + } finally { + queue.close(); } - queue.close(); - await Promise.all(tasks); } // ----- public API ----------------------------------------------------------- diff --git a/js/src/types.js b/js/src/types.js index 61ee461ec..dcd5a2309 100644 --- a/js/src/types.js +++ b/js/src/types.js @@ -40,10 +40,11 @@ export const option = (some) => ({ kind: "option", some }); /** * `future`. Carries a single asynchronous value, multiplexed over the * invocation's framed stream. Requires the async session API (`invoke` / - * `accept`); the synchronous `encode`/`decode` cannot transmit it. - * @param {Type} some + * `accept`); the synchronous `encode`/`decode` cannot transmit it. Pass no + * type for a bare `future` without a payload. + * @param {Type | null} [some] */ -export const future = (some) => ({ kind: "future", some }); +export const future = (some = null) => ({ kind: "future", some }); /** * `stream`. Carries an asynchronous sequence of `elem` chunks, diff --git a/js/src/value.js b/js/src/value.js index 384ead3cd..958554c23 100644 --- a/js/src/value.js +++ b/js/src/value.js @@ -135,19 +135,22 @@ const COMPOSITE = new Set([ /** * Decode a single stream/list chunk of `n` elements of type `elem` from `r`. * `stream` chunks are returned as a `Uint8Array`; all others as an array. + * Elements carrying pending `stream`/`future` values are not supported. * @param {Reader} r * @param {Type} elem * @param {number} n - * @param {import("./index.d.ts").DecodeDeferred[]} [sink] - * @param {number[]} [path] * @returns {Generator} */ -export function* readChunk(r, elem, n, sink, path) { +export function* readChunk(r, elem, n) { if (elem.kind === "u8") return yield* atomic(r, (r) => r.take(n).slice()); const out = []; if (COMPOSITE.has(elem.kind)) { + /** @type {import("./index.d.ts").DecodeDeferred[]} */ + const sink = []; for (let i = 0; i < n; i++) { - out.push(yield* decodeSteps(r, elem, sink, childPath(sink, path, i))); + out.push(yield* decodeSteps(r, elem, sink, [])); + if (sink.length) + throw new Error("pending values in `list` or `stream` elements are not supported"); } return out; } @@ -178,7 +181,7 @@ export function* readChunk(r, elem, n, sink, path) { * @param {Writer} w * @param {Type} ty * @param {any} v - * @param {{ path: number[], kind: "stream" | "future", ty: Type, value: any }[]} [sink] + * @param {{ path: number[], kind: "stream" | "future", ty: Type | null, value: any }[]} [sink] * @param {number[]} [path] relative index path to `v` (only tracked with `sink`) */ export function encodeValue(w, ty, v, sink, path) { @@ -351,8 +354,8 @@ export function encodeValue(w, ty, v, sink, path) { * iterable of chunks for a `stream` — and a descriptor recording the relative * `path` and the placeholder's resolver/queue is appended to `sink`. Without a * `sink`, encountering one throws. A `stream` sent inline as a single ready - * chunk is decoded either way, and recorded in `sink` as `inline` if its - * elements carry pending values. + * chunk is decoded either way. Pending values nested in `list` or `stream` + * elements are not supported. * * @param {Reader} r * @param {Type} ty @@ -378,7 +381,7 @@ export function decodeValue(r, ty, sink, path) { export function* decodeSteps(r, ty, sink, path) { switch (ty.kind) { case "list": - return yield* readChunk(r, ty.elem, yield* atomic(r, readVarU32), sink, path); + return yield* readChunk(r, ty.elem, yield* atomic(r, readVarU32)); case "tuple": { const out = []; for (let i = 0; i < ty.elems.length; i++) { @@ -417,32 +420,23 @@ export function* decodeSteps(r, ty, sink, path) { } case "future": { if (yield* atomic(r, optionStatus)) { - return Promise.resolve(yield* decodeSteps(r, ty.some, sink, path)); + return Promise.resolve(ty.some ? yield* decodeSteps(r, ty.some, sink, path) : undefined); } if (!sink) throw new Error(asyncRequired(ty.kind)); /** @type {(value: any) => void} */ let resolve = () => {}; - /** @type {(err: unknown) => void} */ - let reject = () => {}; - const promise = new Promise((res, rej) => { - resolve = res; - reject = rej; - }); - // The failure also surfaces through `done`, so an ignored future must not be unhandled. - promise.catch(() => {}); - sink.push({ path: path ?? [], kind: "future", ty: ty.some, resolve, reject }); + const promise = new Promise((res) => (resolve = res)); + sink.push({ path: path ?? [], kind: "future", ty: ty.some, resolve }); return promise; } case "stream": { if (yield* atomic(r, optionStatus)) { // A stream sent inline as a single ready chunk. const n = yield* atomic(r, readVarU32); - const before = sink?.length ?? 0; - const chunk = yield* readChunk(r, ty.elem, n, sink, path); + const chunk = yield* readChunk(r, ty.elem, n); const q = new Chan(); if (n > 0) q.push(chunk); q.close(); - if (sink && sink.length > before) sink.push({ path: path ?? [], kind: "inline", queue: q }); return q; } if (!sink) throw new Error(asyncRequired(ty.kind)); diff --git a/js/test/session.test.js b/js/test/session.test.js index 80734ffae..8ccdab0f2 100644 --- a/js/test/session.test.js +++ b/js/test/session.test.js @@ -128,26 +128,6 @@ test("aborting after the peer closed fails the sub-streams declared later", asyn await assert.rejects(mux.incoming([0]).u8(), /boom/); }); -test("finished sub-streams are forgotten", { timeout: 1000 }, async () => { - const [client, server] = duplexPair(); - const ty = t.stream(t.option(t.future(t.u8))); - const elems = Array.from({ length: 1000 }, (_, i) => (i % 2 ? i & 0xff : undefined)); - const serving = (async () => { - const inv = await accept(server); - const { params, done } = await inv.receiveParams([ty]); - const futures = []; - for await (const chunk of params[0]) futures.push(...chunk); - const values = await Promise.all(futures); - await done; - assert.equal(inv.mux.root.children.size, 0); - await inv.sendResults([], []); - return values; - })(); - const { done } = await invoke(client, STREAMS, "get", [ty], [[elems]], []); - await done; - assert.deepEqual(await serving, elems); -}); - test("a failed invocation leaves no unhandled rejection", async () => { const transport = { read: () => Promise.reject(new Error("connection refused")), @@ -237,47 +217,6 @@ test("aborting a web stream transport cancels and aborts both sides", async () = ]); }); -test("a failed stream> parameter leaves no unhandled rejection", async () => { - async function* chunks() { - yield [ - () => { - throw new Error("future failed"); - }, - ]; - throw new Error("stream failed"); - } - const transport = { - read: () => new Promise((resolve) => setTimeout(() => resolve(null), 10)), - write: () => {}, - }; - await assert.rejects( - invoke(transport, STREAMS, "echo", [t.stream(t.future(t.string))], [chunks()], [t.u8]), - /stream failed/, - ); -}); - -test( - "a transport failure rejects pending stream> parameters", - { timeout: 1000 }, - async () => { - const frames = []; - const peer = new Mux({ read: async () => null, write: (bytes) => void frames.push(bytes) }); - await peer.writeHeader(STREAMS, "echo"); - await peer.outgoing([]).write(Uint8Array.of(0)); - await peer.outgoing([0]).write(Uint8Array.of(1, 0)); - const transport = { - read: async () => frames.shift() ?? Promise.reject(new Error("connection reset")), - write: () => {}, - }; - const inv = await accept(transport); - const { params } = await inv.receiveParams([t.stream(t.future(t.string))]); - const chunks = params[0][Symbol.asyncIterator](); - const { value: chunk } = await chunks.next(); - await assert.rejects(chunk[0], /connection reset/); - await assert.rejects(chunks.next(), /connection reset/); - }, -); - test("frame data is routed as it arrives", async () => { const bytes = [0x01, 0x00, 0x03, 0x0a, 0x14, 0x1e, 0x01, 0x00, 0x02, 0x28]; const transport = { @@ -335,15 +274,17 @@ test("invalid arguments fail before anything is written and abort the transport" assert.equal(aborted.length, 2); }); -test("stream chunks must be bytes", async () => { - const transport = { - read: () => new Promise((resolve) => setTimeout(() => resolve(null), 10)), - write: () => {}, - }; - await assert.rejects( - invoke(transport, STREAMS, "echo", [t.stream(t.u8)], [[1, 2, 3]], [t.u8]), - /expected `Uint8Array`/, - ); +test("invalid stream chunks fail the invocation", async () => { + for (const [ty, chunks, error] of [ + [t.stream(t.u8), [1, 2, 3], /expected `Uint8Array`/], + [t.stream(t.future(t.string)), [["foo"]], /not supported/], + ]) { + const transport = { + read: () => new Promise((resolve) => setTimeout(() => resolve(null), 10)), + write: () => {}, + }; + await assert.rejects(invoke(transport, STREAMS, "echo", [ty], [chunks], [t.u8]), error); + } }); for (const [name, values, error] of [ @@ -409,77 +350,6 @@ test("a cancelled stream result is discarded", { timeout: 1000 }, async () => { assert.deepEqual(await results[0].next(), { value: undefined, done: true }); }); -test( - "cancelling a stream keeps the streams nested in the chunks it delivered", - { timeout: 1000 }, - async () => { - const ty = t.tuple(t.stream(t.stream(t.u8)), t.stream(t.u8)); - const bytes = function* () { - for (let i = 0; i < 1000; i++) yield Uint8Array.of(i & 0xff); - }; - const { result, done, serving } = await echo(ty, [[[bytes(), bytes()], [bytes()]], bytes()]); - const [nested, sibling] = result; - let kept = []; - for await (const chunk of nested) { - kept = chunk; - break; - } - assert.equal(kept.length, 2); - kept[1].cancel(); - // Undiscarded nested data would stop the transport from being read. - const [received, siblings] = await Promise.all([collect(kept[0]), collect(sibling)]); - assert.equal(received.length, 1000); - assert.equal(siblings.length, 1000); - await Promise.all([done, serving]); - }, -); - -test( - "cancelling a received stream discards the streams nested in it", - { timeout: 1000 }, - async () => { - const frames = []; - const peer = new Mux({ read: async () => null, write: (bytes) => void frames.push(bytes) }); - await peer.outgoing([]).write(Uint8Array.of(0, 0)); - await peer.outgoing([0]).write(Uint8Array.of(1, 0)); - await peer.outgoing([0]).write(Uint8Array.of(0)); - for (let i = 0; i < 1000; i++) await peer.outgoing([0, 0]).write(Uint8Array.of(1, i & 0xff)); - await peer.outgoing([0, 0]).write(Uint8Array.of(0)); - await peer.outgoing([1]).write(Uint8Array.of(1, 7)); - await peer.outgoing([1]).write(Uint8Array.of(0)); - const transport = { read: async () => frames.shift() ?? null, write: () => {} }; - const ty = [t.stream(t.stream(t.u8)), t.stream(t.u8)]; - const { results, done } = await invoke(transport, STREAMS, "get", [], [], ty); - while (!results[0].closed) await tick(); - let hooked = false; - results[0].onCancel(() => (hooked = true)); - results[0].cancel(); - assert.ok(hooked); - assert.deepEqual(await collect(results[1]), [7]); - await done; - }, -); - -test( - "cancelling an inline stream discards the streams nested in it", - { timeout: 1000 }, - async () => { - const frames = []; - const peer = new Mux({ read: async () => null, write: (bytes) => void frames.push(bytes) }); - await peer.outgoing([]).write(Uint8Array.of(1, 1, 0, 0)); - for (let i = 0; i < 1000; i++) await peer.outgoing([0, 0]).write(Uint8Array.of(1, i & 0xff)); - await peer.outgoing([0, 0]).write(Uint8Array.of(0)); - await peer.outgoing([1]).write(Uint8Array.of(1, 7)); - await peer.outgoing([1]).write(Uint8Array.of(0)); - const transport = { read: async () => frames.shift() ?? null, write: () => {} }; - const ty = [t.stream(t.stream(t.u8)), t.stream(t.u8)]; - const { results, done } = await invoke(transport, STREAMS, "get", [], [], ty); - results[0].cancel(); - assert.deepEqual(await collect(results[1]), [7]); - await done; - }, -); - test("data arriving before its sub-stream is decoded is kept", { timeout: 1000 }, async () => { const frames = []; const peer = new Mux({ read: async () => null, write: (bytes) => void frames.push(bytes) }); @@ -511,35 +381,6 @@ test( }, ); -test("element data arriving ahead of a large chunk is kept", { timeout: 5000 }, async () => { - const n = 4000; - const frames = []; - const peer = new Mux({ read: async () => null, write: (bytes) => void frames.push(bytes) }); - await peer.outgoing([]).write(Uint8Array.of(0)); - for (let i = 0; i < n; i++) await peer.outgoing([0, i]).write(Uint8Array.of(i & 0xff)); - const w = new Writer(); - w.varU(n); - for (let i = 0; i < n; i++) w.u8(0); - await peer.outgoing([0]).write(w.finish()); - await peer.outgoing([0]).write(Uint8Array.of(0)); - const transport = { read: async () => frames.shift() ?? null, write: () => {} }; - const { results, done } = await invoke( - transport, - STREAMS, - "get", - [], - [], - [t.stream(t.future(t.u8))], - ); - const out = []; - for await (const chunk of results[0]) out.push(...(await Promise.all(chunk))); - assert.deepEqual( - out, - Array.from({ length: n }, (_, i) => i & 0xff), - ); - await done; -}); - test( "more data than a sub-stream buffers arriving ahead of its declaration is kept", { timeout: 1000 }, @@ -550,66 +391,36 @@ test( await peer.outgoing([]).write(Uint8Array.of(0)); for (let i = 0; i < n; i++) await peer.outgoing([0, 0]).write(Uint8Array.of(1, i & 0xff)); await peer.outgoing([0, 0]).write(Uint8Array.of(0)); - await peer.outgoing([0]).write(Uint8Array.of(1, 0)); await peer.outgoing([0]).write(Uint8Array.of(0)); const transport = { read: async () => frames.shift() ?? null, write: () => {} }; - const { results, done } = await invoke( - transport, - STREAMS, - "get", - [], - [], - [t.stream(t.stream(t.u8))], - ); - const inner = []; - for await (const chunk of results[0]) inner.push(...chunk); - assert.equal(inner.length, 1); + const ty = [t.future(t.tuple(t.stream(t.u8)))]; + const { results, done } = await invoke(transport, STREAMS, "get", [], [], ty); + const [inner] = await results[0]; assert.deepEqual( - await collect(inner[0]), + await collect(inner), Array.from({ length: n }, (_, i) => i & 0xff), ); await done; }, ); - -test("data for a sub-stream the reply does not carry fails it", { timeout: 1000 }, async () => { - const frames = []; - const peer = new Mux({ read: async () => null, write: (bytes) => void frames.push(bytes) }); - await peer.outgoing([]).write(Uint8Array.of(0, 0)); - for (let i = 0; i < 100; i++) await peer.outgoing([0]).write(Uint8Array.of(1, i)); - await peer.outgoing([1]).write(Uint8Array.of(1, 7)); - await peer.outgoing([1]).write(Uint8Array.of(0)); - let aborted; - const transport = { - read: async () => frames.shift() ?? null, - write: () => {}, - abort: (reason) => void (aborted = reason), - }; - const ty = [t.option(t.stream(t.u8)), t.stream(t.u8)]; - const { results, done } = await invoke(transport, STREAMS, "get", [], [], ty); - assert.equal(results[0], undefined); - await assert.rejects(collect(results[1]), /unexpected data for sub-stream \[0\]/); - await assert.rejects(done, /unexpected data for sub-stream \[0\]/); - assert.match(aborted.message, /unexpected data/); -}); - -test( - "an empty frame for a sub-stream the reply does not carry fails it", - { timeout: 1000 }, - async () => { +test("an invocation sends no frames for empty data", async () => { + for (const [types, args, expected] of [ + [[], [], "\x00\x03foo\x03bar"], + [[t.future()], [Promise.resolve()], "\x00\x03foo\x03bar\x00\x01\x00\x01\x00\x00"], + ]) { const frames = []; - const peer = new Mux({ read: async () => null, write: (bytes) => void frames.push(bytes) }); - await peer.outgoing([]).write(Uint8Array.of(0, 0)); - await peer.outgoing([0]).write(new Uint8Array(0)); - await peer.outgoing([1]).write(Uint8Array.of(1, 7)); - await peer.outgoing([1]).write(Uint8Array.of(0)); - const transport = { read: async () => frames.shift() ?? null, write: () => {} }; - const ty = [t.option(t.stream(t.u8)), t.stream(t.u8)]; - const { results, done } = await invoke(transport, STREAMS, "get", [], [], ty); - await assert.rejects(collect(results[1]), /unexpected data for sub-stream \[0\]/); - await assert.rejects(done, /unexpected data for sub-stream \[0\]/); - }, -); + let closed = false; + const transport = { + read: async () => null, + write: (bytes) => void frames.push(bytes), + closeWrite: () => void (closed = true), + }; + const { done } = await invoke(transport, "foo", "bar", types, args, []); + await done; + assert.ok(closed); + assert.deepEqual(Buffer.concat(frames), Buffer.from(expected, "latin1")); + } +}); /** A transport fed by the frames a peer writes, until `inbound` is closed. */ function feed() { @@ -626,6 +437,24 @@ function feed() { return { inbound, peer, transport, aborted }; } +test("a future without a payload resolves at the end of the transport", async () => { + for (const data of [false, true]) { + const { inbound, peer, transport } = feed(); + await peer.outgoing([]).write(Uint8Array.of(0)); + if (data) await peer.outgoing([0]).write(Uint8Array.of(0)); + inbound.close(); + const { results, done } = await invoke(transport, STREAMS, "get", [], [], [t.future()]); + if (data) { + await assert.rejects(done, /unexpected data for sub-stream \[0\]/); + const pending = Symbol("pending"); + assert.equal(await Promise.race([results[0], pending]), pending); + } else { + assert.equal(await results[0], undefined); + await done; + } + } +}); + test("an empty frame for a finished sub-stream is accepted", { timeout: 1000 }, async () => { const { inbound, peer, transport, aborted } = feed(); await peer.outgoing([]).write(Uint8Array.of(0)); @@ -634,7 +463,7 @@ test("an empty frame for a finished sub-stream is accepted", { timeout: 1000 }, const { results, done } = await invoke(transport, STREAMS, "get", [], [], [t.stream(t.u8)]); assert.deepEqual(await collect(results[0]), [7]); await done; - await peer.outgoing([0]).write(new Uint8Array(0)); + await peer.outgoing([0]).close(); inbound.close(); const outcome = await Promise.race([ aborted, @@ -646,58 +475,22 @@ test("an empty frame for a finished sub-stream is accepted", { timeout: 1000 }, }); test( - "an empty frame for a sub-stream the reply does not carry fails it after a sibling's", - { timeout: 1000 }, - async () => { - const { inbound, peer, transport, aborted } = feed(); - await peer.outgoing([]).write(Uint8Array.of(0, 0)); - await peer.outgoing([1]).write(Uint8Array.of(7)); - const ty = [t.option(t.stream(t.u8)), t.future(t.u8)]; - const { results } = await invoke(transport, STREAMS, "get", [], [], ty); - assert.equal(await results[1], 7); - await peer.outgoing([0]).write(new Uint8Array(0)); - inbound.close(); - assert.match(String(await aborted), /unexpected data for sub-stream \[0\]/); - }, -); - -test( - "an empty frame arriving before its sub-stream finishes is accepted", + "an empty frame arriving before its sub-stream is declared closes it", { timeout: 1000 }, async () => { const frames = []; const peer = new Mux({ read: async () => null, write: (bytes) => void frames.push(bytes) }); await peer.outgoing([]).write(Uint8Array.of(0)); - await peer.outgoing([0, 0]).write(new Uint8Array(0)); - await peer.outgoing([0]).write(Uint8Array.of(1, 0)); + await peer.outgoing([0, 0]).close(); await peer.outgoing([0]).write(Uint8Array.of(0)); const transport = { read: async () => frames.shift() ?? null, write: () => {} }; - const ty = [t.stream(t.option(t.future(t.u8)))]; + const ty = [t.future(t.tuple(t.future()))]; const { results, done } = await invoke(transport, STREAMS, "get", [], [], ty); - assert.deepEqual(await collect(results[0]), [undefined]); + const [inner] = await results[0]; + assert.equal(await inner, undefined); await done; }, ); - -test( - "data for a stream after its end fails it, even once cancelled", - { timeout: 1000 }, - async () => { - const { inbound, peer, transport, aborted } = feed(); - await peer.outgoing([]).write(Uint8Array.of(0)); - await peer.outgoing([0]).write(Uint8Array.of(1, 0)); - await peer.outgoing([0]).write(Uint8Array.of(0)); - const ty = [t.stream(t.future(t.u8))]; - const { results } = await invoke(transport, STREAMS, "get", [], [], ty); - while (!results[0].closed) await tick(); - results[0].cancel(); - await peer.outgoing([0, 0]).write(Uint8Array.of(5)); - await peer.outgoing([0]).write(Uint8Array.of(1, 0)); - inbound.close(); - assert.match(String(await aborted), /unexpected data for sub-stream \[0\]/); - }, -); - test("frame data beyond the end of a sub-stream fails it", { timeout: 1000 }, async () => { for (const [ty, value, extra, split] of [ [t.future(t.u8), [5], 40, true], @@ -733,17 +526,18 @@ test( const peer = new Mux({ read: async () => null, write: (bytes) => void frames.push(bytes) }); await peer.outgoing([]).write(Uint8Array.of(0)); await peer.outgoing([0, 0]).write(Uint8Array.of(1, 0)); - await peer.outgoing([0]).write(Uint8Array.of(1, 0)); + await peer.outgoing([0]).write(Uint8Array.of(0)); for (let i = 0; i < 1000; i++) await peer.outgoing([0, 0]).write(Uint8Array.of(1, i & 0xff)); let reads = 0; const transport = { read: async () => (reads++, frames.shift() ?? null), write: () => {} }; - const { results } = await invoke(transport, STREAMS, "get", [], [], [t.stream(t.stream(t.u8))]); - await results[0][Symbol.asyncIterator]().next(); + const ty = [t.future(t.tuple(t.stream(t.u8)))]; + const { results } = await invoke(transport, STREAMS, "get", [], [], ty); + const [inner] = await results[0]; + await inner[Symbol.asyncIterator]().next(); await new Promise((resolve) => setTimeout(resolve, 50)); assert.ok(reads < 500, `read ${reads} frames`); }, ); - test("data for the root after the reply fails it", { timeout: 1000 }, async () => { const frames = []; const peer = new Mux({ read: async () => null, write: (bytes) => void frames.push(bytes) }); @@ -753,41 +547,28 @@ test("data for the root after the reply fails it", { timeout: 1000 }, async () = await peer.outgoing([0]).write(Uint8Array.of(0)); const transport = { read: async () => frames.shift() ?? null, write: () => {} }; const { results, done } = await invoke(transport, STREAMS, "get", [], [], [t.stream(t.u8)]); - await assert.rejects(collect(results[0]), /unexpected data for sub-stream \[\]/); + await collect(results[0]); await assert.rejects(done, /unexpected data for sub-stream \[\]/); }); -test("cancelling an inline stream discards the sub-streams nested in it", () => { - const mux = new Mux({ read: () => new Promise(() => {}), write: () => {} }); - mux.seal([], [{ path: [0], nested: true }]); - mux.seal([0], [{ path: [0], nested: false }]); - mux.finish([0]); - const nested = mux.incoming([0, 0]); - mux.discard([0]); - return assert.rejects(nested.u8(), /sub-stream discarded/); -}); - -test("a stream truncated mid-frame fails", { timeout: 1000 }, async () => { - const frames = []; - const peer = new Mux({ read: async () => null, write: (bytes) => void frames.push(bytes) }); - await peer.outgoing([]).write(Uint8Array.of(0)); - frames.push(Uint8Array.of(1, 0, 5, 1, 6)); - const transport = { read: async () => frames.shift() ?? null, write: () => {} }; - const { results, done } = await invoke(transport, STREAMS, "get", [], [], [t.stream(t.u8)]); - await assert.rejects(collect(results[0]), /unexpected end of input/); - await assert.rejects(done, /unexpected end of input/); -}); - -test("a stream truncated mid-length fails", { timeout: 1000 }, async () => { - const frames = []; - const peer = new Mux({ read: async () => null, write: (bytes) => void frames.push(bytes) }); - await peer.outgoing([]).write(Uint8Array.of(0)); - await peer.outgoing([0]).write(Uint8Array.of(1, 5)); - await peer.outgoing([0]).write(Uint8Array.of(0x80)); - const transport = { read: async () => frames.shift() ?? null, write: () => {} }; - const { results, done } = await invoke(transport, STREAMS, "get", [], [], [t.stream(t.u8)]); - await assert.rejects(collect(results[0]), /unexpected end of input/); - await assert.rejects(done, /unexpected end of input/); +test("a truncated stream fails", { timeout: 1000 }, async () => { + for (const send of [ + (peer, frames) => frames.push(Uint8Array.of(1, 0, 5, 1, 6)), + (peer) => peer.outgoing([0]).write(Uint8Array.of(1, 5, 0x80)), + async (peer) => { + await peer.outgoing([0]).write(Uint8Array.of(1, 7)); + await peer.outgoing([0]).close(); + }, + ]) { + const frames = []; + const peer = new Mux({ read: async () => null, write: (bytes) => void frames.push(bytes) }); + await peer.outgoing([]).write(Uint8Array.of(0)); + await send(peer, frames); + const transport = { read: async () => frames.shift() ?? null, write: () => {} }; + const { results, done } = await invoke(transport, STREAMS, "get", [], [], [t.stream(t.u8)]); + await collect(results[0]); + await assert.rejects(done, /unexpected end of input/); + } }); const futures = "wrpc-test:async/futures"; @@ -871,28 +652,79 @@ test("future> resolves to a byte stream", async () => { await Promise.all([done, serving]); }); -test("stream> yields chunks of promises", async () => { - const ty = t.stream(t.future(t.string)); - const { result, done, serving } = await echo(ty, [["a", "b"], ["c"]]); - const out = []; - for await (const chunk of result) out.push(...(await Promise.all(chunk))); - assert.deepEqual(out, ["a", "b", "c"]); - await Promise.all([done, serving]); +for (const [kind, ty, chunk] of [ + ["list", t.list(t.future(t.u8)), Uint8Array.of(1, 0)], + ["stream", t.stream(t.future(t.u8)), Uint8Array.of(0)], +]) { + test(`pending values in ${kind} elements are not supported`, { timeout: 1000 }, async () => { + const { inbound, peer, transport, aborted } = feed(); + await peer.outgoing([]).write(chunk); + await peer.outgoing([0]).write(Uint8Array.of(1, 0)); + inbound.close(); + const invoked = invoke(transport, STREAMS, "get", [], [], [ty]); + if (kind === "stream") { + const { results, done } = await invoked; + await collect(results[0]); + await assert.rejects(done, /not supported/); + } else await assert.rejects(invoked, /not supported/); + assert.match(String(await aborted), /not supported/); + }); +} + +test("ready values in stream elements are received", { timeout: 1000 }, async () => { + const { inbound, peer, transport } = feed(); + await peer.outgoing([]).write(Uint8Array.of(0)); + await peer.outgoing([0]).write(Uint8Array.of(2, 1, 5, 1, 6)); + await peer.outgoing([0]).write(Uint8Array.of(0)); + inbound.close(); + const ty = [t.stream(t.future(t.u8))]; + const { results, done } = await invoke(transport, STREAMS, "get", [], [], ty); + assert.deepEqual(await Promise.all(await collect(results[0])), [5, 6]); + await done; }); -test("stream> yields chunks of byte streams", async () => { - const ty = t.stream(t.stream(t.u8)); - // one outer chunk carrying two inner byte streams - const { result, done, serving } = await echo(ty, [ - [new Uint8Array([1, 2]), new Uint8Array([3, 4])], - ]); - const out = []; - for await (const chunk of result) { - for (const inner of chunk) out.push(await collect(inner)); +for (const [name, frame] of [ + ["data", (out) => out.write(Uint8Array.of(7))], + ["an empty frame", (out) => out.close()], +]) { + test( + `${name} for a sub-stream the reply does not carry fails it`, + { timeout: 1000 }, + async () => { + const { inbound, peer, transport, aborted } = feed(); + await peer.outgoing([]).write(Uint8Array.of(0, 0)); + await frame(peer.outgoing([0])); + await peer.outgoing([1]).write(Uint8Array.of(1, 7, 0)); + inbound.close(); + const ty = [t.option(t.stream(t.u8)), t.stream(t.u8)]; + const { results, done } = await invoke(transport, STREAMS, "get", [], [], ty); + await collect(results[1]); + await assert.rejects(done, /unexpected data for sub-stream \[0\]/); + assert.match(String(await aborted), /unexpected data/); + }, + ); + + for (const [declared, path, ty, init] of [ + [true, [0], t.stream(t.u8), [[[0], Uint8Array.of(1, 7, 0)]]], + [false, [0, 0], t.future(t.tuple(t.future(t.u8))), []], + ]) { + test( + `${name} for a closed sub-stream ${declared ? "" : "not yet declared "}fails it`, + { timeout: 1000 }, + async () => { + const { inbound, peer, transport, aborted } = feed(); + await peer.outgoing([]).write(Uint8Array.of(0)); + for (const [p, bytes] of init) await peer.outgoing(p).write(bytes); + await peer.outgoing(path).close(); + await frame(peer.outgoing(path)); + if (!declared) await peer.outgoing([0]).write(Uint8Array.of(0)); + inbound.close(); + await invoke(transport, STREAMS, "get", [], [], [ty]); + assert.match( + String(await aborted), + new RegExp(`unexpected data for sub-stream \\[${path}\\]`), + ); + }, + ); } - assert.deepEqual(out, [ - [1, 2], - [3, 4], - ]); - await Promise.all([done, serving]); -}); +} diff --git a/js/test/value.test.js b/js/test/value.test.js index 68c6b2453..8228f9ecb 100644 --- a/js/test/value.test.js +++ b/js/test/value.test.js @@ -264,7 +264,7 @@ test("stream and future cannot be encoded or decoded synchronously", () => { assert.throws(() => decode(t.future(t.string), new Uint8Array([0])), /async session API/); }); -test("ready streams are decoded synchronously", async () => { +test("ready streams and futures are decoded synchronously", async () => { const collect = async (chunks) => { const out = []; for await (const chunk of chunks) out.push(chunk); @@ -277,6 +277,8 @@ test("ready streams are decoded synchronously", async () => { ["a"], ]); assert.deepEqual(await collect(decode(t.stream(t.u8), Uint8Array.of(1, 0))), []); + assert.equal(await decode(t.future(t.u8), Uint8Array.of(1, 7)), 7); + assert.equal(await decode(t.future(), Uint8Array.of(1)), undefined); }); test("the wasi:keyvalue store types", () => {