ulthon/ulthon_admin
1
0
Fork 0
mirror of https://gitee.com/ulthon/ulthon_admin.git synced 2026-08-30 04:35:33 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(timer): runLoop 热加载并发实例与运行时配置参数

Browse Source 
TimerBase 实现 shouldExecuteTask 手动触发原子消费、reloadRequestList 合并式扩缩容、runLoop 脏标记检测+定时兜底+有效域钳制+drain 优雅缩容+checkTriggerTtl 超时复位。config/timer.php 新增 trigger_ttl/force_reload_interval/drain_max_lifetime 三参数。

Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>
This commit is contained in:
augushong
2026-07-25 21:57:25 +08:00
parent 236a343fa9
commit f478ae0e4e
2 changed files with 545 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
config/timer.php
View File

@@ -12,8 +12,16 @@ $config = [
'max_handles' => 100, // curl multi 最大并发句柄数
'select_timeout' => 0.001, // curl_multi_select 超时（秒）
// T6: runLoop 热加载配置
'force_reload_interval' => (int) Env::get('timer.force_reload_interval', 15), // 兜底 reload 间隔（秒），防 cache 丢失导致脏标记漏检
'trigger_ttl' => (int) Env::get('timer.trigger_ttl', 300), // 手动触发 TTL（秒），超时未消费自动复位 manual_trigger
// 清理日志保留天数（debug_log 表）
'clear_log_days' => Env::get('timer.clear_log_days', 3),
// T7: drain 最长存活上限（秒），draining 实例超过该时长强制移除（防永不 drain）
// 默认与任务 timeout 一致（86400s），可通过 .env timer.drain_max_lifetime 覆盖
'drain_max_lifetime' => (int) Env::get('timer.drain_max_lifetime', 86400),
];
return $config;

556
extend/base/common/command/TimerBase.php
View File

