diff --git a/DEVELOPMENT.md b/DEVELOPMENT.md index 6531d92..7cc455e 100644 --- a/DEVELOPMENT.md +++ b/DEVELOPMENT.md @@ -457,7 +457,8 @@ P1 实现说明: - 时区使用固定 UTC+8(中国无夏令时),避免 Windows 缺少 `tzdata`; - 午夜 `seed_at`(默认 `00:00`)把 `snapshot:last_completed` 的 ok=1 行克隆为当日壳,写入 `snapshot:seeded:=1` / `fans:seeded:=1`,**不**推进完成日; - 也可手动:`python app.py snapshot-seed [--date YYYY-MM-DD]`(Debian / `docker compose exec re-xxm python app.py snapshot-seed`); -- 若全站 `snapshot` 正在跑,或 `last_completed` 暂不可用(`no_source`),seed **不记成功**,约 10 分钟后重试直到真正克隆成功; +- 目标日已有 `snapshot:completed:` 时 seed 记 `already_completed` 成功(即使 `last_completed` 已等于当日),**不得**误判 `no_source`; +- 若全站 `snapshot` 正在跑,或目标日未完成且 `last_completed` 暂不可用(`no_source`),seed **不记成功**,约 10 分钟后重试直到真正克隆成功; - 晚间 `snapshot_at`(默认 `23:15`)对 seeded 日 force 重采;成功后写 `snapshot:completed:=1`、推进 `last_completed` 并清除 seeded; - 当日全部视频与粉丝采集成功后在 `kv` 写入 `snapshot:completed:=1`;启动时未完成会补跑(纯 seed 日不误标完成); - 完成标记不是唯一门槛:每次检查 `pending_snapshot_aids(date)`,新入库或从 `dropped` 恢复的视频会立即补采当天快照; diff --git a/reaction_board/cli.py b/reaction_board/cli.py index 905f59b..e0a7ae3 100644 --- a/reaction_board/cli.py +++ b/reaction_board/cli.py @@ -730,10 +730,20 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int: return _quiet_window_defer( repo, logger, "candidate_cleanup", started_at=started, ) - removed = repo.delete_expired_candidates( - ttl_days=cfg.schedule.candidate_ttl_days - ) - expired = repo.expire_fav_actions() + try: + removed = repo.delete_expired_candidates( + ttl_days=cfg.schedule.candidate_ttl_days + ) + expired = repo.expire_fav_actions() + except Exception as exc: # noqa: BLE001 记录后约 5 分钟补跑 + finished = int(time.time()) + _record_task( + repo, "candidate_cleanup", ok=False, started_at=started, + finished_at=finished, error=str(exc), + ) + return TaskControl( + ok=False, next_run=_retry_soon(float(finished)), error=str(exc), + ) if removed or expired: repo.bump_cache_version("videos") finished = int(time.time()) diff --git a/reaction_board/scheduler.py b/reaction_board/scheduler.py index fbdbb32..2febe37 100644 --- a/reaction_board/scheduler.py +++ b/reaction_board/scheduler.py @@ -87,11 +87,9 @@ class Scheduler: raw = task.fn() except Exception as exc: # noqa: BLE001 单任务失败不终止调度 error = str(exc) - self._log("error", "scheduler.task_error", task=task.name, error=error) else: if isinstance(raw, TaskControl): control = raw - self._log("info", "scheduler.task_ok", task=task.name) finished = self._clock() if control is not None and control.next_run is not None: task.next_run = float(control.next_run) @@ -104,6 +102,23 @@ class Scheduler: ok = bool(control.ok) and error is None if control.error and error is None: error = control.error + # Skip/fail ≠ success:先定 ok/error,再记 scheduler 日志。 + if ok: + self._log("info", "scheduler.task_ok", task=task.name) + elif control is not None: + self._log( + "warning", + "scheduler.task_fail", + task=task.name, + error=error, + ) + else: + self._log( + "error", + "scheduler.task_error", + task=task.name, + error=error, + ) result = TaskResult( name=task.name, started_at=started, diff --git a/reaction_board/services/snapshot.py b/reaction_board/services/snapshot.py index 97c578c..383f5db 100644 --- a/reaction_board/services/snapshot.py +++ b/reaction_board/services/snapshot.py @@ -538,16 +538,8 @@ class SnapshotService: raise ValueError(f"日期格式必须是 YYYY-MM-DD: {stat_date!r}") summary = SnapshotSummary(stat_date=stat_date, started_at=now) - source = self.repo.kv_get("snapshot:last_completed") - if not source or source >= stat_date: - self._log( - "info", "snapshot.seed_skip", - stat_date=stat_date, reason="no_source", source=source, - ) - summary.finished_at = self._now() - summary.completed = False - summary.reason = "no_source" - return summary + # 目标日已完成:先于 no_source。全站快照当天完成后 + # last_completed == stat_date,不可误判为无源再 10 分钟重试。 if self.repo.kv_get(self.completed_key(stat_date)) == "1": self._log( "info", "snapshot.seed_skip", @@ -558,6 +550,16 @@ class SnapshotService: summary.already_completed = True summary.reason = "already_completed" return summary + source = self.repo.kv_get("snapshot:last_completed") + if not source or source >= stat_date: + self._log( + "info", "snapshot.seed_skip", + stat_date=stat_date, reason="no_source", source=source, + ) + summary.finished_at = self._now() + summary.completed = False + summary.reason = "no_source" + return summary if self.repo.kv_get(self.seeded_key(stat_date)) == "1": self._log( "info", "snapshot.seed_skip", diff --git a/tests/test_cli.py b/tests/test_cli.py index 2fe7070..855bca0 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -479,6 +479,115 @@ def test_defer_snapshot_seed_schedules_ten_minutes(): db.close() + + +def test_snapshot_seed_same_day_completed_succeeds(): + """当日已完成时手动 snapshot-seed 应成功 already_completed,不走 no_source。""" + with tempfile.TemporaryDirectory() as tmp: + path = write_config(tmp, cookie="", csrf="") + code, _ = run_cli(["init", "--config", str(path)]) + assert code == 0 + today = "2026-10-31" + db = Database(Path(tmp) / "app.db") + db.initialize() + try: + repo = Repository(db) + repo.kv_set("snapshot:last_completed", today) + repo.kv_set("snapshot:completed:" + today, "1") + finally: + db.close() + code, output = run_cli([ + "snapshot-seed", "--config", str(path), "--date", today, + ]) + assert code == 0, output + assert '"reason": "already_completed"' in output + assert "no_source" not in output + db = Database(Path(tmp) / "app.db") + db.initialize() + try: + repo = Repository(db) + last = json.loads(repo.kv_get("task:last:snapshot_seed")) + assert last["ok"] is True + finally: + db.close() + + +def test_candidate_cleanup_failure_retries_soon(): + """candidate_cleanup 异常应记失败任务并约 5 分钟补跑,而非落到 86400 间隔。""" + class FakeClient: + def check_login(self): + return LoginState(True, 1, "tester", 0, "OK") + + class FakeDiscovery: + def __init__(self, *args, **kwargs): + pass + + def run(self): + from reaction_board.services.discovery import DiscoverySummary + return DiscoverySummary(started_at=1, finished_at=1) + + with tempfile.TemporaryDirectory() as tmp: + path = write_config(tmp) + data = json.loads(path.read_text(encoding="utf-8")) + data["folders"] = [] + data["track_fans"] = [] + path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") + + original_client = cli._client + original_discovery = cli.DiscoveryService + original_delete = Repository.delete_expired_candidates + captured = {} + + def boom(self, *args, **kwargs): + raise RuntimeError("cleanup db boom") + + def fake_client(credentials, *args, **kwargs): + return FakeClient() + + # 只跑 cleanup:把其它任务 next_run 推远;通过 Scheduler 包装拦截。 + original_scheduler = cli.Scheduler + + class CaptureScheduler(original_scheduler): + def __init__(self, tasks, **kwargs): + far = 1_000_000_000_000.0 + for task in tasks: + if task.name != "candidate_cleanup": + task.next_run = far + else: + task.next_run = 0.0 + super().__init__(tasks, **kwargs) + captured["scheduler"] = self + + cli._client = fake_client + cli.DiscoveryService = FakeDiscovery + cli.Scheduler = CaptureScheduler + Repository.delete_expired_candidates = boom + try: + code, output = run_cli(["run", "--once", "--config", str(path)]) + finally: + cli._client = original_client + cli.DiscoveryService = original_discovery + cli.Scheduler = original_scheduler + Repository.delete_expired_candidates = original_delete + + assert code == 0, output + db = Database(Path(tmp) / "app.db") + db.initialize() + try: + repo = Repository(db) + last = json.loads(repo.kv_get("task:last:candidate_cleanup")) + assert last["ok"] is False + assert "cleanup db boom" in last["error"] + finally: + db.close() + scheduler = captured["scheduler"] + cleanup_task = next(t for t in scheduler.tasks if t.name == "candidate_cleanup") + # next_run ≈ finished + 300;允许几秒时钟误差 + assert cleanup_task.next_run > 0 + delta = cleanup_task.next_run - __import__("time").time() + assert 250 <= delta <= 360, delta + + def main_test(): test_init_and_status() test_missing_config() @@ -496,7 +605,9 @@ def main_test(): test_snapshot_seed_cli_clones_offline() test_snapshot_seed_defers_when_snapshot_running() test_snapshot_seed_no_source_is_non_success() + test_snapshot_seed_same_day_completed_succeeds() test_defer_snapshot_seed_schedules_ten_minutes() + test_candidate_cleanup_failure_retries_soon() print("OK: CLI init/sync/snapshot/snapshot-seed/auth-check/run/status/backfill/错误路径测试全部通过") diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py index e2888df..a06e817 100644 --- a/tests/test_scheduler.py +++ b/tests/test_scheduler.py @@ -152,10 +152,87 @@ def test_task_control_overrides_ok_and_next_run(): assert task.next_run == 2_000.0 + + +class _CaptureLogger: + def __init__(self): + self.events = [] + + def info(self, event, **fields): + self.events.append(("info", event, fields)) + + def warning(self, event, **fields): + self.events.append(("warning", event, fields)) + + def error(self, event, **fields): + self.events.append(("error", event, fields)) + + +def test_task_control_ok_false_logs_fail_not_ok(): + """TaskControl(ok=False) 不得记 scheduler.task_ok(Skip ≠ success)。""" + clock = [1_000.0] + logger = _CaptureLogger() + task = ScheduledTask( + "quiet", + 60, + lambda: TaskControl(ok=False, next_run=1_500.0, error="snapshot_quiet_window"), + ) + scheduler = Scheduler([task], clock=lambda: clock[0], logger=logger) + results = scheduler.run_due() + assert results[0].ok is False + assert results[0].error == "snapshot_quiet_window" + assert ("info", "scheduler.task_ok", {"task": "quiet"}) not in logger.events + fail_events = [e for e in logger.events if e[1] == "scheduler.task_fail"] + assert len(fail_events) == 1 + assert fail_events[0][0] == "warning" + assert fail_events[0][2]["task"] == "quiet" + assert fail_events[0][2]["error"] == "snapshot_quiet_window" + + +def test_task_control_ok_true_still_logs_task_ok(): + clock = [1_000.0] + logger = _CaptureLogger() + task = ScheduledTask( + "ok_task", + 60, + lambda: TaskControl(ok=True, next_run=1_200.0), + ) + scheduler = Scheduler([task], clock=lambda: clock[0], logger=logger) + results = scheduler.run_due() + assert results[0].ok is True + assert ("info", "scheduler.task_ok", {"task": "ok_task"}) in logger.events + assert not any(e[1] == "scheduler.task_fail" for e in logger.events) + + +def test_raised_exception_logs_task_fail(): + clock = [100.0] + logger = _CaptureLogger() + + def boom(): + raise RuntimeError("db down") + + scheduler = Scheduler( + [ScheduledTask("bad", 10, boom)], + clock=lambda: clock[0], + logger=logger, + ) + results = scheduler.run_due() + assert results[0].ok is False + assert results[0].error == "db down" + assert not any(e[1] == "scheduler.task_ok" for e in logger.events) + assert not any(e[1] == "scheduler.task_fail" for e in logger.events) + err_events = [e for e in logger.events if e[1] == "scheduler.task_error"] + assert len(err_events) == 1 + assert err_events[0][0] == "error" + assert err_events[0][2]["error"] == "db down" + def main(): test_run_when_due_and_reschedule() test_next_run_fn_controls_reschedule() test_task_control_overrides_ok_and_next_run() + test_task_control_ok_false_logs_fail_not_ok() + test_task_control_ok_true_still_logs_task_ok() + test_raised_exception_logs_task_fail() test_task_error_does_not_block_other_tasks() test_run_forever_with_stop_event() test_max_iterations_and_validation() diff --git a/tests/test_snapshot.py b/tests/test_snapshot.py index 7769bfe..6935ba8 100644 --- a/tests/test_snapshot.py +++ b/tests/test_snapshot.py @@ -537,6 +537,36 @@ def test_full_snapshot_skips_ok_rows_including_hourly_refresh(): +def test_seed_same_day_completed_is_already_completed_not_no_source(): + """全站完成后 last_completed==target:应 already_completed,不误判 no_source。""" + clock = 1_800_000_000 + with tempfile.TemporaryDirectory() as tmp: + transport = SnapshotTransport({1: view_for(1)}) + db, repo, service = make_service(tmp, transport, track_fans=()) + try: + add_video(repo, 1) + today = "2026-10-31" + repo.upsert_snapshot( + today, 1, + metrics={"view": 10, "like": 1, "coin": 0, "favorite": 0, + "share": 0, "reply": 0, "danmaku": 0}, + ok=True, err=None, captured_at=clock, + ) + # 正常收尾态:last_completed 已推进到当日,且 completed 标记在位 + repo.kv_set("snapshot:last_completed", today) + repo.kv_set(service.completed_key(today), "1") + + summary = service.run_seed_from_last_completed(date=today) + assert summary.reason == "already_completed" + assert summary.already_completed is True + assert summary.completed is True + assert summary.reason != "no_source" + assert transport.count("/web-interface/view") == 0 + finally: + db.close() + + + def test_seed_from_last_completed_does_not_mark_completed(): clock = 1_800_000_000 with tempfile.TemporaryDirectory() as tmp: @@ -646,6 +676,7 @@ def main(): test_backup_failure_retries_without_api() test_run_recent_videos_refreshes_without_completion_markers() test_full_snapshot_skips_ok_rows_including_hourly_refresh() + test_seed_same_day_completed_is_already_completed_not_no_source() test_seed_from_last_completed_does_not_mark_completed() test_seeded_day_skips_without_force_and_seals_with_force() print("OK: 每日快照、断点续跑、失败占位与粉丝采集测试全部通过")