# The processing component

> **CocoIndex v1, Rust SDK (experimental).** This page documents the CocoIndex **v1** Rust SDK (the `cocoindex` crate). The Rust SDK is experimental: any release may change or remove its API, and it is not yet published on crates.io. The Python SDK pages are at https://cocoindex.io/docs/.
>
> Source: https://cocoindex.io/docs/rust/programming_guide/processing_component/ · Rust docs index: https://cocoindex.io/docs/rust/llms.txt · Rust SDK tour: https://github.com/cocoindex-io/cocoindex/blob/main/rust/sdk/SHOWCASE.md · Rust examples: https://github.com/cocoindex-io/cocoindex/tree/main/examples/rust

**Note — Prerequisite**
Read [Core Concepts](/docs/rust/programming_guide/core_concepts/) for processing components, target states, and incremental sync. Its per-file fan-out and chunk-embedding diagrams illustrate the mounting APIs below.

Most apps process many independent source items — files, rows, or entities. A **Processing Component** is the unit of execution for one: it runs that item's transformation logic and declares the set of **target states** produced for it.

## Component path

A **component path** is the stable identifier for a processing component across runs (think of it like a path in a tree). CocoIndex uses it to match a component to its previous run, detect what changed for that item, and sync that component's target states as a unit when it finishes. This sync happens per component; CocoIndex does not wait for other components in the same app to complete.

Component paths are hierarchical and form a tree structure. You specify child paths as the key of each mount with stable identifiers like string literals, file names, row keys, or entity IDs:

```rust
// The key is the first argument of the mount macros and methods.
mount!("hello.pdf", process_file(ctx, file, target)).await?;
// A multi-part path: a subpath context, then the key.
ctx.with_subpath(&"user").mount(&user_id, |ctx| async move { /* ... */ Ok(()) }).await?;
```

A key is any `IntoStableKey`: a `&str`, a `String`, an integer, a UUID, bytes or a `StableKey`. `ctx.with_subpath(&"user")` returns a `Ctx` whose mounts go one level down.

Choose paths that are stable for the "same" item (e.g., file path, primary key). If an item disappears and its path is no longer present, CocoIndex cleans up the target states owned by that path (and its sub-paths).

Here's an example component path tree for a pipeline that processes files:

```text
(root)                         ← app_main component
└── process_file
    ├── "hello.pdf"            ← process_file component
    └── "world.pdf"            ← process_file component
```

The tree is populated dynamically as the app runs — each `mount!` / `mount_each!` call adds a subpath.

