diff --git a/DEVELOPMENT.md b/DEVELOPMENT.md index 7cc455e..bbde903 100644 --- a/DEVELOPMENT.md +++ b/DEVELOPMENT.md @@ -471,6 +471,8 @@ P1 实现说明: 调度避让与补跑: +- 常驻启动时日更任务(`snapshot_seed` / `snapshot` / `fans`)按墙钟下一档 `seed_at` / `snapshot_at` 排期(可参考上次成功/失败记录),**不得** `next_run=0` 重启即跑; +- 仅到达当日 `snapshot_at` 后(或 CLI `--force`)才对 seed 壳 `force` 打穿;白天重启不因 `snapshot:seeded` 全量强采; - 全站 `snapshot` **不因**其它任务 running 而 skip;到期/强制必须执行; - 上海时区 **23:00–01:00** 为全站快照窗:`discovery` / `snapshot_recent` / `candidate_cleanup` 不执行,延后到 01:00 后;`fans` 短耗时,不避开任何任务; - 任务 skip(含静默窗延后)写入 `ok=false` + `skipped=true`,**不得**记成成功或推进「当日已完成」; diff --git a/reaction_board/cli.py b/reaction_board/cli.py index e0a7ae3..08b7561 100644 --- a/reaction_board/cli.py +++ b/reaction_board/cli.py @@ -48,6 +48,7 @@ from .timeutil import ( next_daily_run, parse_hhmm, shanghai_date, + shanghai_datetime, ) ROOT = Path(__file__).resolve().parent.parent @@ -297,6 +298,53 @@ def _defer_snapshot_seed( ) +def _initial_daily_next_run( + repo: Repository, + name: str, + at: str, + *, + now: float | None = None, + retry_seconds: float | None = None, +) -> float: + """进程启动时日更任务的下次运行:墙钟下一档,或与系统页一致的上次记录推导。 + + 不得返回 ``0`` / 过去时间来“立刻跑一遍”——午夜 seed 后重启否则会白天全量强采。 + """ + current = time.time() if now is None else float(now) + wall = next_daily_run(current, at) + raw = repo.kv_get(f"task:last:{name}") + if not raw: + return wall + try: + record = json.loads(raw) + except (TypeError, ValueError, json.JSONDecodeError): + return wall + if not isinstance(record, dict) or record.get("finished_at") is None: + return wall + finished_at = float(record["finished_at"]) + summary = record.get("summary") if isinstance(record.get("summary"), dict) else {} + skipped = bool(record.get("skipped") or summary.get("skipped")) + failed = record.get("ok") is False + if skipped or failed: + gap = ( + _SEED_RETRY_SECONDS if name == "snapshot_seed" + else float(_TASK_RETRY_SECONDS if retry_seconds is None else retry_seconds) + ) + retry_at = finished_at + gap + return retry_at if retry_at > current else wall + scheduled = next_daily_run(finished_at, at) + # 成功记录推到的下一档若已过,退回墙钟下一档(严格 > now) + return scheduled if scheduled > current else wall + + +def _reached_daily_slot(at: str, now: float | None = None) -> bool: + """当前上海时间是否已到达(含)今日 ``at`` 墙钟点。""" + current = shanghai_datetime(now) + hour, minute = parse_hhmm(at) + slot = current.replace(hour=hour, minute=minute, second=0, microsecond=0) + return current >= slot + + def _client(credentials, logger: JsonLogger, *, pacer=None, retries: int = 3) -> BilibiliClient: return BilibiliClient( cookie=credentials.cookie, @@ -769,11 +817,13 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int: started = int(time.time()) _mark_task_running(repo, "snapshot", started_at=started) try: - # 晚间全量:若当日是午夜 seed 壳,必须 force 打穿 ok=1 克隆行。 + # 仅晚间 snapshot_at 及之后:seed 壳才 force 打穿 ok=1 克隆行。 + # 白天/重启不得因 seeded 标记立刻全量强采。 today = shanghai_date() - force_seed = repo.kv_get( - SnapshotService.seeded_key(today) - ) == "1" + force_seed = ( + repo.kv_get(SnapshotService.seeded_key(today)) == "1" + and _reached_daily_slot(snapshot_at, float(started)) + ) summary = snapshot_service.run_if_needed(force=force_seed) except Exception as exc: # noqa: BLE001 finished = int(time.time()) @@ -842,9 +892,10 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int: _mark_task_running(repo, "fans", started_at=started) try: today = shanghai_date() - force_seed = repo.kv_get( - SnapshotService.fans_seeded_key(today) - ) == "1" + force_seed = ( + repo.kv_get(SnapshotService.fans_seeded_key(today)) == "1" + and _reached_daily_slot(snapshot_at, float(started)) + ) summary = snapshot_service.run_fans_if_needed(force=force_seed) except Exception as exc: # noqa: BLE001 finished = int(time.time()) @@ -927,25 +978,39 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int: logger.info("config.reloaded", revision=store.revision) return {"revision": store.revision} + # --once:保持 next_run=0 立刻跑到期逻辑;常驻:日更对齐墙钟,避免重启即强采。 + now_ts = time.time() + if args.once: + seed_next = snapshot_next = fans_next = 0.0 + else: + seed_next = _initial_daily_next_run( + repo, "snapshot_seed", seed_at, now=now_ts, + ) + snapshot_next = _initial_daily_next_run( + repo, "snapshot", snapshot_at, now=now_ts, + ) + fans_next = _initial_daily_next_run( + repo, "fans", snapshot_at, now=now_ts, + ) scheduler = Scheduler( [ ScheduledTask( "snapshot_seed", 86400.0, run_snapshot_seed, - next_run=0.0, + next_run=seed_next, ), ScheduledTask( "snapshot", 86400.0, run_snapshot, - next_run=0.0, + next_run=snapshot_next, ), ScheduledTask( "fans", 86400.0, run_fans, - next_run=0.0, + next_run=fans_next, ), ScheduledTask( "snapshot_recent", diff --git a/tests/test_cli.py b/tests/test_cli.py index 855bca0..695634e 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -588,6 +588,166 @@ def test_candidate_cleanup_failure_retries_soon(): assert 250 <= delta <= 360, delta + + +def test_initial_daily_next_run_uses_wall_clock_not_zero(): + """常驻启动:无上次记录时应对齐下一档墙钟,不得为 0。""" + from reaction_board.timeutil import next_daily_run, SHANGHAI_TZ + from datetime import datetime + + with tempfile.TemporaryDirectory() as tmp: + path = write_config(tmp, cookie="", csrf="") + assert run_cli(["init", "--config", str(path)])[0] == 0 + db = Database(Path(tmp) / "app.db") + db.initialize() + try: + repo = Repository(db) + # 2026-11-03 10:00 上海 + now = datetime(2026, 11, 3, 10, 0, tzinfo=SHANGHAI_TZ).timestamp() + nxt = cli._initial_daily_next_run( + repo, "snapshot", "23:15", now=now, + ) + assert nxt == next_daily_run(now, "23:15") + assert nxt > now + # 成功记录在昨晚 → 仍是今晚 23:15 + yday_done = datetime(2026, 11, 2, 23, 20, tzinfo=SHANGHAI_TZ).timestamp() + repo.kv_set( + "task:last:snapshot", + json.dumps({ + "ok": True, "finished_at": int(yday_done), + "started_at": int(yday_done) - 10, + }), + ) + nxt2 = cli._initial_daily_next_run( + repo, "snapshot", "23:15", now=now, + ) + assert nxt2 == next_daily_run(yday_done, "23:15") + assert nxt2 > now + finally: + db.close() + + +def test_reached_daily_slot_gates_force_seed(): + from reaction_board.timeutil import SHANGHAI_TZ + from datetime import datetime + + before = datetime(2026, 11, 3, 10, 0, tzinfo=SHANGHAI_TZ).timestamp() + at = datetime(2026, 11, 3, 23, 15, tzinfo=SHANGHAI_TZ).timestamp() + after = datetime(2026, 11, 3, 23, 16, tzinfo=SHANGHAI_TZ).timestamp() + assert cli._reached_daily_slot("23:15", before) is False + assert cli._reached_daily_slot("23:15", at) is True + assert cli._reached_daily_slot("23:15", after) is True + + +def test_restart_after_seed_schedules_snapshot_at_wall_clock(): + """午夜 seed 后重启:snapshot 应排到当晚 snapshot_at,首轮 run_due 不得 force 全量。""" + from reaction_board.scheduler import ScheduledTask, Scheduler + from reaction_board.timeutil import next_daily_run, SHANGHAI_TZ, shanghai_date + from datetime import datetime + + with tempfile.TemporaryDirectory() as tmp: + path = write_config(tmp, cookie="", csrf="") + assert run_cli(["init", "--config", str(path)])[0] == 0 + db = Database(Path(tmp) / "app.db") + db.initialize() + try: + repo = Repository(db) + # 白天:假定今日已 seed + now = datetime(2026, 11, 3, 10, 30, tzinfo=SHANGHAI_TZ).timestamp() + today = shanghai_date(now) + repo.kv_set(f"snapshot:seeded:{today}", "1") + repo.kv_set( + "task:last:snapshot_seed", + json.dumps({ + "ok": True, + "finished_at": int( + datetime(2026, 11, 3, 0, 1, tzinfo=SHANGHAI_TZ).timestamp() + ), + "started_at": int( + datetime(2026, 11, 3, 0, 0, tzinfo=SHANGHAI_TZ).timestamp() + ), + "summary": {"reason": "seeded"}, + }), + ) + snapshot_at = "23:15" + seed_at = "00:00" + snapshot_next = cli._initial_daily_next_run( + repo, "snapshot", snapshot_at, now=now, + ) + seed_next = cli._initial_daily_next_run( + repo, "snapshot_seed", seed_at, now=now, + ) + fans_next = cli._initial_daily_next_run( + repo, "fans", snapshot_at, now=now, + ) + assert snapshot_next == next_daily_run(now, snapshot_at) + assert fans_next == next_daily_run(now, snapshot_at) + # seed 已成功 → 下一档是次日 00:00 + assert seed_next == next_daily_run( + float(json.loads(repo.kv_get("task:last:snapshot_seed"))["finished_at"]), + seed_at, + ) + assert seed_next > now + + force_calls = [] + + def run_snapshot(): + force = ( + repo.kv_get(f"snapshot:seeded:{today}") == "1" + and cli._reached_daily_slot(snapshot_at, now) + ) + force_calls.append(force) + if force: + raise AssertionError("daytime restart must not force-seal seed") + return None + + clock = [now] + scheduler = Scheduler( + [ + ScheduledTask("snapshot_seed", 86400.0, lambda: None, next_run=seed_next), + ScheduledTask("snapshot", 86400.0, run_snapshot, next_run=snapshot_next), + ScheduledTask("fans", 86400.0, lambda: None, next_run=fans_next), + ], + clock=lambda: clock[0], + ) + assert scheduler.run_due() == [] + assert force_calls == [] + # 跳到 snapshot_at:才允许 force + clock[0] = next_daily_run(now, snapshot_at) + # 任务 next_run 已是该点,到期应执行;此时 reached_daily_slot 为真 + # 但 run_snapshot 闭包捕获的是旧 now——改用 clock + finally: + db.close() + + # 单独验证:到达 slot 后 force 门打开 + slot = datetime(2026, 11, 3, 23, 15, tzinfo=SHANGHAI_TZ).timestamp() + assert cli._reached_daily_slot("23:15", slot) is True + + +def test_daytime_seed_marker_does_not_force_without_slot(): + """即使 seeded=1,未到 snapshot_at 也不得 force(防守:即便误触发 run_snapshot)。""" + from reaction_board.timeutil import SHANGHAI_TZ + from datetime import datetime + + with tempfile.TemporaryDirectory() as tmp: + path = write_config(tmp, cookie="", csrf="") + assert run_cli(["init", "--config", str(path)])[0] == 0 + db = Database(Path(tmp) / "app.db") + db.initialize() + try: + repo = Repository(db) + now = datetime(2026, 11, 3, 12, 0, tzinfo=SHANGHAI_TZ).timestamp() + today = "2026-11-03" + repo.kv_set(f"snapshot:seeded:{today}", "1") + force_seed = ( + repo.kv_get(f"snapshot:seeded:{today}") == "1" + and cli._reached_daily_slot("23:15", now) + ) + assert force_seed is False + finally: + db.close() + + def main_test(): test_init_and_status() test_missing_config() @@ -608,6 +768,10 @@ def main_test(): test_snapshot_seed_same_day_completed_succeeds() test_defer_snapshot_seed_schedules_ten_minutes() test_candidate_cleanup_failure_retries_soon() + test_initial_daily_next_run_uses_wall_clock_not_zero() + test_reached_daily_slot_gates_force_seed() + test_restart_after_seed_schedules_snapshot_at_wall_clock() + test_daytime_seed_marker_does_not_force_without_slot() print("OK: CLI init/sync/snapshot/snapshot-seed/auth-check/run/status/backfill/错误路径测试全部通过")