Building a custom target connector

Connect CocoIndex's declarative target-state system to external systems. Covers TargetHandler, tracking records, actions, sinks, root vs container targets, attachment providers, and idempotent operations.

Version

v 1.0.14

Last reviewed

Nov 4, 2026

A custom target connector is the mechanism that connects CocoIndex’s declarative target state system to external systems. When you call methods like dir_target.declare_file(ctx, name, content) or table_target.declare_row(ctx, &row), a target connector handles the actual synchronization — determining what changed and applying those changes to the external system.

When to create a custom target connector

Most users will use built-in connectors (like localfs or postgres) and never need to create their own. Consider creating a custom target connector when:

• You need to integrate with an external system not covered by existing connectors
• You need custom change detection logic (e.g., content-based fingerprinting)
• You need to manage hierarchical target states (containers with children)

Key data types

The labels distinguish types you implement from types CocoIndex provides.

The types are in cocoindex::prelude: the TargetHandler<V> trait, TargetActionSink<A>, TargetAction<A>, TargetReconcileOutput<A, R>, TargetChildInvalidation, TargetStateProvider<V>, TargetState<V>, ChildTargetDef, StableKey, and the functions register_root_target_states_provider, declare_target_state, declare_target_state_with_child and mount_target. The Rust snippets on this page come from rust/sdk/cocoindex/tests/target_state.rs, which exercises each piece without an external service. For a complete connector against a database, read rust/sdk/cocoindex/src/connectors/valkey.rs (a container target with child documents).

TargetHandler (you implement)

A TargetHandler implements the reconciliation logic. It’s a trait with one required method and one optional method:

pub trait TargetHandler<V>: Send + Sync + 'static
where
    V: Serialize + DeserializeOwned + Send + 'static,
{
    type TrackingRecord: Serialize + DeserializeOwned + Send + Sync + 'static;
    type Action: Serialize + DeserializeOwned + Send + 'static;

    fn reconcile(
        &self,
        key: StableKey,
        desired_target_state: Option<V>,
        prev_possible_records: Vec<Self::TrackingRecord>,
        prev_may_be_missing: bool,
    ) -> Result<Option<TargetReconcileOutput<Self::Action, Self::TrackingRecord>>>;

    // Optional: attachment handlers, keyed by attachment type name.
    fn attachments(&self) -> Result<Vec<(String, ChildTargetDef)>> {
        Ok(Vec::new())
    }
}

The value type V is the trait’s type parameter; the tracking record and the action are associated types. The child handler type is not part of the trait: a container’s sink builds it (see container targets).

Types:

• V: The specification for the target state (e.g., file content, row data). It is Serialize + DeserializeOwned.
• TrackingRecord: What’s stored to detect changes on future runs
• Action: What the sink applies to the external system

Parameters:

• key: StableKey — the key the target state was declared with
• desired_target_state: What the user declared, or None if no longer declared
• prev_possible_records: Tracking records from previous runs (may have multiple due to interrupted updates)
• prev_may_be_missing: If true, the target state might not exist in the external system

Returns:

• Ok(Some(TargetReconcileOutput { .. })) if an action is needed
• Ok(None) if no changes are required

A container handler (one used with declare_target_state_with_child or mount_target) must return Some output whenever desired_target_state is Some, even when nothing changed, and its action must be a Create or an Update: the sink’s child handler factory builds the child handler from that action. Returning None leaves the child handler missing, and declaring a child target state then fails.

Tracking record (you define)

A tracking record captures the essential information needed to detect changes. Good tracking records:

• Are minimal: Only include what’s needed for change detection
• Are deterministic: Same input always produces the same record
• Are serializable: Must be persistable: any Serialize + DeserializeOwned type.

// Example: a table's tracking record (here, the declared table spec itself)
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
struct TableSpec {
    generation: String,
}

Action and TargetActionSink (you implement)

An action (you define) describes what operation to perform on the external system:

/// One row mutation the sink applies. `value: None` is a delete.
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
struct RowAction {
    key: String,
    value: Option<String>,
}

A TargetActionSink batches and executes actions:

