Running in live mode

Keep an App running after its initial sweep so it reacts to source changes continuously. Covers UpdateOptions { live: true }, ctx.mount_each_live, and live-aware connectors like the local filesystem and Kafka.

Version

v 1.0.14

Last reviewed

Nov 4, 2026

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

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:

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 — 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:

• 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() and kafka::topic_as_map.

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).

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. It runs the body once, signals readiness, then re-runs it on a fixed delay:

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 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):

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 connector supports live mode via walk_dir(..).live(), which watches for file system changes:

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.

kafka — consuming a topic with LiveMapView

The 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:

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 — the trait for components that react to changes incrementally, mounted with ctx.mount_live. Most flexible — full control over the lifecycle.
• LiveMapFeed / 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 — 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, kafka::topic_as_map in kafka, iggy::topic_as_map in 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.