Bound schedule audit snapshots and clean up purged schedules

CHANGELOG.md

@@ -2,6 +2,13 @@
 
 ## Unreleased
 
+- Embedded schedule audits default to 1,000 snapshots, preserving the first and
+  newest events. Compact firing receipts keep occurrence deduplication intact;
+  a namespace-scoped prune command removes rows after physical schedule purge.
+  Run the occurrence migration and stop old schedule readers before enabling
+  compaction. Explicitly disable compaction during a rolling upgrade as described
+  in `docs/schedule-audit-retention.md`.
+
 - Add a namespace-scoped open-run version-usage query with legacy/current/unknown
   counts and byte-exact keyset pages. Missing markers and unavailable history
   remain unknown when assessing legacy branch retirement.

docs/schedule-audit-retention.md

@@ -0,0 +1,66 @@
+# Schedule audit retention
+
+Embedded schedules keep at most 1,000 audit snapshots by default. The first
+retained event (normally `ScheduleCreated`) and the newest events remain in
+their original form, with their original sequence numbers. New events continue
+after the latest retained sequence. Set `workflows.v2.schedules.audit_history_limit`
+or `DW_V2_SCHEDULE_AUDIT_HISTORY_LIMIT` to an integer of at least 2 to change the
+cap. This behavior requires the schedule occurrence migration and updated
+scheduler code; it is source work pending publication.
+
+Every audit append compacts at most 100 old snapshots inside the schedule's
+write transaction. A newly created schedule stays within its configured cap.
+A legacy schedule already above the cap approaches it in bounded passes.
+To compact an idle legacy schedule, repeat:
+
+```sh
+php artisan workflow:v2:schedule-audit-prune --namespace=orders --schedule=daily-invoices
+```
+
+The JSON response reports deleted and physical retained counts, including rows
+hidden by application query filters. Stop when the retained count reaches the
+configured limit. An invalid limit refuses compaction.
+
+## Firing receipts and upgrades
+
+An audit snapshot is not the authority for whether an occurrence may start
+again. Compact occurrence receipts preserve that decision after snapshots are
+removed. They contain only the internal schedule ID, namespace and canonical
+UTC occurrence time. They remain until the schedule is physically purged;
+the snapshot cap does not bound this correctness data. Old trigger metadata
+is copied before its audit row is deleted. Malformed occurrence metadata refuses
+the transaction and retains the original audit evidence.
+
+Stop old scheduler and schedule lifecycle writer processes before enabling
+compaction. Older packages read
+audit history for deduplication and cannot read the new compact receipts. For
+a staged upgrade, explicitly set `audit_history_limit` to PHP `null` in all
+new processes until every old scheduler and lifecycle writer has stopped. This temporarily
+disables snapshot compaction. Then enable the configured cap. Setting this
+value to zero or an invalid string does not disable retention; it is an error.
+The migration creates the receipt table without scanning all existing history.
+A schema rollback refuses to discard populated receipt storage.
+
+## Purged schedules
+
+Deleting a schedule through the API marks it deleted and keeps its bounded
+audit history and firing receipts. Once the schedule row is physically removed,
+run the following command in that namespace:
+
+```sh
+php artisan workflow:v2:schedule-audit-prune --namespace=orders --limit=500
+```
+
+Each pass deletes at most the selected number of orphan audit rows and receipts
+combined. Repeat until both deleted counts are zero. A remaining schedule row,
+including a soft deleted schedule, protects its rows. Namespace comparisons are
+exact; the command never sweeps other namespaces. Omission selects the configured
+namespace, or `default` when none is configured. Use `--null-namespace` instead
+of `--namespace` only for legacy rows whose namespace is SQL `NULL`.
+
+Custom schedule, audit and occurrence models use the configured workflow storage
+connection and table prefix. Applications which replace tables must also provide
+the corresponding migration and configure `workflows.v2.schedule_occurrence_model`.
+Application query filters cannot hide firing receipts or make a physically
+existing schedule eligible for orphan cleanup. The maintenance paths enforce
+namespace and internal owner identity explicitly.

src/Commands/V2ScheduleAuditPruneCommand.php

