chakcy
f4de43c743
单任务受控执行（15 态状态机 + 审计轨迹）、DAG 管道、Controller 门面。
执行器三个：线程（默认，含 async def 任务）、内联（同步 / 测试）、进程
（spawn 子进程：硬性停止、绕开 GIL、崩溃隔离）。
调度：并发额度 + 每桶限流 + 具名资源信号量（按标签）+ 优先级排队。
持久层：TaskStore 契约 + 内存 / SQLite / Postgres 实现，契约测试随包发布。
跨进程与多节点：写穿 / 读穿、指令通道与租约、owned_buckets 划界、reconcile。
CLI、12 个可直接运行的示例、docs/ 下 5 份文档。
门禁：ruff / ruff format / mypy(strict) 干净；
pytest 563 passed（44 skipped 是需要真实 Postgres 的那批）；
带一台可连的 Postgres 时 607 passed。
405 lines
13 KiB
Python
405 lines
13 KiB
Python
"""单任务受控执行的测试。
|
|
|
|
覆盖三件用户最关心的事：**失败会怎样**（重试 / 超时 / 最终失败）、**我能不能干预**
|
|
（停止 / 暂停 / 恢复），以及**结果是否诚实**（不配合的任务落 LOST 而不是"已停止"）。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
import time
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from chongming_task.errors import ConfigurationError
|
|
from chongming_task.models import FailureKind
|
|
from chongming_task.policies import FixedBackoff, RetryPolicy, Timeout
|
|
from chongming_task.run import TaskRun, run_task
|
|
from chongming_task.runtime.executors.inline import InlineExecutor
|
|
from chongming_task.states import TaskStatus
|
|
|
|
_WINDOW = 10.0
|
|
|
|
|
|
def _wait_for_status(run: TaskRun, status: TaskStatus, timeout: float = _WINDOW) -> bool:
|
|
"""等状态就位。
|
|
|
|
比 ``sleep`` 可靠：``start()`` 是异步的，"任务已经开始执行"与"状态机已迁移到
|
|
RUNNING"之间存在极短窗口。
|
|
"""
|
|
deadline = time.monotonic() + timeout
|
|
while time.monotonic() < deadline:
|
|
if run.status is status:
|
|
return True
|
|
time.sleep(0.01)
|
|
return False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 成功路径
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_successful_run_returns_value_and_succeeds() -> None:
|
|
outcome = run_task(lambda ctx, a, b: a + b, 2, 3, task_id="t-ok", name="加法")
|
|
assert outcome.ok
|
|
assert outcome.status is TaskStatus.SUCCEEDED
|
|
assert outcome.value == 5
|
|
assert outcome.failure is None
|
|
assert outcome.snapshot.ended_at is not None
|
|
|
|
|
|
def test_transition_history_is_complete_and_ordered() -> None:
|
|
"""状态历史只追加，且每条都带触发源——这是可审计性的基础。"""
|
|
run = TaskRun(lambda ctx: 1, task_id="t-hist", name="demo")
|
|
run.start()
|
|
assert run.wait(timeout=_WINDOW).ok
|
|
|
|
assert [t.to_status for t in run.transitions] == [
|
|
TaskStatus.QUEUED,
|
|
TaskStatus.STARTING,
|
|
TaskStatus.RUNNING,
|
|
TaskStatus.SUCCEEDED,
|
|
]
|
|
assert all(t.trigger for t in run.transitions)
|
|
assert run.transitions[-1].task_id == "t-hist"
|
|
|
|
|
|
def test_exception_is_captured_with_traceback() -> None:
|
|
def boom(ctx: Any) -> None:
|
|
raise ValueError("炸了")
|
|
|
|
outcome = run_task(boom, task_id="t-boom")
|
|
assert outcome.status is TaskStatus.FAILED
|
|
assert outcome.failure is not None
|
|
assert outcome.failure.kind is FailureKind.EXCEPTION
|
|
assert outcome.failure.traceback is not None
|
|
assert "boom" in outcome.failure.traceback
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 重试
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_retries_until_success_and_counts_attempts() -> None:
|
|
attempts: list[int] = []
|
|
|
|
def flaky(ctx: Any, fail_times: int) -> str:
|
|
attempts.append(1)
|
|
if len(attempts) <= fail_times:
|
|
raise RuntimeError("暂时性故障")
|
|
return "ok"
|
|
|
|
outcome = run_task(
|
|
flaky,
|
|
2,
|
|
retry=RetryPolicy(max_attempts=3, backoff=FixedBackoff(0.0)),
|
|
task_id="t-flaky",
|
|
)
|
|
assert outcome.ok
|
|
assert outcome.snapshot.attempt == 3
|
|
assert len(attempts) == 3
|
|
|
|
|
|
def test_retry_exhaustion_lands_failed() -> None:
|
|
def always_fails(ctx: Any) -> None:
|
|
raise RuntimeError("一直失败")
|
|
|
|
outcome = run_task(
|
|
always_fails,
|
|
retry=RetryPolicy(max_attempts=2, backoff=FixedBackoff(0.0)),
|
|
task_id="t-always",
|
|
)
|
|
assert outcome.status is TaskStatus.FAILED
|
|
assert outcome.snapshot.attempt == 2
|
|
assert outcome.failure is not None
|
|
|
|
|
|
def test_retry_wait_is_visible_in_history() -> None:
|
|
seen: list[int] = []
|
|
|
|
def flaky(ctx: Any) -> str:
|
|
if not seen:
|
|
seen.append(1)
|
|
raise RuntimeError("第一次失败")
|
|
return "ok"
|
|
|
|
run = TaskRun(
|
|
flaky,
|
|
retry=RetryPolicy(max_attempts=2, backoff=FixedBackoff(0.0)),
|
|
task_id="t-wait",
|
|
)
|
|
run.start()
|
|
assert run.wait(timeout=_WINDOW).ok
|
|
assert TaskStatus.RETRY_WAIT in [t.to_status for t in run.transitions]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 超时
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_per_attempt_timeout_stops_a_slow_task() -> None:
|
|
def slow(ctx: Any) -> str:
|
|
for _ in range(1_000_000):
|
|
ctx.checkpoint()
|
|
time.sleep(0.01)
|
|
return "永远跑不完"
|
|
|
|
outcome = run_task(slow, timeout=Timeout(per_attempt=0.3), task_id="t-slow")
|
|
assert outcome.status is TaskStatus.FAILED
|
|
assert outcome.failure is not None
|
|
assert outcome.failure.kind is FailureKind.TIMEOUT_ATTEMPT
|
|
|
|
|
|
def test_idle_timeout_is_skipped_while_paused() -> None:
|
|
"""被暂停的任务不推进心跳，若照常判 idle 超时，暂停就会被误判成卡死。"""
|
|
started = threading.Event()
|
|
|
|
def work(ctx: Any) -> str:
|
|
started.set()
|
|
for _ in range(1_000_000):
|
|
ctx.checkpoint()
|
|
time.sleep(0.02)
|
|
return "done"
|
|
|
|
run = TaskRun(
|
|
work,
|
|
timeout=Timeout(idle=0.2),
|
|
task_id="t-idle-paused",
|
|
graceful_stop=2.0,
|
|
)
|
|
run.start()
|
|
assert started.wait(_WINDOW)
|
|
assert _wait_for_status(run, TaskStatus.RUNNING)
|
|
assert run.pause(timeout=_WINDOW) is True
|
|
|
|
# 暂停期间超过 idle 阈值，但不应被判超时。
|
|
time.sleep(0.5)
|
|
assert run.status is TaskStatus.PAUSED
|
|
|
|
run.resume()
|
|
run.stop(reason="测试结束")
|
|
assert run.wait(timeout=_WINDOW).status is TaskStatus.STOPPED
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 停止
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_stop_lands_stopped_and_is_auditable() -> None:
|
|
started = threading.Event()
|
|
|
|
def work(ctx: Any) -> str:
|
|
started.set()
|
|
for _ in range(1_000_000):
|
|
ctx.checkpoint()
|
|
time.sleep(0.005)
|
|
return "done"
|
|
|
|
run = TaskRun(work, task_id="t-stop")
|
|
run.start()
|
|
assert started.wait(_WINDOW)
|
|
assert _wait_for_status(run, TaskStatus.RUNNING)
|
|
|
|
run.stop(reason="发布窗口")
|
|
outcome = run.wait(timeout=_WINDOW)
|
|
assert outcome.status is TaskStatus.STOPPED
|
|
assert outcome.failure is not None
|
|
assert outcome.failure.kind is FailureKind.CANCELLED
|
|
assert outcome.failure.retryable is False, "取消不是失败，不能重试"
|
|
|
|
stopping = [t for t in run.transitions if t.to_status is TaskStatus.STOPPING]
|
|
assert stopping, "应当能看到 STOPPING 这个中间态"
|
|
assert stopping[0].reason == "发布窗口"
|
|
|
|
|
|
def test_uncooperative_task_is_reported_as_lost_not_stopped() -> None:
|
|
"""线程停不掉时结果**未知**：任务可能仍在后台产生副作用。
|
|
|
|
记成"已停止"会让运维误判资源已释放，因此必须落 LOST。
|
|
"""
|
|
|
|
def stubborn(ctx: Any) -> str:
|
|
for _ in range(500):
|
|
time.sleep(0.01)
|
|
return "ignored"
|
|
|
|
run = TaskRun(stubborn, task_id="t-stubborn", graceful_stop=0.2)
|
|
run.start()
|
|
time.sleep(0.3)
|
|
run.stop(reason="不配合")
|
|
|
|
outcome = run.wait(timeout=_WINDOW)
|
|
assert outcome.status is TaskStatus.LOST
|
|
assert outcome.failure is not None
|
|
assert outcome.failure.kind is FailureKind.LOST
|
|
|
|
|
|
def test_stop_before_start_lands_stopped_without_burning_an_attempt() -> None:
|
|
"""都还没跑就要求停止，不该白白消耗一次尝试。"""
|
|
run = TaskRun(lambda ctx: "ok", task_id="t-pre-stop")
|
|
run.stop(reason="还没跑就不要了")
|
|
run.start()
|
|
outcome = run.wait(timeout=_WINDOW)
|
|
assert outcome.status is TaskStatus.STOPPED
|
|
assert outcome.snapshot.attempt == 0
|
|
assert all(t.to_status is not TaskStatus.RUNNING for t in run.transitions)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 暂停与恢复
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_pause_freezes_progress_and_resume_continues() -> None:
|
|
started = threading.Event()
|
|
progress: list[int] = []
|
|
|
|
def work(ctx: Any) -> str:
|
|
started.set()
|
|
for index in range(1_000_000):
|
|
ctx.checkpoint()
|
|
progress.append(index)
|
|
time.sleep(0.005)
|
|
return "done"
|
|
|
|
run = TaskRun(work, task_id="t-pause")
|
|
run.start()
|
|
assert started.wait(_WINDOW)
|
|
assert _wait_for_status(run, TaskStatus.RUNNING)
|
|
|
|
assert run.pause(timeout=_WINDOW) is True
|
|
assert run.status is TaskStatus.PAUSED
|
|
frozen = len(progress)
|
|
time.sleep(0.15)
|
|
assert len(progress) == frozen, "暂停生效后不应继续推进"
|
|
|
|
run.resume()
|
|
assert run.status is TaskStatus.RUNNING, "就地暂停的任务恢复后回到 RUNNING"
|
|
time.sleep(0.15)
|
|
assert len(progress) > frozen, "恢复后应继续推进"
|
|
|
|
run.stop()
|
|
assert run.wait(timeout=_WINDOW).status is TaskStatus.STOPPED
|
|
|
|
|
|
def test_pause_returns_false_when_task_never_checkpoints() -> None:
|
|
"""暂停没生效就必须如实回滚为 RUNNING，不能假装已暂停。"""
|
|
|
|
def work(ctx: Any) -> str:
|
|
for _ in range(500):
|
|
time.sleep(0.01)
|
|
return "ignored"
|
|
|
|
run = TaskRun(work, task_id="t-nopause", graceful_stop=0.2)
|
|
run.start()
|
|
assert _wait_for_status(run, TaskStatus.RUNNING)
|
|
|
|
assert run.pause(timeout=0.2) is False
|
|
assert run.status is TaskStatus.RUNNING
|
|
assert any(t.to_status is TaskStatus.PAUSING for t in run.transitions)
|
|
assert any(
|
|
t.to_status is TaskStatus.RUNNING and t.trigger == "runtime:pause_aborted"
|
|
for t in run.transitions
|
|
), "回滚应当留下 PAUSE_ABORTED 记录"
|
|
|
|
run.stop()
|
|
run.wait(timeout=_WINDOW)
|
|
|
|
|
|
def test_pause_on_a_finished_run_is_rejected() -> None:
|
|
run = TaskRun(lambda ctx: 1, task_id="t-pause-late")
|
|
run.start()
|
|
assert run.wait(timeout=_WINDOW).ok
|
|
assert run.pause(timeout=0.01) is False
|
|
assert run.status is TaskStatus.SUCCEEDED
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# wait 的契约
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_wait_timeout_raises_and_does_not_change_state() -> None:
|
|
"""把"我没等到"当成"任务失败"是最常见的误用，因此这里只抛异常。"""
|
|
release = threading.Event()
|
|
|
|
def work(ctx: Any) -> str:
|
|
release.wait(_WINDOW)
|
|
return "ok"
|
|
|
|
run = TaskRun(work, task_id="t-wait")
|
|
run.start()
|
|
with pytest.raises(TimeoutError, match="状态未被改变"):
|
|
run.wait(timeout=0.2)
|
|
assert run.status in (TaskStatus.STARTING, TaskStatus.RUNNING)
|
|
|
|
release.set()
|
|
assert run.wait(timeout=_WINDOW).ok
|
|
|
|
|
|
def test_wait_before_start_is_an_error() -> None:
|
|
run = TaskRun(lambda ctx: 1, task_id="t-not-started")
|
|
with pytest.raises(RuntimeError, match="start"):
|
|
run.wait()
|
|
|
|
|
|
def test_starting_twice_is_an_error() -> None:
|
|
run = TaskRun(lambda ctx: 1, task_id="t-twice")
|
|
run.start()
|
|
run.wait(timeout=_WINDOW)
|
|
with pytest.raises(RuntimeError, match="已经启动过"):
|
|
run.start()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 执行器可替换
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_executor_is_pluggable() -> None:
|
|
run = TaskRun(
|
|
lambda ctx: "inline",
|
|
task_id="t-inline",
|
|
executor=InlineExecutor(),
|
|
)
|
|
run.start()
|
|
outcome = run.wait(timeout=_WINDOW)
|
|
assert outcome.ok
|
|
assert outcome.value == "inline"
|
|
assert outcome.snapshot.executor == "inline"
|
|
|
|
|
|
class _RefusingExecutor:
|
|
"""拉不起来任何任务，只会抛错——进程执行器的"目标不可 pickle"就长这样。"""
|
|
|
|
name = "refusing"
|
|
|
|
def launch(self, request: Any) -> Any:
|
|
del request
|
|
raise ConfigurationError("目标不可 pickle：这类任务请改用线程执行器")
|
|
|
|
|
|
def test_a_launch_failure_is_recorded_as_the_tasks_own_failure() -> None:
|
|
"""执行器拉不起来时，必须如实记在**任务**上，而不是让异常逃进监督线程。
|
|
|
|
逃进去的后果有两条：任务先停在 ``starting``，最后由兜底说成"监督循环内部错误"
|
|
——那等于把一个任务侧的问题指成框架的 bug；而调用方拿到的诊断里也看不到真正
|
|
该改的东西。
|
|
"""
|
|
outcome = run_task(
|
|
lambda ctx: "never",
|
|
task_id="t-refused",
|
|
executor=_RefusingExecutor(),
|
|
)
|
|
|
|
assert outcome.status is TaskStatus.FAILED
|
|
assert outcome.failure is not None
|
|
assert "不可 pickle" in outcome.failure.message
|
|
assert "监督循环内部错误" not in outcome.failure.message, "别把任务侧的问题说成框架的"
|
|
assert outcome.failure.retryable is False, "配置类错误重试多少次都是同一个结果"
|