/// A sink that records each applied action into `log` as `"<verb> <key>[=<value>]"`.
fn recording_sink(log: Log) -> TargetActionSink<RowAction> {
    TargetActionSink::from_async_fn(move |actions: Vec<TargetAction<RowAction>>| {
        let log = log.clone();
        async move {
            let mut log = log.lock().unwrap();
            for action in actions {
                let (verb, row) = match action {
                    TargetAction::Create(r) => ("create", r),
                    TargetAction::Update(r) => ("update", r),
                    TargetAction::Delete(r) => ("delete", r),
                };
                log.push(match row.value {
                    Some(v) => format!("{verb} {}={}", row.key, v),
                    None => format!("{verb} {}", row.key),
                });
            }
            Ok(())
        }
    })
}

A sink is always async. Its closure receives Vec<TargetAction<A>>, where each item is TargetAction::Create(a), Update(a) or Delete(a), so the operation kind travels beside your action. The action type must be Serialize + DeserializeOwned. The closure gets no connection from the environment: move the connection into it when you build the handler (see Use ContextKey for external resource identity).

The engine batches actions by sink identity: actions returned with the same sink (or a clone of it) are applied together in one call, and every from_async_fn* call creates a new identity. Build the sink once and return a clone from each reconcile.

A container target’s sink also takes a child_handler factory, Fn(&A) -> Result<ChildTargetDef>, which returns the handler for the child target states of a created or updated container. The callback stays the same: it applies the actions. See Implementing container targets.

TargetReconcileOutput (you return)

TargetReconcileOutput bundles what reconcile() returns when an action is needed:

pub struct TargetReconcileOutput<A, R> {
    pub action: TargetAction<A>,                            // Create(A) | Update(A) | Delete(A)
    pub sink: TargetActionSink<A>,                          // How to execute it
    pub tracking_record: Option<R>,                         // What to remember; None = non-existence
    pub child_invalidation: Option<TargetChildInvalidation>, // Destructive | Lossy, for containers
}

The child_invalidation field is only relevant for container targets (those with children). See Child invalidation for details.

TargetStateProvider (CocoIndex provides)

A TargetStateProvider is a factory that creates TargetState objects. You don’t implement this type — CocoIndex gives you one when you register a handler or declare a target state with children.

// You get a provider from registration, for the current run.
let provider = register_root_target_states_provider(ctx, "my.target", handler)?;

// Or from declaring a parent target state (the child provider is ready once
// the enclosing component commits) ...
let child_provider = declare_target_state_with_child(ctx, parent_target_state)?;
// ... or from mounting the parent, which commits it in a foreground child
// component, so the child provider is ready at once.
let child_provider = mount_target(ctx, parent_target_state).await?;

// Or from an attachment on a mounted child provider.
let att_provider = child_provider.attachment(ctx, "vector_index")?;

Every call takes the Ctx: there is no ambient component context in Rust.

TargetState (CocoIndex provides)

A TargetState wraps a key and spec. You create these using the provider, then declare them:

// Create a target state and declare it for reconciliation
declare_target_state(ctx, provider.target_state("row-1", value))?;

The key is anything that implements IntoStableKey: a StableKey, &str, String, i64 or u64.

Implementing root target states

Root target states are not nested inside another target.

Life of a root target state

Understanding what happens at runtime:

1. Registration: You define a TargetHandler and call register_root_target_states_provider() from inside a component, on every run. CocoIndex returns a TargetStateProvider — a factory for creating target states associated with your handler.

2. Declaration: During execution, user code calls provider.target_state(key, spec) to create TargetState objects, then declare_target_state() to declare them. CocoIndex collects all declared target states.

3. Reconciliation: When the processing unit finishes, CocoIndex calls your handler’s reconcile() method for each target state. For declared target states, desired_target_state contains the spec; for previously declared but now missing states, desired_target_state is None (triggering cleanup). Your reconcile() compares the desired state with previous records and returns Some(TargetReconcileOutput { .. }) if an action is needed, or None if no changes are required.

