add aggr

2026-05-07 20:51:38 +08:00
parent 1e7de46c26
commit 349f8e11b0
14 changed files with 730 additions and 1 deletions
@@ -0,0 +1,35 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\Command;
+
+use App\Service\OrderAggregatesRefreshJob;
+use Hyperf\Command\Annotation\Command;
+use Hyperf\Command\Command as HyperfCommand;
+
+/**
+ * 手动触发聚合刷新队列消费
+ *
+ * 与 Crontab 共用 OrderAggregatesRefreshJob，便于运维即时补刷或调试。
+ */
+#[Command]
+class OrderAggregatesRefreshCommand extends HyperfCommand
+{
+    public function __construct(private OrderAggregatesRefreshJob $job)
+    {
+        parent::__construct('orders:refresh-aggregates');
+    }
+
+    public function configure(): void
+    {
+        parent::configure();
+        $this->setDescription('消费 aggregate_refresh_queue，对滞后日期调用 refresh_continuous_aggregate');
+    }
+
+    public function handle(): void
+    {
+        $result = $this->job->run();
+        $this->info(sprintf('Processed: %d, Failed: %d', $result['processed'], $result['failed']));
+    }
+}
@@ -0,0 +1,31 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\Model;
+
+/**
+ * @property string $refresh_date    格式 Y-m-d
+ * @property string $aggregate_view  视图名（如 orders_daily_by_created）
+ * @property \Carbon\Carbon $created_at 入队时间
+ */
+class AggregateRefreshQueue extends Model
+{
+    protected ?string $table = 'aggregate_refresh_queue';
+
+    public bool $timestamps = false;
+
+    public bool $incrementing = false;
+
+    protected string $primaryKey = 'refresh_date';
+
+    protected array $fillable = [
+        'refresh_date',
+        'aggregate_view',
+        'created_at',
+    ];
+
+    protected array $casts = [
+        'created_at' => 'datetime',
+    ];
+}
@@ -5,8 +5,10 @@ declare(strict_types=1);
 namespace App\Platform;

 use App\Entity\Parse\EntityParseFactory;
+use App\Model\AggregateRefreshQueue;
 use App\Platform\Traits\FailedMessageTrait;
 use App\Utils\Log;
+use Carbon\Carbon;
 use Hyperf\Amqp\Annotation\Consumer;
 use Hyperf\Amqp\Builder\QueueBuilder;
 use Hyperf\Amqp\Message\ConsumerMessage;
@@ -152,9 +154,10 @@ class OrderConsumer extends ConsumerMessage
            // 鉴于定义子项为了保留足够的灵活性，因此每次订单更新，我们都需要完整更新 OrderItem
            $this->processOrderItems($items);

+            // 5. 识别 ≥ 3 天前的 created_date 入队，补刷自动策略未覆盖的窗口
+            $this->enqueueAffectedDates($orders_data);

            Db::commit();
-            // @TODO 触发事件通知，更新自动聚合任务

            // 在数据库事务中尝试对  $entityMapResult 中的元素进行持久化，如果没有问题， 则返回 ACK，否则这是 NACK 且 回滚事务。
            return Result::ACK;
@@ -379,4 +382,51 @@ class OrderConsumer extends ConsumerMessage
        }
    }

+    /**
+     * 识别 payload 中 ≥ 3 天前的 created_date，入队 orders_daily_by_created 兜底刷新。
+     *
+     * 自动刷新策略仅覆盖最近 3 天窗口；3 天前的订单变更（补录、追溯调整）需由
+     * aggregate_refresh_queue + Crontab 任务补刷。仅服务 by_created 视图，
+     * by_paid 由全量 REFRESH 覆盖，不入此队列。
+     *
+     * @param array $payloads 来自 entityMap()->all() 的订单数组，每条含 created_date 字段
+     */
+    protected function enqueueAffectedDates(array $payloads): void
+    {
+        if (empty($payloads)) {
+            return;
+        }
+
+        $threshold = Carbon::now()->subDays(3)->toDateString();
+
+        $unique_dates = [];
+        foreach ($payloads as $payload) {
+            $created = $payload['created_date'] ?? null;
+            if ($created === null) {
+                continue;
+            }
+
+            // entityMap 输出 'Y-m-d H:i:sP'；用 Carbon::parse 兼容多种格式
+            $date = Carbon::parse($created)->toDateString();
+
+            // 严格小于阈值才入队（≥ 阈值的部分由自动刷新策略覆盖）
+            if ($date < $threshold) {
+                $unique_dates[$date] = true;
+            }
+        }
+
+        if (empty($unique_dates)) {
+            return;
+        }
+
+        $now = Carbon::now();
+        foreach (array_keys($unique_dates) as $date) {
+            AggregateRefreshQueue::query()->insertOrIgnore([
+                'refresh_date'   => $date,
+                'aggregate_view' => 'orders_daily_by_created',
+                'created_at'     => $now,
+            ]);
+        }
+    }
+
 }
