diff --git a/CHANGELOG.md b/CHANGELOG.md index 14a5a26d..b8f7185f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,8 @@ Preserve completed tasks and snapshot the cancelled attempt's matching task so its worker can acknowledge the stop and repeat that acknowledgement safely. - Apply the same task selection when waiting for cancellation completion. +- Retry rolled-back SQLite application heartbeats after metadata contention, + preserving the current attempt, deadlines and one progress event per call. ## 2.13.0 - 2026-10-24 diff --git a/src/V2/Activity.php b/src/V2/Activity.php index 15241968..5c739744 100644 --- a/src/V2/Activity.php +++ b/src/V2/Activity.php @@ -5,7 +5,6 @@ namespace Workflow\V2; use Illuminate\Container\Container; -use Illuminate\Support\Facades\DB; use Workflow\Traits\ResolvesMethodDependencies; use Workflow\V2\Enums\ActivityAttemptStatus; use Workflow\V2\Enums\ActivityStatus; @@ -26,6 +25,7 @@ use Workflow\V2\Support\ActivityTimeoutEnforcer; use Workflow\V2\Support\HeartbeatProgress; use Workflow\V2\Support\LocalActivityRuntime; +use Workflow\V2\Support\SqliteTransaction; abstract class Activity { @@ -112,7 +112,7 @@ public function heartbeat(array $progress = []): void { $normalizedProgress = HeartbeatProgress::normalizeForWrite($progress); - DB::transaction(function () use ($normalizedProgress): void { + SqliteTransaction::run(function () use ($normalizedProgress): void { $lockedRows = ActivityRowLockOrder::lockForExecution($this->execution->id); $execution = $lockedRows['execution']; diff --git a/tests/Unit/V2/EmbeddedActivityLeaseMaintenanceTest.php b/tests/Unit/V2/EmbeddedActivityLeaseMaintenanceTest.php index 81d49dfb..1da5ef58 100644 --- a/tests/Unit/V2/EmbeddedActivityLeaseMaintenanceTest.php +++ b/tests/Unit/V2/EmbeddedActivityLeaseMaintenanceTest.php @@ -4,6 +4,7 @@ namespace Tests\Unit\V2; +use Illuminate\Database\Events\QueryExecuted; use Illuminate\Database\Schema\Blueprint; use Illuminate\Support\Facades\Artisan; use Illuminate\Support\Facades\DB; @@ -135,6 +136,52 @@ public function testRealUserProgressCanContinueAlongsideRuntimeMaintenance(): vo self::assertSame(0, DB::table('failed_jobs')->count()); } + public function testApplicationHeartbeatRetriesAConcurrentMetadataWriteWithoutRepeatingTheActivity(): void + { + if (DB::connection()->getDriverName() !== 'sqlite') { + self::markTestSkipped('This control forces a SQLite WAL snapshot upgrade conflict.'); + } + [$stub, $task, $execution] = $this->pendingActivity('embedded-progress-snapshot-conflict'); + $execution->forceFill([ + 'retry_policy' => [ + 'maximum_attempts' => 1, + 'heartbeat_timeout' => 2, + ], + ])->save(); + $writer = new \PDO('sqlite:' . $this->path); + $writer->setAttribute(\PDO::ATTR_ERRMODE, \PDO::ERRMODE_EXCEPTION); + $conflicts = 0; + LeaseBlockingActivity::$recordProgress = true; + LeaseBlockingActivity::$beforeWait = static function () use (&$writer, &$conflicts, $task): void { + DB::listen(static function (QueryExecuted $query) use (&$writer, &$conflicts, $task): void { + if ($conflicts !== 0 || $query->connection->transactionLevel() !== 1 + || ! str_starts_with($query->sql, 'select') || ! str_contains($query->sql, 'activity_executions')) { + return; + } + $statement = $writer->prepare('UPDATE workflow_tasks SET updated_at = ? WHERE id = ?'); + $statement->execute(['2000-01-01 00:00:00', $task->id]); + $conflicts++; + }); + }; + + try { + $this->workOnce(); + $this->workOnce(); + + self::assertSame(1, $conflicts); + self::assertSame(1, LeaseBlockingActivity::$invocations); + self::assertSame(TaskStatus::Completed, $task->fresh()->status); + self::assertSame('completed', $stub->refresh()->status()); + $events = WorkflowHistoryEvent::query()->where('workflow_run_id', $stub->runId())->get(); + self::assertSame(0, $events->where('event_type', HistoryEventType::ActivityFailed)->count()); + self::assertSame(4, $events->where('event_type', HistoryEventType::ActivityHeartbeatRecorded)->count()); + self::assertSame(1, $events->where('event_type', HistoryEventType::ActivityCompleted)->count()); + self::assertSame(0, DB::table('failed_jobs')->count()); + } finally { + $writer = null; + } + } + public function testAStoppedMetadataProcessCannotAuthorizeTheInlineResult(): void { if (! function_exists('posix_kill') || ! is_file('/proc/self/task/' . getmypid() . '/children')) {