Merge pull request #839 from durable-workflow/release/2.5.32-sqlite-write-waits

Release 2.5.32: preserve SQLite claim and acknowledgment retry budgets

Author: rmcdaniel

CHANGELOG.md

@@ -1,5 +1,14 @@
 # Changelog
 
+## 2.5.32 - 2026-10-17
+
+- Keep individual SQLite writer waits within the existing claim and queue
+  acknowledgment retry budgets, so a normal busy timeout leaves time to retry.
+- Restore the connection's configured wait before workflow code and after a
+  refused write. Preserve original reservation fences, claims and deadlines.
+- Retry only proven root rollbacks. Do not repeat handlers or activity callbacks,
+  or retry uncertain commit and rollback outcomes.
+
 ## 2.5.31 - 2026-10-17
 
 - Admit SQLite summary writes before projection reads and model hooks so proven

composer.json

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

src/Queue/SqliteAcknowledgingDatabaseQueue.php

@@ -4,11 +4,13 @@
 
 namespace Workflow\Queue;
 
+use Illuminate\Database\Query\Builder;
 use Illuminate\Database\QueryException;
 use Illuminate\Queue\DatabaseQueue;
 use PDO;
 use PDOException;
 use Throwable;
+use Workflow\Support\SqliteWriteWait;
 
 /** Keep Laravel's pop lifecycle while acknowledging only the original reservation. */
 class SqliteAcknowledgingDatabaseQueue extends DatabaseQueue
