# Building live components

> **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/advanced_topics/live_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

By default, a [processing component](/docs/rust/programming_guide/processing_component/) runs in **catch-up mode** — on each app update, it declares all target states and mounts all sub-components from scratch. CocoIndex handles incremental updates by skipping [memoized](/docs/rust/programming_guide/function/) sub-components and reconciling target states at the end, then the component exits. This works well when the dataset is small enough to scan fully each cycle.

When the dataset is large or you need to react to changes continuously (e.g., watching a file system), you want the component itself to stay running and react incrementally. A **live component** does an initial full scan (same as catch-up mode), then keeps running and reacts to individual changes without rescanning everything.

## The LiveComponent trait

A live component is a type that implements the `LiveComponent` trait, with two methods:

```rust
use cocoindex::prelude::*;

struct MyLiveComponent {
    folder: PathBuf,
    target: DirTarget,
}

#[cocoindex::async_trait]
impl LiveComponent for MyLiveComponent {
    /// Full processing: mount all children, declare all target states.
    async fn process(&self, ctx: Ctx) -> Result<()> {
        // ...
        Ok(())
    }

    /// Continuous processing: orchestrate full and incremental updates.
    async fn process_live(&self, operator: LiveComponentOperator) -> Result<()> {
        operator.update_full().await?;
        operator.mark_ready().await;
        // ...
        Ok(())
    }
}
```

- **The fields** hold the component's arguments: you build the value and pass it to `ctx.mount_live(&key, component)`.
- **`process(ctx)`** does a full scan — mounts children via `ctx.mount` / `ctx.mount_each` and declares target states, just like a traditional component function. Called indirectly via `operator.update_full()`.
- **`process_live(operator)`** is the long-running entry point. It orchestrates full and incremental updates using the operator. Its default body runs one full pass, then `mark_ready`.

The trait is the marker, and `ctx.mount_live` is the mount call.

## LiveComponentOperator

The `operator` passed to `process_live()` provides these methods, with the child's body as a closure:

| Method | Description |
|--------|-------------|
| `operator.update_full().await?` | Run `process()` with a full submission phase (GCs removed children). Blocks until fully ready. |
| `operator.update(key, \|ctx\| async move { .. }).await?` | Mount a child component under `key` incrementally. Awaits until its state has synced. |
| `operator.start_update(key, \|ctx\| ..).await?` | The same without waiting: returns a `ReadyHandle` whose `ready()` awaits the sync. |
| `operator.delete(key).await?` / `start_delete(key)` | Delete a child component, waiting for the sync or returning a `ReadyHandle`. |
| `operator.mark_ready().await` | Signal that processing has caught up to the time `process_live()` was called. |
| `operator.is_live()` | Whether the update runs in live mode. |
| `operator.read_committed_state(key)` / `write_committed_state(key, &value)` | A durable value of the component's own, such as a cursor, kept across runs. |

There is no `report_exception` in Rust. Return the error from `process_live` instead: before `mark_ready`, the mount fails with it; after `mark_ready`, it is logged.

### `update_full()`

Triggers a full processing cycle: calls `process()`, submits target states, waits for all children to be ready, and garbage-collects children that are no longer mounted. This is the same mechanism as a traditional component's update cycle.

Errors returned from `process()` are routed through the parent's [exception handler chain](/docs/rust/advanced_topics/exception_handlers/#exception-handlers) — same as background `ctx.mount` failures — and **do not propagate** to the caller. Components that `process()` mounts in the background report their own failures through the chain in effect where they are mounted, as any background mount does. This lets long-running `process_live` loops (such as periodic refreshers) keep going across transient cycle failures while still surfacing the failure to operators.

### `update()` and `delete()`

Mount or delete individual child components without a full scan. These are concurrent with each other but serialized with `update_full()` — if `update_full()` is running, incremental operations wait until it finishes.

When multiple operations target the same key, only the latest one (by invocation order) takes effect.

Failures in the mounted child (or in the delete itself) are routed through the parent's [exception handler chain](/docs/rust/advanced_topics/exception_handlers/#exception-handlers) with `mount_kind` set to `MountKind::Update` or `MountKind::Delete` and `stable_path` set to the *child's* path — same shape as background `ctx.mount` failures. The live loop is unaffected by child failures.

### `mark_ready()`

Signals to the parent that the live component has caught up. The parent's `ctx.mount_live(...)` call returns when `mark_ready()` is called. If `process_live()` returns without calling `mark_ready()`, it is called automatically.

