fix(utils): 原子写重试预算上调至8次/4.75s,索引保存失败留证与日志落盘

- replace_with_retry: attempts 6→8,新增 max_delay 参数(默认1.6s),覆盖 Defender/Search 索引器持续持锁超过原1.55s预算的场景
This commit is contained in:
JOJO 2026-08-03 22:18:41 +08:00
parent 3ceace2762
commit a7a40a8a2f
2 changed files with 54 additions and 26 deletions

View File

@ -7,6 +7,10 @@ Windows Search 索引、同进程并发的读取线程都会造成这类短暂
POSIX rename 不受打开句柄影响无此问题 POSIX rename 不受打开句柄影响无此问题
对策仅对这两种 winerror 做短退避重试其他错误立即抛出 对策仅对这两种 winerror 做短退避重试其他错误立即抛出
2026-08 调整默认预算由 6 / 1.55s 上调至 8 / 4.75s
实测 Defender 实时扫描 / Search 索引器等外部程序偶发持锁超过原预算
如对话索引 index.json 保存失败 WinError 5提高预算以覆盖此类场景
""" """
from __future__ import annotations from __future__ import annotations
@ -25,13 +29,14 @@ def replace_with_retry(
src: PathLike, src: PathLike,
dst: PathLike, dst: PathLike,
*, *,
attempts: int = 6, attempts: int = 8,
initial_delay: float = 0.05, initial_delay: float = 0.05,
max_delay: float = 1.6,
) -> None: ) -> None:
"""os.replace 的 Windows 加固版:对瞬时持锁做指数退避重试。 """os.replace 的 Windows 加固版:对瞬时持锁做指数退避重试。
attempts 为总尝试次数含首次退避序列默认 0.05/0.1/0.2/0.4/0.8s attempts 为总尝试次数含首次退避序列默认 0.05/0.1/0.2/0.4/0.8/1.6/1.6s
最坏情况约 1.55s最后一次失败时原样抛出该 OSError 最坏情况约 4.75s最后一次失败时原样抛出该 OSError
""" """
delay = initial_delay delay = initial_delay
total = max(1, int(attempts)) total = max(1, int(attempts))
@ -43,4 +48,4 @@ def replace_with_retry(
if getattr(exc, "winerror", None) not in _RETRY_WINERRORS or i == total - 1: if getattr(exc, "winerror", None) not in _RETRY_WINERRORS or i == total - 1:
raise raise
time.sleep(delay) time.sleep(delay)
delay = min(delay * 2, 0.8) delay = min(delay * 2, max_delay)

View File

@ -11,15 +11,16 @@ from typing import Dict, List, Optional, Any
from dataclasses import dataclass from dataclasses import dataclass
from utils.atomic_io import replace_with_retry from utils.atomic_io import replace_with_retry
from utils.log_rotation import append_line
try: try:
from config import DATA_DIR, HOST_WORKSPACES_FILE from config import DATA_DIR, HOST_WORKSPACES_FILE, LOGS_DIR
except ImportError: except ImportError:
import sys import sys
from pathlib import Path from pathlib import Path
project_root = Path(__file__).resolve().parents[1] project_root = Path(__file__).resolve().parents[1]
if str(project_root) not in sys.path: if str(project_root) not in sys.path:
sys.path.insert(0, str(project_root)) sys.path.insert(0, str(project_root))
from config import DATA_DIR, HOST_WORKSPACES_FILE from config import DATA_DIR, HOST_WORKSPACES_FILE, LOGS_DIR
try: try:
from utils.perf_log import perf_log from utils.perf_log import perf_log
@ -117,7 +118,18 @@ class IndexMixin:
os.fsync(fh.fileno()) os.fsync(fh.fileno())
# Windows 下目标文件被并发读取/杀软扫描短暂持锁时 replace 会抛 # Windows 下目标文件被并发读取/杀软扫描短暂持锁时 replace 会抛
# WinError 5/32replace_with_retry 做短退避重试POSIX 行为不变) # WinError 5/32replace_with_retry 做短退避重试POSIX 行为不变)
try:
replace_with_retry(temp_path, target_file) replace_with_retry(temp_path, target_file)
except OSError:
# 重试预算耗尽仍失败:把已写好的新内容转移为 .last_failed.tmp 留证,
# 便于事后恢复与频率统计(固定文件名覆盖,不累积垃圾);原异常继续抛出
try:
orphan = target_file.with_name(f".{target_file.name}.last_failed.tmp")
replace_with_retry(temp_path, orphan)
temp_path = None # 已转移finally 不再删除
except OSError:
pass
raise
finally: finally:
if temp_path and temp_path.exists(): if temp_path and temp_path.exists():
try: try:
@ -198,6 +210,8 @@ class IndexMixin:
try: try:
index: Dict = {} index: Dict = {}
if self.index_file.exists(): if self.index_file.exists():
# 只在 with 块内读字节json.loads 期间不持有文件句柄,避免
# Windows 下读取方持锁阻塞写方 os.replace无 FILE_SHARE_DELETE
with open(self.index_file, 'r', encoding='utf-8') as f: with open(self.index_file, 'r', encoding='utf-8') as f:
content = f.read().strip() content = f.read().strip()
if content: if content:
@ -254,6 +268,15 @@ class IndexMixin:
self._atomic_write_json(self.index_file, index) self._atomic_write_json(self.index_file, index)
except Exception as e: except Exception as e:
print(f"⌘ 保存对话索引失败: {e}") print(f"⌘ 保存对话索引失败: {e}")
# 落盘留证(线程安全、自动轮转、异常自吞),便于统计失败频率与诊断
append_line(
Path(LOGS_DIR) / "conversation_index_failures.log",
json.dumps({
"ts": datetime.now().isoformat(timespec="milliseconds"),
"index_file": str(self.index_file),
"error": str(e), # str 含 [WinError N] 与源/目标路径repr 会丢失
}, ensure_ascii=False),
)
def _ensure_index_covering(self, limit: int, offset: int) -> Dict: def _ensure_index_covering(self, limit: int, offset: int) -> Dict:
""" """