The processing component

Processing components are the units of incremental execution and the sync boundaries for target states. Covers mounting APIs (mount, use_mount, mount_each, mount_target), component paths, and granularity tradeoffs.

Version

v 1.0.62

Last reviewed

Jul 4, 2026

Most apps process many independent source items — files, rows, or entities. A Processing Component is the unit of execution for one: it runs that item’s transformation logic and declares the set of target states produced for it.

Component path

A component path is the stable identifier for a processing component across runs (think of it like a path in a tree). CocoIndex uses it to match a component to its previous run, detect what changed for that item, and sync that component’s target states as a unit when it finishes. This sync happens per component; CocoIndex does not wait for other components in the same app to complete.

Component paths are hierarchical and form a tree structure. You specify child paths using coco.component_subpath() with stable identifiers like string literals, file names, row keys, or entity IDs:

coco.component_subpath(filename)           # e.g., coco.component_subpath("hello.pdf")
coco.component_subpath("user", user_id)    # e.g., coco.component_subpath("user", 12345)

Choose paths that are stable for the “same” item (e.g., file path, primary key). If an item disappears and its path is no longer present, CocoIndex cleans up the target states owned by that path (and its sub-paths).

Here’s an example component path tree for a pipeline that processes files:

(root)                         ← app_main component
└── process_file
    ├── "hello.pdf"            ← process_file component
    └── "world.pdf"            ← process_file component

The tree is populated dynamically as the app runs — each mount() / mount_each() call adds a subpath.

See StableKey in the SDK Overview for details on what values can be used in component paths.

Mount

Mounting is how you declare (instantiate) a processing component within an app at a specific path, so CocoIndex knows that component exists, should run, and owns a set of target states — and can match it against its previous run to sync only what changed.

CocoIndex provides two core mounting APIs:

• mount() — sets up a processing component in a child path without depending on data from it. This allows the component to refresh independently in live mode.
• use_mount() — returns a value from the component’s execution to the caller. The component at that path cannot refresh independently without re-executing the caller.

Which one you reach for comes down to a single question: does the caller need a value back? use_mount() consumes the child’s return value, which couples the two — the call blocks until the child finishes and commits its target states, and the child can’t refresh on its own without re-running the caller. mount() takes nothing back and returns as soon as the child is scheduled, so the child runs on its own and can refresh independently in live mode when its inputs change.

And two sugar APIs that simplify common patterns:

• mount_each() — mounts one component per item in a keyed iterable
• mount_target() — mounts a target without an explicit subpath

CocoIndex Plus adds a third ownership mode for components shared across parents:

• mount_shared() / use_mount_shared() — mount a component by a shared key instead of a child path, so several parents can share one instance. See Shared components (Plus).

See also map() for a utility API that operates within a component without creating new ones.

Automatic subpath derivation

mount(), use_mount(), and mount_each() all accept an optional ComponentSubpath as their first argument. When omitted, the subpath is auto-derived from the function name using Symbol(fn.__name__).

# These are equivalent:
await coco.mount(process_file, file, target)
await coco.mount(coco.component_subpath(coco.Symbol("process_file")), process_file, file, target)

This means the component path for a process_file function is parent / Symbol("process_file"). The function must have a __name__ attribute; if it doesn’t (e.g., a lambda), provide an explicit subpath.

Since sibling component paths must not collide, you need an explicit subpath when:

• The same function is mounted more than once — auto-derived paths would be identical, so each call needs a distinct path (e.g., coco.component_subpath("session", youtube_id))
• Different functions happen to share a __name__ — rare, but possible with wrappers or closures
• You want a specific path name — different from the function name

mount()

Use mount() when you don’t need a return value from the processing component. It schedules the processing component to run and returns a handle:

handle = await coco.mount(process_file, file, target)

With an explicit subpath, for example when mounting multiple components of the same function:

handle = await coco.mount(
    coco.component_subpath("process", filename),
    process_file,
    file,
    target,
)

The handle provides a method you can call if you need to wait until the processing component is fully ready — meaning all its target states have been applied to external systems and all components in its sub-paths are ready:

