Add staggered/batched purge release (spec 003)

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('<info>Flushed Debounced Cache Purges.</info>');
             return 0;
         } catch (LocalizedException $e) {

Console/Command/Status.php

@@ -0,0 +1,92 @@
+<?php
+
+declare(strict_types=1);
+
+namespace SamJUK\CacheDebounce\Console\Command;
+
+use Magento\Framework\FlagManager;
+use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Output\OutputInterface;
+use SamJUK\CacheDebounce\Model\Config;
+use SamJUK\CacheDebounce\Model\Entries;
+use SamJUK\CacheDebounce\Model\LagDetector;
+use SamJUK\CacheDebounce\Model\Storage\QueueStorageInterface;
+
+class Status extends Command
+{
+    private $storage;
+    private $config;
+    private $flagManager;
+
+    /**
+     * @param QueueStorageInterface $storage
+     * @param Config $config
+     * @param FlagManager $flagManager
+     * @param string|null $name
+     */
+    public function __construct(
+        QueueStorageInterface $storage,
+        Config $config,
+        FlagManager $flagManager,
+        $name = null
+    ) {
+        $this->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);
+    }
+}

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();
     }
 }

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
      */

Model/Config/Source/StorageDriver.php

@@ -0,0 +1,21 @@
+<?php
+
+declare(strict_types=1);
+
+namespace SamJUK\CacheDebounce\Model\Config\Source;
+
+use Magento\Framework\Data\OptionSourceInterface;
+
+class StorageDriver implements OptionSourceInterface
+{
+    /**
+     * @inheritDoc
+     */
+    public function toOptionArray(): array
+    {
+        return [
+            ['value' => 'db', 'label' => __('Database (default)')],
+            ['value' => 'redis', 'label' => __('Redis')],
+        ];
+    }
+}

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,69 +27,75 @@ 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;
     }
 
     /**
      * Add new tags to the purge queue
      */
     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);
     }
 
     /**
      * Purge all queued tags, and clear the queue
      */
     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));
     }
 }