已合并
feat(service): 新增主备定时任务并接入 SystemContext #426
zhangxiangyu52创建于 8月14日
feat(service): 新增主备定时任务并接入 SystemContext #426
已合并
共 14 个文件变更+1013-2
| @@ -0,0 +1,53 @@ | |||
| 1 | +# coding: utf-8 | ||
| 2 | +# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved | ||
| 3 | + | ||
| 4 | +"""主备定时任务 —— 怎么调用。 | ||
| 5 | + | ||
| 6 | +用 ``SystemContext.create_single_leader_job``：redis / 工号都从上下文取。 | ||
| 7 | +（不传 instance_id 时默认是「主机名+随机号」。） | ||
| 8 | +""" | ||
| 9 | + | ||
| 10 | +from __future__ import annotations | ||
| 11 | + | ||
| 12 | +import asyncio | ||
| 13 | +from typing import Any | ||
| 14 | + | ||
| 15 | +from openjiuwen_runtime.foundation.log import get_logger | ||
| 16 | +from openjiuwen_runtime.service import SystemContext | ||
| 17 | + | ||
| 18 | +logger = get_logger(__name__) | ||
| 19 | + | ||
| 20 | + | ||
| 21 | +async def example_create_single_leader_job(redis: Any) -> None: | ||
| 22 | + async def on_tick() -> None: | ||
| 23 | + # 到点干活；调度时钟是 Redis TIME，业务若要时间自己再读 | ||
| 24 | + logger.info("[demo] tick") | ||
| 25 | + | ||
| 26 | + # 不传 instance_id：上下文自己生成唯一工号 | ||
| 27 | + ctx = SystemContext(redis=redis) | ||
| 28 | + job = ctx.create_single_leader_job( | ||
| 29 | + name="demo", # 【必填】任务名；不传 lock_key 时 → 锁名 lock:demo | ||
| 30 | + on_tick=on_tick, # 【必填】到点回调（无参） | ||
| 31 | + # interval_sec=1, # 【选填】默认 1；执行锁 TTL = 此值 | ||
| 32 | + # gather_window_sec=0.08, # 【选填】默认 0.08；开火前提前醒来报名 | ||
| 33 | + # lock_key="", # 【选填】空则 lock:{name}；一般不用改 | ||
| 34 | + # run_on_start=False, # 【选填】True=启动立刻跑一轮（多用于测试） | ||
| 35 | + ) | ||
| 36 | + await job.start() | ||
| 37 | + try: | ||
| 38 | + await asyncio.sleep(5) # 演示跑几秒；生产里挂在服务生命周期上 | ||
| 39 | + finally: | ||
| 40 | + await job.stop() | ||
| 41 | + | ||
| 42 | + | ||
| 43 | +if __name__ == "__main__": | ||
| 44 | + import redis.asyncio as redis | ||
| 45 | + | ||
| 46 | + async def _main() -> None: | ||
| 47 | + client = redis.from_url("redis://127.0.0.1:6379/0") | ||
| 48 | + try: | ||
| 49 | + await example_create_single_leader_job(client) | ||
| 50 | + finally: | ||
| 51 | + await client.aclose() | ||
| 52 | + | ||
| 53 | + asyncio.run(_main()) | ||
| @@ -74,6 +74,12 @@ from .bootstrap import ( | |||
| 74 | create_system_context, | 74 | create_system_context, |
| 75 | shutdown_system, | 75 | shutdown_system, |
| 76 | ) | 76 | ) |
| 77 | +from .context.periodic import ( | ||
| 78 | + JobRunner, | ||
| 79 | + SingleLeaderCoordinator, | ||
| 80 | + TickLock, | ||
| 81 | + create_single_leader_job, | ||
| 82 | +) | ||
| 77 | from .context.primitives.idempotency import idempotency_guard | 83 | from .context.primitives.idempotency import idempotency_guard |
| 78 | from .routing.handlers import ( | 84 | from .routing.handlers import ( |
| 79 | FunctionMessageHandler, | 85 | FunctionMessageHandler, |
| @@ -161,6 +167,11 @@ __all__ = [ | |||
| 161 | "build_system_context", | 167 | "build_system_context", |
| 162 | "create_system_context", | 168 | "create_system_context", |
| 163 | "shutdown_system", | 169 | "shutdown_system", |
| 170 | + # periodic | ||
| 171 | + "JobRunner", | ||
| 172 | + "SingleLeaderCoordinator", | ||
| 173 | + "TickLock", | ||
| 174 | + "create_single_leader_job", | ||
| 164 | # handlers | 175 | # handlers |
| 165 | "HandlerSpec", | 176 | "HandlerSpec", |
| 166 | "MessageHandler", | 177 | "MessageHandler", |
| @@ -0,0 +1,26 @@ | |||
| 1 | +# coding: utf-8 | ||
| 2 | +# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved | ||
| 3 | + | ||
| 4 | +"""进程内周期任务 SDK（Schedule + Coordinator + JobRunner）。 | ||
| 5 | + | ||
| 6 | +主备模式：``SingleLeaderCoordinator``（等待窗口抽签 + 锁续期）。 | ||
| 7 | +时间：Redis ``TIME`` 对表 + 本机 monotonic 推算。 | ||
| 8 | + | ||
| 9 | +对外入口：``SystemContext.create_single_leader_job``（工号取 ``ctx.instance_id``）。 | ||
| 10 | +""" | ||
| 11 | + | ||
| 12 | +from .coordinator import Coordinator, SingleLeaderCoordinator | ||
| 13 | +from .factory import create_single_leader_job | ||
| 14 | +from .lock import TickLock | ||
| 15 | +from .runner import JobRunner | ||
| 16 | +from .schedule import IntervalSchedule, Schedule | ||
| 17 | + | ||
| 18 | +__all__ = ( | ||
| 19 | + "Coordinator", | ||
| 20 | + "IntervalSchedule", | ||
| 21 | + "JobRunner", | ||
| 22 | + "Schedule", | ||
| 23 | + "SingleLeaderCoordinator", | ||
| 24 | + "TickLock", | ||
| 25 | + "create_single_leader_job", | ||
| 26 | +) | ||
| @@ -0,0 +1,42 @@ | |||
| 1 | +# coding: utf-8 | ||
| 2 | +# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved | ||
| 3 | + | ||
| 4 | +"""用 Redis TIME 对表，之后用本机 monotonic 推算「现在」。 | ||
| 5 | + | ||
| 6 | +避免每个「现在几点」都打一次 TIME（RTT 会吃掉 80ms 集合窗口）。 | ||
| 7 | +""" | ||
| 8 | + | ||
| 9 | +from __future__ import annotations | ||
| 10 | + | ||
| 11 | +import time | ||
| 12 | +from typing import Any | ||
| 13 | + | ||
| 14 | + | ||
| 15 | +def _as_int(v: Any) -> int: | ||
| 16 | + if isinstance(v, bytes): | ||
| 17 | + return int(v) | ||
| 18 | + return int(v) | ||
| 19 | + | ||
| 20 | + | ||
| 21 | +async def redis_unix_now(redis: Any) -> float: | ||
| 22 | + """Redis TIME → unix 秒（含微秒小数）。""" | ||
| 23 | + pair = await redis.time() | ||
| 24 | + sec, usec = pair[0], pair[1] | ||
| 25 | + return float(_as_int(sec)) + float(_as_int(usec)) / 1_000_000.0 | ||
| 26 | + | ||
| 27 | + | ||
| 28 | +class RedisAlignedClock: | ||
| 29 | + """offset = Redis unix − monotonic；now() = monotonic + offset。""" | ||
| 30 | + | ||
| 31 | + def __init__(self, redis: Any) -> None: | ||
| 32 | + self._redis = redis | ||
| 33 | + self._offset = 0.0 | ||
| 34 | + | ||
| 35 | + async def sync(self) -> float: | ||
| 36 | + """打一次 TIME，刷新偏移，返回对表后的现在。""" | ||
| 37 | + rnow = await redis_unix_now(self._redis) | ||
| 38 | + self._offset = rnow - time.monotonic() | ||
| 39 | + return self.now() | ||
| 40 | + | ||
| 41 | + def now(self) -> float: | ||
| 42 | + return time.monotonic() + self._offset | ||
| @@ -0,0 +1,10 @@ | |||
| 1 | +# coding: utf-8 | ||
| 2 | +# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved | ||
| 3 | + | ||
| 4 | +from .base import Coordinator | ||
| 5 | +from .single_leader import SingleLeaderCoordinator | ||
| 6 | + | ||
| 7 | +__all__ = ( | ||
| 8 | + "Coordinator", | ||
| 9 | + "SingleLeaderCoordinator", | ||
| 10 | +) | ||
| @@ -0,0 +1,28 @@ | |||
| 1 | +# coding: utf-8 | ||
| 2 | +# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved | ||
| 3 | + | ||
| 4 | +"""协调器协议。""" | ||
| 5 | + | ||
| 6 | +from __future__ import annotations | ||
| 7 | + | ||
| 8 | +from typing import Optional, Protocol | ||
| 9 | + | ||
| 10 | + | ||
| 11 | +class Coordinator(Protocol): | ||
| 12 | + async def try_claim( | ||
| 13 | + self, | ||
| 14 | + *, | ||
| 15 | + now: float, | ||
| 16 | + instance_id: str, | ||
| 17 | + planned_fire: float | None = None, | ||
| 18 | + ) -> Optional[str]: | ||
| 19 | + """试着领取本轮执行权；成功返回锁 token，失败返回 None。 | ||
| 20 | + | ||
| 21 | + ``planned_fire``：本拍语义上的开火整点（如 10.000）。 | ||
| 22 | + 提前醒来时 ``now`` 可能是 T-窗口，epoch / 等到点应以 ``planned_fire`` 为准。 | ||
| 23 | + """ | ||
| 24 | + ... | ||
| 25 | + | ||
| 26 | + async def release(self, token: str) -> None: | ||
| 27 | + """交回执行权（按 token 校验后放锁）。""" | ||
| 28 | + ... | ||
| @@ -0,0 +1,185 @@ | |||
| 1 | +# coding: utf-8 | ||
| 2 | +# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved | ||
| 3 | + | ||
| 4 | +"""主备协调：提前报名 → 等到开火点 → 抽签选主 → 持锁续期执行。 | ||
| 5 | + | ||
| 6 | +流程（每拍，配合 JobRunner 提前 ``gather_window`` 醒来）： | ||
| 7 | +1. ``planned_fire`` 为本拍整点 T；``now`` 多为 T-窗口 | ||
| 8 | +2. Lua ``SADD`` + ``EXPIRE`` 报名（epoch 取自 T，一次写完避免 key 泄漏） | ||
| 9 | +3. 睡到 T（剩余窗口），让网络慢的实例也能进来 | ||
| 10 | +4. Lua 原子抽签：``SRANDMEMBER`` + ``SET NX winner:{epoch}`` | ||
| 11 | +5. 只有 winner 去 ``SET NX`` 执行锁，并启动续期；别人空转 | ||
| 12 | +""" | ||
| 13 | + | ||
| 14 | +from __future__ import annotations | ||
| 15 | + | ||
| 16 | +import asyncio | ||
| 17 | +import math | ||
| 18 | +from typing import Any, Optional | ||
| 19 | + | ||
| 20 | +from openjiuwen_runtime.foundation.log import get_logger | ||
| 21 | + | ||
| 22 | +from ..clock import RedisAlignedClock, redis_unix_now | ||
| 23 | +from ..lock import TickLock | ||
| 24 | + | ||
| 25 | +logger = get_logger(__name__) | ||
| 26 | + | ||
| 27 | +_ELECT_LUA = """ | ||
| 28 | +local existing = redis.call('GET', KEYS[1]) | ||
| 29 | +if existing then | ||
| 30 | + return existing | ||
| 31 | +end | ||
| 32 | +local pick = redis.call('SRANDMEMBER', KEYS[2]) | ||
| 33 | +if not pick then | ||
| 34 | + return false | ||
| 35 | +end | ||
| 36 | +local ok = redis.call('SET', KEYS[1], pick, 'NX', 'EX', tonumber(ARGV[1])) | ||
| 37 | +if ok then | ||
| 38 | + return pick | ||
| 39 | +end | ||
| 40 | +return redis.call('GET', KEYS[1]) | ||
| 41 | +""" | ||
| 42 | + | ||
| 43 | +# SADD + EXPIRE 一次完成，避免进程在两步之间崩溃留下没有 TTL 的报名 key | ||
| 44 | +_ENROLL_LUA = """ | ||
| 45 | +redis.call('SADD', KEYS[1], ARGV[1]) | ||
| 46 | +redis.call('EXPIRE', KEYS[1], tonumber(ARGV[2])) | ||
| 47 | +return 1 | ||
| 48 | +""" | ||
| 49 | + | ||
| 50 | + | ||
| 51 | +class SingleLeaderCoordinator: | ||
| 52 | + """主备：开火前窗口内集齐候选人，到点后随机选唯一执行者。""" | ||
| 53 | + | ||
| 54 | + def __init__( | ||
| 55 | + self, | ||
| 56 | + redis: Any, | ||
| 57 | + *, | ||
| 58 | + lock_key: str, | ||
| 59 | + lock_ttl_sec: int = 1, | ||
| 60 | + token_prefix: str = "job", | ||
| 61 | + instance_id: str = "", | ||
| 62 | + gather_window_sec: float = 0.08, | ||
| 63 | + meta_ttl_sec: int | None = None, | ||
| 64 | + clock: RedisAlignedClock | None = None, | ||
| 65 | + ) -> None: | ||
| 66 | + self._redis = redis | ||
| 67 | + self._instance_id = instance_id | ||
| 68 | + self._lock_key = lock_key | ||
| 69 | + self._clock = clock | ||
| 70 | + self._gather_window_sec = max(float(gather_window_sec), 0.0) | ||
| 71 | + # 元数据 TTL 至少盖住集合窗口，避免睡醒后 candidates 已过期 | ||
| 72 | + if meta_ttl_sec is None: | ||
| 73 | + meta_ttl_sec = max(3, int(math.ceil(self._gather_window_sec)) + 2) | ||
| 74 | + self._meta_ttl_sec = max(int(meta_ttl_sec), 1) | ||
| 75 | + self._lock = TickLock( | ||
| 76 | + redis, | ||
| 77 | + lock_key=lock_key, | ||
| 78 | + lock_ttl_sec=lock_ttl_sec, | ||
| 79 | + token_prefix=token_prefix, | ||
| 80 | + instance_id=instance_id, | ||
| 81 | + ) | ||
| 82 | + | ||
| 83 | + | ||
| 84 | + def lock_lost_event(self) -> asyncio.Event: | ||
| 85 | + """执行锁失锁事件；Runner 可据此中断 on_tick。""" | ||
| 86 | + return self._lock.lost_event | ||
| 87 | + | ||
| 88 | + def _candidates_key(self, epoch: int) -> str: | ||
| 89 | + return f"{self._lock_key}:candidates:{epoch}" | ||
| 90 | + | ||
| 91 | + def _winner_key(self, epoch: int) -> str: | ||
| 92 | + return f"{self._lock_key}:winner:{epoch}" | ||
| 93 | + | ||
| 94 | + async def _enroll(self, cand_key: str, instance_id: str) -> None: | ||
| 95 | + await self._redis.eval( | ||
| 96 | + _ENROLL_LUA, | ||
| 97 | + 1, | ||
| 98 | + cand_key, | ||
| 99 | + instance_id, | ||
| 100 | + str(self._meta_ttl_sec), | ||
| 101 | + ) | ||
| 102 | + | ||
| 103 | + async def try_claim( | ||
| 104 | + self, | ||
| 105 | + *, | ||
| 106 | + now: float, | ||
| 107 | + instance_id: str, | ||
| 108 | + planned_fire: float | None = None, | ||
| 109 | + ) -> Optional[str]: | ||
| 110 | + iid = instance_id or self._instance_id | ||
| 111 | + fire_at = float(planned_fire) if planned_fire is not None else float(now) | ||
| 112 | + epoch = int(fire_at) | ||
| 113 | + cand_key = self._candidates_key(epoch) | ||
| 114 | + winner_key = self._winner_key(epoch) | ||
| 115 | + | ||
| 116 | + await self._enroll(cand_key, iid) | ||
| 117 | + | ||
| 118 | + # 报名已耗时：用对表时钟（或再问一次 TIME）算剩余，避免按过期 now 睡过 T | ||
| 119 | + if planned_fire is not None: | ||
| 120 | + current = ( | ||
| 121 | + self._clock.now() | ||
| 122 | + if self._clock is not None | ||
| 123 | + else await redis_unix_now(self._redis) | ||
| 124 | + ) | ||
| 125 | + delay = fire_at - current | ||
| 126 | + else: | ||
| 127 | + delay = self._gather_window_sec | ||
| 128 | + if delay > 0: | ||
| 129 | + await asyncio.sleep(delay) | ||
| 130 | + | ||
| 131 | + # 抽签前再续一次 TTL，防止窗口偏大或抖动导致 candidates 已过期 | ||
| 132 | + await self._enroll(cand_key, iid) | ||
| 133 | + | ||
| 134 | + winner = await self._elect(winner_key, cand_key) | ||
| 135 | + if winner is None: | ||
| 136 | + logger.debug("no candidates for epoch=%s instance=%s", epoch, iid) | ||
| 137 | + return None | ||
| 138 | + | ||
| 139 | + winner_s = winner.decode() if isinstance(winner, (bytes, bytearray)) else str(winner) | ||
| 140 | + if winner_s != iid: | ||
| 141 | + logger.debug( | ||
| 142 | + "not elected: epoch=%s instance=%s winner=%s", | ||
| 143 | + epoch, | ||
| 144 | + iid, | ||
| 145 | + winner_s, | ||
| 146 | + ) | ||
| 147 | + return None | ||
| 148 | + | ||
| 149 | + token = await self._lock.try_acquire() | ||
| 150 | + if token is None: | ||
| 151 | + logger.warning( | ||
| 152 | + "elected but lock busy: epoch=%s instance=%s key=%s", | ||
| 153 | + epoch, | ||
| 154 | + iid, | ||
| 155 | + self._lock_key, | ||
| 156 | + ) | ||
| 157 | + return None | ||
| 158 | + | ||
| 159 | + self._lock.start_renew(token) | ||
| 160 | + logger.info( | ||
| 161 | + "single_leader claimed: epoch=%s instance=%s key=%s", | ||
| 162 | + epoch, | ||
| 163 | + iid, | ||
| 164 | + self._lock_key, | ||
| 165 | + ) | ||
| 166 | + return token | ||
| 167 | + | ||
| 168 | + async def _elect(self, winner_key: str, cand_key: str) -> Any: | ||
| 169 | + return await self._redis.eval( | ||
| 170 | + _ELECT_LUA, | ||
| 171 | + 2, | ||
| 172 | + winner_key, | ||
| 173 | + cand_key, | ||
| 174 | + str(self._meta_ttl_sec), | ||
| 175 | + ) | ||
| 176 | + | ||
| 177 | + async def release(self, token: str) -> None: | ||
| 178 | + try: | ||
| 179 | + await self._lock.release_if_owner(token) | ||
| 180 | + except Exception: | ||
| 181 | + logger.exception( | ||
| 182 | + "single_leader release failed: key=%s token=%s", | ||
| 183 | + self._lock.lock_key, | ||
| 184 | + token, | ||
| 185 | + ) | ||
| @@ -0,0 +1,69 @@ | |||
| 1 | +# coding: utf-8 | ||
| 2 | +# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved | ||
| 3 | + | ||
| 4 | +"""工厂：一条调用组装好主备定时任务。 | ||
| 5 | + | ||
| 6 | +对外配置只认本函数参数；内部零件（Runner / Schedule / Coordinator）不另搞 Config 袋。 | ||
| 7 | +redis / instance_id 一律从 ``SystemContext`` 取，不单独传入。 | ||
| 8 | +""" | ||
| 9 | + | ||
| 10 | +from __future__ import annotations | ||
| 11 | + | ||
| 12 | +import math | ||
| 13 | +from typing import TYPE_CHECKING, Any, Awaitable, Callable | ||
| 14 | + | ||
| 15 | +from .clock import RedisAlignedClock | ||
| 16 | +from .coordinator.single_leader import SingleLeaderCoordinator | ||
| 17 | +from .runner import JobRunner | ||
| 18 | +from .schedule.interval import IntervalSchedule | ||
| 19 | + | ||
| 20 | +if TYPE_CHECKING: | ||
| 21 | + from ..system_context import SystemContext | ||
| 22 | + | ||
| 23 | +# 内部常量（不对外暴露） | ||
| 24 | +_META_TTL_FLOOR_SEC = 3 | ||
| 25 | + | ||
| 26 | + | ||
| 27 | +def create_single_leader_job( | ||
| 28 | + ctx: "SystemContext", | ||
| 29 | + *, | ||
| 30 | + name: str, # 任务名；默认锁 key 为 lock:{name} | ||
| 31 | + on_tick: Callable[[], Awaitable[None]], # 到点回调：async def on_tick() -> None | ||
| 32 | + interval_sec: int = 1, # 每隔多少秒响一次；锁 TTL 与此相同 | ||
| 33 | + gather_window_sec: float = 0.08, # 开火前集合窗口：提前醒来报名，到整秒抽签 | ||
| 34 | + lock_key: str = "", # 执行锁 Redis key；空则用 lock:{name} | ||
| 35 | + run_on_start: bool = False, # True 启动后立刻跑一轮（一般仅测试） | ||
| 36 | +) -> JobRunner: | ||
| 37 | + """创建主备周期任务，返回可 start/stop 的 JobRunner。 | ||
| 38 | + | ||
| 39 | + 工号用 ``ctx.instance_id``，Redis 用 ``ctx.require_redis()``。 | ||
| 40 | + 服务框架内优先 ``ctx.create_single_leader_job(...)``。 | ||
| 41 | + """ | ||
| 42 | + redis: Any = ctx.require_redis() | ||
| 43 | + instance_id = ctx.instance_id | ||
| 44 | + interval = max(int(interval_sec), 1) | ||
| 45 | + gather = max(float(gather_window_sec), 0.0) | ||
| 46 | + # 元数据 TTL 盖住集合窗口，避免 candidates 在抽签前过期 | ||
| 47 | + meta_ttl = max(_META_TTL_FLOOR_SEC, int(math.ceil(gather)) + 2) | ||
| 48 | + key = (lock_key or f"lock:{name}").rstrip(":") | ||
| 49 | + clock = RedisAlignedClock(redis) | ||
| 50 | + return JobRunner( | ||
| 51 | + name=name, | ||
| 52 | + schedule=IntervalSchedule(interval), | ||
| 53 | + coordinator=SingleLeaderCoordinator( | ||
| 54 | + redis, | ||
| 55 | + lock_key=key, | ||
| 56 | + lock_ttl_sec=interval, | ||
| 57 | + token_prefix=name, | ||
| 58 | + instance_id=instance_id, | ||
| 59 | + gather_window_sec=gather, | ||
| 60 | + meta_ttl_sec=meta_ttl, | ||
| 61 | + clock=clock, | ||
| 62 | + ), | ||
| 63 | + on_tick=on_tick, | ||
| 64 | + instance_id=instance_id, | ||
| 65 | + redis=redis, | ||
| 66 | + clock=clock, | ||
| 67 | + gather_window_sec=gather, | ||
| 68 | + run_on_start=run_on_start, | ||
| 69 | + ) | ||
| @@ -0,0 +1,326 @@ | |||
| 1 | +# coding: utf-8 | ||
| 2 | +# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved | ||
| 3 | + | ||
| 4 | +"""JobRunner：提前窗口醒来 → 协调 → 回调 → 放锁。 | ||
| 5 | + | ||
| 6 | +调度时钟：Redis TIME 对表 + 本机 monotonic 推算（少打 TIME，避免 RTT 吃窗口）。 | ||
| 7 | +睡觉一次睡到目标点；stop 通过 cancel 打断，或醒来后检查退出。 | ||
| 8 | + | ||
| 9 | +对外配置请看 ``SystemContext.create_single_leader_job``；本类只收组装后的零件。 | ||
| 10 | +""" | ||
| 11 | + | ||
| 12 | +from __future__ import annotations | ||
| 13 | + | ||
| 14 | +import asyncio | ||
| 15 | +import time | ||
| 16 | +from typing import Any, Awaitable, Callable, Optional | ||
| 17 | + | ||
| 18 | +from openjiuwen_runtime.foundation.log import get_logger | ||
| 19 | + | ||
| 20 | +from .clock import RedisAlignedClock | ||
| 21 | +from .coordinator.base import Coordinator | ||
| 22 | +from .schedule.base import Schedule | ||
| 23 | + | ||
| 24 | +logger = get_logger(__name__) | ||
| 25 | + | ||
| 26 | +_STOP_TIMEOUT_SEC = 3.0 | ||
| 27 | +_TIME_FAIL_SLEEP_SEC = 0.5 | ||
| 28 | + | ||
| 29 | + | ||
| 30 | +async def _cancel_and_wait(task: asyncio.Task[Any]) -> None: | ||
| 31 | + """取消后台任务并等到它结束，避免 stop 后回调还在跑。""" | ||
| 32 | + if task.done(): | ||
| 33 | + return | ||
| 34 | + task.cancel() | ||
| 35 | + try: | ||
| 36 | + await task | ||
| 37 | + except asyncio.CancelledError: | ||
| 38 | + pass | ||
| 39 | + | ||
| 40 | + | ||
| 41 | +class JobRunner: | ||
| 42 | + """进程内周期任务生命周期管理。""" | ||
| 43 | + | ||
| 44 | + def __init__( | ||
| 45 | + self, | ||
| 46 | + *, | ||
| 47 | + name: str, | ||
| 48 | + schedule: Schedule, | ||
| 49 | + coordinator: Coordinator, | ||
| 50 | + on_tick: Callable[[], Awaitable[None]], | ||
| 51 | + instance_id: str, | ||
| 52 | + redis: Any, | ||
| 53 | + clock: RedisAlignedClock | None = None, | ||
| 54 | + gather_window_sec: float = 0.0, | ||
| 55 | + run_on_start: bool = False, | ||
| 56 | + stop_timeout_sec: float = _STOP_TIMEOUT_SEC, | ||
| 57 | + ) -> None: | ||
| 58 | + self._name = name | ||
| 59 | + self._schedule = schedule | ||
| 60 | + self._coordinator = coordinator | ||
| 61 | + self._on_tick = on_tick | ||
| 62 | + self._instance_id = instance_id | ||
| 63 | + self._clock = clock or RedisAlignedClock(redis) | ||
| 64 | + self._gather_window_sec = max(float(gather_window_sec), 0.0) | ||
| 65 | + self._run_on_start = bool(run_on_start) | ||
| 66 | + self._stop_timeout_sec = float(stop_timeout_sec) | ||
| 67 | + self._stopped = asyncio.Event() | ||
| 68 | + self._task: Optional[asyncio.Task[Any]] = None | ||
| 69 | + self._last_now: Optional[float] = None | ||
| 70 | + | ||
| 71 | + | ||
| 72 | + def name(self) -> str: | ||
| 73 | + return self._name | ||
| 74 | + | ||
| 75 | + def _now(self) -> float: | ||
| 76 | + return self._clock.now() | ||
| 77 | + | ||
| 78 | + async def _sync_now(self) -> float | None: | ||
| 79 | + """对一次 Redis 表；失败则打日志并睡觉，返回 None 让主循环重来。""" | ||
| 80 | + try: | ||
| 81 | + return await self._clock.sync() | ||
| 82 | + except asyncio.CancelledError: | ||
| 83 | + logger.debug( | ||
| 84 | + "redis TIME cancelled: job=%s instance=%s", | ||
| 85 | + self._name, | ||
| 86 | + self._instance_id, | ||
| 87 | + ) | ||
| 88 | + raise | ||
| 89 | + except Exception: | ||
| 90 | + logger.exception( | ||
| 91 | + "redis TIME failed: job=%s instance=%s", | ||
| 92 | + self._name, | ||
| 93 | + self._instance_id, | ||
| 94 | + ) | ||
| 95 | + await asyncio.sleep(_TIME_FAIL_SLEEP_SEC) | ||
| 96 | + return None | ||
| 97 | + | ||
| 98 | + async def start(self) -> None: | ||
| 99 | + # 旧循环未结束（含 stop 超时）时禁止再起，避免双循环 | ||
| 100 | + if self._task is not None and not self._task.done(): | ||
| 101 | + logger.warning( | ||
| 102 | + "JobRunner start ignored, still running: job=%s instance=%s", | ||
| 103 | + self._name, | ||
| 104 | + self._instance_id, | ||
| 105 | + ) | ||
| 106 | + return | ||
| 107 | + self._stopped.clear() | ||
| 108 | + self._task = asyncio.create_task( | ||
| 109 | + self._run_forever(), | ||
| 110 | + name=f"periodic-{self._name}-{self._instance_id}", | ||
| 111 | + ) | ||
| 112 | + logger.info( | ||
| 113 | + "JobRunner started: job=%s instance=%s", | ||
| 114 | + self._name, | ||
| 115 | + self._instance_id, | ||
| 116 | + ) | ||
| 117 | + | ||
| 118 | + async def stop(self) -> None: | ||
| 119 | + self._stopped.set() | ||
| 120 | + task = self._task | ||
| 121 | + if task is None: | ||
| 122 | + return | ||
| 123 | + task.cancel() | ||
| 124 | + try: | ||
| 125 | + await asyncio.wait_for(task, timeout=self._stop_timeout_sec) | ||
| 126 | + except asyncio.TimeoutError: | ||
| 127 | + pass | ||
| 128 | + except asyncio.CancelledError: | ||
| 129 | + # runner 被我们 cancel 后 await 会冒 CancelledError，这是正常停机； | ||
| 130 | + # 若 stop() 自己被取消，必须继续往上抛，否则关机流程会误以为停干净了。 | ||
| 131 | + me = asyncio.current_task() | ||
| 132 | + if me is not None and me.cancelling(): | ||
| 133 | + raise | ||
| 134 | + except Exception: | ||
| 135 | + logger.exception( | ||
| 136 | + "JobRunner stop wait failed: job=%s instance=%s", | ||
| 137 | + self._name, | ||
| 138 | + self._instance_id, | ||
| 139 | + ) | ||
| 140 | + finally: | ||
| 141 | + # 仅在仍是同一 task 时清空；超时未结束则保留引用，阻止 start 再开一条 | ||
| 142 | + if self._task is task and task.done(): | ||
| 143 | + self._task = None | ||
| 144 | + elif self._task is task and not task.done(): | ||
| 145 | + logger.warning( | ||
| 146 | + "JobRunner stop timed out, loop still alive: job=%s instance=%s", | ||
| 147 | + self._name, | ||
| 148 | + self._instance_id, | ||
| 149 | + ) | ||
| 150 | + logger.info( | ||
| 151 | + "JobRunner stopped: job=%s instance=%s", | ||
| 152 | + self._name, | ||
| 153 | + self._instance_id, | ||
| 154 | + ) | ||
| 155 | + | ||
| 156 | + async def _sleep_until(self, target: float) -> None: | ||
| 157 | + """一次睡到目标时间；stop 靠 cancel 打断，或醒来后由循环检查 _stopped。""" | ||
| 158 | + if self._stopped.is_set(): | ||
| 159 | + return | ||
| 160 | + delay = target - self._now() | ||
| 161 | + if delay <= 0: | ||
| 162 | + return | ||
| 163 | + await asyncio.sleep(delay) | ||
| 164 | + | ||
| 165 | + async def _run_forever(self) -> None: | ||
| 166 | + while not self._stopped.is_set(): | ||
| 167 | + try: | ||
| 168 | + now = await self._sync_now() | ||
| 169 | + if now is None: | ||
| 170 | + continue | ||
| 171 | + | ||
| 172 | + if self._run_on_start: | ||
| 173 | + self._run_on_start = False | ||
| 174 | + await self._safe_tick(planned_fire=now) | ||
| 175 | + continue | ||
| 176 | + | ||
| 177 | + if self._last_now is not None and now < self._last_now: | ||
| 178 | + logger.warning( | ||
| 179 | + "clock went backwards: job=%s last=%s now=%s", | ||
| 180 | + self._name, | ||
| 181 | + self._last_now, | ||
| 182 | + now, | ||
| 183 | + ) | ||
| 184 | + now = await self._sync_now() | ||
| 185 | + if now is None: | ||
| 186 | + continue | ||
| 187 | + self._last_now = now | ||
| 188 | + | ||
| 189 | + next_ts = self._schedule.next_fire_time(now) | ||
| 190 | + if next_ts <= now: | ||
| 191 | + now = await self._sync_now() | ||
| 192 | + if now is None: | ||
| 193 | + continue | ||
| 194 | + next_ts = self._schedule.next_fire_time(now) | ||
| 195 | + | ||
| 196 | + gather = min(self._gather_window_sec, max(next_ts - now, 0.0)) | ||
| 197 | + wake_at = next_ts - gather | ||
| 198 | + if wake_at > now: | ||
| 199 | + await self._sleep_until(wake_at) | ||
| 200 | + if self._stopped.is_set(): | ||
| 201 | + break | ||
| 202 | + | ||
| 203 | + # 长睡之后再对表，集合窗口用较新的偏移 | ||
| 204 | + now2 = await self._sync_now() | ||
| 205 | + if now2 is None: | ||
| 206 | + continue | ||
| 207 | + if self._missed_fire(now2, planned_fire=next_ts, wake_at=wake_at): | ||
| 208 | + logger.info( | ||
| 209 | + "missed fire, skip tick: job=%s planned=%s now=%s", | ||
| 210 | + self._name, | ||
| 211 | + next_ts, | ||
| 212 | + now2, | ||
| 213 | + ) | ||
| 214 | + continue | ||
| 215 | + await self._safe_tick(planned_fire=next_ts, now=now2) | ||
| 216 | + except asyncio.CancelledError: | ||
| 217 | + logger.debug( | ||
| 218 | + "JobRunner loop cancelled: job=%s instance=%s", | ||
| 219 | + self._name, | ||
| 220 | + self._instance_id, | ||
| 221 | + ) | ||
| 222 | + raise | ||
| 223 | + except Exception: | ||
| 224 | + logger.exception( | ||
| 225 | + "JobRunner loop error: job=%s instance=%s", | ||
| 226 | + self._name, | ||
| 227 | + self._instance_id, | ||
| 228 | + ) | ||
| 229 | + await asyncio.sleep(_TIME_FAIL_SLEEP_SEC) | ||
| 230 | + | ||
| 231 | + def _missed_fire(self, now: float, *, planned_fire: float, wake_at: float) -> bool: | ||
| 232 | + """睡过头：本拍不开火，下一圈按新时间对齐到下一拍。 | ||
| 233 | + | ||
| 234 | + - 已进入再下一拍（``now >= next_fire(planned)``）：一定跳过。 | ||
| 235 | + - 本该提前醒来（集合窗口）却已经过了开火点：也跳过。 | ||
| 236 | + - gather=0 时睡到 T 可能有几毫秒抖动，只要还没到下一拍仍打。 | ||
| 237 | + """ | ||
| 238 | + following = self._schedule.next_fire_time(planned_fire) | ||
| 239 | + if now >= following: | ||
| 240 | + return True | ||
| 241 | + return wake_at < planned_fire and now > planned_fire | ||
| 242 | + | ||
| 243 | + async def _invoke_on_tick(self) -> bool: | ||
| 244 | + """跑业务回调；若协调器支持失锁事件，失锁则取消回调。 | ||
| 245 | + | ||
| 246 | + 返回 True 表示正常跑完，False 表示因失锁被中断。 | ||
| 247 | + """ | ||
| 248 | + lost_ev = getattr(self._coordinator, "lock_lost_event", None) | ||
| 249 | + if lost_ev is None: | ||
| 250 | + await self._on_tick() | ||
| 251 | + return True | ||
| 252 | + | ||
| 253 | + tick_task = asyncio.create_task(self._on_tick()) | ||
| 254 | + lost_task = asyncio.create_task(lost_ev.wait()) | ||
| 255 | + try: | ||
| 256 | + done, _pending = await asyncio.wait( | ||
| 257 | + {tick_task, lost_task}, | ||
| 258 | + return_when=asyncio.FIRST_COMPLETED, | ||
| 259 | + ) | ||
| 260 | + if tick_task in done: | ||
| 261 | + await tick_task | ||
| 262 | + return True | ||
| 263 | + | ||
| 264 | + logger.warning( | ||
| 265 | + "on_tick aborted due to lock lost: job=%s instance=%s", | ||
| 266 | + self._name, | ||
| 267 | + self._instance_id, | ||
| 268 | + ) | ||
| 269 | + return False | ||
| 270 | + except asyncio.CancelledError: | ||
| 271 | + # 先抓住取消，把子任务收干净再往上抛（3.11 里 catch 后才能 await 子任务） | ||
| 272 | + await _cancel_and_wait(tick_task) | ||
| 273 | + await _cancel_and_wait(lost_task) | ||
| 274 | + raise | ||
| 275 | + finally: | ||
| 276 | + await _cancel_and_wait(lost_task) | ||
| 277 | + if not tick_task.done(): | ||
| 278 | + await _cancel_and_wait(tick_task) | ||
| 279 | + | ||
| 280 | + async def _safe_tick(self, *, planned_fire: float, now: Optional[float] = None) -> None: | ||
| 281 | + now_v = now if now is not None else self._now() | ||
| 282 | + claim = await self._coordinator.try_claim( | ||
| 283 | + now=now_v, | ||
| 284 | + instance_id=self._instance_id, | ||
| 285 | + planned_fire=planned_fire, | ||
| 286 | + ) | ||
| 287 | + if claim is None: | ||
| 288 | + logger.debug( | ||
| 289 | + "job lock miss: job=%s instance=%s", | ||
| 290 | + self._name, | ||
| 291 | + self._instance_id, | ||
| 292 | + ) | ||
| 293 | + return | ||
| 294 | + | ||
| 295 | + ok = False | ||
| 296 | + aborted = False | ||
| 297 | + t0 = time.monotonic() | ||
| 298 | + delay_ms = max(0.0, (now_v - planned_fire) * 1000) | ||
| 299 | + try: | ||
| 300 | + finished = await self._invoke_on_tick() | ||
| 301 | + ok = finished | ||
| 302 | + aborted = not finished | ||
| 303 | + except Exception: | ||
| 304 | + logger.exception( | ||
| 305 | + "on_tick failed: job=%s instance=%s", | ||
| 306 | + self._name, | ||
| 307 | + self._instance_id, | ||
| 308 | + ) | ||
| 309 | + finally: | ||
| 310 | + duration_ms = (time.monotonic() - t0) * 1000 | ||
| 311 | + try: | ||
| 312 | + await self._coordinator.release(claim) | ||
| 313 | + except Exception: | ||
| 314 | + logger.exception( | ||
| 315 | + "release after tick failed: job=%s", | ||
| 316 | + self._name, | ||
| 317 | + ) | ||
| 318 | + logger.info( | ||
| 319 | + "tick done: job=%s instance=%s ok=%s aborted=%s delay_ms=%.1f duration_ms=%.1f", | ||
| 320 | + self._name, | ||
| 321 | + self._instance_id, | ||
| 322 | + ok, | ||
| 323 | + aborted, | ||
| 324 | + delay_ms, | ||
| 325 | + duration_ms, | ||
| 326 | + ) | ||
| @@ -0,0 +1,7 @@ | |||
| 1 | +# coding: utf-8 | ||
| 2 | +# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved | ||
| 3 | + | ||
| 4 | +from .base import Schedule | ||
| 5 | +from .interval import IntervalSchedule | ||
| 6 | + | ||
| 7 | +__all__ = ("IntervalSchedule", "Schedule") | ||
| @@ -0,0 +1,14 @@ | |||
| 1 | +# coding: utf-8 | ||
| 2 | +# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved | ||
| 3 | + | ||
| 4 | +"""调度器协议。""" | ||
| 5 | + | ||
| 6 | +from __future__ import annotations | ||
| 7 | + | ||
| 8 | +from typing import Protocol | ||
| 9 | + | ||
| 10 | + | ||
| 11 | +class Schedule(Protocol): | ||
| 12 | + def next_fire_time(self, now: float) -> float: | ||
| 13 | + """返回严格大于 now 的下次触发 unix 秒。""" | ||
| 14 | + ... | ||
| @@ -0,0 +1,23 @@ | |||
| 1 | +# coding: utf-8 | ||
| 2 | +# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved | ||
| 3 | + | ||
| 4 | +"""固定间隔、挂钟边界对齐的调度。""" | ||
| 5 | + | ||
| 6 | +from __future__ import annotations | ||
| 7 | + | ||
| 8 | +import math | ||
| 9 | + | ||
| 10 | + | ||
| 11 | +class IntervalSchedule: | ||
| 12 | + """下一个 interval 边界触发（整秒/整 N 秒对齐）。""" | ||
| 13 | + | ||
| 14 | + def __init__(self, interval_sec: int = 1) -> None: | ||
| 15 | + self._interval_sec = max(int(interval_sec), 1) | ||
| 16 | + | ||
| 17 | + | ||
| 18 | + def interval_sec(self) -> int: | ||
| 19 | + return self._interval_sec | ||
| 20 | + | ||
| 21 | + def next_fire_time(self, now: float) -> float: | ||
| 22 | + interval = self._interval_sec | ||
| 23 | + return (math.floor(now / interval) + 1) * interval | ||
| @@ -3,7 +3,7 @@ | |||
| 3 | 3 | ||
| 4 | """SystemContext（设计 §8）。 | 4 | """SystemContext（设计 §8）。 |
| 5 | 5 | ||
| 6 | -- 进程级 SystemContext（lifespan 创建/释放）：db / redis / settings / logger / 原语工厂。 | 6 | +- 进程级 SystemContext（lifespan 创建/释放）：db / redis / settings / logger / 原语工厂 / 周期任务。 |
| 7 | - 请求级 RequestContext 由 ``for_request(envelope)`` 派生。 | 7 | - 请求级 RequestContext 由 ``for_request(envelope)`` 派生。 |
| 8 | - 事务：``async with ctx.transaction() as s`` 取 SQLAlchemy session（多操作原子）。 | 8 | - 事务：``async with ctx.transaction() as s`` 取 SQLAlchemy session（多操作原子）。 |
| 9 | - 硬约束：handler 禁止读写模块级可变状态——无内存状态的多副本。 | 9 | - 硬约束：handler 禁止读写模块级可变状态——无内存状态的多副本。 |
| @@ -18,7 +18,7 @@ import time | |||
| 18 | from dataclasses import replace | 18 | from dataclasses import replace |
| 19 | from inspect import isawaitable | 19 | from inspect import isawaitable |
| 20 | from contextlib import asynccontextmanager | 20 | from contextlib import asynccontextmanager |
| 21 | -from typing import Any, AsyncIterator, Iterable, TypeVar, overload | 21 | +from typing import TYPE_CHECKING, Any, AsyncIterator, Awaitable, Callable, Iterable, TypeVar, overload |
| 22 | from uuid import uuid4 | 22 | from uuid import uuid4 |
| 23 | 23 | ||
| 24 | from sqlalchemy import text | 24 | from sqlalchemy import text |
| @@ -38,6 +38,9 @@ from .kubernetes import KubernetesOperations | |||
| 38 | from .audit import AuditEvent, AuditLogger, LoggingAuditLogger, NoopAuditLogger | 38 | from .audit import AuditEvent, AuditLogger, LoggingAuditLogger, NoopAuditLogger |
| 39 | from .request_context import RequestContext | 39 | from .request_context import RequestContext |
| 40 | 40 | ||
| 41 | +if TYPE_CHECKING: | ||
| 42 | + from .periodic.runner import JobRunner | ||
| 43 | + | ||
| 41 | _logger = logging.getLogger("openjiuwen_runtime.service") | 44 | _logger = logging.getLogger("openjiuwen_runtime.service") |
| 42 | TRequest = TypeVar("TRequest") | 45 | TRequest = TypeVar("TRequest") |
| 43 | 46 | ||
| @@ -393,6 +396,30 @@ class SystemContext: | |||
| 393 | "multi-replica deployment requires a distributed lock backend" | 396 | "multi-replica deployment requires a distributed lock backend" |
| 394 | ) | 397 | ) |
| 395 | 398 | ||
| 399 | + # -------------------------------------------------------------- 周期任务 | ||
| 400 | + def create_single_leader_job( | ||
| 401 | + self, | ||
| 402 | + *, | ||
| 403 | + name: str, | ||
| 404 | + on_tick: Callable[[], Awaitable[None]], | ||
| 405 | + interval_sec: int = 1, | ||
| 406 | + gather_window_sec: float = 0.08, | ||
| 407 | + lock_key: str = "", | ||
| 408 | + run_on_start: bool = False, | ||
| 409 | + ) -> "JobRunner": | ||
| 410 | + """用本进程的 redis / instance_id 组装主备周期任务。""" | ||
| 411 | + from .periodic import create_single_leader_job | ||
| 412 | + | ||
| 413 | + return create_single_leader_job( | ||
| 414 | + self, | ||
| 415 | + name=name, | ||
| 416 | + on_tick=on_tick, | ||
| 417 | + interval_sec=interval_sec, | ||
| 418 | + gather_window_sec=gather_window_sec, | ||
| 419 | + lock_key=lock_key, | ||
| 420 | + run_on_start=run_on_start, | ||
| 421 | + ) | ||
| 422 | + | ||
| 396 | # -------------------------------------------------------------- 请求上下文 | 423 | # -------------------------------------------------------------- 请求上下文 |
| 397 | 424 | ||
| 398 | def for_request(self, request: Envelope[TRequest]) -> RequestContext[TRequest]: | 425 | def for_request(self, request: Envelope[TRequest]) -> RequestContext[TRequest]: |