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)