docs(skill): 修正队列设计技能审核问题（phpredis Stream API 用法、redis store 环境前提、示例缺陷）

.agents/skills/ulthon-queue/SKILL.md

@@ -78,7 +78,7 @@ Db::name('task')
 
 | 字段 | 类型 | 必要性 | 说明 |
 |------|------|--------|------|
-| `status` | tinyint | 必须 | 0待处理 1处理中 2成功 3失败。若业务已有状态字段能表达"待处理"可复用，但独立字段不与业务状态机耦合，更稳 |
+| `status` | tinyint | 必须 | 0待处理 1处理中 2成功 3失败 4已入队（4 仅模式 C 使用）。若业务已有状态字段能表达"待处理"可复用，但独立字段不与业务状态机耦合，更稳 |
 | `claim_time` | int | 必须 | 抢占时间戳，超时回收的唯一依据 |
 | `claim_token` | varchar(32) | 批量抢占需要 | 本轮抢占令牌，区分"这轮我抢到哪几条"。每次只抢 1 条时可省 |
 | `worker_id` | varchar(64) | 可选 | 消费者标识，排查用 |
@@ -105,8 +105,9 @@ use think\facade\Db;
 class NotifyConsumer extends TimerController
 {
     protected $concurrency = 4;    // 消费者数（= 并发上限 cap，后台可在 cap 内动态调整）
-    protected $frequency = 5;
     protected $maxRunTime = 3;     // 时间窗口，不超过 frequency 的一半
+    // 一般不设控制器级 $frequency：防刷会在窗口内拦截手动触发（详见「与 timer 机制的组合」），
+    // 调度节流已由定时器侧 Cache 节流承担；确需防外部刷 URL 再设，且应小于调度 frequency
 
     public function do()
     {
@@ -231,9 +232,18 @@ class NotifyReclaim extends TimerController
 class FixDataTask extends TimerController
 {
     protected $concurrency = 4;      // 并行分段数
-    protected $frequency = 5;
+    protected $frequency = 20;       // 调度频率，与 maxRunTime 保持 >= 2 倍关系
     protected $maxRunTime = 10;
 
+    // 基类只注入 $concurrencyId 属性，concurrencyCount 需自行从请求参数取
+    protected int $concurrencyCount = 1;
+
+    public function initialize()
+    {
+        parent::initialize();
+        $this->concurrencyCount = (int) $this->request->param('concurrency_count', 1);
+    }
+
     public function do()
     {
         // 启动时按 min/max 划分连续区间，分片 i 处理 [lo, hi)，各自独立游标
@@ -293,6 +303,24 @@ DB --[生产者 timer:游标捞取+入队,串行无竞争]--> Redis Stream --[N
 
 引入前提：已有（或多节点必须有的）共享 Redis。不要为此单独引入一套新基础设施，也**不要用 MySQL MEMORY 表替代**（丢持久性、仍是 DB 竞争模式、锁并发不如 Redis 原子操作；不想用 Redis 就用模式 A）。
 
+### 环境前提（缺失会直接报错）
+
+1. **`config/cache.php` 需配置 `redis` store**——框架默认只带 `file` / `login` / `mcp_session`，未配置时 `Cache::store('redis')` 抛 `InvalidArgumentException: Store [redis] not found`：
+
+```php
+// config/cache.php 的 stores 中新增
+'redis' => [
+    'type'     => 'Redis',
+    'host'     => env('redis.host', '127.0.0.1'),
+    'port'     => env('redis.port', 6379),
+    'password' => env('redis.password', ''),
+    'select'   => (int) env('redis.select', 0),
+],
+```
+
+2. **需要 phpredis 扩展**——`handler()` 返回原生 `\Redis` 的前提（未装 phpredis 且未装 predis 时抛 `BadFunctionCallException: not support: redis`）。本节示例的 `xAdd` / `xReadGroup` / `xAck` / `xAutoClaim` 驼峰方法是 phpredis 扩展的 API，Predis 环境（`handler()` 返回 `\Predis\Client`）方法名不同，示例不适用
+3. **版本要求**：Redis 服务器 Stream 与消费组需 5.0+，`XAUTOCLAIM` 需 6.2+（旧版本用 `XPENDING` + `XCLAIM` 组合替代）；phpredis 扩展需 5.3+ 基础 Stream 支持、6.0.0+ 才有 `xAutoClaim`
+
 ### 生产者（concurrency=1，run_type=auto 防多节点重复投递）
 
 ```php
@@ -337,7 +365,6 @@ class NotifyProducer extends TimerController
 class NotifyStreamConsumer extends TimerController
 {
     protected $concurrency = 4;
-    protected $frequency = 5;
     protected $maxRunTime = 3;
 
     public function do()
@@ -348,16 +375,18 @@ class NotifyStreamConsumer extends TimerController
         $max_time = time() + $this->maxRunTime;
 
         do {
+            // 返回结构嵌套流名层：[流名 => [entry_id => [字段 => 值]]]，遍历前先取 $msgs['notify_stream']
+            // 无新消息时旧版 Redis 返回 false（nil 应答）、Redis 8+ 返回空数组，统一 truthy 判断
             $msgs = $redis->xReadGroup('notify_group', $consumer, ['notify_stream' => '>'], 10);
 
-            if (empty($msgs)) {
+            if (empty($msgs['notify_stream'])) {
                 // 新消息为空 -> 认领超时未确认的（崩溃恢复）
                 $this->claimPending($redis, $consumer);
                 usleep(200000);
                 continue;
             }
 
-            foreach ($msgs as $entry_id => $fields) {
+            foreach ($msgs['notify_stream'] as $entry_id => $fields) {
                 $this->processOne($fields['id']);       // 内部含幂等回写
                 $redis->xAck('notify_stream', 'notify_group', [$entry_id]);
             }
@@ -368,16 +397,18 @@ class NotifyStreamConsumer extends TimerController
 
     protected function ensureGroup(\Redis $redis): void
     {
-        try {
-            $redis->xGroupCreate('notify_stream', 'notify_group', '0', true);
-        } catch (\Throwable $e) {
-            // BUSYGROUP: 组已存在，忽略
+        // phpredis 没有 xGroupCreate 方法，统一入口是 xGroup('CREATE', ...)
+        // 组已存在（BUSYGROUP）不抛异常而是返回 false，要用 getLastError() 判断
+        $ret = $redis->xGroup('CREATE', 'notify_stream', 'notify_group', '0-0', true);
+        if ($ret === false && strpos((string) $redis->getLastError(), 'BUSYGROUP') === 0) {
+            $redis->clearLastError();   // 组已存在属预期情况，清除错误标记
         }
     }
 
     protected function claimPending(\Redis $redis, string $consumer): void
     {
-        // XAUTOCLAIM 需要 Redis 6.2+；旧版本用 XPENDING + XCLAIM 组合
+        // 返回 [0 => 下一游标, 1 => [entry_id => [字段 => 值]], 2 => 已删除id（仅 Redis 7.0+ 有第三元素）]
+        // XAUTOCLAIM 需要 phpredis >= 6.0.0 且 Redis 服务器 >= 6.2；旧版本用 XPENDING + XCLAIM 组合
         $result = $redis->xAutoClaim('notify_stream', 'notify_group', $consumer, 600000, '0-0', 10);
         foreach ($result[1] ?? [] as $entry_id => $fields) {
             $this->processOne($fields['id']);
@@ -392,7 +423,7 @@ class NotifyStreamConsumer extends TimerController
 | 原则 | 原因 |
 |------|------|
 | **用 Stream 不用 List** | `RPOP/BRPOP` 是 at-most-once：弹出后消费者崩溃，消息永久丢失。Stream 消费组把"取出未确认"的消息留在 pending 列表，可被 XAUTOCLAIM 重投 |
-| **先标后投** | "从 DB 捞出"与"写入 Redis"之间崩溃会丢数据。先 `status=queued` 再 xAdd，配合回收（`status=4 AND update_time < now-600` 重置为 0）。反过来会重复投递——重复比丢失好处理（幂等可兜底），丢失是静默的 |
+| **先标后投** | "从 DB 捞出"与"写入 Redis"之间崩溃会丢数据。先 `status=4` 再 xAdd，配合回收（`status=4 AND update_time < now-600` 重置为 0；依赖表有 `update_time` 字段，CURD 表默认自带，没有则补）。反过来会重复投递——重复比丢失好处理（幂等可兜底），丢失是静默的 |
 | **幂等消费** | at-least-once 下消息可能投递两次（认领重试、生产者重复入队），回写用条件 UPDATE（`WHERE status != done`）兜底 |
 
 其他要点：
@@ -449,6 +480,7 @@ $item->save();
 | `$this->concurrencyId` | 消费者编号，用于日志排查、worker 标识、Stream consumer 名；**不用于划分数据** |
 | 时间窗口模式 | `maxRunTime` + 循环消费 + 超时退出，见 [ulthon-timer](../ulthon-timer/SKILL.md)；`frequency >= maxRunTime * 2` |
 | `run_type` | 消费者任务默认 `auto`（单节点跑全部消费者）；要各节点分摊时切 `all`。回收/生产者任务恒为 `auto` 或 `main`（全局单份） |
+| 控制器 `$frequency`（防刷） | 消费型任务**一般不设**：手动触发 bypass 的是定时器侧节流，不 bypass 控制器防刷——防刷窗口内手动触发会被"请不要频繁请求"拦截。确需防外部刷再设，且应显著小于调度 frequency |
 | 执行日志 | `/do` 自动重写为 `/execute`，每轮消费的耗时/结果进 `ul_system_timer_log`，配合定时器日志页排查 |
 
 ## 常见坑位（快速自检）
@@ -460,3 +492,7 @@ $item->save();
 - **模式 C 多节点但 Redis 不共享**：各节点消费各自的 Stream，消息错乱。多节点必须共享同一个 Redis 实例
 - **消费者 `sleep` 写在处理分支里**：有数据时 sleep 会拖慢吞吐。只在队列空时休眠
 - **把 `concurrency` 当数据分片数理解**：又绕回 `id % N` 反模式。重读核心原则第 1 条
+- **用了 `$this->concurrencyCount` 却没定义**：`TimerControllerBase` 只注入 `$concurrencyId` 属性，`concurrency_count` 参数需子类自行从 `$this->request->param()` 取并存为属性，直接用未定义属性会在运算时抛 TypeError
+- **未配置 `redis` store 就调 `Cache::store('redis')`**：框架默认 cache 只有 `file`/`login`/`mcp_session`，先按「环境前提」配置，否则抛 `InvalidArgumentException`
+- **phpredis Stream API 两个易错点**：`xReadGroup` 返回结构嵌套流名层（`[流名 => [entry_id => [字段 => 值]]]`，遍历前先取 `$msgs['流名']`，否则会把流名当 entry_id 去 xAck）；phpredis **没有** `xGroupCreate` 方法（统一入口 `xGroup('CREATE', $key, $group, $id, $mkstream)`），且组已存在的 BUSYGROUP 错误**不抛异常**而是返回 false，需用 `getLastError()` 判断
+- **模式 B 的 `frequency` 与 `maxRunTime` 比例失衡**：`maxRunTime >= frequency` 时两次调度窗口重叠、消费节奏错乱，保持 `frequency >= maxRunTime * 2`