diff --git a/src/import_memory.py b/src/import_memory.py index 2efa715428..d8565bc65a 100644 --- a/src/import_memory.py +++ b/src/import_memory.py @@ -810,6 +810,7 @@ def __init__(self, state_dir: str): "memories_skipped": 0, "memories_raw": 0, "chunks_failed": 0, + "failed_chunk_indices": [], "errors": [], "status": "idle", # idle | running | paused | completed | error "job_id": "", @@ -859,6 +860,7 @@ def reset( "memories_skipped": 0, "memories_raw": 0, "chunks_failed": 0, + "failed_chunk_indices": [], "errors": [], "status": "running", "job_id": job_id, @@ -1164,37 +1166,73 @@ async def start( self.release_start_reservation(job_id) async def _process_chunks(self, preserve_raw: bool) -> dict: - """从当前进度开始处理分块。""" + """从当前进度开始处理分块；续传时先补跑之前失败的分块。""" start_idx = self.state.data["processed"] await self._prime_exact_content_hashes() - for i in range(start_idx, len(self._chunks)): - if self._paused: - self.state.data["status"] = "paused" - self.state.save() - logger.info(f"Import paused at chunk {i}/{len(self._chunks)}") - return self.state.to_dict() - - chunk = self._chunks[i] + # processed 表示主游标已经尝试到哪里；失败分块单独持久化，Resume 时 + # 必须先补跑它们，否则单靠 processed 会从失败位置之后继续，静默漏块。 + raw_failed_indices = self.state.data.get("failed_chunk_indices", []) + failed_indices: list[int] = [] + for value in raw_failed_indices if isinstance(raw_failed_indices, list) else []: + if isinstance(value, int) and 0 <= value < len(self._chunks) and value not in failed_indices: + failed_indices.append(value) + self.state.data["failed_chunk_indices"] = failed_indices + + async def process_chunk(index: int) -> bool: + chunk = self._chunks[index] try: - chunk_ok = await self._process_single_chunk(chunk, preserve_raw) + return await self._process_single_chunk(chunk, preserve_raw) except Exception as e: - chunk_ok = False - err_msg = f"分块 {i} 处理失败（{type(e).__name__}）" + err_msg = f"分块 {index} 处理失败（{type(e).__name__}）" logger.warning( "Import chunk failed: index=%s err_type=%s detail=hidden", - i, + index, type(e).__name__, ) if len(self.state.data["errors"]) < _STATE_ERR_LOG_MAX: self.state.data["errors"].append(err_msg) + return False + + # Resume 先重试历史失败分块。成功后解除失败标记；再次失败则保留， + # 但不重复增加 chunks_failed。 + for failed_index in list(failed_indices): + if self._paused: + self.state.data["status"] = "paused" + self.state.save() + logger.info( + f"Import paused before retrying failed chunk " + f"{failed_index}/{len(self._chunks)}" + ) + return self.state.to_dict() + + chunk_ok = await process_chunk(failed_index) + if chunk_ok: + failed_indices.remove(failed_index) + self.state.data["failed_chunk_indices"] = failed_indices + self.state.data["chunks_failed"] = max( + 0, int(self.state.data.get("chunks_failed", 0)) - 1 + ) + self.state.save() + + for i in range(start_idx, len(self._chunks)): + if self._paused: + self.state.data["status"] = "paused" + self.state.save() + logger.info(f"Import paused at chunk {i}/{len(self._chunks)}") + return self.state.to_dict() + + chunk_ok = await process_chunk(i) if not chunk_ok: - self.state.data["chunks_failed"] = int( - self.state.data.get("chunks_failed", 0) - ) + 1 + if i not in failed_indices: + failed_indices.append(i) + self.state.data["failed_chunk_indices"] = failed_indices + self.state.data["chunks_failed"] = int( + self.state.data.get("chunks_failed", 0) + ) + 1 self.state.data["processed"] = i + 1 - # 每处理一个分块就保存进度 + # 每处理一个分块就保存进度；失败索引与主游标一起原子持久化。 self.state.save() failed_chunks = int(self.state.data.get("chunks_failed", 0)) diff --git a/tests/test_import_memory_regressions.py b/tests/test_import_memory_regressions.py index 8e00463435..a06cba17b2 100644 --- a/tests/test_import_memory_regressions.py +++ b/tests/test_import_memory_regressions.py @@ -1,4 +1,4 @@ -"""import_memory.py 回归测试 —— 找茬会话（2026-07-15）发现的四个问题。 +"""import_memory.py 回归测试 —— 覆盖历史导入的断点续传、分块与持久化边界。 1. preserve_raw 断点续传重复导入：进度只在整个 chunk 处理完才落盘，崩溃 重启后同一个 chunk 会被重新提取一遍；preserve_raw 分支原来完全跳过去重。 @@ -7,6 +7,7 @@ 3. 单次提取正文被固定按 12000 字符截断，远小于 chunk 自身 ~10000 token 的目标预算（对英文/混合内容而言），且不留任何痕迹。 4. ImportState.save() 无 fsync/长路径前缀（见 utils.atomic_write_text 用法）。 +5. #137：失败 chunk 推进 processed 后，Resume 不能静默跳过未完成工作。 """ import hashlib import json @@ -381,6 +382,124 @@ async def process(_chunk, _preserve_raw): assert result["chunks_failed"] == outcomes.count(False) +@pytest.mark.asyncio +async def test_resume_retries_failed_chunk_before_advancing(tmp_path, monkeypatch): + engine = ImportEngine( + {"buckets_dir": str(tmp_path), "human": "用户"}, + FakeBucketManager(), + FakeDehydrator(), + ) + engine._chunks = [ + {"content": "chunk-0"}, + {"content": "chunk-1"}, + ] + engine.state.reset("history.txt", "hash", 2, job_id="job") + calls: list[str] = [] + chunk0_attempts = 0 + + async def process(chunk, _preserve_raw): + nonlocal chunk0_attempts + name = chunk["content"] + calls.append(name) + if name == "chunk-0": + chunk0_attempts += 1 + if chunk0_attempts == 1: + # 模拟 chunk 0 失败期间用户点击 Pause：本轮会在进入 chunk 1 前暂停。 + engine._paused = True + return False + return True + + monkeypatch.setattr(engine, "_process_single_chunk", process) + + first = await engine._process_chunks(False) + + assert first["status"] == "paused" + assert first["processed"] == 1 + assert first["chunks_failed"] == 1 + assert first["failed_chunk_indices"] == [0] + assert calls == ["chunk-0"] + + # Resume：必须先补跑失败的 chunk 0，再从主游标继续 chunk 1。 + engine._paused = False + resumed = await engine._process_chunks(False) + + assert calls == ["chunk-0", "chunk-0", "chunk-1"] + assert resumed["processed"] == 2 + assert resumed["chunks_failed"] == 0 + assert resumed["failed_chunk_indices"] == [] + assert resumed["status"] == "completed" + + +@pytest.mark.asyncio +async def test_failed_chunk_retry_does_not_double_count_failure(tmp_path, monkeypatch): + engine = ImportEngine( + {"buckets_dir": str(tmp_path), "human": "用户"}, + FakeBucketManager(), + FakeDehydrator(), + ) + engine._chunks = [ + {"content": "chunk-0"}, + {"content": "chunk-1"}, + ] + engine.state.reset("history.txt", "hash", 2, job_id="job") + engine.state.data.update({ + "processed": 1, + "chunks_failed": 1, + "failed_chunk_indices": [0], + "status": "paused", + }) + + async def process(chunk, _preserve_raw): + return chunk["content"] != "chunk-0" + + monkeypatch.setattr(engine, "_process_single_chunk", process) + + result = await engine._process_chunks(False) + + assert result["processed"] == 2 + assert result["chunks_failed"] == 1 + assert result["failed_chunk_indices"] == [0] + assert result["status"] == "partial" + + +@pytest.mark.asyncio +async def test_resume_keeps_failure_state_consistent_when_later_chunk_fails( + tmp_path, monkeypatch +): + engine = ImportEngine( + {"buckets_dir": str(tmp_path), "human": "用户"}, + FakeBucketManager(), + FakeDehydrator(), + ) + engine._chunks = [ + {"content": "chunk-0"}, + {"content": "chunk-1"}, + {"content": "chunk-2"}, + ] + engine.state.reset("history.txt", "hash", 3, job_id="job") + engine.state.data.update({ + "processed": 1, + "chunks_failed": 1, + "failed_chunk_indices": [0], + "status": "paused", + }) + calls: list[str] = [] + + async def process(chunk, _preserve_raw): + calls.append(chunk["content"]) + return chunk["content"] != "chunk-1" + + monkeypatch.setattr(engine, "_process_single_chunk", process) + + result = await engine._process_chunks(False) + + assert calls == ["chunk-0", "chunk-1", "chunk-2"] + assert result["processed"] == 3 + assert result["chunks_failed"] == 1 + assert result["failed_chunk_indices"] == [1] + assert result["status"] == "partial" + + @pytest.mark.asyncio async def test_import_preflight_failure_replaces_stale_status(tmp_path): dehydrator = FakeDehydrator() diff --git a/update_manifest.json b/update_manifest.json index 9f5ed6ca11..5b51cbf7d8 100644 --- a/update_manifest.json +++ b/update_manifest.json @@ -77,8 +77,8 @@ }, { "path": "src/import_memory.py", - "sha256": "9f89c8ef32b09f0d6ee3274bba1984b8d383ee2ae338ccd4e56dd04171ab5f55", - "size": 60702 + "sha256": "693b7a33db1e2d575d18a2fd1032f5dd5408637e9d958bedf8782adeb57a3956", + "size": 62700 }, { "path": "src/migrate_engine.py",