Finalize v2 timer projection schema

Author: durable-workflow-ops

src/V2/Models/WorkflowRunTimerEntry.php

@@ -12,8 +12,6 @@
 
 class WorkflowRunTimerEntry extends Model
 {
-    public const LEGACY_SCHEMA_VERSION = 0;
-
     public const CURRENT_SCHEMA_VERSION = 1;
 
     public $incrementing = false;
@@ -52,7 +50,6 @@ public function run(): BelongsTo
     public function toTimerPayload(): array
     {
         $payload = is_array($this->payload) ? $this->payload : [];
-        $schemaVersion = $this->schemaVersion();
         $status = self::stringValue($payload['status'] ?? $this->status);
         $sourceStatus = self::stringValue($payload['source_status'] ?? $this->source_status)
             ?? $status;
@@ -75,23 +72,18 @@ public function toTimerPayload(): array
         $payload['condition_definition_fingerprint'] = $this->condition_definition_fingerprint;
         $payload['history_authority'] = $historyAuthority;
         $payload['history_unsupported_reason'] = $historyUnsupportedReason;
-        $payload['row_status'] = self::rowStatus(
-            $payload['row_status'] ?? null,
-            $historyAuthority,
-            $sourceStatus,
-            $schemaVersion,
-        );
+        $payload['row_status'] = self::rowStatus($payload['row_status'] ?? null);
         $payload['diagnostic_only'] = self::diagnosticOnly($historyAuthority);
         $payload['created_at'] = self::timestamp($payload['created_at'] ?? null);
 
         return $payload;
     }
 
-    public function schemaVersion(): int
+    public function schemaVersion(): ?int
     {
         return is_int($this->schema_version)
             ? $this->schema_version
-            : self::LEGACY_SCHEMA_VERSION;
+            : null;
     }
 
     public function usesCurrentSchema(): bool
@@ -117,24 +109,9 @@ private static function diagnosticOnly(mixed $historyAuthority): bool
             && $historyAuthority !== 'typed_history';
     }
 
-    private static function rowStatus(
-        mixed $value,
-        ?string $historyAuthority,
-        ?string $sourceStatus,
-        int $schemaVersion,
-    ): ?string {
-        $rowStatus = self::stringValue($value);
-
-        if ($rowStatus !== null || $schemaVersion !== self::LEGACY_SCHEMA_VERSION) {
-            return $rowStatus;
-        }
-
-        return in_array($historyAuthority, [
-            'mutable_open_fallback',
-            'unsupported_terminal_without_history',
-        ], true)
-            ? $sourceStatus
-            : null;
+    private static function rowStatus(mixed $value): ?string
+    {
+        return self::stringValue($value);
     }
 
     private static function stringValue(mixed $value): ?string

src/V2/Support/HealthCheck.php

@@ -134,8 +134,12 @@ private static function selectedRunProjectionCheck(array $projections): array
                 'timer_needs_rebuild' => $timerNeedsRebuild,
                 'timer_missing_runs_with_timers' => self::integer($timers['missing_runs_with_timers'] ?? 0),
                 'timer_stale_projected_runs' => self::integer($timers['stale_projected_runs'] ?? 0),
-                'timer_legacy_schema_runs' => self::integer($timers['legacy_schema_runs'] ?? 0),
-                'timer_legacy_schema_rows' => self::integer($timers['legacy_schema_rows'] ?? 0),
+                'timer_schema_version_mismatch_runs' => self::integer(
+                    $timers['schema_version_mismatch_runs'] ?? 0
+                ),
+                'timer_schema_version_mismatch_rows' => self::integer(
+                    $timers['schema_version_mismatch_rows'] ?? 0
+                ),
                 'timer_orphaned' => self::integer($timers['orphaned'] ?? 0),
                 'lineage_needs_rebuild' => $lineageNeedsRebuild,
                 'lineage_missing_runs_with_lineage' => self::integer($lineage['missing_runs_with_lineage'] ?? 0),

src/V2/Support/OperatorMetrics.php

@@ -316,9 +316,12 @@ private static function runTimerProjectionMetrics(): array
             'projected_runs_with_timers' => $drift['projected_runs_with_timers'],
             'missing_runs_with_timers' => $drift['missing_runs_with_timers'],
             'stale_projected_runs' => $drift['stale_projected_runs'],
-            'legacy_schema_runs' => $drift['legacy_schema_runs'],
-            'legacy_schema_rows' => $timerModel::query()
-                ->where('schema_version', WorkflowRunTimerEntry::LEGACY_SCHEMA_VERSION)
+            'schema_version_mismatch_runs' => $drift['schema_version_mismatch_runs'],
+            'schema_version_mismatch_rows' => $timerModel::query()
+                ->where(static function ($query): void {
+                    $query->whereNull('schema_version')
+                        ->orWhere('schema_version', '!=', WorkflowRunTimerEntry::CURRENT_SCHEMA_VERSION);
+                })
                 ->count(),
             'orphaned' => $orphaned,
             'needs_rebuild' => $drift['missing_runs_with_timers'] + $drift['stale_projected_runs'] + $orphaned,

src/V2/Support/RunTimerProjector.php

@@ -131,7 +131,7 @@ public static function snapshotForRun(WorkflowRun $run): array
      *     has_canonical: bool,
      *     missing: bool,
      *     stale: bool,
-     *     legacy_schema: bool,
+     *     schema_version_mismatch: bool,
      *     reasons: list<string>
      * }
      */