@@ -0,0 +1,21 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\Service;
+
+/**
+ * 聚合视图刷新接口
+ *
+ * 隔离 PG procedure 调用，便于在单测中注入 mock 实现。
+ */
+interface AggregateRefresherInterface
+{
+    /**
+     * 对指定聚合视图的指定日期刷新
+     *
+     * @param string $view         聚合视图名（如 'orders_daily_by_created'）
+     * @param string $refresh_date Y-m-d 格式日期
+     */
+    public function refresh(string $view, string $refresh_date): void;
+}
@@ -0,0 +1,31 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\Service;
+
+use Hyperf\DbConnection\Db;
+
+/**
+ * TimescaleDB 连续聚合刷新实现
+ *
+ * 调用 PG 的 refresh_continuous_aggregate 存储过程，按整日窗口刷新。
+ */
+class ContinuousAggregateRefresher implements AggregateRefresherInterface
+{
+    public function refresh(string $view, string $refresh_date): void
+    {
+        $start = sprintf("'%s 00:00:00+00'::timestamptz", $refresh_date);
+        $end   = sprintf("'%s 23:59:59.999999+00'::timestamptz", $refresh_date);
+
+        // view 在 P23.2 中硬编码为 'orders_daily_by_created'，但仍走 PDO quote 防御
+        $quoted_view = Db::getPdo()->quote($view);
+
+        Db::statement(sprintf(
+            'CALL refresh_continuous_aggregate(%s, %s, %s)',
+            $quoted_view,
+            $start,
+            $end
+        ));
+    }
+}
@@ -0,0 +1,76 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\Service;
+
+use App\Model\AggregateRefreshQueue;
+use App\Utils\Log;
+use Carbon\Carbon;
+use Throwable;
+
+/**
+ * 订单聚合刷新任务
+ *
+ * 消费 aggregate_refresh_queue 中 created_at < now() - 1 hour 的项，
+ * 逐条调用 AggregateRefresherInterface 刷新对应日期的连续聚合，
+ * 成功后从队列中删除；失败保留队列项由下次任务重试。
+ *
+ * 由 OrderAggregatesRefreshCommand（CLI 入口）和 Crontab（定时入口）共同调用。
+ */
+class OrderAggregatesRefreshJob
+{
+    public function __construct(private AggregateRefresherInterface $refresher)
+    {
+    }
+
+    /**
+     * @return array{processed: int, failed: int} 处理与失败计数
+     */
+    public function run(): array
+    {
+        $items = AggregateRefreshQueue::query()
+            ->where('created_at', '<', Carbon::now()->subHour())
+            ->orderBy('refresh_date')
+            ->orderBy('aggregate_view')
+            ->get();
+
+        if ($items->isEmpty()) {
+            Log::get()->info('orders:refresh-aggregates queue empty');
+            return ['processed' => 0, 'failed' => 0];
+        }
+
+        $processed = 0;
+        $failed    = 0;
+
+        foreach ($items as $item) {
+            $view = $item->aggregate_view;
+            $date = $item->refresh_date;
+
+            try {
+                $this->refresher->refresh($view, $date);
+
+                AggregateRefreshQueue::query()
+                    ->where('refresh_date', $date)
+                    ->where('aggregate_view', $view)
+                    ->delete();
+
+                $processed++;
+            } catch (Throwable $e) {
+                Log::get()->error('orders:refresh-aggregates failed', [
+                    'view'  => $view,
+                    'date'  => $date,
+                    'error' => $e->getMessage(),
+                ]);
+                $failed++;
+            }
+        }
+
+        Log::get()->info('orders:refresh-aggregates done', [
+            'processed' => $processed,
+            'failed'    => $failed,
+        ]);
+
+        return ['processed' => $processed, 'failed' => $failed];
+    }
+}
@@ -20,6 +20,7 @@
        "hyperf/command": "~3.1.0",
        "hyperf/config": "~3.1.0",
        "hyperf/constants": "~3.1.0",
