fix: 提升本地存储写入可靠性#1754
Merged
SengokuCola merged 2 commits intoMay 26, 2026
Merged
Conversation
WalkthroughLocalStoreManager 使用 Path 与 LocalStoreValue 类型，读取时验证根 JSON 为 dict 并在损坏时备份为 .corrupt(.N)，写入改为在同目录写临时文件后用 os.replace 原子替换；新增三项测试覆盖裸文件名、损坏备份与写入失败场景。 Changes本地存储安全写入与损坏恢复
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~20 minutes 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
🧹 Nitpick comments (2)
src/manager/local_store_manager.py (1)
49-49: 💤 Low value建议添加显式返回类型注解
load_local_store方法缺少显式的-> None返回类型注解。根据项目规范，为具有复杂功能的函数添加类型注解可提高代码可读性。♻️ 建议的修改
- def load_local_store(self): + def load_local_store(self) -> None:🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/manager/local_store_manager.py` at line 49, 在 src/manager/local_store_manager.py 中为方法 load_local_store 添加显式返回类型注解，将其签名由 "def load_local_store(self):" 改为 "def load_local_store(self) -> None:"（保持实现不变），以符合项目类型注解规范并提高可读性；只需修改函数定义处的类型注解，确保任何相关静态类型检查或文档生成能识别该返回类型。pytests/common_test/test_local_store_manager.py (1)
9-14: 💤 Low value建议添加返回类型注解
_load_local_store_module有两个参数但缺少返回类型注解。根据编码规范，多参数函数应添加类型注解以提高可读性。♻️ 建议的修改
+from types import ModuleType + + -def _load_local_store_module(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): +def _load_local_store_module(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> ModuleType:🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pytests/common_test/test_local_store_manager.py` around lines 9 - 14, Add a return type annotation to the function _load_local_store_module to indicate it returns a module object (e.g., -> ModuleType); update the signature to accept tmp_path: Path, monkeypatch: pytest.MonkeyPatch and return ModuleType, and ensure you import ModuleType from types (or use typing.Any if you prefer). This change affects the function named _load_local_store_module and the top of the test file where imports live (add "from types import ModuleType" or equivalent) so static type checkers and readers know the function returns the reloaded module.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Nitpick comments:
In `@pytests/common_test/test_local_store_manager.py`:
- Around line 9-14: Add a return type annotation to the function
_load_local_store_module to indicate it returns a module object (e.g., ->
ModuleType); update the signature to accept tmp_path: Path, monkeypatch:
pytest.MonkeyPatch and return ModuleType, and ensure you import ModuleType from
types (or use typing.Any if you prefer). This change affects the function named
_load_local_store_module and the top of the test file where imports live (add
"from types import ModuleType" or equivalent) so static type checkers and
readers know the function returns the reloaded module.
In `@src/manager/local_store_manager.py`:
- Line 49: 在 src/manager/local_store_manager.py 中为方法 load_local_store
添加显式返回类型注解，将其签名由 "def load_local_store(self):" 改为 "def load_local_store(self) ->
None:"（保持实现不变），以符合项目类型注解规范并提高可读性；只需修改函数定义处的类型注解，确保任何相关静态类型检查或文档生成能识别该返回类型。
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: 4c2e7d26-2810-46c0-8b31-a1b59d629644
📒 Files selected for processing (2)
pytests/common_test/test_local_store_manager.pysrc/manager/local_store_manager.py
There was a problem hiding this comment.
🧹 Nitpick comments (2)
pytests/common_test/test_local_store_manager.py (1)
36-52: ⚡ Quick win补一条“合法 JSON 但根节点不是对象”的回归测试。
现在只覆盖了
JSONDecodeError场景，但LocalStoreManager.load_local_store()在 Line 62-63 还单独处理了“语法合法、结构非法”的 JSON（例如[]或"x"）。这个分支正好也是本次修复目标之一，建议单独补一例，避免后续只保住损坏语法文件而漏掉根节点类型校验。🧪 建议补充的测试用例
class TestLocalStoreManager: @@ def test_backs_up_broken_json_before_rebuild( self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: @@ assert json.loads(store_path.read_text(encoding="utf-8")) == {} assert (tmp_path / "data" / "local_store.json.corrupt").read_text(encoding="utf-8") == "{broken" + + def test_backs_up_non_object_json_before_rebuild( + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + """根节点不是对象时也应先备份，再重建为空本地存储。""" + local_store_module = _load_local_store_module(tmp_path, monkeypatch) + store_path = tmp_path / "data" / "local_store.json" + store_path.parent.mkdir(parents=True, exist_ok=True) + store_path.write_text("[]", encoding="utf-8") + + manager = local_store_module.LocalStoreManager(str(store_path)) + + assert manager.store == {} + assert json.loads(store_path.read_text(encoding="utf-8")) == {} + assert (tmp_path / "data" / "local_store.json.corrupt").read_text(encoding="utf-8") == "[]"🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pytests/common_test/test_local_store_manager.py` around lines 36 - 52, Add a regression test to cover the case where the JSON is syntactically valid but the root is not an object: in the test function suite create a store file containing a valid JSON value like "[]" or "\"x\"" (e.g., write_text("[]")), instantiate LocalStoreManager (or call LocalStoreManager.load_local_store()), then assert that manager.store == {}, the on-disk store file has been rewritten to "{}", and the original content was moved to a backup file named local_store.json.corrupt containing the original valid-but-wrong JSON; reference LocalStoreManager and its load_local_store() behavior when adding this test.src/manager/local_store_manager.py (1)
11-22: ⚡ Quick win把
LocalStoreValue改成参数化的递归 JSON 类型。这里的
list/dict是裸泛型，嵌套值类型会直接退化掉，store的类型约束基本失效。既然这里在建模 JSON 存储，建议显式参数化容器并递归描述值类型。♻️ 建议修改
-from typing import TypeAlias +from typing import Dict, List, TypeAlias @@ -LocalStoreValue: TypeAlias = str | list | dict | int | float | bool +LocalStoreValue: TypeAlias = ( + str | int | float | bool | None | List["LocalStoreValue"] | Dict[str, "LocalStoreValue"] +)As per coding guidelines, "For parameterized generics, use type annotations from the
typingmodule to specify the types of generic parameters, such asList[int]for a list of integers orDict[str, Any]for a dictionary with string keys and any-type values."🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/manager/local_store_manager.py` around lines 11 - 22, Change LocalStoreValue into a proper recursive JSON type alias and update LocalStoreManager.store to use it: replace the bare list/dict with a TypeAlias like JSONValue (or LocalStoreValue) defined via typing.Union including str, int, float, bool, None and recursive containers List["JSONValue"] and Dict[str, "JSONValue"] (use typing.TypeAlias, List, Dict, Union and forward references) so nested structures keep their element types; then set store: Dict[str, JSONValue]. Update imports accordingly and keep the symbol names LocalStoreValue (or rename consistently) and LocalStoreManager.store to locate the change.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Nitpick comments:
In `@pytests/common_test/test_local_store_manager.py`:
- Around line 36-52: Add a regression test to cover the case where the JSON is
syntactically valid but the root is not an object: in the test function suite
create a store file containing a valid JSON value like "[]" or "\"x\"" (e.g.,
write_text("[]")), instantiate LocalStoreManager (or call
LocalStoreManager.load_local_store()), then assert that manager.store == {}, the
on-disk store file has been rewritten to "{}", and the original content was
moved to a backup file named local_store.json.corrupt containing the original
valid-but-wrong JSON; reference LocalStoreManager and its load_local_store()
behavior when adding this test.
In `@src/manager/local_store_manager.py`:
- Around line 11-22: Change LocalStoreValue into a proper recursive JSON type
alias and update LocalStoreManager.store to use it: replace the bare list/dict
with a TypeAlias like JSONValue (or LocalStoreValue) defined via typing.Union
including str, int, float, bool, None and recursive containers List["JSONValue"]
and Dict[str, "JSONValue"] (use typing.TypeAlias, List, Dict, Union and forward
references) so nested structures keep their element types; then set store:
Dict[str, JSONValue]. Update imports accordingly and keep the symbol names
LocalStoreValue (or rename consistently) and LocalStoreManager.store to locate
the change.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: 1a857de5-72dd-4249-8c3d-4fbbf1eaa75b
📒 Files selected for processing (2)
pytests/common_test/test_local_store_manager.pysrc/manager/local_store_manager.py
liuwanwan1
force-pushed
the
fix/local-store-atomic-save
branch
from
74194ae to
c963f5e
Compare
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (1)
pytests/common_test/test_local_store_manager.py (1)
36-52: ⚡ Quick win建议补一条“根节点非对象”回归测试
目前损坏恢复只测了语法损坏（
JSONDecodeError），但load_local_store还新增了“根节点不是dict”分支。建议补一条[]或"x"的用例，确保.corrupt备份与重建逻辑在该分支也稳定。可补充的测试示例
class TestLocalStoreManager: @@ def test_backs_up_broken_json_before_rebuild( @@ assert (tmp_path / "data" / "local_store.json.corrupt").read_text(encoding="utf-8") == "{broken" + + def test_backs_up_non_object_json_root_before_rebuild( + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + """根节点不是对象时也应先备份，再重建为空本地存储。""" + local_store_module = _load_local_store_module(tmp_path, monkeypatch) + store_path = tmp_path / "data" / "local_store.json" + store_path.parent.mkdir(parents=True, exist_ok=True) + store_path.write_text("[]", encoding="utf-8") + + manager = local_store_module.LocalStoreManager(str(store_path)) + + assert manager.store == {} + assert json.loads(store_path.read_text(encoding="utf-8")) == {} + assert (tmp_path / "data" / "local_store.json.corrupt").read_text(encoding="utf-8") == "[]"🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pytests/common_test/test_local_store_manager.py` around lines 36 - 52, Add a new test mirroring test_backs_up_broken_json_before_rebuild that exercises the "root node not a dict" branch: create the same local_store.json with a root value like "[]" (or "\"x\""), instantiate LocalStoreManager (or call load_local_store via the local_store_module), then assert manager.store == {}, assert the file was rewritten to an empty JSON object, and assert the backup file local_store.json.corrupt contains the original non-dict content; reference LocalStoreManager and load_local_store to locate where to add this test.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/manager/local_store_manager.py`:
- Around line 59-72: The current load block in LocalStoreManager only catches
json.JSONDecodeError and ValueError, so OSError subclasses (e.g.,
PermissionError, temporary I/O errors) will propagate and can crash startup;
modify the try/except in the method that reads file_path (the block using
file_path.open(...), setting self.store, calling _backup_broken_store and
_write_store_atomically) to also catch OSError separately: on OSError log a
warning/error including the exception, fall back to an empty in-memory store
(self.store = {}), attempt to call _backup_broken_store if appropriate, and
avoid letting the exception propagate so the process can continue; keep the
existing JSON/ValueError handling intact.
---
Nitpick comments:
In `@pytests/common_test/test_local_store_manager.py`:
- Around line 36-52: Add a new test mirroring
test_backs_up_broken_json_before_rebuild that exercises the "root node not a
dict" branch: create the same local_store.json with a root value like "[]" (or
"\"x\""), instantiate LocalStoreManager (or call load_local_store via the
local_store_module), then assert manager.store == {}, assert the file was
rewritten to an empty JSON object, and assert the backup file
local_store.json.corrupt contains the original non-dict content; reference
LocalStoreManager and load_local_store to locate where to add this test.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: 2ae05200-bfbe-4953-8d20-944c18dd0093
📒 Files selected for processing (2)
pytests/common_test/test_local_store_manager.pysrc/manager/local_store_manager.py
Comment on lines
59
to
72
| try: | ||
| with open(self.file_path, "r", encoding="utf-8") as f: | ||
| self.store = json.load(f) | ||
| logger.info("全都记起来了！") | ||
| except json.JSONDecodeError: | ||
| with file_path.open("r", encoding="utf-8") as f: | ||
| loaded_store = json.load(f) | ||
| if not isinstance(loaded_store, dict): | ||
| raise ValueError("本地存储根节点必须是 JSON 对象") | ||
| self.store = loaded_store | ||
| logger.info("全都记起来了！") | ||
| except (json.JSONDecodeError, ValueError) as exc: | ||
| logger.warning("啊咧？记事本被弄脏了，正在重建记事本......") | ||
| logger.debug(f"本地存储文件无法读取: {exc}") | ||
| self._backup_broken_store(file_path) | ||
| self.store = {} | ||
| with open(self.file_path, "w", encoding="utf-8") as f: | ||
| json.dump({}, f, ensure_ascii=False, indent=4) | ||
| self._write_store_atomically(file_path, self.store) | ||
| logger.info("记事本重建成功！") |
There was a problem hiding this comment.
补齐读取阶段的 OSError 兜底，避免启动时直接失败
当前只捕获了 JSONDecodeError/ValueError。如果这里遇到 PermissionError、短暂 I/O 异常等 OSError，会直接中断初始化，影响可用性。建议单独处理 OSError，至少回退到内存空存储并记录日志，而不是让进程在启动阶段失败。
建议修改
try:
with file_path.open("r", encoding="utf-8") as f:
loaded_store = json.load(f)
if not isinstance(loaded_store, dict):
raise ValueError("本地存储根节点必须是 JSON 对象")
self.store = loaded_store
logger.info("全都记起来了！")
except (json.JSONDecodeError, ValueError) as exc:
logger.warning("啊咧？记事本被弄脏了，正在重建记事本......")
logger.debug(f"本地存储文件无法读取: {exc}")
self._backup_broken_store(file_path)
self.store = {}
self._write_store_atomically(file_path, self.store)
logger.info("记事本重建成功！")
+ except OSError as exc:
+ logger.warning("本地存储文件读取失败，将使用空内存存储继续运行。")
+ logger.debug(f"读取本地存储时发生 I/O 异常: {exc}")
+ self.store = {}🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/manager/local_store_manager.py` around lines 59 - 72, The current load
block in LocalStoreManager only catches json.JSONDecodeError and ValueError, so
OSError subclasses (e.g., PermissionError, temporary I/O errors) will propagate
and can crash startup; modify the try/except in the method that reads file_path
(the block using file_path.open(...), setting self.store, calling
_backup_broken_store and _write_store_atomically) to also catch OSError
separately: on OSError log a warning/error including the exception, fall back to
an empty in-memory store (self.store = {}), attempt to call _backup_broken_store
if appropriate, and avoid letting the exception propagate so the process can
continue; keep the existing JSON/ValueError handling intact.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
这次改了什么
这次主要是把
LocalStoreManager的本地存储读写做得更稳一点。以前保存
local_store.json时是直接覆盖目标文件，如果写入过程中出错，有可能把原文件截断。现在改成先在同目录写临时文件，写完后再用os.replace原子替换目标文件，这样保存失败时原文件还能保持原样。另外，启动时如果读到损坏的 JSON，或者文件根节点不是 JSON 对象，会先把原文件备份成
.corrupt，再重建一个空的本地存储，避免恢复过程直接覆盖现场。顺手也处理了自定义路径是
local_store.json这种裸文件名的情况，之前这类路径没有父目录，创建文件时容易出问题。测试覆盖
新增了
LocalStoreManager的单元测试，覆盖这几个场景：.tmp临时文件。破坏性变更
无。
测试命令
UV_CACHE_DIR=/tmp/uv-cache uv run --no-sync pytest pytests/common_test/test_local_store_manager.pyUV_CACHE_DIR=/tmp/uv-cache uv run --no-sync ruff check src/manager/local_store_manager.py pytests/common_test/test_local_store_manager.pyFixes #1753
PR 模板确认
devmainsrc/A_memorix，并已阅读src/A_memorix/MODIFICATION_POLICY.mdSummary by CodeRabbit
发布说明
Bug Fixes
Tests