Sharing resources via context

Use context_key!, EnvironmentBuilder::provide_key() and ctx.get_key() to share connections, models, and config across processing components. Covers change detection on context values and the query path outside an update.

Version
v 1.0.14
Last reviewed
Nov 4, 2026
ExperimentalThe Rust SDK is experimental. It is outside CocoIndex's compatibility promise: any release, including a minor one, may change or remove its API. It is not yet published on crates.io.

CocoIndex provides a context mechanism for sharing resources across your pipeline. This is useful for database connections, API clients, configuration objects, or any resource that multiple processing components need to access.

ContextKey

A ContextKey<T> is a typed key that identifies a resource. Define keys at module level:

use cocoindex::connectors::postgres;
use cocoindex::ops::sentence_transformers::SentenceTransformerEmbedder;

// Database connection — no change detection
cocoindex::context_key!(static DB: postgres::Database);

// Embedding model — with change detection
cocoindex::context_key!(
    static EMBEDDER: SentenceTransformerEmbedder,
    detect_change
);

// An explicit persisted name, when it must differ from the identifier
cocoindex::context_key!(
    static LEGACY_DB: postgres::Database,
    key = "app_database"
);

context_key! declares a static ContextKey<T>. The Rust identifier is the key’s name unless you pass key = "...". When both options are present, write key = "..." before detect_change.

The type parameter (postgres::Database, SentenceTransformerEmbedder) is the type of the value: providing a value of another type does not compile, and retrieving the value returns a &T.

Change detection

By default, context keys have change detection disabled — changing the provided value between runs does not invalidate memoized functions that read it with ctx.get_key(). To opt in to change detection, add detect_change to the context_key! declaration. When enabled, context changes are their own category — tracked by ctx.get_key() at the call site, independent of #[cocoindex::function]. When a fingerprint changes, any memoized function whose execution read that key with ctx.get_key() is invalidated.

Use detect_change for resources that affect computation results — models, configuration objects, etc. This ensures memoized functions re-execute when those values change. Resources that don’t affect computation results (database connections, loggers, debug flags, monitoring clients) can use the default, without detect_change.

A detect_change key fingerprints its value through the MemoInput trait. Derive MemoInput for your own configuration types; each field is handled recursively, so the type needs no Serialize:

#[derive(cocoindex::MemoInput)]
struct AppConfig {
    model: String,
    debug: bool,
}

cocoindex::context_key!(static CONFIG: AppConfig, detect_change);
Tip

Change detection is transitive: if function foo (memoized) calls function bar, and bar reads a change-detected key with ctx.get_key(&KEY), then foo’s memo is also invalidated when the context value changes.

ContextKey as stable identity

Beyond sharing resources, a ContextKey also serves as the stable identity of the resource it points to. When you anchor sources or targets to a ContextKey, CocoIndex treats the key itself (not the underlying value) as the identifier across runs.

This has two consequences:

  1. The underlying value can change without losing tracked state. Rotating credentials, moving a database, or relocating a directory won’t invalidate memoization or managed state, as long as the same ContextKey is used. A key names a logical location: when you point it at a new physical location, CocoIndex assumes the data moved with it. It doesn’t recreate target states that are unchanged, and it doesn’t clean up the old location. If the data doesn’t move with the key, drop the app before you point the key elsewhere.

  2. Renaming a ContextKey is a breaking change. Two different keys are two different resources, even if they point to the same physical backend. Existing tracked state will be treated as orphaned. When migrating code, reuse the previous key name to preserve continuity.

    The same holds within one run: CocoIndex doesn’t detect two keys that reach one backend. If two keys point at the same database and you declare the same table through both, CocoIndex tracks two tables, and syncing one can overwrite or delete the other. Keeping the locations behind different keys apart is your responsibility.

Naming convention

Pick a ContextKey name that reflects the logical role of the resource, not its current address. The name is what CocoIndex persists.

  • Applications: use any descriptive name — e.g., "text_embedding_db", "docs_root".
  • Libraries: prefix with your package name and a / to avoid collisions with application keys or other libraries — e.g., "my_crate/db".

