ARCADIA-Studio/crates/oa-server/src/store_preparation_gate.rs

//! Private synchronization for real-HTTP tests; absent from production builds.
use super::{ApiError, ApiResult, Id, PreparedProject, Project, SourceBundle, Store};
use std::{sync::mpsc, time::Duration};
use tokio::sync::oneshot;

pub(crate) struct PreparedEvidence {
    pub model: Project,
    pub bundle: SourceBundle,
}

pub(crate) struct Armed {
    id: Id,
    expected: String,
    entered: oneshot::Sender<PreparedEvidence>,
    resume: mpsc::Receiver<bool>,
}

/// Dropping an unfinished test cancels the worker instead of stranding it.
pub(crate) struct ReleaseOnDrop(Option<mpsc::Sender<bool>>);

impl ReleaseOnDrop {
    pub(crate) fn release(mut self) {
        self.0.take().unwrap().send(true).unwrap();
    }
}

impl Drop for ReleaseOnDrop {
    fn drop(&mut self) {
        if let Some(sender) = self.0.take() {
            let _ = sender.send(false);
        }
    }
}

impl Store {
    pub(crate) fn pause_next_replacement(
        &self,
        id: Id,
        expected: String,
    ) -> (oneshot::Receiver<PreparedEvidence>, ReleaseOnDrop) {
        let (entered, receiver) = oneshot::channel();
        let (resume, wait) = mpsc::channel();
        let mut slot = self.preparation_gate.lock().unwrap();
        assert!(slot.is_none(), "one gate per store instance");
        *slot = Some(Armed {
            id,
            expected,
            entered,
            resume: wait,
        });
        (receiver, ReleaseOnDrop(Some(resume)))
    }

    pub(super) fn wait_after_preparation(
        &self,
        id: Id,
        expected: &str,
        prepared: &PreparedProject,
    ) -> ApiResult<()> {
        let armed = {
            let mut slot = self.preparation_gate.lock().unwrap();
            if slot
                .as_ref()
                .is_some_and(|gate| gate.id == id && gate.expected == expected)
            {
                slot.take()
            } else {
                None
            }
        };
        let Some(armed) = armed else {
            return Ok(());
        };
        // Neither the catalog nor the gate mutex is held while the real model
        // worker waits. Evidence comes from the already validated candidate.
        armed
            .entered
            .send(PreparedEvidence {
                model: prepared.model().clone(),
                bundle: prepared.bundle().clone(),
            })
            .map_err(|_| ApiError::internal("Preparation test observer disappeared."))?;
        match armed.resume.recv_timeout(Duration::from_secs(15)) {
            Ok(true) => Ok(()),
            _ => Err(ApiError::internal(
                "Preparation test cancelled or timed out.",
            )),
        }
    }
}

#[tokio::test]
async fn preparation_gate_drop_cancels_without_installing_candidate() {
    let directory = tempfile::tempdir().unwrap();
    let path = directory.path().to_owned();
    let (store, initial) = tokio::task::spawn_blocking(move || {
        let store = std::sync::Arc::new(Store::open(path).unwrap());
        let initial = store.create(Project::empty()).unwrap();
        (store, initial)
    })
    .await
    .unwrap();
    let original_bytes = std::fs::read(
        directory
            .path()
            .join(format!("{}.json", initial.resource_id)),
    )
    .unwrap();
    let (entered, release) = store.pause_next_replacement(initial.resource_id, initial.etag());
    let mut model = initial.model.clone();
    model.name = "Must not be installed".into();
    let worker_store = store.clone();
    let id = initial.resource_id;
    let etag = initial.etag();
    let worker = tokio::task::spawn_blocking(move || {
        worker_store.replace_decoded(
            id,
            &etag,
            oa_core::persistence::DecodeResult {
                project: model,
                migration: None,
            },
        )
    });
    tokio::time::timeout(Duration::from_secs(10), entered)
        .await
        .unwrap()
        .unwrap();
    drop(release);
    let error = tokio::time::timeout(Duration::from_secs(2), worker)
        .await
        .expect("dropping the test guard must release its worker")
        .unwrap()
        .unwrap_err();
    assert_eq!(error.detail, "Preparation test cancelled or timed out.");
    assert_eq!(store.get(id).unwrap().model, initial.model);
    assert_eq!(
        std::fs::read(directory.path().join(format!("{id}.json"))).unwrap(),
        original_bytes
    );
}