await handle.ready()

You usually only need to call ready() when you have logic that depends on the processing component’s target states being applied — for example, querying the latest data from a target table after syncing it.

mount() also accepts LiveComponent classes — components that process continuously and react to changes incrementally instead of rescanning everything. See Live Components for details.

use_mount()

Use use_mount() when you need the processing component’s return value. It mounts the component, waits until it’s ready, and returns the value directly:

table = await coco.use_mount(setup_table, table_name="docs")

With an explicit subpath:

table = await coco.use_mount(
    coco.component_subpath("setup"),
    setup_table,
    table_name="docs",
)

A common use of use_mount() is to obtain a target after its container target state is applied.

mount_each()

mount_each() mounts one processing component per item in a keyed iterable.

files = localfs.walk_dir(sourcedir, path_matcher=PatternFilePathMatcher(included_patterns=["**/*.md"]))
await coco.mount_each(process_file, files.items(), target)

Each item in the iterable is a (key, value) tuple. The value is passed as the first argument to the function, and any additional arguments are passed through. Items are mounted under an auto-derived subpath (Symbol(fn.__name__)), so the component path for each item is parent / Symbol("process_file") / key.

In the static case this is just the keyed-loop form of mount() — the same per-file fan-out shown in Core Concepts. The snippet above is equivalent to looping over the items and calling mount(coco.component_subpath(coco.Symbol("process_file"), key), process_file, value, target) for each one. The per-item key is what keeps each component’s path stable across runs, so the engine matches, updates, and cleans up each item individually.

Like mount(), mount_each() does not carry return values from the mounted functions. It returns a single handle covering all mounted components, which you can capture if you need to wait for readiness:

handle = await coco.mount_each(process_file, files.items(), target)
await handle.ready()  # waits until every per-item component is ready

If you want per-item results without creating components, use map().

You can provide an explicit subpath as the first argument:

await coco.mount_each(coco.component_subpath("files"), process_file, files.items(), target)

Source connectors provide an items() method that returns (StableKey, T) pairs. For example, localfs.walk_dir(...).items() yields (relative_path, File) tuples.

It also does one thing a hand-written loop can’t: when a source connector supports live watching, its items() returns a LiveMapView or LiveMapFeed instead of a plain iterable, and mount_each() detects this and automatically adds, updates, and removes per-item components as the source changes — no changes to mount_each() itself are needed. See Live Mode.

mount_target()

mount_target() mounts a target without requiring an explicit subpath.

from cocoindex.connectors import localfs

dir_target = await coco.mount_target(localfs.dir_target(outdir))

The component path is derived automatically from the target’s globally unique key — you don’t need to create a component_subpath for it. This is sugar over calling use_mount() with a target declaration function.

Connectors also provide convenience methods that wrap mount_target():

# Equivalent to the above
dir_target = await localfs.mount_dir_target(outdir)

# PostgreSQL example
table = await postgres.mount_table_target(
    PG_DB,
    table_name="doc_embeddings",
    table_schema=await postgres.TableSchema.from_class(DocEmbedding, primary_key=["id"]),
)

Shared components (Plus)

All the APIs above create components with exclusive ownership: a component lives at one path under one parent, and when that parent stops mounting it, it (and its target states) are cleaned up.

Sometimes several independent parents need the same component — an entity mentioned by many documents, a shared lookup table, a shared connection-backed resource. Mounting it separately under each parent would build it repeatedly and tie its lifetime to a single parent. Shared components (CocoIndex Plus) give it Arc/shared_ptr-style shared ownership instead: the component is identified by a shared key that is unique within the app, several parents can mount it, and it is garbage-collected only once no parent still mounts it.

• use_mount_shared(shared_key, fn, *args, **kwargs) — mount the shared component and return its value (like use_mount()).
• mount_shared(shared_key, fn, *args, **kwargs) — mount it in the background and return a handle (like mount()).

The first argument is the shared key (any stable key — a string, int, tuple, etc.), not a ComponentSubpath. The component lives in its own namespace keyed by that key, independent of which parent mounts it. A component that mounts a shared key is one of its owners — the word used below, since the shared component is not a child of any of them.

