diff --git a/crates/persistence/src/backends/mongodb/search_impl.rs b/crates/persistence/src/backends/mongodb/search_impl.rs index 2973d2e72f..8de0535d08 100644 --- a/crates/persistence/src/backends/mongodb/search_impl.rs +++ b/crates/persistence/src/backends/mongodb/search_impl.rs @@ -655,7 +655,7 @@ fn missing_metadata_hint(filter: &Document) -> Option { has_missing.then(|| Hint::Name(super::schema::RESOURCES_MISSING_INDEX.to_owned())) } -fn missing_resource_condition(name: &str) -> Document { +pub(super) fn missing_resource_condition(name: &str) -> Document { doc! { "$or": [ { "missing_params": name }, { "missing_params": { "$exists": false } }, @@ -3710,7 +3710,14 @@ impl MongoBackend { } if missing_support.is_resource_parameter(param) { candidates = self - .metadata_missing_ids(db, tenant_id, resource_type, param, Some(&candidates)) + .metadata_missing_ids( + db, + tenant_id, + resource_type, + param, + Some(&candidates), + None, + ) .await?; continue; } @@ -3844,7 +3851,14 @@ impl MongoBackend { for param in missing { if missing_support.is_resource_parameter(param) { let ids = self - .metadata_missing_ids(db, tenant_id, resource_type, param, matched.as_ref()) + .metadata_missing_ids( + db, + tenant_id, + resource_type, + param, + matched.as_ref(), + None, + ) .await?; matched = Some(ids); continue; @@ -3942,13 +3956,14 @@ impl MongoBackend { )) } - async fn metadata_missing_ids( + pub(super) async fn metadata_missing_ids( &self, db: &mongodb::Database, tenant_id: &str, resource_type: &str, param: &SearchParameter, candidates: Option<&HashSet>, + session: Option<&mut mongodb::ClientSession>, ) -> StorageResult> { let mut scope = doc! { "tenant_id": tenant_id, "resource_type": resource_type, "is_deleted": false }; @@ -3966,12 +3981,49 @@ impl MongoBackend { param, ))); stages.push(doc! { "$project": { "_id": 0, "id": 1 } }); - let cursor = db - .collection::(Self::RESOURCES_COLLECTION) - .aggregate(stages) - .await - .or_query_error("Failed to check missing resource metadata")?; - collect_id_set(cursor, "id").await + let collection = db.collection::(Self::RESOURCES_COLLECTION); + // A bounded candidate set is a few identity seeks; left to itself the + // planner may walk the shared `missing_params` index instead, reading + // every absent key of the type for each batch (#2233). + let options = mongodb::options::AggregateOptions::builder() + .hint( + candidates + .is_some() + .then(|| Hint::Name(RESOURCES_IDENTITY_INDEX.to_owned())), + ) + .build(); + match session { + Some(session) => { + let mut cursor = collection + .aggregate(stages) + .with_options(options) + .session(&mut *session) + .await + .or_query_error("Failed to check missing resource metadata")?; + let mut ids = HashSet::new(); + while cursor + .advance(&mut *session) + .await + .or_query_error("Failed to advance missing metadata cursor")? + { + let doc = cursor.deserialize_current().map_err(|e| { + internal_error(format!("Failed to deserialize missing metadata: {e}")) + })?; + if let Ok(id) = doc.get_str("id") { + ids.insert(id.to_string()); + } + } + Ok(ids) + } + None => { + let cursor = collection + .aggregate(stages) + .with_options(options) + .await + .or_query_error("Failed to check missing resource metadata")?; + collect_id_set(cursor, "id").await + } + } } pub(super) fn build_search_index_filter( diff --git a/crates/persistence/src/backends/mongodb/storage.rs b/crates/persistence/src/backends/mongodb/storage.rs index 4eae2b0179..55e27a9375 100644 --- a/crates/persistence/src/backends/mongodb/storage.rs +++ b/crates/persistence/src/backends/mongodb/storage.rs @@ -144,6 +144,15 @@ fn stages_its_create(entry: &BundleEntry) -> bool { entry.method == BundleMethod::Post && entry.if_none_exist.is_none() } +fn is_missing_true_param(param: &SearchParameter) -> bool { + !matches!(param.name.as_str(), "_id" | "_lastUpdated") + && matches!(param.modifier, Some(crate::types::SearchModifier::Missing)) + && param + .values + .first() + .is_some_and(|value| value.value == "true") +} + /// Commit retries after a commit whose outcome is unknown /// (`UnknownTransactionCommitResult`), on top of the first commit. Only the /// commit is re-sent: the transaction's writes may already be applied, so the @@ -5956,15 +5965,34 @@ impl MongoBackend { Some(&mut *session), ) .await?; + + // `:missing=true` is applied to each candidate batch below, after the + // other criteria have narrowed it, as search does (#2004): a pass over + // the whole type here would cost in proportion to the type, not to + // the candidates (#2003). + let missing: Vec<&SearchParameter> = typed_params + .iter() + .filter(|p| is_missing_true_param(p)) + .collect(); + let missing_support = self + .classify_missing_in_bundle_transaction(tenant, resource_type, &missing) + .await; + let index_params: Vec<_> = typed_params .iter() - .filter(|p| !matches!(p.name.as_str(), "_id" | "_lastUpdated")) + .filter(|p| { + !matches!(p.name.as_str(), "_id" | "_lastUpdated") && !is_missing_true_param(p) + }) .collect(); if index_params.is_empty() { let query = SearchQuery { resource_type: resource_type.to_string(), - parameters: typed_params, + parameters: typed_params + .iter() + .filter(|p| !is_missing_true_param(p)) + .cloned() + .collect(), count: Some(2), ..Default::default() }; @@ -5978,19 +6006,32 @@ impl MongoBackend { None, )?; let resources = db.collection::(MongoBackend::RESOURCES_COLLECTION); - let cursor = resources - .find(filter) - .limit(2) - .session(&mut *session) - .await - .map_err(|e| { - internal_driver_error("Failed to query resources in transaction", e) - })?; - let docs = collect_session_documents(cursor, session).await?; - return docs - .into_iter() - .map(|doc| document_to_stored_resource(&doc, tenant, resource_type)) - .collect(); + if missing.is_empty() { + let cursor = resources + .find(filter) + .limit(2) + .session(&mut *session) + .await + .map_err(|e| { + internal_driver_error("Failed to query resources in transaction", e) + })?; + let docs = collect_session_documents(cursor, session).await?; + return docs + .into_iter() + .map(|doc| document_to_stored_resource(&doc, tenant, resource_type)) + .collect(); + } + return self + .scan_absent_in_bundle_transaction( + db, + session, + tenant, + resource_type, + filter, + &missing, + &missing_support, + ) + .await; } let tenant_id = tenant.tenant_id().as_str(); @@ -6162,6 +6203,19 @@ impl MongoBackend { candidate_ids.retain(|id| passing.contains(id)); } + if !missing.is_empty() && !candidate_ids.is_empty() { + candidate_ids = self + .retain_absent_in_bundle_transaction( + db, + session, + tenant_id, + resource_type, + &missing, + &missing_support, + candidate_ids, + ) + .await?; + } candidate_ids.retain(|id| !matched_ids.contains(id)); if !candidate_ids.is_empty() { let remaining = (2 - matches.len()) as i64; @@ -6229,6 +6283,177 @@ impl MongoBackend { Ok(matches) } + /// Classifies a conditional entry's `:missing=true` criteria once, after + /// the same freshness probe search runs (#1999, #2007). + async fn classify_missing_in_bundle_transaction( + &self, + tenant: &TenantContext, + resource_type: &str, + missing: &[&SearchParameter], + ) -> MissingSupport { + if missing.is_empty() { + return MissingSupport::none(); + } + let tenant_id = tenant.tenant_id().as_str(); + let query = SearchQuery { + resource_type: resource_type.to_string(), + parameters: missing.iter().map(|p| (*p).clone()).collect(), + ..Default::default() + }; + self.refresh_missing_metadata_freshness(tenant_id, &query) + .await; + self.classify_missing(tenant_id, resource_type, &query) + } + + /// Keeps the `candidates` every `:missing=true` criterion holds for, read + /// in the transaction's session so earlier writes of the Bundle count: + /// `missing_params` with the value-index fallback where it answers the + /// parameter, the absence of a value row otherwise — the same routes + /// search takes. The cost follows the candidate batch, not the type. + #[allow(clippy::too_many_arguments)] + async fn retain_absent_in_bundle_transaction( + &self, + db: &mongodb::Database, + session: &mut ClientSession, + tenant_id: &str, + resource_type: &str, + missing: &[&SearchParameter], + missing_support: &MissingSupport, + mut candidates: HashSet, + ) -> StorageResult> { + let search_index = db.collection::(MongoBackend::SEARCH_INDEX_COLLECTION); + for param in missing { + if candidates.is_empty() { + break; + } + if missing_support.is_resource_parameter(param) { + candidates = self + .metadata_missing_ids( + db, + tenant_id, + resource_type, + param, + Some(&candidates), + Some(&mut *session), + ) + .await?; + continue; + } + let mut presence = + super::search_impl::missing_presence_filter(tenant_id, resource_type, param); + presence.insert( + "resource_id", + doc! { "$in": candidates.iter().cloned().collect::>() }, + ); + let present: HashSet = search_index + .distinct("resource_id", presence) + .session(&mut *session) + .await + .map_err(|e| { + internal_driver_error("Failed presence distinct for missing criteria", e) + })? + .into_iter() + .filter_map(|v| v.as_str().map(str::to_string)) + .collect(); + candidates.retain(|id| !present.contains(id)); + } + Ok(candidates) + } + + /// Criteria with no search-index parameter to drive from: only + /// `_id` / `_lastUpdated` (in `filter`) and `:missing=true`. The entry + /// needs at most two matches, so an unsorted cursor over `filter` is read + /// in batches, each checked by [`Self::retain_absent_in_bundle_transaction`], + /// until two match or the type runs out. Metadata-answered criteria also + /// narrow the cursor itself through `idx_resources_missing_params`. + #[allow(clippy::too_many_arguments)] + async fn scan_absent_in_bundle_transaction( + &self, + db: &mongodb::Database, + session: &mut ClientSession, + tenant: &TenantContext, + resource_type: &str, + filter: Document, + missing: &[&SearchParameter], + missing_support: &MissingSupport, + ) -> StorageResult> { + const BATCH_SIZE: usize = 512; + let tenant_id = tenant.tenant_id().as_str(); + let mut conditions = vec![filter]; + conditions.extend( + missing + .iter() + .filter(|p| missing_support.is_resource_parameter(p)) + .map(|p| super::search_impl::missing_resource_condition(&p.name)), + ); + let resources = db.collection::(MongoBackend::RESOURCES_COLLECTION); + let mut cursor = resources + .find(doc! { "$and": conditions }) + .projection(doc! { "_id": 0, "id": 1 }) + .batch_size(BATCH_SIZE as u32) + .session(&mut *session) + .await + .map_err(|e| { + internal_driver_error("Failed to scan resources for missing criteria", e) + })?; + let mut matched: Vec = Vec::with_capacity(2); + let mut batch: HashSet = HashSet::with_capacity(BATCH_SIZE); + loop { + let more = cursor + .advance(&mut *session) + .await + .map_err(|e| internal_driver_error("Failed to advance missing-criteria scan", e))?; + if more { + let doc = cursor.deserialize_current().map_err(|e| { + internal_error(format!("Failed to deserialize resource id: {e}")) + })?; + if let Ok(id) = doc.get_str("id") { + batch.insert(id.to_string()); + } + if batch.len() < BATCH_SIZE { + continue; + } + } + if !batch.is_empty() { + let absent = self + .retain_absent_in_bundle_transaction( + db, + session, + tenant_id, + resource_type, + missing, + missing_support, + std::mem::take(&mut batch), + ) + .await?; + matched.extend(absent.into_iter().take(2 - matched.len())); + } + if matched.len() >= 2 || !more { + break; + } + } + if matched.is_empty() { + return Ok(Vec::new()); + } + let cursor = resources + .find(doc! { + "tenant_id": tenant_id, + "resource_type": resource_type, + "is_deleted": false, + "id": { "$in": matched }, + }) + .session(&mut *session) + .await + .map_err(|e| { + internal_driver_error("Failed to fetch resources for missing criteria", e) + })?; + collect_session_documents(cursor, session) + .await? + .into_iter() + .map(|doc| document_to_stored_resource(&doc, tenant, resource_type)) + .collect() + } + /// Matches `ifNoneExist` criteria against the raw `resources` documents. /// /// Search is offloaded, so there are no `search_index` rows to consult; diff --git a/crates/persistence/tests/mongodb/missing_index.rs b/crates/persistence/tests/mongodb/missing_index.rs index 7c02660630..847c9c43c1 100644 --- a/crates/persistence/tests/mongodb/missing_index.rs +++ b/crates/persistence/tests/mongodb/missing_index.rs @@ -1489,7 +1489,7 @@ async fn mongodb_missing_metadata_definition_maintenance_restores_fallback_and_r } #[tokio::test] -async fn mongodb_missing_conditional_transactions_do_not_invert_absence() { +async fn mongodb_missing_conditional_transactions_match_absence() { let Some(backend) = create_backend("missing_conditional").await else { return; }; @@ -1541,30 +1541,103 @@ async fn mongodb_missing_conditional_transactions_do_not_invert_absence() { }, ) }; - let error = backend + assert_results( + &backend, + &tenant, + &missing_query(&PARAMETERS[1], true), + &["absent"], + ) + .await; + let result = backend .process_transaction( &tenant, vec![put.clone(), delete(true)], FhirVersion::default(), ) .await - .unwrap_err(); - assert!(format!("{error}").contains("missing=true"), "{error:?}"); + .unwrap(); + assert_eq!(result.entries[0].status, 201); + assert_eq!(result.entries[1].status, 204); assert!( - backend - .read(&tenant, "Observation", "present") - .await - .unwrap() - .is_none(), - "unsupported absence criteria must roll back the preceding write" + matches!( + backend.read(&tenant, "Observation", "absent").await, + Err(StorageError::Resource(ResourceError::Gone { .. })) + ), + "absence criteria must delete the resource GET would have matched" ); assert!( backend - .read(&tenant, "Observation", "absent") + .read(&tenant, "Observation", "present") .await .unwrap() .is_some() ); + + let bare_put = BundleEntry { + method: BundleMethod::Put, + url: "Observation/bare".into(), + resource: Some(observation("bare", vec![])), + if_match: None, + if_none_match: None, + if_none_exist: None, + full_url: None, + criteria: None, + }; + let absent_conditional_create = BundleEntry { + method: BundleMethod::Post, + url: "Observation".into(), + resource: Some(observation("never-created", extensions(false))), + if_match: None, + if_none_match: None, + if_none_exist: Some("missing-token:missing=true".into()), + full_url: None, + criteria: None, + }; + let result = backend + .process_transaction( + &tenant, + vec![bare_put, absent_conditional_create], + FhirVersion::default(), + ) + .await + .unwrap(); + assert_eq!(result.entries[0].status, 201); + assert_eq!( + result.entries[1].status, 200, + "ifNoneExist absence must match the resource written earlier in the same bundle" + ); + assert_eq!( + result.entries[1].reference().as_deref(), + Some("Observation/bare") + ); + + backend + .create( + &tenant, + "Observation", + observation("bare2", vec![]), + FhirVersion::default(), + ) + .await + .unwrap(); + backend + .process_transaction(&tenant, vec![delete(true)], FhirVersion::default()) + .await + .unwrap_err(); + for id in ["bare", "bare2"] { + assert!( + backend + .read(&tenant, "Observation", id) + .await + .unwrap() + .is_some(), + "a multi-match conditional delete must change nothing" + ); + } + backend + .delete(&tenant, "Observation", "bare2") + .await + .unwrap(); // Presence reads the value rows written earlier in the same transaction. let conditional_create = BundleEntry { method: BundleMethod::Post, @@ -1603,7 +1676,7 @@ async fn mongodb_missing_conditional_transactions_do_not_invert_absence() { ); assert!( backend - .read(&tenant, "Observation", "absent") + .read(&tenant, "Observation", "bare") .await .unwrap() .is_some() @@ -1612,7 +1685,7 @@ async fn mongodb_missing_conditional_transactions_do_not_invert_absence() { &backend, &tenant, &missing_query(&PARAMETERS[1], true), - &["absent"], + &["bare"], ) .await; } @@ -2648,3 +2721,204 @@ async fn mongodb_missing_metadata_disabled_keeps_missing_results() { } } } + +#[tokio::test] +async fn mongodb_missing_conditional_matches_same_bundle_write_on_empty_tenant() { + let Some(backend) = create_backend("missing_conditional_empty").await else { + return; + }; + let tenant = create_tenant("missing-conditional-empty"); + register_parameters(&backend, &tenant).await; + let bare_put = BundleEntry { + method: BundleMethod::Put, + url: "Observation/only".into(), + resource: Some(observation("only", vec![])), + if_match: None, + if_none_match: None, + if_none_exist: None, + full_url: None, + criteria: None, + }; + let conditional = BundleEntry { + method: BundleMethod::Post, + url: "Observation".into(), + resource: Some(observation("never", extensions(false))), + if_match: None, + if_none_match: None, + if_none_exist: Some("missing-token:missing=true".into()), + full_url: None, + criteria: None, + }; + let result = backend + .process_transaction(&tenant, vec![bare_put, conditional], FhirVersion::default()) + .await + .unwrap(); + assert_eq!(result.entries[0].status, 201); + assert_eq!( + result.entries[1].status, 200, + "empty tenant: ifNoneExist absence must match the same-bundle write" + ); + + let spec_tenant = create_tenant("missing-conditional-spec"); + let put = BundleEntry { + method: BundleMethod::Put, + url: "Patient/m2".into(), + resource: Some(json!({"resourceType": "Patient", "id": "m2", "active": true})), + if_match: None, + if_none_match: None, + if_none_exist: None, + full_url: None, + criteria: None, + }; + let conditional = BundleEntry { + method: BundleMethod::Post, + url: "Patient".into(), + resource: Some(json!({"resourceType": "Patient", "gender": "other"})), + if_match: None, + if_none_match: None, + if_none_exist: Some("gender:missing=true".into()), + full_url: None, + criteria: None, + }; + let result = backend + .process_transaction(&spec_tenant, vec![put, conditional], FhirVersion::default()) + .await + .unwrap(); + assert_eq!(result.entries[0].status, 201); + assert_eq!( + result.entries[1].status, 200, + "spec param: ifNoneExist absence must match the same-bundle write" + ); + let db = backend.get_database().await.unwrap(); + db.drop().await.unwrap(); +} + +/// #2003: `:missing=true` in conditional criteria is applied per candidate +/// batch, after the other criteria narrow it, on both routes: metadata and, +/// with metadata disabled, the value index. The absent resource is written +/// after 600 present ones, so it lies beyond the first batch of the +/// index-driven pager (128) and of the resource scan (512). +#[tokio::test] +async fn mongodb_missing_conditional_criteria_cross_candidate_batches() { + let Some(connection_string) = shared_mongo::connection_string().await else { + return; + }; + for metadata in [true, false] { + let database_name = build_test_database_name(if metadata { + "missing_cond_batches_md" + } else { + "missing_cond_batches_vi" + }); + let Some(backend) = build_backend(MongoBackendConfig { + connection_string: connection_string.clone(), + database_name, + data_dir: Some(repo_data_dir()), + missing_metadata: metadata, + ..Default::default() + }) + .await + else { + return; + }; + let tenant = create_tenant("missing-cond-batches"); + register_parameters(&backend, &tenant).await; + let put = |id: String, values: Vec| BundleEntry { + method: BundleMethod::Put, + url: format!("Observation/{id}"), + resource: Some(observation(&id, values)), + if_match: None, + if_none_match: None, + if_none_exist: None, + full_url: None, + criteria: None, + }; + for chunk in 0..6 { + let entries = (0..100) + .map(|i| put(format!("present-{}", chunk * 100 + i), extensions(false))) + .collect(); + backend + .process_transaction(&tenant, entries, FhirVersion::default()) + .await + .unwrap(); + } + backend + .process_transaction( + &tenant, + vec![put("absent-a".into(), vec![])], + FhirVersion::default(), + ) + .await + .unwrap(); + + let conditional = |criteria: &str| BundleEntry { + method: BundleMethod::Post, + url: "Observation".into(), + resource: Some(observation("conditional", extensions(false))), + if_match: None, + if_none_match: None, + if_none_exist: Some(criteria.to_owned()), + full_url: None, + criteria: None, + }; + let run = |criteria: &'static str| { + let backend = &backend; + let tenant = &tenant; + async move { + backend + .process_transaction( + tenant, + vec![conditional(criteria)], + FhirVersion::default(), + ) + .await + } + }; + for criteria in [ + "missing-token:missing=true", + "code=fixture&missing-token:missing=true", + "_id=absent-a&missing-token:missing=true", + "missing-token:missing=true&missing-string:missing=true", + ] { + let result = run(criteria).await.unwrap(); + assert_eq!( + (result.entries[0].status, result.entries[0].reference()), + (200, Some("Observation/absent-a".to_owned())), + "metadata={metadata} criteria={criteria}" + ); + } + + backend + .process_transaction( + &tenant, + vec![put("absent-b".into(), vec![])], + FhirVersion::default(), + ) + .await + .unwrap(); + assert_results( + &backend, + &tenant, + &missing_query(&PARAMETERS[1], true), + &["absent-a", "absent-b"], + ) + .await; + for criteria in [ + "missing-token:missing=true", + "code=fixture&missing-token:missing=true", + ] { + let error = run(criteria).await.unwrap_err(); + assert!( + format!("{error}").contains("412"), + "metadata={metadata} criteria={criteria}: {error}" + ); + } + // A present resource does not match, so the entry creates. + let result = run("_id=present-3&missing-token:missing=true") + .await + .unwrap(); + assert_eq!(result.entries[0].status, 201, "metadata={metadata}"); + + let db = backend.get_database().await.unwrap(); + db.drop().await.unwrap(); + } +} diff --git a/docs/mongodb/search-indexes.md b/docs/mongodb/search-indexes.md index d1630b6444..e27538a3d6 100644 --- a/docs/mongodb/search-indexes.md +++ b/docs/mongodb/search-indexes.md @@ -97,7 +97,17 @@ Other parameters can still produce rows and metadata, so a present field does not prove that every parameter extracted successfully. Affected missing-field results can remain incomplete until the extraction problem is corrected and the resource is rewritten or reindexed. This metadata describes top-level -resources; it does not add contained `:missing` support. +resources; it does not add contained `:missing` support. Transaction and +batch conditional criteria (`ifNoneExist`, conditional update/delete URLs) +evaluate `:missing=true` with the same semantics inside the transaction's +session (#2003): the metadata predicate with the value-index fallback when +the fingerprint is fresh, the absence of a value row otherwise. As in search +(#2004), the check runs on each batch of candidates the other criteria +produced, so its cost follows the candidates, not the resource type; criteria +with nothing else to drive from (only `_id`, `_lastUpdated` and +`:missing=true`) scan the type in batches and stop at the second match. +`ifNoneExist` sees resources written earlier in the same bundle; URL-borne +criteria are resolved before any write, as the R4 overlap rule requires. Next to the array, `missing_params_fp` stamps the definition fingerprint it was computed from, and the `idx_resources_missing_params_fp` index serves the