4. Action Execution: CocoIndex batches actions by their TargetActionSink and executes them. The sink applies changes to the external system (database writes, file operations, API calls, etc.). A batch may combine the actions of several processing components that finish around the same time; if the sink fails, CocoIndex retries in smaller batches along component boundaries, so only the components whose actions actually fail are affected. The sink may therefore see an action from a failed batch again — keep actions idempotent.

5. Tracking Persistence: After successful execution, CocoIndex persists the new tracking records. On the next run, these become the prev_possible_records for change detection.

Step 1: Define your types

Start by defining the types for your provider:

use cocoindex::{
    Result, StableKey, TargetAction, TargetActionSink, TargetHandler, TargetReconcileOutput,
    declare_target_state, register_root_target_states_provider,
};
use serde::{Deserialize, Serialize};

// Key: `StableKey`, built from anything that implements `IntoStableKey`
// (`&str`, `String`, `i64`, `u64`, ...).

// Value: what the user declares. Here, a `String` row value.

// Tracking record: what to persist for change detection. Here, the value
// itself (`String`).

/// Action: one row mutation the sink applies. `value: None` is a delete.
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
struct RowAction {
    key: String,
    value: Option<String>,
}

Step 2: Implement the handler

/// A flat row handler with full CRUD + no-change detection (tracking record is
/// the row value itself).
#[derive(Clone)]
struct RowHandler {
    sink: TargetActionSink<RowAction>,
}

impl TargetHandler<String> for RowHandler {
    type TrackingRecord = String;
    type Action = RowAction;

    fn reconcile(
        &self,
        key: StableKey,
        desired: Option<String>,
        prev: Vec<String>,
        prev_may_be_missing: bool,
    ) -> Result<Option<TargetReconcileOutput<RowAction, String>>> {
        let k = key_str(&key);
        match desired {
            Some(value) => {
                // Skip if unchanged
                let unchanged =
                    !prev_may_be_missing && !prev.is_empty() && prev.iter().all(|p| *p == value);
                if unchanged {
                    return Ok(None);
                }
                let row = RowAction {
                    key: k,
                    value: Some(value.clone()),
                };
                let action = if prev.is_empty() {
                    TargetAction::Create(row)
                } else {
                    TargetAction::Update(row)
                };
                Ok(Some(TargetReconcileOutput {
                    action,
                    sink: self.sink.clone(),
                    tracking_record: Some(value),
                    child_invalidation: None,
                }))
            }
            // Handle deletion
            None => {
                if prev.is_empty() && !prev_may_be_missing {
                    return Ok(None); // Nothing to delete
                }
                Ok(Some(TargetReconcileOutput {
                    action: TargetAction::Delete(RowAction {
                        key: k,
                        value: None,
                    }),
                    sink: self.sink.clone(),
                    tracking_record: None,
                    child_invalidation: None,
                }))
            }
        }
    }
}

key_str is a helper of the test that renders the StableKey as a string; sink is the recording_sink above.

Step 3: Register the provider

For root-level target states (not nested within another target), register a provider:

let provider = register_root_target_states_provider(
    &ctx,
    "mycompany.io/mydb/row", // Unique provider name
    RowHandler {
        sink: recording_sink(log),
    },
)?;
for (k, v) in rows {
    declare_target_state(&ctx, provider.target_state(k, v.to_string()))?;
}

The provider is registered for the current run, from inside a component, so register it on every run. Once a run stops registering it, the engine cannot reach its handler, and the target states declared through it are left in place; App::drop_state leaves them in place too.

Step 4: Create user-facing APIs

Wrap the provider in a user-friendly API: a struct that holds the TargetStateProvider<V> and whose methods take &Ctx.

pub struct IndexTarget {
    index_name: Arc<str>,
    documents: TargetStateProvider<Document>,
}

impl IndexTarget {
    /// Declare a document to upsert into the index.
    pub fn declare_document(&self, ctx: &Ctx, document: Document) -> Result<()> {
        // ... validate the document ...
        let key = StableKey::Str(Arc::from(document.id.as_str()));
        declare_target_state(ctx, self.documents.target_state(key, document))
    }
}

Implementing container targets

Container targets (directories, tables) have children (files, rows). A child is a non-root target state.

Non-root target states