+        "hyperf/crontab": "~3.1.0",
        "hyperf/database-pgsql": "^3.1",
        "hyperf/db-connection": "~3.1.0",
        "hyperf/engine": "^2.10",
@@ -0,0 +1,19 @@
+<?php
+
+declare(strict_types=1);
+
+use App\Service\OrderAggregatesRefreshJob;
+use Hyperf\Crontab\Crontab;
+
+return [
+    'enable' => true,
+    'crontab' => [
+        // 每天 02:07 处理 aggregate_refresh_queue 中的滞后聚合刷新。
+        // 02 时段为最低流量段；分钟取 :07 避开整点全互联网定时任务集中触发。
+        (new Crontab())
+            ->setName('OrderAggregatesRefresh')
+            ->setRule('7 2 * * *')
+            ->setCallback([OrderAggregatesRefreshJob::class, 'run'])
+            ->setMemo('每天 02:07 处理 aggregate_refresh_queue 中的滞后聚合刷新'),
+    ],
+];
@@ -11,4 +11,5 @@ declare(strict_types=1);
 */
 return [
    // 可以在这里配置接口到实现的绑定
+    App\Service\AggregateRefresherInterface::class => App\Service\ContinuousAggregateRefresher::class,
 ];
@@ -10,4 +10,5 @@ declare(strict_types=1);
 * @license  https://github.com/hyperf/hyperf/blob/master/LICENSE
 */
 return [
+    Hyperf\Crontab\Process\CrontabDispatcherProcess::class,
 ];