@@ -142,14 +142,14 @@ public static function driftStatusForRun(WorkflowRun $run): array
         $hasProjection = $projected->isNotEmpty();
         $hasCanonical = $canonicalTimers !== [];
         $reasons = self::rebuildReasons($projected, $canonicalTimers);
-        $legacySchema = in_array('legacy_schema', $reasons, true);
+        $schemaVersionMismatch = in_array('schema_version_mismatch', $reasons, true);
 
         return [
             'has_projection' => $hasProjection,
             'has_canonical' => $hasCanonical,
             'missing' => $hasCanonical && ! $hasProjection,
             'stale' => $hasProjection && $reasons !== [],
-            'legacy_schema' => $legacySchema,
+            'schema_version_mismatch' => $schemaVersionMismatch,
             'reasons' => $reasons,
         ];
     }
@@ -225,7 +225,7 @@ private static function rebuildReasons(EloquentCollection $projected, array $can
         }
 
         if ($projected->isNotEmpty() && ! self::projectedRowsUseCurrentSchema($projected)) {
-            $reasons[] = 'legacy_schema';
+            $reasons[] = 'schema_version_mismatch';
         }
 
         if ($projected->isNotEmpty() && ! self::projectionMatchesSnapshot($projected, $canonical)) {

src/V2/Support/SelectedRunProjectionDrift.php

@@ -129,7 +129,7 @@ public static function timelineRunIdsNeedingRebuild(array $runIds = [], ?string
      *     projected_runs_with_timers: int,
      *     missing_runs_with_timers: int,
      *     stale_projected_runs: int,
-     *     legacy_schema_runs: int
+     *     schema_version_mismatch_runs: int
      * }
      */
     public static function timerMetrics(array $runIds = [], ?string $instanceId = null): array
@@ -141,7 +141,7 @@ public static function timerMetrics(array $runIds = [], ?string $instanceId = nu
             'projected_runs_with_timers' => $analysis['projected_runs_with_canonical'],
             'missing_runs_with_timers' => count($analysis['missing_run_ids']),
             'stale_projected_runs' => count($analysis['stale_run_ids']),
-            'legacy_schema_runs' => count($analysis['legacy_schema_run_ids']),
+            'schema_version_mismatch_runs' => count($analysis['schema_version_mismatch_run_ids']),
         ];
     }
 
@@ -296,7 +296,7 @@ private static function runIdsNeedingRebuild(array $analysis): array
      *     projected_runs_with_canonical: int,
      *     missing_run_ids: list<string>,
      *     stale_run_ids: list<string>,