The persisted name is the identifier of the context_key! static by default, so a rename of the static is a rename of the key. Pass key = "..." to keep the old name, or to give a library key its prefix. A key name is unique within a process: constructing a second key with the same name panics.

Providing values

In your lifespan function, use builder.provide_key() to make resources available:

#[cocoindex::lifespan]
async fn lifespan(builder: EnvironmentBuilder) -> Result<EnvironmentBuilder> {
    let db = postgres::Database::connect(&database_url()).await?;
    let pool = db.pool().clone();
    let embedder = load_embedder().await?;
    Ok(builder
        .db_path(PathBuf::from(env!("CARGO_MANIFEST_DIR")).join(".cocoindex_db"))
        .provide_key(&DB, db)
        .provide_key(&EMBEDDER, embedder)
        .on_shutdown(move || async move { pool.close().await }))
}

EnvironmentBuilder::provide_key(&KEY, value) provides a value under a key; on_shutdown registers its cleanup. The same calls work on Environment::builder() when your own code builds the environment.

The resource is available for the lifetime of the environment. When the environment shuts down (Environment::shutdown), it runs the on_shutdown hooks, the most recently registered first, and then closes its state database.

Retrieving values

In processing components, use ctx.get_key() to retrieve provided resources:

#[cocoindex::function]
async fn process_chunk(
    ctx: &Ctx,
    chunk: Chunk,
    text: &str,
    filename: &str,
    id_gen: &IdGenerator,
    table: &postgres::TableTarget,
) -> Result<()> {
    let chunk_text = chunk.text(text);
    let id = i64::try_from(id_gen.next_id(ctx, chunk_text).await?)
        .map_err(|_| Error::engine("generated id does not fit in BIGINT"))?;
    table.declare_row(
        ctx,
        &DocEmbedding {
            id,
            filename: filename.to_string(),
            chunk_start: chunk.start.char_offset as i32,
            chunk_end: chunk.end.char_offset as i32,
            text: chunk_text.to_string(),
            // Retrieve the embedder from context
            embedding: ctx.get_key(&EMBEDDER)?.embed(ctx, chunk_text).await?,
        },
    )
}

ctx.get_key(&KEY)? returns a reference to the value, or an error when the key was not provided.

Connectors also accept ContextKeys directly — for example, postgres::mount_table_target() takes a &ContextKey<postgres::Database> and resolves the connection internally:

async fn index(ctx: Ctx, sourcedir: PathBuf) -> Result<()> {
    // DB is resolved internally by the connector
    let target_table = postgres::mount_table_target(
        &ctx,
        &DB,
        TABLE,
        postgres::TableSchema::from_row::<DocEmbedding>(["id"])?
            .with_vector_schema("embedding", ctx.get_key(&EMBEDDER)?)
            .await?,
        Some(PG_SCHEMA),
    )
    .await?;
    // ... mount processing components ...
    Ok(())
}

Every connector target takes its connection as a &ContextKey<T>, so the key is the target’s stable identity.

Complete example

Here’s a complete pipeline that uses context to share a database connection and an embedding model across processing components:

use std::path::PathBuf;
use std::time::Duration;

use cocoindex::connectors::postgres;
use cocoindex::ops::sentence_transformers::SentenceTransformerEmbedder;
use cocoindex::ops::text::{RecursiveChunkConfig, RecursiveSplitter};
use cocoindex::prelude::*;

const EMBED_MODEL: &str = "sentence-transformers/all-MiniLM-L6-v2";
const PG_SCHEMA: &str = "coco_examples";
const TABLE: &str = "doc_embeddings";
const CHUNK_SIZE: usize = 2000;
const CHUNK_OVERLAP: usize = 500;

// 1. Define context keys at module level
cocoindex::context_key!(static DB: postgres::Database);
cocoindex::context_key!(
    static EMBEDDER: SentenceTransformerEmbedder,
    detect_change
);

