Fix: Unique lock not being released after transaction rollback in ShouldBeUnique jobs with afterCommit() (#55420)

toshitsuna-otsuka · web-flow · commit 1f184bb69948 · 2025-04-18T11:44:40.000-04:00
* add a callback after transaction rollback

* for unique jobs, release the UniqueLock when a transaction rollback

* fix code style
diff --git a/src/Illuminate/Database/DatabaseTransactionRecord.php b/src/Illuminate/Database/DatabaseTransactionRecord.php
@@ -32,6 +32,13 @@ class DatabaseTransactionRecord
      */
     protected $callbacks = [];
 
+    /**
+     * The callbacks that should be executed after rollback.
+     *
+     * @var array
+     */
+    protected $callbacksForRollback = [];
+
     /**
      * Create a new database transaction record instance.
      *
@@ -57,6 +64,17 @@ public function addCallback($callback)
         $this->callbacks[] = $callback;
     }
 
+    /**
+     * Register a callback to be executed after rollback.
+     *
+     * @param  callable  $callback
+     * @return void
+     */
+    public function addCallbackForRollback($callback)
+    {
+        $this->callbacksForRollback[] = $callback;
+    }
+
     /**
      * Execute all of the callbacks.
      *
@@ -69,6 +87,18 @@ public function executeCallbacks()
         }
     }
 
+    /**
+     * Execute all of the callbacks for rollback.
+     *
+     * @return void
+     */
+    public function executeCallbacksForRollback()
+    {
+        foreach ($this->callbacksForRollback as $callback) {
+            $callback();
+        }
+    }
+
     /**
      * Get all of the callbacks.
      *
@@ -78,4 +108,14 @@ public function getCallbacks()
     {
         return $this->callbacks;
     }
+
+    /**
+     * Get all of the callbacks for rollback.
+     *
+     * @return array
+     */
+    public function getCallbacksForRollback()
+    {
+        return $this->callbacksForRollback;
+    }
 }
