Abstract queue storage behind an interface, fix add-during-purge race

Model/Entries.php

@@ -5,17 +5,14 @@
 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 Psr\Log\LoggerInterface;
 
 class Entries
 {
-    /** @var string $tableName */
-    private $tableName;
-
-    /** @var ResourceConnection $resourceConnection */
-    private $resourceConnection;
+    /** @var QueueStorageInterface $storage */
+    private $storage;
 
     /** @var PurgeCache $purgeCache */
     private $purgeCache;
@@ -29,50 +26,38 @@ class Entries
     public function __construct(
         Config $config,
         PurgeCache $purgeCache,
-        ResourceConnection $resourceConnection,
+        QueueStorageInterface $storage,
         LoggerInterface $logger
     ) {
         $this->config = $config;
         $this->purgeCache = $purgeCache;
-        $this->resourceConnection = $resourceConnection;
+        $this->storage = $storage;
         $this->logger = $logger;
-        $this->tableName = $resourceConnection->getTableName('samjuk_cache_debounce');
     }
 
     /**
      * 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) {
+        $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);
 
@@ -81,12 +66,20 @@ public function flush() : void
 
             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);
         }

Model/Storage/Database.php

@@ -0,0 +1,126 @@
+<?php
+
+declare(strict_types=1);
+
+namespace SamJUK\CacheDebounce\Model\Storage;
+
+use Magento\Framework\App\ResourceConnection;
+use Magento\Framework\DB\Adapter\AdapterInterface;
+
+class Database implements QueueStorageInterface
+{
+    private const UNCLAIMED_BATCH_ID = '';
+
+    /** @var string $tableName */
+    private $tableName;
+
+    /** @var ResourceConnection $resourceConnection */
+    private $resourceConnection;
+
+    public function __construct(
+        ResourceConnection $resourceConnection
+    ) {
+        $this->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);
+    }
+
+    /**
+     * @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 : '';
+    }
+
+    private function quoteBatchId(string $batchId): string
+    {
+        return $this->resourceConnection->getConnection()->quoteInto('batch_id = ?', $batchId);
+    }
+}

Model/Storage/QueueStorageInterface.php

@@ -0,0 +1,39 @@
+<?php
+
+declare(strict_types=1);
+
+namespace SamJUK\CacheDebounce\Model\Storage;
+
+interface QueueStorageInterface
+{
+    /**
+     * Queue tags for a future purge. Must be idempotent and no-op on empty.
+     */
+    public function add(array $tags): void;
+
+    /**
+     * Atomically carve off pending tags into a new batch; '' if none.
+     */
+    public function claim(): string;
+
+    /**
+     * Read the tags belonging to a previously claimed batch.
+     */
+    public function tags(string $batchId): array;
+
+    /**
+     * Delete a claimed batch. Never touches pending (unclaimed) rows.
+     */
+    public function clear(string $batchId): void;
+
+    /**
+     * Release a claimed batch back to pending, merging idempotently.
+     */
+    public function release(string $batchId): void;
+
+    /**
+     * Id of an already-claimed, not-yet-cleared batch, if one exists.
+     * Returns '' if nothing is currently claimed.
+     */
+    public function activeBatch(): string;
+}

Setup/Patch/Data/ClearQueueTable.php

@@ -0,0 +1,39 @@
+<?php
+
+declare(strict_types=1);
+
+namespace SamJUK\CacheDebounce\Setup\Patch\Data;
+
+use Magento\Framework\Setup\ModuleDataSetupInterface;
+use Magento\Framework\Setup\Patch\DataPatchInterface;
+
+/**
+ * Clears disposable queue state before a follow-up release adds a PK.
+ */
+class ClearQueueTable implements DataPatchInterface
+{
+    /** @var ModuleDataSetupInterface $moduleDataSetup */
+    private $moduleDataSetup;
+
+    public function __construct(ModuleDataSetupInterface $moduleDataSetup)
+    {
+        $this->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 [];
+    }
+}