Migrate schema 2 ports through frozen readers and expose schema 3 source API

crates/oa-server/Cargo.toml

@@ -15,7 +15,7 @@ axum = { version = "0.8", default-features = false, features = ["http1", "json",
 tokio = { version = "1", features = ["macros", "rt-multi-thread", "net", "signal", "sync", "time", "io-util"] }
 tower-http = { version = "0.6", features = ["limit"] }
 serde.workspace = true
-serde_json.workspace = true
+serde_json = { workspace = true, features = ["raw_value"] }
 uuid.workspace = true
 thiserror.workspace = true
 tempfile = "3"

crates/oa-server/src/routes.rs

@@ -182,13 +182,13 @@ fn not_modified(stored: &StoredProject, headers: &HeaderMap) -> Option<Response>
 fn summary(stored: &StoredProject) -> Value {
     json!({"resource_id":stored.resource_id,"revision":stored.revision,"model_id":stored.model.id,
         "schema_version":stored.model.schema_version,"name":stored.model.name,
-        "elements":stored.model.elements.len(),"relations":stored.model.relations.len()})
+        "elements":stored.model.elements.len(),"relations":stored.model.relations.len(),"migration":stored.migration})
 }
 
-fn parse_source(body: &[u8]) -> ApiResult<Project> {
+fn parse_source(body: &[u8]) -> ApiResult<oa_core::persistence::DecodeResult> {
     let source =
         std::str::from_utf8(body).map_err(|_| ApiError::bad_request("Source must be UTF-8."))?;
-    oa_core::language::parse_limited(source, MODEL_LIMITS).map_err(|diagnostics| {
+    oa_core::language::parse_with_report_limited(source, MODEL_LIMITS).map_err(|diagnostics| {
         let code = if diagnostics.iter().any(|d| d.code == "model_limit") {
             "model_limit"
         } else {
@@ -215,22 +215,41 @@ fn parse_source(body: &[u8]) -> ApiResult<Project> {
     })
 }
 
-fn import_json(body: &[u8]) -> ApiResult<Project> {
-    // Format is chosen by HTTP. The shared decoder owns explicit schema-1 migration.
+fn import_json(body: &[u8]) -> ApiResult<oa_core::persistence::DecodeResult> {
+    // Format is chosen by HTTP. The shared decoder owns explicit schema-1 and schema-2 migration.
     let _: Value = json_body(body)?;
-    oa_core::persistence::decode_limited(body, MODEL_LIMITS).map_err(|error| match error {
-        oa_core::persistence::FileError::Limit(detail) => {
-            ApiError::new(StatusCode::UNPROCESSABLE_ENTITY, "model_limit", detail)
-        }
-        oa_core::persistence::FileError::Json(error) => {
-            ApiError::bad_request(format!("Invalid model JSON: {error}"))
-        }
-        error => ApiError::new(
-            StatusCode::UNPROCESSABLE_ENTITY,
-            "invalid_model",
-            error.to_string(),
-        ),
-    })
+    oa_core::persistence::decode_with_report_limited(body, MODEL_LIMITS).map_err(
+        |error| match error {
+            oa_core::persistence::FileError::Limit(detail) => {
+                ApiError::new(StatusCode::UNPROCESSABLE_ENTITY, "model_limit", detail)
+            }
+            oa_core::persistence::FileError::Model { message, issues } => {
+                let mut error =
+                    ApiError::new(StatusCode::UNPROCESSABLE_ENTITY, "invalid_model", message);
+                error.issues = issues
+                    .into_iter()
+                    .map(|issue| IssueDto {
+                        severity: "error",
+                        code: issue.code,
+                        message: issue.message,
+                        field: issue.field,
+                        object_id: issue.element,
+                        line: None,
+                        column: None,
+                    })
+                    .collect();
+                error
+            }
+            oa_core::persistence::FileError::Json(error) => {
+                ApiError::bad_request(format!("Invalid model JSON: {error}"))
+            }
+            error => ApiError::new(
+                StatusCode::UNPROCESSABLE_ENTITY,
+                "invalid_model",
+                error.to_string(),
+            ),
+        },
+    )
 }
 
 #[derive(Deserialize)]
@@ -282,7 +301,7 @@ async fn import_project(
         } else {
             parse_source(&body)?
         };
-        created(store.create(model)?)
+        created(store.create_decoded(model)?)
     })
     .await
 }
@@ -477,15 +496,15 @@ async fn replace_model(
     blocking(state, move |store| {
         store.checked(id, &expected)?;
         let model = if is_json {
-            json_body(&body)?
+            import_json(&body)?
         } else {
             parse_source(&body)?
         };
-        let (stored, changed) = store.replace(id, &expected, model)?;
+        let (stored, changed) = store.replace_decoded(id, &expected, model)?;
         Ok(response(
             &stored,
             StatusCode::OK,
-            json!({"revision":stored.revision,"changed":changed,"issues":issues(&stored.model)}),
+            json!({"revision":stored.revision,"changed":changed,"issues":issues(&stored.model),"migration":stored.migration,"migration":stored.migration}),
         ))
     })
     .await
@@ -822,10 +841,11 @@ async fn get_validation(State(state): State<AppState>, id: ProjectPath) -> ApiRe
 }
 
 #[derive(Deserialize)]
-#[serde(tag = "format", rename_all = "snake_case", deny_unknown_fields)]
-enum ValidationRequest {
-    Json { model: Project },
-    Arcadia { source: String },
+#[serde(deny_unknown_fields)]
+struct ValidationRequest {
+    format: String,
+    model: Option<Box<serde_json::value::RawValue>>,
+    source: Option<String>,
 }
 
 async fn validate_candidate(
@@ -841,15 +861,16 @@ async fn validate_candidate(
         let snapshot = store.get(id)?;
         let stored = &snapshot;
         let request: ValidationRequest = json_body(&body)?;
-        let parsed = match request {
-            ValidationRequest::Json { model } => Ok(model),
-            ValidationRequest::Arcadia { source } => parse_source(source.as_bytes()),
+        let parsed = match (request.format.as_str(), request.model, request.source) {
+            ("json", Some(model), None) => import_json(model.get().as_bytes()),
+            ("arcadia", None, Some(source)) => parse_source(source.as_bytes()),
+            _ => return Err(ApiError::bad_request("Validation requires either format json with only model, or format arcadia with only source.")),
         };
-        let (valid, issues) = match parsed.and_then(|model| {
-            validate_model(&model)?;
-            Ok(model)
+        let (valid, issues, migration) = match parsed.and_then(|decoded| {
+            validate_model(&decoded.project)?;
+            Ok(decoded)
         }) {
-            Ok(model) => (true, issues(&model)),
+            Ok(decoded) => (true, issues(&decoded.project), decoded.migration),
             Err(error) => {
                 let mut diagnostics = error.issues;
                 if diagnostics.is_empty() {
@@ -863,13 +884,13 @@ async fn validate_candidate(
                         column: None,
                     });
                 }
-                (false, diagnostics)
+                (false, diagnostics, None)
             }
         };
         Ok(response(
             stored,
             StatusCode::OK,
-            json!({"revision":stored.revision,"valid":valid,"issues":issues}),
+            json!({"revision":stored.revision,"valid":valid,"issues":issues,"migration":migration}),
         ))
     })
     .await

crates/oa-server/src/store.rs

@@ -1,6 +1,7 @@
 //! Single-owner on-disk envelopes. All methods run on a blocking worker.
 use crate::error::{ApiError, ApiResult, issues};
 use axum::http::StatusCode;
+use oa_core::persistence::{DecodeResult, MigrationReport};
 use oa_core::{Id, Project};
 use serde::{Deserialize, Serialize};
 use std::{
@@ -29,6 +30,8 @@ pub struct StoredProject {
     pub resource_id: Id,
     pub revision: u64,
     pub model: Project,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub migration: Option<MigrationReport>,
 }
 
 impl StoredProject {
@@ -46,6 +49,7 @@ pub(crate) struct ProjectSummary {
     pub name: String,
     pub elements: usize,
     pub relations: usize,
+    pub migration: Option<MigrationReport>,
 }
 impl From<&StoredProject> for ProjectSummary {
     fn from(stored: &StoredProject) -> Self {
@@ -57,6 +61,7 @@ impl From<&StoredProject> for ProjectSummary {
             name: stored.model.name.clone(),
             elements: stored.model.elements.len(),
             relations: stored.model.relations.len(),
+            migration: stored.migration.clone(),
         }
     }
 }
@@ -88,11 +93,11 @@ struct StoreInner {
 
 /// Constructed only after complete candidate validation, outside the catalog
 /// mutex. Commit methods cannot accidentally accept an unchecked candidate.
-struct PreparedModel(Project);
+struct PreparedModel(Project, Option<MigrationReport>);
 impl PreparedModel {
     fn new(model: Project) -> ApiResult<Self> {
         validate_model(&model)?;
-        Ok(Self(model))
+        Ok(Self(model, None))
     }
 }
 
@@ -129,6 +134,22 @@ impl Store {
         self.lock()?.create(prepared)
     }
 
+    pub(crate) fn create_decoded(&self, decoded: DecodeResult) -> ApiResult<StoredProject> {
+        let mut prepared = PreparedModel::new(decoded.project)?;
+        prepared.1 = decoded.migration;
+        self.lock()?.create(prepared)
+    }
+    pub(crate) fn replace_decoded(
+        &self,
+        id: Id,
+        expected: &str,
+        decoded: DecodeResult,
+    ) -> ApiResult<(StoredProject, bool)> {
+        let mut prepared = PreparedModel::new(decoded.project)?;
+        prepared.1 = decoded.migration;
+        self.lock()?.replace(id, expected, prepared)
+    }
+
     pub(crate) fn replace(
         &self,
         id: Id,
@@ -201,8 +222,26 @@ impl Store {
                     "At most 128 models and 256 MiB of serialized envelopes are supported.".into(),
                 ));
             }
-            let stored: StoredProject =
+            #[derive(Deserialize)]
+            #[serde(deny_unknown_fields)]
+            struct Envelope {
+                resource_id: Id,
+                revision: u64,
+                model: Box<serde_json::value::RawValue>,
+            }
+            let envelope: Envelope =
                 serde_json::from_slice(&bytes).map_err(|error| corrupt(error.to_string()))?;
+            let decoded = oa_core::persistence::decode_with_report_limited(
+                envelope.model.get().as_bytes(),
+                MODEL_LIMITS,
+            )
+            .map_err(|error| corrupt(error.to_string()))?;
+            let stored = StoredProject {
+                resource_id: envelope.resource_id,
+                revision: envelope.revision,
+                model: decoded.project,
+                migration: decoded.migration,
+            };
             if stored.resource_id != filename_id || stored.revision == 0 {
                 return Err(corrupt(
                     "Resource ID/filename mismatch or zero revision.".into(),
@@ -254,6 +293,7 @@ impl StoreInner {
             resource_id: Id::new_v4(),
             revision: 1,
             model,
+            migration: prepared.1,
         };
         let size = self.write(&stored, true)?;
         self.adopt(stored.clone(), size);
@@ -270,7 +310,11 @@ impl StoreInner {
         let current = self.checked(id, expected)?;
         let model = prepared.0;
         if current.model == model {
-            return Ok((current.clone(), false));
+            let mut unchanged = current.clone();
+            if prepared.1.is_some() {
+                unchanged.migration = prepared.1;
+            }
+            return Ok((unchanged, false));
         }
         let revision = current.revision.checked_add(1).ok_or_else(|| {
             ApiError::new(
@@ -283,6 +327,7 @@ impl StoreInner {
             resource_id: id,
             revision,
             model,
+            migration: prepared.1,
         };
         let size = self.write(&stored, false)?;
         self.adopt(stored.clone(), size);
@@ -312,7 +357,9 @@ impl StoreInner {
     }
 
     fn write(&self, stored: &StoredProject, create: bool) -> ApiResult<usize> {
-        let bytes = serde_json::to_vec(stored)
+        // Migration observations are response metadata, never persisted facts.
+        let envelope = serde_json::json!({"resource_id": stored.resource_id, "revision": stored.revision, "model": stored.model});
+        let bytes = serde_json::to_vec(&envelope)
             .map_err(|_| ApiError::internal("Could not serialize model."))?;
         if bytes.len() > MAX_ENVELOPE_BYTES {
             return Err(ApiError::new(

crates/oa-server/tests/http.rs

@@ -8,6 +8,20 @@ use serde_json::{Value, json};
 use std::{path::Path, sync::Arc};
 use tokio::{sync::oneshot, task::JoinHandle};
 
+fn v2_missing_ports() -> Value {
+    let mut old: Value = serde_json::from_slice(include_bytes!(
+        "../../oa-core/tests/fixtures/v2-coastal-rescue.json"
+    ))
+    .unwrap();
+    for relation in old["relations"].as_object_mut().unwrap().values_mut() {
+        if let Some(exchange) = relation["body"].get_mut("FunctionalExchange") {
+            exchange["source_port"] = Value::Null;
+            exchange["target_port"] = Value::Null;
+        }
+    }
+    old
+}
+
 struct TestServer {
     base: String,
     client: Client,
@@ -1318,6 +1332,7 @@ fn storage_rejects_second_owners_corruption_and_identity_mismatch() {
     std::fs::write(&path, "{broken").unwrap();
     assert!(Store::open(directory.path()).is_err());
     let mismatched = StoredProject {
+        migration: None,
         resource_id: Id::new_v4(),
         revision: 1,
         model: Project::empty(),
@@ -1325,6 +1340,7 @@ fn storage_rejects_second_owners_corruption_and_identity_mismatch() {
     std::fs::write(&path, serde_json::to_vec(&mismatched).unwrap()).unwrap();
     assert!(Store::open(directory.path()).is_err());
     let invalid = StoredProject {
+        migration: None,
         resource_id: filename,
         revision: 0,
         model: Project::empty(),
@@ -1484,6 +1500,7 @@ async fn fixed_store_capacity_is_enforced_on_http_and_restart() {
     let mut last = None;
     for _ in 0..oa_server::store::MAX_PROJECTS {
         let envelope = StoredProject {
+            migration: None,
             resource_id: Id::new_v4(),
             revision: 1,
             model: Project::empty(),
@@ -1624,3 +1641,154 @@ async fn legacy_json_import_and_model_reidentification_keep_the_service_address(
     );
     server.stop().await;
 }
+
+#[tokio::test]
+async fn schema3_migration_reports_and_invalid_typed_transaction_preserve_http_state() {
+    let directory = tempfile::tempdir().unwrap();
+    let server = TestServer::start(directory.path(), None).await;
+    let old = v2_missing_ports();
+    let expected =
+        oa_core::persistence::decode_with_report(&serde_json::to_vec(&old).unwrap()).unwrap();
+    let imported = server
+        .client
+        .post(server.url("/projects/import"))
+        .json(&old)
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(imported.status(), StatusCode::CREATED);
+    let etag = imported.headers()["etag"].to_str().unwrap().to_owned();
+    let summary: Value = imported.json().await.unwrap();
+    assert_eq!(summary["schema_version"], 3);
+    assert_eq!(
+        summary["migration"],
+        serde_json::to_value(&expected.migration).unwrap()
+    );
+    let id: Id = summary["resource_id"].as_str().unwrap().parse().unwrap();
+    let stored = server.get(id).await;
+    assert_eq!(stored.model, expected.project);
+    assert_eq!(stored.migration, expected.migration);
+    let filename = directory.path().join(format!("{id}.json"));
+    let before = std::fs::read(&filename).unwrap();
+    let envelope: Value = serde_json::from_slice(&before).unwrap();
+    assert!(envelope.get("migration").is_none());
+    assert_eq!(envelope["model"]["schema_version"], 3);
+    let validation = server
+        .client
+        .post(server.url(&format!("/projects/{id}/validation")))
+        .json(&json!({"format":"json", "model":old}))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(validation.status(), StatusCode::OK);
+    let validation: Value = validation.json().await.unwrap();
+    assert_eq!(validation["valid"], true);
+    assert_eq!(validation["migration"], summary["migration"]);
+    let mut invalid = stored
+        .model
+        .relations
+        .values()
+        .find(|r| r.kind() == RelationKind::FunctionalExchange)
+        .unwrap()
+        .clone();
+    if let oa_core::RelationBody::FunctionalExchange { source_port, .. } = &mut invalid.body {
+        *source_port = Id::nil();
+    }
+    let rejected = server
+        .client
+        .post(server.url(&format!("/projects/{id}/transactions")))
+        .header("if-match", &etag)
+        .json(&json!({"commands":[Command::EditRelation(invalid.id, invalid)]}))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(rejected.status(), StatusCode::UNPROCESSABLE_ENTITY);
+    assert_eq!(server.get(id).await.revision, 1);
+    assert_eq!(std::fs::read(&filename).unwrap(), before);
+    let source = server
+        .client
+        .get(server.url(&format!("/projects/{id}/source")))
+        .send()
+        .await
+        .unwrap()
+        .text()
+        .await
+        .unwrap();
+    assert!(source.starts_with("arcadia 3;"));
+    assert_eq!(oa_core::language::parse(&source).unwrap(), stored.model);
+    let no_op = server
+        .client
+        .put(server.url(&format!("/projects/{id}/source")))
+        .header("content-type", "text/plain")
+        .header("if-match", &etag)
+        .body(source)
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(no_op.status(), StatusCode::OK);
+    assert_eq!(no_op.headers()["etag"].to_str().unwrap(), etag);
+    assert_eq!(std::fs::read(&filename).unwrap(), before);
+    server.stop().await;
+    let restarted = TestServer::start(directory.path(), None).await;
+    let reopened = restarted.get(id).await;
+    assert_eq!(reopened.model, stored.model);
+    assert_eq!(reopened.revision, 1);
+    assert!(reopened.migration.is_none());
+    let no_op_legacy = restarted
+        .client
+        .put(restarted.url(&format!("/projects/{id}/model")))
+        .header("if-match", &etag)
+        .json(&old)
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(no_op_legacy.status(), StatusCode::OK);
+    assert_eq!(no_op_legacy.headers()["etag"].to_str().unwrap(), etag);
+    let no_op_legacy: Value = no_op_legacy.json().await.unwrap();
+    assert_eq!(no_op_legacy["changed"], false);
+    assert_eq!(no_op_legacy["migration"], summary["migration"]);
+    assert_eq!(std::fs::read(&filename).unwrap(), before);
+    restarted.stop().await;
+}
+
+#[tokio::test]
+async fn reopening_v2_envelope_reports_migration_without_saving_or_advancing_revision() {
+    let directory = tempfile::tempdir().unwrap();
+    let id = Id::new_v4();
+    let filename = directory.path().join(format!("{id}.json"));
+    let old = v2_missing_ports();
+    let original = serde_json::to_vec(&json!({"resource_id":id,"revision":7,"model":old})).unwrap();
+    std::fs::write(&filename, &original).unwrap();
+    let server = TestServer::start(directory.path(), None).await;
+    let migrated = server.get(id).await;
+    assert_eq!(migrated.resource_id, id);
+    assert_eq!(migrated.revision, 7);
+    assert_eq!(migrated.model.schema_version, 3);
+    assert_eq!(
+        migrated
+            .migration
+            .as_ref()
+            .unwrap()
+            .generated_endpoints
+            .len(),
+        2
+    );
+    assert_eq!(std::fs::read(&filename).unwrap(), original);
+    let rename = server
+        .client
+        .patch(server.url(&format!("/projects/{id}")))
+        .header("if-match", migrated.etag())
+        .json(&json!({"name":"Reviewed migrated model"}))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(rename.status(), StatusCode::OK);
+    let current = server.get(id).await;
+    assert_eq!(current.revision, 8);
+    assert_eq!(current.model.name, "Reviewed migrated model");
+    let saved: Value = serde_json::from_slice(&std::fs::read(&filename).unwrap()).unwrap();
+    assert_eq!(saved["revision"], 8);
+    assert_eq!(saved["model"]["schema_version"], 3);
+    assert!(saved.get("migration").is_none());
+    server.stop().await;
+}