See [StableKey](/docs/rust/programming_guide/sdk_overview/#stablekey) in the SDK Overview for details on what values can be used in component paths.

## Mount

Mounting is how you declare (instantiate) a processing component within an app at a specific path, so CocoIndex knows that component exists, should run, and owns a set of target states — and can match it against its previous run to sync only what changed.

CocoIndex provides two core mounting APIs:

- **`mount!`** — sets up a processing component in a child path without depending on data from it. This allows the component to refresh independently in live mode.
- **`use_mount!`** — returns a value from the component's execution to the caller. The component at that path cannot refresh independently without re-executing the caller.

**Which one you reach for comes down to a single question: does the caller need a value back?** `use_mount!` consumes the child's return value, which *couples* the two — the call waits until the child finishes and commits its target states, and the child can't refresh on its own without re-running the caller. `mount!` takes nothing back and returns as soon as the child is scheduled, so the child runs on its own and can [refresh independently in live mode](/docs/rust/programming_guide/live_mode/) when its inputs change.

And APIs that simplify common patterns:

- **`mount_each!`** — mounts one component per item in a keyed collection; **`use_mount_each!`** also returns each child's value
- **`mount_target()`** — mounts a target without an explicit subpath; connectors wrap it in their `mount_*_target` functions

See also [`ctx.map`](#using-ctxmap-to-apply-transformations) for a utility API that operates within a component without creating new ones.

Each API has two forms. The macros `mount!`, `use_mount!`, `mount_each!` and `use_mount_each!` take a `#[cocoindex::function]` call, substitute the child's `&Ctx` for its first argument, and fingerprint the function's logic and arguments so an unchanged child is skipped. The methods `ctx.mount`, `ctx.scope`, `ctx.mount_each` and `ctx.use_mount_each` take a key and a closure, and always run it. `ctx.mount_each_live` is `ctx.mount_each` for a live source.

The Python API names map to the Rust forms as follows:

| Python | Rust macro | Rust method |
|--------|------------|-------------|
| `coco.mount(fn, ...)` | `mount!(fn(ctx, ...))` | `ctx.mount(&key, \|ctx\| async move { ... })` |
| `coco.use_mount(fn, ...)` | `use_mount!(fn(ctx, ...))` | `ctx.scope(&key, \|ctx\| async move { ... })` |
| `coco.mount_each(fn, items, ...)` | `mount_each!(items, \|item\| fn(ctx, item, ...))` | `ctx.mount_each(items, key_fn, \|ctx, item\| async move { ... })`, `ctx.mount_each_stream(stream, ...)`, `ctx.mount_each_live(&key, feed, ...)` |
| — | `use_mount_each!(items, \|item\| fn(ctx, item, ...))` | `ctx.use_mount_each(items, key_fn, \|ctx, item\| async move { ... })` |
| `coco.mount_target(target)` | — | `mount_target(&ctx, target_state)`, or a connector's `mount_*_target` |
| `coco.map(fn, items, ...)` | — | `ctx.map(items, async \|item\| { ... })` |

### Automatic subpath derivation
`mount!`, `use_mount!`, `mount_each!` and `use_mount_each!` all accept an optional key as their first argument. When omitted, the subpath is **auto-derived** from the function name, as a plain string key.

```rust
// These are equivalent:
mount!(process_file(ctx, file, target)).await?;
mount!("process_file", process_file(ctx, file, target)).await?;
```

The methods `ctx.mount`, `ctx.scope` and `ctx.mount_each` always take an explicit key.

This means the component path for a `process_file` function is `parent / "process_file"`.

Since sibling component paths must not collide, you need an explicit key when:
- **The same function is mounted more than once** — auto-derived paths would be identical, so each call needs a distinct key
- **Functions in different modules share a name** — the derived key is the bare function name
- **You want a specific path name** — different from the function name

### `mount`

Use `mount!` when you don't need a return value from the processing component. It schedules the processing component to run and returns a handle:

```rust
let handle = mount!(process_file(ctx, file, target)).await?;
```

With an explicit subpath, for example when mounting multiple components of the same function:

```rust
let process = ctx.with_subpath(&"process");
let ctx = &process;
let handle = mount!(file.key(), process_file(ctx, file, target.clone())).await?;
```

The macro reads the parent context from the `ctx` argument of the call, so bind the subpath context to `ctx` first. `ctx.mount(&key, |ctx| async move { .. })` is the closure form, for a key and body decided at run time; it always runs the body.

The handle provides a method you can call if you need to wait until the processing component is fully ***ready*** — meaning all its target states have been applied to external systems and all components in its sub-paths are ready:

```rust
handle.ready().await?;
```

The handle is a `ComponentMountHandle`. Without `ready()`, the parent still waits for its children before the update completes.

You usually only need to call `ready()` when you have logic that depends on the processing component's target states being applied — for example, querying the latest data from a target table after syncing it.

`ctx.mount_live(&key, component)` mounts a **`LiveComponent`** — a component that processes continuously and reacts to changes incrementally instead of rescanning everything. See [Live Components](/docs/rust/advanced_topics/live_component/) for details.

### `use_mount`

Use `use_mount!` when you need the processing component's return value. It mounts the component, waits until it's ready, and returns the value directly:

```rust
let summary = use_mount!(summarize(ctx, document)).await?;
```

With an explicit subpath:

```rust
let summary = use_mount!("setup", summarize(ctx, document)).await?;

// The closure form, always run:
let val = ctx.scope(&"child", |child_ctx| async move {
    Ok(42)
}).await?;
```

The return value of a `use_mount!` child must implement `Serialize` and `Deserialize`: the engine stores it with the component's memo and returns it on a hit.

A common use of a foreground mount is to obtain a [target](/docs/rust/programming_guide/target_state/#where-do-targets-come-from) after its container target state is applied.

### `mount_each`

`mount_each!` mounts one processing component per item in a keyed collection.

```rust
let files = cocoindex::resources::fs::walk_items(&source_dir, &["**/*.md"])?;
mount_each!(files, |file| process_file(ctx, file, target)).await?;
```

`walk_items` returns `(relative_path, FileEntry)` pairs. The closure binds each item's value; `ctx` in the call is each child's context, and `target` is a captured extra that is fingerprinted with the value.

Each item is a `(key, value)` pair. The closure's call receives the value, and any other arguments are passed through. Items are mounted under an [auto-derived subpath](#automatic-subpath-derivation) (the function name), so the component path for each item is `parent / "process_file" / key`.

In the static case this is just the keyed-loop form of [`mount!`](#mount) — the same per-file fan-out shown in [Core Concepts](/docs/rust/programming_guide/core_concepts/#processing-component). The per-item `key` is what keeps each component's path stable across runs, so the engine matches, updates, and cleans up each item individually.

Like `mount!`, `mount_each!` and `ctx.mount_each` do not carry return values from the mounted functions. They mount the children in the background as the items are read, and return one `ComponentMountHandle` whose `ready()` waits for all of them. For per-item values, the foreground forms `use_mount_each!` and `ctx.use_mount_each` wait for every child and return their values as a `Vec` in input order. `ctx.mount_each_live` returns once the feed's first scan is mounted and ready; in live mode its children keep reacting in the background.

If you want per-item results without creating components, use [`ctx.map`](#using-ctxmap-to-apply-transformations).

You can provide an explicit subpath as the first argument:

```rust
mount_each!("files", files, |file| process_file(ctx, file, target)).await?;

// A live source takes the closure form, with the prefix first:
ctx.mount_each_live(&"process_file", files, move |ctx, file: FileEntry| {
    let target_table = target_table.clone();
    async move { process_file(&ctx, &file, &target_table).await }
})
.await?;
```

Source connectors return `(key, value)` pairs. For example, `walk_items(dir, patterns)` returns `(relative_path, FileEntry)` pairs as a `Vec`.

Keys must be unique within one `mount_each!` call. If the items yield the same key twice, the call returns an error that names the duplicate key.

An item's key is relative to its source, so two sources can yield the same key. When you mount several sources under one parent component, give each `mount_each!` its own prefix, as in `mount_each!("files", files, ..)` above. Without the prefixes, both calls mount under `"process_file"`, so a `README.md` in each source gets the same component path, and the second call returns the same error. The target states that the items declare need distinct keys too, so give each source its own target.

A live source takes the closure form: `walk_dir(dir).live()` returns a live feed, and `ctx.mount_each_live` adds, updates, and removes per-item components as the source changes. See [Live Mode](/docs/rust/programming_guide/live_mode/).

### `mount_target()`
`mount_target()` mounts a target without requiring an explicit subpath.

`mount_target(&ctx, target_state)` derives the component path from the target's globally unique key, so you don't choose a key for it. It is the generic form, for connector authors; pipelines use the connectors' `mount_*` functions below.

Connectors also provide convenience methods that wrap `mount_target()`:

```rust
// A directory target; `DirTarget::mount(&ctx, &output_dir)?` is the same call.
let target = mount_dir_target(&ctx, out)?;

// PostgreSQL example
let target_table = postgres::mount_table_target(
    &ctx,
    &DB,
    TABLE,
    postgres::TableSchema::from_row::<DocEmbedding>(["id"])?
        .with_vector_schema("embedding", ctx.get_key(&EMBEDDER)?)
        .await?,
    Some(PG_SCHEMA),
)
.await?;
```

<span id="using-component_subpath-as-a-context-manager"></span>

### Nesting paths with `with_subpath`

You can use `ctx.with_subpath()` to create nested paths without repeating common prefixes:

```rust
let process = ctx.with_subpath(&"process");
for file in files {
    let ctx = &process;
    mount!(file.key(), process_file(ctx, file, target.clone())).await?;
}
```

`ctx.with_subpath(&"process")` returns a scoped `Ctx`; the original is unchanged. Chain calls for a multi-part prefix. The scoped context still belongs to the same component, so target states declared through it are the component's own. `use_state` is an error on a subpath context: the state belongs to the component, not to the subpath.

**Tip**
When iterating over keyed items, prefer [`mount_each!`](#mount_each) — it handles the loop and subpath creation for you.

## Memoized components

`memo` on a function you **mount** memoizes the whole component — the *processing-component-level* memoization introduced in [Core Concepts](/docs/rust/programming_guide/core_concepts/#function-memoization-skip-unchanged-computations). On a cache hit the body doesn't run, and everything the previous run produced is carried over as a unit: the component's declared [target states](/docs/rust/programming_guide/target_state/) **and** its child mounts. Cleanup of no-longer-mounted children only applies when the body actually *runs* — a run that no longer mounts a child removes that child's subtree, while a cache hit retains it.

```rust
#[cocoindex::function(memo, memo_key(table = skip))]
async fn process_file(ctx: &Ctx, file: &FileEntry, table: &postgres::TableTarget) -> Result<()> {
    // ...
}

ctx.mount_each_live(&"process_file", files, move |ctx, file: FileEntry| {
    let target_table = target_table.clone();
    async move { process_file(&ctx, &file, &target_table).await }
})
.await?;
```

The closure forms (`ctx.mount`, `ctx.scope`, `ctx.mount_each`, `ctx.mount_each_live`) always run the body, so put `memo` on the function they call, as above. The table target is not `MemoInput`; `memo_key(table = skip)` leaves it out of the key.

The macros `mount!`, `use_mount!` and `mount_each!` memoize the component on the function's logic and arguments even without `memo`: an unchanged child is skipped on the next run.

Skipping a stable item this way costs almost nothing in storage, provided you follow one principle:

- **Declare large values as target states; don't return them.** Every memo entry — whether for a memoized component or for a memoized function called directly inside one (the *transform-level* memoization from Core Concepts) — must store its return value in order to serve a cache hit. A component that outputs via [`declare_row(...)`](/docs/rust/programming_guide/target_state/) and returns `()` keeps its entry tiny: the declared values live in the component's own tracking, which [sync](#how-target-states-sync) needs regardless. A memoized function that *returns* a large value (an embedding vector, a parsed document) stores that value in every entry. A secondary difference: a directly-called memoized function's entry also records a reference to each target state it declared, so a hit can keep those states alive — a per-state cost that mounting avoids. See the cost note in [When to memoize](/docs/rust/programming_guide/function/#when-to-memoize).
- **A memoized component may mount children.** A memoized function *called directly* cannot mount — mounting is a side effect its cache can't replay ([memoization constraints](/docs/rust/programming_guide/function/#memoization)) — but the restriction doesn't apply once the function is mounted as a component. A memoized parent that mounts children still invalidates when a child's logic changes — the child's fingerprint is part of the parent's recorded logic deps.
- **A component that reads beyond its arguments records what it read.** With [`ctx.use_data_dep`](/docs/rust/programming_guide/function/#data-dependencies-datadep-and-use_data_dep) (experimental) a body records the files (or anything else) it chose to read while running, and its entry is reused only while those reads validate. The records stay in that component's own entry; a memoized ancestor of such a component is reused only after validating those records from that entry too, through the path it stored, so one invalid read anywhere beneath it re-runs the ancestor's body (which is what re-runs the component that read). Its memoized children are carried over with it on a hit, and still hit on their own entries when its body runs.

Where to put `memo` follows the placement rule in [When to memoize](/docs/rust/programming_guide/function/#when-to-memoize) — *memoize a boundary only if its key can hit while every enclosing memo boundary misses*: the per-file pipeline above composes a file-level component memo with a chunk-level embedding memo, while a *memoized* component whose path is keyed by a content hash of its input makes any inner memo over the same content mostly redundant.

## How target states sync

The component path tree determines ownership. When a component is no longer mounted at a path (e.g., a source file is deleted), CocoIndex automatically cleans up its target states — and recursively for all its sub-paths.

**Info — Sync Mechanism**

After a processing component finishes, CocoIndex syncs its target states:

1. **Compares** the target states declared in this run against those from the previous run at the same path
2. **Applies changes** as a unit — creating, updating, or deleting target states as needed
3. **Recursively cleans up** sub-paths where components are no longer mounted

All writes happen strictly after processing completes — you never see partial effects from a processing failure or interrupt. Each target backend applies its batch atomically when supported (e.g., within a database transaction), but changes across different target backends are not transactional with each other. When several components finish around the same time, a backend may receive their changes in one batch; that is only an optimization — if the batch fails, CocoIndex retries per component so one component's failure never undoes another's.

## What happens when a component fails

CocoIndex processes each component in two phases: **processing** (running your function, declaring target states) and **submit** (writing changes to target backends).

### Failure isolation

The framework's general rule: **a call returns an error on failure iff the failed work was on the critical path for the call to return.**

- **`use_mount!`** (and `use_mount_each!`, `ctx.scope`) — you're awaiting the child's *value*, so the child succeeding is on the critical path. The child's error propagates to the parent.
- **`mount!` and `mount_each!`** (and `ctx.mount`, `ctx.mount_each`) — these return as soon as the work is *scheduled* (you get a handle back). The child's execution runs in the background, off your critical path. A failure in one child does **not** affect the parent or siblings — by default the error is logged and other components continue. One bad file shouldn't take down the entire pipeline.

To react to background failures, you can:

- Install [exception handlers](/docs/rust/advanced_topics/exception_handlers/#exception-handlers) — global or scoped — to send alerts, record metrics, or implement custom logic. `EnvironmentBuilder::exception_handler` installs a global handler and `ctx.with_exception_handler(handler)` returns a scoped `Ctx`; a child mounted from it inherits the handler chain. A handler returns `Ok(())` to swallow the failure or `Err(_)` to pass it on, and the last error reaches `handle.ready()`.
- [Monitor app progress](/docs/rust/advanced_topics/progress_monitoring/) — `RunStats` exposes per-component stats including error counts, so you can detect failures programmatically.

For the full picture (including the critical-path principle applied to every API, interrupted update recovery, and the exception handler API) see [Error Handling](/docs/rust/advanced_topics/exception_handlers/).

### No rollback, convergent roll-forward

CocoIndex does not roll back partial writes. The two-phase design makes this safe:

- **Processing** is side-effect-free — it only declares target states in memory. If processing fails (e.g., a parsing error), no writes were attempted, so there's nothing to undo.
- **Submit** writes changes to target backends. If a submit fails partway through (e.g., a database connection drops), some writes may have been applied. CocoIndex does not attempt to undo them. Instead, on the next run CocoIndex computes the current desired state, and the target connector reconciles against all possible previous states — converging the target to the correct state regardless of what was partially applied. This is why built-in connectors use convergent operations like upserts (`INSERT ... ON CONFLICT DO UPDATE`) rather than plain inserts.

## How big should a processing component be?

When defining processing components, think about granularity (what one path represents) because it determines the sync boundary for target states.

For example, if you're processing files:

- **Coarse**: one component for "all files" (path `process`)
- **Medium**: one component per file (path `process / file_path`)
- **Fine**: one component per chunk (path `process / file_path / chunk_id`)

In general:

- **Coarse-grained** (fewer, larger components): More target states sync together as a unit, but you only see updates after the larger component finishes.
- **Fine-grained** (more, smaller components): Each component syncs its target states as soon as it finishes, but target states owned by different components do not sync together as a unit.

For small datasets, a single processing component that owns all target states is simple and ensures all target states sync together. As data grows, consider breaking it down into one component per source item (e.g., one per file) to reduce latency: you see each item's target states synced as soon as it's processed, without waiting for the full dataset to complete. This also helps isolate failures to that item.

## Passing the context to other tasks

The component context is the `Ctx` value your function receives, so it goes wherever you pass it. A `Ctx` is `Clone`: pass a clone into a spawned task, and CocoIndex still tracks component relationships and target state ownership there:

```rust
let ctx = ctx.clone();
tokio::spawn(async move { limited(&ctx, item).await })
```

There is no ambient context to capture and attach: every function takes the `Ctx` explicitly.

## Processing helpers

### Using `ctx.map` to apply transformations

`ctx.map(items, f)` applies an async closure to each item in a collection, running all calls concurrently within the current processing component. Unlike [`mount!`](#mount) and [`mount_each!`](#mount_each), it does **not** create child processing components — it's purely concurrent execution (similar to `futures::future::join_all`).

```rust
#[cocoindex::function(memo, memo_key(table = skip))]
async fn process_file(ctx: &Ctx, file: &FileEntry, table: &postgres::TableTarget) -> Result<()> {
    let text = file.read_text().await?;
    let chunks = RecursiveSplitter::new()?.split_with(
        &text,
        RecursiveChunkConfig {
            chunk_size: CHUNK_SIZE,
            min_chunk_size: None,
            chunk_overlap: Some(CHUNK_OVERLAP),
            language: Some("markdown".to_string()),
        },
    );
    let filename = file.key();
    let id_gen = IdGenerator::new();
    ctx.map(chunks, async |chunk| {
        process_chunk(ctx, chunk, &text, &filename, &id_gen, &table).await
    })
    .await?;
    Ok(())
}
```

The futures run in place, so the closure may borrow the caller's values (the text, the `IdGenerator`, the target).

The closure receives each item. `ctx.map` returns a `Vec` of the results, in the same order as the input items. Every started item runs to completion, then the first error in input order is returned.

#### When to use `ctx.map` vs `mount_each!`

- Use **`mount_each!`** when each item should be its own processing component — with its own component path, target state ownership, and target states sync boundary.
- Use **`ctx.map`** when you want to process items concurrently *within* the current component, without creating new component boundaries. This is common for sub-item work like processing chunks within a file — the same within-component chunk work the [chunk-embedding pipeline in Core Concepts](/docs/rust/programming_guide/core_concepts/#processing-component) walks through.