From 809d6c59752fb39058d4b444dcb9dcd6de6123b2 Mon Sep 17 00:00:00 2001 From: WuYongJie <549727955@qq.com> Date: Wed, 22 Jul 2026 10:25:52 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96=EF=BC=9A=E5=BC=82=E6=AD=A5?= =?UTF-8?q?=E5=88=B7=E6=96=B0=E4=B8=8A=E6=B8=B8=E8=B4=A2=E5=8A=A1=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E5=BF=AB=E7=85=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- AGENTS.md | 15 +- .../JianSunRefreshActiveSnapshots.php | 11 +- config/jiansun.php | 29 +- deploy/activate-incremental-release.sh | 2 +- deploy/activate-release.sh | 2 +- deploy/bootstrap-server.sh | 1 + docs/开发与部署SOP.md | 36 +- .../Controllers/Api/GoodsReconciliations.php | 17 +- .../Http/Controllers/Api/SupplierPayables.php | 17 +- .../SupplierPayableRepository.php | 64 +-- .../Services/JianSunQueryCache.php | 374 ++++++++++++++---- .../JianSunQueryCacheTest.php | 134 +++++++ 12 files changed, 563 insertions(+), 139 deletions(-) create mode 100644 tests/Unit/EcommerceFinance/JianSunQueryCacheTest.php diff --git a/AGENTS.md b/AGENTS.md index 8f80836c..146632c1 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -123,12 +123,15 @@ ### 8.6 上游数据库读取规则 -- 业务页面禁止直接、无缓存地查询上游数据库；统一通过 `JianSunQueryCache` 读取本地活跃查询快照。 -- 相同页面和筛选条件默认复用 15 分钟，选项类数据默认复用 60 分钟；不得通过前端轮询缩短该周期。 -- 所有上游查询全局单线程执行，必须设置连接超时、查询超时和每 15 分钟刷新预算；失败时只允许使用已有旧快照，不得无限重试。 -- 定时任务只刷新最近 24 小时内实际使用过的查询，默认每轮最多 12 个；闲置页面不得为了“预热”持续访问上游。 -- `JIANSUN_ENABLED` 默认关闭。只有确认生产服务器网络白名单、只读账号、ODBC 驱动和查询索引后，才能在服务器私有 `runtime/.env` 中开启。 -- 真正的行级增量同步必须先确认每张源表的稳定主键、更新时间字段、删除标记和对应索引；未确认前禁止用全表扫描冒充增量同步。 +- 上游是长期运行且不可修改的旧系统，按只读黑盒处理；不得要求修改其账号、表结构、索引、字段或启用 Change Tracking 等功能。 +- 业务页面禁止直接、无缓存地查询上游数据库；统一通过 `JianSunQueryCache` 读取持久化在本机 `storage` 卷中的活跃查询快照。 +- 快照复用按“数据范围 + 查询条件”判断，与账户数量无关；权限或数据范围不同的结果不得错误共用。 +- 列表和汇总快照默认 2 分钟，选项类数据默认 60 分钟。页面命中过期快照时立即返回本地值并排入后台刷新，不得等待上游，也不得通过前端高频轮询制造压力。 +- 相同查询使用独立单飞锁；不同冷查询最多占用 2 个上游槽位。每分钟最多 6 个查询组，每分钟后台最多刷新 2 个近期活跃查询；不得恢复为一把锁串行所有页面请求。 +- 必须设置连接超时、查询超时、慢查询冷却和连接失败熔断；一次失败后进入冷却，已有旧快照时继续提供本地数据，不得连续重试。 +- 上游 API 必须保留 `X-Caiwu-Data-Status` 和 `X-Caiwu-Data-Synced-At` 响应头；验收同时检查数据状态和同步时间，不能只看 HTTP 200。 +- `JIANSUN_ENABLED` 默认关闭；只允许在现有只读访问方式完成受控连通性和单查询探测后，通过服务器私有 `runtime/.env` 开启。 +- 当前方案是近实时查询快照，不得冒充完整行级增量复制。没有可靠变更标记时，禁止定时全表扫描；历史一致性只能采用低频、小批量、可中止的核对。 - 首次全量历史数据同步必须单独评估行数、时间窗口和上游低峰期，不得在日常发布时自动执行。 - 上游地址、账号、密码只允许存在于服务器私有环境配置；代码、提交、PR、共享文档和日志中一律不得出现真实值。 diff --git a/app/Console/Commands/JianSunRefreshActiveSnapshots.php b/app/Console/Commands/JianSunRefreshActiveSnapshots.php index eacfe48e..7fd14de7 100644 --- a/app/Console/Commands/JianSunRefreshActiveSnapshots.php +++ b/app/Console/Commands/JianSunRefreshActiveSnapshots.php @@ -13,7 +13,7 @@ class JianSunRefreshActiveSnapshots extends Command { protected $signature = 'jiansun:snapshots:refresh {--limit= : 本轮最多刷新的活跃查询数}'; - protected $description = '单线程刷新最近使用过的 JianSun 上游查询快照'; + protected $description = '在时间和并发预算内刷新最近使用过的 JianSun 本地快照'; public function handle(JianSunQueryCache $cache, JianSunSnapshotRefresher $refresher): int { @@ -23,9 +23,11 @@ class JianSunRefreshActiveSnapshots extends Command return self::SUCCESS; } - $configuredLimit = max((int) config('jiansun.snapshots.refresh_limit', 12), 1); + $configuredLimit = max((int) config('jiansun.snapshots.refresh_limit', 2), 1); $requestedLimit = (int) ($this->option('limit') ?: $configuredLimit); $entries = $cache->activeEntries(min(max($requestedLimit, 1), $configuredLimit)); + $timeBudget = max((int) config('jiansun.snapshots.refresh_time_budget_seconds', 45), 10); + $startedAt = microtime(true); if ($entries === []) { $this->components->info('没有需要刷新的活跃查询。'); @@ -37,6 +39,11 @@ class JianSunRefreshActiveSnapshots extends Command $stale = 0; foreach ($entries as $entry) { + if ((microtime(true) - $startedAt) >= $timeBudget) { + $this->components->info('本轮刷新已达到时间预算，剩余查询留到下一轮。'); + break; + } + try { $status = $refresher->refresh($entry); $status === 'fresh' ? $fresh++ : $stale++; diff --git a/config/jiansun.php b/config/jiansun.php index 7a29db11..8681a2cf 100644 --- a/config/jiansun.php +++ b/config/jiansun.php @@ -29,22 +29,31 @@ return [ | 活跃查询快照 |-------------------------------------------------------------------------- | - | 页面只在快照过期时访问一次上游；相同页面和筛选条件由所有用户共享。 - | 全局锁保证同一时间只有一个查询组访问上游，预算则限制十五分钟内的 - | 最大刷新次数。旧快照仅在上游临时不可用时作为明确的降级数据使用。 + | 页面优先读取持久化在本机 storage 卷中的快照；快照过期后先返回旧值， + | 再由后台刷新。相同数据范围和筛选条件复用同一快照，与账户数量无关。 + | 同一查询只允许一个刷新任务，不同查询最多占用少量上游槽位，并受 + | 每分钟预算、慢查询冷却和连接失败熔断共同保护。 | */ 'snapshots' => [ - 'ttl_seconds' => (int) env('JIANSUN_SNAPSHOT_TTL', 900), + 'ttl_seconds' => (int) env('JIANSUN_SNAPSHOT_TTL', 120), + 'summary_ttl_seconds' => (int) env('JIANSUN_SUMMARY_TTL', 120), 'options_ttl_seconds' => (int) env('JIANSUN_OPTIONS_TTL', 3600), 'stale_ttl_seconds' => (int) env('JIANSUN_STALE_TTL', 86400), - 'active_ttl_seconds' => (int) env('JIANSUN_ACTIVE_QUERY_TTL', 86400), - 'registry_limit' => (int) env('JIANSUN_ACTIVE_QUERY_LIMIT', 60), - 'refresh_limit' => (int) env('JIANSUN_REFRESH_LIMIT', 12), - 'miss_budget_per_window' => (int) env('JIANSUN_MISS_BUDGET', 20), - 'budget_window_seconds' => (int) env('JIANSUN_BUDGET_WINDOW', 900), + 'active_ttl_seconds' => (int) env('JIANSUN_ACTIVE_QUERY_TTL', 21600), + 'registry_limit' => (int) env('JIANSUN_ACTIVE_QUERY_LIMIT', 300), + 'refresh_limit' => (int) env('JIANSUN_REFRESH_LIMIT', 2), + 'miss_budget_per_window' => (int) env('JIANSUN_MISS_BUDGET', 6), + 'budget_window_seconds' => (int) env('JIANSUN_BUDGET_WINDOW', 60), 'lock_seconds' => (int) env('JIANSUN_LOCK_SECONDS', 45), - 'lock_wait_seconds' => (int) env('JIANSUN_LOCK_WAIT', 12), + 'lock_wait_seconds' => (int) env('JIANSUN_LOCK_WAIT', 3), + 'max_concurrency' => (int) env('JIANSUN_MAX_CONCURRENCY', 2), + 'slot_wait_ms' => (int) env('JIANSUN_SLOT_WAIT_MS', 1500), + 'failure_cooldown_seconds' => (int) env('JIANSUN_FAILURE_COOLDOWN', 300), + 'query_failure_cooldown_seconds' => (int) env('JIANSUN_QUERY_FAILURE_COOLDOWN', 300), + 'slow_query_ms' => (int) env('JIANSUN_SLOW_QUERY_MS', 5000), + 'slow_query_cooldown_seconds' => (int) env('JIANSUN_SLOW_QUERY_COOLDOWN', 300), + 'refresh_time_budget_seconds' => (int) env('JIANSUN_REFRESH_TIME_BUDGET', 45), 'refresh_pause_ms' => (int) env('JIANSUN_REFRESH_PAUSE_MS', 300), ], ]; diff --git a/deploy/activate-incremental-release.sh b/deploy/activate-incremental-release.sh index f1c7847f..422c3d73 100644 --- a/deploy/activate-incremental-release.sh +++ b/deploy/activate-incremental-release.sh @@ -78,7 +78,7 @@ mv "$REMOTE_DIR/.deployed-sha.tmp" "$REMOTE_DIR/.deployed-sha" # 只运行 JianSun 快照刷新命令，不启用整个 Laravel 调度器，避免意外启动 # 账单提醒等既有任务。命令在 JIANSUN_ENABLED=false 时会安全跳过。 cat > /etc/cron.d/caiwu-jiansun-snapshots <<'EOF' -*/15 * * * * root /usr/bin/flock -n /run/caiwu-jiansun-sync.lock /usr/bin/docker exec -u www-data caiwu_app php artisan jiansun:snapshots:refresh >/dev/null 2>&1 +* * * * * root /usr/bin/flock -n /run/caiwu-jiansun-sync.lock /usr/bin/docker exec -u www-data caiwu_app php artisan jiansun:snapshots:refresh >/dev/null 2>&1 EOF chmod 644 /etc/cron.d/caiwu-jiansun-snapshots diff --git a/deploy/activate-release.sh b/deploy/activate-release.sh index 2d0a360a..1e2441bd 100644 --- a/deploy/activate-release.sh +++ b/deploy/activate-release.sh @@ -82,7 +82,7 @@ mv "$REMOTE_DIR/.deployed-sha.tmp" "$REMOTE_DIR/.deployed-sha" # 只运行 JianSun 快照刷新命令，不启用整个 Laravel 调度器，避免意外启动 # 账单提醒等既有任务。命令在 JIANSUN_ENABLED=false 时会安全跳过。 cat > /etc/cron.d/caiwu-jiansun-snapshots <<'EOF' -*/15 * * * * root /usr/bin/flock -n /run/caiwu-jiansun-sync.lock /usr/bin/docker exec -u www-data caiwu_app php artisan jiansun:snapshots:refresh >/dev/null 2>&1 +* * * * * root /usr/bin/flock -n /run/caiwu-jiansun-sync.lock /usr/bin/docker exec -u www-data caiwu_app php artisan jiansun:snapshots:refresh >/dev/null 2>&1 EOF chmod 644 /etc/cron.d/caiwu-jiansun-snapshots diff --git a/deploy/bootstrap-server.sh b/deploy/bootstrap-server.sh index 14b42581..3f399476 100644 --- a/deploy/bootstrap-server.sh +++ b/deploy/bootstrap-server.sh @@ -58,6 +58,7 @@ if [ ! -f "$REMOTE_DIR/runtime/.env" ]; then 'CACHE_DRIVER=file' \ 'SESSION_DRIVER=file' \ 'QUEUE_CONNECTION=sync' \ + 'JIANSUN_ENABLED=false' \ 'MARIADB_DATABASE=caiwu' \ 'MARIADB_USER=caiwu' \ "MARIADB_PASSWORD=$DB_PASSWORD" \ diff --git a/docs/开发与部署SOP.md b/docs/开发与部署SOP.md index ba86b174..2441605e 100644 --- a/docs/开发与部署SOP.md +++ b/docs/开发与部署SOP.md @@ -137,17 +137,20 @@ git pull --ff-only ## 8. 上游财务数据库同步 -生产环境默认设置 `JIANSUN_ENABLED=false`。网络白名单、只读账号和索引未核实前，即使镜像已经包含 ODBC 驱动，也不会连接上游。 +生产环境默认设置 `JIANSUN_ENABLED=false`。上游按不可修改的只读旧系统处理，不要求对方修改账号、表结构、索引、字段或数据库功能；只有现有访问方式通过受控连通性和单查询探测后，才在服务器私有配置中开启。 启用后采用“活跃查询快照”策略： -1. 页面相同查询 15 分钟内直接读取服务器本地文件缓存，6 个测试账号共享同一份结果； -2. 下拉选项等低频数据缓存 60 分钟； -3. `/etc/cron.d/caiwu-jiansun-snapshots` 每 15 分钟触发一次刷新，只处理最近使用过的查询； -4. 所有刷新使用同一把互斥锁，上游同时最多只有 1 个查询组； -5. 每轮默认最多刷新 12 个活跃查询，每 15 分钟默认最多产生 20 次实际缓存未命中； -6. 连接 5 秒未建立或单次查询超过 15 秒即失败，不连续重试；已有旧快照时明确降级为旧数据； -7. 服务器和上游不通时保持关闭，页面不会绕过缓存直接连接。 +1. 页面优先读取持久化在 `caiwu_storage` 卷中的本地快照；相同数据范围和查询条件复用同一结果，与系统账户数量无关； +2. 列表和汇总默认保持 2 分钟新鲜，下拉选项等低频数据缓存 60 分钟； +3. 新鲜快照直接返回；过期快照也立即返回，同时登记为紧急后台刷新，页面不等待上游；只有本地从未存在该查询时，才允许受控同步获取一次； +4. `/etc/cron.d/caiwu-jiansun-snapshots` 每分钟触发一次，只处理最近 6 小时实际使用且已经到期的查询； +5. 相同查询使用独立单飞锁，不同查询最多占用 2 个上游槽位，因此大量缓存命中不会被全局串行； +6. 每轮默认最多刷新 2 个活跃查询，每分钟默认最多产生 6 个实际上游查询组；列表优先于汇总，汇总优先于选项； +7. 连接 5 秒未建立或单次查询超过 15 秒即失败；慢查询和失败查询进入 5 分钟冷却，连接失败同时触发全局熔断，不连续重试； +8. 服务器和上游不通时继续使用已有本地快照；完全没有本地数据时明确返回暂不可用，不绕过保护直接连接。 + +相关 API 使用 `X-Caiwu-Data-Status`（`fresh` / `stale`）和 `X-Caiwu-Data-Synced-At` 响应头报告数据状态，验收时不能只看 HTTP 200。 服务器私有 `runtime/.env` 只使用下列变量名，真实值不得写入 Git 或本文档： @@ -159,13 +162,22 @@ DB_JIANSUN_DATABASE= DB_JIANSUN_USERNAME= DB_JIANSUN_PASSWORD= DB_JIANSUN_ODBC_DRIVER=FreeTDS -JIANSUN_SNAPSHOT_TTL=900 +JIANSUN_SNAPSHOT_TTL=120 +JIANSUN_SUMMARY_TTL=120 JIANSUN_OPTIONS_TTL=3600 -JIANSUN_REFRESH_LIMIT=12 -JIANSUN_MISS_BUDGET=20 +JIANSUN_ACTIVE_QUERY_TTL=21600 +JIANSUN_ACTIVE_QUERY_LIMIT=300 +JIANSUN_REFRESH_LIMIT=2 +JIANSUN_MISS_BUDGET=6 +JIANSUN_BUDGET_WINDOW=60 +JIANSUN_MAX_CONCURRENCY=2 +JIANSUN_FAILURE_COOLDOWN=300 +JIANSUN_QUERY_FAILURE_COOLDOWN=300 +JIANSUN_SLOW_QUERY_MS=5000 +JIANSUN_SLOW_QUERY_COOLDOWN=300 ``` -这套策略控制的是当前页面查询压力，不等同于复制上游全库。若以后需要真正的行级增量镜像，必须先由负责人确认每张表的稳定主键、更新时间、删除标记和索引，再安排一次低峰期初始化；禁止对数百万行大表每 10～15 分钟做全表扫描。 +这套策略提供的是近实时查询快照，不等同于复制上游全库。对于没有可靠更新时间或删除标记的旧表，不能声称实现了完整行级增量；禁止为了追求秒级实时而周期性扫描数百万行。页面实时性依靠本地快照和一分钟后台刷新，历史一致性只能通过低频、小批量、可随时停止的核对补齐。 ## 9. 禁止事项 diff --git a/modules/EcommerceFinance/Http/Controllers/Api/GoodsReconciliations.php b/modules/EcommerceFinance/Http/Controllers/Api/GoodsReconciliations.php index c15b73bd..dfa53775 100644 --- a/modules/EcommerceFinance/Http/Controllers/Api/GoodsReconciliations.php +++ b/modules/EcommerceFinance/Http/Controllers/Api/GoodsReconciliations.php @@ -41,7 +41,7 @@ class GoodsReconciliations extends ApiController fn () => $this->repo->list($filters, $page, $limit) ); - return response()->json([ + return $this->snapshotResponse([ 'data' => $result['data'], 'total' => $result['total'], 'page' => $page, @@ -62,7 +62,7 @@ class GoodsReconciliations extends ApiController fn () => $this->repo->summary($filters) ); - return response()->json($summary); + return $this->snapshotResponse($summary); } /** @@ -76,13 +76,24 @@ class GoodsReconciliations extends ApiController fn () => $this->repo->filterOptions() ); - return response()->json($options); + return $this->snapshotResponse($options); } // =================================================================== // Helpers // =================================================================== + private function snapshotResponse(array $payload) + { + $response = response()->json($payload); + + foreach ($this->cache->responseHeaders() as $name => $value) { + $response->headers->set($name, $value); + } + + return $response; + } + /** * 从请求参数构建筛选条件 */ diff --git a/modules/EcommerceFinance/Http/Controllers/Api/SupplierPayables.php b/modules/EcommerceFinance/Http/Controllers/Api/SupplierPayables.php index a43be292..c8401e4d 100644 --- a/modules/EcommerceFinance/Http/Controllers/Api/SupplierPayables.php +++ b/modules/EcommerceFinance/Http/Controllers/Api/SupplierPayables.php @@ -339,7 +339,7 @@ class SupplierPayables extends ApiController fn () => $resolver($page, $limit) ); - return response()->json([ + return $this->snapshotResponse([ 'data' => $result['data'], 'total' => $result['total'], 'page' => $page, @@ -350,7 +350,7 @@ class SupplierPayables extends ApiController private function summaryResponse(string $scope, array $filters, Closure $resolver) { - return response()->json($this->cache->remember( + return $this->snapshotResponse($this->cache->remember( $scope . '.summary', ['filters' => $filters], $resolver @@ -359,13 +359,24 @@ class SupplierPayables extends ApiController private function optionsResponse(string $scope, Closure $resolver) { - return response()->json($this->cache->remember( + return $this->snapshotResponse($this->cache->remember( $scope . '.options', [], $resolver )); } + private function snapshotResponse(array $payload) + { + $response = response()->json($payload); + + foreach ($this->cache->responseHeaders() as $name => $value) { + $response->headers->set($name, $value); + } + + return $response; + } + private function filters(string $type): array { $filters = []; diff --git a/modules/EcommerceFinance/Repositories/SupplierPayableRepository.php b/modules/EcommerceFinance/Repositories/SupplierPayableRepository.php index 4af4bdc4..8de2a03c 100644 --- a/modules/EcommerceFinance/Repositories/SupplierPayableRepository.php +++ b/modules/EcommerceFinance/Repositories/SupplierPayableRepository.php @@ -29,7 +29,14 @@ class SupplierPayableRepository /** * 通用工具：带筛选 + 分页的查询 */ - private function runQuery(string $sql, array $params, int $page, int $pageSize, int $countThreshold = 5000): array + private function runQuery( + string $sql, + array $params, + int $page, + int $pageSize, + string $orderBy = '(SELECT 0)', + int $countThreshold = 5000 + ): array { $offset = ($page - 1) * $pageSize; @@ -45,8 +52,9 @@ class SupplierPayableRepository $total = -1; } - // 2. 分页数据（用 TOP + OFFSET/FETCH 不可行，统一使用 ROW_NUMBER() OVER 包装） - $pagedSql = 'SELECT * FROM (SELECT *, ROW_NUMBER() OVER (ORDER BY (SELECT 0)) AS rn FROM (' . $sql . ') AS src) AS t WHERE rn BETWEEN ? AND ?'; + // 2. 分页数据：排序字段只允许由仓库内部固定传入，不接受请求参数。 + // ROW_NUMBER 位于无 ORDER BY 的派生表外层，兼容旧版 SQL Server。 + $pagedSql = 'SELECT * FROM (SELECT src.*, ROW_NUMBER() OVER (ORDER BY ' . $orderBy . ') AS rn FROM (' . $sql . ') AS src) AS t WHERE rn BETWEEN ? AND ? ORDER BY rn'; $paramsWithRange = array_merge($params, [$offset + 1, $offset + $pageSize]); $stmt = $this->db()->prepare($pagedSql); @@ -61,10 +69,17 @@ class SupplierPayableRepository /** * 计算分页的 SELECT TOP N（无 COUNT），加速大表查询 */ - private function runQueryFast(string $sql, array $params, int $page, int $pageSize, int $total = -1): array + private function runQueryFast( + string $sql, + array $params, + int $page, + int $pageSize, + int $total = -1, + string $orderBy = '(SELECT 0)' + ): array { $offset = ($page - 1) * $pageSize; - $pagedSql = 'SELECT * FROM (SELECT *, ROW_NUMBER() OVER (ORDER BY (SELECT 0)) AS rn FROM (' . $sql . ') AS src) AS t WHERE rn BETWEEN ? AND ?'; + $pagedSql = 'SELECT * FROM (SELECT src.*, ROW_NUMBER() OVER (ORDER BY ' . $orderBy . ') AS rn FROM (' . $sql . ') AS src) AS t WHERE rn BETWEEN ? AND ? ORDER BY rn'; $paramsWithRange = array_merge($params, [$offset + 1, $offset + $pageSize]); $stmt = $this->db()->prepare($pagedSql); $stmt->execute($paramsWithRange); @@ -135,7 +150,7 @@ class SupplierPayableRepository FROM finance_purchaseReturn $whereSql"; - $result = $this->runQuery($sql, $params, $page, $pageSize); + $result = $this->runQuery($sql, $params, $page, $pageSize, 'send_time DESC, wms_pick_no DESC'); return [ 'data' => $result['data'], @@ -217,7 +232,7 @@ class SupplierPayableRepository FROM finance_invoiceDeduction $whereSql"; - return $this->runQuery($sql, $params, $page, $pageSize); + return $this->runQuery($sql, $params, $page, $pageSize, 'month_str DESC, create_time DESC'); } public function invoiceDeductionSummary(array $filters): array @@ -292,10 +307,9 @@ class SupplierPayableRepository create_time FROM fys_pressing_factory_compensation_detail ) AS t - $whereSql - ORDER BY month DESC, create_time DESC"; + $whereSql"; - return $this->runQuery($sql, $params, $page, $pageSize); + return $this->runQuery($sql, $params, $page, $pageSize, 'month DESC, create_time DESC'); } public function compensationDetailSummary(array $filters): array @@ -374,10 +388,9 @@ class SupplierPayableRepository CONVERT(varchar(19), createTime, 120) AS create_time, creator FROM finance_advanceLoans - $whereSql - ORDER BY month DESC, createTime DESC"; + $whereSql"; - return $this->runQuery($sql, $params, $page, $pageSize); + return $this->runQuery($sql, $params, $page, $pageSize, 'month_str DESC, create_time DESC'); } public function advanceDetailSummary(array $filters): array @@ -455,10 +468,9 @@ class SupplierPayableRepository businessId, CONVERT(varchar(19), updateAt, 120) AS update_time FROM finance_sampleDeduction - $whereSql - ORDER BY finishTime DESC"; + $whereSql"; - return $this->runQuery($sql, $params, $page, $pageSize); + return $this->runQuery($sql, $params, $page, $pageSize, 'finish_time DESC'); } public function sampleDeductionSummary(array $filters): array @@ -542,10 +554,9 @@ class SupplierPayableRepository CONVERT(varchar(10), shipDate, 120) AS ship_date, expressName FROM pm_unOrderDetails - $whereSql - ORDER BY date DESC, Id DESC"; + $whereSql"; - return $this->runQuery($sql, $params, $page, $pageSize); + return $this->runQuery($sql, $params, $page, $pageSize, 'date DESC, id DESC'); } public function deliveryUnpricedSummary(array $filters): array @@ -622,10 +633,9 @@ class SupplierPayableRepository labels, creator_name AS creator FROM finance_purchaseReturn - $whereSql - ORDER BY date DESC"; + $whereSql"; - return $this->runQuery($sql, $params, $page, $pageSize); + return $this->runQuery($sql, $params, $page, $pageSize, 'send_time DESC'); } public function returnUnpricedSummary(array $filters): array @@ -753,10 +763,9 @@ class SupplierPayableRepository mark AS remark2, checker AS person_charge FROM fys_supplierPaymentSummary - $whereSql - ORDER BY Month DESC, supplier"; + $whereSql"; - return $this->runQuery($sql, $params, $page, $pageSize); + return $this->runQuery($sql, $params, $page, $pageSize, 'month_str DESC, supplier'); } public function otherAdjustmentsSummary(array $filters): array @@ -830,10 +839,9 @@ class SupplierPayableRepository creator, CONVERT(varchar(19), create_time, 120) AS create_time FROM fys_express_courier_settlement - $whereSql - ORDER BY month DESC, expressname"; + $whereSql"; - return $this->runQuery($sql, $params, $page, $pageSize); + return $this->runQuery($sql, $params, $page, $pageSize, 'month_str DESC, express_name'); } public function dropshippingExpressSummary(array $filters): array diff --git a/modules/EcommerceFinance/Services/JianSunQueryCache.php b/modules/EcommerceFinance/Services/JianSunQueryCache.php index 2b225f07..414339c3 100644 --- a/modules/EcommerceFinance/Services/JianSunQueryCache.php +++ b/modules/EcommerceFinance/Services/JianSunQueryCache.php @@ -3,6 +3,7 @@ namespace Modules\EcommerceFinance\Services; use Closure; +use Illuminate\Contracts\Cache\Lock; use Illuminate\Contracts\Cache\LockTimeoutException; use Illuminate\Support\Facades\Cache; use Illuminate\Support\Facades\Log; @@ -10,19 +11,23 @@ use Symfony\Component\HttpKernel\Exception\ServiceUnavailableHttpException; use Throwable; /** - * 上游查询的本地快照与单飞保护。 + * 上游查询的本地快照、请求合并和压力保护。 * - * 这不是上游全库镜像：它只保留最近实际使用过的页面查询，避免 2GB - * 服务器复制数百万行，同时确保多个用户不会重复冲击上游数据库。 + * 页面命中新鲜快照时立即返回；命中过期快照时也立即返回，并把刷新登记给 + * 后台任务。只有完全没有本地快照的冷查询才会在请求内受控访问一次上游。 */ class JianSunQueryCache { - private const CACHE_VERSION = 'v1'; + private const CACHE_VERSION = 'v2'; - private const REGISTRY_KEY = 'jiansun:snapshots:active'; + private const REGISTRY_KEY = 'jiansun:snapshots:active:v2'; + + private const GLOBAL_CIRCUIT_KEY = 'jiansun:upstream-circuit:v2'; + + private ?array $lastReadState = null; /** - * 读取新鲜快照；未命中时在全局锁内查询一次上游。 + * 读取本地快照；过期快照采用 stale-while-revalidate，不阻塞页面。 */ public function remember(string $scope, array $parameters, Closure $resolver): mixed { @@ -30,20 +35,50 @@ class JianSunQueryCache $key = $this->cacheKey($scope, $parameters); $cached = Cache::get($key); - if (is_array($cached) && array_key_exists('value', $cached)) { - $this->register($scope, $parameters); + if ($this->hasValue($cached)) { + $this->register($scope, $parameters, false); + $this->setReadState('fresh', $cached); return $cached['value']; } + $stale = Cache::get($key . ':stale'); + + if ($this->hasValue($stale)) { + // 页面不等待上游；下一轮后台任务优先刷新这个查询。 + $this->register($scope, $parameters, true); + $this->setReadState('stale', $stale); + + return $stale['value']; + } + + // 首次访问没有任何本地数据时才允许同步获取一次；先登记，失败后后台 + // 仍可在熔断结束后继续补齐快照。 + $this->register($scope, $parameters, true); $resolved = $this->resolve($scope, $parameters, $resolver, false); - $this->register($scope, $parameters); + $payload = Cache::get($key) ?: Cache::get($key . ':stale'); + $this->setReadState($resolved['status'], is_array($payload) ? $payload : []); return $resolved['value']; } /** - * 定时刷新一个已经使用过的快照。 + * 供 API 响应暴露数据新鲜度，不改变既有 JSON 结构。 + */ + public function responseHeaders(): array + { + if ($this->lastReadState === null) { + return []; + } + + return array_filter([ + 'X-Caiwu-Data-Status' => $this->lastReadState['status'] ?? null, + 'X-Caiwu-Data-Synced-At' => $this->lastReadState['cached_at'] ?? null, + ]); + } + + /** + * 后台刷新一个最近使用过的快照。 * * @return string fresh|stale */ @@ -55,12 +90,12 @@ class JianSunQueryCache } /** - * 返回最近实际访问过的查询，供十五分钟定时任务逐个刷新。 + * 返回最近实际访问且已经到期的查询，紧急项和列表优先。 */ public function activeEntries(?int $limit = null): array { $now = time(); - $activeTtl = max((int) config('jiansun.snapshots.active_ttl_seconds', 86400), 900); + $activeTtl = max((int) config('jiansun.snapshots.active_ttl_seconds', 21600), 900); $entries = Cache::get(self::REGISTRY_KEY, []); if (! is_array($entries)) { @@ -74,9 +109,23 @@ class JianSunQueryCache && $this->isDue($entry, $now); }); - usort($entries, static fn (array $a, array $b): int => $b['seen_at'] <=> $a['seen_at']); + usort($entries, function (array $a, array $b): int { + $urgent = ((int) ($b['urgent'] ?? 0)) <=> ((int) ($a['urgent'] ?? 0)); - $limit ??= max((int) config('jiansun.snapshots.refresh_limit', 12), 1); + if ($urgent !== 0) { + return $urgent; + } + + $priority = $this->priorityFor((string) $b['scope']) <=> $this->priorityFor((string) $a['scope']); + + if ($priority !== 0) { + return $priority; + } + + return ((int) $b['seen_at']) <=> ((int) $a['seen_at']); + }); + + $limit ??= max((int) config('jiansun.snapshots.refresh_limit', 2), 1); return array_slice($entries, 0, $limit); } @@ -85,38 +134,53 @@ class JianSunQueryCache { $key = $this->cacheKey($scope, $parameters); $staleKey = $key . ':stale'; - $lock = Cache::lock( - 'jiansun:upstream-query-lock', + $queryId = hash('sha256', $scope . '|' . json_encode($parameters, JSON_UNESCAPED_UNICODE)); + $queryLock = Cache::lock( + 'jiansun:query-lock:' . $queryId, max((int) config('jiansun.snapshots.lock_seconds', 45), 15) ); try { - return $lock->block( - max((int) config('jiansun.snapshots.lock_wait_seconds', 12), 1), - function () use ($scope, $parameters, $resolver, $force, $key, $staleKey): array { + return $queryLock->block( + max((int) config('jiansun.snapshots.lock_wait_seconds', 3), 1), + function () use ($scope, $parameters, $resolver, $force, $key, $staleKey, $queryId): array { if (! $force) { $cached = Cache::get($key); - if (is_array($cached) && array_key_exists('value', $cached)) { + if ($this->hasValue($cached)) { + $this->markRefreshed($scope, $parameters); + return ['value' => $cached['value'], 'status' => 'fresh']; } } + if ($this->isCircuitOpen() || Cache::has($this->queryCooldownKey($queryId))) { + return $this->staleOrFail($staleKey, '上游数据正在安全冷却，请稍后再试。'); + } + if (! $this->consumeBudget()) { return $this->staleOrFail($staleKey, '上游查询刷新已达到安全上限，请稍后再试。'); } + $slot = $this->acquireUpstreamSlot(); + + if ($slot === null) { + return $this->staleOrFail($staleKey, '上游同步任务繁忙，请稍后再试。'); + } + + $startedAt = microtime(true); + try { $value = $resolver(); } catch (Throwable $exception) { - Log::warning('JianSun 上游查询失败，尝试读取旧快照。', [ - 'scope' => $scope, - 'exception' => $exception::class, - ]); + $this->recordFailure($scope, $queryId, $exception); return $this->staleOrFail($staleKey, '上游数据暂时不可用，请稍后再试。', $exception); + } finally { + $slot->release(); } + $durationMs = (int) round((microtime(true) - $startedAt) * 1000); $payload = [ 'value' => $value, 'cached_at' => now()->toIso8601String(), @@ -130,11 +194,168 @@ class JianSunQueryCache max((int) config('jiansun.snapshots.stale_ttl_seconds', 86400), $this->ttlFor($scope)) ); + $this->recordSuccess($scope, $parameters, $queryId, $durationMs, $value); + return ['value' => $value, 'status' => 'fresh']; } ); } catch (LockTimeoutException $exception) { - return $this->staleOrFail($staleKey, '上游数据正在同步，请稍后再试。', $exception); + return $this->staleOrFail($staleKey, '相同数据正在同步，请稍后再试。', $exception); + } + } + + private function acquireUpstreamSlot(): ?Lock + { + $slotCount = min(max((int) config('jiansun.snapshots.max_concurrency', 2), 1), 4); + $waitMs = min(max((int) config('jiansun.snapshots.slot_wait_ms', 1500), 0), 5000); + $deadline = microtime(true) + ($waitMs / 1000); + $lockSeconds = max( + (int) config('jiansun.connection.connect_timeout', 5) + + (int) config('jiansun.connection.query_timeout', 15) + + 10, + 15 + ); + + do { + for ($slot = 0; $slot < $slotCount; $slot++) { + $lock = Cache::lock('jiansun:upstream-slot:' . $slot, $lockSeconds); + + if ($lock->get()) { + return $lock; + } + } + + if (microtime(true) >= $deadline) { + break; + } + + usleep(50000); + } while (true); + + return null; + } + + private function consumeBudget(): bool + { + $window = max((int) config('jiansun.snapshots.budget_window_seconds', 60), 60); + $limit = max((int) config('jiansun.snapshots.miss_budget_per_window', 6), 1); + $bucket = intdiv(time(), $window); + $key = 'jiansun:upstream-budget:' . $bucket; + $lock = Cache::lock('jiansun:upstream-budget-lock', 3); + + try { + return $lock->block(1, function () use ($key, $limit, $window): bool { + $count = (int) Cache::get($key, 0); + + if ($count >= $limit) { + return false; + } + + Cache::put($key, $count + 1, $window + 60); + + return true; + }); + } catch (LockTimeoutException) { + // 预算状态不确定时宁可少查一次，也不冒险放大上游流量。 + return false; + } + } + + private function recordFailure(string $scope, string $queryId, Throwable $exception): void + { + $queryCooldown = max((int) config('jiansun.snapshots.query_failure_cooldown_seconds', 300), 60); + Cache::put($this->queryCooldownKey($queryId), true, $queryCooldown); + + if ($exception instanceof ServiceUnavailableHttpException) { + $globalCooldown = max((int) config('jiansun.snapshots.failure_cooldown_seconds', 300), 60); + Cache::put(self::GLOBAL_CIRCUIT_KEY, true, $globalCooldown); + } + + Log::warning('JianSun 上游查询失败，已进入冷却。', [ + 'scope' => $scope, + 'exception' => $exception::class, + ]); + } + + private function recordSuccess( + string $scope, + array $parameters, + string $queryId, + int $durationMs, + mixed $value + ): void { + $slowMs = max((int) config('jiansun.snapshots.slow_query_ms', 5000), 1000); + + if ($durationMs >= $slowMs) { + $cooldown = max((int) config('jiansun.snapshots.slow_query_cooldown_seconds', 300), 60); + Cache::put($this->queryCooldownKey($queryId), true, $cooldown); + Log::warning('JianSun 上游查询较慢，已降低该查询刷新频率。', [ + 'scope' => $scope, + 'duration_ms' => $durationMs, + ]); + } + + $this->markRefreshed($scope, $parameters, [ + 'last_duration_ms' => $durationMs, + 'last_row_count' => $this->rowCount($value), + 'last_refreshed_at' => time(), + ]); + } + + private function register(string $scope, array $parameters, bool $urgent): void + { + $entryKey = $this->entryKey($scope, $parameters); + $touchSeconds = $urgent ? 30 : 300; + + if (! Cache::add('jiansun:registry-touch:' . $entryKey . ':' . (int) $urgent, true, $touchSeconds)) { + return; + } + + $this->updateRegistry(function (array $entries) use ($entryKey, $scope, $parameters, $urgent): array { + $existing = (array) ($entries[$entryKey] ?? []); + $entries[$entryKey] = array_merge($existing, [ + 'scope' => $scope, + 'parameters' => $parameters, + 'seen_at' => time(), + 'urgent' => $urgent || (bool) ($existing['urgent'] ?? false), + ]); + + return $entries; + }); + } + + private function markRefreshed(string $scope, array $parameters, array $metrics = []): void + { + $entryKey = $this->entryKey($scope, $parameters); + + $this->updateRegistry(function (array $entries) use ($entryKey, $metrics): array { + if (! isset($entries[$entryKey])) { + return $entries; + } + + $entries[$entryKey] = array_merge($entries[$entryKey], $metrics, ['urgent' => false]); + + return $entries; + }); + } + + private function updateRegistry(Closure $callback): void + { + $lock = Cache::lock('jiansun:registry-lock:v2', 5); + + try { + $lock->block(2, function () use ($callback): void { + $entries = Cache::get(self::REGISTRY_KEY, []); + $entries = is_array($entries) ? $entries : []; + $entries = $callback($entries); + + uasort($entries, static fn (array $a, array $b): int => ((int) $b['seen_at']) <=> ((int) $a['seen_at'])); + $limit = max((int) config('jiansun.snapshots.registry_limit', 300), 1); + + Cache::forever(self::REGISTRY_KEY, array_slice($entries, 0, $limit, true)); + }); + } catch (LockTimeoutException) { + // 登记失败不能阻塞页面；下一次访问或刷新会再次尝试。 } } @@ -142,62 +363,26 @@ class JianSunQueryCache { $stale = Cache::get($staleKey); - if (is_array($stale) && array_key_exists('value', $stale)) { + if ($this->hasValue($stale)) { return ['value' => $stale['value'], 'status' => 'stale']; } throw new ServiceUnavailableHttpException(30, $message, $previous); } - /** - * 所有刷新都在同一把全局锁内执行，因此这里的计数读写不会并发冲突。 - */ - private function consumeBudget(): bool + private function isCircuitOpen(): bool { - $window = max((int) config('jiansun.snapshots.budget_window_seconds', 900), 60); - $limit = max((int) config('jiansun.snapshots.miss_budget_per_window', 20), 1); - $bucket = intdiv(time(), $window); - $key = 'jiansun:upstream-budget:' . $bucket; - $count = (int) Cache::get($key, 0); - - if ($count >= $limit) { - return false; - } - - Cache::put($key, $count + 1, $window + 60); - - return true; + return Cache::has(self::GLOBAL_CIRCUIT_KEY); } - private function register(string $scope, array $parameters): void + private function queryCooldownKey(string $queryId): string { - $entryKey = hash('sha256', $scope . '|' . json_encode($parameters, JSON_UNESCAPED_UNICODE)); + return 'jiansun:query-cooldown:' . $queryId; + } - // 同一查询五分钟内只更新一次活跃时间，减少文件缓存写入。 - if (! Cache::add('jiansun:registry-touch:' . $entryKey, true, 300)) { - return; - } - - $lock = Cache::lock('jiansun:registry-lock', 5); - - try { - $lock->block(2, function () use ($entryKey, $scope, $parameters): void { - $entries = Cache::get(self::REGISTRY_KEY, []); - $entries = is_array($entries) ? $entries : []; - $entries[$entryKey] = [ - 'scope' => $scope, - 'parameters' => $parameters, - 'seen_at' => time(), - ]; - - uasort($entries, static fn (array $a, array $b): int => $b['seen_at'] <=> $a['seen_at']); - $limit = max((int) config('jiansun.snapshots.registry_limit', 60), 1); - - Cache::forever(self::REGISTRY_KEY, array_slice($entries, 0, $limit, true)); - }); - } catch (LockTimeoutException) { - // 注册失败不应阻塞页面；下一次请求会再次尝试。 - } + private function entryKey(string $scope, array $parameters): string + { + return hash('sha256', $scope . '|' . json_encode($parameters, JSON_UNESCAPED_UNICODE)); } private function cacheKey(string $scope, array $parameters): string @@ -212,11 +397,19 @@ class JianSunQueryCache return (int) config('jiansun.snapshots.options_ttl_seconds', 3600); } - return (int) config('jiansun.snapshots.ttl_seconds', 900); + if (str_ends_with($scope, '.summary')) { + return (int) config('jiansun.snapshots.summary_ttl_seconds', 120); + } + + return (int) config('jiansun.snapshots.ttl_seconds', 120); } private function isDue(array $entry, int $now): bool { + if ((bool) ($entry['urgent'] ?? false)) { + return true; + } + $scope = (string) $entry['scope']; $parameters = (array) $entry['parameters']; $cached = Cache::get($this->cacheKey($scope, $parameters)); @@ -225,8 +418,43 @@ class JianSunQueryCache return true; } - // 给 cron 一分钟余量，避免刚好在过期边界前运行后再等十五分钟。 - return ((int) $cached['cached_at_ts'] + max($this->ttlFor($scope), 60)) <= ($now + 60); + // 给一分钟任务留少量余量，使页面看到的数据通常保持在两分钟以内。 + return ((int) $cached['cached_at_ts'] + max($this->ttlFor($scope), 60)) <= ($now + 15); + } + + private function priorityFor(string $scope): int + { + return match (true) { + str_ends_with($scope, '.list') => 3, + str_ends_with($scope, '.summary') => 2, + default => 1, + }; + } + + private function rowCount(mixed $value): int + { + if (! is_array($value)) { + return 1; + } + + if (isset($value['data']) && is_array($value['data'])) { + return count($value['data']); + } + + return count($value); + } + + private function hasValue(mixed $payload): bool + { + return is_array($payload) && array_key_exists('value', $payload); + } + + private function setReadState(string $status, array $payload): void + { + $this->lastReadState = [ + 'status' => $status, + 'cached_at' => $payload['cached_at'] ?? null, + ]; } private function normalize(array $parameters): array diff --git a/tests/Unit/EcommerceFinance/JianSunQueryCacheTest.php b/tests/Unit/EcommerceFinance/JianSunQueryCacheTest.php new file mode 100644 index 00000000..29a3daea --- /dev/null +++ b/tests/Unit/EcommerceFinance/JianSunQueryCacheTest.php @@ -0,0 +1,134 @@ +set('jiansun.snapshots.ttl_seconds', 60); + config()->set('jiansun.snapshots.summary_ttl_seconds', 60); + config()->set('jiansun.snapshots.options_ttl_seconds', 3600); + config()->set('jiansun.snapshots.stale_ttl_seconds', 3600); + config()->set('jiansun.snapshots.active_ttl_seconds', 3600); + config()->set('jiansun.snapshots.registry_limit', 300); + config()->set('jiansun.snapshots.refresh_limit', 2); + config()->set('jiansun.snapshots.miss_budget_per_window', 1000); + config()->set('jiansun.snapshots.budget_window_seconds', 60); + config()->set('jiansun.snapshots.max_concurrency', 2); + config()->set('jiansun.snapshots.slot_wait_ms', 0); + config()->set('jiansun.snapshots.lock_wait_seconds', 1); + config()->set('jiansun.snapshots.slow_query_ms', 5000); + } + + protected function tearDown(): void + { + Carbon::setTestNow(); + Cache::flush(); + + parent::tearDown(); + } + + public function test_identical_queries_are_shared_without_user_count_limit(): void + { + $cache = app(JianSunQueryCache::class); + $calls = 0; + + for ($i = 0; $i < 100; $i++) { + $value = $cache->remember('goods.list', ['page' => 1], function () use (&$calls): array { + $calls++; + + return ['data' => [['id' => 1]], 'total' => 1]; + }); + + $this->assertSame(1, $value['total']); + } + + $this->assertSame(1, $calls); + } + + public function test_expired_snapshot_returns_immediately_and_is_refreshed_later(): void + { + $cache = app(JianSunQueryCache::class); + $calls = 0; + + $first = $cache->remember('goods.list', ['page' => 1], function () use (&$calls): array { + $calls++; + + return ['data' => [['version' => 1]], 'total' => 1]; + }); + + Carbon::setTestNow(now()->addSeconds(61)); + + $stale = $cache->remember('goods.list', ['page' => 1], function () use (&$calls): array { + $calls++; + + return ['data' => [['version' => 999]], 'total' => 1]; + }); + + $this->assertSame($first, $stale); + $this->assertSame(1, $calls, '页面读取过期快照时不应同步访问上游。'); + $this->assertSame('stale', $cache->responseHeaders()['X-Caiwu-Data-Status']); + $this->assertTrue((bool) ($cache->activeEntries(1)[0]['urgent'] ?? false)); + + $status = $cache->refresh('goods.list', ['page' => 1], function () use (&$calls): array { + $calls++; + + return ['data' => [['version' => 2]], 'total' => 1]; + }); + + $fresh = $cache->remember('goods.list', ['page' => 1], fn (): array => []); + + $this->assertSame('fresh', $status); + $this->assertSame(2, $fresh['data'][0]['version']); + $this->assertSame('fresh', $cache->responseHeaders()['X-Caiwu-Data-Status']); + $this->assertSame(2, $calls); + } + + public function test_unrelated_cold_queries_do_not_share_one_global_query_lock(): void + { + $cache = app(JianSunQueryCache::class); + $parametersA = ['page' => 1]; + $queryA = hash('sha256', 'goods.list|' . json_encode($parametersA, JSON_UNESCAPED_UNICODE)); + $lockA = Cache::lock('jiansun:query-lock:' . $queryA, 30); + + $this->assertTrue($lockA->get()); + + try { + $value = $cache->remember('goods.summary', [], fn (): array => ['total' => 7]); + $this->assertSame(7, $value['total']); + } finally { + $lockA->release(); + } + } + + public function test_upstream_slots_bound_parallel_cold_queries(): void + { + $cache = app(JianSunQueryCache::class); + $slot0 = Cache::lock('jiansun:upstream-slot:0', 30); + $slot1 = Cache::lock('jiansun:upstream-slot:1', 30); + + $this->assertTrue($slot0->get()); + $this->assertTrue($slot1->get()); + + try { + $this->expectException(ServiceUnavailableHttpException::class); + $cache->remember('goods.list', ['page' => 2], fn (): array => ['data' => [], 'total' => 0]); + } finally { + $slot0->release(); + $slot1->release(); + } + } +}