Let an async source write an owned attribute through its claim

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<Ident> = (0..write_markers.len()).map(|index| format_ident!("__attr_{index}")).collect();
-		let slots: Vec<Ident> = (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)
 			})
 		}
 	});

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<Label>) {
+					(val, OwnedAttr::new(""))
+				}
+			),
+		);
+		assert!(node.output.shape.attrs.iter().all(|attr| attr.owned), "an `OwnedAttr` slot crosses the boundary owned");
+	}
+
 	/// An async source's element is the value its slot stores, so a byte-carried
 	/// generic token has no form here.
 	#[test]

node-graph/node-macro/src/parsing.rs

@@ -51,13 +51,20 @@ pub(crate) struct AttributeRead {
 	pub(crate) marker: Type,
 }
 
+/// One attribute write slot: the marker, and whether it crosses as an owned
+/// copy (`OwnedAttr<M>`) instead of an evaluation-lifetime value (`Attr<M>`).
+pub(crate) struct AttrWrite {
+	pub(crate) marker: Type,
+	pub(crate) owned: bool,
+}
+
 /// The write half of a record kernel's return: the element type in the first
 /// tuple slot, then the attribute markers written and the ones removed. `None`
 /// unless the value is a well-formed write tuple (a non-marker element first,
-/// then only `Attr` and `RemoveAttr` slots, at least one).
+/// then only `Attr`, `OwnedAttr` and `RemoveAttr` slots, at least one).
 pub(crate) struct RecordWrites {
 	pub(crate) element: Type,
-	pub(crate) markers: Vec<Type>,
+	pub(crate) markers: Vec<AttrWrite>,
 	pub(crate) removes: Vec<Type>,
 }
 
@@ -65,14 +72,16 @@ pub(crate) fn record_writes(value: &Type) -> Option<RecordWrites> {
 	let Type::Tuple(tuple) = value else { return None };
 	let mut slots = tuple.elems.iter();
 	let element = slots.next()?;
-	if attr_marker(element).is_some() || remove_attr_marker(element).is_some() {
+	if attr_marker(element).is_some() || owned_attr_marker(element).is_some() || remove_attr_marker(element).is_some() {
 		return None;
 	}
 	let mut markers = Vec::new();
 	let mut removes = Vec::new();
 	for slot in slots {
 		if let Some(marker) = attr_marker(slot) {
-			markers.push(marker);
+			markers.push(AttrWrite { marker, owned: false });
+		} else if let Some(marker) = owned_attr_marker(slot) {
+			markers.push(AttrWrite { marker, owned: true });
 		} else if let Some(marker) = remove_attr_marker(slot) {
 			removes.push(marker);
 		} else {
@@ -91,6 +100,11 @@ pub(crate) fn attr_marker(ty: &Type) -> Option<Type> {
 	marker_of(ty, "Attr")
 }
 
+/// Returns the marker type of an `OwnedAttr<Marker>` type, if `ty` is one.
+pub(crate) fn owned_attr_marker(ty: &Type) -> Option<Type> {
+	marker_of(ty, "OwnedAttr")
+}
+
 /// Returns the marker type of a `RemoveAttr<Marker>` type, if `ty` is one.
 pub(crate) fn remove_attr_marker(ty: &Type) -> Option<Type> {
 	marker_of(ty, "RemoveAttr")

node-graph/node-macro/src/validation.rs

@@ -162,8 +162,11 @@ fn validate_record_io(parsed: &ParsedNodeFn) {
 	}
 	if let Some(writes) = &writes {
 		let mut seen_writes: Vec<String> = Vec::new();
-		for marker in &writes.markers {
-			let written = marker.to_token_stream().to_string();
+		for write in &writes.markers {
+			if write.owned && !async_source {
+				emit_error!(parsed.output_type.span(), "an owned attribute crossing belongs to an async source; a synchronous write parks its value in the kernel");
+			}
+			let written = write.marker.to_token_stream().to_string();
 			if seen_writes.contains(&written) {
 				emit_error!(parsed.output_type.span(), "attribute `{}` is written twice", written);
 			}