Merge pull request #813 from durable-workflow/release/2.5.25-admission

Release 2.5.25 bounded SQLite task admission

Author: rmcdaniel

.github/pr-test-selection.json

@@ -12,6 +12,7 @@
     "tests/Unit/V2/PlatformProtocolSpecsTest.php",
     "tests/Unit/V2/ReadmeBranchDestinationsTest.php",
     "tests/Unit/V2/ScopedCancellationContextTest.php",
+    "tests/Unit/V2/SqliteTransactionTest.php",
     "tests/Unit/V2/StartOptionsTest.php",
     "tests/Unit/V2/VersionMarkerWireFormatTest.php",
     "tests/Unit/V2/WorkflowFiberRunnerTest.php",

.github/queue-contention/README.md

@@ -39,6 +39,11 @@ expected for an unsuccessful control. Completion with repairs or failed
 deliveries also fails the clean acceptance check. Do not reinterpret that as a
 failed workflow if the result says it completed.
 
+Set `DW_PROBE_REPAIR_OBSERVATIONS=true` for bounded task and native-row metadata
+before repair updates, at most eight observations per process. Records can
+precede a rolled-back transaction. The final repair count reports committed
+effects. These diagnostic reads do not grant claim or execution authority.
+
 `business_invariants_pass` reports exact typed completion, callback/attempt
 counts and preserved original timing/history. `clean_acceptance_pass` also
 requires zero repairs, failed deliveries and leftover jobs. Both remain visible.

.github/queue-contention/runner.php

@@ -199,8 +199,32 @@ protected function defineEnvironment($app): void
 $diagnostics = (object) [
     'rollback' => null,
     'errors' => 0,
+    'repairs' => 0,
 ];
 $deliveryErrorPath = $directory . '/delivery-errors.jsonl';
+$repairObservationPath = $directory . '/repair-observations.jsonl';
+if (getenv('DW_PROBE_REPAIR_OBSERVATIONS') === 'true') {
+    WorkflowTask::updating(static function (WorkflowTask $task) use ($diagnostics, $repairObservationPath): void {
+        if (! $task->isDirty('repair_count') || ++$diagnostics->repairs > 8) {
+            return;
+        }
+        $reference = $task->queue_delivery_reference;
+        $row = is_array($reference) && isset($reference['job_id'])
+            ? DB::table('jobs')->useWritePdo()->where('id', $reference['job_id'])
+                ->first(['id', 'reserved_at', 'attempts', 'available_at'])
+            : null;
+        file_put_contents($repairObservationPath, json_encode([
+            'stage' => 'before_update',
+            'pid' => getmypid(),
+            'task_id' => $task->id,
+            'status' => $task->status->value,
+            'repair_before' => $task->getOriginal('repair_count'),
+            'repair_after' => $task->repair_count,
+            'native_row' => $row,
+            'observed_unix' => time(),
+        ], JSON_THROW_ON_ERROR) . "\n", FILE_APPEND | LOCK_EX);
+    });
+}
 $app['events']->listen(Illuminate\Queue\Events\JobProcessing::class, static function () use ($diagnostics): void {
     $diagnostics->rollback = null;
 });
@@ -330,6 +354,12 @@ protected function defineEnvironment($app): void
     'ready_jobs' => DB::table('jobs')->whereNull('reserved_at')->where('available_at', '<=', time())->count(),
     'reserved_jobs' => DB::table('jobs')->whereNotNull('reserved_at')->count(),
     'failed_jobs' => DB::table('failed_jobs')->count(),
