Use isolated document extraction in shared indexing and desktop reconciliation

crates/searchi-core/src/extraction.rs

@@ -1,4 +1,4 @@
-//! Bounded inputs and outputs; parser allocations and CPU are not process isolated.
+//! Bounded snapshots; documents use the configured isolated process adapter.
 use crate::SourceLocation;
 use anyhow::{bail, Context, Result};
 use quick_xml::{events::Event, name::ResolveResult, NsReader, XmlVersion};
@@ -7,8 +7,36 @@ use std::{
     fs::File,
     io::{Cursor, Read},
     path::Path,
+    sync::atomic::AtomicBool,
+    time::Duration,
 };
 
+pub(crate) enum DocumentExtraction {
+    Worker(crate::extraction_process::ProcessConfig),
+    InternalInProcess,
+}
+
+impl DocumentExtraction {
+    pub fn normal() -> Result<Self> {
+        let executable = match std::env::var_os("SEARCHI_EXTRACT_WORKER") {
+            Some(path) => std::path::PathBuf::from(path),
+            None => std::env::current_exe()?.with_file_name(if cfg!(windows) {
+                "searchi-extract-worker.exe"
+            } else {
+                "searchi-extract-worker"
+            }),
+        };
+        Ok(Self::worker(executable))
+    }
+
+    pub fn worker(executable: std::path::PathBuf) -> Self {
+        Self::Worker(crate::extraction_process::ProcessConfig {
+            executable,
+            timeout: Duration::from_secs(30),
+        })
+    }
+}
+
 const MAX_TEXT: usize = 2 * 1024 * 1024;
 const MAX_DOCUMENT: usize = 32 * 1024 * 1024;
 const MAX_XML: usize = 8 * 1024 * 1024;
@@ -75,17 +103,65 @@ pub(crate) fn read_source(path: &Path) -> Result<Option<SourceInput>> {
     Ok(Some(SourceInput { bytes, hash }))
 }
 
