diff --git a/.github/workflows/server-parity.yml b/.github/workflows/server-parity.yml index 624958ec4..66ed1f1ab 100644 --- a/.github/workflows/server-parity.yml +++ b/.github/workflows/server-parity.yml @@ -46,6 +46,7 @@ on: - tests/Unit/ServerParityActivityCancellationAckContractTest.mjs - tests/Unit/ServerParityServiceCatalogContractTest.mjs - tests/Unit/ServerParityWorkflowImportContractTest.mjs + - tests/Unit/ServerParityWaterlineImportContractTest.mjs - tests/Unit/ServerParitySystemMetricsContractTest.mjs - tests/Unit/ServerParityOperatorDashboardTypesContractTest.mjs - tests/Unit/ServerParityPrometheusMetricsContractTest.mjs @@ -1801,7 +1802,7 @@ jobs: bash scripts/ci/run-composer.sh --rm -v "$PWD:/source" -w /source -- \ install --working-dir=scripts/conformance/server-parity --no-interaction --no-progress docker run --rm --user=1000:1000 --entrypoint node -v "$PWD:/source:ro" -w /source \ - "$image" --test tests/Unit/ServerParityRunnerTest.mjs tests/Unit/ServerParitySdkDirectionLanesTest.mjs tests/Unit/ServerParityCodecDeadlineCellsTest.mjs tests/Unit/ServerParityCooperativeContractTest.mjs tests/Unit/ServerParityChildCancellationContractTest.mjs tests/Unit/ServerParityScheduleContractTest.mjs tests/Unit/ServerParityVisibilityContractTest.mjs tests/Unit/ServerParityAdmissionContractTest.mjs tests/Unit/ServerParityRoleTokenContractTest.mjs tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs tests/Unit/ServerParitySearchAttributeContractTest.mjs tests/Unit/ServerParityWorkerListingContractTest.mjs tests/Unit/ServerParityWorkerReplacementContractTest.mjs tests/Unit/ServerParityPatchDeploymentContractTest.mjs tests/Unit/ServerParityWaitingHistoryContractTest.mjs tests/Unit/ServerParityTaskRetryContractTest.mjs tests/Unit/ServerParityPublishedPatchContractTest.mjs tests/Unit/ServerParityPublishedPatchTransportTest.mjs tests/Unit/ServerParityRecordFileTest.mjs tests/Unit/ServerParityFailureLocationTest.mjs tests/Unit/ServerParityFailureExcerptTest.mjs tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs tests/Unit/ServerParityStorageTestContractTest.mjs tests/Unit/ServerParitySchedulePoliciesContractTest.mjs tests/Unit/ServerParitySignalsBeforeWorkerContractTest.mjs tests/Unit/ServerParityScheduleCatalogContractTest.mjs tests/Unit/ServerParityOperatorMetricsContractTest.mjs tests/Unit/ServerParityActivityLeaseRenewalContractTest.mjs tests/Unit/ServerParityActivityCancellationAckContractTest.mjs tests/Unit/ServerParityServiceCatalogContractTest.mjs tests/Unit/ServerParityWorkflowImportContractTest.mjs tests/Unit/ServerParitySystemMetricsContractTest.mjs tests/Unit/ServerParityOperatorDashboardTypesContractTest.mjs tests/Unit/ServerParityPrometheusMetricsContractTest.mjs tests/Unit/ServerParityPayloadCleanupContractTest.mjs tests/Unit/ServerParityRepairPassContractTest.mjs tests/Unit/ServerParityClusterInfoContractTest.mjs tests/Unit/ServerParityNamespaceDeleteContractTest.mjs tests/Unit/ServerParityMessageStreamsContractTest.mjs tests/Unit/ServerParityWorkflowPollRequestContractTest.mjs tests/Unit/ServerParityUpdateValidationContractTest.mjs tests/Unit/ServerParityFixtureSelectionTest.mjs tests/Unit/ServerParityUpdateValidationRetryTest.mjs tests/Unit/ServerParityKilledAuthorPollTest.mjs tests/Unit/ServerParityImagePullTest.mjs tests/Unit/ServerParityWorkerSessionsContractTest.mjs tests/Unit/ServerParityWorkerSessionActivitiesContractTest.mjs > parity-evidence/comparator-tests.txt + "$image" --test tests/Unit/ServerParityRunnerTest.mjs tests/Unit/ServerParitySdkDirectionLanesTest.mjs tests/Unit/ServerParityCodecDeadlineCellsTest.mjs tests/Unit/ServerParityCooperativeContractTest.mjs tests/Unit/ServerParityChildCancellationContractTest.mjs tests/Unit/ServerParityScheduleContractTest.mjs tests/Unit/ServerParityVisibilityContractTest.mjs tests/Unit/ServerParityAdmissionContractTest.mjs tests/Unit/ServerParityRoleTokenContractTest.mjs tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs tests/Unit/ServerParitySearchAttributeContractTest.mjs tests/Unit/ServerParityWorkerListingContractTest.mjs tests/Unit/ServerParityWorkerReplacementContractTest.mjs tests/Unit/ServerParityPatchDeploymentContractTest.mjs tests/Unit/ServerParityWaitingHistoryContractTest.mjs tests/Unit/ServerParityTaskRetryContractTest.mjs tests/Unit/ServerParityPublishedPatchContractTest.mjs tests/Unit/ServerParityPublishedPatchTransportTest.mjs tests/Unit/ServerParityRecordFileTest.mjs tests/Unit/ServerParityFailureLocationTest.mjs tests/Unit/ServerParityFailureExcerptTest.mjs tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs tests/Unit/ServerParityStorageTestContractTest.mjs tests/Unit/ServerParitySchedulePoliciesContractTest.mjs tests/Unit/ServerParitySignalsBeforeWorkerContractTest.mjs tests/Unit/ServerParityScheduleCatalogContractTest.mjs tests/Unit/ServerParityOperatorMetricsContractTest.mjs tests/Unit/ServerParityActivityLeaseRenewalContractTest.mjs tests/Unit/ServerParityActivityCancellationAckContractTest.mjs tests/Unit/ServerParityServiceCatalogContractTest.mjs tests/Unit/ServerParityWorkflowImportContractTest.mjs tests/Unit/ServerParityWaterlineImportContractTest.mjs tests/Unit/ServerParitySystemMetricsContractTest.mjs tests/Unit/ServerParityOperatorDashboardTypesContractTest.mjs tests/Unit/ServerParityPrometheusMetricsContractTest.mjs tests/Unit/ServerParityPayloadCleanupContractTest.mjs tests/Unit/ServerParityRepairPassContractTest.mjs tests/Unit/ServerParityClusterInfoContractTest.mjs tests/Unit/ServerParityNamespaceDeleteContractTest.mjs tests/Unit/ServerParityMessageStreamsContractTest.mjs tests/Unit/ServerParityWorkflowPollRequestContractTest.mjs tests/Unit/ServerParityUpdateValidationContractTest.mjs tests/Unit/ServerParityFixtureSelectionTest.mjs tests/Unit/ServerParityUpdateValidationRetryTest.mjs tests/Unit/ServerParityKilledAuthorPollTest.mjs tests/Unit/ServerParityImagePullTest.mjs tests/Unit/ServerParityWorkerSessionsContractTest.mjs tests/Unit/ServerParityWorkerSessionActivitiesContractTest.mjs > parity-evidence/comparator-tests.txt docker run --rm --user=1000:1000 --cpus=1 --memory=256m --memory-swap=256m \ --network=none --entrypoint node -v "$PWD:/source:ro" -w /source \ "$image" --test tests/Unit/ServerParityCodecDeadlineLoadTest.mjs > parity-evidence/codec-load-tests.txt diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md index 1723f93c4..1e7858d07 100644 --- a/docs/rust-server-port.md +++ b/docs/rust-server-port.md @@ -82,13 +82,17 @@ Updated October 24, 2026. ([#737](https://github.com/durable-workflow/server/issues/737)). No expectation is edited to make native pass. -- **Parity surface.** Native serves 139 of the 175 API route methods in +- **Parity surface.** Native serves 140 of the 175 API route methods in published 2.11.1's route table, with message streams, update validation, - activity cancellation acknowledgement, worker sessions and embedded v2 - workflow import. An execution test sends each one through the native - router, and the ones still answering `501 rust_capability_not_implemented` - must match + activity cancellation acknowledgement, worker sessions, and embedded v2 + and Waterline v1 workflow import. An execution test sends each one through + the native router, and the ones still answering + `501 rust_capability_not_implemented` must match [`unserved-php-routes.v2.11.1.txt`](../rust/tests/fixtures/unserved-php-routes.v2.11.1.txt). + Unlike published PHP, native's embedded v2 import refuses a bundle whose + rows reference rows outside it, links only runs of one namespace, and reads + the signing key only from the request body + ([#765](https://github.com/durable-workflow/server/issues/765)). - **Next action.** Land the items above in order, keep porting the unserved routes fixture first, and package the native binary into the server image. @@ -149,12 +153,6 @@ any other unknown path or method answers PHP's own 404 or 405 `update-validation` fixture. Native does not replay a validation task by `poll_request_id` or bind a request id to its task kinds; see `docs/rust-server-port-checkpoints/2026-10-23-update-validation.md`.) -- **Run operations.** - - Waterline v1 import. (Embedded v2 import is served: - `POST /api/workflows/import/embedded-v2`. Unlike published PHP, native - refuses a bundle whose rows reference rows outside it, links only runs of - one namespace, and reads the signing key only from the request body - ([#765](https://github.com/durable-workflow/server/issues/765)).) - **Other families.** - Message streams are served ([checkpoint](rust-server-port-checkpoints/2026-10-23-message-streams.md)), @@ -234,9 +232,28 @@ contract fixture before its correction. batch size, and still caps parallel groups at 100 members and a run at 64 updates over its lifetime. #612 tracks the rest ([checkpoint](rust-server-port-checkpoints/2026-10-23-structural-limits.md)). -- **Environment settings.** Of PHP's 166 contract settings, native honors - 43 and ignores 57 that change nothing it serves. It refuses to start on a - non-default value of the other 66 +- **Waterline v1 projections, ahead of published PHP.** + - Native refuses a message stream append to a projection with + `409 v1_projection_read_only`, at the route before the body is read and + again under the store's lock. Published 2.11.1 accepts it and creates a + signal and a workflow task + ([#769](https://github.com/durable-workflow/server/issues/769)). + - Native refuses a projection that would write more history events or + bytes than one transaction may (`DW_V2_LIMIT_HISTORY_TRANSACTION_SIZE`, + `_BYTES`) with `409 v1_projection_too_large`, before reading or writing + any row. The count is checked from the source's row counts before any + event is built; then each event is built and measured in turn, and the + first event past the byte limit stops the build. Published 2.11.1 has no limit + ([#770](https://github.com/durable-workflow/server/issues/770)). + - Operator metrics, the health snapshot and the backend capabilities + publish the active `history_transaction_size` and + `history_transaction_bytes` (0 when disabled), as Workflow #934's + `StructuralLimits::snapshot()` does. +- **Environment settings.** Of PHP's 164 contract settings, native honors + 49 and ignores 51 that change nothing it serves. It also honors + `DW_V2_LIMIT_HISTORY_TRANSACTION_BYTES`, which the Workflow package reads + and PHP's contract does not list yet. It refuses to start on a + non-default value of the other 64 ([checkpoint](rust-server-port-checkpoints/2026-10-22-environment-contract.md)). Several of those are cheap to honor: repair timing and the multi-node declaration. diff --git a/rust/src/config_audit.rs b/rust/src/config_audit.rs index c8f35a5ea..bb6bd30d2 100644 --- a/rust/src/config_audit.rs +++ b/rust/src/config_audit.rs @@ -474,7 +474,7 @@ pub fn audit(environment: &BTreeMap) -> Report { .keys() .map(String::as_str) .filter(|name| name.starts_with("DW_")) - .filter(|name| !SETTINGS.iter().any(|setting| setting.name == *name)) + .filter(|name| !SETTINGS.iter().chain(NATIVE_HONORED).any(|setting| setting.name == *name)) .filter(|name| !native_only(name)) .collect(); if !unknown.is_empty() { @@ -484,7 +484,7 @@ pub fn audit(environment: &BTreeMap) -> Report { )); } for name in environment.keys() { - if let Some(setting) = SETTINGS.iter().find(|setting| setting.legacy == Some(name.as_str())) { + if let Some(setting) = SETTINGS.iter().chain(NATIVE_HONORED).find(|setting| setting.legacy == Some(name.as_str())) { report.warnings.push(format!( "Legacy env var {name} is still honored but deprecated — rename to {}.", setting.name @@ -542,6 +542,14 @@ const UNLISTED: &[Setting] = &[ Setting { name: "DW_V2_LEGACY_SCHEDULE_STORAGE_TIMEZONE", legacy: Some("WORKFLOW_V2_LEGACY_SCHEDULE_STORAGE_TIMEZONE"), disposition: Refused(Empty, LEGACY_SCHEDULE_TIMEZONE) }, ]; +/// Settings native honors that PHP's contract does not list yet: the +/// Workflow package reads them (Workflow #930), and PHP's contract gains them +/// 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 }, +]; + /// Every `vars` entry of `config/dw-contract.php`, with its contract legacy name. pub const SETTINGS: &[Setting] = &[ Setting { name: "DW_ACTIVITY_TASK_MAX_ACTIVE_LEASES_PER_NAMESPACE", legacy: Some("WORKFLOW_SERVER_ACTIVITY_TASK_MAX_ACTIVE_LEASES_PER_NAMESPACE"), disposition: Refused(NoPositiveInt, TASK_ADMISSION) }, @@ -664,7 +672,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_SIZE", legacy: Some("WORKFLOW_V2_LIMIT_HISTORY_TRANSACTION_SIZE"), disposition: Ignored }, + 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 }, Setting { name: "DW_V2_LIMIT_PENDING_ACTIVITIES", legacy: Some("WORKFLOW_V2_LIMIT_PENDING_ACTIVITIES"), disposition: Refused(Int(2000), STRUCTURAL) }, @@ -1074,6 +1082,33 @@ mod tests { assert!(report.refusal().is_some_and(|refusal| !refusal.contains(secret))); } + #[test] + fn the_history_transaction_limits_are_honored_settings() { + let honored = |name: &str| SETTINGS.iter().chain(NATIVE_HONORED).find(|setting| setting.name == name) + .is_some_and(|setting| matches!(setting.disposition, Honored)); + for name in ["DW_V2_LIMIT_HISTORY_TRANSACTION_SIZE", "DW_V2_LIMIT_HISTORY_TRANSACTION_BYTES"] { + assert!(honored(name), "{name}"); + } + // NATIVE_HONORED names are outside PHP's contract and listed once. + for setting in NATIVE_HONORED { + assert!(!contract().iter().any(|(name, ..)| name == setting.name), "{} is now in the contract", setting.name); + assert!(!SETTINGS.iter().any(|listed| listed.name == setting.name)); + } + // Strict mode starts with each primary name set. + let mut primary = environment(&[("DW_V2_LIMIT_HISTORY_TRANSACTION_SIZE", "123"), ("DW_V2_LIMIT_HISTORY_TRANSACTION_BYTES", "4567")]); + primary.insert("DW_ENV_AUDIT_STRICT".into(), "1".into()); + let report = audit(&primary); + assert_eq!((report.warnings.as_slice(), report.refusal()), (&[] as &[String], None)); + // A legacy name warns as deprecated, naming its primary, and nothing else. + let report = audit(&environment(&[("WORKFLOW_V2_LIMIT_HISTORY_TRANSACTION_SIZE", "123"), + ("WORKFLOW_V2_LIMIT_HISTORY_TRANSACTION_BYTES", "4567")])); + assert_eq!(report.warnings, [ + "Legacy env var WORKFLOW_V2_LIMIT_HISTORY_TRANSACTION_BYTES is still honored but deprecated — rename to DW_V2_LIMIT_HISTORY_TRANSACTION_BYTES.", + "Legacy env var WORKFLOW_V2_LIMIT_HISTORY_TRANSACTION_SIZE is still honored but deprecated — rename to DW_V2_LIMIT_HISTORY_TRANSACTION_SIZE.", + ]); + assert_eq!(report.refusal(), None); + } + #[test] fn unknown_and_legacy_names_warn_and_strict_mode_refuses_them() { let values = [ diff --git a/rust/src/runtime/continuations.rs b/rust/src/runtime/continuations.rs index e8590924a..84097ff43 100644 --- a/rust/src/runtime/continuations.rs +++ b/rust/src/runtime/continuations.rs @@ -45,7 +45,7 @@ where /// `formatRunListEntry` renders it. The current run is the highest /// numbered one. pub(super) async fn list_runs(&self, workflow: &str) -> Result { - let rows=Self::query("SELECT r.id,r.workflow_type,r.business_key,r.status,r.run_number,r.queue,r.compatibility,r.started_at,r.closed_at FROM workflow_runs r JOIN workflow_instances i ON i.id=r.workflow_instance_id WHERE i.id=$1 AND i.namespace=$2 AND r.namespace=$3 ORDER BY r.run_number LIMIT 1025") + let rows=Self::query("SELECT r.id,r.workflow_type,r.business_key,r.status,r.run_number,r.queue,r.compatibility,r.started_at,r.closed_at,r.import_source FROM workflow_runs r JOIN workflow_instances i ON i.id=r.workflow_instance_id WHERE i.id=$1 AND i.namespace=$2 AND r.namespace=$3 ORDER BY r.run_number LIMIT 1025") .bind(workflow).bind(self.namespace.as_ref()).bind(self.namespace.as_ref()).fetch_all(&self.pool).await?; if rows.is_empty() { return Err(refuse(StatusCode::NOT_FOUND, "instance_not_found")); @@ -67,7 +67,7 @@ where let required = compatibility.as_deref().map(str::trim).filter(|value| !value.is_empty()); let (supported, reason) = super::run_diagnostics::fleet_supports(&fleet, required, queue.as_deref()); let memo = self.run_memo(&run).await?; - runs.push(json!({"run_id":run,"run_number":DB::number(&row,"run_number")?, + let mut entry = json!({"run_id":run,"run_number":DB::number(&row,"run_number")?, "workflow_type":DB::string(&row,"workflow_type")?,"business_key":DB::optional_string(&row,"business_key")?, "status":status,"status_bucket":super::visibility::status_bucket(&status)?, "is_terminal":matches!(status.as_str(), "completed" | "failed" | "cancelled" | "terminated"), @@ -75,7 +75,12 @@ where "compatibility_status":if supported { "compatible" } else { "no_compatible_worker" }, "compatibility_supported_in_fleet":supported,"compatibility_fleet_reason":reason, "started_at":super::task_queues::json_time(DB::optional_instant(&row,"started_at")?), - "closed_at":super::task_queues::json_time(DB::optional_instant(&row,"closed_at")?),"memo":memo})); + "closed_at":super::task_queues::json_time(DB::optional_instant(&row,"closed_at")?),"memo":memo}); + // `formatRunListEntry` leaves a Waterline v1 projection's memo out. + if DB::optional_string(&row, "import_source")?.as_deref() == Some(super::waterline_import::IMPORT_SOURCE) { + entry.as_object_mut().expect("run entry").remove("memo"); + } + runs.push(entry); } Ok(json!({"workflow_id":workflow,"run_count":runs.len(),"runs":runs})) } diff --git a/rust/src/runtime/control_plane.rs b/rust/src/runtime/control_plane.rs index 8467f7058..e0b512860 100644 --- a/rust/src/runtime/control_plane.rs +++ b/rust/src/runtime/control_plane.rs @@ -21,6 +21,7 @@ pub(super) enum Operation<'a> { Signal(&'a str), Query(&'a str), ImportEmbeddedV2, + ImportWaterlineV1, } impl Operation<'_> { @@ -139,6 +140,7 @@ impl Operation<'_> { ), // PHP defines no contract for the import, so its defaults apply. Self::ImportEmbeddedV2 => ("import_embedded_v2", &[], &[]), + Self::ImportWaterlineV1 => ("import_waterline_v1", &[], &["status", "identity"]), }; let mut metadata = json!({"schema":"durable-workflow.v2.control-plane-response", "version":1, "operation":operation, "workflow_id":body.get("workflow_id"), @@ -194,6 +196,14 @@ impl Operation<'_> { } } } + // Its contract's success and rejection fields. + Self::ImportWaterlineV1 => { + for field in ["status", "identity"] { + if let Some(value) = body.get(field) { + metadata[field] = value.clone(); + } + } + } _ => {} } // ControlPlaneResponseContract::PROJECTED_FIELDS: each present field. @@ -271,6 +281,13 @@ impl Operation<'_> { "control_plane_internal_error", ), Self::Redrive => ("workflow_id continued_from_run_id run_id reason message", ""), + Self::ImportWaterlineV1 => ( + "status reason message remediation identity", + "invalid_source_id unqualified_v1_identity v1_identity_mismatch unsupported_source_engine \ + missing_v1_workflow_class unsupported_v1_status v1_projection_changed \ + v1_projection_namespace_collision v1_projection_storage_inconsistent \ + native_v2_identity_collision target_write_failed", + ), Self::Cancel | Self::Terminate | Self::Archive | Self::Repair => { ("workflow_id run_id reason message remediation", "v1_projection_read_only") } @@ -319,6 +336,7 @@ pub(super) async fn admission_response(request: Request, mut body: Value) -> Val let segments: Vec<_> = path.trim_start_matches('/').split('/').collect(); let operation = match (request.method().as_str(), segments.as_slice()) { ("POST", ["api", "workflows", "import", "embedded-v2"]) => Operation::ImportEmbeddedV2, + ("POST", ["api", "workflows", "import", "waterline-v1"]) => Operation::ImportWaterlineV1, ("POST", ["api", "workflows"]) => Operation::Start, ("POST", ["api", "workflows", _, "signal", name]) | ("POST", ["api", "workflows", _, "runs", _, "signal", name]) => Operation::Signal(name), diff --git a/rust/src/runtime/cooperative.rs b/rust/src/runtime/cooperative.rs index b3aa73d38..cf55e7ca5 100644 --- a/rust/src/runtime/cooperative.rs +++ b/rust/src/runtime/cooperative.rs @@ -1000,6 +1000,12 @@ where if run_id != current { return Err(refuse(StatusCode::CONFLICT, "selected_run_not_current")); } + // A Waterline v1 projection is read-only; its v1 runtime owns execution. + if DB::optional_string(&run, "import_source")?.as_deref() == Some(super::waterline_import::IMPORT_SOURCE) { + return Err(super::RuntimeError::Protocol { status: StatusCode::CONFLICT, response: json!({ + "reason": "legacy_v1_operation_not_supported", + "message": "Imported legacy workflows do not support cooperative cancellation."}) }); + } let run_status = DB::string(&run, "status")?; let response = |duplicate: bool, pending: Value| { json!({"workflow_id":workflow_id,"run_id":run_id, diff --git a/rust/src/runtime/execution.rs b/rust/src/runtime/execution.rs index bf16f1b1a..e611ccf36 100644 --- a/rust/src/runtime/execution.rs +++ b/rust/src/runtime/execution.rs @@ -600,6 +600,14 @@ impl Runtime { response: json!({"accepted":false,"workflow_id":workflow_id,"stream_name":stream,"outcome":"rejected","reason":"instance_not_found"}), }); } + // A read-only Waterline projection takes no messages, whatever the + // body; the store checks again under its lock. + if let Some(run_id) = self.projected_current_run(workflow_id).await? { + return Err(super::RuntimeError::Protocol { + status: StatusCode::CONFLICT, + response: super::waterline_import::read_only_append(workflow_id, &run_id, stream), + }); + } let body = super::laravel_rules::normalize(&body); let blob = match self.message_payload(&body).await { Ok(blob) => blob, @@ -619,6 +627,19 @@ impl Runtime { } } + /// The store's append alone, without the service's checks: tests of the + /// store's own guard call it. + #[cfg(test)] + pub(crate) async fn store_append_message(&self, workflow_id: &str, stream: &str, message_id: &str, blob: &str) + -> Result<(StatusCode, Value)> { + let (codec, structural, context) = (self.signal_codec.clone(), self.structural, json!({})); + match &*self.storage { + Storage::Sqlite(store) => store.append_message(workflow_id, stream, message_id, "avro", blob, &context, codec, structural).await, + Storage::Postgres(store) => store.append_message(workflow_id, stream, message_id, "avro", blob, &context, codec, structural).await, + Storage::MySql(store) => store.append_message(workflow_id, stream, message_id, "avro", blob, &context, codec, structural).await, + } + } + /// The request rules, then `AvroPayloadEnvelopeResolver::resolve`: no /// input is an empty argument list, plain values are encoded, and an /// inline envelope keeps its exact bytes. @@ -1580,6 +1601,10 @@ impl Runtime { delegate!(renew_activity_lease(task_id: &str, body: Value) -> Result); delegate!(import_existing_run(run_id: &str) -> Result>); delegate!(import_write(plan: &super::workflow_import::Plan) -> std::result::Result); + delegate!(waterline_existing(workflow_id: &str, run_id: &str) -> Result); + delegate!(projected_current_run(workflow_id: &str) -> Result>); + delegate!(projected_run(workflow_id: &str, run_id: &str) -> Result>); + delegate!(waterline_write(projection: &super::waterline_import::Normalized, imported_at: chrono::DateTime) -> std::result::Result<(), bool>); /// The namespace this runtime's store serves. pub(crate) fn namespace_name(&self) -> String { match &*self.storage { diff --git a/rust/src/runtime/history_export.rs b/rust/src/runtime/history_export.rs index 8a0238d3a..e34dcdba2 100644 --- a/rust/src/runtime/history_export.rs +++ b/rust/src/runtime/history_export.rs @@ -64,6 +64,7 @@ pub(super) const RUN: &[(&str, Column)] = &[ ("queue", Text), ("last_history_sequence", Int), ("started_at", Time), ("closed_at", Time), ("archived_at", Time), ("archive_command_id", Text), ("archive_reason", Text), ("last_progress_at", Time), ("updated_at", Time), ("current_run_id", Text), ("instance_business_key", Text), ("instance_visibility_labels", Json), ("continuations", Int), + ("import_source", Text), ("imported_at", Time), ("import_dedupe_key", Text), ]; pub(super) const RUN_SQL: &str = "SELECT r.*,i.current_run_id,i.business_key AS instance_business_key,i.visibility_labels AS instance_visibility_labels,(SELECT COUNT(*) FROM workflow_links l WHERE l.child_workflow_instance_id=r.workflow_instance_id AND l.link_type='continue_as_new') AS continuations FROM workflow_runs r JOIN workflow_instances i ON i.id=r.workflow_instance_id WHERE i.id=$1 AND r.id=$2 AND r.namespace=$3"; pub(super) const SUMMARY: &[(&str, Column)] = &[ @@ -605,10 +606,11 @@ fn codec_schemas() -> Value { } /// `withIntegrity`: SHA-256, and HMAC-SHA-256 when a key is configured, over -/// the bundle's canonical JSON. Native objects keep their keys sorted, so -/// compact serialization is the recursive-ksort canonical form. +/// the bundle's canonical JSON as PHP encodes it: maps key-sorted, lists in +/// order, slashes unescaped, zero fractions kept, and non-ASCII text escaped, +/// so PHP's importer verifies a native export. fn integrity(bundle: &Value, signing: Option<&Signing>) -> Value { - let canonical = serde_json::to_string(bundle).expect("JSON values serialize"); + let canonical = canonical_json(bundle); let checksum = hex(&Sha256::digest(canonical.as_bytes())); let signature = signing.map(|signing| { let mut mac = Hmac::::new_from_slice(signing.key.as_bytes()).expect("HMAC takes any key length"); @@ -620,6 +622,26 @@ fn integrity(bundle: &Value, signing: Option<&Signing>) -> Value { "key_id": signing.and_then(|signing| signing.key_id.clone())}) } +/// `BundleIntegrityVerifier::canonicalJson` of a bundle. +pub(super) fn canonical_json(bundle: &Value) -> String { + let text = serde_json::to_vec(bundle).expect("JSON values serialize"); + super::php_value::decode(&text).expect("serialized JSON decodes").canonical().encode(super::php_value::CANONICAL) +} + +/// `LegacyV1Projection::metadataForRun` from the export's own rows. +fn projection_metadata(rows: &Rows) -> Value { + let migration = rows.events.first().map(|event| event["payload"]["migration_projection"].clone()).unwrap_or(Value::Null); + json!({ + "origin": migration["origin"], + "identity": migration["identity"], + "task_queue_context": migration["task_queue_context"], + "unsupported_fields": migration["unsupported_fields"].as_array().cloned().unwrap_or_default(), + "projection": {"schema": super::waterline_import::SCHEMA, "version": super::waterline_import::VERSION, + "read_only": true, "execution_owner": "v1", "imported_at": rows.run["imported_at"], + "dedupe_key": rows.run["import_dedupe_key"]}, + }) +} + /// `HistoryExport::forRun`. pub(super) fn bundle(rows: &Rows, exported_at: &str, signing: Option<&Signing>) -> Value { let run = &rows.run; @@ -655,6 +677,11 @@ pub(super) fn bundle(rows: &Rows, exported_at: &str, signing: Option<&Signing>) "codec_schemas": codec_schemas(), }); bundle["payload_manifest"] = manifest(&bundle); + // `decorateHistoryExport`: a Waterline v1 projection carries its + // provenance, and the integrity block covers it. + if run["import_source"] == super::waterline_import::IMPORT_SOURCE { + bundle["migration_projection"] = projection_metadata(rows); + } bundle["integrity"] = integrity(&bundle, signing); bundle } @@ -687,11 +714,18 @@ mod tests { let signed = bundle(&rows, "now", Some(&signing)); let mut unsigned = signed.clone(); unsigned.as_object_mut().unwrap().remove("integrity"); - let canonical = serde_json::to_string(&unsigned).unwrap(); + let canonical = canonical_json(&unsigned); assert_eq!(signed["integrity"]["checksum"], hex(&Sha256::digest(canonical.as_bytes()))); assert_eq!(signed["integrity"]["signature_algorithm"], "hmac-sha256"); assert_eq!(signed["integrity"]["key_id"], "v1"); assert_eq!(signed["dedupe_key"], "r:2:t"); assert_eq!(signed["history_complete"], true); } + + #[test] + fn canonical_json_matches_php() { + // PHP printed {"a":[2,1],"b":1.0,"k":"\u03bb/x"} for this value with + // JSON_UNESCAPED_SLASHES | JSON_PRESERVE_ZERO_FRACTION after ksort. + assert_eq!(canonical_json(&json!({"k": "λ/x", "b": 1.0, "a": [2, 1]})), r#"{"a":[2,1],"b":1.0,"k":"\u03bb/x"}"#); + } } diff --git a/rust/src/runtime/http.rs b/rust/src/runtime/http.rs index 23fcc6c81..ec8d68d7c 100644 --- a/rust/src/runtime/http.rs +++ b/rust/src/runtime/http.rs @@ -68,6 +68,7 @@ pub fn router(runtime: Runtime) -> Router { ) .route("/api/workflows", get(list_workflows).post(start)) .route("/api/workflows/import/embedded-v2", post(import_embedded_v2)) + .route("/api/workflows/import/waterline-v1", post(import_waterline_v1)) .route("/api/activities", get(list_activities).post(start_activity)) .route("/api/activities/{activity_id}", get(show_activity)) .route( @@ -319,6 +320,7 @@ pub fn router(runtime: Runtime) -> Router { ) .fallback(route_fallback) .method_not_allowed_fallback(route_fallback) + .layer(middleware::from_fn(super::waterline_import::read_only_guard)) .layer(middleware::from_fn(empty_list_body)) // `payload_limits` reads and bounds every body first. .layer(DefaultBodyLimit::disable()) @@ -709,7 +711,8 @@ fn exact_namespace_route(path: &str) -> Option { // The metrics describe the namespace's own state, read by GET or POST. "/api/system/metrics" => READ_WRITE, // An import writes into the requested namespace. - "/api/external-payloads/v1" | "/api/storage/test" | "/api/workflows/import/embedded-v2" => WRITE, + "/api/external-payloads/v1" | "/api/storage/test" | "/api/workflows/import/embedded-v2" + | "/api/workflows/import/waterline-v1" => WRITE, "/api/task-queues" | "/api/deployments" | "/api/workers" @@ -1111,6 +1114,15 @@ async fn import_embedded_v2( Ok((status, Json(answer))) } +/// `POST /api/workflows/import/waterline-v1`. +async fn import_waterline_v1( + Extension(runtime): Extension, + body: axum::body::Bytes, +) -> Result<(StatusCode, Json)> { + let (status, answer) = runtime.import_waterline_v1(&body).await?; + Ok((status, Json(answer))) +} + /// `POST /api/activities`: a standalone activity in its own host run. async fn start_activity( Extension(runtime): Extension, diff --git a/rust/src/runtime/memo_limits.rs b/rust/src/runtime/memo_limits.rs index fbf931e67..dd5efdd89 100644 --- a/rust/src/runtime/memo_limits.rs +++ b/rust/src/runtime/memo_limits.rs @@ -90,6 +90,13 @@ pub struct StructuralLimits { pub pending_updates: Option, /// Bytes of one serialized payload blob (DW_V2_LIMIT_PAYLOAD_SIZE_BYTES). pub payload_bytes: Option, + /// History events one transaction may write + /// (DW_V2_LIMIT_HISTORY_TRANSACTION_SIZE; Workflow #930): here, a + /// Waterline v1 projection import. + pub history_events: Option, + /// History bytes one transaction may write, measured as PHP measures an + /// event (DW_V2_LIMIT_HISTORY_TRANSACTION_BYTES). + pub history_bytes: Option, } impl Default for StructuralLimits { @@ -98,6 +105,8 @@ impl Default for StructuralLimits { pending_signals: Some(5000), pending_updates: Some(500), payload_bytes: Some(2_097_152), + history_events: Some(5000), + history_bytes: Some(16_777_216), } } } @@ -110,6 +119,8 @@ impl StructuralLimits { pending_signals: configured("PENDING_SIGNALS", defaults.pending_signals.unwrap(), invalid)?, pending_updates: configured("PENDING_UPDATES", defaults.pending_updates.unwrap(), invalid)?, payload_bytes: configured("PAYLOAD_SIZE_BYTES", defaults.payload_bytes.unwrap(), invalid)?, + history_events: configured("HISTORY_TRANSACTION_SIZE", defaults.history_events.unwrap(), invalid)?, + history_bytes: configured("HISTORY_TRANSACTION_BYTES", defaults.history_bytes.unwrap(), invalid)?, }) } } diff --git a/rust/src/runtime/message_streams.rs b/rust/src/runtime/message_streams.rs index 42e87ab1a..2f3294731 100644 --- a/rust/src/runtime/message_streams.rs +++ b/rust/src/runtime/message_streams.rs @@ -308,6 +308,11 @@ where Ok(open.then_some(run)) } + /// Whether the run is a read-only Waterline v1 projection. + fn projected(run: &DB::Row) -> Result { + Ok(DB::optional_string(run, "import_source")?.as_deref() == Some(super::waterline_import::IMPORT_SOURCE)) + } + /// `MessageStreamService::append`, in one transaction. #[allow(clippy::too_many_arguments)] pub(crate) async fn append_message( @@ -330,6 +335,10 @@ where let Some(run) = Self::active_current_run(&mut tx, workflow_id).await? else { return Err(rejection(StatusCode::CONFLICT, "run_not_active", workflow_id, stream, message_id, json!({}))); }; + if Self::projected(&run)? { + return Err(RuntimeError::Protocol { status: StatusCode::CONFLICT, + response: super::waterline_import::read_only_append(workflow_id, &DB::string(&run, "id")?, stream) }); + } let row = self.locked_stream(&mut tx, workflow_id, stream).await?; let stream_id = DB::number(&row, "id")?; let cursor = DB::number(&row, "cursor_position")?; @@ -388,8 +397,10 @@ where let mut tx = self.begin().await?; let bound = Self::query("SELECT id FROM workflow_instances WHERE id=$1 AND namespace=$2") .bind(workflow_id).bind(self.namespace.as_ref()).fetch_optional(&mut *tx).await?; - if bound.is_none() || Self::active_current_run(&mut tx, workflow_id).await?.is_none() { - return Ok(()); + // A read-only projection keeps no stream state, malformed or not. + match Self::active_current_run(&mut tx, workflow_id).await? { + Some(run) if bound.is_some() && !Self::projected(&run)? => {} + _ => return Ok(()), } let row = self.locked_stream(&mut tx, workflow_id, stream).await?; let at = now(); diff --git a/rust/src/runtime/mod.rs b/rust/src/runtime/mod.rs index 8f444e3d6..fa5bbe531 100644 --- a/rust/src/runtime/mod.rs +++ b/rust/src/runtime/mod.rs @@ -111,6 +111,7 @@ mod task_queue_build_ids; mod task_queues; mod query_route; mod run_describe; +mod waterline_import; mod workflow_import; mod workflow_streams; mod schedule_list; diff --git a/rust/src/runtime/operator_metrics.rs b/rust/src/runtime/operator_metrics.rs index 02fa06405..9fe1c20b3 100644 --- a/rust/src/runtime/operator_metrics.rs +++ b/rust/src/runtime/operator_metrics.rs @@ -270,7 +270,7 @@ where "projections": self.projection_metrics(now).await?, "schedules": self.schedule_metrics(now).await?, "workers": self.worker_metrics().await?, - "backend": backend_capabilities(now, DB::PHP_CONNECTION), + "backend": backend_capabilities(now, DB::PHP_CONNECTION, &config.structural, self.memo_size_limit), "structural_limits": structural_limits(DB::PHP_CONNECTION, &config.structural, self.memo_size_limit), "update_wait": {"completion_timeout_seconds": 10, "poll_interval_milliseconds": 50}, "repair_policy": repair_policy(), @@ -1491,7 +1491,8 @@ fn structural_limits(connection: &str, limits: &super::StructuralLimits, memo: O "payload_size_bytes": shown(limits.payload_bytes), "memo_size_bytes": shown(memo), "search_attribute_size_bytes": 40960, - "history_transaction_size": 5000, + "history_transaction_size": shown(limits.history_events), + "history_transaction_bytes": shown(limits.history_bytes), "warning_threshold_percent": 80, }); let sqlite = connection == "sqlite"; @@ -1508,9 +1509,9 @@ fn structural_limits(connection: &str, limits: &super::StructuralLimits, memo: O } /// What this runtime needs from its backends, in PHP's capability shape. -fn backend_capabilities(now: DateTime, connection: &str) -> Value { - // Only the backend's issues are reported here, which no ceiling changes. - let limits = structural_limits(connection, &super::StructuralLimits::default(), None); +fn backend_capabilities(now: DateTime, connection: &str, structural: &super::StructuralLimits, memo: Option) -> Value { + // The active ceilings, with the backend's adjustments and issues. + let limits = structural_limits(connection, structural, memo); let issues = limits["issues"].clone(); let severity = if issues.as_array().is_some_and(|issues| !issues.is_empty()) { "info" diff --git a/rust/src/runtime/run_describe.rs b/rust/src/runtime/run_describe.rs index 6320efecf..72ba500fb 100644 --- a/rust/src/runtime/run_describe.rs +++ b/rust/src/runtime/run_describe.rs @@ -286,9 +286,16 @@ where } fields.extend(compatibility); // `ServerWorkflowControlPlane::withDefinitionRouting`. - let mut tx = self.begin_read().await?; - let routing = self.definition_inspect(&mut tx, run, None).await?; - tx.commit().await?; + // A Waterline v1 projection has no standalone routing, and may have + // no task queue at all. + let routing = if DB::optional_string(run, "import_source")?.as_deref() == Some(super::waterline_import::IMPORT_SOURCE) { + Value::Null + } else { + let mut tx = self.begin_read().await?; + let routing = self.definition_inspect(&mut tx, run, None).await?; + tx.commit().await?; + routing + }; fields.insert("definition_routing".into(), routing); fields.insert("payload_previews".into(), json!({"input_omitted":input_omitted,"output_omitted":output_omitted, "max_encoded_bytes":max_payload_bytes()})); diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index 964fd1135..3dfabf634 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -704,7 +704,7 @@ where let output_envelope = output.as_deref().map(wire); let mut described = json!({"workflow_id": workflow_id, "run_id": DB::string(&row,"id")?, "workflow_type": DB::string(&row,"workflow_type")?, - "namespace": self.namespace.as_ref(), "task_queue": DB::string(&row,"queue")?, "status": status, + "namespace": self.namespace.as_ref(), "task_queue": DB::optional_string(&row,"queue")?, "status": status, "commands_scope":"selected_run","commands":commands, "status_bucket": super::visibility::status_bucket(&status)?, "is_current_run": current, @@ -715,7 +715,7 @@ where "input_envelope": input_envelope, "output_envelope": output_envelope, // `toJSON()` for start and close; the run description's // `toIso8601String()` for deadlines. - "started_at": super::task_queues::json_time(Some(DB::instant(&row,"started_at")?)), + "started_at": super::task_queues::json_time(DB::optional_instant(&row,"started_at")?), "closed_at": super::task_queues::json_time(DB::optional_instant(&row,"closed_at")?), "execution_timeout_seconds": DB::optional_number(&row,"execution_timeout_seconds")?, "run_timeout_seconds": DB::optional_number(&row,"run_timeout_seconds")?, @@ -727,6 +727,38 @@ where .await?; described.as_object_mut().expect("description object").extend(fields); } + // `formatRun` for a Waterline v1 projection: no control-plane + // description, every action off, the stored codec and progress time, + // and the projection's metadata. + if DB::optional_string(&row, "import_source")?.as_deref() == Some(super::waterline_import::IMPORT_SOURCE) { + let metadata = self.projection_metadata(&DB::string(&row, "id")?, DB::optional_instant(&row, "imported_at")?, + DB::optional_string(&row, "import_dedupe_key")?).await?; + let failure = if stale_after.is_some() && described["status"] == "failed" { + self.projected_failure(&DB::string(&row, "id")?).await? + } else { + Value::Null + }; + let fields = described.as_object_mut().expect("description object"); + fields.insert("payload_codec".into(), json!(DB::optional_string(&row, "payload_codec")?)); + if stale_after.is_some() { + fields.insert("definition_routing".into(), Value::Null); + fields.insert("last_progress_at".into(), super::task_queues::json_time(DB::optional_instant(&row, "last_progress_at")?)); + if let Some(Value::Object(actions)) = fields.get_mut("actions") { + actions.values_mut().for_each(|value| *value = json!(false)); + } + fields.insert("migration_projection".into(), metadata); + fields.insert("action_blocked_reason".into(), json!("v1_projection_read_only")); + // `terminalFailurePayload`: a projection keeps no failure + // snapshots, so a failed one reports its projected activity + // failures only. + if !failure.is_null() { + fields.insert("error".into(), failure["message"].clone()); + fields.insert("exception".into(), failure["exception"].clone()); + fields.insert("failures".into(), json!([])); + fields.insert("failure".into(), failure); + } + } + } Ok(described) } @@ -738,7 +770,7 @@ where page_size: i64, ) -> Result { self.describe(workflow_id, Some(run_id)).await?; - let run = Self::query("SELECT compatibility,queue FROM workflow_runs WHERE id=$1") + let run = Self::query("SELECT compatibility,queue,import_source,imported_at,import_dedupe_key FROM workflow_runs WHERE id=$1") .bind(run_id) .fetch_one(&self.pool) .await?; @@ -757,6 +789,11 @@ where // describe does. let mut body = json!({"workflow_id": workflow_id, "run_id": run_id, "events": public_events, "next_page_token": next}); body.as_object_mut().expect("history object").extend(compatibility); + // `LegacyV1Projection::metadataForRun` on a projected run. + if DB::optional_string(&run, "import_source")?.as_deref() == Some(super::waterline_import::IMPORT_SOURCE) { + body["migration_projection"] = self.projection_metadata(run_id, DB::optional_instant(&run, "imported_at")?, + DB::optional_string(&run, "import_dedupe_key")?).await?; + } Ok(body) } @@ -4005,7 +4042,8 @@ where Ok( json!({"id": DB::string(&row,"id")?, "sequence": DB::number(&row,"sequence")?, "event_type": DB::string(&row,"event_type")?, "payload": payload, - "recorded_at": DB::instant(&row,"recorded_at")?.to_rfc3339_opts(chrono::SecondsFormat::Micros, true), + // A Waterline v1 projection may hold untimed events; PHP reads them as null. + "recorded_at": DB::optional_instant(&row,"recorded_at")?.map(|at| at.to_rfc3339_opts(chrono::SecondsFormat::Micros, true)), "workflow_task_id": DB::optional_string(&row,"workflow_task_id")?, "workflow_command_id": DB::optional_string(&row,"workflow_command_id")?}), ) } diff --git a/rust/src/runtime/visibility.rs b/rust/src/runtime/visibility.rs index a9346da7c..97341d4d3 100644 --- a/rust/src/runtime/visibility.rs +++ b/rust/src/runtime/visibility.rs @@ -125,15 +125,18 @@ fn validation(field: &'static str, message: &'static str) -> RuntimeError { } } -pub(super) const VISIBILITY_PAGE: &str = "SELECT r.id,r.workflow_instance_id,r.workflow_type,r.queue,r.status,r.compatibility,r.started_at,r.closed_at,COALESCE(r.business_key,i.business_key) AS business_key FROM workflow_runs r JOIN workflow_instances i ON i.id=r.workflow_instance_id WHERE r.namespace=$1 AND i.namespace=$2 AND ($3 IS NULL OR r.workflow_type=$4) AND ($5 IS NULL OR CASE WHEN r.status IN ('pending','running','waiting') THEN 'running' WHEN r.status='completed' THEN 'completed' ELSE 'failed' END=$6 OR r.status=$7) AND ($8 IS NULL OR r.workflow_instance_id LIKE $9 OR COALESCE(r.business_key,i.business_key) LIKE $10) ORDER BY r.started_at DESC,r.id DESC LIMIT $11 OFFSET $12"; +pub(super) const VISIBILITY_PAGE: &str = "SELECT r.id,r.workflow_instance_id,r.workflow_type,r.queue,r.status,r.compatibility,r.started_at,r.closed_at,COALESCE(r.business_key,i.business_key) AS business_key FROM workflow_runs r JOIN workflow_instances i ON i.id=r.workflow_instance_id LEFT JOIN workflow_run_summaries s ON s.id=r.id AND r.import_source='waterline_v1_projection' WHERE r.namespace=$1 AND i.namespace=$2 AND ($3 IS NULL OR r.workflow_type=$4) AND ($5 IS NULL OR CASE WHEN r.status IN ('pending','running','waiting') THEN 'running' WHEN r.status='completed' THEN 'completed' ELSE 'failed' END=$6 OR r.status=$7) AND ($8 IS NULL OR r.workflow_instance_id LIKE $9 OR COALESCE(r.business_key,i.business_key) LIKE $10) ORDER BY COALESCE(s.sort_timestamp,r.started_at) DESC,r.id DESC LIMIT $11 OFFSET $12"; // Preserve decimal micro/nanosecond ordering across PHP SQL UTC and earlier // native RFC3339 UTC strings without SQLite floating-point date conversion. +// A Waterline v1 projection sorts by its summary's sort time, as PHP's +// summary-ordered list places it; other runs keep their start time. pub(super) const SQLITE_VISIBILITY_PAGE: &str = r#" WITH runs AS ( SELECT r.id,r.workflow_instance_id,r.workflow_type,r.queue,r.status,r.compatibility,r.started_at,r.closed_at,COALESCE(r.business_key,i.business_key) AS business_key, - replace(replace(replace(r.started_at,'T',' '),'Z',''),'+00:00','') AS _dw_started_utc + replace(replace(replace(COALESCE(s.sort_timestamp,r.started_at),'T',' '),'Z',''),'+00:00','') AS _dw_started_utc FROM workflow_runs r JOIN workflow_instances i ON i.id=r.workflow_instance_id + LEFT JOIN workflow_run_summaries s ON s.id=r.id AND r.import_source='waterline_v1_projection' WHERE r.namespace=$1 AND i.namespace=$2 AND ($3 IS NULL OR r.workflow_type=$4) AND ($5 IS NULL OR CASE WHEN r.status IN ('pending','running','waiting') THEN 'running' WHEN r.status='completed' THEN 'completed' ELSE 'failed' END=$6 OR r.status=$7) AND ($8 IS NULL OR r.workflow_instance_id LIKE $9 OR COALESCE(r.business_key,i.business_key) LIKE $10) @@ -275,10 +278,10 @@ where }; workflows.push(json!({"workflow_id":DB::string(row,"workflow_instance_id")?, "run_id":run_id, "workflow_type":DB::string(row,"workflow_type")?, "business_key":DB::optional_string(row,"business_key")?, - "status":status, "status_bucket":bucket, "task_queue":DB::string(row,"queue")?, + "status":status, "status_bucket":bucket, "task_queue":DB::optional_string(row,"queue")?, "is_terminal":bucket!="running", "compatibility":null, "compatibility_status":"compatible", "compatibility_supported_in_fleet":true, "compatibility_fleet_reason":null, - "started_at":DB::instant(row,"started_at")?, "closed_at":DB::optional_instant(row,"closed_at")?, + "started_at":super::task_queues::json_time(DB::optional_instant(row,"started_at")?), "closed_at":super::task_queues::json_time(DB::optional_instant(row,"closed_at")?), "search_attributes":search_attributes})); } let next = diff --git a/rust/src/runtime/waterline_import.rs b/rust/src/runtime/waterline_import.rs new file mode 100644 index 000000000..af36e38ee --- /dev/null +++ b/rust/src/runtime/waterline_import.rs @@ -0,0 +1,1398 @@ +//! `LegacyV1ProjectionController` and `LegacyV1Projection`, as published +//! Server 2.11.1 serves them: a Waterline v1 flow detail projected into one +//! read-only standalone run. The identity is a deterministic hash of the +//! source id and the qualified v1 id; the report, events, dedupe key and +//! summary follow PHP's construction, including Eloquent storing ordinary +//! date columns as the source's wall-clock time and history times as UTC. +use super::execution::Runtime; +use super::php_value::{Flags, Key, Php, PhpArray}; +use super::store::now; +use super::workflow_import::{Stamp, parse_time}; +use super::Result; +use axum::http::StatusCode; +use chrono::{DateTime, Utc}; +use serde_json::{Value, json}; +use sha2::{Digest, Sha256}; + +pub(super) const IMPORT_SOURCE: &str = "waterline_v1_projection"; +pub(super) const ENGINE_SOURCE: &str = "v1_projection"; +pub(super) const SCHEMA: &str = "durable-workflow.v2.waterline-v1-projection.contract"; +const REPORT_SCHEMA: &str = "durable-workflow.v2.waterline-v1-projection.report"; +pub(super) const VERSION: i64 = 1; +const OPAQUE_REASON: &str = "legacy_php_serialization_not_portable"; +const OPAQUE_REMEDIATION: &str = + "Treat the preserved value as opaque; export decoded JSON from the source application if portable payload access is required."; + +/// A PHP array literal with string keys, in order. +fn object(entries: Vec<(&str, Php)>) -> Php { + Php::Array(PhpArray(entries.into_iter().map(|(key, value)| (Key::of(key), value)).collect())) +} + +fn list(items: Vec) -> Php { + Php::Array(PhpArray(items.into_iter().enumerate().map(|(index, value)| (Key::Int(index as i64), value)).collect())) +} + +fn text(value: impl Into) -> Php { + Php::Str(value.into()) +} + +fn optional(value: Option) -> Php { + value.map_or(Php::Null, Php::Str) +} + +/// PHP's `trim()` with its default characters. +fn php_trim(value: &str) -> &str { + value.trim_matches([' ', '\t', '\n', '\r', '\0', '\x0b']) +} + +/// `LegacyV1Projection::string`: a string that is not blank, trimmed. +fn string(value: Option<&Php>) -> Option { + match value { + Some(Php::Str(text)) if !php_trim(text).is_empty() => Some(php_trim(text).to_owned()), + _ => None, + } +} + +/// PHP's `$array[$key] ?? $default`: null and absent both fall through. +fn coalesce<'a>(array: &'a PhpArray, keys: &[&str]) -> Option<&'a Php> { + keys.iter().find_map(|key| array.get(key).filter(|value| !matches!(value, Php::Null))) +} + +/// PHP's `(string)` of a float: `precision` 14, as `%.14G` prints it. +fn float_string(value: f64) -> String { + if value.is_nan() { + return "NAN".into(); + } + if value.is_infinite() { + return if value > 0.0 { "INF".into() } else { "-INF".into() }; + } + if value == 0.0 { + return if value.is_sign_negative() { "-0".into() } else { "0".into() }; + } + let scientific = format!("{value:.13e}"); + let (mantissa, exponent) = scientific.split_once('e').expect("scientific notation"); + let exponent: i32 = exponent.parse().expect("exponent"); + let (sign, digits) = mantissa.strip_prefix('-').map_or(("", mantissa), |digits| ("-", digits)); + let digits: String = digits.chars().filter(char::is_ascii_digit).collect(); + let digits = digits.trim_end_matches('0'); + let digits = if digits.is_empty() { "0" } else { digits }; + if exponent < -4 || exponent >= 14 { + let fraction = if digits.len() > 1 { &digits[1..] } else { "0" }; + return format!("{sign}{}.{fraction}E{}{}", &digits[..1], if exponent < 0 { '-' } else { '+' }, exponent.abs()); + } + if exponent < 0 { + return format!("{sign}0.{}{digits}", "0".repeat((-exponent - 1) as usize)); + } + let point = exponent as usize + 1; + if digits.len() <= point { + format!("{sign}{digits}{}", "0".repeat(point - digits.len())) + } else { + format!("{sign}{}.{}", &digits[..point], &digits[point..]) + } +} + +/// `is_scalar($value) ? (string) $value : null`. +fn scalar_string(value: Option<&Php>) -> Option { + match value? { + Php::Str(text) => Some(text.clone()), + Php::Int(number) => Some(number.to_string()), + Php::Float(number) => Some(float_string(*number)), + Php::Bool(true) => Some("1".into()), + Php::Bool(false) => Some(String::new()), + _ => None, + } +} + +/// `LegacyV1Projection::timestamp`: a non-blank string Carbon can parse. +fn timestamp(value: Option<&Php>) -> Option { + match value { + Some(Php::Str(text)) if !php_trim(text).is_empty() => parse_time(text).ok(), + _ => None, + } +} + +/// Carbon's `toJSON()`: the UTC instant with microseconds. +fn json_time(stamp: Option) -> Option { + stamp.map(|stamp| stamp.utc.format("%Y-%m-%dT%H:%M:%S%.6fZ").to_string()) +} + +fn sha256(text: &str) -> String { + super::bundle_verifier::hex(&Sha256::digest(text.as_bytes())) +} + +/// `rows`: the array members of a list, reindexed. +fn rows(value: Option<&Php>) -> Vec { + match value { + Some(Php::Array(array)) => array.values().filter_map(|value| value.array().cloned()).collect(), + _ => Vec::new(), + } +} + +/// How many rows `rows` would return, without copying them. +fn row_count(value: Option<&Php>) -> usize { + match value { + Some(Php::Array(array)) => array.values().filter(|value| value.array().is_some()).count(), + _ => 0, + } +} + +/// A projection larger than one transaction may write: its history events +/// against `DW_V2_LIMIT_HISTORY_TRANSACTION_SIZE`, or its history bytes +/// against `DW_V2_LIMIT_HISTORY_TRANSACTION_BYTES`. Published PHP has no +/// such limit; native refuses before reading or writing any row. +fn too_large(dimension: &str, used: usize, limit: usize) -> Php { + let (measure, setting) = match dimension { + "history_transaction_size" => ("history events", "DW_V2_LIMIT_HISTORY_TRANSACTION_SIZE"), + _ => ("history bytes", "DW_V2_LIMIT_HISTORY_TRANSACTION_BYTES"), + }; + object(vec![ + ("reason", text("v1_projection_too_large")), + ("message", text(format!("The projection would write {used} {measure} in one transaction; at most {limit} are allowed ({dimension})."))), + ("remediation", text(format!("Keep the workflow on its v1 runtime, or raise {setting} on this server before importing it."))), + ("limit", object(vec![("dimension", text(dimension)), ("used", Php::Int(used as i64)), ("limit", Php::Int(limit as i64))])), + ]) +} + +/// The first limit a projection of `events` events and `bytes` bytes +/// exceeds, if any; 0 disables a limit. +fn exceeded(limits: &super::StructuralLimits, events: usize, bytes: usize) -> Option { + let over = |limit: Option, used: usize| limit.filter(|limit| *limit > 0 && used > *limit); + over(limits.history_events, events) + .map(|limit| too_large("history_transaction_size", events, limit)) + .or_else(|| over(limits.history_bytes, bytes).map(|limit| too_large("history_transaction_bytes", bytes, limit))) +} + +/// The body that refuses an operation on a read-only projection. +pub(super) fn read_only_body(workflow_id: &str, run_id: &str, operation: &str) -> Value { + json!({ + "message": format!("Operation [{operation}] is unavailable because this execution is a read-only projection owned by the source v1 runtime."), + "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": workflow_id, + "run_id": run_id, + "operation": operation, + "execution_owner": "v1", + }) +} + +/// A message stream append refused on a read-only projection, in the shape +/// of the stream's own refusals. Published PHP accepts it. +pub(super) fn read_only_append(workflow_id: &str, run_id: &str, stream: &str) -> Value { + let mut body = read_only_body(workflow_id, run_id, "message_stream_append"); + for (key, value) in [("accepted", json!(false)), ("duplicate", json!(false)), ("stream_name", json!(stream)), + ("outcome", json!("rejected"))] { + body[key] = value; + } + body +} + +fn error(reason: &str, message: &str, remediation: &str) -> Php { + object(vec![("reason", text(reason)), ("message", text(message)), ("remediation", text(remediation))]) +} + +fn unsupported(field: &str, reason: &str, remediation: &str) -> Php { + object(vec![("field", text(field)), ("reason", text(reason)), ("remediation", text(remediation))]) +} + +/// `opaquePayload`. +fn opaque(value: Option<&Php>) -> Php { + let value = value.filter(|value| !matches!(value, Php::Null)); + object(vec![ + ("available", Php::Bool(value.is_some())), + ("encoding", text("php-serialize")), + ("value", optional(scalar_string(value))), + ("portable", Php::Bool(false)), + ("unsupported_reason", text(OPAQUE_REASON)), + ("remediation", text(OPAQUE_REMEDIATION)), + ]) +} + +/// `mapStatus`. +fn map_status(status: &str) -> Option<&'static str> { + Some(match status { + "created" | "pending" => "pending", + "running" => "running", + "waiting" => "waiting", + "completed" | "continued" => "completed", + "failed" => "failed", + "cancelled" => "cancelled", + "terminated" => "terminated", + _ => return None, + }) +} + +/// `statusBucket`. +fn status_bucket(status: &str) -> &'static str { + match status { + "completed" => "completed", + "failed" | "cancelled" | "terminated" => "failed", + _ => "running", + } +} + +/// `mappedIdentity`. The source id passed its URL-safe pattern, so Laravel's +/// `Str::lower`, `replaceMatches`, `trim('-')` and `limit(32, '')` act on +/// ASCII only. +pub(super) fn mapped_identity(source_id: &str, qualified_id: &str) -> (String, String, String) { + let key = format!("{}\0{}", php_trim(source_id), php_trim(qualified_id)); + let lowered = source_id.to_lowercase(); + let mut slug = String::new(); + let mut replaced = false; + for character in lowered.chars() { + if character.is_ascii_lowercase() || character.is_ascii_digit() || matches!(character, '.' | '_' | '-') { + slug.push(character); + replaced = false; + } else if !replaced { + slug.push('-'); + replaced = true; + } + } + let mut slug: String = slug.trim_matches('-').chars().take(32).collect::().trim_end().to_owned(); + if slug.is_empty() { + slug = "source".into(); + } + ( + format!("v1:{slug}:{}", &sha256(&format!("workflow\0{key}"))[..24]), + sha256(&format!("run\0{key}"))[..26].to_owned(), + sha256(&format!("import\0{key}")), + ) +} + +/// One projected history event before it is stored. +pub(crate) struct Event { + pub(super) id: String, + pub(super) sequence: i64, + pub(super) event_type: &'static str, + pub(super) recorded_at: Option, + pub(super) payload: Php, +} + +/// A valid source, normalized as `LegacyV1Projection::normalize` does. +pub(crate) struct Normalized { + pub(super) namespace: String, + pub(super) workflow_id: String, + pub(super) run_id: String, + pub(super) import_id: String, + pub(super) dedupe_key: String, + pub(super) class: String, + pub(super) connection: Option, + pub(super) queue: Option, + pub(super) status: &'static str, + pub(super) closed_reason: Option, + pub(super) started_at: Option, + pub(super) last_progress_at: Option, + pub(super) closed_at: Option, + pub(super) exception_count: i64, + pub(super) events: Vec, + pub(super) history_size_bytes: i64, + identity: Php, + unsupported: Php, +} + +pub(crate) enum Outcome { + Invalid { errors: Vec }, + /// Larger than one transaction may write. `built` is how many events + /// were built first: 0 for a refusal by count. Only the controls read it. + TooLarge { + error: Php, + #[cfg_attr(not(test), allow(dead_code))] + built: usize, + }, + Valid(Box), +} + +/// `LegacyV1Projection::normalize`, without size limits. +#[cfg(test)] +pub(crate) fn normalize(source_id: &str, source: &PhpArray, namespace: Option<&str>) -> Outcome { + let unlimited = super::StructuralLimits { history_events: None, history_bytes: None, ..super::StructuralLimits::default() }; + normalize_bounded(source_id, source, namespace, &unlimited) +} + +/// `LegacyV1Projection::normalize`, refusing a projection larger than one +/// transaction may write. The event count is known from the source's row +/// counts and is checked before any row is copied or any event is built; +/// history bytes are added up event by event and the first event past the +/// limit stops the measure. +pub(crate) fn normalize_bounded(source_id: &str, source: &PhpArray, namespace: Option<&str>, + limits: &super::StructuralLimits) -> Outcome { + let source_id = php_trim(source_id).to_owned(); + let namespace = namespace.map(php_trim).filter(|namespace| !namespace.is_empty()).unwrap_or("default").to_owned(); + let qualified = string(coalesce(source, &["operator_id", "id"])); + let legacy = scalar_string(source.get("legacy_id")); + let class = string(source.get("class")); + let source_status = string(source.get("status")).unwrap_or_default().to_lowercase(); + let status = map_status(&source_status); + let mut errors = Vec::new(); + let source_id_valid = (1..=64).contains(&source_id.len()) + && source_id.bytes().all(|byte| byte.is_ascii_alphanumeric() || b"._:-".contains(&byte)); + if !source_id_valid { + errors.push(error("invalid_source_id", "source_id must be 1-64 stable URL-safe characters.", + "Use a deployment-stable identifier such as legacy-prod-us-east.")); + } + let qualified_valid = qualified.as_deref() + .and_then(|qualified| qualified.strip_prefix("v1:")) + .is_some_and(|rest| !rest.is_empty() && !rest.contains('\n')); + if !qualified_valid { + errors.push(error("unqualified_v1_identity", "The Waterline workflow must carry operator_id or id in v1: form.", + "Export the detail from the Waterline hybrid migration API.")); + } + if let (Some(legacy), Some(qualified)) = (&legacy, &qualified) + && *qualified != format!("v1:{legacy}") + { + errors.push(error("v1_identity_mismatch", "legacy_id does not match the qualified Waterline identity.", + "Re-export the workflow detail without editing its identity fields.")); + } + if !matches!(source.get("engine_source"), Some(Php::Str(engine)) if engine == "v1") { + errors.push(error("unsupported_source_engine", "Only Waterline engine_source=v1 projections are accepted.", + "Use embedded-v2 import for v2-origin history bundles.")); + } + if class.is_none() { + errors.push(error("missing_v1_workflow_class", "The Waterline projection does not identify the workflow class.", + "Export the complete v1 workflow detail response.")); + } + if status.is_none() { + errors.push(error("unsupported_v1_status", "The v1 workflow status cannot be represented by the standalone v2 visibility model.", + "Complete or normalize the workflow on v1, then export it again.")); + } + let (Some(qualified), Some(class), Some(status), true) = (qualified, class, status, errors.is_empty()) else { + return Outcome::Invalid { errors }; + }; + + // The start event, one event per log, signal and exception, and a + // terminal event when the status has one. + let terminal = matches!(status, "completed" | "failed" | "cancelled" | "terminated"); + let projected = ["logs", "signals", "exceptions"].iter() + .fold(1usize + usize::from(terminal), |total, key| total.saturating_add(row_count(source.get(key)))); + if let Some(error) = exceeded(limits, projected, 0) { + return Outcome::TooLarge { error, built: 0 }; + } + let (workflow_id, run_id, import_id) = mapped_identity(&source_id, &qualified); + let connection = string(source.get("connection")); + let queue = string(source.get("queue")); + let logs = rows(source.get("logs")); + let signals = rows(source.get("signals")); + let exceptions = rows(source.get("exceptions")); + let unsupported = unsupported_fields(source, &logs, &signals, &exceptions, &source_status); + let legacy_workflow_id = legacy.clone().unwrap_or_else(|| qualified[3..].to_owned()); + let waterline = object(vec![ + ("source_id", text(&source_id)), + ("qualified_workflow_id", text(&qualified)), + ("legacy_workflow_id", text(legacy_workflow_id)), + ]); + let identity = object(vec![ + ("waterline", waterline.clone()), + ("standalone", object(vec![ + ("namespace", text(&namespace)), + ("workflow_id", text(&workflow_id)), + ("run_id", text(&run_id)), + ])), + ("relationship", text("deterministic_source_qualified_projection")), + ("collision_policy", text("reject_without_overwrite")), + ]); + let origin = object(vec![ + ("engine_source", text("v1")), + ("engine_version", text(string(source.get("engine_version")).unwrap_or_else(|| "1.x".into()))), + ("execution_engine", text(string(source.get("execution_engine")).unwrap_or_else(|| "finish-on-v1".into()))), + ]); + let task_queue = object(vec![ + ("connection", optional(connection.clone())), + ("name", optional(queue.clone())), + ("execution_owner", text("v1")), + ("standalone_dispatch", text("disabled")), + ]); + let started_at = timestamp(source.get("created_at")); + let last_progress_at = timestamp(source.get("updated_at")).or(started_at); + let closed_at = if terminal { last_progress_at } else { None }; + let raw = |key: &str| source.get(key).cloned().unwrap_or(Php::Null); + let rows_value = |rows: &[PhpArray]| list(rows.iter().cloned().map(Php::Array).collect()); + let for_hash = object(vec![ + ("origin", origin.clone()), + ("identity", waterline), + ("class", text(&class)), + ("status", text(&source_status)), + ("connection", optional(connection.clone())), + ("queue", optional(queue.clone())), + ("arguments", raw("arguments")), + ("output", raw("output")), + ("logs", rows_value(&logs)), + ("signals", rows_value(&signals)), + ("exceptions", rows_value(&exceptions)), + ("created_at", optional(json_time(started_at))), + ("updated_at", optional(json_time(last_progress_at))), + ]); + let dedupe_key = sha256(&for_hash.canonical().encode(Flags { unescaped_slashes: true, ..Flags::default() })); + let migration = object(vec![ + ("origin", origin), + ("identity", identity.clone()), + ("task_queue_context", task_queue), + ("unsupported_fields", unsupported.clone()), + ]); + // Events are built one at a time, and the first past the byte limit + // stops the build. + let (events, history_size_bytes) = match events(source, &logs, &signals, &exceptions, migration, status, started_at, + last_progress_at, &run_id, limits) { + Ok(built) => built, + Err((error, built)) => return Outcome::TooLarge { error, built }, + }; + let history_size_bytes = i64::try_from(history_size_bytes).unwrap_or(i64::MAX); + Outcome::Valid(Box::new(Normalized { + namespace, + workflow_id, + run_id, + import_id, + dedupe_key, + class, + connection, + queue, + status, + closed_reason: terminal.then(|| source_status.clone()), + started_at, + last_progress_at, + closed_at, + exception_count: exceptions.len() as i64, + events, + history_size_bytes, + identity, + unsupported, + })) +} + +/// `unsupportedFields`. +fn unsupported_fields(source: &PhpArray, logs: &[PhpArray], signals: &[PhpArray], exceptions: &[PhpArray], status: &str) -> Php { + let mut fields = vec![ + unsupported("runtime.replay", "v1_history_not_replayable_as_v2", + "Retain the v1 application storage for v1 replay; use the projected typed history for operator inspection only."), + unsupported("timers", "legacy_timer_state_not_exposed_by_waterline", + "Inspect timers in the source application and allow open executions to finish on v1."), + ]; + for field in ["arguments", "output"] { + if source.get(field).is_some_and(|value| !matches!(value, Php::Null)) { + fields.push(unsupported(&format!("payloads.{field}"), OPAQUE_REASON, OPAQUE_REMEDIATION)); + } + } + for (field, rows, key) in [ + ("payloads.activity_results", logs, "result"), + ("payloads.signal_arguments", signals, "arguments"), + ("payloads.exception_values", exceptions, "exception"), + ] { + if rows.iter().any(|row| row.get(key).is_some_and(|value| !matches!(value, Php::Null))) { + fields.push(unsupported(field, OPAQUE_REASON, OPAQUE_REMEDIATION)); + } + } + if !matches!(status, "completed" | "continued" | "failed" | "cancelled" | "terminated") { + fields.push(unsupported("runtime.execution", "finish_on_v1_execution_not_transferable", + "Keep the v1 worker running until completion; the standalone projection is deliberately read-only and never dispatches work.")); + } + list(fields) +} + +/// `events`: started, then each log, signal and exception, then the +/// terminal event, ordered by boundary, whole-second time (untimed first) +/// and source order, with sequence and id assigned after sorting. +#[allow(clippy::too_many_arguments)] +#[cfg(test)] +thread_local! { + /// How many events `events` has built, for the controls that prove a + /// refused build stops at the first event past the limit. + static EVENTS_BUILT: std::cell::Cell = const { std::cell::Cell::new(0) }; +} + +/// Where a projected event comes from. +#[derive(Clone, Copy)] +enum Origin { + Start, + Log(usize), + Signal(usize), + Exception(usize), + Terminal(&'static str), +} + +/// The projected events, ordered as PHP orders them, with their history +/// bytes (`strlen(json_encode($event, JSON_UNESCAPED_SLASHES))` over each +/// event as PHP holds it: type, Carbon time, payload, sequence and id). +/// The order is decided from each event's boundary, time and source order +/// alone; then each event is built and measured in turn, and the first one +/// past the byte limit stops the build with how many were built. +#[allow(clippy::too_many_arguments)] +fn events(source: &PhpArray, logs: &[PhpArray], signals: &[PhpArray], exceptions: &[PhpArray], migration: Php, + status: &str, started_at: Option, last_progress_at: Option, run_id: &str, + limits: &super::StructuralLimits) -> std::result::Result<(Vec, usize), (Php, usize)> { + let field = |row: &PhpArray, key: &str| row.get(key).cloned().unwrap_or(Php::Null); + let terminal = match status { + "completed" => Some("WorkflowCompleted"), + "failed" => Some("WorkflowFailed"), + "cancelled" => Some("WorkflowCancelled"), + "terminated" => Some("WorkflowTerminated"), + _ => None, + }; + let mut order: Vec<(u8, i64, Option, Origin)> = + Vec::with_capacity(2 + logs.len() + signals.len() + exceptions.len()); + order.push((0, 0, started_at, Origin::Start)); + for (index, log) in logs.iter().enumerate() { + order.push((1, 100 + index as i64, timestamp(coalesce(log, &["created_at", "now"])), Origin::Log(index))); + } + for (index, signal) in signals.iter().enumerate() { + order.push((1, 200 + index as i64, timestamp(coalesce(signal, &["created_at", "received_at"])), Origin::Signal(index))); + } + for (index, exception) in exceptions.iter().enumerate() { + order.push((1, 300 + index as i64, timestamp(exception.get("created_at")), Origin::Exception(index))); + } + if let Some(event_type) = terminal { + order.push((2, i64::MAX, last_progress_at, Origin::Terminal(event_type))); + } + // Carbon's getTimestamp: whole seconds; PHP_INT_MIN for an untimed event. + order.sort_by_key(|(boundary, order, at, _)| (*boundary, at.map_or(i64::MIN, |at| at.utc.timestamp()), *order)); + + let mut migration = Some(migration); + let mut built = Vec::with_capacity(order.len()); + let mut bytes: usize = 0; + for (index, (_, _, recorded_at, origin)) in order.into_iter().enumerate() { + let (event_type, payload) = match origin { + Origin::Start => ("WorkflowStarted", object(vec![ + ("migration_projection", migration.take().unwrap_or(Php::Null)), + ("legacy_payloads", object(vec![("arguments", opaque(source.get("arguments")))])), + ])), + Origin::Log(index) => { + let log = &logs[index]; + ("ActivityCompleted", object(vec![ + ("migration_projection", object(vec![ + ("source_record", text("workflow_log")), + ("legacy_id", field(log, "id")), + ("legacy_index", field(log, "index")), + ])), + ("activity", object(vec![ + ("class", field(log, "class")), + ("logical_time", field(log, "now")), + ("result", opaque(log.get("result"))), + ])), + ])) + } + Origin::Signal(index) => { + let signal = &signals[index]; + ("SignalReceived", object(vec![ + ("migration_projection", object(vec![ + ("source_record", text("workflow_signal")), + ("legacy_id", field(signal, "id")), + ])), + ("signal_name", coalesce(signal, &["name", "method"]).cloned().unwrap_or(Php::Null)), + ("arguments", opaque(signal.get("arguments"))), + ])) + } + Origin::Exception(index) => { + let exception = &exceptions[index]; + ("ActivityFailed", object(vec![ + ("migration_projection", object(vec![ + ("source_record", text("workflow_exception")), + ("legacy_id", field(exception, "id")), + ])), + ("exception", object(vec![ + ("class", field(exception, "class")), + ("value", opaque(exception.get("exception"))), + ])), + ])) + } + Origin::Terminal(event_type) => (event_type, object(vec![ + ("migration_projection", object(vec![ + ("source_status", source.get("status").cloned().unwrap_or(Php::Null)), + ("execution_owner", text("v1")), + ])), + ("legacy_payloads", object(vec![("output", opaque(source.get("output")))])), + ])), + }; + #[cfg(test)] + EVENTS_BUILT.with(|count| count.set(count.get() + 1)); + let sequence = index as i64 + 1; + let event = Event { + id: sha256(&format!("{run_id}\0{sequence}\0{event_type}"))[..26].to_owned(), + sequence, + event_type, + recorded_at, + payload, + }; + let size = object(vec![ + ("event_type", text(event.event_type)), + ("recorded_at", optional(json_time(event.recorded_at))), + ("payload", event.payload.clone()), + ("sequence", Php::Int(event.sequence)), + ("id", text(&event.id)), + ]).encode(Flags { unescaped_slashes: true, ..Flags::default() }).len(); + bytes = bytes.checked_add(size).unwrap_or(usize::MAX); + if let Some(error) = exceeded(limits, 0, bytes) { + return Err((error, index + 1)); + } + built.push(event); + } + Ok((built, bytes)) +} + +/// `report`: the projection report, with the first error's reason, message +/// and remediation at the top of a rejection, then `extra`. +fn report(outcome: &Outcome, status: &str, extra: Vec<(&str, Value)>) -> Value { + let (identity, dedupe, unsupported, errors) = match outcome { + Outcome::Invalid { errors } => (Value::Null, Value::Null, json!([]), list(errors.clone()).to_json()), + Outcome::TooLarge { error, .. } => (Value::Null, Value::Null, json!([]), list(vec![error.clone()]).to_json()), + Outcome::Valid(normalized) => (normalized.identity.to_json(), json!(normalized.dedupe_key), + normalized.unsupported.to_json(), json!([])), + }; + let mut body = json!({"schema": REPORT_SCHEMA, "schema_version": VERSION, "status": status, "projection_only": true, + "identity": identity, "dedupe_key": dedupe, "unsupported_fields": unsupported, "errors": errors}); + if status == "rejected" + && let Some(first) = body["errors"].get(0).cloned() + { + for key in ["reason", "message", "remediation"] { + body[key] = first[key].clone(); + } + } + for (key, value) in extra { + body[key] = value; + } + body +} + +/// What the importer needs to know about rows already holding the identity. +pub(crate) struct Existing { + pub(super) run: Option, + pub(super) instance_namespace: Option>, +} + +pub(crate) struct ExistingRun { + pub(super) workflow_instance_id: String, + pub(super) import_source: Option, + pub(super) import_id: Option, + pub(super) dedupe_key: Option, + pub(super) namespace: Option, + pub(super) imported_at: Option>, +} + +/// The validated request: `source_id` required|string|max:64, `workflow` +/// required|array, `dry_run` nullable|boolean, after Laravel's TrimStrings +/// and ConvertEmptyStringsToNull. +fn validate(body: &PhpArray) -> std::result::Result<(String, PhpArray, bool), Value> { + let present = |value: Option<&Php>| match value { + None | Some(Php::Null) => false, + Some(Php::Str(text)) => !text.trim().is_empty(), + Some(Php::Array(array)) => !array.is_empty(), + Some(_) => true, + }; + let mut errors: Vec<(String, Vec)> = Vec::new(); + let source = body.get("source_id"); + let mut source_messages = Vec::new(); + if !present(source) { + source_messages.push("The source id field is required.".to_owned()); + } else { + if !matches!(source, Some(Php::Str(_))) { + source_messages.push("The source id field must be a string.".to_owned()); + } + match source { + Some(Php::Array(array)) if array.len() > 64 => { + source_messages.push("The source id field must not have more than 64 items.".to_owned()); + } + Some(Php::Str(text)) if text.chars().count() > 64 => { + source_messages.push("The source id field must not be greater than 64 characters.".to_owned()); + } + other if !matches!(other, Some(Php::Array(_)) | Some(Php::Str(_))) + && scalar_string(other).is_some_and(|text| text.chars().count() > 64) => + { + source_messages.push("The source id field must not be greater than 64 characters.".to_owned()); + } + _ => {} + } + } + errors.push(("source_id".into(), source_messages)); + let workflow = body.get("workflow"); + let workflow_messages = if !present(workflow) { + vec!["The workflow field is required.".to_owned()] + } else if !matches!(workflow, Some(Php::Array(_))) { + vec!["The workflow field must be an array.".to_owned()] + } else { + Vec::new() + }; + errors.push(("workflow".into(), workflow_messages)); + let dry_run = body.get("dry_run"); + let boolean = match dry_run { + None | Some(Php::Null) => Some(false), + Some(Php::Bool(value)) => Some(*value), + Some(Php::Int(0)) => Some(false), + Some(Php::Int(1)) => Some(true), + Some(Php::Str(text)) if text == "0" => Some(false), + Some(Php::Str(text)) if text == "1" => Some(true), + _ => None, + }; + errors.push(("dry_run".into(), if boolean.is_none() { vec!["The dry run field must be true or false.".to_owned()] } else { Vec::new() })); + if let Some(super::RuntimeError::Protocol { response, .. }) = super::laravel_validation::failure_messages(errors) { + return Err(response); + } + let (Some(Php::Str(source_id)), Some(Php::Array(workflow)), Some(dry_run)) = (source, workflow, boolean) else { + unreachable!("validated request"); + }; + Ok((source_id.clone(), workflow.clone(), dry_run)) +} + +/// A path segment percent-decoded, as the router decodes its parameters. +fn decoded(segment: &str) -> Option { + let bytes = segment.as_bytes(); + let mut out = Vec::with_capacity(bytes.len()); + let mut index = 0; + while index < bytes.len() { + if bytes[index] == b'%' { + let hex = std::str::from_utf8(bytes.get(index + 1..index + 3)?).ok()?; + out.push(u8::from_str_radix(hex, 16).ok()?); + index += 3; + } else { + out.push(bytes[index]); + index += 1; + } + } + String::from_utf8(out).ok() +} + +/// `projectedV1OperationResponse` for the control operations PHP refuses +/// on a projection before reading their bodies: signal, update, cancel, +/// terminate, repair and archive check the instance's current run; redrive +/// checks its source run. A query is refused once its `input` passes +/// validation, as PHP validates it first. Cooperative cancellation refuses +/// in its own transaction. +pub(super) async fn read_only_guard(request: axum::extract::Request, next: axum::middleware::Next) -> axum::response::Response { + use axum::response::IntoResponse; + use super::control_plane::Operation; + if request.method() != axum::http::Method::POST { + return next.run(request).await; + } + let path = request.uri().path().to_owned(); + let segments: Vec<&str> = path.trim_start_matches('/').split('/').collect(); + let (workflow, run, operation, name) = match segments.as_slice() { + ["api", "workflows", workflow, operation @ ("cancel" | "terminate" | "repair" | "archive")] => (*workflow, None, *operation, None), + ["api", "workflows", workflow, operation @ ("signal" | "query" | "update"), name] => (*workflow, None, *operation, Some(*name)), + ["api", "workflows", workflow, "runs", run, operation @ ("cancel" | "terminate" | "repair" | "archive" | "redrive")] => { + (*workflow, Some(*run), *operation, None) + } + ["api", "workflows", workflow, "runs", run, operation @ ("signal" | "query" | "update"), name] => { + (*workflow, Some(*run), *operation, Some(*name)) + } + // Refused before its body is read: an append would deliver a signal + // and create a workflow task for the projection. + ["api", "workflows", workflow, "message-streams", stream, "messages"] => { + (*workflow, None, "message_stream_append", Some(*stream)) + } + _ => return next.run(request).await, + }; + let (Some(runtime), Some(workflow)) = (request.extensions().get::().cloned(), decoded(workflow)) else { + return next.run(request).await; + }; + let name = name.and_then(decoded).unwrap_or_default(); + let projected = match (operation, run.and_then(decoded)) { + ("redrive", Some(run)) => runtime.projected_run(&workflow, &run).await, + _ => runtime.projected_current_run(&workflow).await, + }; + let Ok(Some(run_id)) = projected else { + return next.run(request).await; + }; + let request = if operation == "query" { + // `input` nullable|array: an invalid input answers its validation error. + let (parts, body) = request.into_parts(); + let Ok(bytes) = axum::body::to_bytes(body, usize::MAX).await else { + return next.run(axum::extract::Request::from_parts(parts, axum::body::Body::empty())).await; + }; + let input_valid = match super::php_value::request_body(&bytes).get("input") { + None | Some(Php::Null | Php::Array(_)) => true, + Some(_) => false, + }; + let request = axum::extract::Request::from_parts(parts, axum::body::Body::from(bytes)); + if !input_valid { + return next.run(request).await; + } + request + } else { + request + }; + drop(request); + if operation == "message_stream_append" { + return (StatusCode::CONFLICT, axum::Json(read_only_append(&workflow, &run_id, &name))).into_response(); + } + let body = read_only_body(&workflow, &run_id, operation); + let body = match operation { + "signal" => Operation::Signal(&name).response(body), + "query" => Operation::Query(&name).response(body), + "update" => Operation::Update(&name).response(body), + "cancel" => Operation::Cancel.response(body), + "terminate" => Operation::Terminate.response(body), + "repair" => Operation::Repair.response(body), + "archive" => Operation::Archive.response(body), + _ => body, + }; + (StatusCode::CONFLICT, axum::Json(body)).into_response() +} + +impl Runtime { + /// `LegacyV1ProjectionController::store` once the gates have passed. + pub(crate) async fn import_waterline_v1(&self, body: &[u8]) -> Result<(StatusCode, Value)> { + let operation = || super::control_plane::Operation::ImportWaterlineV1; + let request = super::php_value::request_body(body); + let (source_id, workflow, dry_run) = match validate(&request) { + Ok(validated) => validated, + Err(response) => return Ok((StatusCode::UNPROCESSABLE_ENTITY, operation().response(response))), + }; + let namespace = self.namespace_name(); + // Hashing and JSON encoding of the whole source stay off the async workers. + let limits = self.structural; + let outcome = tokio::task::spawn_blocking(move || normalize_bounded(&source_id, &workflow, Some(&namespace), &limits)) + .await + .map_err(|_| super::refuse(StatusCode::SERVICE_UNAVAILABLE, "waterline_import_unavailable"))?; + let answer = |status: &str, extra: Vec<(&str, Value)>| { + let code = match status { + "projected" => StatusCode::CREATED, + "already_projected" | "dry_run" => StatusCode::OK, + _ => StatusCode::CONFLICT, + }; + (code, operation().response(report(&outcome, status, extra))) + }; + // A projection too large for one transaction is refused before any + // row is read or written. + if let Outcome::TooLarge { error, .. } = &outcome { + return Ok(answer("rejected", vec![("limit", error.to_json()["limit"].clone())])); + } + let Outcome::Valid(normalized) = &outcome else { + return Ok(answer("rejected", Vec::new())); + }; + let existing = self.waterline_existing(&normalized.workflow_id, &normalized.run_id).await?; + if existing.run.is_some() || existing.instance_namespace.is_some() { + let same = existing.run.as_ref().is_some_and(|run| existing.instance_namespace.is_some() + && run.workflow_instance_id == normalized.workflow_id + && run.import_source.as_deref() == Some(IMPORT_SOURCE) + && run.import_id.as_deref() == Some(normalized.import_id.as_str()) + && run.dedupe_key.as_deref() == Some(normalized.dedupe_key.as_str())); + let instance_namespace = existing.instance_namespace.clone().flatten(); + let run_namespace = existing.run.as_ref().and_then(|run| run.namespace.clone()); + if same && run_namespace.as_deref() == Some(normalized.namespace.as_str()) + && instance_namespace.as_deref() == Some(normalized.namespace.as_str()) + { + let imported = existing.run.as_ref().and_then(|run| run.imported_at); + return Ok(answer("already_projected", vec![("imported_at", super::task_queues::json_time(imported))])); + } + if same { + let stored = instance_namespace.filter(|namespace| Some(namespace) == run_namespace.as_ref()); + let mut identity = normalized.identity.to_json(); + identity["standalone"]["namespace"] = json!(stored); + let (reason, message, remediation) = match &stored { + None => ("v1_projection_storage_inconsistent".to_owned(), + "The existing v1 projection has inconsistent instance and run namespaces.".to_owned(), + "Repair or remove the inconsistent projection rows before repeating the import.".to_owned()), + Some(stored) => ("v1_projection_namespace_collision".to_owned(), + "This v1 identity is already projected in a different standalone namespace.".to_owned(), + format!("Repeat the operation in the existing {stored} namespace, or choose a different stable source_id for an independent projection.")), + }; + return Ok(answer("rejected", vec![("identity", identity), ("requested_namespace", json!(normalized.namespace)), + ("reason", json!(reason)), ("message", json!(message)), ("remediation", json!(remediation))])); + } + let changed = existing.run.as_ref().is_some_and(|run| run.import_source.as_deref() == Some(IMPORT_SOURCE)); + return Ok(answer("rejected", if changed { + vec![("reason", json!("v1_projection_changed")), + ("message", json!("This v1 identity was already projected from different source state.")), + ("remediation", json!("Inspect the existing projection and resolve it before importing a different snapshot of the same v1 identity."))] + } else { + vec![("reason", json!("native_v2_identity_collision")), + ("message", json!("The mapped standalone identity is already owned by a native or unrelated v2 execution.")), + ("remediation", json!("Choose a different stable source_id; the server will not overwrite the colliding v2 execution."))] + })); + } + if dry_run { + return Ok(answer("dry_run", Vec::new())); + } + let imported_at = now(); + Ok(match self.waterline_write(normalized, imported_at).await { + Ok(()) => answer("projected", vec![("imported_at", super::task_queues::json_time(Some(imported_at)))]), + Err(collision) => { + let (reason, message, remediation) = if collision { + ("native_v2_identity_collision", "The mapped standalone identity was claimed while the projection was being written.", + "Retry with a different stable source_id; no existing execution was overwritten.") + } else { + ("target_write_failed", "The standalone server could not persist the v1 projection transaction.", + "Verify server storage readiness, then repeat the same idempotent operation.") + }; + answer("rejected", vec![("reason", json!(reason)), ("message", json!(message)), ("remediation", json!(remediation))]) + } + }) + } +} + +store_methods! { + /// `terminalFailurePayload` for a failed projection, which has no failure + /// snapshots: its activity failure summaries and their last message, or + /// null when it has none. + pub(crate) async fn projected_failure(&self, run_id: &str) -> Result { + let rows = Self::query("SELECT sequence,event_type,payload FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type IN ('ActivityFailed','ActivityTimedOut') ORDER BY sequence") + .bind(run_id) + .fetch_all(&self.pool) + .await?; + let mut summaries = Vec::with_capacity(rows.len()); + for row in &rows { + let payload = DB::document_row(row, "payload")?; + let mut summary = serde_json::Map::new(); + summary.insert("event_sequence".into(), json!(DB::number(row, "sequence")?)); + summary.insert("event_type".into(), json!(DB::string(row, "event_type")?)); + for field in ["activity_type", "activity_class", "activity_execution_id", "activity_attempt_id", "failure_id", + "failure_category", "exception_type", "exception_class", "message"] { + if let Some(text) = payload[field].as_str().filter(|text| !text.is_empty()) { + summary.insert(field.into(), json!(text)); + } + } + if !payload["non_retryable"].is_null() { + summary.insert("non_retryable".into(), payload["non_retryable"].clone()); + } + summaries.push(Value::Object(summary)); + } + if summaries.is_empty() { + return Ok(Value::Null); + } + let message = summaries.iter().rev().find_map(|summary| summary["message"].as_str().map(str::to_owned)); + let mut failure = serde_json::Map::new(); + if let Some(message) = message { + failure.insert("message".into(), json!(message)); + } + failure.insert("activity_failures".into(), Value::Array(summaries)); + Ok(Value::Object(failure)) + } + + /// `LegacyV1Projection::metadataForRun` for a projected run: the started + /// event's projection context and the projection's own description. + pub(crate) async fn projection_metadata(&self, run_id: &str, imported_at: Option>, dedupe_key: Option) -> Result { + let started = Self::query("SELECT payload FROM workflow_history_events WHERE workflow_run_id=$1 ORDER BY sequence LIMIT 1") + .bind(run_id) + .fetch_optional(&self.pool) + .await? + .map(|row| DB::document_row(&row, "payload")) + .transpose()? + .unwrap_or(Value::Null); + let migration = &started["migration_projection"]; + let unsupported = migration["unsupported_fields"].as_array().cloned().unwrap_or_default(); + Ok(json!({ + "origin": migration["origin"], + "identity": migration["identity"], + "task_queue_context": migration["task_queue_context"], + "unsupported_fields": unsupported, + "projection": {"schema": SCHEMA, "version": VERSION, "read_only": true, "execution_owner": "v1", + "imported_at": super::task_queues::json_time(imported_at), "dedupe_key": dedupe_key}, + })) + } + + /// The current run of an instance in this namespace when it is a + /// Waterline v1 projection (`NamespaceWorkflowScope::currentRun`). + pub(crate) async fn projected_current_run(&self, workflow_id: &str) -> Result> { + Self::query("SELECT r.id FROM workflow_instances i JOIN workflow_runs r ON r.id=i.current_run_id WHERE i.id=$1 AND i.namespace=$2 AND r.import_source=$3") + .bind(workflow_id).bind(self.namespace.as_ref()).bind(IMPORT_SOURCE) + .fetch_optional(&self.pool).await? + .map(|row| DB::string(&row, "id")).transpose() + } + + /// A selected run of this namespace when it is a Waterline v1 projection. + pub(crate) async fn projected_run(&self, workflow_id: &str, run_id: &str) -> Result> { + Self::query("SELECT id FROM workflow_runs WHERE id=$1 AND workflow_instance_id=$2 AND namespace=$3 AND import_source=$4") + .bind(run_id).bind(workflow_id).bind(self.namespace.as_ref()).bind(IMPORT_SOURCE) + .fetch_optional(&self.pool).await? + .map(|row| DB::string(&row, "id")).transpose() + } + + /// The run and instance already holding a mapped identity, in any + /// namespace: workflow and run ids are global. + pub(crate) async fn waterline_existing(&self, workflow_id: &str, run_id: &str) -> Result { + let run = Self::query("SELECT workflow_instance_id,import_source,import_id,import_dedupe_key,namespace,imported_at FROM workflow_runs WHERE id=$1") + .bind(run_id) + .fetch_optional(&self.pool) + .await? + .map(|row| -> Result { + Ok(ExistingRun { + workflow_instance_id: DB::string(&row, "workflow_instance_id")?, + import_source: DB::optional_string(&row, "import_source")?, + import_id: DB::optional_string(&row, "import_id")?, + dedupe_key: DB::optional_string(&row, "import_dedupe_key")?, + namespace: DB::optional_string(&row, "namespace")?, + imported_at: DB::optional_instant(&row, "imported_at")?, + }) + }) + .transpose()?; + let instance_namespace = Self::query("SELECT namespace FROM workflow_instances WHERE id=$1") + .bind(workflow_id) + .fetch_optional(&self.pool) + .await? + .map(|row| DB::optional_string(&row, "namespace")) + .transpose()?; + Ok(Existing { run, instance_namespace }) + } + + /// The projection's instance, run, history and summary in one + /// transaction. `Err(true)` when the identity was claimed meanwhile. + pub(crate) async fn waterline_write(&self, projection: &Normalized, imported_at: DateTime) -> std::result::Result<(), bool> { + #[cfg(test)] + install_probe::reached(&projection.workflow_id); + let mut tx = self.begin().await.map_err(|_| false)?; + let claimed = Self::query("SELECT id FROM workflow_instances WHERE id=$1") + .bind(&projection.workflow_id).fetch_optional(&mut *tx).await.map_err(|_| false)?.is_some() + || Self::query("SELECT id FROM workflow_runs WHERE id=$1") + .bind(&projection.run_id).fetch_optional(&mut *tx).await.map_err(|_| false)?.is_some(); + if claimed { + return Err(true); + } + // Eloquent stores ordinary date columns as the parsed time's own + // wall-clock time; history times go through the UTC cast. + let wall = |stamp: Option| stamp.map(|stamp| stamp.wall); + let created = wall(projection.started_at).unwrap_or(imported_at); + let empty = json!([]); + Self::query("INSERT INTO workflow_instances(id,workflow_class,workflow_type,namespace,visibility_labels,memo,current_run_id,run_count,last_message_sequence,started_at,created_at,updated_at) VALUES ($1,$2,$3,$4,$5,$6,$7,1,0,$8,$9,$10)") + .bind(&projection.workflow_id).bind(&projection.class).bind(&projection.class).bind(&projection.namespace) + .bind(DB::document(&empty)).bind(DB::document(&empty)).bind(&projection.run_id) + .bind(DB::optional_time(wall(projection.started_at))).bind(DB::bind_time(created)).bind(DB::bind_time(imported_at)) + .execute(&mut *tx).await.map_err(|_| false)?; + Self::query("INSERT INTO workflow_runs(id,workflow_instance_id,run_number,workflow_class,workflow_type,namespace,visibility_labels,status,closed_reason,payload_codec,arguments,output,connection,queue,last_history_sequence,last_command_sequence,message_cursor_position,started_at,closed_at,last_progress_at,import_source,import_id,import_dedupe_key,import_contract_version,imported_at,created_at,updated_at) VALUES ($1,$2,1,$3,$4,$5,$6,$7,$8,NULL,NULL,NULL,$9,$10,$11,0,0,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21)") + .bind(&projection.run_id).bind(&projection.workflow_id).bind(&projection.class).bind(&projection.class) + .bind(&projection.namespace).bind(DB::document(&empty)).bind(projection.status).bind(projection.closed_reason.as_deref()) + .bind(projection.connection.as_deref()).bind(projection.queue.as_deref()).bind(projection.events.len() as i64) + .bind(DB::optional_time(wall(projection.started_at))).bind(DB::optional_time(wall(projection.closed_at))) + .bind(DB::optional_time(wall(projection.last_progress_at))).bind(IMPORT_SOURCE).bind(&projection.import_id) + .bind(&projection.dedupe_key).bind(VERSION).bind(DB::bind_time(imported_at)).bind(DB::bind_time(created)) + .bind(DB::bind_time(imported_at)) + .execute(&mut *tx).await.map_err(|_| false)?; + for event in &projection.events { + let stored = wall(event.recorded_at).unwrap_or(imported_at); + Self::query("INSERT INTO workflow_history_events(id,workflow_run_id,sequence,event_type,payload,recorded_at,recorded_at_utc,created_at,updated_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9)") + .bind(&event.id).bind(&projection.run_id).bind(event.sequence).bind(event.event_type) + .bind(DB::document(&event.payload.to_json())) + .bind(DB::optional_time(event.recorded_at.map(|at| at.utc))) + .bind(DB::optional_time(event.recorded_at.map(|at| at.utc))) + .bind(DB::bind_time(stored)).bind(DB::bind_time(stored)) + .execute(&mut *tx).await.map_err(|_| false)?; + } + let bucket = status_bucket(projection.status); + let running = bucket == "running"; + let sort = wall(projection.last_progress_at).or(wall(projection.started_at)).unwrap_or(imported_at); + Self::query("INSERT INTO workflow_run_summaries(id,workflow_instance_id,run_number,is_current_run,engine_source,class,workflow_type,namespace,visibility_labels,status,status_bucket,closed_reason,connection,queue,started_at,sort_timestamp,closed_at,liveness_state,liveness_reason,repair_blocked_reason,repair_attention,task_problem,exception_count,history_event_count,history_size_bytes,continue_as_new_recommended,created_at,updated_at) VALUES ($1,$2,1,TRUE,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,'v1_projection_read_only',FALSE,FALSE,$18,$19,$20,FALSE,$21,$22)") + .bind(&projection.run_id).bind(&projection.workflow_id).bind(ENGINE_SOURCE) + .bind(&projection.class).bind(&projection.class).bind(&projection.namespace).bind(DB::document(&empty)) + .bind(projection.status).bind(bucket).bind(projection.closed_reason.as_deref()) + .bind(projection.connection.as_deref()).bind(projection.queue.as_deref()) + .bind(DB::optional_time(wall(projection.started_at))).bind(DB::bind_time(sort)) + .bind(DB::optional_time(wall(projection.closed_at))) + .bind(running.then_some("external_v1")) + .bind(running.then_some("Execution remains owned by the source v1 runtime; this standalone row is read-only.")) + .bind(projection.exception_count).bind(projection.events.len() as i64).bind(projection.history_size_bytes) + .bind(DB::bind_time(created)).bind(DB::bind_time(imported_at)) + .execute(&mut *tx).await.map_err(|_| false)?; + tx.commit().await.map_err(|_| false)?; + self.wake.notify_waiters(); + Ok(()) + } +} + +/// The point where a projection's install asks for the transition lock, +/// for the controls that prove an append cannot cross an install waiting on +/// it. Keyed by workflow id, so tests running in parallel do not see each +/// other's installs. +#[cfg(test)] +mod install_probe { + use std::collections::HashMap; + use std::sync::{Arc, Mutex, OnceLock}; + use tokio::sync::Notify; + + fn watched() -> &'static Mutex>> { + static WATCHED: OnceLock>>> = OnceLock::new(); + WATCHED.get_or_init(Mutex::default) + } + + /// Notified once the install of `workflow_id` reaches the transition lock. + pub(super) fn watch(workflow_id: &str) -> Arc { + let notify = Arc::new(Notify::new()); + watched().lock().unwrap().insert(workflow_id.to_owned(), notify.clone()); + notify + } + + pub(super) fn reached(workflow_id: &str) { + if let Some(notify) = watched().lock().unwrap().get(workflow_id) { + notify.notify_one(); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn direct_appends_and_malformed_counts_never_touch_a_projection() { + let directory = tempfile::tempdir().unwrap(); + let runtime = super::super::Runtime::open(directory.path().join("projection.sqlite").to_str().unwrap(), "test-token".into()) + .await + .unwrap(); + let source = json!({"id": "v1:4201", "legacy_id": "4201", "engine_source": "v1", "class": "App\\Workflows\\Legacy\\Subscription", + "status": "running", "arguments": "a:0:{}", "logs": [{"id": 9101, "index": 0, "class": "App\\Activities\\Bill", + "result": "i:42;", "created_at": "2026-09-30T09:00:02.000000Z"}], "created_at": "2026-09-30T09:00:00.000000Z"}); + let body = serde_json::to_vec(&json!({"source_id": "guard-src", "workflow": source})).unwrap(); + let (status, report) = runtime.import_waterline_v1(&body).await.unwrap(); + assert_eq!(status, StatusCode::CREATED, "{report}"); + let workflow_id = report["identity"]["standalone"]["workflow_id"].as_str().unwrap().to_owned(); + let run_id = report["identity"]["standalone"]["run_id"].clone(); + let pool = runtime.sqlite_pool(); + let snapshot = || async { + let mut counts = Vec::new(); + for table in ["workflow_inbound_streams", "workflow_inbound_stream_items", "workflow_signal_records", "workflow_commands", + "workflow_history_events", "workflow_tasks", "workflow_run_summaries"] { + let count: i64 = sqlx::query_scalar(sqlx::AssertSqlSafe(format!("SELECT COUNT(*) FROM {table}"))).fetch_one(pool).await.unwrap(); + counts.push((table, count)); + } + let summary: Option = sqlx::query_scalar("SELECT CAST(updated_at AS TEXT) FROM workflow_run_summaries") + .fetch_one(pool).await.unwrap(); + (counts, summary) + }; + let before = snapshot().await; + + // The service refuses before reading the body, valid or not. + for body in [json!({"message_id": "m-1", "input": ["x"]}), json!({"message_id": 7, "input": "not a list"})] { + match runtime.append_message(&workflow_id, "orders", body, json!({})).await { + Err(super::super::RuntimeError::Protocol { status, response }) => { + assert_eq!(status, StatusCode::CONFLICT, "{response}"); + assert_eq!((&response["reason"], &response["run_id"]), (&json!("v1_projection_read_only"), &run_id), "{response}"); + assert_eq!(response["operation"], "message_stream_append"); + } + other => panic!("{other:?}"), + } + } + // The store's own guard refuses under its lock, and malformed input + // is not counted against a projection. + match runtime.store_append_message(&workflow_id, "orders", "m-2", "wwHioz3/VYAiNwA=").await { + Err(super::super::RuntimeError::Protocol { status, response }) => { + assert_eq!((status, &response["reason"]), (StatusCode::CONFLICT, &json!("v1_projection_read_only")), "{response}"); + } + other => panic!("{other:?}"), + } + runtime.record_malformed_message(&workflow_id, "orders", &json!("m-3")).await.unwrap(); + + assert_eq!(snapshot().await, before); + runtime.close().await; + } + + /// SQLite's writer waits inside its own busy handler, which no other + /// connection can see; the install probe shows it reached the transition + /// lock instead. With `BEGIN IMMEDIATE` held, an append finds no instance + /// and writes nothing; once the install commits, an append is refused. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn an_append_cannot_cross_an_install_waiting_on_the_sqlite_lock() { + use sqlx::Connection; + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("barrier.sqlite"); + let runtime = super::super::Runtime::open(path.to_str().unwrap(), "test-token".into()).await.unwrap(); + let source = json!({"id": "v1:4201", "legacy_id": "4201", "engine_source": "v1", "class": "App\\Workflows\\Legacy\\Subscription", + "status": "running", "arguments": "a:0:{}", "logs": [{"id": 9101, "index": 0, "class": "App\\Activities\\Bill", + "result": "i:42;", "created_at": "2026-09-30T09:00:02.000000Z"}], "created_at": "2026-09-30T09:00:00.000000Z"}); + let body = |dry_run: bool| serde_json::to_vec(&json!({"source_id": "barrier-src", "workflow": source, "dry_run": dry_run})).unwrap(); + let (status, dry) = runtime.import_waterline_v1(&body(true)).await.unwrap(); + assert_eq!(status, StatusCode::OK, "{dry}"); + let workflow_id = dry["identity"]["standalone"]["workflow_id"].as_str().unwrap().to_owned(); + let pool = runtime.sqlite_pool().clone(); + let counts = || async { + let tables: Vec = sqlx::query_scalar("SELECT name FROM sqlite_master WHERE type='table' ORDER BY name") + .fetch_all(&pool).await.unwrap(); + let mut counts = Vec::new(); + for table in tables { + let count: i64 = sqlx::query_scalar(sqlx::AssertSqlSafe(format!("SELECT COUNT(*) FROM {table}"))).fetch_one(&pool).await.unwrap(); + counts.push((table, count)); + } + counts + }; + let before = counts().await; + let append = |message: &str| json!({"message_id": message, "input": ["x"]}); + + let reached = install_probe::watch(&workflow_id); + let mut holder = sqlx::SqliteConnection::connect(&format!("sqlite://{}", path.display())).await.unwrap(); + sqlx::query("BEGIN IMMEDIATE").execute(&mut holder).await.unwrap(); + let install = tokio::spawn({ + let runtime = runtime.clone(); + let body = body(false); + async move { runtime.import_waterline_v1(&body).await } + }); + tokio::time::timeout(std::time::Duration::from_secs(10), reached.notified()).await + .expect("the install never reached the transition lock"); + + // The install is at its transaction and cannot commit. + match runtime.append_message(&workflow_id, "orders", append("m-pending"), json!({})).await { + Err(super::super::RuntimeError::Protocol { status, response }) => { + assert_eq!((status, &response["reason"]), (StatusCode::NOT_FOUND, &json!("instance_not_found")), "{response}"); + } + other => panic!("{other:?}"), + } + assert!(!install.is_finished(), "the install committed while BEGIN IMMEDIATE was held"); + sqlx::query("ROLLBACK").execute(&mut holder).await.unwrap(); + + let (status, report) = install.await.unwrap().unwrap(); + assert_eq!(status, StatusCode::CREATED, "{report}"); + match runtime.append_message(&workflow_id, "orders", append("m-after"), json!({})).await { + Err(super::super::RuntimeError::Protocol { status, response }) => { + assert_eq!((status, &response["reason"]), (StatusCode::CONFLICT, &json!("v1_projection_read_only")), "{response}"); + } + other => panic!("{other:?}"), + } + // Only the projection's rows were written. + let after = counts().await; + assert_eq!(before.len(), after.len()); + for ((table, was), (_, now)) in before.iter().zip(&after) { + let grew = match table.as_str() { + "workflow_instances" | "workflow_runs" | "workflow_run_summaries" => 1, + "workflow_history_events" => 2, + _ => 0, + }; + assert_eq!(*now, was + grew, "{table}"); + } + runtime.close().await; + } + + #[test] + fn size_limits_name_the_first_dimension_exceeded() { + let limits = |events, bytes| super::super::StructuralLimits { history_events: events, history_bytes: bytes, + ..super::super::StructuralLimits::default() }; + assert!(exceeded(&limits(Some(3), Some(100)), 3, 100).is_none()); + let refusal = exceeded(&limits(Some(3), Some(100)), 4, 101).unwrap().to_json(); + assert_eq!((&refusal["reason"], &refusal["limit"]), (&json!("v1_projection_too_large"), + &json!({"dimension": "history_transaction_size", "used": 4, "limit": 3}))); + let refusal = exceeded(&limits(Some(3), Some(100)), 3, 101).unwrap().to_json(); + assert_eq!(refusal["limit"]["dimension"], "history_transaction_bytes"); + // 0 disables a limit. + assert!(exceeded(&limits(Some(0), Some(0)), usize::MAX, usize::MAX).is_none()); + } + + #[test] + fn mapped_identities_follow_php() { + // PHP: LegacyV1Projection::mappedIdentity('legacy-prod', 'v1:42'). + let (workflow, run, import) = mapped_identity("legacy-prod", "v1:42"); + assert!(workflow.starts_with("v1:legacy-prod:") && workflow.len() == "v1:legacy-prod:".len() + 24); + assert_eq!(run.len(), 26); + assert_eq!(import.len(), 64); + assert_eq!(mapped_identity("Legacy:Prod..X", "v1:1").0.split(':').nth(1), Some("legacy-prod..x")); + assert_eq!(mapped_identity(&"a".repeat(40), "v1:1").0.split(':').nth(1).map(str::len), Some(32)); + assert_eq!(mapped_identity("::", "v1:1").0.split(':').nth(1), Some("source")); + } + + /// A completed source with `logs` logs, parsed as the request body is. + fn completed_source(logs: usize) -> Php { + let logs: Vec = (0..logs).map(|index| json!({"id": index, "class": "App\\Activities\\Step", + "result": "s:4:\"done\";", "created_at": "2026-09-30T08:00:01.000000Z"})).collect(); + let body = json!({"operator_id": "v1:4501", "engine_source": "v1", "class": "App\\Workflows\\Legacy\\Bare", + "status": "completed", "logs": logs, "signals": [{"id": 1, "name": "go"}], "exceptions": "not rows"}); + super::super::php_value::decode(body.to_string().as_bytes()).unwrap() + } + + fn limited(events: Option, bytes: Option) -> super::super::StructuralLimits { + super::super::StructuralLimits { history_events: events, history_bytes: bytes, ..super::super::StructuralLimits::default() } + } + + /// Events built by `events` while `run` runs on this thread. + fn events_built(run: impl FnOnce() -> T) -> (T, usize) { + EVENTS_BUILT.with(|count| count.set(0)); + let result = run(); + (result, EVENTS_BUILT.with(std::cell::Cell::get)) + } + + #[test] + fn the_event_count_is_refused_before_any_event_is_built() { + let source = completed_source(10); + let source = source.array().unwrap(); + // Start, ten logs, one signal and the terminal event; a non-row + // exceptions value counts nothing. + let (outcome, built) = events_built(|| normalize_bounded("src-1", source, None, &limited(Some(13), None))); + let Outcome::Valid(normalized) = outcome else { panic!("exact count") }; + assert_eq!((normalized.events.len(), built), (13, 13)); + let (outcome, built) = events_built(|| normalize_bounded("src-1", source, None, &limited(Some(12), None))); + match outcome { + Outcome::TooLarge { error, built: reported } => { + // Refused from the row counts: no event built. + assert_eq!((reported, built), (0, 0)); + assert_eq!(error.to_json()["limit"], json!({"dimension": "history_transaction_size", "used": 13, "limit": 12})); + } + _ => panic!("one over the event limit"), + } + // An invalid source still answers its own errors first. + let invalid = super::super::php_value::decode(br#"{"engine_source":"v1","logs":[{},{},{}]}"#).unwrap(); + assert!(matches!(normalize_bounded("src-1", invalid.array().unwrap(), None, &limited(Some(1), None)), Outcome::Invalid { .. })); + } + + #[test] + fn history_bytes_stop_the_build_at_the_first_event_over_the_limit() { + let source = completed_source(10); + let source = source.array().unwrap(); + let Outcome::Valid(normalized) = normalize_bounded("src-1", source, None, &limited(None, None)) else { panic!("valid") }; + let total = usize::try_from(normalized.history_size_bytes).unwrap(); + let Outcome::Valid(exact) = normalize_bounded("src-1", source, None, &limited(None, Some(total))) else { panic!("exact bytes") }; + assert_eq!(exact.history_size_bytes, normalized.history_size_bytes); + // One byte short: every event is built, and the last passes the limit. + match normalize_bounded("src-1", source, None, &limited(None, Some(total - 1))) { + Outcome::TooLarge { error, built } => { + assert_eq!(built, normalized.events.len()); + assert_eq!(error.to_json()["limit"], json!({"dimension": "history_transaction_bytes", "used": total, "limit": total - 1})); + } + _ => panic!("one byte over"), + } + } + + #[test] + fn an_early_event_over_the_byte_limit_builds_no_later_event() { + // The start event is first and carries the source's arguments; many + // valid logs follow it. + let mut source = completed_source(2_000); + if let Php::Array(array) = &mut source { + array.0.push((Key::of("arguments"), text("x".repeat(50_000)))); + } + let source = source.array().unwrap(); + let (outcome, built) = events_built(|| normalize_bounded("src-1", source, None, &limited(None, Some(10_000)))); + match outcome { + Outcome::TooLarge { error, built: reported } => { + // Only the start event was built; none of the 2,002 after it. + assert_eq!((reported, built), (1, 1)); + assert_eq!(error.to_json()["limit"]["dimension"], "history_transaction_bytes"); + } + _ => panic!("the start event alone passes the limit"), + } + // With room for the start event and a few more, the build stops at + // the first event past the limit. + let (outcome, start) = events_built(|| normalize_bounded("src-1", source, None, &limited(None, None))); + let Outcome::Valid(all) = outcome else { panic!("valid") }; + assert_eq!(start, all.events.len()); + let first = |count: usize| all.events[..count].iter().map(|event| object(vec![ + ("event_type", text(event.event_type)), + ("recorded_at", optional(json_time(event.recorded_at))), + ("payload", event.payload.clone()), + ("sequence", Php::Int(event.sequence)), + ("id", text(&event.id)), + ]).encode(Flags { unescaped_slashes: true, ..Flags::default() }).len()).sum::(); + let (outcome, built) = events_built(|| normalize_bounded("src-1", source, None, &limited(None, Some(first(4))))); + assert!(matches!(outcome, Outcome::TooLarge { built: 5, .. })); + assert_eq!(built, 5); + } + + #[test] + fn events_and_history_size_follow_php() { + // PHP 2.11.1: LegacyV1Projection::normalize('src-1', bare source, 'ns-1') + // printed these event encodings (json_encode, JSON_UNESCAPED_SLASHES). + let source = super::super::php_value::decode(br#"{"operator_id":"v1:4501","engine_source":"v1","class":"App\\Workflows\\Legacy\\Bare","status":"completed"}"#) + .unwrap(); + let Outcome::Valid(normalized) = normalize("src-1", source.array().unwrap(), Some("ns-1")) else { panic!("valid") }; + let encoded: Vec = normalized.events.iter().map(|event| object(vec![ + ("event_type", text(event.event_type)), + ("recorded_at", optional(json_time(event.recorded_at))), + ("payload", event.payload.clone()), + ("sequence", Php::Int(event.sequence)), + ("id", text(&event.id)), + ]).encode(Flags { unescaped_slashes: true, ..Flags::default() })).collect(); + assert_eq!(encoded[0].len(), 1374, "{}", encoded[0]); + assert_eq!(encoded[1].len(), 496, "{}", encoded[1]); + assert_eq!(normalized.history_size_bytes, 1870); + assert_eq!(normalized.workflow_id, "v1:src-1:c8c25815a3332910ba072598"); + assert_eq!(normalized.run_id, "b0ee56b2b89080ea9181490e2f"); + assert_eq!(normalized.events[0].id, "9730c0bb4149e0a120cc8a8d7e"); + } + + #[test] + fn floats_cast_to_strings_as_php() { + assert_eq!(float_string(4101.0), "4101"); + assert_eq!(float_string(1.5), "1.5"); + assert_eq!(float_string(0.1 + 0.2), "0.3"); + assert_eq!(float_string(1e25), "1.0E+25"); + assert_eq!(float_string(-0.00001), "-1.0E-5"); + assert_eq!(float_string(0.0001), "0.0001"); + } +} diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs index 8d41575e4..c86070a05 100644 --- a/rust/tests/execution.rs +++ b/rust/tests/execution.rs @@ -204,6 +204,8 @@ mod command_requests; mod describe_actions; #[path = "execution/run_deadlines.rs"] mod run_deadlines; +#[path = "execution/waterline_projection_guards.rs"] +mod waterline_projection_guards; #[path = "execution/operator_dashboard.rs"] mod operator_dashboard; #[path = "execution/waterline_reads.rs"] @@ -322,6 +324,44 @@ enum TestDatabase { }, } +/// The rows of a two-column text query, written for each backend. +async fn text_rows(database: &TestDatabase, sqlite: &str, postgres: &str, mysql: &str) -> Vec<(String, String)> { + match database { + TestDatabase::Sqlite(dir) => { + let pool = sqlx::SqlitePool::connect(&format!("sqlite://{}", dir.path().join("runtime.sqlite").display())).await.unwrap(); + let rows = sqlx::query_as(sqlx::AssertSqlSafe(sqlite.to_owned())).fetch_all(&pool).await.unwrap(); + pool.close().await; + rows + } + TestDatabase::Postgres { options, .. } => { + let pool = sqlx::postgres::PgPoolOptions::new().connect_with(options.as_ref().clone()).await.unwrap(); + let rows = sqlx::query_as(sqlx::AssertSqlSafe(postgres.to_owned())).fetch_all(&pool).await.unwrap(); + pool.close().await; + rows + } + TestDatabase::MySql { options, .. } => { + let pool = sqlx::mysql::MySqlPoolOptions::new().connect_with(options.as_ref().clone()).await.unwrap(); + let rows = sqlx::query_as(sqlx::AssertSqlSafe(mysql.to_owned())).fetch_all(&pool).await.unwrap(); + pool.close().await; + rows + } + } +} + +/// Every table of the test database, by name, with its row count. +async fn table_counts(database: &TestDatabase) -> Vec<(String, i64)> { + let tables = text_rows(database, + "SELECT name, name FROM sqlite_master WHERE type='table' ORDER BY name", + "SELECT table_name::text, table_name::text FROM information_schema.tables WHERE table_schema='public' ORDER BY 1", + "SELECT CAST(table_name AS CHAR), CAST(table_name AS CHAR) FROM information_schema.tables WHERE table_schema=DATABASE() ORDER BY 1").await; + let mut counts = Vec::with_capacity(tables.len()); + for (table, _) in tables { + let sql: &'static str = Box::leak(format!("SELECT COUNT(*) FROM {table}").into_boxed_str()); + counts.push((table, task_retry::rows(database, sql).await)); + } + counts +} + impl TestDatabase { async fn new() -> Self { assert!( diff --git a/rust/tests/execution/structural_limits.rs b/rust/tests/execution/structural_limits.rs index 42534b31e..2a6d49b4d 100644 --- a/rust/tests/execution/structural_limits.rs +++ b/rust/tests/execution/structural_limits.rs @@ -11,6 +11,7 @@ async fn limited(database: &TestDatabase, signals: usize, updates: usize, payloa pending_signals: Some(signals), pending_updates: Some(updates), payload_bytes: Some(payload), + ..StructuralLimits::default() }; let runtime = database.open().await.unwrap().with_structural_limits(limits); (runtime.clone(), router(runtime)) @@ -221,3 +222,44 @@ async fn an_oversized_activity_result_fails_the_activity_like_php() { runtime.close().await; database.remove().await; } + +/// The history transaction ceilings as the operator metrics, the health +/// snapshot and the nested backend capabilities report them. +async fn published_history_limits(app: &Router) -> Vec<(String, Value, Value)> { + let (status, metrics) = request(app, "GET", "/api/system/operator-metrics", Value::Null).await; + assert_eq!(status, StatusCode::OK, "{metrics}"); + let (status, health) = request(app, "GET", "/api/system/health", Value::Null).await; + assert_eq!(status, StatusCode::OK, "{health}"); + let metrics = &metrics["operator_metrics"]; + let surfaces = [ + ("metrics configured", &metrics["structural_limits"]["configured"]), + ("metrics effective", &metrics["structural_limits"]["effective"]), + ("backend configured", &metrics["backend"]["structural_limits"]["configured"]), + ("backend effective", &metrics["backend"]["structural_limits"]["effective"]), + ("health configured", &health["health"]["structural_limits"]["configured"]), + ("health effective", &health["health"]["structural_limits"]["effective"]), + ]; + surfaces.into_iter().map(|(surface, limits)| { + (surface.to_owned(), limits["history_transaction_size"].clone(), limits["history_transaction_bytes"].clone()) + }).collect() +} + +#[tokio::test] +async fn every_surface_reports_the_active_history_transaction_limits() { + let database = TestDatabase::new().await; + for (events, bytes, shown) in [ + (Some(4321), Some(987_654), (json!(4321), json!(987_654))), + (Some(0), None, (json!(0), json!(0))), + ] { + let limits = StructuralLimits { history_events: events, history_bytes: bytes, ..StructuralLimits::default() }; + let app = router(database.open().await.unwrap().with_structural_limits(limits)); + for (surface, size, published_bytes) in published_history_limits(&app).await { + assert_eq!((&size, &published_bytes), (&shown.0, &shown.1), "{surface}"); + } + } + // The defaults, from no configuration. + let app = router(database.open().await.unwrap()); + for (surface, size, bytes) in published_history_limits(&app).await { + assert_eq!((size, bytes), (json!(5000), json!(16_777_216)), "{surface}"); + } +} diff --git a/rust/tests/execution/waterline_projection_guards.rs b/rust/tests/execution/waterline_projection_guards.rs new file mode 100644 index 000000000..5edb30911 --- /dev/null +++ b/rust/tests/execution/waterline_projection_guards.rs @@ -0,0 +1,296 @@ +use super::*; +use durable_workflow_server::runtime::StructuralLimits; + +// A Waterline v1 projection is one standalone, read-only run. Native refuses +// a message stream append to it before reading the body (published 2.11.1 +// accepts one and creates a signal and a workflow task), and refuses a +// projection larger than one transaction may write before reading or +// writing any row (published 2.11.1 has no limit). + +const ROUTE: &str = "/api/workflows/import/waterline-v1"; + +/// The running source of the waterline-import fixture, with `logs` logs. +fn running(logs: usize) -> Value { + json!({"id": "v1:4201", "legacy_id": "4201", "engine_source": "v1", "class": "App\\Workflows\\Legacy\\Subscription", + "status": "running", "connection": "database", "queue": "legacy-billing", "arguments": "a:0:{}", + "logs": (0..logs).map(|index| json!({"id": 9101 + index, "index": index, "now": "2026-09-30T09:00:01.000000Z", + "class": "App\\Activities\\Bill", "result": "i:42;", "created_at": "2026-09-30T09:00:02.000000Z"})).collect::>(), + "signals": [], "exceptions": [], "created_at": "2026-09-30T09:00:00.000000Z"}) +} + +async fn import(app: &Router, source_id: &str, workflow: Value) -> (StatusCode, Value) { + request(app, "POST", ROUTE, json!({"source_id": source_id, "workflow": workflow})).await +} + +/// Every table's row count: a projection or a stream append may write no +/// other row than the ones each test expects. +async fn rows(database: &TestDatabase) -> Vec<(String, i64)> { + let counts = table_counts(database).await; + for table in ["workflow_links", "workflow_inbound_streams", "workflow_inbound_stream_items", "workflow_signal_records"] { + assert!(counts.iter().any(|(name, _)| name == table), "{table} is not in {counts:?}"); + } + counts +} + +#[tokio::test] +async fn appends_to_a_projection_are_refused_before_their_body_is_read() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + let (status, report) = import(&app, "guard-src", running(1)).await; + assert_eq!(status, StatusCode::CREATED, "{report}"); + let workflow_id = report["identity"]["standalone"]["workflow_id"].as_str().unwrap().to_owned(); + let run_id = report["identity"]["standalone"]["run_id"].clone(); + let path = format!("/api/workflows/{}/message-streams/orders/messages", workflow_id.replace(':', "%3A")); + let before = rows(&database).await; + let (status, described) = request(&app, "GET", &format!("/api/workflows/{}", workflow_id.replace(':', "%3A")), Value::Null).await; + assert_eq!(status, StatusCode::OK, "{described}"); + + for body in [json!({"message_id": "m-1", "input": ["x"]}), json!({"message_id": 7, "input": "not a list"}), json!("not an object")] { + let (status, refusal) = request(&app, "POST", &path, body.clone()).await; + assert_eq!(status, StatusCode::CONFLICT, "{body}: {refusal}"); + assert_eq!(refusal["reason"], "v1_projection_read_only", "{refusal}"); + assert_eq!((&refusal["operation"], &refusal["run_id"], &refusal["stream_name"]), + (&json!("message_stream_append"), &run_id, &json!("orders")), "{refusal}"); + assert_eq!((&refusal["accepted"], &refusal["outcome"]), (&json!(false), &json!("rejected")), "{refusal}"); + } + + // No stream, item, signal, command, history, task or summary changed. + assert_eq!(rows(&database).await, before); + let (_, after) = request(&app, "GET", &format!("/api/workflows/{}", workflow_id.replace(':', "%3A")), Value::Null).await; + assert_eq!(after, described); +} + +#[tokio::test] +async fn a_projection_past_the_event_limit_is_refused_before_any_row() { + let database = TestDatabase::new().await; + let limits = StructuralLimits { history_events: Some(3), ..StructuralLimits::default() }; + let app = router(database.open().await.unwrap().with_structural_limits(limits)); + + // The start event and two logs: exactly the limit. + let (status, report) = import(&app, "events-a", running(2)).await; + assert_eq!(status, StatusCode::CREATED, "{report}"); + let before = rows(&database).await; + + // One log more, and many more: refused, nothing written. + for (source, logs, used) in [("events-b", 3, 4), ("events-c", 500, 501)] { + let (status, refusal) = import(&app, source, running(logs)).await; + assert_eq!(status, StatusCode::CONFLICT, "{refusal}"); + assert_eq!((&refusal["status"], &refusal["reason"]), (&json!("rejected"), &json!("v1_projection_too_large")), "{refusal}"); + assert_eq!(refusal["limit"], json!({"dimension": "history_transaction_size", "used": used, "limit": 3}), "{refusal}"); + assert_eq!(rows(&database).await, before); + } +} + +#[tokio::test] +async fn a_projection_past_the_byte_limit_is_refused_before_any_row() { + let database = TestDatabase::new().await; + // Measure one projection's history bytes as the summary records them. + let open = router(database.open().await.unwrap()); + let (status, report) = import(&open, "bytes-a", running(2)).await; + assert_eq!(status, StatusCode::CREATED, "{report}"); + let standalone = &report["identity"]["standalone"]; + let export = format!("/api/workflows/{}/runs/{}/history/export", + standalone["workflow_id"].as_str().unwrap().replace(':', "%3A"), standalone["run_id"].as_str().unwrap()); + let (status, bundle) = request(&open, "GET", &export, Value::Null).await; + assert_eq!(status, StatusCode::OK, "{bundle}"); + let bytes = usize::try_from(bundle["summary"]["history_size_bytes"].as_i64().unwrap()).unwrap(); + assert!(bytes > 0, "{bundle}"); + + // Source ids of one length give projections of one size. + let exact = router(database.open().await.unwrap() + .with_structural_limits(StructuralLimits { history_bytes: Some(bytes), ..StructuralLimits::default() })); + let (status, report) = import(&exact, "bytes-b", running(2)).await; + assert_eq!(status, StatusCode::CREATED, "{report}"); + let before = rows(&database).await; + + let short = router(database.open().await.unwrap() + .with_structural_limits(StructuralLimits { history_bytes: Some(bytes - 1), ..StructuralLimits::default() })); + let (status, refusal) = import(&short, "bytes-c", running(2)).await; + assert_eq!(status, StatusCode::CONFLICT, "{refusal}"); + assert_eq!(refusal["reason"], "v1_projection_too_large", "{refusal}"); + assert_eq!(refusal["limit"], json!({"dimension": "history_transaction_bytes", "used": bytes, "limit": bytes - 1}), "{refusal}"); + assert_eq!(rows(&database).await, before); +} + +/// Appends racing the import that creates a projection: each answers that +/// the instance does not exist yet or that it is read-only, and no stream, +/// item, signal, command or task is ever written. The store decides under +/// the lock every durable transaction takes, so the race has one outcome. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn appends_racing_a_projection_import_never_write() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + let (status, dry) = request(&app, "POST", ROUTE, json!({"source_id": "race-src", "workflow": running(1), "dry_run": true})).await; + assert_eq!(status, StatusCode::OK, "{dry}"); + let workflow_id = dry["identity"]["standalone"]["workflow_id"].as_str().unwrap().to_owned(); + let path = format!("/api/workflows/{}/message-streams/orders/messages", workflow_id.replace(':', "%3A")); + let before = rows(&database).await; + + let mut appends = tokio::task::JoinSet::new(); + for index in 0..24 { + let (app, path) = (app.clone(), path.clone()); + appends.spawn(async move { + request(&app, "POST", &path, json!({"message_id": format!("m-{index}"), "input": ["x"]})).await + }); + } + let (status, report) = import(&app, "race-src", running(1)).await; + assert_eq!(status, StatusCode::CREATED, "{report}"); + while let Some(answer) = appends.join_next().await { + let (status, body) = answer.unwrap(); + let expected = match status { + StatusCode::NOT_FOUND => "instance_not_found", + StatusCode::CONFLICT => "v1_projection_read_only", + other => panic!("{other}: {body}"), + }; + assert_eq!(body["reason"], expected, "{body}"); + } + // After the race, the projection and only the projection was written. + let after = rows(&database).await; + assert_eq!(before.len(), after.len()); + for ((table, was), (_, now)) in before.iter().zip(&after) { + let grew = match table.as_str() { + "workflow_instances" | "workflow_runs" | "workflow_run_summaries" => 1, + "workflow_history_events" => 2, + _ => 0, + }; + assert_eq!(*now, was + grew, "{table}"); + } +} + +/// A connection holding the lock every durable transaction takes on this +/// backend: while it is held, no transaction can begin its writes. +enum TransitionLock { + Sqlite(sqlx::SqliteConnection), + Postgres(sqlx::PgConnection), + MySql(sqlx::MySqlConnection), +} + +async fn hold_transition_lock(database: &TestDatabase) -> TransitionLock { + use sqlx::Connection; + match database { + TestDatabase::Sqlite(dir) => { + let url = format!("sqlite://{}", dir.path().join("runtime.sqlite").display()); + let mut connection = sqlx::SqliteConnection::connect(&url).await.unwrap(); + sqlx::query("BEGIN IMMEDIATE").execute(&mut connection).await.unwrap(); + TransitionLock::Sqlite(connection) + } + TestDatabase::Postgres { options, .. } => { + let mut connection = sqlx::PgConnection::connect_with(options).await.unwrap(); + sqlx::query("SELECT pg_advisory_lock(4924501473315676163)").execute(&mut connection).await.unwrap(); + TransitionLock::Postgres(connection) + } + TestDatabase::MySql { options, .. } => { + let mut connection = sqlx::MySqlConnection::connect_with(options).await.unwrap(); + // MySQL runs transaction control only outside prepared statements. + sqlx::raw_sql("START TRANSACTION").execute(&mut connection).await.unwrap(); + sqlx::query("SELECT engine FROM dw_server_schema WHERE engine='rust-development' FOR UPDATE") + .execute(&mut connection).await.unwrap(); + TransitionLock::MySql(connection) + } + } +} + +/// Wait until another transaction waits on the transition lock, where the +/// backend shows it: PostgreSQL's ungranted advisory lock, or an InnoDB +/// transaction in LOCK WAIT, in this test's database only (other tests run +/// in parallel). One connection polls, so the observer adds no connection +/// churn. SQLite's writer waits inside its own busy handler, which no other +/// connection can see; returns false there. +async fn lock_waiter_observed(database: &TestDatabase) -> bool { + use sqlx::Connection; + const POSTGRES: &str = "SELECT COUNT(*) FROM pg_locks WHERE locktype='advisory' AND NOT granted \ + AND database=(SELECT oid FROM pg_database WHERE datname=current_database())"; + const MYSQL: &str = "SELECT COUNT(*) FROM information_schema.innodb_trx t \ + JOIN information_schema.processlist p ON p.id=t.trx_mysql_thread_id WHERE t.trx_state='LOCK WAIT' AND p.db=DATABASE()"; + let pause = || tokio::time::sleep(std::time::Duration::from_millis(50)); + match database { + TestDatabase::Sqlite(_) => return false, + TestDatabase::Postgres { options, .. } => { + let mut connection = sqlx::PgConnection::connect_with(options).await.unwrap(); + for _ in 0..200 { + let waiting: i64 = sqlx::query_scalar(POSTGRES).fetch_one(&mut connection).await.unwrap(); + if waiting > 0 { + return true; + } + pause().await; + } + } + TestDatabase::MySql { options, .. } => { + // On MariaDB the install did not reach its lock wait while this + // observer polled information_schema from the start; it waits + // a moment first, then still requires the wait to be seen. + tokio::time::sleep(std::time::Duration::from_millis(1000)).await; + let mut connection = sqlx::MySqlConnection::connect_with(options).await.unwrap(); + for _ in 0..200 { + let waiting: i64 = sqlx::query_scalar(MYSQL).fetch_one(&mut connection).await.unwrap(); + if waiting > 0 { + return true; + } + pause().await; + } + } + } + panic!("no transaction waited on the transition lock within ten seconds"); +} + +async fn release(lock: TransitionLock) { + match lock { + TransitionLock::Sqlite(mut connection) => { sqlx::query("ROLLBACK").execute(&mut connection).await.unwrap(); } + TransitionLock::Postgres(mut connection) => { + sqlx::query("SELECT pg_advisory_unlock(4924501473315676163)").execute(&mut connection).await.unwrap(); + } + TransitionLock::MySql(mut connection) => { sqlx::raw_sql("ROLLBACK").execute(&mut connection).await.unwrap(); } + } +} + +/// The projection's install waits on the transition lock, so its outcome is +/// fixed on both sides: an append made while the install is pending finds +/// no instance and writes nothing, and one made after it commits is refused. +/// An append reaches the store only for an existing instance, and a +/// projection's instance and run are written in one transaction, so no +/// append can pass the store's check and then see the projection appear. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn an_append_cannot_cross_a_projection_install_waiting_on_the_lock() { + let database = TestDatabase::new().await; + if matches!(database, TestDatabase::Sqlite(_)) { + // SQLite's writer waits where no other connection can see it; the + // library control an_append_cannot_cross_an_install_waiting_on_the_sqlite_lock + // observes the install reaching the lock through a test probe. + return; + } + let app = router(database.open().await.unwrap()); + let (status, dry) = request(&app, "POST", ROUTE, json!({"source_id": "barrier-src", "workflow": running(1), "dry_run": true})).await; + assert_eq!(status, StatusCode::OK, "{dry}"); + let workflow_id = dry["identity"]["standalone"]["workflow_id"].as_str().unwrap().to_owned(); + let path = format!("/api/workflows/{}/message-streams/orders/messages", workflow_id.replace(':', "%3A")); + let before = rows(&database).await; + + let lock = hold_transition_lock(&database).await; + let install = tokio::spawn({ + let app = app.clone(); + async move { import(&app, "barrier-src", running(1)).await } + }); + // On PostgreSQL and MySQL the install is seen waiting on the lock; on + // SQLite it cannot commit while the lock is held, wherever it is. + assert!(lock_waiter_observed(&database).await); + let (status, body) = request(&app, "POST", &path, json!({"message_id": "m-pending", "input": ["x"]})).await; + assert_eq!((status, &body["reason"]), (StatusCode::NOT_FOUND, &json!("instance_not_found")), "{body}"); + assert!(!install.is_finished(), "the install committed while the transition lock was held"); + release(lock).await; + + let (status, report) = install.await.unwrap(); + assert_eq!(status, StatusCode::CREATED, "{report}"); + let (status, body) = request(&app, "POST", &path, json!({"message_id": "m-after", "input": ["x"]})).await; + assert_eq!((status, &body["reason"]), (StatusCode::CONFLICT, &json!("v1_projection_read_only")), "{body}"); + + let after = rows(&database).await; + assert_eq!(before.len(), after.len()); + for ((table, was), (_, now)) in before.iter().zip(&after) { + let grew = match table.as_str() { + "workflow_instances" | "workflow_runs" | "workflow_run_summaries" => 1, + "workflow_history_events" => 2, + _ => 0, + }; + assert_eq!(*now, was + grew, "{table}"); + } +} diff --git a/rust/tests/fixtures/unserved-php-routes.v2.11.1.txt b/rust/tests/fixtures/unserved-php-routes.v2.11.1.txt index 75af9ff47..364ee1de6 100644 --- a/rust/tests/fixtures/unserved-php-routes.v2.11.1.txt +++ b/rust/tests/fixtures/unserved-php-routes.v2.11.1.txt @@ -32,7 +32,6 @@ POST /api/worker/workflow-tasks/{taskId}/local-activities/{attemptId}/acknowledg POST /api/worker/workflow-tasks/{taskId}/local-activities/{attemptId}/control POST /api/worker/workflow-tasks/{taskId}/local-activities/{attemptId}/heartbeat POST /api/worker/workflow-tasks/{taskId}/local-activities/{attemptId}/outcome -POST /api/workflows/import/waterline-v1 GET /api/workflows/{workflowId}/nexus-operations GET /api/workflows/{workflowId}/runs/{runId}/nexus-operations POST /api/workflows/{workflowId}/runs/{runId}/streams/{streamName}/close diff --git a/scripts/conformance/server-parity/contract.mjs b/scripts/conformance/server-parity/contract.mjs index eee792f66..7f4ec2835 100644 --- a/scripts/conformance/server-parity/contract.mjs +++ b/scripts/conformance/server-parity/contract.mjs @@ -20,6 +20,7 @@ import {checkWorkerSessionActivities} from './worker-session-activities-contract import {checkActivityCancellationAcknowledgement} from './activity-cancellation-ack-contract.mjs'; import {checkServiceCatalog} from './service-catalog-contract.mjs'; import {checkWorkflowImport} from './workflow-import-contract.mjs'; +import {checkWaterlineImport} from './waterline-import-contract.mjs'; import {checkSystemMetrics} from './system-metrics-contract.mjs'; import {checkOperatorDashboardTypes} from './operator-dashboard-types-contract.mjs'; import {checkPrometheusMetrics} from './prometheus-metrics-contract.mjs'; @@ -167,6 +168,7 @@ export function checkObservation(fixture, observation, workflowId, artifacts) { ? checkActivityCancellationAcknowledgement(fixture, observation, workflowId) : null; const serviceCatalog = fixture.service_catalog ? checkServiceCatalog(fixture, observation, workflowId) : null; const workflowImport = fixture.workflow_import ? checkWorkflowImport(fixture, observation, workflowId) : null; + const waterlineImport = fixture.waterline_import ? checkWaterlineImport(fixture, observation, workflowId) : null; const systemMetrics = fixture.system_metrics ? checkSystemMetrics(fixture, observation, workflowId) : null; const operatorDashboardTypes = fixture.operator_dashboard_types ? checkOperatorDashboardTypes(fixture, observation, workflowId) : null; const prometheusMetrics = fixture.prometheus_metrics ? checkPrometheusMetrics(fixture, observation, workflowId) : null; @@ -730,6 +732,7 @@ export function checkObservation(fixture, observation, workflowId, artifacts) { ...(activityCancellationAcknowledgement ? {activity_cancellation_acknowledgement: activityCancellationAcknowledgement} : {}), ...(serviceCatalog ? {service_catalog: serviceCatalog} : {}), ...(workflowImport ? {workflow_import: workflowImport} : {}), + ...(waterlineImport ? {waterline_import: waterlineImport} : {}), ...(systemMetrics ? {system_metrics: systemMetrics} : {}), ...(operatorDashboardTypes ? {operator_dashboard_types: operatorDashboardTypes} : {}), ...(prometheusMetrics ? {prometheus_metrics: prometheusMetrics} : {}), diff --git a/scripts/conformance/server-parity/probe.php b/scripts/conformance/server-parity/probe.php index 4f293fc4a..219c5ca7a 100644 --- a/scripts/conformance/server-parity/probe.php +++ b/scripts/conformance/server-parity/probe.php @@ -439,6 +439,7 @@ function httpObservation(array $fixture, string $workflowId, string $namespace, ? finishHttpActivityCancellationAcknowledgement($client, $fixture, $workflowId, $url) : null; $serviceCatalog = isset($fixture['service_catalog']) ? finishHttpServiceCatalog($client, $fixture, $workflowId, $url) : null; $workflowImport = isset($fixture['workflow_import']) ? finishHttpWorkflowImport($client, $fixture, $workflowId, $url) : null; + $waterlineImport = isset($fixture['waterline_import']) ? finishHttpWaterlineImport($client, $fixture, $workflowId, $url) : null; $systemMetrics = isset($fixture['system_metrics']) ? finishHttpSystemMetrics($client, $fixture, $workflowId, $url) : null; $operatorDashboardTypes = isset($fixture['operator_dashboard_types']) ? finishHttpOperatorDashboardTypes($client, $fixture, $workflowId, $url) : null; $prometheusMetrics = isset($fixture['prometheus_metrics']) ? finishHttpPrometheusMetrics($client, $fixture, $workflowId, $url) : null; @@ -555,6 +556,7 @@ function httpObservation(array $fixture, string $workflowId, string $namespace, 'activity_cancellation_acknowledgement' => $activityCancellationAcknowledgement, 'service_catalog' => $serviceCatalog, 'workflow_import' => $workflowImport, + 'waterline_import' => $waterlineImport, 'system_metrics' => $systemMetrics, 'operator_dashboard_types' => $operatorDashboardTypes, 'prometheus_metrics' => $prometheusMetrics, @@ -875,6 +877,7 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa require __DIR__.'/activity-cancellation-ack-probe.php'; require __DIR__.'/service-catalog-probe.php'; require __DIR__.'/workflow-import-probe.php'; + require __DIR__.'/waterline-import-probe.php'; require __DIR__.'/system-metrics-probe.php'; require __DIR__.'/operator-dashboard-types-probe.php'; require __DIR__.'/prometheus-metrics-probe.php'; @@ -1066,6 +1069,9 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa if ($mode === 'embedded' && isset($fixture['workflow_import'])) { $observation['workflow_import'] = ['applicable' => false, 'reason' => 'embedded_has_no_http_workflow_import']; } + if ($mode === 'embedded' && isset($fixture['waterline_import'])) { + $observation['waterline_import'] = ['applicable' => false, 'reason' => 'embedded_has_no_http_waterline_import']; + } if ($mode === 'embedded' && isset($fixture['system_metrics'])) { $observation['system_metrics'] = ['applicable' => false, 'reason' => 'embedded_has_no_http_metrics']; } diff --git a/scripts/conformance/server-parity/waterline-import-contract.mjs b/scripts/conformance/server-parity/waterline-import-contract.mjs new file mode 100644 index 000000000..4f9bf6255 --- /dev/null +++ b/scripts/conformance/server-parity/waterline-import-contract.mjs @@ -0,0 +1,25 @@ +import assert from 'node:assert/strict'; + +// Waterline v1 projection import (LegacyV1ProjectionController), recorded +// against published Server 2.11.1. The fixture's `expected` holds the whole +// normalized observation, with the probe's workflow prefix as `{id}`, +// projected identities, event ids, dedupe keys and export checksums as +// `{