For targets nested inside another target (e.g., files inside a directory), the lifecycle is similar to root targets but how you get the provider is different.

For root targets, you call register_root_target_states_provider() and immediately get a provider with your handler. For non-root targets, the handler comes from the child_handler factory of the parent’s sink, which CocoIndex calls with the parent’s action:

1. Declaration: Call declare_target_state_with_child(ctx, parent_ts) — returns an unresolved child provider immediately. Or call mount_target(ctx, parent_ts).await, which declares and commits the parent in a foreground child component and returns a child provider that is already resolved.
2. Resolution: When the parent reconciles, CocoIndex calls the sink’s child_handler factory with the parent’s action and resolves the child provider with the handler it returns. It does this before the sink applies the action, and in a preview (app.preview(..)) in place of applying it — so the factory must build the handler from the action alone, with no side effects.
3. Usage: The child provider can now create child target states, which follow the same reconciliation → execution → tracking flow as root targets.

Some containers only get their identity when created: the external system assigns it (a Notion database or a Google Drive folder gets a server-assigned id). The factory cannot know that id, so let the handler look it up when it applies: the container sink records key → id in a connector-level map when it creates (or finds) the container, and the child handler’s sink reads the id from that map by the container’s key. Containers reconcile and apply on every run, before their children apply, so the map is filled when the children need it; a preview never applies the children, so it never needs the id.

Child invalidation

When a container target undergoes certain changes, the child target states may be affected. The child_invalidation field in TargetReconcileOutput lets you signal this to CocoIndex:

• TargetChildInvalidation::Destructive — The container change destroys all existing children (e.g., a primary key change that requires dropping and recreating a table). CocoIndex will ignore all previous tracking records for children under this container and treat them as new.

• TargetChildInvalidation::Lossy — The container change may cause data loss for existing children (e.g., a schema change that removes columns). CocoIndex will force an upsert for all children by setting prev_may_be_missing to true, even if their data appears unchanged.

• None (default) — No impact on children. Normal change detection applies.

Set child_invalidation in the parent handler’s reconcile() method when you detect that the container itself has changed in a way that affects its children:

#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
struct TableSpec {
    generation: String,
}

#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
struct TableAction {
    name: String,
    drop: bool,
}

/// A container ("table") handler whose sink produces a child row handler per
/// declared table, and whose child invalidation is configurable.
struct TableHandler {
    sink: TargetActionSink<TableAction>,
    invalidation: Option<TargetChildInvalidation>,
}

impl TargetHandler<TableSpec> for TableHandler {
    type TrackingRecord = TableSpec;
    type Action = TableAction;

    fn reconcile(
        &self,
        key: StableKey,
        desired: Option<TableSpec>,
        prev: Vec<TableSpec>,
        _prev_may_be_missing: bool,
    ) -> Result<Option<TargetReconcileOutput<TableAction, TableSpec>>> {
        let name = key_str(&key);
        match desired {
            // Always emit an output when the table is declared, so the sink
            // builds the child handler.
            Some(spec) => {
                let changed = !prev.is_empty() && !prev.contains(&spec);
                let action = if prev.is_empty() {
                    TargetAction::Create(TableAction { name, drop: false })
                } else {
                    TargetAction::Update(TableAction { name, drop: false })
                };
                Ok(Some(TargetReconcileOutput {
                    action,
                    sink: self.sink.clone(),
                    tracking_record: Some(spec),
                    child_invalidation: changed.then_some(self.invalidation).flatten(),
                }))
            }
            None => Ok(Some(TargetReconcileOutput {
                action: TargetAction::Delete(TableAction { name, drop: true }),
                sink: self.sink.clone(),
                tracking_record: None,
                child_invalidation: Some(TargetChildInvalidation::Destructive),
            })),
        }
    }
}

Deleting a container

When a container target state is no longer declared, CocoIndex runs the container handler’s non-existence action and does not reconcile the children individually. The container action is therefore responsible for its children, in one of two shapes:

