{"record":{"id":"4b6098ca4bb75216","repo":"666ghj/MiroFish","slug":"zep-updater-worker-did-not-stop-within-join-timeo","errorCode":null,"errorMessage":"Zep updater worker did not stop within {join_timeout:.0f}s","messagePattern":"Zep updater worker did not stop within (.+?)s","errorType":"exception","errorClass":"TimeoutError","httpStatus":null,"severity":"error","filePath":"backend/app/services/zep_graph_memory_updater.py","lineNumber":326,"sourceCode":"            name=f\"ZepMemoryUpdater-{self.graph_id[:8]}\"\n        )\n        self._worker_thread.start()\n        logger.info(f\"ZepGraphMemoryUpdater 已启动: graph_id={self.graph_id}\")\n    \n    def stop(self):\n        \"\"\"Drain the worker, flush tail events, and wait for Cloud ingestion.\"\"\"\n        deadline = time.time() + ZEP_INGESTION_WAIT_TIMEOUT_SECONDS\n        # Serialize the accepting->closed transition with add_activity's\n        # check+enqueue operation. This closes the small race where a producer\n        # could enqueue after both the worker and final flush had exited.\n        with self._acceptance_lock:\n            self._running = False\n\n        if self._worker_thread and self._worker_thread.is_alive():\n            join_timeout = max(0.0, deadline - time.time())\n            self._worker_thread.join(timeout=join_timeout)\n            if self._worker_thread.is_alive():\n                raise TimeoutError(\n                    f\"Zep updater worker did not stop within {join_timeout:.0f}s\"\n                )\n\n        # The worker has drained the queue. Only now is it safe to flush\n        # buffers; doing this before join loses an item already dequeued by the\n        # worker but not yet buffered.\n        self._flush_remaining(deadline=deadline)\n\n        if self._failed_batches:\n            raise RuntimeError(\n                f\"{len(self._failed_batches)} Zep activity batch(es) failed; \"\n                \"simulation graph ingestion is incomplete\"\n            )\n\n        self._wait_for_pending_episodes(deadline=deadline)\n        \n        logger.info(f\"ZepGraphMemoryUpdater 已停止: graph_id={self.graph_id}, \"\n                   f\"total_activities={self._total_activities}, \"","sourceCodeStart":308,"sourceCodeEnd":344,"githubUrl":"https://github.com/666ghj/MiroFish/blob/b5b53acc57189a4a42e44a23e149dc655c98fe82/backend/app/services/zep_graph_memory_updater.py#L308-L344","documentation":"stop() on the updater drains its worker thread within a deadline derived from ZEP_INGESTION_WAIT_TIMEOUT_SECONDS; if the worker thread is still alive after join_timeout seconds it raises TimeoutError('Zep updater worker did not stop within Ns'). The worker is stuck — typically blocked in a Zep HTTP call or retry backoff — so orderly shutdown cannot be guaranteed.","triggerScenarios":"Calling stop() while the worker is mid-batch against a slow/hung Zep API; retry backoff (initial_delay 2.0s, 3 retries per call in the reader/updater pattern) stacking beyond the deadline; very large queued backlog exceeding the ingestion wait budget.","commonSituations":"Zep Cloud latency spikes or outages at simulation stop; on-prem proxy stalling connections; simulations with heavy activity volume so drain time legitimately exceeds the default budget.","solutions":["Retry stop() after checking Zep API health — the worker may finish its in-flight request and a second stop can complete.","Increase ZEP_INGESTION_WAIT_TIMEOUT_SECONDS (and ZEP_HTTP_REQUEST_TIMEOUT_SECONDS) to match your batch sizes and network conditions.","If permanently stuck, capture thread dump/logs for the worker, then abandon the thread (daemon) and reconcile the graph later — expect the 'N batch(es) failed' RuntimeError or partial ingestion."],"exampleFix":"# before\nupdater.stop()  # TimeoutError: worker did not stop within 60s\n\n# after\ntry:\n    updater.stop()\nexcept TimeoutError:\n    logger.warning(\"zep worker stuck; waiting one more cycle\")\n    time.sleep(30)\n    updater.stop()  # second attempt after in-flight HTTP completes","handlingStrategy":"retry","validationCode":null,"typeGuard":null,"tryCatchPattern":"try:\n    updater.stop()\nexcept TimeoutError:\n    time.sleep(30)\n    updater.stop()  # second attempt; if it still fails, escalate with logs","preventionTips":["Tune ZEP_INGESTION_WAIT_TIMEOUT_SECONDS to worst-case batch drain time.","Monitor Zep API latency during runs; slow API = slow stop.","Keep per-batch durations logged so the deadline can be set from data."],"tags":["zep","updater","timeout","shutdown","worker-thread"],"backgroundTag":null,"analyzedSha":"b5b53acc57189a4a42e44a23e149dc655c98fe82","analyzedAt":"2026-08-14T22:29:33.146Z","schemaVersion":2},"datasetVersion":"2026-08-15T22:17:37.221Z"}