+    'repair_observations' => is_file($repairObservationPath)
+        ? array_map(
+            static fn (string $line): array => json_decode($line, true, flags: JSON_THROW_ON_ERROR),
+            file($repairObservationPath, FILE_IGNORE_NEW_LINES | FILE_SKIP_EMPTY_LINES)
+        )
+        : [],
     'failures' => $failures,
     'callbacks' => count($callbacks),
     'callback_members' => array_count_values(

.github/workflows/php.yml

@@ -335,6 +335,7 @@ jobs:
       run: |
         mkdir -p build/test-results
         vendor/bin/phpunit tests/Unit/V2/DatabaseQueueRecoveryTest.php \
+          tests/Unit/V2/EmbeddedSqliteTaskContentionTest.php \
           --fail-on-warning --log-junit build/test-results/pr-database-queue-recovery.xml
 
     - name: Resolve bounded pull-request test selection

CHANGELOG.md

@@ -1,5 +1,15 @@
 # Changelog
 
+## 2.5.25 - 2026-10-16
+
+- Allow additional proven SQLite root rollbacks during task write admission
+  within the existing retry window, before the claim or execution body starts.
+- Preserve explicit attempt limits, ordinary mutation retries, other database
+  drivers and uncertain commit/rollback handling. Activity callbacks are not
+  repeated to publish their results.
+- Keep original tasks, histories, retry policies and deadlines. Native contention
+  still propagates when the bounded admission budget is exhausted.
+
 ## 2.5.24 - 2026-10-16
 
 - Avoid redundant embedded recovery dispatches while an exact unreserved

composer.json

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

src/V2/Support/SqliteTransaction.php

@@ -47,7 +47,10 @@ public static function run(Closure $operation, ?int $attempts = null, ?Closure $
         }
 
         $deadline = hrtime(true) + 5_000_000_000;
-        $maximumAttempts = max(1, $attempts ?? 3);
+        // Before admission, no execution body has run. Keep retrying proven
+        // rollbacks within the existing window instead of exhausting three
+        // fast lock refusals. Preserve ordinary mutation and explicit limits.
+        $maximumAttempts = max(1, $attempts ?? ($admission === null ? 3 : 64));
         for ($attempt = 1; ; $attempt++) {
             $failedInRoot = false;
             $returned = false;
@@ -89,7 +92,7 @@ public static function run(Closure $operation, ?int $attempts = null, ?Closure $
                     throw $exception;
                 }
 
-                usleep($attempt * 20_000);
+                usleep($admission === null ? $attempt * 20_000 : min($attempt * 20_000, 100_000));
                 if (hrtime(true) >= $deadline) {
                     throw $exception;
                 }

tests/Unit/V2/EmbeddedSqliteTaskContentionTest.php

@@ -65,6 +65,10 @@ public static function phases(): array
         foreach ($phases as $name => [$phase]) {
             $cases[$name] = [$phase, false];
             $cases[$name . ' with separate application database'] = [$phase, true];
+            if (! in_array($phase, ['start', 'signal'], true)) {
+                $cases[$name . ' after three rollbacks'] = [$phase, false, 3];
+                $cases[$name . ' after three rollbacks with separate application database'] = [$phase, true, 3];
+            }
         }
 
         return $cases;
@@ -177,8 +181,11 @@ static function (QueryExecuted $query) use (
     }
 
     #[DataProvider('phases')]
-    public function testOriginalRunCompletesOnceAfterRealWalWrite(string $phase, bool $separate): void
-    {
+    public function testOriginalRunCompletesOnceAfterRealWalWrite(
+        string $phase,
+        bool $separate,
+        int $releaseAfter = 1
+    ): void {
         if ($separate) {
             config()->set('database.connections.application_sqlite', [
                 'driver' => 'sqlite',
@@ -199,9 +206,10 @@ public function testOriginalRunCompletesOnceAfterRealWalWrite(string $phase, boo
         $armed = $phase === 'start';
         $injected = false;
         $admissions = 0;
+        $rollbacks = 0;
         $connection->getEventDispatcher()
-            ->listen(TransactionRolledBack::class, static function () use ($writer): void {
-                if ($writer->inTransaction()) {
+            ->listen(TransactionRolledBack::class, static function () use ($writer, $releaseAfter, &$rollbacks): void {
+                if ($writer->inTransaction() && ++$rollbacks >= $releaseAfter) {
                     $writer->commit();
                 }
             });

tests/Unit/V2/SqliteTransactionTest.php

@@ -415,6 +415,39 @@ public function testWriteAdmissionRetriesBeforeEnteringBody(): void
         $this->assertSame(1, (int) $this->connection->selectOne('SELECT value FROM state WHERE id=1')->value);
     }
 
+    public function testWriteAdmissionCanRecoverAfterThreeProvenRootRollbacks(): void
+    {
+        $writer = $this->writer();
+        $writer->beginTransaction();
+        $writer->exec('UPDATE state SET value=1 WHERE id=2');
+        $rollbacks = 0;
+        $this->events->listen(TransactionRolledBack::class, function () use ($writer, &$rollbacks): void {
+            $rollbacks++;
+            $this->assertSame(0, $this->connection->transactionLevel());
+            $this->assertFalse($this->connection->getPdo()->inTransaction());
+            if ($rollbacks === 3) {
+                $writer->commit();
+            }
+        });
+        $calls = 0;
+        try {
+            SqliteTransaction::run(function () use (&$calls): void {
+                $calls++;
+                $this->connection->update('UPDATE state SET value=value+1 WHERE id=1');
+            }, admission: function (): void {
+                $this->connection->update('UPDATE state SET id=id WHERE id=1');
+            });
+        } finally {
+            if ($writer->inTransaction()) {
+                $writer->rollBack();
+            }
+        }
+        $this->assertSame(3, $rollbacks);
+        $this->assertSame(1, $calls);
+        $this->assertSame(1, (int) $this->connection->selectOne('SELECT value FROM state WHERE id=1')->value);
+        $this->assertSame(1, (int) $this->connection->selectOne('SELECT value FROM state WHERE id=2')->value);
+    }
+
     public function testExhaustedAdmissionNeverEntersBody(): void
     {
         $writer = $this->writer();
@@ -425,7 +458,7 @@ public function testExhaustedAdmissionNeverEntersBody(): void
         try {
             SqliteTransaction::run(static function () use (&$calls): void {
                 $calls++;
-            }, admission: function () use (&$admissions): void {
+            }, 3, admission: function () use (&$admissions): void {
                 $admissions++;
                 $this->connection->update('UPDATE state SET id=id WHERE id=1');
             });
@@ -441,6 +474,33 @@ public function testExhaustedAdmissionNeverEntersBody(): void
         $this->assertFalse($this->connection->getPdo()->inTransaction());
     }
 
+    public function testDefaultAdmissionRemainsBoundedWithoutExecutingTheBody(): void
+    {
+        $writer = $this->writer();
+        $writer->beginTransaction();
+        $writer->exec('UPDATE state SET value=1 WHERE id=2');
+        $admissions = 0;
+        $calls = 0;
+        try {
+            SqliteTransaction::run(static function () use (&$calls): void {
+                $calls++;
+            }, admission: function () use (&$admissions): void {
+                $admissions++;
+                $this->connection->update('UPDATE state SET id=id WHERE id=1');
+            });
+            $this->fail('Persistent contention must keep its original exception.');
+        } catch (PDOException $exception) {
+            $this->assertSame(5, $exception->errorInfo[1] & 255);
+        } finally {
+            $writer->rollBack();
+        }
+        $this->assertGreaterThan(3, $admissions);
+        $this->assertLessThanOrEqual(64, $admissions);
+        $this->assertSame(0, $calls);
+        $this->assertSame(0, $this->connection->transactionLevel());
+        $this->assertFalse($this->connection->getPdo()->inTransaction());
+    }
+
     public function testAdmittedBodyDoesNotRetryItsOwnContention(): void
     {
         $writer = $this->writer();