• Destroy — for a container CocoIndex owns, remove it together with its contents (DROP TABLE, removing a directory tree, dropping an index, …). The TableHandler above returns a Delete action with drop: true for this.
• Abandon — for a user-managed container (a message topic, a table the user created), leave the contents untouched. Once the container is no longer declared, CocoIndex has released its claim on it and must not write to it again: another writer may now own those keys, writing to a topic the user has since deleted would recreate it on brokers that auto-create topics, and the deletion policy captured from the old declaration (such as a deletion_value_fn closure) no longer exists.

The Valkey connector (rust/sdk/cocoindex/src/connectors/valkey.rs) implements both: it drops a system-managed index and leaves a user-managed one alone (ManagedBy::User, see statediff). Document on your connector page which shape it uses.

The user-facing contract is in How CocoIndex syncs target states.

Step 1: Define parent and child handlers

The parent handler reconciles the container itself. The child handler reconciles entries within it:

The parent is the TableHandler above, a TargetHandler<TableSpec>. The child is the RowHandler from Step 2 of root target states, a TargetHandler<String>: a child handler is a regular TargetHandler for the child value type. The child value type is not part of the parent’s trait implementation; the parent’s sink names it when it wraps the child handler in a ChildTargetDef.

Step 2: Build child handlers with a child_handler factory

The parent’s sink is built with TargetActionSink::from_async_fn_with_child_handler(apply, child_handler). The callback creates and drops containers; the factory returns the handler for a container’s child target states: it takes &A and returns a ChildTargetDef, which wraps a typed child handler:

/// Build a container sink whose child handler factory gives each
/// created/updated table a fresh child row handler (recording into `row_log`).
fn table_sink(table_log: Log, row_log: Log) -> TargetActionSink<TableAction> {
    TargetActionSink::from_async_fn_with_child_handler(
        move |actions: Vec<TargetAction<TableAction>>| {
            let table_log = table_log.clone();
            async move {
                for action in actions {
                    log_table_action(&table_log, action);
                }
                Ok(())
            }
        },
        move |_: &TableAction| {
            Ok(ChildTargetDef::new::<String, _>(RowHandler {
                sink: recording_sink(row_log.clone()),
            }))
        },
    )
}

fn log_table_action(table_log: &Log, action: TargetAction<TableAction>) {
    let entry = match action {
        TargetAction::Create(t) | TargetAction::Update(t) => format!("ensure {}", t.name),
        TargetAction::Delete(t) => format!("drop {}", t.name),
    };
    table_log.lock().unwrap().push(entry);
}

CocoIndex calls the factory once per action whose target state was declared with a child (a created or updated container, never a delete) before the callback applies the batch, and in a preview in place of the callback. So the factory is sync, free of side effects, and builds the handler from the action alone (plus what the closure captured when the sink was built). Actions without a child provider (orphan deletes, or leaf actions when one sink is shared between a container and its children) don’t reach the factory.

The container handler’s reconcile() returns an action on every run while the container is declared, even when nothing about it changed: the factory builds the child handler from that action. Preview lists this action too: equality of arbitrary tracking records does not guarantee that applying the action has no external effect.

Step 3: Create user-facing API

The user-facing API uses declare_target_state_with_child() or mount_target() and exposes methods for declaring children:

let table_provider = register_root_target_states_provider(
    &ctx,
    "test/table",
    TableHandler {
        sink: make_sink(table_log, row_log),
        invalidation,
    },
)?;
// Commit the table in a foreground child component; the child provider is
// ready at once.
let child: TargetStateProvider<String> = mount_target(
    &ctx,
    table_provider.target_state(
        "docs",
        TableSpec {
            generation: generation.to_string(),
        },
    ),
)
.await?;
for (k, v) in rows {
    declare_target_state(&ctx, child.target_state(k, v.to_string()))?;
}

A connector wraps the child provider in a user-facing struct, as Valkey’s IndexTarget above does: declare_index_target returns one built from declare_target_state_with_child (its documents resolve when the component commits), and mount_index_target one built from mount_target (ready at once).

Migrating from child slots and child definitions

