From 60c42603098d33c57134e5e1d02df1e1567abfd0 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 14 Nov 2026 10:39:28 +0000 Subject: [PATCH 01/11] docs(versioning): define open-run version usage query contract --- docs/version-usage.md | 107 +++++++++ .../contracts/version-usage-v1.schema.json | 204 ++++++++++++++++++ 2 files changed, 311 insertions(+) create mode 100644 docs/version-usage.md create mode 100644 resources/contracts/version-usage-v1.schema.json diff --git a/docs/version-usage.md b/docs/version-usage.md new file mode 100644 index 000000000..2a18d092a --- /dev/null +++ b/docs/version-usage.md @@ -0,0 +1,107 @@ +# Open-run version usage (draft v1) + +This read-only contract implements durable-workflow/.github#220. It answers which +open runs still have a recorded decision below a proposed minimum supported +version. It does not replay workflows, run application code, write a projection, +change markers, or authorize deleting a branch. + +The native control-plane route is `GET /workflows/version-usage`, relative to the +existing control-plane API root. Authentication and namespace selection use the +existing runtime request context. The embedded entry point is +`Workflow\V2\Support\VersionUsageQuery::page()`; its namespace argument is required, +and `null` selects only the embedded default namespace, never all namespaces. +Namespace, workflow type and change ID comparisons are exact and case-sensitive. + +## Request + +- `workflow_type`: required nonempty UTF-8 string, at most 255 bytes. +- `change_id`: required nonempty UTF-8 string, at most 1024 bytes. +- `min_version`: required signed 32-bit integer, from -2147483648 to 2147483647. + This is the proposed new `minSupported`, not the largest version deployed. + For removing a `patched()` legacy branch, use 1. +- `state`: optional `legacy`, `current`, or `unknown`; omitted means all states. +- `limit`: integer from 1 to 200, default 50. +- `cursor`: optional opaque continuation from this exact namespace/type/change/ + minimum/state query. It is not an authorization token. Invalid or mismatched + cursors are errors; they never fall back to a first page. + +Invalid arguments return the existing invalid-request error envelope (HTTP 422 +for the native endpoint). Storage or connection failures are errors, not empty +successful answers. Implementations must use the configured storage models and +connection, including custom table names and prefixes. + +## Classification + +Only runs whose canonical status is open participate (embedded: pending, +running, waiting). Retained closed, archived terminal and prior continued runs +are excluded. An archived run which is still open remains included. + +- `legacy`: a retained, valid marker for this exact change ID has version less + than `min_version`. +- `current`: its recorded version is greater than or equal to `min_version`. +- `unknown`: no marker exists, or retained history cannot establish the decision. + +Each run has `recorded_version` only for legacy/current. Unknown reasons are: + +- `marker_absent`: retained history has no decision for this change ID, including + an older definition that has not reached the change point. A matching current + code fingerprint, build label, admission time or deployment time never upgrades + this to current. +- `history_unavailable`: details are pruned, the initial history/prefix is absent + or incomplete, or the stored event format is not supported. +- `history_invalid`: malformed decision data, invalid recorded range, or + conflicting decisions for the same change ID. Repeated identical historical + decisions remain valid. Marker integer fields use the signed 32-bit range. + +Marker readers must understand their runtime's retained format. Embedded marker +fields are `change_id`, `version`, `min_supported`, `max_supported` and positive +logical `sequence`; the range must contain the recorded version. No application +argument, output, memo or payload-store fetch is necessary for this query. + +## Response and paging + +The response shape is in +[`resources/contracts/version-usage-v1.schema.json`](../resources/contracts/version-usage-v1.schema.json). +`counts.total` equals legacy + current + unknown for the entire bounded open-run +cohort, before the optional state filter and independently of page size/cursor. +`runs` contains only the selected state when a state filter is present. + +Runs are ordered by run ID ascending, with a strict keyset continuation. The +first request captures the greatest matching open run ID as `through_run_id`; +all continuations retain that upper bound and advance after the last returned +ID. New admissions beyond that bound cannot appear on later pages. Cursors bind +all identity/classification filters, and every page re-applies authorization and +exact namespace predicates. Namespace changes cannot reuse a cursor. + +This is an explicitly **live**, read-only observation, not a durable fleet +snapshot or deployment fence. Runs may close or record a marker between pages; +counts are refreshed over the same upper-bounded cohort and may change. Returned +IDs do not repeat within a cursor traversal, but a newly matching state behind +the cursor requires a new traversal. Counts and a page are derived from the same +per-run observations in each response. No response contains a `safe_to_remove` +flag. Operators must keep unknowns visible and restart the query after draining +or otherwise fencing admissions before making a rollout/removal decision. + +Example shape (the cursor is opaque): + +```json +{ + "schema_version": 1, + "namespace": "orders", + "workflow_type": "shipment", + "change_id": "carrier-api", + "min_version": 2, + "state": null, + "consistency": "live", + "through_run_id": "01J00000000000000000000002", + "counts": {"total": 3, "legacy": 1, "current": 1, "unknown": 1}, + "runs": [{"instance_id": "shipment-1", "run_id": "01J00000000000000000000001", "status": "waiting", "state": "legacy", "recorded_version": 1, "unknown_reason": null}], + "limit": 1, + "returned_count": 1, + "has_more": true, + "next_cursor": "opaque-runtime-cursor" +} +``` + +This draft is source-contract work only. Native endpoint/CLI/Waterline adoption +and installed compatibility remain owned by their respective implementations. diff --git a/resources/contracts/version-usage-v1.schema.json b/resources/contracts/version-usage-v1.schema.json new file mode 100644 index 000000000..bf87b1885 --- /dev/null +++ b/resources/contracts/version-usage-v1.schema.json @@ -0,0 +1,204 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "title": "Workflow open-run version usage v1", + "type": "object", + "additionalProperties": false, + "required": [ + "schema_version", + "namespace", + "workflow_type", + "change_id", + "min_version", + "state", + "consistency", + "through_run_id", + "counts", + "runs", + "limit", + "returned_count", + "has_more", + "next_cursor" + ], + "properties": { + "schema_version": { + "const": 1 + }, + "namespace": { + "type": [ + "string", + "null" + ] + }, + "workflow_type": { + "type": "string", + "minLength": 1 + }, + "change_id": { + "type": "string", + "minLength": 1 + }, + "min_version": { + "type": "integer", + "minimum": -2147483648, + "maximum": 2147483647 + }, + "state": { + "enum": [ + null, + "legacy", + "current", + "unknown" + ] + }, + "consistency": { + "const": "live" + }, + "through_run_id": { + "type": [ + "string", + "null" + ] + }, + "counts": { + "type": "object", + "additionalProperties": false, + "required": [ + "total", + "legacy", + "current", + "unknown" + ], + "properties": { + "total": { + "type": "integer", + "minimum": 0 + }, + "legacy": { + "type": "integer", + "minimum": 0 + }, + "current": { + "type": "integer", + "minimum": 0 + }, + "unknown": { + "type": "integer", + "minimum": 0 + } + } + }, + "runs": { + "type": "array", + "maxItems": 200, + "items": { + "$ref": "#/$defs/run" + } + }, + "limit": { + "type": "integer", + "minimum": 1, + "maximum": 200 + }, + "returned_count": { + "type": "integer", + "minimum": 0, + "maximum": 200 + }, + "has_more": { + "type": "boolean" + }, + "next_cursor": { + "type": [ + "string", + "null" + ] + } + }, + "$defs": { + "run": { + "type": "object", + "additionalProperties": false, + "required": [ + "instance_id", + "run_id", + "status", + "state", + "recorded_version", + "unknown_reason" + ], + "properties": { + "instance_id": { + "type": "string" + }, + "run_id": { + "type": "string" + }, + "status": { + "enum": [ + "pending", + "running", + "waiting" + ] + }, + "state": { + "enum": [ + "legacy", + "current", + "unknown" + ] + }, + "recorded_version": { + "type": [ + "integer", + "null" + ], + "minimum": -2147483648, + "maximum": 2147483647 + }, + "unknown_reason": { + "enum": [ + null, + "marker_absent", + "history_unavailable", + "history_invalid" + ] + } + }, + "allOf": [ + { + "if": { + "properties": { + "state": { + "const": "unknown" + } + } + }, + "then": { + "properties": { + "recorded_version": { + "type": "null" + }, + "unknown_reason": { + "enum": [ + "marker_absent", + "history_unavailable", + "history_invalid" + ] + } + } + }, + "else": { + "properties": { + "recorded_version": { + "type": "integer" + }, + "unknown_reason": { + "type": "null" + } + } + } + } + ] + } + } +} From 4ae0b68f367eaa1311e60f529474984f780a6951 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 14 Nov 2026 11:13:27 +0000 Subject: [PATCH 02/11] feat(versioning): query retained open-run version usage --- CHANGELOG.md | 4 + docs/version-usage.md | 11 +- src/V2/Support/VersionUsageCursor.php | 53 ++ src/V2/Support/VersionUsageDecision.php | 83 ++++ src/V2/Support/VersionUsageQuery.php | 257 ++++++++++ tests/Feature/V2/V2VersionUsageQueryTest.php | 479 +++++++++++++++++++ tests/Unit/V2/VersionUsageCursorTest.php | 59 +++ tests/Unit/V2/VersionUsageDecisionTest.php | 175 +++++++ 8 files changed, 1116 insertions(+), 5 deletions(-) create mode 100644 src/V2/Support/VersionUsageCursor.php create mode 100644 src/V2/Support/VersionUsageDecision.php create mode 100644 src/V2/Support/VersionUsageQuery.php create mode 100644 tests/Feature/V2/V2VersionUsageQueryTest.php create mode 100644 tests/Unit/V2/VersionUsageCursorTest.php create mode 100644 tests/Unit/V2/VersionUsageDecisionTest.php diff --git a/CHANGELOG.md b/CHANGELOG.md index 937a17772..451505245 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,10 @@ ## Unreleased +- 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 (#220). + - Experimental embedded compensation scopes (#1063): retained versioned obligations, native Saga recovery, sequential child participation and explicit release. Each scope keeps its originating run; continue-as-new, origin redrive diff --git a/docs/version-usage.md b/docs/version-usage.md index 2a18d092a..9e4ad91ea 100644 --- a/docs/version-usage.md +++ b/docs/version-usage.md @@ -5,7 +5,7 @@ open runs still have a recorded decision below a proposed minimum supported version. It does not replay workflows, run application code, write a projection, change markers, or authorize deleting a branch. -The native control-plane route is `GET /workflows/version-usage`, relative to the +The native control-plane route is `GET /workflow-version-usage`, relative to the existing control-plane API root. Authentication and namespace selection use the existing runtime request context. The embedded entry point is `Workflow\V2\Support\VersionUsageQuery::page()`; its namespace argument is required, @@ -48,10 +48,11 @@ Each run has `recorded_version` only for legacy/current. Unknown reasons are: code fingerprint, build label, admission time or deployment time never upgrades this to current. - `history_unavailable`: details are pruned, the initial history/prefix is absent - or incomplete, or the stored event format is not supported. + or incomplete, the stored event format is not supported, or the bounded + inspection budget (1,000 version markers per run) is exceeded. - `history_invalid`: malformed decision data, invalid recorded range, or - conflicting decisions for the same change ID. Repeated identical historical - decisions remain valid. Marker integer fields use the signed 32-bit range. + conflicting decisions for the same change ID. Repeated decisions with the same version remain valid when each recorded + range is valid; range metadata may differ across repeated historical calls. Marker integer fields use the signed 32-bit range. Marker readers must understand their runtime's retained format. Embedded marker fields are `change_id`, `version`, `min_supported`, `max_supported` and positive @@ -66,7 +67,7 @@ The response shape is in cohort, before the optional state filter and independently of page size/cursor. `runs` contains only the selected state when a state filter is present. -Runs are ordered by run ID ascending, with a strict keyset continuation. The +Runs are ordered by byte-exact run ID ascending, with a strict keyset continuation. The first request captures the greatest matching open run ID as `through_run_id`; all continuations retain that upper bound and advance after the last returned ID. New admissions beyond that bound cannot appear on later pages. Cursors bind diff --git a/src/V2/Support/VersionUsageCursor.php b/src/V2/Support/VersionUsageCursor.php new file mode 100644 index 000000000..8f4269611 --- /dev/null +++ b/src/V2/Support/VersionUsageCursor.php @@ -0,0 +1,53 @@ + 1, + 'query' => $query, + 'through' => $through, + 'after' => $after, + ], JSON_THROW_ON_ERROR)), '+/', '-_'), '='); + } + + public static function decode(string $cursor, string $query): array + { + if (strlen($cursor) > 2048 || preg_match('/\A[A-Za-z0-9_-]+\z/D', $cursor) !== 1) { + throw new InvalidArgumentException('Invalid version-usage cursor.'); + } + $decoded = base64_decode(strtr($cursor, '-_', '+/'), true); + try { + $value = $decoded === false ? null : json_decode($decoded, true, 8, JSON_THROW_ON_ERROR); + } catch (JsonException) { + $value = null; + } + if (! is_array($value)) { + throw new InvalidArgumentException('Invalid version-usage cursor.'); + } + $keys = array_keys($value); + sort($keys); + if ($keys !== ['after', 'query', 'through', 'version'] || $value['version'] !== 1 + || ! is_string($value['query']) || ! hash_equals($query, $value['query']) + || ! self::runId($value['after']) || ! self::runId($value['through']) + || strcmp($value['after'], $value['through']) > 0 + || self::encode($query, $value['through'], $value['after']) !== $cursor) { + throw new InvalidArgumentException('Version-usage cursor does not match this query.'); + } + return $value; + } + + private static function runId(mixed $value): bool + { + return is_string($value) && preg_match('/\A[A-Za-z0-9][A-Za-z0-9_.:-]{0,190}\z/D', $value) === 1; + } +} diff --git a/src/V2/Support/VersionUsageDecision.php b/src/V2/Support/VersionUsageDecision.php new file mode 100644 index 000000000..cc81f3d78 --- /dev/null +++ b/src/V2/Support/VersionUsageDecision.php @@ -0,0 +1,83 @@ +reason = $historyAvailable ? 'marker_absent' : 'history_unavailable'; + } + + public function observe(mixed $payload): void + { + if (in_array($this->reason, ['history_invalid', 'history_unavailable'], true)) { + return; + } + if (count($this->sequences) >= 1000) { + $this->reason = 'history_unavailable'; + return; + } + if (! is_array($payload) + || ! is_string($payload['change_id'] ?? null) || $payload['change_id'] === '' + || ! is_int($payload['sequence'] ?? null) || $payload['sequence'] < 1 + || isset($this->sequences[$payload['sequence']]) + || ! self::integer($payload['version'] ?? null) + || ! self::integer($payload['min_supported'] ?? null) + || ! self::integer($payload['max_supported'] ?? null) + || $payload['min_supported'] > $payload['max_supported'] + || $payload['version'] < $payload['min_supported'] + || $payload['version'] > $payload['max_supported']) { + $this->reason = 'history_invalid'; + return; + } + if (array_diff(array_keys($payload), [ + 'sequence', 'change_id', 'version', 'min_supported', 'max_supported', + 'reused_from_run_id', 'reused_recorded_at', + ]) !== []) { + $this->reason = 'history_unavailable'; + return; + } + $this->sequences[$payload['sequence']] = true; + if ($payload['change_id'] !== $this->changeId) { + return; + } + if ($this->version !== null && $this->version !== $payload['version']) { + $this->reason = 'history_invalid'; + return; + } + $this->version = $payload['version']; + $this->reason = null; + } + + public function unavailable(): void + { + $this->reason = 'history_unavailable'; + } + + public function result(): array + { + return [ + 'state' => $this->reason !== null ? 'unknown' : ($this->version < $this->minVersion ? 'legacy' : 'current'), + 'recorded_version' => $this->reason === null ? $this->version : null, + 'unknown_reason' => $this->reason, + ]; + } + + public static function integer(mixed $value): bool + { + return is_int($value) && $value >= -2147483648 && $value <= 2147483647; + } +} diff --git a/src/V2/Support/VersionUsageQuery.php b/src/V2/Support/VersionUsageQuery.php new file mode 100644 index 000000000..7b48b8546 --- /dev/null +++ b/src/V2/Support/VersionUsageQuery.php @@ -0,0 +1,257 @@ + 200 + || ! in_array($state, [null, 'legacy', 'current', 'unknown'], true)) { + throw new InvalidArgumentException('Invalid version-usage threshold, state or page limit.'); + } + $identity = hash('sha256', json_encode([ + $namespace, $workflowType, $changeId, $minVersion, $state, + ], JSON_THROW_ON_ERROR)); + $continuation = $cursor === null ? null : VersionUsageCursor::decode($cursor, $identity); + $base = $this->runs($namespace, $workflowType); + $through = $continuation['through'] ?? (clone $base)->orderByRaw($this->idExpression($base) . ' DESC')->value( + 'id' + ); + $after = $continuation['after'] ?? null; + $counts = [ + 'total' => 0, + 'legacy' => 0, + 'current' => 0, + 'unknown' => 0, + ]; + $results = []; + if ($through !== null) { + $model = $base->getModel(); + foreach ($this->batches($base, $through) as $runs) { + $decisions = $this->decisions($runs, $model, $changeId, $minVersion); + foreach ($runs as $run) { + $decision = $decisions[$run->id]->result(); + ++$counts['total']; + ++$counts[$decision['state']]; + if (($state === null || $decision['state'] === $state) + && ($after === null || strcmp($run->id, $after) > 0) + && count($results) <= $limit) { + $results[] = [ + 'instance_id' => $run->workflow_instance_id, + 'run_id' => $run->id, + 'status' => $run->status, + ] + $decision; + } + } + } + } + $hasMore = count($results) > $limit; + $results = array_slice($results, 0, $limit); + return [ + 'schema_version' => 1, + 'namespace' => $namespace, + 'workflow_type' => $workflowType, + 'change_id' => $changeId, + 'min_version' => $minVersion, + 'state' => $state, + 'consistency' => 'live', + 'through_run_id' => $through, + 'counts' => $counts, + 'runs' => $results, + 'limit' => $limit, + 'returned_count' => count($results), + 'has_more' => $hasMore, + 'next_cursor' => $hasMore ? VersionUsageCursor::encode( + $identity, + $through, + $results[array_key_last($results)]['run_id'] + ) : null, + ]; + } + + /** + * @return Generator + */ + private function batches(Builder $base, string $through): Generator + { + $id = $this->idExpression($base); + $parameter = self::binary('?', $base->getQuery()->getConnection()->getDriverName()); + $after = null; + do { + $query = (clone $base)->select([ + 'id', 'workflow_instance_id', 'status', 'details_pruned_at', 'last_history_sequence', + ])->whereRaw($id . ' <= ' . $parameter, [$through])->orderByRaw($id . ' ASC'); + if ($after !== null) { + $query->whereRaw($id . ' > ' . $parameter, [$after]); + } + $rows = $query->limit(100) + ->toBase() + ->get(); + if ($rows->isEmpty()) { + return; + } + yield $rows; + $after = $rows->last() +->id; + } while ($rows->count() === 100); + } + + private function idExpression(Builder $query): string + { + return self::binary( + $query->getQuery() + ->getGrammar() + ->wrap($query->getModel()->qualifyColumn('id')), + $query->getQuery() + ->getConnection() + ->getDriverName(), + ); + } + + /** + * @return Builder + */ + private function runs(?string $namespace, string $workflowType): Builder + { + $model = ConfiguredV2Models::resolve('run_model', WorkflowRun::class); + $query = NamespaceIdentity::constrain($model::query()->setEagerLoads([]), $namespace) + ->where('workflow_type', $workflowType) + ->whereIn('status', [RunStatus::Pending->value, RunStatus::Running->value, RunStatus::Waiting->value]); + $column = $query->getQuery() + ->getGrammar() + ->wrap($query->getModel()->qualifyColumn('workflow_type')); + $driver = $query->getQuery() + ->getConnection() + ->getDriverName(); + $query->whereRaw(self::binary($column, $driver) . ' = ' . self::binary('?', $driver), [$workflowType]); + return $query; + } + + /** + * @return array + */ + private function decisions(Collection $runs, WorkflowRun $runModel, string $changeId, int $minVersion): array + { + $eventModel = ConfiguredV2Models::resolve('history_event_model', WorkflowHistoryEvent::class); + $history = (new $eventModel())->setConnection($runModel->getConnectionName()) + ->newQuery() + ->setEagerLoads([])->whereIn('workflow_run_id', $runs->pluck('id')->all())->toBase(); + $driver = $history->getConnection() + ->getDriverName(); + $sequence = $history->getGrammar() + ->wrap('sequence'); + $kind = self::binary($history->getGrammar()->wrap('event_type'), $driver); + $marker = self::binary('?', $driver); + $statistics = (clone $history)->select('workflow_run_id') + ->selectRaw( + "COUNT(*) AS event_count, MIN({$sequence}) AS first_sequence, MAX({$sequence}) AS last_sequence, " + . "SUM(CASE WHEN {$kind} = {$marker} THEN 1 ELSE 0 END) AS marker_count", + [HistoryEventType::VersionMarkerRecorded->value] + ) + ->groupBy('workflow_run_id') + ->get() + ->keyBy('workflow_run_id'); + $starts = (clone $history)->where('sequence', 1) + ->pluck('event_type', 'workflow_run_id'); + $types = array_map(static fn (HistoryEventType $type): string => $type->value, HistoryEventType::cases()); + $driver = $history->getConnection() + ->getDriverName(); + $kind = self::binary($history->getGrammar()->wrap('event_type'), $driver); + $parameters = implode(', ', array_fill(0, count($types), self::binary('?', $driver))); + $unsupported = (clone $history)->whereRaw($kind . ' NOT IN (' . $parameters . ')', $types) + ->distinct() + ->pluck('workflow_run_id') + ->all(); + $startedType = HistoryEventType::WorkflowStarted->value; + $decisions = []; + $inspect = []; + foreach ($runs as $run) { + $stats = $statistics->get($run->id); + $available = $run->details_pruned_at === null && $stats !== null + && (int) $stats->first_sequence === 1 + && (int) $stats->last_sequence === (int) $run->last_history_sequence + && (int) $stats->event_count === (int) $run->last_history_sequence + && (int) $stats->marker_count <= 1000 + && $starts->get($run->id) === $startedType + && ! in_array($run->id, $unsupported, true); + $decisions[$run->id] = new VersionUsageDecision($changeId, $minVersion, $available); + if ($available) { + $inspect[] = $run->id; + } + } + foreach ((clone $history)->whereIn('workflow_run_id', $inspect)->where( + 'event_type', + HistoryEventType::VersionMarkerRecorded->value + ) + ->lazyById(250) as $event) { + if ($event->event_type !== HistoryEventType::VersionMarkerRecorded->value) { + $decisions[$event->workflow_run_id]->unavailable(); + continue; + } + try { + $payload = is_string($event->payload) + ? json_decode($event->payload, true, 32, JSON_THROW_ON_ERROR) : $event->payload; + } catch (JsonException) { + $payload = null; + } + $decisions[$event->workflow_run_id]->observe($payload); + } + // A prune or append racing this observation cannot turn a partial read into current usage. + $latest = $runModel->newQuery() + ->setEagerLoads([])->whereIn('id', $runs->pluck('id')->all()) + ->toBase() + ->get(['id', 'last_history_sequence', 'details_pruned_at'])->keyBy('id'); + foreach ($runs as $run) { + $end = $latest->get($run->id); + if ($end === null || $end->details_pruned_at !== null + || (int) $end->last_history_sequence !== (int) $run->last_history_sequence) { + $decisions[$run->id]->unavailable(); + } + } + return $decisions; + } + + private static function binary(string $expression, string $driver): string + { + return match ($driver) { + 'mysql', 'mariadb' => "CAST({$expression} AS BINARY)", + 'sqlite' => "CAST({$expression} AS BLOB)", + 'pgsql' => "CAST({$expression} AS TEXT) COLLATE \"C\"", + 'sqlsrv' => "CONVERT(varbinary(max), CAST({$expression} AS nvarchar(max)))", + default => throw new InvalidArgumentException('Unsupported storage driver for exact version usage.'), + }; + } + + private static function text(string $value, int $limit): void + { + if ($value === '' || strlen($value) > $limit || preg_match('//u', $value) !== 1) { + throw new InvalidArgumentException('Version-usage identities must be nonempty bounded UTF-8 strings.'); + } + } +} diff --git a/tests/Feature/V2/V2VersionUsageQueryTest.php b/tests/Feature/V2/V2VersionUsageQueryTest.php new file mode 100644 index 000000000..0538534c7 --- /dev/null +++ b/tests/Feature/V2/V2VersionUsageQueryTest.php @@ -0,0 +1,479 @@ +createRun(1, 1); + $this->createRun(2, 2); + $this->createRun(3, null); + $this->createRun(4, -1); + $query = new VersionUsageQuery(); + $first = $query->page('orders', 'shipment', 'shipping', 2, 1); + $this->assertSame([ + 'total' => 4, + 'legacy' => 2, + 'current' => 1, + 'unknown' => 1, + ], $first['counts']); + $this->assertSame('legacy', $first['runs'][0]['state']); + $this->assertTrue($first['has_more']); + $current = $query->page('orders', 'shipment', 'shipping', 2, 1, state: 'current'); + $this->assertSame($first['counts'], $current['counts']); + $this->assertSame(2, $current['runs'][0]['recorded_version']); + $this->assertFalse($current['has_more']); + $unknown = $query->page('orders', 'shipment', 'shipping', 2, state: 'unknown'); + $this->assertSame('marker_absent', $unknown['runs'][0]['unknown_reason']); + $this->assertNull($unknown['runs'][0]['recorded_version']); + } + + public function testNamespacesWorkflowTypesAndChangeIdsUseExactIdentities(): void + { + foreach ([null, 'default', 'orders', 'ORDERS', 'orders '] as $index => $namespace) { + $this->createRun($index + 1, 1, $namespace); + } + $this->createRun(6, 2, 'orders', 'Shipment'); + $this->createRun(7, 2, 'orders', 'shipment '); + $query = new VersionUsageQuery(); + foreach ([null, 'default', 'orders', 'ORDERS', 'orders '] as $namespace) { + $page = $query->page($namespace, 'shipment', 'shipping', 2); + $this->assertSame(1, $page['counts']['total']); + $this->assertSame(1, $page['counts']['legacy']); + $this->assertSame($namespace, $page['namespace']); + } + $this->assertSame(1, $query->page('orders', 'Shipment', 'shipping', 2)['counts']['current']); + $this->assertSame(1, $query->page('orders', 'shipment', 'Shipping', 2)['counts']['unknown']); + } + + public function testClosedRunsAreExcludedButOpenArchivedRunsAreStillCounted(): void + { + $this->createRun(1, 1) + ->update([ + 'status' => 'completed', + ]); + $this->createRun(2, 1) + ->update([ + 'status' => 'failed', + ]); + $this->createRun(3, 1) + ->update([ + 'status' => 'cancelled', + ]); + $this->createRun(4, 1) + ->update([ + 'status' => 'terminated', + ]); + $this->createRun(5, null) + ->update([ + 'archived_at' => now(), + ]); + $page = (new VersionUsageQuery())->page('orders', 'shipment', 'shipping', 2); + $this->assertSame([ + 'total' => 1, + 'legacy' => 0, + 'current' => 0, + 'unknown' => 1, + ], $page['counts']); + } + + public function testPrunedMissingGappedAndUnsupportedHistoriesStayUnknownEvenWithNewMarkers(): void + { + $pruned = $this->createRun(1, 2); + $pruned->update([ + 'details_pruned_at' => now(), + ]); + $missing = $this->createRun(2, 2); + WorkflowHistoryEvent::query()->where('workflow_run_id', $missing->id)->delete(); + $gap = $this->createRun(3, 2); + WorkflowHistoryEvent::query()->where('workflow_run_id', $gap->id)->where('sequence', 1)->delete(); + $future = $this->createRun(4, 2); + $this->event($future, 'FutureHistoryEvent', []); + $caseVariant = $this->createRun(5, 2); + DB::table((new WorkflowHistoryEvent())->getTable())->where('workflow_run_id', $caseVariant->id) + ->where('sequence', 2) + ->update([ + 'event_type' => 'versionmarkerrecorded', + ]); + $page = (new VersionUsageQuery())->page('orders', 'shipment', 'shipping', 2); + $this->assertSame([ + 'total' => 5, + 'legacy' => 0, + 'current' => 0, + 'unknown' => 5, + ], $page['counts']); + $this->assertSame( + ['history_unavailable'], + array_values(array_unique(array_column($page['runs'], 'unknown_reason'))) + ); + } + + public function testOversizedMarkerHistoryIsUnknownWithoutReadingAPartialDecision(): void + { + $run = $this->createRun(1, null); + $events = []; + for ($sequence = 1; $sequence <= 1001; ++$sequence) { + $events[] = [ + 'id' => (string) \Illuminate\Support\Str::ulid(), + 'workflow_run_id' => $run->id, + 'sequence' => $sequence + 1, + 'event_type' => 'VersionMarkerRecorded', + 'payload' => json_encode($this->marker(2, $sequence), JSON_THROW_ON_ERROR), + 'recorded_at' => now(), + ]; + } + DB::table((new WorkflowHistoryEvent())->getTable())->insert($events); + $run->update([ + 'last_history_sequence' => 1002, + ]); + $page = (new VersionUsageQuery())->page('orders', 'shipment', 'shipping', 2); + $this->assertSame([ + 'total' => 1, + 'legacy' => 0, + 'current' => 0, + 'unknown' => 1, + ], $page['counts']); + $this->assertSame('history_unavailable', $page['runs'][0]['unknown_reason']); + } + + public function testConflictingAndMalformedMarkersNeverBecomeCurrent(): void + { + $conflict = $this->createRun(1, 2); + $this->event($conflict, HistoryEventType::VersionMarkerRecorded->value, $this->marker(1, 2)); + $malformed = $this->createRun(2, 2); + $this->event($malformed, HistoryEventType::VersionMarkerRecorded->value, array_replace($this->marker(2, 2), [ + 'version' => '2', + ])); + $page = (new VersionUsageQuery())->page('orders', 'shipment', 'shipping', 2); + $this->assertSame(2, $page['counts']['unknown']); + $this->assertSame(['history_invalid', 'history_invalid'], array_column($page['runs'], 'unknown_reason')); + } + + public function testPagingHasAStableUpperBoundAndCannotCrossTenantOrFilterScope(): void + { + $one = $this->createRun(1, 1); + $two = $this->createRun(2, 2); + $three = $this->createRun(3, null); + $query = new VersionUsageQuery(); + $first = $query->page('orders', 'shipment', 'shipping', 2, 1); + $this->createRun(4, 1); + $two->update([ + 'status' => 'completed', + ]); + $next = $query->page('orders', 'shipment', 'shipping', 2, 1, $first['next_cursor']); + $this->assertSame([$three->id], array_column($next['runs'], 'run_id')); + $this->assertSame($first['through_run_id'], $next['through_run_id']); + $this->assertSame(2, $next['counts']['total']); + $this->assertFalse($next['has_more']); + $this->assertNotContains($one->id, array_column($next['runs'], 'run_id')); + foreach ([['ORDERS', 'shipment', 'shipping', 2, null], + ['orders', 'shipment', 'shipping', 3, null], + ['orders', 'shipment', 'shipping', 2, 'legacy'], + ] as [$namespace, $type, $change, $minimum, $state]) { + try { + $query->page($namespace, $type, $change, $minimum, 1, $first['next_cursor'], $state); + $this->fail('A cursor crossed its query identity.'); + } catch (InvalidArgumentException) { + $this->assertTrue(true); + } + } + } + + public function testRunIdKeysetsUseByteOrderOnEveryStorageCollation(): void + { + foreach ([[1, 'b', 1], [2, 'C', 2]] as [$number, $id, $version]) { + $run = $this->createRun($number, $version); + DB::table((new WorkflowHistoryEvent())->getTable())->where('workflow_run_id', $run->id) + ->update([ + 'workflow_run_id' => $id, + ]); + $run->update([ + 'id' => $id, + ]); + } + $query = new VersionUsageQuery(); + $first = $query->page('orders', 'shipment', 'shipping', 2, 1); + $this->assertSame(['C'], array_column($first['runs'], 'run_id')); + $this->assertSame('b', $first['through_run_id']); + $second = $query->page('orders', 'shipment', 'shipping', 2, 1, $first['next_cursor']); + $this->assertSame(['b'], array_column($second['runs'], 'run_id')); + $this->assertFalse($second['has_more']); + $this->assertSame(2, $second['counts']['total']); + } + + public function testCurrentDefinitionFingerprintDoesNotReplaceAnUnreachedMarker(): void + { + $run = $this->createRun(1, null, 'orders', 'test-version-workflow'); + $run->update([ + 'workflow_class' => TestVersionWorkflow::class, + ]); + DB::table((new WorkflowHistoryEvent())->getTable())->where('workflow_run_id', $run->id) + ->where('sequence', 1) + ->update([ + 'payload' => json_encode([ + 'workflow_definition_fingerprint' => WorkflowDefinition::fingerprint(TestVersionWorkflow::class), + ], JSON_THROW_ON_ERROR), + ]); + $this->assertTrue(WorkflowDefinitionFingerprint::matchesCurrent($run)); + $page = (new VersionUsageQuery())->page('orders', $run->workflow_type, 'shipping', 2); + $this->assertSame([ + 'total' => 1, + 'legacy' => 0, + 'current' => 0, + 'unknown' => 1, + ], $page['counts']); + $this->assertSame('marker_absent', $page['runs'][0]['unknown_reason']); + } + + public function testPruningDuringAnObservationCannotLeaveAnOptimisticCurrentResult(): void + { + $run = $this->createRun(1, 2); + $pruned = false; + DB::listen(static function (QueryExecuted $event) use ($run, &$pruned): void { + if (! $pruned && str_contains($event->sql, 'COUNT(*) AS event_count')) { + $pruned = true; + WorkflowRun::query()->whereKey($run->id)->update([ + 'details_pruned_at' => now(), + ]); + } + }); + $page = (new VersionUsageQuery())->page('orders', 'shipment', 'shipping', 2); + $this->assertTrue($pruned); + $this->assertSame([ + 'total' => 1, + 'legacy' => 0, + 'current' => 0, + 'unknown' => 1, + ], $page['counts']); + $this->assertSame('history_unavailable', $page['runs'][0]['unknown_reason']); + } + + public function testReadDoesNotReplayDecodeApplicationPayloadsOrWriteProjections(): void + { + $this->createRun(1, 2); + $before = WorkflowHistoryEvent::query()->count(); + $statements = []; + DB::listen(static function (QueryExecuted $event) use (&$statements): void { + $statements[] = $event->sql; + }); + Json::decodeUsing(static fn () => throw new LogicException('Application decoder must not run.')); + try { + $page = (new VersionUsageQuery())->page('orders', 'shipment', 'shipping', 2); + $this->assertSame(1, $page['counts']['current']); + } finally { + Json::decodeUsing(null); + } + $this->assertSame($before, WorkflowHistoryEvent::query()->count()); + $this->assertNotEmpty($statements); + foreach ($statements as $statement) { + $this->assertStringStartsWith('select', strtolower(ltrim($statement))); + } + } + + public function testEmptyCohortIsExplicitAndInvalidRequestsAreRefused(): void + { + $query = new VersionUsageQuery(); + $page = $query->page(null, 'shipment', 'shipping', 2); + $this->assertSame([ + 'total' => 0, + 'legacy' => 0, + 'current' => 0, + 'unknown' => 0, + ], $page['counts']); + $this->assertSame([], $page['runs']); + $this->assertNull($page['through_run_id']); + $this->assertNull($page['next_cursor']); + $this->assertArrayNotHasKey('safe_to_remove', $page); + $this->expectException(InvalidArgumentException::class); + $query->page('orders', 'shipment', 'shipping', 2147483648); + } + + public function testCountsAndPagesTraverseMoreThanOneStorageBatch(): void + { + for ($number = 1; $number <= 205; ++$number) { + $this->createRun($number, $number % 3 === 0 ? null : ($number % 2 === 0 ? 2 : 1)); + } + $query = new VersionUsageQuery(); + $ids = []; + $cursor = null; + do { + $page = $query->page('orders', 'shipment', 'shipping', 2, 70, $cursor); + $this->assertSame(205, $page['counts']['total']); + $ids = [...$ids, ...array_column($page['runs'], 'run_id')]; + $cursor = $page['next_cursor']; + } while ($cursor !== null); + $this->assertCount(205, $ids); + $this->assertCount(205, array_unique($ids)); + $sorted = $ids; + sort($sorted, SORT_STRING); + $this->assertSame($sorted, $ids); + } + + public function testConfiguredModelsConnectionAndPrefixAreUsedWithoutReadingDefaultTables(): void + { + $connection = 'version_usage_fixture'; + config() + ->set('database.connections.' . $connection, DB::connection()->getConfig() + [ + 'prefix' => '', + ]); + config() + ->set('database.connections.' . $connection . '.prefix', 'usage_fixture_'); + $database = DB::connection($connection); + $schema = $database->getSchemaBuilder(); + $schema->create('runs', static function (Blueprint $table): void { + $table->string('id') + ->primary(); + $table->string('workflow_instance_id'); + $table->string('namespace') + ->nullable(); + $table->string('workflow_type'); + $table->string('status'); + $table->timestamp('details_pruned_at') + ->nullable(); + $table->integer('last_history_sequence'); + }); + $schema->create('history', static function (Blueprint $table): void { + $table->string('id') + ->primary(); + $table->string('workflow_run_id'); + $table->integer('sequence'); + $table->string('event_type'); + $table->text('payload'); + }); + try { + $this->createRun(1, 1); + config() + ->set('workflows.v2.run_model', VersionUsageCustomRun::class); + config() + ->set('workflows.v2.history_event_model', VersionUsageCustomHistory::class); + $database->table('runs') + ->insert([ + 'id' => 'custom-1', + 'workflow_instance_id' => 'custom-instance', + 'namespace' => 'orders', + 'workflow_type' => 'shipment', + 'status' => 'waiting', + 'last_history_sequence' => 2, + ]); + $database->table('history') + ->insert([ + [ + 'id' => 'event-1', + 'workflow_run_id' => 'custom-1', + 'sequence' => 1, + 'event_type' => 'WorkflowStarted', + 'payload' => '{}', + ], + [ + 'id' => 'event-2', + 'workflow_run_id' => 'custom-1', + 'sequence' => 2, + 'event_type' => 'VersionMarkerRecorded', + 'payload' => json_encode($this->marker(2), JSON_THROW_ON_ERROR), + ], + ]); + $page = (new VersionUsageQuery())->page('orders', 'shipment', 'shipping', 2); + $this->assertSame([ + 'total' => 1, + 'legacy' => 0, + 'current' => 1, + 'unknown' => 0, + ], $page['counts']); + $this->assertSame('custom-1', $page['runs'][0]['run_id']); + } finally { + config()->set('workflows.v2.run_model', WorkflowRun::class); + config() + ->set('workflows.v2.history_event_model', WorkflowHistoryEvent::class); + $schema->dropIfExists('history'); + $schema->dropIfExists('runs'); + DB::purge($connection); + } + } + + private function createRun( + int $number, + ?int $version, + ?string $namespace = 'orders', + string $type = 'shipment' + ): WorkflowRun { + $id = str_pad((string) $number, 26, '0', STR_PAD_LEFT); + $instance = WorkflowInstance::query()->create([ + 'id' => 'usage-instance-' . $number, + 'namespace' => $namespace, + 'workflow_type' => $type, + 'workflow_class' => 'NotLoadedByVersionUsage', + ]); + $run = WorkflowRun::query()->create([ + 'id' => $id, + 'workflow_instance_id' => $instance->id, + 'run_number' => 1, + 'namespace' => $namespace, + 'workflow_type' => $type, + 'workflow_class' => 'NotLoadedByVersionUsage', + 'status' => 'waiting', + ]); + $this->event($run, HistoryEventType::WorkflowStarted->value, []); + if ($version !== null) { + $this->event($run, HistoryEventType::VersionMarkerRecorded->value, $this->marker($version)); + } + return $run; + } + + private function event(WorkflowRun $run, string $kind, array $payload): void + { + $sequence = (int) $run->last_history_sequence + 1; + DB::table((new WorkflowHistoryEvent())->getTable())->insert([ + 'id' => (string) \Illuminate\Support\Str::ulid(), + 'workflow_run_id' => $run->id, + 'sequence' => $sequence, + 'event_type' => $kind, + 'payload' => json_encode($payload, JSON_THROW_ON_ERROR), + 'recorded_at' => now(), + ]); + $run->update([ + 'last_history_sequence' => $sequence, + ]); + } + + private function marker(int $version, int $sequence = 1): array + { + return [ + 'sequence' => $sequence, + 'change_id' => 'shipping', + 'version' => $version, + 'min_supported' => -1, + 'max_supported' => 3, + ]; + } +} + +final class VersionUsageCustomRun extends WorkflowRun +{ + protected $table = 'runs'; + + protected $connection = 'version_usage_fixture'; +} + +final class VersionUsageCustomHistory extends WorkflowHistoryEvent +{ + protected $table = 'history'; +} diff --git a/tests/Unit/V2/VersionUsageCursorTest.php b/tests/Unit/V2/VersionUsageCursorTest.php new file mode 100644 index 000000000..00c357c99 --- /dev/null +++ b/tests/Unit/V2/VersionUsageCursorTest.php @@ -0,0 +1,59 @@ +assertSame('run-9', $value['through']); + $this->assertSame('run-3', $value['after']); + $this->expectException(InvalidArgumentException::class); + VersionUsageCursor::decode($encoded, hash('sha256', 'another-namespace')); + } + + public static function malformed(): iterable + { + yield 'empty' => ['']; + yield 'base64 punctuation' => ['%%']; + yield 'oversized' => [str_repeat('A', 2049)]; + yield 'not json' => ['YWJj']; + yield 'scalar' => ['MQ']; + $query = hash('sha256', 'query'); + foreach ([[ + 'version' => 2, + ], [ + 'after' => [], + ], [ + 'through' => '', + ], [ + 'after' => 'run-z', + ], [ + 'extra' => true, + ]] as $mutation) { + yield json_encode($mutation) => [rtrim(strtr(base64_encode(json_encode(array_replace([ + 'version' => 1, + 'query' => $query, + 'through' => 'run-9', + 'after' => 'run-1', + ], $mutation), JSON_THROW_ON_ERROR)), '+/', '-_'), '=')]; + } + } + + #[DataProvider('malformed')] + public function testMalformedCursorsAreErrorsRatherThanFirstPages(string $cursor): void + { + $this->expectException(InvalidArgumentException::class); + VersionUsageCursor::decode($cursor, hash('sha256', 'query')); + } +} diff --git a/tests/Unit/V2/VersionUsageDecisionTest.php b/tests/Unit/V2/VersionUsageDecisionTest.php new file mode 100644 index 000000000..9e8700939 --- /dev/null +++ b/tests/Unit/V2/VersionUsageDecisionTest.php @@ -0,0 +1,175 @@ + [-1, 1, 'legacy']; + yield 'older' => [1, 2, 'legacy']; + yield 'minimum' => [2, 2, 'current']; + yield 'newer' => [3, 2, 'current']; + yield 'signed minimum' => [-2147483648, -2147483648, 'current']; + yield 'signed maximum' => [2147483647, 2147483647, 'current']; + } + + #[DataProvider('versions')] + public function testRecordedDecisionIsComparedWithTheProposedMinimum( + int $version, + int $minimum, + string $state + ): void { + $decision = new VersionUsageDecision('shipping', $minimum, true); + $decision->observe($this->marker($version)); + $this->assertSame([ + 'state' => $state, + 'recorded_version' => $version, + 'unknown_reason' => null, + ], $decision->result()); + } + + public function testAbsentAndUnavailableHistoryCannotBeClassifiedCurrent(): void + { + $missing = new VersionUsageDecision('shipping', 1, true); + $missing->observe($this->marker(2) + [ + 'ignored' => true, + ]); + $this->assertSame('history_unavailable', $missing->result()['unknown_reason']); + $empty = new VersionUsageDecision('shipping', 1, true); + $this->assertSame('marker_absent', $empty->result()['unknown_reason']); + $unavailable = new VersionUsageDecision('shipping', 1, false); + $unavailable->observe($this->marker(2)); + $this->assertSame([ + 'state' => 'unknown', + 'recorded_version' => null, + 'unknown_reason' => 'history_unavailable', + ], $unavailable->result()); + } + + public function testIdenticalHistoricalDecisionsRemainValidButConflictsCannotBeOverwritten(): void + { + $decision = new VersionUsageDecision('shipping', 2, true); + $decision->observe($this->marker(1)); + $decision->observe(array_replace($this->marker(1), [ + 'sequence' => 2, + ])); + $this->assertSame('legacy', $decision->result()['state']); + $decision->observe(array_replace($this->marker(2), [ + 'sequence' => 3, + ])); + $decision->observe(array_replace($this->marker(1), [ + 'sequence' => 4, + ])); + $this->assertSame([ + 'state' => 'unknown', + 'recorded_version' => null, + 'unknown_reason' => 'history_invalid', + ], $decision->result()); + } + + public function testRepeatedVersionMayRetainDifferentIndividuallyValidRanges(): void + { + $decision = new VersionUsageDecision('shipping', 2, true); + $decision->observe(array_replace($this->marker(1), [ + 'min_supported' => -1, + 'max_supported' => 1, + ])); + $decision->observe(array_replace($this->marker(1), [ + 'sequence' => 2, + 'min_supported' => 0, + 'max_supported' => 4, + ])); + $this->assertSame([ + 'state' => 'legacy', + 'recorded_version' => 1, + 'unknown_reason' => null, + ], $decision->result()); + } + + public static function invalidMarkers(): iterable + { + yield 'not data' => [null]; + yield 'missing bounds' => [[ + 'change_id' => 'shipping', + 'version' => 2, + 'sequence' => 1, + ]]; + foreach ([ + 'sequence' => 0, + 'change_id' => '', + 'version' => '2', + 'min_supported' => false, + 'max_supported' => 2147483648, + ] as $key => $value) { + yield $key => [ + array_replace(self::validMarker(), [ + $key => $value, + ])]; + } + yield 'outside recorded range' => [ + array_replace(self::validMarker(), [ + 'version' => 5, + 'max_supported' => 4, + ])]; + } + + #[DataProvider('invalidMarkers')] + public function testInvalidHistoryRemainsUnknown(mixed $marker): void + { + $decision = new VersionUsageDecision('shipping', 1, true); + $decision->observe($marker); + $this->assertSame('history_invalid', $decision->result()['unknown_reason']); + $this->assertNull($decision->result()['recorded_version']); + } + + public function testChangeIdsAreExactAndDuplicateLogicalPositionsAreInvalid(): void + { + $decision = new VersionUsageDecision('shipping', 1, true); + $decision->observe(array_replace($this->marker(2), [ + 'change_id' => 'Shipping', + ])); + $this->assertSame('marker_absent', $decision->result()['unknown_reason']); + $decision->observe($this->marker(2)); + $this->assertSame('history_invalid', $decision->result()['unknown_reason']); + } + + public function testInspectionBudgetCannotReturnAnOptimisticPartialDecision(): void + { + $decision = new VersionUsageDecision('shipping', 1, true); + for ($sequence = 1; $sequence <= 1001; ++$sequence) { + $decision->observe(array_replace($this->marker(2), [ + 'sequence' => $sequence, + ])); + } + $this->assertSame([ + 'state' => 'unknown', + 'recorded_version' => null, + 'unknown_reason' => 'history_unavailable', + ], $decision->result()); + } + + private function marker(int $version): array + { + return array_replace(self::validMarker(), [ + 'version' => $version, + ]); + } + + private static function validMarker(): array + { + return [ + 'sequence' => 1, + 'change_id' => 'shipping', + 'version' => 2, + 'min_supported' => -2147483648, + 'max_supported' => 2147483647, + ]; + } +} From 0f8f30f2094c175f8432e0d0150e189f13738870 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 14 Nov 2026 11:25:11 +0000 Subject: [PATCH 03/11] fix(versioning): retain unknown ownership and provenance in usage counts --- docs/version-usage.md | 7 +-- src/V2/Support/VersionUsageQuery.php | 28 ++++++++-- tests/Feature/V2/V2VersionUsageQueryTest.php | 55 ++++++++++++++++++++ 3 files changed, 82 insertions(+), 8 deletions(-) diff --git a/docs/version-usage.md b/docs/version-usage.md index 9e4ad91ea..d8489ab08 100644 --- a/docs/version-usage.md +++ b/docs/version-usage.md @@ -14,8 +14,8 @@ Namespace, workflow type and change ID comparisons are exact and case-sensitive. ## Request -- `workflow_type`: required nonempty UTF-8 string, at most 255 bytes. -- `change_id`: required nonempty UTF-8 string, at most 1024 bytes. +- `workflow_type`: required nonblank UTF-8 string, at most 255 bytes. +- `change_id`: required nonblank UTF-8 string, at most 1024 bytes. - `min_version`: required signed 32-bit integer, from -2147483648 to 2147483647. This is the proposed new `minSupported`, not the largest version deployed. For removing a `patched()` legacy branch, use 1. @@ -47,7 +47,8 @@ Each run has `recorded_version` only for legacy/current. Unknown reasons are: an older definition that has not reached the change point. A matching current code fingerprint, build label, admission time or deployment time never upgrades this to current. -- `history_unavailable`: details are pruned, the initial history/prefix is absent +- `history_unavailable`: details are pruned or owning-instance/source provenance is unavailable, + the initial history/prefix is absent or incomplete, the stored event format is not supported, or the bounded inspection budget (1,000 version markers per run) is exceeded. - `history_invalid`: malformed decision data, invalid recorded range, or diff --git a/src/V2/Support/VersionUsageQuery.php b/src/V2/Support/VersionUsageQuery.php index 7b48b8546..617c39936 100644 --- a/src/V2/Support/VersionUsageQuery.php +++ b/src/V2/Support/VersionUsageQuery.php @@ -12,6 +12,7 @@ use Workflow\V2\Enums\HistoryEventType; use Workflow\V2\Enums\RunStatus; use Workflow\V2\Models\WorkflowHistoryEvent; +use Workflow\V2\Models\WorkflowInstance; use Workflow\V2\Models\WorkflowRun; /** Read-only, namespace-bound observations of retained open-run version decisions. */ @@ -54,7 +55,7 @@ public function page( if ($through !== null) { $model = $base->getModel(); foreach ($this->batches($base, $through) as $runs) { - $decisions = $this->decisions($runs, $model, $changeId, $minVersion); + $decisions = $this->decisions($runs, $model, $namespace, $changeId, $minVersion); foreach ($runs as $run) { $decision = $decisions[$run->id]->result(); ++$counts['total']; @@ -105,7 +106,7 @@ private function batches(Builder $base, string $through): Generator $after = null; do { $query = (clone $base)->select([ - 'id', 'workflow_instance_id', 'status', 'details_pruned_at', 'last_history_sequence', + 'id', 'workflow_instance_id', 'status', 'details_pruned_at', 'last_history_sequence', 'import_source', ])->whereRaw($id . ' <= ' . $parameter, [$through])->orderByRaw($id . ' ASC'); if ($after !== null) { $query->whereRaw($id . ' > ' . $parameter, [$after]); @@ -156,8 +157,23 @@ private function runs(?string $namespace, string $workflowType): Builder /** * @return array */ - private function decisions(Collection $runs, WorkflowRun $runModel, string $changeId, int $minVersion): array - { + private function decisions( + Collection $runs, + WorkflowRun $runModel, + ?string $namespace, + string $changeId, + int $minVersion + ): array { + $instanceModel = ConfiguredV2Models::resolve('instance_model', WorkflowInstance::class); + $owners = NamespaceIdentity::constrain( + (new $instanceModel())->setConnection($runModel->getConnectionName()) + ->newQuery() + ->setEagerLoads([]), + $namespace, + )->whereIn('id', $runs->pluck('workflow_instance_id')->all()) + ->toBase() + ->pluck('id') + ->all(); $eventModel = ConfiguredV2Models::resolve('history_event_model', WorkflowHistoryEvent::class); $history = (new $eventModel())->setConnection($runModel->getConnectionName()) ->newQuery() @@ -194,6 +210,8 @@ private function decisions(Collection $runs, WorkflowRun $runModel, string $chan foreach ($runs as $run) { $stats = $statistics->get($run->id); $available = $run->details_pruned_at === null && $stats !== null + && in_array($run->workflow_instance_id, $owners, true) + && in_array($run->import_source, [null, EmbeddedV2ImportContract::IMPORT_SOURCE], true) && (int) $stats->first_sequence === 1 && (int) $stats->last_sequence === (int) $run->last_history_sequence && (int) $stats->event_count === (int) $run->last_history_sequence @@ -250,7 +268,7 @@ private static function binary(string $expression, string $driver): string private static function text(string $value, int $limit): void { - if ($value === '' || strlen($value) > $limit || preg_match('//u', $value) !== 1) { + if ($value === '' || strlen($value) > $limit || preg_match('/(*UCP)\S/u', $value) !== 1) { throw new InvalidArgumentException('Version-usage identities must be nonempty bounded UTF-8 strings.'); } } diff --git a/tests/Feature/V2/V2VersionUsageQueryTest.php b/tests/Feature/V2/V2VersionUsageQueryTest.php index 0538534c7..1ac3ef92d 100644 --- a/tests/Feature/V2/V2VersionUsageQueryTest.php +++ b/tests/Feature/V2/V2VersionUsageQueryTest.php @@ -127,6 +127,31 @@ public function testPrunedMissingGappedAndUnsupportedHistoriesStayUnknownEvenWit ); } + public function testUnavailableOwnershipAndUnrecognizedImportProvenanceStayInUnknownCounts(): void + { + $missing = $this->createRun(1, 2); + WorkflowInstance::query()->whereKey($missing->workflow_instance_id)->delete(); + $foreign = $this->createRun(2, 2); + WorkflowInstance::query()->whereKey($foreign->workflow_instance_id)->update([ + 'namespace' => 'private', + ]); + $imported = $this->createRun(3, 2); + $imported->update([ + 'import_source' => 'unsupported-source', + ]); + $page = (new VersionUsageQuery())->page('orders', 'shipment', 'shipping', 2); + $this->assertSame([ + 'total' => 3, + 'legacy' => 0, + 'current' => 0, + 'unknown' => 3, + ], $page['counts']); + $this->assertSame( + ['history_unavailable'], + array_values(array_unique(array_column($page['runs'], 'unknown_reason'))) + ); + } + public function testOversizedMarkerHistoryIsUnknownWithoutReadingAPartialDecision(): void { $run = $this->createRun(1, null); @@ -350,6 +375,14 @@ public function testConfiguredModelsConnectionAndPrefixAreUsedWithoutReadingDefa $table->timestamp('details_pruned_at') ->nullable(); $table->integer('last_history_sequence'); + $table->string('import_source') + ->nullable(); + }); + $schema->create('instances', static function (Blueprint $table): void { + $table->string('id') + ->primary(); + $table->string('namespace') + ->nullable(); }); $schema->create('history', static function (Blueprint $table): void { $table->string('id') @@ -365,6 +398,13 @@ public function testConfiguredModelsConnectionAndPrefixAreUsedWithoutReadingDefa ->set('workflows.v2.run_model', VersionUsageCustomRun::class); config() ->set('workflows.v2.history_event_model', VersionUsageCustomHistory::class); + config() + ->set('workflows.v2.instance_model', VersionUsageCustomInstance::class); + $database->table('instances') + ->insert([ + 'id' => 'custom-instance', + 'namespace' => 'orders', + ]); $database->table('runs') ->insert([ 'id' => 'custom-1', @@ -403,7 +443,10 @@ public function testConfiguredModelsConnectionAndPrefixAreUsedWithoutReadingDefa config()->set('workflows.v2.run_model', WorkflowRun::class); config() ->set('workflows.v2.history_event_model', WorkflowHistoryEvent::class); + config() + ->set('workflows.v2.instance_model', WorkflowInstance::class); $schema->dropIfExists('history'); + $schema->dropIfExists('instances'); $schema->dropIfExists('runs'); DB::purge($connection); } @@ -477,3 +520,15 @@ final class VersionUsageCustomHistory extends WorkflowHistoryEvent { protected $table = 'history'; } + +final class VersionUsageCustomInstance extends WorkflowInstance +{ + protected $table = 'instances'; + + protected $connection = 'version_usage_fixture'; + + public function getForeignKey(): string + { + return 'workflow_instance_id'; + } +} From a6b8fd4102b0ad1a53befd93b1c26234e6610002 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 14 Nov 2026 11:39:21 +0000 Subject: [PATCH 04/11] docs(versioning): keep usage contract product-facing --- CHANGELOG.md | 2 +- docs/version-usage.md | 20 +++++++++++--------- 2 files changed, 12 insertions(+), 10 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 451505245..ea1a8f346 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,7 +4,7 @@ - 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 (#220). + remain unknown when assessing legacy branch retirement. - Experimental embedded compensation scopes (#1063): retained versioned obligations, native Saga recovery, sequential child participation and explicit diff --git a/docs/version-usage.md b/docs/version-usage.md index d8489ab08..fe49610e6 100644 --- a/docs/version-usage.md +++ b/docs/version-usage.md @@ -1,7 +1,7 @@ # Open-run version usage (draft v1) -This read-only contract implements durable-workflow/.github#220. It answers which -open runs still have a recorded decision below a proposed minimum supported +This read-only contract answers which open runs still have a recorded decision +below a proposed minimum supported version. It does not replay workflows, run application code, write a projection, change markers, or authorize deleting a branch. @@ -47,13 +47,15 @@ Each run has `recorded_version` only for legacy/current. Unknown reasons are: an older definition that has not reached the change point. A matching current code fingerprint, build label, admission time or deployment time never upgrades this to current. -- `history_unavailable`: details are pruned or owning-instance/source provenance is unavailable, - the initial history/prefix is absent - or incomplete, the stored event format is not supported, or the bounded +- `history_unavailable`: details are pruned, owning-instance/source provenance is + unavailable, the initial history/prefix is absent or incomplete, the stored + event format is not supported, or the bounded inspection budget (1,000 version markers per run) is exceeded. - `history_invalid`: malformed decision data, invalid recorded range, or - conflicting decisions for the same change ID. Repeated decisions with the same version remain valid when each recorded - range is valid; range metadata may differ across repeated historical calls. Marker integer fields use the signed 32-bit range. + conflicting decisions for the same change ID. Repeated decisions with the same + version remain valid when each recorded range is valid; range metadata may + differ across repeated historical calls. Marker integer fields use the signed + 32-bit range. Marker readers must understand their runtime's retained format. Embedded marker fields are `change_id`, `version`, `min_supported`, `max_supported` and positive @@ -68,8 +70,8 @@ The response shape is in cohort, before the optional state filter and independently of page size/cursor. `runs` contains only the selected state when a state filter is present. -Runs are ordered by byte-exact run ID ascending, with a strict keyset continuation. The -first request captures the greatest matching open run ID as `through_run_id`; +Runs are ordered by byte-exact run ID ascending, with a strict keyset continuation. +The first request captures the greatest matching open run ID as `through_run_id`; all continuations retain that upper bound and advance after the last returned ID. New admissions beyond that bound cannot appear on later pages. Cursors bind all identity/classification filters, and every page re-applies authorization and From b540761484cbe96ce46746b77e2741b725b427d1 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 14 Nov 2026 11:43:21 +0000 Subject: [PATCH 05/11] fix(versioning): use exact canonical open status predicates --- src/V2/Support/VersionUsageQuery.php | 7 +++++++ tests/Feature/V2/V2VersionUsageQueryTest.php | 22 ++++++++++++++++++++ 2 files changed, 29 insertions(+) diff --git a/src/V2/Support/VersionUsageQuery.php b/src/V2/Support/VersionUsageQuery.php index 617c39936..ebbd0094d 100644 --- a/src/V2/Support/VersionUsageQuery.php +++ b/src/V2/Support/VersionUsageQuery.php @@ -151,6 +151,13 @@ private function runs(?string $namespace, string $workflowType): Builder ->getConnection() ->getDriverName(); $query->whereRaw(self::binary($column, $driver) . ' = ' . self::binary('?', $driver), [$workflowType]); + $status = $query->getQuery() + ->getGrammar() + ->wrap($query->getModel()->qualifyColumn('status')); + $parameters = implode(', ', array_fill(0, 3, self::binary('?', $driver))); + $query->whereRaw(self::binary($status, $driver) . ' IN (' . $parameters . ')', [ + RunStatus::Pending->value, RunStatus::Running->value, RunStatus::Waiting->value, + ]); return $query; } diff --git a/tests/Feature/V2/V2VersionUsageQueryTest.php b/tests/Feature/V2/V2VersionUsageQueryTest.php index 1ac3ef92d..0887509bc 100644 --- a/tests/Feature/V2/V2VersionUsageQueryTest.php +++ b/tests/Feature/V2/V2VersionUsageQueryTest.php @@ -127,6 +127,28 @@ public function testPrunedMissingGappedAndUnsupportedHistoriesStayUnknownEvenWit ); } + public function testNoncanonicalStatusSpellingsCannotEnterTheOpenRunCohort(): void + { + $this->createRun(1, 1); + foreach ([ + 2 => 'WAITING', + 3 => 'waiting ', + ] as $number => $status) { + $run = $this->createRun($number, 2); + DB::table((new WorkflowRun())->getTable())->where('id', $run->id)->update([ + 'status' => $status, + ]); + } + $page = (new VersionUsageQuery())->page('orders', 'shipment', 'shipping', 2); + $this->assertSame([ + 'total' => 1, + 'legacy' => 1, + 'current' => 0, + 'unknown' => 0, + ], $page['counts']); + $this->assertSame(['waiting'], array_column($page['runs'], 'status')); + } + public function testUnavailableOwnershipAndUnrecognizedImportProvenanceStayInUnknownCounts(): void { $missing = $this->createRun(1, 2); From 9af1eb53f698f8dd764dfd7841c1851d79555b9d Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 14 Nov 2026 11:51:14 +0000 Subject: [PATCH 06/11] fix(versioning): authenticate usage cursors and read producer history --- docs/version-usage.md | 12 +- src/V2/Support/VersionUsageCursor.php | 28 ++++- src/V2/Support/VersionUsageDecision.php | 5 +- src/V2/Support/VersionUsageQuery.php | 112 ++++++++++++++++++- tests/Feature/V2/V2VersionUsageQueryTest.php | 61 ++++++++++ tests/Unit/V2/VersionUsageCursorTest.php | 70 ++++++++++-- tests/Unit/V2/VersionUsageDecisionTest.php | 23 ++++ 7 files changed, 287 insertions(+), 24 deletions(-) diff --git a/docs/version-usage.md b/docs/version-usage.md index fe49610e6..490ad8df9 100644 --- a/docs/version-usage.md +++ b/docs/version-usage.md @@ -23,7 +23,10 @@ Namespace, workflow type and change ID comparisons are exact and case-sensitive. - `limit`: integer from 1 to 200, default 50. - `cursor`: optional opaque continuation from this exact namespace/type/change/ minimum/state query. It is not an authorization token. Invalid or mismatched - cursors are errors; they never fall back to a first page. + cursors are errors; they never fall back to a first page. Cursor bodies are + authenticated before parsing. The embedded reader uses a domain-separated HMAC + with the configured application key; missing keys are errors and key rotation + invalidates prior cursors. Invalid arguments return the existing invalid-request error envelope (HTTP 422 for the native endpoint). Storage or connection failures are errors, not empty @@ -57,9 +60,14 @@ Each run has `recorded_version` only for legacy/current. Unknown reasons are: differ across repeated historical calls. Marker integer fields use the signed 32-bit range. +The v1 reader freezes its supported retained event-kind set; adding a runtime +event enum case does not silently expand reader support. + Marker readers must understand their runtime's retained format. Embedded marker fields are `change_id`, `version`, `min_supported`, `max_supported` and positive -logical `sequence`; the range must contain the recorded version. No application +logical `sequence`; the range must contain the recorded version. The known +producer-injected `task` snapshot is supported; unknown marker extensions require +a reader upgrade and remain unavailable. No application argument, output, memo or payload-store fetch is necessary for this query. ## Response and paging diff --git a/src/V2/Support/VersionUsageCursor.php b/src/V2/Support/VersionUsageCursor.php index 8f4269611..51dda8836 100644 --- a/src/V2/Support/VersionUsageCursor.php +++ b/src/V2/Support/VersionUsageCursor.php @@ -6,26 +6,34 @@ use InvalidArgumentException; use JsonException; +use LogicException; /** @internal A continuation binds query identity; authorization is reapplied by every query. */ final class VersionUsageCursor { - public static function encode(string $query, string $through, string $after): string + private const DOMAIN = "durable-workflow.version-usage-cursor.v1\0"; + + public static function encode(string $query, string $through, string $after, string $key): string { - return rtrim(strtr(base64_encode(json_encode([ + $body = rtrim(strtr(base64_encode(json_encode([ 'version' => 1, 'query' => $query, 'through' => $through, 'after' => $after, ], JSON_THROW_ON_ERROR)), '+/', '-_'), '='); + return $body . '.' . self::signature($body, $key); } - public static function decode(string $cursor, string $query): array + public static function decode(string $cursor, string $query, string $key): array { - if (strlen($cursor) > 2048 || preg_match('/\A[A-Za-z0-9_-]+\z/D', $cursor) !== 1) { + if (strlen($cursor) > 2048 || preg_match('/\A([A-Za-z0-9_-]+)\.([0-9a-f]{64})\z/D', $cursor, $parts) !== 1) { throw new InvalidArgumentException('Invalid version-usage cursor.'); } - $decoded = base64_decode(strtr($cursor, '-_', '+/'), true); + // Authenticate the exact transport bytes before decoding or parsing them. + if (! hash_equals(self::signature($parts[1], $key), $parts[2])) { + throw new InvalidArgumentException('Invalid version-usage cursor signature.'); + } + $decoded = base64_decode(strtr($parts[1], '-_', '+/'), true); try { $value = $decoded === false ? null : json_decode($decoded, true, 8, JSON_THROW_ON_ERROR); } catch (JsonException) { @@ -40,12 +48,20 @@ public static function decode(string $cursor, string $query): array || ! is_string($value['query']) || ! hash_equals($query, $value['query']) || ! self::runId($value['after']) || ! self::runId($value['through']) || strcmp($value['after'], $value['through']) > 0 - || self::encode($query, $value['through'], $value['after']) !== $cursor) { + || self::encode($query, $value['through'], $value['after'], $key) !== $cursor) { throw new InvalidArgumentException('Version-usage cursor does not match this query.'); } return $value; } + private static function signature(string $body, string $key): string + { + if ($key === '') { + throw new LogicException('A non-empty application key is required for version-usage cursors.'); + } + return hash_hmac('sha256', self::DOMAIN . $body, $key); + } + private static function runId(mixed $value): bool { return is_string($value) && preg_match('/\A[A-Za-z0-9][A-Za-z0-9_.:-]{0,190}\z/D', $value) === 1; diff --git a/src/V2/Support/VersionUsageDecision.php b/src/V2/Support/VersionUsageDecision.php index cc81f3d78..bf362a6e9 100644 --- a/src/V2/Support/VersionUsageDecision.php +++ b/src/V2/Support/VersionUsageDecision.php @@ -45,8 +45,9 @@ public function observe(mixed $payload): void } if (array_diff(array_keys($payload), [ 'sequence', 'change_id', 'version', 'min_supported', 'max_supported', - 'reused_from_run_id', 'reused_recorded_at', - ]) !== []) { + 'reused_from_run_id', 'reused_recorded_at', 'task', + ]) !== [] || (array_key_exists('task', $payload) && (! is_array($payload['task']) + || array_is_list($payload['task'])))) { $this->reason = 'history_unavailable'; return; } diff --git a/src/V2/Support/VersionUsageQuery.php b/src/V2/Support/VersionUsageQuery.php index ebbd0094d..95c985799 100644 --- a/src/V2/Support/VersionUsageQuery.php +++ b/src/V2/Support/VersionUsageQuery.php @@ -9,6 +9,7 @@ use Illuminate\Support\Collection; use InvalidArgumentException; use JsonException; +use LogicException; use Workflow\V2\Enums\HistoryEventType; use Workflow\V2\Enums\RunStatus; use Workflow\V2\Models\WorkflowHistoryEvent; @@ -18,6 +19,92 @@ /** Read-only, namespace-bound observations of retained open-run version decisions. */ final class VersionUsageQuery { + /** + * Frozen v1 retained format; enum growth does not silently expand reader support. + */ + private const EVENT_TYPES = [ + 'StartAccepted', + 'StartRejected', + 'WorkflowStarted', + 'WorkflowContinuedAsNew', + 'ChildWorkflowScheduled', + 'ChildWorkflowAwaited', + 'ChildCancellationRequested', + 'ChildCancellationResolved', + 'ChildRunStarted', + 'ChildRunCompleted', + 'ChildRunFailed', + 'ChildRunCancelled', + 'ChildRunTerminated', + 'ServiceOperationScheduled', + 'ServiceCallStarted', + 'ServiceCallCompleted', + 'ServiceCallFailed', + 'ServiceCallCancelled', + 'ConditionWaitOpened', + 'ConditionWaitSatisfied', + 'ConditionWaitTimedOut', + 'ConditionWaitCancelled', + 'SignalWaitOpened', + 'SignalWaitCancelled', + 'SignalReceived', + 'SignalApplied', + 'SignalRejected', + 'UpdateAccepted', + 'UpdateRejected', + 'UpdateApplied', + 'UpdateCompleted', + 'RepairRequested', + 'CancelRequested', + 'CooperativeCancellationRequested', + 'CooperativeCancellationDelivered', + 'CancellationScopeOpened', + 'CancellationScopeRequested', + 'CancellationScopeDeliveryPrepared', + 'CancellationScopeDelivered', + 'CancellationScopeRequestConflicted', + 'WorkflowCancelled', + 'TerminateRequested', + 'WorkflowTerminated', + 'ArchiveRequested', + 'WorkflowArchived', + 'ActivityScheduled', + 'ActivityStarted', + 'ActivityHeartbeatRecorded', + 'ActivityRetryScheduled', + 'ActivityCompleted', + 'ActivityFailed', + 'ActivityOutcomeRejected', + 'ActivityCancelled', + 'ActivityCancellationAcknowledged', + 'ActivityTimedOut', + 'FailureHandled', + 'SideEffectRecorded', + 'VersionMarkerRecorded', + 'TimerScheduled', + 'TimerFired', + 'TimerCancelled', + 'SelectionResolved', + 'SelectionOperationCancelled', + 'SearchAttributesUpserted', + 'MemoUpserted', + 'WorkflowTimedOut', + 'WorkflowTimeoutCleanupStarted', + 'WorkflowCompleted', + 'WorkflowFailed', + 'ParentClosePolicyApplied', + 'ParentClosePolicyFailed', + 'ParentCloseCancellationRequested', + 'MessageCursorAdvanced', + 'ScheduleCreated', + 'SchedulePaused', + 'ScheduleResumed', + 'ScheduleUpdated', + 'ScheduleTriggered', + 'ScheduleDeleted', + 'ScheduleTriggerSkipped', + ]; + public function page( ?string $namespace, string $workflowType, @@ -39,7 +126,11 @@ public function page( $identity = hash('sha256', json_encode([ $namespace, $workflowType, $changeId, $minVersion, $state, ], JSON_THROW_ON_ERROR)); - $continuation = $cursor === null ? null : VersionUsageCursor::decode($cursor, $identity); + $key = config('app.key'); + if (! is_string($key) || $key === '') { + throw new LogicException('A non-empty application key is required for version-usage cursors.'); + } + $continuation = $cursor === null ? null : VersionUsageCursor::decode($cursor, $identity, $key); $base = $this->runs($namespace, $workflowType); $through = $continuation['through'] ?? (clone $base)->orderByRaw($this->idExpression($base) . ' DESC')->value( 'id' @@ -91,7 +182,8 @@ public function page( 'next_cursor' => $hasMore ? VersionUsageCursor::encode( $identity, $through, - $results[array_key_last($results)]['run_id'] + $results[array_key_last($results)]['run_id'], + $key ) : null, ]; } @@ -194,15 +286,20 @@ private function decisions( $statistics = (clone $history)->select('workflow_run_id') ->selectRaw( "COUNT(*) AS event_count, MIN({$sequence}) AS first_sequence, MAX({$sequence}) AS last_sequence, " - . "SUM(CASE WHEN {$kind} = {$marker} THEN 1 ELSE 0 END) AS marker_count", - [HistoryEventType::VersionMarkerRecorded->value] + . "SUM(CASE WHEN {$kind} = {$marker} THEN 1 ELSE 0 END) AS marker_count, " + . "SUM(CASE WHEN {$kind} = {$marker} THEN 1 ELSE 0 END) AS started_count, " + . "SUM(CASE WHEN {$kind} = {$marker} THEN 1 ELSE 0 END) AS accepted_count", + [HistoryEventType::VersionMarkerRecorded->value, HistoryEventType::WorkflowStarted->value, + HistoryEventType::StartAccepted->value] ) ->groupBy('workflow_run_id') ->get() ->keyBy('workflow_run_id'); $starts = (clone $history)->where('sequence', 1) ->pluck('event_type', 'workflow_run_id'); - $types = array_map(static fn (HistoryEventType $type): string => $type->value, HistoryEventType::cases()); + $seconds = (clone $history)->where('sequence', 2) + ->pluck('event_type', 'workflow_run_id'); + $types = self::EVENT_TYPES; $driver = $history->getConnection() ->getDriverName(); $kind = self::binary($history->getGrammar()->wrap('event_type'), $driver); @@ -223,7 +320,10 @@ private function decisions( && (int) $stats->last_sequence === (int) $run->last_history_sequence && (int) $stats->event_count === (int) $run->last_history_sequence && (int) $stats->marker_count <= 1000 - && $starts->get($run->id) === $startedType + && (int) $stats->started_count === 1 + && (($starts->get($run->id) === $startedType && (int) $stats->accepted_count === 0) + || ($starts->get($run->id) === HistoryEventType::StartAccepted->value + && $seconds->get($run->id) === $startedType && (int) $stats->accepted_count === 1)) && ! in_array($run->id, $unsupported, true); $decisions[$run->id] = new VersionUsageDecision($changeId, $minVersion, $available); if ($available) { diff --git a/tests/Feature/V2/V2VersionUsageQueryTest.php b/tests/Feature/V2/V2VersionUsageQueryTest.php index 0887509bc..e41628cf8 100644 --- a/tests/Feature/V2/V2VersionUsageQueryTest.php +++ b/tests/Feature/V2/V2VersionUsageQueryTest.php @@ -8,6 +8,7 @@ use Illuminate\Database\Events\QueryExecuted; use Illuminate\Database\Schema\Blueprint; use Illuminate\Support\Facades\DB; +use Illuminate\Support\Facades\Queue; use InvalidArgumentException; use LogicException; use Tests\Fixtures\V2\TestVersionWorkflow; @@ -16,12 +17,21 @@ use Workflow\V2\Models\WorkflowHistoryEvent; use Workflow\V2\Models\WorkflowInstance; use Workflow\V2\Models\WorkflowRun; +use Workflow\V2\Models\WorkflowTask; use Workflow\V2\Support\VersionUsageQuery; use Workflow\V2\Support\WorkflowDefinition; use Workflow\V2\Support\WorkflowDefinitionFingerprint; +use Workflow\V2\WorkflowStub; final class V2VersionUsageQueryTest extends TestCase { + protected function setUp(): void + { + parent::setUp(); + config() + ->set('app.key', 'version-usage-feature-signing-key'); + } + public function testCountsIncludeOldNewAndNoMarkerBeforeStateFilteringOrPagination(): void { $this->createRun(1, 1); @@ -47,6 +57,57 @@ public function testCountsIncludeOldNewAndNoMarkerBeforeStateFilteringOrPaginati $this->assertNull($unknown['runs'][0]['recorded_version']); } + public function testActualRecordedMarkerWithTaskSnapshotClassifiesItsDecision(): void + { + config()->set('queue.default', 'database'); + Queue::fake(); + $workflow = WorkflowStub::make(TestVersionWorkflow::class, 'actual-version-usage', 'orders'); + $workflow->start(); + $run = WorkflowRun::query()->findOrFail($workflow->runId()); + $this->assertSame(['StartAccepted', 'WorkflowStarted'], $run->historyEvents->pluck('event_type')->map( + static fn (HistoryEventType $type): string => $type->value + )->all()); + $task = WorkflowTask::query()->create([ + 'workflow_run_id' => $run->id, + 'namespace' => $run->namespace, + 'task_type' => 'workflow', + 'status' => 'ready', + 'available_at' => now(), + 'payload' => [], + ]); + $event = WorkflowHistoryEvent::record($run, HistoryEventType::VersionMarkerRecorded, $this->marker(2), $task); + $this->assertSame($task->id, $event->payload['task']['id']); + $page = (new VersionUsageQuery())->page('orders', 'test-version-workflow', 'shipping', 2); + $this->assertSame('current', $page['runs'][0]['state']); + $this->assertSame(2, $page['runs'][0]['recorded_version']); + $this->assertNull($page['runs'][0]['unknown_reason']); + } + + public function testIncompleteOrRepeatedStartPrefixesStayUnavailable(): void + { + $missingStarted = $this->createRun(1, 2); + DB::table((new WorkflowHistoryEvent())->getTable())->where('workflow_run_id', $missingStarted->id) + ->where('sequence', 1) + ->update([ + 'event_type' => 'StartAccepted', + ]); + $repeated = $this->createRun(2, 2); + $this->event($repeated, 'WorkflowStarted', []); + $page = (new VersionUsageQuery())->page('orders', 'shipment', 'shipping', 2); + $this->assertSame(2, $page['counts']['unknown']); + $this->assertSame( + ['history_unavailable', 'history_unavailable'], + array_column($page['runs'], 'unknown_reason') + ); + } + + public function testPagingRequiresTheConfiguredApplicationSigningKey(): void + { + config()->set('app.key', ''); + $this->expectException(LogicException::class); + (new VersionUsageQuery())->page('orders', 'shipment', 'shipping', 2); + } + public function testNamespacesWorkflowTypesAndChangeIdsUseExactIdentities(): void { foreach ([null, 'default', 'orders', 'ORDERS', 'orders '] as $index => $namespace) { diff --git a/tests/Unit/V2/VersionUsageCursorTest.php b/tests/Unit/V2/VersionUsageCursorTest.php index 00c357c99..019dcc488 100644 --- a/tests/Unit/V2/VersionUsageCursorTest.php +++ b/tests/Unit/V2/VersionUsageCursorTest.php @@ -5,21 +5,69 @@ namespace Tests\Unit\V2; use InvalidArgumentException; +use LogicException; use PHPUnit\Framework\Attributes\DataProvider; use PHPUnit\Framework\TestCase; use Workflow\V2\Support\VersionUsageCursor; final class VersionUsageCursorTest extends TestCase { + private const KEY = 'version-usage-test-signing-key'; + public function testCursorBindsTheQueryAndPreservesBothKeysetBoundaries(): void { $fingerprint = hash('sha256', 'namespace/type/change/minimum/state'); - $encoded = VersionUsageCursor::encode($fingerprint, 'run-9', 'run-3'); - $value = VersionUsageCursor::decode($encoded, $fingerprint); + $encoded = VersionUsageCursor::encode($fingerprint, 'run-9', 'run-3', self::KEY); + $value = VersionUsageCursor::decode($encoded, $fingerprint, self::KEY); $this->assertSame('run-9', $value['through']); $this->assertSame('run-3', $value['after']); $this->expectException(InvalidArgumentException::class); - VersionUsageCursor::decode($encoded, hash('sha256', 'another-namespace')); + VersionUsageCursor::decode($encoded, hash('sha256', 'another-namespace'), self::KEY); + } + + public function testTamperedBoundariesQueryAndTagCannotAuthenticate(): void + { + $query = hash('sha256', 'query'); + $cursor = VersionUsageCursor::encode($query, 'run-9', 'run-3', self::KEY); + [$body, $tag] = explode('.', $cursor); + $value = json_decode(base64_decode(strtr($body, '-_', '+/'), true), true, flags: JSON_THROW_ON_ERROR); + foreach ([ + 'after' => 'run-4', + 'through' => 'run-8', + 'query' => hash('sha256', 'other'), + ] as $field => $replacement) { + $mutated = $value; + $mutated[$field] = $replacement; + $transport = rtrim(strtr(base64_encode(json_encode($mutated, JSON_THROW_ON_ERROR)), '+/', '-_'), '='); + try { + VersionUsageCursor::decode($transport . '.' . $tag, $query, self::KEY); + $this->fail('Modified cursor body was accepted.'); + } catch (InvalidArgumentException $exception) { + $this->assertSame('Invalid version-usage cursor signature.', $exception->getMessage()); + } + } + foreach ([$body, $body . '.' . str_repeat('0', 64), 'bm90LWpzb24.' . $tag] as $mutated) { + try { + VersionUsageCursor::decode($mutated, $query, self::KEY); + $this->fail('Unsigned or modified cursor was accepted.'); + } catch (InvalidArgumentException) { + $this->addToAssertionCount(1); + } + } + } + + public function testRotatedKeyRefusesAnOldCursor(): void + { + $query = hash('sha256', 'query'); + $cursor = VersionUsageCursor::encode($query, 'run-9', 'run-3', self::KEY); + $this->expectException(InvalidArgumentException::class); + VersionUsageCursor::decode($cursor, $query, 'another-application-key'); + } + + public function testAnEmptySigningKeyIsRefused(): void + { + $this->expectException(LogicException::class); + VersionUsageCursor::encode(hash('sha256', 'query'), 'run-9', 'run-3', ''); } public static function malformed(): iterable @@ -27,8 +75,8 @@ public static function malformed(): iterable yield 'empty' => ['']; yield 'base64 punctuation' => ['%%']; yield 'oversized' => [str_repeat('A', 2049)]; - yield 'not json' => ['YWJj']; - yield 'scalar' => ['MQ']; + yield 'not json' => [self::signed('abc')]; + yield 'scalar' => [self::signed('1')]; $query = hash('sha256', 'query'); foreach ([[ 'version' => 2, @@ -41,12 +89,12 @@ public static function malformed(): iterable ], [ 'extra' => true, ]] as $mutation) { - yield json_encode($mutation) => [rtrim(strtr(base64_encode(json_encode(array_replace([ + yield json_encode($mutation) => [self::signed(json_encode(array_replace([ 'version' => 1, 'query' => $query, 'through' => 'run-9', 'after' => 'run-1', - ], $mutation), JSON_THROW_ON_ERROR)), '+/', '-_'), '=')]; + ], $mutation), JSON_THROW_ON_ERROR))]; } } @@ -54,6 +102,12 @@ public static function malformed(): iterable public function testMalformedCursorsAreErrorsRatherThanFirstPages(string $cursor): void { $this->expectException(InvalidArgumentException::class); - VersionUsageCursor::decode($cursor, hash('sha256', 'query')); + VersionUsageCursor::decode($cursor, hash('sha256', 'query'), self::KEY); + } + + private static function signed(string $json): string + { + $body = rtrim(strtr(base64_encode($json), '+/', '-_'), '='); + return $body . '.' . hash_hmac('sha256', "durable-workflow.version-usage-cursor.v1\0" . $body, self::KEY); } } diff --git a/tests/Unit/V2/VersionUsageDecisionTest.php b/tests/Unit/V2/VersionUsageDecisionTest.php index 9e8700939..c83578e1c 100644 --- a/tests/Unit/V2/VersionUsageDecisionTest.php +++ b/tests/Unit/V2/VersionUsageDecisionTest.php @@ -140,6 +140,29 @@ public function testChangeIdsAreExactAndDuplicateLogicalPositionsAreInvalid(): v $this->assertSame('history_invalid', $decision->result()['unknown_reason']); } + public function testTaskSnapshotIsKnownButOtherMarkerExtensionsRemainUnavailable(): void + { + $decision = new VersionUsageDecision('shipping', 2, true); + $decision->observe($this->marker(2) + [ + 'task' => [ + 'id' => 'task-1', + 'status' => 'leased', + ], + ]); + $this->assertSame('current', $decision->result()['state']); + foreach ([[ + 'task' => 'bad', + ], [ + 'task' => ['list'], + ], [ + 'future_format' => 2, + ]] as $extension) { + $other = new VersionUsageDecision('shipping', 2, true); + $other->observe($this->marker(2) + $extension); + $this->assertSame('history_unavailable', $other->result()['unknown_reason']); + } + } + public function testInspectionBudgetCannotReturnAnOptimisticPartialDecision(): void { $decision = new VersionUsageDecision('shipping', 1, true); From 79b2f531491da2472f31972c071f612a35003b4d Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 14 Nov 2026 12:09:36 +0000 Subject: [PATCH 07/11] fix(ci): distinguish read-only version usage from replay changes --- .../ci/classify-version-usage-observer.php | 122 ++++++++++++++++++ scripts/ci/test-regression-corpus-policy.py | 59 +++++++++ scripts/ci/validate-regression-corpus.py | 26 +++- 3 files changed, 206 insertions(+), 1 deletion(-) create mode 100644 scripts/ci/classify-version-usage-observer.php diff --git a/scripts/ci/classify-version-usage-observer.php b/scripts/ci/classify-version-usage-observer.php new file mode 100644 index 000000000..feb13dd95 --- /dev/null +++ b/scripts/ci/classify-version-usage-observer.php @@ -0,0 +1,122 @@ + + ! is_array($token) || ! in_array($token[0], [T_WHITESPACE, T_COMMENT, T_DOC_COMMENT], true))); + $imports = [ + 'Generator', 'Illuminate\\Database\\Eloquent\\Builder', 'Illuminate\\Support\\Collection', + 'InvalidArgumentException', 'JsonException', 'LogicException', + 'Workflow\\V2\\Enums\\HistoryEventType', 'Workflow\\V2\\Enums\\RunStatus', + 'Workflow\\V2\\Models\\WorkflowHistoryEvent', 'Workflow\\V2\\Models\\WorkflowInstance', + 'Workflow\\V2\\Models\\WorkflowRun', + ]; + $functions = [ + 'config', 'strlen', 'preg_match', 'hash', 'hash_hmac', 'hash_equals', 'json_encode', 'json_decode', + 'is_string', 'is_int', 'is_array', 'in_array', 'strcmp', 'count', 'sort', 'array_slice', + 'array_key_last', 'array_keys', 'array_key_exists', 'array_diff', 'array_is_list', 'array_fill', + 'implode', 'rtrim', 'strtr', 'base64_encode', 'base64_decode', + ]; + $methods = [ + 'getQuery', 'getConnection', 'getDriverName', 'getGrammar', 'wrap', 'getModel', 'qualifyColumn', + 'setEagerLoads', 'where', 'whereIn', 'whereRaw', 'orderByRaw', 'select', 'selectRaw', 'limit', + 'groupBy', 'toBase', 'get', 'pluck', 'value', 'keyBy', 'distinct', 'lazyById', 'isEmpty', 'last', + 'count', 'all', 'setConnection', 'getConnectionName', 'newQuery', + 'runs', 'idExpression', 'batches', 'decisions', 'observe', 'result', 'unavailable', + ]; + $static = [ + 'ConfiguredV2Models' => ['resolve'], 'NamespaceIdentity' => ['constrain'], + 'VersionUsageDecision' => ['integer'], 'VersionUsageCursor' => ['encode', 'decode'], + 'self' => ['text', 'binary', 'runId', 'encode', 'signature', 'integer'], + ]; + $namespaces = 0; + $classes = 0; + foreach ($tokens as $index => $token) { + $previous = $tokens[$index - 1] ?? null; + $next = $tokens[$index + 1] ?? null; + $text = is_array($token) ? $token[1] : $token; + $previousKind = is_array($previous) ? $previous[0] : null; + if (is_array($token) && in_array($token[0], [ + T_EVAL, T_INCLUDE, T_INCLUDE_ONCE, T_REQUIRE, T_REQUIRE_ONCE, T_EXIT, + T_GLOBAL, T_GOTO, T_PRINT, T_ECHO, T_HALT_COMPILER, T_EXTENDS, T_IMPLEMENTS, + T_INTERFACE, T_TRAIT, T_ENUM, + ], true)) { + return false; + } + if ($text === '`' || (in_array($text, [')', ']', '}'], true) && $next === '(') + || (is_array($token) && $token[0] === T_VARIABLE && $next === '(' + && $previousKind !== T_NEW)) { + return false; + } + if (is_array($token) && $token[0] === T_NAMESPACE) { + ++$namespaces; + if (! is_array($next) || $next[1] !== 'Workflow\\V2\\Support') { + return false; + } + } + if (is_array($token) && $token[0] === T_CLASS) { + // ::class is metadata, not a class declaration. + if ($previousKind === T_DOUBLE_COLON) { + continue; + } + ++$classes; + if (! is_array($next) || $next[1] !== $class) { + return false; + } + } + if (is_array($token) && $token[0] === T_USE) { + if (! is_array($next) || ! in_array($next[1], $imports, true) + || ($tokens[$index + 2] ?? null) !== ';') { + return false; + } + } + if ($previousKind === T_OBJECT_OPERATOR || $previousKind === T_NULLSAFE_OBJECT_OPERATOR) { + if (! is_array($token) || $token[0] !== T_STRING || ! in_array($text, $methods, true)) { + // Reading a field is safe; a variable selector/call is not. + if (! is_array($token) || $token[0] !== T_STRING || $next === '(') { + return false; + } + } + } + if ($previousKind === T_DOUBLE_COLON && $next === '(') { + $owner = $tokens[$index - 2] ?? null; + $owner = is_array($owner) ? $owner[1] : $owner; + if (! in_array($text, $static[$owner] ?? [], true) + && ! ($owner === '$model' && $text === 'query')) { + return false; + } + } + if (is_array($token) && in_array($token[0], [T_STRING, T_NAME_FULLY_QUALIFIED, T_NAME_QUALIFIED], true) + && $next === '(' && ! in_array($previousKind, [T_FUNCTION, T_NEW, T_OBJECT_OPERATOR, + T_NULLSAFE_OBJECT_OPERATOR, T_DOUBLE_COLON], true) + && ! in_array($text, $functions, true)) { + return false; + } + if (is_array($token) && $token[0] === T_NEW) { + $type = is_array($next) ? $next[1] : null; + if (! in_array($type, ['InvalidArgumentException', 'LogicException', 'VersionUsageDecision', + '$instanceModel', '$eventModel'], true)) { + return false; + } + } + if (is_array($token) && in_array($token[0], [T_CONSTANT_ENCAPSED_STRING, T_ENCAPSED_AND_WHITESPACE], true) + && preg_match('/\b(INSERT|UPDATE|DELETE|REPLACE|ALTER|DROP|CREATE|TRUNCATE|CALL|EXECUTE|INTO|COPY|PRAGMA|ATTACH|DETACH|VACUUM|GRANT|REVOKE)\b/i', $text)) { + return false; + } + } + return $namespaces === 1 && $classes === 1; +} + +try { + $input = json_decode(stream_get_contents(STDIN), true, flags: JSON_THROW_ON_ERROR); + echo json_encode(versionUsageObserver($input['path'], $input['source']), JSON_THROW_ON_ERROR); +} catch (Throwable) { + echo 'false'; +} diff --git a/scripts/ci/test-regression-corpus-policy.py b/scripts/ci/test-regression-corpus-policy.py index 132a5a2c3..ce6052663 100644 --- a/scripts/ci/test-regression-corpus-policy.py +++ b/scripts/ci/test-regression-corpus-policy.py @@ -2141,6 +2141,65 @@ def test_core_replay_surfaces_require_growth_without_diff_keywords(self) -> None ) target.write_text(" None: + for name in ("VersionUsageQuery", "VersionUsageDecision", "VersionUsageCursor"): + source = self.root / f"src/V2/Support/{name}.php" + source.write_text( + "last_history_sequence; }\n}", + encoding="utf-8", + ) + self.git("add", "src/V2/Support") + result = self.validate() + self.assertEqual(0, result.returncode, result.stderr) + + def test_version_usage_names_cannot_hide_writes_replay_or_dynamic_execution(self) -> None: + source = self.root / "src/V2/Support/VersionUsageQuery.php" + for body in ( + "$run->save(); return $run->last_history_sequence;", + "WorkflowHistoryEvent::record($run, $type); return $run->last_history_sequence;", + "WorkflowFiberRunner::forClass('demo'); return $run->last_history_sequence;", + "$function(); return $run->last_history_sequence;", + "$functions[0](); return $run->last_history_sequence;", + "($function)(); return $run->last_history_sequence;", + "$run->$method(); return $run->last_history_sequence;", + "eval('return 1;'); return $run->last_history_sequence;", + "$run->selectRaw('DELETE FROM workflow_runs'); return $run->last_history_sequence;", + "$run->selectRaw('id INTO OUTFILE name'); return $run->last_history_sequence;", + ): + with self.subTest(body=body): + source.write_text( + "last_history_sequence; } }", + encoding="utf-8", + ) + result = self.validate() + self.assertNotEqual(0, result.returncode, result.stdout) + self.assertIn("replay implementation changed but its corpus did not grow", result.stderr) + + def test_version_usage_read_exception_does_not_exempt_other_support_changes(self) -> None: + source = self.root / "src/V2/Support/VersionUsageQuery.php" + source.write_text( + "last_history_sequence; } }", + encoding="utf-8", + ) + companion = self.root / "src/V2/Support/AnotherHistoryHelper.php" + companion.write_text("last_history_sequence + 1;", encoding="utf-8") + result = self.validate() + self.assertNotEqual(0, result.returncode, result.stdout) + self.assertIn("replay implementation changed but its corpus did not grow", result.stderr) + def test_task_discovery_helper_body_does_not_require_replay_growth(self) -> None: source = self.root / "src/V2/Support/DefaultWorkflowTaskBridge.php" original = """ Date: Sat, 14 Nov 2026 12:27:04 +0000 Subject: [PATCH 08/11] fix(versioning): validate reuse provenance and isolate retained identities --- docs/version-usage.md | 8 +- .../ci/classify-version-usage-observer.php | 39 ++++-- scripts/ci/test-regression-corpus-policy.py | 2 + src/V2/Support/VersionUsageDecision.php | 33 +++++ src/V2/Support/VersionUsageQuery.php | 131 +++++++++++------- tests/Feature/V2/V2VersionUsageQueryTest.php | 98 +++++++++++++ tests/Unit/V2/VersionUsageDecisionTest.php | 52 +++++++ 7 files changed, 302 insertions(+), 61 deletions(-) diff --git a/docs/version-usage.md b/docs/version-usage.md index 490ad8df9..3f5e5e07a 100644 --- a/docs/version-usage.md +++ b/docs/version-usage.md @@ -67,7 +67,13 @@ Marker readers must understand their runtime's retained format. Embedded marker fields are `change_id`, `version`, `min_supported`, `max_supported` and positive logical `sequence`; the range must contain the recorded version. The known producer-injected `task` snapshot is supported; unknown marker extensions require -a reader upgrade and remain unavailable. No application +a reader upgrade and remain unavailable. Reused decisions must retain both +`reused_from_run_id` and `reused_recorded_at`, or neither. The source ID is a +nonempty bounded run identity; the clock is a valid RFC 3339 instant with an +explicit offset and the runtime's supported precision (at most six fractional +digits for embedded). Malformed or incomplete reuse +provenance remains `history_invalid`; readers never rewrite retained clock strings. +No application argument, output, memo or payload-store fetch is necessary for this query. ## Response and paging diff --git a/scripts/ci/classify-version-usage-observer.php b/scripts/ci/classify-version-usage-observer.php index feb13dd95..355361b78 100644 --- a/scripts/ci/classify-version-usage-observer.php +++ b/scripts/ci/classify-version-usage-observer.php @@ -12,7 +12,8 @@ function versionUsageObserver(string $path, string $source): bool $tokens = array_values(array_filter(token_get_all($source, TOKEN_PARSE), static fn ($token): bool => ! is_array($token) || ! in_array($token[0], [T_WHITESPACE, T_COMMENT, T_DOC_COMMENT], true))); $imports = [ - 'Generator', 'Illuminate\\Database\\Eloquent\\Builder', 'Illuminate\\Support\\Collection', + 'DateTimeImmutable', 'Generator', 'Illuminate\\Database\\Query\\Builder', + 'Illuminate\\Database\\Eloquent\\Builder', 'Illuminate\\Support\\Collection', 'InvalidArgumentException', 'JsonException', 'LogicException', 'Workflow\\V2\\Enums\\HistoryEventType', 'Workflow\\V2\\Enums\\RunStatus', 'Workflow\\V2\\Models\\WorkflowHistoryEvent', 'Workflow\\V2\\Models\\WorkflowInstance', @@ -20,22 +21,28 @@ function versionUsageObserver(string $path, string $source): bool ]; $functions = [ 'config', 'strlen', 'preg_match', 'hash', 'hash_hmac', 'hash_equals', 'json_encode', 'json_decode', - 'is_string', 'is_int', 'is_array', 'in_array', 'strcmp', 'count', 'sort', 'array_slice', + 'str_contains', 'is_string', 'is_int', 'is_array', 'in_array', 'strcmp', 'count', 'sort', 'array_slice', 'array_key_last', 'array_keys', 'array_key_exists', 'array_diff', 'array_is_list', 'array_fill', 'implode', 'rtrim', 'strtr', 'base64_encode', 'base64_decode', ]; $methods = [ 'getQuery', 'getConnection', 'getDriverName', 'getGrammar', 'wrap', 'getModel', 'qualifyColumn', 'setEagerLoads', 'where', 'whereIn', 'whereRaw', 'orderByRaw', 'select', 'selectRaw', 'limit', - 'groupBy', 'toBase', 'get', 'pluck', 'value', 'keyBy', 'distinct', 'lazyById', 'isEmpty', 'last', + 'groupByRaw', 'groupBy', 'toBase', 'get', 'pluck', 'value', 'keyBy', 'distinct', 'lazyById', 'isEmpty', 'last', 'count', 'all', 'setConnection', 'getConnectionName', 'newQuery', - 'runs', 'idExpression', 'batches', 'decisions', 'observe', 'result', 'unavailable', + 'runs', 'idExpression', 'batches', 'rows', 'identities', 'decisions', 'observe', 'result', 'unavailable', ]; $static = [ 'ConfiguredV2Models' => ['resolve'], 'NamespaceIdentity' => ['constrain'], 'VersionUsageDecision' => ['integer'], 'VersionUsageCursor' => ['encode', 'decode'], - 'self' => ['text', 'binary', 'runId', 'encode', 'signature', 'integer'], + 'DateTimeImmutable' => ['createFromFormat', 'getLastErrors'], + 'self' => ['text', 'binary', 'runId', 'encode', 'signature', 'integer', 'validReuse'], ]; + if ($class !== 'VersionUsageQuery') { + $methods = []; + $static = ['self' => $static['self']] + ($class === 'VersionUsageDecision' + ? ['DateTimeImmutable' => $static['DateTimeImmutable']] : []); + } $namespaces = 0; $classes = 0; foreach ($tokens as $index => $token) { @@ -50,7 +57,8 @@ function versionUsageObserver(string $path, string $source): bool ], true)) { return false; } - if ($text === '`' || (in_array($text, [')', ']', '}'], true) && $next === '(') + if (($class !== 'VersionUsageCursor' && $text === '.') || $text === '`' + || (in_array($text, [')', ']', '}'], true) && $next === '(') || (is_array($token) && $token[0] === T_VARIABLE && $next === '(' && $previousKind !== T_NEW)) { return false; @@ -72,8 +80,14 @@ function versionUsageObserver(string $path, string $source): bool } } if (is_array($token) && $token[0] === T_USE) { + $alias = $tokens[$index + 2] ?? null; + $aliasName = $tokens[$index + 3] ?? null; + $queryAlias = is_array($next) && $next[1] === 'Illuminate\\Database\\Query\\Builder' + && is_array($alias) && $alias[0] === T_AS + && is_array($aliasName) && $aliasName[1] === 'QueryBuilder' + && ($tokens[$index + 4] ?? null) === ';'; if (! is_array($next) || ! in_array($next[1], $imports, true) - || ($tokens[$index + 2] ?? null) !== ';') { + || (($tokens[$index + 2] ?? null) !== ';' && ! $queryAlias)) { return false; } } @@ -89,7 +103,7 @@ function versionUsageObserver(string $path, string $source): bool $owner = $tokens[$index - 2] ?? null; $owner = is_array($owner) ? $owner[1] : $owner; if (! in_array($text, $static[$owner] ?? [], true) - && ! ($owner === '$model' && $text === 'query')) { + && ! ($class === 'VersionUsageQuery' && $owner === '$model' && $text === 'query')) { return false; } } @@ -101,13 +115,16 @@ function versionUsageObserver(string $path, string $source): bool } if (is_array($token) && $token[0] === T_NEW) { $type = is_array($next) ? $next[1] : null; - if (! in_array($type, ['InvalidArgumentException', 'LogicException', 'VersionUsageDecision', - '$instanceModel', '$eventModel'], true)) { + $types = ['InvalidArgumentException', 'LogicException']; + if ($class === 'VersionUsageQuery') { + $types = array_merge($types, ['VersionUsageDecision', '$instanceModel', '$eventModel']); + } + if (! in_array($type, $types, true)) { return false; } } if (is_array($token) && in_array($token[0], [T_CONSTANT_ENCAPSED_STRING, T_ENCAPSED_AND_WHITESPACE], true) - && preg_match('/\b(INSERT|UPDATE|DELETE|REPLACE|ALTER|DROP|CREATE|TRUNCATE|CALL|EXECUTE|INTO|COPY|PRAGMA|ATTACH|DETACH|VACUUM|GRANT|REVOKE)\b/i', $text)) { + && preg_match('/\b(INSERT|UPDATE|DELETE|REPLACE|ALTER|DROP|CREATE|TRUNCATE|CALL|EXECUTE|INTO|OUTFILE|DUMPFILE|COPY|PRAGMA|ATTACH|DETACH|VACUUM|GRANT|REVOKE)\b/i', $text)) { return false; } } diff --git a/scripts/ci/test-regression-corpus-policy.py b/scripts/ci/test-regression-corpus-policy.py index ce6052663..45ee50b05 100644 --- a/scripts/ci/test-regression-corpus-policy.py +++ b/scripts/ci/test-regression-corpus-policy.py @@ -2167,6 +2167,8 @@ def test_version_usage_names_cannot_hide_writes_replay_or_dynamic_execution(self "eval('return 1;'); return $run->last_history_sequence;", "$run->selectRaw('DELETE FROM workflow_runs'); return $run->last_history_sequence;", "$run->selectRaw('id INTO OUTFILE name'); return $run->last_history_sequence;", + "$run->selectRaw('id IN'.'TO OUTFILE \\'/tmp/version-usage-write\\'')->get(); return $run->last_history_sequence;", + "$run->selectRaw('id IN'.('TO OUTFILE \\'/tmp/version-usage-write\\''))->get(); return $run->last_history_sequence;", ): with self.subTest(body=body): source.write_text( diff --git a/src/V2/Support/VersionUsageDecision.php b/src/V2/Support/VersionUsageDecision.php index bf362a6e9..61ff16cce 100644 --- a/src/V2/Support/VersionUsageDecision.php +++ b/src/V2/Support/VersionUsageDecision.php @@ -4,6 +4,8 @@ namespace Workflow\V2\Support; +use DateTimeImmutable; + /** @internal Reduces retained marker data without replay or application payload decoding. */ final class VersionUsageDecision { @@ -43,6 +45,10 @@ public function observe(mixed $payload): void $this->reason = 'history_invalid'; return; } + if (! self::validReuse($payload)) { + $this->reason = 'history_invalid'; + return; + } if (array_diff(array_keys($payload), [ 'sequence', 'change_id', 'version', 'min_supported', 'max_supported', 'reused_from_run_id', 'reused_recorded_at', 'task', @@ -81,4 +87,31 @@ public static function integer(mixed $value): bool { return is_int($value) && $value >= -2147483648 && $value <= 2147483647; } + + private static function validReuse(array $payload): bool + { + $hasSource = array_key_exists('reused_from_run_id', $payload); + if ($hasSource !== array_key_exists('reused_recorded_at', $payload)) { + return false; + } + if (! $hasSource) { + return true; + } + $source = $payload['reused_from_run_id']; + $clock = $payload['reused_recorded_at']; + if (! is_string($source) || preg_match('/\A[A-Za-z0-9][A-Za-z0-9_.:-]{0,190}\z/D', $source) !== 1 + || ! is_string($clock) || strlen($clock) > 64 + || preg_match( + '/\A[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}(?:\.[0-9]{1,6})?(?:Z|[+-](?:[01][0-9]|2[0-3]):[0-5][0-9])\z/D', + $clock + ) !== 1) { + return false; + } + $time = DateTimeImmutable::createFromFormat( + str_contains($clock, '.') ? '!Y-m-d\TH:i:s.uP' : '!Y-m-d\TH:i:sP', + $clock, + ); + $errors = DateTimeImmutable::getLastErrors(); + return $time !== false && ($errors === false || ($errors['warning_count'] === 0 && $errors['error_count'] === 0)); + } } diff --git a/src/V2/Support/VersionUsageQuery.php b/src/V2/Support/VersionUsageQuery.php index 95c985799..0e1a68b9d 100644 --- a/src/V2/Support/VersionUsageQuery.php +++ b/src/V2/Support/VersionUsageQuery.php @@ -6,6 +6,7 @@ use Generator; use Illuminate\Database\Eloquent\Builder; +use Illuminate\Database\Query\Builder as QueryBuilder; use Illuminate\Support\Collection; use InvalidArgumentException; use JsonException; @@ -132,9 +133,8 @@ public function page( } $continuation = $cursor === null ? null : VersionUsageCursor::decode($cursor, $identity, $key); $base = $this->runs($namespace, $workflowType); - $through = $continuation['through'] ?? (clone $base)->orderByRaw($this->idExpression($base) . ' DESC')->value( - 'id' - ); + $idExpression = $this->idExpression($base); + $through = $continuation['through'] ?? (clone $base)->orderByRaw("{$idExpression} DESC")->value('id'); $after = $continuation['after'] ?? null; $counts = [ 'total' => 0, @@ -195,16 +195,28 @@ private function batches(Builder $base, string $through): Generator { $id = $this->idExpression($base); $parameter = self::binary('?', $base->getQuery()->getConnection()->getDriverName()); + $query = (clone $base)->select([ + 'id', 'workflow_instance_id', 'status', 'details_pruned_at', 'last_history_sequence', 'import_source', + ])->whereRaw("{$id} <= {$parameter}", [$through])->toBase(); + yield from $this->rows($query, 100); + } + + /** + * @return Generator + */ + private function rows(QueryBuilder $base, int $size): Generator + { + $driver = $base->getConnection() + ->getDriverName(); + $id = self::binary($base->getGrammar()->wrap('id'), $driver); + $parameter = self::binary('?', $driver); $after = null; do { - $query = (clone $base)->select([ - 'id', 'workflow_instance_id', 'status', 'details_pruned_at', 'last_history_sequence', 'import_source', - ])->whereRaw($id . ' <= ' . $parameter, [$through])->orderByRaw($id . ' ASC'); + $query = (clone $base)->orderByRaw("{$id} ASC"); if ($after !== null) { - $query->whereRaw($id . ' > ' . $parameter, [$after]); + $query->whereRaw("{$id} > {$parameter}", [$after]); } - $rows = $query->limit(100) - ->toBase() + $rows = $query->limit($size) ->get(); if ($rows->isEmpty()) { return; @@ -212,7 +224,20 @@ private function batches(Builder $base, string $through): Generator yield $rows; $after = $rows->last() ->id; - } while ($rows->count() === 100); + } while ($rows->count() === $size); + } + + private function identities(QueryBuilder $query, string $column, array $values): QueryBuilder + { + $query->whereIn($column, $values); + if ($values === []) { + return $query; + } + $driver = $query->getConnection() + ->getDriverName(); + $field = self::binary($query->getGrammar()->wrap($column), $driver); + $parameters = implode(', ', array_fill(0, count($values), self::binary('?', $driver))); + return $query->whereRaw("{$field} IN ({$parameters})", $values); } private function idExpression(Builder $query): string @@ -242,12 +267,15 @@ private function runs(?string $namespace, string $workflowType): Builder $driver = $query->getQuery() ->getConnection() ->getDriverName(); - $query->whereRaw(self::binary($column, $driver) . ' = ' . self::binary('?', $driver), [$workflowType]); + $column = self::binary($column, $driver); + $parameter = self::binary('?', $driver); + $query->whereRaw("{$column} = {$parameter}", [$workflowType]); $status = $query->getQuery() ->getGrammar() ->wrap($query->getModel()->qualifyColumn('status')); $parameters = implode(', ', array_fill(0, 3, self::binary('?', $driver))); - $query->whereRaw(self::binary($status, $driver) . ' IN (' . $parameters . ')', [ + $status = self::binary($status, $driver); + $query->whereRaw("{$status} IN ({$parameters})", [ RunStatus::Pending->value, RunStatus::Running->value, RunStatus::Waiting->value, ]); return $query; @@ -264,35 +292,35 @@ private function decisions( int $minVersion ): array { $instanceModel = ConfiguredV2Models::resolve('instance_model', WorkflowInstance::class); - $owners = NamespaceIdentity::constrain( + $owners = $this->identities(NamespaceIdentity::constrain( (new $instanceModel())->setConnection($runModel->getConnectionName()) ->newQuery() ->setEagerLoads([]), $namespace, - )->whereIn('id', $runs->pluck('workflow_instance_id')->all()) - ->toBase() - ->pluck('id') + )->toBase(), 'id', $runs->pluck('workflow_instance_id') + ->all())->pluck('id') ->all(); $eventModel = ConfiguredV2Models::resolve('history_event_model', WorkflowHistoryEvent::class); - $history = (new $eventModel())->setConnection($runModel->getConnectionName()) - ->newQuery() - ->setEagerLoads([])->whereIn('workflow_run_id', $runs->pluck('id')->all())->toBase(); + $history = $this->identities( + (new $eventModel())->setConnection($runModel->getConnectionName()) + ->newQuery() + ->setEagerLoads([])->toBase(), + 'workflow_run_id', + $runs->pluck('id') + ->all(), + ); $driver = $history->getConnection() ->getDriverName(); $sequence = $history->getGrammar() ->wrap('sequence'); $kind = self::binary($history->getGrammar()->wrap('event_type'), $driver); $marker = self::binary('?', $driver); - $statistics = (clone $history)->select('workflow_run_id') - ->selectRaw( - "COUNT(*) AS event_count, MIN({$sequence}) AS first_sequence, MAX({$sequence}) AS last_sequence, " - . "SUM(CASE WHEN {$kind} = {$marker} THEN 1 ELSE 0 END) AS marker_count, " - . "SUM(CASE WHEN {$kind} = {$marker} THEN 1 ELSE 0 END) AS started_count, " - . "SUM(CASE WHEN {$kind} = {$marker} THEN 1 ELSE 0 END) AS accepted_count", - [HistoryEventType::VersionMarkerRecorded->value, HistoryEventType::WorkflowStarted->value, - HistoryEventType::StartAccepted->value] - ) - ->groupBy('workflow_run_id') + $runId = self::binary($history->getGrammar()->wrap('workflow_run_id'), $driver); + $statistics = (clone $history)->selectRaw( + "{$runId} AS workflow_run_id, COUNT(*) AS event_count, MIN({$sequence}) AS first_sequence, MAX({$sequence}) AS last_sequence, SUM(CASE WHEN {$kind} = {$marker} THEN 1 ELSE 0 END) AS marker_count, SUM(CASE WHEN {$kind} = {$marker} THEN 1 ELSE 0 END) AS started_count, SUM(CASE WHEN {$kind} = {$marker} THEN 1 ELSE 0 END) AS accepted_count", + [HistoryEventType::VersionMarkerRecorded->value, HistoryEventType::WorkflowStarted->value, + HistoryEventType::StartAccepted->value], + )->groupByRaw($runId) ->get() ->keyBy('workflow_run_id'); $starts = (clone $history)->where('sequence', 1) @@ -304,8 +332,10 @@ private function decisions( ->getDriverName(); $kind = self::binary($history->getGrammar()->wrap('event_type'), $driver); $parameters = implode(', ', array_fill(0, count($types), self::binary('?', $driver))); - $unsupported = (clone $history)->whereRaw($kind . ' NOT IN (' . $parameters . ')', $types) + $unsupported = (clone $history)->whereRaw("{$kind} NOT IN ({$parameters})", $types) + ->selectRaw("{$runId} AS workflow_run_id") ->distinct() + ->get() ->pluck('workflow_run_id') ->all(); $startedType = HistoryEventType::WorkflowStarted->value; @@ -330,28 +360,31 @@ private function decisions( $inspect[] = $run->id; } } - foreach ((clone $history)->whereIn('workflow_run_id', $inspect)->where( - 'event_type', - HistoryEventType::VersionMarkerRecorded->value - ) - ->lazyById(250) as $event) { - if ($event->event_type !== HistoryEventType::VersionMarkerRecorded->value) { - $decisions[$event->workflow_run_id]->unavailable(); - continue; - } - try { - $payload = is_string($event->payload) - ? json_decode($event->payload, true, 32, JSON_THROW_ON_ERROR) : $event->payload; - } catch (JsonException) { - $payload = null; + $markers = $this->identities(clone $history, 'workflow_run_id', $inspect) + ->where('event_type', HistoryEventType::VersionMarkerRecorded->value) + ->select(['id', 'workflow_run_id', 'event_type', 'payload']); + foreach ($this->rows($markers, 250) as $events) { + foreach ($events as $event) { + if ($event->event_type !== HistoryEventType::VersionMarkerRecorded->value) { + $decisions[$event->workflow_run_id]->unavailable(); + continue; + } + try { + $payload = is_string($event->payload) + ? json_decode($event->payload, true, 32, JSON_THROW_ON_ERROR) : $event->payload; + } catch (JsonException) { + $payload = null; + } + $decisions[$event->workflow_run_id]->observe($payload); } - $decisions[$event->workflow_run_id]->observe($payload); } // A prune or append racing this observation cannot turn a partial read into current usage. - $latest = $runModel->newQuery() - ->setEagerLoads([])->whereIn('id', $runs->pluck('id')->all()) - ->toBase() - ->get(['id', 'last_history_sequence', 'details_pruned_at'])->keyBy('id'); + $latest = $this->identities( + NamespaceIdentity::constrain($runModel->newQuery()->setEagerLoads([]), $namespace)->toBase(), + 'id', + $runs->pluck('id') + ->all(), + )->get(['id', 'last_history_sequence', 'details_pruned_at'])->keyBy('id'); foreach ($runs as $run) { $end = $latest->get($run->id); if ($end === null || $end->details_pruned_at !== null diff --git a/tests/Feature/V2/V2VersionUsageQueryTest.php b/tests/Feature/V2/V2VersionUsageQueryTest.php index e41628cf8..95495d038 100644 --- a/tests/Feature/V2/V2VersionUsageQueryTest.php +++ b/tests/Feature/V2/V2VersionUsageQueryTest.php @@ -276,6 +276,104 @@ public function testConflictingAndMalformedMarkersNeverBecomeCurrent(): void $this->assertSame(['history_invalid', 'history_invalid'], array_column($page['runs'], 'unknown_reason')); } + public function testMalformedReusedMarkersRemainUnknownThroughTheStoredQuery(): void + { + foreach ([[ + 'reused_from_run_id' => [], + ], [ + 'reused_recorded_at' => '2026-11-14T01:02:03Z', + ], + [ + 'reused_from_run_id' => '', + 'reused_recorded_at' => '2026-11-14T01:02:03Z', + ], + [ + 'reused_from_run_id' => 'original', + 'reused_recorded_at' => '', + ], + [ + 'reused_from_run_id' => 'original', + 'reused_recorded_at' => '2026-02-30T01:02:03Z', + ], + ] as $index => $reuse) { + $run = $this->createRun($index + 1, null); + $this->event($run, 'VersionMarkerRecorded', $this->marker(2) + $reuse); + } + $page = (new VersionUsageQuery())->page('orders', 'shipment', 'shipping', 2); + $this->assertSame([ + 'total' => 5, + 'legacy' => 0, + 'current' => 0, + 'unknown' => 5, + ], $page['counts']); + $this->assertSame( + ['history_invalid'], + array_values(array_unique(array_column($page['runs'], 'unknown_reason'))) + ); + } + + public function testCaseVariantHistoryCannotContaminateOrCompleteAnotherRunsPrefix(): void + { + foreach ([ + 1 => 'Case-A', + 2 => 'Gap-A', + ] as $number => $id) { + $run = $this->createRun($number, 2); + DB::table((new WorkflowHistoryEvent())->getTable())->where('workflow_run_id', $run->id)->update([ + 'workflow_run_id' => $id, + ]); + DB::table((new WorkflowRun())->getTable())->where('id', $run->id)->update([ + 'id' => $id, + 'last_history_sequence' => $number === 2 ? 3 : 2, + ]); + DB::table((new WorkflowHistoryEvent())->getTable())->insert([ + 'id' => (string) \Illuminate\Support\Str::ulid(), + 'workflow_run_id' => strtolower($id), + 'sequence' => 3, + 'event_type' => 'VersionMarkerRecorded', + 'payload' => json_encode($this->marker(1, 2), JSON_THROW_ON_ERROR), + 'recorded_at' => now(), + ]); + } + $page = (new VersionUsageQuery())->page('orders', 'shipment', 'shipping', 2); + $this->assertSame([ + 'total' => 2, + 'legacy' => 0, + 'current' => 1, + 'unknown' => 1, + ], $page['counts']); + $this->assertSame(['Case-A', 'Gap-A'], array_column($page['runs'], 'run_id')); + $this->assertSame('current', $page['runs'][0]['state']); + $this->assertSame('history_unavailable', $page['runs'][1]['unknown_reason']); + } + + public function testValidMarkerEvidenceIsCompleteAcrossMarkerReadBatches(): void + { + $run = $this->createRun(1, null); + $events = []; + for ($sequence = 1; $sequence <= 251; ++$sequence) { + $events[] = [ + 'id' => (string) \Illuminate\Support\Str::ulid(), + 'workflow_run_id' => $run->id, + 'sequence' => $sequence + 1, + 'event_type' => 'VersionMarkerRecorded', + 'payload' => json_encode($this->marker(2, $sequence), JSON_THROW_ON_ERROR), + 'recorded_at' => now(), + ]; + } + DB::table((new WorkflowHistoryEvent())->getTable())->insert($events); + $run->update([ + 'last_history_sequence' => 252, + ]); + $page = (new VersionUsageQuery())->page('orders', 'shipment', 'shipping', 2); + $this->assertSame([ + 'total' => 1, + 'legacy' => 0, + 'current' => 1, + 'unknown' => 0, + ], $page['counts']); + } + public function testPagingHasAStableUpperBoundAndCannotCrossTenantOrFilterScope(): void { $one = $this->createRun(1, 1); diff --git a/tests/Unit/V2/VersionUsageDecisionTest.php b/tests/Unit/V2/VersionUsageDecisionTest.php index c83578e1c..b0c345a7e 100644 --- a/tests/Unit/V2/VersionUsageDecisionTest.php +++ b/tests/Unit/V2/VersionUsageDecisionTest.php @@ -178,6 +178,58 @@ public function testInspectionBudgetCannotReturnAnOptimisticPartialDecision(): v ], $decision->result()); } + public static function invalidReuse(): iterable + { + yield 'source only' => [[ + 'reused_from_run_id' => 'original-run', + ]]; + yield 'clock only' => [[ + 'reused_recorded_at' => '2026-11-14T01:02:03+00:00', + ]]; + foreach ([[], null, '', ' '] as $index => $source) { + yield 'source value ' . $index => [[ + 'reused_from_run_id' => $source, + 'reused_recorded_at' => '2026-11-14T01:02:03+00:00', + ]]; + } + foreach ([[], null, '', ' ', 'tomorrow', 'not-a-clock', '2026-02-30T01:02:03+00:00', + '2026-11-14T25:02:03+00:00', '2026-11-14T01:02:03+25:00', + '2026-11-14T01:02:03.1234567+00:00'] as $index => $clock) { + yield 'clock value ' . $index => [[ + 'reused_from_run_id' => 'original-run', + 'reused_recorded_at' => $clock, + ]]; + } + } + + #[DataProvider('invalidReuse')] + public function testMalformedReuseProvenanceCannotBecomeCurrent(array $reuse): void + { + $decision = new VersionUsageDecision('shipping', 2, true); + $decision->observe($this->marker(2) + $reuse); + $this->assertSame([ + 'state' => 'unknown', + 'recorded_version' => null, + 'unknown_reason' => 'history_invalid', + ], $decision->result()); + } + + public function testValidRetainedClocksKeepTheirOriginalVersionWithoutRewritingStrings(): void + { + foreach (['2026-11-14T01:02:03+00:00', '2026-11-14T01:02:03.123456+00:00', + '2026-11-14T01:02:03.123Z', '2026-11-14T01:02:03Z', '2026-11-14T01:02:03-05:30'] as $clock) { + $decision = new VersionUsageDecision('shipping', 2, true); + $marker = $this->marker(2) + [ + 'reused_from_run_id' => 'original-run', + 'reused_recorded_at' => $clock, + ]; + $original = $marker; + $decision->observe($marker); + $this->assertSame('current', $decision->result()['state']); + $this->assertSame($original, $marker); + } + } + private function marker(int $version): array { return array_replace(self::validMarker(), [ From daa00a65b597f43f1cd4b9eb15c00b3cec2497ed Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 14 Nov 2026 12:54:35 +0000 Subject: [PATCH 09/11] test(versioning): prove retained usage through the guarded corpus consumer --- .../ci/classify-version-usage-observer.php | 139 ------------------ scripts/ci/test-regression-corpus-policy.py | 70 +++------ scripts/ci/validate-regression-corpus.py | 22 --- .../V2EmbeddedReplayRegressionCorpusTest.php | 38 +++++ ...ersion-usage-retained-reused-decision.json | 56 +++++++ 5 files changed, 113 insertions(+), 212 deletions(-) delete mode 100644 scripts/ci/classify-version-usage-observer.php create mode 100644 tests/Fixtures/V2/ReplayRegression/version-usage-retained-reused-decision.json diff --git a/scripts/ci/classify-version-usage-observer.php b/scripts/ci/classify-version-usage-observer.php deleted file mode 100644 index 355361b78..000000000 --- a/scripts/ci/classify-version-usage-observer.php +++ /dev/null @@ -1,139 +0,0 @@ - - ! is_array($token) || ! in_array($token[0], [T_WHITESPACE, T_COMMENT, T_DOC_COMMENT], true))); - $imports = [ - 'DateTimeImmutable', 'Generator', 'Illuminate\\Database\\Query\\Builder', - 'Illuminate\\Database\\Eloquent\\Builder', 'Illuminate\\Support\\Collection', - 'InvalidArgumentException', 'JsonException', 'LogicException', - 'Workflow\\V2\\Enums\\HistoryEventType', 'Workflow\\V2\\Enums\\RunStatus', - 'Workflow\\V2\\Models\\WorkflowHistoryEvent', 'Workflow\\V2\\Models\\WorkflowInstance', - 'Workflow\\V2\\Models\\WorkflowRun', - ]; - $functions = [ - 'config', 'strlen', 'preg_match', 'hash', 'hash_hmac', 'hash_equals', 'json_encode', 'json_decode', - 'str_contains', 'is_string', 'is_int', 'is_array', 'in_array', 'strcmp', 'count', 'sort', 'array_slice', - 'array_key_last', 'array_keys', 'array_key_exists', 'array_diff', 'array_is_list', 'array_fill', - 'implode', 'rtrim', 'strtr', 'base64_encode', 'base64_decode', - ]; - $methods = [ - 'getQuery', 'getConnection', 'getDriverName', 'getGrammar', 'wrap', 'getModel', 'qualifyColumn', - 'setEagerLoads', 'where', 'whereIn', 'whereRaw', 'orderByRaw', 'select', 'selectRaw', 'limit', - 'groupByRaw', 'groupBy', 'toBase', 'get', 'pluck', 'value', 'keyBy', 'distinct', 'lazyById', 'isEmpty', 'last', - 'count', 'all', 'setConnection', 'getConnectionName', 'newQuery', - 'runs', 'idExpression', 'batches', 'rows', 'identities', 'decisions', 'observe', 'result', 'unavailable', - ]; - $static = [ - 'ConfiguredV2Models' => ['resolve'], 'NamespaceIdentity' => ['constrain'], - 'VersionUsageDecision' => ['integer'], 'VersionUsageCursor' => ['encode', 'decode'], - 'DateTimeImmutable' => ['createFromFormat', 'getLastErrors'], - 'self' => ['text', 'binary', 'runId', 'encode', 'signature', 'integer', 'validReuse'], - ]; - if ($class !== 'VersionUsageQuery') { - $methods = []; - $static = ['self' => $static['self']] + ($class === 'VersionUsageDecision' - ? ['DateTimeImmutable' => $static['DateTimeImmutable']] : []); - } - $namespaces = 0; - $classes = 0; - foreach ($tokens as $index => $token) { - $previous = $tokens[$index - 1] ?? null; - $next = $tokens[$index + 1] ?? null; - $text = is_array($token) ? $token[1] : $token; - $previousKind = is_array($previous) ? $previous[0] : null; - if (is_array($token) && in_array($token[0], [ - T_EVAL, T_INCLUDE, T_INCLUDE_ONCE, T_REQUIRE, T_REQUIRE_ONCE, T_EXIT, - T_GLOBAL, T_GOTO, T_PRINT, T_ECHO, T_HALT_COMPILER, T_EXTENDS, T_IMPLEMENTS, - T_INTERFACE, T_TRAIT, T_ENUM, - ], true)) { - return false; - } - if (($class !== 'VersionUsageCursor' && $text === '.') || $text === '`' - || (in_array($text, [')', ']', '}'], true) && $next === '(') - || (is_array($token) && $token[0] === T_VARIABLE && $next === '(' - && $previousKind !== T_NEW)) { - return false; - } - if (is_array($token) && $token[0] === T_NAMESPACE) { - ++$namespaces; - if (! is_array($next) || $next[1] !== 'Workflow\\V2\\Support') { - return false; - } - } - if (is_array($token) && $token[0] === T_CLASS) { - // ::class is metadata, not a class declaration. - if ($previousKind === T_DOUBLE_COLON) { - continue; - } - ++$classes; - if (! is_array($next) || $next[1] !== $class) { - return false; - } - } - if (is_array($token) && $token[0] === T_USE) { - $alias = $tokens[$index + 2] ?? null; - $aliasName = $tokens[$index + 3] ?? null; - $queryAlias = is_array($next) && $next[1] === 'Illuminate\\Database\\Query\\Builder' - && is_array($alias) && $alias[0] === T_AS - && is_array($aliasName) && $aliasName[1] === 'QueryBuilder' - && ($tokens[$index + 4] ?? null) === ';'; - if (! is_array($next) || ! in_array($next[1], $imports, true) - || (($tokens[$index + 2] ?? null) !== ';' && ! $queryAlias)) { - return false; - } - } - if ($previousKind === T_OBJECT_OPERATOR || $previousKind === T_NULLSAFE_OBJECT_OPERATOR) { - if (! is_array($token) || $token[0] !== T_STRING || ! in_array($text, $methods, true)) { - // Reading a field is safe; a variable selector/call is not. - if (! is_array($token) || $token[0] !== T_STRING || $next === '(') { - return false; - } - } - } - if ($previousKind === T_DOUBLE_COLON && $next === '(') { - $owner = $tokens[$index - 2] ?? null; - $owner = is_array($owner) ? $owner[1] : $owner; - if (! in_array($text, $static[$owner] ?? [], true) - && ! ($class === 'VersionUsageQuery' && $owner === '$model' && $text === 'query')) { - return false; - } - } - if (is_array($token) && in_array($token[0], [T_STRING, T_NAME_FULLY_QUALIFIED, T_NAME_QUALIFIED], true) - && $next === '(' && ! in_array($previousKind, [T_FUNCTION, T_NEW, T_OBJECT_OPERATOR, - T_NULLSAFE_OBJECT_OPERATOR, T_DOUBLE_COLON], true) - && ! in_array($text, $functions, true)) { - return false; - } - if (is_array($token) && $token[0] === T_NEW) { - $type = is_array($next) ? $next[1] : null; - $types = ['InvalidArgumentException', 'LogicException']; - if ($class === 'VersionUsageQuery') { - $types = array_merge($types, ['VersionUsageDecision', '$instanceModel', '$eventModel']); - } - if (! in_array($type, $types, true)) { - return false; - } - } - if (is_array($token) && in_array($token[0], [T_CONSTANT_ENCAPSED_STRING, T_ENCAPSED_AND_WHITESPACE], true) - && preg_match('/\b(INSERT|UPDATE|DELETE|REPLACE|ALTER|DROP|CREATE|TRUNCATE|CALL|EXECUTE|INTO|OUTFILE|DUMPFILE|COPY|PRAGMA|ATTACH|DETACH|VACUUM|GRANT|REVOKE)\b/i', $text)) { - return false; - } - } - return $namespaces === 1 && $classes === 1; -} - -try { - $input = json_decode(stream_get_contents(STDIN), true, flags: JSON_THROW_ON_ERROR); - echo json_encode(versionUsageObserver($input['path'], $input['source']), JSON_THROW_ON_ERROR); -} catch (Throwable) { - echo 'false'; -} diff --git a/scripts/ci/test-regression-corpus-policy.py b/scripts/ci/test-regression-corpus-policy.py index 45ee50b05..37dc9acee 100644 --- a/scripts/ci/test-regression-corpus-policy.py +++ b/scripts/ci/test-regression-corpus-policy.py @@ -2141,66 +2141,34 @@ def test_core_replay_surfaces_require_growth_without_diff_keywords(self) -> None ) target.write_text(" None: - for name in ("VersionUsageQuery", "VersionUsageDecision", "VersionUsageCursor"): - source = self.root / f"src/V2/Support/{name}.php" - source.write_text( - "last_history_sequence; }\n}", - encoding="utf-8", - ) - self.git("add", "src/V2/Support") - result = self.validate() - self.assertEqual(0, result.returncode, result.stderr) - - def test_version_usage_names_cannot_hide_writes_replay_or_dynamic_execution(self) -> None: - source = self.root / "src/V2/Support/VersionUsageQuery.php" - for body in ( - "$run->save(); return $run->last_history_sequence;", - "WorkflowHistoryEvent::record($run, $type); return $run->last_history_sequence;", - "WorkflowFiberRunner::forClass('demo'); return $run->last_history_sequence;", - "$function(); return $run->last_history_sequence;", - "$functions[0](); return $run->last_history_sequence;", - "($function)(); return $run->last_history_sequence;", - "$run->$method(); return $run->last_history_sequence;", - "eval('return 1;'); return $run->last_history_sequence;", - "$run->selectRaw('DELETE FROM workflow_runs'); return $run->last_history_sequence;", - "$run->selectRaw('id INTO OUTFILE name'); return $run->last_history_sequence;", - "$run->selectRaw('id IN'.'TO OUTFILE \\'/tmp/version-usage-write\\'')->get(); return $run->last_history_sequence;", - "$run->selectRaw('id IN'.('TO OUTFILE \\'/tmp/version-usage-write\\''))->get(); return $run->last_history_sequence;", - ): - with self.subTest(body=body): + def test_untracked_history_observers_keep_the_existing_replay_guard(self) -> None: + for name in ("VersionUsageQuery", "VersionUsageDecision", "VersionUsageCursor", "AnotherHistoryHelper"): + with self.subTest(name=name): + source = self.root / f"src/V2/Support/{name}.php" source.write_text( "last_history_sequence; }} }}", encoding="utf-8", ) result = self.validate() self.assertNotEqual(0, result.returncode, result.stdout) self.assertIn("replay implementation changed but its corpus did not grow", result.stderr) - source.write_text( - "last_history_sequence; } }", - encoding="utf-8", - ) - result = self.validate() - self.assertNotEqual(0, result.returncode, result.stdout) - self.assertIn("replay implementation changed but its corpus did not grow", result.stderr) + source.unlink() - def test_version_usage_read_exception_does_not_exempt_other_support_changes(self) -> None: + def test_replacing_observation_with_transformed_raw_sql_still_requires_evidence(self) -> None: source = self.root / "src/V2/Support/VersionUsageQuery.php" - source.write_text( - "last_history_sequence; } }", - encoding="utf-8", - ) - companion = self.root / "src/V2/Support/AnotherHistoryHelper.php" - companion.write_text("last_history_sequence + 1;", encoding="utf-8") - result = self.validate() - self.assertNotEqual(0, result.returncode, result.stdout) - self.assertIn("replay implementation changed but its corpus did not grow", result.stderr) + header = "last_history_sequence; } }", encoding="utf-8") + self.commit_current_as_base() + for body in ( + "$sql = base64_decode('aWQgSU5UTyBPVVRGSUxFICcvdG1wL3ZlcnNpb24tdXNhZ2Utd3JpdGUn'); return $run->selectRaw($sql)->get();", + "$sql = implode('', ['id IN', 'TO OUT', 'FILE name']); return $run->selectRaw($sql)->get();", + ): + with self.subTest(body=body): + source.write_text(header + body + " } }", encoding="utf-8") + result = self.validate() + self.assertNotEqual(0, result.returncode, result.stdout) + self.assertIn("replay implementation changed but its corpus did not grow", result.stderr) def test_task_discovery_helper_body_does_not_require_replay_growth(self) -> None: source = self.root / "src/V2/Support/DefaultWorkflowTaskBridge.php" diff --git a/scripts/ci/validate-regression-corpus.py b/scripts/ci/validate-regression-corpus.py index 80212618a..3aab48d6f 100644 --- a/scripts/ci/validate-regression-corpus.py +++ b/scripts/ci/validate-regression-corpus.py @@ -2056,28 +2056,6 @@ def _guard_matches( ) if classification.returncode == 0 and classification.stdout == "true": return False - if guard["glob"] == "src/V2/Support/*.php": - # Metadata observers do not reconstruct workflow commands. Closed call - # classification keeps writes and replay dependencies under this guard. - observers = { - "src/V2/Support/VersionUsageQuery.php", - "src/V2/Support/VersionUsageDecision.php", - "src/V2/Support/VersionUsageCursor.php", - } - for path in list(matching): - if path not in observers: - continue - classification = subprocess.run( - ["php", str(Path(__file__).with_name("classify-version-usage-observer.php"))], - input=json.dumps({"path": path, "source": (root / path).read_text(encoding="utf-8")}), - capture_output=True, - text=True, - check=False, - ) - if classification.returncode == 0 and classification.stdout == "true": - matching.remove(path) - if not matching: - return False patterns = guard.get("content_patterns") if patterns is None: return True diff --git a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php index 492176af9..348dc36d2 100644 --- a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php +++ b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php @@ -102,6 +102,7 @@ use Workflow\V2\Support\ScheduleManager; use Workflow\V2\Support\ServiceCallPrincipal; use Workflow\V2\Support\ServiceCallScheduler; +use Workflow\V2\Support\VersionUsageQuery; use Workflow\V2\Support\WorkflowCommandNormalizer; use Workflow\V2\Support\WorkflowDefinition; use Workflow\V2\Support\WorkflowDefinitionFingerprint; @@ -123,6 +124,43 @@ final class V2EmbeddedReplayRegressionCorpusTest extends TestCase private int $workflowNumber = 0; + public function testVersionUsageCorpusObservesRetainedDecisionsWithoutExecutingWork(): void + { + $path = self::FIXTURE_DIR . '/version-usage-retained-reused-decision.json'; + if (! is_file($path)) { + $this->markTestSkipped('The retained version-usage evidence is absent.'); + } + $this->assertTrue( + class_exists(VersionUsageQuery::class), + 'The retained version-usage reader capability is unavailable.' + ); + $fixture = json_decode(file_get_contents($path), true, flags: JSON_THROW_ON_ERROR); + self::stopWorkers(); + Queue::fake(); + config([ + 'queue.default' => 'database', + 'app.key' => 'version-usage-corpus-signing-key', + ]); + $run = $this->createRunFromFixture($fixture); + $history = $run->historyEvents() + ->get() + ->toArray(); + $attempts = ActivityAttempt::query()->count(); + $query = new VersionUsageQuery(); + foreach ([[1, 'current'], [2, 'legacy']] as [$minimum, $state]) { + $page = $query->page($run->namespace, $run->workflow_type, 'same', $minimum); + $this->assertSame(1, $page['counts']['total']); + $this->assertSame(1, $page['counts'][$state]); + $this->assertSame($state, $page['runs'][0]['state']); + $this->assertSame(1, $page['runs'][0]['recorded_version']); + } + $missing = $query->page($run->namespace, $run->workflow_type, 'unreached-change', 1); + $this->assertSame(1, $missing['counts']['unknown']); + $this->assertSame('marker_absent', $missing['runs'][0]['unknown_reason']); + $this->assertSame($attempts, ActivityAttempt::query()->count()); + $this->assertSame($history, $run->historyEvents()->get()->toArray()); + } + public function testEntityReferenceAttemptCorpusLoadsCurrentStateAndNeverLoadsOnReplay(): void { $path = self::FIXTURE_DIR . '/entity-reference-attempt-current-state.json'; diff --git a/tests/Fixtures/V2/ReplayRegression/version-usage-retained-reused-decision.json b/tests/Fixtures/V2/ReplayRegression/version-usage-retained-reused-decision.json new file mode 100644 index 000000000..be8497be0 --- /dev/null +++ b/tests/Fixtures/V2/ReplayRegression/version-usage-retained-reused-decision.json @@ -0,0 +1,56 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "version-usage-retained-reused-decision", + "protocol_version": "1.20", + "bindings": [ + "php" + ], + "consumers": [ + "workflow-fiber-runner", + "query-state-replayer", + "workflow-executor" + ], + "workflow": { + "type": "Tests\\Fixtures\\V2\\TestSinglePatchReplayWorkflow", + "arguments": [], + "payload_codec": "avro" + }, + "history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "payload": {}, + "recorded_at": "2026-11-14T01:00:00.123456+00:00" + }, + { + "sequence": 2, + "event_type": "VersionMarkerRecorded", + "payload": { + "sequence": 1, + "change_id": "same", + "version": 1, + "min_supported": 0, + "max_supported": 3, + "reused_from_run_id": "retained-source-run", + "reused_recorded_at": "2026-11-13T23:59:59.654321+00:00" + }, + "recorded_at": "2026-11-14T01:00:01.123456+00:00" + }, + { + "sequence": 3, + "event_type": "ActivityScheduled", + "payload": { + "sequence": 2, + "activity_type": "Tests\\Fixtures\\V2\\TestQueryReplayGuardActivity", + "activity_execution_id": "01j00000000000000000000009" + }, + "recorded_at": "2026-11-14T01:00:02.123456+00:00" + } + ], + "expected": { + "completed": false, + "result": null, + "commands": [] + } +} From 7e1706ef770285d337f7cb91d7fa24109b70a7fc Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 14 Nov 2026 13:06:48 +0000 Subject: [PATCH 10/11] test(versioning): execute every declared retained-history consumer --- .../V2EmbeddedReplayRegressionCorpusTest.php | 78 +++++++++++++------ 1 file changed, 55 insertions(+), 23 deletions(-) diff --git a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php index 348dc36d2..e92c63533 100644 --- a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php +++ b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php @@ -124,16 +124,12 @@ final class V2EmbeddedReplayRegressionCorpusTest extends TestCase private int $workflowNumber = 0; - public function testVersionUsageCorpusObservesRetainedDecisionsWithoutExecutingWork(): void + public function testVersionUsageCorpusAlsoExecutesDeclaredReplayConsumers(): void { $path = self::FIXTURE_DIR . '/version-usage-retained-reused-decision.json'; if (! is_file($path)) { $this->markTestSkipped('The retained version-usage evidence is absent.'); } - $this->assertTrue( - class_exists(VersionUsageQuery::class), - 'The retained version-usage reader capability is unavailable.' - ); $fixture = json_decode(file_get_contents($path), true, flags: JSON_THROW_ON_ERROR); self::stopWorkers(); Queue::fake(); @@ -141,24 +137,7 @@ class_exists(VersionUsageQuery::class), 'queue.default' => 'database', 'app.key' => 'version-usage-corpus-signing-key', ]); - $run = $this->createRunFromFixture($fixture); - $history = $run->historyEvents() - ->get() - ->toArray(); - $attempts = ActivityAttempt::query()->count(); - $query = new VersionUsageQuery(); - foreach ([[1, 'current'], [2, 'legacy']] as [$minimum, $state]) { - $page = $query->page($run->namespace, $run->workflow_type, 'same', $minimum); - $this->assertSame(1, $page['counts']['total']); - $this->assertSame(1, $page['counts'][$state]); - $this->assertSame($state, $page['runs'][0]['state']); - $this->assertSame(1, $page['runs'][0]['recorded_version']); - } - $missing = $query->page($run->namespace, $run->workflow_type, 'unreached-change', 1); - $this->assertSame(1, $missing['counts']['unknown']); - $this->assertSame('marker_absent', $missing['runs'][0]['unknown_reason']); - $this->assertSame($attempts, ActivityAttempt::query()->count()); - $this->assertSame($history, $run->historyEvents()->get()->toArray()); + $this->assertVersionUsageRetainedDecision($fixture); } public function testEntityReferenceAttemptCorpusLoadsCurrentStateAndNeverLoadsOnReplay(): void @@ -2510,6 +2489,12 @@ public function testFixturesExecuteThroughDeclaredReplayConsumers(): void foreach ($this->fixtures() as $fixture) { $this->executeColdReplayFixture($fixture); + if ($fixture['id'] === 'version-usage-retained-reused-decision') { + config()->set('app.key', 'version-usage-corpus-signing-key'); + $this->assertVersionUsageRetainedDecision($fixture); + continue; + } + if ($fixture['id'] === 'cooperative-cleanup-terminal-notification') { // Its live cleanup, commit and notification proof runs in the dedicated corpus test. continue; @@ -3050,6 +3035,53 @@ public function testSelectionLoserCorpusIsSettledWhenTheRunCloses(): void } } + /** + * @param array $fixture + */ + private function assertVersionUsageRetainedDecision(array $fixture): void + { + $this->assertTrue( + class_exists(VersionUsageQuery::class), + 'The retained version-usage reader capability is unavailable.' + ); + $this->clearWorkflowState(); + $run = $this->createRunFromFixture($fixture); + $history = $run->historyEvents() + ->get() + ->toArray(); + $attempts = ActivityAttempt::query()->count(); + $query = new VersionUsageQuery(); + foreach ([[1, 'current'], [2, 'legacy']] as [$minimum, $state]) { + $page = $query->page($run->namespace, $run->workflow_type, 'same', $minimum); + $this->assertSame(1, $page['counts']['total']); + $this->assertSame(1, $page['counts'][$state]); + $this->assertSame($state, $page['runs'][0]['state']); + $this->assertSame(1, $page['runs'][0]['recorded_version']); + } + $missing = $query->page($run->namespace, $run->workflow_type, 'unreached-change', 1); + $this->assertSame(1, $missing['counts']['unknown']); + $this->assertSame('marker_absent', $missing['runs'][0]['unknown_reason']); + $this->assertSame($attempts, ActivityAttempt::query()->count()); + $this->assertSame($history, $run->historyEvents()->get()->toArray()); + + $state = (new QueryStateReplayer())->replayState($run->fresh()); + $this->assertInstanceOf(TestSinglePatchReplayWorkflow::class, $state->workflow); + $this->assertSame([true, null], $state->workflow->currentState()); + $this->assertSame(2, $state->sequence); + $this->assertNotNull($state->current); + $this->assertSame($history, $run->historyEvents()->get()->toArray()); + TestQueryReplayGuardActivity::$executions = 0; + $this->runReadyWorkflowTask($run); + $this->assertSame(RunStatus::Waiting, $run->fresh()->status); + $this->assertSame(0, TestQueryReplayGuardActivity::$executions); + $this->assertSame( + 1, + $run->historyEvents()->where('event_type', HistoryEventType::VersionMarkerRecorded)->count() + ); + $this->assertSame(1, $run->historyEvents()->where('event_type', HistoryEventType::ActivityScheduled)->count()); + $this->assertSame($history, array_slice($run->historyEvents()->get()->toArray(), 0, count($history))); + } + private function runReadyTaskOfType(string $runId, TaskType $taskType): void { $task = WorkflowTask::query() From 04d863046eb2f9e7e2d9cee006124fb728254d77 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 14 Nov 2026 13:20:08 +0000 Subject: [PATCH 11/11] style: align version usage corpus method chain --- tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php index e92c63533..023486327 100644 --- a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php +++ b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php @@ -3076,7 +3076,8 @@ class_exists(VersionUsageQuery::class), $this->assertSame(0, TestQueryReplayGuardActivity::$executions); $this->assertSame( 1, - $run->historyEvents()->where('event_type', HistoryEventType::VersionMarkerRecorded)->count() + $run->historyEvents() + ->where('event_type', HistoryEventType::VersionMarkerRecorded)->count() ); $this->assertSame(1, $run->historyEvents()->where('event_type', HistoryEventType::ActivityScheduled)->count()); $this->assertSame($history, array_slice($run->historyEvents()->get()->toArray(), 0, count($history)));