diff --git a/CHANGELOG.md b/CHANGELOG.md index 937a17772..ea1a8f346 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. + - 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 new file mode 100644 index 000000000..3f5e5e07a --- /dev/null +++ b/docs/version-usage.md @@ -0,0 +1,125 @@ +# Open-run version usage (draft v1) + +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. + +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, +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 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. +- `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. 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 +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, 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. + +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. The known +producer-injected `task` snapshot is supported; unknown marker extensions require +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 + +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 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 +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" + } + } + } + } + ] + } + } +} diff --git a/scripts/ci/test-regression-corpus-policy.py b/scripts/ci/test-regression-corpus-policy.py index 132a5a2c3..37dc9acee 100644 --- a/scripts/ci/test-regression-corpus-policy.py +++ b/scripts/ci/test-regression-corpus-policy.py @@ -2141,6 +2141,35 @@ def test_core_replay_surfaces_require_growth_without_diff_keywords(self) -> None ) target.write_text(" 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.unlink() + + def test_replacing_observation_with_transformed_raw_sql_still_requires_evidence(self) -> None: + source = self.root / "src/V2/Support/VersionUsageQuery.php" + 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" original = """ 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, string $key): array + { + 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.'); + } + // 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) { + $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'], $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 new file mode 100644 index 000000000..61ff16cce --- /dev/null +++ b/src/V2/Support/VersionUsageDecision.php @@ -0,0 +1,117 @@ +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 (! 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', + ]) !== [] || (array_key_exists('task', $payload) && (! is_array($payload['task']) + || array_is_list($payload['task'])))) { + $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; + } + + 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 new file mode 100644 index 000000000..0e1a68b9d --- /dev/null +++ b/src/V2/Support/VersionUsageQuery.php @@ -0,0 +1,415 @@ + 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)); + $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); + $idExpression = $this->idExpression($base); + $through = $continuation['through'] ?? (clone $base)->orderByRaw("{$idExpression} 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, $namespace, $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'], + $key + ) : null, + ]; + } + + /** + * @return Generator + */ + 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)->orderByRaw("{$id} ASC"); + if ($after !== null) { + $query->whereRaw("{$id} > {$parameter}", [$after]); + } + $rows = $query->limit($size) + ->get(); + if ($rows->isEmpty()) { + return; + } + yield $rows; + $after = $rows->last() +->id; + } 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 + { + 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(); + $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))); + $status = self::binary($status, $driver); + $query->whereRaw("{$status} IN ({$parameters})", [ + RunStatus::Pending->value, RunStatus::Running->value, RunStatus::Waiting->value, + ]); + return $query; + } + + /** + * @return array + */ + private function decisions( + Collection $runs, + WorkflowRun $runModel, + ?string $namespace, + string $changeId, + int $minVersion + ): array { + $instanceModel = ConfiguredV2Models::resolve('instance_model', WorkflowInstance::class); + $owners = $this->identities(NamespaceIdentity::constrain( + (new $instanceModel())->setConnection($runModel->getConnectionName()) + ->newQuery() + ->setEagerLoads([]), + $namespace, + )->toBase(), 'id', $runs->pluck('workflow_instance_id') + ->all())->pluck('id') + ->all(); + $eventModel = ConfiguredV2Models::resolve('history_event_model', WorkflowHistoryEvent::class); + $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); + $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) + ->pluck('event_type', 'workflow_run_id'); + $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); + $parameters = implode(', ', array_fill(0, count($types), self::binary('?', $driver))); + $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; + $decisions = []; + $inspect = []; + 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 + && (int) $stats->marker_count <= 1000 + && (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) { + $inspect[] = $run->id; + } + } + $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); + } + } + // A prune or append racing this observation cannot turn a partial read into current usage. + $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 + || (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('/(*UCP)\S/u', $value) !== 1) { + throw new InvalidArgumentException('Version-usage identities must be nonempty bounded UTF-8 strings.'); + } + } +} diff --git a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php index 492176af9..023486327 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,22 @@ final class V2EmbeddedReplayRegressionCorpusTest extends TestCase private int $workflowNumber = 0; + 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.'); + } + $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', + ]); + $this->assertVersionUsageRetainedDecision($fixture); + } + public function testEntityReferenceAttemptCorpusLoadsCurrentStateAndNeverLoadsOnReplay(): void { $path = self::FIXTURE_DIR . '/entity-reference-attempt-current-state.json'; @@ -2472,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; @@ -3012,6 +3035,54 @@ 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() diff --git a/tests/Feature/V2/V2VersionUsageQueryTest.php b/tests/Feature/V2/V2VersionUsageQueryTest.php new file mode 100644 index 000000000..95495d038 --- /dev/null +++ b/tests/Feature/V2/V2VersionUsageQueryTest.php @@ -0,0 +1,715 @@ +set('app.key', 'version-usage-feature-signing-key'); + } + + public function testCountsIncludeOldNewAndNoMarkerBeforeStateFilteringOrPagination(): void + { + $this->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 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) { + $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 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); + 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); + $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 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); + $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'); + $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') + ->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); + config() + ->set('workflows.v2.instance_model', VersionUsageCustomInstance::class); + $database->table('instances') + ->insert([ + 'id' => 'custom-instance', + 'namespace' => 'orders', + ]); + $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); + config() + ->set('workflows.v2.instance_model', WorkflowInstance::class); + $schema->dropIfExists('history'); + $schema->dropIfExists('instances'); + $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'; +} + +final class VersionUsageCustomInstance extends WorkflowInstance +{ + protected $table = 'instances'; + + protected $connection = 'version_usage_fixture'; + + public function getForeignKey(): string + { + return 'workflow_instance_id'; + } +} 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": [] + } +} diff --git a/tests/Unit/V2/VersionUsageCursorTest.php b/tests/Unit/V2/VersionUsageCursorTest.php new file mode 100644 index 000000000..019dcc488 --- /dev/null +++ b/tests/Unit/V2/VersionUsageCursorTest.php @@ -0,0 +1,113 @@ +assertSame('run-9', $value['through']); + $this->assertSame('run-3', $value['after']); + $this->expectException(InvalidArgumentException::class); + 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 + { + yield 'empty' => ['']; + yield 'base64 punctuation' => ['%%']; + yield 'oversized' => [str_repeat('A', 2049)]; + yield 'not json' => [self::signed('abc')]; + yield 'scalar' => [self::signed('1')]; + $query = hash('sha256', 'query'); + foreach ([[ + 'version' => 2, + ], [ + 'after' => [], + ], [ + 'through' => '', + ], [ + 'after' => 'run-z', + ], [ + 'extra' => true, + ]] as $mutation) { + 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))]; + } + } + + #[DataProvider('malformed')] + public function testMalformedCursorsAreErrorsRatherThanFirstPages(string $cursor): void + { + $this->expectException(InvalidArgumentException::class); + 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 new file mode 100644 index 000000000..b0c345a7e --- /dev/null +++ b/tests/Unit/V2/VersionUsageDecisionTest.php @@ -0,0 +1,250 @@ + [-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 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); + 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()); + } + + 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(), [ + 'version' => $version, + ]); + } + + private static function validMarker(): array + { + return [ + 'sequence' => 1, + 'change_id' => 'shipping', + 'version' => 2, + 'min_supported' => -2147483648, + 'max_supported' => 2147483647, + ]; + } +}