Two older container sink constructors are deprecated: TargetActionSink::from_async_fn_with_children, whose closure receives each action with an Option<ChildSlot> and calls slot.fulfill(ChildTargetDef::new(..)) on each, and TargetActionSink::from_async_fn_with_child_defs, whose closure returns a Vec<Option<ChildTargetDef>> index-aligned with the actions. A sink built with from_async_fn_with_children cannot be previewed.

To migrate, move the construction of the ChildTargetDef into a closure Fn(&A) -> Result<ChildTargetDef> that builds it from the action alone, drop the slots (or the returned definitions) from the apply closure so it returns Ok(()), and build the sink with from_async_fn_with_child_handler(apply, child_handler), as table_sink above does.

Implementing attachment providers

Attachment providers let a child handler expose auxiliary target states alongside its regular children. For example, a PostgreSQL table handler manages rows as regular children, but can also manage vector indexes and SQL command attachments as separate attachment types — each tracked independently.

When to use attachments

Use attachments when a target has auxiliary state beyond its primary children — indexes, triggers, materialized views, or any side-resource that should be managed alongside the main data. Attachments use symbol keys as namespace separators so they never conflict with regular child keys.

Target state path hierarchy

Attachments create additional levels in the target state path using symbol keys (denoted with @ prefix in documentation):

@my_connector/table                    [level 1 — root provider]
  (db_key, schema, table_name)         [level 2 — table state]
    @vector_index                      [level 3 — attachment namespace (symbol key)]
      index_name_1                     [level 4 — attachment instance]
      index_name_2                     [level 4]
    @sql_command_attachment             [level 3 — another attachment namespace]
      cmd_name_1                       [level 4]
    row_pk_1                           [level 3 — regular child (row)]
    row_pk_2                           [level 3]

The symbol keys (@vector_index, @sql_command_attachment) are path namespaces — not target states themselves. They separate attachment instances from regular children at the same level.

How it works

1. Handler implements attachments(): The child handler (e.g., row handler) returns a Vec of (name, ChildTargetDef) pairs for all supported attachment types and their handlers.
2. Engine eagerly registers attachment providers: When the child provider is resolved with the handler from the child_handler factory, the engine calls attachments() and registers all returned types. This ensures orphaned attachments can be cleaned up even when not declared in the current run.
3. User code calls provider.attachment(): On a resolved child provider, this retrieves the cached attachment sub-provider.
4. Target states declared under the attachment provider are tracked independently from regular children.

Step 1: Implement the attachment handler

An attachment handler is just a regular TargetHandler — it implements reconcile() and has an action sink. The example below reuses the RowHandler from Step 2 of root target states as the attachment handler, with its own recording sink.

Step 2: Add attachments() to the parent handler

The parent handler (which manages regular children) returns all supported attachment types:

/// A main handler (CRUD like `RowHandler`) that also exposes an "index"
/// attachment handler.
struct MainHandler {
    sink: TargetActionSink<RowAction>,
    att_log: Log,
}

impl TargetHandler<String> for MainHandler {
    type TrackingRecord = String;
    type Action = RowAction;

    fn reconcile(
        &self,
        key: StableKey,
        desired: Option<String>,
        prev: Vec<String>,
        prev_may_be_missing: bool,
    ) -> Result<Option<TargetReconcileOutput<RowAction, String>>> {
        RowHandler {
            sink: self.sink.clone(),
        }
        .reconcile(key, desired, prev, prev_may_be_missing)
    }

    fn attachments(&self) -> Result<Vec<(String, ChildTargetDef)>> {
        Ok(vec![(
            "index".to_string(),
            ChildTargetDef::new::<String, _>(RowHandler {
                sink: recording_sink(self.att_log.clone()),
            }),
        )])
    }
}

Step 3: Expose attachment APIs on the user-facing target

The user-facing target struct calls provider.attachment() to get the attachment sub-provider, then declares target states on it:

let provider = register_root_target_states_provider(
    &ctx,
    "test/main",
    MainHandler {
        sink: recording_sink(main_log),
        att_log,
    },
)?;
declare_target_state(&ctx, provider.target_state("row", "v1".to_string()))?;
if declare_attachment {
    let att: TargetStateProvider<String> = provider.attachment(&ctx, "index")?;
    declare_target_state(&ctx, att.target_state("idx", "spec1".to_string()))?;
}

