diff --git a/node-graph/node-macro/src/codegen.rs b/node-graph/node-macro/src/codegen.rs index ed01a7f548..a9de96885c 100644 --- a/node-graph/node-macro/src/codegen.rs +++ b/node-graph/node-macro/src/codegen.rs @@ -1859,14 +1859,28 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn // carried frame when the node has a carrier. // A writing source stores the kernel's whole tuple as that plain value: // the lift writes the attributes through the claim, then lifts the - // element, the shape the sync record tail closes with. + // element, the shape the sync record tail closes with. An owned crossing + // parks into the serving arena first, so its exhaustion is a poll. let source_writes = (record_io && async_source && !write_markers.is_empty()).then(|| { let binders: Vec = (0..write_markers.len()).map(|index| format_ident!("__attr_{index}")).collect(); - let slots: Vec = (0..write_markers.len()).map(|index| format_ident!("__write_{index}")).collect(); + let stores = node.output.shape.attrs.iter().enumerate().map(|(index, attr)| { + let binder = &binders[index]; + let slot = format_ident!("__write_{index}"); + match attr.owned { + false => quote!(unsafe { __frame.attr_at(self.#slot, #binder.0) };), + true => quote! { + let #binder = match #binder.park(#core_types::context::ExtractArena::arena(__input)) { + ::core::option::Option::Some(value) => value, + ::core::option::Option::None => return #core_types::gpoll::GPoll::arena_exhausted(), + }; + unsafe { __frame.attr_at(self.#slot, #binder) }; + }, + } + }); quote! { - .map(|(__element #(, #core_types::attribute::Attr(#binders))*)| { - #(unsafe { __frame.attr_at(self.#slots, #binders) };)* - __element + .and_then(|(__element #(, #binders)*)| { + #(#stores)* + #core_types::gpoll::GPoll::Final(__element) }) } }); diff --git a/node-graph/node-macro/src/codegen/ir.rs b/node-graph/node-macro/src/codegen/ir.rs index 643d291dee..d7750d58c8 100644 --- a/node-graph/node-macro/src/codegen/ir.rs +++ b/node-graph/node-macro/src/codegen/ir.rs @@ -96,9 +96,23 @@ fn output(parsed: &ParsedNodeFn, generics: &[Ident]) -> Output { shape: ItemShape { element: element_of(&element, generics), depth, - attrs: writes.into_iter().map(|marker| LevelAttr { marker, level: 0 }).collect(), + attrs: writes + .into_iter() + .map(|write| LevelAttr { + marker: write.marker, + level: 0, + owned: write.owned, + }) + .collect(), }, - removes: removes.into_iter().map(|marker| LevelAttr { marker, level: 0 }).collect(), + removes: removes + .into_iter() + .map(|marker| LevelAttr { + marker, + level: 0, + owned: false, + }) + .collect(), gathers, } } @@ -158,6 +172,7 @@ fn item_shape(element: &Type, depth: u8, reads: &[AttributeRead], generics: &[Id .map(|read| LevelAttr { marker: read.marker.clone(), level: 0, + owned: false, }) .collect(), } @@ -557,6 +572,8 @@ pub(crate) enum Element { pub(crate) struct LevelAttr { pub(crate) marker: Type, pub(crate) level: u8, + /// Writes only: the value crosses as an owned copy that parks at the lift. + pub(crate) owned: bool, } pub(crate) enum Effect { @@ -680,7 +697,7 @@ mod tests { Facts { sources: if skips_carrier(parsed) { vec![] } else { vec![0] }, carried: token_carrier(parsed), - writes: markers(write_markers.iter()), + writes: markers(write_markers.iter().map(|write| &write.marker)), removes: markers(removes.iter()), delta: 0, } @@ -780,6 +797,19 @@ mod tests { ); } + #[test] + fn bridge_record_owned_write_async_source() { + let node = assert_bridge( + quote!(category("")), + quote!( + async fn tag_async(_: impl Ctx, _: (), val: f64) -> (f64, OwnedAttr