fix(bindgen): host read on stream, determinism, expand stream tests

crates/js-component-bindgen/src/intrinsics/lift.rs

@@ -903,24 +903,36 @@ impl LiftIntrinsic {
                 let external_stream_class =
                     Intrinsic::AsyncStream(AsyncStreamIntrinsic::ExternalStreamClass).name();
                 let lift_flat_stream_fn = self.name();
+                let global_stream_table_map = AsyncStreamIntrinsic::GlobalStreamTableMap.name();
+
                 output.push_str(&format!(r#"
-                    function {lift_flat_stream_fn}(componentTableIdx, ctx) {{
-                        {debug_log_fn}('[{lift_flat_stream_fn}()] args', {{ componentTableIdx, ctx }});
-                        const {{ memory, useDirectParams, params, componentIdx }} = ctx;
+                    function {lift_flat_stream_fn}(streamTableIdx, ctx) {{
+                        {debug_log_fn}('[{lift_flat_stream_fn}()] args', {{ streamTableIdx, ctx }});
+                        const {{ memory, useDirectParams, params }} = ctx;
 
-                        const streamEndIdx = params[0];
-                        if (!streamEndIdx) {{ throw new Error('missing stream idx'); }}
+                        const {{ table, componentIdx }} = {global_stream_table_map}[streamTableIdx];
+                        if (componentIdx === undefined || !table) {{
+                            throw new Error(`invalid global stream table state for table [${{tableIdx}}]`);
+                        }}
 
                         const cstate = {get_or_create_async_state_fn}(componentIdx);
                         if (!cstate) {{ throw new Error(`missing async state for component [${{componentIdx}}]`); }}
 
-                        const streamEnd = cstate.getStreamEnd({{ tableIdx: componentTableIdx, streamEndIdx }});
+                        const streamEndWaitableIdx = params[0];
+                        if (!streamEndWaitableIdx) {{ throw new Error('missing stream idx'); }}
+
+                        const streamEnd = cstate.getStreamEnd({{ tableIdx: streamTableIdx, streamEndWaitableIdx }});
                         if (!streamEnd) {{
-                            throw new Error(`missing stream end [${{streamEndIdx}}] (table [${{componentTableIdx}}]) in component [${{componentIdx}}] during lift`);
+                            throw new Error(`missing stream end [${{streamEndWaitableIdx}}] (table [${{streamTableIdx}}]) in component [${{componentIdx}}] during lift`);
                         }}
 
+                        // TODO: check for borrowed type
+                        // TODO: check for readable only
+                        // TODO: confirm shared type matches tyep for lift
+                        // TODO: check for IDLE state
+
                         const stream = new {external_stream_class}({{
-                            hostStreamRep: streamEnd.streamRep(),
+                            globalRep: streamEnd.globalStreamMapRep(),
                             isReadable: streamEnd.isReadable(),
                             isWritable: streamEnd.isWritable(),
                             writeFn: (v) => {{ return streamEnd.write(v); }},

crates/js-component-bindgen/src/intrinsics/lower.rs

@@ -1,9 +1,9 @@
 //! Intrinsics that represent helpers that enable Lower integration
 
-use crate::{
-    intrinsics::{Intrinsic, p3::error_context::ErrCtxIntrinsic, string::StringIntrinsic},
-    source::Source,
-};
+use crate::intrinsics::Intrinsic;
+use crate::intrinsics::p3::{async_stream::AsyncStreamIntrinsic, error_context::ErrCtxIntrinsic};
+use crate::intrinsics::string::StringIntrinsic;
+use crate::source::Source;
 
 use super::conversion::ConversionIntrinsic;
 
@@ -708,12 +708,53 @@ impl LowerIntrinsic {
 
             Self::LowerFlatStream => {
                 let debug_log_fn = Intrinsic::DebugLog.name();
-                output.push_str(&format!("
-                    function _lowerFlatStream(size, memory, vals, storagePtr, storageLen) {{
-                        {debug_log_fn}('[_lowerFlatStream()] args', {{ size, memory, vals, storagePtr, storageLen }});
-                        throw new Error('flat lower for streams not yet implemented!');
+                let global_stream_map = AsyncStreamIntrinsic::GlobalStreamMap.name();
+                let external_stream_class = AsyncStreamIntrinsic::ExternalStreamClass.name();
+                let internal_stream_class = AsyncStreamIntrinsic::InternalStreamClass.name();
+
+                // TODO: fix writable is getting dropped before it can be read!!
+                // We need to do some waiting?
+                // Last write should have been triggering the reader to progress...
+                // Then the last reads return all the data???
+
+                output.push_str(&format!(
+                    r#"
+                    function _lowerFlatStream(streamTableIdx, ctx) {{
+                        {debug_log_fn}('[_lowerFlatStream()] args', {{ streamTableIdx, ctx }});
+                        const {{
+                            memory,
+                            realloc,
+                            vals,
+                            storagePtr: resultPtr,
+                        }} = ctx;
+
+                        const externalStream = vals[0];
+                        if (!externalStream || !(externalStream instanceof {external_stream_class})) {{
+                            throw new Error("invalid external stream value");
+                        }}
+
+                        const globalRep = externalStream.globalRep();
+                        const internalStream = {global_stream_map}.get(globalRep);
+                        if (!internalStream || !(internalStream instanceof {internal_stream_class})) {{
+                            throw new Error(`failed to find internal stream with rep [${{globalRep}}]`);
+                        }}
+
+                        const readEnd = internalStream.readEnd();
+                        const waitableIdx = readEnd.waitableIdx();
+
+                        // Write the idx of the waitable to memory (a waiting async task or caller)
+                        if (resultPtr) {{
+                            new DataView(memory.buffer).setUint32(resultPtr, waitableIdx, true);
+                        }}
+
+                        // TODO: if we flat lower another way (host -> guest async) we need to actually
+                        // modify the guests table's afresh, we can't just use the global rep!
+                        // (can detect this by whether the external stream has a rep or not)
+
+                        return waitableIdx
                     }}
-                "));
+                "#
+                ));
             }
 
             // When a component-model level error context is lowered, it contains the global error-context