@coco.fn(memo=True)
def load_lookup_table(path: str) -> LookupTable:
    return LookupTable.load(path)

@coco.fn
async def process_file(file: FileLike) -> None:
    # Every file's component mounts the same shared "lookup" — built once,
    # shared by all, and dropped only when no file references it anymore.
    table = await coco.use_mount_shared("lookup", load_lookup_table, "data/lookup.csv")
    ...

Key points:

• Identified by key, shared by all. Every mount_shared("lookup", ...) call across the app resolves to the same component instance. Different owners mounting the same key share it. Concurrent mounts of a key that is not built yet build it once; the other mounts wait for that build and reuse its result.
• Lifetime is reference-counted. Each owner that mounts the key holds the shared component. An owner lets go when it runs again without mounting the key, or when it is removed itself; an owner that is a memo hit keeps its holds without running. When the last owner lets go, the shared component — its target states and everything mounted under it — is garbage-collected, just like a regular component whose parent stops mounting it.
• Make the processor memoizable. Because multiple owners (and repeat runs) mount the same key, mark fn with @coco.fn(memo=True) so redundant mounts short-circuit instead of re-running the work. Each mount is its own memo check: the first mount of a key in a run builds it, later ones hit.
• Latest wins. Pass the same fn and arguments for a given key. If owners disagree, the most recent mount’s result is what’s kept — so a use_mount_shared caller may receive a value produced by another owner’s mount. Keeping owners in agreement below shows how to make them agree.
• Owners inherit its dependencies. A memoized owner’s memo entry records the dependencies of everything it mounts, shared components included. A shared component that reads a change-detected context value makes every owner that mounts it depend on that value, so a change re-runs them all. Pass per-instance inputs as arguments instead, and let the owner that detects the change re-mount with new arguments.

Sub-pipelines under a shared component

A shared component is a full processing component: besides declaring target states, it can mount regular children with mount(), use_mount(), and mount_each(), and those children can have children of their own. This is the shape for “compute once per shared entity, with a pipeline of its own” — here, each entity’s reference page is fetched and split once, however many documents mention the entity:

@coco.fn(memo=True)
async def embed_section(section: Section, table: postgres.TableTarget[SectionEmbedding]) -> None:
    table.declare_row(row=SectionEmbedding(id=section.id, text=section.text, embedding=await embed(section.text)))

@coco.fn(memo=True)
async def index_entity(key: str, table: postgres.TableTarget[SectionEmbedding]) -> None:
    page = await fetch_entity_page(key)          # the expensive per-entity work, done once
    sections = split_into_sections(page)
    await coco.mount_each(coco.component_subpath("section"), embed_section, enumerate(sections), table)

@coco.fn(memo=True)
async def process_document(doc: Document, table: postgres.TableTarget[SectionEmbedding]) -> None:
    for key in doc.mentioned_entities:
        await coco.use_mount_shared(("entity", key), index_entity, key, table)

The shared component’s children live under its key, not under any owner: with the key ("entity", "zvec"), the shared component’s path is {["entity","zvec"]} and its sections are {["entity","zvec"]}/"section"/0, {["entity","zvec"]}/"section"/1, and so on, while each document stays at its own path such as /"doc"/"a.md". The children are owned by the shared component exclusively, so when the last document stops mentioning zvec, the whole {["entity","zvec"]} subtree — the entity’s own target states and every section’s rows — is collected together.

Readiness

use_mount_shared() returns once the shared component is ready — its target states applied and its sub-components ready — like use_mount(). mount_shared() returns a handle as soon as the run is scheduled, like mount(), and as with mount() the owner’s own commit waits: an owner’s target states are written only after everything it mounted this run, shared components included, is ready, and await handle.ready() on the owner covers them as well. So when a shared component declares a graph node and its owners declare edges pointing at that node, the node is written before any of the edges.

Paths, stats, and inspection