@@ -27,6 +27,10 @@ class TimerBase extends Command
protected $requestList;
protected $callList;
// T6: 热加载状态（跨 runLoop 轮次保持）
protected $lastVersion; // 上次观测到的 timer_config_version（脏标记比对基准）
protected $lastForceReload; // 上次兜底 reload 的时间戳
protected function configure()
{
parent::configure();
@@ -81,6 +85,12 @@ class TimerBase extends Command
$system_host_call_list = TimerService::generateTaskInstanceFromConfig($system_host_register);
$call_list = array_merge($call_list, $system_host_call_list);
// 运行时可变结构基础：每个 site 实例注入 state 字段（active/draining）
// reload 时按 task_name 粒度 diff 合并，不全量重建（D15/M10）
foreach ($request_list as $state_key => $state_item) {
$request_list[$state_key]['state'] = 'active';
}
$this->host = $host;
$this->siteDomain = $site_domain;
$this->siteHost = $site_host;
@@ -103,7 +113,7 @@ class TimerBase extends Command
*/
protected function shouldExecuteTask($task_item): bool
{
// call 类型任务不受 run_type 调度层影响
// (1) call 类型任务不受 run_type 调度层影响
if (($task_item['type'] ?? '') === 'call') {
return true;
}
@@ -125,13 +135,48 @@ class TimerBase extends Command
return true;
}
// status=0 表示任务已禁用
// (2) 手动触发优先（穿透 status）：跨 run_type 通用，定向到本节点。
// 原子条件 UPDATE —— WHERE 含 trigger_node_id = current_node_id（D19：无 IS NULL 回退分支，
// 单标志位 manual_trigger + 单赢家 UPDATE 只能定向单节点）。
// affected_rows=1 表示本节点消费了这次触发，立即 return true；
// affected_rows=0 表示本节点不是目标节点 → fall through 到步骤 (3)/(4) 走正常调度
// （manual 类型会在步骤 (4) return false，即仅目标节点执行）。
if (!empty($config['manual_trigger'])) {
$current_node_id = HostService::getNodeId();
try {
$affected = Db::name('system_timer_config')
->where('task_name', $task_name)
->where('manual_trigger', 1)
->where('trigger_node_id', $current_node_id)
->update([
'manual_trigger' => 0,
'trigger_node_id' => null,
'last_trigger_time' => null,
'update_time' => time(),
]);
} catch (\Throwable $e) {
Log::warning('shouldExecuteTask manual trigger consume exception: ' . $e->getMessage());
$affected = 0;
}
if ($affected === 1) {
Log::info('[timer] trigger task=' . $task_name . ' node=' . $current_node_id . ' consumed');
return true;
}
// affected=0：本节点不是目标，fall through（不 return false）
}
// (3) status=0 表示任务已禁用（只拦截自动调度，手动已在步骤 (2) 穿透）
if (isset($config['status']) && (int) $config['status'] === 0) {
return false;
}
$run_type = $config['run_type'] ?? 'auto';
// (4) run_type 调度
switch ($run_type) {
case 'main':
// 仅主节点执行
@@ -180,18 +225,7 @@ class TimerBase extends Command
return true;
case 'manual':
// 手动触发：检查 manual_trigger=1，执行一次后重置为 0
if (!empty($config['manual_trigger'])) {
Db::name('system_timer_config')
->where('task_name', $task_name)
->update([
'manual_trigger' => 0,
'update_time' => time(),
]);
return true;
}
// manual 类型：永不自动执行，执行全靠步骤 (2) 的手动触发
return false;
default:
@@ -204,10 +238,12 @@ class TimerBase extends Command
{
$host = $this->host;
$site_host = $this->siteHost;
$request_list = $this->requestList;
$output = $this->output;
$input = $this->input;
// D20: 不再用局部副本（$request_list = $this->requestList 是 COW 拷贝），
// 直接遍历 $this->requestList 属性。PHP foreach 进入时拷贝数组值，
// while 每轮重新 foreach → reload 修改属性后下一轮可见；mid-iteration append 安全。
$handler = new CurlMultiHandler([
'select_timeout' => Config::get('timer.select_timeout', 0.001),
'max_handles' => Config::get('timer.max_handles', 100),
@@ -228,14 +264,65 @@ class TimerBase extends Command
$pending = [];
// T6: 热加载状态初始化（进程启动时快照当前 version，避免首轮误触发 reload）
if (!isset($this->lastVersion)) {
$this->lastVersion = Cache::get('timer_config_version', 0); // 故意不打 tag（m1，防 Cache::tag('system_timer')->clear() 连带丢）
$this->lastForceReload = time();
}
while (true) {
try {
// T6: 热加载触发（脏标记检测 + 定时兜底）
// 脏标记：timer_config_version 变化（T4 trigger/编辑 commit 时 +1）
// 兜底：每 force_reload_interval 秒强制 reload（防 cache 丢失导致脏标记漏检）
$current_version = Cache::get('timer_config_version', 0); // 故意不打 tag（m1）
$force_interval = (int) Config::get('timer.force_reload_interval', 15);
if ($current_version !== $this->lastVersion || (time() - $this->lastForceReload) >= $force_interval) {
$effective_map = $this->computeEffectiveConcurrency(); // 钳制有效域
$this->reloadRequestList($effective_map); // T5 合并方法（不全量重建）
$this->checkTriggerTtl(); // TTL 复位检查
$this->lastVersion = $current_version;
$this->lastForceReload = time();
}
$has_new_task = false;
// T6/D22: 批量预取 task configs（manual 预检用）
// shouldExecuteTask 内部自查询不变（T3 区域不改），此处仅服务 runLoop 层 manual 预检
$task_configs_batch = [];
if (!empty($this->requestList)) {
try {
$batch_rows = Db::name('system_timer_config')->select();
foreach ($batch_rows as $r) {
$task_configs_batch[$r['task_name']] = $r;
}
} catch (\Throwable $e) {
// 降级：manual 预检失效，走正常 throttle（不阻塞主循环）
}
}
// --- site 任务：非阻塞发火 ---
foreach ($request_list as $request_item) {
// D20: 直接遍历 $this->requestList 属性（非局部副本），reload 后下一轮可见
foreach ($this->requestList as $arr_key => $request_item) {
$name = $request_item['name'];
$key = $name . '_' . $request_item['concurrency_id'];
$state = $request_item['state'] ?? 'active';
// draining 实例：不发新请求；pending 无此 key 时移除（drain 完成）
// 保留在结构中直到在飞请求结束，保证 drain 连续性
// T7: 超过 drain_max_lifetime 强制移除（防永不 drain，如任务卡死/异常）
if ($state === 'draining') {
$drain_max = (int) Config::get('timer.drain_max_lifetime', Config::get('timer.timeout', 86400));
$drain_started = $request_item['drain_started_at'] ?? 0;
$timed_out = ($drain_started > 0 && (time() - $drain_started) > $drain_max);
if (!isset($pending[$key]) || $timed_out) {
// drain 完成（pending 空）或超时（防 86400s 永不 drain）
unset($this->requestList[$arr_key]);
Log::info('[timer] drain-done task=' . $name . ' id=' . $request_item['concurrency_id'] . ($timed_out ? ' (timeout)' : ''));
}
continue;
}
// 已在飞 -> 跳过
if (isset($pending[$key])) {
@@ -244,10 +331,22 @@ class TimerBase extends Command
$cache_key = 'timer_request_' . $name . '_' . $request_item['concurrency_id'];
$cache_tag = 'system_timer';
$last_exec_time = Cache::get($cache_key, 0);
if ($last_exec_time >= time() - $request_item['frequency']) {
continue;
// T6/D22: manual 预检（在 cache throttle 检查之前）
// manual_trigger=1 且定向本节点 → bypass 频率节流检查，立即可达 shouldExecuteTask
// 关键：仍写节流 cache（防 all/main 无 DB 频率兜底的任务跨 tick 双发）
$task_cfg = $task_configs_batch[$name] ?? null;
$bypass_throttle = !empty($task_cfg['manual_trigger'])
&& !empty($task_cfg['trigger_node_id'])
&& $task_cfg['trigger_node_id'] === HostService::getNodeId();
if (!$bypass_throttle) {
$last_exec_time = Cache::get($cache_key, 0);
if ($last_exec_time >= time() - $request_item['frequency']) {
continue;
}
}
// 无论 bypass 与否都写节流 cache（D22：仅 bypass 节流检查，不 bypass 节流写入）
Cache::tag($cache_tag)->set($cache_key, time());
// run_type 调度检查（Cache 节流之后、实际执行之前）
@@ -328,4 +427,423 @@ class TimerBase extends Command
}
}
}
/**
* 合并式 reload：按 task_name 粒度 diff 当前 requestList 与目标 concurrency，
* 扩容 append（concurrency_id 从 max+1 续编，D3），缩容把高 id 置 state=draining（保留在结构中）。
*
* 不全量重建（D15/M10）：保留其他 task 的 draining 实例，不影响其 drain 连续性。
* drain 完成的移除由 runLoop 遍历时处理（pending 空 + state=draining → unset）。
*
* 崩溃恢复契约（M4）：drain 状态纯内存，进程重启后以 DB concurrency 为准全重建
* （generateAllRequestList 产出 0..N-1 完整分片），放弃 drain 连续性——
* 重启前正在 draining 的实例不会恢复 draining，直接按新 DB 值重建为 active。
*
* 注意：本方法只提供数据结构 + 合并逻辑，**不实现 reload 触发**
* （脏标记检测/定时兜底/有效域钳制 = T6）。T6 在 runLoop 检测到变化时调用本方法，
* 可传入已钳制的 effective_concurrency_map 覆盖默认的 DB ?? code 解析。
*
* @param array|null $effective_concurrency_map [task_name => effective_int]
* 调用方（T6）钳制后的有效并发映射；
* null 时本方法内部读 DB concurrency ?? 代码默认（T5 简化版）。
* 未出现的 task_name 保持不变（保守）。
*/
protected function reloadRequestList(?array $effective_concurrency_map = null): void
{
// (1) 收集当前 requestList 中各 task 的实例（按 name 分组）
// 记录每个实例的数组 key（用于修改 state / append 定位）与 state
$by_task = []; // [task_name => [concurrency_id => ['arr_key' => int, 'state' => string]]]
foreach ($this->requestList as $arr_key => $item) {
$name = $item['name'];
$cid = $item['concurrency_id'];
$state = $item['state'] ?? 'active';
$by_task[$name][$cid] = ['arr_key' => $arr_key, 'state' => $state];
}
if (empty($by_task)) {
return;
}
// (2) 确定每个 task 的目标 concurrency
// 优先用调用方传入的 effective map（T6 钳制后），否则内部读 DB ?? code（T5 简化）
$code_defaults = $this->collectCodeConcurrencyDefaults($by_task);
$target_map = $effective_concurrency_map ?? $this->resolveEffectiveConcurrencyFromDb($code_defaults);
// (3) 对每个已存在的 task 做 diff（合并，不重建）
foreach ($by_task as $name => $instances) {
if (!isset($target_map[$name])) {
// 目标映射未覆盖该 task（可能是新增任务，归 T7 处理；这里保守不动）
continue;
}
$target = (int) $target_map[$name];
if ($target < 0) {
$target = 0;
}
// 分离 active / draining
$active_ids = [];
foreach ($instances as $cid => $meta) {
if ($meta['state'] === 'active') {
$active_ids[$cid] = $meta;
}
}
$active_count = count($active_ids);
if ($target > $active_count) {
// 扩容：append，concurrency_id 从 max(所有现有 id)+1 续编（D3）
// 含 draining 实例的 id 也参与 max 计算，保证 id 单调递增不重用
$all_ids = array_keys($instances);
$max_id = empty($all_ids) ? -1 : max($all_ids);
$need = $target - $active_count;
$code_concurrency_count = $code_defaults[$name] ?? $target;
for ($i = 0; $i < $need; $i++) {
$max_id++;
$new_item = $this->buildScaledInstance($name, $max_id, $target, $code_concurrency_count);
$this->requestList[] = $new_item;
}
$msg = '[timer] reload task=' . $name . ' concurrency ' . $active_count . '->' . $target . ' effective=' . $target;
Log::info($msg);
$this->output->writeln($msg);
} elseif ($target < $active_count) {
// 缩容：把高 id 置 draining（concurrency_id 降序，直到 active_count == target）
// D3：缩容先 drain 高 id，concurrency_id=0 永远第一分片（最后才 drain）
$sorted_active = $active_ids;
krsort($sorted_active); // 降序
$to_drain_count = $active_count - $target;
$drained = 0;
foreach ($sorted_active as $cid => $meta) {
if ($drained >= $to_drain_count) {
break;
}
$arr_key = $meta['arr_key'];
// T7: 置 draining 时清理节流 cache（防下次扩容回原值被旧窗口误判跳过）
// 注意：Cache::tag()->delete() 不存在（TagSet 无 delete 方法），用 Cache::delete() 直接删
$cache_key = 'timer_request_' . $name . '_' . $cid;
Cache::delete($cache_key);
$this->requestList[$arr_key]['state'] = 'draining';
// T7: 记录 drain 开始时间，runLoop 据此判断 drain 超时强制移除
$this->requestList[$arr_key]['drain_started_at'] = time();
$msg = '[timer] drain task=' . $name . ' id=' . $cid;
Log::info($msg);
$this->output->writeln($msg);
$drained++;
}
}
// target == active_count：无变化
}
}
/**
* 从当前 requestList 实例中提取各 task 的代码默认 concurrency
* （实例的 concurrency_count 字段，由 generateTaskInstanceFromConfig 注入）。
*
* @param array $by_task reloadRequestList 内部分组结构
* @return array [task_name => code_default_concurrency_int]
*/
protected function collectCodeConcurrencyDefaults(array $by_task): array
{
$defaults = [];
foreach ($by_task as $name => $instances) {
// 取任意一个实例的 concurrency_count 作为代码默认（同 task 所有实例一致）
foreach ($instances as $meta) {
if (isset($this->requestList[$meta['arr_key']]['concurrency_count'])) {
$defaults[$name] = (int) $this->requestList[$meta['arr_key']]['concurrency_count'];
}
break;
}
if (!isset($defaults[$name])) {
$defaults[$name] = 1;
}
}
return $defaults;
}
/**
* 读 DB system_timer_config 的 concurrency 字段，NULL 用代码默认（T5 简化版）。
* T6 会替换为钳制后的 effective map（max/min/Σ 预算），本方法仅作 fallback。
*
* @param array $code_defaults [task_name => code_default_int]
* @return array [task_name => effective_int]
*/
protected function resolveEffectiveConcurrencyFromDb(array $code_defaults): array
{
if (empty($code_defaults)) {
return [];
}
try {
$rows = Db::name('system_timer_config')
->where('task_name', 'in', array_keys($code_defaults))
->column('concurrency', 'task_name');
} catch (\Throwable $e) {
Log::warning('reloadRequestList: read system_timer_config failed - ' . $e->getMessage() . ' (fallback to code defaults)');
return $code_defaults;
}
$map = [];
foreach ($code_defaults as $name => $code_val) {
$db_val = $rows[$name] ?? null;
// NULL → 代码默认（字段注释：NULL=继承代码默认）
$map[$name] = ($db_val === null) ? $code_val : (int) $db_val;
}
return $map;
}
/**
* 基于 requestList 中同 task 的现有实例构造一个新的分片实例（用于扩容 append）。
* 克隆模板实例，重写 target 的 query 参数（concurrency_id/concurrency_count/host_id），
* 保证 target URL 与 generateTaskInstanceFromConfig 的产出同构。
*
* @param string $task_name 任务名
* @param int $new_cid 新 concurrency_id（max+1 续编）
* @param int $new_count 新 concurrency_count（目标 concurrency）
* @param int $code_concurrency_count 代码默认 concurrency（fallback）
* @return \app\common\model\VirtualModel 带新分片参数的实例（state=active）
*/
protected function buildScaledInstance(string $task_name, int $new_cid, int $new_count, int $code_concurrency_count = 1)
{
// 从 requestList 找一个同 task 的现有实例作为模板（取第一个）
$template = null;
foreach ($this->requestList as $item) {
if ($item['name'] === $task_name) {
$template = $item;
break;
}
}
if ($template === null) {
throw new \RuntimeException('TimerBase::buildScaledInstance: cannot find template instance for task: ' . $task_name);
}
// clone 后修改（think\Model 浅拷贝，data 数组 COW，修改不影响原对象）
$new_item = clone $template;
// 重写 target 的 query 参数：concurrency_id / concurrency_count / host_id
$target = $new_item['target'];
$target_info = parse_url($target);
$query_params = [];
if (isset($target_info['query'])) {
parse_str($target_info['query'], $query_params);
}
$query_params['concurrency_id'] = $new_cid;
$query_params['concurrency_count'] = $new_count;
$query_params['host_id'] = HostService::getNodeId();
$query_params['task_name'] = $task_name;
$target_info['query'] = http_build_query($query_params);
$new_item['target'] = unparse_url($target_info);
$new_item['concurrency_id'] = $new_cid;
$new_item['concurrency_count'] = $new_count;
$new_item['state'] = 'active';
return $new_item;
}
/**
* T6: 计算各 task 钳制后的有效并发数（reload 前）.
*
* effective = max(1, min(DB concurrency ?? 代码默认, 控制器 cap, 剩余 max_handles 预算))
*
* - DB concurrency = NULL → 继承代码默认（字段语义）
* - DB concurrency <= 0 → 钳为 1（防静默吞任务，M5）
* - 控制器 cap：本期用"代码默认 concurrency_count"近似（实例携带），精确控制器 $concurrency 反射留 TODO（T10+）
* - Σ 预算：max_handles - 其他 task 已占 active 数（本 task 当前 active 先释放再重分配）
*
* 超限（effective != raw）时 Log::info + writeln 告警（D21 clamp 证据）。
*
* @return array [task_name => effective_int] 传给 reloadRequestList 的 effective_map
*/
protected function computeEffectiveConcurrency(): array
{
// (1) 收集当前 requestList 各 task 分组（复用 T5 的分组结构）
$by_task = [];
foreach ($this->requestList as $arr_key => $item) {
$name = $item['name'];
$state = $item['state'] ?? 'active';
$by_task[$name][$item['concurrency_id']] = ['arr_key' => $arr_key, 'state' => $state];
}
if (empty($by_task)) {
return [];
}
// (2) 代码默认 concurrency（实例的 concurrency_count，T5 collectCodeConcurrencyDefaults）
$code_defaults = $this->collectCodeConcurrencyDefaults($by_task);
// (3) 读 DB concurrency（NULL → 代码默认）
try {
$rows = Db::name('system_timer_config')
->where('task_name', 'in', array_keys($code_defaults))
->column('concurrency', 'task_name');
} catch (\Throwable $e) {
Log::warning('computeEffectiveConcurrency: read system_timer_config failed - ' . $e->getMessage() . ' (fallback to code defaults)');
$rows = [];
}
// (4) Σ 预算：max_handles 减去其他 task 已占 active 数
$max_handles = (int) Config::get('timer.max_handles', 100);
$active_total = 0;
foreach ($this->requestList as $item) {
if (($item['state'] ?? 'active') === 'active') {
$active_total++;
}
}
// (5) 逐 task 钳制
$effective_map = [];
foreach ($code_defaults as $name => $code_val) {
$db_val = $rows[$name] ?? null;
// NULL → 代码默认；否则取 DB 值
$raw = ($db_val === null) ? (int) $code_val : (int) $db_val;
// 控制器 cap：反射 site target 对应控制器类的 $concurrency 属性（含继承链）
// TODO: 当前仅解析 {module}/{dir.Controller}/{action} 格式，多级模块/自定义路由留 T10+ 增强
$cap = $this->resolveControllerCap($name, $by_task[$name] ?? []);
// 本 task 当前 active 数（预算计算时先释放本 task 占用，再按 effective 重分配）
$current_active = 0;
if (isset($by_task[$name])) {
foreach ($by_task[$name] as $meta) {
if ($meta['state'] === 'active') {
$current_active++;
}
}
}
$budget_after_release = $max_handles - ($active_total - $current_active);
$effective = max(1, min($raw, $cap, $budget_after_release));
if ($effective !== $raw) {
$db_display = $db_val === null ? 'null' : (string) $db_val;
$msg = '[timer] clamp task=' . $name . ' db=' . $db_display . ' effective=' . $effective;
Log::info($msg);
$this->output->writeln($msg);
}
$effective_map[$name] = $effective;
}
return $effective_map;
}
/**
* 反射 site target 对应控制器类的 $concurrency 属性作为 cap.
*
* 解析 target 路径 /{module}/{dir.Controller}/{action} → app\{module}\controller\{dir\Controller}，
* 用 ReflectionClass::getDefaultProperties() 读取继承链上的 $concurrency 默认值（含 Base 层）。
* 反射结果缓存到 $controllerCapCache（控制器类不变，进程生命周期内稳定）。
*
* @param string $task_name 任务名（缓存 key）
* @param array $instances reloadRequestList 分组结构（取 target 用）
* @return int 控制器 cap（反射失败 fallback 1）
*/
protected $controllerCapCache = [];
protected function resolveControllerCap(string $task_name, array $instances): int
{
if (isset($this->controllerCapCache[$task_name])) {
return $this->controllerCapCache[$task_name];
}
$cap = 1; // fallback（保守）
// 从分组实例取 target（取第一个实例）
$target = null;
foreach ($instances as $meta) {
$arr_key = $meta['arr_key'] ?? null;
if ($arr_key !== null && isset($this->requestList[$arr_key]['target'])) {
$target = $this->requestList[$arr_key]['target'];
break;
}
}
if ($target) {
try {
$path = parse_url($target, PHP_URL_PATH);
$path = trim($path, '/');
$parts = explode('/', $path);
if (count($parts) >= 2) {
// {module}/{dir.Controller}/{action} → app\{module}\controller\{dir\Controller}
// 点表示子目录：timer.ClearLog → timer\ClearLog
$controller_segment = str_replace('.', '\\', $parts[1]);
$controller_class = 'app\\' . $parts[0] . '\\controller\\' . $controller_segment;
if (class_exists($controller_class)) {
$defaults = (new \ReflectionClass($controller_class))->getDefaultProperties();
if (isset($defaults['concurrency'])) {
$cap = (int) $defaults['concurrency'];
}
}
}
} catch (\Throwable $e) {
// 反射失败：保守 fallback cap=1（控制器层 T8 cap 防御仍兜底）
Log::warning('resolveControllerCap: reflect failed for task=' . $task_name . ' - ' . $e->getMessage());
}
}
if ($cap < 1) {
$cap = 1;
}
$this->controllerCapCache[$task_name] = $cap;
return $cap;
}
/**
* T6: 手动触发 TTL 复位检查（reload 兜底分支调用）.
*
* manual_trigger=1 但 last_trigger_time 超过 trigger_ttl（默认 300s）未消费的任务，
* 视为触发请求丢失（目标节点离线/异常），自动复位 manual_trigger=0/trigger_node_id=NULL/last_trigger_time=NULL。
*
* 防 manual_trigger 永久卡住（D17）。
*/
protected function checkTriggerTtl(): void
{
$ttl = (int) Config::get('timer.trigger_ttl', 300);
try {
$rows = Db::name('system_timer_config')
->where('manual_trigger', 1)
->select();
} catch (\Throwable $e) {
Log::warning('checkTriggerTtl: read system_timer_config failed - ' . $e->getMessage());
return;
}
$now = time();
foreach ($rows as $r) {
$last_trigger = $r['last_trigger_time'] ?? null;
if (empty($last_trigger)) {
continue; // 无时间戳（异常数据），跳过等下次
}
if (($now - (int) $last_trigger) <= $ttl) {
continue; // 未超时
}
try {
Db::name('system_timer_config')
->where('id', $r['id'])
->where('manual_trigger', 1) // 防并发消费后误复位
->update([
'manual_trigger' => 0,
'trigger_node_id' => null,
'last_trigger_time' => null,
'update_time' => $now,
]);
$msg = '[timer] trigger-expire task=' . ($r['task_name'] ?? '') . ' node=' . ($r['trigger_node_id'] ?? '') . ' reset';
Log::info($msg);
$this->output->writeln($msg);
} catch (\Throwable $e) {
Log::warning('checkTriggerTtl: update failed for task=' . ($r['task_name'] ?? '') . ' - ' . $e->getMessage());
}
}
}
}