Building live components

Implement the LiveComponent trait with process() for full scans and process_live(operator) for incremental updates — covers LiveComponentOperator (update_full, update, delete, mark_ready), the LiveMapFeed / LiveMapView / LiveStream traits over event sources, and the in-process LiveMap.

Version

v 1.0.14

Last reviewed

Nov 4, 2026

By default, a 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 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:

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:

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

Example: file system watcher

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

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:

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:

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:

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

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.

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 (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 for the underlying model.

Here process_fn is the closure passed to ctx.auto_refresh.

LiveMapFeed and LiveMapView

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

#[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.

LiveMapSubscriber

The subscriber passed to watch() mirrors LiveComponentOperator:

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:

/// 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 walker — walk_dir(dir).live() (feature fs_live) returns a LiveMapView.
• LiveMapView over a stream: The 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.

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:

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.

#[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 — with SingleWatcherGuard.

Example: The oci_object_storage connector’s list_objects_live takes a LiveStream<Vec<u8>> of event payloads (typically kafka::topic_as_stream(..).payloads()) 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 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.

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.