diff --git a/app/Http/Controllers/Api/MessageStreamController.php b/app/Http/Controllers/Api/MessageStreamController.php index 003ad47d1..3233a8d71 100644 --- a/app/Http/Controllers/Api/MessageStreamController.php +++ b/app/Http/Controllers/Api/MessageStreamController.php @@ -4,6 +4,7 @@ use App\Support\AvroPayloadEnvelopeResolver; use App\Support\ControlPlaneProtocol; +use App\Support\LegacyV1Projection; use App\Support\MessageStreamService; use App\Support\NamespaceExternalPayloadStorage; use App\Support\NamespaceWorkflowScope; @@ -44,6 +45,15 @@ public function append(Request $request, string $workflowId, string $streamName) ], 404); } + $run = NamespaceWorkflowScope::currentRun($namespace, $workflowId); + if (LegacyV1Projection::isProjectedRun($run)) { + return ControlPlaneProtocol::jsonForRequest( + $request, + LegacyV1Projection::readOnlyMessageAppend($run, $streamName), + 409, + ); + } + $validator = Validator::make($request->all(), [ 'message_id' => ['required', 'string', 'max:191', 'regex:/^[A-Za-z0-9._:-]+$/'], 'input' => ['nullable', 'array'], diff --git a/app/Http/Controllers/Api/WorkflowController.php b/app/Http/Controllers/Api/WorkflowController.php index 9511c8ab2..1848156b7 100644 --- a/app/Http/Controllers/Api/WorkflowController.php +++ b/app/Http/Controllers/Api/WorkflowController.php @@ -1319,18 +1319,7 @@ private function rejectProjectedV1Operation( private function projectedV1OperationResponse(Request $request, WorkflowRun $run, string $operation): JsonResponse { - return ControlPlaneProtocol::jsonForRequest($request, [ - 'message' => sprintf( - 'Operation [%s] is unavailable because this execution is a read-only projection owned by the source v1 runtime.', - $operation, - ), - 'reason' => 'v1_projection_read_only', - 'remediation' => 'Run commands against the source v1 application until it completes; repeat projection never dispatches or advances standalone execution.', - 'workflow_id' => $run->workflow_instance_id, - 'run_id' => $run->id, - 'operation' => $operation, - 'execution_owner' => 'v1', - ], 409); + return ControlPlaneProtocol::jsonForRequest($request, LegacyV1Projection::readOnlyOperation($run, $operation), 409); } /** diff --git a/app/Support/LegacyV1Projection.php b/app/Support/LegacyV1Projection.php index f1af98fce..549ac869f 100644 --- a/app/Support/LegacyV1Projection.php +++ b/app/Support/LegacyV1Projection.php @@ -10,6 +10,8 @@ use Workflow\V2\Models\WorkflowInstance; use Workflow\V2\Models\WorkflowRun; use Workflow\V2\Models\WorkflowRunSummary; +use Workflow\V2\Support\HistoryBudget; +use Workflow\V2\Support\StructuralLimits; final class LegacyV1Projection { @@ -18,7 +20,7 @@ final class LegacyV1Projection private const OPAQUE_PAYLOAD_REMEDIATION = 'Treat the preserved value as opaque; export decoded JSON from the source application if portable payload access is required.'; /** - * @param array $source + * @param array $source * @return array */ public static function import(string $sourceId, array $source, ?string $namespace, bool $dryRun = false): array @@ -112,10 +114,7 @@ public static function import(string $sourceId, array $source, ?string $namespac $events = $normalized['events']; $status = $normalized['status']; $closedAt = $normalized['closed_at']; - $historySize = array_sum(array_map( - static fn (array $event): int => strlen((string) json_encode($event, JSON_UNESCAPED_SLASHES)), - $events, - )); + $historySize = $normalized['history_size_bytes']; WorkflowInstance::query()->create([ 'id' => $workflowId, @@ -233,6 +232,35 @@ public static function isProjectedRun(?WorkflowRun $run): bool && $run->import_source === LegacyV1ProjectionContract::IMPORT_SOURCE; } + /** @return array */ + public static function readOnlyOperation(WorkflowRun $run, string $operation): array + { + return [ + 'message' => sprintf( + 'Operation [%s] is unavailable because this execution is a read-only projection owned by the source v1 runtime.', + $operation, + ), + 'reason' => 'v1_projection_read_only', + 'remediation' => 'Run commands against the source v1 application until it completes; repeat projection never dispatches or advances standalone execution.', + 'workflow_id' => $run->workflow_instance_id, + 'run_id' => $run->id, + 'operation' => $operation, + 'execution_owner' => 'v1', + ]; + } + + /** @return array */ + public static function readOnlyMessageAppend(WorkflowRun $run, string $streamName): array + { + return [ + ...self::readOnlyOperation($run, 'message_stream_append'), + 'accepted' => false, + 'duplicate' => false, + 'stream_name' => $streamName, + 'outcome' => 'rejected', + ]; + } + /** * @return array|null */ @@ -274,7 +302,7 @@ public static function metadataForRun(WorkflowRun $run): ?array * Add provenance to a history bundle and refresh its checksum so the * exported document remains self-consistent. * - * @param array $bundle + * @param array $bundle * @return array */ public static function decorateHistoryExport(array $bundle, WorkflowRun $run): array @@ -333,7 +361,7 @@ public static function mappedIdentity(string $sourceId, string $qualifiedId): ar } /** - * @param array $source + * @param array $source * @return array */ private static function normalize(string $sourceId, array $source, ?string $namespace): array @@ -375,6 +403,26 @@ private static function normalize(string $sourceId, array $source, ?string $name ]; } + $terminal = in_array($status, ['completed', 'failed', 'cancelled', 'terminated'], true); + $eventCount = 1 + (int) $terminal; + foreach (['logs', 'signals', 'exceptions'] as $collection) { + if (! is_array($source[$collection] ?? null)) { + continue; + } + foreach ($source[$collection] as $row) { + $eventCount += (int) is_array($row); + } + } + $eventLimit = StructuralLimits::historyTransactionSizeLimit(); + if ($eventLimit > 0 && $eventCount > $eventLimit) { + return [ + 'valid' => false, + 'namespace' => $namespace, + 'errors' => [self::limitError('history_transaction_size', $eventCount, $eventLimit)], + 'unsupported_fields' => [], + ]; + } + $mapped = self::mappedIdentity($sourceId, $qualifiedId); $source['class'] = $class; $source['connection'] = self::string($source['connection'] ?? null); @@ -410,7 +458,6 @@ private static function normalize(string $sourceId, array $source, ?string $name ]; $startedAt = self::timestamp($source['created_at'] ?? null); $lastProgressAt = self::timestamp($source['updated_at'] ?? null) ?? $startedAt; - $terminal = in_array($status, ['completed', 'failed', 'cancelled', 'terminated'], true); $closedAt = $terminal ? $lastProgressAt : null; $normalizedForHash = [ 'origin' => $origin, @@ -427,13 +474,22 @@ private static function normalize(string $sourceId, array $source, ?string $name 'created_at' => self::timestampString($startedAt), 'updated_at' => self::timestampString($lastProgressAt), ]; - $dedupeKey = hash('sha256', json_encode(self::canonicalize($normalizedForHash), JSON_UNESCAPED_SLASHES | JSON_THROW_ON_ERROR)); $migration = [ 'origin' => $origin, 'identity' => $identity, 'task_queue_context' => $taskQueue, 'unsupported_fields' => $unsupported, ]; + $built = self::events($source, $migration, $status, $startedAt, $lastProgressAt, $mapped['run_id']); + if (isset($built['limit_error'])) { + return [ + 'valid' => false, + 'namespace' => $namespace, + 'errors' => [$built['limit_error']], + 'unsupported_fields' => [], + ]; + } + $dedupeKey = hash('sha256', json_encode(self::canonicalize($normalizedForHash), JSON_UNESCAPED_SLASHES | JSON_THROW_ON_ERROR)); return [ 'valid' => true, @@ -451,14 +507,15 @@ private static function normalize(string $sourceId, array $source, ?string $name 'import_id' => $mapped['import_id'], 'dedupe_key' => $dedupeKey, 'unsupported_fields' => $unsupported, - 'events' => self::events($source, $migration, $status, $startedAt, $lastProgressAt, $mapped['run_id']), + 'events' => $built['events'], + 'history_size_bytes' => $built['history_size_bytes'], ]; } /** - * @param array $source - * @param array $migration - * @return list> + * @param array $source + * @param array $migration + * @return array{events: list>, history_size_bytes: int}|array{limit_error: array} */ private static function events(array $source, array $migration, string $status, ?Carbon $startedAt, ?Carbon $lastProgressAt, string $runId): array { @@ -467,7 +524,7 @@ private static function events(array $source, array $migration, string $status, 'recorded_at' => $startedAt, 'boundary' => 0, 'order' => 0, - 'payload' => [ + 'build_payload' => static fn (): array => [ 'migration_projection' => $migration, 'legacy_payloads' => [ 'arguments' => self::opaquePayload($source['arguments'] ?? null), @@ -481,7 +538,7 @@ private static function events(array $source, array $migration, string $status, 'recorded_at' => self::timestamp($log['created_at'] ?? $log['now'] ?? null), 'boundary' => 1, 'order' => 100 + $index, - 'payload' => [ + 'build_payload' => static fn (): array => [ 'migration_projection' => [ 'source_record' => 'workflow_log', 'legacy_id' => $log['id'] ?? null, @@ -502,7 +559,7 @@ private static function events(array $source, array $migration, string $status, 'recorded_at' => self::timestamp($signal['created_at'] ?? $signal['received_at'] ?? null), 'boundary' => 1, 'order' => 200 + $index, - 'payload' => [ + 'build_payload' => static fn (): array => [ 'migration_projection' => [ 'source_record' => 'workflow_signal', 'legacy_id' => $signal['id'] ?? null, @@ -519,7 +576,7 @@ private static function events(array $source, array $migration, string $status, 'recorded_at' => self::timestamp($exception['created_at'] ?? null), 'boundary' => 1, 'order' => 300 + $index, - 'payload' => [ + 'build_payload' => static fn (): array => [ 'migration_projection' => [ 'source_record' => 'workflow_exception', 'legacy_id' => $exception['id'] ?? null, @@ -545,7 +602,7 @@ private static function events(array $source, array $migration, string $status, 'recorded_at' => $lastProgressAt, 'boundary' => 2, 'order' => PHP_INT_MAX, - 'payload' => [ + 'build_payload' => static fn (): array => [ 'migration_projection' => [ 'source_status' => $source['status'] ?? null, 'execution_owner' => 'v1', @@ -564,19 +621,42 @@ private static function events(array $source, array $migration, string $status, return [$left['boundary'], $leftTime, $left['order']] <=> [$right['boundary'], $rightTime, $right['order']]; }); + $historySize = 0; + $byteLimit = StructuralLimits::historyTransactionBytesLimit(); foreach ($events as $index => &$event) { + $buildPayload = $event['build_payload']; + unset($event['boundary'], $event['order'], $event['build_payload']); + $event['payload'] = $buildPayload(); $sequence = $index + 1; $event['sequence'] = $sequence; $event['id'] = substr(hash('sha256', $runId."\0".$sequence."\0".$event['event_type']), 0, 26); - unset($event['boundary'], $event['order']); + $historySize += HistoryBudget::sizeOfEvent(new WorkflowHistoryEvent($event)); + if ($byteLimit > 0 && $historySize > $byteLimit) { + return ['limit_error' => self::limitError('history_transaction_bytes', $historySize, $byteLimit)]; + } } unset($event); - return $events; + return ['events' => $events, 'history_size_bytes' => $historySize]; + } + + /** @return array */ + private static function limitError(string $dimension, int $used, int $limit): array + { + [$measure, $setting] = $dimension === 'history_transaction_size' + ? ['history events', 'DW_V2_LIMIT_HISTORY_TRANSACTION_SIZE'] + : ['history bytes', 'DW_V2_LIMIT_HISTORY_TRANSACTION_BYTES']; + + return [ + 'reason' => 'v1_projection_too_large', + 'message' => "The projection would write {$used} {$measure} in one transaction; at most {$limit} are allowed ({$dimension}).", + 'remediation' => "Keep the workflow on its v1 runtime, or raise {$setting} on this server before importing it.", + 'limit' => compact('dimension', 'used', 'limit'), + ]; } /** - * @param array $source + * @param array $source * @return list */ private static function unsupportedFields(array $source, string $status): array @@ -652,8 +732,8 @@ private static function opaquePayload(mixed $value): array } /** - * @param array $normalized - * @param array $extra + * @param array $normalized + * @param array $extra * @return array */ private static function report(array $normalized, string $status, array $extra = []): array @@ -661,7 +741,7 @@ private static function report(array $normalized, string $status, array $extra = $errors = is_array($normalized['errors'] ?? null) ? array_values($normalized['errors']) : []; $rejection = []; if ($status === 'rejected' && is_array($errors[0] ?? null)) { - $rejection = array_intersect_key($errors[0], array_flip(['reason', 'message', 'remediation'])); + $rejection = array_intersect_key($errors[0], array_flip(['reason', 'message', 'remediation', 'limit'])); } return [ diff --git a/app/Support/MessageStreamService.php b/app/Support/MessageStreamService.php index a714e1169..d58068033 100644 --- a/app/Support/MessageStreamService.php +++ b/app/Support/MessageStreamService.php @@ -56,6 +56,9 @@ public function append( } $run = NamespaceWorkflowScope::currentRun($namespace, $workflowId); + if (LegacyV1Projection::isProjectedRun($run)) { + return ['status' => 409, ...LegacyV1Projection::readOnlyMessageAppend($run, $streamName)]; + } if (! $run instanceof WorkflowRun || $run->status->isTerminal()) { return $this->rejection('run_not_active', 409, $workflowId, $streamName, $messageId); } @@ -168,7 +171,7 @@ public function recordMalformed( } $run = NamespaceWorkflowScope::currentRun($namespace, $workflowId); - if (! $run instanceof WorkflowRun || $run->status->isTerminal()) { + if (! $run instanceof WorkflowRun || LegacyV1Projection::isProjectedRun($run) || $run->status->isTerminal()) { return; } diff --git a/config/dw-contract.php b/config/dw-contract.php index e1fce5c97..c072304d3 100644 --- a/config/dw-contract.php +++ b/config/dw-contract.php @@ -1013,6 +1013,12 @@ 'since' => '2.0.0', 'legacy' => 'WORKFLOW_V2_LIMIT_HISTORY_TRANSACTION_SIZE', ], + 'DW_V2_LIMIT_HISTORY_TRANSACTION_BYTES' => [ + 'description' => 'Package-level history-transaction byte ceiling, also applied to Waterline v1 projection imports. Zero disables this dimension.', + 'default' => '16777216', + 'since' => '2.11.5', + 'legacy' => 'WORKFLOW_V2_LIMIT_HISTORY_TRANSACTION_BYTES', + ], 'DW_V2_LIMIT_WARNING_THRESHOLD_PERCENT' => [ 'description' => 'Percent of a structural limit at which the package emits an approaching-limit warning.', 'default' => '80', diff --git a/docs/bounded-growth.md b/docs/bounded-growth.md index 45f0c2779..eab1c2a28 100644 --- a/docs/bounded-growth.md +++ b/docs/bounded-growth.md @@ -58,6 +58,28 @@ retention-deleted history. Retention and rebuild maintenance should leave a marked run without recreating its deleted summary. Check `/api/system/health`, `/api/system/metrics` and `/api/system/operator-metrics` after an upgrade. +## Waterline v1 projection imports + +A projection import writes its history in one transaction. It uses +`DW_V2_LIMIT_HISTORY_TRANSACTION_SIZE` (5,000 events by default) and +`DW_V2_LIMIT_HISTORY_TRANSACTION_BYTES` (16 MiB by default). Setting either +dimension to `0` disables that dimension. These limits supplement the HTTP +request payload limit. + +The event count includes the start event, each log, signal and exception row, +and a terminal event when applicable. The byte budget and stored summary use +Core's canonical history metric: the event type's byte length plus its payload's +JSON byte length, with unescaped slashes and Unicode. The server checks the count +before copying or hashing source rows, then builds payloads in history order and stops at the +first event exceeding the byte budget. + +An oversized import returns HTTP 409 with `reason: v1_projection_too_large` +and `limit: {dimension, used, limit}` before reading existing workflow rows or +starting the write transaction. Dry runs enforce the same limits. Keep the +workflow on its v1 runtime or deliberately raise the reported limit before +importing it. Imported projections remain read-only, including message-stream +appends and malformed-input accounting. + ## Cache Inventory | Policy ID | Prefix | Owner | Growth Bound | diff --git a/docs/server-reference.md b/docs/server-reference.md index b2eb8faa6..2fd08cb71 100644 --- a/docs/server-reference.md +++ b/docs/server-reference.md @@ -1930,6 +1930,7 @@ inside the package's `config/workflows.php` via | `DW_V2_LIMIT_MEMO_SIZE_BYTES` | `262144` | Package-level memo byte ceiling. | | `DW_V2_LIMIT_SEARCH_ATTRIBUTE_SIZE_BYTES` | `40960` | Package-level search-attribute byte ceiling. | | `DW_V2_LIMIT_HISTORY_TRANSACTION_SIZE` | `5000` | Package-level history-transaction event ceiling. | +| `DW_V2_LIMIT_HISTORY_TRANSACTION_BYTES` | `16777216` | Package-level history-transaction byte ceiling, also applied to Waterline v1 projection imports. Zero disables this dimension. | | `DW_V2_LIMIT_WARNING_THRESHOLD_PERCENT` | `80` | Percent of a structural limit at which the package warns. | | `DW_V2_TASK_DISPATCH_MODE` | `queue` | Package-level workflow-task dispatch mode. Usually overridden by the server via `DW_TASK_DISPATCH_MODE`. | | `DW_V2_MATCHING_ROLE_QUEUE_WAKE` | `true` | Whether queue workers run the in-worker matching-role wake on every Looping event. In the supplied Compose stacks, `false` also skips the scheduler's unscoped repair pass. Set it to `false` only when a dedicated `php artisan workflow:v2:repair-pass --loop` daemon owns the sweep. | diff --git a/rust/src/config_audit.rs b/rust/src/config_audit.rs index 97085af75..f246b16d5 100644 --- a/rust/src/config_audit.rs +++ b/rust/src/config_audit.rs @@ -556,7 +556,6 @@ const UNLISTED: &[Setting] = &[ /// when the PHP Server takes that release. They are not unknown names, and /// their legacy names warn as the contract's do. pub const NATIVE_HONORED: &[Setting] = &[ - Setting { name: "DW_V2_LIMIT_HISTORY_TRANSACTION_BYTES", legacy: Some("WORKFLOW_V2_LIMIT_HISTORY_TRANSACTION_BYTES"), disposition: Honored }, Setting { name: "DW_V2_LIMIT_HISTORY_TRANSACTION_ROWS", legacy: Some("WORKFLOW_V2_LIMIT_HISTORY_TRANSACTION_ROWS"), disposition: Honored }, ]; @@ -682,6 +681,7 @@ pub const SETTINGS: &[Setting] = &[ Setting { name: "DW_V2_HISTORY_EXPORT_SIGNING_KEY", legacy: Some("WORKFLOW_V2_HISTORY_EXPORT_SIGNING_KEY"), disposition: Honored }, Setting { name: "DW_V2_HISTORY_EXPORT_SIGNING_KEY_ID", legacy: Some("WORKFLOW_V2_HISTORY_EXPORT_SIGNING_KEY_ID"), disposition: Honored }, Setting { name: "DW_V2_LIMIT_COMMAND_BATCH_SIZE", legacy: Some("WORKFLOW_V2_LIMIT_COMMAND_BATCH_SIZE"), disposition: Refused(Int(1000), STRUCTURAL) }, + Setting { name: "DW_V2_LIMIT_HISTORY_TRANSACTION_BYTES", legacy: Some("WORKFLOW_V2_LIMIT_HISTORY_TRANSACTION_BYTES"), disposition: Honored }, Setting { name: "DW_V2_LIMIT_HISTORY_TRANSACTION_SIZE", legacy: Some("WORKFLOW_V2_LIMIT_HISTORY_TRANSACTION_SIZE"), disposition: Honored }, Setting { name: "DW_V2_LIMIT_MEMO_SIZE_BYTES", legacy: Some("WORKFLOW_V2_LIMIT_MEMO_SIZE_BYTES"), disposition: Honored }, Setting { name: "DW_V2_LIMIT_PAYLOAD_SIZE_BYTES", legacy: Some("WORKFLOW_V2_LIMIT_PAYLOAD_SIZE_BYTES"), disposition: Honored }, @@ -766,7 +766,7 @@ mod tests { #[test] fn every_contract_setting_has_one_disposition() { let contract = contract(); - assert_eq!(contract.len(), 164, "PHP's contract changed; classify the new settings"); + assert_eq!(contract.len(), 165, "PHP's contract changed; classify the new settings"); let mut names: Vec<&str> = SETTINGS.iter().map(|setting| setting.name).collect(); names.sort_unstable(); names.dedup(); diff --git a/tests/Feature/LegacyV1ProjectionBudgetTest.php b/tests/Feature/LegacyV1ProjectionBudgetTest.php new file mode 100644 index 000000000..49c740354 --- /dev/null +++ b/tests/Feature/LegacyV1ProjectionBudgetTest.php @@ -0,0 +1,197 @@ +assertSame(5000, StructuralLimits::historyTransactionSizeLimit()); + $this->assertSame(16 * 1024 * 1024, StructuralLimits::historyTransactionBytesLimit()); + } + + public function test_the_exact_event_limit_is_admitted_and_one_over_is_refused_before_queries(): void + { + $this->createNamespace('default'); + $source = $this->source(); + $source['logs'] = [[], 'not a row', null]; + $source['signals'] = [[]]; + $source['exceptions'] = [[]]; + config(['workflows.v2.structural_limits.history_transaction_size' => 4]); + + foreach ([false, true] as $dryRun) { + $this->assertQueryFreeRefusal($source, 'history_transaction_size', 5, 4, $dryRun); + } + $this->assertDatabaseCount('workflow_runs', 0); + $this->assertDatabaseCount('workflow_history_events', 0); + + config(['workflows.v2.structural_limits.history_transaction_size' => 5]); + $report = LegacyV1Projection::import('budget-test', $source, 'default'); + $this->assertSame('projected', $report['status']); + $this->assertDatabaseCount('workflow_history_events', 5); + $this->assertDatabaseCount('workflow_tasks', 0); + } + + public function test_count_refusal_precedes_source_serialization_and_event_building(): void + { + $source = $this->source(); + $source['arguments'] = $this->unserializable(); + $source['logs'] = array_fill(0, 10, []); + config(['workflows.v2.structural_limits.history_transaction_size' => 2]); + + $this->assertQueryFreeRefusal($source, 'history_transaction_size', 12, 2); + } + + public function test_the_exact_encoded_byte_limit_is_admitted_and_one_byte_over_is_refused(): void + { + $this->createNamespace('default'); + $source = $this->source(); + $source['arguments'] = "Unicode: 雪😀\u{2028}\u{2029}, slash: /, quote: \""; + $source['logs'] = [['result' => 'a:1:{s:5:"value";i:42;}', 'created_at' => '2026-07-10T10:01:00+03:00']]; + config(['workflows.v2.structural_limits.history_transaction_bytes' => 0]); + DB::beginTransaction(); + $initial = LegacyV1Projection::import('budget-test', $source, 'default'); + $runId = $initial['identity']['standalone']['run_id']; + $bytes = WorkflowHistoryEvent::query()->where('workflow_run_id', $runId)->get()->sum( + static fn (WorkflowHistoryEvent $event): int => strlen($event->event_type->value) + + strlen(json_encode($event->payload, JSON_THROW_ON_ERROR | JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE)), + ); + $summaryBytes = (int) WorkflowRunSummary::query()->findOrFail($runId)->history_size_bytes; + DB::rollBack(); + $this->assertSame($bytes, $summaryBytes); + + config(['workflows.v2.structural_limits.history_transaction_bytes' => $bytes - 1]); + $this->assertQueryFreeRefusal($source, 'history_transaction_bytes', $bytes, $bytes - 1); + $this->assertDatabaseCount('workflow_history_events', 0); + + config(['workflows.v2.structural_limits.history_transaction_bytes' => $bytes]); + $exact = LegacyV1Projection::import('budget-test', $source, 'default'); + $this->assertSame('projected', $exact['status']); + $this->assertSame($bytes, (int) WorkflowRunSummary::query()->findOrFail($exact['identity']['standalone']['run_id'])->history_size_bytes); + $this->assertDatabaseCount('workflow_tasks', 0); + } + + public function test_an_early_byte_refusal_does_not_serialize_later_source_payloads(): void + { + $source = $this->source(); + $source['arguments'] = str_repeat('x', 1000); + $source['logs'] = [['result' => $this->unserializable()]]; + config(['workflows.v2.structural_limits.history_transaction_bytes' => 1]); + + $queries = []; + DB::listen(static function (QueryExecuted $query) use (&$queries): void { + $queries[] = $query->sql; + }); + $report = LegacyV1Projection::import('budget-test', $source, 'default'); + $this->assertSame('rejected', $report['status']); + $this->assertSame('v1_projection_too_large', $report['reason']); + $this->assertSame('history_transaction_bytes', $report['limit']['dimension']); + $this->assertGreaterThan(1000, $report['limit']['used']); + $this->assertSame([], $queries); + } + + public function test_zero_disables_both_dimensions(): void + { + $this->createNamespace('default'); + config([ + 'workflows.v2.structural_limits.history_transaction_size' => 0, + 'workflows.v2.structural_limits.history_transaction_bytes' => 0, + ]); + $source = $this->source(); + $source['logs'] = array_fill(0, 10, ['result' => str_repeat('x', 1000)]); + + $report = LegacyV1Projection::import('budget-test', $source, 'default'); + $this->assertSame('projected', $report['status']); + $this->assertSame(12, WorkflowHistoryEvent::query()->count()); + } + + public function test_disabling_one_dimension_preserves_the_other_limit(): void + { + config([ + 'workflows.v2.structural_limits.history_transaction_size' => 0, + 'workflows.v2.structural_limits.history_transaction_bytes' => 1, + ]); + $report = LegacyV1Projection::import('budget-test', $this->source(), 'default'); + $this->assertSame('rejected', $report['status']); + $this->assertSame('history_transaction_bytes', $report['limit']['dimension']); + + config([ + 'workflows.v2.structural_limits.history_transaction_size' => 1, + 'workflows.v2.structural_limits.history_transaction_bytes' => 0, + ]); + $this->assertQueryFreeRefusal($this->source(), 'history_transaction_size', 2, 1); + } + + public function test_the_http_report_exposes_the_rejected_dimension_and_zero_writes(): void + { + $this->createNamespace('default'); + config(['workflows.v2.structural_limits.history_transaction_size' => 1]); + + $this->withHeaders($this->apiHeaders()) + ->postJson('/api/workflows/import/waterline-v1', [ + 'source_id' => 'budget-test', + 'workflow' => $this->source(), + ]) + ->assertConflict() + ->assertJsonPath('status', 'rejected') + ->assertJsonPath('reason', 'v1_projection_too_large') + ->assertJsonPath('limit', ['dimension' => 'history_transaction_size', 'used' => 2, 'limit' => 1]); + + foreach (['workflow_instances', 'workflow_runs', 'workflow_history_events', 'workflow_run_summaries', 'workflow_tasks'] as $table) { + $this->assertDatabaseCount($table, 0); + } + } + + private function assertQueryFreeRefusal(array $source, string $dimension, int $used, int $limit, bool $dryRun = false): void + { + $queries = []; + DB::listen(static function (QueryExecuted $query) use (&$queries): void { + $queries[] = $query->sql; + }); + $report = LegacyV1Projection::import('budget-test', $source, 'default', $dryRun); + $this->assertSame('rejected', $report['status']); + $this->assertSame('v1_projection_too_large', $report['reason']); + $this->assertSame(compact('dimension', 'used', 'limit'), $report['limit']); + $this->assertSame([], $queries); + } + + private function source(): array + { + return [ + 'id' => 'v1:42', + 'engine_source' => 'v1', + 'class' => 'LegacyOrderWorkflow', + 'status' => 'completed', + 'created_at' => '2026-07-10T10:00:00Z', + 'updated_at' => '2026-07-10T10:05:00Z', + ]; + } + + private function unserializable(): JsonSerializable + { + return new class implements JsonSerializable + { + public function jsonSerialize(): mixed + { + throw new LogicException('A later source value must not be serialized after a budget refusal.'); + } + }; + } +} diff --git a/tests/Feature/LegacyV1ProjectionSafetyTest.php b/tests/Feature/LegacyV1ProjectionSafetyTest.php new file mode 100644 index 000000000..57cd38eee --- /dev/null +++ b/tests/Feature/LegacyV1ProjectionSafetyTest.php @@ -0,0 +1,182 @@ + [$state], [ + 'pending', 'running', 'waiting', 'completed', 'failed', 'cancelled', 'terminated', + ]); + } + + #[DataProvider('projectionStates')] + public function test_stream_append_refuses_valid_and_malformed_bodies_without_writes(string $state): void + { + $run = $this->projection($state); + $before = $this->projectionState(); + $writes = []; + DB::listen(static function (QueryExecuted $query) use (&$writes): void { + if (preg_match('/^\s*(insert|update|delete)\b/i', $query->sql)) { + $writes[] = $query->sql; + } + }); + + foreach ([ + ['message_id' => 'message-1', 'input' => []], + [], + ['message_id' => ['invalid']], + ['message_id' => 'message-1', 'input' => 'invalid'], + ['message_id' => 'message-1', 'input' => ['codec' => 'unsupported']], + ] as $body) { + $this->withHeaders($this->apiHeaders()) + ->postJson('/api/workflows/'.rawurlencode($run->workflow_instance_id).'/message-streams/orders/messages', $body) + ->assertConflict() + ->assertJsonPath('reason', 'v1_projection_read_only') + ->assertJsonPath('accepted', false) + ->assertJsonPath('duplicate', false) + ->assertJsonPath('outcome', 'rejected') + ->assertJsonPath('operation', 'message_stream_append') + ->assertJsonPath('stream_name', 'orders') + ->assertJsonPath('run_id', $run->id) + ->assertJsonPath('execution_owner', 'v1'); + $this->assertSame($before, $this->projectionState()); + } + + foreach (['null', '"not an object"'] as $body) { + $this->call( + 'POST', + '/api/workflows/'.rawurlencode($run->workflow_instance_id).'/message-streams/orders/messages', + server: [ + ...$this->transformHeadersToServerVars($this->apiHeaders()), + 'CONTENT_TYPE' => 'application/json', + 'HTTP_ACCEPT' => 'application/json', + ], + content: $body, + )->assertConflict()->assertJsonPath('reason', 'v1_projection_read_only'); + $this->assertSame($before, $this->projectionState()); + } + + $this->call( + 'POST', + '/api/workflows/'.rawurlencode($run->workflow_instance_id).'/message-streams/orders/messages', + server: [ + ...$this->transformHeadersToServerVars($this->apiHeaders()), + 'CONTENT_TYPE' => 'application/json', + 'HTTP_ACCEPT' => 'application/json', + ], + content: '{', + )->assertStatus(400)->assertJsonPath('reason', 'malformed_json'); + $this->assertSame($before, $this->projectionState()); + + $this->assertSame([], $writes); + } + + public function test_a_direct_append_cannot_create_a_stream_or_update_an_existing_duplicate(): void + { + $run = $this->projection('waiting'); + $this->mock(RuntimeSignalControlPlane::class, static function (MockInterface $mock): void { + $mock->shouldNotReceive('runtimeSignal'); + }); + $service = app(MessageStreamService::class); + $payload = 'unchanged-payload'; + $hash = hash('sha256', $payload); + + foreach ([false, true] as $existing) { + if ($existing) { + $stream = WorkflowInboundStream::query()->create([ + 'namespace' => 'default', + 'workflow_instance_id' => $run->workflow_instance_id, + 'stream_name' => 'orders', + 'last_position' => 1, + 'cursor_position' => 0, + ]); + WorkflowInboundStreamItem::query()->create([ + 'stream_id' => $stream->id, + 'namespace' => 'default', + 'workflow_instance_id' => $run->workflow_instance_id, + 'stream_name' => 'orders', + 'message_id' => 'message-1', + 'position' => 1, + 'payload_codec' => 'avro', + 'payload_blob' => $payload, + 'payload_hash' => $hash, + ]); + } + $before = $this->projectionState(); + $result = $service->append('default', $run->workflow_instance_id, 'orders', 'message-1', 'avro', $payload, $hash); + $this->assertSame(409, $result['status']); + $this->assertSame('v1_projection_read_only', $result['reason']); + $this->assertFalse($result['accepted']); + $this->assertSame($before, $this->projectionState()); + + $service->recordMalformed('default', $run->workflow_instance_id, 'orders', 'message-1'); + $this->assertSame($before, $this->projectionState()); + } + } + + public function test_the_projection_fence_preserves_namespace_non_disclosure(): void + { + $run = $this->projection('waiting'); + $this->createNamespace('other'); + $before = $this->projectionState(); + + $this->withHeaders($this->apiHeaders('other')) + ->postJson('/api/workflows/'.rawurlencode($run->workflow_instance_id).'/message-streams/orders/messages', []) + ->assertNotFound() + ->assertJsonPath('reason', 'instance_not_found') + ->assertJsonMissingPath('run_id'); + + $this->assertSame($before, $this->projectionState()); + } + + private function projection(string $status): WorkflowRun + { + $this->createNamespace('default'); + $report = LegacyV1Projection::import('projection-safety', [ + 'id' => 'v1:42', + 'engine_source' => 'v1', + 'class' => 'LegacyOrderWorkflow', + 'status' => $status, + 'created_at' => '2026-07-10T10:00:00Z', + 'updated_at' => '2026-07-10T10:05:00Z', + ], 'default'); + $this->assertSame('projected', $report['status']); + + return WorkflowRun::query()->findOrFail($report['identity']['standalone']['run_id']); + } + + private function projectionState(): array + { + $state = []; + foreach ([ + 'workflow_instances', 'workflow_runs', 'workflow_run_summaries', 'workflow_history_events', + 'workflow_inbound_streams', 'workflow_inbound_stream_items', 'workflow_signal_records', + 'workflow_commands', 'workflow_tasks', + ] as $table) { + $state[$table] = DB::table($table)->orderBy('id')->get()->map(static fn (object $row): array => (array) $row)->all(); + } + + return $state; + } +}