-pub(crate) fn extract(path: &Path, input: SourceInput) -> Result<Option<Vec<Passage>>> {
+pub(crate) fn verify_document_snapshot(
+    path: &Path,
+    expected_hash: &str,
+    initial: &std::fs::Metadata,
+) -> Result<()> {
+    let verified = read_source(path)?.context("Document source no longer readable")?;
+    let after = std::fs::metadata(path)?;
+    if verified.hash != expected_hash
+        || after.len() != initial.len()
+        || after.modified()? != initial.modified()?
+    {
+        bail!("Document changed during extraction; retry indexing; reconciliation rolled back");
+    }
+    Ok(())
+}
+
+pub(crate) fn extract(
+    path: &Path,
+    input: SourceInput,
+    documents: &DocumentExtraction,
+    cancelled: Option<&AtomicBool>,
+) -> Result<Option<Vec<Passage>>> {
     let extension = path
         .extension()
         .and_then(|value| value.to_str())
         .unwrap_or("")
         .to_ascii_lowercase();
-    if extension == "pdf" {
-        return pdf(&input.bytes).map(Some);
-    }
-    if extension == "docx" {
-        return docx(&input.bytes).map(Some);
+    let format = match extension.as_str() {
+        "pdf" => Some(crate::extraction_protocol::DocumentFormat::Pdf),
+        "docx" => Some(crate::extraction_protocol::DocumentFormat::Docx),
+        _ => None,
+    };
+    if let Some(format) = format {
+        if matches!(documents, DocumentExtraction::InternalInProcess) {
+            return document_bytes(format, &input.bytes).map(Some);
+        }
+        let DocumentExtraction::Worker(config) = documents else {
+            unreachable!()
+        };
+        let request = crate::extraction_protocol::RequestHeader::from_bytes(format, &input.bytes)?;
+        let never_cancelled = AtomicBool::new(false);
+        let response = crate::extraction_process::run(
+            config,
+            request,
+            input.bytes,
+            cancelled.unwrap_or(&never_cancelled),
+        )?;
+        return match response.outcome {
+            crate::extraction_protocol::ExtractionOutcome::Extracted { passages } => Ok(Some(
+                passages
+                    .into_iter()
+                    .map(|p| Passage {
+                        text: p.text,
+                        source: p.source,
+                    })
+                    .collect(),
+            )),
+            crate::extraction_protocol::ExtractionOutcome::Rejected { reason } => bail!("{reason}"),
+        };
     }
     let code = crate::is_code_path(path);
     if input.bytes.contains(&0) {
@@ -327,8 +403,28 @@ mod provenance_tests {
             format!("{:x}", Sha256::digest(b"original parser input"))
         );
         std::fs::write(&path, "replacement on disk")?;
-        let passages = extract(&path, input)?.unwrap();
+        let passages =
+            extract(&path, input, &DocumentExtraction::InternalInProcess, None)?.unwrap();
         assert_eq!(passages[0].text, "original parser input");
         Ok(())
     }
+
+    #[test]
+    fn document_publication_rejects_same_stat_byte_changes() -> Result<()> {
+        let directory = tempfile::tempdir()?;
+        let path = directory.path().join("snapshot.pdf");
+        std::fs::write(&path, b"original")?;
+        let metadata = std::fs::metadata(&path)?;
+        let input = read_source(&path)?.unwrap();
+        verify_document_snapshot(&path, &input.hash, &metadata)?;
+        std::fs::write(&path, b"modified")?;
+        File::options()
+            .write(true)
+            .open(&path)?
+            .set_times(std::fs::FileTimes::new().set_modified(metadata.modified()?))?;
+        assert_eq!(std::fs::metadata(&path)?.len(), metadata.len());
+        assert_eq!(std::fs::metadata(&path)?.modified()?, metadata.modified()?);
+        assert!(verify_document_snapshot(&path, &input.hash, &metadata).is_err());
+        Ok(())
+    }
 }

crates/searchi-core/src/lib.rs

@@ -1,8 +1,8 @@
 //! Offline lexical search engine shared by the CLI and desktop shell.
 //!
 //! Reconciliation is atomic. Incomplete directory walks never delete records.
-//! Text and document extraction have byte limits, but still run in this process.
-//! Worker isolation remains a production milestone.
+//! Document extraction uses a bounded helper process; plain text/code snapshots
+//! remain in-process. Native platform release validation remains a milestone.
 mod code;
 mod extraction;
 pub mod extraction_process;
@@ -28,6 +28,8 @@ use walkdir::WalkDir;
 const SCHEMA_VERSION: i64 = 13;
 const MAX_RESULTS: usize = 1000;
 
+type PreviousExtraction = (i64, i64, String, Option<String>, Option<String>);
+
 #[derive(Debug, Default, Serialize)]
 pub struct IndexReport {
     pub discovered: usize,
@@ -113,6 +115,7 @@ pub struct SearchIndex {
     connection: Connection,
     database_path: Option<PathBuf>,
     gram_frequencies: RefCell<HashMap<String, i64>>,
+    document_extraction: extraction::DocumentExtraction,
 }
 
 impl SearchIndex {
@@ -287,9 +290,22 @@ impl SearchIndex {
             connection,
             database_path,
             gram_frequencies: RefCell::new(HashMap::new()),
+            document_extraction: extraction::DocumentExtraction::normal()?,
         })
     }
 
+    /// Select the trusted packaged document helper (for example desktop resources).
+    /// Failure to launch never falls back to parsing documents in this process.
+    pub fn set_document_worker(&mut self, executable: impl Into<PathBuf>) {
+        self.document_extraction = extraction::DocumentExtraction::worker(executable.into());
+    }
+
+    /// Internal parser-fixture adapter. Never enable in a distributed application.
+    #[doc(hidden)]
+    pub fn use_in_process_document_parser_for_tests(&mut self) {
+        self.document_extraction = extraction::DocumentExtraction::InternalInProcess;
+    }
+
     /// SHA-256 of the exact raw bytes used for the indexed extraction. Legacy
     /// records return None until reindexed; callers must not infer provenance
     /// from size/mtime or hash a later file read on their behalf.
@@ -653,11 +669,11 @@ impl SearchIndex {
                     .and_then(|value| value.to_str())
                     .unwrap_or("")
                     .to_ascii_lowercase();
-                let previous: Option<(i64, i64, String, Option<String>)> = transaction
+                let previous: Option<PreviousExtraction> = transaction
                     .query_row(
-                        "SELECT size,mtime,root,extracted_sha256 FROM files WHERE path=?1",
+                        "SELECT size,mtime,root,extracted_sha256,extraction_error FROM files WHERE path=?1",
                         [&path],
-                        |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
+                        |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?, row.get(4)?)),
                     )
                     .optional()?;
                 // Supported inputs are bounded and fingerprinted even when file
@@ -685,25 +701,41 @@ impl SearchIndex {
                     .and_then(|input| input.as_ref())
                     .map(|input| input.hash.clone());
                 if !media_hint
-                    && previous
-                        .as_ref()
-                        .is_some_and(|(old_size, old_mtime, old_root, old_hash)| {
+                    && previous.as_ref().is_some_and(
+                        |(old_size, old_mtime, old_root, old_hash, old_error)| {
                             *old_size == size
                                 && *old_mtime == mtime
                                 && old_root == &root_key
                                 && input.is_ok()
                                 && old_hash == &source_hash
-                        })
+                                && old_error.is_none()
+                        },
+                    )
                 {
                     report.unchanged += 1;
                     continue;
                 }
                 let extracted = input.and_then(|input| match input {
-                    Some(input) => extraction::extract(entry.path(), input),
+                    Some(input) => extraction::extract(
+                        entry.path(),
+                        input,
+                        &self.document_extraction,
+                        cancelled,
+                    ),
                     None => Ok(None),
                 });
                 let (extracted, extraction_error) = match extracted {
                     Ok(extracted) => (extracted, None),
+                    Err(error)
+                        if error
+                            .downcast_ref::<extraction_process::ProcessError>()
+                            .is_some_and(|error| {
+                                matches!(error, extraction_process::ProcessError::Cancelled)
+                            }) =>
+                    {
+                        return Err(error)
+                            .context("Index reconciliation cancelled; transaction rolled back");
+                    }
                     Err(error) if matches!(extension.as_str(), "pdf" | "docx") => {
                         report.failures += 1;
                         let reason: String = format!("{error:#}").chars().take(2048).collect();
@@ -714,6 +746,12 @@ impl SearchIndex {
                             .with_context(|| format!("read {path}; reconciliation rolled back"))
                     }
                 };
+                check_index_cancelled(cancelled)?;
+                if matches!(extension.as_str(), "pdf" | "docx") {
+                    if let Some(hash) = &source_hash {
+                        extraction::verify_document_snapshot(entry.path(), hash, &metadata)?;
+                    }
+                }
                 if extracted.is_none() {
                     report.skipped += 1;
                 }

crates/searchi-core/tests/document_publication.rs

@@ -0,0 +1,107 @@
+//! 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(())
+}

crates/searchi-core/tests/documents.rs

@@ -57,6 +57,7 @@ fn compressed_document_limits_are_enforced() -> Result<()> {
     );
     docx(&root.path().join("over-limit.docx"), &xml)?;
     let mut index = SearchIndex::open(":memory:")?;
+    index.use_in_process_document_parser_for_tests();
     assert_eq!(index.index_folder(root.path())?.failures, 2);
     assert_eq!(index.stats()?.content_files, 0);
     assert!(index.search("bomb.pdf", 10)?[0].extraction_error.is_some());
@@ -90,6 +91,7 @@ fn encrypted_pdf_is_explicitly_unsupported() -> Result<()> {
     document.encrypt(&state)?;
     document.save(&path)?;
     let mut index = SearchIndex::open(":memory:")?;
+    index.use_in_process_document_parser_for_tests();
     assert_eq!(index.index_folder(root.path())?.failures, 1);
     assert!(index.search("protected.pdf", 10)?[0]
         .extraction_error
@@ -122,6 +124,72 @@ fn docx(path: &Path, xml: &str) -> Result<()> {
     Ok(())
 }
 
+#[test]
+fn missing_worker_preserves_filename_clears_passages_and_retries() -> Result<()> {
+    let root = tempfile::tempdir()?;
+    pdf(&root.path().join("agreement.pdf"))?;
+    let mut index = SearchIndex::open(":memory:")?;
+    index.use_in_process_document_parser_for_tests();
+    index.index_folder(root.path())?;
+    assert!(!index.search("cancellation", 10)?.is_empty());
+    index.set_document_worker(root.path().join("missing-helper.exe"));
+    // Force a changed snapshot so a previous successful extraction is replaced.
+    pdf_with_text(
+        &root.path().join("agreement.pdf"),
+        &["different cancellation"],
+    )?;
+    assert_eq!(index.index_folder(root.path())?.failures, 1);
+    assert!(index.search("cancellation", 10)?.is_empty());
+    assert!(index.search("agreement.pdf", 10)?[0]
+        .extraction_error
+        .is_some());
+    // A same-stat, same-byte failed source must not be skipped on retry.
+    index.use_in_process_document_parser_for_tests();
+    assert_eq!(index.index_folder(root.path())?.indexed, 1);
+    assert!(!index.search("cancellation", 10)?.is_empty());
+    Ok(())
+}
+
+#[test]
+#[ignore = "requires built helper in SEARCHI_EXTRACT_WORKER"]
+fn normal_adapter_uses_real_document_worker_and_cancellation_rolls_back() -> Result<()> {
+    anyhow::ensure!(
+        std::env::var_os("SEARCHI_EXTRACT_WORKER").is_some(),
+        "Set actual document helper path"
+    );
+    let root = tempfile::tempdir()?;
+    pdf(&root.path().join("agreement.pdf"))?;
+    let mut index = SearchIndex::open(":memory:")?;
+    assert_eq!(index.index_folder(root.path())?.failures, 0);
+    assert_eq!(
+        index.search("cancellation", 10)?[0]
+            .source
+            .as_ref()
+            .unwrap()
+            .number,
+        2
+    );
+    let before = index.stats()?.files;
+    let cancelled = std::sync::atomic::AtomicBool::new(true);
+    assert!(index
+        .index_folder_cancellable(
+            root.path(),
+            &searchi_core::IndexOptions::default(),
+            &cancelled
+        )
+        .is_err());
+    assert_eq!(index.stats()?.files, before);
+    assert_eq!(
+        index.search("cancellation", 10)?[0]
+            .source
+            .as_ref()
+            .unwrap()
+            .number,
+        2
+    );
+    Ok(())
+}
+
 #[test]
 fn real_pdf_and_docx_return_matching_locations_and_persist() -> Result<()> {
     let root = tempfile::tempdir()?;
@@ -132,6 +200,7 @@ fn real_pdf_and_docx_return_matching_locations_and_persist() -> Result<()> {
     )?;
     let db = root.path().join("index.db");
     let mut index = SearchIndex::open(&db)?;
+    index.use_in_process_document_parser_for_tests();
     assert_eq!(index.index_folder(root.path())?.indexed, 2);
     let result = index.search("cancellation", 10)?;
     assert_eq!(result[0].source.as_ref().unwrap().kind, "page");
@@ -180,6 +249,7 @@ fn corrupt_document_reports_failure_and_preserves_filename_search() -> Result<()
     let root = tempfile::tempdir()?;
     fs::write(root.path().join("notes.txt"), "persistent evidence")?;
     let mut index = SearchIndex::open(":memory:")?;
+    index.use_in_process_document_parser_for_tests();
     index.index_folder(root.path())?;
     fs::write(root.path().join("broken.pdf"), b"not a PDF")?;
     assert_eq!(index.index_folder(root.path())?.failures, 1);
@@ -199,7 +269,8 @@ fn corrupt_document_reports_failure_and_preserves_filename_search() -> Result<()
     index.index_folder(root.path())?;
     assert_eq!(index.search("cancellation", 10)?.len(), 1);
     fs::write(root.path().join("agreement.pdf"), "corrupted replacement")?;
-    assert_eq!(index.index_folder(root.path())?.failures, 1);
+    // Previously rejected DOCX is retried too; failures are per current scan.
+    assert_eq!(index.index_folder(root.path())?.failures, 2);
     assert!(index.search("cancellation", 10)?.is_empty());
     assert!(index.search("agreement.pdf", 10)?[0]
         .extraction_error