Serve the external payload storage diagnostic natively

.github/workflows/server-parity.yml

@@ -26,6 +26,7 @@ on:
       - tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs
       - tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs
       - tests/Unit/ServerParitySearchAttributeContractTest.mjs
+      - tests/Unit/ServerParityStorageTestContractTest.mjs
       - tests/Unit/ServerParityWorkerListingContractTest.mjs
       - tests/Unit/ServerParityWorkerReplacementContractTest.mjs
       - tests/Unit/ServerParityWaitingHistoryContractTest.mjs
@@ -67,6 +68,7 @@ on:
       - tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs
       - tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs
       - tests/Unit/ServerParitySearchAttributeContractTest.mjs
+      - tests/Unit/ServerParityStorageTestContractTest.mjs
       - tests/Unit/ServerParityWorkerListingContractTest.mjs
       - tests/Unit/ServerParityWorkerReplacementContractTest.mjs
       - tests/Unit/ServerParityWaitingHistoryContractTest.mjs
@@ -1458,7 +1460,7 @@ jobs:
             -e COMPOSER_HOME=/tmp/composer -v "$PWD:/source" -w /source "$image" \
             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/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/ServerParityLateWorkflowCompletionContractTest.mjs > parity-evidence/comparator-tests.txt
+            "$image" --test tests/Unit/ServerParityRunnerTest.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/ServerParityLateWorkflowCompletionContractTest.mjs tests/Unit/ServerParityStorageTestContractTest.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

docs/rust-server-port.md

@@ -229,6 +229,29 @@ support is not evidence of multi-node safety.
 Newest checkpoints first. Each entry is exact to the source, tree and runs it
 names, including unsuccessful observations.
 