// 3. Derive the table schema from the row type; the vector dimension comes from EMBEDDER
#[derive(Clone, Serialize, Deserialize, SchemaFields)]
struct DocEmbedding {
    id: i64,
    filename: String,
    chunk_start: i32,
    chunk_end: i32,
    text: String,
    #[coco(vector)]
    embedding: Vec<f32>,
}

// 4. Retrieve values in processing functions
#[cocoindex::function]
async fn process_chunk(
    ctx: &Ctx,
    chunk: Chunk,
    text: &str,
    filename: &str,
    id_gen: &IdGenerator,
    table: &postgres::TableTarget,
) -> Result<()> {
    let chunk_text = chunk.text(text);
    let id = i64::try_from(id_gen.next_id(ctx, chunk_text).await?)
        .map_err(|_| Error::engine("generated id does not fit in BIGINT"))?;
    table.declare_row(
        ctx,
        &DocEmbedding {
            id,
            filename: filename.to_string(),
            chunk_start: chunk.start.char_offset as i32,
            chunk_end: chunk.end.char_offset as i32,
            text: chunk_text.to_string(),
            embedding: ctx.get_key(&EMBEDDER)?.embed(ctx, chunk_text).await?,
        },
    )
}

#[cocoindex::function(memo, memo_key(table = skip))]
async fn process_file(ctx: &Ctx, file: &FileEntry, table: &postgres::TableTarget) -> Result<()> {
    let text = file.read_text().await?;
    let chunks = RecursiveSplitter::new()?.split_with(
        &text,
        RecursiveChunkConfig {
            chunk_size: CHUNK_SIZE,
            min_chunk_size: None,
            chunk_overlap: Some(CHUNK_OVERLAP),
            language: Some("markdown".to_string()),
        },
    );
    let filename = file.key();
    let id_gen = IdGenerator::new();
    ctx.map(chunks, async |chunk| {
        process_chunk(ctx, chunk, &text, &filename, &id_gen, &table).await
    })
    .await?;
    Ok(())
}

// 2. Provide values in the lifespan
#[cocoindex::lifespan]
async fn lifespan(builder: EnvironmentBuilder) -> Result<EnvironmentBuilder> {
    let db = postgres::Database::connect(&database_url()).await?;
    let pool = db.pool().clone();
    let embedder = load_embedder().await?;
    Ok(builder
        .db_path(PathBuf::from(env!("CARGO_MANIFEST_DIR")).join(".cocoindex_db"))
        .provide_key(&DB, db)
        .provide_key(&EMBEDDER, embedder)
        .on_shutdown(move || async move { pool.close().await }))
}

// 5. DB used directly by the connector (resolved internally)
#[cocoindex::app(name = "TextEmbeddingRust")]
async fn app_main(ctx: Ctx) -> Result<()> {
    index(ctx, source_dir()).await
}

async fn index(ctx: Ctx, sourcedir: PathBuf) -> Result<()> {
    let target_table = postgres::mount_table_target(
        &ctx,
        &DB,
        TABLE,
        postgres::TableSchema::from_row::<DocEmbedding>(["id"])?
            .with_vector_schema("embedding", ctx.get_key(&EMBEDDER)?)
            .await?,
        Some(PG_SCHEMA),
    )
    .await?;
    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
}

fn database_url() -> String {
    std::env::var("POSTGRES_URL")
        .unwrap_or_else(|_| "postgres://cocoindex:cocoindex@localhost/cocoindex".to_string())
}

fn source_dir() -> PathBuf {
    PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("markdown_files")
}

async fn load_embedder() -> Result<SentenceTransformerEmbedder> {
    SentenceTransformerEmbedder::load(EMBED_MODEL).await
}

The full program, with main and a pgvector query path, is examples/rust/text_embedding/src/main.rs.

Accessing context outside processing components

Context values are read through a Ctx, and a Ctx exists only inside an update. Environment has no method that returns a provided value. For query or serving logic that runs outside an update, keep the resource yourself: the query branch of examples/rust/text_embedding/src/main.rs connects its own pool and loads its own embedder.

PreviousFunctions
CocoIndex DocsEdit this pageReport issue