-     *     legacy_schema_run_ids: list<string>
+     *     schema_version_mismatch_run_ids: list<string>
      * }
      */
     private static function timerAnalysis(array $runIds, ?string $instanceId): array
@@ -305,13 +305,13 @@ private static function timerAnalysis(array $runIds, ?string $instanceId): array
         $projectedRunsWithCanonical = 0;
         $missingRunIds = [];
         $staleRunIds = [];
-        $legacySchemaRunIds = [];
+        $schemaVersionMismatchRunIds = [];
 
         self::runQuery(['timerEntries', 'timers', 'historyEvents'], $runIds, $instanceId)
             ->chunkById(100, static function ($runs) use (
-                &$legacySchemaRunIds,
                 &$missingRunIds,
                 &$projectedRunsWithCanonical,
+                &$schemaVersionMismatchRunIds,
                 &$runsWithCanonical,
                 &$staleRunIds,
             ): void {
@@ -334,8 +334,8 @@ private static function timerAnalysis(array $runIds, ?string $instanceId): array
                         $staleRunIds[] = $run->id;
                     }
 
-                    if ($status['legacy_schema']) {
-                        $legacySchemaRunIds[] = $run->id;
+                    if ($status['schema_version_mismatch']) {
+                        $schemaVersionMismatchRunIds[] = $run->id;
                     }
                 }
             }, 'id');
@@ -345,7 +345,7 @@ private static function timerAnalysis(array $runIds, ?string $instanceId): array
             'projected_runs_with_canonical' => $projectedRunsWithCanonical,
             'missing_run_ids' => array_values(array_unique($missingRunIds)),
             'stale_run_ids' => array_values(array_unique($staleRunIds)),
-            'legacy_schema_run_ids' => array_values(array_unique($legacySchemaRunIds)),
+            'schema_version_mismatch_run_ids' => array_values(array_unique($schemaVersionMismatchRunIds)),
         ];
     }
 

src/V2/Support/SelectedRunSnapshot.php

@@ -116,7 +116,7 @@ public static function timelineDriftStatus(WorkflowRun $run): array
      *     has_canonical: bool,
      *     missing: bool,
      *     stale: bool,
-     *     legacy_schema: bool,
+     *     schema_version_mismatch: bool,
      *     reasons: list<string>
      * }
      */

src/migrations/2026_04_11_000140_create_workflow_run_timer_entries_table.php

@@ -19,7 +19,7 @@ public function up(): void
                 ->index();
             $table->string('timer_id', 191);
             $table->unsignedSmallInteger('schema_version')
-                ->default(WorkflowRunTimerEntry::LEGACY_SCHEMA_VERSION);
+                ->default(WorkflowRunTimerEntry::CURRENT_SCHEMA_VERSION);
             $table->unsignedInteger('position');
             $table->unsignedInteger('sequence')
                 ->nullable()

tests/Feature/V2/V2OperatorMetricsTest.php

