Async recursion, async closures, and streams

Three things that each stop working code from compiling, and all three follow from the state-machine picture.

pub fn run(tree: Vec<Vec<usize>>, x: u32, factor: u32, fail_first: u32, take: usize)
    -> (u64, u32, Option<u32>, Vec<u64>)

run returns (tree sum, apply_twice result, retry result, collected stream). Each piece is a separate small exercise.

(a) Recursion needs a box, and you can derive why

Sum the node indices of a tree given as an adjacency list, with an async function that awaits itself on each child.

Write it the obvious way and rustc says:

error[E0733]: recursion in an async fn requires boxing
   = note: a recursive `async fn` call must introduce indirection
           such as `Box::pin` to avoid an infinitely sized future

You can now derive that error instead of memorising it. An async fn‘s future is a struct containing every local live across an .await. If the body awaits itself, one of those locals is a future of the same type — so size_of::<F>() = size_of::<F>() + something. There is no finite answer, and the compiler says so.

Pin<Box<dyn Future<Output = T>>> fixes it because a box has a fixed size whatever it points at. The cost is one heap allocation per level of recursion, which is why the trees here are shallow and why deep async recursion is a real performance question rather than a free abstraction.

Two practical notes: hide the type behind an alias so clippy::type_complexity stays quiet, and note the boxed form does not trip clippy::manual_async_fn — that lint only fires on -> impl Future returning a bare async block.

(b) Async closures

Rust 1.85, alongside edition 2024, stabilised async |x| { .. } and the AsyncFn / AsyncFnMut / AsyncFnOnce traits in std::ops. Write two higher-order async functions with them:

pub async fn apply_twice<F: AsyncFn(u32) -> u32>(f: F, x: u32) -> u32;
pub async fn retry<F: AsyncFnMut() -> Option<u32>>(mut f: F, attempts: u32) -> Option<u32>;

retry calls f up to attempts times (3 here) and returns the first Some. f counts its own calls and starts succeeding after fail_first failures.

They filled a real gap. Before them the workaround was a plain closure returning an async block:

|x| async move { .. }

which works right up until the returned future needs to borrow from the closure’s own captures. Then it cannot: the closure returns a future that must outlive the borrow, the lifetimes cannot be expressed, and you get an error about borrowed data escaping (E0525 on the closure, or an E0596-shaped message about the capture) that is genuinely hard to read. Higher-ranked async signatures — “for any lifetime, this takes a &'a T and returns a future valid for 'a“ — were simply inexpressible.

AsyncFnMut is what lets retry take a closure that mutates its own captured counter, exactly as FnMut does in synchronous code.

While you are here: clippy::redundant_async_block fires on async { fut.await }, which is what over-wrapping looks like once you have the syntax.

(c) Streams

A stream is to Iterator what Future is to a single value:

fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>>

Read the return type carefully, because it is doing two independent jobs. Pending means “no item yet“. Ready(None) means “no items ever again“. Ready(Some(v)) is an item. Collapsing those into one nullable answer is the design mistake this signature exists to avoid.

AsyncIterator is still unstable in Rust 1.95. It lives in std::async_iter behind #![feature(async_iterator)], and the ecosystem uses futures::Stream instead. Neither is available here, so you define the trait yourself — which is arguably better, because it makes the Poll<Option<T>> decision visible instead of inherited.

Define SimpleStream, implement a Counter that yields 0, 1, 2, ..., plus Map and Take adaptors, and collect the first take items of Counter.map(* factor). Box the inner stream in each adaptor so the adaptors stay Unpin — otherwise every one of them needs the pin projection from item 16.16, and this jumps a whole tier.

You cannot for over it: for desugars to IntoIterator, and that is a different trait with a synchronous next. The idiom is

while let Some(value) = next(&mut stream).await { .. }

and Take must return Ready(None) without polling its inner stream once its budget is spent — the contract after Ready(None) is “do not poll again”, and an adaptor is the first place people break it.