@@ -0,0 +1,70 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Workflow\Commands;
+
+use Illuminate\Console\Command;
+use LogicException;
+use Workflow\V2\Models\WorkflowSchedule;
+use Workflow\V2\Support\ConfiguredV2Models;
+use Workflow\V2\Support\NamespaceIdentity;
+use Workflow\V2\Support\ScheduleAuditRetention;
+
+class V2ScheduleAuditPruneCommand extends Command
+{
+    protected $signature = 'workflow:v2:schedule-audit-prune
+        {--namespace= : Exact namespace; defaults to configured namespace or default}
+        {--null-namespace : Select only legacy NULL-namespace rows}
+        {--schedule= : Compact this existing schedule ID instead of cleaning orphaned rows}
+        {--limit=500 : Maximum orphan rows deleted in this pass (1 to 1000)}';
+
+    protected $description = 'Bound schedule audit snapshots or clean audit rows left by purged schedules';
+
+    public function handle(): int
+    {
+        $namespace = $this->option('namespace');
+        $nullNamespace = (bool) $this->option('null-namespace');
+        if ($nullNamespace && $namespace !== null) {
+            $this->error('--namespace and --null-namespace cannot be combined.');
+            return self::FAILURE;
+        }
+        $namespace = $nullNamespace ? null : ($namespace ?? config('workflows.v2.namespace') ?? 'default');
+        if ($namespace !== null && (! is_string($namespace) || strlen($namespace) > 255
+            || preg_match('/(*UCP)\S/u', $namespace) !== 1)) {
+            $this->error('The selected namespace must be a nonempty UTF-8 string within 255 bytes.');
+            return self::FAILURE;
+        }
+        $limit = $this->option('limit');
+        if (is_string($limit) && preg_match('/\A[1-9][0-9]*\z/D', $limit) === 1) {
+            $limit = filter_var($limit, FILTER_VALIDATE_INT);
+        }
+        if (! is_int($limit) || $limit < 1 || $limit > 1000) {
+            $this->error('--limit must be an integer between 1 and 1000.');
+            return self::FAILURE;
+        }
+
+        try {
+            $id = $this->option('schedule');
+            if ($id === null) {
+                $report = ScheduleAuditRetention::cleanup($namespace, $limit);
+            } else {
+                $schedule = NamespaceIdentity::constrain(
+                    ConfiguredV2Models::query('schedule_model', WorkflowSchedule::class)->withoutGlobalScopes(),
+                    $namespace,
+                )->where('schedule_id', $id)
+                    ->first();
+                if ($schedule === null || $schedule->schedule_id !== $id) {
+                    $this->error('The exact schedule ID does not exist in the selected namespace.');
+                    return self::FAILURE;
+                }
+                $report = ScheduleAuditRetention::pruneWithReport($schedule);
+            }
+            $this->line(json_encode($report, JSON_THROW_ON_ERROR));
+        } catch (LogicException $error) {
+            $this->error($error->getMessage());
+            return self::FAILURE;
+        }
+        return self::SUCCESS;
+    }
+}

src/Providers/WorkflowServiceProvider.php

@@ -24,6 +24,7 @@
 use Workflow\Commands\V2ReplayConformanceCommand;
 use Workflow\Commands\V2ReplaySimulateCommand;
 use Workflow\Commands\V2ReplayVerifyCommand;
+use Workflow\Commands\V2ScheduleAuditPruneCommand;
 use Workflow\Commands\V2ScheduleTickCommand;
 use Workflow\Commands\V2UpgradeStatusCommand;
 use Workflow\Commands\WorkflowMakeCommand;
@@ -206,6 +207,7 @@ public function boot(): void
             V2ReplaySimulateCommand::class,
             V2ReplayVerifyCommand::class,
             V2ScheduleTickCommand::class,
+            V2ScheduleAuditPruneCommand::class,
             V2UpgradeStatusCommand::class,
         ]);
 

src/V2/Models/WorkflowScheduleHistoryEvent.php

@@ -15,6 +15,7 @@
 use Workflow\V2\Enums\HistoryEventType;
 use Workflow\V2\Support\ConfiguredV2Models;
 use Workflow\V2\Support\HistoryEventPayloadContract;
