From d0a14b5c1bef502e631698526727a9abeccfbb08 Mon Sep 17 00:00:00 2001 From: shy3130 <415333856@qq.com> Date: Wed, 9 Sep 2026 16:30:20 +0800 Subject: [PATCH] =?UTF-8?q?fix(ext-pull):=20=E6=8B=89=E5=8F=96=E5=BE=AA?= =?UTF-8?q?=E7=8E=AF=E7=9A=84=E7=8A=B6=E6=80=81=E5=9B=9E=E5=86=99=E4=B8=8D?= =?UTF-8?q?=E5=86=8D=E6=B8=85=E7=AD=96=E7=95=A5=E7=BC=93=E5=AD=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 上修复 (keep_strategy_cache 沿数据写入链放行) 后线上复现: 每轮拉取 成功后 12ms 缓存仍被清空。clear_cache 新增调用链日志抓到真凶 — ExtConfigStore.upsert 无条件触发 _invalidate_ext_derived: 调度器每轮 拉取要回写 last_run/last_status/next_run 共 2-3 次配置, 每次都全清 策略结果缓存, 绕过了已放行的数据写入链路。 upsert 增加 keep_strategy_cache 参数 (默认 False 保持 UI 保存配置/ 手动变更的全清语义), 拉取循环内 4 处例行回写全部传 True。 --- backend/app/services/ext_data.py | 8 +++++--- backend/app/services/ext_pull.py | 8 ++++---- backend/tests/test_ext_factors.py | 27 +++++++++++++++++++++++++++ 3 files changed, 36 insertions(+), 7 deletions(-) diff --git a/backend/app/services/ext_data.py b/backend/app/services/ext_data.py index 25a6e7b..39165f3 100644 --- a/backend/app/services/ext_data.py +++ b/backend/app/services/ext_data.py @@ -293,7 +293,7 @@ class ExtConfigStore: except Exception: return None - def upsert(self, config: ExtConfig) -> None: + def upsert(self, config: ExtConfig, *, keep_strategy_cache: bool = False) -> None: config.updated_at = datetime.now().isoformat() cp = self._config_path(config.id) cp.parent.mkdir(parents=True, exist_ok=True) @@ -301,8 +301,10 @@ class ExtConfigStore: json.dumps(config.to_dict(), ensure_ascii=False, indent=2), encoding="utf-8", ) - # 字段集/模式变化会改变扩展列集合: 失效扩展帧缓存与策略结果缓存 - _invalidate_ext_derived(self._base.parent) + # 字段集/模式变化会改变扩展列集合: 失效扩展帧缓存与策略结果缓存。 + # 定时拉取循环的 last_run/next_run 例行回写传 keep_strategy_cache=True, + # 否则每轮拉取后策略页缓存被状态回写清空 (数据写入链路已另行放行)。 + _invalidate_ext_derived(self._base.parent, keep_strategy_cache=keep_strategy_cache) def delete(self, config_id: str) -> bool: import shutil diff --git a/backend/app/services/ext_pull.py b/backend/app/services/ext_pull.py index 65f1786..d64de72 100644 --- a/backend/app/services/ext_pull.py +++ b/backend/app/services/ext_pull.py @@ -496,7 +496,7 @@ class PullScheduler: fresh.pull.last_run = datetime.now(timezone.utc).isoformat() fresh.pull.last_status = "skipped" fresh.pull.last_message = "不在拉取时间窗口内" - store.upsert(fresh) + store.upsert(fresh, keep_strategy_cache=True) logger.info("PullScheduler: %s skipped (outside time window)", config.id) interval = max(pull.schedule_minutes * 60, 60) await asyncio.sleep(interval) @@ -513,7 +513,7 @@ class PullScheduler: fresh.pull.last_status = "success" fresh.pull.last_message = f"{n} rows @ {d}" fresh.pull.last_rows = n - store.upsert(fresh) + store.upsert(fresh, keep_strategy_cache=True) logger.info("PullScheduler: %s success, %d rows", config.id, n) except Exception as e: fresh2 = store.get(config.id) @@ -521,7 +521,7 @@ class PullScheduler: fresh2.pull.last_run = datetime.now(timezone.utc).isoformat() fresh2.pull.last_status = "error" fresh2.pull.last_message = str(e)[:200] - store.upsert(fresh2) + store.upsert(fresh2, keep_strategy_cache=True) logger.warning("PullScheduler: %s error: %s", config.id, e) # 间隔取自最新配置 (每次重新读取, 修复改间隔不生效) @@ -533,7 +533,7 @@ class PullScheduler: latest.pull.next_run = datetime.fromtimestamp( next_dt, tz=UTC ).isoformat() - store.upsert(latest) + store.upsert(latest, keep_strategy_cache=True) await asyncio.sleep(interval) if not self._running: diff --git a/backend/tests/test_ext_factors.py b/backend/tests/test_ext_factors.py index 483b5b9..b38ae33 100644 --- a/backend/tests/test_ext_factors.py +++ b/backend/tests/test_ext_factors.py @@ -275,6 +275,33 @@ def test_routine_pull_keeps_strategy_cache_but_default_clears(data_dir): assert strategy_cache.read_cache(data_dir) is None +def test_scheduler_status_upsert_keeps_strategy_cache(data_dir): + """定时拉取循环的 last_run/next_run 例行回写 (keep_strategy_cache=True) + 不清策略结果缓存; UI 保存配置 (默认) 仍全清。 + + 线上事故: 每轮拉取成功后 store.upsert(fresh) 状态回写触发全清, 数据 + 写入链路放行后 12ms 缓存仍被清空 —— 拉取循环内所有回写都须放行。 + """ + from app.services import strategy_cache + from app.services.ext_data import ExtConfigStore + + cfg = _mk_config(data_dir, mode="timeseries") + store = ExtConfigStore(data_dir) + strategy_cache.write_cache( + data_dir, "2026-01-05", + {"s1": {"total": 1, "as_of": "2026-01-05", "rows": []}}, + ) + + # 调度器例行回写 (last_run/next_run): 保留策略结果 + store.upsert(cfg, keep_strategy_cache=True) + cached = strategy_cache.read_cache(data_dir) or {} + assert cached.get("results", {}).get("s1", {}).get("total") == 1 + + # UI 保存配置 (字段集可能变化): 默认全清 + store.upsert(cfg) + assert strategy_cache.read_cache(data_dir) is None + + def test_config_field_change_invalidates_sync(data_dir): _mk_config(data_dir) ext_factors.ensure_synced(data_dir)