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。
403 lines
15 KiB
Python
403 lines
15 KiB
Python
"""写穿（write-through）的恢复能力测试。
|
||
|
||
这一组用例存在的理由：**写穿失败曾经会让任务的存储状态永久冻结**。
|
||
|
||
故障链是：``_persist`` 用 ``expect_version = transition.version - 1`` 写入，这个式子
|
||
假设此前的每一次迁移都成功落库。一旦某次失败，存储停在 k-1 而内存已到 k，此后**每一次**
|
||
写入都因版本对不上而冲突——一次瞬时故障让任务在存储里永远停住，最后被 ``reconcile()``
|
||
判成 ``LOST``：把一个真正跑完的任务报成"结果未知"。
|
||
|
||
所以这里的重点不是"失败时不会崩"，而是"**失败之后能追上**"。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import threading
|
||
import time
|
||
from typing import Any
|
||
|
||
import pytest
|
||
|
||
from chongming_task.controller import Controller
|
||
from chongming_task.errors import ConfigurationError
|
||
from chongming_task.models import FailureKind, TaskSnapshot
|
||
from chongming_task.states import TaskStatus
|
||
from chongming_task.stores.memory import InMemoryTaskStore
|
||
from chongming_task.stores.sqlite import SqliteTaskStore
|
||
|
||
_WINDOW = 15.0
|
||
|
||
|
||
class FlakyStore(InMemoryTaskStore):
|
||
"""可以按需让 ``save`` 抖动若干次，之后恢复正常。
|
||
|
||
``fail_saves`` 计数：每次 ``save`` 尝试消耗一个；为 0 时完全正常。这样能精确模拟
|
||
"网络抖了一下"以及"存储长时间不可用"两种情况。
|
||
"""
|
||
|
||
def __init__(self, *, fail_saves: int = 0, fail_forever: bool = False) -> None:
|
||
super().__init__()
|
||
self.fail_saves = fail_saves
|
||
self.fail_forever = fail_forever
|
||
self.save_calls = 0
|
||
self._mutex = threading.RLock()
|
||
|
||
def save(self, snapshot: TaskSnapshot, *, expect_version: int) -> None:
|
||
with self._mutex:
|
||
self.save_calls += 1
|
||
if self.fail_forever or self.fail_saves > 0:
|
||
if not self.fail_forever:
|
||
self.fail_saves -= 1
|
||
raise OSError("模拟存储抖动")
|
||
super().save(snapshot, expect_version=expect_version)
|
||
|
||
|
||
def _wait_until_running(controller: Controller, task_id: str, *, timeout: float = 5.0) -> None:
|
||
"""等任务真的进入 ``RUNNING``。
|
||
|
||
**为什么必须显式等**：``TaskRun.pause()`` 对还没进入 ``RUNNING`` 的任务会
|
||
**立刻返回 False 且不产生任何迁移**。提交后马上开循环，若执行线程还没被调度起来，
|
||
整个循环会在微秒级跑完、一次都没暂停成功——用例看起来"通过到了断言那一步"，
|
||
实则前提从未成立（它曾因此间歇性失败）。机器越忙，这个窗口越大。
|
||
|
||
**为什么不能用 ``controller.get()``**：它**优先读存储**，而本文件的存储多半是故意
|
||
写不进去的 ``FlakyStore``——那样永远等不到 ``RUNNING``。这里只看本地执行句柄。
|
||
"""
|
||
deadline = time.monotonic() + timeout
|
||
while time.monotonic() < deadline:
|
||
for run in controller.runs():
|
||
if run.snapshot.task_id == task_id and run.status is TaskStatus.RUNNING:
|
||
return
|
||
time.sleep(0.005)
|
||
raise AssertionError(f"{task_id} 没能在 {timeout}s 内进入 RUNNING")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 核心：一次抖动不能让存储永久冻结
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_a_single_transient_failure_does_not_freeze_the_store() -> None:
|
||
"""**这条是本次修复的核心回归用例。**
|
||
|
||
以前：第 3 次 save 抖动一次 → 存储永久停在 ``starting``，而任务其实跑完了。
|
||
现在：下一次迁移会把存储一次追上。
|
||
"""
|
||
store = FlakyStore(fail_saves=1)
|
||
controller = Controller(store=store)
|
||
controller.register("ok", _ok)
|
||
controller.submit("ok", task_id="t1")
|
||
outcome = controller.wait("t1", timeout=_WINDOW)
|
||
|
||
assert outcome.ok, "存储抖动不该让任务失败"
|
||
snapshot = store.get("t1")
|
||
assert snapshot is not None
|
||
assert snapshot.status is TaskStatus.SUCCEEDED, "存储必须追上内存"
|
||
assert snapshot.outcome == {"value": "done"}
|
||
|
||
history = [record.to_status for record in store.history("t1")]
|
||
assert history == [
|
||
TaskStatus.QUEUED,
|
||
TaskStatus.STARTING,
|
||
TaskStatus.RUNNING,
|
||
TaskStatus.SUCCEEDED,
|
||
], "迁移轨迹必须完整——缓冲的意义就在这里"
|
||
|
||
assert controller.desynced_tasks == {}, "瞬时故障不该造成永久分叉"
|
||
assert controller.persist_failures, "但失败必须留痕"
|
||
controller.close(timeout=_WINDOW)
|
||
|
||
|
||
def test_the_store_version_catches_up_after_a_gap() -> None:
|
||
"""版本会**跳号**（这是允许的），但必须单调且最终一致。"""
|
||
store = FlakyStore(fail_saves=2)
|
||
controller = Controller(store=store)
|
||
controller.register("ok", _ok)
|
||
controller.submit("ok", task_id="t1")
|
||
controller.wait("t1", timeout=_WINDOW)
|
||
|
||
snapshot = store.get("t1")
|
||
assert snapshot is not None
|
||
assert snapshot.status is TaskStatus.SUCCEEDED
|
||
assert snapshot.status_version == 4, "内存算出的版本"
|
||
|
||
versions = [record.version for record in store.history("t1")]
|
||
assert versions == [1, 2, 3, 4], "补写后轨迹不缺项"
|
||
assert versions == sorted(versions), "版本必须单调"
|
||
controller.close(timeout=_WINDOW)
|
||
|
||
|
||
def test_failures_spread_over_several_transitions_still_converge() -> None:
|
||
"""抖动散布在多次迁移上（而不是集中一次）也要能追上。"""
|
||
store = FlakyStore(fail_saves=3)
|
||
controller = Controller(store=store)
|
||
controller.register("ok", _ok)
|
||
controller.submit("ok", task_id="t1")
|
||
assert controller.wait("t1", timeout=_WINDOW).ok
|
||
|
||
snapshot = store.get("t1")
|
||
assert snapshot is not None
|
||
assert snapshot.status is TaskStatus.SUCCEEDED
|
||
controller.close(timeout=_WINDOW)
|
||
|
||
|
||
def test_a_long_outage_recovered_by_close_catches_up() -> None:
|
||
"""存储整个任务期间都不可用：任务照常成功，``close()`` 时尽力补一次。
|
||
|
||
补不上的话，任务在存储里就是落后的——这仍然比"任务失败"好，但必须在
|
||
``persist_failures`` 里留痕。
|
||
"""
|
||
store = FlakyStore(fail_forever=True)
|
||
controller = Controller(store=store)
|
||
controller.register("ok", _ok)
|
||
controller.submit("ok", task_id="t1")
|
||
outcome = controller.wait("t1", timeout=_WINDOW)
|
||
|
||
assert outcome.ok, "存储不可用不该让任务失败"
|
||
# 一次都没写进去：行停在 create 时的初始状态（pending）。
|
||
assert store.get("t1").status is TaskStatus.PENDING # type: ignore[union-attr]
|
||
assert len(controller.persist_failures) >= 1
|
||
|
||
# 存储恢复后 close() 会补上
|
||
store.fail_forever = False
|
||
controller.close(timeout=_WINDOW)
|
||
assert store.get("t1").status is TaskStatus.SUCCEEDED # type: ignore[union-attr]
|
||
|
||
|
||
def test_the_pending_buffer_is_bounded() -> None:
|
||
"""存储长期不可用时内存不能跟着无限增长。"""
|
||
from chongming_task.controller import _PENDING_TRANSITIONS
|
||
|
||
store = FlakyStore(fail_forever=True)
|
||
controller = Controller(store=store)
|
||
controller.register("spinner", _spinner)
|
||
controller.submit("spinner", task_id="t1")
|
||
_wait_until_running(controller, "t1")
|
||
|
||
# 让任务反复暂停/恢复，制造远超上限的迁移次数。
|
||
for _ in range(_PENDING_TRANSITIONS + 20):
|
||
assert controller.pause("t1", timeout=2.0), "任务没到达检查点：迁移没发生，用例失去意义"
|
||
controller.resume("t1")
|
||
|
||
state = controller._persist_states["t1"]
|
||
assert len(state.pending) <= _PENDING_TRANSITIONS
|
||
assert state.dropped > 0, "丢弃必须被计数，否则轨迹缺口就不可见了"
|
||
|
||
controller.stop("t1")
|
||
controller.wait_all(timeout=_WINDOW)
|
||
controller.close(timeout=_WINDOW)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 外部写入者：该放手时放手
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_an_external_writer_marks_the_task_desynced_instead_of_being_overwritten() -> None:
|
||
"""用"上次确实写进去的版本"仍冲突 → 是**别人**写的，覆盖它会抹掉对方的状态。
|
||
|
||
这与"我们自己漏写了一笔"必须区分开：前者要放手，后者要追上去。
|
||
"""
|
||
store = InMemoryTaskStore()
|
||
controller = Controller(store=store, command_poll_interval=0.05)
|
||
controller.register("block", _block)
|
||
run = controller.submit("block", task_id="t1")
|
||
assert _wait_until(lambda: run.status is TaskStatus.RUNNING)
|
||
|
||
# 模拟另一个写入者把版本推进了（例如另一个进程认领了同一个任务）。
|
||
current = store.get("t1")
|
||
assert current is not None
|
||
intruder = _bump(current)
|
||
store.save(intruder, expect_version=current.status_version)
|
||
|
||
# 触发一次我们自己的迁移：应当发现外部写入并放手。
|
||
controller.stop("t1")
|
||
assert (
|
||
_wait_until(lambda: bool(controller.desynced_tasks))
|
||
or controller.wait_all(timeout=_WINDOW) is not None
|
||
)
|
||
|
||
desynced = controller.desynced_tasks
|
||
assert "t1" in desynced, "必须检测到外部写入者并停止写穿"
|
||
assert "外部写入者" in desynced["t1"]
|
||
|
||
# 对方的状态没有被我们覆盖。
|
||
assert store.get("t1").status is TaskStatus.RUNNING # type: ignore[union-attr]
|
||
controller.close(timeout=_WINDOW)
|
||
|
||
|
||
def test_desynced_tasks_are_reported_not_hidden() -> None:
|
||
"""永久分叉必须可见——只靠 persist_failures（最近 20 条）是不够的。"""
|
||
store = InMemoryTaskStore()
|
||
controller = Controller(store=store)
|
||
assert controller.desynced_tasks == {}
|
||
|
||
state = controller._persist_state("ghost")
|
||
assert controller.desynced_tasks == {}
|
||
state.desynced = "测试"
|
||
assert controller.desynced_tasks == {"ghost": "测试"}
|
||
controller.close(timeout=_WINDOW)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 非法返回值：响亮失败，而不是静默损坏
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_a_non_serializable_return_value_fails_the_task_loudly() -> None:
|
||
"""以前：进程内 succeeded、存储停在 running、重启后被判 LOST。
|
||
|
||
现在：任务**失败**，原因写明是返回值不满足持久化契约。
|
||
"""
|
||
store = InMemoryTaskStore()
|
||
controller = Controller(store=store)
|
||
controller.register("bad_result", _bad_result)
|
||
controller.submit("bad_result", task_id="t1")
|
||
outcome = controller.wait("t1", timeout=_WINDOW)
|
||
|
||
assert not outcome.ok, "违反契约必须是失败，而不是静默的'成功'"
|
||
assert outcome.status is TaskStatus.FAILED
|
||
snapshot = store.get("t1")
|
||
assert snapshot is not None
|
||
assert snapshot.status is TaskStatus.FAILED, "存储与内存一致，不再是 running"
|
||
assert snapshot.failure is not None
|
||
assert snapshot.failure.kind is FailureKind.RESULT_INVALID
|
||
assert "不可 JSON 序列化" in snapshot.failure.message
|
||
assert snapshot.failure.retryable is False, "重试解决不了：同一个值还是不可序列化"
|
||
controller.close(timeout=_WINDOW)
|
||
|
||
|
||
def test_a_non_serializable_result_is_not_retried() -> None:
|
||
"""重试没有意义，所以只尝试一次。"""
|
||
store = InMemoryTaskStore()
|
||
controller = Controller(store=store)
|
||
controller.register("bad_result", _bad_result)
|
||
controller.submit("bad_result", task_id="t1", retry=_retry_three())
|
||
controller.wait("t1", timeout=_WINDOW)
|
||
|
||
snapshot = store.get("t1")
|
||
assert snapshot is not None
|
||
assert snapshot.attempt == 1, "不可序列化不是暂时性问题，不该重试"
|
||
controller.close(timeout=_WINDOW)
|
||
|
||
|
||
def test_serializable_results_are_unaffected() -> None:
|
||
"""正常路径不能被这次收紧影响。"""
|
||
store = InMemoryTaskStore()
|
||
controller = Controller(store=store)
|
||
controller.register("rich", _rich_result)
|
||
controller.submit("rich", task_id="t1")
|
||
outcome = controller.wait("t1", timeout=_WINDOW)
|
||
assert outcome.ok
|
||
|
||
snapshot = store.get("t1")
|
||
assert snapshot is not None
|
||
assert snapshot.outcome is not None
|
||
assert snapshot.outcome["value"]["nested"] == [1, 2, 3]
|
||
controller.close(timeout=_WINDOW)
|
||
|
||
|
||
def test_a_non_string_tag_is_rejected_at_submit_time() -> None:
|
||
"""标签会被写进存储（JSON 数组），所以必须在**入口**按类型契约拒绝。
|
||
|
||
真实类型是 ``frozenset[str]``。非字符串元素既不能正确往返，又只会在写穿路径上才
|
||
炸——而那条路径的失败不会让任务失败，问题会以"看板上的标签不对"的形式出现。
|
||
|
||
注意这条**必须用会序列化的后端**才测得出存储层的反应：``InMemoryTaskStore`` 直接
|
||
持有对象、从不编码。"内存后端能跑"不等于"真后端能跑"。
|
||
"""
|
||
store = SqliteTaskStore(":memory:")
|
||
try:
|
||
controller = Controller(store=store)
|
||
controller.register("ok", _ok)
|
||
with pytest.raises(ConfigurationError, match="标签必须是字符串"):
|
||
controller.submit("ok", task_id="t1", tags=[object()])
|
||
controller.close(timeout=_WINDOW)
|
||
finally:
|
||
store.close()
|
||
|
||
|
||
def test_a_non_serializable_progress_message_raises_at_the_call_site() -> None:
|
||
"""进度字段会被写进存储；类型错误必须在调用点炸掉，而不是流到写穿路径上。"""
|
||
store = InMemoryTaskStore()
|
||
controller = Controller(store=store)
|
||
controller.register("bad_progress", _bad_progress)
|
||
outcome = controller.wait(
|
||
controller.submit("bad_progress", task_id="t1").snapshot.task_id, timeout=_WINDOW
|
||
)
|
||
|
||
assert not outcome.ok
|
||
assert "message 必须是 str" in (outcome.failure.message if outcome.failure else "")
|
||
|
||
snapshot = store.get("t1")
|
||
assert snapshot is not None
|
||
assert snapshot.status is TaskStatus.FAILED
|
||
controller.close(timeout=_WINDOW)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 工具
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _bump(snapshot: TaskSnapshot) -> TaskSnapshot:
|
||
"""构造一个"别人写的"快照：版本前进一步。"""
|
||
from dataclasses import replace
|
||
|
||
return replace(snapshot, status_version=snapshot.status_version + 1)
|
||
|
||
|
||
def _retry_three() -> Any:
|
||
from chongming_task.policies import FixedBackoff, RetryPolicy
|
||
|
||
return RetryPolicy(max_attempts=3, backoff=FixedBackoff(0.0))
|
||
|
||
|
||
def _wait_until(predicate: Any, timeout: float = _WINDOW) -> bool:
|
||
import time
|
||
|
||
deadline = time.monotonic() + timeout
|
||
while time.monotonic() < deadline:
|
||
if predicate():
|
||
return True
|
||
time.sleep(0.01)
|
||
return False
|
||
|
||
|
||
def _ok(ctx: Any) -> str:
|
||
del ctx
|
||
return "done"
|
||
|
||
|
||
def _bad_result(ctx: Any) -> Any:
|
||
del ctx
|
||
return {1, 2, 3}
|
||
|
||
|
||
def _rich_result(ctx: Any) -> dict[str, Any]:
|
||
del ctx
|
||
return {"nested": [1, 2, 3], "flag": True, "text": "中文"}
|
||
|
||
|
||
def _bad_progress(ctx: Any) -> str:
|
||
ctx.progress(1, total=10, message=object()) # type: ignore[arg-type]
|
||
return "unreachable"
|
||
|
||
|
||
def _spinner(ctx: Any) -> str:
|
||
import time
|
||
|
||
for _ in range(1_000_000):
|
||
ctx.checkpoint()
|
||
time.sleep(0.002)
|
||
return "never"
|
||
|
||
|
||
def _block(ctx: Any) -> str:
|
||
import time
|
||
|
||
for _ in range(1_000_000):
|
||
ctx.checkpoint()
|
||
time.sleep(0.005)
|
||
return "never"
|