+use Workflow\V2\Support\ScheduleAuditRetention;
 
 class WorkflowScheduleHistoryEvent extends Model
 {
@@ -63,6 +64,28 @@ public static function record(
         array $payload = [],
     ): self {
         HistoryEventPayloadContract::assertKnownPayloadKeys($eventType, $payload);
+        $snapshot = self::snapshotPayload($schedule, $payload);
+
+        return ScheduleAuditRetention::locked(
+            $schedule,
+            static fn (WorkflowSchedule $locked): self => self::recordLocked($locked, $eventType, $snapshot),
+        );
+    }
+
+    public static function utcOccurrenceKey(DateTimeInterface $occurrence): string
+    {
+        return \Carbon\CarbonImmutable::instance($occurrence)
+            ->utc()
+            ->format('Y-m-d H:i:s.u');
+    }
+
+    private static function recordLocked(
+        WorkflowSchedule $schedule,
+        HistoryEventType $eventType,
+        array $payload,
+    ): self {
+        // Validate configuration before appending an event.
+        ScheduleAuditRetention::limit();
 
         $snapshot = self::snapshotPayload($schedule, $payload);
         $workflowInstanceId = self::stringValue($payload['workflow_instance_id'] ?? null);
@@ -73,48 +96,56 @@ public static function record(
         $occurrenceAtUtc = $occurrenceTime !== null
             ? self::utcOccurrenceKey(new DateTimeImmutable($occurrenceTime))
             : null;
+        $historyModel = ConfiguredV2Models::resolve('schedule_history_event_model', self::class);
+        $history = (new $historyModel())->setConnection($schedule->getConnectionName())
+            ->newQueryWithoutScopes();
 
         for ($attempt = 1; $attempt <= self::SEQUENCE_RETRY_LIMIT; $attempt++) {
-            $sequence = ((int) ConfiguredV2Models::query('schedule_history_event_model', self::class)
-                ->where('workflow_schedule_id', $schedule->id)
-                ->max('sequence')) + 1;
+            $sequenceQuery = (clone $history)->where('workflow_schedule_id', $schedule->id);
+            // A committed old writer can win after our first read. On retry,
+            // bypass a repeatable-read snapshot without leaving this transaction.
+            $lastSequence = $attempt === 1 ? $sequenceQuery->max('sequence')
+                : $sequenceQuery->orderByDesc('sequence')
+                    ->lockForUpdate()
+                    ->value('sequence');
+            $sequence = ((int) $lastSequence) + 1;
+            $attributes = [
+                'workflow_schedule_id' => $schedule->id,
+                'schedule_id' => $schedule->schedule_id,
+                'namespace' => $schedule->namespace,
+                'sequence' => $sequence,
+                'event_type' => $eventType->value,
+                'payload' => $snapshot,
+                'workflow_instance_id' => $workflowInstanceId,
+                'workflow_run_id' => $workflowRunId,
+                'occurrence_at_utc' => $occurrenceAtUtc,
+                'recorded_at' => now(),
+            ];
 
             try {
                 /** @var self $event */
-                $event = ConfiguredV2Models::query('schedule_history_event_model', self::class)->create([
-                    'workflow_schedule_id' => $schedule->id,
-                    'schedule_id' => $schedule->schedule_id,
-                    'namespace' => $schedule->namespace,
-                    'sequence' => $sequence,
-                    'event_type' => $eventType->value,
-                    'payload' => $snapshot,
-                    'workflow_instance_id' => $workflowInstanceId,
-                    'workflow_run_id' => $workflowRunId,
-                    'occurrence_at_utc' => $occurrenceAtUtc,
-                    'recorded_at' => now(),
-                ]);
-
-                return $event;
+                $event = $schedule->getConnection()
+                    ->transaction(static fn (): self => (clone $history)->create($attributes));
             } catch (UniqueConstraintViolationException $e) {
                 if ($attempt === self::SEQUENCE_RETRY_LIMIT) {
                     throw $e;
                 }
                 // Another writer claimed this sequence first. Re-read the max
                 // and try the next slot.
+                continue;
             }
+            if ($occurrenceAtUtc !== null) {
+                ScheduleAuditRetention::retainOccurrence($schedule, $occurrenceAtUtc);
+            }
+            ScheduleAuditRetention::pruneLocked($schedule);
+
+            return $event;
         }
 
         // Unreachable: the loop either returns or re-throws at the final attempt.
         throw new \LogicException('Schedule history sequence allocation exhausted retries.');
     }
 
-    public static function utcOccurrenceKey(DateTimeInterface $occurrence): string
-    {
-        return \Carbon\CarbonImmutable::instance($occurrence)
-            ->utc()
-            ->format('Y-m-d H:i:s.u');
-    }
-
     /**
      * @param  array<string, mixed>  $payload
      * @return array<string, mixed>

src/V2/Models/WorkflowScheduleOccurrence.php

@@ -0,0 +1,26 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Workflow\V2\Models;
+
+use Illuminate\Database\Eloquent\Concerns\HasUlids;
+use Illuminate\Database\Eloquent\Model;
+use Workflow\Traits\ResolvesStorageConnection;
+
+/** Compact firing receipts outlive the bounded audit snapshots. */
+class WorkflowScheduleOccurrence extends Model
+{
+    use HasUlids;
+    use ResolvesStorageConnection;
+
+    public $incrementing = false;
+
+    public $timestamps = false;
+
+    protected $table = 'workflow_schedule_occurrences';
+
+    protected $guarded = [];
+
+    protected $keyType = 'string';
+}