@@ -277,14 +277,14 @@ public function testSnapshotCountsStaleRunSummaryProjectionRows(): void
     public function testSnapshotCountsSelectedRunProjectionDrift(): void
     {
         $missingWaitRun = $this->createRunWithSummary(
-            instanceId: 'metrics-wait-missing-instance',
+            instanceId: 'metrics-wait-missing-i',
             runId: '01JMETRICSPROJWAITMISS01',
             status: 'waiting',
             statusBucket: 'running',
             livenessState: 'waiting_for_signal',
         );
         $projectedWaitRun = $this->createRunWithSummary(
-            instanceId: 'metrics-wait-projected-instance',
+            instanceId: 'metrics-wait-projected-i',
             runId: '01JMETRICSPROJWAITDONE01',
             status: 'waiting',
             statusBucket: 'running',
@@ -395,6 +395,7 @@ public function testSnapshotCountsSelectedRunProjectionDrift(): void
             'workflow_run_id' => $projectedWaitRun->id,
             'workflow_instance_id' => $projectedWaitRun->workflow_instance_id,
             'timer_id' => 'projection-timer-projected',
+            'schema_version' => WorkflowRunTimerEntry::CURRENT_SCHEMA_VERSION - 1,
             'position' => 0,
             'sequence' => 12,
             'status' => 'fired',
@@ -424,6 +425,7 @@ public function testSnapshotCountsSelectedRunProjectionDrift(): void
             'workflow_run_id' => '01JMETRICSPROJTIMERGONE01',
             'workflow_instance_id' => 'metrics-timer-orphan-instance',
             'timer_id' => 'projection-timer-orphan',
+            'schema_version' => WorkflowRunTimerEntry::CURRENT_SCHEMA_VERSION - 1,
             'position' => 0,
             'status' => 'pending',
             'source_status' => 'pending',
@@ -513,8 +515,8 @@ public function testSnapshotCountsSelectedRunProjectionDrift(): void
         $this->assertSame(1, $snapshot['projections']['run_timer_entries']['projected_runs_with_timers']);
         $this->assertSame(1, $snapshot['projections']['run_timer_entries']['missing_runs_with_timers']);
         $this->assertSame(1, $snapshot['projections']['run_timer_entries']['stale_projected_runs']);
-        $this->assertSame(1, $snapshot['projections']['run_timer_entries']['legacy_schema_runs']);
-        $this->assertSame(2, $snapshot['projections']['run_timer_entries']['legacy_schema_rows']);
+        $this->assertSame(1, $snapshot['projections']['run_timer_entries']['schema_version_mismatch_runs']);
+        $this->assertSame(2, $snapshot['projections']['run_timer_entries']['schema_version_mismatch_rows']);
         $this->assertSame(1, $snapshot['projections']['run_timer_entries']['orphaned']);
         $this->assertSame(3, $snapshot['projections']['run_timer_entries']['needs_rebuild']);
 

tests/Feature/V2/V2RunDetailViewTest.php

@@ -727,10 +727,10 @@ public function testRunDetailViewReadsTimerRowsFromRebuildableProjection(): void
         ]);
     }
 
