Make projector upserts idempotent under concurrent dispatch (#438, #436)

#438: parallel fan-out (e.g. 100 children completing in one all([...]) group)
strands runs because RunTimelineProjector.updateOrCreate races on the
projection PK. firstOrNew + save splits SELECT and INSERT, so two workers
both observe no row and both INSERT, and the loser hits SQLSTATE 23000
ER_DUP_ENTRY for the timeline row's primary key. The dispatch path
propagates the exception, the workflow task fails, and the parent stays in
waiting indefinitely while the queue length keeps growing.

Wrap the per-row updateOrCreate in IdempotentProjectionUpsert, which
retries on unique-key violations with bounded jittered backoff. After the
racing writer wins, our retry's firstOrNew finds the row and falls
through to UPDATE — no analogous race exists on UPDATE because it
matches an existing PK rather than asserting absence. Apply the helper
to the four sibling projectors that share the same updateOrCreate shape
(timeline / wait / timer / lineage / summary).

Sibling DELETE-side races for these projectors were already addressed
by StaleProjectionCleanup (#425).

#436: a bare PHP Error (e.g. "Non-static method ... cannot be called
statically") could not be restored for replay because is_subclass_of
returns false when the class itself IS Error — Error and Exception
implement Throwable independently, so the base-class fallback picked
Exception::class and reflection raised "Cannot access protected property
Error::$message" against the rehydrated instance. Switch to
is_a(..., Error::class, true) so Error itself, not just its subclasses,
picks the Error reflection target.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: durable-workflow-ops

src/V2/Support/FailureFactory.php

@@ -602,7 +602,12 @@ private static function restoreThrowable(string $class, array $payload): Throwab
 
         /** @var Throwable $throwable */
         $throwable = $reflection->newInstanceWithoutConstructor();
-        $baseClass = is_subclass_of($class, Error::class) ? Error::class : Exception::class;
+        // Throwable's protected message/code/file/line/trace properties are declared
+        // independently on Error and Exception (siblings, not parent/child). Use is_a
+        // with the allow_string flag so Error itself — not just its subclasses — picks
+        // the Error::class reflection target. Falling back to Exception here would
+        // raise "Cannot access protected property Error::$message" (#436).
+        $baseClass = is_a($class, Error::class, true) ? Error::class : Exception::class;
 
         self::setThrowableProperty($throwable, $baseClass, 'message', $payload['message']);
         self::setThrowableProperty($throwable, $baseClass, 'code', $payload['code']);

src/V2/Support/IdempotentProjectionUpsert.php

@@ -0,0 +1,100 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Workflow\V2\Support;
+
+use Illuminate\Database\Eloquent\Model;
+use Illuminate\Database\QueryException;
+use LogicException;
+use Throwable;
+
+/**
+ * Wraps a per-row updateOrCreate with retry on unique-key violations so the
+ * timeline / wait / timer / lineage / summary projectors stay correct when
+ * two workers race the same projection row.
+ *
+ * The race shape (#438): updateOrCreate is firstOrNew + save. Two concurrent
+ * callers both observe no row, both INSERT, and the loser hits SQLSTATE 23000
+ * (MySQL/SQLite) or 23505 (Postgres) on the projection's primary or unique
+ * key. On retry, firstOrNew sees the row the winning side just wrote and the
+ * second call falls through to UPDATE, which has no analogous race because
+ * UPDATE does not collide on existing primary keys.
+ *
+ * Sibling DELETE-side races for these projectors are handled by
+ * {@see StaleProjectionCleanup} (#425).
+ */
+final class IdempotentProjectionUpsert
+{
+    private const MAX_ATTEMPTS = 5;
+
+    /**
+     * @template TModel of Model
+     * @param class-string<TModel> $model
+     * @param array<string, mixed> $key
+     * @param array<string, mixed> $values
+     * @return TModel
+     */
+    public static function upsert(string $model, array $key, array $values): Model
+    {
+        for ($attempt = 1; $attempt <= self::MAX_ATTEMPTS; $attempt++) {
+            try {
+                /** @var TModel $row */
+                $row = $model::query()->updateOrCreate($key, $values);
+
+                return $row;
+            } catch (QueryException $e) {
+                if (! self::isUniqueViolation($e) || $attempt === self::MAX_ATTEMPTS) {
+                    throw $e;
+                }
+
+                usleep(self::backoffMicroseconds($attempt));
+            }
+        }
+
+        throw new LogicException('IdempotentProjectionUpsert exhausted attempts without resolving.');
+    }
+
+    private static function isUniqueViolation(Throwable $e): bool
+    {
+        $errorInfo = $e instanceof QueryException ? ($e->errorInfo ?? null) : null;
+        $sqlState = is_array($errorInfo) ? (string) ($errorInfo[0] ?? '') : '';
+        $driverCode = is_array($errorInfo) ? (int) ($errorInfo[1] ?? 0) : 0;
+
+        // 23505 is the Postgres-specific unique_violation SQLSTATE; it never
+        // overlaps with other constraint failures so we can short-circuit.
+        if ($sqlState === '23505') {
+            return true;
+        }
+
+        // 23000 is the generic integrity-constraint family (MySQL, SQLite,
+        // SQL Server). Narrow to driver codes that mean "duplicate key" so we
+        // don't retry e.g. foreign-key or NOT NULL violations that won't clear
+        // on the next pass.
+        if ($sqlState === '23000') {
+            // MySQL 1062 = ER_DUP_ENTRY
+            // SQLite 19 = SQLITE_CONSTRAINT (covers UNIQUE; message-disambiguated below)
+            // SQL Server 2627 = unique-constraint violation; 2601 = duplicate index key
+            if ($driverCode === 1062 || $driverCode === 2627 || $driverCode === 2601) {
+                return true;
+            }
+
+            $message = strtolower($e->getMessage());
+
+            return str_contains($message, 'duplicate entry')
+                || str_contains($message, 'unique constraint failed')
+                || str_contains($message, 'duplicate key');
+        }
+
+        return false;
+    }
+
+    private static function backoffMicroseconds(int $attempt): int
+    {
+        // 2ms, 4ms, 8ms, 16ms (capped) with small jitter to break ties between
+        // racing retriers.
+        $base = min(16_000, 2_000 * (1 << ($attempt - 1)));
+
+        return $base + random_int(0, 1_000);
+    }
+}

src/V2/Support/RunLineageProjector.php

@@ -131,7 +131,8 @@ private static function projectEntry(
         $runId = self::stringValue($entry['workflow_run_id'] ?? null);
 
         /** @var WorkflowRunLineageEntry $row */
-        $row = $lineageModel::query()->updateOrCreate(
+        $row = IdempotentProjectionUpsert::upsert(
+            $lineageModel,
             [
                 'id' => $projectionId,
             ],

src/V2/Support/RunSummaryProjector.php

@@ -302,7 +302,8 @@ public static function project(WorkflowRun $run): WorkflowRunSummary
         $selectedNextTask = $openUpdateWait !== null ? $openUpdateTask : $nextTask;
 
         /** @var WorkflowRunSummary $summary */
-        $summary = $summaryModel::query()->updateOrCreate(
+        $summary = IdempotentProjectionUpsert::upsert(
+            $summaryModel,
             [
                 'id' => $run->id,
             ],

src/V2/Support/RunTimelineProjector.php

@@ -34,7 +34,8 @@ public static function project(WorkflowRun $run, ?array $entries = null): array
             $seen[] = $projectionId;
 
             /** @var WorkflowTimelineEntry $row */
-            $row = $entryModel::query()->updateOrCreate(
+            $row = IdempotentProjectionUpsert::upsert(
+                $entryModel,
                 [
                     'id' => $projectionId,
                 ],

src/V2/Support/RunTimerProjector.php

@@ -34,7 +34,8 @@ public static function project(WorkflowRun $run, ?array $timers = null): array
             $seen[] = $projectionId;
 
             /** @var WorkflowRunTimerEntry $row */
-            $row = $entryModel::query()->updateOrCreate(
+            $row = IdempotentProjectionUpsert::upsert(
+                $entryModel,
                 [
                     'id' => $projectionId,
                 ],

src/V2/Support/RunWaitProjector.php

@@ -30,7 +30,8 @@ public static function project(WorkflowRun $run, ?array $waits = null): array
             $seen[] = $projectionId;
 
             /** @var WorkflowRunWait $row */
-            $row = $waitModel::query()->updateOrCreate(
+            $row = IdempotentProjectionUpsert::upsert(
+                $waitModel,
                 [
                     'id' => $projectionId,
                 ],

tests/Unit/V2/FailureFactoryRestoreTest.php

@@ -0,0 +1,69 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tests\Unit\V2;
+
+use ArgumentCountError;
+use Error;
+use Exception;
+use RuntimeException;
+use Tests\TestCase;
+use TypeError;
+use Workflow\V2\Support\FailureFactory;
+
+final class FailureFactoryRestoreTest extends TestCase
+{
+    /**
+     * Regression for #436. PHP's Throwable interface is implemented independently
+     * by Exception and Error (siblings, not parent/child). The restorer used
+     * is_subclass_of($class, Error::class) to decide which base-class reflection
+     * surface owns the protected message/code/file/line/trace properties — but
+     * is_subclass_of returns false when $class IS Error, so an activity that
+     * threw a bare Error fell through to Exception's reflection target and
+     * raised "Cannot access protected property Error::$message" during replay,
+     * stranding the run in waiting.
+     *
+     * @dataProvider errorSubclassesProvider
+     */
+    public function testRestoresErrorSubclassesWithoutFallingBackToExceptionBaseClass(string $class, string $message): void
+    {
+        $payload = FailureFactory::payload(new $class($message));
+
+        $restored = FailureFactory::restoreForReplay($payload);
+
+        $this->assertInstanceOf($class, $restored);
+        $this->assertSame($message, $restored->getMessage());
+    }
+
+    /**
+     * @return iterable<string, array{0: string, 1: string}>
+     */
+    public static function errorSubclassesProvider(): iterable
+    {
+        yield 'bare Error' => [Error::class, 'static call against instance method'];
+        yield 'TypeError' => [TypeError::class, 'argument 1 must be of type string'];
+        yield 'ArgumentCountError' => [ArgumentCountError::class, 'too few arguments to function'];
+    }
+
+    public function testRestoresExceptionSubclassesUnchanged(): void
+    {
+        $original = new RuntimeException('still works for Exception side', 42);
+
+        $restored = FailureFactory::restoreForReplay(FailureFactory::payload($original));
+
+        $this->assertInstanceOf(RuntimeException::class, $restored);
+        $this->assertSame('still works for Exception side', $restored->getMessage());
+        $this->assertSame(42, $restored->getCode());
+    }
+
+    public function testRestoresBaseException(): void
+    {
+        $original = new Exception('base exception sanity check');
+
+        $restored = FailureFactory::restoreForReplay(FailureFactory::payload($original));
+
+        $this->assertInstanceOf(Exception::class, $restored);
+        $this->assertSame('base exception sanity check', $restored->getMessage());
+    }
+}