A run that stops declaring the attachment deletes it, even though it was not declared in that run: the engine registered the attachment provider eagerly.

Connector implementation guidelines

Use ContextKey for external resource identity

When a target connector manages state in an external resource (a database, a search index, an object store, etc.), use a ContextKey string as part of the target state key — not connection parameters like host, port, or credentials.

Target state keys must be stable across runs for correct reconciliation. CocoIndex uses keys to match current declarations with previously tracked states. If the key is stable, previously tracked states are associated with the current target, so CocoIndex can correctly reconcile — e.g., deleting rows that are no longer declared. If the key changes (because a connection parameter changed), CocoIndex cannot associate previous tracked states with the current target, and treats the target as being in a cleared state — losing the ability to clean up old data.

ContextKey solves this by providing a user-defined stable logical name (e.g., "my_pg") that is decoupled from transient connection details:

Put the key’s name (ContextKey::name(), the string "my_pg") in the target state key. The Valkey connector’s index key holds the database’s ContextKey name plus the index name:

/// An index target state's key: the database's `ContextKey` name plus the
/// index name.
struct IndexKey {
    db_key: String,
    index_name: String,
}

impl IndexKey {
    fn stable_key(self) -> StableKey {
        StableKey::Array(Arc::from([
            StableKey::Str(Arc::from(self.db_key)),
            StableKey::Str(Arc::from(self.index_name)),
        ]))
    }
}

This way, changing a password, switching replicas, or rotating credentials won’t invalidate tracked states or break reconciliation.

A sink built with TargetActionSink::from_async_fn receives only the actions, not the environment’s context values. So resolve the connection with ctx.get_key(&key) in the run that registers the provider, and move it into the handler and its sink. The run that registers the provider also produces the delete actions for target states it no longer declares, so the handler has its connection for them too. Pass the typed connection down to child handlers directly, not the key name.

Idempotent actions

Actions should be idempotent — applying the same action multiple times should have the same effect as applying it once: for example CREATE TABLE IF NOT EXISTS, DROP INDEX IF EXISTS and INSERT ... ON CONFLICT DO UPDATE rather than a plain CREATE TABLE or INSERT, which fail when the object already exists. A sink may see an action again after a failed batch is retried.

Handle multiple previous states

Due to interrupted updates, prev_possible_records may contain multiple records. Design your reconciliation logic to handle this:

// Check if ALL previous states match (conservative approach)
let unchanged =
    !prev_may_be_missing && !prev.is_empty() && prev.iter().all(|p| *p == value);
if unchanged {
    return Ok(None); // Safe to skip
}

cocoindex::statediff packages this decision. TrackingRecordTransition { desired, prev, prev_may_be_missing } holds the inputs of one reconcile call, and diff(Some(&transition)) returns None (skip) or a DiffAction: Insert, Upsert, Replace or Delete. Replace means a previous record differs from the desired one; Upsert means the records match but the target may be missing. For a target the user may own, wrap records in MutualTrackingRecord::new(record, managed_by) with ManagedBy::System or ManagedBy::User; resolve_system_transition(desired, prev, prev_may_be_missing) returns None for a user-managed target, so it is left alone.

For a container with keyed parts (a table’s primary key and its columns), CompositeTrackingRecord and diff_composite return the main action plus one transition per part.

Efficient change detection

Choose tracking records that enable efficient change detection without storing full content:

Shared action sinks

If all instances of a handler use the same action logic, create a shared sink. Build it once, store it in the handler, and return self.sink.clone() from each reconcile, as RowHandler above does: a clone keeps the sink’s identity, so the engine applies the actions of every target state that returned it in one call.

Complete examples

• rust/sdk/cocoindex/tests/target_state.rs — the handler, sink and provider snippets on this page, runnable without an external service: flat targets, containers with mount_target, child invalidation, attachments, and the deprecated sink shapes
• rust/sdk/cocoindex/src/resources/fs.rs — DirTarget, file system targets
• rust/sdk/cocoindex/src/connectors/valkey.rs — a Valkey index (container) and its documents, with statediff and ManagedBy