Introduction

Welcome to Typed Effects in Rust.

If you already use async Rust, you know the model: Futures are polled by an executor; work runs when those futures are driven (for example with .await). That foundation is sound, and this book does not ask you to unlearn it.

What teams often hit next is organization at scale: error types that grow without structure, dependencies threaded through long call chains, and background work whose lifetime is hard to reason about. Those problems are not unique to Rust, but they show up in every non-trivial async codebase.

id_effect is a library for writing async programs where the shape of the work—success type, error type, and required environment—is carried in one place, and where much of the program is built as composable descriptions (Effect<A, E, R>) that you run only when you choose how and with which dependencies.

You still run on ordinary async runtimes. You still use .await inside bridges to third-party code. What changes is how you structure domain logic, tests, and dependency boundaries.

Who This Book Is For

You should know Rust basics: ownership, borrowing, traits, and how async/await and Future fit together. You do not need prior experience with category theory or functional programming jargon—we introduce terms only when they help.

If you want a typed, compositional style for async Rust—with explicit requirements in the type system and a clear split between “what to run” and “how to run it”—this book is for you.

How to Read This Book

Part I: Foundations explains why effects are useful and teaches the core types. Start here.

Part II: Environment & Dependencies covers the R parameter and compile-time dependency injection patterns, then walks the workspace integration crates (Tokio bridge, platform I/O, reqwest, Axum, Tower, config, logging) so you can wire a real binary without leaving the book.

Part III: Real Programs covers error handling, concurrency, resources, and scheduling for production code.

Part IV: Advanced covers STM, streams, schemas, and testing—read when you need those topics.

Code examples are intended to compile unless marked otherwise.

Let's begin.

Why Effects?

Before we write effect code in detail, it helps to agree on what problem we are solving and where id_effect sits relative to ordinary async Rust.

Rust’s async model is built on Future and executors: futures are lazy until polled, and .await is how async functions compose. That model is not a mistake—it is the standard way to express non-blocking I/O and concurrency.

At application scale, the difficulties are usually engineering ones:

• Errors — mapping and aggregating failures across layers without losing structure.
• Dependencies — passing clients, configuration, and context without turning every function signature into a long parameter list (or hiding the same behind globals).
• Concurrency — knowing who owns a task, how it shuts down, and what happens on cancellation.

This chapter names those patterns, relates them to how Effect<A, E, R> is designed, and sets up the rest of Part I.

By the end of the chapter you should understand:

• Why teams reach for a declarative layer on top of hand-written async fn chains.
• What “effect” means in this book: a description of work, separate from running it with a chosen environment.
• Why the type has three parameters (A, E, R) and why that matters for APIs and tests.

We start with a concrete look at those recurring challenges.

Challenges in Large Async Codebases

Async Rust gives you non-blocking I/O and structured concurrency primitives. In production, the same strengths can become painful when composition and boundaries are not planned: errors, dependencies, and spawned work all tend to accumulate complexity.

This section is not a claim that “async is broken.” It is a concise picture of problems id_effect is meant to help with—so the rest of the book has a shared vocabulary.

Challenge 1: Error mapping and noise

A typical async workflow chains several operations. Each step may fail in its own way, so you map errors into a domain type and propagate:

#![allow(unused)]
fn main() {
async fn process_order(order: Order) -> Result<Receipt, ProcessError> {
    let config = get_config()
        .await
        .map_err(|e| ProcessError::Config(e))?;

    let user = fetch_user(&config, order.user_id)
        .await
        .map_err(|e| ProcessError::User(e))?;

    let inventory = check_inventory(&config, &order.items)
        .await
        .map_err(|e| ProcessError::Inventory(e))?;

    let payment = charge_payment(&config, &user, order.total)
        .await
        .map_err(|e| ProcessError::Payment(e))?;

    let shipment = create_shipment(&config, &order, &user)
        .await
        .map_err(|e| ProcessError::Shipment(e))?;

    Ok(Receipt::new(order, payment, shipment))
}
}

The business steps are clear, but the .map_err noise is repetitive. The domain ProcessError enum often grows with every new integration. Policy (retries, fallbacks) may live in callers or ad hoc helpers, which makes behavior harder to see in one place.

What effects add: failure and recovery can be expressed as transformations on a description (for example retry, map_error, structured Exit types), so policies are easier to reuse and test without rewriting the core flow.

Challenge 2: Explicit dependency parameters

Another common shape is the handler that needs many clients and cross-cutting services:

#![allow(unused)]
fn main() {
async fn handle_request(
    db: &DatabasePool,
    cache: &RedisClient,
    logger: &Logger,
    config: &AppConfig,
    metrics: &MetricsClient,
    tracer: &Tracer,
    request: Request,
) -> Response {
    // ...
}
}

Dependencies are explicit, which is good for honesty, but every layer between here and main must repeat or forward them. Tests must build or mock the same bundle repeatedly. Alternatives (globals, implicit context) trade one problem for another.

What effects add: required capabilities can be expressed in R (the environment type) and satisfied in one place at the edge, while inner functions stay focused on logic.

Challenge 3: Background work and lifetimes

Fire-and-forget background tasks are easy to start and harder to reason about:

#![allow(unused)]
fn main() {
fn start_background_worker(db: DatabasePool) {
    tokio::spawn(async move {
        loop {
            match process_queue(&db).await {
                Ok(_) => {}
                Err(e) => eprintln!("Worker error: {}", e),
            }
            tokio::time::sleep(Duration::from_secs(5)).await;
        }
    });
}
}

Questions that matter in production—shutdown, cancellation, panic behavior, and resource cleanup—need explicit design. That is true in any async system; the goal is to make ownership and intent visible in the program structure.

What effects add: structured concurrency patterns (fibers, scopes, handles) integrate with the same Effect abstraction so “what runs” and “how it ends” can be expressed consistently.

How this relates to Future and async

Remember: async fn bodies compile to Futures; nothing runs until a future is polled (for example via .await on an async caller). The difficulties above are usually about how we organize async code—signatures, error types, and where side effects are allowed—not about rejecting the Future model.

In practice, hand-written async often reads like a straight-line script: await one step, then the next. That is appropriate for many functions. It becomes harder when you want the same logical workflow to be inspected, wrapped (retries, timeouts), or tested with a substituted environment without threading mocks through every layer.

Effects push the “script” into a value: Effect<A, E, R> is a description that you run with run_async, run_blocking, or test harnesses—after you have composed and configured it.

That does not replace understanding executors or Future. It adds a layer for domain structure: answer type A, error type E, requirements R, and explicit execution.

Next we define what an Effect is in this library and how that description differs from calling async fn directly—without exaggerating either side.

What Even Is an Effect?

An Effect is a description of a computation, not the computation itself.

The rest of the API—map, flat_map, environment types, runners—is there to work with that description in a type-safe way.

The Recipe Analogy

Think about a recipe for chocolate cake.

A recipe is not a cake. You can hold a recipe in your hands without any flour appearing. You can read a recipe without preheating an oven. You can photocopy a recipe, modify it (less sugar, more cocoa), combine it with a frosting recipe, and share it with a friend — all without a single cake coming into existence.

The cake only appears when someone executes the recipe. Takes out the ingredients, follows the steps, waits for the oven.

An Effect is a recipe for a computation.

When you write succeed(42), you're not "succeeding" at anything. You're writing down a recipe that says "when executed, produce the value 42." The 42 doesn't exist yet. No computation has happened. You just have a piece of paper with instructions on it.

#![allow(unused)]
fn main() {
use id_effect::{Effect, succeed};

// This doesn't compute anything — it's a description
let recipe: Effect<i32, String, ()> = succeed(42);

// Still nothing has happened. `recipe` is just a value.
// We can pass it around, store it, inspect its type.
}

The computation only happens when you explicitly run it:

#![allow(unused)]
fn main() {
use id_effect::run_blocking;

// NOW something happens
let result: Result<i32, String> = run_blocking(recipe);
assert_eq!(result, Ok(42));
}

Building Up Descriptions

Because an Effect is just data — a description — you can transform it without running it.

#![allow(unused)]
fn main() {
let recipe: Effect<i32, String, ()> = succeed(42);

// Transform the description: "when executed, produce 42, then double it"
let doubled: Effect<i32, String, ()> = recipe.map(|x| x * 2);

// Still nothing has happened! `doubled` is just a modified recipe.

// Now run it
let result = run_blocking(doubled);
assert_eq!(result, Ok(84));
}

The .map() call didn't execute anything. It took one recipe and produced a new recipe that includes an extra step. Like writing "double the result" at the bottom of your cake recipe — the cake doesn't change until someone bakes it.

A More Realistic Example

Let's see what this looks like with actual I/O:

#![allow(unused)]
fn main() {
use id_effect::{Effect, effect, run_blocking};

// This function doesn't fetch anything — it returns a DESCRIPTION
// of how to fetch a user
fn fetch_user(id: u64) -> Effect<User, DbError, ()> {
    effect! {
        let conn = ~ connect_to_db();
        let user = ~ query_user(&conn, id);
        Ok(user)
    }
}

// Calling the function doesn't open any connections
let description = fetch_user(42);

// `description` is a value we can hold, pass around, combine with others
// No database has been touched

// Only when we run it does the I/O happen
let user = run_blocking(description)?;
}

That effect! block looks imperative — it looks like it's doing things. But it's not. It's building a description of things to do. The ~ operator means "this step depends on the previous step completing" — it's describing sequencing, not executing it.

The Key Insight: Separation of Concerns

This separation — description vs execution — is how the challenges from the previous section (errors, dependencies, task structure) get a consistent home in the type system.

Error handling becomes part of the description itself. When you write:

#![allow(unused)]
fn main() {
let resilient = risky_operation.retry(Schedule::exponential(100.ms(), 3));
}

You're not adding retry logic to running code. You're modifying the description to say "when executed, retry up to 3 times with exponential backoff." The retry logic is baked into the recipe.

Dependencies become part of the type signature. When you write:

#![allow(unused)]
fn main() {
fn get_user(id: u64) -> Effect<User, DbError, caps!(Database)>
}

That caps!(Database) in the type says "this recipe requires a database capability to execute." The compiler enforces it. You can't run the effect without wiring Database at the edge. No runtime surprises.

Structured concurrency becomes possible because the runtime knows what each effect intends to do before it does it. Spawning an effect doesn't fire and forget — it creates a handle to a structured task with clear ownership and cancellation semantics.

What's in an Effect?

An Effect<A, E, R> carries three pieces of information in its type:

• A — the Answer: what you get if it succeeds
• E — the Error: what you get if it fails
• R — the Requirements: what environment is needed to run it

We'll explore all three in the next section. For now, just notice that an Effect's type tells you everything about what it does — success, failure, and dependencies — without you having to read the implementation.

#![allow(unused)]
fn main() {
// This type signature tells the whole story:
fn process_payment(
    amount: Money
) -> Effect<Receipt, PaymentError, caps!(PaymentGateway, EffectLogger)>

// - Produces a Receipt on success
// - Can fail with PaymentError
// - Requires PaymentGateway and EffectLogger to run
}

No need to read the function body to know what resources it needs or what errors it can produce. The type is the documentation.

Style: imperative async vs effect descriptions

Typical async fn code is written as a sequence of steps: each .await drives the next piece of work. That is clear and idiomatic Rust.

Effect code in this library is often written so that many domain functions return Effect<…>: a value that describes work and only runs when you pass it to a runner with an environment. The style emphasizes composition (map, flat_map, layers, retries) before execution.

Both approaches run on the same Future machinery underneath. Use effects where you want environment and error structure in the type, shared policies, and test substitution at the boundary; use plain async fn where a small linear function is enough.

Let's look at those three type parameters in detail.

The Three Type Parameters

Every Effect carries three type parameters: Effect<A, E, R>. These aren't arbitrary — they answer the three fundamental questions every computation must address:

• A — What do I produce when I succeed?
• E — What do I produce when I fail?
• R — What do I need in order to run?

Let's examine each one.

A: The Answer

The A parameter is the success type — what you get back when everything goes right.

#![allow(unused)]
fn main() {
use id_effect::{Effect, succeed};

// This effect produces an i32 on success
let answer: Effect<i32, String, ()> = succeed(42);

// This effect produces a User on success
let user_effect: Effect<User, DbError, ()> = succeed(User::new("Alice"));
}

If you're familiar with Result<T, E>, think of A as the T. It's what you're hoping to get.

When you transform an effect with .map(), you're changing the A:

#![allow(unused)]
fn main() {
let numbers: Effect<i32, String, ()> = succeed(21);
let doubled: Effect<i32, String, ()> = numbers.map(|n| n * 2);
let stringified: Effect<String, String, ()> = doubled.map(|n| n.to_string());
}

Each .map() transforms the success value while preserving the error type and requirements.

E: The Error

The E parameter is the failure type — what you get back when something goes wrong.

#![allow(unused)]
fn main() {
use id_effect::{Effect, fail};

// This effect always fails with a String error
let failure: Effect<i32, String, ()> = fail("something went wrong".to_string());

// This effect can fail with a DbError
let user: Effect<User, DbError, ()> = fetch_user_from_db(42);
}

Again, if you know Result<T, E>, think of E as the E. It's what you're worried might happen.

You can transform error types with .map_error():

#![allow(unused)]
fn main() {
let db_effect: Effect<User, DbError, ()> = fetch_user(42);

// Convert DbError to a more general AppError
let app_effect: Effect<User, AppError, ()> = db_effect.map_error(|e| AppError::Database(e));
}

Unlike traditional error handling where you sprinkle .map_err() everywhere, with effects you typically handle error transformation at specific boundaries — when composing larger effects from smaller ones, or when exposing an API.

R: The Requirements

Here is where effects get interesting. The R parameter represents the environment — the dependencies this effect needs in order to run.

When an effect needs services, express R with caps! and capability keys (Chapter 5 names keys fully; use Database, not a bare Database type):

#![allow(unused)]
fn main() {
use id_effect::{Effect, caps, effect, provide, require, run_with, succeed};

// Self-contained — R is ()
let standalone: Effect<i32, String, ()> = succeed(42);

// Needs Database at the edge
fn get_user(id: u64) -> Effect<User, DbError, caps!(Database)> {
    effect!(|r| {
        let db = ~Database;
        Ok(db.fetch_user(id))
    })
}

// Needs two keys
fn get_user_logged(id: u64) -> Effect<User, DbError, caps!(Database, EffectLogger)> {
    effect!(|r| {
        let db = ~Database;
        let log = ~EffectLogger;
        let user = db.fetch_user(id)?;
        log.info(&format!("fetched {}", user.id));
        Ok(user)
    })
}
}

You cannot run an effect until its capabilities exist. Satisfy them at the program edge with run_with, not inside library code:

#![allow(unused)]
fn main() {
run_with([provide!(DatabaseLive)], get_user(42))?;
}

There is no .provide() on effects. run_with builds an Env and executes the program.

Why R matters

The R parameter is why id_effect offers compile-time dependency injection.

#![allow(unused)]
fn main() {
fn process_order(order: Order) -> Effect<
    Receipt,
    OrderError,
    caps!(Database, PaymentGateway, EmailService, EffectLogger),
>
}

Just from the type you know success, error, and which capability services must be wired before run_with.

R flows through composition

#![allow(unused)]
fn main() {
fn get_user(id: u64) -> Effect<User, DbError, caps!(Database)> { ... }
fn send_email(to: &str, body: &str) -> Effect<(), EmailError, caps!(EmailService)> { ... }

fn notify_user(id: u64) -> Effect<(), AppError, caps!(Database, EmailService)> {
    effect!(|r| {
        let user = ~ get_user(id).map_error(AppError::Db);
        ~ send_email(&user.email, "Hello!").map_error(AppError::Email);
        ()
    })
}
}

The unit environment: ()

When R = (), the effect is self-contained:

#![allow(unused)]
fn main() {
let standalone: Effect<i32, String, ()> = succeed(42);
let result = run_blocking(standalone, ());
}

Effects with dependencies keep caps!(…) on the type until the edge:

fn main() -> Result<(), AppError> {
    run_with(
        [
            provide!(DatabaseLive),
            provide!(CacheLive),
            provide!(LoggerLive),
            provide!(ConfigLive),
        ],
        business_logic(),
    )
}

Reading Effect Signatures

Let's practice reading some signatures:

#![allow(unused)]
fn main() {
// Produces String, never fails, needs nothing
Effect<String, Never, ()>

// Produces i32, can fail with ParseError, needs nothing
Effect<i32, ParseError, ()>

// Produces User, can fail with DbError, needs Database
Effect<User, DbError, caps!(Database)>

// Produces (), can fail with AppError, needs four capability services
Effect<(), AppError, caps!(Database, Cache, EffectLogger)>
}

With practice, you'll read these as fluently as you read Result<T, E>. The extra R parameter becomes second nature.

What's Next

We've seen that effects are descriptions, not actions. We've seen that Effect<A, E, R> encodes success type, error type, and requirements.

But we haven't answered the obvious question: why does this matter? Why is it better to describe computations than to just do them?

The answer is laziness. And laziness, it turns out, is a superpower.

Laziness as a Superpower

So far we've established that Effect<A, E, R> is a description of a computation — a recipe that does nothing until someone executes it. You might be thinking: "OK, but why is that good? I have to run it eventually. What do I gain by waiting?"

Quite a bit, if your program benefits from composing and testing before execution.

Here is what you can do with a computation you have not run yet.

Effect values vs driving an async fn

Rust futures are lazy: calling an async fn returns a Future; the body runs when that future is polled (for example with .await).

The contrast here is about what your API returns—a raw Future you must await immediately in the caller, versus an Effect value you can store, compose, and run later.

#![allow(unused)]
fn main() {
// Returns a Future; the HTTP work runs when this future is awaited / polled
async fn fetch_user_async(id: u64) -> Result<User, HttpError> {
    http_get(&format!("https://api.example.com/users/{id}")).await
}

// Returns a description; I/O runs when the effect is executed with an environment
fn fetch_user(id: u64) -> Effect<User, HttpError, HttpClient> {
    effect! {
        let user = ~ http_get(&format!("https://api.example.com/users/{id}"));
        user
    }
}
}

Calling fetch_user_async(1) only builds the future; the request runs when something polls it (typically at .await). Calling fetch_user(1) returns an Effect—still no I/O until you run that effect with a runner and the needed HttpClient.

The point is not that async fn is “eager.” It is that effects give you a first-class value to combine (retries, timeouts, tests) before you commit to a particular run.

Superpower #1: Compose First, Run Later

Because effects are values, you can build an entire program before running any of it:

#![allow(unused)]
fn main() {
fn load_dashboard(user_id: u64) -> Effect<DashboardPage, AppError, caps!(Database, Cache, EffectLogger)> {
    effect!(|r| {
        let user    = ~ fetch_user(user_id).map_error(AppError::Db);
        let posts   = ~ fetch_posts(user.id).map_error(AppError::Db);
        let profile = ~ build_profile(&user, &posts).map_error(AppError::Render);
        profile
    })
}

// Nothing has run yet. We have a value.
let page = load_dashboard(42);

// Chain more work onto it — still nothing runs
let logged_page = page.flat_map(|p| log_view(p));

// Only now does any of this execute — wire every key once at the edge
run_with(
    [provide!(DatabaseLive), provide!(CacheLive), provide!(LoggerLive)],
    logged_page,
)?;
}

Every line before run_blocking is pure data manipulation. You're assembling a pipeline. The pipeline can be inspected, transformed, passed to other functions, stored in a struct. The laws of composition apply cleanly because there are no side-effects sneaking in.

Superpower #2: Retry Without Rewriting

Because an effect is a description, you can wrap it with new behavior without touching the original:

#![allow(unused)]
fn main() {
let flaky = call_payment_api(order);

// Add exponential back-off retry — no changes to call_payment_api
let resilient = flaky.retry(Schedule::exponential(Duration::from_millis(100), 3));

// Add a timeout on top of that — still no changes
let bounded = resilient.timeout(Duration::from_secs(5));
}

Compare this to the async version: to add retries to an async fn, you'd either modify the function body, wrap it in a helper that calls it in a loop, or reach for an external crate. The retry logic gets tangled with the business logic.

With effects, retry is just another transformation. retry takes a lazy description, produces a new lazy description that runs the original up to N times. No surgery on the original required.

Superpower #3: Test Without Mocking the Universe

Because nothing runs until you provide the environment, tests can substitute controlled implementations without rewriting a single line of production code:

#![allow(unused)]
fn main() {
#[test]
fn user_not_found_returns_error() {
    let test_env = TestEnv::new()
        .with_http(stub_http_404_for("/users/99"));

    let result = run_test(fetch_user(99), test_env);

    assert!(matches!(result, Err(HttpError::NotFound)));
}
}

The same fetch_user function used in production runs in the test — just against a different environment. No #[cfg(test)] stubs. No Arc<dyn Trait> that you only swap out in tests. The type system ensures you've provided every dependency the effect declared.

Sequential async vs bundled descriptions

Sequential async fn code is natural for linear flows: each .await advances the next step, and control matches the source order.

Effect-oriented APIs often bundle those steps into a single Effect value first, then apply cross-cutting behavior (retry, timeout, tracing) as transformations on that value before calling run_*.

That separation is useful when the same workflow must be reused under different policies or tested with a substituted environment, without copying the body of the async function.

When Does It Actually Run?

There are exactly three places where an Effect executes:

#![allow(unused)]
fn main() {
// In a binary or application entry point
run_blocking(program, ());

// In an async context
run_async(program, ()).await;

// In tests
run_test(program, test_env);
}

Everywhere else, you're building, transforming, or combining descriptions. The runtime boundary is explicit. You know exactly where the side-effects begin.

Until run_* is called, your effect is just data: composable and easy to substitute in tests.

That's Chapter 1. You now have a picture of why teams adopt effects (errors, dependencies, concurrency structure), what an Effect is (a description executed with an environment), what the type parameters mean (A = success, E = failure, R = requirements), and why keeping work in description form matters for composition and testing.

Chapter 2 gets hands-on: first effects, map, flat_map, and a small end-to-end program.

Your First Effect

Chapter 1 was all philosophy. We established what effects are, why they exist, and why laziness is useful. Now we get our hands dirty.

By the end of this chapter you will have written real effects, transformed them, chained them together, and run a complete small program. You'll use succeed, fail, map, map_error, and flat_map — the five operations that cover the vast majority of day-to-day effect work.

Let's start with the simplest question: how do you create an effect in the first place?

Creating Effects — succeed, fail, and pure

Every effect starts as either a success or a failure. The two constructors that express this are succeed and fail.

succeed

succeed wraps a value into an effect that, when run, immediately produces that value:

#![allow(unused)]
fn main() {
use id_effect::{Effect, succeed};

let answer: Effect<i32, String, ()> = succeed(42);
let greeting: Effect<String, String, ()> = succeed("Hello, world!".to_string());
}

Nothing happens when you call succeed. You get back a description — a lazy recipe that says "produce this value when someone asks." The 42 is already there, but no computation has been executed.

The type parameters are important:

• A = i32 — the value we produce
• E = String — the error type (unused here, but we still have to pick one)
• R = () — no environment needed

If you prefer the FP vocabulary, pure is an alias for succeed:

#![allow(unused)]
fn main() {
use id_effect::pure;

let effect = pure(42_i32);
}

Both names refer to exactly the same thing. Use whichever feels natural in context.

fail

fail wraps an error into an effect that, when run, immediately fails with that error:

#![allow(unused)]
fn main() {
use id_effect::{Effect, fail};

let oops: Effect<i32, String, ()> = fail("something went wrong".to_string());
}

Again, nothing executes. oops is a description of a failure, not the failure itself. You can pass it around, store it, and transform it without triggering any error handling.

The type annotation matters: Effect<i32, String, ()> says this would have produced an i32 on success — we just know it won't.

From a Closure

For cases where you want to capture some computation in an effect (but still defer it):

#![allow(unused)]
fn main() {
use id_effect::{Effect, effect};

let computed: Effect<i32, String, ()> = effect!(|_r: &mut ()| {
    let x = expensive_calculation();
    x * 2
});
}

The body of effect! runs lazily — only when the effect is executed. This is the workhorse macro we'll cover thoroughly in Chapter 3.

Type Inference

Rust's type inference often lets you skip the annotations:

#![allow(unused)]
fn main() {
// Types inferred from usage
let answer = succeed(42);      // Effect<i32, _, ()>
let greeting = succeed("hi"); // Effect<&str, _, ()>
}

The error type E is usually inferred from how the effect is used later — when you chain it with other effects that can fail, the error type propagates. You'll only need to annotate explicitly when the compiler asks.

Quick Reference

#![allow(unused)]
fn main() {
succeed(value)    // Effect that produces value
pure(value)       // Alias for succeed
fail(error)       // Effect that fails with error
effect!(|_r| { … }) // Effect from a lazy closure
}

These three constructors cover every starting point. Everything else is transformation and composition.

Transforming Success — map and its Friends

You have an effect. It produces some value. But you want a different value — or a different error. That's what map and map_error are for.

map

map transforms the success value without running any new effects:

#![allow(unused)]
fn main() {
use id_effect::{succeed, Effect};

let number: Effect<i32, String, ()> = succeed(21);
let doubled: Effect<i32, String, ()> = number.map(|n| n * 2);
let text: Effect<String, String, ()> = doubled.map(|n| n.to_string());
}

None of these .map() calls executes anything. Each one wraps the previous description in a new layer: "and then transform the result with this function." The chain of transformations only runs when you call run_blocking or similar.

The type of the effect changes with each map. The A parameter shifts:

#![allow(unused)]
fn main() {
// Effect<i32, String, ()>
//   .map(|n: i32| n.to_string())
// → Effect<String, String, ()>
}

The E (error type) and R (requirements) stay the same. .map touches only the success path.

map_error

map_error transforms the failure type, leaving the success path untouched:

#![allow(unused)]
fn main() {
use id_effect::fail;

#[derive(Debug)]
struct AppError(String);

let db_err: Effect<String, String, ()> = fail("db connection failed".to_string());
let app_err: Effect<String, AppError, ()> = db_err.map_error(|s| AppError(s));
}

This is typically used at module boundaries when you need to unify error types. A database layer might return DbError, but your application layer needs AppError. map_error does the conversion without touching anything else.

Why These Don't Execute Anything

It's worth repeating: neither map nor map_error runs any computation.

#![allow(unused)]
fn main() {
let effect = succeed(42)
    .map(|n| { println!("mapping!"); n + 1 })
    .map(|n| n * 2);

// At this point: nothing has printed, nothing has computed.
// We have a description of three steps.

let result = run_blocking(effect, ());
// NOW the effect runs. "mapping!" prints once. Result is 86.
}

This is the promise of laziness: you can build pipelines of transformations without triggering side effects until the moment you choose.

Combining map and map_error

A common pattern is calling both to normalise an effect into your domain's types:

#![allow(unused)]
fn main() {
fn fetch_user_record(id: u64) -> Effect<User, AppError, ()> {
    raw_db_fetch(id)
        .map(|row| User::from_row(row))
        .map_error(|e| AppError::Database(e))
}
}

The effect goes in with raw DB types; it comes out with domain types. The transformation chain documents the conversion at a glance.

and_then / tap (convenience)

Two more helpers are worth knowing:

#![allow(unused)]
fn main() {
// and_then: map + flatten (when your mapper returns an Option or Result)
let validated: Effect<i32, String, ()> = succeed(42)
    .and_then(|n| if n > 0 { Some(n) } else { None });

// tap: inspect the success value without changing it
let logged: Effect<i32, String, ()> = succeed(42)
    .tap(|n| println!("value: {n}"));  // side-effect, same type flows through
}

tap is particularly useful for debugging — add it anywhere in a chain without disrupting the types.

Summary

None of them execute the effect. They all return new, larger descriptions.

Chaining Effects — flat_map and the Bind

map handles the case where your transformation is a pure function: A → B. But often the next step is itself an effect. You don't want Effect<Effect<B, E, R>, E, R> — you want Effect<B, E, R>. That's flat_map.

The Problem with map for Effects

Say you want to fetch a user and then fetch their posts:

#![allow(unused)]
fn main() {
fn get_user(id: u64) -> Effect<User, DbError, caps!(Database)> { ... }
fn get_posts(user_id: u64) -> Effect<Vec<Post>, DbError, Database> { ... }
}

If you try to use map:

#![allow(unused)]
fn main() {
// This gives Effect<Effect<Vec<Post>, DbError, Database>, DbError, Database>
// — a nested effect, not what we want
let wrong = get_user(1).map(|user| get_posts(user.id));
}

map's function must return a plain value. If it returns an Effect, you get nesting.

flat_map: Chain Without Nesting

