{"record":{"id":"32e4d11399a317a9","repo":"666ghj/MiroFish","slug":"zep-updater-drain-deadline-elapsed-before-flushing","errorCode":null,"errorMessage":"Zep updater drain deadline elapsed before flushing all activities","messagePattern":"Zep updater drain deadline elapsed before flushing all activities","errorType":"exception","errorClass":"TimeoutError","httpStatus":null,"severity":"error","filePath":"backend/app/services/zep_graph_memory_updater.py","lineNumber":582,"sourceCode":"            try:\n                activity = self._activity_queue.get_nowait()\n                platform = activity.platform.lower()\n                with self._buffer_lock:\n                    if platform not in self._platform_buffers:\n                        self._platform_buffers[platform] = []\n                    self._platform_buffers[platform].append(activity)\n            except Empty:\n                break\n        \n        for platform in list(self._platform_buffers):\n            with self._buffer_lock:\n                buffer = list(self._platform_buffers.get(platform, []))\n            if not buffer:\n                continue\n            display_name = self._get_platform_display_name(platform)\n            logger.info(f\"发送{display_name}平台剩余的 {len(buffer)} 条活动\")\n            if deadline is not None and time.time() >= deadline:\n                raise TimeoutError(\n                    \"Zep updater drain deadline elapsed before flushing all activities\"\n                )\n            try:\n                processed_count = self._send_batch_activities(\n                    buffer,\n                    platform,\n                    deadline=deadline,\n                )\n            except _DrainDeadlineExceeded as error:\n                with self._buffer_lock:\n                    del self._platform_buffers[platform][:error.processed_count]\n                raise TimeoutError(str(error)) from error\n            else:\n                with self._buffer_lock:\n                    del self._platform_buffers[platform][:processed_count]\n\n    def _wait_for_pending_episodes(self, *, deadline: float | None = None) -> None:\n        pending = set(self._pending_episode_uuids)","sourceCodeStart":564,"sourceCodeEnd":600,"githubUrl":"https://github.com/666ghj/MiroFish/blob/b5b53acc57189a4a42e44a23e149dc655c98fe82/backend/app/services/zep_graph_memory_updater.py#L564-L600","documentation":"Raised by the updater's drain routine when the caller-supplied deadline (time.time() >= deadline) is reached before every per-platform activity buffer has been flushed to Zep. It is a cooperative cancellation signal: remaining buffered activities stay in _platform_buffers so the failure is surfaced to SimulationRunner rather than silently dropped.","triggerScenarios":"Calling the drain/flush path with a deadline after slow Zep writes have consumed most of the budget: by the time the loop reaches a platform whose buffer is non-empty, time.time() >= deadline. Large backlogs, slow network, or an unrealistically tight deadline trigger it before _send_batch_activities is even attempted.","commonSituations":"Simulation shutdown with a fixed timeout, Zep Cloud latency spikes or rate limiting (each retry burns deadline budget), or a burst of activity that filled buffers faster than batches could be sent.","solutions":["Increase or remove the drain deadline if the data must be flushed (accept longer shutdown)","Check the updater stats (total_sent vs buffered counts) to size batches/deadline realistically","Address the root cause of slowness: Zep rate limits, network latency, oversized batches","Treat the TimeoutError as 'batch incomplete': buffers are preserved, so re-run drain with a fresh deadline instead of resending manually"],"exampleFix":"# before\ndrain(deadline=time.time() + 5)  # too tight for a large backlog\n\n# after\nbudget = max(30, estimated_batches * avg_batch_seconds * 2)\ndrain(deadline=time.time() + budget)","handlingStrategy":"retry","validationCode":"remaining = sum(len(b) for b in updater._platform_buffers.values())\nif remaining and deadline_soon:\n    # extend or re-plan instead of letting drain raise\n    deadline = time.time() + new_budget","typeGuard":null,"tryCatchPattern":"try:\n    updater.drain(deadline=deadline)\nexcept TimeoutError:\n    # buffers are preserved; retry drain with a fresh deadline\n    updater.drain(deadline=time.time() + EXTENDED_BUDGET)","preventionTips":["Size the drain deadline to the buffered backlog, not a constant","Monitor total_sent vs buffered counts during the run to predict drain duration","Send activities incrementally instead of one large end-of-run flush"],"tags":["zep","timeout","drain","backpressure"],"backgroundTag":null,"analyzedSha":"b5b53acc57189a4a42e44a23e149dc655c98fe82","analyzedAt":"2026-08-14T22:29:33.146Z","schemaVersion":2},"datasetVersion":"2026-08-15T22:17:37.221Z"}