chongming-utils-task/examples/09_long_running.py

"""示例 9：无限循环任务的**启与停**。

一个"永远跑下去"的任务该怎么写、怎么启动、怎么停，以及三种收尾方式的终态差异。

运行::

    uv run python examples/09_long_running.py

## 三条规则

1. **任务体每轮必须调 ``ctx.checkpoint()``。** 停止是**协作式**的：只有调用了它，任务
   才会看到停止请求。不调用 = 停不掉（见文件最后的对照）。
2. **停止请求有两种收尾方式，终态不同**：

   ================================================  ===========  ==========
   写法                                              终态         返回值
   ================================================  ===========  ==========
   ``while True: ctx.checkpoint()``（让它抛）        ``stopped``  无
   检查 ``ctx.is_stop_requested`` 后正常 ``return``  ``succeeded`` ✅ 有
   完全不调 ``checkpoint()``                         ``lost``     无
   ================================================  ===========  ==========

   "停止"在第一种写法里是"立刻中止"，在第二种里是"**请收工**"——后者会把已经处理好的
   结果留下来，对"跑到下一个安全点再退出"这类需求更合适。
3. **停止是终态。** 目前没有 ``retry``，停止/失败之后想再跑要换一个 ``task_id``
   重新提交（``create`` 会拒绝重复 id）。所以日常"先停一下"应该用 ``pause`` / ``resume``，
   ``stop`` 留给"我要它彻底结束"。
"""

from __future__ import annotations

import sys
import threading
import time
from typing import Any

from chongming_task import Controller, InMemoryTaskStore, render_table


def worker(ctx: Any, source: str) -> int:
    """无限循环：持续处理，直到收到停止请求后**优雅收尾**。"""
    processed = 0
    while not ctx.is_stop_requested:
        ctx.checkpoint()
        processed += 1
        ctx.progress(processed, unit="batch", message=f"{source} 第 {processed} 批")
        if processed % 20 == 0:
            ctx.log(f"已处理 {processed} 批")
        time.sleep(0.02)

    # 走到这里说明收到停止请求了：做收尾，然后正常返回——
    # 返回值会被保留下来，终态是 succeeded 而不是 stopped。
    ctx.log(f"收到停止请求，收尾：本次共处理 {processed} 批")
    return processed


def main() -> None:
    store = InMemoryTaskStore()
    controller = Controller(store=store, owner="node-a")
    controller.register("worker", worker)

    # ---- 启动 ----
    run = controller.submit("worker", "kafka://orders", task_id="orders-worker")
    print(f"已启动 {run.snapshot.task_id}  状态={run.status.value}")
    time.sleep(0.5)

    print("\n--- 运行中 ---")
    print(render_table(controller.list_tasks()))

    # ---- 日常先停一下：pause / resume（不是 stop） ----
    print("\n--- pause（就地挂起，仍占用执行槽）---")
    print(f"  pause 是否确认: {controller.pause('orders-worker', timeout=5)}")
    print(f"  状态: {run.status.value}")

    print("\n--- resume ---")
    controller.resume("orders-worker")
    time.sleep(0.2)
    print(f"  状态: {run.status.value}")

    # ---- 让它彻底结束：stop ----
    print("\n--- stop（终态，不可逆）---")
    controller.stop("orders-worker", reason="发布窗口")
    outcome = controller.wait("orders-worker", timeout=10)

    print(f"  终态: {outcome.status.value}")
    print(f"  返回值: {outcome.value}  ← 优雅收尾把已处理的数量带了出来")
    print("\n--- 结束后 ---")
    print(render_table(controller.list_tasks()))

    print("\n--- 审计轨迹（谁按的、为什么，都在这里）---")
    for record in store.history("orders-worker"):
        reason = f"  原因={record.reason}" if record.reason else ""
        print(
            f"  {record.version:>2}  {record.from_status.value:>9} -> "
            f"{record.to_status.value:<9} {record.trigger}{reason}"
        )

    # ---- 停止是终态：同一个 task_id 不能复活 ----
    print("\n--- 想再跑一次？同一个 task_id 会被拒绝 ---")
    try:
        controller.submit("worker", "kafka://orders", task_id="orders-worker")
    except Exception as exc:
        print(f"  {type(exc).__name__}: {exc}")
    print("  换一个 task_id 重新提交即可（或让进程一直活着，用 pause/resume 做日常启停）。")

    controller.close(timeout=5)

    _contrast_uncooperative()
    _print_cli_usage()


def _contrast_uncooperative() -> None:
    """对照：不调 ``checkpoint()`` 的任务**停不掉**。

    这不是缺陷，是诚实：线程无法被强杀，所以系统只能如实报 ``lost``（"结果未知"），
    而任务体仍在后台继续跑。真正终止它只能靠结束进程。
    """
    print("\n" + "=" * 70)
    print("对照：一个从不调 checkpoint() 的无限任务")
    print("=" * 70)

    release = threading.Event()
    ticks = {"n": 0}

    def uncooperative(ctx: Any) -> int:
        while not release.is_set():
            ticks["n"] += 1
            time.sleep(0.02)
        return ticks["n"]

    controller = Controller(store=InMemoryTaskStore(), graceful_stop=0.5)
    controller.register("stubborn", uncooperative)
    controller.submit("stubborn", task_id="stubborn")
    time.sleep(0.3)

    print("  停止请求已发出，等待载体响应……")
    controller.stop("stubborn")
    outcome = controller.wait("stubborn", timeout=8)
    print(f"  终态: {outcome.status.value}   ← 不是 stopped，而是 lost（结果未知）")
    print(f"  失败类型: {outcome.failure.kind.value if outcome.failure else None}")

    before = ticks["n"]
    time.sleep(0.4)
    alive = [t.name for t in threading.enumerate() if t.is_alive() and t.name.startswith("task-")]
    print(f"  任务体的线程还在吗: {alive}")
    print(f"  它还在继续干活吗: {ticks['n'] > before}（{before} -> {ticks['n']}）")
    print("  → 线程无法被强杀，只能靠结束进程来真正终止它。")

    release.set()
    controller.close(timeout=3)


def _print_cli_usage() -> None:
    """运维视角：命令行怎么启、怎么停。"""
    print("\n" + "=" * 70)
    print("命令行（另一个进程，通过共享存储启停）")
    print("=" * 70)
    print(
        """
# 启动：前台执行，交给 systemd / nohup 保持存活
# 注意 --bucket-key：限定了 --owned-buckets 时它是**必需**的，否则任务没有归属桶、
# 不会有任何节点认领它（CLI 会在提交时报错）。
chongming-task --db /var/lib/tasks.db --num-buckets 64 --owned-buckets 0-15 \\
    run --tasks myapp.tasks worker --arg source=kafka://orders \\
        --task-id orders-worker --bucket-key orders

# 另一个终端：看状态、停止
chongming-task --db /var/lib/tasks.db --owned-buckets 0-15 ps --all --mine
chongming-task --db /var/lib/tasks.db --owned-buckets 0-15 status orders-worker
chongming-task --db /var/lib/tasks.db --owned-buckets 0-15 stop orders-worker --reason 发布窗口

# Ctrl-C 也可以：main() 会走 finally 里的 controller.close()，把在跑的任务停干净
""".rstrip()
    )
    print("  注意：`run` 是前台进程——它退出时任务就没人管了，请交给进程管理器。")
    sys.stdout.write("")


if __name__ == "__main__":
    main()