diff --git a/CHANGELOG.md b/CHANGELOG.md index a4acba4b4..7d22a0c17 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,14 @@ # Changelog -## 2.12.1 - Unreleased +## 2.12.2 - 2026-10-24 + +- Refuse expired or settled activity callbacks under the original run lock + before acquiring task or execution locks. Preserve committed renewals and + the normal live-outcome path. Concurrent retention can finish without a + reverse lock dependency or restored details. +- Acquire SQLite writer admission before reading a retained run for pruning, + within the existing transaction policy and original publication budget. + This correction emits no new history event and needs no new event reader. - Require the upstream TokenGuard string-validation fix on Laravel 11, raising its minimum to 11.56.1. Keep the maintained Laravel 12/13 minimums unchanged. diff --git a/composer.json b/composer.json index 618cd1637..81915cc0a 100644 --- a/composer.json +++ b/composer.json @@ -87,7 +87,7 @@ "dev-main": "2.0.x-dev" }, "durable-workflow": { - "product-train": "2.12.1", + "product-train": "2.12.2", "laravel-embedded-upgrade-contract": "resources/laravel-embedded-upgrade-contract.json", "laravel-dependency-security-policy": "resources/laravel-dependency-security-policy.json" }, diff --git a/docs/activity-outcome-retention.md b/docs/activity-outcome-retention.md new file mode 100644 index 000000000..cd7994754 --- /dev/null +++ b/docs/activity-outcome-retention.md @@ -0,0 +1,33 @@ +# Activity outcome refusals and retention + +A late activity callback and pruning can reach the same terminal run at the +same time. Pruning locks the run before deleting detail rows. A callback must +not hold task or execution locks while waiting for that run lock if its attempt +has already lost authority. + +The prospective Core 2.12.2 correction first checks an expired or settled +attempt under the run lock, using fresh stored task, execution and attempt +bindings. If pruning has completed, it refuses the callback without recreating +rows. If a renewal committed before the lock, its current lease remains +authoritative and normal outcome recording performs its existing checks. +Ordinary live attempts keep their existing outcome path and deadlines. + +This refusal does not serialize a result, resolve an external payload, record a +new diagnostic, alter the retry policy or grant business-outcome authority. It +is independent of the proposed rejected-outcome audit and its reader rollout. +The canonical timeout and cancellation mechanisms remain responsible for +settling the original attempt. + +SQLite pruning acquires the writer before reading its conditional snapshot. +Admission does not change the run's identity or ORM timestamps, and already +claimed outcomes retain their original monotonic publication deadline. + +The regression controls use two actual database processes. Retention holds the +run lock before the callback requests it, then deletes the same original +details. Both inline and service-bridge paths must finish without a deadlock or +restored detail, across completed, failed, cancelled and terminated runs. Other +controls cover committed renewal, result-preparation refusal and SQLite writer +ownership before reading the retention snapshot. + +Source qualification is not publication. Normal package installation and any +affected Server image adoption require their own exact-artifact acceptance. diff --git a/src/V2/Support/ActivityOutcomeRecorder.php b/src/V2/Support/ActivityOutcomeRecorder.php index 275419f4e..f40719d33 100644 --- a/src/V2/Support/ActivityOutcomeRecorder.php +++ b/src/V2/Support/ActivityOutcomeRecorder.php @@ -41,6 +41,15 @@ public static function record( ?string $codec = null, ?int $admissionDeadline = null, ): array { + $sqlite = (new ActivityAttempt())->getConnection() + ->getDriverName() === 'sqlite'; + if (! $sqlite) { + $refused = self::refusedExpiredAttempt($taskId, $attemptId, $attemptCount, $admissionDeadline); + if ($refused !== null) { + return $refused; + } + } + return SqliteTransaction::forTask($taskId, static function () use ( $taskId, $attemptId, @@ -50,7 +59,18 @@ public static function record( $maxAttempts, $backoffSeconds, $codec, + $sqlite, + $admissionDeadline, ): array { + // Even an unlocked SQLite read can wait behind an exclusive writer. + // Keep the refusal metadata behind the original bounded admission. + if ($sqlite) { + $refused = self::refusedExpiredAttempt($taskId, $attemptId, $attemptCount, $admissionDeadline); + if ($refused !== null) { + return $refused; + } + } + /** @var WorkflowTask|null $task */ $task = WorkflowTask::query() ->lockForUpdate() @@ -508,6 +528,94 @@ public static function recordForAttempt( ); } + /** + * Fence an already lost outcome without taking task/execution locks in the + * reverse order to retention. A normal live outcome keeps its existing path. + * + * @return array{recorded: bool, reason: string|null, next_task: WorkflowTask|null}|null + */ + private static function refusedExpiredAttempt( + string $taskId, + string $attemptId, + int $attemptNumber, + ?int $admissionDeadline, + ): ?array { + $original = ActivityAttempt::query()->find($attemptId); + if (! $original instanceof ActivityAttempt || $original->id !== $attemptId + || $original->workflow_task_id !== $taskId || $original->attempt_number !== $attemptNumber) { + return null; + } + if ($original->status === ActivityAttemptStatus::Running + && $original->lease_expires_at !== null && now() + ->lt($original->lease_expires_at)) { + return null; + } + + $connection = $original->getConnection(); + $runId = (string) $original->workflow_run_id; + + return SqliteTransaction::run(static function () use ($original): ?array { + return self::refusedExpiredAttemptUnderRunLock($original); + }, admission: static function () use ($connection, $runId): void { + WorkflowRun::on($connection->getName())->whereKey($runId)->toBase()->update([ + 'id' => $runId, + ]); + }, admissionDeadline: $admissionDeadline, connection: $connection); + } + + /** + * @return array{recorded: bool, reason: string|null, next_task: WorkflowTask|null}|null + */ + private static function refusedExpiredAttemptUnderRunLock(ActivityAttempt $original): ?array + { + $connection = $original->getConnectionName(); + $run = WorkflowRun::on($connection)->whereKey($original->workflow_run_id)->lockForUpdate()->first(); + if (! $run instanceof WorkflowRun || $run->details_pruned_at !== null) { + return self::ignored('stale_attempt'); + } + $attempt = ActivityAttempt::on($connection)->find($original->id); + $task = WorkflowTask::on($connection)->find($original->workflow_task_id); + $execution = ActivityExecution::on($connection)->find($original->activity_execution_id); + if (! $attempt instanceof ActivityAttempt || ! $task instanceof WorkflowTask + || ! $execution instanceof ActivityExecution + || $attempt->id !== $original->id || $attempt->attempt_number !== $original->attempt_number + || $attempt->workflow_task_id !== $task->id || $attempt->activity_execution_id !== $execution->id + || $attempt->workflow_run_id !== $run->id || $execution->workflow_run_id !== $run->id + || $task->workflow_run_id !== $run->id || $task->namespace !== $run->namespace + || ($task->payload['activity_execution_id'] ?? null) !== $execution->id) { + return self::ignored('stale_attempt'); + } + + $reason = self::expiredAttemptReason($run, $execution, $task, $attempt); + + return $reason === null ? null : self::ignored($reason); + } + + private static function expiredAttemptReason( + WorkflowRun $run, + ActivityExecution $execution, + WorkflowTask $task, + ActivityAttempt $attempt, + ): ?string { + if (in_array($run->status, [RunStatus::Cancelled, RunStatus::Terminated], true) + && ! ActivityAbandonment::allows($run, $execution)) { + return $run->status === RunStatus::Terminated ? 'run_terminated' : 'run_cancelled'; + } + if ($attempt->status !== ActivityAttemptStatus::Running || $task->status !== TaskStatus::Leased + || $task->attempt_count !== $attempt->attempt_number + || $execution->attempt_count !== $attempt->attempt_number + || $execution->current_attempt_id !== $attempt->id) { + return 'stale_attempt'; + } + if (ActivityTimeoutEnforcer::hasExpiredDeadline($execution, now()) + && ActivityAbandonment::allows($run, $execution)) { + return 'activity_deadline_elapsed'; + } + + // Renewals that committed before the run lock must keep their authority. + return ActivityAttemptAuthority::expiredReason($task, $execution, now(), $run); + } + /** * Close a standalone-activity host run with the activity's terminal * outcome. On success the run is marked Completed and the run output diff --git a/src/V2/Support/WorkflowRunRetentionCleanup.php b/src/V2/Support/WorkflowRunRetentionCleanup.php index f3902ab25..edf93e46c 100644 --- a/src/V2/Support/WorkflowRunRetentionCleanup.php +++ b/src/V2/Support/WorkflowRunRetentionCleanup.php @@ -75,7 +75,7 @@ public static function pruneRun(WorkflowRun|string $run): array ->getModel() ->getConnection(); - return $runConnection->transaction(static function () use ($runId): array { + return SqliteTransaction::run(static function () use ($runId): array { /** @var WorkflowRun $lockedRun */ $lockedRun = ConfiguredV2Models::query('run_model', WorkflowRun::class) ->whereKey($runId) @@ -171,7 +171,16 @@ public static function pruneRun(WorkflowRun|string $run): array } return $report; - }); + }, admission: static function () use ($runId): void { + $runs = ConfiguredV2Models::query('run_model', WorkflowRun::class); + $key = $runs->getModel() + ->getKeyName(); + $runs->whereKey($runId) + ->toBase() + ->update([ + $key => $runId, + ]); + }, connection: $runConnection); } /** diff --git a/tests/Support/ActivityOutcomeRetentionProcessTest.php b/tests/Support/ActivityOutcomeRetentionProcessTest.php new file mode 100644 index 000000000..26b1ed7e1 --- /dev/null +++ b/tests/Support/ActivityOutcomeRetentionProcessTest.php @@ -0,0 +1,106 @@ +input(); + Date::setTestNow($input['observed_at']); + $connection = DB::connection(); + if ($input['outcome'] === 'pruned') { + $connection->listen(static function (QueryExecuted $event) use ($input): void { + if (! str_contains(strtolower($event->sql), 'for update') + || ! str_contains($event->sql, 'workflow_runs')) { + return; + } + file_put_contents($input['ready'], 'run locked'); + while (! is_file($input['continue'])) { + if (hrtime(true) >= $input['deadline']) { + self::fail('Retention exceeded the original process budget.'); + } + usleep(1_000); + } + }); + $report = WorkflowRunRetentionCleanup::pruneRun($input['run_id']); + self::assertGreaterThan(0, $report['history_events_deleted']); + $reply = [ + 'recorded' => false, + 'reason' => 'pruned', + ]; + } else { + // Signal before this query can wait on retention's run lock. The + // preceding implementation already holds task/execution locks here. + $connection->beforeExecuting(static function (string $query) use ($input): void { + if (str_contains(strtolower($query), 'for update') + && str_contains($query, 'workflow_runs')) { + file_put_contents($input['ready'], 'requesting run lock'); + } + }); + if ($input['direct']) { + $reply = ActivityOutcomeRecorder::record( + $input['task_id'], + $input['attempt_id'], + $input['attempt_number'], + 'discarded late result', + $input['outcome'] === 'failed' ? new RuntimeException('discarded late failure') : null, + 2, + 5, + ); + } else { + $bridge = $this->app->make(ActivityTaskBridge::class); + $reply = $input['outcome'] === 'completed' + ? $bridge->complete($input['attempt_id'], 'discarded late result') + : $bridge->fail($input['attempt_id'], [ + 'message' => 'discarded late failure', + ]); + } + self::assertFalse($reply['recorded']); + } + + file_put_contents($input['report'], json_encode([ + 'pid' => getmypid(), + 'recorded' => $reply['recorded'], + 'reason' => $reply['reason'], + ], JSON_THROW_ON_ERROR)); + } + + protected function getPackageProviders($app): array + { + return [WorkflowServiceProvider::class]; + } + + protected function defineEnvironment($app): void + { + $input = $this->input(); + $app['config']->set('database.default', 'retention_process'); + $app['config']->set('database.connections.retention_process', $input['database']); + $app['config']->set('workflows.storage.connection', 'retention_process'); + $app['config']->set('queue.default', 'sync'); + $app['config']->set('cache.default', 'array'); + $app['config']->set('logging.default', 'null'); + } + + private function input(): array + { + $path = getenv('DWF_RETENTION_PROCESS_INPUT'); + if (! is_string($path) || $path === '') { + $this->markTestSkipped('Invoked only by the concurrent retention control.'); + } + + return json_decode(file_get_contents($path), true, flags: JSON_THROW_ON_ERROR); + } +} diff --git a/tests/Unit/V2/ActivityOutcomeRetentionTest.php b/tests/Unit/V2/ActivityOutcomeRetentionTest.php new file mode 100644 index 000000000..8c93faacc --- /dev/null +++ b/tests/Unit/V2/ActivityOutcomeRetentionTest.php @@ -0,0 +1,386 @@ +freezeTime(); + config([ + 'workflows.v2.task_dispatch_mode' => 'poll', + ]); + } + + public static function retainedOutcomes(): array + { + $cases = []; + foreach ([RunStatus::Completed, RunStatus::Failed, RunStatus::Cancelled, RunStatus::Terminated] as $status) { + foreach ([false, true] as $reclaimed) { + foreach (['completed', 'failed'] as $outcome) { + $key = $status->value . '/' . (int) $reclaimed . '/' . $outcome; + $cases[$key . '/bridge'] = [$status, $reclaimed, $outcome, false]; + $cases[$key . '/inline'] = [$status, $reclaimed, $outcome, true]; + } + } + } + + return $cases; + } + + #[DataProvider('retainedOutcomes')] + public function testPruningAndRefusedOutcomeCannotDeadlockOrRestoreDetails( + RunStatus $status, + bool $reclaimed, + string $outcome, + bool $direct, + ): void { + if (DB::connection()->getDriverName() === 'sqlite') { + $this->markTestSkipped('This deterministic row-lock interleaving requires PostgreSQL or MySQL.'); + } + $deadline = hrtime(true) + 30_000_000_000; + [, $claim, $run] = $this->claimedOriginal(); + $this->travel(301) + ->seconds(); + if ($reclaimed) { + self::assertTrue(ActivityTimeoutEnforcer::enforceLeaseExpiry($claim['task_id'])['enforced']); + } + $run->forceFill([ + 'status' => $status, + 'closed_at' => now(), + ])->save(); + $reports = $this->interleavePruning($run, $claim, $outcome, $deadline, $direct); + + self::assertNotSame($reports[0]['pid'], $reports[1]['pid']); + self::assertSame('pruned', $reports[0]['reason']); + self::assertFalse($reports[1]['recorded']); + self::assertContains($reports[1]['reason'], [ + 'stale_attempt', 'attempt_not_found', 'task_not_found', 'activity_execution_missing', + ]); + self::assertSame($status, $run->fresh()->status); + self::assertNotNull($run->fresh()->details_pruned_at); + self::assertSame(0, $run->historyEvents()->count()); + foreach ([ + ActivityAttempt::class, + ActivityExecution::class, + WorkflowTask::class, + WorkflowTimelineEntry::class, + ] as $model) { + self::assertSame(0, $model::query()->where('workflow_run_id', $run->id)->count()); + } + self::assertNull(WorkflowRunSummary::query()->find($run->id)); + } + + public function testRefusedExpiredOutcomeDoesNotPreparePayloadOrWriteHistory(): void + { + [$bridge, $claim, $run] = $this->claimedOriginal(); + $this->travel(301) + ->seconds(); + $before = $run->historyEvents() + ->get() + ->map->getAttributes() + ->all(); + $reply = $bridge->complete($claim['activity_attempt_id'], new class() { + public function __serialize(): array + { + throw new \RuntimeException('An expired result must not be serialized.'); + } + }); + + self::assertFalse($reply['recorded']); + self::assertSame('lease_expired', $reply['reason']); + self::assertSame($before, $run->historyEvents()->get()->map->getAttributes()->all()); + self::assertSame( + ActivityAttemptStatus::Running, + ActivityAttempt::query()->findOrFail($claim['activity_attempt_id'])->status + ); + } + + public static function outcomeEntryPoints(): array + { + return [ + 'service bridge' => [false], + 'inline callback' => [true], + ]; + } + + #[DataProvider('outcomeEntryPoints')] + public function testACommittedRenewalKeepsAuthorityAfterAnOlderRead(bool $direct): void + { + [$bridge, $claim] = $this->claimedOriginal(); + self::assertInstanceOf(ActivityLeaseRenewal::class, $bridge); + $this->travel(295) + ->seconds(); + $armed = true; + $observedExpiry = null; + ActivityAttempt::retrieved(function (ActivityAttempt $attempt) use ( + &$armed, + &$observedExpiry, + $bridge, + $claim, + ): void { + if (! $armed || $attempt->id !== $claim['activity_attempt_id']) { + return; + } + $armed = false; + $observedExpiry = $attempt->lease_expires_at; + self::assertTrue( + $bridge->renewLease($claim['activity_attempt_id'], $claim['lease_owner'])['lease_renewed'] + ); + $this->travel(10) + ->seconds(); + }); + + $reply = $direct + ? ActivityOutcomeRecorder::record( + $claim['task_id'], + $claim['activity_attempt_id'], + $claim['attempt_number'], + 'accepted', + null, + 2, + 5 + ) + : $bridge->complete($claim['activity_attempt_id'], 'accepted'); + + self::assertFalse($armed); + self::assertNotNull($observedExpiry); + self::assertTrue(now()->gte($observedExpiry)); + self::assertTrue($reply['recorded']); + self::assertSame( + ActivityAttemptStatus::Completed, + ActivityAttempt::query()->findOrFail($claim['activity_attempt_id'])->status + ); + } + + public function testSqliteRetentionOwnsTheWriterBeforeReadingItsRunSnapshot(): void + { + $connection = DB::connection(); + if ($connection->getDriverName() !== 'sqlite' || $connection->getDatabaseName() === ':memory:') { + $this->markTestSkipped('Writer admission requires two connections to the same SQLite file.'); + } + [, , $run] = $this->claimedOriginal(); + $run->forceFill([ + 'status' => RunStatus::Completed, + 'closed_at' => now(), + ])->save(); + $checked = false; + $writerHeld = false; + $database = $connection->getDatabaseName(); + $connection->listen(static function ($event) use (&$checked, &$writerHeld, $database): void { + if ($checked || ! str_starts_with(strtolower(ltrim($event->sql)), 'select') + || ! str_contains($event->sql, 'workflow_runs')) { + return; + } + $checked = true; + $other = new PDO('sqlite:' . $database, options: [ + PDO::ATTR_TIMEOUT => 0, + ]); + $other->exec('PRAGMA busy_timeout=0'); + try { + $other->exec('BEGIN IMMEDIATE'); + $other->exec('ROLLBACK'); + } catch (PDOException $error) { + self::assertContains(((int) ($error->errorInfo[1] ?? 0)) & 255, [5, 6]); + $writerHeld = true; + } + }); + + WorkflowRunRetentionCleanup::pruneRun($run); + + self::assertTrue($checked); + self::assertTrue($writerHeld, 'The retention snapshot must already own SQLite writer admission.'); + self::assertNotNull($run->fresh()->details_pruned_at); + self::assertSame(0, $run->historyEvents()->count()); + } + + public function testSqliteRefusalMetadataKeepsTheOriginalPublicationBudget(): void + { + $connection = DB::connection(); + if ($connection->getDriverName() !== 'sqlite' || $connection->getDatabaseName() === ':memory:') { + $this->markTestSkipped('Publication contention requires two connections to one SQLite file.'); + } + [, $claim, $run] = $this->claimedOriginal(); + $pdo = $connection->getPdo(); + $previousTimeout = (int) $pdo->query('PRAGMA busy_timeout') + ->fetchColumn(); + $previousJournal = (string) $pdo->query('PRAGMA journal_mode') + ->fetchColumn(); + self::assertContains(strtolower($previousJournal), ['delete', 'wal', 'truncate', 'persist', 'memory', 'off']); + $pdo->exec('PRAGMA journal_mode=DELETE'); + $pdo->exec('PRAGMA busy_timeout=3000'); + $other = new PDO('sqlite:' . $connection->getDatabaseName()); + $other->exec('BEGIN EXCLUSIVE'); + $started = hrtime(true); + try { + try { + ActivityOutcomeRecorder::record( + $claim['task_id'], + $claim['activity_attempt_id'], + $claim['attempt_number'], + 'must not be published', + null, + 2, + 5, + admissionDeadline: $started + 250_000_000, + ); + self::fail('The held writer cannot admit this outcome.'); + } catch (QueryException $error) { + self::assertInstanceOf(PDOException::class, $error->getPrevious()); + self::assertContains(((int) ($error->getPrevious()->errorInfo[1] ?? 0)) & 255, [5, 6]); + } + $elapsed = (hrtime(true) - $started) / 1_000_000_000; + } finally { + $other->exec('ROLLBACK'); + $pdo->exec('PRAGMA busy_timeout=' . $previousTimeout); + $pdo->exec('PRAGMA journal_mode=' . $previousJournal); + } + + self::assertLessThan( + 1.0, + $elapsed, + 'A metadata read must not consume the unrelated three-second SQLite timeout.' + ); + self::assertSame(RunStatus::Waiting, $run->fresh()->status); + self::assertSame( + ActivityAttemptStatus::Running, + ActivityAttempt::query()->findOrFail($claim['activity_attempt_id'])->status + ); + } + + private function claimedOriginal(): array + { + $workflow = WorkflowStub::make(TestRetryWorkflow::class, 'outcome-retention'); + $workflow->start('Ada'); + $task = WorkflowTask::query()->where('task_type', TaskType::Workflow)->where( + 'status', + TaskStatus::Ready + )->sole(); + $this->app->call([new RunWorkflowTask($task->id), 'handle']); + $bridge = $this->app->make(ActivityTaskBridge::class); + $task = $bridge->poll(null, null, 1)[0]; + $claim = $bridge->claimStatus($task['task_id'], 'original-worker'); + self::assertTrue($claim['claimed']); + + return [$bridge, $claim, WorkflowRun::query()->findOrFail($workflow->runId())]; + } + + private function interleavePruning( + WorkflowRun $run, + array $claim, + string $outcome, + int $deadline, + bool $direct + ): array { + $directory = sys_get_temp_dir() . '/workflow-outcome-retention-' . bin2hex(random_bytes(8)); + mkdir($directory, 0700); + $workers = []; + try { + foreach (['pruned', $outcome] as $index => $operation) { + $path = $directory . '/input-' . $index . '.json'; + file_put_contents($path, json_encode([ + 'database' => DB::connection($run->getConnectionName())->getConfig(), + 'run_id' => $run->id, + 'attempt_id' => $claim['activity_attempt_id'], + 'task_id' => $claim['task_id'], + 'attempt_number' => $claim['attempt_number'], + 'direct' => $direct, + 'outcome' => $operation, + 'observed_at' => now() + ->toISOString(), + 'deadline' => $deadline, + 'ready' => $directory . '/ready-' . $index, + 'continue' => $directory . '/continue', + 'report' => $directory . '/report-' . $index . '.json', + ], JSON_THROW_ON_ERROR)); + $worker = new Process([ + PHP_BINARY, + dirname(__DIR__, 3) . '/vendor/bin/phpunit', + dirname(__DIR__, 2) . '/Support/ActivityOutcomeRetentionProcessTest.php', + '--do-not-cache-result', + ], dirname(__DIR__, 3), [ + 'DWF_RETENTION_PROCESS_INPUT' => $path, + 'APP_SERVICES_CACHE' => $directory . '/services-' . $index . '.php', + 'APP_PACKAGES_CACHE' => $directory . '/packages-' . $index . '.php', + 'XDEBUG_MODE' => 'off', + ], timeout: 30); + $workers[] = $worker; + $worker->start(); + $this->awaitFile($directory . '/ready-' . $index, $workers, $deadline); + } + file_put_contents($directory . '/continue', 'continue'); + while ($workers[0]->isRunning() || $workers[1]->isRunning()) { + if (hrtime(true) >= $deadline) { + self::fail('The original concurrent retention budget expired.'); + } + usleep(1_000); + } + foreach ($workers as $worker) { + self::assertSame( + 0, + $worker->getExitCode(), + substr($worker->getErrorOutput() . $worker->getOutput(), -8_192) + ); + } + + return [ + json_decode(file_get_contents($directory . '/report-0.json'), true, flags: JSON_THROW_ON_ERROR), + json_decode(file_get_contents($directory . '/report-1.json'), true, flags: JSON_THROW_ON_ERROR), + ]; + } finally { + foreach ($workers as $worker) { + $worker->stop(2); + } + foreach (glob($directory . '/*') ?: [] as $path) { + unlink($path); + } + rmdir($directory); + } + } + + private function awaitFile(string $path, array $workers, int $deadline): void + { + while (! is_file($path)) { + foreach ($workers as $worker) { + if (! $worker->isRunning()) { + self::fail(substr($worker->getErrorOutput() . $worker->getOutput(), -8_192)); + } + } + if (hrtime(true) >= $deadline) { + self::fail('The original retention interleaving budget expired.'); + } + usleep(1_000); + } + } +}