A shared component’s path is its key in braces, {key}, rather than a path under an owner, and its children extend it. The same form appears wherever CocoIndex prints component paths — cocoindex show, the stable_path of an exception context, log lines. In cocoindex show, the {key} namespace is listed after the regular tree:

Stable paths:
  /
  /"doc"
  /"doc"/"a.md"
  /"doc"/"b.md"
  {["entity","neo4j"]}
  {["entity","neo4j"]}/"section"
  {["entity","neo4j"]}/"section"/0
  {["entity","neo4j"]}/"section"/1

With --tree, each {key} namespace is a top-level node of its own, after the app’s tree:

Stable paths:
- / [component]
- doc
  - a.md [component]
  - b.md [component]
- {["entity","neo4j"]} [component]
  - section
    - 0 [component]
    - 1 [component]

In update stats, a shared component is counted under its processor’s name like any other component, once per mount: in a run where three documents mount ("entity", "neo4j"), the first mount counts as an add (or a reprocess) and the other two as unchanged. An owner inside a stats_group reports the shared components it mounts in that group.

Restrictions

• Not inside process_live(). Like mount(), use_mount(), and mount_each(), the shared mount APIs raise RuntimeError when called from a live component’s process_live(). Mount shared components from process() or from regular components; operator.update() has no shared form.
• Functions only. The processor must be a @coco.fn function. Passing a LiveComponent class raises TypeError.
• Plus only. On the open-source build, both calls raise RuntimeError: Shared components (mount_shared / use_mount_shared) require CocoIndex Plus.

Cost

Compared with a regular child, a shared component adds bookkeeping for the holds:

• Taking a hold. The first time an owner mounts a key it did not hold before, the hold is recorded in the state store at the mount call — before the shared component’s memo is consulted, so a new owner is registered even when the key is a memo hit. On a cold run the cost therefore scales with the number of owner–key pairs, not with the number of distinct keys.
• Keeping a hold. An owner that runs again and mounts the keys it already holds writes nothing for them: the holds are loaded along with its memoization state when its run starts. An owner that is a memo hit keeps its holds without running at all. The shared component’s own memo check still runs once per mount.
• Releasing a hold. When an owner runs without a key it held, the release is part of the owner’s commit. If that was the last hold, the garbage collection of the shared component — its target states and its subtree — follows.

When many owners share few keys, the per-hold bookkeeping, not the shared work, is the dominant state-store cost of the feature: the shared component builds once, but every owner–key pair is recorded.

Keeping owners in agreement

Pass the same processor and arguments from every owner. Owners that disagree make the shared component flip between versions as they run: the race is resolved by latest wins, but every flip re-runs the processor and rewrites its target states. Two shapes keep owners in agreement:

• Identity only. Pass only the key, and let the processor derive everything else from it. Owners agree trivially, and the processor’s memo hits on every mount after the first. If the processor derives properties from a context value with detect_change=True, remember the inheritance rule above: every owner re-runs when that value changes. That is fine for a small, rarely changing configuration and costly for many owners holding entities from a frequently updated catalog.
• Properties as arguments, from one snapshot. Read the source of the properties once per run into a context value that is not change-detected, and have every owner build the arguments from it. Owners agree within a run because they read the same snapshot. Put change detection where the property lives: a component that tracks the source row (memoized on the row’s version) re-runs when the row changes and re-mounts the shared component with new arguments, while the other owners stay memo hits and keep their holds. The knowledge graph example below uses this shape.

Shared entities across sources

The main use case for shared components is an entity that several sources know about: a product mentioned in support tickets, a code module named in docs and pull requests, a person who authored issues and releases. Each source processes its own items; whenever an item refers to an entity, its component mounts the shared entity component by the entity’s canonical key and declares only its own edge to it. The entity node then exists as long as any source still mentions it, and goes away with the last mention — no source owns it, and no source has to coordinate with another:

@dataclass(frozen=True)
class KnowledgeGraph:
    entities: neo4j.TableTarget[EntityNode]
    documents: neo4j.TableTarget[DocumentNode]
    mentions: neo4j.RelationTarget

