//! Deterministic parent/index publication tests using an owned protocol fixture. //! Genuine parser execution is covered separately in documents.rs. use anyhow::Result; use searchi_core::{ extraction_protocol::{DocumentFormat, RequestHeader}, IndexOptions, SearchIndex, }; use std::{ fs, sync::atomic::{AtomicBool, Ordering}, time::{Duration, Instant}, }; #[test] fn mid_worker_same_stat_changes_and_cancellation_roll_back_index() -> Result<()> { for cancel in [false, true] { let directory = tempfile::tempdir()?; let root = directory.path().join("files"); fs::create_dir(&root)?; let source = root.join("contract.pdf"); fs::write(root.join("notes.txt"), "preserved evidence")?; let mut index = SearchIndex::open(":memory:")?; index.index_folder(&root)?; fs::write(&source, b"original")?; let initial = fs::metadata(&source)?; let digest = RequestHeader::from_bytes(DocumentFormat::Pdf, b"original")?.input_sha256; let marker = directory.path().join("admitted"); let release = directory.path().join("release"); let fixture_source = directory.path().join("worker.rs"); let executable = directory .path() .join(format!("fixture{}", std::env::consts::EXE_SUFFIX)); let response = serde_json::json!({ "version":1,"input_sha256":digest, "outcome":{"status":"extracted","passages":[{"text":"stale document text","source":{"kind":"page","number":1}}]} }).to_string(); fs::write( &fixture_source, format!( r#" use std::io::{{Read,Write}}; fn main() {{ let mut bytes=Vec::new(); std::io::stdin().read_to_end(&mut bytes).unwrap(); std::fs::write({marker:?}, b"admitted").unwrap(); while !std::path::Path::new({release:?}).exists() {{ std::thread::sleep(std::time::Duration::from_millis(5)); }} let response={response:?}.as_bytes(); let mut out=std::io::stdout(); out.write_all(&(response.len() as u32).to_le_bytes()).unwrap(); out.write_all(response).unwrap(); }} "# ), )?; let status = std::process::Command::new("rustc") .arg(&fixture_source) .arg("-o") .arg(&executable) .status()?; anyhow::ensure!(status.success(), "Compile owned protocol fixture"); index.set_document_worker(executable); let cancelled = AtomicBool::new(false); std::thread::scope(|scope| { let changer = scope.spawn(|| -> Result<()> { let deadline = Instant::now() + Duration::from_secs(10); while !marker.exists() && Instant::now() < deadline { std::thread::sleep(Duration::from_millis(5)); } if !marker.exists() { cancelled.store(true, Ordering::Release); anyhow::bail!("Worker did not admit fixture input"); } if cancel { cancelled.store(true, Ordering::Release); } else { fs::write(&source, b"modified")?; fs::File::options() .write(true) .open(&source)? .set_times(fs::FileTimes::new().set_modified(initial.modified()?))?; assert_eq!(fs::metadata(&source)?.len(), initial.len()); assert_eq!(fs::metadata(&source)?.modified()?, initial.modified()?); fs::write(&release, b"release")?; } Ok(()) }); let result = index.index_folder_cancellable(&root, &IndexOptions::default(), &cancelled); assert!( result.is_err(), "Mutation or cancellation must roll back publication" ); let message = format!("{:#}", result.unwrap_err()); assert!( message.contains(if cancel { "cancelled" } else { "changed during extraction" }), "{message}" ); changer.join().unwrap().unwrap(); }); assert_eq!(index.stats()?.files, 1); assert_eq!(index.search("evidence", 10)?.len(), 1); assert!(index.search("stale", 10)?.is_empty()); assert!(!index.contains_path(&source.to_string_lossy())?); } Ok(()) }