diff --git a/Console/Command/Flush.php b/Console/Command/Flush.php index d1d01a3..2c63713 100644 --- a/Console/Command/Flush.php +++ b/Console/Command/Flush.php @@ -8,21 +8,21 @@ use Symfony\Component\Console\Command\Command; use Symfony\Component\Console\Input\InputInterface; use Symfony\Component\Console\Output\OutputInterface; -use SamJUK\CacheDebounce\Model\Entries as CacheDebouncedEntries; +use SamJUK\CacheDebounce\Model\StaggeredFlush; class Flush extends Command { - private $cacheDebouncedEntries; + private $staggeredFlush; /** - * @param CacheDebouncedEntries $cacheDebouncedEntries + * @param StaggeredFlush $staggeredFlush * @param string|null $name */ public function __construct( - CacheDebouncedEntries $cacheDebouncedEntries, + StaggeredFlush $staggeredFlush, $name = null ) { - $this->cacheDebouncedEntries = $cacheDebouncedEntries; + $this->staggeredFlush = $staggeredFlush; parent::__construct($name); } @@ -37,7 +37,7 @@ protected function configure(): void protected function execute(InputInterface $input, OutputInterface $output): int { try { - $this->cacheDebouncedEntries->flush(); + $this->staggeredFlush->execute(); $output->writeln('Flushed Debounced Cache Purges.'); return 0; } catch (LocalizedException $e) { diff --git a/Console/Command/Status.php b/Console/Command/Status.php new file mode 100644 index 0000000..33240c6 --- /dev/null +++ b/Console/Command/Status.php @@ -0,0 +1,92 @@ +storage = $storage; + $this->config = $config; + $this->flagManager = $flagManager; + parent::__construct($name); + } + + protected function configure(): void + { + $this->setName('samjuk:cache-debounce:status'); + $this->setDescription('Shows the current state of the debounced purge queue.'); + + parent::configure(); + } + + protected function execute(InputInterface $input, OutputInterface $output): int + { + $pending = $this->storage->pendingCount(); + $oldestAgeSeconds = $this->storage->oldestPendingAgeSeconds(); + $activeBatch = $this->storage->activeBatch(); + $activeBatchTagCount = $activeBatch !== '' ? count($this->storage->tags($activeBatch)) : 0; + + $lastFlushAt = $this->flagManager->getFlagData(Entries::FLAG_LAST_FLUSH_AT); + $lastFlushDuration = $this->flagManager->getFlagData(Entries::FLAG_LAST_FLUSH_DURATION); + $consecutiveLaggingRuns = (int)$this->flagManager->getFlagData(LagDetector::FLAG_CONSECUTIVE_LAGGING_RUNS); + + $oldestAgeLabel = $oldestAgeSeconds !== null + ? sprintf(' (oldest queued %s ago)', $this->formatDuration($oldestAgeSeconds)) + : ''; + + $output->writeln(sprintf('Pending tags: %d%s', $pending, $oldestAgeLabel)); + + $output->writeln(sprintf( + 'Active batch: %s', + $activeBatch === '' ? 'none' : sprintf('%s (%d tags)', $activeBatch, $activeBatchTagCount) + )); + + $output->writeln(sprintf( + 'Last flush: %s', + $lastFlushAt ? sprintf('%s (duration %.1fs)', $lastFlushAt, (float)$lastFlushDuration) : 'never' + )); + + $output->writeln(sprintf( + 'Consecutive lagging runs: %d / %d', + $consecutiveLaggingRuns, + $this->config->getStaggerLagAlertAfterRuns() + )); + + return 0; + } + + private function formatDuration(int $seconds): string + { + if ($seconds < 60) { + return $seconds . 's'; + } + + return sprintf('%dm%02ds', intdiv($seconds, 60), $seconds % 60); + } +} diff --git a/Cron/Flush.php b/Cron/Flush.php index 4c0d163..64dc919 100644 --- a/Cron/Flush.php +++ b/Cron/Flush.php @@ -4,21 +4,21 @@ namespace SamJUK\CacheDebounce\Cron; -use SamJUK\CacheDebounce\Model\Entries as CacheDebouncedEntries; +use SamJUK\CacheDebounce\Model\StaggeredFlush; class Flush { - /** @var CacheDebouncedEntries $cacheDebouncedEntries */ - private $cacheDebouncedEntries; + /** @var StaggeredFlush $staggeredFlush */ + private $staggeredFlush; public function __construct( - CacheDebouncedEntries $cacheDebouncedEntries + StaggeredFlush $staggeredFlush ) { - $this->cacheDebouncedEntries = $cacheDebouncedEntries; + $this->staggeredFlush = $staggeredFlush; } public function execute() : void { - $this->cacheDebouncedEntries->flush(); + $this->staggeredFlush->execute(); } } diff --git a/Model/Config.php b/Model/Config.php index 6deb9f7..72fa082 100644 --- a/Model/Config.php +++ b/Model/Config.php @@ -9,6 +9,12 @@ class Config { private const XML_PATH_GENERAL_ENABLED = 'samjuk_cache_debounce/general/enabled'; + private const XML_PATH_STAGGER_ENABLED = 'samjuk_cache_debounce/stagger/enabled'; + private const XML_PATH_STAGGER_BATCH_SIZE = 'samjuk_cache_debounce/stagger/batch_size'; + private const XML_PATH_STAGGER_INTERVAL_MS = 'samjuk_cache_debounce/stagger/interval_ms'; + private const XML_PATH_STAGGER_MAX_RUNTIME_SECONDS = 'samjuk_cache_debounce/stagger/max_runtime_seconds'; + private const XML_PATH_STAGGER_LAG_RATIO_THRESHOLD = 'samjuk_cache_debounce/stagger/lag_ratio_threshold'; + private const XML_PATH_STAGGER_LAG_ALERT_AFTER_RUNS = 'samjuk_cache_debounce/stagger/lag_alert_after_runs'; /** @var bool $shouldDebouncePurgeRequest */ private $shouldDebouncePurgeRequest = true; @@ -46,6 +52,54 @@ public function setShouldDebouncePurgeRequest(bool $state) : void $this->shouldDebouncePurgeRequest = $state; } + /** + * Check if staggered/batched purge release is enabled + */ + public function isStaggerEnabled() : bool + { + return $this->getFlag(self::XML_PATH_STAGGER_ENABLED); + } + + /** + * Tags per staggered purge chunk + */ + public function getStaggerBatchSize() : int + { + return (int)$this->scopeConfig->getValue(self::XML_PATH_STAGGER_BATCH_SIZE); + } + + /** + * Pause, in milliseconds, between staggered purge chunks + */ + public function getStaggerIntervalMs() : int + { + return (int)$this->scopeConfig->getValue(self::XML_PATH_STAGGER_INTERVAL_MS); + } + + /** + * Safety budget, in seconds, per staggered release invocation + */ + public function getStaggerMaxRuntimeSeconds() : int + { + return (int)$this->scopeConfig->getValue(self::XML_PATH_STAGGER_MAX_RUNTIME_SECONDS); + } + + /** + * Ratio of arrived-during-drain to just-drained tags that counts as lagging + */ + public function getStaggerLagRatioThreshold() : float + { + return (float)$this->scopeConfig->getValue(self::XML_PATH_STAGGER_LAG_RATIO_THRESHOLD); + } + + /** + * Consecutive lagging runs before an admin notification is raised + */ + public function getStaggerLagAlertAfterRuns() : int + { + return (int)$this->scopeConfig->getValue(self::XML_PATH_STAGGER_LAG_ALERT_AFTER_RUNS); + } + /** * Fetch a system config flag */ diff --git a/Model/Config/Source/StorageDriver.php b/Model/Config/Source/StorageDriver.php new file mode 100644 index 0000000..dcc37b2 --- /dev/null +++ b/Model/Config/Source/StorageDriver.php @@ -0,0 +1,21 @@ + 'db', 'label' => __('Database (default)')], + ['value' => 'redis', 'label' => __('Redis')], + ]; + } +} diff --git a/Model/Entries.php b/Model/Entries.php index 7e37bb5..3cb1d2e 100644 --- a/Model/Entries.php +++ b/Model/Entries.php @@ -5,17 +5,18 @@ namespace SamJUK\CacheDebounce\Model; use SamJUK\CacheDebounce\Model\Config; +use SamJUK\CacheDebounce\Model\Storage\QueueStorageInterface; use Magento\CacheInvalidate\Model\PurgeCache; -use Magento\Framework\App\ResourceConnection; +use Magento\Framework\FlagManager; use Psr\Log\LoggerInterface; class Entries { - /** @var string $tableName */ - private $tableName; + public const FLAG_LAST_FLUSH_AT = 'samjuk_cache_debounce_last_flush_at'; + public const FLAG_LAST_FLUSH_DURATION = 'samjuk_cache_debounce_last_flush_duration'; - /** @var ResourceConnection $resourceConnection */ - private $resourceConnection; + /** @var QueueStorageInterface $storage */ + private $storage; /** @var PurgeCache $purgeCache */ private $purgeCache; @@ -26,17 +27,21 @@ class Entries /** @var LoggerInterface $logger */ private $logger; + /** @var FlagManager $flagManager */ + private $flagManager; + public function __construct( Config $config, PurgeCache $purgeCache, - ResourceConnection $resourceConnection, - LoggerInterface $logger + QueueStorageInterface $storage, + LoggerInterface $logger, + FlagManager $flagManager ) { $this->config = $config; $this->purgeCache = $purgeCache; - $this->resourceConnection = $resourceConnection; + $this->storage = $storage; $this->logger = $logger; - $this->tableName = $resourceConnection->getTableName('samjuk_cache_debounce'); + $this->flagManager = $flagManager; } /** @@ -44,22 +49,7 @@ public function __construct( */ public function add(array $tags) : void { - if (!$tags) { - return; - } - - $this->resourceConnection->getConnection() - ->insertArray($this->tableName, ['tag'], $tags); - } - - /** - * Get all tags from the purge queue - */ - public function get() : array - { - $connection = $this->resourceConnection->getConnection(); - $query = $connection->select()->from($this->tableName, 'tag')->distinct(); - return $connection->fetchCol($query); + $this->storage->add($tags); } /** @@ -67,28 +57,45 @@ public function get() : array */ public function flush() : void { - $tags = $this->get(); - if (count($tags) === 0) { + // Resume an already-claimed batch before claiming new work. + $batchId = $this->storage->activeBatch(); + if ($batchId === '') { + $batchId = $this->storage->claim(); + } + if ($batchId === '') { $this->logger->debug("[CacheDebounce] Nothing to flush"); return; } + $tags = $this->storage->tags($batchId); $this->logger->debug("[CacheDebounce] Flushing Tags: " . json_encode($tags)); $this->config->setShouldDebouncePurgeRequest(false); + $start = microtime(true); try { $success = $this->purgeCache->sendPurgeRequest($tags); if (!$success) { $this->logger->error( - "[CacheDebounce] Purge request failed — leaving tags queued for retry: " . json_encode($tags) + "[CacheDebounce] Purge request failed — releasing batch $batchId for retry: " + . json_encode($tags) ); + $this->storage->release($batchId); return; } - $this->resourceConnection->getConnection()->delete($this->tableName); + $this->storage->clear($batchId); + } catch (\Throwable $e) { + $this->logger->error( + "[CacheDebounce] Purge request threw — releasing batch $batchId for retry: " . $e->getMessage() + ); + $this->storage->release($batchId); + throw $e; } finally { $this->config->setShouldDebouncePurgeRequest(true); } + + $this->flagManager->saveFlag(self::FLAG_LAST_FLUSH_AT, date('Y-m-d H:i:s')); + $this->flagManager->saveFlag(self::FLAG_LAST_FLUSH_DURATION, round(microtime(true) - $start, 1)); } } diff --git a/Model/LagDetector.php b/Model/LagDetector.php new file mode 100644 index 0000000..94b3f72 --- /dev/null +++ b/Model/LagDetector.php @@ -0,0 +1,71 @@ +config = $config; + $this->flagManager = $flagManager; + $this->notifier = $notifier; + $this->logger = $logger; + } + + /** + * Tracks consecutive lagging runs, re-notifying every N runs while it lasts. + */ + public function recordSample(int $claimed, int $arrivedDuring): void + { + $isLagging = $claimed > 0 + && ($arrivedDuring / $claimed) >= $this->config->getStaggerLagRatioThreshold(); + + if (!$isLagging) { + $this->flagManager->deleteFlag(self::FLAG_CONSECUTIVE_LAGGING_RUNS); + return; + } + + $consecutiveRuns = (int)$this->flagManager->getFlagData(self::FLAG_CONSECUTIVE_LAGGING_RUNS) + 1; + $this->flagManager->saveFlag(self::FLAG_CONSECUTIVE_LAGGING_RUNS, $consecutiveRuns); + + $alertAfterRuns = $this->config->getStaggerLagAlertAfterRuns(); + if ($alertAfterRuns > 0 && $consecutiveRuns % $alertAfterRuns === 0) { + $message = sprintf( + 'Purge release is lagging behind ingestion: %d tags arrived while %d were being drained ' + . '(%d consecutive lagging runs).', + $arrivedDuring, + $claimed, + $consecutiveRuns + ); + + $this->logger->warning("[CacheDebounce] $message"); + + // NotifierInterface has no addWarning() — addMajor is the closest severity. + $this->notifier->addMajor('Cache Debounce: purge release is lagging', $message); + } + } +} diff --git a/Model/StaggeredFlush.php b/Model/StaggeredFlush.php new file mode 100644 index 0000000..77ccda3 --- /dev/null +++ b/Model/StaggeredFlush.php @@ -0,0 +1,143 @@ +config = $config; + $this->storage = $storage; + $this->purgeCache = $purgeCache; + $this->lockManager = $lockManager; + $this->lagDetector = $lagDetector; + $this->flagManager = $flagManager; + $this->entries = $entries; + $this->logger = $logger; + } + + /** + * Releases a batch paced by the `stagger` config; single-shot when disabled. + */ + public function execute(): void + { + if (!$this->config->isStaggerEnabled()) { + $this->entries->flush(); + return; + } + + if (!$this->lockManager->lock(self::LOCK_NAME, self::LOCK_TIMEOUT_FAIL_FAST)) { + $this->logger->debug('[CacheDebounce] Staggered release already running, skipping.'); + return; + } + + try { + $this->release(); + } finally { + $this->lockManager->unlock(self::LOCK_NAME); + } + } + + private function release(): void + { + $wasResumed = true; + $batchId = $this->storage->activeBatch(); + if ($batchId === '') { + $wasResumed = false; + $batchId = $this->storage->claim(); + if ($batchId === '') { + $this->logger->debug('[CacheDebounce] Nothing to flush.'); + return; + } + } + + // Always re-purges the full batch, even on resume — a BAN is idempotent. + $tags = $this->storage->tags($batchId); + $claimedCount = count($tags); + $chunks = array_chunk($tags, max(1, $this->config->getStaggerBatchSize())); + $lastChunkIndex = count($chunks) - 1; + $intervalMicroseconds = max(0, $this->config->getStaggerIntervalMs()) * 1000; + $maxRuntimeSeconds = $this->config->getStaggerMaxRuntimeSeconds(); + $start = microtime(true); + + $this->config->setShouldDebouncePurgeRequest(false); + try { + foreach ($chunks as $index => $chunk) { + // Budget only checked from chunk 2 on, so a bad config can't block all progress. + if ($index > 0 && (microtime(true) - $start) >= $maxRuntimeSeconds) { + $this->logger->debug( + "[CacheDebounce] Staggered release hit its runtime budget; " + . "batch $batchId stays claimed for the next run." + ); + return; + } + + if (!$this->purgeCache->sendPurgeRequest($chunk)) { + $this->logger->error( + "[CacheDebounce] Staggered purge chunk failed — leaving batch $batchId queued for retry." + ); + return; + } + + if ($index !== $lastChunkIndex) { + usleep($intervalMicroseconds); + } + } + } finally { + $this->config->setShouldDebouncePurgeRequest(true); + } + + $this->storage->clear($batchId); + + $this->flagManager->saveFlag(Entries::FLAG_LAST_FLUSH_AT, date('Y-m-d H:i:s')); + $this->flagManager->saveFlag(Entries::FLAG_LAST_FLUSH_DURATION, round(microtime(true) - $start, 1)); + + // Skip on resume — pendingCount() would include pre-run backlog too. + if (!$wasResumed) { + $this->lagDetector->recordSample($claimedCount, $this->storage->pendingCount()); + } + } +} diff --git a/Model/Storage/Database.php b/Model/Storage/Database.php new file mode 100644 index 0000000..5dbb005 --- /dev/null +++ b/Model/Storage/Database.php @@ -0,0 +1,154 @@ +resourceConnection = $resourceConnection; + $this->tableName = $resourceConnection->getTableName('samjuk_cache_debounce'); + } + + /** + * @inheritDoc + * + * Dedupes against pending tags at the app level; no PK for this yet. + */ + public function add(array $tags): void + { + $tags = array_values(array_unique(array_filter($tags, fn ($tag) => $tag !== null && $tag !== ''))); + if (!$tags) { + return; + } + + $newTags = array_diff($tags, $this->tags(self::UNCLAIMED_BATCH_ID)); + if (!$newTags) { + return; + } + + $rows = []; + foreach ($newTags as $tag) { + $rows[] = [self::UNCLAIMED_BATCH_ID, $tag]; + } + + $this->resourceConnection->getConnection()->insertArray( + $this->tableName, + ['batch_id', 'tag'], + $rows, + AdapterInterface::INSERT_IGNORE + ); + } + + /** + * @inheritDoc + */ + public function claim(): string + { + $batchId = bin2hex(random_bytes(16)); + + $affectedRows = $this->resourceConnection->getConnection()->update( + $this->tableName, + ['batch_id' => $batchId], + $this->quoteBatchId(self::UNCLAIMED_BATCH_ID) + ); + + return $affectedRows > 0 ? $batchId : ''; + } + + /** + * @inheritDoc + */ + public function tags(string $batchId): array + { + $connection = $this->resourceConnection->getConnection(); + $query = $connection->select() + ->from($this->tableName, 'tag') + ->where('batch_id = ?', $batchId); + + return $connection->fetchCol($query); + } + + /** + * @inheritDoc + */ + public function clear(string $batchId): void + { + $this->resourceConnection->getConnection()->delete( + $this->tableName, + $this->quoteBatchId($batchId) + ); + } + + /** + * @inheritDoc + */ + public function release(string $batchId): void + { + $this->add($this->tags($batchId)); + $this->clear($batchId); + } + + private function quoteBatchId(string $batchId): string + { + return $this->resourceConnection->getConnection()->quoteInto('batch_id = ?', $batchId); + } + + /** + * @inheritDoc + */ + public function pendingCount(): int + { + $connection = $this->resourceConnection->getConnection(); + $query = $connection->select() + ->from($this->tableName, ['count' => 'COUNT(*)']) + ->where('batch_id = ?', self::UNCLAIMED_BATCH_ID); + + return (int)$connection->fetchOne($query); + } + + /** + * @inheritDoc + */ + public function activeBatch(): string + { + $connection = $this->resourceConnection->getConnection(); + $query = $connection->select() + ->from($this->tableName, ['batch_id']) + ->where('batch_id != ?', self::UNCLAIMED_BATCH_ID) + ->limit(1); + + $batchId = $connection->fetchOne($query); + + return $batchId !== false ? (string)$batchId : ''; + } + + /** + * @inheritDoc + */ + public function oldestPendingAgeSeconds(): ?int + { + $connection = $this->resourceConnection->getConnection(); + $query = $connection->select() + ->from($this->tableName, ['age' => 'TIMESTAMPDIFF(SECOND, MIN(created_at), NOW())']) + ->where('batch_id = ?', self::UNCLAIMED_BATCH_ID); + + $age = $connection->fetchOne($query); + + return ($age === null || $age === false) ? null : max(0, (int)$age); + } +} diff --git a/Model/Storage/Pool.php b/Model/Storage/Pool.php new file mode 100644 index 0000000..1bba76b --- /dev/null +++ b/Model/Storage/Pool.php @@ -0,0 +1,111 @@ +objectManager = $objectManager; + $this->scopeConfig = $scopeConfig; + $this->drivers = $drivers; + } + + /** + * @inheritDoc + */ + public function add(array $tags): void + { + $this->driver()->add($tags); + } + + /** + * @inheritDoc + */ + public function claim(): string + { + return $this->driver()->claim(); + } + + /** + * @inheritDoc + */ + public function tags(string $batchId): array + { + return $this->driver()->tags($batchId); + } + + /** + * @inheritDoc + */ + public function clear(string $batchId): void + { + $this->driver()->clear($batchId); + } + + /** + * @inheritDoc + */ + public function release(string $batchId): void + { + $this->driver()->release($batchId); + } + + /** + * @inheritDoc + */ + public function pendingCount(): int + { + return $this->driver()->pendingCount(); + } + + /** + * @inheritDoc + */ + public function activeBatch(): string + { + return $this->driver()->activeBatch(); + } + + /** + * @inheritDoc + */ + public function oldestPendingAgeSeconds(): ?int + { + return $this->driver()->oldestPendingAgeSeconds(); + } + + private function driver(): QueueStorageInterface + { + if ($this->resolved === null) { + $key = (string)$this->scopeConfig->getValue(self::XML_PATH_STORAGE_DRIVER) ?: self::DEFAULT_DRIVER; + $class = $this->drivers[$key] ?? $this->drivers[self::DEFAULT_DRIVER]; + $this->resolved = $this->objectManager->create($class); + } + + return $this->resolved; + } +} diff --git a/Model/Storage/QueueStorageInterface.php b/Model/Storage/QueueStorageInterface.php new file mode 100644 index 0000000..cd7fa5a --- /dev/null +++ b/Model/Storage/QueueStorageInterface.php @@ -0,0 +1,49 @@ +connectionResolver = $connectionResolver; + } + + /** + * @inheritDoc + */ + public function add(array $tags): void + { + if (!$tags) { + return; + } + + $this->connectionResolver->getClient()->sAdd(self::KEY_LIVE, ...$tags); + } + + /** + * @inheritDoc + */ + public function claim(): string + { + $batchId = bin2hex(random_bytes(16)); + $client = $this->connectionResolver->getClient(); + + try { + $client->rename(self::KEY_LIVE, self::KEY_BATCH_PREFIX . $batchId); + } catch (\CredisException $e) { + if (stripos($e->getMessage(), 'no such key') !== false) { + return ''; + } + throw $e; + } + + $client->set(self::KEY_ACTIVE_BATCH, $batchId); + + return $batchId; + } + + /** + * @inheritDoc + */ + public function tags(string $batchId): array + { + return $this->connectionResolver->getClient()->sMembers(self::KEY_BATCH_PREFIX . $batchId); + } + + /** + * @inheritDoc + */ + public function clear(string $batchId): void + { + $client = $this->connectionResolver->getClient(); + $client->del(self::KEY_BATCH_PREFIX . $batchId); + $client->del(self::KEY_ACTIVE_BATCH); + } + + /** + * @inheritDoc + */ + public function release(string $batchId): void + { + $client = $this->connectionResolver->getClient(); + $client->sUnionStore(self::KEY_LIVE, self::KEY_LIVE, self::KEY_BATCH_PREFIX . $batchId); + $client->del(self::KEY_BATCH_PREFIX . $batchId); + $client->del(self::KEY_ACTIVE_BATCH); + } + + /** + * @inheritDoc + */ + public function pendingCount(): int + { + return (int)$this->connectionResolver->getClient()->sCard(self::KEY_LIVE); + } + + /** + * @inheritDoc + */ + public function activeBatch(): string + { + $batchId = $this->connectionResolver->getClient()->get(self::KEY_ACTIVE_BATCH); + + return $batchId !== false && $batchId !== null ? (string)$batchId : ''; + } + + /** + * @inheritDoc + * + * A Redis Set has no per-member insertion time, so this driver can't + * answer this — always null. + */ + public function oldestPendingAgeSeconds(): ?int + { + return null; + } +} diff --git a/Model/Storage/Redis/ConnectionResolver.php b/Model/Storage/Redis/ConnectionResolver.php new file mode 100644 index 0000000..15e6520 --- /dev/null +++ b/Model/Storage/Redis/ConnectionResolver.php @@ -0,0 +1,88 @@ +deploymentConfig = $deploymentConfig; + } + + /** + * @throws LocalizedException when no usable connection can be resolved. + */ + public function getClient(): \Credis_Client + { + if ($this->client !== null) { + return $this->client; + } + + $dedicated = $this->deploymentConfig->get(self::CONFIG_PATH_DEDICATED); + $config = $dedicated ? $this->normalize($dedicated) : $this->resolveFallback(); + + if (!$config) { + throw new LocalizedException(__( + 'SamJUK_CacheDebounce: no Redis connection could be resolved. Configure the ' + . '"cache_debounce.redis" block in app/etc/env.php, or configure a Redis backend ' + . 'for the "page_cache" cache frontend.' + )); + } + + $this->client = new \Credis_Client( + $config['host'], + (int)$config['port'], + null, + '', + (int)$config['database'], + $config['password'] ?: null + ); + + return $this->client; + } + + private function resolveFallback(): ?array + { + $backendOptions = $this->deploymentConfig->get(self::CONFIG_PATH_SHARED_FALLBACK); + + if (!$backendOptions) { + return null; + } + + return [ + 'host' => $backendOptions['server'] ?? '127.0.0.1', + 'port' => $backendOptions['port'] ?? 6379, + 'password' => $backendOptions['password'] ?? '', + 'database' => $backendOptions['database'] ?? 0, + ]; + } + + /** + * Fill in defaults for whichever of host/port/database/password the + * "cache_debounce.redis" env.php block leaves out. + */ + private function normalize(array $config): array + { + return [ + 'host' => $config['host'] ?? '127.0.0.1', + 'port' => $config['port'] ?? 6379, + 'password' => $config['password'] ?? '', + 'database' => $config['database'] ?? 0, + ]; + } +} diff --git a/README.md b/README.md index 35ecd2c..660a2dc 100644 --- a/README.md +++ b/README.md @@ -38,6 +38,66 @@ Option | Config Path | Default | Description --- | --- | --- | --- Enabled | `samjuk_cache_debounce/general/enabled` | `0` | Feature flag to toggle functionality of the module Flush Schedule | `samjuk_cache_debounce/cron/flush_schedule` | `*/5 0 0 0 0` | Cron schedule to run the scheduled flush +Stagger Enabled | `samjuk_cache_debounce/stagger/enabled` | `0` | Release a claimed batch gradually instead of purging it all at once — see below +Stagger Batch Size | `samjuk_cache_debounce/stagger/batch_size` | `50` | Tags per purge chunk +Stagger Interval (ms) | `samjuk_cache_debounce/stagger/interval_ms` | `1000` | Pause between chunks +Stagger Max Runtime (s) | `samjuk_cache_debounce/stagger/max_runtime_seconds` | `240` | Safety budget per cron/CLI invocation +Stagger Lag Ratio Threshold | `samjuk_cache_debounce/stagger/lag_ratio_threshold` | `1.0` | A run counts as "lagging" if tags arriving during the drain ≥ this ratio of tags just drained +Stagger Lag Alert After Runs | `samjuk_cache_debounce/stagger/lag_alert_after_runs` | `3` | Consecutive lagging runs before an admin notification is raised +Storage Driver | `samjuk_cache_debounce_advanced/general/storage_driver` | `db` | Queue storage backend — `db` or `redis`. A deploy-time infra decision, not a store setting — see below + +This one is a deliberate exception to "Configuration can be handled via System configuration" above: it's hidden from Stores > Configuration entirely (`showInDefault`/`showInWebsite`/`showInStore` all `0`) and gated behind its own ACL resource (`SamJUK_CacheDebounce::storage_driver`) that isn't granted to any role by default. It exists only so `config:set` has a registered path to write to. Set it with: +```sh +php bin/magento config:set samjuk_cache_debounce_advanced/general/storage_driver redis +``` +Add `-e` to lock it into `app/etc/env.php` instead of `core_config_data`, removing Admin overridability entirely. + +**Only switch drivers when the queue is empty** (`samjuk:cache-debounce:status` shows zero pending and no active batch). Each driver only sees its own storage — switching mid-flight leaves whatever was queued in the old backend stranded there, invisible to the new one, until something re-queues the same tags. + +### Redis storage driver + +The `redis` driver connects via a dedicated `app/etc/env.php` block: +```php +'cache_debounce' => [ + 'redis' => [ + 'host' => '127.0.0.1', + 'port' => '6379', + 'password' => '', + 'database' => '2', + ], +], +``` + +If that block is absent, the module falls back to whatever Redis backend is already configured for the `page_cache` cache frontend, so it works zero-config once the driver is switched on. + +> **Note:** sharing the page cache's Redis instance means pending purge tags are subject to that instance's `maxmemory-policy`. If it's `allkeys-lru`/`allkeys-lfu` (common for FPC), tags can be silently evicted under memory pressure. Configure a dedicated block above to avoid this. + +### Staggered / batched purge release + +On a large debounce window, a single `flush()` can send hundreds of tags to Varnish in one instant — a CPU/IO spike exactly when the store is under load. Enabling `stagger` releases a claimed batch gradually instead: chunked into `batch_size`-sized purge requests, paced by `interval_ms`, within a `max_runtime_seconds` budget per cron/CLI invocation. Both the cron job and `samjuk:cache-debounce:flush` route through this — a single-shot flush when disabled (the pre-existing behavior), a paced release when enabled. + +An interrupted run (deploy, OOM-kill, timeout) leaves its batch claimed but not cleared; the next invocation resumes it — by re-purging its full tag list, not from a persisted chunk offset. A Varnish BAN is idempotent, so the cost of that is, at worst, a handful of redundant purges. + +Only one release runs at a time, guarded by `Magento\Framework\Lock\LockManagerInterface`. If release throughput can't keep up with ingestion, it's logged as a warning, surfaced in the admin bell-icon notifications after `lag_alert_after_runs` consecutive lagging runs, and re-raised every `lag_alert_after_runs` runs for as long as the condition persists. + +Check current state with: +```sh +php bin/magento samjuk:cache-debounce:status +``` + +### Adding your own storage driver + +Drivers are resolved from an array bound in `etc/di.xml`, keyed by the value of `storage_driver`. Any module can add an entry to extend it, without touching this module's `di.xml`: +```xml + + + + Vendor\Module\Model\Storage\MyDriver + + + +``` +`MyDriver` just needs to implement `QueueStorageInterface`. Then `bin/magento config:set samjuk_cache_debounce_advanced/general/storage_driver my_driver`. ## Will this help my store? diff --git a/Setup/Patch/Data/ClearQueueTable.php b/Setup/Patch/Data/ClearQueueTable.php new file mode 100644 index 0000000..f7b7bac --- /dev/null +++ b/Setup/Patch/Data/ClearQueueTable.php @@ -0,0 +1,39 @@ +moduleDataSetup = $moduleDataSetup; + } + + public function apply(): void + { + // DELETE not TRUNCATE: data patches run inside a transaction. + $connection = $this->moduleDataSetup->getConnection(); + $connection->delete($this->moduleDataSetup->getTable('samjuk_cache_debounce')); + } + + public static function getDependencies(): array + { + return []; + } + + public function getAliases(): array + { + return []; + } +} diff --git a/Test/Integration/Model/EntriesTest.php b/Test/Integration/Model/EntriesTest.php index f6b54d2..c607a40 100644 --- a/Test/Integration/Model/EntriesTest.php +++ b/Test/Integration/Model/EntriesTest.php @@ -3,7 +3,10 @@ namespace SamJUK\CacheDebounce\Test\Integration\Model; use PHPUnit\Framework\TestCase; +use Magento\Framework\App\ResourceConnection; use Magento\TestFramework\ObjectManager; +use SamJUK\CacheDebounce\Model\Entries; +use SamJUK\CacheDebounce\Model\Storage\QueueStorageInterface; class EntriesTest extends TestCase { @@ -11,19 +14,74 @@ class EntriesTest extends TestCase protected $objectManager; private $cacheDebouncedEntries; + private $storage; + private $resourceConnection; protected function setUp(): void { $this->objectManager = ObjectManager::getInstance(); - $this->cacheDebouncedEntries = $this->objectManager->get(\SamJUK\CacheDebounce\Model\Entries::class); + $this->cacheDebouncedEntries = $this->objectManager->get(Entries::class); + $this->storage = $this->objectManager->get(QueueStorageInterface::class); + + // Independent of the classes under test, so a prior test failing + // mid-assertion can never leak rows into the next test. + $this->resourceConnection = $this->objectManager->get(ResourceConnection::class); + $this->truncateTable(); + } + + protected function tearDown(): void + { + $this->truncateTable(); + } + + private function truncateTable(): void + { + $connection = $this->resourceConnection->getConnection(); + $connection->delete($this->resourceConnection->getTableName('samjuk_cache_debounce')); } public function testDebouncedEntriesStorage() { $this->cacheDebouncedEntries->add(self::CACHE_TAGS); - $this->assertEquals(self::CACHE_TAGS, $this->cacheDebouncedEntries->get()); - $this->cacheDebouncedEntries->flush(); - $this->assertEquals([], $this->cacheDebouncedEntries->get()); + $batchId = $this->storage->claim(); + $this->assertEquals(self::CACHE_TAGS, $this->storage->tags($batchId)); + + $this->storage->clear($batchId); + $this->assertEquals([], $this->storage->tags($batchId)); + } + + /** + * Regression test for the add-during-purge race: a tag queued after a + * batch has been claimed must survive into the *next* batch, not be + * silently dropped by the trailing clear() of the batch currently + * being purged. + */ + public function testTagAddedWhileBatchIsClaimedSurvivesIntoNextBatch() + { + $this->cacheDebouncedEntries->add(['cat_c_1']); + + $batchId = $this->storage->claim(); + + $this->cacheDebouncedEntries->add(['cat_c_2']); + + $this->assertEquals(['cat_c_1'], $this->storage->tags($batchId)); + + $this->storage->clear($batchId); + + $nextBatchId = $this->storage->claim(); + $this->assertEquals(['cat_c_2'], $this->storage->tags($nextBatchId)); + } + + public function testSecondConcurrentClaimReturnsEmptyString() + { + $this->cacheDebouncedEntries->add(self::CACHE_TAGS); + + $batchId = $this->storage->claim(); + $this->assertNotSame('', $batchId); + + $this->assertSame('', $this->storage->claim()); + + $this->storage->clear($batchId); } } diff --git a/Test/Integration/Model/StaggeredFlushTest.php b/Test/Integration/Model/StaggeredFlushTest.php new file mode 100644 index 0000000..edaf5b5 --- /dev/null +++ b/Test/Integration/Model/StaggeredFlushTest.php @@ -0,0 +1,123 @@ +objectManager = ObjectManager::getInstance(); + $this->storage = $this->objectManager->get(QueueStorageInterface::class); + $this->entries = $this->objectManager->get(Entries::class); + + // Independent of the classes under test, so a prior test failing + // mid-assertion can never leak rows into the next test. + $this->resourceConnection = $this->objectManager->get(ResourceConnection::class); + $this->truncateTable(); + } + + protected function tearDown(): void + { + $this->truncateTable(); + $this->objectManager->removeSharedInstance(PurgeCache::class); + } + + private function truncateTable(): void + { + $connection = $this->resourceConnection->getConnection(); + $connection->delete($this->resourceConnection->getTableName('samjuk_cache_debounce')); + } + + /** + * Double must be shared-instanced before create() resolves it. + */ + private function staggeredFlushWithPurgeCache(PurgeCache $purgeCache): StaggeredFlush + { + $this->objectManager->addSharedInstance($purgeCache, PurgeCache::class); + + return $this->objectManager->create(StaggeredFlush::class); + } + + public function testInterruptedRunLeavesBatchClaimedAndTheNextRunResumesAndDrainsIt() + { + $this->entries->add(['cat_c_1', 'cat_c_2', 'cat_c_3']); + + $calls = 0; + $crashingPurgeCache = $this->createMock(PurgeCache::class); + $crashingPurgeCache->method('sendPurgeRequest')->willReturnCallback(function ($tags) use (&$calls) { + $calls++; + if ($calls > 2) { + throw new \RuntimeException('simulated crash mid-loop'); + } + return true; + }); + + try { + $this->staggeredFlushWithPurgeCache($crashingPurgeCache)->execute(); + $this->fail('Expected the simulated crash to propagate.'); + } catch (\RuntimeException $e) { + $this->assertSame('simulated crash mid-loop', $e->getMessage()); + } + + $batchId = $this->storage->activeBatch(); + $this->assertNotSame('', $batchId, 'An interrupted batch must stay claimed for the next run.'); + + $healthyPurgeCache = $this->createMock(PurgeCache::class); + $healthyPurgeCache->method('sendPurgeRequest')->willReturn(true); + + $this->staggeredFlushWithPurgeCache($healthyPurgeCache)->execute(); + + $this->assertSame('', $this->storage->activeBatch()); + $this->assertSame(0, $this->storage->pendingCount()); + } + + public function testConcurrentInvocationsOnlyOneProceedsPastTheLock() + { + $this->entries->add(['cat_c_1']); + + $purgeCache = $this->createMock(PurgeCache::class); + $purgeCache->method('sendPurgeRequest')->willReturn(true); + $staggeredFlush = $this->staggeredFlushWithPurgeCache($purgeCache); + + $lockManager = $this->objectManager->get(LockManagerInterface::class); + $this->assertTrue($lockManager->lock(self::LOCK_NAME, 0)); + + try { + $staggeredFlush->execute(); + + // Lock held by us — the concurrent invocation must have backed + // off without touching the queue at all. + $this->assertSame(1, $this->storage->pendingCount()); + $this->assertSame('', $this->storage->activeBatch()); + } finally { + $lockManager->unlock(self::LOCK_NAME); + } + + // Lock released — the next invocation proceeds and drains normally. + $staggeredFlush->execute(); + $this->assertSame(0, $this->storage->pendingCount()); + $this->assertSame('', $this->storage->activeBatch()); + } +} diff --git a/Test/Integration/Model/Storage/DatabaseTest.php b/Test/Integration/Model/Storage/DatabaseTest.php new file mode 100644 index 0000000..bd025b1 --- /dev/null +++ b/Test/Integration/Model/Storage/DatabaseTest.php @@ -0,0 +1,156 @@ +objectManager = ObjectManager::getInstance(); + $this->storage = $this->objectManager->create(Database::class); + + // Independent of the class under test, so a prior test failing + // mid-assertion can never leak rows into the next test. + $this->resourceConnection = $this->objectManager->get(ResourceConnection::class); + $this->truncateTable(); + } + + protected function tearDown(): void + { + $this->truncateTable(); + } + + private function truncateTable(): void + { + $connection = $this->resourceConnection->getConnection(); + $connection->delete($this->resourceConnection->getTableName('samjuk_cache_debounce')); + } + + public function testAddClaimTagsClearRoundTrip() + { + $this->storage->add(['cat_c_1', 'cat_c_2']); + + $batchId = $this->storage->claim(); + $this->assertNotSame('', $batchId); + + $this->assertEquals(['cat_c_1', 'cat_c_2'], $this->storage->tags($batchId)); + + $this->storage->clear($batchId); + $this->assertEquals([], $this->storage->tags($batchId)); + } + + public function testAddDuringClaimIsNotLostOrDuplicated() + { + $this->storage->add(['cat_c_1']); + + $batchId = $this->storage->claim(); + + // Overlapping tag added after the claim snapshot — must land in the + // next batch, not be silently merged into the one already claimed. + $this->storage->add(['cat_c_1', 'cat_c_2']); + + $this->assertEquals(['cat_c_1'], $this->storage->tags($batchId)); + + $this->storage->clear($batchId); + + $nextBatchId = $this->storage->claim(); + $this->assertEquals(['cat_c_1', 'cat_c_2'], $this->storage->tags($nextBatchId)); + } + + public function testAddDoesNotDuplicateAlreadyPendingTags() + { + $this->storage->add(['cat_c_1']); + $this->storage->add(['cat_c_1', 'cat_c_2']); + + $batchId = $this->storage->claim(); + + $this->assertEquals(['cat_c_1', 'cat_c_2'], $this->storage->tags($batchId)); + } + + public function testAddIgnoresNullAndEmptyTags() + { + $this->storage->add(['cat_c_1', null, '']); + + $batchId = $this->storage->claim(); + + $this->assertEquals(['cat_c_1'], $this->storage->tags($batchId)); + } + + public function testSecondConcurrentClaimReturnsEmptyString() + { + $this->storage->add(['cat_c_1']); + + $firstBatchId = $this->storage->claim(); + $this->assertNotSame('', $firstBatchId); + + $this->assertSame('', $this->storage->claim()); + + $this->storage->clear($firstBatchId); + } + + public function testReleaseMakesTheBatchClaimableAgain() + { + $this->storage->add(['cat_c_1', 'cat_c_2']); + $batchId = $this->storage->claim(); + + $this->storage->release($batchId); + + $this->assertEquals([], $this->storage->tags($batchId)); + + $nextBatchId = $this->storage->claim(); + $this->assertEquals(['cat_c_1', 'cat_c_2'], $this->storage->tags($nextBatchId)); + } + + public function testReleaseMergesCleanlyWithTagsReQueuedWhileClaimed() + { + $this->storage->add(['cat_c_1']); + $batchId = $this->storage->claim(); + + // Same tag re-queued while the original batch is still in flight — + // release() must not end up with two pending rows for it. + $this->storage->add(['cat_c_1']); + + $this->storage->release($batchId); + + $nextBatchId = $this->storage->claim(); + $this->assertEquals(['cat_c_1'], $this->storage->tags($nextBatchId)); + } + + public function testPendingCountAndActiveBatchReflectClaimState() + { + $this->assertSame(0, $this->storage->pendingCount()); + $this->assertSame('', $this->storage->activeBatch()); + + $this->storage->add(['cat_c_1', 'cat_c_2']); + $this->assertSame(2, $this->storage->pendingCount()); + $this->assertSame('', $this->storage->activeBatch()); + + $batchId = $this->storage->claim(); + $this->assertSame(0, $this->storage->pendingCount()); + $this->assertSame($batchId, $this->storage->activeBatch()); + + $this->storage->clear($batchId); + $this->assertSame('', $this->storage->activeBatch()); + } + + public function testOldestPendingAgeSecondsReflectsWhenTagsWereQueued() + { + $this->assertNull($this->storage->oldestPendingAgeSeconds()); + + $this->storage->add(['cat_c_1']); + + $age = $this->storage->oldestPendingAgeSeconds(); + $this->assertIsInt($age); + $this->assertGreaterThanOrEqual(0, $age); + $this->assertLessThan(10, $age); + } +} diff --git a/Test/Integration/Model/Storage/Redis/ConnectionResolverTest.php b/Test/Integration/Model/Storage/Redis/ConnectionResolverTest.php new file mode 100644 index 0000000..9c8bd9e --- /dev/null +++ b/Test/Integration/Model/Storage/Redis/ConnectionResolverTest.php @@ -0,0 +1,69 @@ +get(DeploymentConfig\Reader::class); + + $dedicatedConfig = new DeploymentConfig($reader, ['cache_debounce' => ['redis' => [ + 'host' => '127.0.0.1', + 'port' => '6379', + 'password' => '', + 'database' => (string)self::DEDICATED_DATABASE, + ]]]); + $fallbackConfig = new DeploymentConfig($reader, ['cache' => ['frontend' => ['page_cache' => [ + 'backend_options' => [ + 'server' => '127.0.0.1', + 'port' => '6379', + 'password' => '', + 'database' => (string)self::FALLBACK_DATABASE, + ], + ]]]]); + + $this->dedicatedResolver = new ConnectionResolver($dedicatedConfig); + $this->fallbackResolver = new ConnectionResolver($fallbackConfig); + + $this->dedicatedResolver->getClient()->flushDb(); + $this->fallbackResolver->getClient()->flushDb(); + } + + protected function tearDown(): void + { + $this->dedicatedResolver->getClient()->flushDb(); + $this->fallbackResolver->getClient()->flushDb(); + } + + public function testResolvesTheDedicatedConnection() + { + $client = $this->dedicatedResolver->getClient(); + + $client->set(self::CANARY_KEY, '1'); + $this->assertSame('1', $client->get(self::CANARY_KEY)); + } + + public function testFallsBackToThePageCacheConnectionWhenNoDedicatedBlockIsConfigured() + { + $client = $this->fallbackResolver->getClient(); + + $client->set(self::CANARY_KEY, '1'); + $this->assertSame('1', $client->get(self::CANARY_KEY)); + } +} diff --git a/Test/Integration/Model/Storage/RedisTest.php b/Test/Integration/Model/Storage/RedisTest.php new file mode 100644 index 0000000..7449e1c --- /dev/null +++ b/Test/Integration/Model/Storage/RedisTest.php @@ -0,0 +1,143 @@ + '127.0.0.1', + 'port' => '6379', + 'password' => '', + 'database' => '2', + ]; + + private $storage; + private $rawClient; + + protected function setUp(): void + { + $reader = ObjectManager::getInstance()->get(DeploymentConfig\Reader::class); + $deploymentConfig = new DeploymentConfig($reader, ['cache_debounce' => ['redis' => self::CONNECTION_CONFIG]]); + + $this->storage = new Redis(new ConnectionResolver($deploymentConfig)); + + // Independent resolver instance, not the class under test. + $this->rawClient = (new ConnectionResolver($deploymentConfig))->getClient(); + $this->rawClient->flushDb(); + } + + protected function tearDown(): void + { + $this->rawClient->flushDb(); + } + + public function testAddClaimTagsClearRoundTrip() + { + $this->storage->add(['cat_c_1', 'cat_c_2']); + + $batchId = $this->storage->claim(); + $this->assertNotSame('', $batchId); + + $tags = $this->storage->tags($batchId); + sort($tags); + $this->assertEquals(['cat_c_1', 'cat_c_2'], $tags); + + $this->storage->clear($batchId); + $this->assertEquals([], $this->storage->tags($batchId)); + } + + public function testAddDuringClaimIsNotLostOrDuplicated() + { + $this->storage->add(['cat_c_1']); + + $batchId = $this->storage->claim(); + + $this->storage->add(['cat_c_1', 'cat_c_2']); + + $this->assertEquals(['cat_c_1'], $this->storage->tags($batchId)); + + $this->storage->clear($batchId); + + $nextBatchId = $this->storage->claim(); + $tags = $this->storage->tags($nextBatchId); + sort($tags); + $this->assertEquals(['cat_c_1', 'cat_c_2'], $tags); + + $this->storage->clear($nextBatchId); + } + + public function testSecondConcurrentClaimReturnsEmptyString() + { + $this->storage->add(['cat_c_1']); + + $firstBatchId = $this->storage->claim(); + $this->assertNotSame('', $firstBatchId); + + $this->assertSame('', $this->storage->claim()); + + $this->storage->clear($firstBatchId); + } + + public function testReleaseMakesTheBatchClaimableAgain() + { + $this->storage->add(['cat_c_1', 'cat_c_2']); + $batchId = $this->storage->claim(); + + $this->storage->release($batchId); + + $this->assertEquals([], $this->storage->tags($batchId)); + + $nextBatchId = $this->storage->claim(); + $tags = $this->storage->tags($nextBatchId); + sort($tags); + $this->assertEquals(['cat_c_1', 'cat_c_2'], $tags); + } + + public function testReleaseMergesCleanlyWithTagsReQueuedWhileClaimed() + { + $this->storage->add(['cat_c_1']); + $batchId = $this->storage->claim(); + + // Same tag re-queued while the original batch is still in flight — + // release() must not duplicate it; Redis sets dedupe natively. + $this->storage->add(['cat_c_1']); + + $this->storage->release($batchId); + + $nextBatchId = $this->storage->claim(); + $this->assertEquals(['cat_c_1'], $this->storage->tags($nextBatchId)); + } + + public function testPendingCountAndActiveBatchReflectClaimState() + { + $this->assertSame(0, $this->storage->pendingCount()); + $this->assertSame('', $this->storage->activeBatch()); + + $this->storage->add(['cat_c_1', 'cat_c_2']); + $this->assertSame(2, $this->storage->pendingCount()); + $this->assertSame('', $this->storage->activeBatch()); + + $batchId = $this->storage->claim(); + $this->assertSame(0, $this->storage->pendingCount()); + $this->assertSame($batchId, $this->storage->activeBatch()); + + $this->storage->clear($batchId); + $this->assertSame('', $this->storage->activeBatch()); + } + + public function testOldestPendingAgeSecondsIsAlwaysNull() + { + $this->storage->add(['cat_c_1']); + + $this->assertNull($this->storage->oldestPendingAgeSeconds()); + } +} diff --git a/Test/Unit/Model/Config/Source/StorageDriverTest.php b/Test/Unit/Model/Config/Source/StorageDriverTest.php new file mode 100644 index 0000000..9d65f05 --- /dev/null +++ b/Test/Unit/Model/Config/Source/StorageDriverTest.php @@ -0,0 +1,16 @@ +toOptionArray(), 'value'); + + $this->assertSame(['db', 'redis'], $values); + } +} diff --git a/Test/Unit/Model/ConfigTest.php b/Test/Unit/Model/ConfigTest.php index 2266d1b..264e729 100644 --- a/Test/Unit/Model/ConfigTest.php +++ b/Test/Unit/Model/ConfigTest.php @@ -37,4 +37,33 @@ public function testShouldDebouncePurgeRequestEnabledFlag() $config->setShouldDebouncePurgeRequest(false); $this->assertFalse($config->shouldDebouncePurgeRequest()); } + + public function testIsStaggerEnabled() + { + $scopeConfig = $this->createMock(ScopeConfigInterface::class); + $config = new CacheDebounceConfig($scopeConfig); + $this->assertFalse($config->isStaggerEnabled()); + + $scopeConfig->method('isSetFlag')->willReturn(1); + $this->assertTrue($config->isStaggerEnabled()); + } + + public function testStaggerNumericGetters() + { + $scopeConfig = $this->createMock(ScopeConfigInterface::class); + $scopeConfig->method('getValue')->willReturnMap([ + ['samjuk_cache_debounce/stagger/batch_size', 'default', null, '50'], + ['samjuk_cache_debounce/stagger/interval_ms', 'default', null, '1000'], + ['samjuk_cache_debounce/stagger/max_runtime_seconds', 'default', null, '240'], + ['samjuk_cache_debounce/stagger/lag_ratio_threshold', 'default', null, '1.5'], + ['samjuk_cache_debounce/stagger/lag_alert_after_runs', 'default', null, '3'], + ]); + $config = new CacheDebounceConfig($scopeConfig); + + $this->assertSame(50, $config->getStaggerBatchSize()); + $this->assertSame(1000, $config->getStaggerIntervalMs()); + $this->assertSame(240, $config->getStaggerMaxRuntimeSeconds()); + $this->assertSame(1.5, $config->getStaggerLagRatioThreshold()); + $this->assertSame(3, $config->getStaggerLagAlertAfterRuns()); + } } diff --git a/Test/Unit/Model/EntriesTest.php b/Test/Unit/Model/EntriesTest.php index 85a40f7..b4c0625 100644 --- a/Test/Unit/Model/EntriesTest.php +++ b/Test/Unit/Model/EntriesTest.php @@ -2,69 +2,119 @@ namespace SamJUK\CacheDebounce\Test\Unit\Model; +use Magento\CacheInvalidate\Model\PurgeCache; +use Magento\Framework\App\Config\ScopeConfigInterface; +use Magento\Framework\FlagManager; use PHPUnit\Framework\TestCase; +use Psr\Log\LoggerInterface; +use SamJUK\CacheDebounce\Model\Config as CacheDebounceConfig; use SamJUK\CacheDebounce\Model\Entries as CacheDebounceEntries; +use SamJUK\CacheDebounce\Model\Storage\QueueStorageInterface; class EntriesTest extends TestCase { + private const BATCH_ID = 'batch-123'; private const CACHE_TAGS = ['cat_c_1', 'cat_c_2', 'cat_c_p_1']; private $cacheDebounceConfig; private $purgeCacheModel; - private $connection; - private $resourceConnection; + private $storage; private $loggerInterface; + private $flagManager; private $cacheDebounceEntries; - private $select; protected function setUp(): void { - $this->select = $this->createMock(\Magento\Framework\DB\Select::class); - $this->select->method('from')->willReturn($this->select); - $this->select->method('distinct')->willReturn($this->select); - $this->connection = $this->createMock(\Magento\Framework\DB\Adapter\AdapterInterface::class); - $this->connection->method('select')->willReturn($this->select); - $this->cacheDebounceConfig = $this->createMock(\SamJUK\CacheDebounce\Model\Config::class); - $this->resourceConnection = $this->createMock(\Magento\Framework\App\ResourceConnection::class); - $this->resourceConnection->method('getConnection')->willReturn($this->connection); - $this->purgeCacheModel = $this->createMock(\Magento\CacheInvalidate\Model\PurgeCache::class); - $this->loggerInterface = $this->createMock(\Psr\Log\LoggerInterface::class); + $this->cacheDebounceConfig = $this->createMock(CacheDebounceConfig::class); + $this->storage = $this->createMock(QueueStorageInterface::class); + $this->purgeCacheModel = $this->createMock(PurgeCache::class); + $this->loggerInterface = $this->createMock(LoggerInterface::class); + $this->flagManager = $this->createMock(FlagManager::class); $this->cacheDebounceEntries = new CacheDebounceEntries( $this->cacheDebounceConfig, $this->purgeCacheModel, - $this->resourceConnection, - $this->loggerInterface + $this->storage, + $this->loggerInterface, + $this->flagManager ); } - public function testFlushTags() + public function testFlushDoesNothingWhenNothingIsClaimed() { - $this->connection->method('fetchCol')->willReturn([ - self::CACHE_TAGS - ]); + $this->storage->method('claim')->willReturn(''); + + $this->purgeCacheModel->expects($this->never())->method('sendPurgeRequest'); + + $this->cacheDebounceEntries->flush(); + } + + public function testFlushPurgesClaimedTagsAndClearsBatch() + { + $this->storage->method('claim')->willReturn(self::BATCH_ID); + $this->storage->method('tags')->with(self::BATCH_ID)->willReturn(self::CACHE_TAGS); $this->purgeCacheModel->expects($this->once()) ->method('sendPurgeRequest') - ->with([self::CACHE_TAGS]) + ->with(self::CACHE_TAGS) ->willReturn(true); - $this->connection->expects($this->once())->method('delete'); + $this->storage->expects($this->once())->method('clear')->with(self::BATCH_ID); + + $this->cacheDebounceEntries->flush(); + } + + public function testFlushResumesAnAlreadyClaimedBatchInsteadOfClaimingNew() + { + $this->storage->method('activeBatch')->willReturn(self::BATCH_ID); + $this->storage->method('tags')->with(self::BATCH_ID)->willReturn(self::CACHE_TAGS); + $this->storage->expects($this->never())->method('claim'); + + $this->purgeCacheModel->method('sendPurgeRequest')->willReturn(true); + + $this->storage->expects($this->once())->method('clear')->with(self::BATCH_ID); + + $this->cacheDebounceEntries->flush(); + } + + public function testFlushWritesLastFlushFlagsOnSuccess() + { + $this->storage->method('claim')->willReturn(self::BATCH_ID); + $this->storage->method('tags')->willReturn(self::CACHE_TAGS); + $this->purgeCacheModel->method('sendPurgeRequest')->willReturn(true); + + $this->flagManager->expects($this->exactly(2))->method('saveFlag')->with( + $this->logicalOr( + CacheDebounceEntries::FLAG_LAST_FLUSH_AT, + CacheDebounceEntries::FLAG_LAST_FLUSH_DURATION + ) + ); $this->cacheDebounceEntries->flush(); } - public function testFlushDoesNotClearQueueWhenPurgeRequestFails() + public function testFlushDoesNotWriteLastFlushFlagsWhenPurgeRequestFails() { - $this->connection->method('fetchCol')->willReturn([ - self::CACHE_TAGS - ]); + $this->storage->method('claim')->willReturn(self::BATCH_ID); + $this->storage->method('tags')->willReturn(self::CACHE_TAGS); + $this->purgeCacheModel->method('sendPurgeRequest')->willReturn(false); + + $this->flagManager->expects($this->never())->method('saveFlag'); + + $this->cacheDebounceEntries->flush(); + } + + public function testFlushReleasesBatchInsteadOfClearingWhenPurgeRequestFails() + { + $this->storage->method('claim')->willReturn(self::BATCH_ID); + $this->storage->method('tags')->willReturn(self::CACHE_TAGS); $this->purgeCacheModel->expects($this->once()) ->method('sendPurgeRequest') - ->with([self::CACHE_TAGS]) + ->with(self::CACHE_TAGS) ->willReturn(false); - $this->connection->expects($this->never())->method('delete'); + $this->storage->expects($this->never())->method('clear'); + $this->storage->expects($this->once())->method('release')->with(self::BATCH_ID); $this->loggerInterface->expects($this->once())->method('error'); @@ -73,20 +123,25 @@ public function testFlushDoesNotClearQueueWhenPurgeRequestFails() public function testFlushResetsDebounceFlagEvenWhenPurgeRequestThrows() { - $scopeConfig = $this->createMock(\Magento\Framework\App\Config\ScopeConfigInterface::class); + $scopeConfig = $this->createMock(ScopeConfigInterface::class); $scopeConfig->method('isSetFlag')->willReturn(true); - $config = new \SamJUK\CacheDebounce\Model\Config($scopeConfig); + $config = new CacheDebounceConfig($scopeConfig); - $this->connection->method('fetchCol')->willReturn(self::CACHE_TAGS); + $this->storage->method('claim')->willReturn(self::BATCH_ID); + $this->storage->method('tags')->willReturn(self::CACHE_TAGS); $this->purgeCacheModel->method('sendPurgeRequest')->willThrowException(new \RuntimeException('varnish down')); $entries = new CacheDebounceEntries( $config, $this->purgeCacheModel, - $this->resourceConnection, - $this->loggerInterface + $this->storage, + $this->loggerInterface, + $this->flagManager ); + $this->storage->expects($this->never())->method('clear'); + $this->storage->expects($this->once())->method('release')->with(self::BATCH_ID); + $this->assertTrue($config->shouldDebouncePurgeRequest()); try { @@ -99,19 +154,17 @@ public function testFlushResetsDebounceFlagEvenWhenPurgeRequestThrows() $this->assertTrue($config->shouldDebouncePurgeRequest()); } - public function testAddWithEmptyTagsDoesNotTouchConnection() + public function testAddDelegatesToStorage() { - $this->connection->expects($this->never())->method('insertArray'); + $this->storage->expects($this->once())->method('add')->with(self::CACHE_TAGS); - $this->cacheDebounceEntries->add([]); + $this->cacheDebounceEntries->add(self::CACHE_TAGS); } - public function testAddWithTagsInsertsThem() + public function testAddWithEmptyTagsStillDelegatesToStorage() { - $this->connection->expects($this->once()) - ->method('insertArray') - ->with($this->anything(), ['tag'], self::CACHE_TAGS); + $this->storage->expects($this->once())->method('add')->with([]); - $this->cacheDebounceEntries->add(self::CACHE_TAGS); + $this->cacheDebounceEntries->add([]); } } diff --git a/Test/Unit/Model/LagDetectorTest.php b/Test/Unit/Model/LagDetectorTest.php new file mode 100644 index 0000000..ea6dfdd --- /dev/null +++ b/Test/Unit/Model/LagDetectorTest.php @@ -0,0 +1,116 @@ +flagState = null; + + $this->config = $this->createMock(Config::class); + $this->config->method('getStaggerLagRatioThreshold')->willReturn(1.0); + $this->config->method('getStaggerLagAlertAfterRuns')->willReturn(3); + + $this->flagManager = $this->createMock(FlagManager::class); + $this->flagManager->method('getFlagData')->willReturnCallback(function () { + return $this->flagState; + }); + $this->flagManager->method('saveFlag')->willReturnCallback(function ($code, $value) { + $this->flagState = $value; + return true; + }); + $this->flagManager->method('deleteFlag')->willReturnCallback(function () { + $this->flagState = null; + return true; + }); + + $this->notifier = $this->createMock(NotifierInterface::class); + $this->logger = $this->createMock(LoggerInterface::class); + + $this->detector = new LagDetector($this->config, $this->flagManager, $this->notifier, $this->logger); + } + + /** + * @param array $samples pairs of [claimed, arrivedDuring] + */ + private function runSequence(array $samples): void + { + foreach ($samples as [$claimed, $arrivedDuring]) { + $this->detector->recordSample($claimed, $arrivedDuring); + } + } + + public function testNonLaggingSampleDeletesTheCounterAndNeverNotifies() + { + $this->notifier->expects($this->never())->method('addMajor'); + + $this->runSequence([[10, 5]]); + + $this->assertNull($this->flagState); + } + + public function testRatioAtExactlyTheThresholdCountsAsLagging() + { + $this->runSequence([[10, 10]]); + + $this->assertSame(1, $this->flagState); + } + + public function testZeroClaimedNeverCountsAsLaggingRegardlessOfArrivals() + { + $this->notifier->expects($this->never())->method('addMajor'); + + $this->runSequence([[0, 5]]); + + $this->assertNull($this->flagState); + } + + public function testConsecutiveLaggingRunsAccumulateAndNotifyOnlyOnReachingTheThreshold() + { + $this->notifier->expects($this->once())->method('addMajor'); + $this->logger->expects($this->once())->method('warning'); + + // lag_alert_after_runs = 3: first two lagging runs stay silent. + $this->runSequence([[10, 20], [10, 20], [10, 20]]); + + $this->assertSame(3, $this->flagState); + } + + public function testANonLaggingRunResetsTheCounterSoTheNextLagStartsFromOne() + { + $this->notifier->expects($this->never())->method('addMajor'); + + $this->runSequence([[10, 20], [10, 20], [10, 5], [10, 20]]); + + $this->assertSame(1, $this->flagState); + } + + public function testSustainedLagReNotifiesEveryAlertAfterRunsWithoutResetting() + { + $this->notifier->expects($this->exactly(2))->method('addMajor'); + + $this->runSequence([ + [10, 20], [10, 20], [10, 20], // 3rd run: 1st notification + [10, 20], [10, 20], [10, 20], // 6th run: 2nd notification + ]); + + $this->assertSame(6, $this->flagState); + } +} diff --git a/Test/Unit/Model/StaggeredFlushTest.php b/Test/Unit/Model/StaggeredFlushTest.php new file mode 100644 index 0000000..724597f --- /dev/null +++ b/Test/Unit/Model/StaggeredFlushTest.php @@ -0,0 +1,200 @@ +config = $this->createMock(Config::class); + $this->config->method('isStaggerEnabled')->willReturn(true); + $this->config->method('getStaggerBatchSize')->willReturn(50); + $this->config->method('getStaggerIntervalMs')->willReturn(0); + $this->config->method('getStaggerMaxRuntimeSeconds')->willReturn(999); + + $this->storage = $this->createMock(QueueStorageInterface::class); + $this->storage->method('activeBatch')->willReturn(''); + + $this->purgeCache = $this->createMock(PurgeCache::class); + $this->purgeCache->method('sendPurgeRequest')->willReturn(true); + + $this->lockManager = $this->createMock(LockManagerInterface::class); + $this->lockManager->method('lock')->willReturn(true); + + $this->lagDetector = $this->createMock(LagDetector::class); + $this->flagManager = $this->createMock(FlagManager::class); + $this->entries = $this->createMock(Entries::class); + $this->logger = $this->createMock(LoggerInterface::class); + } + + private function staggeredFlush(): StaggeredFlush + { + return new StaggeredFlush( + $this->config, + $this->storage, + $this->purgeCache, + $this->lockManager, + $this->lagDetector, + $this->flagManager, + $this->entries, + $this->logger + ); + } + + public function testFallsBackToASingleShotFlushWhenStaggerIsDisabled() + { + $this->config = $this->createMock(Config::class); + $this->config->method('isStaggerEnabled')->willReturn(false); + + $this->entries->expects($this->once())->method('flush'); + $this->lockManager->expects($this->never())->method('lock'); + + $this->staggeredFlush()->execute(); + } + + public function testFailedLockAcquisitionExitsWithoutCallingClaim() + { + $this->lockManager = $this->createMock(LockManagerInterface::class); + $this->lockManager->method('lock')->willReturn(false); + + $this->storage->expects($this->never())->method('claim'); + $this->lockManager->expects($this->never())->method('unlock'); + + $this->staggeredFlush()->execute(); + } + + public function testNothingPendingUnlocksWithoutPurging() + { + $this->storage->method('claim')->willReturn(''); + + $this->purgeCache->expects($this->never())->method('sendPurgeRequest'); + $this->lockManager->expects($this->once())->method('unlock'); + + $this->staggeredFlush()->execute(); + } + + public function testResumesAnAlreadyClaimedActiveBatchInsteadOfReclaiming() + { + $this->storage = $this->createMock(QueueStorageInterface::class); + $this->storage->method('activeBatch')->willReturn(self::BATCH_ID); + $this->storage->method('tags')->with(self::BATCH_ID)->willReturn(['cat_c_1']); + $this->storage->expects($this->never())->method('claim'); + $this->storage->expects($this->once())->method('clear')->with(self::BATCH_ID); + + // A resumed batch's post-drain pendingCount() can include tags that + // arrived before this run started, so it's not sampled for lag. + $this->lagDetector->expects($this->never())->method('recordSample'); + + $this->staggeredFlush()->execute(); + } + + public function testChunksTagsAccordingToBatchSize() + { + $this->config = $this->createMock(Config::class); + $this->config->method('isStaggerEnabled')->willReturn(true); + $this->config->method('getStaggerBatchSize')->willReturn(2); + $this->config->method('getStaggerIntervalMs')->willReturn(0); + $this->config->method('getStaggerMaxRuntimeSeconds')->willReturn(999); + $this->storage->method('claim')->willReturn(self::BATCH_ID); + $this->storage->method('tags')->willReturn(['a', 'b', 'c', 'd', 'e']); + + $chunks = []; + $this->purgeCache->expects($this->exactly(3)) + ->method('sendPurgeRequest') + ->willReturnCallback(function ($chunk) use (&$chunks) { + $chunks[] = $chunk; + return true; + }); + + $this->staggeredFlush()->execute(); + + $this->assertSame([['a', 'b'], ['c', 'd'], ['e']], $chunks); + } + + public function testStopsAndDoesNotClearOnceTheRuntimeBudgetIsExceededMidRun() + { + // Multiple chunks, so the budget check (which only applies from the + // second chunk onward) has a chance to actually stop the run. + $this->config = $this->createMock(Config::class); + $this->config->method('isStaggerEnabled')->willReturn(true); + $this->config->method('getStaggerBatchSize')->willReturn(1); + $this->config->method('getStaggerIntervalMs')->willReturn(0); + $this->config->method('getStaggerMaxRuntimeSeconds')->willReturn(0); + $this->storage->method('claim')->willReturn(self::BATCH_ID); + $this->storage->method('tags')->willReturn(['cat_c_1', 'cat_c_2']); + + $this->purgeCache->expects($this->once())->method('sendPurgeRequest')->willReturn(true); + $this->storage->expects($this->never())->method('clear'); + $this->lagDetector->expects($this->never())->method('recordSample'); + $this->lockManager->expects($this->once())->method('unlock'); + + $this->staggeredFlush()->execute(); + } + + public function testAlwaysSendsTheFirstChunkEvenWithAnExhaustedRuntimeBudget() + { + // A misconfigured (e.g. zero) budget must never block *every* + // chunk — otherwise the batch can never drain, ever, on any run. + $this->config = $this->createMock(Config::class); + $this->config->method('isStaggerEnabled')->willReturn(true); + $this->config->method('getStaggerBatchSize')->willReturn(50); + $this->config->method('getStaggerIntervalMs')->willReturn(0); + $this->config->method('getStaggerMaxRuntimeSeconds')->willReturn(0); + $this->storage->method('claim')->willReturn(self::BATCH_ID); + $this->storage->method('tags')->willReturn(['cat_c_1']); + $this->storage->method('pendingCount')->willReturn(0); + + $this->purgeCache->expects($this->once())->method('sendPurgeRequest')->with(['cat_c_1'])->willReturn(true); + $this->storage->expects($this->once())->method('clear')->with(self::BATCH_ID); + + $this->staggeredFlush()->execute(); + } + + public function testAFailedPurgeChunkLeavesTheBatchClaimedAndStops() + { + $this->storage->method('claim')->willReturn(self::BATCH_ID); + $this->storage->method('tags')->willReturn(['cat_c_1', 'cat_c_2']); + $this->purgeCache = $this->createMock(PurgeCache::class); + $this->purgeCache->method('sendPurgeRequest')->willReturn(false); + + $this->storage->expects($this->never())->method('clear'); + $this->lagDetector->expects($this->never())->method('recordSample'); + + $this->staggeredFlush()->execute(); + } + + public function testFullyDrainedBatchIsClearedAndRecordedForLagDetection() + { + $this->storage->method('claim')->willReturn(self::BATCH_ID); + $this->storage->method('tags')->willReturn(['cat_c_1', 'cat_c_2']); + $this->storage->method('pendingCount')->willReturn(1); + + $this->storage->expects($this->once())->method('clear')->with(self::BATCH_ID); + $this->lagDetector->expects($this->once())->method('recordSample')->with(2, 1); + $this->flagManager->expects($this->exactly(2))->method('saveFlag'); + + $this->staggeredFlush()->execute(); + } +} diff --git a/Test/Unit/Model/Storage/DatabaseTest.php b/Test/Unit/Model/Storage/DatabaseTest.php new file mode 100644 index 0000000..180f440 --- /dev/null +++ b/Test/Unit/Model/Storage/DatabaseTest.php @@ -0,0 +1,226 @@ +connection = $this->createMock(AdapterInterface::class); + $this->connection->method('quoteInto')->willReturnCallback( + fn ($text, $value) => str_replace('?', "'$value'", $text) + ); + $this->resourceConnection = $this->createMock(ResourceConnection::class); + $this->resourceConnection->method('getConnection')->willReturn($this->connection); + $this->resourceConnection->method('getTableName')->willReturn(self::TABLE_NAME); + + $this->storage = new Database($this->resourceConnection); + } + + private function stubSelect(): void + { + $select = $this->createMock(\Magento\Framework\DB\Select::class); + $select->method('from')->willReturn($select); + $select->method('where')->willReturn($select); + $this->connection->method('select')->willReturn($select); + } + + public function testAddWithEmptyTagsDoesNotTouchConnection() + { + $this->connection->expects($this->never())->method('insertArray'); + + $this->storage->add([]); + } + + public function testAddInsertsTagsAsUnclaimedWithInsertIgnore() + { + $this->stubSelect(); + $this->connection->method('fetchCol')->willReturn([]); + + $this->connection->expects($this->once()) + ->method('insertArray') + ->with( + self::TABLE_NAME, + ['batch_id', 'tag'], + [['', 'cat_c_1'], ['', 'cat_c_2']], + AdapterInterface::INSERT_IGNORE + ); + + $this->storage->add(['cat_c_1', 'cat_c_2']); + } + + public function testAddFiltersOutNullAndEmptyTags() + { + $this->stubSelect(); + $this->connection->method('fetchCol')->willReturn([]); + + $this->connection->expects($this->once()) + ->method('insertArray') + ->with( + self::TABLE_NAME, + ['batch_id', 'tag'], + [['', 'cat_c_1']], + AdapterInterface::INSERT_IGNORE + ); + + $this->storage->add(['cat_c_1', null, '']); + } + + public function testAddSkipsTagsAlreadyPending() + { + $this->stubSelect(); + $this->connection->method('fetchCol')->willReturn(['cat_c_1']); + + $this->connection->expects($this->once()) + ->method('insertArray') + ->with( + self::TABLE_NAME, + ['batch_id', 'tag'], + [['', 'cat_c_2']], + AdapterInterface::INSERT_IGNORE + ); + + $this->storage->add(['cat_c_1', 'cat_c_2']); + } + + public function testAddInsertsNothingWhenEveryTagIsAlreadyPending() + { + $this->stubSelect(); + $this->connection->method('fetchCol')->willReturn(['cat_c_1']); + + $this->connection->expects($this->never())->method('insertArray'); + + $this->storage->add(['cat_c_1']); + } + + public function testClaimReturnsEmptyStringWhenNothingPending() + { + $this->connection->method('update')->willReturn(0); + + $this->assertSame('', $this->storage->claim()); + } + + public function testClaimReturnsBatchIdWhenRowsClaimed() + { + $this->connection->method('update')->willReturn(2); + + $batchId = $this->storage->claim(); + + $this->assertNotSame('', $batchId); + $this->assertSame(32, strlen($batchId)); + } + + public function testTagsSelectsScopedToBatchId() + { + $select = $this->createMock(\Magento\Framework\DB\Select::class); + $select->method('from')->willReturn($select); + $select->method('where')->with('batch_id = ?', 'batch-123')->willReturn($select); + $this->connection->method('select')->willReturn($select); + $this->connection->method('fetchCol')->willReturn(['cat_c_1']); + + $this->assertSame(['cat_c_1'], $this->storage->tags('batch-123')); + } + + public function testClearDeletesOnlyThatBatch() + { + $this->connection->expects($this->once()) + ->method('delete') + ->with(self::TABLE_NAME, "batch_id = 'batch-123'"); + + $this->storage->clear('batch-123'); + } + + public function testReleaseReQueuesBatchTagsAsUnclaimedThenClearsTheBatch() + { + $this->stubSelect(); + // First call is tags($batchId) reading the claimed batch; second is + // add()'s own dedup check against currently-pending tags. + $this->connection->method('fetchCol')->willReturnOnConsecutiveCalls(['cat_c_1', 'cat_c_2'], []); + + $this->connection->expects($this->once()) + ->method('insertArray') + ->with( + self::TABLE_NAME, + ['batch_id', 'tag'], + [['', 'cat_c_1'], ['', 'cat_c_2']], + AdapterInterface::INSERT_IGNORE + ); + $this->connection->expects($this->once()) + ->method('delete') + ->with(self::TABLE_NAME, "batch_id = 'batch-123'"); + + $this->storage->release('batch-123'); + } + + public function testPendingCountCountsOnlyUnclaimedRows() + { + $select = $this->createMock(\Magento\Framework\DB\Select::class); + $select->method('from')->with(self::TABLE_NAME, ['count' => 'COUNT(*)'])->willReturn($select); + $select->method('where')->with('batch_id = ?', '')->willReturn($select); + $this->connection->method('select')->willReturn($select); + $this->connection->method('fetchOne')->with($select)->willReturn('5'); + + $this->assertSame(5, $this->storage->pendingCount()); + } + + public function testActiveBatchReturnsEmptyStringWhenNothingClaimed() + { + $select = $this->createMock(\Magento\Framework\DB\Select::class); + $select->method('from')->willReturn($select); + $select->method('where')->with('batch_id != ?', '')->willReturn($select); + $select->method('limit')->with(1)->willReturn($select); + $this->connection->method('select')->willReturn($select); + $this->connection->method('fetchOne')->willReturn(false); + + $this->assertSame('', $this->storage->activeBatch()); + } + + public function testActiveBatchReturnsIdOfAlreadyClaimedBatch() + { + $select = $this->createMock(\Magento\Framework\DB\Select::class); + $select->method('from')->willReturn($select); + $select->method('where')->willReturn($select); + $select->method('limit')->willReturn($select); + $this->connection->method('select')->willReturn($select); + $this->connection->method('fetchOne')->willReturn('batch-123'); + + $this->assertSame('batch-123', $this->storage->activeBatch()); + } + + public function testOldestPendingAgeSecondsReturnsNullWhenNothingPending() + { + // Aggregate query returns one row of SQL NULL, which PDO surfaces as null. + $select = $this->createMock(\Magento\Framework\DB\Select::class); + $select->method('from')->willReturn($select); + $select->method('where')->willReturn($select); + $this->connection->method('select')->willReturn($select); + $this->connection->method('fetchOne')->willReturn(null); + + $this->assertNull($this->storage->oldestPendingAgeSeconds()); + } + + public function testOldestPendingAgeSecondsReturnsAgeComputedBySql() + { + $select = $this->createMock(\Magento\Framework\DB\Select::class); + $select->method('from')->with( + self::TABLE_NAME, + ['age' => 'TIMESTAMPDIFF(SECOND, MIN(created_at), NOW())'] + )->willReturn($select); + $select->method('where')->willReturn($select); + $this->connection->method('select')->willReturn($select); + $this->connection->method('fetchOne')->willReturn('10'); + + $this->assertSame(10, $this->storage->oldestPendingAgeSeconds()); + } +} diff --git a/Test/Unit/Model/Storage/PoolTest.php b/Test/Unit/Model/Storage/PoolTest.php new file mode 100644 index 0000000..08a4af8 --- /dev/null +++ b/Test/Unit/Model/Storage/PoolTest.php @@ -0,0 +1,114 @@ + Database::class, + 'redis' => Redis::class, + ]; + + private $objectManager; + private $scopeConfig; + private $database; + private $redis; + + protected function setUp(): void + { + $this->objectManager = $this->createMock(ObjectManagerInterface::class); + $this->scopeConfig = $this->createMock(ScopeConfigInterface::class); + $this->database = $this->createMock(Database::class); + $this->redis = $this->createMock(Redis::class); + + $this->objectManager->method('create')->willReturnMap([ + [Database::class, [], $this->database], + [Redis::class, [], $this->redis], + ]); + } + + private function pool(): Pool + { + return new Pool($this->objectManager, $this->scopeConfig, self::DRIVERS); + } + + public function testUsesDatabaseDriverByDefault() + { + $this->scopeConfig->method('getValue')->with(self::XML_PATH_STORAGE_DRIVER)->willReturn(null); + + $this->database->expects($this->once())->method('add')->with(self::CACHE_TAGS); + + $this->pool()->add(self::CACHE_TAGS); + } + + public function testUsesDatabaseDriverForAnyValueOtherThanRedis() + { + $this->scopeConfig->method('getValue')->willReturn('db'); + + $this->database->expects($this->once())->method('claim')->willReturn('batch-123'); + + $this->pool()->claim(); + } + + public function testUsesRedisDriverWhenConfiguredAndNeverTouchesDatabase() + { + $this->scopeConfig->method('getValue')->willReturn('redis'); + + $this->database->expects($this->never())->method('add'); + $this->redis->expects($this->once())->method('add')->with(self::CACHE_TAGS); + + $this->pool()->add(self::CACHE_TAGS); + } + + public function testDriverIsOnlyResolvedOnceAcrossMultipleCalls() + { + $objectManager = $this->createMock(ObjectManagerInterface::class); + $objectManager->expects($this->once())->method('create')->with(Redis::class)->willReturn($this->redis); + $this->scopeConfig->method('getValue')->willReturn('redis'); + + $pool = new Pool($objectManager, $this->scopeConfig, self::DRIVERS); + $pool->add(['cat_c_1']); + $pool->claim(); + } + + public function testReleaseDelegatesToTheResolvedDriver() + { + $this->scopeConfig->method('getValue')->willReturn('db'); + $this->database->expects($this->once())->method('release')->with('batch-123'); + + $this->pool()->release('batch-123'); + } + + public function testPendingCountDelegatesToTheResolvedDriver() + { + $this->scopeConfig->method('getValue')->willReturn('db'); + $this->database->expects($this->once())->method('pendingCount')->willReturn(5); + + $this->assertSame(5, $this->pool()->pendingCount()); + } + + public function testActiveBatchDelegatesToTheResolvedDriver() + { + $this->scopeConfig->method('getValue')->willReturn('redis'); + $this->redis->expects($this->once())->method('activeBatch')->willReturn('batch-123'); + + $this->assertSame('batch-123', $this->pool()->activeBatch()); + } + + public function testOldestPendingAgeSecondsDelegatesToTheResolvedDriver() + { + $this->scopeConfig->method('getValue')->willReturn('db'); + $this->database->expects($this->once())->method('oldestPendingAgeSeconds')->willReturn(42); + + $this->assertSame(42, $this->pool()->oldestPendingAgeSeconds()); + } +} diff --git a/Test/Unit/Model/Storage/Redis/ConnectionResolverTest.php b/Test/Unit/Model/Storage/Redis/ConnectionResolverTest.php new file mode 100644 index 0000000..5a5e756 --- /dev/null +++ b/Test/Unit/Model/Storage/Redis/ConnectionResolverTest.php @@ -0,0 +1,69 @@ + '10.0.0.1', 'port' => '6390', 'password' => 'x', 'database' => '2']; + private const FALLBACK_CONFIG = ['server' => '10.0.0.2', 'port' => '6379', 'password' => '', 'database' => '0']; + + private $deploymentConfig; + + protected function setUp(): void + { + $this->deploymentConfig = $this->createMock(DeploymentConfig::class); + } + + public function testDedicatedConfigTakesPriorityOverTheSharedFallback() + { + $this->deploymentConfig->method('get')->willReturnMap([ + ['cache_debounce/redis', null, self::DEDICATED_CONFIG], + ['cache/frontend/page_cache/backend_options', null, self::FALLBACK_CONFIG], + ]); + + $this->assertResolvesToClient(); + } + + public function testFallsBackToThePageCacheBackendWhenDedicatedConfigIsAbsent() + { + $this->deploymentConfig->method('get')->willReturnMap([ + ['cache_debounce/redis', null, null], + ['cache/frontend/page_cache/backend_options', null, self::FALLBACK_CONFIG], + ]); + + $this->assertResolvesToClient(); + } + + public function testThrowsLocalizedExceptionWhenNeitherConfigIsAvailable() + { + $this->deploymentConfig->method('get')->willReturn(null); + + $resolver = new ConnectionResolver($this->deploymentConfig); + + $this->expectException(LocalizedException::class); + + $resolver->getClient(); + } + + public function testDedicatedConfigMissingKeysFallsBackToDefaultsInsteadOfErroring() + { + $this->deploymentConfig->method('get')->willReturnMap([ + ['cache_debounce/redis', null, ['host' => '10.0.0.9']], + ['cache/frontend/page_cache/backend_options', null, null], + ]); + + $this->assertResolvesToClient(); + } + + private function assertResolvesToClient(): void + { + $resolver = new ConnectionResolver($this->deploymentConfig); + + $this->assertInstanceOf(\Credis_Client::class, $resolver->getClient()); + } +} diff --git a/Test/Unit/Model/Storage/RedisTest.php b/Test/Unit/Model/Storage/RedisTest.php new file mode 100644 index 0000000..641d610 --- /dev/null +++ b/Test/Unit/Model/Storage/RedisTest.php @@ -0,0 +1,168 @@ +client = $this->getMockBuilder(\Credis_Client::class) + ->disableOriginalConstructor() + ->addMethods(['sAdd', 'rename', 'sMembers', 'del', 'sUnionStore', 'sCard', 'set', 'get']) + ->getMock(); + $this->connectionResolver = $this->createMock(ConnectionResolver::class); + $this->connectionResolver->method('getClient')->willReturn($this->client); + + $this->storage = new Redis($this->connectionResolver); + } + + public function testAddWithEmptyTagsDoesNotTouchConnection() + { + $this->client->expects($this->never())->method('sAdd'); + + $this->storage->add([]); + } + + public function testAddSaddsTagsIntoTheLiveSet() + { + $this->client->expects($this->once()) + ->method('sAdd') + ->with(self::KEY_LIVE, 'cat_c_1', 'cat_c_2'); + + $this->storage->add(['cat_c_1', 'cat_c_2']); + } + + public function testClaimReturnsEmptyStringWhenLiveKeyDoesNotExist() + { + $this->client->method('rename')->willThrowException(new \CredisException('ERR no such key')); + + $this->assertSame('', $this->storage->claim()); + } + + public function testClaimDoesNotSwallowOtherRedisErrors() + { + $this->client->method('rename')->willThrowException(new \CredisException('ERR something else')); + + $this->expectException(\CredisException::class); + $this->expectExceptionMessage('ERR something else'); + + $this->storage->claim(); + } + + public function testClaimRenamesLiveKeyToABatchKeyAndReturnsItsId() + { + $this->client->expects($this->once()) + ->method('rename') + ->with(self::KEY_LIVE, $this->callback(function (string $key) { + return strpos($key, self::KEY_BATCH_PREFIX) === 0 + && strlen($key) === strlen(self::KEY_BATCH_PREFIX) + 32; + })); + + $batchId = $this->storage->claim(); + + $this->assertNotSame('', $batchId); + $this->assertSame(32, strlen($batchId)); + } + + public function testClaimRecordsTheActiveBatchMarker() + { + $this->client->expects($this->once()) + ->method('set') + ->with(self::KEY_ACTIVE_BATCH, $this->isType('string')); + + $this->storage->claim(); + } + + public function testClaimDoesNotSetTheActiveBatchMarkerWhenNothingWasPending() + { + $this->client->method('rename')->willThrowException(new \CredisException('ERR no such key')); + $this->client->expects($this->never())->method('set'); + + $this->storage->claim(); + } + + public function testTagsReadsFromTheBatchKey() + { + $this->client->expects($this->once()) + ->method('sMembers') + ->with(self::KEY_BATCH_PREFIX . 'batch-123') + ->willReturn(['cat_c_1']); + + $this->assertSame(['cat_c_1'], $this->storage->tags('batch-123')); + } + + public function testClearDeletesTheBatchKeyAndTheActiveBatchMarker() + { + $deletedKeys = []; + $this->client->expects($this->exactly(2)) + ->method('del') + ->willReturnCallback(function (string $key) use (&$deletedKeys) { + $deletedKeys[] = $key; + return true; + }); + + $this->storage->clear('batch-123'); + + $this->assertSame([self::KEY_BATCH_PREFIX . 'batch-123', self::KEY_ACTIVE_BATCH], $deletedKeys); + } + + public function testReleaseMergesBatchSetBackIntoLiveSetThenDeletesTheBatchAndMarker() + { + $this->client->expects($this->once()) + ->method('sUnionStore') + ->with(self::KEY_LIVE, self::KEY_LIVE, self::KEY_BATCH_PREFIX . 'batch-123'); + + $deletedKeys = []; + $this->client->expects($this->exactly(2)) + ->method('del') + ->willReturnCallback(function (string $key) use (&$deletedKeys) { + $deletedKeys[] = $key; + return true; + }); + + $this->storage->release('batch-123'); + + $this->assertSame([self::KEY_BATCH_PREFIX . 'batch-123', self::KEY_ACTIVE_BATCH], $deletedKeys); + } + + public function testPendingCountReadsCardinalityOfTheLiveSet() + { + $this->client->expects($this->once()) + ->method('sCard') + ->with(self::KEY_LIVE) + ->willReturn(3); + + $this->assertSame(3, $this->storage->pendingCount()); + } + + public function testActiveBatchReturnsEmptyStringWhenNoBatchIsClaimed() + { + $this->client->method('get')->with(self::KEY_ACTIVE_BATCH)->willReturn(false); + + $this->assertSame('', $this->storage->activeBatch()); + } + + public function testActiveBatchReturnsIdOfAlreadyClaimedBatch() + { + $this->client->method('get')->with(self::KEY_ACTIVE_BATCH)->willReturn('batch-123'); + + $this->assertSame('batch-123', $this->storage->activeBatch()); + } + + public function testOldestPendingAgeSecondsIsAlwaysNull() + { + $this->assertNull($this->storage->oldestPendingAgeSeconds()); + } +} diff --git a/etc/adminhtml/acl.xml b/etc/adminhtml/acl.xml new file mode 100644 index 0000000..6bda1e9 --- /dev/null +++ b/etc/adminhtml/acl.xml @@ -0,0 +1,18 @@ + + + + + + + + + + + + + + + + + + diff --git a/etc/adminhtml/system.xml b/etc/adminhtml/system.xml index c51d38b..9d3004b 100644 --- a/etc/adminhtml/system.xml +++ b/etc/adminhtml/system.xml @@ -22,6 +22,49 @@ + + + + + Release a claimed batch gradually instead of purging it all at once. Disabled = single-shot flush() (current behavior). + Magento\Config\Model\Config\Source\Yesno + + + + Tags per purge chunk. + + + + Pause between chunks. + + + + Safety budget per cron/CLI invocation — must comfortably clear before the Flush Schedule interval elapses again. + + + + A run counts as "lagging" if tags arriving during the drain are >= this ratio of tags just drained. + + + + Consecutive lagging runs before an admin notification is raised. + + + + + +
+ + samjuk + SamJUK_CacheDebounce::storage_driver + + + + + Do not change unless you know what you are doing. See README.md. + SamJUK\CacheDebounce\Model\Config\Source\StorageDriver + +
diff --git a/etc/config.xml b/etc/config.xml index fcd33e3..48d5d7f 100644 --- a/etc/config.xml +++ b/etc/config.xml @@ -8,6 +8,19 @@ */15 * * * * + + 0 + 50 + 1000 + 240 + 1 + 3 + + + + db + + diff --git a/etc/db_schema.xml b/etc/db_schema.xml index 41f0d15..80c6838 100644 --- a/etc/db_schema.xml +++ b/etc/db_schema.xml @@ -1,6 +1,9 @@ + + +
diff --git a/etc/db_schema_whitelist.json b/etc/db_schema_whitelist.json new file mode 100644 index 0000000..fcb747e --- /dev/null +++ b/etc/db_schema_whitelist.json @@ -0,0 +1,9 @@ +{ + "samjuk_cache_debounce": { + "column": { + "batch_id": true, + "tag": true, + "created_at": true + } + } +} diff --git a/etc/di.xml b/etc/di.xml index 36fd265..457698c 100644 --- a/etc/di.xml +++ b/etc/di.xml @@ -1,5 +1,16 @@ + + + + + + SamJUK\CacheDebounce\Model\Storage\Database + SamJUK\CacheDebounce\Model\Storage\Redis + + + + @@ -8,6 +19,7 @@ SamJUK\CacheDebounce\Console\Command\Flush + SamJUK\CacheDebounce\Console\Command\Status