-    public function testRunDetailViewRebuildsLegacyProjectedTimerRowsWithoutRowStatus(): void
+    public function testRunDetailViewRebuildsNonCurrentProjectedTimerRowsWithoutRowStatus(): void
     {
         $instance = WorkflowInstance::query()->create([
-            'id' => 'detail-projected-timer-compat',
+            'id' => 'detail-timer-proj-compat',
             'workflow_class' => TestTimerWorkflow::class,
             'workflow_type' => 'test-timer-workflow',
             'run_count' => 1,
@@ -789,7 +789,7 @@ public function testRunDetailViewRebuildsLegacyProjectedTimerRowsWithoutRowStatu
                 'workflow_run_id' => $run->id,
                 'workflow_instance_id' => $instance->id,
                 'timer_id' => $timer->id,
-                'schema_version' => WorkflowRunTimerEntry::LEGACY_SCHEMA_VERSION,
+                'schema_version' => WorkflowRunTimerEntry::CURRENT_SCHEMA_VERSION - 1,
                 'position' => 0,
                 'sequence' => 1,
                 'status' => 'pending',
@@ -809,7 +809,10 @@ public function testRunDetailViewRebuildsLegacyProjectedTimerRowsWithoutRowStatu
         $detail = RunDetailView::forRun($run->fresh(['summary', 'timerEntries']));
 
         $this->assertSame('workflow_run_timer_entries_rebuilt', $detail['timers_projection_source']);
-        $this->assertSame(['legacy_schema'], $detail['timers_projection_rebuild_reasons']);
+        $this->assertSame(
+            ['schema_version_mismatch', 'stale_projection'],
+            $detail['timers_projection_rebuild_reasons'],
+        );
         $this->assertCount(1, $detail['timers']);
         $this->assertSame('pending', $detail['timers'][0]['status']);
         $this->assertSame('pending', $detail['timers'][0]['source_status']);

tests/Unit/Commands/V2RebuildProjectionsCommandTest.php

@@ -421,13 +421,13 @@ public function testNeedsRebuildOptionIncludesRunsMissingLineageProjectionRows()
 
     public function testNeedsRebuildOptionIncludesRunsWithStaleSelectedRunProjectionPayloads(): void
     {
-        [, $waitRun] = $this->createWaitingRun('projection-command-stale-wait-selected');
-        [, $timelineRun] = $this->createCompletedRun('projection-command-stale-timeline-selected');
-        [, $lineageRun] = $this->createCompletedRun('projection-command-stale-lineage-selected');
-        [, $timerRun] = $this->createWaitingRun('projection-command-stale-timer-selected');
+        [, $waitRun] = $this->createWaitingRun('proj-stale-wait-sel');
+        [, $timelineRun] = $this->createCompletedRun('proj-stale-timeline-sel');
+        [, $lineageRun] = $this->createCompletedRun('proj-stale-lineage-sel');
+        [, $timerRun] = $this->createWaitingRun('proj-stale-timer-sel');
 
         ActivityExecution::query()->create([
-            'id' => 'projection-command-stale-wait-activity',
+            'id' => 'proj-stale-wait-activity',
             'workflow_run_id' => $waitRun->id,
             'sequence' => 1,
             'activity_class' => 'ProjectionSelectedWaitActivity',
@@ -451,7 +451,7 @@ public function testNeedsRebuildOptionIncludesRunsWithStaleSelectedRunProjection
             ],
         );
         WorkflowTimer::query()->create([
-            'id' => 'projection-command-stale-timer',
+            'id' => 'proj-stale-timer',
             'workflow_run_id' => $timerRun->id,
             'sequence' => 1,
             'status' => 'pending',
@@ -483,15 +483,15 @@ public function testNeedsRebuildOptionIncludesRunsWithStaleSelectedRunProjection
         WorkflowRunLineageEntry::query()
             ->where('workflow_run_id', $lineageRun->id)
             ->update([
-                'related_workflow_run_id' => 'projection-stale-child-run-wrong',
+                'related_workflow_run_id' => 'projection-stale-child-bad',
             ]);
         $timerEntry = WorkflowRunTimerEntry::query()
             ->where('workflow_run_id', $timerRun->id)
             ->firstOrFail();
         $timerPayload = $timerEntry->payload;
         unset($timerPayload['row_status']);
         $timerEntry->forceFill([
-            'schema_version' => WorkflowRunTimerEntry::LEGACY_SCHEMA_VERSION,
+            'schema_version' => WorkflowRunTimerEntry::CURRENT_SCHEMA_VERSION - 1,
             'payload' => $timerPayload,
         ])->save();
 
@@ -507,7 +507,7 @@ public function testNeedsRebuildOptionIncludesRunsWithStaleSelectedRunProjection
         ]);
         $this->assertDatabaseHas('workflow_run_waits', [
             'workflow_run_id' => $waitRun->id,
-            'wait_id' => 'activity:projection-command-stale-wait-activity',
+            'wait_id' => 'activity:proj-stale-wait-activity',
             'target_type' => 'projection.selected.wait',
         ]);
         $this->assertDatabaseMissing('workflow_run_timeline_entries', [
@@ -521,7 +521,7 @@ public function testNeedsRebuildOptionIncludesRunsWithStaleSelectedRunProjection
         ]);
         $this->assertDatabaseMissing('workflow_run_lineage_entries', [
             'workflow_run_id' => $lineageRun->id,
-            'related_workflow_run_id' => 'projection-stale-child-run-wrong',
+            'related_workflow_run_id' => 'projection-stale-child-bad',
         ]);
         $this->assertDatabaseHas('workflow_run_lineage_entries', [
             'workflow_run_id' => $lineageRun->id,
@@ -530,7 +530,7 @@ public function testNeedsRebuildOptionIncludesRunsWithStaleSelectedRunProjection
         ]);
         $this->assertDatabaseHas('workflow_run_timer_entries', [
             'workflow_run_id' => $timerRun->id,
-            'timer_id' => 'projection-command-stale-timer',
+            'timer_id' => 'proj-stale-timer',
             'schema_version' => WorkflowRunTimerEntry::CURRENT_SCHEMA_VERSION,
         ]);
     }