diff --git a/DEVELOPMENT.md b/DEVELOPMENT.md index 766f429..ddc07dd 100644 --- a/DEVELOPMENT.md +++ b/DEVELOPMENT.md @@ -466,6 +466,13 @@ P1 实现说明： - Cookie 登录态通过 `x/web-interface/nav` 检查，失效只告警，不阻塞匿名 `view`； - Telegram 摘要推迟到 P3，P1 先写结构化 JSON 日志和 `kv` 状态。 +调度避让与补跑： + +- 全站 `snapshot` **不因**其它任务 running 而 skip；到期/强制必须执行； +- 上海时区 **23:00–01:00** 为全站快照窗：`discovery` / `snapshot_recent` / `candidate_cleanup` 不执行，延后到 01:00 后；`fans` 短耗时，不避开任何任务； +- 任务 skip（含静默窗延后）写入 `ok=false` + `skipped=true`，**不得**记成成功或推进「当日已完成」； +- skip / fail 后短间隔补跑（约 5 分钟）；日更成功才推到次日 `snapshot_at` / `seed_at`。 + --- ## 8. Telegram 与入库操作 diff --git a/project.md b/project.md index 75f5548..c7ccfce 100644 --- a/project.md +++ b/project.md @@ -800,7 +800,7 @@ secrets.example.json 凭据样例（auth 回写目标） **进程结构** ``` -scheduler 线程：discovery(每 60min) / snapshot_seed(每日 00:00) / snapshot+fans(每日 23:15) / snapshot_recent(小时) / candidate_cleanup(每日) +scheduler 线程：discovery(每 discover_minutes，避开 23:00–01:00) / snapshot_seed(每日 00:00) / snapshot(每日 23:15，不因其它任务 skip) / fans(每日 23:15，不避开) / snapshot_recent(小时，避开 23:00–01:00) / candidate_cleanup(每日，避开 23:00–01:00) web 线程：ThreadingHTTPServer :8080 telegram：getUpdates long polling（回调 + 私聊文本入库） storage：SQLite WAL，单写线程 diff --git a/reaction_board/cli.py b/reaction_board/cli.py index 90d996c..0f778c9 100644 --- a/reaction_board/cli.py +++ b/reaction_board/cli.py @@ -33,13 +33,20 @@ from .database import Database from .logging_json import JsonLogger from .notify.telegram import TelegramClient from .repository import Repository -from .scheduler import ScheduledTask, Scheduler +from .scheduler import ScheduledTask, Scheduler, TaskControl from .services.discovery import DiscoveryService from .services.library import LibraryService from .services.snapshot import SnapshotService from .services.telegram_poller import TelegramPoller from .services.up_profile import UpProfileService -from .timeutil import next_daily_run, parse_hhmm, shanghai_date +from .timeutil import ( + in_snapshot_quiet_window, + interval_next_run, + next_after_snapshot_quiet_window, + next_daily_run, + parse_hhmm, + shanghai_date, +) ROOT = Path(__file__).resolve().parent.parent DEFAULT_CONFIG = default_config_path(ROOT) @@ -163,13 +170,6 @@ def _clear_task_running(repo: Repository, name: str) -> None: pass -def _is_task_running(repo: Repository, name: str) -> bool: - try: - return bool(repo.kv_get(_task_running_key(name))) - except Exception: # noqa: BLE001 - return False - - def _record_task( repo: Repository, name: str, @@ -179,12 +179,19 @@ def _record_task( finished_at: int | None = None, payload=None, error: str | None = None, + skipped: bool = False, ) -> None: - """把最近一次任务结果写入 kv，并清除 running 标记。""" + """把最近一次任务结果写入 kv，并清除 running 标记。 + + ``skipped=True`` 表示主动延后/避让，不得记成成功，也不推进「当日已完成」。 + """ now = int(time.time()) + if skipped: + ok = False record = { "name": name, "ok": bool(ok), + "skipped": bool(skipped), "started_at": int(started_at if started_at is not None else now), "finished_at": int(finished_at if finished_at is not None else now), "error": error, @@ -197,6 +204,42 @@ def _record_task( _clear_task_running(repo, name) +# 全站快照 / 粉丝失败后的补跑间隔（秒） +_TASK_RETRY_SECONDS = 300.0 +_QUIET_SKIP_REASON = "snapshot_quiet_window" + + +def _quiet_window_defer( + repo: Repository, + logger: JsonLogger, + name: str, + *, + started_at: int, +) -> TaskControl: + """静默窗内不执行：记 skip（非成功），并排到 01:00 之后。""" + finished = int(time.time()) + logger.info(f"{name}.skip", reason=_QUIET_SKIP_REASON) + _record_task( + repo, + name, + ok=False, + skipped=True, + started_at=started_at, + finished_at=finished, + payload={"skipped": True, "reason": _QUIET_SKIP_REASON}, + error=_QUIET_SKIP_REASON, + ) + return TaskControl( + ok=False, + next_run=next_after_snapshot_quiet_window(float(finished)), + error=_QUIET_SKIP_REASON, + ) + + +def _retry_soon(now: float | None = None) -> float: + return (time.time() if now is None else now) + _TASK_RETRY_SECONDS + + def _client(credentials, logger: JsonLogger, *, pacer=None, retries: int = 3) -> BilibiliClient: return BilibiliClient( cookie=credentials.cookie, @@ -504,66 +547,83 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int: logger=logger, ) + hh, mm = parse_hhmm(cfg.schedule.snapshot_at) + snapshot_at = f"{hh:02d}:{mm:02d}" + sh, sm = parse_hhmm(cfg.schedule.seed_at) + seed_at = f"{sh:02d}:{sm:02d}" + recent_minutes = int(store.get().schedule.recent_snapshot_minutes) + discover_minutes = int(store.get().schedule.discover_minutes) + def run_discover(): started = int(time.time()) + if in_snapshot_quiet_window(float(started)): + return _quiet_window_defer( + repo, logger, "discovery", started_at=started, + ) _mark_task_running(repo, "discovery", started_at=started) + interval = float(store.get().schedule.discover_minutes * 60) try: summary = discovery_service.run() - except Exception as exc: # noqa: BLE001 记录后交给调度器隔离 + except Exception as exc: # noqa: BLE001 记录后补跑，不抛给次日 + finished = int(time.time()) _record_task( repo, "discovery", ok=False, started_at=started, - finished_at=int(time.time()), error=str(exc), + finished_at=finished, error=str(exc), + ) + return TaskControl( + ok=False, next_run=_retry_soon(float(finished)), error=str(exc), ) - raise payload = summary.as_dict() + ok = not summary.errors + finished = int(summary.finished_at or time.time()) _record_task( - repo, "discovery", ok=not summary.errors, + repo, "discovery", ok=ok, started_at=summary.started_at, - finished_at=summary.finished_at or int(time.time()), + finished_at=finished, payload=payload, error="；".join(summary.errors) if summary.errors else None, ) last["discover"] = summary logger.info("discovery.summary", **payload) - return summary + next_run = ( + interval_next_run(float(finished), interval) if ok + else _retry_soon(float(finished)) + ) + return TaskControl(ok=ok, next_run=next_run) def run_cleanup(): started = int(time.time()) + if in_snapshot_quiet_window(float(started)): + 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() if removed or expired: repo.bump_cache_version("videos") + finished = int(time.time()) logger.info( "cleanup.done", candidates_removed=removed, fav_actions_expired=expired, ) _record_task( repo, "candidate_cleanup", ok=True, started_at=started, - finished_at=int(time.time()), + finished_at=finished, payload={ "candidates_removed": removed, "fav_actions_expired": expired, }, ) - return { - "candidates_removed": removed, - "fav_actions_expired": expired, - } + return TaskControl( + ok=True, + next_run=interval_next_run(float(finished), 86400.0), + ) def run_snapshot(): + """全站快照：不因其它任务 running 而 skip；失败/未完成则短间隔补跑。""" started = int(time.time()) - if _is_task_running(repo, "snapshot_recent"): - logger.info( - "snapshot.skip", reason="snapshot_recent_running", - ) - _record_task( - repo, "snapshot", ok=True, started_at=started, - finished_at=int(time.time()), - payload={"skipped": True, "reason": "snapshot_recent_running"}, - ) - return None _mark_task_running(repo, "snapshot", started_at=started) try: # 晚间全量：若当日是午夜 seed 壳，必须 force 打穿 ok=1 克隆行。 @@ -573,56 +633,68 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int: ) == "1" summary = snapshot_service.run_if_needed(force=force_seed) except Exception as exc: # noqa: BLE001 + finished = int(time.time()) _record_task( repo, "snapshot", ok=False, started_at=started, - finished_at=int(time.time()), error=str(exc), + finished_at=finished, error=str(exc), + ) + return TaskControl( + ok=False, next_run=_retry_soon(float(finished)), error=str(exc), ) - raise payload = summary.as_dict() ok = bool(summary.completed and not summary.backup_error) + finished = int(summary.finished_at or time.time()) _record_task( repo, "snapshot", ok=ok, started_at=summary.started_at, - finished_at=summary.finished_at or int(time.time()), + finished_at=finished, payload=payload, error=summary.backup_error, ) last["snapshot"] = summary logger.info("snapshot.summary", **payload) - return summary + next_run = ( + next_daily_run(float(finished), snapshot_at) if ok + else _retry_soon(float(finished)) + ) + return TaskControl(ok=ok, next_run=next_run) def run_snapshot_recent(): started = int(time.time()) - if _is_task_running(repo, "snapshot"): - logger.info( - "snapshot.recent_skip", reason="snapshot_running", + if in_snapshot_quiet_window(float(started)): + return _quiet_window_defer( + repo, logger, "snapshot_recent", started_at=started, ) - _record_task( - repo, "snapshot_recent", ok=True, started_at=started, - finished_at=int(time.time()), - payload={"skipped": True, "reason": "snapshot_running"}, - ) - return None _mark_task_running(repo, "snapshot_recent", started_at=started) + interval = float(store.get().schedule.recent_snapshot_minutes * 60) try: summary = snapshot_service.run_recent_videos() except Exception as exc: # noqa: BLE001 + finished = int(time.time()) _record_task( repo, "snapshot_recent", ok=False, started_at=started, - finished_at=int(time.time()), error=str(exc), + finished_at=finished, error=str(exc), + ) + return TaskControl( + ok=False, next_run=_retry_soon(float(finished)), error=str(exc), ) - raise payload = summary.as_dict() ok = summary.failed == 0 error = "；".join(summary.errors) if summary.errors else None + finished = int(summary.finished_at or time.time()) _record_task( repo, "snapshot_recent", ok=ok, started_at=summary.started_at, - finished_at=summary.finished_at or int(time.time()), + finished_at=finished, payload=payload, error=error, ) last["snapshot_recent"] = summary logger.info("snapshot.recent_summary", **payload) - return summary + next_run = ( + interval_next_run(float(finished), interval) if ok + else _retry_soon(float(finished)) + ) + return TaskControl(ok=ok, next_run=next_run) def run_fans(): + """粉丝更新短耗时：不避开任何任务，可与全量并行。""" started = int(time.time()) _mark_task_running(repo, "fans", started_at=started) try: @@ -632,24 +704,32 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int: ) == "1" summary = snapshot_service.run_fans_if_needed(force=force_seed) except Exception as exc: # noqa: BLE001 + finished = int(time.time()) _record_task( repo, "fans", ok=False, started_at=started, - finished_at=int(time.time()), error=str(exc), + finished_at=finished, error=str(exc), + ) + return TaskControl( + ok=False, next_run=_retry_soon(float(finished)), error=str(exc), ) - raise payload = summary.as_dict() ok = summary.fans_failed == 0 error = "；".join( err for err in summary.errors if str(err).startswith("fans ") ) or None + finished = int(summary.finished_at or time.time()) _record_task( repo, "fans", ok=ok, started_at=summary.started_at, - finished_at=summary.finished_at or int(time.time()), + finished_at=finished, payload=payload, error=error, ) last["fans"] = summary logger.info("fans.summary", **payload) - return summary + next_run = ( + next_daily_run(float(finished), snapshot_at) if ok + else _retry_soon(float(finished)) + ) + return TaskControl(ok=ok, next_run=next_run) def run_snapshot_seed(): started = int(time.time()) @@ -657,20 +737,26 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int: try: summary = snapshot_service.run_seed_from_last_completed() except Exception as exc: # noqa: BLE001 + finished = int(time.time()) _record_task( repo, "snapshot_seed", ok=False, started_at=started, - finished_at=int(time.time()), error=str(exc), + finished_at=finished, error=str(exc), + ) + return TaskControl( + ok=False, next_run=_retry_soon(float(finished)), error=str(exc), ) - raise payload = summary.as_dict() + finished = int(summary.finished_at or time.time()) _record_task( repo, "snapshot_seed", ok=True, started_at=summary.started_at, - finished_at=summary.finished_at or int(time.time()), + finished_at=finished, payload=payload, ) last["snapshot_seed"] = summary logger.info("snapshot.seed_summary", **payload) - return summary + return TaskControl( + ok=True, next_run=next_daily_run(float(finished), seed_at), + ) def run_config_watch(): before = store.revision @@ -679,11 +765,6 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int: logger.info("config.reloaded", revision=store.revision) return {"revision": store.revision} - hh, mm = parse_hhmm(cfg.schedule.snapshot_at) - snapshot_at = f"{hh:02d}:{mm:02d}" - sh, sm = parse_hhmm(cfg.schedule.seed_at) - seed_at = f"{sh:02d}:{sm:02d}" - recent_minutes = int(store.get().schedule.recent_snapshot_minutes) scheduler = Scheduler( [ ScheduledTask( @@ -691,21 +772,18 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int: 86400.0, run_snapshot_seed, next_run=0.0, - next_run_fn=lambda started: next_daily_run(started, seed_at), ), ScheduledTask( "snapshot", 86400.0, run_snapshot, next_run=0.0, - next_run_fn=lambda started: next_daily_run(started, snapshot_at), ), ScheduledTask( "fans", 86400.0, run_fans, next_run=0.0, - next_run_fn=lambda started: next_daily_run(started, snapshot_at), ), ScheduledTask( "snapshot_recent", @@ -714,7 +792,7 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int: ), ScheduledTask( "discovery", - float(cfg.schedule.discover_minutes * 60), + float(discover_minutes * 60), run_discover, ), ScheduledTask("candidate_cleanup", 86400.0, run_cleanup), @@ -722,6 +800,7 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int: ], logger=logger, ) + if serve_web and not args.once: from .web.server import create_server, serve_forever diff --git a/reaction_board/scheduler.py b/reaction_board/scheduler.py index 03a2b1c..fbdbb32 100644 --- a/reaction_board/scheduler.py +++ b/reaction_board/scheduler.py @@ -3,6 +3,9 @@ P0 只注册收藏夹同步任务；P1/P3 会在同一调度器上追加快照、粉丝和发现任务。 调度只负责“什么时候运行”，具体任务保持幂等，并在自己的服务层处理重试。 + +任务 ``fn`` 可返回 :class:`TaskControl` 以覆盖 ``ok`` / ``next_run`` +（用于静默窗延后、失败补跑等）；未返回时仍按 ``next_run_fn`` / ``interval`` 排期。 """ from __future__ import annotations @@ -12,6 +15,15 @@ from dataclasses import dataclass from typing import Any, Callable, Sequence +@dataclass +class TaskControl: + """任务可选返回值：覆盖成功标记与下次运行时间。""" + + ok: bool = True + next_run: float | None = None + error: str | None = None + + @dataclass class ScheduledTask: name: str @@ -70,23 +82,33 @@ class Scheduler: continue started = current error: str | None = None + control: TaskControl | None = None try: - task.fn() + 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 task.next_run_fn is not None: + if control is not None and control.next_run is not None: + task.next_run = float(control.next_run) + elif task.next_run_fn is not None: task.next_run = task.next_run_fn(started) else: task.next_run = started + task.interval + ok = error is None + if control is not None: + ok = bool(control.ok) and error is None + if control.error and error is None: + error = control.error result = TaskResult( name=task.name, started_at=started, finished_at=finished, - ok=error is None, + ok=ok, error=error, ) results.append(result) diff --git a/reaction_board/timeutil.py b/reaction_board/timeutil.py index 74e02e9..30c234a 100644 --- a/reaction_board/timeutil.py +++ b/reaction_board/timeutil.py @@ -57,3 +57,39 @@ def day_bounds(date: str) -> tuple[float, float]: raise ValueError(f"日期格式必须是 YYYY-MM-DD: {date!r}") start = datetime.strptime(date, "%Y-%m-%d").replace(tzinfo=SHANGHAI_TZ) return start.timestamp(), (start + timedelta(days=1)).timestamp() + + +# 全站快照静默窗：上海时区 [23:00, 01:00)。此间 discovery / recent / cleanup 不执行。 +SNAPSHOT_QUIET_START_HOUR = 23 +SNAPSHOT_QUIET_END_HOUR = 1 + + +def in_snapshot_quiet_window(ts: float | None = None) -> bool: + """是否处于全站快照静默窗 [23:00, 01:00)（上海时区，跨日）。""" + dt = shanghai_datetime(ts) + return dt.hour >= SNAPSHOT_QUIET_START_HOUR or dt.hour < SNAPSHOT_QUIET_END_HOUR + + +def next_after_snapshot_quiet_window(ts: float) -> float: + """若 ``ts`` 落在静默窗内，推到窗结束后的 01:00；否则原样返回。""" + if not in_snapshot_quiet_window(ts): + return ts + dt = shanghai_datetime(ts) + if dt.hour >= SNAPSHOT_QUIET_START_HOUR: + target = (dt + timedelta(days=1)).replace( + hour=SNAPSHOT_QUIET_END_HOUR, minute=0, second=0, microsecond=0, + ) + else: + target = dt.replace( + hour=SNAPSHOT_QUIET_END_HOUR, minute=0, second=0, microsecond=0, + ) + if target.timestamp() <= ts: + target += timedelta(days=1) + return target.timestamp() + + +def interval_next_run(now: float, interval: float) -> float: + """``now + interval``，若落入静默窗则推到 01:00 之后。""" + if interval <= 0: + raise ValueError("interval 必须大于 0") + return next_after_snapshot_quiet_window(now + interval) diff --git a/reaction_board/web/queries.py b/reaction_board/web/queries.py index 67a5c13..baeea07 100644 --- a/reaction_board/web/queries.py +++ b/reaction_board/web/queries.py @@ -34,7 +34,14 @@ from ..analytics import ( ) from ..config import ConfigStore from ..database import Database -from ..timeutil import day_bounds, next_daily_run, shanghai_date, shanghai_datetime +from ..timeutil import ( + day_bounds, + interval_next_run, + next_after_snapshot_quiet_window, + next_daily_run, + shanghai_date, + shanghai_datetime, +) from .cache import DASHBOARD_CACHE, TTL_FANS, TTL_VIDEOS, build_cache_key TRACKED_STATUSES = ("active", "unfaved") @@ -1971,9 +1978,10 @@ class DashboardQueries: "fans": int(next_daily_run(now, cfg.schedule.snapshot_at)), "snapshot": int(next_daily_run(now, cfg.schedule.snapshot_at)), "snapshot_seed": int(next_daily_run(now, cfg.schedule.seed_at)), - "discover": int(now + discover_interval), - "snapshot_recent": int(now + recent_interval), + "discover": int(interval_next_run(now, float(discover_interval))), + "snapshot_recent": int(interval_next_run(now, float(recent_interval))), } + retry_seconds = 300 for name, key, at in ( ("fans", "fans", cfg.schedule.snapshot_at), ("snapshot", "snapshot", cfg.schedule.snapshot_at), @@ -1985,19 +1993,41 @@ class DashboardQueries: if not (isinstance(record, dict) and record.get("finished_at")): continue finished_at = int(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 name in {"discover", "snapshot_recent"}: - # 只按上次完成推下一档；进行中由 running 展示，不按墙钟抢跑。 + # 上次完成推下一档；skip/失败短补跑；静默窗推到 01:00 后。 interval = ( recent_interval if name == "snapshot_recent" else discover_interval ) - next_runs[name] = finished_at + interval + if skipped: + next_runs[name] = int( + next_after_snapshot_quiet_window(float(finished_at)) + ) + elif failed: + next_runs[name] = int(finished_at + retry_seconds) + else: + next_runs[name] = int( + interval_next_run(float(finished_at), float(interval)) + ) else: - next_runs[name] = int(next_daily_run(finished_at, at)) + if skipped or failed: + # 日更任务失败/skip：短间隔补跑，绝不按成功推到次日。 + next_runs[name] = int(finished_at + retry_seconds) + else: + next_runs[name] = int(next_daily_run(finished_at, at)) errors = [ {"task": name, **record} for name, record in tasks.items() - if isinstance(record, dict) and record.get("ok") is False + if isinstance(record, dict) + and record.get("ok") is False + and not record.get("skipped") + and not ( + isinstance(record.get("summary"), dict) + and record["summary"].get("skipped") + ) ] counts = {str(row["status"]): int(row["n"]) for row in conn.execute( "SELECT status, COUNT(*) AS n FROM video GROUP BY status" diff --git a/reaction_board/web/static/views/system.js b/reaction_board/web/static/views/system.js index f4920d7..1145244 100644 --- a/reaction_board/web/static/views/system.js +++ b/reaction_board/web/static/views/system.js @@ -5,6 +5,15 @@ import { escapeHtml, formatCount, formatDateTime, formatNumber, } from "../format.js"; +function taskResultLabel(record) { + const summary = record.summary && typeof record.summary === "object" ? record.summary : {}; + if (record.skipped || summary.skipped) { + return `跳过`; + } + if (record.ok) return "成功"; + return `失败`; +} + function taskRows(tasks) { const entries = Object.entries(tasks || {}); if (!entries.length) return emptyState("暂无任务记录", "运行一次同步、快照或发现后会写入最近结果。"); @@ -15,12 +24,16 @@ function taskRows(tasks) { ${entries.map(([name, record]) => { const summary = record.summary && typeof record.summary === "object" ? Object.entries(record.summary) - .filter(([key]) => ["added", "removed", "candidates", "notified", "ok", "failed", "total", "completed", "backup_error"].includes(key)) + .filter(([key]) => [ + "added", "removed", "candidates", "notified", "ok", "failed", + "total", "completed", "backup_error", "skipped", "already_completed", + "reason", "elapsed", "success", + ].includes(key)) .map(([key, value]) => `${key}=${value}`).join(" ") : ""; return ` ${escapeHtml(name)} - ${record.ok ? "成功" : `失败`} + ${taskResultLabel(record)} ${escapeHtml(formatDateTime(record.finished_at))} ${escapeHtml(summary || record.error || "—")} `; @@ -100,7 +113,7 @@ export async function renderSystem(root) {
新视频小时同步${running.snapshot_recent ? "运行中" : escapeHtml(formatDateTime(next.snapshot_recent))}
新视频搜索${running.discover || running.discovery ? "运行中" : escapeHtml(formatDateTime(next.discover))}
-