@coco.fn(memo=True)
def entity_node(entity: Entity, graph: KnowledgeGraph) -> None:
    # The shared component {["entity", key]}: one node per entity, whoever mentions it.
    graph.entities.declare_record(row=EntityNode(key=entity.key, name=entity.name, kind=entity.kind))

@coco.fn
async def add_document(doc: Document, graph: KnowledgeGraph) -> None:
    # Runs inside each document's own component, whichever source produced the document.
    graph.documents.declare_record(row=DocumentNode(id=doc.id, source=doc.source, title=doc.title))
    entries = coco.use_context(CATALOG_ENTRIES)      # one snapshot per run; not change-detected
    for key in coco.use_context(RESOLVER).resolve(doc.text):
        await coco.use_mount_shared(("entity", key), entity_node, entries[key], graph)
        graph.mentions.declare_relation(from_id=doc.id, to_id=key)

A document owns its Document node and its MENTIONS edges outright; the Entity node it points at is held, not owned. Readiness guarantees the node is written before the edge, and the holds guarantee the node outlives every edge pointing at it.

The multi-source knowledge graph example is the worked version: a Postgres catalog, a folder of Markdown docs, and the issues, pull requests, and releases of a GitHub repo land in one Neo4j graph, with entity and person nodes as shared components.

Using component_subpath as a context manager

You can use component_subpath() as a context manager to create nested paths without repeating common prefixes:

with coco.component_subpath("process"):
    for f in files:
        await coco.mount(
            coco.component_subpath(str(f.relative_path)),
            process_file,
            f,
            target,
        )

This is equivalent to:

for f in files:
    await coco.mount(
        coco.component_subpath("process", str(f.relative_path)),
        process_file,
        f,
        target,
    )

Memoized components

memo=True on a function you mount memoizes the whole component — the processing-component-level memoization introduced in Core Concepts. On a cache hit the body doesn’t run, and everything the previous run produced is carried over as a unit: the component’s declared target states and its child mounts. Cleanup of no-longer-mounted children only applies when the body actually runs — a run that no longer mounts a child removes that child’s subtree, while a cache hit retains it.

@coco.fn(memo=True)
async def process_file(file: FileLike, table: postgres.TableTarget[DocEmbedding]) -> None:
    ...

await coco.mount_each(process_file, files.items(), table)

Skipping a stable item this way costs almost nothing in storage, provided you follow one principle:

• Declare large values as target states; don’t return them. Every memo entry — whether for a memoized component or for a memoized function called directly inside one (the transform-level memoization from Core Concepts) — must store its return value in order to serve a cache hit. A component that outputs via declare_row(...) and returns None keeps its entry tiny: the declared values live in the component’s own tracking, which sync needs regardless. A memoized function that returns a large value (an embedding vector, a parsed document) stores that value in every entry. A secondary difference: a directly-called memoized function’s entry also records a reference to each target state it declared, so a hit can keep those states alive — a per-state cost that mounting avoids. See the cost note in When to memoize.
• A memoized component may mount children. A memoized function called directly cannot mount — mounting is a side effect its cache can’t replay (memo=True constraints) — but the restriction doesn’t apply once the function is mounted as a component. A memoized parent that mounts children still invalidates when a child’s logic changes — the child’s fingerprint is part of the parent’s recorded logic deps.

Where to put memo=True follows the placement rule in When to memoize — memoize a boundary only if its key can hit while every enclosing memo boundary misses: the per-file pipeline above composes a file-level component memo with a chunk-level embedding memo, while a memoized component whose path is keyed by a content hash of its input makes any inner memo over the same content mostly redundant.

How target states sync

The component path tree determines ownership. When a component is no longer mounted at a path (e.g., a source file is deleted), CocoIndex automatically cleans up its target states — and recursively for all its sub-paths.

What happens when a component fails

CocoIndex processes each component in two phases: processing (running your function, declaring target states) and submit (writing changes to target backends).

Failure isolation

The framework’s general rule: a call raises on failure iff the failed work was on the critical path for the call to return.

