feat(automation): 拆开「多久跑一次」与「同一账号多久重检」—— 调度频率终于可配
用户反馈:「单账号重检间隔(小时) 很别扭,直接设置多久跑一次、一次跑多少个更符合直觉」。
**对,而且比"别扭"更严重:调度频率以前根本改不了。**
## 问题:两个概念挤在一个字段里
`_run_one` 里原本写着:
job.next_run_at = finished + timedelta(seconds=JOB_INTERVAL_SECONDS) # 硬编码 300 秒
也就是不管界面上填什么,下一轮永远是「跑完 + 5 分钟」。
而界面那个「单账号重检间隔(小时)」对应的是 `interval_seconds` ——
它只被作业**自己的选行逻辑**使用("这行多久该重检"),**完全不参与调度**。
于是出现最坏的情况:**看起来能配调度,其实配的是一个不相干的参数**。
上一轮我还拿它算过吞吐("10 分钟一轮"),算错了 —— 已在调度器注释里纠正。
## 拆法
| 概念 | 字段 | 界面 |
| --- | --- | --- |
| **多久跑一次** | `run_interval_seconds`(迁移 0051,默认 300)| 「多久跑一次(分钟)」|
| 同一账号多久重检 | `interval_seconds`(不变)| 「同一账号多久重检(小时)」|
- `next_run_at = finished + run_interval_seconds`(有下限 30 秒,避免忙轮询)
- 接口:`GET` 返回新字段、`PATCH` 可改
- 前端:两个输入框分开,并把原来那段绕口的 tooltip
("它不决定调度频率 —— 调度周期见页面顶部说明")改掉 ——
那句话本身就是这个设计缺陷的自白
## 默认 300 秒
与改动前**实际行为一致** —— 不会因为这次迁移让已有作业突然变快或变慢。
想调快就自己填(比如「账号资料定时补全」填 5 分钟 + 每轮 400 条 ≈ 3 小时补完)。
## ⚠️ 顺带说明一个你截图里可能看到的现象
截图里「账号资料定时补全」显示的是 **0.5 小时 / 30 条** —— 那是我**最初的默认值**。
因为 `ensure_jobs` **只创建不更新**,我后面改成 10 分钟/400 **不会作用到已存在的行**。
这是刻意的(否则会覆盖运维的手工调整),但意味着**默认值改了要手动在界面上改**,
或者删掉那行让 `ensure_jobs` 重建。现在有了「多久跑一次」这个框,直接填就行。
## 测试(+3)
- 下一轮按**运行间隔**排,不是按「同一行重检间隔」(两个值刻意设成不同来钉住)
- 不填时默认 300 秒(保持旧行为)
- 过小的间隔被夹到下限
## 验证
TZ=Asia/Shanghai **639 passed / 6 skipped**;ruff 全绿;前端 build 通过、lint 0 errors。
This commit is contained in:
@@ -916,3 +916,81 @@ class TestSchedulerRunsJobsConcurrently:
|
||||
time.sleep(0.1)
|
||||
assert not scheduler._IN_FLIGHT, "失败后没清在途标记,作业会被永久跳过"
|
||||
assert calls, "作业应该真的跑过一次"
|
||||
|
||||
|
||||
class TestRunIntervalIsScheduling:
|
||||
"""「多久跑一次」与「同一行多久重检」是**两个独立概念**。
|
||||
|
||||
用户反馈:界面上那个「单账号重检间隔(小时)」很别扭,直接设「多久跑一次、
|
||||
一次跑多少个」更符合直觉 —— 对,因为**调度频率以前根本改不了**:
|
||||
|
||||
job.next_run_at = finished + timedelta(seconds=JOB_INTERVAL_SECONDS) # 硬编码 5 分钟
|
||||
|
||||
于是不管界面上填什么,下一轮永远是「跑完 + 5 分钟」。
|
||||
"""
|
||||
|
||||
def setup_method(self):
|
||||
self.engine = get_test_engine()
|
||||
self.session = make_test_session(self.engine)
|
||||
reset_registry()
|
||||
|
||||
def teardown_method(self):
|
||||
reset_registry()
|
||||
self.session.close()
|
||||
|
||||
def _job(self, run_interval: int | None):
|
||||
calls: list[int] = []
|
||||
|
||||
def probe(db, job):
|
||||
calls.append(1)
|
||||
return {"scanned": 1}
|
||||
|
||||
register_job(
|
||||
JobSpec(
|
||||
key="probe",
|
||||
title="probe",
|
||||
interval_seconds=6 * 3600, # 同一行重检间隔(不参与调度)
|
||||
batch_size=10,
|
||||
concurrency=1,
|
||||
run=probe,
|
||||
**({"run_interval_seconds": run_interval} if run_interval else {}),
|
||||
)
|
||||
)
|
||||
self.session.add(
|
||||
AutomationJob(
|
||||
key="probe",
|
||||
title="probe",
|
||||
interval_seconds=6 * 3600,
|
||||
run_interval_seconds=run_interval or 300,
|
||||
batch_size=10,
|
||||
concurrency=1,
|
||||
enabled=True,
|
||||
)
|
||||
)
|
||||
self.session.commit()
|
||||
return calls
|
||||
|
||||
def _next_run_delay(self) -> float:
|
||||
run_due_jobs(self.session)
|
||||
job = (
|
||||
self.session.query(AutomationJob).filter(AutomationJob.key == "probe").one()
|
||||
)
|
||||
assert job.last_run_at is not None and job.next_run_at is not None
|
||||
return (job.next_run_at - job.last_run_at).total_seconds()
|
||||
|
||||
def test_next_run_follows_run_interval_not_the_row_interval(self):
|
||||
"""下一轮按**运行间隔**排,不是按「同一行重检间隔」。"""
|
||||
self._job(run_interval=1800)
|
||||
delay = self._next_run_delay()
|
||||
assert delay == 1800, f"应按运行间隔 1800 排,实际 {delay}"
|
||||
assert delay != 6 * 3600, "不能再用「同一行重检间隔」当调度周期"
|
||||
|
||||
def test_default_run_interval_keeps_old_behaviour(self):
|
||||
"""不填时默认 300 秒 —— 与改动前的硬编码行为一致(不会突然变快变慢)。"""
|
||||
self._job(run_interval=None)
|
||||
assert self._next_run_delay() == 300
|
||||
|
||||
def test_too_small_interval_is_clamped(self):
|
||||
"""太小的间隔是忙轮询,没有意义 —— 有下限。"""
|
||||
self._job(run_interval=1)
|
||||
assert self._next_run_delay() >= scheduler.MIN_RUN_INTERVAL_SECONDS
|
||||
|
||||
@@ -26,7 +26,7 @@ TESTS_DIR = Path(__file__).resolve().parent
|
||||
PROJECT_ROOT = TESTS_DIR.parent
|
||||
VERSIONS_DIR = PROJECT_ROOT / "web" / "backend" / "migrations" / "versions"
|
||||
|
||||
HEAD_REVISION = "20261004_0050"
|
||||
HEAD_REVISION = "20261004_0051"
|
||||
|
||||
# 迁移新增、但不声明在模型里的复合查询索引。
|
||||
EXTRA_INDEXES = (
|
||||
|
||||
@@ -0,0 +1,64 @@
|
||||
"""作业的「运行间隔」独立出来:`automation_jobs.run_interval_seconds`
|
||||
|
||||
## 问题
|
||||
|
||||
界面上只有一个「单账号重检间隔(小时)」,但它其实承担了两个互不相干的概念,
|
||||
而且**调度频率根本改不了**:
|
||||
|
||||
| 概念 | 之前 | 之后 |
|
||||
| --- | --- | --- |
|
||||
| **多久跑一次** | **硬编码** `JOB_INTERVAL_SECONDS`(5 分钟)| `run_interval_seconds` |
|
||||
| 同一行多久重检 | `interval_seconds` | `interval_seconds`(不变,只是不再兼任调度)|
|
||||
|
||||
`_run_one` 里原本写着:
|
||||
|
||||
job.next_run_at = finished + timedelta(seconds=JOB_INTERVAL_SECONDS)
|
||||
|
||||
也就是不管在界面上填什么,下一轮永远是「跑完 + 5 分钟」。
|
||||
用户反馈「直接设置多久跑一次、一次跑多少个更符合直觉」—— 对。
|
||||
|
||||
默认 300 秒,与改动前的实际行为一致(不会因为这次迁移让已有作业突然变快/变慢)。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sqlalchemy as sa
|
||||
from alembic import op
|
||||
|
||||
revision: str = "20261004_0051"
|
||||
down_revision: str | None = "20261004_0050"
|
||||
branch_labels: str | None = None
|
||||
depends_on: str | None = None
|
||||
|
||||
COLUMN = "run_interval_seconds"
|
||||
|
||||
|
||||
def _column_exists(name: str) -> bool:
|
||||
conn = op.get_bind()
|
||||
count = conn.execute(
|
||||
sa.text(
|
||||
"SELECT COUNT(*) FROM information_schema.columns "
|
||||
"WHERE table_schema = DATABASE() "
|
||||
"AND table_name = 'automation_jobs' AND column_name = :name"
|
||||
),
|
||||
{"name": name},
|
||||
).scalar()
|
||||
return bool(count)
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
if not _column_exists(COLUMN):
|
||||
op.add_column(
|
||||
"automation_jobs",
|
||||
sa.Column(
|
||||
COLUMN,
|
||||
sa.Integer(),
|
||||
nullable=False,
|
||||
server_default="300",
|
||||
comment="多久跑一次(秒);与「同一行多久重检」是两回事",
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
op.drop_column("automation_jobs", COLUMN)
|
||||
@@ -897,6 +897,15 @@ class AutomationJob(Base):
|
||||
title: Mapped[str] = mapped_column(String(128), default="")
|
||||
# 默认关闭:按「渐进上线」原则,先观察再打开。
|
||||
enabled: Mapped[bool] = mapped_column(Boolean, default=False, server_default="0")
|
||||
# **多久跑一次**(秒)。这是调度频率,与"同一行多久重检"是两回事。
|
||||
# 以前调度频率是**硬编码**的 JOB_INTERVAL_SECONDS(5 分钟),
|
||||
# UI 上那个"单账号重检间隔"其实是 `interval_seconds` ——
|
||||
# 两个概念挤在一个输入框里,很别扭(用户指出)。
|
||||
run_interval_seconds: Mapped[int] = mapped_column(
|
||||
Integer, default=300, server_default="300"
|
||||
)
|
||||
# 同一**行**(账号)多久重检一次(秒)—— 由作业自己的选行逻辑使用,
|
||||
# **不参与调度**。别和上面那个混。
|
||||
interval_seconds: Mapped[int] = mapped_column(
|
||||
Integer, default=21600, server_default="21600"
|
||||
)
|
||||
|
||||
@@ -37,6 +37,8 @@ class JobUpdate(BaseModel):
|
||||
|
||||
enabled: bool | None = None
|
||||
interval_seconds: int | None = Field(default=None, ge=MIN_INTERVAL_SECONDS)
|
||||
# 多久跑一次(秒)。下限 30 秒,避免变成忙轮询。
|
||||
run_interval_seconds: int | None = Field(default=None, ge=30)
|
||||
batch_size: int | None = Field(default=None, ge=1, le=MAX_BATCH_SIZE)
|
||||
concurrency: int | None = Field(default=None, ge=1, le=16)
|
||||
|
||||
@@ -47,6 +49,7 @@ def _job_out(job: AutomationJob) -> dict[str, Any]:
|
||||
"title": job.title,
|
||||
"enabled": bool(job.enabled),
|
||||
"interval_seconds": int(job.interval_seconds or 0),
|
||||
"run_interval_seconds": int(job.run_interval_seconds or 0),
|
||||
"batch_size": int(job.batch_size or 0),
|
||||
"concurrency": int(job.concurrency or 0),
|
||||
"last_run_at": job.last_run_at.isoformat() if job.last_run_at else None,
|
||||
@@ -125,6 +128,8 @@ def update_job(
|
||||
changed["enabled"] = job.enabled
|
||||
# 刚打开时排一轮:不早于一个 tick,避免打开瞬间就压满 CPU。
|
||||
job.next_run_at = datetime.now(UTC) + _tick_delta() if job.enabled else None
|
||||
if payload.run_interval_seconds is not None:
|
||||
job.run_interval_seconds = int(payload.run_interval_seconds)
|
||||
if payload.interval_seconds is not None:
|
||||
changed["interval_seconds"] = payload.interval_seconds
|
||||
job.interval_seconds = payload.interval_seconds
|
||||
|
||||
@@ -57,10 +57,14 @@ class JobSpec:
|
||||
|
||||
key: str
|
||||
title: str
|
||||
#: 同一**行**多久重检(由作业自己的选行逻辑用,不参与调度)
|
||||
interval_seconds: int
|
||||
batch_size: int
|
||||
concurrency: int
|
||||
run: Callable[[Session, AutomationJob], dict[str, Any]]
|
||||
#: **多久跑一次**(调度频率)。默认 5 分钟,与改动前的实际行为一致。
|
||||
#: ⚠️ 必须放在**有默认值**的位置(dataclass 不允许默认值字段排在无默认值之前)。
|
||||
run_interval_seconds: int = 300
|
||||
|
||||
|
||||
_REGISTRY: dict[str, JobSpec] = {}
|
||||
@@ -68,6 +72,9 @@ _REGISTRY: dict[str, JobSpec] = {}
|
||||
#: 同时最多几个作业在跑。作业之间互不依赖,并行是为了**慢作业不挡住快作业**
|
||||
#: (以前串行:一个作业跑 4 分钟,CK 巡检就得等 4 分钟)。
|
||||
_JOB_MAX_WORKERS = 4
|
||||
|
||||
#: 运行间隔的下限:太短会变成忙轮询,没有意义
|
||||
MIN_RUN_INTERVAL_SECONDS = 30
|
||||
_EXECUTOR: ThreadPoolExecutor | None = None
|
||||
_EXECUTOR_LOCK = threading.Lock()
|
||||
|
||||
@@ -115,6 +122,7 @@ def ensure_jobs(db: Session) -> None:
|
||||
title=spec.title,
|
||||
enabled=False,
|
||||
interval_seconds=spec.interval_seconds,
|
||||
run_interval_seconds=spec.run_interval_seconds,
|
||||
batch_size=spec.batch_size,
|
||||
concurrency=spec.concurrency,
|
||||
config=None,
|
||||
@@ -305,9 +313,12 @@ def _run_one(db: Session, job: AutomationJob, spec: JobSpec) -> dict[str, Any]:
|
||||
run.detail = counters.get("detail") or None
|
||||
|
||||
job.last_run_at = finished
|
||||
# 下一轮固定在一个 tick 之后;interval_seconds 是「同一行多久重检」,
|
||||
# 由作业自己在选行时用,不参与调度。
|
||||
job.next_run_at = finished + timedelta(seconds=JOB_INTERVAL_SECONDS)
|
||||
# 下一轮 = 跑完 + **运行间隔**(可配置)。
|
||||
# ⚠️ 别再用 `interval_seconds` —— 那是「同一行多久重检」,
|
||||
# 由作业自己的选行逻辑使用,**不参与调度**(两个概念以前挤在一个字段里)。
|
||||
job.next_run_at = finished + timedelta(
|
||||
seconds=max(int(job.run_interval_seconds or 0), MIN_RUN_INTERVAL_SECONDS)
|
||||
)
|
||||
job.last_status = status
|
||||
job.last_message = message
|
||||
job.updated_at = finished
|
||||
|
||||
@@ -876,6 +876,9 @@ export interface HuyaPasswordLoginSelectedRequest {
|
||||
has_cookie?: boolean;
|
||||
force_new_device?: boolean;
|
||||
max_attempts?: number;
|
||||
/** 多久跑一次(秒)—— 调度频率 */
|
||||
run_interval_seconds?: number;
|
||||
/** 同一账号多久重检(秒)—— 不参与调度 */
|
||||
interval_seconds?: number;
|
||||
retry_delay_seconds?: number;
|
||||
}
|
||||
@@ -1031,6 +1034,8 @@ export interface AutomationRunItem {
|
||||
}
|
||||
|
||||
export interface AutomationJobItem {
|
||||
/** 多久跑一次(秒)—— 调度频率 */
|
||||
run_interval_seconds: number;
|
||||
key: string;
|
||||
title: string;
|
||||
enabled: boolean;
|
||||
@@ -1068,6 +1073,9 @@ export interface AutomationRunsResponse {
|
||||
|
||||
export interface AutomationJobUpdate {
|
||||
enabled?: boolean;
|
||||
/** 多久跑一次(秒)—— 调度频率 */
|
||||
run_interval_seconds?: number;
|
||||
/** 同一账号多久重检(秒)—— 不参与调度 */
|
||||
interval_seconds?: number;
|
||||
batch_size?: number;
|
||||
concurrency?: number;
|
||||
|
||||
@@ -99,6 +99,11 @@ export default function AutomationPage() {
|
||||
job.key,
|
||||
{
|
||||
enabled: job.enabled,
|
||||
// 「多久跑一次」与「同一行多久重检」是两件事,分开两个输入框
|
||||
run_interval_minutes: Math.max(
|
||||
Math.round((job.run_interval_seconds ?? 300) / 60),
|
||||
1,
|
||||
),
|
||||
interval_hours: secondsToHours(job.interval_seconds),
|
||||
batch_size: job.batch_size,
|
||||
concurrency: job.concurrency,
|
||||
@@ -123,6 +128,10 @@ export default function AutomationPage() {
|
||||
try {
|
||||
await automationApi.updateJob(job.key, {
|
||||
enabled: values.enabled,
|
||||
run_interval_seconds: Math.max(
|
||||
Math.round((values.run_interval_minutes ?? 5) * 60),
|
||||
30,
|
||||
),
|
||||
interval_seconds: Math.max(
|
||||
Math.round((values.interval_hours ?? 6) * 3600),
|
||||
300,
|
||||
@@ -312,10 +321,17 @@ export default function AutomationPage() {
|
||||
>
|
||||
<Switch disabled={!canManage} />
|
||||
</Form.Item>
|
||||
<Form.Item
|
||||
name={[job.key, "run_interval_minutes"]}
|
||||
label="多久跑一次(分钟)"
|
||||
tooltip="调度频率:上一轮跑完后隔多久开始下一轮。这是作业「多久跑一次」。"
|
||||
>
|
||||
<InputNumber min={0.5} step={5} disabled={!canManage} />
|
||||
</Form.Item>
|
||||
<Form.Item
|
||||
name={[job.key, "interval_hours"]}
|
||||
label="单账号重检间隔(小时)"
|
||||
tooltip="同一个账号多久重新检测/重试一次。它不决定调度频率 —— 调度周期见页面顶部说明。"
|
||||
label="同一账号多久重检(小时)"
|
||||
tooltip="同一个账号多久重新检测/重试一次。它**不决定**作业多久跑一次 —— 那是左边那一项。"
|
||||
>
|
||||
<InputNumber min={0.1} step={1} disabled={!canManage} />
|
||||
</Form.Item>
|
||||
|
||||
Reference in New Issue
Block a user