flat_map (also known as and_then on effects) takes a function A → Effect<B, E, R> and "flattens" the result:

#![allow(unused)]
fn main() {
let combined: Effect<Vec<Post>, DbError, Database> =
    get_user(1).flat_map(|user| get_posts(user.id));
}

Now you have one flat effect that, when run, first fetches the user, then uses the result to fetch posts. The nesting is gone.

Chaining Multiple Steps

flat_map chains read left-to-right, but deep chains get noisy:

#![allow(unused)]
fn main() {
// Gets unwieldy quickly
let program = get_user(1)
    .flat_map(|user| get_posts(user.id)
        .flat_map(|posts| render_page(user, posts)));
}

This is where the effect! macro comes in.

The effect! Macro as Syntactic Sugar

The effect! macro turns flat_map chains into readable sequential code using the ~ operator:

#![allow(unused)]
fn main() {
use id_effect::effect;

let program: Effect<Page, AppError, caps!(Database)> = effect!(|r| {
    let user  = ~ get_user(1).map_error(AppError::Db);
    let posts = ~ get_posts(user.id).map_error(AppError::Db);
    let page  = render_page(user, posts);
    page
};
}

The ~ operator is the bind: "run this effect and give me its success value." Each ~ expr desugars to a flat_map. The whole block is one effect.

Note that render_page (a pure function with no ~) is just a normal Rust expression — it runs inside the macro body during execution.

Error Short-Circuiting

Like ? in Result, if any ~ step fails, the whole effect! exits early with that error:

#![allow(unused)]
fn main() {
let program: Effect<Page, AppError, caps!(Database)> = effect!(|r| {
    let user = ~ get_user(999).map_error(AppError::Db);
    // If get_user fails, execution stops here.
    // The rest never runs.
    let posts = ~ get_posts(user.id).map_error(AppError::Db);
    render_page(user, posts)
};
}

This is sequential, not parallel. Each step waits for the previous.

map vs flat_map — When to Use Each

A rule of thumb: if you find yourself writing effect.map(|v| another_effect(v)) and noticing the nested type, switch to flat_map or the macro.

The Full Picture

#![allow(unused)]
fn main() {
// All equivalent:

// 1. Explicit flat_map
get_user(1)
    .flat_map(|user| get_posts(user.id))

// 2. Using effect! with ~
effect! {
    let user = ~ get_user(1);
    ~ get_posts(user.id)
}

// 3. Short form for single bind
effect! { ~ get_user(1).flat_map(|u| get_posts(u.id)) }
}

The effect! macro is the idiomatic choice for anything more than one step. Chapter 3 covers it in full detail.

Your First Real Program

Let's build something complete: a small program that loads configuration, connects to a database, queries a user, and formats a greeting. It's simple enough to fit on one page, but real enough to demonstrate the full effect workflow.

The Domain

#![allow(unused)]
fn main() {
#[derive(Debug)]
struct Config {
    db_url: String,
    app_name: String,
}

#[derive(Debug)]
struct User {
    id: u64,
    name: String,
    email: String,
}

#[derive(Debug)]
enum AppError {
    Config(String),
    Database(String),
}
}

The Individual Steps

Each step is a focused effect:

#![allow(unused)]
fn main() {
use id_effect::{Effect, effect, succeed, fail};

fn load_config() -> Effect<Config, AppError, ()> {
    // In a real app, read from a file or env vars
    succeed(Config {
        db_url: "postgres://localhost/myapp".to_string(),
        app_name: "Greeter".to_string(),
    })
}

fn connect_db(config: &Config) -> Effect<Database, AppError, ()> {
    Database::connect(&config.db_url)
        .map_error(|e| AppError::Database(format!("connect: {e}")))
}

fn fetch_user(db: &Database, id: u64) -> Effect<User, AppError, ()> {
    db.query_user(id)
        .map_error(|e| AppError::Database(format!("query: {e}")))
}

fn format_greeting(config: &Config, user: &User) -> String {
    format!("{}: Hello, {}! ({})", config.app_name, user.name, user.email)
}
}

Composing the Program

Now we compose these steps into one effect using effect!:

#![allow(unused)]
fn main() {
fn greet_user(user_id: u64) -> Effect<String, AppError, ()> {
    effect! {
        let config = ~ load_config();
        let db     = ~ connect_db(&config);
        let user   = ~ fetch_user(&db, user_id);
        format_greeting(&config, &user)
    }
}
}

Read it like a recipe:

1. Load config — if it fails, stop with AppError::Config
2. Connect to DB — if it fails, stop with AppError::Database
3. Fetch user — if it fails, stop with AppError::Database
4. Format the greeting — this is pure, always succeeds

Nothing has run yet. greet_user(42) is a value.

Running It

At the edge of the program — in main — we execute:

fn main() {
    match run_blocking(greet_user(42)) {
        Ok(greeting) => println!("{greeting}"),
        Err(AppError::Config(msg)) => eprintln!("Config error: {msg}"),
        Err(AppError::Database(msg)) => eprintln!("DB error: {msg}"),
    }
}

Testing It

Because the effect is a description, testing is straightforward — just swap out the underlying steps:

#![allow(unused)]
fn main() {
#[test]
fn test_greeting_format() {
    let effect = effect! {
        let config = ~ succeed(Config {
            db_url: "unused".into(),
            app_name: "TestApp".into(),
        });
        let user = ~ succeed(User {
            id: 1,
            name: "Alice".into(),
            email: "alice@example.com".into(),
        });
        format_greeting(&config, &user)
    };

    let result = run_test(effect);
    assert_eq!(result.unwrap(), "TestApp: Hello, Alice! (alice@example.com)");
}
}

No mocking framework. No Arc<dyn Trait> plumbing. Just substitute different succeed values for the steps you want to control.

What You Just Learned

You've written a complete effect-based program. Along the way you used:

• succeed and fail to construct effects from values
• .map and .map_error to transform success and error types
• effect! { ~ ... } to sequence effects without callback nesting
• run_blocking to execute at the program edge
• run_test to verify behaviour in tests

That's the core of 90% of what you'll write day-to-day. The next two chapters go deeper: Chapter 3 explores the effect! macro in detail, and Chapter 4 begins the tour of R — the environment type that makes dependency injection a compile-time guarantee.

You just wrote your first effect-based program. It won't be your last.

The effect! Macro — Do-Notation for Mortals

Chapter 2 introduced the effect! macro as "syntactic sugar for flat_map." That's technically accurate, but undersells it. In practice, effect! is how you write almost every multi-step computation in id_effect.

This chapter covers the why, the how, and the limits of the macro. By the end you'll be fluent in ~, comfortable handling errors inside the macro, and clear on when not to use it.

Why Do-Notation Exists

Consider three steps that each depend on the previous result:

#![allow(unused)]
fn main() {
fn step_a() -> Effect<i32, Err, ()>   { succeed(1) }
fn step_b(n: i32) -> Effect<i32, Err, ()> { succeed(n * 2) }
fn step_c(n: i32) -> Effect<String, Err, ()> { succeed(n.to_string()) }
}

Written with raw flat_map:

#![allow(unused)]
fn main() {
let program = step_a()
    .flat_map(|a| step_b(a)
        .flat_map(|b| step_c(b)));
}

Two steps: readable. Five steps: a pyramid. Ten steps: indistinguishable from callback hell.

Haskell solved this decades ago with do-notation. Scala's for-comprehensions do the same thing. Rust doesn't have built-in do-notation, so id_effect provides it via a macro.

Do-Notation as a Concept

Do-notation lets you write sequential effectful code that looks like imperative code:

do
  a ← step_a
  b ← step_b(a)
  c ← step_c(b)
  return c

Each ← means "run this effect and bind its result to this name." If any step fails, the whole computation short-circuits.

Rust can't use the ← symbol, so id_effect uses ~ (prefix tilde):

#![allow(unused)]
fn main() {
effect! {
    let a = ~ step_a();
    let b = ~ step_b(a);
    let c = ~ step_c(b);
    c
}
}

Same semantics. Rust syntax. Zero nesting.

How the Desugaring Works

The macro transforms each ~ expr into a flat_map:

#![allow(unused)]
fn main() {
// Written:
effect! {
    let a = ~ step_a();
    let b = ~ step_b(a);
    b.to_string()
}

// Roughly expands to:
step_a().flat_map(|a| {
    step_b(a).flat_map(|b| {
        succeed(b.to_string())
    })
})
}

The macro generates exactly the nested flat_map chain you'd write by hand — just without the visual noise.

One Body, One block

One discipline matters: use one effect! block per function. Don't branch between two macro bodies:

#![allow(unused)]
fn main() {
// BAD — two separate effect! blocks for one computation
if flag {
    effect! { let x = ~ a(); x }
} else {
    effect! { let y = ~ b(); y }
}

// GOOD — one block, branching inside
effect! {
    if flag {
        ~ a()
    } else {
        ~ b()
    }
}
}

A single effect! block is a single description. Splitting it into multiple blocks loses the composition guarantee.

Pure Expressions

Not every line inside effect! has to be an effect. Pure Rust expressions work normally:

#![allow(unused)]
fn main() {
effect! {
    let user = ~ fetch_user(id);
    let name = user.name.to_uppercase();  // pure — no ~
    let posts = ~ fetch_posts(user.id);
    (name, posts)
}
}

Only use ~ when the expression has type Effect<_, _, _>. Pure expressions just run inline.

The ~ Operator Explained

The ~ (tilde) is the bind operator inside effect!. It means: "execute this effect and give me its success value; if it fails, propagate the failure and stop."

Basic Usage

#![allow(unused)]
fn main() {
effect! {
    let user = ~ fetch_user(42);   // bind the result to `user`
    user.name
}
}

~ fetch_user(42) desugars to a flat_map. The rest of the block becomes the body of the closure.

Discarding Results

When you don't need the value, use ~ without a binding:

#![allow(unused)]
fn main() {
effect! {
    ~ log_event("processing started");   // run for side effect, discard result
    let result = ~ do_work();
    ~ log_event("processing done");
    result
}
}

Both ~ log_event(...) expressions run for their effects and the () return is discarded.

Method Calls on Effects

~ works on any expression that evaluates to an Effect. That includes method chains:

#![allow(unused)]
fn main() {
effect! {
    let user = ~ fetch_user(id).map_error(AppError::Database);
    let posts = ~ fetch_posts(user.id)
        .map_error(AppError::Database)
        .retry(Schedule::exponential(100.ms()).take(3));
    (user, posts)
}
}

The ~ applies to the entire expression, including any .map_error(), .retry(), etc. that follow.

~ in Conditionals and Loops

You can use ~ inside if expressions and loops:

#![allow(unused)]
fn main() {
effect! {
    let value = if condition {
        ~ compute_a()
    } else {
        ~ compute_b()
    };
    process(value)
}
}

Both branches are effects; the macro handles either path.

#![allow(unused)]
fn main() {
effect! {
    for id in user_ids {
        ~ process_user(id);  // sequential: one at a time
    }
    "done"
}
}

Note: this is sequential iteration. For concurrent processing, use fiber_all (Chapter 9).

What ~ Cannot Do

~ only works inside an effect! block. Calling it outside is a compile error:

#![allow(unused)]
fn main() {
// Does not compile — ~ is not valid here
let x = ~ fetch_user(42);

// Must be inside effect!
let x = effect! { ~ fetch_user(42) };
}

Also, ~ cannot bind across an async closure boundary. If you're calling from_async, the body of the async block is separate:

#![allow(unused)]
fn main() {
effect! {
    let result = ~ from_async(|_r| async move {
        // Inside here, you're in regular Rust async — no ~
        let data = some_future().await?;
        Ok(data)
    });
    result
}
}

Use ~ outside the async move block; use .await inside it.

The Old Postfix Syntax (Deprecated)

Early versions of id_effect used a postfix tilde: expr ~. This is no longer valid. Always use the prefix form:

#![allow(unused)]
fn main() {
// OLD — do not use
step_a() ~;

// GOOD
~ step_a();
let x = ~ step_b();
}

If you see postfix tilde in older code, update it to the prefix form.

Error Handling Inside effect!

The ~ operator short-circuits on failure — if a bound effect fails, the whole effect! block fails with that error. But you can also handle errors within the block.

The Default: Short-Circuit

#![allow(unused)]
fn main() {
effect! {
    let a = ~ step_a();    // if this fails → whole block fails
    let b = ~ step_b(a);   // if this fails → whole block fails
    b
}
}

This matches ? in Result. You get clean sequencing at the cost of aborting early. For most code, that's exactly what you want.

Catching Errors Mid-block

To handle an error inline and continue, use .catch before the ~:

#![allow(unused)]
fn main() {
effect! {
    let user = ~ fetch_user(id).catch(|_| succeed(User::anonymous()));
    // If fetch_user fails, we get User::anonymous() and continue
    render_user(user)
}
}

.catch converts a failure into a success (or a different effect). The ~ then sees a successful effect.

Converting Errors with map_error

Often you have multiple effect types with different E parameters and need to unify them:

#![allow(unused)]
fn main() {
#[derive(Debug)]
enum AppError {
    Db(DbError),
    Network(HttpError),
}

effect! {
    let user = ~ fetch_user(id).map_error(AppError::Db);
    let data = ~ fetch_external_data(user.id).map_error(AppError::Network);
    process(user, data)
}
}

Both effects are converted to the same AppError before binding. The block's E parameter is AppError throughout.

Handling Errors with fold

fold handles both success and failure paths:

#![allow(unused)]
fn main() {
effect! {
    let outcome = ~ risky_operation().fold(
        |err| format!("Error: {err}"),
        |val| format!("Success: {val}"),
    );
    // outcome is always Ok(String), never fails here
    log_outcome(outcome)
}
}

fold is like pattern matching on the effect — you handle both arms and produce a uniform success value.

Re-raising Errors

Inside a .catch handler, you can inspect the error and decide whether to recover or re-fail:

#![allow(unused)]
fn main() {
effect! {
    let result = ~ db_operation().catch(|error| {
        if error.is_transient() {
            // Transient: retry once with a fallback
            fallback_db_operation()
        } else {
            // Permanent: re-raise
            fail(error)
        }
    });
    result
}
}

fail(error) inside a handler produces a failing effect — the outer ~ then propagates it.

Accumulating Multiple Errors

Short-circuit stops at the first error. When you need all errors (like form validation), use validate_all outside the macro:

#![allow(unused)]
fn main() {
// Not inside effect! — runs all regardless of failures
let results = validate_all(vec![
    validate_name(&input.name),
    validate_email(&input.email),
    validate_age(input.age),
]);

// results is Effect<Vec<Ok>, Vec<Err>, ()>
}

Chapter 8 covers validate_all and error accumulation patterns in detail.

The Rule of Thumb

When Not to Use the Macro

effect! is the idiomatic choice for most multi-step computations. But it's a macro — which means it has edges. Knowing when to reach for raw flat_map instead saves debugging time.

Use Raw flat_map for Single-Step Transforms

When there's exactly one effectful step and you're transforming its result, flat_map is cleaner:

#![allow(unused)]
fn main() {
// Unnecessarily verbose
effect! {
    let id = ~ parse_id(raw);
    id
}

// Clear and direct
parse_id(raw).flat_map(|id| succeed(id))
// or just:
parse_id(raw)
}

Use effect! when you have two or more sequential steps. For one, flat_map or .map is usually enough.

Use Combinators for Structural Patterns

Some patterns have named combinators that are more expressive than macros:

#![allow(unused)]
fn main() {
// Instead of:
effect! {
    let a = ~ step_a();
    let b = ~ step_b();
    (a, b)
}

// Consider (when steps are independent):
step_a().zip(step_b())
}

zip communicates intent: "I need both, in any order." The effect! version implies sequential dependency. For independent steps, prefer explicit combinators. (For concurrent independent steps, see fiber_all in Chapter 9.)

Avoid Deep Nesting Within the Block

The macro eliminates nesting between flat_map chains. But you can still create nested effect! blocks, which gets confusing:

#![allow(unused)]
fn main() {
// CONFUSING — nested macro bodies
effect! {
    let result = ~ effect! {      // inner macro
        let x = ~ inner_step();
        x * 2
    };
    result + 1
}

// BETTER — flatten it
effect! {
    let x = ~ inner_step();
    let result = x * 2;
    result + 1
}
}

If you feel the urge to nest effect! inside effect!, flatten the outer block instead.

The Macro and Type Inference

The macro occasionally confuses the type inferencer, especially when the error type isn't pinned early. If you see cryptic "can't infer type" errors inside effect!:

1. Annotate the return type of the enclosing function explicitly
2. Add a .map_error(Into::into) on the first ~ binding to anchor E
3. As a last resort, break out the inner logic into a named helper function

When Generic Returns Are Needed

Library code with polymorphic A, E, R sometimes can't use the macro cleanly:

#![allow(unused)]
fn main() {
// This works fine with explicit function + effect!
pub fn load_config<A, E, R>() -> Effect<A, E, R>
where
    A: From<Config> + 'static,
    E: From<ConfigError> + 'static,
    R: 'static,
{
    effect!(|_r: &mut R| {
        let cfg = read_env_config()?;
        A::from(cfg)
    })
}
}

The closure form of effect! (with |_r: &mut R|) is the right tool for generic graph-builder functions. It's still the macro, just in its raw form.

Summary

The macro is a tool, not a religion. Use it when it makes the code read like a story; use combinators when they express intent more directly.

The R Parameter — Your Dependencies, Encoded in Types

Chapter 1 introduced R as "what an effect needs to run." We kept it vague on purpose — you needed to understand effects before worrying about their environment.

For effects that use dependencies, R is written with caps!: a compile-time list of capability services. Pure effects use R = ().

What R means

#![allow(unused)]
fn main() {
use id_effect::{Effect, caps, effect, require};

fn get_user(id: u64) -> Effect<User, DbError, caps!(Database)> {
    effect!(|r| {
        let db = ~Database;
        Ok(db.fetch_user(id))
    })
}
}

Implicit |r|

When the enclosing function already returns Effect<_, _, caps!(…)> , you can write effect!(|r| { … }) and omit the environment type on r. Rust infers &mut caps!(…) from the return type. Use an explicit |r: &mut caps!(…)| when you want the macro to validate that body keys (~Key, require!(Key)) match that list.

Library code can stay generic over any R that exposes the same keys:

#![allow(unused)]
fn main() {
fn get_user<R>(id: u64) -> Effect<User, DbError, R>
where
    R: id_effect::Needs<Database> + 'static,
{
    effect!(|r: &mut R| {
        let db = ~Database;
        Ok(db.fetch_user(id))
    })
}
}

R is a promise to the compiler: this effect may only run where Database is available. caps! documents which keys the effect touches; at runtime run_with builds an Env that satisfies them.

How requirements are satisfied

Library code does not wire dependencies. Provide at the program edge with run_with:

#![allow(unused)]
fn main() {
use id_effect::{provide, run_with};

run_with(
    [provide!(DatabaseLive), provide!(LoggerLive)],
    get_user(42),
)?;
}

run_with builds a CapabilityGraph, constructs an Env, and runs the effect with run_blocking.

For tests you can skip the graph and build Env directly:

#![allow(unused)]
fn main() {
use id_effect::{Env, caps, run_blocking};

let mut env = Env::new();
env.insert::<Cap<Database>>(mock_db);
run_blocking(get_user(42), caps!(Database)::from_env(env))?;
}

Or use build_env when you still want provider types but not a full app run.

The next sections cover how R flows through composition, how to wire dependencies at the edge, and how capability services replace positional tuples.

R Revisited — More Than Just a Type Parameter

You've seen R in function signatures:

#![allow(unused)]
fn main() {
fn get_user(id: u64) -> Effect<User, DbError, caps!(Database)>
}

It looks like "this needs Database." But what does that mean precisely?

R as a contract

R is a promise to the compiler. When you write:

#![allow(unused)]
fn main() {
fn get_user(id: u64) -> Effect<User, DbError, caps!(Database)> {
    effect!(|r| {
        let db = ~Database;
        Ok(db.fetch_user(id))
    })
}
}

You are declaring: "To run this effect, you must supply Database in the environment." The compiler holds you to that promise. You cannot call run_with without a provider for Database.

#![allow(unused)]
fn main() {
// Missing DatabaseLive in the provider list → runtime CapabilityError at run_with
// run_with([], get_user(1))?;

// Correct — graph builds Database before the effect runs
run_with([provide!(DatabaseLive)], get_user(1))?;
}

The contract is not a comment. It is enforced by caps!(…) on the effect type and by run_with at the edge.

R flows through composition

When you combine effects with effect!, their capability requirements merge:

#![allow(unused)]
fn main() {
fn get_user(id: u64) -> Effect<User, DbError, caps!(Database)> { ... }
fn get_posts(user_id: u64) -> Effect<Vec<Post>, DbError, caps!(Database)> { ... }

// Combined: still caps!(Database) — both needed the same key
fn get_user_with_posts(id: u64) -> Effect<(User, Vec<Post>), DbError, caps!(Database)> {
    effect!(|r| {
        let user = ~ get_user(id);
        let posts = ~ get_posts(user.id);
        (user, posts)
    })
}
}

When effects need different keys:

#![allow(unused)]
fn main() {
fn log(msg: &str) -> Effect<(), LogError, caps!(EffectLogger)> { ... }
fn get_user(id: u64) -> Effect<User, DbError, caps!(Database)> { ... }

// Combined: caps!(Database, EffectLogger) — needs BOTH
fn get_user_logged(id: u64) -> Effect<User, AppError, caps!(Database, EffectLogger)> {
    effect!(|r| {
        ~ log(&format!("Fetching user {id}")).map_error(AppError::Log);
        let user = ~ get_user(id).map_error(AppError::Db);
        user
    })
}
}

The composed effect's caps!(…) list is the union of what each step needs. You wire every key once at main or in tests:

#![allow(unused)]
fn main() {
run_with(
    [provide!(DatabaseLive), provide!(LoggerLive)],
    get_user_logged(42),
)?;
}

Multiple requirements

As functions grow, they naturally accumulate keys:

#![allow(unused)]
fn main() {
fn process_order(order: Order) -> Effect<
    Receipt,
    AppError,
    caps!(Database, PaymentGateway, EmailService, EffectLogger),
> {
    effect!(|r| {
        ~ log("Processing order").map_error(AppError::Log);
        let user = ~ get_user(order.user_id).map_error(AppError::Db);
        let payment = ~ charge(order.total).map_error(AppError::Payment);
        ~ send_confirmation(&user.email).map_error(AppError::Email);
        Receipt::new(payment)
    })
}
}

Just from the type signature you know this function touches four capability services. No need to read the implementation.

Why R instead of parameters?

Traditional Rust would thread dependencies as function parameters:

#![allow(unused)]
fn main() {
fn process_order(
    order: Order,
    db: &Database,
    pay: &PaymentGateway,
    email: &EmailService,
    log: &Logger,
) -> Result<Receipt, AppError> { ... }
}

That works, but it forces every layer of your call stack to accept and forward dependencies it may not directly use. The R parameter encodes the same information in the return type — and caps!(…) names each dependency so two services of the same Rust type remain distinct.

Foreshadowing

You may be wondering: how does the runtime store Database and EffectLogger in one place?

Env is an order-independent map keyed by capability identity — not a positional tuple. Chapter 5 shows how `` generates each *Key type. For now: R = caps!(…) is the compile-time list; Env is the runtime container.

Providing Dependencies — run_with and Env

An effect with R = caps!(…) cannot run until its capabilities exist. Provide at the edge, not inside library code.

run_with — the main entrypoint

use id_effect::{Effect, ProviderSpecDerive, caps, effect, provide, require, run_with, succeed};
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Counter(pub u32);

#[derive(ProviderSpecDerive)]
#[provides(Counter)]
struct CounterLive;

impl CounterLive {
    fn new() -> Counter {
        Counter(42)
    }
}

fn app() -> Effect<u32, (), caps!(Counter)> {
    effect!(|r| {
        let counter = ~Counter;
        counter.0
    })
}

fn main() {
    let n = run_with([provide!(CounterLive)], app()).expect("run");
}

provide! wraps a ProviderSpec as a ProviderBox. run_with collects providers, plans build order via CapabilityGraph, and runs the effect.

Multiple providers

Pass every provider the app needs in one list — order in the array does not matter; the graph topologically sorts by requires():

#![allow(unused)]
fn main() {
run_with(
    [
        provide!(ConfigLive),
        provide!(DatabaseLive),
        provide!(LoggerLive),
        provide!(UserRepoLive),
    ],
    my_application(),
)?;
}

Manual Env for tests

When you only need a handful of values, build Env by hand:

#![allow(unused)]
fn main() {
let mut env = Env::new();
env.insert::<Cap<Database>>(mock_pool);
env.insert::<Cap<EffectLogger>>(test_logger);

let user = run_blocking(get_user(42), env)?;
}

build_env is the middle ground — same provider types as production, but you get the Env back without running an effect:

#![allow(unused)]
fn main() {
let env = build_env([provide!(MockUserRepoLive)])?;
run_test(get_user(1), env)?;
}

Where to provide

Provide at main, test setup, or HTTP/Tokio boundaries — not inside library functions.

#![allow(unused)]
fn main() {
// BAD — library reaches for concrete deps
pub fn process_order(order: Order) -> Effect<Receipt, AppError, ()> {
    let db = connect("hardcoded-url");
    // ...
}

// GOOD — library declares needs; caller wires them
pub fn process_order(order: Order) -> Effect<Receipt, AppError, caps!(Database, EffectLogger)> {
    // ...
}
}

Summary

None of these execute effect steps until run_with / run_blocking / run_async is called — wiring stays lazy until the boundary.

Widening and Narrowing — Environment Transformations

Sometimes your effect needs part of an environment, but you have the whole thing. Or you need to thread an effect through a context that provides more than required. This is where zoom_env, contramap_env, and capability subtyping come in.

The Mismatch Problem

Imagine your application needs several capabilities:

#![allow(unused)]
fn main() {
// Effect needs only EffectLogger
fn log_event(msg: &str) -> Effect<(), LogError, caps!(EffectLogger)> { ... }

// Caller has Database + EffectLogger + Config
fn process(data: Data) -> Effect<(), AppError, caps!(Database, EffectLogger, Config)> { ... }
}

You can't call log_event inside process without adapting the environment — the R types don't match. You need to narrow or widen.

zoom_env: Narrow the Environment

zoom_env adapts an effect to work with a larger environment by providing a lens from the larger type to the smaller one:

#![allow(unused)]
fn main() {
// Adapt log_event to work with a struct holding logger + other fields
let app_log = log_event("hello").zoom_env(|env: &AppEnv| &env.logger);
}

Now app_log has type Effect<(), LogError, AppEnv>. The function extracts the Logger from AppEnv and feeds it to the original effect.

Inside effect!, the pattern looks like:

#![allow(unused)]
fn main() {
fn process(data: Data) -> Effect<(), AppError, caps!(Database, EffectLogger, Config)> {
    effect!(|r| {
        ~ log_event("start").zoom_env(|e| extract_logger(e)).map_error(AppError::Log);
        ~ db_query(data).zoom_env(|e| extract_db(e)).map_error(AppError::Db);
        Ok(())
    })
}
}

contramap_env: Transform the Environment

While zoom_env narrows, contramap_env transforms. It applies a function to convert whatever environment the caller provides into what the effect actually needs:

#![allow(unused)]
fn main() {
// Effect needs a raw string URL
fn connect(url: &str) -> Effect<Database, DbError, String> { ... }

// You have a Config that contains the URL
let with_config = connect_raw.contramap_env(|cfg: &Config| cfg.db_url.clone());
// Now type is Effect<Database, DbError, Config>
}

contramap_env is the formal name for "adapt the environment type." In practice, most code uses zoom_env for the common case of extracting a field.

caps! and automatic subtyping

For capability DI, prefer caps! and the ~ bind operator inside effect!. A wider runtime environment satisfies a narrower R: every CapList shares one Env; cap_into_bind clones that env and verifies the inner keys (see ADR 0005).

#![allow(unused)]
fn main() {
use id_effect::{Effect, caps, effect, run_with, provide};

fn query(id: u64) -> Effect<User, DbError, caps!(Database)> {
    effect!(|r| {
        let db = ~Database;
        db.fetch_user(id)
    })
}

fn log_event(msg: &str) -> Effect<(), LogError, caps!(EffectLogger)> {
    effect!(|r| {
        let log = ~EffectLogger;
        log.info(msg);
    })
}

fn app() -> Effect<User, AppError, caps!(Database, EffectLogger)> {
    effect!(|r| {
        ~log_event("start");
        ~query(42)
    })
}

run_with(
    [provide!(DatabaseLive), provide!(LoggerLive)],
    app(),
)?;
}

Automatic binding with ~: when the inner effect needs any single key from the outer caps!(…) list, effect! expands ~inner(...) to cap_into_bind — no manual projection.

Capability lookup: ~Database inside effect! borrows that capability from r (same as require!(Database)).

Implicit |r|: write effect!(|r| { … }); Rust infers &mut caps!(…) from the enclosing function's Effect<_, _, caps!(…)> return type. You still declare caps!(…) once on the signature.

Outside effect!, narrow explicitly with CapWiden::widen or project_at_*:

#![allow(unused)]
fn main() {
let wide: caps!(Database, EffectLogger) = /* from build_env or run_with */;
let narrow: caps!(Database) = wide.widen();
run_blocking(query(1), narrow)?;
}

When an inner function declares caps!(Database) and the caller holds caps!(Database, EffectLogger), the compiler checks that every required key is present — no positional tuple indexing, no runtime downcasts.

R as Documentation Revisited

These combinators highlight why R is valuable as documentation. When you see:

#![allow(unused)]
fn main() {
fn log_event(msg: &str) -> Effect<(), LogError, caps!(EffectLogger)>
}

You know exactly what this function needs. You don't need to read its body to see if it also touches the database. Adaptation at the call site stays explicit.

Compare to the pre-effect alternative:

#![allow(unused)]
fn main() {
// Traditional: you'd need to read the body to know what `env` is used for
fn log_event(env: &AppEnv, msg: &str) -> Result<(), LogError> { ... }
}

With R, the function declares what it needs. With zoom_env or CapWiden, the caller declares how to satisfy it.

When to Use These

In practice, zoom_env and contramap_env appear most often in library code — when writing reusable utilities that should work with any environment containing the right piece. Application code typically uses caps! / CapWiden and named capability services (Chapters 5–6), which avoid the need for explicit projection.

Think of zoom_env as the manual fallback when automatic capability subtyping isn't the right fit.

R as Documentation — Self-Describing Functions

The R parameter is often described as "the environment type." That's true, but it undersells the practical benefit. R is living documentation that the compiler enforces.

The Signature Tells the Story

Consider two versions of the same function:

#![allow(unused)]
fn main() {
// Version A: traditional async
async fn process_order(order: Order) -> Result<Receipt, Error> {
    // What does this use? Read the body to find out.
    // Database? PaymentGateway? Email? Metrics?
    // You'll have to trace through 200 lines to know.
}

// Version B: effect-based
fn process_order(order: Order) -> Effect<
    Receipt,
    OrderError,
    caps!(Database, PaymentGateway, EmailService, EffectLogger),
> {
    // What does this use? Look at the signature.
    // Database ✓, PaymentGateway ✓, EmailService ✓, Logger ✓
    // Done.
}
}

Version B's type is self-describing. You don't need to read the implementation to understand its dependency surface.

Code Review Benefits

In a pull request, R changes are visible in the diff. If someone adds a call to send_metrics() inside process_order and MetricsClient wasn't previously in R, the function signature must change:

- fn process_order(order: Order) -> Effect<Receipt, OrderError, caps!(Database, PaymentGateway, EmailService, EffectLogger)>
+ fn process_order(order: Order) -> Effect<Receipt, OrderError, caps!(Database, PaymentGateway, EmailService, EffectLogger, MetricsClient)>

This diff is in the function signature — impossible to miss. With traditional parameters or singletons, new dependencies can silently appear in implementation bodies.

Refactoring Safety

When you refactor and remove a dependency, the compiler finds all the places that were providing the now-unnecessary value. The R type shrinks, and provider lists that still register removed keys become visible during review.

#![allow(unused)]
fn main() {
// After removing EffectLogger from process_order's caps! list:

// Callers must drop LoggerLive from run_with when nothing in the graph needs it
run_with(
    [provide!(DatabaseLive)], // LoggerLive no longer required by this effect
    process_order(order),
)?;
}

The compiler guides you to clean up wiring. Traditional code leaves stale dependencies silently lingering.

Testing Clarity

When writing a test, R tells you exactly what you need to mock:

#![allow(unused)]
fn main() {
#[test]
fn test_process_order() {
    // R = caps!(Database, PaymentGateway, EmailService, EffectLogger)
    // So the test needs these four keys — no more, no less
    let mut env = Env::new();
    env.insert::<Cap<Database>>(mock_db());
    env.insert::<Cap<PaymentGateway>>(mock_payment());
    env.insert::<Cap<EmailService>>(mock_email());
    env.insert::<Cap<EffectLogger>>(test_logger());

    let result = run_test(process_order(test_order()), env);
    assert!(result.is_ok());
}
}

There's no "I wonder if this also touches the metrics service" uncertainty. The type says it doesn't. If you're missing a mock, the code won't compile or require! fails at runtime.

R is Not Magic

It's important to understand that R is just a type parameter. The "compile-time DI" property comes from:

1. Functions declaring what they need in R (usually via caps!)
2. Capability services identifying services in Env
3. Composition automatically merging requirements
4. Wiring centralized at run_with / build_env boundaries

There's no reflection, no registration, no framework. Just types.

The next chapter shows how capability services and Env make this scale beyond simple capability lists — handling large, complex dependency graphs without positional ambiguity.

Capability services and Env — Compile-Time Service Lookup

Chapter 4 showed how R encodes dependencies. For small programs a single service in caps!(T) is enough. As the graph grows you need named capability services so the compiler can distinguish dependencies — even when they share the same Rust type.

This chapter covers:

• Why positional/tuple R breaks down
• `` — declaring key types
• Env — the order-independent runtime container
• Needs<K> and ~Key — accessing services inside effects

By the end you'll know how capability lookup works and why insertion order in Env never matters.

The Problem with Positional Types

Early effect code sometimes used tuples as R. That works briefly, then becomes fragile.

The tuple explosion

Two dependencies: readable.

#![allow(unused)]
fn main() {
Effect<A, E, (Database, Logger)>
}

Five dependencies: which is which?

#![allow(unused)]
fn main() {
Effect<A, E, (Pool, Pool, Logger, Config, HttpClient)>
//            ^^^^ two Pools — main DB or cache?
}

Tuples are positional. (Pool, Pool, …) is ambiguous when both fields share a type.

Fragility under change

#![allow(unused)]
fn main() {
fn foo() -> Effect<A, E, (Database, Logger)>
//                        0         1
}

A teammate inserts Config:

#![allow(unused)]
fn main() {
fn foo() -> Effect<A, E, (Database, Config, Logger)>
//                        0         1       2
}

Every caller that built (db, log) must become (db, config, log). The type system does not point at stale indices — it's a silent refactor hazard.

Same-type collision

Rust cannot distinguish a main-database Pool from a cache Pool in a tuple:

#![allow(unused)]
fn main() {
run_blocking(effect, (cache_pool, main_pool)); // compiles, wrong at runtime
}

What we need

Each dependency needs a compile-time name independent of position:

• Database → the primary Pool
• Cache → the cache Pool

Different keys, same underlying type — the compiler catches swaps.

That's what `` generates.

Capability services

A capability service is a Rust type that names a dependency in Env. The Cap<T> wrapper implements CapabilityKey for any cloneable T, so you use the service type directly in caps!, require!, and #[provides].

Declaring a service

#![allow(unused)]
fn main() {
// Concrete value type
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Counter(pub u32);

// Trait-backed service (typical for ports/adapters)
pub type Database = Arc<dyn DbClient>;
pub type UserRepo = Arc<dyn UserRepository>;
}

Database and Cache can both wrap a Pool but remain distinct capability identities — you cannot pass one where the other is required.

Registering values

At runtime, values live in Env:

#![allow(unused)]
fn main() {
use id_effect::{Cap, Env};

let mut env = Env::new();
env.insert::<Cap<Database>>(main_pool);
env.insert::<Cap<Cache>>(cache_pool);
}

Or let a ProviderSpec insert them during run_with.

Why named services help the compiler

#![allow(unused)]
fn main() {
fn needs_database() -> Effect<A, E, caps!(Database)> { ... }
fn needs_cache() -> Effect<A, E, caps!(Cache)> { ... }
}

Providing the wrong service is a type error at the call site, not a silent runtime swap.

Service traits

Define a focused trait, then a type alias for the handle stored in Env:

#![allow(unused)]
fn main() {
pub trait UserRepository: Send + Sync {
    fn get_user(&self, id: u64) -> Effect<User, DbError, ()>;
}
pub type UserRepo = Arc<dyn UserRepository>;
}

Trait methods keep R = () — the caller carries Database / UserRepo in its caps! list.

Summary

Named services eliminate the positional problem. The next section introduces Env — how those services are stored at runtime.

Env — The Runtime Capability Container

Multi-capability effects use Env at runtime: a map from capability identity to service value. Insertion order does not matter.

Structure

#![allow(unused)]
fn main() {
use id_effect::Env;

let mut env = Env::new();
env.insert::<Cap<Database>>(pool);
env.insert::<Cap<EffectLogger>>(logger);

assert!(env.has::<Cap<Database>>());
let pool = env.get::<Cap<Database>>();
}

Env stores cloneable, Send + Sync values keyed by CapabilityId (derived from the key type). Lookups are O(1); there is no positional indexing.

Building Env

Three common paths:

1. Application entry — providers + graph

#![allow(unused)]
fn main() {
run_with([provide!(ConfigLive), provide!(DatabaseLive)], app())?;
}

2. Providers only — reuse in tests

#![allow(unused)]
fn main() {
let env = build_env([provide!(MockDatabaseLive)])?;
}

3. Manual — fast unit tests

#![allow(unused)]
fn main() {
let mut env = Env::new();
env.insert::<Cap<Database>>(MockPool::new());
}

Why not a plain HashMap<TypeId, Box<dyn Any>>?

You could store dyn Any and downcast. Env + Capability keeps:

• Compile-time requirements via Needs<K> bounds and caps!
• Typed access — get::<Cap<Database>>() returns &Pool, not &dyn Any
• Stable diagnostics — missing capabilities produce CapabilityError::Missing with the key name

Application code should think in Env and capability services, not positional tuples.

Order independence

These two sequences produce equivalent lookup behaviour:

#![allow(unused)]
fn main() {
env.insert::<Cap<Database>>(db).insert::<Cap<EffectLogger>>(log);
// vs
env.insert::<Cap<EffectLogger>>(log).insert::<Cap<Database>>(db);
}

Adding a new capability never changes how existing keys are accessed — refactor-safe in a way tuples never were.

When you touch Env directly

• Test fixtures with one or two mocks
• Tokio/async examples that pass Env to run_async
• HTTP hosts that store State<Env> (see Axum host)

Production apps usually list provide!(…) values once at the top level and let CapabilityGraph assemble Env.

Needs and ~Key — Reading from Env

To use a capability inside an effect, declare required keys in R with caps! and borrow with ~Key, require!, or Needs::need.

Declaring requirements with caps!

Put the keys your effect needs in the third type parameter:

#![allow(unused)]
fn main() {
fn get_user(id: u64) -> Effect<User, DbError, caps!(Database)> { ... }
fn notify_user(id: u64, msg: &str) -> Effect<(), AppError, caps!(UserRepo, EmailService)> { ... }
}

Library helpers can stay generic when callers supply a wider Env:

#![allow(unused)]
fn main() {
fn get_user<R>(id: u64) -> Effect<User, DbError, R>
where
    R: id_effect::Needs<Database> + 'static,
{
    effect!(|r: &mut R| {
        let db = ~Database;
        db.fetch_user(id)
    })
}
}

~Key — capability lookup inside effect bodies

#![allow(unused)]
fn main() {
use id_effect::{effect, require, caps};

fn get_user(id: u64) -> Effect<User, DbError, caps!(Database)> {
    effect!(|r| {
        let db = ~Database;
        db.fetch_user(id)
    })
}
}

~Key inside effect! expands to a typed borrow from r. require!(K) is equivalent sugar.

If get_user requires Database but you run it with an empty Env, you get a runtime missing-capability error when the effect executes — not a silent None. For static verification, keep caps!(…) or Needs<K> bounds on public APIs so callers must wire providers before run_with.

When building tests manually:

#![allow(unused)]
fn main() {
let mut env = Env::new();
// forgot env.insert::<Cap<Database>>(...)
run_blocking(get_user(42), env); // panics on ~Key / get
}

Prefer build_env or typed test helpers so incomplete wiring fails at setup time.

Summary

An application that satisfies all Needs bounds at the edge and passes a complete provider list to run_with is an application where every dependency is explicit — no service-locator globals.

Providers — Building Your Dependency Graph

You've seen how R encodes what an effect needs and how Env holds values at runtime. But who builds the environment?

In small programs you can call Env::insert by hand. In real applications you declare providers: types with #[derive(ProviderSpecDerive)] and #[provides(Key)] that know how to construct each capability, optionally reading other capabilities from a partially-built Env.

Pass providers to run_with and CapabilityGraph topologically sorts them from each provider's requires() metadata. No manual ordering.

This chapter covers ProviderSpec, dependent providers, and composing provider lists for production and tests.

What Is a Provider?

An Effect describes a computation that needs an environment. A ProviderSpec describes how to build one capability and register it in Env.

Effect<User, DbError, caps!(Database)>
  └── "I need Database to produce a User"

ProviderSpec for DatabaseLive
  └── "I need Config (via requires) to produce Database"

Effects declare needs; providers declare construction.

The ProviderSpec trait

#![allow(unused)]
fn main() {
use id_effect::{CapabilityId, Env, ProviderError, ProviderSpec};

struct DatabaseLive;

impl ProviderSpec for DatabaseLive {
    type Key = Database;
    type Output = Pool;

    fn provider_id() -> &'static str { "database-live" }

    fn requires() -> &'static [CapabilityId] {
        // Declare Config as a dependency for graph ordering
        // (return a static slice of CapabilityId values)
        &[]
    }

    fn provide(deps: &Env) -> Result<Pool, ProviderError> {
        let config = deps.get::<Cap<Config>>();
        Ok(connect_pool(config.db_url()))
    }
}
}

• type Key — which capability this provider registers
• type Output — the concrete value stored in Env
• provide(deps) — build the value; read dependencies with deps.get::<Cap<K>>()
• requires() — capability ids this provider needs built first (used by CapabilityGraph)

Providers are lazy recipes

Implementing ProviderSpec does nothing by itself. Construction runs when you call run_with, build_env, or CapabilityGraph::build.

The key insight

• Effects are programs — what to do with capabilities
• Providers are constructors — how to build capabilities

provide!(DatabaseLive) registers the recipe; run_with executes the graph and passes the resulting Env to your effect.

Lifecycle and cleanup

Providers run synchronously at startup. For resources with async teardown, build handles in provide and register finalizers in effect Scope code (see Chapter 10). The provider graph itself focuses on construction order, not fiber lifetimes.

Building Providers — From Simple to Complex

Leaf providers — no dependencies

#![allow(unused)]
fn main() {
struct ConfigLive;

impl ProviderSpec for ConfigLive {
    type Key = Config;
    type Output = Config;

    fn provider_id() -> &'static str { "config-from-env" }

    fn provide(_deps: &Env) -> Result<Config, ProviderError> {
        Config::from_env().map_err(|e| ProviderError {
            provider: Self::provider_id(),
            message: e.to_string(),
        })
    }
}
}

Dependent providers

Read already-built capabilities from deps:

#![allow(unused)]
fn main() {
struct UserRepoLive;

impl ProviderSpec for UserRepoLive {
    type Key = UserRepo;
    type Output = Arc<dyn UserRepository>;

    fn provider_id() -> &'static str { "user-repo-postgres" }

    fn provide(deps: &Env) -> Result<Arc<dyn UserRepository>, ProviderError> {
        let pool = deps.get::<Cap<Database>>().clone();
        Ok(Arc::new(PostgresUserRepository { pool }))
    }
}
}

Override requires() to return [Database::id(), …] so the graph builds the database before the repository.

Test doubles

Same key, different provider type:

#![allow(unused)]
fn main() {
struct MockUserRepoLive;

impl ProviderSpec for MockUserRepoLive {
    type Key = UserRepo;
    type Output = Arc<dyn UserRepository>;

    fn provider_id() -> '&'static str { "user-repo-mock" }

    fn provide(_deps: &Env) -> Result<Arc<dyn UserRepository>, ProviderError> {
        Ok(Arc::new(MockUserRepository::default_fixture()))
    }
}
}

When a test needs custom data, use Env::insert instead of provide!. Business logic still uses Needs<UserRepo> — only the wiring at the edge changes.

Custom values

Workspace crates expose helpers that return ProviderBox directly — e.g. id_effect_config::provide_config_provider, id_effect_platform::http::reqwest::provide_reqwest_client. Use these when you already hold a concrete handle and don't need a zero-sized ProviderSpec type.

The pattern in practice

ConfigLive          (no deps)
  → DatabaseLive    (needs Config)
  → CacheLive       (needs Config)
  → UserRepoLive    (needs Database)

The next section shows how CapabilityGraph wires the list together.

Composing Providers — CapabilityGraph and run_with

Individual providers build one capability. Applications pass a list to run_with; CapabilityGraph plans build order from each provider's requires() and provides().

Basic composition

#![allow(unused)]
fn main() {
use id_effect::{provide, run_with};

run_with(
    [
        provide!(ConfigLive),
        provide!(DatabaseLive),
        provide!(LoggerLive),
        provide!(UserRepoLive),
    ],
    my_application(),
)?;
}

The array order is irrelevant — the graph topologically sorts providers. Cycles or missing dependencies surface as CapabilityPlannerError with diagnostics from CapabilityGraph::diagnostics.

Building Env without running

#![allow(unused)]
fn main() {
let env = build_env([
    provide!(ConfigLive),
    provide!(DatabaseLive),
    provide!(UserRepoLive),
])?;

run_test(get_user(1), env)?;
}

Or inspect the plan explicitly:

#![allow(unused)]
fn main() {
let mut graph = CapabilityGraph::new();
graph = graph.add(provide!(ConfigLive).0);
graph = graph.add(provide!(DatabaseLive).0);
let order = graph.plan()?;
let env = graph.build()?;
}

Production vs test stacks

#![allow(unused)]
fn main() {
// Production
run_with(
    [provide!(ConfigLive), provide!(DatabaseLive), provide!(UserRepoLive)],
    my_app(),
)?;

// Test — swap implementations, same keys
run_with([provide!(MockUserRepoLive)], get_user(1))?;
}

Application code is unchanged. Only the provider list differs. Needs<K> bounds ensure each effect's requirements are met.

Subset wiring in tests

Tests need not mirror the full production graph — only the capabilities the effect under test requires:

#![allow(unused)]
fn main() {
#[test]
fn test_get_user() {
    let env = build_env([provide!(MockUserRepoLive)]).unwrap();
    let user = run_test(get_user(1), env).unwrap();
    assert_eq!(user.name, "Alice");
}
}

If get_user also needs EffectLogger, the test must include provide!(TestLoggerLive) or insert that key manually — incomplete wiring fails when the effect runs.

Capability Graphs — Automatic Dependency Resolution

For small applications, passing a short provider list to run_with is enough. For larger apps, CapabilityGraph plans build order from each provider's requires() / provides() metadata and surfaces diagnostics via CapabilityGraph::diagnostics.

Declaring a provider graph

#![allow(unused)]
fn main() {
use id_effect::{CapabilityGraph, provide};

let graph = CapabilityGraph::new()
    .add(provide!(ConfigLive).0)
    .add(provide!(DatabaseLive).0)
    .add(provide!(CacheLive).0);
}

Each ProviderSpec declares dependencies via requires(); CapabilityGraph topologically sorts providers before calling build / build_from.

Planning and building

#![allow(unused)]
fn main() {
let order = graph.plan()?;
let env = graph.build()?;
}

The planner returns node indices in dependency order. Independent branches may appear in any stable topological order.

Cycle detection

graph.plan() returns an error if there are circular dependencies:

#![allow(unused)]
fn main() {
let diags = bad_graph.diagnostics();
assert!(!diags.is_empty()); // e.g. cycle-detected
}

Cycles and missing providers are detected at plan time via CapabilityPlannerError::to_diagnostic. Use cargo run -p id_effect_cli --bin id-effect-diagnose -- example cycle to print a sample report.

Conditional providers

Layers can be added conditionally:

#![allow(unused)]
fn main() {
let mut providers = vec![provide!(ConfigLive)];
if cfg!(feature = "metrics") {
    providers.push(provide!(MetricsLive));
}
let env = build_env(providers)?;
}

Feature flags and environment-based configuration compose naturally with the graph API.

When to use graphs vs hand-built Env

Services — The Complete DI Pattern

The previous chapters established the building blocks: capability services (identities), Env (the runtime container), and providers (constructors). Now we put them together into the complete Service pattern.

A Service in id_effect is the combination of:

1. A trait defining the interface
2. A capability service identifying it in the environment
3. One or more implementations (production and test)
4. A provider that wires an implementation into Env

This is the full dependency injection story. By the end of this chapter you'll have a working multi-service application wired entirely at compile time.

The sections Tokio bridge through Logging extend the same pattern to workspace integration crates: how you run effects on Tokio, swap platform and HTTP implementations, host handlers under Axum / Tower, and inject configuration and logging as services. Read them in order when wiring a production binary.

Service Traits — Defining Interfaces

The first step in defining a service is the trait — the contract between implementation and callers.

Define the interface

#![allow(unused)]
fn main() {
use id_effect::Effect;

pub trait UserRepository: Send + Sync {
    fn get_user(&self, id: u64) -> Effect<User, DbError, ()>;
    fn save_user(&self, user: &User) -> Effect<(), DbError, ()>;
}
}

Conventions:

• Methods return Effect<_, _, ()> — the service method itself has no extra environment; callers carry UserRepo in caps!.
• Send + Sync on the trait so handles like Arc<dyn UserRepository> work across fibers.
• Small, verb-oriented methods (get_user, not users).

Define the capability service

#![allow(unused)]
fn main() {
struct UserRepo;
}

This generates UserRepo: Capability with Value = Arc<dyn UserRepository>.

Use ~Key in callers

#![allow(unused)]
fn main() {
use id_effect::{effect, require, caps, succeed};

fn get_user_profile(id: u64) -> Effect<UserProfile, AppError, caps!(UserRepo)> {
    effect!(|r| {
        let repo = ~UserRepo;
        let user = ~ repo.get_user(id).map_error(AppError::Db);
        UserProfile::from(user)
    })
}
}

Keep traits focused

#![allow(unused)]
fn main() {
// BAD — one god trait
trait AppService {
    fn get_user(&self, id: u64) -> Effect<User, AppError, ()>;
    fn send_email(&self, to: &str, body: &str) -> Effect<(), AppError, ()>;
}

// GOOD — separate capabilities
trait UserRepository { /* … */ }
trait EmailService { /* … */ }
struct UserRepo;
struct Email;
}

Functions declare exactly what they need: caps!(UserRepo, Email).

Accessing Services — Needs and ~Key

Application effects access services with Needs<K> bounds, caps! in R, and ~Key (or require!) inside effect!.

Single service

#![allow(unused)]
fn main() {
use id_effect::{effect, require, caps, succeed};

fn get_user(id: u64) -> Effect<User, DbError, caps!(UserRepo)> {
    effect!(|r| {
        let repo = ~UserRepo;
        ~ repo.get_user(id)
    })
}
}

Generic over any environment that implements the bound:

#![allow(unused)]
fn main() {
fn get_user<R>(id: u64) -> Effect<User, DbError, R>
where
    R: Needs<UserRepo> + 'static,
{
    effect!(|r: &mut R| {
        let repo = ~UserRepo;
        ~ repo.get_user(id)
    })
}
}

Multiple services

#![allow(unused)]
fn main() {
fn notify_user(id: u64, message: &str) -> Effect<(), AppError, caps!(UserRepo, Email)> {
    effect!(|r| {
        let repo = ~UserRepo;
        let email = ~Email;
        let user = ~ repo.get_user(id).map_error(AppError::Db);
        ~ email.send(&user.email, message).map_error(AppError::Email);
        ()
    })
}
}

Direct Env access

Prefer effect! + ~Config in application code. For small sync helpers outside effect!, Needs::<Config>::need(env) is available.

caps! vs generic R

All styles run against the same Env built by run_with or build_env.

Providing Services — ProviderSpec Implementations

You have a trait and an implementation. A provider wires the impl into Env.

Production provider

#![allow(unused)]
fn main() {
use id_effect::{Env, ProviderSpecDerive, caps, effect, provide, run_with};
use std::sync::Arc;

struct PostgresUserRepository { pool: Pool }

impl UserRepository for PostgresUserRepository {
    fn get_user(&self, id: u64) -> Effect<User, DbError, ()> {
        effect! { /* query via self.pool */ }
    }
}

#[derive(ProviderSpecDerive)]
#[provides(UserRepo)]
struct UserRepoLive;

impl UserRepoLive {
    fn new(deps: &Env) -> Arc<dyn UserRepository> {
        let pool = deps.get::<Cap<Database>>().clone();
        Arc::new(PostgresUserRepository { pool })
    }
}
}

Register alongside dependencies:

#![allow(unused)]
fn main() {
run_with(
    [provide!(ConfigLive), provide!(DatabaseLive), provide!(UserRepoLive)],
    get_user_profile(42),
)?;
}

Mock provider

provide! takes a type (provide!(MockUserRepoLive)), not a value. For custom fixture data, insert into Env directly:

#![allow(unused)]
fn main() {
let mut env = Env::new();
env.insert::<Cap<UserRepo>>(Arc::new(MockUserRepository { users: test_data() }));
run_blocking(get_user(1), env)?;
}

For a fixed zero-config mock, use a unit struct:

#![allow(unused)]
fn main() {
#[derive(ProviderSpecDerive)]
#[provides(UserRepo)]
struct MockUserRepoLive;

impl MockUserRepoLive {
    fn new() -> Arc<dyn UserRepository> {
        Arc::new(MockUserRepository::default_fixture())
    }
}
}

Same UserRepo, no real database — swap at the edge:

#![allow(unused)]
fn main() {
// Production
run_with([provide!(DatabaseLive), provide!(UserRepoLive)], app())?;

// Test (fixed fixture)
run_with([provide!(MockUserRepoLive)], get_user(1))?;
}

Application code using caps!(UserRepo) and ~UserRepo stays identical.

A Complete DI Example — Putting It All Together

A small blog API with three services, a provider graph, and production vs test wiring.

Domain

#![allow(unused)]
fn main() {
struct User { id: u64, name: String, email: String }
struct Post { id: u64, author_id: u64, title: String, body: String }
enum AppError { Db(DbError), Notify(NotifyError) }
}

Three service traits + keys

#![allow(unused)]
fn main() {
use id_effect::Effect;
use std::sync::Arc;

pub trait UserRepository: Send + Sync {
    fn get_user(&self, id: u64) -> Effect<User, DbError, ()>;
}

pub trait PostRepository: Send + Sync {
    fn get_posts_by_author(&self, author_id: u64) -> Effect<Vec<Post>, DbError, ()>;
}

pub trait NotificationService: Send + Sync {
    fn send_welcome(&self, to: &str) -> Effect<(), NotifyError, ()>;
}
struct UserRepo;
struct PostRepo;
struct Notifier;
}

Business logic

#![allow(unused)]
fn main() {
use id_effect::{effect, require, caps, succeed};

fn get_author_feed(author_id: u64) -> Effect<(User, Vec<Post>), AppError, caps!(UserRepo, PostRepo)> {
    effect!(|r| {
        let user_repo = ~UserRepo;
        let post_repo = ~PostRepo;
        let user  = ~ user_repo.get_user(author_id).map_error(AppError::Db);
        let posts = ~ post_repo.get_posts_by_author(author_id).map_error(AppError::Db);
        (user, posts)
    })
}

fn register_user(name: &str, email: &str) -> Effect<User, AppError, caps!(UserRepo, Notifier)> {
    effect!(|r| {
        let repo = ~UserRepo;
        let notifier = ~Notifier;
        let user = ~ repo.create_user(name, email).map_error(AppError::Db);
        ~ notifier.send_welcome(&user.email).map_error(AppError::Notify);
        user
    })
}
}

Production wiring

use id_effect::{provide, run_with};

fn main() {
    run_with(
        [
            provide!(ConfigLive),
            provide!(DatabaseLive),
            provide!(PgUserRepoLive),
            provide!(PgPostRepoLive),
            provide!(SmtpNotifierLive),
        ],
        get_author_feed(1),
    )
    .expect("app failed");
}

CapabilityGraph ensures DatabaseLive runs before repo providers that read Database from Env.

Test wiring

#![allow(unused)]
fn main() {
#[test]
fn feed_includes_authors_posts() {
    let mut env = Env::new();
    env.insert::<Cap<UserRepo>>(Arc::new(mock_user_repo(&[alice(), bob()])));
    env.insert::<Cap<PostRepo>>(Arc::new(mock_post_repo(&[alice_post()])));

    let (_user, posts) = run_test(get_author_feed(1), env).unwrap();
    assert_eq!(posts.len(), 1);
    assert_eq!(posts[0].title, "Alice's Post");
}
}

What this demonstrates

• Business logic declares caps!(…) and uses ~Key — no Postgres, SMTP, or concrete types in domain code.
• Providers swap at the edge via provide!(…).
• The dependency graph is explicit in provider requires() + the effect's capability list.

That's compile-time dependency injection: requirements are typed; wiring is centralized at main and in tests.

Tokio bridge (id_effect_tokio)

The core id_effect crate defines interpreters (run_blocking, run_async, run_fork, …) and the Runtime trait. Workspace crate id_effect_tokio supplies the Tokio-backed implementation for binaries on #[tokio::main].

Read this section before Platform I/O, HTTP via reqwest, and Axum.

What id_effect_tokio provides

• TokioRuntime — implements [id_effect::Runtime]: cooperative sleep/yield; forked fibers on Tokio's blocking pool.
• Re-exports — run_async, run_blocking, run_fork, yield_now for use at async boundaries.
• spawn_blocking_run_async — when the effect graph is not Send but must be driven by run_async (Axum uses the same pattern).

Capability DI with Tokio

Build Env manually or via run_with / build_env, then pass it to run_async:

use id_effect::{Env, caps, effect, require, run_async, succeed};
struct ApiToken;

fn fetch() -> Effect<Vec<Quote>, AppError, caps!(ApiToken)> {
    effect!(|r| {
        let token = ~ApiToken;
        // async steps…
        Ok(quotes)
    })
}

#[tokio::main]
async fn main() {
    let mut env = Env::new();
    env.insert::<Cap<ApiToken>>("secret");
    let quotes = run_async(fetch(), env).await?;
}

For provider-based apps, use build_env + run_async:

#![allow(unused)]
fn main() {
let env = build_env([provide!(ConfigLive), provide!(HttpClientLive)])?;
let res = run_async(my_handler(), env).await?;
}

Mental model

Sharp edges

1. Send: run_async futures are often not Send — use Axum/Tower adapters or spawn_blocking_run_async.
2. Router tests: prefer #[tokio::test(flavor = "multi_thread", worker_threads = 2)] when exercising Axum + effects.

Further reading

• Example: crates/id_effect_tokio/examples/109_tokio_end_to_end.rs
• cargo doc --open -p id_effect_tokio

Platform I/O (id_effect_platform)

Workspace crate id_effect_platform mirrors Effect.ts @effect/platform: HTTP, filesystem, and process as typed capabilities in Env, not ad hoc reqwest / std::fs calls in domain code.

Why separate from id_effect?

• Ports, not drivers — traits like HttpClient, FileSystem, ProcessRuntime describe what you need; provider impls install live or test doubles.
• Test doubles — TestFileSystem is in-memory; production uses LiveFileSystemProvider.
• Stable HTTP boundary — domain code depends on the HttpClient trait + execute, not reqwest::RequestBuilder (the capability service is crate-private).

Modules

Wiring pattern

Each module declares a capability service and ships a default provider:

#![allow(unused)]
fn main() {
// in id_effect_platform::http (simplified; HttpClientService is pub(crate))
#[derive(ProviderSpecDerive)]
#[provides(HttpClientService)]
pub struct ReqwestHttpClientProvider;

impl ReqwestHttpClientProvider {
    fn new() -> Arc<dyn HttpClient> {
        Arc::new(ReqwestHttpClient::default_client())
    }
}
}

Application entry:

#![allow(unused)]
fn main() {
use id_effect::{run_with, RunError};
use id_effect_platform::http::{HttpRequest, execute, provide_reqwest_http_client};

let res = run_with(
    [provide_reqwest_http_client()],
    execute(HttpRequest::get("https://example.com")),
)
.map_err(|e| match e {
    RunError::Effect(e) => e,
    e => panic!("planner: {e}"),
})?;
}

Effects returned by execute carry the correct Needs bound internally; application code usually calls run_with([provide_reqwest_http_client()], execute(req)) without naming the key.

Drive async platform effects with id_effect::run_async (see Tokio bridge).

Security note (filesystem)

TestFileSystem rejects paths containing ... Live I/O follows OS semantics — sandbox untrusted paths at a higher layer.

Runnable example

cargo run -p id_effect_platform --example 010_platform_http_get

Next

For reqwest-specific pools and JSON helpers, see HTTP via reqwest.

HTTP via reqwest (id_effect_platform::http::reqwest)

The id_effect_platform::http::reqwest module integrates reqwest::Client with the effect environment: the client lives in R behind ReqwestClient, and HTTP work is expressed as Effect values (send, text, bytes, json, …).

Relation to portable HTTP

Prefer id_effect_platform::http for new application boundaries that should stay stack-agnostic. Use id_effect_platform::http::reqwest when you depend on RequestBuilder pipelines, need pooled clients (provide_reqwest_pool, send_pooled), or want json_schema so parse failures carry field paths (id_effect::schema::ParseError).

Import reqwest types (Client, Error, …) from the reqwest crate; import helpers from id_effect_platform::http::reqwest. Drive async steps with id_effect::run_async per Tokio bridge.

Core pieces

• ReqwestClient — tag for the active reqwest::Client in R.
• provide_reqwest_client / ReqwestClientLive — construct the client at the composition root.
• send, text, bytes, json, json_schema — effectful helpers over RequestBuilder.
• Optional pools — TTL pools of clients for connection churn scenarios.

Further reading

• Crate-level docs: cargo doc --open -p id_effect_platform
• Migrating from async for the general async fn → Effect move; combine with this module at HTTP boundaries when you choose the reqwest adapter.

Axum host (id_effect_axum)

Workspace crate id_effect_axum runs Effect<A, E, R> programs inside Axum handlers on the same Tokio runtime as #[tokio::main] / axum::serve.

Mental model

• Axum stays async fn at the wire edge; your domain stays in Effect with environment R (often caps!(…) or State<Env> at the boundary).
• The bridge takes State<Env>, builds an effect from &mut Env, then drives it to completion with id_effect::run_async using tokio::task::block_in_place + Handle::block_on so the Effect value never crosses a Send async boundary incorrectly.

Runtime requirements

Workspace effects are intentionally not Send in the general case. Axum handlers must return Send futures; the id_effect_axum adapter satisfies that contract by running the interpreter on the multi-thread runtime's blocking integration path. Use a multi-thread Tokio runtime (default for #[tokio::main]). On current_thread, prefer driving effects outside this adapter or supply a dedicated integration.

Tests: #[tokio::test] defaults to current-thread and can panic with this bridge—use e.g. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] for router tests (see crate tests for the pattern).

API surface

• routing::get / post / … — ergonomic wrappers when the handler is Fn(&mut Env) -> Effect<…>.
• run_with_caps — you already have State<Env>; build and run one effect per request.
• execute — Axum handler: State + IntoResponse for success and failure.
• json::decode_json_schema — validate JSON bodies with Schema::decode_unknown and map errors to HTTP responses (e.g. 422 with paths).

Capability DI end-to-end

Run the reference example:

cargo run -p id_effect_axum --example 020_capability_run_with

Pattern:

1. build_env([provide!(…) , …]) at startup
2. Router::with_state(env)
3. run_with_caps(State(env), |env| my_effect()) per route

#![allow(unused)]
fn main() {
use axum::{Router, extract::State, routing::get};
use id_effect::{Env, build_env, caps, effect, provide, require, Effect};
use id_effect_axum::run_with_caps;
struct Counter;

#[derive(::id_effect::ProviderSpecDerive)]
#[provides(Counter)]
struct CounterLive;

impl CounterLive {
    fn new() -> u32 { 7 }
}

fn handler(_env: &mut Env) -> Effect<String, (), caps!(Counter)> {
    effect!(|r| {
        let n = ~Counter;
        format!("count={n}")
    })
}

let env = build_env([provide!(CounterLive)]).expect("env");
let app = Router::new()
    .route("/", get(|State(env): State<Env>| async move {
        run_with_caps(State(env), handler).await.unwrap()
    }))
    .with_state(env);
}

Further reading

• RPC boundaries — id_effect_rpc envelopes, correlation ids, tracing
• cargo doc --open -p id_effect_axum
• Examples: cargo run -p id_effect_axum --example 010_routing_hello
• Tokio bridge for interpreter semantics; Tower for generic Service composition without Axum.

Tower service (id_effect_tower)

Workspace crate id_effect_tower implements tower::Service for Effect-based handlers. Effects are driven with id_effect::run_async, so you can compose Tower middleware (timeouts, retries, load balancing, …) around the same domain style as Axum, without tying to Axum’s router first.

When to use it

• You already have a Tower stack (custom servers, gRPC gateways, middleware chains) and want poll_ready / call semantics.
• You need per-service concurrency limits or request metrics hooks at the Tower boundary.

EffectService

• EffectService::new(state, f) — f(&mut state, request) returns an Effect; unlimited concurrency unless configured otherwise.
• with_max_in_flight — semaphore-gated poll_ready / call so at most n handler effects run concurrently; exposes an in-flight counter for observability.
• with_request_metrics — wraps each call with Metric::track_duration and increments an error counter on typed failure.

Dependencies

The crate depends on id_effect_tokio for the async driver; it does not re-export TokioRuntime—compose runtime and layers at your application root.

Further reading

• cargo doc --open -p id_effect_tower
• moon run effect-tower:examples or cargo run -p id_effect_tower --example 001_effect_service
• Axum host if you are on Axum specifically; Tokio bridge for Send and runtime caveats.

Configuration (id_effect_config)

Workspace crate id_effect_config loads configuration in a way aligned with Effect.ts configuration: lazy descriptors, optional Figment layering, and low-level effectful reads from a provider installed in R.

Three complementary styles

1. Config<T> descriptors (recommended for parity with Effect Config.*)
   Compose values like Config::string("HOST"), Config::integer("PORT").with_default(3000), then evaluate with Config::load against a concrete provider or Config::run as an Effect with config capabilities in R.

2. Figment + serde
   Build a Figment (TOML + env + …), then extract / provider helpers for whole-document deserialization when you prefer serde-shaped config files.

3. Low-level read_* helpers
   Declare caps!(ConfigProviderService) (or the crate's config key) and call read_string, read_integer, … for imperative-style reads that still stay inside the effect environment.

Wiring

At the stack root, install a config provider with provide!(…) and combine with Providers the same way as databases or loggers:

#![allow(unused)]
fn main() {
run_with([provide!(ConfigLive), provide!(DatabaseLive)], app())?;
}

Further reading

• cargo doc --open -p id_effect_config — extensive crate-level examples
• Schema for validating structured values after config strings become wire data

Logging (id_effect_logger)

Workspace crate id_effect_logger provides an injectable EffectLogger service: log lines are effect steps that read the logger capability from R, so formatting and sinks stay composable and testable.

Usage shape

• ~EffectLogger inside effect! (with |r|) to obtain the handle, then info, warn, … on static or dynamic messages (impl Into<Cow<'static, str>>).
• provide_effect_logger (and related constructors) build a ProviderSpec you pass to run_with / build_env.

#![allow(unused)]
fn main() {
use id_effect::{Effect, effect, caps, provide, run_with};

fn app() -> Effect<(), (), caps!(EffectLogger)> {
    effect!(|r| {
        let log = *~EffectLogger;
        ~ log.info("hello");
        ()
    })
}

run_with([provide!(TracingLoggerLive)], app())?;
}

Backends

The crate ships pipeline pieces such as structured JSON, tracing integration, and composite backends—see cargo doc -p id_effect_logger for LogBackend, JsonLogBackend, TracingLogBackend, CompositeLogBackend.

Relation to the rest of Part II

Logging is just another capability: same mental model as Capability services and Providing Services. Swap backends in tests via provide!(…) instead of silencing println!.

Further reading

• cargo doc --open -p id_effect_logger

RPC boundaries with id_effect (@effect/rpc parity)

Effect.ts @effect/rpc ties Schema, Layer, and HTTP so typed contracts cross process boundaries. In Rust there is no single blessed stack; this chapter documents patterns and the workspace id_effect_rpc crate for RPC-shaped HTTP on top of id_effect_axum and id_effect::schema.

Choosing a wire stack (tonic, tarpc, HTTP+JSON)

Recommendation for id_effect today: use HTTP + JSON for public APIs where you already host Axum, validate bodies with Schema::decode_unknown via id_effect_axum::json::decode_json_schema, and return structured failures with id_effect_rpc::RpcError. Add gRPC (tonic) when you need IDL-first codegen or cross-language streaming at scale.

Effect, R, and errors at the edge

• Keep Effect<A, E, R> for domain logic; E stays rich inside the process.
• At the Axum route, map E (and schema failures) into RpcError or 422 JSON from decode_json_schema — clients see a stable wire contract, not internal enums.
• Propagate x-correlation-id (read or generate with id_effect_rpc::correlation) and attach it to responses so logs and traces line up across hops.

Tracing and OpenTelemetry

Use id_effect_rpc::span::rpc_request_span for a stable rpc.request span name and fields (http.method, http.route, rpc.operation, correlation.id). When your binary installs a subscriber with an OpenTelemetry layer (see Phase B docs), these fields map cleanly to semantic conventions without id_effect_rpc depending on OTEL crates.

Workspace crate: id_effect_rpc

API index: cargo doc -p id_effect_rpc --open.

Example: Axum + Schema + RpcError

Runnable example:

cargo run -p id_effect_rpc --example 010_json_greet

It accepts POST /greet with JSON {"name": string, "enthusiasm": i64}, validates with Schema, rejects bad enthusiasm with RpcError::invalid_argument, and echoes x-correlation-id.

Stage D3 — codegen (optional)

Proc-macros or build.rs stubs for service definitions are backlog until the D2 operational model is proven in production. See docs/effect-ts-parity/phases/phase-d-rpc.md slugs iep-d-030 / iep-d-031.

Further reading

• Axum host — id_effect_axum, execute, JSON + schema
• Schema — wire types and ParseError
• Phase spec: docs/effect-ts-parity/phases/phase-d-rpc.md

Durable workflow spike (id_effect_workflow)

This chapter describes the experimental id_effect_workflow crate: a SQLite-backed, append-only log of completed steps that supports single-process restart resume — the Phase G spike aligned with docs/effect-ts-parity/phases/phase-g-cluster-workflow.md.

What problem it solves

Long-running business processes often need at-least-once execution with stable outputs per logical step. After a crash, a process should not repeat side effects for steps that already completed successfully.

DurableWorkflowLog persists each completed (workflow_id, seq) with a JSON payload. On restart, run_step_typed returns the stored value instead of invoking the closure again.

What it deliberately does not solve

• Distributed cluster execution (no membership, no shard routing).
• Multi-writer correctness without external coordination (add your own leasing or use an orchestrator).
• Compensation / saga policies beyond what your application encodes in ordinary Rust.

For most production multi-service workflows, prefer Temporal, Cadence, or cloud Step Functions — see docs/effect-ts-parity/phases/phase-g/adr-iep-g-011-temporal-vs-saga-vs-out-of-scope.md.

Composition with Effect

The log API is synchronous (Rusqlite). Keep SQLite on a blocking boundary:

• Inside effect!: perform log IO directly when your interpreter runs on a blocking runtime (see unit tests in id_effect_workflow).
• Async hosts: wrap calls in from_async + spawn_blocking (or your platform’s blocking pool) so you do not block async executors.

Security and replay notes

Persisted JSON may contain PII — treat it like any other database column (encryption, retention, access control are application concerns). If stored JSON is corrupted, replay surfaces WorkflowError::Json; recovery is policy-driven (do not silently re-run financial side effects). Details: docs/effect-ts-parity/phases/phase-g/iep-g-012-security-replay.md.

Semver

The crate ships at 0.1.x as experimental; storage layout may change without migrations until a stabilization ADR lands.

Error Handling — Cause, Exit, and Recovery

Part II gave you the full dependency injection story. Part III is about what happens when things go wrong — and in production, things always go wrong.

Rust's Result<T, E> is excellent for expected errors: outcomes you anticipated and typed. But real programs also encounter unexpected failures: panics, OOM conditions, and cancelled fibers. id_effect models all of these with a richer type hierarchy.

This chapter introduces Cause (the full error taxonomy), Exit (the terminal outcome of any effect), and the combinators for recovering from both.

Beyond Result — Why Cause Exists

Result<T, E> handles the errors you expect. But what about the errors that aren't your E?

Expected vs. Unexpected

#![allow(unused)]
fn main() {
// Expected: you planned for this
Effect<User, UserNotFound, Db>

// But what if the database panics?
// What if the fiber is cancelled?
// What if the process runs out of memory?
// None of these are UserNotFound.
}

Traditional Rust handles unexpected failures through panics, which unwind (or abort) and bypass all your error handling. In async code, panics in tasks can silently swallow errors or leave resources unreleased.

The Cause Type

Cause<E> is id_effect's complete taxonomy of failure:

#![allow(unused)]
fn main() {
use id_effect::Cause;

enum Cause<E> {
    Fail(E),                // Your typed, expected error
    Die(Box<dyn Any>),      // A panic or defect — something that shouldn't happen
    Interrupt,              // The fiber was cancelled
}
}

Every failure in the effect runtime is one of these three. Together they cover the full space of "things that can go wrong."

• Cause::Fail(e) — an error you declared in E, handled with catch or map_error
• Cause::Die(payload) — a panic, logic bug, or fatal error; should be logged and treated as a defect
• Cause::Interrupt — clean cancellation; the fiber was asked to stop and cooperated

Why This Matters

Without Cause, you can only handle Cause::Fail. The other two propagate invisibly up the fiber tree and may silently swallow logs or leave resources unreleased.

With Cause, you can handle all failure modes in a structured way:

#![allow(unused)]
fn main() {
my_effect.catch_all(|cause| match cause {
    Cause::Fail(e)    => recover_from_expected(e),
    Cause::Die(panic) => log_defect_and_fail(panic),
    Cause::Interrupt  => succeed(default_value()),
})
}

Resource finalizers (Chapter 10) use this same model — they run on any Cause, ensuring cleanup regardless of how the fiber ends.

Day-to-Day Usage

In normal application code you rarely inspect Cause directly. You use:

• .catch(f) for handling Cause::Fail
• .catch_all(f) when you need to handle panics or interruption too
• Exit (next section) when you need to inspect the terminal outcome

The Cause type is mostly visible at infrastructure boundaries — resource finalizers, fiber supervisors, and top-level error handlers.

Exit — Terminal Outcomes

Every effect execution ends with an Exit. It's the final word on what happened.

The Exit Type

#![allow(unused)]
fn main() {
use id_effect::Exit;

enum Exit<A, E> {
    Success(A),         // Effect completed, produced A
    Failure(Cause<E>),  // Effect failed with Cause<E>
}
}

Exit combines the success type and the full failure taxonomy.

• run_blocking returns Result<A, E> — you only see typed E failures; defects and interrupts are not represented as values there.
• run_test returns Exit<A, E> and is the right harness for unit tests that must assert on defects, interrupts, and typed failures.

For CLI / process exit codes, map an Exit with id_effect_cli::exit_code_for_exit (see CLI with clap).

Converting Exit to Result

Most application code wants Result. The conversion is straightforward:

#![allow(unused)]
fn main() {
let result: Result<User, AppError> = exit.into_result(|cause| match cause {
    Cause::Fail(e) => AppError::Expected(e),
    Cause::Die(_)  => AppError::Defect,
    Cause::Interrupt(_) => AppError::Cancelled,
});
}

Or use the convenience method that maps Cause::Fail(e) → Err(e) and treats other causes as panics:

#![allow(unused)]
fn main() {
let result: Result<User, DbError> = exit.into_result_or_panic();
}

Exit in Fiber Joins

When you join a fiber (Chapter 9), you get an Exit back:

#![allow(unused)]
fn main() {
let fiber = my_effect.fork();
let exit: Exit<A, E> = fiber.join().await;
}

This lets you inspect whether the fiber succeeded, failed with a typed error, panicked, or was cancelled — and respond appropriately in the parent fiber.

Practical Rule

Use run_blocking (which returns Result<A, E>) for most application logic. Use run_test when you need the full Exit taxonomy in tests. At the process edge (binaries), combine run_main / exit_code_for_exit from id_effect_cli with the table in the CLI exit codes chapter.

Recovery Combinators — catch, fold, and Friends

Knowing about Cause and Exit is only useful if you can act on them. id_effect provides a focused set of recovery combinators.

catch: Handle Expected Errors

catch intercepts Cause::Fail(e) and gives you a chance to recover:

#![allow(unused)]
fn main() {
let resilient = risky_db_call()
    .catch(|error: DbError| {
        match error {
            DbError::NotFound => succeed(User::anonymous()),
            other => fail(other),  // re-raise anything else
        }
    });
}

If risky_db_call fails with Cause::Fail(e), the closure runs. If it fails with Cause::Die or Cause::Interrupt, those propagate unchanged — catch only handles typed failures.

catch_all: Handle Everything

catch_all intercepts any Cause:

#![allow(unused)]
fn main() {
let bulletproof = my_effect.catch_all(|cause| match cause {
    Cause::Fail(e)   => handle_expected_error(e),
    Cause::Die(_)    => succeed(fallback_value()),
    Cause::Interrupt => succeed(cancelled_gracefully()),
});
}

Use catch_all when you genuinely need to handle panics or cancellation — typically at resource boundaries or top-level handlers. Don't use it to swallow defects silently.

fold: Handle Both Paths

fold transforms both success and failure into a uniform success type:

#![allow(unused)]
fn main() {
let always_string: Effect<String, Never, ()> = risky_call()
    .fold(
        |error| format!("Error: {error}"),
        |value| format!("Success: {value}"),
    );
}

After fold, the effect never fails (E = Never). Both arms produce the same type. This is useful for logging, metrics, or converting to a neutral representation.

or_else: Try an Alternative

or_else runs an alternative effect on failure:

#![allow(unused)]
fn main() {
let with_fallback = primary_source()
    .or_else(|_err| secondary_source());
}

If primary_source fails, secondary_source runs. If that also fails, the combined effect fails with the second error. Useful for fallback chains.

ignore_error: Discard Failures

When you genuinely don't care about an operation's success:

#![allow(unused)]
fn main() {
// Log "best effort" — failure is acceptable
let logged_effect = log_metrics()
    .ignore_error()
    .flat_map(|_| actual_work());
}

ignore_error converts Effect<A, E, R> to Effect<Option<A>, Never, R>. The effect always "succeeds" — with Some(value) on success or None on failure.

The Recovery Hierarchy

catch(f)      — handles Cause::Fail only
catch_all(f)  — handles all Cause variants
fold(on_e, on_a) — transforms both paths to success
or_else(f)    — runs alternative on failure
ignore_error  — converts failure to Option

Prefer the narrowest combinator that solves your problem. catch for expected errors. catch_all only when you need to touch panics or cancellation.

Error Accumulation — Collecting All Failures

catch and fold handle errors sequentially: one effect, one error, one handler. But sometimes you need to run many operations and collect all their failures — not just the first.

The Fail-Fast Problem

Sequential effect! short-circuits on the first failure:

#![allow(unused)]
fn main() {
effect! {
    let _ = ~ validate_name(&input.name);    // fails here →
    let _ = ~ validate_email(&input.email);  // never runs
    let _ = ~ validate_age(input.age);       // never runs
}
}

For form validation or batch imports, you want to report all errors to the user, not just the first one.

validate_all

validate_all runs a collection of effects and accumulates all failures:

#![allow(unused)]
fn main() {
use id_effect::validate_all;

let results = validate_all(vec![
    validate_name(&input.name),
    validate_email(&input.email),
    validate_age(input.age),
]);
// Type: Effect<Vec<Name, Email, Age>, Vec<ValidationError>, ()>
}

If any validations fail, all errors are collected and returned as a Vec. If all succeed, you get all the success values.

partition

partition runs effects and splits the results into successes and failures:

#![allow(unused)]
fn main() {
use id_effect::partition;

let (successes, failures): (Vec<User>, Vec<ImportError>) =
    run_blocking(partition(records.iter().map(import_record)))?;

println!("{} imported, {} failed", successes.len(), failures.len());
}

partition never fails. It always returns two lists: what worked and what didn't. Useful for batch operations where partial success is acceptable.

Or: Combining Two Error Types

When composing effects with different error types, Or avoids flattening into a single error type before you're ready:

#![allow(unused)]
fn main() {
use id_effect::Or;

// Instead of converting both to AppError immediately:
type BothErrors = Or<DbError, NetworkError>;

fn combined() -> Effect<Data, BothErrors, ()> {
    db_fetch()
        .map_error(Or::Left)
        .zip(network_fetch().map_error(Or::Right))
        .map(|(a, b)| merge(a, b))
}
}

Or<A, B> is the coproduct of two error types. It defers the decision of how to combine them until you actually need to handle them.

The ParseErrors Type

The Schema module (Chapter 14) uses ParseErrors — a structured accumulator for parsing failures with field paths:

#![allow(unused)]
fn main() {
let result: Result<User, ParseErrors> = user_schema.parse(data);

if let Err(errors) = result {
    for e in errors.iter() {
        eprintln!("At {}: {}", e.path(), e.message());
    }
}
}

ParseErrors is specialised for schema validation, but the pattern — collect all, report all — applies whenever you validate structured input.

When to Accumulate vs. Short-Circuit

The choice is about what makes sense to the caller. Short-circuit is efficient; accumulation is informative. Use the one your users need.

Concurrency & Fibers — Structured Async

Async Rust gives you the ability to do many things concurrently. The challenge is doing it safely — without fire-and-forget tasks that outlive their parent, without silent failures when a task panics, and without resource leaks when tasks are cancelled.

id_effect uses Fibers for structured concurrency. A Fiber is a lightweight, interruptible async task with a typed result, an explicit lifecycle, and guaranteed cleanup.

Fibers spawn through Compute Fabric: ThreadSleepRuntime admits work on a shared FiberPool instead of one OS thread per fiber, and TokioRuntime gates blocking-pool tasks with the same admission controller.

This chapter covers spawning fibers, joining them, cancelling them gracefully, using FiberRef for fiber-local state, and supervision (restart policies tied to Scope and CancellationToken).

What Are Fibers? — Lightweight Structured Tasks

A Fiber is an effect-managed async task. It's lighter than an OS thread and safer than a raw tokio::spawn.

Fibers vs. Raw Tasks

#![allow(unused)]
fn main() {
// Raw tokio::spawn — fire and forget
// Who owns this? What happens if it panics?
// When does it stop? Who cleans up?
tokio::spawn(async {
    do_something().await;
});

// Effect Fiber — explicit lifecycle
let handle: FiberHandle<Result, Error> = my_effect.fork();
// You hold the handle. The fiber is yours.
let exit: Exit<Error, Result> = handle.join().await;
// The fiber stops when you join or drop the handle.
}

With tokio::spawn, the task runs independently. If it panics, the panic is captured by Tokio and may or may not surface to you. There's no built-in way to cancel it or guarantee its resources are cleaned up.

With fork, you get a FiberHandle. When you join(), you get the full Exit — success, typed failure, panic, or cancellation. When you drop the handle without joining, the fiber is cancelled automatically.

Structured Concurrency

The key property of Fibers is structured lifecycle:

• A fiber cannot outlive its parent scope without explicit permission
• All spawned fibers are joined (or cancelled) before the parent effect completes
• Panics and failures propagate through the fiber tree, not silently into the void

This makes concurrent code much easier to reason about. When process_batch completes, all its helper fibers have completed too — or been cancelled and cleaned up.

FiberId

Each Fiber has a unique FiberId. You can use it for logging, tracing, and correlation:

#![allow(unused)]
fn main() {
use id_effect::FiberId;

effect! {
    let id = ~ current_fiber_id();
    ~ log(&format!("[fiber:{id}] starting work"));
    // ...
}
}

FiberId flows through the fiber's execution automatically. You don't thread it manually.

FiberHandle and FiberStatus

FiberHandle<E, A> is the control interface for a spawned fiber:

#![allow(unused)]
fn main() {
let handle = my_effect.fork();

// Check status without blocking
let status: FiberStatus = handle.status();

// Join — blocks until the fiber completes
let exit: Exit<E, A> = handle.join().await;

// Interrupt — ask the fiber to stop
handle.interrupt();
}

FiberStatus can be Running, Completed, or Interrupted. Unlike tokio::JoinHandle, you can inspect status without consuming the handle.

Spawning and Joining — fiber_all and Friends

Running a single fiber is useful; running many concurrently is where Fibers shine.

fork: Spawn One Fiber

#![allow(unused)]
fn main() {
let handle = compute_expensive_result().fork();

// Do other work while the fiber runs
let local_result = local_computation();

// Now join the fiber
let remote_result = handle.join().await.into_result_or_panic()?;

(local_result, remote_result)
}

fork spawns the effect as a concurrent fiber. You can do other work and join later.

fiber_all: Run Many, Collect All

#![allow(unused)]
fn main() {
use id_effect::fiber_all;

// Run all concurrently; collect all results
let results: Vec<User> = run_blocking(
    fiber_all(user_ids.iter().map(|&id| fetch_user(id)))
)?;
}

fiber_all takes an iterable of effects, runs them all concurrently, and waits for every one to complete. If any fails, the first failure is returned (and any remaining fibers are cancelled).

For independent work where all results are needed, fiber_all is the idiomatic choice.

fiber_race: First to Complete Wins

#![allow(unused)]
fn main() {
use id_effect::fiber_race;

// Try primary and backup concurrently — take whichever responds first
let data = run_blocking(
    fiber_race(vec![fetch_from_primary(), fetch_from_backup()])
)?;
// The slower fiber is automatically cancelled
}

fiber_race returns as soon as any fiber succeeds. The others are interrupted. Useful for timeout patterns, geographic failover, and speculative execution.

fiber_any: First Success

#![allow(unused)]
fn main() {
use id_effect::fiber_any;

// Try all; return first success (ignore failures until all done)
let result = fiber_any(vec![
    try_region_us(),
    try_region_eu(),
    try_region_ap(),
])?;
}

fiber_any differs from fiber_race in that it ignores failures and waits for the first success. Only if all fail does it return an error.

run_fork: Low-Level Spawn

For cases where you need to spawn with full control:

#![allow(unused)]
fn main() {
use id_effect::run_fork;

let runtime = Runtime::current();
let handle = run_fork(runtime, || (my_effect, my_env));
}

run_fork is the low-level primitive. effect.fork() is syntactic sugar over it when you're already inside an effect context.

Compute Fabric placement

When you pass a ThreadSleepRuntime or TokioRuntime that holds a ComputeFabric, every run_fork / spawn_with call:

1. Runs a supervisor tick (telemetry vs ResourcePolicy)
2. Acquires an admission permit before starting the worker
3. Releases the permit when the fiber completes

Under memory pressure the supervisor throttles permits; under headroom it admits up to the pool size. See Compute Fabric and example 120_compute_fabric_memory_cap.rs.

Error Behaviour

Choose based on whether partial success is acceptable and whether you want to wait for everyone.

Cancellation — Interrupting Gracefully

Cancellation in async code is notoriously difficult to get right. id_effect makes it explicit and cooperative.

The Model: Cooperative Cancellation

Fibers aren't forcibly killed. They're interrupted — given a signal to stop at the next safe checkpoint. The fiber cooperates by checking for interruption at yield points.

The simplest yield point is check_interrupt:

#![allow(unused)]
fn main() {
use id_effect::check_interrupt;

effect! {
    for chunk in large_dataset.chunks(1000) {
        ~ check_interrupt();  // yields; if interrupted, stops here
        process_chunk(chunk);
    }
    "done"
}
}

Every ~ binding is also an implicit yield point. If a fiber is interrupted while awaiting an effect, the interruption propagates to the next ~ bind.

CancellationToken

For external cancellation (e.g., from HTTP request handlers or UI cancel buttons):

#![allow(unused)]
fn main() {
use id_effect::CancellationToken;

// Create a cancellation token
let token = CancellationToken::new();

// Pass it to a long-running effect
let effect = long_running_job().with_cancellation(&token);

// Spawn it
let handle = effect.fork();

// Later, cancel from outside
token.cancel();

// The fiber will stop at the next check_interrupt
let exit = handle.join().await;
// exit will be Exit::Failure(Cause::Interrupt)
}

Tokens can be cloned and shared. Cancelling any clone cancels all effects sharing that token.

Interrupting Directly via FiberHandle

#![allow(unused)]
fn main() {
let handle = background_work().fork();

// After some timeout or external event:
handle.interrupt();

let exit = handle.join().await;
// Cleanup (finalizers, scopes) runs before the handle resolves
}

.interrupt() sends the interruption signal. The fiber's finalizers (Chapter 10) still run. .join() waits for them to complete.

Graceful Shutdown

Interruption is the mechanism for graceful shutdown. The pattern:

1. Signal all top-level fibers with .interrupt()
2. Wait for all handles to join (with a timeout)
3. If any fiber doesn't stop within the timeout, escalate

#![allow(unused)]
fn main() {
let handles: Vec<FiberHandle<_, _>> = workers.iter().map(|w| w.fork()).collect();

// Shutdown signal received
for h in &handles { h.interrupt(); }

// Wait with timeout
for h in handles {
    tokio::time::timeout(Duration::from_secs(5), h.join()).await;
}
}

Because effect finalizers run on interruption, all resources are cleaned up as fibers stop — no manual cleanup required at the shutdown handler.

Uninterruptible Regions

Some operations shouldn't be interrupted mid-way (e.g., writing to a database inside a transaction). Mark them uninterruptible:

#![allow(unused)]
fn main() {
use id_effect::uninterruptible;

// This block runs to completion even if interrupted
let committed = uninterruptible(effect! {
    ~ begin_transaction();
    ~ insert_records();
    ~ commit();
});
}

The interruption is deferred until committed completes. Use sparingly — long uninterruptible regions delay shutdown.

FiberRef — Fiber-Local State

FiberRef is the effect equivalent of thread-local storage. It holds a value that's scoped to the current fiber — each fiber has its own independent copy.

Defining a FiberRef

#![allow(unused)]
fn main() {
use id_effect::FiberRef;

// A fiber-local trace ID, defaulting to "none"
static TRACE_ID: FiberRef<String> = FiberRef::new(|| "none".to_string());
}

FiberRef::new takes a factory closure that produces the initial value for each fiber. The static variable is the key; each fiber has its own value.

Reading and Writing

#![allow(unused)]
fn main() {
effect! {
    // Set the trace ID for this fiber
    ~ TRACE_ID.set("req-abc-123".to_string());

    // Read it anywhere in this fiber's call stack
    let id = ~ TRACE_ID.get();
    ~ log(&format!("[{id}] processing request"));

    ~ process_request();
    Ok(())
}
}

set and get are both effects (they need the fiber context). Inside effect!, use ~ to bind them.

FiberRef Doesn't Cross Fiber Boundaries

When you fork a new fiber, it starts with its own copy of the FiberRef value (the factory closure runs again):

#![allow(unused)]
fn main() {
effect! {
    ~ TRACE_ID.set("parent-123".to_string());
    
    let child = effect! {
        let id = ~ TRACE_ID.get();
        // id is "none" — the fork starts fresh
        println!("child trace id: {id}");
    }.fork();
    
    ~ child.join();
    Ok(())
}
}

If you want the child to inherit the parent's value, pass it explicitly or use FiberRef::inherit:

#![allow(unused)]
fn main() {
let child_with_inherited = effect! {
    let id = ~ TRACE_ID.get();
    TRACE_ID.locally(id, child_effect())  // child sees parent's value
};
}

locally(value, effect) runs the effect with a temporarily overridden FiberRef value, then restores the previous value when done.

Common Use Cases

FiberRef makes it easy to carry contextual information through deep call stacks without threading extra parameters everywhere — the fiber equivalent of request-scoped context in traditional web frameworks.

Supervision — Restart Policies and Scope

Raw run_fork gives you a FiberHandle and full control. Long-lived servers usually need policies: when a child effect fails, should you retry, back off, give up, or substitute a default? Effect.ts encodes these ideas in supervisors. id_effect provides the same vocabulary wired to Scope, CancellationToken, and Schedule.

Why not only retry?

retry re-runs an Effect<A, E, R> until success or the schedule stops. Supervision adds:

• A stable shutdown channel ([CancellationToken]) installed when a child [Scope] closes, so cooperative loops exit with Cause::Interrupt instead of spinning forever.
• Declarative policies (Terminate, Restart, RestartWithLimit, Escalate, Ignore) that compose with virtual time (TestClock) for deterministic tests.

Supervisor and Scope

Supervisor::attach(parent_scope) forks a child [Scope]. When the parent closes, the child closes; a finalizer on the child cancels the supervisor token so any supervised loop observes cancellation on the next iteration header.

Use Supervisor::detached() for examples and unit tests that do not need a parent tree.

Policies in one table

RestartWithLimit counts retries after failure: limit == 0 means “no retries” (fail on the first Err without sleeping). A positive limit allows that many retry attempts after the initial failing run.

Typed failures vs interrupts vs defects

• A supervised child that returns [Err] becomes [Cause::Fail].
• Token cancellation (scope teardown or explicit cancel) surfaces as [Cause::Interrupt].
• Panics inside the interpreter are still defects at the runtime boundary; supervision does not “recover” unwinding std panics—document and test the happy paths your runtime actually exposes.

supervised vs FiberHandle::scoped

FiberHandle::scoped ties one handle to one scope via a finalizer that interrupts the fiber. supervised runs the factory inline on your environment R, so it suits retry loops without an extra FiberHandle until you opt into Supervisor::spawn.

Example sketch

#![allow(unused)]
fn main() {
use id_effect::{
  Supervisor, SupervisorPolicy, supervised,
  Schedule, TestClock, succeed,
};
use std::time::Instant;

let parent = id_effect::Scope::make();
let sup = Supervisor::attach(&parent);
let clock = TestClock::new(Instant::now());
let body = supervised(
  &sup,
  SupervisorPolicy::Restart {
    schedule: Schedule::spaced(std::time::Duration::ZERO),
  },
  clock,
  || succeed::<u32, &str, ()>(42),
);
// run `body` with your environment; close `parent` to cancel via token.
}

For production delays, prefer a non-zero [Schedule] and a real Clock (for example the Tokio bridge clock in server code).

See also

• Phase F parity doc — epic breakdown and acceptance notes.
• supervisor.rs — full API and unit tests.

Resources & Scopes — Deterministic Cleanup

RAII works beautifully in synchronous Rust: resources are released when they fall out of scope, Drop runs deterministically. In async code, the picture gets complicated.

This chapter shows why RAII breaks down in async contexts, introduces Scope and finalizers as the solution, covers the acquire_release pattern for RAII-style resource management, and concludes with Pool for reusing expensive connections.

The Resource Problem — Cleanup in Async

RAII in synchronous code:

#![allow(unused)]
fn main() {
{
    let file = File::open("data.txt")?;
    process(&file)?;
}  // file.drop() runs here, always, unconditionally
}

Reliable. Simple. The drop happens when the scope ends — no exceptions (unless you have exceptions).

The Async Complication

#![allow(unused)]
fn main() {
async fn process_data() -> Result<(), Error> {
    let conn = open_connection().await?;
    let data = fetch_data(&conn).await?;  // What if this is cancelled?
    transform_and_save(data).await?;      // Or this?
    conn.close().await?;                  // May never reach here
    Ok(())
}
}

Three problems:

1. Cancellation: If this async function is cancelled mid-execution, conn.close() never runs.
2. Panic: If transform_and_save panics, the async task is dropped. conn.close() is skipped.
3. Async Drop: impl Drop for Connection can only do synchronous cleanup. If closing a connection requires .await, you can't do it in Drop.

conn.close() must be an async call, but Drop can't be async. This is a fundamental mismatch.

The Root Cause

RAII relies on Drop running synchronously when a value goes out of scope. In async code, "going out of scope" and "running cleanup" can be decoupled — by cancellation, by executor scheduling, or by the fact that async closures are state machines that might never reach certain states.

The Solution Preview

id_effect solves this with:

• Scope — a region where finalizers are registered and guaranteed to run (even on cancellation or panic)
• acquire_release — a combinator that pairs acquisition with its cleanup
• Pool — for long-lived resources that need controlled reuse

All three run cleanup effects (not just synchronous Drop), and all three run them unconditionally — success, failure, or interruption.

Scopes and Finalizers — Guaranteed Cleanup

A Scope is a region of execution with a finalizer registry. Any cleanup effects registered in the scope run when the scope exits — regardless of how it exits.

Creating a Scope

#![allow(unused)]
fn main() {
use id_effect::{Scope, Finalizer, scoped};

let result = scoped(|scope| {
    effect! {
        let conn = ~ open_connection();

        // Register cleanup — runs when scope exits
        ~ scope.add_finalizer(Finalizer::new(move || {
            conn.close()
        }));

        // Do work
        let data = ~ fetch_data(&conn);
        process(data)
    }
});
}

scoped creates a scope, runs the inner effect, and then — in all cases — runs the registered finalizers in reverse order.

Finalizers Always Run

#![allow(unused)]
fn main() {
let result = scoped(|scope| {
    effect! {
        let conn = ~ open_connection();
        ~ scope.add_finalizer(Finalizer::new(move || conn.close()));

        ~ risky_operation();  // may panic or fail
        "done"
    }
});
// Whether risky_operation succeeds, fails, or panics:
// conn.close() ALWAYS runs before result is returned
}

This is the guarantee that RAII can't provide in async: the finalizer is an async effect that runs in the right context, at the right time, always.

Multiple Finalizers

Finalizers run in reverse registration order (last-in, first-out — like RAII destructors):

#![allow(unused)]
fn main() {
scoped(|scope| {
    effect! {
        let conn   = ~ open_connection();
        let txn    = ~ begin_transaction(&conn);
        let cursor = ~ open_cursor(&txn);

        ~ scope.add_finalizer(Finalizer::new(move || close_cursor(cursor)));   // runs 3rd
        ~ scope.add_finalizer(Finalizer::new(move || rollback_or_commit(txn))); // runs 2nd... wait
        ~ scope.add_finalizer(Finalizer::new(move || close_connection(conn)));  // wait — read below
    }
})
}

Actually, the first registered finalizer runs last. Register cleanup in the order you want it to run, reversed: register connection first (so it closes last), cursor last (so it closes first).

Scope Inheritance

Scopes nest. A child scope's finalizers run before the parent's:

#![allow(unused)]
fn main() {
scoped(|outer| {
    scoped(|inner| {
        effect! {
            ~ inner.add_finalizer(Finalizer::new(|| cleanup_inner()));
            ~ outer.add_finalizer(Finalizer::new(|| cleanup_outer()));
            work()
        }
    })
})
// Execution order: cleanup_inner(), then cleanup_outer()
}

Layers use scopes internally — every resource a Layer builds can register its own finalizer, and the whole graph tears down cleanly when the application shuts down.

acquire_release — The RAII Pattern

acquire_release is a convenience wrapper around Scope that pairs acquisition and release into a single value.

The Pattern

#![allow(unused)]
fn main() {
use id_effect::acquire_release;

let managed_connection = acquire_release(
    // Acquire: run this to get the resource
    open_connection(),
    // Release: run this when done (always runs)
    |conn| conn.close(),
);
}

managed_connection is itself an effect that:

1. When executed, opens the connection
2. Registers conn.close() as a finalizer in the current scope
3. Produces the connection for use

Use it with flat_map or effect!:

#![allow(unused)]
fn main() {
let result = managed_connection.flat_map(|conn| {
    do_work_with_conn(&conn)
});
// conn.close() runs after do_work_with_conn, regardless of outcome
}

Or inline:

#![allow(unused)]
fn main() {
effect! {
    let conn = ~ managed_connection;
    let data = ~ fetch_data(&conn);   // conn closes after this block
    process(data)
}
}

Why This Is Better Than Manual Scope

acquire_release makes the acquisition-release pair inseparable. You can't accidentally call open_connection() without also registering its cleanup. The resource and its lifecycle are coupled at the point of creation.

#![allow(unused)]
fn main() {
// With manual scope: easy to forget the finalizer
let conn = ~ open_connection();
// ... (forgot: ~ scope.add_finalizer(...))

// With acquire_release: cleanup is mandatory, automatic
let conn = ~ acquire_release(open_connection(), |c| c.close());
}

Resource Wrapping Pattern

A common convention is to wrap acquire_release in a helper function:

#![allow(unused)]
fn main() {
fn managed_db_connection(url: &str) -> Effect<Connection, DbError, ()> {
    acquire_release(
        Connection::open(url),
        |conn| conn.close(),
    )
}

// Usage
effect! {
    let conn = ~ managed_db_connection(config.db_url());
    ~ run_query(&conn, "SELECT 1")
}
}

The helper function documents that open_connection() must always be paired with close(). Callers don't think about the lifecycle; it's handled.

Comparison with Drop

acquire_release is not a replacement for impl Drop — it's a complement:

• impl Drop: synchronous cleanup for types that own simple resources
• acquire_release: async cleanup for effects that acquire and release through the effect runtime

Use both appropriately. A TcpStream closing its OS file descriptor in Drop is fine. Closing a database connection pool that requires async coordination belongs in acquire_release.

Pools — Reusing Expensive Resources

Creating a database connection takes time: DNS lookup, TCP handshake, TLS, authentication. Creating one per request is wasteful. A pool maintains a set of connections and lends them out, returning them when done.

id_effect provides Pool and KeyedPool as first-class effect constructs.

Pool: Basic Connection Pool

#![allow(unused)]
fn main() {
use id_effect::Pool;

// Create a pool of up to 10 connections
let pool: Pool<Connection> = Pool::new(
    || open_connection("postgres://localhost/app"),  // factory
    10,                                               // max size
);
}

The pool lazily creates connections up to the max. Idle connections are kept alive for reuse.

Using a Pool Connection

#![allow(unused)]
fn main() {
pool.with_resource(|conn: &Connection| {
    effect! {
        let rows = ~ conn.query("SELECT * FROM users");
        rows.into_iter().map(User::from_row).collect::<Vec<_>>()
    }
})
}

with_resource acquires a connection from the pool, runs the effect, and returns the connection automatically when done — regardless of success, failure, or cancellation. No acquire_release boilerplate; the pool handles it.

Waiting for Availability

If all connections are in use, with_resource waits until one becomes available:

#![allow(unused)]
fn main() {
// Concurrent requests share the pool; each waits its turn
fiber_all(vec![
    pool.with_resource(|c| query_a(c)),
    pool.with_resource(|c| query_b(c)),
    pool.with_resource(|c| query_c(c)),
])
}

The pool queues waiters and notifies them as connections are returned.

KeyedPool: Multiple Named Pools

For scenarios with multiple distinct pools (e.g., read replica + write primary):

#![allow(unused)]
fn main() {
use id_effect::KeyedPool;

let pools: KeyedPool<&str, Connection> = KeyedPool::new(
    |key: &&str| open_connection(key),
    5,  // max per key
);

// Get a connection for the write primary
pools.with_resource("write-primary", |conn| { ... })

// Get a connection for the read replica
pools.with_resource("read-replica", |conn| { ... })
}

Each key has its own independently bounded pool.

Pool as a capability

In practice, pools live in Env and are wired at the program edge:

#![allow(unused)]
fn main() {
use id_effect::{Effect, caps, effect, provide, require, run_with, Needs, ProviderSpecDerive};
struct DbPool;

#[derive(ProviderSpecDerive)]
#[provides(DbPool)]
struct DbPoolLive;

impl DbPoolLive {
    fn new(config: &Config) -> Pool<Connection> {
        Pool::new(|| Connection::open(config.db_url()), config.pool_size())
    }
}

fn query_users() -> Effect<Vec<User>, DbError, caps!(DbPool)> {
    effect!(|r| {
        let pool = ~DbPool;
        ~ pool.with_resource(|conn| {
            effect!(|r| {
                let rows = ~ conn.query("SELECT * FROM users");
                rows.iter().map(User::from_row).collect::<Vec<_>>()
            })
        })
    })
}

// main or test
run_with([provide!(ConfigLive), provide!(DbPoolLive)], query_users())?;
}

Pool creation runs in the provider graph; business code declares caps!(DbPool) and uses ~DbPool inside effect!.

Scheduling — Retry, Repeat, and Time

Production services fail. Networks are unreliable. Downstream APIs go down. The database gets overwhelmed. Defensive engineering means anticipating failure and building policies for what to do when it happens.

id_effect models these policies with Schedule — a type that describes when to retry, how long to wait between attempts, and when to give up. Combined with Clock injection, scheduling logic becomes testable without real-time delays.

Temporal scheduling (this chapter) is separate from Compute Fabric compute scheduling: Schedule governs when to retry; Fabric governs where and how concurrently effects run against CPU and memory policy. Both can apply in the same program — e.g. retry a remote step while the supervisor throttles local fiber admission.

Schedule — The Retry/Repeat Policy Type

A Schedule is not just a number of retries or a delay. It's a policy — a function that takes the current state (attempt count, elapsed time, last output) and decides whether to continue and how long to wait.

The Core Concept

#![allow(unused)]
fn main() {
use id_effect::Schedule;

// A Schedule answers: "Given where we are, should we continue? And after how long?"
// Input: attempt number, elapsed time, last result
// Output: Continue(delay) or Done
}

This abstraction is more powerful than "retry 3 times with 1-second delay." A Schedule can:

• Increase delay exponentially (standard backoff)
• Cap the total time regardless of attempts
• Stop after a maximum number of attempts
• Adjust based on the error type or last result
• Combine policies with && and ||

Creating Schedules

#![allow(unused)]
fn main() {
use id_effect::Schedule;

// Fixed delay: always wait the same amount
let fixed = Schedule::spaced(Duration::from_secs(1));

// Exponential: 100ms, 200ms, 400ms, 800ms, ...
let exponential = Schedule::exponential(Duration::from_millis(100));

// Fibonacci: 100ms, 100ms, 200ms, 300ms, 500ms, 800ms, ...
let fibonacci = Schedule::fibonacci(Duration::from_millis(100));

// Forever: repeat indefinitely with no delay
let forever = Schedule::forever();

// Once: run exactly once (useful for testing)
let once = Schedule::once();
}

Combining Schedules

Schedules compose:

#![allow(unused)]
fn main() {
// Retry up to 5 times
let max_5 = Schedule::exponential(100.ms()).take(5);

// But stop after 30 seconds total
let bounded = Schedule::exponential(100.ms()).until_total_duration(Duration::from_secs(30));

// Combine with &&: both conditions must agree to continue
let safe = Schedule::exponential(100.ms())
    .take(5)
    .until_total_duration(Duration::from_secs(30));
}

Schedule as a Value

Like effects, schedules are values. You can define them once and reuse them:

#![allow(unused)]
fn main() {
const DEFAULT_RETRY: Schedule = Schedule::exponential(Duration::from_millis(100))
    .take(5)
    .with_jitter(Duration::from_millis(50));

fn call_external_api() -> Effect<Response, ApiError, HttpClient> {
    make_request().retry(DEFAULT_RETRY)
}
}

Jitter (random delay variation) reduces thundering-herd problems when many processes retry simultaneously. .with_jitter(d) adds a random delay in [0, d) to each wait.

Built-in Schedules — exponential, fibonacci, forever

id_effect provides a library of common schedule types. This section catalogs them with typical use cases.

exponential

#![allow(unused)]
fn main() {
Schedule::exponential(Duration::from_millis(100))
// Delays: 100ms, 200ms, 400ms, 800ms, 1600ms, ...
}

The standard backoff pattern. Each delay doubles. Use for most network retry scenarios — it backs off quickly and gives the downstream service time to recover.

#![allow(unused)]
fn main() {
// With cap: 100ms, 200ms, 400ms, 800ms, 800ms, 800ms (capped at 800ms)
Schedule::exponential(Duration::from_millis(100))
    .with_max_delay(Duration::from_millis(800))
}

fibonacci

#![allow(unused)]
fn main() {
Schedule::fibonacci(Duration::from_millis(100))
// Delays: 100ms, 100ms, 200ms, 300ms, 500ms, 800ms, 1300ms, ...
}

Fibonacci backoff grows more gradually than exponential — useful when you want more retries in the early attempts before backing off significantly.

spaced

#![allow(unused)]
fn main() {
Schedule::spaced(Duration::from_secs(5))
// Delays: 5s, 5s, 5s, 5s, ... (constant)
}

Fixed interval. Use for polling (checking status every N seconds), heartbeats, or scenarios where the delay should be predictable.

forever

#![allow(unused)]
fn main() {
Schedule::forever()
// No delay between repetitions; runs indefinitely
}

Run an effect as fast as possible, forever. Use with repeat for continuous background jobs (e.g., metrics collection, background syncs). Almost always combined with .take(n) or .until_total_duration(d).

Limiters

Limiters constrain any schedule:

#![allow(unused)]
fn main() {
// Stop after N attempts
schedule.take(5)

// Stop after N successes (for repeat)
schedule.take_successes(3)

// Stop after total elapsed time
schedule.until_total_duration(Duration::from_secs(30))

// Stop after N total elapsed retries including initial
schedule.take_with_initial(6)  // 1 initial + 5 retries
}

Jitter

#![allow(unused)]
fn main() {
// Add random delay in [0, jitter_max) to each wait
schedule.with_jitter(Duration::from_millis(50))

// Alternatively, full jitter (random in [0, delay])
schedule.with_full_jitter()
}

Jitter prevents retry storms. When 1000 clients all hit a failing service and all retry at the same time, they create a "thundering herd." Jitter spreads the retries out.

Combining with &&

#![allow(unused)]
fn main() {
let bounded_exponential = Schedule::exponential(Duration::from_millis(100))
    .take(5)
    .with_jitter(Duration::from_millis(20))
    .until_total_duration(Duration::from_secs(10));
}

The && (chain) operation means "continue only if both schedules agree to continue." The first schedule that says "done" wins.

retry and repeat — Applying Policies

Schedule is a policy description. retry and repeat are the two operations that apply it.

retry: On Failure, Try Again

#![allow(unused)]
fn main() {
use id_effect::Schedule;

let result = flaky_api_call()
    .retry(Schedule::exponential(Duration::from_millis(100)).take(3));
}

retry runs the effect. If it fails, it checks the schedule. If the schedule says "continue", it waits the indicated delay and tries again. When the schedule says "done" or the effect succeeds, retry returns.

Return value: the success value on success, or the last error if all retries are exhausted.

retry_while: Conditional Retry

Not all errors are retriable. Retry only when the error matches a condition:

#![allow(unused)]
fn main() {
let result = api_call()
    .retry_while(
        Schedule::exponential(Duration::from_millis(100)).take(5),
        |error| error.is_transient(),  // only retry transient errors
    );
}

Permanent errors (e.g., 404 Not Found, permission denied) shouldn't be retried — they won't go away. .retry_while lets you distinguish them.

repeat: On Success, Run Again

#![allow(unused)]
fn main() {
let polling = check_job_status()
    .repeat(Schedule::spaced(Duration::from_secs(5)));
}

repeat runs the effect. When it succeeds, it checks the schedule. If the schedule says "continue", it waits and runs again. This is the complement of retry: same mechanism, triggered by success instead of failure.

Use cases:

• Poll for job completion every 5 seconds
• Send heartbeats every 30 seconds
• Refresh a cache on a fixed interval

repeat_until: Stop When Condition Met

#![allow(unused)]
fn main() {
let waiting_for_ready = poll_service()
    .repeat_until(
        Schedule::spaced(Duration::from_secs(1)),
        |status| status == ServiceStatus::Ready,
    );
}

repeat_until repeats until the success value satisfies a predicate. When the condition is met, it stops and returns the value.

Composition with Other Operations

retry and repeat return effects — they compose like everything else:

#![allow(unused)]
fn main() {
// Retry the individual call, then repeat the whole batch
let batch = process_single_item(item)
    .retry(Schedule::exponential(100.ms()).take(3));

let continuous = batch
    .repeat(Schedule::spaced(Duration::from_secs(60)));
}

Error Information in Retry

If you need to inspect errors during retry (for logging, metrics, etc.):

#![allow(unused)]
fn main() {
let instrumented = risky_call()
    .retry_with_feedback(
        Schedule::exponential(100.ms()).take(3),
        |attempt, error| {
            // Called before each retry
            println!("Attempt {attempt} failed: {error:?}");
        },
    );
}

retry_with_feedback passes the attempt number and the error to a side-effectful callback before each retry. Useful for structured logging of retry behaviour.

Compute Fabric

Every effect in id_effect passes through Compute Fabric at the runtime boundary. You write lazy Effect descriptions and declare a ResourcePolicy; Fabric decides where, when, and how concurrently work runs.

Declarative policy

#![allow(unused)]
fn main() {
use id_effect::compute::{ResourcePolicy, MetricMode, MetricPolicy, RebalanceStrategy};

let policy = ResourcePolicy {
  memory: MetricPolicy::new(MetricMode::Max { ceiling: 0.85 }),
  cpu: MetricPolicy::new(MetricMode::Max { ceiling: 1.0 }),
  rebalance: RebalanceStrategy::ThrottleAdmission,
};
}

Memory capped at 85% while CPU may use all cores is:

#![allow(unused)]
fn main() {
let policy = ResourcePolicy::memory_cap_max_cpu(0.85);
}

Unlimited memory with even CPU spread across workers:

#![allow(unused)]
fn main() {
let policy = ResourcePolicy::unlimited_memory_cpu_spread(0.25);
}

Supervisor loop

ComputeSupervisor polls TelemetryEngine (live sysinfo on host; mock in tests) and compares readings to policy:

1. Monitor — TelemetrySnapshot { cpu_pct, mem_pct }
2. Admit — AdmissionController adjusts permit count
3. Place — fibers on FiberPool, collections via Rayon, I/O via Tokio
4. Rebalance — throttle, shed, scale-out, or scale-in

Each tick emits a ComputeEvent (SupervisorTick) for observability alongside FiberEvent.

Example: memory at 61% with an 85% ceiling → headroom → admit many fibers. Memory at 86% → throttle admission.

Installing Fabric

#![allow(unused)]
fn main() {
use id_effect::{ThreadSleepRuntime, run_fork, succeed};
use id_effect::compute::ComputeFabric;

let fabric = ComputeFabric::memory_cap_max_cpu(0.85);
let rt = ThreadSleepRuntime::with_fabric(fabric);
// fibers spawned via rt use the shared pool + admission
}

run_with installs a default fabric for the duration of each application run. run_blocking refreshes the thread-local AdaptiveContext on every entry.

See example 120_compute_fabric_memory_cap.rs.

Adaptive parallelism

AdaptiveContext holds the current admission budget, Rayon thread count, and auto-parallel element threshold. Bulk dispatch via parallel_if_profitable consults should_parallelize_current(len) instead of a fixed 1024 default when Fabric is installed.

• configure_rayon_threads — supervisor-sized Rayon pool
• CpuSpreadBucket — token bucket for MetricMode::Spread
• Example: 121_compute_fabric_cpu_spread.rs

See also Implicit Parallelism for the full 0.4.0 API surface (map / map_effect + *_serial).

Effect binds

Independent ~ steps in an effect! block may run concurrently when the Effect Dependency Graph finds no data or capability conflict. Dependent steps stay ordered. Use #[effect(serial)] to opt out.

Example: 122_compute_fabric_effect_parallel.rs (Stream::map_effect under admission budget).

Streams and collections

• Element threshold and Rayon pool size follow AdaptiveContext
• Stream::map_effect caps concurrent effect mappers from the admission budget
• Pure chunk ops (map, filter, …) use the same Fabric snapshot via parallel_if_profitable
• CPU spread mode caps per-worker share via CpuSpreadBucket

Cluster

When local Fabric saturates, RebalanceStrategy::ScaleOut builds a FabricJobSpec via ComputeSupervisor::scale_out. Convert with id_effect_jobs::JobSpec::from_fabric and enqueue on a FabricJobRunner. Durable cross-node steps hook through DistributedStepJournal (stub).

Example: 123_compute_fabric_cluster.rs (two-node offload with MemoryJobRunner).

ClusterResourcePolicy combines global and per-node caps with PlacementMode (LocalFirst, Spread, Affinity).

Further reading

• ADR 0007
• ADR 0008 — Fabric-only parallelism (0.4.0)
• Implicit Parallelism
• Concurrency
• Scheduling — temporal vs compute scheduling

Clock Injection — Testable Time

Schedule::exponential(100.ms()).take(3) with real retries takes ~700ms to test. Multiply by hundreds of tests and your test suite takes minutes. Clock injection solves this.

The Clock Trait

#![allow(unused)]
fn main() {
use id_effect::{Clock};

// The Clock trait abstracts time
trait Clock: Send + Sync {
    fn now(&self) -> UtcDateTime;
    fn sleep(&self, duration: Duration) -> Effect<(), Never, ()>;
}
}

All time-related operations in id_effect go through the current fiber's Clock. Replace the clock, and "time" moves as fast as you drive it.

Production: LiveClock

#![allow(unused)]
fn main() {
use id_effect::LiveClock;

// Uses real system time and tokio::time::sleep
let live_clock = LiveClock::new();
}

In production, inject LiveClock through the environment. Effect code never calls std::time::SystemTime::now() directly — it uses the injected clock.

Testing: TestClock

#![allow(unused)]
fn main() {
use id_effect::TestClock;

let clock = TestClock::new();

// The clock starts at epoch and doesn't advance on its own
assert_eq!(clock.now(), EPOCH);

// Advance time instantly
clock.advance(Duration::from_secs(60));
assert_eq!(clock.now(), EPOCH + 60s);
}

TestClock is deterministic. It advances only when you tell it to. Sleep effects don't wait — they check the clock, and if the clock is past their wake time, they return immediately.

Test Example

#![allow(unused)]
fn main() {
#[test]
fn exponential_retry_makes_three_attempts() {
    let clock = TestClock::new();
    let attempts = Arc::new(AtomicU32::new(0));

    let effect = {
        let attempts = attempts.clone();
        failing_operation(attempts.clone())
            .retry(Schedule::exponential(Duration::from_secs(1)).take(3))
    };

    // Fork the effect with the test clock
    let handle = effect.fork_with_clock(&clock);

    // Advance time to trigger each retry
    clock.advance(Duration::from_secs(1));   // retry 1
    clock.advance(Duration::from_secs(2));   // retry 2
    clock.advance(Duration::from_secs(4));   // retry 3 (exhausted)

    let exit = handle.join_blocking();
    assert!(matches!(exit, Exit::Failure(_)));
    assert_eq!(attempts.load(Ordering::Relaxed), 4);  // initial + 3 retries
}
}

The test runs in microseconds despite testing multi-second retry behaviour. No tokio::time::pause() hacks. No sleep(Duration::ZERO) workarounds.

Clock in the environment

Like other dependencies, Clock is a capability in Env:

#![allow(unused)]
fn main() {
use id_effect::{Effect, Clock, LiveClock, TestClock, caps, effect, provide, require, run_with};
use std::sync::Arc;
struct AppClock;

#[derive(::id_effect::ProviderSpecDerive)]
#[provides(AppClock)]
struct LiveClockLive;

impl LiveClockLive {
    fn new() -> Arc<dyn Clock> {
        Arc::new(LiveClock::new())
    }
}

fn now() -> Effect<UtcDateTime, Never, caps!(AppClock)> {
    effect!(|r| {
        let clock = ~AppClock;
        ~ clock.now()
    })
}
}

Production uses provide!(LiveClockLive); tests insert Arc::new(TestClock::new(...)) via build_env or a mock provider. Business logic is identical in both contexts.

CLI entrypoints with clap + Effect

Rust’s standard answer for parsing is clap. For id_effect, treat the binary as a thin shell:

1. Parse argv into a struct (#[derive(clap::Parser)]).
2. Assemble layers / MapConfigProvider / other R values from flags and environment.
3. Run your program Effect<A, E, R> with run_blocking (or an async driver when you integrate Tokio).
4. Exit with [std::process::ExitCode] using id_effect_cli helpers (see the next sections).

This mirrors Effect.ts @effect/cli: declarative argument models composed into a single program — here the “program” is an Effect value rather than a Future chain.

Recommended main shape (Rust 1.61+)

Returning ExitCode keeps main testable and avoids calling std::process::exit deep inside library code:

use clap::Parser;
use id_effect_cli::{run_main, RunMainConfig};

#[derive(Parser)]
struct Cli {
    #[arg(long)]
    name: String,
}

fn main() -> std::process::ExitCode {
    let cli = Cli::parse();
    let eff = my_app(cli.name);
    run_main(eff, my_env(), RunMainConfig::with_tracing())
}

my_app returns an Effect<…>; my_env() is whatever R your effect needs (often a Context built from Layer stacks).

Workspace helpers

The id_effect_cli crate (same repository) provides:

• Optional clap dependency (feature flag; on by default for convenience).
• run_main — optional tracing install, run_blocking, stderr logging for Err, ExitCode mapping.
• exit_code_for_exit / exit_code_for_cause when you already have an Exit from run_test or a supervisor.

Template

See the checked-in example examples/cli-minimal (--token …, Secret via id_effect_config).

Further reading

• Exit codes for main — Exit / Cause → ExitCode table
• Config + Secret from flags — wiring id_effect_config at the edge
• Error handling — Cause vs plain E
• Configuration (id_effect_config)

Exit codes for main

When a CLI finishes, the OS only sees an 8-bit exit status. id_effect distinguishes richer outcomes (Exit, Cause); at the process edge you collapse that into std::process::ExitCode.

Result from run_blocking

run_blocking returns Result<A, E>:

id_effect_cli::run_main uses this mapping and prints Err with Debug to stderr.

Full Exit / Cause

When you have an Exit (for example from tests or a custom driver), map leaf causes as follows — composite Cause::Both / Cause::Then use the maximum byte so “stronger” failures dominate:

Helpers: exit_code_for_exit, exit_code_for_cause, cause_max_exit_byte.

Practical guidance

• Use 1 for “the command could not complete successfully” (missing flag, validation error, upstream HTTP 4xx mapped to your E, …).
• Reserve 101 for “the program detected an internal defect” — corresponds to Cause::Die in structured runs (see cause_max_exit_byte).
• Use 130 only when you surface fiber interruption to the process edge (rare in simple CLIs).

If you need richer machine output, prefer stderr JSON or a dedicated output file — do not overload exit codes beyond what your operators can act on.

Config + Secret from flags

id_effect_config already documents providers and Config descriptors. At a CLI edge, you usually:

1. Parse a --token / --api-key flag (or read a path to a file).
2. Seed a MapConfigProvider (or EnvConfigProvider) so the key matches what your Config::* descriptors expect.
3. Evaluate Config::… inside an Effect using config_env in R.
4. Wrap sensitive strings with Secret via Config::secret so logs and Debug never print raw material.

Minimal snippet

#![allow(unused)]
fn main() {
use id_effect::Effect;
use id_effect_config::{
    Config, ConfigError, MapConfigProvider, Secret, config_env,
};

fn load_api_token() -> Effect<Secret<String>, ConfigError, id_effect_config::ConfigEnv> {
    Config::string("API_TOKEN").secret().run::<Secret<String>, ConfigError, _>()
}

// In main: build provider from CLI flag, then `config_env(provider)` and `run_blocking`.
}

The repository ships a full runnable layout under examples/cli-minimal (cli_minimal package in the workspace).

Operational notes

• Prefer short-lived exposure of secrets: parse → wrap in Secret → pass to effects; avoid storing raw String copies in globals.
• For file-based secrets, read bytes in an effect and wrap immediately; redact paths in error messages if they reveal usernames.
• Combine with CLI with clap for argv parsing and exit codes for main.

Software Transactional Memory — Optimistic Concurrency

Shared mutable state is hard. Mutexes work but compose poorly: lock two mutexes in the wrong order and you deadlock. Lock them separately and you get torn reads. Lock the whole world and you serialise unnecessarily.

Software Transactional Memory (STM) takes a different approach: every operation on shared state runs inside a transaction. Transactions commit atomically or roll back and retry. No explicit locks. No deadlocks. No torn reads.

This chapter covers id_effect's STM implementation: Stm, TRef, commit, and the transactional collection types.

Why STM? — The Shared State Problem

Consider transferring money between two accounts. With mutexes:

#![allow(unused)]
fn main() {
fn transfer(from: &Mutex<Account>, to: &Mutex<Account>, amount: u64) {
    let from_guard = from.lock().unwrap();
    // Thread B might be doing transfer(to, from, ...) right here
    let to_guard = to.lock().unwrap();
    // DEADLOCK: Thread A holds from, waiting for to.
    //           Thread B holds to, waiting for from.
    from_guard.balance -= amount;
    to_guard.balance += amount;
}
}

The standard fix (always lock in a consistent order) requires global coordination across your codebase. Add a third account and you need to sort three locks. It doesn't compose.

STM: Optimistic Concurrency

STM operates on the assumption that conflicts are rare. Instead of locking, it:

1. Reads current values into a local transaction log
2. Computes new values based on those reads
3. Attempts to commit: checks that nothing changed since the reads, then atomically writes

If anything changed between step 1 and step 3, the transaction retries automatically from step 1.

#![allow(unused)]
fn main() {
use id_effect::{TRef, stm, commit};

fn transfer(from: &TRef<Account>, to: &TRef<Account>, amount: u64)
-> Effect<(), TransferError, ()>
{
    commit(stm! {
        let from_acct = ~ from.read_stm();
        let to_acct   = ~ to.read_stm();

        if from_acct.balance < amount {
            ~ stm::fail(TransferError::InsufficientFunds);
        }

        ~ from.write_stm(Account { balance: from_acct.balance - amount, ..from_acct });
        ~ to.write_stm(Account { balance: to_acct.balance + amount, ..to_acct });
        ()
    })
}
}

No locks. No deadlock risk. The transaction retries automatically if another transaction modified either account between our read and our write.

When STM Wins

STM shines when:

• You need to update multiple values atomically
• You're composing smaller transactional operations into larger ones
• Contention is low (retries are cheap)

Avoid STM for long-running operations that do I/O — transactions should be short and pure. The stm! macro is for read-modify-write, not for network calls.

TRef — Transactional References

TRef<T> is the fundamental mutable cell in id_effect's STM system. It wraps a value that can be read and written inside transactions.

Creating a TRef

#![allow(unused)]
fn main() {
use id_effect::TRef;

let counter: TRef<i32> = TRef::new(0);
let balance: TRef<f64> = TRef::new(1000.0);
}

TRef::new(value) creates a transactional reference with an initial value. TRefs are typically created once (at startup or when initialising shared state) and then shared across fibers via Arc.

Transactional Operations

All TRef operations return Stm<_> — transactional descriptions, not effects. They only work inside stm! (or when run through commit/atomically):

#![allow(unused)]
fn main() {
use id_effect::{TRef, stm};

let counter = TRef::new(0);

// Read inside a transaction
let read_op: Stm<i32> = counter.read_stm();

// Write inside a transaction
let write_op: Stm<()> = counter.write_stm(42);

// Modify (read-write atomically)
let modify_op: Stm<()> = counter.modify_stm(|n| n + 1);
}

These are descriptions. Nothing happens until they're committed.

Running Inside stm!

The stm! macro provides do-notation for composing Stm operations, exactly like effect! does for Effect:

#![allow(unused)]
fn main() {
use id_effect::{stm, TRef};

let counter = TRef::new(0_i32);
let total   = TRef::new(0_i32);

let transaction: Stm<()> = stm! {
    let count = ~ counter.read_stm();
    let sum   = ~ total.read_stm();
    ~ counter.write_stm(count + 1);
    ~ total.write_stm(sum + count);
    ()
};
}

Sharing TRefs

TRefs are Clone + Send + Sync. Wrap in Arc to share across fibers:

#![allow(unused)]
fn main() {
use std::sync::Arc;

let shared: Arc<TRef<i32>> = Arc::new(TRef::new(0));

// Clone for each fiber
let clone1 = Arc::clone(&shared);
let clone2 = Arc::clone(&shared);

fiber_all(vec![
    increment_n_times(clone1, 1000),
    increment_n_times(clone2, 1000),
])
// Result: counter = 2000 (atomically, without locks)
}

TRef vs. Mutex

Use TRef for short, composable state mutations. Use Mutex when you need to hold a lock across I/O (though ideally you redesign to avoid that).

Stm and commit — Building Transactions

The stm! macro produces Stm<A> values — descriptions of transactional computations. To execute them, you use commit or atomically.

commit: Lift Stm into Effect

#![allow(unused)]
fn main() {
use id_effect::{commit, Stm, Effect};

let transaction: Stm<i32> = stm! {
    let a = ~ ref_a.read_stm();
    let b = ~ ref_b.read_stm();
    a + b
};

// Lift into Effect
let effect: Effect<i32, Never, ()> = commit(transaction);

// Now run it
let result = run_blocking(effect)?;
}

commit wraps a Stm in an effect that, when run, executes the transaction and retries if there's a conflict. The E type of commit(stm) is Never unless the Stm can fail (see stm::fail).

atomically: Direct Execution

#![allow(unused)]
fn main() {
use id_effect::atomically;

// Run a transaction immediately in the current context
let value: i32 = atomically(stm! {
    ~ counter.modify_stm(|n| n + 1);
    ~ counter.read_stm()
});
}

atomically is the synchronous equivalent of commit + run_blocking. Use it when you're already outside the effect system and need a quick transactional update.

stm::fail: Transactional Errors

Transactions can fail with typed errors:

#![allow(unused)]
fn main() {
use id_effect::stm;

fn withdraw(account: &TRef<u64>, amount: u64) -> Stm<u64> {
    stm! {
        let balance = ~ account.read_stm();
        if balance < amount {
            ~ stm::fail(InsufficientFunds);  // abort the transaction
        }
        ~ account.write_stm(balance - amount);
        balance - amount
    }
}

// commit propagates the error into E
let effect: Effect<u64, InsufficientFunds, ()> = commit(withdraw(&account, 100));
}

stm::fail(e) aborts the current transaction with error e. The transaction is not retried — it fails immediately with the given error.

stm::retry: Block Until Condition

Sometimes a transaction should wait until a condition is true rather than failing:

#![allow(unused)]
fn main() {
// Block (retry) until the queue has items
fn dequeue(queue: &TRef<Vec<Item>>) -> Stm<Item> {
    stm! {
        let items = ~ queue.read_stm();
        if items.is_empty() {
            ~ stm::retry();  // block until queue changes, then retry
        }
        let item = items[0].clone();
        ~ queue.write_stm(items[1..].to_vec());
        item
    }
}
}

stm::retry() doesn't mean "try again immediately." It means "block until any TRef I read has changed, then try again." This is how TQueue implements blocking dequeue without busy-waiting.

Composing Transactions

Transactions compose by sequencing stm! blocks:

#![allow(unused)]
fn main() {
let big_transaction: Stm<()> = stm! {
    // Sub-transaction 1
    let _ = ~ transfer_funds(&from, &to, amount);
    // Sub-transaction 2
    let _ = ~ record_audit_log(&from, &to, amount);
    ()
};

// Both operations commit atomically or neither does
let effect = commit(big_transaction);
}

The composed transaction retries as a unit — if either sub-operation sees a conflict, the whole thing restarts from the beginning.

TQueue, TMap, TSemaphore — Transactional Collections

id_effect provides STM-aware versions of common collection types. They compose with other STM operations and integrate with stm!.

TQueue: Bounded Transactional Queue

#![allow(unused)]
fn main() {
use id_effect::TQueue;

let queue: TQueue<Job> = TQueue::bounded(100);

// Enqueue (blocks/retries if full)
let offer: Stm<()> = queue.offer_stm(job);

// Dequeue (blocks/retries if empty)
let take: Stm<Job> = queue.take_stm();

// Peek without removing
let peek: Stm<Option<Job>> = queue.peek_stm();

// Non-blocking try
let try_take: Stm<Option<Job>> = queue.try_take_stm();
}

TQueue::bounded(n) creates a queue with capacity n. offer_stm blocks (via stm::retry) when full; take_stm blocks when empty. Both integrate naturally with stm!.

Producer-Consumer Pattern

#![allow(unused)]
fn main() {
fn producer(queue: Arc<TQueue<Job>>, jobs: Vec<Job>) -> Effect<(), Never, ()> {
    effect! {
        for job in jobs {
            ~ commit(queue.offer_stm(job));
        }
        Ok(())
    }
}

fn consumer(queue: Arc<TQueue<Job>>) -> Effect<Never, Never, ()> {
    effect! {
        loop {
            let job = ~ commit(queue.take_stm());  // blocks if empty
            ~ process_job(job);
        }
    }
}
}

TMap: Transactional Hash Map

#![allow(unused)]
fn main() {
use id_effect::TMap;

let map: TMap<String, User> = TMap::new();

// Inside stm!:
let insert: Stm<()> = map.insert_stm("alice".into(), alice_user);
let get:    Stm<Option<User>> = map.get_stm("alice");
let remove: Stm<Option<User>> = map.remove_stm("alice");
let update: Stm<()> = map.modify_stm("alice", |u| { u.name = "ALICE".into(); u });
}

TMap is a concurrent hash map where all operations participate in STM transactions. Reading from TMap and TRef in the same transaction is atomic:

#![allow(unused)]
fn main() {
commit(stm! {
    let user = ~ user_map.get_stm("alice");
    let count = ~ access_counter.read_stm();
    ~ access_counter.write_stm(count + 1);
    user
})
// Either the map read AND the counter increment happen, or neither does
}

TSemaphore: Transactional Semaphore

#![allow(unused)]
fn main() {
use id_effect::TSemaphore;

// Create a semaphore with 10 permits
let sem: TSemaphore = TSemaphore::new(10);

// Acquire 1 permit (blocks if none available)
let acquire: Stm<()> = sem.acquire_stm(1);

// Release 1 permit
let release: Stm<()> = sem.release_stm(1);
}

TSemaphore limits concurrent access to a resource. Use it with acquire_release for resource pools where you want transactional semantics:

#![allow(unused)]
fn main() {
commit(stm! {
    ~ sem.acquire_stm(1);  // blocks until permit available
    ()
}).flat_map(|()| {
    do_limited_work().flat_map(|result| {
        commit(sem.release_stm(1)).map(|()| result)
    })
})
}

Summary

All compose inside stm! and commit atomically with other STM operations.

Streams — Backpressure and Chunked Processing

An Effect produces one value. A Stream produces many values over time. When you need to process a potentially infinite or very large sequence — database result sets, event logs, file lines, sensor readings — Stream is the right abstraction.

This chapter covers when to use Stream vs Effect, how streams process data in Chunks for efficiency, how to control flow with backpressure policies, and how to consume streams with Sink.

Stream vs Effect — When to Use Each

The choice is simple: how many values does your computation produce?

Effect<A, E, R>  → produces exactly one A (or fails)
Stream<A, E, R>  → produces zero or more A values over time (or fails)

Concrete Examples

#![allow(unused)]
fn main() {
// Effect: get one user
fn get_user(id: u64) -> Effect<User, DbError, Db>

// Stream: all users, one at a time
fn all_users() -> Stream<User, DbError, Db>

// Effect: count rows
fn count_orders() -> Effect<u64, DbError, Db>

// Stream: export all orders for a report
fn export_orders() -> Stream<Order, DbError, Db>
}

If you fetch 10 million rows into a Vec and return it as an Effect, you'll run out of memory. A Stream loads and processes them incrementally.

Stream Transformations

Stream has the same transformation API as Effect:

#![allow(unused)]
fn main() {
all_users()
    .filter(|u| u.is_active())
    .map(|u| UserSummary::from(u))
    .take(100)
}

.map, .filter, .flat_map, .take, .drop, .zip — all work on streams. None of them load the whole stream into memory; they process elements as they arrive.

Collecting a Stream into an Effect

When you do need all the results:

#![allow(unused)]
fn main() {
let users: Effect<Vec<User>, DbError, Db> = all_users().collect();
}

.collect() consumes the stream and accumulates into a Vec. Use only when the full result fits in memory.

For large results, prefer a fold or a sink:

#![allow(unused)]
fn main() {
let count: Effect<usize, DbError, Db> = all_users().fold(0, |acc, _| acc + 1);
}

Converting Effect to Stream

Wrap an Effect in a single-element stream when you need to compose with streaming operators:

#![allow(unused)]
fn main() {
use id_effect::Stream;

let single_user_stream: Stream<User, DbError, Db> = Stream::from_effect(get_user(1));

// Now compose with other streams
let combined = single_user_stream.chain(all_users());
}

The Rule

• Need one result: Effect
• Need to process many results without loading all at once: Stream
• Need to compose multiple streams: Stream with chain, zip, merge
• Need all results in memory: Stream + .collect() (with appropriate size caution)

Chunks — Efficient Batched Processing

A Stream doesn't emit elements one at a time at the memory level. It emits them in Chunks — contiguous, fixed-capacity batches. Most of the time you don't interact with Chunk directly; the stream API works element-wise and handles chunking internally. But understanding chunks helps you tune performance.

What Is a Chunk

#![allow(unused)]
fn main() {
use id_effect::Chunk;

// A Chunk is a fixed-capacity contiguous sequence
let chunk: Chunk<i32> = Chunk::from_vec(vec![1, 2, 3, 4, 5]);

// Access elements
let first: Option<&i32> = chunk.first();
let len: usize = chunk.len();

// Iterate
for item in &chunk {
    println!("{item}");
}
}

A Chunk<A> is essentially a smart Arc<[A]> slice: cheap to clone (reference-counted), cache-friendly (contiguous layout), and zero-copy when slicing.

Why Chunks Exist

Processing elements one at a time through a chain of .map and .filter calls has overhead: each step is a separate allocation or indirection. Chunks amortize that cost:

Single-element model:
  elem1 → map → filter → emit → elem2 → map → filter → emit → ...
  (N function calls for N elements through each operator)

Chunk model:
  chunk[1..64] → map_chunk → filter_chunk → emit_chunk → ...
  (N/64 overhead calls; SIMD-friendly layout)

The default chunk size is 64 elements. You can change it when constructing a stream:

#![allow(unused)]
fn main() {
let stream = Stream::from_iter(0..1_000_000)
    .with_chunk_size(256);
}

Larger chunks improve throughput for CPU-bound map/filter operations. Smaller chunks reduce latency when downstream consumers need to act quickly.

Working with Chunks Directly

Most operators are element-wise, but a few operate at the chunk level for efficiency:

#![allow(unused)]
fn main() {
// map_chunks: apply a function to entire chunks at once
stream.map_chunks(|chunk| {
    chunk.map(|x| x * 2)  // vectorizable
})

// flat_map_chunks: emit a new chunk per input chunk
stream.flat_map_chunks(|chunk| {
    Chunk::from_iter(chunk.iter().flat_map(expand))
})
}

Use map_chunks when your transformation is pure and benefits from batching (e.g., numeric processing, serialisation).

Chunk in Sinks and Collectors

When a Sink receives data, it receives Chunks. Custom sinks that write to a file or network socket often want to write whole chunks at once:

#![allow(unused)]
fn main() {
impl Sink<Bytes> for FileSink {
    fn on_chunk(&mut self, chunk: Chunk<Bytes>) -> Effect<(), IoError, ()> {
        // write all bytes in one system call
        effect! {
            for bytes in &chunk {
                ~ self.write(bytes);
            }
            ()
        }
    }
}
}

Building Chunks

#![allow(unused)]
fn main() {
// From an iterator
let chunk = Chunk::from_iter([1, 2, 3]);

// From a Vec (no copy if Vec capacity matches)
let chunk = Chunk::from_vec(v);

// Empty chunk
let empty: Chunk<i32> = Chunk::empty();

// Single element
let one = Chunk::single(42);

// Concatenate two chunks (zero-copy if they're adjacent)
let combined = Chunk::concat(chunk_a, chunk_b);
}

Summary

You rarely construct Chunk by hand in application code. The stream runtime handles chunking for you. Understand chunks when:

• You're writing a custom Sink and want efficient writes
• You're tuning throughput with .with_chunk_size(n)
• You're implementing a library operator with map_chunks

Backpressure Policies — Controlling Flow

A stream is a pipeline: a producer emits data, operators transform it, a consumer processes it. Problems arise when the producer is faster than the consumer. Backpressure is the mechanism that handles this mismatch.

The Problem

Producer: emits 10,000 events/sec
Consumer: processes 1,000 events/sec

What happens to the 9,000 surplus events per second?

Your options are: block the producer, drop events, or buffer them. Each is correct in different contexts. id_effect makes the choice explicit via BackpressurePolicy.

BackpressurePolicy

#![allow(unused)]
fn main() {
use id_effect::BackpressurePolicy;

// Block the producer until the consumer catches up (default for bounded channels)
BackpressurePolicy::Block

// Drop the newest events when the buffer is full
BackpressurePolicy::DropLatest

// Drop the oldest events when the buffer is full (keep the freshest data)
BackpressurePolicy::DropOldest

// Unbounded buffering — never drop, never block (use carefully)
BackpressurePolicy::Unbounded
}

Applying a Policy to a Channel-Backed Stream

The most common place to specify backpressure is when bridging from a channel to a Stream:

#![allow(unused)]
fn main() {
use id_effect::{stream_from_channel_with_policy, BackpressurePolicy};
use std::sync::mpsc;

let (tx, rx) = mpsc::channel::<Event>();

// Drop old events; always reflect the latest state
let stream = stream_from_channel_with_policy(rx, 1024, BackpressurePolicy::DropOldest);
}

Contrast with stream_from_channel, which uses Block by default. If you don't think about backpressure at this point, Block is the safe choice — you won't lose data, but a slow consumer will slow down the producer.

Choosing a Policy

Stream-Level Backpressure

Streams composed with flat_map or merge also have implicit backpressure: downstream operators signal upstream when they can accept more work. This happens automatically and doesn't require a policy setting — the Stream runtime handles it.

For explicit control over concurrency in flat_map:

#![allow(unused)]
fn main() {
stream
    .flat_map_with_concurrency(4, |id| fetch_record(id))
    // Only 4 fetch_record effects run concurrently
    // Others wait until a slot frees — natural backpressure
}

Monitoring Drops

When using DropLatest or DropOldest, you often want to know how many events were dropped:

#![allow(unused)]
fn main() {
let (stream, dropped_counter) = stream_from_channel_with_policy_and_counter(
    rx,
    1024,
    BackpressurePolicy::DropOldest,
);

// Periodically log the counter
effect! {
    loop {
        let n = dropped_counter.load(Ordering::Relaxed);
        if n > 0 {
            ~ log.warn(format!("Dropped {n} events due to backpressure"));
        }
        ~ sleep(Duration::from_secs(10));
    }
}
}

Summary

Always choose a backpressure policy explicitly. The default (Block) is safe but can stall producers. DropOldest is often right for real-time data. DropLatest is right when order matters but throughput doesn't. Unbounded is only acceptable when the rate is truly bounded by the domain.

Sinks — Consuming Streams

A Stream describes a sequence of values. A Sink describes how to consume them. Together they form a complete pipeline: Stream → operators → Sink.

Built-in Sinks

Most of the time you don't write a Sink explicitly — you use one of the consuming methods on Stream:

#![allow(unused)]
fn main() {
// Collect all elements into a Vec
let users: Effect<Vec<User>, DbError, Db> = all_users().collect();

// Fold into a single value
let total: Effect<u64, DbError, Db> = orders()
    .fold(0u64, |acc, order| acc + order.amount);

// Run a side-effecting action for each element
let logged: Effect<(), DbError, Db> = events()
    .for_each(|event| log_event(event));

// Drain (discard all values, run for side effects only)
let drained: Effect<(), DbError, Db> = events()
    .map(|e| emit_metric(e))
    .drain();

// Take the first N elements
let first_ten: Effect<Vec<User>, DbError, Db> = all_users()
    .take(10)
    .collect();
}

Each of these methods turns a Stream<A, E, R> into an Effect<B, E, R>, which you can then run with run_blocking or compose further.

The Sink Trait

When the built-in consumers aren't enough, implement Sink:

#![allow(unused)]
fn main() {
use id_effect::{Sink, Chunk, Effect};

struct CsvWriter {
    path: PathBuf,
    written: usize,
}

impl Sink<Record> for CsvWriter {
    type Error = IoError;
    type Env   = ();

    fn on_chunk(
        &mut self,
        chunk: Chunk<Record>,
    ) -> Effect<(), IoError, ()> {
        effect! {
            for record in &chunk {
                ~ self.write_csv_line(record);
            }
            ()
        }
    }

    fn on_done(&mut self) -> Effect<(), IoError, ()> {
        effect! {
            ~ self.flush();
            ()
        }
    }
}
}

on_chunk is called for each chunk of elements. on_done is called once when the stream ends — use it to flush buffers or close handles.

Running a Stream into a Sink

#![allow(unused)]
fn main() {
let writer = CsvWriter::new("output.csv");

let effect: Effect<(), IoError, Db> = all_records()
    .run_into_sink(writer);

run_blocking(effect, env)?;
}

run_into_sink drives the stream and feeds each chunk to the sink. If the stream fails, on_done is not called — use resource scopes around the sink when cleanup is unconditionally required.

Sink Composition

Sinks can be composed: a ZipSink feeds the same stream to two sinks simultaneously:

#![allow(unused)]
fn main() {
let count_sink  = CountSink::new();
let csv_sink    = CsvWriter::new("out.csv");

// Both sinks receive every element
let combined = ZipSink::new(count_sink, csv_sink);

all_records().run_into_sink(combined)
}

Each element is delivered to both sinks in order. If either sink fails, the whole pipeline fails.

Finite vs Infinite Streams and Sinks

A Sink doesn't know whether its stream is finite or infinite. Combine with take, take_while, or take_until to bound an infinite stream before running it into a sink:

#![allow(unused)]
fn main() {
// Process at most 1 hour of events
let one_hour = Duration::from_secs(3600);

event_stream()
    .take_until(sleep(one_hour))
    .for_each(|event| process(event))
}

Summary

Implicit Parallelism — Compute Fabric

Bulk transforms and effectful stream steps parallelize implicitly through Compute Fabric. There is no public Parallelism type and no *_with / *_par policy surface — Fabric decides when Rayon or bounded effect concurrency is profitable.

Use the primary method name (map, filter, sort_with, …). Opt out with *_serial when you need FnMut, non-Send closures, or deterministic ordering.

How dispatch works

Every run_blocking / run_async / run_with boundary refreshes the thread-local AdaptiveContext from the installed supervisor. Bulk paths call parallel_if_profitable, which consults should_parallelize_current(len) and runs Rayon via install_parallel when the snapshot says it pays off.

Under memory pressure the supervisor lowers admission and raises the element threshold; under headroom it admits more fibers and parallelizes smaller chunks. See example 120_compute_fabric_memory_cap.rs.

Collections and vec

#![allow(unused)]
fn main() {
use id_effect::vec;

let doubled = vec::map(vec![1, 2, 3], |x| x * 2);
// Fabric may use Rayon when len is large enough and headroom allows.
}

Escape hatch for captured mutable state or non-Send closures:

#![allow(unused)]
fn main() {
use id_effect::vec;

let mut acc = 0;
vec::map_serial(v, |x| { acc += x; acc });
}

The same pattern applies to HashMap::map_values, filter, red-black tree scans, sort_with, and similar bulk APIs.

Streams — pure transforms

Stream::map and Stream::filter apply Fabric-aware Rayon per upstream chunk:

#![allow(unused)]
fn main() {
use id_effect::Stream;

Stream::from_iterable(0..10_000)
    .map(|n| n * 2)
    .filter(|n| *n % 2 == 0)
    .run_collect();
}

For FnMut or ordering-sensitive work, use map_serial / filter_serial:

#![allow(unused)]
fn main() {
let mut offset = 0;
stream.map_serial(move |n| { offset += 1; n + offset });
}

See example 071_stream_map_serial.rs.

Streams — effectful steps

When each element runs an Effect, use Stream::map_effect. Concurrency is capped by the current admission budget (at least one permit):

#![allow(unused)]
fn main() {
use id_effect::{Stream, succeed};

Stream::from_iterable(items)
    .map_effect(|item| process(item))
    .run_collect();
}

Output order matches stream order. Each mapper receives a clone of the environment R, so R must be Clone + Send + Sync.

With Compute Fabric installed, a memory cap or throttle directly reduces how many effect mappers run in flight. See example 122_compute_fabric_effect_parallel.rs.

effect! and the Effect Dependency Graph

Independent ~ steps with no data or capability conflict may run concurrently when the EDG finds a parallel bind set. Dependent steps and overlapping capability borrows stay sequential.

Opt out per block:

#![allow(unused)]
fn main() {
#[effect(serial)]
effect! { |r| {
    ~step_a();
    ~step_b();
}}
}

Example: 122_compute_fabric_effect_parallel.rs (stream effect mapping under Fabric). For multi-step programs, see Compute Fabric § Effect binds.

Migration from 0.3.x

See ADR 0008 (supersedes the public surface of ADR 0006).

Schema — Parse, Don't Validate

Data enters your program from the outside world: HTTP request bodies, database rows, configuration files, message queue payloads. All of it is untrusted. All of it needs to be checked.

The naive approach is to deserialise first and validate later — accept a User struct via serde, then check that email is non-empty and age is positive in a separate step. The problem: your type says User but your program has a User that might have an empty email. The type lies.

The better approach is parse, don't validate: transform untrusted input into trusted types in one step. If the parse succeeds, you have a valid User. If it fails, you have a structured ParseError that tells you exactly what was wrong.

id_effect's schema module is built on this principle.

What This Chapter Covers

• Unknown — the type for unvalidated wire data (next section)
• Schema combinators — the building blocks for describing data shapes (ch14-02)
• Validation and refinement — refine, filter, and Brand for domain constraints (ch14-03)
• ParseErrors — structured, accumulating error reports (ch14-04)

The Unknown Type — Unvalidated Wire Data

Unknown is the type for data that hasn't been validated yet. Think of it as a typed serde_json::Value — it can hold any shape of data, but you can't do anything useful with it until you run it through a schema.

Creating Unknown Values

#![allow(unused)]
fn main() {
use id_effect::schema::Unknown;

// From a JSON string
let u: Unknown = Unknown::from_json_str(r#"{"name": "Alice", "age": 30}"#)?;

// From a serde_json Value
let v: serde_json::Value = serde_json::json!({ "name": "Alice" });
let u: Unknown = Unknown::from_serde_json(v);

// From raw parts
let u: Unknown = Unknown::object([
    ("name", Unknown::string("Alice")),
    ("age",  Unknown::integer(30)),
]);

// Primitives
let s: Unknown = Unknown::string("hello");
let n: Unknown = Unknown::integer(42);
let b: Unknown = Unknown::boolean(true);
let null: Unknown = Unknown::null();
let arr: Unknown = Unknown::array([Unknown::integer(1), Unknown::integer(2)]);
}

Why Not serde_json::Value Directly?

serde_json::Value is an excellent data type, but it's stringly typed: value["name"] gives you an Option<&Value> and there's no structure around parse errors, path tracking, or accumulation. Unknown wraps the same idea but integrates with id_effect's schema parser, which gives you:

• Path tracking — "error at .users[3].email"
• Accumulated errors — all failures in one parse, not just the first
• Composable schemas — build complex validators from simple primitives

Inspecting Unknown Values

You don't normally inspect Unknown directly — you run it through a schema. But when debugging:

#![allow(unused)]
fn main() {
// Check what shape the value has
match u.kind() {
    UnknownKind::Object(fields) => { /* … */ }
    UnknownKind::Array(elems)   => { /* … */ }
    UnknownKind::String(s)      => { /* … */ }
    UnknownKind::Integer(n)     => { /* … */ }
    UnknownKind::Float(f)       => { /* … */ }
    UnknownKind::Boolean(b)     => { /* … */ }
    UnknownKind::Null            => { /* … */ }
}

// Access a field without parsing (returns Option<&Unknown>)
let name: Option<&Unknown> = u.field("name");
}

The Parse Boundary

Unknown is your import type. At every IO boundary — HTTP handler, NATS message, config file, database row — convert incoming data to Unknown first, then parse it with a schema:

#![allow(unused)]
fn main() {
async fn handle_request(body: Bytes) -> Effect<CreateUserResponse, ApiError, Deps> {
    effect! {
        // Convert raw bytes to Unknown
        let raw = Unknown::from_json_bytes(&body)
            .map_err(ApiError::InvalidJson)?;

        // Parse Unknown into a typed, validated struct
        let req: CreateUserRequest = ~ parse_schema(create_user_schema(), raw);

        // Now req is fully trusted — proceed with domain logic
        ~ create_user(req)
    }
}
}

Nothing beyond the parse boundary sees Unknown. Domain functions only accept validated types.

Unknown and Serde

If you have existing serde-deserializable types, use the serde bridge (requires the schema-serde feature):

#![allow(unused)]
fn main() {
use id_effect::schema::serde_bridge::unknown_from_serde_json;

// Deserialise via serde, then convert to Unknown for schema validation
let value: serde_json::Value = serde_json::from_str(input)?;
let u: Unknown = unknown_from_serde_json(value);
}

This lets you incrementally adopt the schema system without rewriting all your serde impls at once.

Schema Combinators — Describing Data Shapes

A schema is a value that describes how to parse an Unknown into a typed result. Schemas compose: build small schemas for primitive types, then combine them into schemas for complex structures.

Primitive Schemas

#![allow(unused)]
fn main() {
use id_effect::schema::{string, integer, i64, f64, boolean, null};

// Parse a string
let name_schema = string();

// Parse an integer (i64)
let age_schema = i64();

// Parse a float
let price_schema = f64();

// Parse a boolean
let active_schema = boolean();
}

Each schema has type Schema<T> — string() is a Schema<String>, i64() is a Schema<i64>, and so on.

Struct Schemas

#![allow(unused)]
fn main() {
use id_effect::schema::struct_;

#[derive(Debug)]
struct User {
    name: String,
    age:  i64,
}

let user_schema = struct_!(User {
    name: string(),
    age:  i64(),
});
}

struct_! maps field names to their schemas and constructs the target type. If any field is missing or has the wrong type, parsing fails with a ParseError that includes the field path.

For schemas without a derive macro, use object:

#![allow(unused)]
fn main() {
use id_effect::schema::object;

let user_schema = object([
    ("name", string().map(|s| s)),
    ("age",  i64()),
]).map(|(name, age)| User { name, age });
}

Optional Fields

#![allow(unused)]
fn main() {
use id_effect::schema::optional;

struct Config {
    host:    String,
    port:    Option<u16>,
    timeout: Option<Duration>,
}

let config_schema = struct_!(Config {
    host:    string(),
    port:    optional(u16()),
    timeout: optional(duration_ms()),
});
}

optional(schema) produces Schema<Option<T>>. A missing field or null both parse as None.

Array Schemas

#![allow(unused)]
fn main() {
use id_effect::schema::array;

// Vec of strings
let tags_schema: Schema<Vec<String>> = array(string());

// Vec of User
let users_schema: Schema<Vec<User>> = array(user_schema);
}

array(item_schema) parses a JSON array where each element is validated by item_schema. Errors include the index: "[2].email: expected string, got null".

Union Schemas

#![allow(unused)]
fn main() {
use id_effect::schema::{union_, literal_string};

#[derive(Debug)]
enum Status { Active, Inactive, Pending }

let status_schema = union_![
    literal_string("active")   => Status::Active,
    literal_string("inactive") => Status::Inactive,
    literal_string("pending")  => Status::Pending,
];
}

union_! tries each branch in order and returns the first that succeeds. Errors report all branches that failed.

Transforming Schemas

Schemas are values — you can .map them:

#![allow(unused)]
fn main() {
// Parse a string and convert it to uppercase
let upper_schema: Schema<String> = string().map(|s| s.to_uppercase());

// Parse a string and try to convert to a domain type
let email_schema: Schema<Email> = string().try_map(|s| {
    Email::parse(s).map_err(ParseError::custom)
});
}

.map transforms on success. .try_map can fail and produce a ParseError.

Running a Schema

#![allow(unused)]
fn main() {
use id_effect::schema::parse;

let raw: Unknown = Unknown::from_json_str(r#"{"name":"Alice","age":30}"#)?;

match parse(user_schema, raw) {
    Ok(user)  => println!("Got: {user:?}"),
    Err(errs) => println!("Errors: {errs}"),
}
}

parse returns Result<T, ParseErrors>. ParseErrors accumulates all errors — not just the first — so a caller gets the complete picture of what's wrong.

Schema as a Type Contract

A schema is documentation. Where you use Schema<CreateUserRequest>, readers know: this function requires exactly this shape of data, checked at runtime. The schema is the spec.

#![allow(unused)]
fn main() {
pub fn create_user_handler() -> impl Fn(Unknown) -> Effect<User, ApiError, Db> {
    let schema = create_user_schema();
    move |raw| {
        effect! {
            let req = parse(schema.clone(), raw)
                .map_err(ApiError::Validation)?;
            ~ db_create_user(req)
        }
    }
}
}

Validation and Refinement — Constrained Types

Schemas parse structure. Validation adds constraints: an age must be positive, an email must contain @, a price must have at most two decimal places. Refinement goes further: a validated Email is a different type from a raw String, so you can never accidentally pass an unvalidated string where an email is expected.

refine: Attach a Predicate

refine takes a schema and a predicate. Parsing succeeds only if both the schema's parse and the predicate pass:

#![allow(unused)]
fn main() {
use id_effect::schema::{string, i64, refine};

// Age must be between 0 and 150
let age_schema = refine(
    i64(),
    |n| (0..=150).contains(n),
    "age must be between 0 and 150",
);

// Non-empty string
let non_empty = refine(
    string(),
    |s: &String| !s.is_empty(),
    "must not be empty",
);
}

If the predicate returns false, parsing fails with a ParseError containing the message you provided.

filter: Same as refine, Different Style

filter is an alias for refine with a closure-first signature, matching Rust iterator conventions:

#![allow(unused)]
fn main() {
let positive = i64().filter(|n| *n > 0, "must be positive");
let trimmed  = string().filter(|s| s == s.trim(), "must not have leading/trailing whitespace");
}

Use whichever reads more naturally.

try_map: Fallible Transformation

When conversion logic can fail — parsing a date, constructing a URL, validating an email — use .try_map:

#![allow(unused)]
fn main() {
use id_effect::schema::ParseError;

let url_schema = string().try_map(|s| {
    url::Url::parse(&s).map_err(|e| ParseError::custom(format!("invalid URL: {e}")))
});

let date_schema = string().try_map(|s| {
    chrono::NaiveDate::parse_from_str(&s, "%Y-%m-%d")
        .map_err(|e| ParseError::custom(format!("invalid date: {e}")))
});
}

.try_map runs after the base schema succeeds. The closure returns Result<NewType, ParseError>.

Brand — Newtypes with Zero Cost

A Brand is a newtype wrapper that exists only at the type level. At runtime it's transparent. At compile time it prevents mixing up bare primitives with domain values:

#![allow(unused)]
fn main() {
use id_effect::schema::Brand;

// Define branded types
type UserId   = Brand<i64,    UserIdMarker>;
type Email    = Brand<String, EmailMarker>;
type PosPrice = Brand<f64,    PosPriceMarker>;

struct UserIdMarker;
struct EmailMarker;
struct PosPriceMarker;
}

Build schemas that produce branded types:

#![allow(unused)]
fn main() {
let user_id_schema: Schema<UserId> = i64()
    .filter(|n| *n > 0, "user id must be positive")
    .map(Brand::new);

let email_schema: Schema<Email> = string()
    .try_map(|s| {
        if s.contains('@') {
            Ok(Brand::new(s))
        } else {
            Err(ParseError::custom("invalid email"))
        }
    });
}

Now functions that need an Email won't compile with a bare String:

#![allow(unused)]
fn main() {
fn send_welcome(to: Email) -> Effect<(), MailError, Mailer> { /* … */ }

// This compiles:
send_welcome(parsed_email);

// This doesn't:
send_welcome("alice@example.com".to_string()); // type error: expected Email, found String
}

HasSchema — Attaching Schemas to Types

When a type always has the same schema, implement HasSchema:

#![allow(unused)]
fn main() {
use id_effect::schema::HasSchema;

impl HasSchema for User {
    fn schema() -> Schema<Self> {
        struct_!(User {
            id:    user_id_schema(),
            email: email_schema(),
            name:  non_empty_string_schema(),
        })
    }
}

// Now parse using the impl
let user: User = User::schema().run(raw)?;
}

HasSchema types work with generic tooling (exporters, documentation generators, UI scaffolding) that needs to know a type's schema without being parameterised over it.

Summary

ParseErrors — Structured Error Accumulation

When a user submits a form with five invalid fields, they deserve to know about all five — not just the first one you found. ParseErrors is id_effect's solution: errors accumulate across an entire parse, and you report them all at once.

ParseError vs ParseErrors

#![allow(unused)]
fn main() {
use id_effect::schema::{ParseError, ParseErrors};

// One error
let e: ParseError = ParseError::custom("age must be positive");

// Many errors
let es: ParseErrors = ParseErrors::single(e);
}

ParseError is a single failure. ParseErrors is a non-empty collection of failures with path information.

What a ParseError Contains

#![allow(unused)]
fn main() {
// A parse error has:
// - a message
// - a path (where in the data structure it occurred)
// - optionally, the value that failed

let err = ParseError::builder()
    .message("expected integer, got string")
    .path(["users", "0", "age"])
    .received(Unknown::string("thirty"))
    .build();

println!("{err}");
// → users[0].age: expected integer, got string (received: "thirty")
}

Path Tracking

Paths are built automatically as schemas descend into nested structures. You don't need to set them manually:

#![allow(unused)]
fn main() {
let raw = Unknown::from_json_str(r#"
  {
    "users": [
      { "name": "Alice", "age": 30 },
      { "name": "Bob",   "age": "thirty" }
    ]
  }
"#)?;

let result = parse(users_schema, raw);
// Err(ParseErrors {
//   errors: [
//     ParseError { path: "users[1].age", message: "expected integer" }
//   ]
// })
}

The struct_! macro and array combinator push path segments automatically. Custom schemas using .try_map or .filter inherit the current path.

Accumulation

The key property of ParseErrors is accumulation. When parsing a struct with multiple fields, failures from different fields are collected, not short-circuited:

#![allow(unused)]
fn main() {
let raw = Unknown::from_json_str(r#"
  { "name": "", "age": -5, "email": "not-an-email" }
"#)?;

let result: Result<User, ParseErrors> = parse(user_schema, raw);
// Err(ParseErrors {
//   errors: [
//     ParseError { path: "name",  message: "must not be empty" },
//     ParseError { path: "age",   message: "age must be between 0 and 150" },
//     ParseError { path: "email", message: "invalid email" },
//   ]
// })
}

All three errors reported in one call. No round-trips.

Using ParseErrors at API Boundaries

Convert ParseErrors to your API's error type:

#![allow(unused)]
fn main() {
#[derive(Debug)]
enum ApiError {
    Validation(Vec<FieldError>),
    Internal(String),
}

#[derive(Debug)]
struct FieldError {
    field:   String,
    message: String,
}

fn to_api_errors(errs: ParseErrors) -> ApiError {
    ApiError::Validation(
        errs.into_iter()
            .map(|e| FieldError {
                field:   e.path().to_string(),
                message: e.message().to_string(),
            })
            .collect()
    )
}
}

ParseErrors in Effects

parse returns a plain Result. To lift into an Effect:

#![allow(unused)]
fn main() {
effect! {
    let raw = Unknown::from_json_bytes(&body)
        .map_err(ApiError::InvalidJson)?;

    let req = parse(create_user_schema(), raw)
        .map_err(to_api_errors)?;

    ~ create_user(req)
}
}

The ? operator on a Result<T, ParseErrors> inside effect! maps the error into E via From. Define impl From<ParseErrors> for YourError to make this ergonomic.

Displaying ParseErrors

ParseErrors implements Display with a human-readable multiline format:

Validation failed (3 errors):
  name: must not be empty
  age: age must be between 0 and 150
  email: invalid email

And Debug for the raw structure when inspecting in tests.

Summary

• ParseError = one failure with a message and a path
• ParseErrors = all failures from a complete parse attempt
• Paths are tracked automatically by schema combinators
• Accumulation means the user sees all problems at once
• Convert to your API error type at the boundary; keep the path information

Testing — Effects Are Easy to Test

Testing async code often means standing up infrastructure, dealing with timing, and wide mocks—then chasing occasional flakes in CI.

Effect programs can be tested differently. Because an Effect is a description of what to do — not the doing itself — you control everything about how it runs. Swap in a test clock. Provide fake services through the Layer system. Detect fiber leaks automatically. Run in microseconds instead of seconds.

This chapter covers the testing tools id_effect provides.

What Makes Effects Testable

Three properties make effect programs easy to test:

1. Services are injected, not ambient. Your code doesn't call DatabaseClient::global(). It declares caps!(Database) and uses ~Database inside effect!. In tests, you provide a different environment — one with a fake database.

2. Time is injectable. Code that uses Clock instead of std::time::SystemTime::now() can be tested with TestClock, which advances only when you tell it to.

3. Effects don't run until you run them. An Effect is inert. You can inspect, compose, and modify it before running. run_test runs it in a harness that adds leak detection and deterministic scheduling.

What This Chapter Covers

• run_test — the test harness that replaces run_blocking in tests (next section)
• TestClock — deterministic time control in tests (ch15-02)
• Mocking services — injecting test doubles via layers (ch15-03)
• Property testing — generating inputs and checking invariants (ch15-04)

run_test — The Test Harness

run_test is the test equivalent of run_blocking. Use it in every #[test] that runs an effect.

Basic Usage

#![allow(unused)]
fn main() {
use id_effect::{run_test, succeed, Exit};

#[test]
fn simple_effect_succeeds() {
    let exit = run_test(succeed(42), ());
    assert_eq!(exit, Exit::Success(42));
}
}

run_test returns an Exit<A, E> rather than a Result<A, E>. This lets you assert on the exact exit reason — success, typed failure, defect, or cancellation.

Why Not run_blocking in Tests?

run_blocking is correct but missing test-specific guarantees:

Fiber leaks — effects that spawn children and don't join them — are silent in production but become test failures under run_test. This catches a class of resource leak bugs at unit-test time.

Asserting on Exit

#![allow(unused)]
fn main() {
#[test]
fn division_by_zero_fails() {
    let eff = divide(10, 0);
    let exit = run_test(eff, ());

    // Assert specific failure
    assert!(matches!(exit, Exit::Failure(Cause::Fail(DivError::DivisionByZero))));
}

#[test]
fn effect_that_panics_is_a_defect() {
    let eff = effect!(|_r: &mut ()| {
        panic!("oops");
    });
    let exit = run_test(eff, ());

    assert!(matches!(exit, Exit::Failure(Cause::Die(_))));
}
}

Exit::Success(a) — the effect succeeded with value a Exit::Failure(Cause::Fail(e)) — the effect failed with typed error e Exit::Failure(Cause::Die(s)) — the effect panicked or encountered a defect Exit::Failure(Cause::Interrupt) — the effect was cancelled

run_test with an Environment

When your effect needs capabilities, build a test environment with build_env or manual Env::insert:

#![allow(unused)]
fn main() {
struct Database;

mock_capability!(MockDb, Database, Arc<dyn Db>, "db/mock", || {
    Arc::new(FakeDatabase::new()) as Arc<dyn Db>
});

#[test]
fn create_user_inserts_into_db() {
    let env = build_env([provide!(MockDb)]).expect("env");
    let fake_db = env.get::<Cap<Database>>().clone();

    let eff = create_user(NewUser { name: "Alice".into(), age: 30 });
    let exit = run_test(eff, env);

    assert!(matches!(exit, Exit::Success(_)));
    assert_eq!(fake_db.users().len(), 1);
}
}

run_test(effect, env) is the full signature. Pass () as the second argument when the effect requires no environment.

Unwrapping success in tests

When you're confident an effect succeeds and just want the value, match on Exit:

#![allow(unused)]
fn main() {
#[test]
fn addition_works() {
    let exit = run_test(succeed(1 + 1), ());
    let Exit::Success(result) = exit else {
        panic!("expected success, got {exit:?}");
    };
    assert_eq!(result, 2);
}
}

Explicit matching keeps failure variants visible in the test — useful when a non-success Exit indicates a bug in the test setup.

Fiber Leak Detection

#![allow(unused)]
fn main() {
#[test]
fn this_test_will_fail_due_to_leak() {
    let eff = effect!(|_r: &mut ()| {
        // Spawns a fiber but never joins it
        run_fork(/* … */);
        ()
    });

    // run_test detects the leaked fiber and fails the test
    let exit = run_test(eff, ());
    // exit: Exit::Failure(Cause::Die("fiber leak detected: 1 fiber(s) not joined"))
}
}

Fix leaks by joining fibers or explicitly cancelling them before the effect completes.

TestClock — Deterministic Time in Tests

TestClock was introduced in Clock Injection from a scheduling perspective. This section focuses on how to use it in tests — specifically with run_test_with_clock and multi-step scenarios.

The Problem with Real Time in Tests

#![allow(unused)]
fn main() {
// This test takes 7 seconds to run
#[test]
fn retry_exhaustion_slow() {
    let eff = failing_call()
        .retry(Schedule::exponential(Duration::from_secs(1)).take(3));
    let exit = run_blocking(eff, ());
    assert!(matches!(exit, Exit::Failure(_)));
}
}

Multiply this by dozens of tests and your suite is unusable. TestClock makes it instant.

run_test_with_clock

#![allow(unused)]
fn main() {
use id_effect::{run_test_with_clock, TestClock, Exit};

#[test]
fn retry_exhaustion_fast() {
    let clock = TestClock::new();
    let eff = failing_call()
        .retry(Schedule::exponential(Duration::from_secs(1)).take(3));

    // run_test_with_clock runs the effect with the supplied clock handle
    let exit = run_test_with_clock(eff, (), clock.clone());

    assert!(matches!(exit, Exit::Failure(_)));
}
}

run_test_with_clock(effect, env, clock) runs an effect in the deterministic test harness with an explicit TestClock. Inject a clock capability via your provider graph when the effect reads time from Env.

TestClock API

#![allow(unused)]
fn main() {
let clock = TestClock::new();

// Read the current (fake) time — starts at Unix epoch
let now: UtcDateTime = clock.now();

// Advance by a duration
clock.advance(Duration::from_millis(500));

// Jump to an absolute time
clock.set_time(UtcDateTime::from_unix_secs(1_700_000_000));

// How many sleeps are currently waiting?
let pending: usize = clock.pending_sleeps();
}

pending_sleeps() is useful in tests to assert that an effect is blocked on a timer rather than having silently completed or failed.

Testing Scheduled Work

#![allow(unused)]
fn main() {
#[test]
fn cron_job_runs_every_minute() {
    let counter = Arc::new(AtomicU32::new(0));
    let c = counter.clone();

    let job = effect!(|_r: &mut ()| {
        c.fetch_add(1, Ordering::Relaxed);
    })
    .repeat(Schedule::fixed(Duration::from_secs(60)));

    let clock = TestClock::new();
    let _handle = job.fork();

    // Advance through 3 minutes
    clock.advance(Duration::from_secs(60));
    clock.advance(Duration::from_secs(60));
    clock.advance(Duration::from_secs(60));

    assert_eq!(counter.load(Ordering::Relaxed), 3);
}
}

Time and Race Conditions

TestClock is deterministic — time moves only when you call advance. This means tests that use TestClock have no time-based race conditions: the scheduler runs wake-up callbacks synchronously when you advance.

If your effect spawns multiple fibers that all sleep, advancing time wakes all fibers whose sleep deadline has passed, in a consistent order.

Combining TestClock with Fake Services

#![allow(unused)]
fn main() {
struct RateLimitStoreCap;

mock_capability!(MockRateLimitStore, RateLimitStoreCap, Arc<dyn RateLimitStore>, "ratelimit/mock", || {
    Arc::new(InMemoryRateLimitStore::new()) as Arc<dyn RateLimitStore>
});

#[test]
fn rate_limiter_enforces_window() {
    let env = build_env([provide!(MockRateLimitStore)]).expect("env");
    let clock = TestClock::new();

    let eff = check_rate_limit_flow("alice");
    let exit = run_test_with_clock(eff, env, clock.clone());

    assert!(matches!(exit, Exit::Success(_)));
}
}

Build the capability env with build_env([provide!(…), …]), then pass it to run_test_with_clock alongside your TestClock.

Mocking Services — Test Doubles via Providers

In id_effect, "mocking" isn't a special testing concept — it's just providing a different ProviderSpec. Production code gets PostgresDbLive. Test code gets InMemoryDbMock. Business logic never knows the difference.

No mock frameworks. No #[automock]. No vi.mock() equivalent. Just providers.

The Pattern

Declare a capability service and a trait object (or concrete type):

#![allow(unused)]
fn main() {
struct Database;

trait Db: Send + Sync {
    fn get_user(&self, id: UserId) -> Effect<User, DbError, ()>;
    fn save_user(&self, user: User) -> Effect<(), DbError, ()>;
}
}

Provide two implementations — one for production, one for tests:

#![allow(unused)]
fn main() {
// Production
#[derive(::id_effect::ProviderSpecDerive)]
#[provides(Database)]
struct PostgresDbLive;

struct PostgresDb { pool: PgPool }
impl Db for PostgresDb { /* real SQL queries */ }

// Test double
struct InMemoryDb { users: Mutex<HashMap<UserId, User>> }
impl Db for InMemoryDb {
    fn get_user(&self, id: UserId) -> Effect<User, DbError, ()> {
        let users = self.users.lock().unwrap();
        match users.get(&id) {
            Some(u) => succeed(u.clone()),
            None    => fail(DbError::NotFound(id)),
        }
    }

    fn save_user(&self, user: User) -> Effect<(), DbError, ()> {
        self.users.lock().unwrap().insert(user.id, user);
        succeed(())
    }
}

mock_capability!(InMemoryDbMock, Database, Arc<dyn Db>, "db/inmemory", || {
    Arc::new(InMemoryDb::new()) as Arc<dyn Db>
});
}

Injecting the Test Double

#![allow(unused)]
fn main() {
#[test]
fn get_user_returns_saved_user() {
    let env = build_env([provide!(InMemoryDbMock)]).expect("env");

    let eff = effect!(|r| {
        let db = ~Database;
        ~ db.save_user(User { id: UserId::new(1), name: "Alice".into() });
        ~ db.get_user(UserId::new(1))
    });

    let exit = run_test(eff, env);
    let Exit::Success(user) = exit else { panic!("expected success") };
    assert_eq!(user.name, "Alice");
}
}

The business logic (save_user then get_user) is identical to production. Only the provider list differs.

Asserting on Calls

When you need to verify that a service was called with specific arguments, add tracking to the test double:

#![allow(unused)]
fn main() {
struct SpyMailer {
    sent: Mutex<Vec<Email>>,
}

impl Mailer for SpyMailer {
    fn send(&self, email: Email) -> Effect<(), MailError, ()> {
        self.sent.lock().unwrap().push(email.clone());
        succeed(())
    }
}
struct MailerCap;

mock_capability!(SpyMailerMock, MailerCap, Arc<dyn Mailer>, "mailer/spy", || {
    Arc::new(SpyMailer::new()) as Arc<dyn Mailer>
});

#[test]
fn registration_sends_welcome_email() {
    let env = build_env([provide!(SpyMailerMock)]).expect("env");
    let spy = env.get::<Cap<MailerCap>>().clone();

    let exit = run_test(register_user("alice@example.com"), env);
    assert!(matches!(exit, Exit::Success(_)));

    let sent = spy.sent.lock().unwrap();
    assert_eq!(sent.len(), 1);
    assert_eq!(sent[0].to, "alice@example.com");
}
}

Failing Services

Test that your code handles service failures correctly by providing a failing test double:

#![allow(unused)]
fn main() {
struct FailingDb;
impl Db for FailingDb {
    fn get_user(&self, _id: UserId) -> Effect<User, DbError, ()> {
        fail(DbError::ConnectionLost)
    }
    fn save_user(&self, _user: User) -> Effect<(), DbError, ()> {
        fail(DbError::ConnectionLost)
    }
}

mock_capability!(FailingDbMock, Database, Arc<dyn Db>, "db/failing", || {
    Arc::new(FailingDb) as Arc<dyn Db>
});

#[test]
fn get_user_propagates_db_errors() {
    let env = build_env([provide!(FailingDbMock)]).expect("env");
    let exit = run_test(get_user(UserId::new(1)), env);
    assert!(matches!(exit, Exit::Failure(Cause::Fail(DbError::ConnectionLost))));
}
}

Provider-Based Test Setup

For more complex scenarios, build a shared test env:

#![allow(unused)]
fn main() {
fn test_env() -> Env {
    build_env([
        provide!(InMemoryDbMock),
        provide!(SpyMailerMock),
        provide!(TestClockLive),
    ])
    .expect("test env")
}

#[test]
fn full_registration_flow() {
    let env = test_env();
    let exit = run_test(full_registration_flow(), env);
    assert!(matches!(exit, Exit::Success(_)));
}
}

The test provider list mirrors your production wiring in structure but with test implementations. Add new providers in one place and all tests pick them up.

What You Don't Need

• No mockall, no mock! macros
• No #[cfg(test)] on business logic
• No Box<dyn Fn(…)> callback injection patterns
• No global state reset between tests

The capability provider graph is the mock framework.

Property Testing — Invariants over Inputs

Unit tests check specific cases. Property tests check invariants: statements that must be true for any valid input. Effect programs are excellent targets for property testing because their inputs and outputs are well-typed, their schemas define exactly what's valid, and the provider system makes it easy to run thousands of executions cheaply.

Setup

id_effect works with both proptest and quickcheck. The examples below use proptest.

[dev-dependencies]
proptest = "1"

Testing Pure Effects

#![allow(unused)]
fn main() {
use proptest::prelude::*;
use id_effect::{run_test, Exit};

proptest! {
    #[test]
    fn addition_is_commutative(a: i64, b: i64) {
        let eff_ab = add(a, b);
        let eff_ba = add(b, a);

        let Exit::Success(r_ab) = run_test(eff_ab, ()) else { return Ok(()); };
        let Exit::Success(r_ba) = run_test(eff_ba, ()) else { return Ok(()); };

        prop_assert_eq!(r_ab, r_ba);
    }
}
}

proptest! generates hundreds of (a, b) pairs. Each iteration calls run_test, which is cheap for pure effects.

Testing Schema Round-Trips

Schemas have a round-trip property: if you serialise a valid value and re-parse it, you get the same value back.

#![allow(unused)]
fn main() {
proptest! {
    #[test]
    fn user_schema_round_trips(
        name in "[a-zA-Z]{1,50}",
        age in 0i64..=120,
    ) {
        let original = User {
            name: name.clone(),
            age,
        };

        // Serialise to Unknown
        let raw = User::schema().encode(&original);

        // Re-parse
        let parsed = User::schema().run(raw);

        prop_assert!(parsed.is_ok());
        prop_assert_eq!(parsed.unwrap(), original);
    }
}
}

Round-trip tests catch asymmetries between your serialiser and parser that unit tests often miss.

Testing Error Invariants

Property tests are excellent for verifying that your error handling is consistent:

#![allow(unused)]
fn main() {
proptest! {
    #[test]
    fn withdraw_never_goes_negative(
        balance in 0u64..=1_000_000,
        amount  in 0u64..=1_000_000,
    ) {
        let account = TRef::new(balance);
        let exit = run_test(commit(withdraw(&account, amount)), ());

        if amount <= balance {
            // Should succeed and balance should be reduced
            assert!(matches!(exit, Exit::Success(_)));
            let new_balance = atomically(account.read_stm());
            assert_eq!(new_balance, balance - amount);
        } else {
            // Should fail — balance must not go negative
            assert!(matches!(exit, Exit::Failure(Cause::Fail(InsufficientFunds))));
            let new_balance = atomically(account.read_stm());
            assert_eq!(new_balance, balance);  // unchanged
        }
    }
}
}

Generating Arbitrary Service Environments

For integration-style property tests, generate random state in the fake service:

#![allow(unused)]
fn main() {
struct Database;

mock_capability!(InMemoryDbMock, Database, Arc<dyn Db>, "db/inmemory", || {
    Arc::new(InMemoryDb::new()) as Arc<dyn Db>
});

proptest! {
    #[test]
    fn get_user_returns_what_was_saved(user in arbitrary_user()) {
        let env = build_env([provide!(InMemoryDbMock)]).expect("env");

        // Save
        run_test(save_user(user.clone()), env.clone());

        // Retrieve
        let exit = run_test(get_user(user.id), env);
        let Exit::Success(retrieved) = exit else { return Ok(()); };

        prop_assert_eq!(retrieved, user);
    }
}
}

Define arbitrary_user() as a proptest Strategy:

#![allow(unused)]
fn main() {
fn arbitrary_user() -> impl Strategy<Value = User> {
    (
        "[a-zA-Z ]{1,50}",
        0i64..=120,
        any::<u64>().prop_map(UserId::new),
    ).prop_map(|(name, age, id)| User { id, name, age })
}
}

Schema-Driven Generation

When a type has HasSchema, you can derive a generator that always produces valid inputs:

#![allow(unused)]
fn main() {
// generate_valid::<User>() produces Users that would pass User::schema()
let strategy = generate_valid::<User>();

proptest! {
    #[test]
    fn valid_users_are_always_accepted(user in generate_valid::<User>()) {
        let raw = User::schema().encode(&user);
        prop_assert!(User::schema().run(raw).is_ok());
    }
}
}

This ensures the generator and schema stay in sync: if you tighten a refine constraint, generate_valid starts producing inputs that satisfy the new constraint.

Shrinking

proptest automatically shrinks failing inputs to the smallest example that still fails. Since run_test is fast (no I/O, no real timers), shrinking runs quickly even with hundreds of iterations.

When a property fails, you'll see the minimal failing case:

Test failed. Minimal failing input:
  name = ""
  age = -1
Reason: must not be empty (path: name)

This is far more actionable than a raw failure trace from a specific hand-chosen test case.

Optics with id_effect

Part V · Chapter 18 — functional optics for immutable focus and update.

The id_effect_optics crate provides:

• Lens — total field access (get, set, modify, compose, as_traversal)
• Prism — partial variant access (preview, review)
• Optional — helpers for Option<T> fields
• Traversal — map over vectors, optional fields, and composed optics
• Iso — bidirectional isomorphisms
• Transducer — composable reducer transforms
• Schema bridge — dot-path and JSON Pointer access on Unknown
• JSON Patch — RFC 6902 operations on Unknown
• TrieZipper — navigable persistent trie with rebuild
• #[derive(Optics)] — codegen via id_effect_proc_macro

When to reach for optics

Use optics when you need composable, reusable focus into nested data — especially persistent im structures or Unknown documents at boundaries.

Example

cargo run -p id_effect_optics --example 010_lens
cargo run -p id_effect_optics --example 030_schema_patch

Sections

• Lenses
• Prisms and Optionals
• Traversals and schema bridge

Lenses

A Lens<S, A> is a total optic: every S has exactly one focused A.

#![allow(unused)]
fn main() {
use id_effect_optics::{Lens, field};

#[derive(Clone)]
struct Person { name: String }

let name = field(
    |p: &Person| &p.name,
    |mut p, name| { p.name = name; p },
);

let updated = name.modify(Person { name: "ada".into() }, |n| n.to_uppercase());
}

Compose nested lenses with Lens::compose.

Stub: derive-generated field lenses land in Plan 04 (FP DX).

Prisms and Optionals

A Prism<S, A> focuses a sum-type variant:

#![allow(unused)]
fn main() {
use id_effect_optics::Prism;

enum Shape { Circle(f64), Rect { w: f64, h: f64 } }

let circle = Prism::new(
    |s: &Shape| match s { Shape::Circle(r) => Some(*r), _ => None },
    Shape::Circle,
);
}

Optional wraps a Lens<S, Option<T>> for fallible inner access (set_some, set_none, modify).

Stub: prism composition with schema tagged unions — ch20 parse codecs.

Traversals and schema bridge

Traversals

Traversal maps over zero or many inner values:

#![allow(unused)]
fn main() {
use id_effect_optics::{vector_each, at_vec, at_option, field};

let doubled = vector_each::<i32>().over(vec![1, 2, 3], |n| n * 2);
}

Compose a field lens with collection traversals via at_vec and at_option.

Schema field paths

get_at_path, set_at_path, and create_at_path navigate Unknown with dot-separated segments (user.name, tags.0) or JSON Pointer paths (/user/name/0).

JSON Patch

apply_patch supports add, replace, remove, move, copy, and test.

Trie zipper

TrieZipper navigates persistent tries, supports rebuild, insert_child, and remove_child.

State Machines (id_effect_fsm)

The id_effect_fsm crate adds typed finite-state machines, sagas, and linear session types on top of id_effect. It is the Phase FP FSM slice from the FP Kitchen Sink roadmap.

Why FSMs in an effect library?

Business logic often has explicit states (order lifecycle, connection handshake, approval gates). A pure transition table keeps routing declarative; Effect programs attach side effects at the edges via run_blocking.

Crate layout

• TransitionTable — immutable edge map
• StateMachine — mutable current state + step
• Interpreter — effectful stepping
• to_mermaid — documentation diagrams
• Saga — compensate on failure
• SessionSend / SessionRecv — linear protocols
• step_durable — SQLite snapshots via id_effect_workflow

When to use workflow bridge

For single-process restart resume, pair FSM stepping with id_effect_workflow. For multi-service orchestration, prefer an external engine (Temporal, Step Functions).

Transition tables

Build tables fluently, then freeze them into a StateMachine:

#![allow(unused)]
fn main() {
use id_effect_fsm::{StateMachine, TransitionTable};

let table = TransitionTable::new()
    .on("idle", "start", "running")
    .on("running", "stop", "idle");

let mut m = StateMachine::new("idle", table);
m.step("start")?;
}

Missing edges return FsmError::NoTransition. Reset with StateMachine::reset or set state directly after loading a durable snapshot.

Effect interpreter

Interpreter registers effect factories keyed by (state, event). Each factory returns a fresh Effect (effects are not Clone).

#![allow(unused)]
fn main() {
use id_effect::{Effect, run_blocking};
use id_effect_fsm::{Interpreter, StateMachine, TransitionTable};

let interp = Interpreter::new().on_transition("idle", "tick", || {
    Effect::new(|_| Ok(()))
});
let mut m = StateMachine::new("idle", table);
interp.run(&mut m, ["tick"], ())?;
}

Use run_blocking at the application boundary — the same rule as other synchronous IO in effect! blocks.

Sagas and session types

Saga compensation

Saga runs forward Effect steps via run_blocking. On failure it invokes compensation factories in reverse order.

Linear session types

[SessionSend

](../../id_effect_fsm/src/session.rs) and ``SessionRecv<P> are phantom markers encoding alternating send/receive phases. The included ping/pong protocol demonstrates type-directed hand-off without runtime protocol state machines.

Workflow bridge

register_fsm and step_durable persist FsmSnapshot JSON rows through DurableWorkflowLog.

After a restart, restore_state reloads the latest snapshot into a StateMachine. Cached steps skip re-execution — the same resume semantics as workflow run_step_typed.

Parser Combinators

id_effect_parse brings small, composable parsers to the workspace — the same functional pattern as schema combinators in Part IV, but oriented toward text and byte streams rather than JSON Unknown values.

The Parser type

A Parser<I, O, E> wraps a function I -> Result<(O, I), E>: parsed output plus remaining input.

#![allow(unused)]
fn main() {
use id_effect_parse::{char, int, parse_str, Parser};

let number = char('(')
    .and_then(|_| int())
    .and_then(|n| char(')').map(move |_| n));

let (value, rest) = parse_str(&number, "(42) rest").unwrap();
assert_eq!(value, 42);
assert_eq!(rest, " rest");
}

Core combinators:

Built-ins such as char, tag, int, and ws cover common text needs.

Pretty printing

The Pretty trait builds Doc values — a Wadler-style document tree rendered with a line width budget:

#![allow(unused)]
fn main() {
use id_effect_parse::{Doc, Pretty};

let doc = Doc::text("users")
    .cat(Doc::line())
    .cat(["alice", "bob"].pretty())
    .group();

println!("{}", doc.render(40));
}

Use pretty printers for debug output, REPLs, and human-readable config — not for wire formats (use Codec instead).

Invertible codecs

Codec pairs parse and print so formats round-trip:

#![allow(unused)]
fn main() {
use id_effect_parse::codec::quoted_string;

let codec = quoted_string();
let wire = codec.print(&"hello".to_string());
let (parsed, _) = codec.parse(wire).unwrap();
assert_eq!(parsed, "hello");
}

When you need lossless serialization with a parser-shaped API, start with Codec::new.

Diffs

Diff<T> describes value-level changes (Unchanged, Added, Removed, Changed). Helpers like diff_values and diff_option support config drift and snapshot tests.

Parsing Stream chunks

Collect stream chunks, flatten, then parse — parse_stream for typed buffers, parse_text_stream for UTF-8 text:

#![allow(unused)]
fn main() {
use id_effect::{Chunk, Stream, run_blocking};
use id_effect_parse::{parse_text_stream, tag};

let parser = tag("ping");
let stream = Stream::from_iterable(vec![Chunk::from_vec(b"ping".to_vec())]);
let value = run_blocking(parse_text_stream(&parser, stream), ()).unwrap();
assert_eq!(value, "ping");
}

Schema bridge

SchemaBridge connects Schema values to text parsers:

• parser_for_json — parse JSON text, then decode_unknown
• parser_for_string_wire — when the wire type is String
• parser_for::<T>() — for types implementing HasSchema

Boundary validation for external data still belongs to id_effect::schema (Part IV). Use schema at the edge; use SchemaBridge when a text protocol should reuse the same schema.

#![allow(unused)]
fn main() {
use id_effect::schema::{HasSchema, i64, string, struct_};
use id_effect_parse::SchemaBridge;

let schema = struct_("name", string(), "age", i64());
let parser = SchemaBridge::parser_for_json(schema);
let (person, _) = parser.parse(r#"{"name":"Ada","age":36}"#.to_string()).unwrap();
assert_eq!(person, ("Ada".to_string(), 36));
}

#[derive(SchemaParser)]

The proc macro generates schema(), parser(), and HasSchema for structs with named fields (see Part V chapter 24).

Next steps

• Part IV Schema for JSON and API boundaries
• Part IV Streams for chunk backpressure and collection
• Workspace crate: crates/id_effect_parse/

Advanced Streaming

Part V · Chapter 22 — windowing, joins, replay fanout, FSM scans, and transducers.

Part IV introduced pull-based Stream processing with chunks, sinks, and backpressure. This chapter covers multi-stream patterns: grouping events into windows, joining live sources, replaying history to fanout branches, stepping simple state machines over elements, and applying composable transducers.

Module map

When to use what

• Windows — aggregate or batch events by count, time, or session gaps.
• Joins — correlate two live sources (merge for fair interleave, combine_latest for dashboards).
• Replay fanout — same as broadcast, plus a retained tail buffer per branch.
• state_scan — emit only on FSM transitions (contrast with scan which emits every step).
• Transducers — reusable map/filter pipelines (API-compatible with id_effect_optics::Transducer).

See the section pages for examples and tests in each module.

Windowing

window adds count- and time-based grouping on Stream.

Count windows

#![allow(unused)]
fn main() {
use id_effect::Stream;

let sums = Stream::from_iterable(1..=10)
    .tumbling(3)   // non-overlapping chunks → Vec<_>
    .map(|chunk| chunk.iter().sum::<i32>());
}

• tumbling(n) — alias of grouped; last chunk may be short.
• sliding(size, step) — overlapping windows; size == 0 or step == 0 yields an empty stream.

Time and session windows

Provide a timestamp extractor Fn(&A) -> Instant:

#![allow(unused)]
fn main() {
use id_effect::Stream;
use std::time::{Duration, Instant};

let events: Vec<(Instant, char)> = /* ... */;
let sessions = Stream::from_iterable(events)
    .session_by_gap(Duration::from_secs(30), |(ts, _)| *ts);
}

• tumbling_by_time(duration, ts) — fixed-width buckets aligned to Instant::UNIX_EPOCH.
• sliding_by_time(duration, step, ts) — overlapping time ranges stepped by step.

Time buckets use merge_time_bucket helpers internally for ordered aggregation maps.

Stream joins

join combines multiple streams.

Fair merge

Stream::merge alternates elements from two sources:

#![allow(unused)]
fn main() {
use id_effect::Stream;

let merged = Stream::from_iterable([1, 2]).merge(Stream::from_iterable([10, 20]));
// [1, 10, 2, 20]
}

Combine latest

combine_latest keeps the latest value from each side and emits whenever either updates (after both have emitted at least once):

#![allow(unused)]
fn main() {
use id_effect::{Stream, combine_latest};

let pairs = combine_latest(
    Stream::from_iterable([1, 2]),
    Stream::from_iterable(['a', 'b']),
);
// [(2, 'a'), (2, 'b')]
}

Keyed join

keyed_join performs an inner join on the latest value per key:

#![allow(unused)]
fn main() {
use id_effect::{Stream, keyed_join};

let joined = keyed_join(
    Stream::from_iterable([("user", 1), ("other", 2)]),
    Stream::from_iterable([("user", 'x')]),
);
// [("user", 1, 'x')]
}

Replay fanout

broadcast_with_replay extends Stream::broadcast with a sliding replay tail.

#![allow(unused)]
fn main() {
use id_effect::{Stream, broadcast_with_replay, run_async};

let src = Stream::from_iterable(1..=100);
let (mut branches, pump) = run_async(
    broadcast_with_replay(src, /* hub */ 64, /* replay tail */ 16, /* branches */ 2),
    (),
)
.await?;

// Run `pump` concurrently with pulls on each branch (same pattern as `broadcast`).
}

• hub_capacity — sliding PubSub ring size.
• replay_len — number of recent items retained in the shared replay buffer (also seeded into each branch buffer as the pump runs).
• Stream::broadcast_replay(cap, branches) — convenience wrapper using replay_len == cap.

Return shape matches broadcast: (Vec<Stream<…>>, pump_effect).

state_scan — FSM stepping

state_scan folds state while emitting only when the step returns Some.

Contrast with Stream::scan, which emits on every element:

#![allow(unused)]
fn main() {
use id_effect::Stream;

#[derive(Clone, Copy, PartialEq)]
enum Mode { Idle, Active }

let transitions = Stream::from_iterable([0u8, 1, 1, 0, 1]).state_scan(Mode::Idle, |mode, x| {
    let next = if x > 0 { Mode::Active } else { Mode::Idle };
    let emit = matches!((mode, next), (Mode::Idle, Mode::Active));
    (next, emit.then_some(x))
});
// emits on Idle → Active edges only
}

Use this for simple finite-state interpretations without pulling in the full id_effect_fsm interpreter.

Transducers on streams

transducer provides Transducer with the same shape as id_effect_optics::Transducer (kept local to avoid a crate dependency cycle).

#![allow(unused)]
fn main() {
use id_effect::{Stream, Transducer, transducer_filter, transducer_map};

let xf = transducer_map(|n: i32| n + 1).compose(transducer_filter(|n: &i32| n % 2 == 0));
let out = Stream::from_iterable(1..=5).via_transducer(xf);
// [2, 4, 6]
}

• via_transducer — alias of transduce_items: map/filter pipeline per element, preserving order.
• Transducer::compose — inner transducer runs first, then outer (same as optics).
• Built-ins: transducer_map, transducer_filter.

For list-only transduction without streams, use id_effect_optics directly or Transducer::transduce on iterators.

Runtime Resilience

Production services need more than happy-path effects. Timeouts, overload, and flaky dependencies are normal. This chapter covers coordination primitives and the id_effect_resilience crate for keeping programs responsive under stress.

What This Chapter Covers

• RequestResolver — batch parallel lookups through a single fetch
• SubscriptionRef — shared state plus change notifications
• Redacted — schema-layer secrets with masked Debug
• match_effect! — enum match helper with path-prefixed arms
• id_effect_resilience — circuit breaker, rate limiter, bulkhead, hedged requests

RequestResolver

Effect.ts batches data-source lookups so N parallel getUser(id) calls become one SQL WHERE id IN (…). In id_effect, each pending lookup is a [RequestEntry] with a [Deferred] result slot. A [RequestResolver::run_all] receives sequential batches; entries inside a batch may run in parallel.

batching deduplicates keys per batch and calls your fetch function once:

#![allow(unused)]
fn main() {
use id_effect::{Deferred, RequestEntry, batching, run_async};

let resolver = batching(|keys| Effect::new(move |_r| {
    // build HashMap from keys …
    Ok(map)
}));
}

SubscriptionRef

[SubscriptionRef] combines [Ref] with [PubSub]. Every set / update publishes the new value; subscribe returns a [Queue] that receives the current value first, then every subsequent change.

#![allow(unused)]
fn main() {
use id_effect::{Scope, SubscriptionRef, run_async};

let cell = run_async(SubscriptionRef::make(0u32), ()).await?;
let scope = Scope::make();
let changes = run_async(cell.subscribe(), scope.clone()).await?;
}

Redacted values

Use [Redacted<T>] anywhere a schema or domain type might reach logs. Debug and Display print <redacted>; call expose only at trust boundaries.

match_effect!

The match_effect! proc macro prefixes variant names so the compiler checks exhaustiveness without repeating the enum path:

#![allow(unused)]
fn main() {
use id_effect::match_effect;

match_effect!(Color, paint, {
    Red(n) => n,
    Green => 0,
    Blue => 1,
})
}

id_effect_resilience

Add id_effect_resilience to your workspace dependency when you need operational guardrails:

#![allow(unused)]
fn main() {
use id_effect_resilience::{CircuitBreaker, hedged};
use id_effect::run_async;

let breaker = run_async(CircuitBreaker::make(5, Duration::from_secs(30)), ()).await?;
let out = run_async(breaker.call(fetch_user(id)), ()).await?;
}

Pair resilience primitives with Schedule retry policies from Part III — breakers shed load, schedules handle transient faults.

Verification and Metaprogramming

Part V closes with tools that keep functional patterns honest: law checks, property tests, golden snapshots, and derive stubs that will grow into full codegen.

What This Chapter Covers