进行中显示「运行中」；空闲时按该任务上次完成时间推算下次运行，不按墙钟抢跑。新视频小时同步与全站快照互斥。

+

进行中显示「运行中」；空闲时按该任务上次完成时间推算下次运行。23:00–01:00（上海）为全站快照窗：发现 / 小时同步 / 候选清理会延后到 01:00 后；粉丝更新不避开。跳过或失败不会记成成功，会短间隔补跑。

diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py index 6376a64..e2888df 100644 --- a/tests/test_scheduler.py +++ b/tests/test_scheduler.py @@ -8,7 +8,7 @@ import threading sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) sys.stdout.reconfigure(encoding="utf-8") -from reaction_board.scheduler import ScheduledTask, Scheduler # noqa: E402 +from reaction_board.scheduler import ScheduledTask, Scheduler, TaskControl # noqa: E402 def test_run_when_due_and_reschedule(): @@ -128,9 +128,34 @@ def test_next_run_fn_controls_reschedule(): assert calls == [1_000.0, 1_123.0] + +def test_task_control_overrides_ok_and_next_run(): + clock = [1_000.0] + task = ScheduledTask( + "defer", + 60, + lambda: TaskControl(ok=False, next_run=1_500.0, error="quiet"), + ) + scheduler = Scheduler([task], clock=lambda: clock[0]) + results = scheduler.run_due() + assert len(results) == 1 + assert results[0].ok is False + assert results[0].error == "quiet" + assert task.next_run == 1_500.0 + # 未到 next_run 不重跑 + clock[0] = 1_499.0 + assert scheduler.run_due() == [] + clock[0] = 1_500.0 + task.fn = lambda: TaskControl(ok=True, next_run=2_000.0) + results = scheduler.run_due() + assert results[0].ok is True + assert task.next_run == 2_000.0 + + def main(): test_run_when_due_and_reschedule() test_next_run_fn_controls_reschedule() + test_task_control_overrides_ok_and_next_run() test_task_error_does_not_block_other_tasks() test_run_forever_with_stop_event() test_max_iterations_and_validation() diff --git a/tests/test_timeutil.py b/tests/test_timeutil.py index bfe4060..7a5bd20 100644 --- a/tests/test_timeutil.py +++ b/tests/test_timeutil.py @@ -11,7 +11,10 @@ sys.stdout.reconfigure(encoding="utf-8") from reaction_board.timeutil import ( # noqa: E402 SHANGHAI_TZ, day_bounds, + in_snapshot_quiet_window, + interval_next_run, is_date, + next_after_snapshot_quiet_window, next_daily_run, parse_hhmm, shanghai_date, @@ -72,11 +75,40 @@ def test_day_bounds_and_date_validation(): raise AssertionError("非法日期应失败") + +def test_snapshot_quiet_window(): + assert in_snapshot_quiet_window(ts("2026-11-02 23:00:00")) + assert in_snapshot_quiet_window(ts("2026-11-02 23:59:59")) + assert in_snapshot_quiet_window(ts("2026-11-03 00:00:00")) + assert in_snapshot_quiet_window(ts("2026-11-03 00:59:59")) + assert not in_snapshot_quiet_window(ts("2026-11-03 01:00:00")) + assert not in_snapshot_quiet_window(ts("2026-11-02 22:59:59")) + assert not in_snapshot_quiet_window(ts("2026-11-02 12:00:00")) + + assert next_after_snapshot_quiet_window(ts("2026-11-02 22:00:00")) == ts( + "2026-11-02 22:00:00" + ) + assert next_after_snapshot_quiet_window(ts("2026-11-02 23:15:00")) == ts( + "2026-11-03 01:00:00" + ) + assert next_after_snapshot_quiet_window(ts("2026-11-03 00:30:00")) == ts( + "2026-11-03 01:00:00" + ) + # 22:50 + 60min = 23:50 → 推到次日 01:00 + assert interval_next_run(ts("2026-11-02 22:50:00"), 3600) == ts( + "2026-11-03 01:00:00" + ) + assert interval_next_run(ts("2026-11-02 12:00:00"), 3600) == ts( + "2026-11-02 13:00:00" + ) + + def main(): test_shanghai_date_boundary() test_parse_hhmm() test_next_daily_run() test_day_bounds_and_date_validation() + test_snapshot_quiet_window() print("OK: 上海时区与每日时间计算测试全部通过") diff --git a/tests/test_web_queries.py b/tests/test_web_queries.py index fbb185c..f0b6b05 100644 --- a/tests/test_web_queries.py +++ b/tests/test_web_queries.py @@ -610,12 +610,15 @@ def test_growth_auto_falls_back_to_one_day(): def test_snapshot_recent_next_run_follows_completion(): import json + from datetime import datetime + from reaction_board.timeutil import SHANGHAI_TZ, interval_next_run with tempfile.TemporaryDirectory() as tmp: db, q = setup(tmp) try: repo = Repository(db) - finished = FIXED_NOW - 10 + # 午后完成：间隔后仍在静默窗外，应严格 finished+3600 + finished = int(datetime(2026, 10, 31, 15, 0, 0, tzinfo=SHANGHAI_TZ).timestamp()) repo.kv_set( "task:last:snapshot_recent", json.dumps({ @@ -627,6 +630,83 @@ def test_snapshot_recent_next_run_follows_completion(): DASHBOARD_CACHE.clear() system = q.system() assert system["next_runs"]["snapshot_recent"] == finished + 3600 + assert system["next_runs"]["snapshot_recent"] == int( + interval_next_run(float(finished), 3600) + ) + finally: + db.close() + + +def test_snapshot_recent_next_run_defers_quiet_window_and_retry_on_skip(): + import json + from datetime import datetime + from reaction_board.timeutil import SHANGHAI_TZ, next_after_snapshot_quiet_window + + with tempfile.TemporaryDirectory() as tmp: + db, q = setup(tmp) + try: + repo = Repository(db) + # 22:50 成功 → 下一档 23:50 落入静默窗 → 推到 01:00 + finished = int(datetime(2026, 10, 31, 22, 50, 0, tzinfo=SHANGHAI_TZ).timestamp()) + repo.kv_set( + "task:last:snapshot_recent", + json.dumps({ + "name": "snapshot_recent", "ok": True, + "started_at": finished - 5, "finished_at": finished, + "error": None, "summary": {}, + }), + ) + DASHBOARD_CACHE.clear() + system = q.system() + expected = int(datetime(2026, 11, 1, 1, 0, 0, tzinfo=SHANGHAI_TZ).timestamp()) + assert system["next_runs"]["snapshot_recent"] == expected + + # skip 记录：补跑到窗后 01:00，不得当成功推进 + quiet = int(datetime(2026, 10, 31, 23, 20, 0, tzinfo=SHANGHAI_TZ).timestamp()) + repo.kv_set( + "task:last:snapshot_recent", + json.dumps({ + "name": "snapshot_recent", "ok": False, "skipped": True, + "started_at": quiet, "finished_at": quiet, + "error": "snapshot_quiet_window", + "summary": {"skipped": True, "reason": "snapshot_quiet_window"}, + }), + ) + DASHBOARD_CACHE.clear() + system = q.system() + assert system["next_runs"]["snapshot_recent"] == int( + next_after_snapshot_quiet_window(float(quiet)) + ) + assert not any(e["task"] == "snapshot_recent" for e in system["errors"]) + finally: + db.close() + + +def test_snapshot_failed_next_run_retries_soon_not_next_day(): + import json + from datetime import datetime + from reaction_board.timeutil import SHANGHAI_TZ, next_daily_run + + with tempfile.TemporaryDirectory() as tmp: + db, q = setup(tmp) + try: + repo = Repository(db) + finished = int(datetime(2026, 10, 31, 23, 15, 0, tzinfo=SHANGHAI_TZ).timestamp()) + repo.kv_set( + "task:last:snapshot", + json.dumps({ + "name": "snapshot", "ok": False, + "started_at": finished - 1, "finished_at": finished, + "error": "boom", "summary": {"completed": False}, + }), + ) + DASHBOARD_CACHE.clear() + system = q.system() + assert system["next_runs"]["snapshot"] == finished + 300 + assert system["next_runs"]["snapshot"] != int( + next_daily_run(finished, "23:15") + ) + assert any(e["task"] == "snapshot" for e in system["errors"]) finally: db.close() @@ -882,6 +962,8 @@ def main_test(): test_system_tasks_sorted_by_started_at_desc() test_growth_auto_falls_back_to_one_day() test_snapshot_recent_next_run_follows_completion() + test_snapshot_recent_next_run_defers_quiet_window_and_retry_on_skip() + test_snapshot_failed_next_run_retries_soon_not_next_day() test_overview_adjacent_previous_with_seed_and_dual_fields() test_same_day_growth_baselines() test_growth_rate_sort_uses_prior() @@ -948,13 +1030,15 @@ def test_system_status_uses_completed_date_and_running_flags(): def test_system_next_runs_follow_last_completion_not_wall_clock(): import json + from datetime import datetime + from reaction_board.timeutil import SHANGHAI_TZ with tempfile.TemporaryDirectory() as tmp: db, q = setup(tmp) try: repo = Repository(db) - # 发现任务刚完成不久：下次 = finished + 60min，即使 now 已超过也不跳档 - finished = FIXED_NOW - 30 + # 午后完成：下次 = finished + 60min，不跳档、不误入静默窗推算 + finished = int(datetime(2026, 10, 31, 15, 0, 0, tzinfo=SHANGHAI_TZ).timestamp()) repo.kv_set( "task:last:discovery", json.dumps({ @@ -966,7 +1050,6 @@ def test_system_next_runs_follow_last_completion_not_wall_clock(): DASHBOARD_CACHE.clear() system = q.system() assert system["next_runs"]["discover"] == finished + 3600 - # 仍小于/等于 now 也可以（表示本轮后的下一档已到期或未到），但绝不能跳到 +7200 assert system["next_runs"]["discover"] != finished + 7200 finally: db.close()