### Handling failures

`update_full`, `update` and `delete` route their failures through the handler chain, with `mount_kind` set to `MountKind::UpdateFull`, `MountKind::Update` or `MountKind::Delete`. Register a handler with `ctx.mount_live_with_handler(&key, component, handler)` for one component, with `ctx.with_exception_handler` for the mounts of one scope, or on the environment builder for all of them (see [Exception handlers](/docs/rust/advanced_topics/exception_handlers/#exception-handlers)).

## Example: file system watcher

A component that watches a local folder and processes each file:

```rust
struct FolderWatcher {
    source: Vec<(String, String)>,
    provider: TargetStateProvider<String>,
}

#[cocoindex::async_trait]
impl LiveComponent for FolderWatcher {
    /// Full scan: one child per entry of the source.
    async fn process(&self, ctx: Ctx) -> Result<()> {
        let provider = self.provider.clone();
        ctx.mount_each(
            self.source.clone(),
            |(k, _v): &(String, String)| k.clone(),
            move |child, (k, v)| {
                let provider = provider.clone();
                async move { declare_target_state(&child, provider.target_state(k, v)) }
            },
        )
        .await?;
        Ok(())
    }

    async fn process_live(&self, operator: LiveComponentOperator) -> Result<()> {
        // 1. Full scan.
        operator.update_full().await?;
        // 2. Signal readiness: the parent can proceed.
        operator.mark_ready().await;
        // 3. React to changes: delete the child mounted for key "a".
        operator.delete("a").await?;
        Ok(())
    }
}
```

For a real directory, `cocoindex` ships the watcher: `walk_dir(dir).live()` returns a `LiveMapView`, and `ctx.mount_each_live` mounts one child per file (see `examples/rust/text_embedding`). Write a `LiveComponent` by hand when the change source is your own.

The parent mounts it like any other component:

```rust
app.update(move |ctx| async move {
    ctx.mount_live(&"comp", FolderWatcher { source, provider }).await
})
.await?;
```

`mount_live` returns when the component marks itself ready. In live mode the component keeps running in the background until the app is dropped.

## Example: traditional component equivalent

A traditional single-function component:

```rust
app.update(move |ctx| async move {
    for (k, v) in rows {
        declare_target_state(&ctx, provider.target_state(k, v.to_string()))?;
    }
    Ok(())
})
.await?;
```

is equivalent to this LiveComponent:

```rust
struct ProcessAll {
    provider: TargetStateProvider<String>,
    rows: Vec<(&'static str, &'static str)>,
}

#[cocoindex::async_trait]
impl LiveComponent for ProcessAll {
    async fn process(&self, ctx: Ctx) -> Result<()> {
        for (k, v) in &self.rows {
            declare_target_state(&ctx, self.provider.target_state(*k, v.to_string()))?;
        }
        Ok(())
    }
    // The default `process_live` runs one full pass and marks the component ready.
}
```

The key difference: the LiveComponent version can later be extended to handle incremental changes in `process_live()` without changing `process()`.

## Example: periodic refresh with `ctx.auto_refresh`

<span id="example-periodic-refresh-with-cocoauto_refresh"></span>

A common pattern is "run a regular processor on a fixed schedule." `ctx.auto_refresh` mounts a live component that re-runs a closure on an interval — no need to write `process_live` yourself.

```rust
use std::time::Duration;

ctx.auto_refresh(&"sync_users", Duration::from_secs(5 * 60), move |ctx| {
    let table = table.clone();
    async move {
        let rows = fetch_all_users(&ctx).await?;
        for row in rows {
            table.declare_row(&ctx, &row)?;
        }
        Ok(())
    }
})
.await?;
```

`ctx.auto_refresh(&key, interval, f)` mounts the refresh component under `key`. `f` is called with a fresh `Ctx` each cycle, so it must be `Fn` and `Clone`; clone the handles it needs into the closure.

Behavior:
- The first cycle runs immediately, then `mark_ready` is signaled to the parent.
- In live mode, the loop continues: `sleep(interval) → run process_fn → ...` with a **fixed delay** between cycles (cycles never overlap; a slow cycle just stretches the gap).
- In catch-up mode, the first cycle runs and `auto_refresh` terminates — observationally identical to mounting `process_fn` directly. The interval is ignored.
- Cycle failures are routed through the parent's [exception handler chain](/docs/rust/advanced_topics/exception_handlers/#exception-handlers) (see `update_full()` above); the loop keeps going.
- **Each cycle reconciles target states against the previous cycle.** If `process_fn` no longer declares some target state (e.g. a row that vanished from your source table), CocoIndex automatically deletes the corresponding target — same garbage-collection model as a regular component re-run. You write the function as if processing from scratch each time; the framework handles the diff. See [Target State](/docs/rust/programming_guide/target_state/) for the underlying model.

Here `process_fn` is the closure passed to `ctx.auto_refresh`.

## LiveMapFeed and LiveMapView
**Tip**
For an introduction to live mode, see [Live Mode](/docs/rust/programming_guide/live_mode/). Connector authors can implement the protocols below to supply live collections.

Two traits represent live keyed collections that `ctx.mount_each_live` can consume. Choose based on whether your source can enumerate its current state:

- **`LiveMapView`** — the source has enumerable current state that can be scanned (e.g., a directory listing, database table). Supports both full scans via `scan()` and incremental changes via `watch()`.
- **`LiveMapFeed`** — the source only streams changes, with no "current state" to scan (e.g., a Kafka consumer, a webhook event stream). Provides only `watch()`.

```rust
#[async_trait]
pub trait LiveMapFeed<K, V>: Send + Sync + 'static {
    /// Stream changes to `subscriber` until cancelled. Call
    /// `subscriber.mark_ready()` once caught up.
    async fn watch(&self, subscriber: LiveMapSubscriber<K, V>) -> Result<()>;

    /// Whether to skip the initial catch-up scan in live mode. Defaults to `false`.
    async fn skip_initial_scan(&self, _operator: &LiveComponentOperator) -> Result<bool> {
        Ok(false)
    }
}

#[async_trait]
pub trait LiveMapView<K, V>: LiveMapFeed<K, V> {
    /// All current `(key, value)` pairs, for the catch-up full pass and re-scans.
    async fn scan(&self) -> Result<Vec<(K, V)>>;
}
```

`LiveMapView` extends `LiveMapFeed`. `ctx.mount_each_live(&key, feed, process_fn)` takes a `LiveMapView` and creates an internal `LiveComponent`: `scan()` yields all `(key, value)` pairs for full scans (called inside the internal `process()`), and `watch()` handles the incremental changes. A watch-only source implements `scan` as an empty snapshot, so everything arrives through `watch()`. A durable feed can override `skip_initial_scan` to resume from a cursor it kept with `operator.read_committed_state`.

The live mount is a separate method, `ctx.mount_each_live`; `ctx.mount_each` and `ctx.mount_each_stream` only take a plain collection or stream of items.

### LiveMapSubscriber

The `subscriber` passed to `watch()` mirrors `LiveComponentOperator`:

| LiveMapSubscriber | LiveComponentOperator | Description |
|---|---|---|
| `subscriber.update_all().await?` | `operator.update_full().await?` | Full re-scan of all items |
| `subscriber.mark_ready().await` | `operator.mark_ready().await` | Signal readiness |
| `subscriber.update(key, value).await?` | `operator.update(key, f).await?` | Incremental update; awaits until the child's state has synced |
| `subscriber.delete(key).await?` | `operator.delete(key).await?` | Incremental delete; awaits until the deletion has synced |
| `subscriber.start_update(key, value).await?` / `start_delete(key)` | `operator.start_update` / `start_delete` | The same without waiting; returns a `ReadyHandle` |

The subscriber also has `is_live()`, and `read_committed_state` / `write_committed_state` for the feed's own durable cursor.

A typical `watch()` implementation for a `LiveMapView`:

```rust
/// Watch-only stream: emits a fixed set of incremental updates after readiness.
struct StreamFeed {
    items: Vec<(String, i64)>,
}

#[async_trait]
impl LiveMapFeed<String, i64> for StreamFeed {
    async fn watch(&self, subscriber: LiveMapSubscriber<String, i64>) -> Result<()> {
        subscriber.mark_ready().await;
        for (k, v) in &self.items {
            subscriber.update(k.clone(), *v).await?;
        }
        Ok(())
    }
}

#[async_trait]
impl LiveMapView<String, i64> for StreamFeed {
    async fn scan(&self) -> Result<Vec<(String, i64)>> {
        // Empty initial snapshot; everything arrives via watch().
        Ok(Vec::new())
    }
}
```

The framework runs the initial scan itself, before `watch()`, so `watch()` starts with `mark_ready`.

### Implementing live support for a connector

To add live support to a source connector, make your source return a type that implements `LiveMapView` (if the source has scannable state) or `LiveMapFeed` (if it only streams changes):

- **`LiveMapView` example**: The [`localfs`](/docs/rust/connectors/localfs/) walker — `walk_dir(dir).live()` (feature `fs_live`) returns a `LiveMapView`.
- **`LiveMapView` over a stream**: The [`kafka`](/docs/rust/connectors/kafka/) connector — `kafka::topic_as_map(&consumer, topic)` implements `LiveMapView`, with a scan that reads the latest message per key.
- **In-process**: `LiveMap<K, V>`, described below.

### LiveMap: an in-process live source

`LiveMap<K, V>` joins live data-*producing* logic and live data-*consuming* logic inside one session. The producer declares `(key, value)` entries as target states with `declare_entry`; the consumer reads the map as a `LiveMapView` through `ctx.mount_each_live`. An entry exists as long as some component declares it; when its declaring component stops declaring it, the entry is removed, and the consumer's child for it is deleted.

```rust
use cocoindex::prelude::*;

/// Declares every entry of `source`.
struct Producer {
    lm: LiveMap<String, String>,
    source: Vec<(String, String)>,
}

#[cocoindex::async_trait]
impl LiveComponent for Producer {
    async fn process(&self, ctx: Ctx) -> Result<()> {
        for (key, value) in self.source.clone() {
            self.lm.declare_entry(&ctx, key, value)?;
        }
        Ok(())
    }
}

/// Consumes the map: one child per entry.
struct Consumer {
    lm: LiveMap<String, String>,
}

#[cocoindex::async_trait]
impl LiveComponent for Consumer {
    async fn process(&self, ctx: Ctx) -> Result<()> {
        ctx.mount_each_live(&"each", self.lm.clone(), move |_ctx, value: String| async move {
            // ... process one entry ...
            Ok(())
        })
        .await
    }
}

// Inside the component tree:
let lm: LiveMap<String, String> = LiveMap::create(&ctx).await?;
```

`LiveMap::create(&ctx)` mounts the map's backing container target. The map is an in-memory, live-mode primitive, not state that persists across runs.

### Single-subscriber contract

A live feed (`LiveMapView`, `LiveMapFeed`, or `LiveStream`) is **single-subscriber**. Its `watch()` typically owns an exclusive underlying resource (a consumer subscription, an OS file watch, a single in-memory change channel) that can't be fanned out to two concurrent callers — e.g. two subscriptions would race one consumer's offset commits. So a given feed instance supports **one active `watch()` (one `mount_each_live`) at a time**; consuming it from two places concurrently is unsupported. Fan-out to multiple subscribers, if ever needed, belongs in a layer above the feed.

Enforce this in your `watch()` with `cocoindex::live_component::SingleWatcherGuard`, which returns an error on a second concurrent call:

```rust
struct MyFeed { watch_guard: SingleWatcherGuard, /* ... */ }

#[async_trait]
impl LiveMapFeed<String, Vec<u8>> for MyFeed {
    async fn watch(&self, subscriber: LiveMapSubscriber<String, Vec<u8>>) -> Result<()> {
        let _token = self.watch_guard.enter()?;
        // ... body ...
    }
}
```

Build the guard with `SingleWatcherGuard::new("MyFeed")`. `enter()` returns an error on a second concurrent `watch()`; the flag clears when the token drops, on return, error or cancellation.

## LiveStream

Some change sources are not naturally keyed maps — they deliver opaque event payloads (e.g. cloud-storage event notifications, raw message buses) that the consuming connector translates into map updates on its own terms. For these, CocoIndex provides **`LiveStream`**: a keyless sibling of `LiveMapFeed` — an unkeyed sequence of messages without any "key" or "delete" semantics. Connectors typically wrap a user-supplied `LiveStream` into their own `LiveMapView` (or `LiveMapFeed`) by translating each event into the appropriate key-level operation.

```rust
#[async_trait]
pub trait LiveStream<M: Send + 'static>: Send + Sync + 'static {
    /// Deliver messages to `subscriber` until cancelled.
    async fn watch(&self, subscriber: &dyn LiveStreamSubscriber<M>) -> Result<()>;
}

#[async_trait]
pub trait LiveStreamSubscriber<M>: Send + Sync {
    async fn send(&self, message: M) -> Result<ReadyHandle>;
    async fn mark_ready(&self);
    // Also `read_committed_state` / `write_committed_state`, with
    // defaults that keep nothing, for a stream's own durable cursor.
}
```

A subscriber returns `ReadyHandle::immediate()` for a message that needs no work, and `ReadyHandle::from_future(..)` (for example the handle of `LiveMapSubscriber::start_update`) otherwise.

`send()` is awaited by the producer to obtain a `ReadyHandle` handle; the handle's `ready()` is then awaited later (often by an internal offset/commit tracker) to signal that the message has been fully processed. The stream awaits each `send()` inline, so a slow subscriber naturally throttles the producer — at most one message is in flight per stream, bounding adapter memory regardless of producer rate.

`mark_ready()` must return promptly — it is awaited inline in the producer's loop. Subscribers that need to wait on additional preconditions (e.g. a pending scan) should spawn a background task and return.

### Implementing a LiveStream-based connector

A connector built on `LiveStream` typically:

1. Accepts a user-supplied `LiveStream<...>` as an argument — the user wires up the underlying transport.
2. Wraps the stream into a `LiveMapView` (or `LiveMapFeed`): `scan()` runs an initial scan if the source has scannable state; `watch()` subscribes to the stream and translates each event into `subscriber.update(key, value)` / `subscriber.delete(key)` (or their `start_*` forms, whose `ReadyHandle` the stream awaits before it acknowledges the event).
3. Lets the underlying `LiveStream` handle offset commits, readiness signaling, and back-pressure.
4. Guards its `watch()` against concurrent subscription — see [Single-subscriber contract](#single-subscriber-contract) — with `SingleWatcherGuard`.

**Example**: The [`oci_object_storage`](/docs/rust/connectors/oci_object_storage/#live-bucket-watching) connector's `list_objects_live` takes a `LiveStream<Vec<u8>>` of event payloads (typically [`kafka::topic_as_stream(..).payloads()`](/docs/rust/connectors/kafka/#as-a-live-stream)) and translates OCI Object Storage events into bucket-keyed updates.

The Kafka connector exposes `topic_as_stream()` precisely so that downstream connectors like this one can plug into Kafka without inheriting the keyed-map semantics of `topic_as_map()`.

## Live mode vs catch-up mode

See [Live Mode](/docs/rust/programming_guide/live_mode/) for how the two modes are enabled at the app level and an overview of how they work.

For a manual `LiveComponent`, the mode controls whether `process_live()` continues running after `mark_ready()`:

- **Live mode** (`UpdateOptions { live: true, .. }` on `app.update_with_options(..)`, or `cocoindex update -L`): `process_live()` continues after `mark_ready()` — the component keeps watching for changes.
- **Catch-up mode** (the default): `process_live()` terminates as soon as `mark_ready()` is awaited. No code after `operator.mark_ready().await` executes, so the component behaves like a traditional one-shot processor.

This lets you use the same `LiveComponent` type in both modes without code changes. `operator.is_live()` tells the body which mode it runs in.

## Restrictions

### No `use_mount!` inside `process()`

`process()` may only mount background children (`ctx.mount`, `mount!`, `ctx.mount_each`, `mount_each!`). Any setup that requires `ctx.scope` / `use_mount!` (such as declaring target tables) must be done in the **parent** component before mounting the LiveComponent. This keeps the controller's provider set stable across full and incremental updates.

### Mounting a LiveComponent

A `LiveComponent` is mounted only with `ctx.mount_live` (or `ctx.mount_live_with_handler`). To mount one live component per item, use `ctx.mount_each_live_component(items, key_fn, make_component)`; to fan one feed out into closure-driven children, use `ctx.mount_each_live`.

## Readiness

The parent's `ctx.mount_live(...)` call returns when `mark_ready()` is called inside `process_live()`, regardless of whether `process_live()` is still running.

```mermaid
sequenceDiagram
    participant Parent
    participant LC as LiveComponent
    participant Op as Operator
    participant Children

    Parent->>LC: mount()
    activate LC
    Note over LC: process_live(operator)

    LC->>Op: update_full()
    activate Op
    Op->>LC: process()
    activate LC
    LC->>Children: mount(A), mount(B), ...
    deactivate LC
    Note over Op: submit + GC<br/>wait children ready
    Op-->>LC: done
    deactivate Op

    LC->>Op: mark_ready()
    Op-->>Parent: readiness resolved

    rect rgb(240, 248, 255)
        Note over LC,Children: Live mode only
        LC->>Op: update(A)
        Op->>Children: run A
        LC->>Op: delete(B)
        Op->>Children: delete B
        Note over LC: continues...
    end
    deactivate LC
```

If `process_live()` returns without calling `mark_ready()`, it is called automatically — the parent will not hang.