cocoindex-io/cocoindex · error · RuntimeError

LiveMap supports a single active watch() at a time.

Error message

LiveMap supports a single active watch() at a time.

What it means

Raised when `LiveMap.watch()` is called while another watcher is already active on the same LiveMap. LiveMap supports exactly one concurrent consumer: it drives a single subscriber via an internal queue, and a second concurrent watch would receive duplicated/conflicting deliveries.

Source

Thrown at python/cocoindex/resources/live_map.py:287

        _coco.declare_target_state(self._entry_provider.target_state(key, value))

    def __aiter__(self) -> _AsyncIterator[tuple[_K, _V]]:
        return self._scan()

    async def _scan(self) -> _AsyncIterator[tuple[_K, _V]]:
        # Snapshot synchronously so a sink firing between yields can't mutate mid-iteration.
        snapshot = list(self._entries.items())
        if self._watcher_queue is not None and self._watch_scan_seq is None:
            # First scan after a watch armed its queue = the watcher's initial
            # snapshot (`subscriber.update_all`): record how far it reached.
            self._watch_scan_seq = self._seq
        for item in snapshot:
            yield item

    async def watch(self, subscriber: "_coco.LiveMapSubscriber[_K, _V]") -> None:
        """Deliver an initial snapshot then incremental changes. Drives one consumer."""
        if self._watcher_queue is not None:
            raise RuntimeError("LiveMap supports a single active watch() at a time.")
        queue: _asyncio.Queue[_Change] = _asyncio.Queue()
        # Arm before the scan so changes concurrent with it aren't lost. The mirror
        # image of that choice is a change landing between arming and the snapshot:
        # it gets queued AND included in the snapshot. The seq gate below drops such
        # already-delivered changes at drain time (they'd otherwise re-notify the
        # consumer with an equal value, defeating the `==` gate).
        self._watcher_queue = queue
        self._watch_scan_seq = None
        try:
            await subscriber.update_all()
            await subscriber.mark_ready()
            snapshot_seq = self._watch_scan_seq
            while True:
                change = await queue.get()
                if snapshot_seq is not None and change.seq <= snapshot_seq:
                    continue  # already reflected in the initial snapshot
                if change.deleted:
                    handle = await subscriber.delete(change.key)

View on GitHub (pinned to e84aa99b32)

Solutions

1. Ensure only one task calls watch() per LiveMap; route changes to additional consumers via your own fan-out (queue/broadcast).
2. Cancel or await completion of the previous watcher before starting a new watch() call.
3. If you need multiple consumers, create a separate LiveMap per consumer or multiplex from the single subscriber yourself.
4. Guard startup code so a retry/reconnect path doesn't spawn a second watcher while the first lives.

Example fix

// before
asyncio.create_task(lm.watch(sub1))
asyncio.create_task(lm.watch(sub2))  # RuntimeError
// after
async def fan_out():
    async for change in single_watcher_events:
        await sub1(change)
        await sub2(change)

if live_map._watcher_queue is not None:
    raise RuntimeError('watch() already active; skip starting another')

watcher_task = None
async def start_watcher(lm, sub):
    global watcher_task
    try:
        await lm.watch(sub)
    except RuntimeError as e:
        if 'single active watch()' in str(e):
            logging.warning('watcher already running; not starting another')
        else:
            raise

When it happens

Trigger: Calling `live_map.watch(subscriber)` (typically from an app_main consumer task) while a previous `watch()` is still running and has not been cancelled/finished, so `_watcher_queue` is still set.

Common situations: Starting two consumer tasks in app_main that both watch the same LiveMap; restarting a consumer without cancelling the old one (e.g. on reconnect logic); running the app twice in the same process sharing one LiveMap instance.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

• Cannot use sync 'with coco.runtime()' from within an async e
• Memo state function returned an awaitable from a sync contex
• async_to_sync_iter must not be called from a running event l
• use_mount() requires a ComponentSubpath when the function ha
• LiveComponent classes cannot be used with use_mount(). Use m

AI-assisted analysis of cocoindex-io/cocoindex@e84aa99b32 (2026-09-08). Data as JSON: /api/errors/ef330445c1445f4a. Report an issue: GitHub.