{"record":{"id":"ef330445c1445f4a","repo":"cocoindex-io/cocoindex","slug":"livemap-supports-a-single-active-watch-at-a-time","errorCode":null,"errorMessage":"LiveMap supports a single active watch() at a time.","messagePattern":"LiveMap supports a single active watch\\(\\) at a time\\.","errorType":"exception","errorClass":"RuntimeError","httpStatus":null,"severity":"error","filePath":"python/cocoindex/resources/live_map.py","lineNumber":287,"sourceCode":"        _coco.declare_target_state(self._entry_provider.target_state(key, value))\n\n    def __aiter__(self) -> _AsyncIterator[tuple[_K, _V]]:\n        return self._scan()\n\n    async def _scan(self) -> _AsyncIterator[tuple[_K, _V]]:\n        # Snapshot synchronously so a sink firing between yields can't mutate mid-iteration.\n        snapshot = list(self._entries.items())\n        if self._watcher_queue is not None and self._watch_scan_seq is None:\n            # First scan after a watch armed its queue = the watcher's initial\n            # snapshot (`subscriber.update_all`): record how far it reached.\n            self._watch_scan_seq = self._seq\n        for item in snapshot:\n            yield item\n\n    async def watch(self, subscriber: \"_coco.LiveMapSubscriber[_K, _V]\") -> None:\n        \"\"\"Deliver an initial snapshot then incremental changes. Drives one consumer.\"\"\"\n        if self._watcher_queue is not None:\n            raise RuntimeError(\"LiveMap supports a single active watch() at a time.\")\n        queue: _asyncio.Queue[_Change] = _asyncio.Queue()\n        # Arm before the scan so changes concurrent with it aren't lost. The mirror\n        # image of that choice is a change landing between arming and the snapshot:\n        # it gets queued AND included in the snapshot. The seq gate below drops such\n        # already-delivered changes at drain time (they'd otherwise re-notify the\n        # consumer with an equal value, defeating the `==` gate).\n        self._watcher_queue = queue\n        self._watch_scan_seq = None\n        try:\n            await subscriber.update_all()\n            await subscriber.mark_ready()\n            snapshot_seq = self._watch_scan_seq\n            while True:\n                change = await queue.get()\n                if snapshot_seq is not None and change.seq <= snapshot_seq:\n                    continue  # already reflected in the initial snapshot\n                if change.deleted:\n                    handle = await subscriber.delete(change.key)","sourceCodeStart":269,"sourceCodeEnd":305,"githubUrl":"https://github.com/cocoindex-io/cocoindex/blob/e84aa99b3292c5270a4b313b2a7137ad9ce8ab3b/python/cocoindex/resources/live_map.py#L269-L305","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Ensure only one task calls watch() per LiveMap; route changes to additional consumers via your own fan-out (queue/broadcast).","Cancel or await completion of the previous watcher before starting a new watch() call.","If you need multiple consumers, create a separate LiveMap per consumer or multiplex from the single subscriber yourself.","Guard startup code so a retry/reconnect path doesn't spawn a second watcher while the first lives."],"exampleFix":"// before\nasyncio.create_task(lm.watch(sub1))\nasyncio.create_task(lm.watch(sub2))  # RuntimeError\n// after\nasync def fan_out():\n    async for change in single_watcher_events:\n        await sub1(change)\n        await sub2(change)","handlingStrategy":"try-catch","validationCode":"if live_map._watcher_queue is not None:\n    raise RuntimeError('watch() already active; skip starting another')","typeGuard":null,"tryCatchPattern":"watcher_task = None\nasync def start_watcher(lm, sub):\n    global watcher_task\n    try:\n        await lm.watch(sub)\n    except RuntimeError as e:\n        if 'single active watch()' in str(e):\n            logging.warning('watcher already running; not starting another')\n        else:\n            raise","preventionTips":["Start the watcher in exactly one place in your app (e.g. a single supervised task).","Cancel and await the previous watcher task before restarting on reconnect.","Fan out to multiple consumers with your own asyncio.Queue/broadcast rather than multiple watch() calls."],"tags":["python","asyncio","live-map","concurrency"],"backgroundTag":"invalid-state-transition","analyzedSha":"e84aa99b3292c5270a4b313b2a7137ad9ce8ab3b","analyzedAt":"2026-09-08T15:59:19.997Z","contentChangedAt":"2026-09-08T15:59:19.997Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}