feat: add stream.forward canon builtin

Implement parsing, validation, encoding, and printing support for the
`stream.forward` built-in specified alongside `stream.splice` in
WebAssembly/component-model#658:

    (canon stream.forward $streamT (core func $f))

where `$f` has type `(func (param i32 i32))`, taking a readable stream
end and a writable stream end.  Unlike `stream.splice`, which borrows
both ends and reports the outcome back to the caller, `stream.forward`
transfers both ends out of the calling instance and returns immediately
with no result, so it takes neither a `n` parameter nor an `async`
immediate.  Like `stream.splice` it takes no `canonopt`s, since no
elements pass through the caller's linear memory.

The binary encoding uses the currently-unassigned opcode 0x2f, and
validation gates the built-in behind the component model async and
"more async builtins" features since it is not yet part of the
Component Model specification.

Assisted-by: claude:claude-fable-5
Signed-off-by: Roman Volosatovs <rvolosatovs@riseup.net>

Author: rvolosatovs

crates/wasm-encoder/src/component/builder.rs

@@ -562,6 +562,12 @@ impl ComponentBuilder {
         self.core_funcs.add(Some("stream.splice"))
     }
 
+    /// Declares a new `stream.forward` intrinsic.
+    pub fn stream_forward(&mut self, ty: u32) -> u32 {
+        self.canonical_functions().stream_forward(ty);
+        self.core_funcs.add(Some("stream.forward"))
+    }
+
     /// Declares a new `stream.cancel-read` intrinsic.
     pub fn stream_cancel_read(&mut self, ty: u32, async_: bool) -> u32 {
         self.canonical_functions().stream_cancel_read(ty, async_);

crates/wasm-encoder/src/component/canonicals.rs

@@ -308,6 +308,16 @@ impl CanonicalFunctionSection {
         self
     }
 
+    /// Defines a function to forward all remaining elements from the readable
+    /// end of one `stream` to the writable end of another `stream` of the
+    /// specified type, transferring both ends out of the calling instance.
+    pub fn stream_forward(&mut self, ty: u32) -> &mut Self {
+        self.bytes.push(0x2f);
+        ty.encode(&mut self.bytes);
+        self.num_added += 1;
+        self
+    }
+
     /// Defines a function to cancel an in-progress read from a `stream` of the
     /// specified type.
     pub fn stream_cancel_read(&mut self, ty: u32, async_: bool) -> &mut Self {

crates/wasm-encoder/src/reencode/component.rs

@@ -1033,6 +1033,9 @@ pub mod component_utils {
             wasmparser::CanonicalFunction::StreamSplice { ty, async_ } => {
                 section.stream_splice(reencoder.component_type_index(ty), async_);
             }
+            wasmparser::CanonicalFunction::StreamForward { ty } => {
+                section.stream_forward(reencoder.component_type_index(ty));
+            }
             wasmparser::CanonicalFunction::StreamCancelRead { ty, async_ } => {
                 section.stream_cancel_read(ty, async_);
             }

crates/wasmparser/src/readers/component/canonicals.rs

@@ -169,6 +169,17 @@ pub enum CanonicalFunction {
         /// `BLOCKED`.
         async_: bool,
     },
+    /// A function to forward all remaining elements from the readable end of
+    /// one `stream` to the writable end of another `stream` of the same
+    /// specified type, transferring both ends out of the calling instance.
+    ///
+    /// 🚧 This is an experimental builtin sketched in
+    /// <https://github.com/WebAssembly/component-model/issues/658> and not yet
+    /// part of the Component Model specification.
+    StreamForward {
+        /// The `stream` type to expect.
+        ty: u32,
+    },
     /// A function to cancel an in-progress read from a `stream` of the
     /// specified type.
     StreamCancelRead {
@@ -388,6 +399,7 @@ impl<'a> FromReader<'a> for CanonicalFunction {
                 ty: reader.read()?,
                 async_: reader.read()?,
             },
+            0x2f => CanonicalFunction::StreamForward { ty: reader.read()? },
             0x11 => CanonicalFunction::StreamCancelRead {
                 ty: reader.read()?,
                 async_: reader.read()?,

crates/wasmparser/src/validator/component.rs

@@ -1245,6 +1245,7 @@ impl ComponentState {
             CanonicalFunction::StreamSplice { ty, async_ } => {
                 self.stream_splice(ty, async_, types, offset)
             }
+            CanonicalFunction::StreamForward { ty } => self.stream_forward(ty, types, offset),
             CanonicalFunction::StreamCancelRead { ty, async_ } => {
                 self.stream_cancel_read(ty, async_, types, offset)
             }
@@ -1759,6 +1760,28 @@ impl ComponentState {
         Ok(())
     }
 
+    fn stream_forward(&mut self, ty: u32, types: &mut TypeAlloc, offset: u64) -> Result<()> {
+        require_feature::cm_async(
+            self.features,
+            "`stream.forward` requires the component model async feature",
+            offset,
+        )?;
+        require_feature::cm_more_async_builtins(
+            self.features,
+            "`stream.forward` requires the component model more async builtins feature",
+            offset,
+        )?;
+
+        let ty = self.defined_type_at(ty, offset)?;
+        let ComponentDefinedType::Stream { .. } = &types[ty] else {
+            bail!(offset, "`stream.forward` requires a stream type")
+        };
+
+        self.core_funcs
+            .push(types.intern_func_type(FuncType::new([ValType::I32; 2], []), offset));
+        Ok(())
+    }
+
     fn stream_cancel_read(
         &mut self,
         ty: u32,

crates/wasmprinter/src/component.rs

@@ -1057,6 +1057,11 @@ impl Printer<'_, '_> {
                         Ok(())
                     })?;
                 }
+                CanonicalFunction::StreamForward { ty } => {
+                    self.print_intrinsic(state, "canon stream.forward ", &|me, state| {
+                        me.print_idx(&state.component.type_names, ty)
+                    })?;
+                }
                 CanonicalFunction::StreamCancelRead { ty, async_ } => {
                     self.print_intrinsic(state, "canon stream.cancel-read ", &|me, state| {
                         me.print_idx(&state.component.type_names, ty)?;

crates/wast/src/component/binary.rs

@@ -422,6 +422,10 @@ impl<'a> Encoder<'a> {
                     self.core_func_names.push(name);
                     self.funcs.stream_splice((&info.ty).into(), info.async_);
                 }
+                CoreFuncKind::StreamForward(info) => {
+                    self.core_func_names.push(name);
+                    self.funcs.stream_forward((&info.ty).into());
+                }
                 CoreFuncKind::StreamCancelRead(info) => {
                     self.core_func_names.push(name);
                     self.funcs

crates/wast/src/component/func.rs

@@ -66,6 +66,7 @@ pub enum CoreFuncKind<'a> {
     StreamRead(CanonStreamRead<'a>),
     StreamWrite(CanonStreamWrite<'a>),
     StreamSplice(CanonStreamSplice<'a>),
+    StreamForward(CanonStreamForward<'a>),
     StreamCancelRead(CanonStreamCancelRead<'a>),
     StreamCancelWrite(CanonStreamCancelWrite<'a>),
     StreamDropReadable(CanonStreamDropReadable<'a>),
@@ -160,6 +161,8 @@ impl<'a> CoreFuncKind<'a> {
             Ok(CoreFuncKind::StreamWrite(parser.parse()?))
         } else if l.peek::<kw::stream_splice>()? {
             Ok(CoreFuncKind::StreamSplice(parser.parse()?))
+        } else if l.peek::<kw::stream_forward>()? {
+            Ok(CoreFuncKind::StreamForward(parser.parse()?))
         } else if l.peek::<kw::stream_cancel_read>()? {
             Ok(CoreFuncKind::StreamCancelRead(parser.parse()?))
         } else if l.peek::<kw::stream_cancel_write>()? {
@@ -744,6 +747,23 @@ impl<'a> Parse<'a> for CanonStreamSplice<'a> {
     }
 }
 
+/// Information relating to the `stream.forward` intrinsic.
+#[derive(Debug)]
+pub struct CanonStreamForward<'a> {
+    /// The stream type to forward.
+    pub ty: ItemRef<'a, kw::r#type>,
+}
+
+impl<'a> Parse<'a> for CanonStreamForward<'a> {
+    fn parse(parser: Parser<'a>) -> Result<Self> {
+        parser.parse::<kw::stream_forward>()?;
+
+        Ok(Self {
+            ty: parser.parse::<IndexOrRef<'_, _>>()?.0,
+        })
+    }
+}
+
 /// Information relating to the `stream.cancel-read` intrinsic.
 #[derive(Debug)]
 pub struct CanonStreamCancelRead<'a> {

crates/wast/src/component/resolve.rs

@@ -423,6 +423,9 @@ impl<'a> Resolver<'a> {
                 CoreFuncKind::StreamSplice(info) => {
                     self.component_item_ref(&mut info.ty)?;
                 }
+                CoreFuncKind::StreamForward(info) => {
+                    self.component_item_ref(&mut info.ty)?;
+                }
                 CoreFuncKind::StreamCancelRead(info) => {
                     self.component_item_ref(&mut info.ty)?;
                 }

crates/wast/src/lib.rs

@@ -576,6 +576,7 @@ pub mod kw {
     custom_keyword!(stream_read = "stream.read");
     custom_keyword!(stream_write = "stream.write");
     custom_keyword!(stream_splice = "stream.splice");
+    custom_keyword!(stream_forward = "stream.forward");
     custom_keyword!(stream_cancel_read = "stream.cancel-read");
     custom_keyword!(stream_cancel_write = "stream.cancel-write");
     custom_keyword!(stream_drop_readable = "stream.drop-readable");