diff --git a/src/Illuminate/Database/DatabaseTransactionsManager.php b/src/Illuminate/Database/DatabaseTransactionsManager.php
@@ -139,6 +139,8 @@ public function rollback($connection, $newTransactionLevel)
                 do {
                     $this->removeCommittedTransactionsThatAreChildrenOf($this->currentTransaction[$connection]);
 
+                    $this->currentTransaction[$connection]->executeCallbacksForRollback();
+
                     $this->currentTransaction[$connection] = $this->currentTransaction[$connection]->parent;
                 } while (
                     isset($this->currentTransaction[$connection]) &&
@@ -156,6 +158,12 @@ public function rollback($connection, $newTransactionLevel)
      */
     protected function removeAllTransactionsForConnection($connection)
     {
+        if ($this->currentTransaction) {
+            for ($currentTransaction = $this->currentTransaction[$connection]; isset($currentTransaction); $currentTransaction = $currentTransaction->parent) {
+                $currentTransaction->executeCallbacksForRollback();
+            }
+        }
+
         $this->currentTransaction[$connection] = null;
 
         $this->pendingTransactions = $this->pendingTransactions->reject(
@@ -203,6 +211,19 @@ public function addCallback($callback)
         $callback();
     }
 
+    /**
+     * Register a callback for transaction rollback.
+     *
+     * @param  callable  $callback
+     * @return void
+     */
+    public function addCallbackForRollback($callback)
+    {
+        if ($current = $this->callbackApplicableTransactions()->last()) {
+            return $current->addCallbackForRollback($callback);
+        }
+    }
+
     /**
      * Get the transactions that are applicable to callbacks.
      *
diff --git a/src/Illuminate/Queue/Queue.php b/src/Illuminate/Queue/Queue.php
@@ -4,9 +4,12 @@
 
 use Closure;
 use DateTimeInterface;
+use Illuminate\Bus\UniqueLock;
 use Illuminate\Container\Container;
+use Illuminate\Contracts\Cache\Repository as Cache;
 use Illuminate\Contracts\Encryption\Encrypter;
 use Illuminate\Contracts\Queue\ShouldBeEncrypted;
+use Illuminate\Contracts\Queue\ShouldBeUnique;
 use Illuminate\Contracts\Queue\ShouldQueueAfterCommit;
 use Illuminate\Queue\Events\JobQueued;
 use Illuminate\Queue\Events\JobQueueing;
@@ -324,6 +327,14 @@ protected function enqueueUsing($job, $payload, $queue, $delay, $callback)
     {
         if ($this->shouldDispatchAfterCommit($job) &&
             $this->container->bound('db.transactions')) {
+            if ($job instanceof ShouldBeUnique) {
+                $this->container->make('db.transactions')->addCallbackForRollback(
+                    function () use ($job) {
+                        (new UniqueLock($this->container->make(Cache::class)))->release($job);
+                    }
+                );
+            }
+
             return $this->container->make('db.transactions')->addCallback(
                 function () use ($queue, $job, $payload, $delay, $callback) {
                     $this->raiseJobQueueingEvent($queue, $job, $payload, $delay);
diff --git a/src/Illuminate/Queue/SyncQueue.php b/src/Illuminate/Queue/SyncQueue.php
@@ -2,8 +2,11 @@
 
 namespace Illuminate\Queue;
 
+use Illuminate\Bus\UniqueLock;
+use Illuminate\Contracts\Cache\Repository as Cache;
 use Illuminate\Contracts\Queue\Job;
 use Illuminate\Contracts\Queue\Queue as QueueContract;
+use Illuminate\Contracts\Queue\ShouldBeUnique;
 use Illuminate\Queue\Events\JobExceptionOccurred;
 use Illuminate\Queue\Events\JobProcessed;
 use Illuminate\Queue\Events\JobProcessing;
@@ -47,6 +50,14 @@ public function push($job, $data = '', $queue = null)
     {
         if ($this->shouldDispatchAfterCommit($job) &&
             $this->container->bound('db.transactions')) {
+            if ($job instanceof ShouldBeUnique) {
+                $this->container->make('db.transactions')->addCallbackForRollback(
+                    function () use ($job) {
+                        (new UniqueLock($this->container->make(Cache::class)))->release($job);
+                    }
+                );
+            }
+
             return $this->container->make('db.transactions')->addCallback(
                 fn () => $this->executeJob($job, $data, $queue)
             );
diff --git a/tests/Database/DatabaseTransactionsManagerTest.php b/tests/Database/DatabaseTransactionsManagerTest.php
@@ -177,6 +177,96 @@ public function testCallbackIsExecutedIfNoTransactions()
         $this->assertEquals(['default', 1], $callbacks[0]);
     }
 
+    public function testCallbacksForRollbackAreAddedToTheCurrentTransaction()
+    {
+        $callbacks = [];
+
+        $manager = (new DatabaseTransactionsManager);
+
+        $manager->begin('default', 1);
+
+        $manager->addCallbackForRollback(function () use (&$callbacks) {
+        });
+
+        $manager->begin('default', 2);
+
+        $manager->begin('admin', 1);
+
+        $manager->addCallbackForRollback(function () use (&$callbacks) {
+        });
+
+        $this->assertCount(1, $manager->getPendingTransactions()[0]->getCallbacksForRollback());
+        $this->assertCount(0, $manager->getPendingTransactions()[1]->getCallbacksForRollback());
+        $this->assertCount(1, $manager->getPendingTransactions()[2]->getCallbacksForRollback());
+    }
+
+    public function testRollbackTransactionsExecutesCallbacks()
+    {
+        $callbacks = [];
+
+        $manager = (new DatabaseTransactionsManager);
+
+        $manager->begin('default', 1);
+
+        $manager->addCallbackForRollback(function () use (&$callbacks) {
+            $callbacks[] = ['default', 1];
+        });
+
+        $manager->begin('default', 2);
+
+        $manager->addCallbackForRollback(function () use (&$callbacks) {
+            $callbacks[] = ['default', 2];
+        });
+
+        $manager->begin('admin', 1);
+
+        $manager->rollback('default', 1);
+        $manager->rollback('default', 0);
+
+        $this->assertCount(2, $callbacks);
+        $this->assertEquals(['default', 2], $callbacks[0]);
+        $this->assertEquals(['default', 1], $callbacks[1]);
+    }
+
+    public function testRollbackExecutesOnlyCallbacksOfTheConnection()
+    {
+        $callbacks = [];
+
+        $manager = (new DatabaseTransactionsManager);
+
+        $manager->begin('default', 1);
+
+        $manager->addCallbackForRollback(function () use (&$callbacks) {
+            $callbacks[] = ['default', 1];
+        });
+
+        $manager->begin('default', 2);
+        $manager->begin('admin', 1);
+
+        $manager->addCallbackForRollback(function () use (&$callbacks) {
+            $callbacks[] = ['admin', 1];
+        });
+
+        $manager->rollback('default', 1);
+        $manager->rollback('default', 0);
+
+        $this->assertCount(1, $callbacks);
+        $this->assertEquals(['default', 1], $callbacks[0]);
+    }
+
+    public function testCallbackForRollbackIsNotExecutedIfNoTransactions()
+    {
+        $callbacks = [];
+
+        $manager = (new DatabaseTransactionsManager);
+
+        $manager->addCallbackForRollback(function () use (&$callbacks) {
+            $callbacks[] = ['default', 1];
+        });
+
+        $this->assertCount(0, $callbacks);
+    }
+
     public function testStageTransactions()
     {
         $manager = (new DatabaseTransactionsManager);
diff --git a/tests/Integration/Queue/QueueConnectionTest.php b/tests/Integration/Queue/QueueConnectionTest.php
@@ -3,6 +3,7 @@
 namespace Illuminate\Tests\Integration\Queue;
 
 use Illuminate\Bus\Queueable;
+use Illuminate\Contracts\Queue\ShouldBeUnique;
 use Illuminate\Contracts\Queue\ShouldQueue;
 use Illuminate\Database\DatabaseTransactionsManager;
 use Illuminate\Foundation\Bus\Dispatchable;
@@ -19,6 +20,7 @@ class QueueConnectionTest extends TestCase
     protected function tearDown(): void
     {
         QueueConnectionTestJob::$ran = false;
+        QueueConnectionTestUniqueJob::$ran = false;
 
         parent::tearDown();
     }
@@ -28,6 +30,7 @@ public function testJobWontGetDispatchedInsideATransaction()
         $this->app->singleton('db.transactions', function () {
             $transactionManager = m::mock(DatabaseTransactionsManager::class);
             $transactionManager->shouldReceive('addCallback')->once()->andReturn(null);
+            $transactionManager->shouldNotReceive('addCallbackForRollback');
 
             return $transactionManager;
         });
@@ -40,6 +43,7 @@ public function testJobWillGetDispatchedInsideATransactionWhenExplicitlyIndicate
         $this->app->singleton('db.transactions', function () {
             $transactionManager = m::mock(DatabaseTransactionsManager::class);
             $transactionManager->shouldNotReceive('addCallback')->andReturn(null);
+            $transactionManager->shouldNotReceive('addCallbackForRollback');
 
             return $transactionManager;
         });
@@ -58,6 +62,7 @@ public function testJobWontGetDispatchedInsideATransactionWhenExplicitlyIndicate
         $this->app->singleton('db.transactions', function () {
             $transactionManager = m::mock(DatabaseTransactionsManager::class);
             $transactionManager->shouldReceive('addCallback')->once()->andReturn(null);
+            $transactionManager->shouldNotReceive('addCallbackForRollback');
 
             return $transactionManager;
         });
@@ -68,6 +73,55 @@ public function testJobWontGetDispatchedInsideATransactionWhenExplicitlyIndicate
             // This job was dispatched
         }
     }
+
+    public function testUniqueJobWontGetDispatchedInsideATransaction()
+    {
+        $this->app->singleton('db.transactions', function () {
+            $transactionManager = m::mock(DatabaseTransactionsManager::class);
+            $transactionManager->shouldReceive('addCallback')->once()->andReturn(null);
+            $transactionManager->shouldReceive('addCallbackForRollback')->once()->andReturn(null);
+
+            return $transactionManager;
+        });
+
+        Bus::dispatch(new QueueConnectionTestUniqueJob);
+    }
+
+    public function testUniqueJobWillGetDispatchedInsideATransactionWhenExplicitlyIndicated()
+    {
+        $this->app->singleton('db.transactions', function () {
+            $transactionManager = m::mock(DatabaseTransactionsManager::class);
+            $transactionManager->shouldNotReceive('addCallback')->andReturn(null);
+            $transactionManager->shouldNotReceive('addCallbackForRollback')->andReturn(null);
+
+            return $transactionManager;
+        });
+
+        try {
+            Bus::dispatch((new QueueConnectionTestUniqueJob)->beforeCommit());
+        } catch (Throwable) {
+            // This job was dispatched
+        }
+    }
+
+    public function testUniqueJobWontGetDispatchedInsideATransactionWhenExplicitlyIndicated()
+    {
+        $this->app['config']->set('queue.connections.sqs.after_commit', false);
+
+        $this->app->singleton('db.transactions', function () {
+            $transactionManager = m::mock(DatabaseTransactionsManager::class);
+            $transactionManager->shouldReceive('addCallback')->once()->andReturn(null);
+            $transactionManager->shouldReceive('addCallbackForRollback')->once()->andReturn(null);
+
+            return $transactionManager;
+        });
+
+        try {
+            Bus::dispatch((new QueueConnectionTestUniqueJob)->afterCommit());
+        } catch (SqsException) {
+            // This job was dispatched
+        }
+    }
 }
 
 class QueueConnectionTestJob implements ShouldQueue
@@ -81,3 +135,15 @@ public function handle()
         static::$ran = true;
     }
 }
+
+class QueueConnectionTestUniqueJob implements ShouldQueue, ShouldBeUnique
+{
+    use Dispatchable, Queueable;
+
+    public static $ran = false;
+
+    public function handle()
+    {
+        static::$ran = true;
+    }
+}
diff --git a/tests/Queue/QueueSyncQueueTest.php b/tests/Queue/QueueSyncQueueTest.php
