# Running in live mode

> **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/live_mode/ · 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, an update (`app.update(body)`) runs in **catch-up mode**: it scans all sources, processes what changed since the last run (memoized components are skipped, so unchanged work is not redone), syncs target states, and returns. Targets are caught up to the moment the run started, and that's it — to pick up further changes, you run an update again.

**Live mode** keeps the app running after catch-up finishes and lets components stream changes continuously from their sources (e.g. a file system watcher or a database change feed), applying them to target states with low latency. This is useful when:

- You want near-real-time reactions to source changes, instead of waiting for the next update
- Your sources can push changes more efficiently than a full rescan can discover them

Two things are needed for live mode to work: the app must be **enabled** to stay running, and somewhere in the component tree a component must **react** to changes.

## Enabling live mode

Set `live: true` in the `UpdateOptions` of the update:

```rust
let handle = app.start_update_with_options(
    UpdateOptions {
        full_reprocess: false,
        live: true,
        ..UpdateOptions::default()
    },
    body,
)?;
handle.result().await?;
```

`app.update_with_options(options, body).await?` is the form without a handle.

From the CLI:

```bash
cocoindex update --live
# or
cocoindex update -L
```

The `live` flag propagates top-down through the component tree — both `mount!` and `use_mount!` (and the `ctx.mount` / `ctx.scope` methods behind them) inherit `live` from the parent, so children are live when the app is live.

Without `live: true` on the update, the app runs in catch-up mode — everything completes after the initial scan, even if a source supports live watching.

## Reacting to changes

Enabling live mode keeps the app running, but something in the component tree needs to actually watch for changes. That something is a [**LiveComponent**](/docs/rust/advanced_topics/live_component/) — a component with a long-running `process_live()` method that delivers incremental updates.

You rarely need to write a `LiveComponent` manually. The two most common patterns are:

### Sources with `LiveMapView`

