use crate::{ app, store::{Store, StoredProject}, }; use oa_core::{Id, Project}; use reqwest::{Client, StatusCode}; use serde_json::{Value, json}; use std::{sync::Arc, time::Duration}; use tokio::{sync::oneshot, time::timeout}; #[tokio::test] async fn prepared_source_http_rechecks_revision_without_holding_catalog() { let directory = tempfile::tempdir().unwrap(); let path = directory.path().to_owned(); let store = Arc::new( tokio::task::spawn_blocking(move || Store::open(path)) .await .unwrap() .unwrap(), ); let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let base = format!("http://{}/api/v1", listener.local_addr().unwrap()); let (stop, done) = oneshot::channel(); let router = app(store.clone(), None); let task = tokio::spawn(async move { axum::serve(listener, router) .with_graceful_shutdown(async { let _ = done.await; }) .await .unwrap(); }); let client = Client::new(); let created = client .post(format!("{base}/projects")) .json(&json!({"name":"Initial review"})) .send() .await .unwrap(); assert_eq!(created.status(), StatusCode::CREATED); let etag = created.headers()["etag"].to_str().unwrap().to_owned(); let summary: Value = created.json().await.unwrap(); let id: Id = summary["resource_id"].as_str().unwrap().parse().unwrap(); let project_url = format!("{base}/projects/{id}"); let initial: StoredProject = client .get(&project_url) .send() .await .unwrap() .error_for_status() .unwrap() .json() .await .unwrap(); let (entered, release) = store.pause_next_replacement(id, etag.clone()); // Real parsing, legacy admission and source preparation all run. Runtime // cost is irrelevant: the gate proves the exact pre-commit phase instead. let source = "arcadia 2; project prepared \"Prepared stale head\" { oa_activity root \"Root\" {} oa_activity child \"Child\" { parent @root; } }"; let proposed_client = client.clone(); let proposed_url = format!("{project_url}/source"); let proposed_etag = etag.clone(); let proposed = tokio::spawn(async move { proposed_client .put(proposed_url) .header("if-match", proposed_etag) .header("content-type", "text/plain") .body(source) .send() .await .unwrap() }); let evidence = timeout(Duration::from_secs(10), entered) .await .expect("source did not reach validated pre-commit phase") .unwrap(); assert_eq!(evidence.model.name, "Prepared stale head"); assert_eq!(evidence.model.elements.len(), 2); let compiled = oa_core::source_project::compile_bundle(&evidence.bundle, crate::store::SOURCE_LIMITS) .unwrap(); assert_eq!(compiled.decoded.project, evidence.model); assert!( !proposed.is_finished(), "prepared request must still overlap the winner" ); let catalog = timeout( Duration::from_secs(2), client.get(format!("{base}/projects")).send(), ) .await .expect("catalog locked during admitted candidate lifetime") .unwrap(); assert_eq!(catalog.status(), StatusCode::OK); let winner = timeout( Duration::from_secs(2), client .patch(&project_url) .header("if-match", &etag) .json(&json!({"name":"Concurrent accepted review"})) .send(), ) .await .expect("prepared worker held the commit lock") .unwrap(); assert_eq!(winner.status(), StatusCode::OK); let winner_etag = winner.headers()["etag"].to_str().unwrap().to_owned(); assert!( !proposed.is_finished(), "stale candidate must await explicit release" ); release.release(); let loser = timeout(Duration::from_secs(10), proposed) .await .expect("released candidate did not finish") .unwrap(); assert_eq!(loser.status(), StatusCode::PRECONDITION_FAILED); let failure: Value = loser.json().await.unwrap(); assert_eq!(failure["code"], "revision_conflict"); assert_eq!(failure["current_revision"], 2); let committed_response = client.get(&project_url).send().await.unwrap(); assert_eq!( committed_response.headers()["etag"].to_str().unwrap(), winner_etag ); let committed: StoredProject = committed_response .error_for_status() .unwrap() .json() .await .unwrap(); let mut expected_model: Project = initial.model; expected_model.name = "Concurrent accepted review".into(); assert_eq!(committed.model, expected_model); assert_eq!(committed.revision, 2); assert!(committed.model.elements.is_empty()); let installed: StoredProject = serde_json::from_slice( &std::fs::read(directory.path().join(format!("{id}.json"))).unwrap(), ) .unwrap(); assert_eq!(installed.model, committed.model); assert_eq!(installed.source_project, committed.source_project); assert_eq!(installed.revision, 2); assert_eq!(installed.resource_id, committed.resource_id); assert_eq!(installed.migration, committed.migration); let _ = stop.send(()); drop(client); timeout(Duration::from_secs(5), task) .await .unwrap() .unwrap(); }