Merge pull request #795 from durable-workflow/release/2.5.20-watchdog

Release 2.5.20 with bounded database watchdog throttles

Author: rmcdaniel

composer.json

@@ -79,7 +79,7 @@
             "dev-main": "2.0.x-dev"
         },
         "durable-workflow": {
-            "product-train": "2.5.19",
+            "product-train": "2.5.20",
             "laravel-embedded-upgrade-contract": "resources/laravel-embedded-upgrade-contract.json",
             "laravel-dependency-security-policy": "resources/laravel-dependency-security-policy.json"
         },

src/Support/CacheThrottle.php

@@ -0,0 +1,62 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Workflow\Support;
+
+use Illuminate\Cache\DatabaseStore;
+use Illuminate\Database\SqlServerConnection;
+use Illuminate\Support\Carbon;
+use Illuminate\Support\Facades\Cache;
+
+final class CacheThrottle
+{
+    public static function acquire(string $key, string|bool $value, int $seconds, ?string $activeMarker = null): bool
+    {
+        $cache = Cache::store();
+        $store = $cache->getStore();
+        if (! $store instanceof DatabaseStore || $store->getConnection() instanceof SqlServerConnection) {
+            return ($activeMarker === null || ! $cache->has($activeMarker))
+                && $cache->add($key, $value, $seconds);
+        }
+
+        // DatabaseStore::add deletes expired rows before inserting. Renew this
+        // throttle in place so competing queue polls do not delete the same row.
+        $table = config('cache.stores.' . Cache::getDefaultDriver() . '.table', 'cache');
+        $prefixedKey = $store->getPrefix() . $key;
+        $query = $store->getConnection()
+            ->table($table)
+            ->useWritePdo()
+            ->where('key', $prefixedKey);
+        $now = Carbon::now()->getTimestamp();
+        if ($activeMarker !== null) {
+            $markerExpiration = $store->getConnection()
+                ->table($table)
+                ->useWritePdo()
+                ->where('key', $store->getPrefix() . $activeMarker)->value('expiration');
+            if ($markerExpiration !== null && $markerExpiration >= $now) {
+                // A wake just before chain expiry must not delay its replacement.
+                $seconds = min($seconds, max(1, (int) $markerExpiration - $now));
+            }
+        }
+        $expiration = $query->value('expiration');
+        $values = [
+            'value' => serialize($value),
+            'expiration' => $now + $seconds,
+        ];
+        if ($expiration === null) {
+            return $store->getConnection()
+                ->table($table)
+                ->insertOrIgnore([
+                    'key' => $prefixedKey,
+                    ...$values,
+                ]) === 1;
+        }
+        if ($expiration > $now) {
+            return false;
+        }
+
+        return $query->where('expiration', '<=', $now)
+            ->update($values) === 1;
+    }
+}

src/V2/TaskWatchdog.php

@@ -8,6 +8,7 @@
 use Illuminate\Support\Facades\DB;
 use Illuminate\Support\Facades\Schema;
 use Throwable;
+use Workflow\Support\CacheThrottle;
 use Workflow\V2\Contracts\HistoryProjectionRole;
 use Workflow\V2\Contracts\WorkflowTaskRepairProjectionRole;
 use Workflow\V2\Enums\RunStatus;
@@ -97,7 +98,7 @@ public static function runPass(
         }
 
         if ($respectThrottle) {
-            if (! Cache::add(self::LOOP_THROTTLE_KEY, true, TaskRepairPolicy::loopThrottleSeconds())) {
+            if (! CacheThrottle::acquire(self::LOOP_THROTTLE_KEY, true, TaskRepairPolicy::loopThrottleSeconds())) {
                 return self::emptyReport(
                     $connection,
                     $queue,

src/Watchdog.php

@@ -18,6 +18,7 @@
 use Illuminate\Support\Str;
 use Workflow\Models\StoredWorkflow;
 use Workflow\States\WorkflowPendingStatus;
+use Workflow\Support\CacheThrottle;
 
 class Watchdog implements ShouldBeEncrypted, ShouldQueue
 {
@@ -60,16 +61,21 @@ public static function wake(string $connection, ?string $queue = null): void
         DB::afterCommit(static function () use ($connection, $queue, $timeout): void {
             $generation = self::newGeneration();
 
+            if (! CacheThrottle::acquire(
+                self::LOOP_THROTTLE_KEY,
+                $generation,
+                self::bootstrapWindow($timeout),
+                self::CACHE_KEY,
+            ) || Cache::has(self::CACHE_KEY)) {
+                return;
+            }
+
             $watchdog = Cache::lock(self::CHAIN_LOCK_KEY, self::leaseDuration($timeout))
                 ->get(static function () use ($connection, $queue, $timeout, $generation): ?self {
                     if (! self::storedWorkflowTableExists() || Cache::has(self::CACHE_KEY)) {
                         return null;
                     }
 
-                    if (! Cache::add(self::LOOP_THROTTLE_KEY, $generation, self::bootstrapWindow($timeout))) {
-                        return null;
-                    }
-
                     if (! self::hasRecoverablePendingWorkflows($timeout) || Cache::has(self::CACHE_KEY)) {
                         return null;
                     }