• use_mount() — you’re awaiting the child’s value, so the child succeeding is on the critical path. The child’s exception propagates to the parent.
• mount() and mount_each() — these return as soon as the work is scheduled (you get a handle back). The child’s execution runs in the background, off your critical path. A failure in one child does not affect the parent or siblings — by default the exception is logged and other components continue. One bad file shouldn’t take down the entire pipeline.

To react to background failures, you can:

• Install exception handlers — global or scoped — to send alerts, record metrics, or implement custom logic. A handler that raises propagates the failure through await handle.ready() if you choose to await it.
• Monitor app progress — UpdateStats exposes per-component stats including error counts, so you can detect failures programmatically.

For the full picture — including the critical-path principle applied to every API, interrupted update recovery, and the exception handler API — see Error Handling.

No rollback, convergent roll-forward

CocoIndex does not roll back partial writes. The two-phase design makes this safe:

• Processing is side-effect-free — it only declares target states in memory. If processing fails (e.g., a parsing error), no writes were attempted, so there’s nothing to undo.
• Submit writes changes to target backends. If a submit fails partway through (e.g., a database connection drops), some writes may have been applied. CocoIndex does not attempt to undo them. Instead, on the next run CocoIndex computes the current desired state, and the target connector reconciles against all possible previous states — converging the target to the correct state regardless of what was partially applied. This is why built-in connectors use convergent operations like upserts (INSERT ... ON CONFLICT DO UPDATE) rather than plain inserts.

How big should a processing component be?

When defining processing components, think about granularity — what one path represents — because it determines the sync boundary for target states.

For example, if you’re processing files:

• Coarse: one component for “all files” (coco.component_subpath("process"))
• Medium: one component per file (coco.component_subpath("process", file_path))
• Fine: one component per chunk (coco.component_subpath("process", file_path, chunk_id))

In general:

• Coarse-grained (fewer, larger components): More target states sync together as a unit, but you only see updates after the larger component finishes.
• Fine-grained (more, smaller components): Each component syncs its target states as soon as it finishes, but target states owned by different components do not sync together as a unit.

For small datasets, a single processing component that owns all target states is simple and ensures all target states sync together. As data grows, consider breaking it down into one component per source item (e.g., one per file) to reduce latency: you see each item’s target states synced as soon as it’s processed, without waiting for the full dataset to complete. This also helps isolate failures to that item.

Explicit context management

CocoIndex automatically propagates component context through Python’s contextvars, which works for ordinary function calls (both sync and async). However, in situations where context variables are not preserved (for example, when using concurrent.futures.ThreadPoolExecutor), you need to explicitly capture and attach the context.

Use coco.get_component_context() to capture the current context, and context.attach() to restore it:

from concurrent.futures import ThreadPoolExecutor

@coco.fn
def app_main() -> None:
    # Capture the current context
    ctx = coco.get_component_context()

    def worker(item):
        # Attach the context in the worker thread
        with ctx.attach():
            # Now CocoIndex APIs work correctly
            process_item(item)

    with ThreadPoolExecutor() as executor:
        executor.map(worker, items)

This pattern ensures that CocoIndex can track component relationships and target state ownership even across thread boundaries.

Processing helpers

map()

map() applies an async function to each item in a collection, running all calls concurrently within the current processing component. Unlike mount() and mount_each(), it does not create child processing components — it’s purely concurrent execution (similar to asyncio.gather()).

@coco.fn(memo=True)
async def process_file(file: FileLike, table: postgres.TableTarget[DocEmbedding]) -> None:
    chunks = splitter.split(await file.read_text())
    id_gen = IdGenerator()
    await coco.map(process_chunk, chunks, file.file_path.path, id_gen, table)

The first argument to the function receives each item; additional arguments are passed through to every call. map() returns a list of the results, in the same order as the input items.

When to use map() vs mount_each()

• Use mount_each() when each item should be its own processing component — with its own component path, target state ownership, and target states sync boundary.
• Use map() when you want to process items concurrently within the current component, without creating new component boundaries. This is common for sub-item work like processing chunks within a file — the same within-component chunk work the chunk-embedding pipeline in Core Concepts walks through.