+October 20, external payload storage diagnostic:
+
+- **Source.** Native serves `POST /api/storage/test` for operators, in the
+  default or a named namespace, as PHP's `StorageController::test` does:
+  - Laravel's validation messages and error counts for `driver` and the two
+    payload sizes (shared with the search attribute registry through
+    `laravel_validation`);
+  - `external_storage_not_configured`, `external_storage_disabled` and
+    `storage_driver_unavailable` refusals with PHP's bodies;
+  - a small and large generated payload written to, read back from and
+    removed from the local driver, with each one's size, SHA-256 and the
+    digest of its object URI.
+- **Shared objects.** Objects are content-addressed, so a diagnostic payload
+  can have the same bytes as a registered external payload. Native holds the
+  payload object lock across the round trip and removes the object only when
+  no registered payload refers to it. PHP removes it unconditionally.
+- **Not covered.** S3, GCS, Azure and custom drivers are refused as
+  unavailable in this runtime, as published PHP does without their SDKs.
+- **Fixture.** `storage-test` drives a fresh namespace through no policy, a
+  disabled policy, another driver, three invalid bodies and two passing round
+  trips. Published PHP 2.5.14 and native returned identical bodies, including
+  the reference identities, and embedded PHP declares it inapplicable.
+
 October 20, database TLS settings (PHP Server 2.9.10, #545):
 
 - **Source.** Native resolves PostgreSQL, MySQL and MariaDB TLS with

rust/src/external_payload/local.rs

@@ -149,6 +149,19 @@ impl LocalStorage {
         .map_err(unavailable)?
     }
 
+    /// Removes the object. The caller holds the object lock and has checked
+    /// that no registered payload shares it.
+    pub async fn delete(&self, reference: &Reference) -> Result<(), SnapshotError> {
+        let path = self.path(reference)?;
+        tokio::task::spawn_blocking(move || match fs::remove_file(&path) {
+            Ok(()) => Ok(()),
+            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
+            Err(error) => Err(unavailable(error)),
+        })
+        .await
+        .map_err(unavailable)?
+    }
+
     pub async fn fetch(
         &self,
         reference: Reference,

rust/src/runtime/execution.rs

@@ -295,6 +295,15 @@ impl Runtime {
     delegate!(describe_namespace(name: &str) -> Result<Value>);
     delegate!(create_namespace(body: Value) -> Result<Value>);
     delegate!(list_search_attributes() -> Result<Value>);
+    delegate!(
+        storage_test_round_trip(
+            local: &crate::external_payload::LocalStorage,
+            reference: crate::external_payload::Reference,
+            payload: Vec<u8>,
+            max_bytes: u64,
+            workers: Arc<Semaphore>,
+        ) -> Result<()>
+    );
     delegate!(create_search_attribute(body: Value) -> Result<Value>);
     delegate!(delete_search_attribute(name: &str) -> Result<Value>);
     delegate!(set_external_policy(name: &str, body: Value) -> Result<Value>);

rust/src/runtime/external_registry.rs

@@ -100,6 +100,59 @@ where
         Ok(())
     }
 
+    /// The storage diagnostic's write, read and delete of a generated object,
+    /// under the object lock that uploads and cleanup take. Local objects are
+    /// content-addressed, so a registered payload with the same bytes keeps
+    /// its object: the diagnostic deletes only an object nothing references.
+    pub(super) async fn storage_test_round_trip(
+        &self,
+        local: &crate::external_payload::LocalStorage,
+        reference: Reference,
+        payload: Vec<u8>,
+        max_bytes: u64,
+        workers: std::sync::Arc<tokio::sync::Semaphore>,
+    ) -> Result<()> {
+        use super::external_transport::snapshot_error;
+        let uri = local.uri_for(&reference).map_err(snapshot_error)?;
+        let mut tx = self.begin().await?;
+        Self::payload_object_lock(&mut tx, &uri).await?;
+        let registered =
+            Self::scalar("SELECT COUNT(*) FROM runtime_external_payloads WHERE storage_uri=$1")
+                .bind(&uri)
+                .fetch_one(&mut *tx)
+                .await?
+                > 0;
+        let snapshot = crate::external_payload::capture(
+            std::io::Cursor::new(payload),
+            reference.clone(),
+            max_bytes,
+            std::env::temp_dir(),
+            workers.clone(),
+        )
+        .await
+        .map_err(snapshot_error)?;
+        local
+            .put(reference.clone(), snapshot)
+            .await
+            .map_err(snapshot_error)?;
+        // Fetch verifies the stored size and digest against the reference.
+        local
+            .fetch(
+                reference.clone(),
+                &uri,
+                max_bytes,
+                std::env::temp_dir(),
+                workers,
+            )
+            .await
+            .map_err(snapshot_error)?;
+        if !registered {
+            local.delete(&reference).await.map_err(snapshot_error)?;
+        }
+        tx.commit().await?;
+        Ok(())
+    }
+
     /// The caller commits this writing row before any provider effect. A retry
     /// keeps the existing ID, ready status and retained ownership; abandoned
     /// unclaimed retries renew only their upload expiry, matching PHP.

rust/src/runtime/http.rs

@@ -45,6 +45,7 @@ pub fn router(runtime: Runtime) -> Router {
             "/api/external-payloads/v1/{reference_id}",
             get(super::external_transport::fetch),
         )
+        .route("/api/storage/test", post(storage_test))
         .route(
             "/api/search-attributes",
             get(list_search_attributes).post(create_search_attribute),
@@ -291,6 +292,7 @@ async fn guard(State(runtime): State<Runtime>, mut request: Request, next: Next)
         &["worker"]
     } else if path.starts_with("/api/workers/") && request.method() != Method::GET
         || path == "/api/namespaces" && request.method() == Method::POST
+        || path == "/api/storage/test"
         // Updating a namespace or its external storage needs an admin.
         || path.starts_with("/api/namespaces/") && request.method() == Method::PUT
     {
@@ -427,6 +429,9 @@ fn named_namespace_operation(method: &str, path: &str) -> bool {
     if path == "/api/workflows" {
         return method == "POST";
     }
+    if path == "/api/storage/test" {
+        return method == "POST";
+    }
     // The registry is namespace-scoped state with no default-namespace selector.
     if path == "/api/search-attributes" {
         return matches!(method, "GET" | "POST");
@@ -526,6 +531,13 @@ async fn create_namespace(
     ))
 }
 
+async fn storage_test(
+    Extension(runtime): Extension<Runtime>,
+    Json(body): Json<Value>,
+) -> Result<Json<Value>> {
+    Ok(Json(super::storage_test::storage_test(&runtime, body).await?))
+}
+
 async fn list_search_attributes(Extension(runtime): Extension<Runtime>) -> Result<Json<Value>> {
     Ok(Json(runtime.list_search_attributes().await?))
 }

rust/src/runtime/laravel_validation.rs

@@ -0,0 +1,60 @@
+//! Laravel's request normalization and validation failure body, as published
+//! PHP renders them for control-plane requests.
+use super::RuntimeError;
+use axum::http::StatusCode;
+use serde_json::{Map, Value, json};
+
+/// Laravel's TrimStrings and ConvertEmptyStringsToNull run before validation.
+pub(super) fn normalized(value: Option<&Value>) -> Option<Value> {
+    let trimmed = |c: char| c.is_whitespace() || matches!(c, '\u{feff}' | '\u{200b}' | '\u{200e}');
+    match value {
+        Some(Value::String(text)) => {
+            let text = text.trim_matches(trimmed);
+            Some(if text.is_empty() {
+                Value::Null
+            } else {
+                json!(text)
+            })
+        }
+        other => other.cloned(),
+    }
+}
+
+/// Laravel's `required`: null, an empty string and an empty array fail.
+pub(super) fn present(value: Option<&Value>) -> bool {
+    match value {
+        None | Some(Value::Null) => false,
+        Some(Value::String(text)) => !text.trim().is_empty(),
+        Some(Value::Array(items)) => !items.is_empty(),
+        Some(Value::Object(items)) => !items.is_empty(),
+        Some(_) => true,
+    }
+}
+
+/// Laravel's 422 ValidationException body for control-plane requests. Fields
+/// without messages are left out; the summary counts the remaining messages.
+pub(super) fn failure(errors: Vec<(&'static str, Vec<&'static str>)>) -> Option<RuntimeError> {
+    let errors: Vec<_> = errors
+        .into_iter()
+        .filter(|(_, messages)| !messages.is_empty())
+        .collect();
+    let all: Vec<&str> = errors
+        .iter()
+        .flat_map(|(_, messages)| messages.iter().copied())
+        .collect();
+    let mut message = (*all.first()?).to_owned();
+    match all.len() - 1 {
+        0 => {}
+        1 => message.push_str(" (and 1 more error)"),
+        more => message.push_str(&format!(" (and {more} more errors)")),
+    }
+    let errors: Map<String, Value> = errors
+        .into_iter()
+        .map(|(field, messages)| (field.to_owned(), json!(messages)))
+        .collect();
+    Some(RuntimeError::Protocol {
+        status: StatusCode::UNPROCESSABLE_ENTITY,
+        response: json!({"message":message,"reason":"validation_failed",
+            "errors":errors.clone(),"validation_errors":errors}),
+    })
+}

rust/src/runtime/mod.rs

@@ -30,6 +30,7 @@ mod failure_commands;
 mod history_budget;
 mod history_presentation;
 mod http;
+mod laravel_validation;
 mod memo_json;
 mod memo_limits;
 mod memos;
@@ -47,11 +48,12 @@ mod schedule_execution;
 mod schedule_http;
 mod schedule_storage;
 mod schedules;
-mod search_attributes;
 mod schema;
+mod search_attributes;
 mod selection_summary;
 mod selections;
 mod sqlite;
+mod storage_test;
 mod store;
 mod task_failures;
 mod task_repair;