Retry embedded application heartbeats after SQLite contention

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
 

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'];
 

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')) {