@@ -0,0 +1,32 @@
+<?php
+
+declare(strict_types=1);
+
+use Hyperf\Database\Schema\Schema;
+use Hyperf\Database\Schema\Blueprint;
+use Hyperf\Database\Migrations\Migration;
+use Hyperf\DbConnection\Db;
+
+return new class extends Migration
+{
+    public function up(): void
+    {
+        Schema::create('aggregate_refresh_queue', function (Blueprint $table) {
+            $table->date('refresh_date')->comment('待刷新的聚合日');
+            $table->string('aggregate_view', 100)->comment('视图名（orders_daily_by_created / orders_daily_by_paid）');
+            $table->timestampTz('created_at')->useCurrent()->comment('入队时间');
+
+            $table->primary(['refresh_date', 'aggregate_view'], 'pk_aggregate_refresh_queue');
+            $table->index('created_at', 'idx_aggregate_refresh_queue_created_at');
+        });
+
+        // Hyperf PostgresGrammar::compilePrimary 丢弃自定义 index 名（vendor/hyperf/database-pgsql/.../PostgresGrammar.php:230），
+        // PG 默认用 {table}_pkey；这里显式重命名以满足计划验收要求（名称 pk_aggregate_refresh_queue）
+        Db::statement('ALTER TABLE aggregate_refresh_queue RENAME CONSTRAINT aggregate_refresh_queue_pkey TO pk_aggregate_refresh_queue');
+    }
+
+    public function down(): void
+    {
+        Schema::dropIfExists('aggregate_refresh_queue');
+    }
+};
@@ -0,0 +1,89 @@
+<?php
+
+declare(strict_types=1);
+
+namespace HyperfTest\Cases\Unit\Model;
+
+use App\Model\AggregateRefreshQueue;
+use Carbon\Carbon;
+use PHPUnit\Framework\TestCase;
+
+/**
+ * @internal
+ * @coversNothing
+ */
+class AggregateRefreshQueueTest extends TestCase
+{
+    protected function runInCoroutine(callable $callback): void
+    {
+        if (\Swoole\Coroutine::getCid() > 0) {
+            $callback();
+            return;
+        }
+
+        $exception = null;
+        \Swoole\Coroutine\run(static function () use ($callback, &$exception): void {
+            try {
+                $callback();
+            } catch (\Throwable $e) {
+                $exception = $e;
+            }
+        });
+        if ($exception) {
+            throw $exception;
+        }
+    }
+
+    private function uniqueRefreshDate(): string
+    {
+        // 与 ApiKeyTest 的 bin2hex(random_bytes(4)) 同源策略：用唯一性而非全表清理来隔离用例
+        return sprintf('20%02d-%02d-%02d', random_int(50, 99), random_int(1, 12), random_int(1, 28));
+    }
+
+    public function test_attributes_persist_correctly(): void
+    {
+        $this->runInCoroutine(function (): void {
+            $unique_date = $this->uniqueRefreshDate();
+
+            try {
+                AggregateRefreshQueue::query()->create([
+                    'refresh_date' => $unique_date,
+                    'aggregate_view' => 'orders_daily_by_created',
+                    'created_at' => Carbon::now(),
+                ]);
+
+                $row = AggregateRefreshQueue::query()->where('refresh_date', $unique_date)->first();
+
+                $this->assertNotNull($row);
+                $this->assertSame($unique_date, $row->refresh_date);
+                $this->assertSame('orders_daily_by_created', $row->aggregate_view);
+                $this->assertInstanceOf(Carbon::class, $row->created_at);
+            } finally {
+                AggregateRefreshQueue::query()->where('refresh_date', $unique_date)->delete();
+            }
+        });
+    }
+
+    public function test_insert_or_ignore_dedups_on_composite_pk(): void
+    {
+        $this->runInCoroutine(function (): void {
+            $unique_date = $this->uniqueRefreshDate();
+
+            $payload = [
+                'refresh_date' => $unique_date,
+                'aggregate_view' => 'orders_daily_by_created',
+                'created_at' => Carbon::now(),
+            ];
+
+            try {
+                AggregateRefreshQueue::query()->insertOrIgnore($payload);
+                AggregateRefreshQueue::query()->insertOrIgnore($payload);
+
+                $count = AggregateRefreshQueue::query()->where('refresh_date', $unique_date)->count();
+                $this->assertSame(1, $count);
+            } finally {
+                AggregateRefreshQueue::query()->where('refresh_date', $unique_date)->delete();
+            }
+        });
+    }
+}
@@ -0,0 +1,164 @@
+<?php
+
+declare(strict_types=1);
+
+namespace HyperfTest\Cases\Unit\Platform;
+
+use App\Model\AggregateRefreshQueue;
+use App\Platform\OrderConsumer;
+use Carbon\Carbon;
+use PHPUnit\Framework\TestCase;
+use ReflectionMethod;
+
+/**
+ * @internal
+ * @coversNothing
+ */
+class OrderConsumerEnqueueTest extends TestCase
+{
+    protected function runInCoroutine(callable $callback): void
+    {
+        if (\Swoole\Coroutine::getCid() > 0) {
+            $callback();
+            return;
+        }
+
+        $exception = null;
+        \Swoole\Coroutine\run(static function () use ($callback, &$exception): void {
+            try {
+                $callback();
+            } catch (\Throwable $e) {
+                $exception = $e;
+            }
+        });
+        if ($exception) {
+            throw $exception;
+        }
+    }
+
+    /**
+     * 调用 protected enqueueAffectedDates 方法
+     */
+    protected function invokeEnqueue(array $payloads): void
+    {
+        $consumer = new OrderConsumer();
+        $method = new ReflectionMethod($consumer, 'enqueueAffectedDates');
+        $method->setAccessible(true);
+        $method->invoke($consumer, $payloads);
+    }
+
+    /**
+     * 清理指定日期的队列条目（前置 + 后置双保险）
+     */
+    protected function cleanupQueue(string $date): void
+    {
+        AggregateRefreshQueue::query()
+            ->where('refresh_date', $date)
+            ->where('aggregate_view', 'orders_daily_by_created')
+            ->delete();
+    }
+
+    /**
+     * 当日订单不应入队（在自动刷新策略覆盖窗口内）
+     */
+    public function test_today_payload_does_not_enqueue(): void
+    {
+        $this->runInCoroutine(function (): void {
+            $today = Carbon::now()->toDateString();
+            $payloads = [['created_date' => Carbon::now()->format('Y-m-d H:i:sP')]];
+
+            $this->cleanupQueue($today);
+            try {
+                $this->invokeEnqueue($payloads);
+
+                $row = AggregateRefreshQueue::query()
+                    ->where('refresh_date', $today)
+                    ->where('aggregate_view', 'orders_daily_by_created')
+                    ->first();
+                $this->assertNull($row);
+            } finally {
+                $this->cleanupQueue($today);
+            }
+        });
+    }
+
+    /**
+     * 2 天前订单不应入队（仍在自动刷新策略 [now-3d, now-1h] 窗口内）
+     */
+    public function test_two_days_ago_payload_does_not_enqueue(): void
+    {
+        $this->runInCoroutine(function (): void {
+            $two_days_ago = Carbon::now()->subDays(2)->toDateString();
+            $payloads = [['created_date' => Carbon::now()->subDays(2)->format('Y-m-d H:i:sP')]];
+
+            $this->cleanupQueue($two_days_ago);
+            try {
+                $this->invokeEnqueue($payloads);
+
+                $row = AggregateRefreshQueue::query()
+                    ->where('refresh_date', $two_days_ago)
+                    ->where('aggregate_view', 'orders_daily_by_created')
+                    ->first();
+                $this->assertNull($row);
+            } finally {
+                $this->cleanupQueue($two_days_ago);
+            }
+        });
+    }
+
+    /**
+     * 30 天前订单应入队 by_created（自动策略未覆盖，需补刷）
+     */
+    public function test_old_payload_enqueues_by_created(): void
+    {
+        $this->runInCoroutine(function (): void {
+            $old_date = Carbon::now()->subDays(30)->toDateString();
+            $payloads = [['created_date' => Carbon::now()->subDays(30)->format('Y-m-d H:i:sP')]];
+
+            $this->cleanupQueue($old_date);
+            try {
+                $this->invokeEnqueue($payloads);
+
+                $row = AggregateRefreshQueue::query()
+                    ->where('refresh_date', $old_date)
+                    ->where('aggregate_view', 'orders_daily_by_created')
+                    ->first();
+                $this->assertNotNull($row);
+                $this->assertSame($old_date, $row->refresh_date);
+                $this->assertSame('orders_daily_by_created', $row->aggregate_view);
+            } finally {
+                $this->cleanupQueue($old_date);
+            }
+        });
+    }
+
+    /**
+     * 同日期重复入队由复合主键 + insertOrIgnore 防重
+     */
+    public function test_dedup_via_insert_or_ignore(): void
+    {
+        $this->runInCoroutine(function (): void {
+            $old_date = Carbon::now()->subDays(45)->toDateString();
+            $payloads = [
+                ['created_date' => Carbon::now()->subDays(45)->format('Y-m-d H:i:sP')],
+                ['created_date' => Carbon::now()->subDays(45)->format('Y-m-d H:i:sP')],
+            ];
+
+            $this->cleanupQueue($old_date);
+            try {
+                // 单批次内同日期 → $unique_dates map 去重
+                $this->invokeEnqueue($payloads);
+                // 跨批次同日期 → 复合主键 insertOrIgnore 防重
+                $this->invokeEnqueue($payloads);
+
+                $count = AggregateRefreshQueue::query()
+                    ->where('refresh_date', $old_date)
+                    ->where('aggregate_view', 'orders_daily_by_created')
+                    ->count();
+                $this->assertSame(1, $count);
+            } finally {
+                $this->cleanupQueue($old_date);
+            }
+        });
+    }
+}
@@ -0,0 +1,178 @@
+<?php
+
+declare(strict_types=1);
+
+namespace HyperfTest\Cases\Unit\Service;
+
+use App\Model\AggregateRefreshQueue;
+use App\Service\AggregateRefresherInterface;
+use App\Service\OrderAggregatesRefreshJob;
+use Carbon\Carbon;
+use PHPUnit\Framework\TestCase;
+use RuntimeException;
+
+/**
+ * 测试用 stub Refresher，记录每次 refresh 调用
+ */
+class RecordingRefresher implements AggregateRefresherInterface
+{
+    /** @var array<int, array{view: string, date: string}> */
+    public array $calls = [];
+
+    public function refresh(string $view, string $refresh_date): void
+    {
+        $this->calls[] = ['view' => $view, 'date' => $refresh_date];
+    }
+}
+
+/**
+ * 测试用 stub Refresher，每次调用抛异常（模拟刷新失败）
+ */
+class ThrowingRefresher implements AggregateRefresherInterface
+{
+    public function refresh(string $view, string $refresh_date): void
+    {
+        throw new RuntimeException('refresh failed');
+    }
+}
+
+/**
+ * @internal
+ * @coversNothing
+ */
+class OrderAggregatesRefreshJobTest extends TestCase
+{
+    protected function runInCoroutine(callable $callback): void
+    {
+        if (\Swoole\Coroutine::getCid() > 0) {
+            $callback();
+            return;
+        }
+
+        $exception = null;
+        \Swoole\Coroutine\run(static function () use ($callback, &$exception): void {
+            try {
+                $callback();
+            } catch (\Throwable $e) {
+                $exception = $e;
+            }
+        });
+        if ($exception) {
+            throw $exception;
+        }
+    }
+
+    /**
+     * 沿用 P23.1 AggregateRefreshQueueTest 的随机未来日期隔离策略，
+     * 确保不同测试用例间无冲突，也不会污染生产数据。
+     */
+    private function uniqueRefreshDate(): string
+    {
+        return sprintf('20%02d-%02d-%02d', random_int(50, 99), random_int(1, 12), random_int(1, 28));
+    }
+
+    /**
+     * 队列项 created_at < now()-1h 应被处理：调用 refresher 后从队列删除
+     */
+    public function test_processes_old_items_and_clears_queue(): void
+    {
+        $this->runInCoroutine(function (): void {
+            $date_a = $this->uniqueRefreshDate();
+            $date_b = $this->uniqueRefreshDate();
+            while ($date_b === $date_a) {
+                $date_b = $this->uniqueRefreshDate();
+            }
+
+            // 前置清理（防 random 撞历史残留）
+            AggregateRefreshQueue::query()->whereIn('refresh_date', [$date_a, $date_b])->delete();
+
+            try {
+                $two_hours_ago = Carbon::now()->subHours(2);
+                AggregateRefreshQueue::query()->insertOrIgnore([
+                    ['refresh_date' => $date_a, 'aggregate_view' => 'orders_daily_by_created', 'created_at' => $two_hours_ago],
+                    ['refresh_date' => $date_b, 'aggregate_view' => 'orders_daily_by_created', 'created_at' => $two_hours_ago],
+                ]);
+
+                $refresher = new RecordingRefresher();
+                $job = new OrderAggregatesRefreshJob($refresher);
+                $result = $job->run();
+
+                $this->assertSame(2, $result['processed']);
+                $this->assertSame(0, $result['failed']);
+                $this->assertCount(2, $refresher->calls);
+
+                // 队列已清空（针对本次测试的两条）
+                $remaining = AggregateRefreshQueue::query()
+                    ->whereIn('refresh_date', [$date_a, $date_b])
+                    ->count();
+                $this->assertSame(0, $remaining);
+            } finally {
+                AggregateRefreshQueue::query()->whereIn('refresh_date', [$date_a, $date_b])->delete();
+            }
+        });
+    }
+
+    /**
+     * 队列项 created_at >= now()-1h 应被跳过：refresher 无调用，队列保留
+     */
+    public function test_recently_enqueued_items_are_skipped(): void
+    {
+        $this->runInCoroutine(function (): void {
+            $date = $this->uniqueRefreshDate();
+            AggregateRefreshQueue::query()->where('refresh_date', $date)->delete();
+
+            try {
+                AggregateRefreshQueue::query()->insertOrIgnore([
+                    'refresh_date'   => $date,
+                    'aggregate_view' => 'orders_daily_by_created',
+                    'created_at'     => Carbon::now()->subMinutes(30), // < 1 hour
+                ]);
+
+                $refresher = new RecordingRefresher();
+                $job = new OrderAggregatesRefreshJob($refresher);
+                $result = $job->run();
+
+                $this->assertSame(0, $result['processed']);
+                $this->assertSame(0, $result['failed']);
+                $this->assertCount(0, $refresher->calls);
+
+                // 队列项保留
+                $exists = AggregateRefreshQueue::query()->where('refresh_date', $date)->exists();
+                $this->assertTrue($exists);
+            } finally {
+                AggregateRefreshQueue::query()->where('refresh_date', $date)->delete();
+            }
+        });
+    }
+
+    /**
+     * refresher 抛异常时：队列项保留（下次任务重试），失败计数 +1
+     */
+    public function test_failure_keeps_queue_item(): void
+    {
+        $this->runInCoroutine(function (): void {
+            $date = $this->uniqueRefreshDate();
+            AggregateRefreshQueue::query()->where('refresh_date', $date)->delete();
+
+            try {
+                AggregateRefreshQueue::query()->insertOrIgnore([
+                    'refresh_date'   => $date,
+                    'aggregate_view' => 'orders_daily_by_created',
+                    'created_at'     => Carbon::now()->subHours(2),
+                ]);
+
+                $job = new OrderAggregatesRefreshJob(new ThrowingRefresher());
+                $result = $job->run();
+
+                $this->assertSame(0, $result['processed']);
+                $this->assertSame(1, $result['failed']);
+
+                // 队列项保留以待重试
+                $exists = AggregateRefreshQueue::query()->where('refresh_date', $date)->exists();
+                $this->assertTrue($exists);
+            } finally {
+                AggregateRefreshQueue::query()->where('refresh_date', $date)->delete();
+            }
+        });
+    }
+}