resonate-recursive-fan-out-pattern-rust

# Resonate Recursive Fan-Out Pattern — Rust > **SDK note.** The Rust SDK is in active development (v0.6.0, on crates.io). This pattern uses `ctx.run().spawn()` and `ctx.rpc().spawn()` — both synchronous as of 0.5.0+. Verify against current SDK source before shipping. ## Overview Recursive fan-out spawns multiple child invocations in parallel, awaits them individually, and (optionally) recurses deeper. The Rust expression uses `.spawn()` on a Context execution builder to get a `DurableFuture`, collects those futures into a `Vec`, then awaits each in turn. For the language-agnostic mental model, see `resonate-recursive-fan-out-pattern-typescript`. ## When to use - Batch processing where items are independent - Web crawling / tree traversal with dynamic depth - Map-reduce shaped workflows - Any fan-out where each leaf is a discrete, retryable unit ## Parallel fan-out in the same process ```rust use resonate::prelude::*; #[resonate::function] async fn enrich_batch(ctx: &Context, order_ids: Vec<String>) -> Result<Vec<String>> { let mut futures = Vec::with_capacity(order_ids.len()); // fire off N children in parallel for id in order_ids { let fut = ctx.run(enrich_one, id).spawn()?; futures.push(fut); } // await all; order preserved let mut results = Vec::with_capacity(futures.len()); for f in futures { results.push(f.await?); } Ok(results) } #[resonate::function] async fn enrich_one(order_id: String) -> Result<String> { Ok(format!("enriched-{}", order_id)) } ``` Each `enrich_one` is a durable promise. If the parent crashes mid-loop, the children are stored server-side; on parent replay, `f.await?` lookups hit the stored results. ## Parallel fan-out across workers ```rust #[resonate::function] async fn parallel_enrich(ctx: &Context, ids: Vec<String>) -> Result<Vec<String>> { let mut futures = Vec::with_capacity(ids.len()); for id in ids { let fut = ctx .rpc::<String>("enrich_one", id) .target("poll://any@enrichment-workers") .spawn()?; futures.push(fut); } let mut results = Vec::with_capacity(futures.len()); for f in futures { results.push(f.await?); } Ok(results) } ``` The Resonate server handles fair dispatch across any process polling `enrichment-workers`. ## Recursive fan-out A durable function calling itself for tree traversal: ```rust #[derive(serde::Serialize, serde::Deserialize)] struct CrawlInput { url: String, depth: u32, } #[derive(serde::Serialize, serde::Deserialize)] struct CrawlOutput { url: String, depth: u32, links: Vec<String>, children: Vec<CrawlOutput>, } #[resonate::function] async fn crawl(ctx: &Context, input: CrawlInput) -> Result<CrawlOutput> { let links = ctx.run(fetch_links, input.url.clone()).await?; if input.depth == 0 { return Ok(CrawlOutput { url: input.url, depth: 0, links, children: vec![], }); } // fan out to each link, one depth deeper let mut futures = Vec::with_capacity(links.len()); for link in &links { let sub = CrawlInput { url: link.clone(), depth: input.depth - 1 }; let fut = ctx.rpc::<CrawlOutput>("crawl", sub).spawn()?; futures.push(fut); } let mut children = Vec::with_capacity(futures.len()); for f in futures { children.push(f.await?); } Ok(CrawlOutput { url: input.url, depth: input.depth, links, children }) } #[resonate::function] async fn fetch_links(url: String) -> Result<Vec<String>> { // ... Ok(vec![]) } ``` ## Bounded parallelism Process in chunks of K to cap in-flight invocations: ```rust #[resonate::function] async fn process_bounded( ctx: &Context, (items, concurrency): (Vec<String>, usize), ) -> Result<Vec<String>> { let mut results = Vec::with_capacity(items.len()); for chunk in items.chunks(concurrency) { let mut futures = Vec::with_capacity(chunk.len()); for item in chunk { let fut = ctx.run(process_one, item.clone()).spawn()?; futures.push(fut); } for f in futures { results.push(f.await?); } } Ok(results) } #[resonate::function] async fn process_one(item: String) -> Result<String> { Ok(format!("processed-{}", item)) } ``` Each chunk of up to `concurrency` runs in parallel; the next chunk only starts after the previous fully settles. ## Partial-failure tolerance By default, a `?` on `.await?` bubbles the first child error up. To tolerate individual failures: ```rust #[resonate::function] async fn enrich_tolerant(ctx: &Context, ids: Vec<String>) -> Result<Vec<std::result::Result<String, String>>> { let mut futures = Vec::with_capacity(ids.len()); for id in ids { let fut = ctx.run(enrich_one, id).spawn()?; futures.push(fut); } let mut out = Vec::with_capacity(futures.len()); for f in futures { match f.await { Ok(v) => out.push(Ok(v)), Err(e) => out.push(Err(e.to_string())), } } Ok(out) } ``` (Note the return type uses `std::result::Result` to disambiguate from Resonate's aliased `Result`.) ## Distinct Rust idioms - **`.spawn()?`** — synchronous: returns `Result<DurableFuture<T>>` immediately (no await needed); unwrap with `?`, then await the `DurableFuture` later to get the actual `T` - **`Vec::with_capacity(n)`** — idiomatic capacity pre-allocation for known-size collections - **`.chunks(n)` slice method** — bounded parallelism cleaner than iterator fiddling - **Turbofish `ctx.rpc::<T>(...)`** — needed when the return type isn't inferable from context - **`.map_err(|e| e.to_string())`** for error-type narrowing in tolerant-fan-out cases - **`Clone`** on input types — borrowed slices (`&[String]`) can't cross `ctx.run` boundaries; own the data ## Avoid - Unbounded recursion without a depth counter — the promise tree grows; Rust's default stack is fine but the server-side promise graph isn't free - Holding borrowed references across `.await?` — `.spawn()` and `.await?` mean owned data only ## Related skills - `resonate-basic-durable-world-usage-rust` — `ctx.run`, `ctx.rpc`, `.spawn()`, builder options - `resonate-saga-pattern-rust` — fan-out inside or alongside sagas - `durable-execution` — foundational replay semantics - `resonate-recursive-fan-out-pattern-typescript` / `-python` — sibling SDKs