Source connectors provide live capabilities through two [traits](/docs/rust/advanced_topics/live_component/#livemapfeed-and-livemapview):

- **`LiveMapFeed`** — `watch` streams changes to a subscriber. A feed with no snapshot to scan works only in live mode.
- **`LiveMapView`** — a `LiveMapFeed` that also has a `scan` of its current state, so it works in both catch-up and live modes. Examples: [`walk_dir(..).live()`](/docs/rust/connectors/localfs/#live-file-watching) and [`kafka::topic_as_map`](/docs/rust/connectors/kafka/#as-source).

`ctx.mount_each_live(&key, feed, f)` takes a `LiveMapView` and mounts one child component per key. It creates a `LiveComponent` internally that:

1. **Scans current state** (if available) — iterates all items and mounts a processing component for each
2. **Signals readiness** — the initial scan is complete (or the stream has caught up), target states are synced
3. **Watches for changes** — the source delivers incremental updates:
   - New or modified items → re-mount the affected component
   - Deleted items → remove the component and its target states

CocoIndex handles change detection, memoization, and target state reconciliation the same way as in catch-up mode.

In catch-up mode, `mount_each_live` scans the feed once and returns. `mount_each!` and `ctx.mount_each` take a plain iterator instead, and always do a one-time iteration over the items. A `LiveComponent` of your own implements the `LiveComponent` trait (`process`, and `process_live` for the reactive body) and is mounted with `ctx.mount_live(&key, component)`.

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

### Periodic refresh with `ctx.auto_refresh`

When the source isn't change-aware but you still want fresh data (say, polling a REST endpoint or re-reading a database table that doesn't emit change events) mount your processing body with [`ctx.auto_refresh`](/docs/rust/advanced_topics/live_component/#example-periodic-refresh-with-cocoauto_refresh). It runs the body once, signals readiness, then re-runs it on a fixed delay:

```rust
ctx.auto_refresh(&"poller", Duration::from_secs(60), |ctx| async move {
    // declare the rows that exist now ...
    Ok(())
})
.await?;
```

`ctx.auto_refresh(&key, interval, f)` mounts the refresh component under `key`. In catch-up mode it runs `f` once and returns. In live mode it runs once, marks the component ready, then runs again after each `interval`.

**Catch-up compatibility:** in catch-up mode (the default), `auto_refresh` runs the body once and returns. The interval is ignored. Same pipeline, choose catch-up or live at run time.

**Handling deletes:** each cycle's declarations are reconciled against the previous run. If a row disappears from the source table between polls, the body simply doesn't declare it that cycle — CocoIndex automatically deletes the corresponding target. You don't need to track deletions yourself.

### Decoupling stages with an in-memory `LiveMap`

When one part of your pipeline produces keyed data and another consumes it (without going through an external system) put an in-memory [`LiveMap`](/docs/rust/common_resources/live_map/) between them. Producers declare `(key, value)` entries; the consumer reads it as a `LiveMapView` via `ctx.mount_each_live`, reacting to adds, updates, and deletes just like a live source.

`LiveMap<K, V>` takes any `LiveMapKey` as `K` (`String`, `i64`, `u64` or a `StableKey`):

```rust
let lm: LiveMap<i64, String> = LiveMap::create(&ctx).await?;
lm.declare_entry(&ctx, 7, "seven".to_string())?;           // producer side
ctx.mount_each_live(&"entries", lm.clone(), process_entry).await?; // consumer side
```

## Examples

### `localfs` — file watching with `LiveMapView`

The [`localfs`](/docs/rust/connectors/localfs/) connector supports live mode via `walk_dir(..).live()`, which watches for file system changes:

```rust
async fn index(ctx: Ctx, sourcedir: PathBuf) -> Result<()> {
    // ...
    // The source supports live watch; pass `-L` to `cocoindex update` to run live.
    let files = walk_dir(sourcedir)
        .recursive(true)
        .path_matcher(PatternFilePathMatcher::include(["**/*.md"])?)
        .live()
        .poll_interval(Duration::from_secs(1));
    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
}
```

`walk_dir(..).live()` needs the `fs_live` feature. It follows the directory with the platform's native file events (each event updates or removes one file's component), polls at `poll_interval` where those are unavailable, and walks the whole directory every `rescan_interval` (an hour by default) as a safety net.

**Catch-up compatibility:** `LiveMapView` sources also work in catch-up mode — they do the initial full scan and exit cleanly. You can write your pipeline once and choose catch-up or live at run time.

For a complete working example, see [`examples/rust/text_embedding`](https://github.com/cocoindex-io/cocoindex/tree/main/examples/rust/text_embedding).

### `kafka` — consuming a topic with `LiveMapView`

The [`kafka`](/docs/rust/connectors/kafka/) connector treats a topic as a live keyed map — each message is an upsert or delete for a key. `kafka::topic_as_map` returns a `LiveMapView`: its scan reads every partition up to its high-watermark, keeping the latest message per key, and its watch tails new records from there. With a consumer group, a live run resumes from the committed offsets instead of scanning:

```rust
use cocoindex::connectors::kafka::{self, KafkaMessage};

let consumer = kafka::KafkaConsumer::connect_with_options(
    &broker_refs,
    kafka::KafkaConsumerOptions {
        consumer_group: Some("kafka_consume".to_string()),
        ..Default::default()
    },
)
.await?;

let feed = kafka::topic_as_map(&consumer, topic);
ctx.mount_each_live(&"messages", feed, move |_ctx, message: KafkaMessage| {
    async move {
        println!("message {}@{}/{}", message.key_str().unwrap_or_default(), message.partition, message.offset);
        Ok(())
    }
})
.await
```

The full program, with the `--live` switch, is `examples/rust/kafka_consume/src/main.rs`.

## Going deeper

The abstractions behind live mode, from most general to most specific:

- **[LiveComponent](/docs/rust/advanced_topics/live_component/)** — the trait for components that react to changes incrementally, mounted with `ctx.mount_live`. Most flexible — full control over the lifecycle.
- **[LiveMapFeed / LiveMapView](/docs/rust/advanced_topics/live_component/#livemapfeed-and-livemapview)** — traits for a changing collection of keyed items. `ctx.mount_each_live` uses a `LiveMapView` to construct a `LiveComponent` automatically. Connector authors implement them to add live support.
- **[`ctx.auto_refresh`](/docs/rust/advanced_topics/live_component/#example-periodic-refresh-with-cocoauto_refresh)** — mounts a regular processing body as a `LiveComponent` that re-runs on a fixed interval. Use when there's no change-event source.
- **Source connectors** — provide a `LiveMapView` from their source APIs (e.g., `walk_dir(..).live()` in [`localfs`](/docs/rust/connectors/localfs/), `kafka::topic_as_map` in [`kafka`](/docs/rust/connectors/kafka/), `iggy::topic_as_map` in [`iggy`](/docs/rust/connectors/iggy/)). Users just pass the result to `ctx.mount_each_live()`.

For custom change feeds, fine-grained lifecycle control, or implementing live map traits on your own connector, see [Live Components](/docs/rust/advanced_topics/live_component/).