From 0274490f33c74e23b58b6ecefc08bc12a6b02085 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Sat, 17 Oct 2026 20:21:20 +0000 Subject: [PATCH 1/6] test: reproduce exhausted SQLite claim admission with stock busy timeout --- .../V2/EmbeddedSqliteTaskContentionTest.php | 86 +++++++++++++++++++ 1 file changed, 86 insertions(+) diff --git a/tests/Unit/V2/EmbeddedSqliteTaskContentionTest.php b/tests/Unit/V2/EmbeddedSqliteTaskContentionTest.php index 23ece22e4..9229c72d3 100644 --- a/tests/Unit/V2/EmbeddedSqliteTaskContentionTest.php +++ b/tests/Unit/V2/EmbeddedSqliteTaskContentionTest.php @@ -87,6 +87,92 @@ public static function claims(): array ]; } + #[DataProvider('claims')] + public function testOrdinaryBusyTimeoutLeavesTimeToRetryTheUnclaimedDelivery(string $kind, bool $separate): void + { + 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'); From df64ff07baa618b512ffbf4163334e1d8427063b Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Sat, 17 Oct 2026 20:24:16 +0000 Subject: [PATCH 2/6] fix: preserve the SQLite claim admission budget for root retries --- src/V2/Support/SqliteTransaction.php | 41 ++++++++++-- .../V2/EmbeddedSqliteTaskContentionTest.php | 65 ++++++++++++------- tests/Unit/V2/SqliteTransactionTest.php | 43 ++++++++++++ 3 files changed, 121 insertions(+), 28 deletions(-) diff --git a/src/V2/Support/SqliteTransaction.php b/src/V2/Support/SqliteTransaction.php index eea6598c8..f5b65eab6 100644 --- a/src/V2/Support/SqliteTransaction.php +++ b/src/V2/Support/SqliteTransaction.php @@ -49,7 +49,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 +124,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,6 +140,7 @@ private static function attempt(Connection $connection, PDO $pdo, Closure $opera $operation, $admission, $pdo, + $deadline, &$failedInRoot, &$returned, &$entered, @@ -142,7 +148,7 @@ private static function attempt(Connection $connection, PDO $pdo, Closure $opera ): mixed { try { if ($admission !== null) { - $admission($database); + self::admit($database, $pdo, $admission, $deadline); } $entered = true; $result = $operation($database); @@ -173,6 +179,33 @@ private static function attempt(Connection $connection, PDO $pdo, Closure $opera } } + /** + * @param Closure(Connection): void $admission + */ + private static function admit(Connection $database, PDO $pdo, Closure $admission, int $deadline): void + { + $statement = $pdo->query('PRAGMA busy_timeout'); + if ($statement === false) { + throw new \RuntimeException('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) { + $admission($database); + return; + } + + // A single driver wait must not consume the root retry budget. Only the + // admission uses a short wait; application code keeps its original setting. + $pdo->exec('PRAGMA busy_timeout=' . $waitMilliseconds); + try { + $admission($database); + } finally { + $pdo->exec('PRAGMA busy_timeout=' . $originalTimeout); + } + } + private static function isContention(Throwable $exception): bool { do { diff --git a/tests/Unit/V2/EmbeddedSqliteTaskContentionTest.php b/tests/Unit/V2/EmbeddedSqliteTaskContentionTest.php index 9229c72d3..a765301e0 100644 --- a/tests/Unit/V2/EmbeddedSqliteTaskContentionTest.php +++ b/tests/Unit/V2/EmbeddedSqliteTaskContentionTest.php @@ -90,6 +90,7 @@ 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', @@ -99,7 +100,8 @@ public function testOrdinaryBusyTimeoutLeavesTimeToRetryTheUnclaimedDelivery(str ]); DB::setDefaultConnection('application_sqlite'); $this->createQueueTables(DB::connection()); - config()->set('queue.failed.database', 'application_sqlite'); + 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)); @@ -107,13 +109,17 @@ public function testOrdinaryBusyTimeoutLeavesTimeToRetryTheUnclaimedDelivery(str $this->workOnce(); } $task = WorkflowTask::query()->where('workflow_run_id', $stub->runId()) - ->where('task_type', $kind)->sole(); + ->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(); + ->get() + ->map->getAttributes() + ->all(); $connection = DB::connection('contention_sqlite'); - $connection->getPdo()->exec('PRAGMA busy_timeout=5000'); + $connection->getPdo() + ->exec('PRAGMA busy_timeout=5000'); $writer = new PDO('sqlite:' . $this->path); $writer->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION); $injected = false; @@ -126,19 +132,16 @@ public function testOrdinaryBusyTimeoutLeavesTimeToRetryTheUnclaimedDelivery(str $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(); - } - }); + $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 { @@ -149,7 +152,11 @@ public function testOrdinaryBusyTimeoutLeavesTimeToRetryTheUnclaimedDelivery(str 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( + 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++) { @@ -163,11 +170,17 @@ public function testOrdinaryBusyTimeoutLeavesTimeToRetryTheUnclaimedDelivery(str ], $stub->output()); self::assertSame(1, SqliteContentionActivity::$invocations); self::assertSame($deadlines, [ - $run->fresh()->execution_deadline_at->toJSON(), - $run->fresh()->run_deadline_at->toJSON(), + $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()); + ->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()); @@ -195,6 +208,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, @@ -216,13 +230,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); @@ -235,7 +252,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']); diff --git a/tests/Unit/V2/SqliteTransactionTest.php b/tests/Unit/V2/SqliteTransactionTest.php index 2c979d7e6..a5af51b74 100644 --- a/tests/Unit/V2/SqliteTransactionTest.php +++ b/tests/Unit/V2/SqliteTransactionTest.php @@ -126,6 +126,49 @@ public function connection(): never $this->assertFalse($this->connection->getPdo()->inTransaction()); } + public function testAdmissionRestoresTheOriginalBusyTimeoutBeforeTheBody(): void + { + $pdo = $this->connection->getPdo(); + $pdo->exec('PRAGMA busy_timeout=5000'); + $calls = 0; + SqliteTransaction::run(function () use ($pdo, &$calls): void { + self::assertSame(5000, (int) $pdo->query('PRAGMA busy_timeout')->fetchColumn()); + $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'); + }); + self::assertSame(1, $calls); + self::assertSame(5000, (int) $pdo->query('PRAGMA busy_timeout')->fetchColumn()); + self::assertSame(1, (int) $this->connection->selectOne('SELECT value FROM state WHERE id=1')->value); + } + + public function testRejectedAdmissionRestoresTheOriginalBusyTimeoutWithoutEnteringTheBody(): void + { + $pdo = $this->connection->getPdo(); + $pdo->exec('PRAGMA busy_timeout=5000'); + $writer = $this->writer(); + $writer->beginTransaction(); + $writer->exec('UPDATE state SET value=1 WHERE id=2'); + $calls = 0; + try { + SqliteTransaction::run(static function () use (&$calls): void { + $calls++; + }, 1, admission: function (): void { + $this->connection->update('UPDATE state SET id=id WHERE id=1'); + }); + self::fail('Persistent contention must keep the original admission failure.'); + } catch (PDOException $exception) { + self::assertSame(5, $exception->errorInfo[1] & 255); + } finally { + $writer->rollBack(); + } + self::assertSame(0, $calls); + self::assertSame(5000, (int) $pdo->query('PRAGMA busy_timeout')->fetchColumn()); + self::assertSame(0, $this->connection->transactionLevel()); + self::assertFalse($pdo->inTransaction()); + } + public function testConfiguredStorageMutationRollsBackItsOwnRoot(): void { $primary = $this->separateApplicationConnection(); From eb588b9a64793a1f52d5d43de064c98e5bbfea43 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Sat, 17 Oct 2026 20:28:52 +0000 Subject: [PATCH 3/6] fix: classify refused SQLite admission settings and finalize style --- .../Exceptions/SqliteAdmissionException.php | 11 ++++++ src/V2/Support/SqliteTransaction.php | 3 +- .../V2/EmbeddedSqliteTaskContentionTest.php | 6 +++- tests/Unit/V2/SqliteTransactionTest.php | 34 +++++++++++++++++++ 4 files changed, 52 insertions(+), 2 deletions(-) create mode 100644 src/V2/Exceptions/SqliteAdmissionException.php diff --git a/src/V2/Exceptions/SqliteAdmissionException.php b/src/V2/Exceptions/SqliteAdmissionException.php new file mode 100644 index 000000000..7f355cc78 --- /dev/null +++ b/src/V2/Exceptions/SqliteAdmissionException.php @@ -0,0 +1,11 @@ +query('PRAGMA busy_timeout'); if ($statement === false) { - throw new \RuntimeException('Cannot inspect the SQLite admission busy timeout.'); + throw new SqliteAdmissionException('Cannot inspect the SQLite admission busy timeout.'); } $originalTimeout = (int) $statement->fetchColumn(); $statement->closeCursor(); diff --git a/tests/Unit/V2/EmbeddedSqliteTaskContentionTest.php b/tests/Unit/V2/EmbeddedSqliteTaskContentionTest.php index a765301e0..627e8ecf0 100644 --- a/tests/Unit/V2/EmbeddedSqliteTaskContentionTest.php +++ b/tests/Unit/V2/EmbeddedSqliteTaskContentionTest.php @@ -133,7 +133,11 @@ public function testOrdinaryBusyTimeoutLeavesTimeToRetryTheUnclaimedDelivery(str $injected = true; }); $connection->getEventDispatcher() - ->listen(TransactionRolledBack::class, static function ($event) use ($connection, $writer, &$rollbacks,): void { + ->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()); diff --git a/tests/Unit/V2/SqliteTransactionTest.php b/tests/Unit/V2/SqliteTransactionTest.php index a5af51b74..55bf6c038 100644 --- a/tests/Unit/V2/SqliteTransactionTest.php +++ b/tests/Unit/V2/SqliteTransactionTest.php @@ -17,6 +17,7 @@ use PHPUnit\Framework\Attributes\DataProvider; use PHPUnit\Framework\TestCase; use Throwable; +use Workflow\V2\Exceptions\SqliteAdmissionException; use Workflow\V2\Support\SqliteTransaction; final class SqliteTransactionTest extends TestCase @@ -143,6 +144,39 @@ public function testAdmissionRestoresTheOriginalBusyTimeoutBeforeTheBody(): void self::assertSame(1, (int) $this->connection->selectOne('SELECT value FROM state WHERE id=1')->value); } + public function testUnavailableAdmissionSettingsRefuseBeforeAdmissionOrBody(): void + { + $pdo = new class('sqlite:' . $this->path) extends PDO { + public int $queries = 0; + + public function query(string $query, ?int $fetchMode = null, mixed ...$fetchModeArgs): \PDOStatement|false + { + $this->queries++; + return false; + } + }; + $connection = new SQLiteConnection($pdo, $this->path, '', [ + 'driver' => 'sqlite', + ]); + $calls = 0; + $admissions = 0; + try { + SqliteTransaction::run(static function () use (&$calls): void { + $calls++; + }, admission: static function () use (&$admissions): void { + $admissions++; + }, connection: $connection); + self::fail('Unavailable settings must refuse before writer admission.'); + } catch (SqliteAdmissionException $exception) { + self::assertSame('Cannot inspect the SQLite admission busy timeout.', $exception->getMessage()); + } + self::assertSame(1, $pdo->queries); + self::assertSame(0, $admissions); + self::assertSame(0, $calls); + self::assertSame(0, $connection->transactionLevel()); + self::assertFalse($pdo->inTransaction()); + } + public function testRejectedAdmissionRestoresTheOriginalBusyTimeoutWithoutEnteringTheBody(): void { $pdo = $this->connection->getPdo(); From 7954cdc5a4bce53e37c24dc51771b9de73e14b0a Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Sat, 17 Oct 2026 20:33:20 +0000 Subject: [PATCH 4/6] fix: preserve the fenced SQLite acknowledgment retry budget --- .../SqliteAcknowledgingDatabaseQueue.php | 19 ++++++++- src/Support/SqliteWriteWait.php | 39 +++++++++++++++++++ src/Support/SqliteWriteWaitException.php | 12 ++++++ .../Exceptions/SqliteAdmissionException.php | 11 ------ src/V2/Support/SqliteTransaction.php | 33 ++-------------- .../SqliteDatabaseAcknowledgementTest.php | 38 ++++++++++++++++++ tests/Unit/V2/SqliteTransactionTest.php | 4 +- 7 files changed, 112 insertions(+), 44 deletions(-) create mode 100644 src/Support/SqliteWriteWait.php create mode 100644 src/Support/SqliteWriteWaitException.php delete mode 100644 src/V2/Exceptions/SqliteAdmissionException.php diff --git a/src/Queue/SqliteAcknowledgingDatabaseQueue.php b/src/Queue/SqliteAcknowledgingDatabaseQueue.php index ff05e10d5..8b4a5b7bd 100644 --- a/src/Queue/SqliteAcknowledgingDatabaseQueue.php +++ b/src/Queue/SqliteAcknowledgingDatabaseQueue.php @@ -9,6 +9,7 @@ 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 +21,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 +36,23 @@ 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(); + if ($boundedWait) { + SqliteWriteWait::run($pdo, static function () use ($query): void { + $query->delete(); + }, $deadline); + } else { + $query->delete(); + } } catch (QueryException $exception) { $deleteFailure = $exception; throw $exception; diff --git a/src/Support/SqliteWriteWait.php b/src/Support/SqliteWriteWait.php new file mode 100644 index 000000000..eae1568f2 --- /dev/null +++ b/src/Support/SqliteWriteWait.php @@ -0,0 +1,39 @@ +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); + } + } +} diff --git a/src/Support/SqliteWriteWaitException.php b/src/Support/SqliteWriteWaitException.php new file mode 100644 index 000000000..db67ee615 --- /dev/null +++ b/src/Support/SqliteWriteWaitException.php @@ -0,0 +1,12 @@ +query('PRAGMA busy_timeout'); - if ($statement === false) { - throw new SqliteAdmissionException('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) { - $admission($database); - return; - } - - // A single driver wait must not consume the root retry budget. Only the - // admission uses a short wait; application code keeps its original setting. - $pdo->exec('PRAGMA busy_timeout=' . $waitMilliseconds); - try { - $admission($database); - } finally { - $pdo->exec('PRAGMA busy_timeout=' . $originalTimeout); - } - } - private static function isContention(Throwable $exception): bool { do { diff --git a/tests/Unit/Queue/SqliteDatabaseAcknowledgementTest.php b/tests/Unit/Queue/SqliteDatabaseAcknowledgementTest.php index 7402c5f4d..d3c0fceec 100644 --- a/tests/Unit/Queue/SqliteDatabaseAcknowledgementTest.php +++ b/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 { diff --git a/tests/Unit/V2/SqliteTransactionTest.php b/tests/Unit/V2/SqliteTransactionTest.php index 55bf6c038..921429835 100644 --- a/tests/Unit/V2/SqliteTransactionTest.php +++ b/tests/Unit/V2/SqliteTransactionTest.php @@ -17,7 +17,7 @@ use PHPUnit\Framework\Attributes\DataProvider; use PHPUnit\Framework\TestCase; use Throwable; -use Workflow\V2\Exceptions\SqliteAdmissionException; +use Workflow\Support\SqliteWriteWaitException; use Workflow\V2\Support\SqliteTransaction; final class SqliteTransactionTest extends TestCase @@ -167,7 +167,7 @@ public function query(string $query, ?int $fetchMode = null, mixed ...$fetchMode $admissions++; }, connection: $connection); self::fail('Unavailable settings must refuse before writer admission.'); - } catch (SqliteAdmissionException $exception) { + } catch (SqliteWriteWaitException $exception) { self::assertSame('Cannot inspect the SQLite admission busy timeout.', $exception->getMessage()); } self::assertSame(1, $pdo->queries); From ccee93d845d2a22962eaa5e83b9a308862f25ece Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Sat, 17 Oct 2026 20:36:09 +0000 Subject: [PATCH 5/6] chore: prepare compatible 2.5.32 SQLite writer wait fix --- CHANGELOG.md | 9 +++++++++ composer.json | 2 +- 2 files changed, 10 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 359643de1..62f2d1a98 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/composer.json b/composer.json index 863849b9b..0c5c6b0b5 100644 --- a/composer.json +++ b/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" }, From 299d74827685ec182835026f2c3de4786c9fd0b8 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Sat, 17 Oct 2026 20:38:03 +0000 Subject: [PATCH 6/6] refactor: separate the fenced delete from rollback-proof checks --- .../SqliteAcknowledgingDatabaseQueue.php | 20 ++++++++++++------- 1 file changed, 13 insertions(+), 7 deletions(-) diff --git a/src/Queue/SqliteAcknowledgingDatabaseQueue.php b/src/Queue/SqliteAcknowledgingDatabaseQueue.php index 8b4a5b7bd..85fdddffc 100644 --- a/src/Queue/SqliteAcknowledgingDatabaseQueue.php +++ b/src/Queue/SqliteAcknowledgingDatabaseQueue.php @@ -4,6 +4,7 @@ namespace Workflow\Queue; +use Illuminate\Database\Query\Builder; use Illuminate\Database\QueryException; use Illuminate\Queue\DatabaseQueue; use PDO; @@ -46,13 +47,7 @@ public function deleteClaimedJob(string $queue, int|string $id, int $reservedAt, try { // One conditional write avoids SQLite's SELECT-to-write upgrade. // A newer reservation cannot match the original claim. - if ($boundedWait) { - SqliteWriteWait::run($pdo, static function () use ($query): void { - $query->delete(); - }, $deadline); - } else { - $query->delete(); - } + self::deleteWithBoundedWait($query, $pdo, $boundedWait, $deadline); } catch (QueryException $exception) { $deleteFailure = $exception; throw $exception; @@ -98,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(); + } }