diff --git a/crates/persistence/src/backends/mongodb/reindex_pipeline.rs b/crates/persistence/src/backends/mongodb/reindex_pipeline.rs index a550f2d7cd..31e4e91f81 100644 --- a/crates/persistence/src/backends/mongodb/reindex_pipeline.rs +++ b/crates/persistence/src/backends/mongodb/reindex_pipeline.rs @@ -532,6 +532,17 @@ impl PageFailure { }) .collect() } + + /// [`Self::fan_out`] for a metadata-only page (#2008): every resource + /// fails and none had its metadata written. + pub(super) fn fan_out_metadata(&self, n: usize) -> (Vec>, u64) { + let results = self + .fan_out(n) + .into_iter() + .map(|outcome| outcome.map(|_| ())) + .collect(); + (results, 0) + } } impl From for PageFailure { @@ -679,7 +690,19 @@ async fn update_missing_metadata( db: &mongodb::Database, updates: &[(usize, Document)], ) -> Result, PageFailure> { + update_missing_metadata_counted(db, updates) + .await + .map(|(failures, _)| failures) +} + +/// [`update_missing_metadata`], also returning how many resources the updates +/// matched: what a metadata-only run reports as written (#2008). +async fn update_missing_metadata_counted( + db: &mongodb::Database, + updates: &[(usize, Document)], +) -> Result<(HashMap, u64), PageFailure> { let mut failures = HashMap::new(); + let mut matched_total = 0u64; for chunk in updates.chunks(REINDEX_SUBBATCH_MAX) { let statements: Vec<_> = chunk.iter().map(|(_, update)| update.clone()).collect(); let reply = db @@ -693,11 +716,47 @@ async fn update_missing_metadata( PageFailure::driver("Failed to update reindex missing metadata", error) .with_permanent(failures.clone()) })?; - let (_, write_failures) = metadata_update_result(&reply, chunk) + let (matched, write_failures) = metadata_update_result(&reply, chunk) .map_err(|failure| failure.with_permanent(failures.clone()))?; + matched_total += matched as u64; failures.extend(write_failures); } - Ok(failures) + Ok((failures, matched_total)) +} + +/// One resource's metadata-only update (#2008): the rebuild's +/// [`SubBatchDocs::missing_updates`] statement, plus, with `skip_existing`, a +/// guard that leaves alone a resource whose `missing_params` already carries +/// this extraction's fingerprint (#1999). An unstamped array (written before +/// #1999) or a stale stamp is rewritten, since searches distrust both. +fn metadata_only_update( + tenant_id: &str, + resource: &StoredResource, + missing_params: Option<&[String]>, + missing_fp: Option<&str>, + skip_existing: bool, +) -> Document { + let mut filter = resource_version_filter(tenant_id, resource); + if skip_existing { + // Skip only arrays already stamped with this extraction's + // fingerprint; unstamped (#1984-era) or stale stamps are rewritten. + match missing_fp { + Some(fp) => filter.insert( + "$or", + vec![ + doc! { "missing_params": { "$exists": false } }, + doc! { "missing_params_fp": { "$ne": fp } }, + ], + ), + None => filter.insert("missing_params", doc! { "$exists": false }), + }; + } + doc! { + "q": filter, + "u": super::storage::with_missing_metadata_update(doc! {}, missing_params, missing_fp), + "multi": false, + "upsert": false, + } } /// What a database task hands back to the page's own thread when joined: @@ -1560,6 +1619,63 @@ impl super::MongoBackend { self.search_index_documents_checked(tenant_id, r.resource_type(), r.id(), r.content()) } + /// Metadata-only page write (#2008): extracts every resource exactly as a + /// rebuild does, then writes only its `missing_params`, under the same + /// version guard, and never touches `search_index` or + /// `search_index_contained`. A resource whose extraction failed has its + /// metadata removed rather than claimed, as in a rebuild, and is reported + /// as failed. Returns the per-resource outcomes and how many resources the + /// updates matched. + pub(super) async fn write_metadata_page( + &self, + db: &mongodb::Database, + tenant_id: &str, + resources: &[StoredResource], + skip_existing: bool, + ) -> Result<(Vec>, u64), PageFailure> { + let n = resources.len(); + let prepare = |i: usize| self.extract_one(tenant_id, &resources[i]); + let pool = if tokio_multi_thread_runtime() && n >= REINDEX_SERIAL_POOL_MIN_PAGE { + self.reindex_prepare_pool() + } else { + None + }; + let prepared: Vec<(SearchIndexDocuments, Option)> = if pool.is_some() { + let env = PrepareEnv { + pool, + gate: self.reindex_prepare_gate(), + }; + extract_range(&env, 0..n, &prepare).0 + } else { + (0..n).map(&prepare).collect() + }; + + let mut extract_failures: Vec> = Vec::with_capacity(n); + let mut updates: Vec<(usize, Document)> = Vec::with_capacity(n); + for (i, (docs, failure)) in prepared.into_iter().enumerate() { + updates.push(( + i, + metadata_only_update( + tenant_id, + &resources[i], + docs.missing_params.as_deref(), + docs.missing_fp.as_deref(), + skip_existing, + ), + )); + extract_failures.push(failure); + } + + let (write_failures, matched) = update_missing_metadata_counted(db, &updates) + .await + .map_err(|failure| failure.with_outcomes(extract_failures.clone(), HashMap::new()))?; + let outcomes = page_outcomes(extract_failures, write_failures, &vec![0; n]) + .into_iter() + .map(|outcome| outcome.map(|_| ())) + .collect(); + Ok((outcomes, matched)) + } + /// Serial writer: extracts every resource, deletes the page's stale /// entries, then inserts the new ones, one command at a time (own rows, /// then contained rows). Uses the rayon pool only for pages of at least @@ -2116,6 +2232,20 @@ mod tests { )); } + #[test] + fn metadata_fan_out_fails_every_resource_and_writes_none() { + let mut permanent = HashMap::new(); + permanent.insert(1, "metadata rejected".to_string()); + let failure = + PageFailure::driver("metadata page", command_error(2)).with_permanent(permanent); + let (results, written) = failure.fan_out_metadata(3); + assert_eq!(written, 0); + assert_eq!(results.len(), 3); + assert!(matches!(&results[0], Err(e) if e.to_string().contains("metadata page"))); + assert!(matches!(&results[1], Err(e) if e.to_string().contains("metadata rejected"))); + assert!(matches!(&results[2], Err(e) if e.to_string().contains("metadata page"))); + } + #[test] fn fan_out_copies_one_classification_and_shares_its_original_source() { let failure = PageFailure::driver("delete page", command_error(2)); diff --git a/crates/persistence/src/backends/mongodb/storage.rs b/crates/persistence/src/backends/mongodb/storage.rs index def78b92dd..18dbd2d2bb 100644 --- a/crates/persistence/src/backends/mongodb/storage.rs +++ b/crates/persistence/src/backends/mongodb/storage.rs @@ -7722,6 +7722,33 @@ impl ReindexTarget for MongoBackend { self.write_search_entries_page_timed(tenant, resources, &mut stats) .await } + + /// `missing_params` is this backend's search metadata (#1984), and only + /// while it owns search: an offloaded deployment never reads it. + fn supports_search_metadata(&self) -> bool { + !self.is_search_offloaded() + } + + /// Backfills `missing_params` without a rebuild (#2008); see + /// `write_metadata_page`. + async fn write_search_metadata_page( + &self, + tenant: &TenantContext, + resources: &[StoredResource], + skip_existing: bool, + ) -> (Vec>, u64) { + if resources.is_empty() { + return (Vec::new(), 0); + } + let n = resources.len(); + let db = match self.get_database().await { + Ok(db) => db, + Err(e) => return super::reindex_pipeline::PageFailure::storage(e).fan_out_metadata(n), + }; + self.write_metadata_page(&db, tenant.tenant_id().as_str(), resources, skip_existing) + .await + .unwrap_or_else(|failure| failure.fan_out_metadata(n)) + } } /// Documents per `insert_many` when the serial or overlapped writer's diff --git a/crates/persistence/src/search/reindex.rs b/crates/persistence/src/search/reindex.rs index e1838e1580..94779a44b8 100644 --- a/crates/persistence/src/search/reindex.rs +++ b/crates/persistence/src/search/reindex.rs @@ -571,6 +571,45 @@ pub trait ReindexTarget: Send + Sync { ) -> Vec> { self.write_search_entries_page(tenant, resources).await } + + /// Whether this writer keeps per-resource search metadata that a + /// metadata-only run can rewrite on its own + /// ([`ReindexRequest::metadata_only`], #2008). MongoDB's `missing_params` + /// is the only such metadata today; the default is `false`, and a + /// metadata-only run skips a writer that says so. + fn supports_search_metadata(&self) -> bool { + false + } + + /// Recomputes and writes only the per-resource search metadata of a page, + /// leaving the search index itself untouched (#2008). Returns one result + /// per resource, in input order, and how many resources had their + /// metadata written. A resource whose stored version is no longer the one + /// in `resources` is left alone: the write that superseded it wrote its + /// own metadata. With `skip_existing`, a resource whose metadata is + /// already current is left alone too. + /// + /// Called only on a writer whose [`Self::supports_search_metadata`] is + /// `true`; the default refuses every resource. + async fn write_search_metadata_page( + &self, + tenant: &TenantContext, + resources: &[StoredResource], + skip_existing: bool, + ) -> (Vec>, u64) { + let _ = (tenant, skip_existing); + let results = resources + .iter() + .map(|_| { + Err(crate::error::BackendError::UnsupportedCapability { + backend_name: "reindex target".to_string(), + capability: "metadata-only reindex".to_string(), + } + .into()) + }) + .collect(); + (results, 0) + } } /// A backend that is both a [`ReindexSource`] and a [`ReindexTarget`] — @@ -605,6 +644,19 @@ pub struct ReindexRequest { #[serde(default)] pub clear_only: bool, + /// Recompute and write only each resource's search metadata (MongoDB's + /// `missing_params`), with the same extraction and version guard as a + /// rebuild, and leave the search index untouched (#2008). Runs only on + /// writers whose [`ReindexTarget::supports_search_metadata`] is `true`. + #[serde(default)] + pub metadata_only: bool, + + /// With [`Self::metadata_only`], leave alone the resources whose metadata + /// is already current, so only missing, unstamped or stale metadata is + /// written (#2008). Every resource in scope is still read and extracted. + #[serde(default)] + pub skip_existing_metadata: bool, + /// First resource id to reindex (included). Requires exactly one /// resource type (#1739). #[serde(default)] @@ -652,6 +704,8 @@ impl Default for ReindexRequest { batch_size: default_batch_size(), clear_existing: false, clear_only: false, + metadata_only: false, + skip_existing_metadata: false, id_start: None, id_end: None, bulk_index_rebuild: false, @@ -725,6 +779,19 @@ impl ReindexRequest { self } + /// Writes only search metadata, not the search index (see the field). + pub fn metadata_only(mut self) -> Self { + self.metadata_only = true; + self + } + + /// With [`Self::metadata_only`], skips resources whose metadata is + /// already current (see the field). + pub fn skip_existing_metadata(mut self) -> Self { + self.skip_existing_metadata = true; + self + } + /// The id range this request selects, or `None` when it has no bounds. pub fn id_range(&self) -> Option { (self.id_start.is_some() || self.id_end.is_some()).then(|| ReindexIdRange { @@ -747,6 +814,20 @@ impl ReindexRequest { if self.clear_only && self.bulk_index_rebuild { return invalid("clearOnly cannot be combined with a bulk index rebuild"); } + if self.skip_existing_metadata && !self.metadata_only { + return invalid("skipExistingMetadata requires metadataOnly"); + } + if self.metadata_only { + if self.clear_existing { + return invalid("metadataOnly cannot be combined with clearExisting"); + } + if self.clear_only { + return invalid("metadataOnly cannot be combined with clearOnly"); + } + if self.bulk_index_rebuild { + return invalid("metadataOnly cannot be combined with a bulk index rebuild"); + } + } if self.id_range().is_none() { return Ok(()); } @@ -867,6 +948,15 @@ pub struct ReindexProgress { /// must not let it clear them (#1125). #[serde(default, skip_serializing_if = "std::ops::Not::not")] pub resource_scoped: bool, + + /// Whether the job wrote only search metadata + /// ([`ReindexRequest::metadata_only`], #2008). + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + pub metadata_only: bool, + + /// Resources whose search metadata a metadata-only job wrote. + #[serde(default)] + pub metadata_updated: u64, } /// An error encountered during reindexing. @@ -1010,6 +1100,8 @@ impl ReindexProgress { error_message: None, current_resource_type: None, resource_scoped: false, + metadata_only: false, + metadata_updated: 0, } } @@ -1053,6 +1145,12 @@ impl ReindexProgress { serde_json::json!({"name": "errorCount", "valueInteger": self.errors.len()}), serde_json::json!({"name": "percentage", "valueDecimal": self.percentage()}), ]; + if self.metadata_only { + parameter.push(serde_json::json!({"name": "metadataOnly", "valueBoolean": true})); + parameter.push( + serde_json::json!({"name": "metadataUpdated", "valueInteger": self.metadata_updated}), + ); + } if let Some(started_at) = &self.started_at { parameter.push(serde_json::json!({"name": "startedAt", "valueDateTime": started_at})); } @@ -1495,6 +1593,11 @@ impl ReindexOperation { agent: Option, ) -> Result<(String, oneshot::Receiver<()>), ReindexError> { request.validate()?; + if request.metadata_only && !self.writers.iter().any(|w| w.supports_search_metadata()) { + return Err(ReindexError::Unsupported { + capability: "metadata-only reindex".to_string(), + }); + } let source = match request.id_range() { Some(range) => self .source @@ -1516,6 +1619,7 @@ impl ReindexOperation { let mut progress = ReindexProgress::new(&job_id); progress.tenant_id = Some(tenant.tenant_id().as_str().to_string()); progress.resource_scoped = request.resource_ids.is_some(); + progress.metadata_only = request.metadata_only; // Store the job self.jobs.write().insert(job_id.clone(), progress); @@ -2364,7 +2468,22 @@ async fn write_resource_batch( resource_type: &str, resources: &[StoredResource], extra_processed: u64, + metadata: Option, ) -> BatchOutcome { + if let Some(metadata) = metadata { + return write_metadata_batch( + tenant, + writers, + jobs, + job_id, + failures, + resource_type, + resources, + extra_processed, + metadata, + ) + .await; + } let mut wrote_any: Vec = vec![false; resources.len()]; let mut entry_counts: Vec = vec![0; resources.len()]; let mut batch = BatchOutcome::default(); @@ -2418,6 +2537,66 @@ async fn write_resource_batch( batch } +/// A metadata-only run's options (#2008); see +/// [`ReindexRequest::metadata_only`]. +#[derive(Clone, Copy)] +struct MetadataOnly { + skip_existing: bool, +} + +/// The metadata-only counterpart of [`write_resource_batch`] (#2008): writes +/// each resource's search metadata through every writer, and advances +/// `metadata_updated` instead of `entries_created`. `writers` holds only those +/// that support it. +#[allow(clippy::too_many_arguments)] +async fn write_metadata_batch( + tenant: &TenantContext, + writers: &[Arc], + jobs: &Arc>>, + job_id: &str, + failures: &mut ResourceFailureLog, + resource_type: &str, + resources: &[StoredResource], + extra_processed: u64, + metadata: MetadataOnly, +) -> BatchOutcome { + let mut batch = BatchOutcome::default(); + let mut updated = 0; + if !resources.is_empty() { + for writer in writers { + let started = Instant::now(); + let (outcomes, written) = writer + .write_search_metadata_page(tenant, resources, metadata.skip_existing) + .await; + batch.write += started.elapsed(); + updated = updated.max(written); + for (i, outcome) in outcomes.into_iter().enumerate() { + if let Err(e) = outcome { + batch.failed += 1; + record_resource_failure( + jobs, + job_id, + failures, + resource_type, + resources[i].id(), + format!("Failed to write search metadata: {e}"), + is_transient_error(&e), + ) + } + } + } + } + + let mut jobs_guard = jobs.write(); + if let Some(progress) = jobs_guard.get_mut(job_id) { + progress.processed_resources += resources.len() as u64 + extra_processed; + progress.metadata_updated += updated; + } + drop(jobs_guard); + + batch +} + /// How long the driver stands back between two page writes. /// /// Pages are written back-to-back, and on the SQLite backend each one is an @@ -2947,6 +3126,20 @@ async fn run_reindex( }, }; + // A metadata-only run writes through the writers that keep metadata, and + // only those; `start` refused it if there are none (#2008). + let metadata = request.metadata_only.then_some(MetadataOnly { + skip_existing: request.skip_existing_metadata, + }); + let writers: Vec> = if metadata.is_some() { + writers + .into_iter() + .filter(|writer| writer.supports_search_metadata()) + .collect() + } else { + writers + }; + // A clear-only run pages no type. let paged_types: &[String] = if request.clear_only { &[] @@ -3142,6 +3335,7 @@ async fn run_reindex( resource_type, &resources, missing, + metadata, ) .await; record_and_log_page( @@ -3269,6 +3463,7 @@ async fn run_reindex( resource_type, &page.resources, page.skipped.len() as u64, + metadata, ) .await; @@ -6234,6 +6429,231 @@ mod tests { } } + /// A writer with search metadata (#2008): records each metadata page and + /// fails `fail` transiently; its index writes must never be called. + #[derive(Default)] + struct MetadataTarget { + pages: parking_lot::Mutex, bool)>>, + fail: BTreeSet, + } + + #[async_trait] + impl ReindexTarget for MetadataTarget { + async fn delete_search_entries( + &self, + _: &TenantContext, + _: &str, + _: &str, + ) -> StorageResult { + panic!("a metadata-only run deletes no index entries") + } + + async fn write_search_entries( + &self, + _: &TenantContext, + _: &StoredResource, + ) -> StorageResult { + panic!("a metadata-only run writes no index entries") + } + + async fn clear_search_index(&self, _: &TenantContext) -> StorageResult { + panic!("a metadata-only run clears nothing") + } + + fn supports_search_metadata(&self) -> bool { + true + } + + async fn write_search_metadata_page( + &self, + _: &TenantContext, + resources: &[StoredResource], + skip_existing: bool, + ) -> (Vec>, u64) { + self.pages.lock().push(( + resources.iter().map(|r| r.id().to_string()).collect(), + skip_existing, + )); + let results: Vec> = resources + .iter() + .map(|r| { + if self.fail.contains(r.id()) { + Err(crate::error::BackendError::Unavailable { + backend_name: "metadata".into(), + message: format!("injected failure of {}", r.id()), + } + .into()) + } else { + Ok(()) + } + }) + .collect(); + let written = results.iter().filter(|r| r.is_ok()).count() as u64; + (results, written) + } + } + + /// A metadata-only run (#2008) writes only through the writers that keep + /// metadata, counts what they wrote as `metadata_updated` rather than + /// entries, records their per-resource failures, and says so in + /// `$reindex-status`. + #[tokio::test] + async fn metadata_only_run_writes_only_metadata_through_supporting_writers() { + let metadata = Arc::new(MetadataTarget { + fail: BTreeSet::from(["p1".to_string()]), + ..MetadataTarget::default() + }); + let index = Arc::new(RecordingTarget::default()); + let op = ReindexOperation::with_parts( + Arc::new(PagedSource::new(5)), + vec![index.clone(), metadata.clone()], + Arc::new(crate::search::TenantSearchRegistries::base_only()), + ); + let request = ReindexRequest::all() + .metadata_only() + .skip_existing_metadata() + .with_batch_size(2); + let id = op + .start(TenantContext::system(), request, None) + .await + .unwrap(); + // Pages are spaced by `REINDEX_PAGE_YIELD`, longer than + // `await_terminal`'s yields last. + let progress = tokio::time::timeout(Duration::from_secs(10), async { + loop { + match op.get_progress(&id).await { + Some(progress) if progress.status.is_finished() => break progress, + _ => tokio::time::sleep(Duration::from_millis(5)).await, + } + } + }) + .await + .expect("job reaches a terminal status"); + + assert_eq!(progress.status, ReindexStatus::Completed); + assert!(progress.metadata_only); + assert_eq!(progress.processed_resources, 5); + assert_eq!(progress.metadata_updated, 4); + assert_eq!(progress.entries_created, 0); + assert_eq!(progress.errors.len(), 1); + assert_eq!(progress.errors[0].resource_id, "p1"); + assert!(progress.errors[0].retryable); + assert!( + progress.errors[0] + .error + .starts_with("Failed to write search metadata:") + ); + assert_eq!( + *metadata.pages.lock(), + vec![ + (vec!["p0".to_string(), "p1".to_string()], true), + (vec!["p2".to_string(), "p3".to_string()], true), + (vec!["p4".to_string()], true), + ] + ); + assert!( + index.written.lock().is_empty(), + "the index writer is skipped" + ); + + let parameters = progress.to_parameters(); + let parameter = |name: &str| { + parameters["parameter"] + .as_array() + .unwrap() + .iter() + .find(|p| p["name"] == name) + .cloned() + }; + assert_eq!(parameter("metadataOnly").unwrap()["valueBoolean"], true); + assert_eq!(parameter("metadataUpdated").unwrap()["valueInteger"], 4); + let rebuild = ReindexProgress::new("rebuild").to_parameters(); + assert!( + !rebuild.to_string().contains("metadataOnly"), + "a rebuild's status has no metadata fields" + ); + } + + /// Without a writer that keeps metadata, a metadata-only run is refused + /// before a job id is issued, and the default page write refuses every + /// resource (#2008). + #[tokio::test] + async fn metadata_only_run_is_unsupported_without_a_metadata_writer() { + let index = Arc::new(RecordingTarget::default()); + assert!(!index.supports_search_metadata()); + let op = ReindexOperation::with_parts( + Arc::new(PagedSource::new(2)), + vec![index.clone()], + Arc::new(crate::search::TenantSearchRegistries::base_only()), + ); + let error = op + .start( + TenantContext::system(), + ReindexRequest::all().metadata_only(), + None, + ) + .await + .unwrap_err(); + assert!( + matches!(&error, ReindexError::Unsupported { capability } if capability == "metadata-only reindex"), + "{error:?}" + ); + assert!(op.list_jobs().is_empty()); + + let resources = PagedSource::new(2) + .fetch_resources_page(&TenantContext::system(), "Patient", None, 2) + .await + .unwrap() + .resources; + let (results, written) = index + .write_search_metadata_page(&TenantContext::system(), &resources, false) + .await; + assert_eq!(written, 0); + assert_eq!(results.len(), 2); + assert!(results.iter().all(|r| matches!( + r, + Err(crate::error::StorageError::Backend( + crate::error::BackendError::UnsupportedCapability { .. } + )) + ))); + } + + #[test] + fn metadata_only_rejects_what_would_touch_the_index() { + for (request, expected) in [ + ( + ReindexRequest::all().metadata_only().clear_existing(), + "metadataOnly cannot be combined with clearExisting", + ), + ( + ReindexRequest::all().metadata_only().clear_only(), + "metadataOnly cannot be combined with clearOnly", + ), + ( + ReindexRequest::all() + .metadata_only() + .with_bulk_index_rebuild(true), + "metadataOnly cannot be combined with a bulk index rebuild", + ), + ( + ReindexRequest::all().skip_existing_metadata(), + "skipExistingMetadata requires metadataOnly", + ), + ] { + let error = request.validate().unwrap_err(); + assert!(error.to_string().contains(expected), "{error}"); + } + assert!( + ReindexRequest { + id_start: Some("4".to_string()), + ..ReindexRequest::for_types(["Patient"]).metadata_only() + } + .validate() + .is_ok(), + "an id range combines with metadataOnly" + ); + } + /// Records the scope of every clear (#1624). Its page write replaces /// without deleting first, as PostgreSQL and Elasticsearch do, so every /// per-resource delete it records came from the clear, not the rebuild. diff --git a/crates/persistence/tests/mongodb/reindex_metadata_only.rs b/crates/persistence/tests/mongodb/reindex_metadata_only.rs new file mode 100644 index 0000000000..4b1302e98b --- /dev/null +++ b/crates/persistence/tests/mongodb/reindex_metadata_only.rs @@ -0,0 +1,402 @@ +//! #2008: a metadata-only `$reindex` backfills `missing_params` on resources +//! written before #1984 and leaves `search_index` untouched. Child module of +//! the `mongodb_tests` root, the same arrangement as +//! `tests/mongodb/reindex_id_range.rs`. + +use super::*; + +use std::collections::BTreeSet; + +use helios_persistence::search::{ReindexOperation, ReindexRequest, ReindexStatus}; + +use super::reindex_id_walk::wait_for_terminal; + +/// Six Patients, every other one without a `birthDate`, and one Observation. +async fn seed(backend: &MongoBackend, tenant: &TenantContext) { + for n in 0..6 { + let mut patient = + json!({"resourceType": "Patient", "id": format!("p{n}"), "gender": "female"}); + if n % 2 == 0 { + patient["birthDate"] = json!("1970-01-01"); + } + backend + .create(tenant, "Patient", patient, FhirVersion::default()) + .await + .unwrap(); + } + backend + .create( + tenant, + "Observation", + json!({"resourceType": "Observation", "id": "o1", "status": "final", "code": {"text": "metadata"}}), + FhirVersion::default(), + ) + .await + .unwrap(); +} + +async fn resources(backend: &MongoBackend) -> mongodb::Collection { + backend + .get_database() + .await + .unwrap() + .collection::("resources") +} + +/// Removes `missing_params` from every resource of `tenant`: what a dataset +/// written before #1984 looks like, and what the definition-maintenance +/// procedure in `docs/mongodb/search-indexes.md` leaves behind. +async fn unset_metadata(backend: &MongoBackend, tenant: &TenantContext) { + resources(backend) + .await + .update_many( + doc! { "tenant_id": tenant.tenant_id().as_str() }, + doc! { "$unset": { "missing_params": "" } }, + ) + .await + .unwrap(); +} + +/// Each live resource of `tenant` as `Type/id`, with its `missing_params` +/// (`None` when the field is absent). +async fn metadata( + backend: &MongoBackend, + tenant: &TenantContext, +) -> Vec<(String, Option>)> { + let mut cursor = resources(backend) + .await + .find(doc! { "tenant_id": tenant.tenant_id().as_str(), "is_deleted": false }) + .await + .unwrap(); + let mut out = Vec::new(); + while cursor.advance().await.unwrap() { + let row: Document = cursor.deserialize_current().unwrap(); + let names = row.get_array("missing_params").ok().map(|names| { + names + .iter() + .map(|name| name.as_str().unwrap().to_string()) + .collect() + }); + out.push(( + format!( + "{}/{}", + row.get_str("resource_type").unwrap(), + row.get_str("id").unwrap() + ), + names, + )); + } + out.sort(); + out +} + +/// Every `search_index` and `search_index_contained` row of `tenant`, whole, +/// so a run that rewrote a row (new `_id`) is told apart from one that left it. +async fn index_rows(backend: &MongoBackend, tenant: &TenantContext) -> BTreeSet { + let db = backend.get_database().await.unwrap(); + let mut rows = BTreeSet::new(); + for collection in ["search_index", "search_index_contained"] { + let mut cursor = db + .collection::(collection) + .find(doc! { "tenant_id": tenant.tenant_id().as_str() }) + .await + .unwrap(); + while cursor.advance().await.unwrap() { + let row: Document = cursor.deserialize_current().unwrap(); + rows.insert(format!("{collection}:{row}")); + } + } + rows +} + +/// Ids of the Patients `birthdate:missing=` finds. +async fn birthdate_missing( + backend: &MongoBackend, + tenant: &TenantContext, + missing: bool, +) -> BTreeSet { + let query = SearchQuery::new("Patient") + .with_count(100) + .with_parameter(SearchParameter { + name: "birthdate".to_string(), + param_type: SearchParamType::Date, + modifier: Some(SearchModifier::Missing), + values: vec![SearchValue::eq(if missing { "true" } else { "false" })], + chain: vec![], + components: vec![], + }); + backend + .search(tenant, &query) + .await + .unwrap() + .resources + .items + .iter() + .map(|resource| resource.id().to_string()) + .collect() +} + +async fn run( + backend: &Arc, + tenant: &TenantContext, + request: ReindexRequest, +) -> helios_persistence::search::ReindexProgress { + let op = ReindexOperation::new(backend.clone(), backend.tenant_registries().clone()); + let job = op.start(tenant.clone(), request, None).await.unwrap(); + wait_for_terminal(&op, &job).await +} + +/// After a metadata-only run on a pre-#1984 dataset every live resource has +/// `missing_params`, matching what a write would have stored, the value +/// index is byte-for-byte unchanged, `:missing` finds the same resources, and +/// another tenant is not touched. +#[tokio::test] +async fn mongodb_metadata_only_reindex_backfills_without_touching_the_index() { + let Some(backend) = create_backend_with_full_registry("reindex_metadata_only").await else { + eprintln!("Skipping (requires Docker or HFS_TEST_MONGODB_URL)"); + return; + }; + let backend = Arc::new(backend); + let tenant = create_tenant("metadata-only"); + let other = create_tenant("metadata-only-other"); + seed(&backend, &tenant).await; + seed(&backend, &other).await; + + let written = metadata(&backend, &tenant).await; + assert!( + written.iter().all(|(_, names)| names.is_some()), + "writes store missing_params: {written:?}" + ); + let found_missing = birthdate_missing(&backend, &tenant, true).await; + let found_present = birthdate_missing(&backend, &tenant, false).await; + assert_eq!(found_missing.len(), 3); + assert_eq!(found_present.len(), 3); + + unset_metadata(&backend, &tenant).await; + unset_metadata(&backend, &other).await; + let rows = index_rows(&backend, &tenant).await; + assert!(!rows.is_empty()); + assert_eq!( + birthdate_missing(&backend, &tenant, true).await, + found_missing + ); + + let progress = run(&backend, &tenant, ReindexRequest::all().metadata_only()).await; + assert_eq!(progress.status, ReindexStatus::Completed, "{progress:?}"); + assert!(progress.errors.is_empty(), "{progress:?}"); + assert_eq!(progress.processed_resources, 7); + assert_eq!(progress.metadata_updated, 7); + assert_eq!(progress.entries_created, 0); + assert!(progress.metadata_only); + + assert_eq!( + metadata(&backend, &tenant).await, + written, + "the backfill stores what the writes stored" + ); + assert_eq!( + index_rows(&backend, &tenant).await, + rows, + "search_index is untouched" + ); + assert_eq!( + birthdate_missing(&backend, &tenant, true).await, + found_missing + ); + assert_eq!( + birthdate_missing(&backend, &tenant, false).await, + found_present + ); + assert!( + metadata(&backend, &other) + .await + .iter() + .all(|(_, names)| names.is_none()), + "another tenant is not backfilled" + ); +} + +/// A page large enough to extract on the reindex prepare pool, on a +/// multi-thread runtime, stores the same metadata as the writes (#2008). +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn mongodb_metadata_only_reindex_extracts_a_large_page_on_the_pool() { + let Some(backend) = create_backend_with_full_registry("reindex_metadata_pool").await else { + eprintln!("Skipping (requires Docker or HFS_TEST_MONGODB_URL)"); + return; + }; + let backend = Arc::new(backend); + let tenant = create_tenant("metadata-pool"); + for n in 0..40 { + let mut patient = json!({"resourceType": "Patient", "id": format!("p{n:02}")}); + if n % 3 == 0 { + patient["gender"] = json!("male"); + } + backend + .create(&tenant, "Patient", patient, FhirVersion::default()) + .await + .unwrap(); + } + let written = metadata(&backend, &tenant).await; + unset_metadata(&backend, &tenant).await; + let rows = index_rows(&backend, &tenant).await; + + let progress = run( + &backend, + &tenant, + ReindexRequest::for_types(["Patient"]).metadata_only(), + ) + .await; + assert_eq!(progress.status, ReindexStatus::Completed, "{progress:?}"); + assert!(progress.errors.is_empty(), "{progress:?}"); + assert_eq!(progress.metadata_updated, 40); + assert_eq!(metadata(&backend, &tenant).await, written); + assert_eq!(index_rows(&backend, &tenant).await, rows); +} + +/// `skipExistingMetadata` fills in only the resources without the field, and +/// a type-scoped run only that type. +#[tokio::test] +async fn mongodb_metadata_only_reindex_can_skip_existing_metadata() { + let Some(backend) = create_backend_with_full_registry("reindex_metadata_skip").await else { + eprintln!("Skipping (requires Docker or HFS_TEST_MONGODB_URL)"); + return; + }; + let backend = Arc::new(backend); + let tenant = create_tenant("metadata-skip"); + seed(&backend, &tenant).await; + let written = metadata(&backend, &tenant).await; + unset_metadata(&backend, &tenant).await; + // A sentinel no extraction produces, so a rewrite would show. + resources(&backend) + .await + .update_one( + doc! { "tenant_id": tenant.tenant_id().as_str(), "resource_type": "Patient", "id": "p0" }, + doc! { "$set": { "missing_params": ["sentinel"] } }, + ) + .await + .unwrap(); + + let progress = run( + &backend, + &tenant, + ReindexRequest::for_types(["Patient"]) + .metadata_only() + .skip_existing_metadata(), + ) + .await; + assert_eq!(progress.status, ReindexStatus::Completed, "{progress:?}"); + assert_eq!(progress.processed_resources, 6); + assert_eq!(progress.metadata_updated, 5, "p0 already had metadata"); + + let after = metadata(&backend, &tenant).await; + for ((key, names), (_, expected)) in after.iter().zip(&written) { + match key.as_str() { + "Patient/p0" => assert_eq!(names.as_deref(), Some(&["sentinel".to_string()][..])), + "Observation/o1" => assert_eq!(names, &None, "out of the run's scope"), + _ => assert_eq!(names, expected, "{key}"), + } + } +} + +/// With search offloaded, MongoDB keeps no metadata a run could write, so a +/// metadata-only run is refused before a job starts. +#[tokio::test] +async fn mongodb_metadata_only_reindex_is_unsupported_when_search_is_offloaded() { + let Some(backend) = + create_backend_with_search_offloaded("reindex_metadata_offloaded", true).await + else { + eprintln!("Skipping (requires Docker or HFS_TEST_MONGODB_URL)"); + return; + }; + let backend = Arc::new(backend); + let op = ReindexOperation::new(backend.clone(), backend.tenant_registries().clone()); + let error = op + .start( + create_tenant("metadata-offloaded"), + ReindexRequest::all().metadata_only(), + None, + ) + .await + .unwrap_err(); + assert!( + matches!( + error, + helios_persistence::search::ReindexError::Unsupported { .. } + ), + "{error:?}" + ); +} + +/// `skipExistingMetadata` skips only arrays stamped with the current +/// fingerprint (#1999). An array written before the stamp existed (v0.2.5, +/// #1984) or under other definitions sends every `:missing=true` on the type +/// to the value index, so the backfill must restamp it. +#[tokio::test] +async fn mongodb_metadata_only_skip_existing_restamps_unstamped_and_stale_arrays() { + let Some(backend) = create_backend_with_full_registry("reindex_metadata_restamp").await else { + eprintln!("Skipping (requires Docker or HFS_TEST_MONGODB_URL)"); + return; + }; + let backend = Arc::new(backend); + let tenant = create_tenant("metadata-restamp"); + seed(&backend, &tenant).await; + let written = metadata(&backend, &tenant).await; + let patient = |id: &str| { + doc! { "tenant_id": tenant.tenant_id().as_str(), "resource_type": "Patient", "id": id } + }; + let collection = resources(&backend).await; + let fp_of = |row: &Document| row.get_str("missing_params_fp").ok().map(str::to_string); + let current = fp_of(&collection.find_one(patient("p0")).await.unwrap().unwrap()) + .expect("writes stamp a fingerprint"); + // p0: current stamp, sentinel array: must be left alone. + collection + .update_one( + patient("p0"), + doc! { "$set": { "missing_params": ["sentinel"] } }, + ) + .await + .unwrap(); + // p1: a #1984-era array without a stamp. + collection + .update_one( + patient("p1"), + doc! { "$unset": { "missing_params_fp": "" } }, + ) + .await + .unwrap(); + // p2: a stamp from other definitions. + collection + .update_one( + patient("p2"), + doc! { "$set": { "missing_params": ["sentinel"], "missing_params_fp": "stale" } }, + ) + .await + .unwrap(); + + let progress = run( + &backend, + &tenant, + ReindexRequest::for_types(["Patient"]) + .metadata_only() + .skip_existing_metadata(), + ) + .await; + assert_eq!(progress.status, ReindexStatus::Completed, "{progress:?}"); + assert!(progress.errors.is_empty(), "{progress:?}"); + assert_eq!( + progress.metadata_updated, 2, + "only p1 and p2 were not current" + ); + + for id in ["p0", "p1", "p2"] { + let row = collection.find_one(patient(id)).await.unwrap().unwrap(); + assert_eq!(fp_of(&row).as_deref(), Some(current.as_str()), "{id}"); + } + let after = metadata(&backend, &tenant).await; + for ((key, names), (_, expected)) in after.iter().zip(&written) { + match key.as_str() { + "Patient/p0" => assert_eq!(names.as_deref(), Some(&["sentinel".to_string()][..])), + _ => assert_eq!(names, expected, "{key}"), + } + } +} diff --git a/crates/persistence/tests/mongodb_tests.rs b/crates/persistence/tests/mongodb_tests.rs index 6cbe667416..43c8a3f502 100644 --- a/crates/persistence/tests/mongodb_tests.rs +++ b/crates/persistence/tests/mongodb_tests.rs @@ -1256,6 +1256,10 @@ mod reindex_fetch_by_ids; #[path = "mongodb/reindex_id_range.rs"] mod reindex_id_range; +/// #2008: metadata-only `$reindex` backfills `missing_params`. +#[path = "mongodb/reindex_metadata_only.rs"] +mod reindex_metadata_only; + /// #1586: a transaction bundle the server aborts with a /// `TransientTransactionError` is re-run instead of failing with a 400. #[path = "mongodb/transaction_retry.rs"] diff --git a/crates/rest/src/handlers/reindex.rs b/crates/rest/src/handlers/reindex.rs index bc9871a69a..0cefabb51c 100644 --- a/crates/rest/src/handlers/reindex.rs +++ b/crates/rest/src/handlers/reindex.rs @@ -22,6 +22,8 @@ //! |---|---|---| //! | `clearExisting` | `valueBoolean` | Clear the run's scope before rebuilding it | //! | `clearOnly` | `valueBoolean` | Clear the run's scope and do not rebuild it | +//! | `metadataOnly` | `valueBoolean` | Recompute only per-resource search metadata (MongoDB's `missing_params`) and leave the search index untouched | +//! | `skipExistingMetadata` | `valueBoolean` | With `metadataOnly`, leave alone resources whose metadata is already current | //! | `batchSize` | `valueInteger` | Resources per page | //! | `idStart` | `valueString` | First id to reindex (included); `POST /{type}/$reindex` only | //! | `idEnd` | `valueString` | Id where the run stops (not included); `POST /{type}/$reindex` only | @@ -30,7 +32,9 @@ //! them `COLLATE "C"`, not in the database's collation), so ranges laid end to //! end (each `idEnd` the next `idStart`) cover a type exactly once. An invalid //! combination is `400` before a job starts; an id range on a backend without -//! range support (S3, Elasticsearch as the source) is `501`. +//! range support (S3, Elasticsearch as the source) is `501`, and so is +//! `metadataOnly` on a deployment without per-resource search metadata (every +//! backend but MongoDB, including MongoDB with search offloaded, #2008). //! //! On SQLite, ranges started together mostly queue on its single writer, and //! can see the busy → `503` path, rather than rebuild in parallel. They are @@ -215,6 +219,17 @@ fn parse_reindex_request( .and_then(serde_json::Value::as_bool) .ok_or_else(|| bad_request("clearOnly requires valueBoolean".to_string()))?; } + Some(name @ ("metadataOnly" | "skipExistingMetadata")) => { + let on = param + .get("valueBoolean") + .and_then(serde_json::Value::as_bool) + .ok_or_else(|| bad_request(format!("{name} requires valueBoolean")))?; + if name == "metadataOnly" { + request.metadata_only = on; + } else { + request.skip_existing_metadata = on; + } + } Some(name @ ("idStart" | "idEnd")) => { let id = param .get("valueString") @@ -467,6 +482,59 @@ mod tests { assert!(!request.clear_only); } + #[test] + fn metadata_only_parameters_are_parsed() { + let body = parameters(json!([ + { "name": "metadataOnly", "valueBoolean": true }, + { "name": "skipExistingMetadata", "valueBoolean": true }, + { "name": "idStart", "valueString": "4" } + ])); + let request = parse_reindex_request(patient(), Some(&body)).unwrap(); + assert!(request.metadata_only); + assert!(request.skip_existing_metadata); + assert_eq!(request.id_start.as_deref(), Some("4")); + + let request = parse_reindex_request(None, None).unwrap(); + assert!(!request.metadata_only); + assert!(!request.skip_existing_metadata); + } + + #[test] + fn invalid_metadata_only_parameters_are_rejected() { + let cases = [ + ( + json!([{ "name": "skipExistingMetadata", "valueBoolean": true }]), + "skipExistingMetadata requires metadataOnly", + ), + ( + json!([{ "name": "metadataOnly", "valueString": "true" }]), + "metadataOnly requires valueBoolean", + ), + ( + json!([ + { "name": "metadataOnly", "valueBoolean": true }, + { "name": "clearExisting", "valueBoolean": true } + ]), + "metadataOnly cannot be combined with clearExisting", + ), + ( + json!([ + { "name": "metadataOnly", "valueBoolean": true }, + { "name": "clearOnly", "valueBoolean": true } + ]), + "metadataOnly cannot be combined with clearOnly", + ), + ]; + for (parameter, expected) in cases { + let body = parameters(parameter); + let error = parse_reindex_request(patient(), Some(&body)).unwrap_err(); + assert!( + matches!(&error, RestError::BadRequest { message } if message.contains(expected)), + "{body}: {error:?}" + ); + } + } + #[test] fn invalid_id_range_and_clear_only_parameters_are_rejected() { let cases = [ diff --git a/docs/mongodb/search-indexes.md b/docs/mongodb/search-indexes.md index ec2915e6dc..d1630b6444 100644 --- a/docs/mongodb/search-indexes.md +++ b/docs/mongodb/search-indexes.md @@ -210,9 +210,48 @@ time is included in the insert timing statistic. before judging `:missing=true` latency. Existing resources without metadata remain searchable through the compatibility path while their value indexes remain valid. -3. Run the tenant's `POST /$reindex` to populate metadata for existing resources. +3. Populate metadata for existing resources with a metadata-only reindex + (#2008), which writes `missing_params` and leaves `search_index` and + `search_index_contained` untouched: + + ```http + POST /$reindex + Content-Type: application/fhir+json + + {"resourceType": "Parameters", "parameter": [ + {"name": "metadataOnly", "valueBoolean": true}, + {"name": "skipExistingMetadata", "valueBoolean": true} + ]} + ``` + Poll `GET /$reindex-status/{job_id}` on the same node and resolve reported - failures before treating the rebuild as complete. + failures before treating the backfill as complete. A full `POST /$reindex` + (no parameters) populates metadata too, but also deletes and rewrites every + value-index row. + +### Metadata-only reindex + +`metadataOnly` runs the same walk, extraction and version guard as a rebuild: +each page's resources are extracted exactly as a rebuild would extract them, +and `missing_params` is written only where the stored version is still the +one the page read. It writes nothing to `search_index` or +`search_index_contained`. Because the metadata comes from the same extraction +that produced the value-index rows, `:missing=true` results are unchanged by +the backfill as long as those rows were written under the current +definitions; only the plan moves from the compatibility path to +`idx_resources_missing_params`. + +| Parameter | Effect | +|---|---| +| `metadataOnly` | Write only `missing_params`. Cannot be combined with `clearExisting`, `clearOnly` or a bulk index rebuild. Combines with a resource type (`POST /{type}/$reindex`) and with `idStart`/`idEnd`. | +| `skipExistingMetadata` | With `metadataOnly`, leave alone resources whose `missing_params` already carries the current `missing_params_fp`. Resources without the array, with an unstamped array (written before #1999) or with a stale stamp are written. Every resource in scope is still read and extracted; only the write is skipped. | + +`$reindex-status` reports `metadataOnly: true` and `metadataUpdated`, the +number of resources whose metadata was written; `entriesCreated` stays `0`. +A resource whose extraction fails has its metadata removed, as in a rebuild, +and is listed as an error. On a deployment where Elasticsearch owns search +(`mongodb-elasticsearch`), `missing_params` is never read and +`metadataOnly` is `501 Not Implemented`, as it is on every other backend. ### Definition changes (#1999) @@ -257,8 +296,10 @@ value-index route, which is about the cost it had before #1984. Resources written before the stamp existed (or by a node that predates it) carry `missing_params` without `missing_params_fp`. The probe finds them in one key and every `:missing=true` on the type takes the value-index route, so -results are correct after an upgrade without any manual step. Run one -`$reindex` after upgrading to stamp them and restore the metadata route. +results are correct after an upgrade without any manual step. Run the +metadata-only reindex from the upgrade steps above (`metadataOnly`, with +`skipExistingMetadata`) to stamp them and restore the metadata route without +rewriting the value index. Recovery, when metadata is suspected wrong despite the stamp (a restore from backup, a manual edit): `$unset` both fields together for the affected tenant