diff --git a/config/timer.php b/config/timer.php index 92c64f4..80d4c10 100644 --- a/config/timer.php +++ b/config/timer.php @@ -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; diff --git a/extend/base/common/command/TimerBase.php b/extend/base/common/command/TimerBase.php index c182377..3e19bc0 100644 --- a/extend/base/common/command/TimerBase.php +++ b/extend/base/common/command/TimerBase.php @@ -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()); + } + } + } }