@@ -20,6 +22,8 @@ public function deleteClaimedJob(string $queue, int|string $id, int $reservedAt,
         $rootTransaction = $this->database->transactionLevel() === 0;
         $pdo = $this->database->getPdo();
         $deadline = hrtime(true) + 5_000_000_000;
+        $boundedWait = $rootTransaction && $pdo instanceof PDO
+            && $pdo->getAttribute(PDO::ATTR_DRIVER_NAME) === 'sqlite';
 
         for ($attempt = 0; $attempt < self::MAX_DELETE_ATTEMPTS; $attempt++) {
             $deleteFailure = null;
@@ -33,11 +37,17 @@ public function deleteClaimedJob(string $queue, int|string $id, int $reservedAt,
             $bindings = $query->getBindings();
 
             try {
-                $this->database->transaction(static function () use ($query, &$deleteFailure): void {
+                $this->database->transaction(static function () use (
+                    $query,
+                    &$deleteFailure,
+                    $boundedWait,
+                    $pdo,
+                    $deadline,
+                ): void {
                     try {
                         // One conditional write avoids SQLite's SELECT-to-write upgrade.
                         // A newer reservation cannot match the original claim.
-                        $query->delete();
+                        self::deleteWithBoundedWait($query, $pdo, $boundedWait, $deadline);
                     } catch (QueryException $exception) {
                         $deleteFailure = $exception;
                         throw $exception;
@@ -83,4 +93,15 @@ protected function marshalJob($queue, $job)
             $queue,
         );
     }
+
+    private static function deleteWithBoundedWait(Builder $query, PDO $pdo, bool $boundedWait, int $deadline): void
+    {
+        if ($boundedWait) {
+            SqliteWriteWait::run($pdo, static function () use ($query): void {
+                $query->delete();
+            }, $deadline);
+            return;
+        }
+        $query->delete();
+    }
 }

src/Support/SqliteWriteWait.php

@@ -0,0 +1,39 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Workflow\Support;
+
+use Closure;
+use PDO;
+
+/** @internal The caller owns transaction proof, write identity and retry policy. */
+final class SqliteWriteWait
+{
+    /**
+     * @param Closure(): void $write
+     */
+    public static function run(PDO $pdo, Closure $write, int $deadline): void
+    {
+        $statement = $pdo->query('PRAGMA busy_timeout');
+        if ($statement === false) {
+            throw new SqliteWriteWaitException('Cannot inspect the SQLite admission busy timeout.');
+        }
+        $originalTimeout = (int) $statement->fetchColumn();
+        $statement->closeCursor();
+        $waitMilliseconds = min(100, max(1, (int) (($deadline - hrtime(true)) / 1_000_000)));
+        if ($originalTimeout <= $waitMilliseconds) {
+            $write();
+            return;
+        }
+
+        // A single driver wait must not consume the caller's retry budget.
+        // Restore the setting before application code or the next operation.
+        $pdo->exec('PRAGMA busy_timeout=' . $waitMilliseconds);
+        try {
+            $write();
+        } finally {
+            $pdo->exec('PRAGMA busy_timeout=' . $originalTimeout);
+        }
+    }
+}

src/Support/SqliteWriteWaitException.php

@@ -0,0 +1,12 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Workflow\Support;
+
+use RuntimeException;
+
+/** @internal A safe write cannot proceed without inspecting its driver wait. */
+final class SqliteWriteWaitException extends RuntimeException
+{
+}

src/V2/Support/SqliteTransaction.php

@@ -11,6 +11,7 @@
 use PDO;
 use PDOException;
 use Throwable;
+use Workflow\Support\SqliteWriteWait;
 
 final class SqliteTransaction
 {
@@ -49,7 +50,7 @@ public static function run(
 
         [$maximumAttempts, $deadline] = self::retryLimits($attempts, $admission !== null, $admissionDeadline);
         for ($attempt = 1; ; $attempt++) {
-            $outcome = self::attempt($connection, $pdo, $operation, $admission);
+            $outcome = self::attempt($connection, $pdo, $operation, $admission, $deadline);
             if (! isset($outcome['error'])) {
                 return $outcome['value'];
             }
@@ -124,8 +125,13 @@ private static function retryLimits(?int $attempts, bool $admission, ?int $admis
      * @param Closure(Connection): void|null $admission
      * @return array{value: T}|array{error: Throwable, retry: bool}
      */
-    private static function attempt(Connection $connection, PDO $pdo, Closure $operation, ?Closure $admission): array
-    {
+    private static function attempt(
+        Connection $connection,
+        PDO $pdo,
+        Closure $operation,
+        ?Closure $admission,
+        int $deadline,
+    ): array {
         $failedInRoot = false;
         $returned = false;
         $entered = false;
@@ -135,14 +141,17 @@ private static function attempt(Connection $connection, PDO $pdo, Closure $opera
                 $operation,
                 $admission,
                 $pdo,
+                $deadline,
                 &$failedInRoot,
                 &$returned,
                 &$entered,
                 &$mutationError
             ): mixed {
                 try {
                     if ($admission !== null) {
-                        $admission($database);
+                        SqliteWriteWait::run($pdo, static function () use ($admission, $database): void {
+                            $admission($database);
+                        }, $deadline);
                     }
                     $entered = true;
                     $result = $operation($database);

tests/Unit/Queue/SqliteDatabaseAcknowledgementTest.php

@@ -123,6 +123,44 @@ public function testSustainedBusyStopsWithinTheRetryWindowWithTheOriginalReserva
         self::assertEquals($original, $this->connection->table('jobs')->find($job->getJobId()));
     }
 
+    #[DataProvider('writerReleasePoints')]
+    public function testNormalBusyTimeoutLeavesTimeForFencedAcknowledgementRetries(int $releaseAfter): void
+    {
+        $pdo = $this->connection->getPdo();
+        $pdo->exec('PRAGMA busy_timeout=5000');
+        $job = $this->job();
+        $neighbor = $this->push();
+        $neighborBefore = $this->connection->table('jobs')
+            ->find($neighbor);
+        $writer = $this->blockDelete();
+        $rollbacks = 0;
+        $this->events->listen(TransactionRolledBack::class, function () use (
+            $writer,
+            $releaseAfter,
+            &$rollbacks
+        ): void {
+            self::assertSame(1, AcknowledgementReceiptJob::$handled);
+            self::assertSame(0, $this->connection->transactionLevel());
+            self::assertFalse($this->connection->getPdo()->inTransaction());
+            if (++$rollbacks === $releaseAfter) {
+                $writer->rollBack();
+            }
+        });
+        try {
+            $job->fire();
+        } finally {
+            if ($writer->inTransaction()) {
+                $writer->rollBack();
+            }
+        }
+        self::assertSame($releaseAfter, $rollbacks);
+        self::assertSame($releaseAfter + 1, $this->deleteStatements);
+        self::assertSame(1, AcknowledgementReceiptJob::$handled);
+        self::assertNull($this->connection->table('jobs')->find($job->getJobId()));
+        self::assertEquals($neighborBefore, $this->connection->table('jobs')->find($neighbor));
+        self::assertSame(5000, (int) $pdo->query('PRAGMA busy_timeout')->fetchColumn());
+    }
+
     #[DataProvider('writerReleasePoints')]
     public function testAReclaimedReservationSurvivesTheOriginalAcknowledgementRetry(int $releaseAfter): void
     {

tests/Unit/V2/EmbeddedSqliteTaskContentionTest.php

@@ -87,6 +87,109 @@ public static function claims(): array
         ];
     }
 
+    #[DataProvider('claims')]
+    public function testOrdinaryBusyTimeoutLeavesTimeToRetryTheUnclaimedDelivery(string $kind, bool $separate): void
+    {
+        config()->set('logging.default', 'null');
+        if ($separate) {
+            config()->set('database.connections.application_sqlite', [
+                'driver' => 'sqlite',
+                'database' => ':memory:',
+                'prefix' => '',
+                'foreign_key_constraints' => true,
+            ]);
+            DB::setDefaultConnection('application_sqlite');
+            $this->createQueueTables(DB::connection());
+            config()
+                ->set('queue.failed.database', 'application_sqlite');
+        }
+        $stub = WorkflowStub::make(SqliteContentionWorkflow::class, 'ordinary-busy-' . $kind, 'contention-test');
+        $stub->start(false, false, new StartOptions(executionTimeoutSeconds: 120, runTimeoutSeconds: 100));
+        if ($kind === 'activity') {
+            $this->workOnce();
+        }
+        $task = WorkflowTask::query()->where('workflow_run_id', $stub->runId())
+            ->where('task_type', $kind)
+            ->sole();
+        $run = $task->run;
+        $deadlines = [$run->execution_deadline_at->toJSON(), $run->run_deadline_at->toJSON()];
+        $prefix = WorkflowHistoryEvent::query()->where('workflow_run_id', $run->id)->orderBy('sequence')
+            ->get()
+            ->map->getAttributes()
+            ->all();
+        $connection = DB::connection('contention_sqlite');
+        $connection->getPdo()
+            ->exec('PRAGMA busy_timeout=5000');
+        $writer = new PDO('sqlite:' . $this->path);
+        $writer->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);
+        $injected = false;
+        $rollbacks = 0;
+        $connection->beforeExecuting(static function (string $sql) use ($writer, &$injected): void {
+            if ($injected || ! str_starts_with($sql, 'update "workflow_tasks" set "id" =')) {
+                return;
+            }
+            $writer->beginTransaction();
+            $writer->exec('INSERT INTO contention_probe (id) VALUES (1)');
+            $injected = true;
+        });
+        $connection->getEventDispatcher()
+            ->listen(TransactionRolledBack::class, static function ($event) use (
+                $connection,
+                $writer,
+                &$rollbacks
+            ): void {
+                if ($event->connection === $connection && $writer->inTransaction()) {
+                    self::assertSame(0, $connection->transactionLevel());
+                    self::assertFalse($connection->getPdo()->inTransaction());
+                    self::assertSame(0, SqliteContentionActivity::$invocations);
+                    $rollbacks++;
+                    $writer->commit();
+                }
+            });
+        try {
+            $this->workOnce();
+        } finally {
+            if ($writer->inTransaction()) {
+                $writer->rollBack();
+            }
+        }
+        self::assertTrue($injected);
+        self::assertSame(1, $rollbacks);
+        self::assertSame(5000, (int) $connection->getPdo()->query('PRAGMA busy_timeout')->fetchColumn());
+        self::assertSame(
+            0,
+            DB::table('failed_jobs')->count(),
+            'A rolled-back pre-claim wait must leave time to retry.'
+        );
+        self::assertSame('completed', $task->fresh()->status->value);
+        self::assertSame(1, (int) $task->fresh()->attempt_count);
+        for ($remaining = 0; $remaining < 3 && $stub->refresh()->status() !== 'completed'; $remaining++) {
+            $this->workOnce();
+        }
+        self::assertSame('completed', $stub->refresh()->status());
+        self::assertSame([
+            'integer' => 9007199254740993,
+            'optional' => null,
+            'unicode' => 'λ',
+        ], $stub->output());
+        self::assertSame(1, SqliteContentionActivity::$invocations);
+        self::assertSame($deadlines, [
+            $run->fresh()
+                ->execution_deadline_at->toJSON(),
+            $run->fresh()
+                ->run_deadline_at->toJSON(),
+        ]);
+        self::assertSame($prefix, WorkflowHistoryEvent::query()->where('workflow_run_id', $run->id)
+            ->orderBy('sequence')
+            ->limit(count($prefix))
+            ->get()
+            ->map->getAttributes()
+            ->all());
+        self::assertSame(0, WorkflowTask::query()->where('workflow_run_id', $run->id)->sum('repair_count'));
+        self::assertSame(0, DB::table('jobs')->count());
+        self::assertSame(0, DB::table('failed_jobs')->count());
+    }
+
     public function testActivityOutcomeWaitsForTheWriterWithinItsOriginalClaim(): void
     {
         config()->set('logging.default', 'null');
@@ -109,6 +212,7 @@ public function testActivityOutcomeWaitsForTheWriterWithinItsOriginalClaim(): vo
         $admissions = 0;
         $rollbacks = 0;
         $claim = null;
+        $releaseAt = hrtime(true) + 5_100_000_000;
         $connection->beforeExecuting(static function (string $sql) use (
             $writer,
             $task,
@@ -130,13 +234,16 @@ public function testActivityOutcomeWaitsForTheWriterWithinItsOriginalClaim(): vo
             ->listen(TransactionRolledBack::class, static function ($event) use (
                 $writer,
                 $connection,
-                &$rollbacks
+                &$rollbacks,
+                $releaseAt,
             ): void {
                 if ($event->connection === $connection && $writer->inTransaction()) {
                     self::assertSame(0, $connection->transactionLevel());
                     self::assertFalse($connection->getPdo()->inTransaction());
                     $rollbacks++;
-                    $writer->commit();
+                    if (hrtime(true) >= $releaseAt) {
+                        $writer->commit();
+                    }
                 }
             });
         $started = hrtime(true);
@@ -149,7 +256,7 @@ public function testActivityOutcomeWaitsForTheWriterWithinItsOriginalClaim(): vo
         }
         $elapsed = (hrtime(true) - $started) / 1_000_000_000;
         self::assertGreaterThanOrEqual(5, $elapsed);
-        self::assertSame(1, $rollbacks);
+        self::assertGreaterThanOrEqual(1, $rollbacks);
         self::assertSame(1, SqliteContentionActivity::$invocations);
         self::assertIsArray($claim);
         self::assertSame('leased', $claim['status']);