CaiWuXiTong/modules/EcommerceFinance/Services/UpstreamQueryCache.php

671 lines
22 KiB
PHP
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

<?php
namespace Modules\EcommerceFinance\Services;
use Closure;
use Illuminate\Cache\CacheManager;
use Illuminate\Contracts\Cache\Repository;
use Illuminate\Contracts\Cache\Lock;
use Illuminate\Contracts\Cache\LockTimeoutException;
use Illuminate\Support\Facades\Log;
use Modules\EcommerceFinance\Exceptions\UpstreamSnapshotPendingException;
use Symfony\Component\HttpKernel\Exception\ServiceUnavailableHttpException;
use Throwable;
/**
* 任意上游数据源共用的共享快照、请求合并、限流和熔断实现。
*
* 所有缓存键和压力预算都包含 source,确保一个上游故障或繁忙时不会
* 阻塞其他上游。页面命中过期快照时立即返回旧值,由后台任务刷新。
*/
class UpstreamQueryCache
{
private const CACHE_VERSION = 'v2';
private readonly string $source;
private readonly string $label;
private readonly UpstreamSourceRegistry $sources;
private readonly Repository $cache;
private ?array $lastReadState = null;
private ?array $accessContext = null;
public function __construct(
UpstreamSourceRegistry $sources,
string $source,
CacheManager $cacheManager,
) {
$definition = $sources->definition($source);
$this->sources = $sources;
$this->source = $definition['key'];
$this->label = $definition['label'];
$this->cache = $cacheManager->store(
(string) config('upstreams.snapshots.cache_store', 'upstream_shared')
);
}
public function sourceKey(): string
{
return $this->source;
}
public function setting(string $path, mixed $default = null): mixed
{
return $this->sources->value($this->source, $path, $default);
}
/**
* 读取共享快照;过期快照采用 stale-while-revalidate,不阻塞页面。
*/
public function remember(string $scope, array $parameters, Closure $resolver): mixed
{
if (! (bool) $this->setting('enabled', false)) {
throw new ServiceUnavailableHttpException(60, '该上游数据同步尚未启用,请联系管理员。');
}
if ($this->usesDirectDevelopmentRead()) {
$value = $resolver();
$this->setReadState('direct', [
'cached_at' => now()->toIso8601String(),
]);
return $value;
}
if (! $this->isScopeAllowed($scope)) {
throw new ServiceUnavailableHttpException(30, '该数据范围仍在低压力验证中,请稍后再试。');
}
$parameters = $this->prepareParameters($parameters);
$key = $this->cacheKey($scope, $parameters);
$cached = $this->cache->get($key);
if ($this->hasValue($cached)) {
$this->register($scope, $parameters, false);
$this->setReadState('fresh', $cached);
return $cached['value'];
}
$stale = $this->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);
if (! (bool) config('upstreams.snapshots.allow_inline_refresh', false)) {
throw new UpstreamSnapshotPendingException();
}
$resolved = $this->resolve($scope, $parameters, $resolver, false);
$payload = $this->cache->get($key) ?: $this->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
*/
public function refresh(string $scope, array $parameters, Closure $resolver): string
{
$parameters = $this->prepareParameters($parameters);
if (! (bool) $this->setting('enabled', false) || ! $this->isScopeAllowed($scope)) {
return 'stale';
}
return $this->resolve($scope, $parameters, $resolver, true)['status'];
}
/**
* 返回本来源最近实际访问且已经到期的查询。
*/
public function activeEntries(?int $limit = null): array
{
if (! (bool) $this->setting('enabled', false)) {
return [];
}
$now = time();
$activeTtl = max((int) $this->setting('snapshots.active_ttl_seconds', 21600), 900);
$entries = $this->cache->get($this->registryKey(), []);
if (! is_array($entries)) {
return [];
}
$entries = array_filter($entries, function (mixed $entry) use ($now, $activeTtl): bool {
return is_array($entry)
&& isset($entry['seen_at'], $entry['scope'], $entry['parameters'])
&& $this->isScopeAllowed((string) $entry['scope'])
&& ($now - (int) $entry['seen_at']) <= $activeTtl
&& $this->isDue($entry, $now);
});
usort($entries, function (array $a, array $b): int {
$urgent = ((int) ($b['urgent'] ?? 0)) <=> ((int) ($a['urgent'] ?? 0));
if ($urgent !== 0) {
return $urgent;
}
// 首份快照和过期快照都标记为 urgent。对于这些请求,必须让
// 等待最久的查询先执行;否则持续出现的新列表查询会凭借下方
// 的类型优先级,长期挤掉同一页面的 summary/options,页面就会
// 一直停在“列表有了,但汇总或筛选项仍 503”的半完成状态。
if ((bool) ($a['urgent'] ?? false)) {
$oldest = ((int) ($a['seen_at'] ?? 0)) <=> ((int) ($b['seen_at'] ?? 0));
if ($oldest !== 0) {
return $oldest;
}
}
$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) $this->setting('snapshots.refresh_limit', 2), 1);
return array_slice($entries, 0, $limit);
}
private function resolve(string $scope, array $parameters, Closure $resolver, bool $force): array
{
$key = $this->cacheKey($scope, $parameters);
$staleKey = $key . ':stale';
$queryId = hash('sha256', $scope . '|' . json_encode($parameters, JSON_UNESCAPED_UNICODE));
$queryLock = $this->cache->lock(
$this->key('query-lock:' . $queryId),
max((int) $this->setting('snapshots.lock_seconds', 45), 15)
);
try {
return $queryLock->block(
max((int) $this->setting('snapshots.lock_wait_seconds', 3), 1),
function () use ($scope, $parameters, $resolver, $force, $key, $staleKey, $queryId): array {
if (! $force) {
$cached = $this->cache->get($key);
if ($this->hasValue($cached)) {
$this->markRefreshed($scope, $parameters);
return ['value' => $cached['value'], 'status' => 'fresh'];
}
}
if ($this->isCircuitOpen() || $this->cache->has($this->queryCooldownKey($queryId))) {
return $this->staleOrFail($staleKey, '上游数据正在安全冷却,请稍后再试。');
}
if (! $this->consumeBudget()) {
return $this->staleOrFail($staleKey, '上游查询刷新已达到安全上限,请稍后再试。');
}
$slots = $this->acquireUpstreamSlots();
if ($slots === null) {
return $this->staleOrFail($staleKey, '上游同步任务繁忙,请稍后再试。');
}
$startedAt = microtime(true);
try {
$value = $resolver();
} catch (Throwable $exception) {
$this->recordFailure($scope, $queryId, $exception);
return $this->staleOrFail($staleKey, '上游数据暂时不可用,请稍后再试。', $exception);
} finally {
$this->releaseUpstreamSlots($slots);
}
$durationMs = (int) round((microtime(true) - $startedAt) * 1000);
$payload = [
'value' => $value,
'cached_at' => now()->toIso8601String(),
'cached_at_ts' => time(),
];
$this->cache->put($key, $payload, max($this->ttlFor($scope), 60));
$this->cache->put(
$staleKey,
$payload,
max((int) $this->setting('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 array<int,Lock>|null */
private function acquireUpstreamSlots(): ?array
{
$sourceSlotCount = min(max((int) $this->setting('snapshots.max_concurrency', 2), 1), 4);
$globalSlotCount = min(max((int) config('upstreams.runtime.max_concurrency', 4), 1), 8);
$waitMs = min(max(
min(
(int) $this->setting('snapshots.slot_wait_ms', 1500),
(int) config('upstreams.runtime.slot_wait_ms', 1500)
),
0
), 5000);
$deadline = microtime(true) + ($waitMs / 1000);
$lockSeconds = max(
(int) $this->setting('connection.connect_timeout', 5)
+ (int) $this->setting('connection.query_timeout', 15)
+ 10,
15
);
do {
for ($sourceSlot = 0; $sourceSlot < $sourceSlotCount; $sourceSlot++) {
$sourceLock = $this->cache->lock($this->key('upstream-slot:' . $sourceSlot), $lockSeconds);
if (! $sourceLock->get()) {
continue;
}
for ($globalSlot = 0; $globalSlot < $globalSlotCount; $globalSlot++) {
$globalLock = $this->cache->lock('upstream:global-slot:' . $globalSlot, $lockSeconds);
if ($globalLock->get()) {
return [$sourceLock, $globalLock];
}
}
$sourceLock->release();
}
if (microtime(true) >= $deadline) {
break;
}
usleep(50000);
} while (true);
return null;
}
/** @param array<int,Lock> $slots */
private function releaseUpstreamSlots(array $slots): void
{
foreach (array_reverse($slots) as $slot) {
$slot->release();
}
}
private function consumeBudget(): bool
{
$window = max((int) $this->setting('snapshots.budget_window_seconds', 60), 60);
$limit = max((int) $this->setting('snapshots.miss_budget_per_window', 6), 1);
$bucket = intdiv(time(), $window);
$key = $this->key('upstream-budget:' . $bucket);
$lock = $this->cache->lock($this->key('upstream-budget-lock'), 3);
try {
return $lock->block(1, function () use ($key, $limit, $window): bool {
$count = (int) $this->cache->get($key, 0);
if ($count >= $limit) {
return false;
}
$this->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) $this->setting('snapshots.query_failure_cooldown_seconds', 300), 60);
$this->cache->put($this->queryCooldownKey($queryId), true, $queryCooldown);
if ($exception instanceof ServiceUnavailableHttpException) {
$globalCooldown = max((int) $this->setting('snapshots.failure_cooldown_seconds', 300), 60);
$this->cache->put($this->circuitKey(), true, $globalCooldown);
}
Log::warning('上游查询失败,已进入冷却。', [
'source' => $this->source,
'source_label' => $this->label,
'scope' => $scope,
'exception' => $exception::class,
]);
}
private function recordSuccess(
string $scope,
array $parameters,
string $queryId,
int $durationMs,
mixed $value
): void {
$slowMs = max((int) $this->setting('snapshots.slow_query_ms', 5000), 1000);
if ($durationMs >= $slowMs) {
$cooldown = max((int) $this->setting('snapshots.slow_query_cooldown_seconds', 300), 60);
$this->cache->put($this->queryCooldownKey($queryId), true, $cooldown);
Log::warning('上游查询较慢,已降低该查询刷新频率。', [
'source' => $this->source,
'source_label' => $this->label,
'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 (! $this->cache->add($this->key('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, [
'source' => $this->source,
'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 = $this->cache->lock($this->key('registry-lock:' . self::CACHE_VERSION), 5);
try {
$lock->block(2, function () use ($callback): void {
$entries = $this->cache->get($this->registryKey(), []);
$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) $this->setting('snapshots.registry_limit', 300), 1);
$this->cache->forever($this->registryKey(), array_slice($entries, 0, $limit, true));
});
} catch (LockTimeoutException) {
// 登记失败不能阻塞页面;下一次访问或刷新会再次尝试。
}
}
private function staleOrFail(string $staleKey, string $message, ?Throwable $previous = null): array
{
$stale = $this->cache->get($staleKey);
if ($this->hasValue($stale)) {
return ['value' => $stale['value'], 'status' => 'stale'];
}
throw new ServiceUnavailableHttpException(30, $message, $previous);
}
private function isCircuitOpen(): bool
{
return $this->cache->has($this->circuitKey());
}
private function queryCooldownKey(string $queryId): string
{
return $this->key('query-cooldown:' . $queryId);
}
private function isScopeAllowed(string $scope): bool
{
$allowed = (array) $this->setting('allowed_scopes', []);
foreach ($allowed as $pattern) {
$pattern = trim((string) $pattern);
if ($pattern === '*' || $pattern === $scope) {
return true;
}
if (str_ends_with($pattern, '.*') && str_starts_with($scope, substr($pattern, 0, -1))) {
return true;
}
}
return false;
}
private function usesDirectDevelopmentRead(): bool
{
return (bool) config('upstreams.development.direct_read', false);
}
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
{
return $this->key('snapshot:' . self::CACHE_VERSION . ':' . $scope . ':'
. hash('sha256', json_encode($parameters, JSON_UNESCAPED_UNICODE | JSON_PRESERVE_ZERO_FRACTION)));
}
private function ttlFor(string $scope): int
{
if (str_ends_with($scope, '.options')) {
return (int) $this->setting('snapshots.options_ttl_seconds', 3600);
}
if (str_ends_with($scope, '.summary')) {
return (int) $this->setting('snapshots.summary_ttl_seconds', 120);
}
return (int) $this->setting('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 = $this->cache->get($this->cacheKey($scope, $parameters));
if (! is_array($cached) || ! isset($cached['cached_at_ts'])) {
return true;
}
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
{
foreach ($parameters as &$value) {
if (is_array($value)) {
$value = $this->normalize($value);
}
}
unset($value);
ksort($parameters);
return $parameters;
}
/**
* 快照必须按公司和权限范围隔离。后台刷新沿用首次访问时登记的上下文,
* 不依赖无登录用户的 CLI 进程重新推断权限。
*/
private function prepareParameters(array $parameters): array
{
$contextKey = '__caiwu_snapshot_context';
$context = isset($parameters[$contextKey]) && is_array($parameters[$contextKey])
? $this->normalizeContext($parameters[$contextKey])
: $this->currentAccessContext();
unset($parameters[$contextKey]);
$parameters[$contextKey] = $context;
return $this->normalize($parameters);
}
private function currentAccessContext(): array
{
if ($this->accessContext !== null) {
return $this->accessContext;
}
$companyId = function_exists('company_id') ? (int) (company_id() ?? 0) : 0;
$user = auth()->user();
if ($user === null) {
return $this->accessContext = [
'company_id' => $companyId,
'access_hash' => 'system',
];
}
$permissionIds = $user->allPermissions(['id'])
->pluck('id')
->map(static fn (mixed $id): int => (int) $id)
->sort()
->values()
->all();
return $this->accessContext = [
'company_id' => $companyId,
'access_hash' => hash('sha256', json_encode(
$permissionIds,
JSON_UNESCAPED_UNICODE | JSON_PRESERVE_ZERO_FRACTION
)),
];
}
private function normalizeContext(array $context): array
{
return [
'company_id' => max((int) ($context['company_id'] ?? 0), 0),
'access_hash' => preg_match('/^[a-f0-9]{64}$/', (string) ($context['access_hash'] ?? '')) === 1
? (string) $context['access_hash']
: 'system',
];
}
private function key(string $suffix): string
{
return $this->source . ':' . $suffix;
}
private function registryKey(): string
{
return $this->key('snapshots:active:' . self::CACHE_VERSION);
}
private function circuitKey(): string
{
return $this->key('upstream-circuit:' . self::CACHE_VERSION);
}
}