From 3e606f3a594e1afd9a4e99b20fbe403afb9eeabf Mon Sep 17 00:00:00 2001 From: bomanaps Date: Sun, 25 Oct 2026 12:08:38 +0100 Subject: [PATCH 1/2] missing=true in conditional criteria --- .../src/backends/mongodb/search_impl.rs | 60 +++++-- .../src/backends/mongodb/storage.rs | 167 ++++++++++++++++- .../tests/mongodb/missing_index.rs | 170 ++++++++++++++++-- docs/mongodb/search-indexes.md | 8 +- 4 files changed, 378 insertions(+), 27 deletions(-) diff --git a/crates/persistence/src/backends/mongodb/search_impl.rs b/crates/persistence/src/backends/mongodb/search_impl.rs index 2973d2e72f..421e35ac12 100644 --- a/crates/persistence/src/backends/mongodb/search_impl.rs +++ b/crates/persistence/src/backends/mongodb/search_impl.rs @@ -264,7 +264,7 @@ const MAX_SEARCH_PARAM_SORT_KEYS: usize = 15; const CANDIDATE_BATCH_SIZE: usize = 512; const PROBE_ROW_LIMIT: u64 = 100_000; // 300k × ~45 bytes/UUID ≈ 13.5 MB — safely under the 16 MB BSON document cap. -const MAX_RESULT_ID_SET: usize = 300_000; +pub(super) const MAX_RESULT_ID_SET: usize = 300_000; /// Targets a reference `:identifier` search may resolve to (#1408). They /// travel in one `$in`, two entries each; 10 000 stays near 1 MB, well under @@ -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,37 @@ 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); + match session { + Some(session) => { + let mut cursor = collection + .aggregate(stages) + .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) + .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..b863e401cc 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,20 +5965,34 @@ impl MongoBackend { Some(&mut *session), ) .await?; + + let absent_ids = self + .absent_ids_in_bundle_transaction(db, session, tenant, resource_type, &typed_params) + .await?; + if absent_ids.as_ref().is_some_and(HashSet::is_empty) { + return Ok(Vec::new()); + } + 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() }; let tenant_id = tenant.tenant_id().as_str(); - let filter = self.build_resource_filter( + let mut filter = self.build_resource_filter( tenant_id, resource_type, &query, @@ -5977,6 +6000,14 @@ impl MongoBackend { None, None, )?; + if let Some(ids) = &absent_ids { + filter = doc! { + "$and": [ + filter, + { "id": { "$in": ids.iter().cloned().collect::>() } } + ] + }; + } let resources = db.collection::(MongoBackend::RESOURCES_COLLECTION); let cursor = resources .find(filter) @@ -6162,6 +6193,9 @@ impl MongoBackend { candidate_ids.retain(|id| passing.contains(id)); } + if let Some(absent) = &absent_ids { + candidate_ids.retain(|id| absent.contains(id)); + } candidate_ids.retain(|id| !matched_ids.contains(id)); if !candidate_ids.is_empty() { let remaining = (2 - matches.len()) as i64; @@ -6229,6 +6263,133 @@ impl MongoBackend { Ok(matches) } + async fn absent_ids_in_bundle_transaction( + &self, + db: &mongodb::Database, + session: &mut ClientSession, + tenant: &TenantContext, + resource_type: &str, + typed_params: &[SearchParameter], + ) -> StorageResult>> { + let missing: Vec<&SearchParameter> = typed_params + .iter() + .filter(|p| is_missing_true_param(p)) + .collect(); + if missing.is_empty() { + return Ok(None); + } + let tenant_id = tenant.tenant_id().as_str(); + let probe_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, &probe_query) + .await; + let support = self.classify_missing(tenant_id, resource_type, &probe_query); + let mut combined: Option> = None; + for param in missing { + let ids = if support.is_resource_parameter(param) { + self.metadata_missing_ids( + db, + tenant_id, + resource_type, + param, + combined.as_ref(), + Some(&mut *session), + ) + .await? + } else { + self.complement_absent_ids_in_transaction( + db, + session, + tenant_id, + resource_type, + param, + combined.as_ref(), + ) + .await? + }; + if ids.is_empty() { + return Ok(Some(ids)); + } + combined = Some(ids); + } + Ok(combined) + } + + async fn complement_absent_ids_in_transaction( + &self, + db: &mongodb::Database, + session: &mut ClientSession, + tenant_id: &str, + resource_type: &str, + param: &SearchParameter, + candidates: Option<&HashSet>, + ) -> StorageResult> { + let search_index = db.collection::(MongoBackend::SEARCH_INDEX_COLLECTION); + let mut presence = + super::search_impl::missing_presence_filter(tenant_id, resource_type, param); + if let Some(ids) = candidates { + presence.insert( + "resource_id", + doc! { "$in": ids.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(); + match candidates { + Some(ids) => Ok(ids + .iter() + .filter(|id| !present.contains(*id)) + .cloned() + .collect()), + None => { + let resources = db.collection::(MongoBackend::RESOURCES_COLLECTION); + let mut cursor = resources + .find(doc! { + "tenant_id": tenant_id, + "resource_type": resource_type, + "is_deleted": false, + }) + .projection(doc! { "_id": 0, "id": 1 }) + .limit((super::search_impl::MAX_RESULT_ID_SET + 1) as i64) + .session(&mut *session) + .await + .map_err(|e| { + internal_driver_error("Failed to list live ids for missing criteria", e) + })?; + let mut all: HashSet = HashSet::new(); + while cursor + .advance(&mut *session) + .await + .map_err(|e| internal_driver_error("Failed to advance live id cursor", e))? + { + let doc = cursor.deserialize_current().map_err(|e| { + internal_error(format!("Failed to deserialize live id: {e}")) + })?; + if let Ok(id) = doc.get_str("id") { + all.insert(id.to_string()); + } + } + if all.len() > super::search_impl::MAX_RESULT_ID_SET { + return Err(StorageError::Search(SearchError::TooManyResults { + count: all.len(), + max: super::search_impl::MAX_RESULT_ID_SET, + })); + } + all.retain(|id| !present.contains(id)); + Ok(all) + } + } + } + /// 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..3b73edb010 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,74 @@ 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(); +} diff --git a/docs/mongodb/search-indexes.md b/docs/mongodb/search-indexes.md index d1630b6444..45ff80d77b 100644 --- a/docs/mongodb/search-indexes.md +++ b/docs/mongodb/search-indexes.md @@ -97,7 +97,13 @@ 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 typed complement otherwise. `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 From 2563f80f558059323ca8248a754a77dc1d7c1d56 Mon Sep 17 00:00:00 2001 From: smunini Date: Sun, 25 Oct 2026 10:02:03 -0400 Subject: [PATCH 2/2] perf(mongodb): apply conditional :missing=true per candidate batch (#2003) Conditional criteria (ifNoneExist, conditional update/delete) computed the absent-id set for the whole resource type before any other criterion narrowed it. At 310k Patients, `identifier=X&gender:missing=true` took 6-13 s per entry on the metadata route. On the value route it failed with TooManyResults even though identifier matched one resource. The id sets also had no path below the 16 MB document limit. The check now follows search (#2004): - The criteria are classified once per entry, after the same freshness probe search runs (#1999, #2007). - In the index-driven pager, :missing=true is applied to each 128-id candidate batch after the other criteria. - With nothing else to drive from (only _id, _lastUpdated and :missing=true), an unsorted cursor is read in 512-id batches and stops at the second match. Metadata-answered criteria also narrow the cursor. - metadata_missing_ids hints idx_resources_identity when it is given candidates. Otherwise the planner walked idx_resources_missing_params, reading every absent key of the type for each batch (#2233). At 310k Patients (release, mongo 7.0 rs): - identifier+:missing=true: 6.4-13 s -> 3 ms with metadata; with metadata off it no longer fails (2 ms). - broad :missing=true alone: 11.9 s -> 23 ms. - rare :missing=true alone: 2 ms with metadata; 1.2 s with metadata off (a full type pass). - Criteria without :missing and the GET shapes are unchanged. --- .../src/backends/mongodb/search_impl.rs | 16 +- .../src/backends/mongodb/storage.rs | 314 +++++++++++------- .../tests/mongodb/missing_index.rs | 130 ++++++++ docs/mongodb/search-indexes.md | 10 +- 4 files changed, 340 insertions(+), 130 deletions(-) diff --git a/crates/persistence/src/backends/mongodb/search_impl.rs b/crates/persistence/src/backends/mongodb/search_impl.rs index 421e35ac12..8de0535d08 100644 --- a/crates/persistence/src/backends/mongodb/search_impl.rs +++ b/crates/persistence/src/backends/mongodb/search_impl.rs @@ -264,7 +264,7 @@ const MAX_SEARCH_PARAM_SORT_KEYS: usize = 15; const CANDIDATE_BATCH_SIZE: usize = 512; const PROBE_ROW_LIMIT: u64 = 100_000; // 300k × ~45 bytes/UUID ≈ 13.5 MB — safely under the 16 MB BSON document cap. -pub(super) const MAX_RESULT_ID_SET: usize = 300_000; +const MAX_RESULT_ID_SET: usize = 300_000; /// Targets a reference `:identifier` search may resolve to (#1408). They /// travel in one `$in`, two entries each; 10 000 stays near 1 MB, well under @@ -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 } }, @@ -3982,10 +3982,21 @@ impl MongoBackend { ))); stages.push(doc! { "$project": { "_id": 0, "id": 1 } }); 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")?; @@ -4007,6 +4018,7 @@ impl MongoBackend { 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 diff --git a/crates/persistence/src/backends/mongodb/storage.rs b/crates/persistence/src/backends/mongodb/storage.rs index b863e401cc..55e27a9375 100644 --- a/crates/persistence/src/backends/mongodb/storage.rs +++ b/crates/persistence/src/backends/mongodb/storage.rs @@ -5966,12 +5966,17 @@ impl MongoBackend { ) .await?; - let absent_ids = self - .absent_ids_in_bundle_transaction(db, session, tenant, resource_type, &typed_params) - .await?; - if absent_ids.as_ref().is_some_and(HashSet::is_empty) { - return Ok(Vec::new()); - } + // `: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() @@ -5992,7 +5997,7 @@ impl MongoBackend { ..Default::default() }; let tenant_id = tenant.tenant_id().as_str(); - let mut filter = self.build_resource_filter( + let filter = self.build_resource_filter( tenant_id, resource_type, &query, @@ -6000,28 +6005,33 @@ impl MongoBackend { None, None, )?; - if let Some(ids) = &absent_ids { - filter = doc! { - "$and": [ - filter, - { "id": { "$in": ids.iter().cloned().collect::>() } } - ] - }; - } 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(); @@ -6193,8 +6203,18 @@ impl MongoBackend { candidate_ids.retain(|id| passing.contains(id)); } - if let Some(absent) = &absent_ids { - candidate_ids.retain(|id| absent.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() { @@ -6263,131 +6283,175 @@ impl MongoBackend { Ok(matches) } - async fn absent_ids_in_bundle_transaction( + /// 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, - db: &mongodb::Database, - session: &mut ClientSession, tenant: &TenantContext, resource_type: &str, - typed_params: &[SearchParameter], - ) -> StorageResult>> { - let missing: Vec<&SearchParameter> = typed_params - .iter() - .filter(|p| is_missing_true_param(p)) - .collect(); + missing: &[&SearchParameter], + ) -> MissingSupport { if missing.is_empty() { - return Ok(None); + return MissingSupport::none(); } let tenant_id = tenant.tenant_id().as_str(); - let probe_query = SearchQuery { + 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, &probe_query) + self.refresh_missing_metadata_freshness(tenant_id, &query) .await; - let support = self.classify_missing(tenant_id, resource_type, &probe_query); - let mut combined: Option> = None; - for param in missing { - let ids = if support.is_resource_parameter(param) { - self.metadata_missing_ids( - db, - tenant_id, - resource_type, - param, - combined.as_ref(), - Some(&mut *session), - ) - .await? - } else { - self.complement_absent_ids_in_transaction( - db, - session, - tenant_id, - resource_type, - param, - combined.as_ref(), - ) - .await? - }; - if ids.is_empty() { - return Ok(Some(ids)); - } - combined = Some(ids); - } - Ok(combined) + self.classify_missing(tenant_id, resource_type, &query) } - async fn complement_absent_ids_in_transaction( + /// 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, - param: &SearchParameter, - candidates: Option<&HashSet>, + missing: &[&SearchParameter], + missing_support: &MissingSupport, + mut candidates: HashSet, ) -> StorageResult> { let search_index = db.collection::(MongoBackend::SEARCH_INDEX_COLLECTION); - let mut presence = - super::search_impl::missing_presence_filter(tenant_id, resource_type, param); - if let Some(ids) = candidates { + 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": ids.iter().cloned().collect::>() }, + 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)); } - let present: HashSet = search_index - .distinct("resource_id", presence) + 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 presence distinct for missing criteria", e))? - .into_iter() - .filter_map(|v| v.as_str().map(str::to_string)) - .collect(); - match candidates { - Some(ids) => Ok(ids - .iter() - .filter(|id| !present.contains(*id)) - .cloned() - .collect()), - None => { - let resources = db.collection::(MongoBackend::RESOURCES_COLLECTION); - let mut cursor = resources - .find(doc! { - "tenant_id": tenant_id, - "resource_type": resource_type, - "is_deleted": false, - }) - .projection(doc! { "_id": 0, "id": 1 }) - .limit((super::search_impl::MAX_RESULT_ID_SET + 1) as i64) - .session(&mut *session) - .await - .map_err(|e| { - internal_driver_error("Failed to list live ids for missing criteria", e) - })?; - let mut all: HashSet = HashSet::new(); - while cursor - .advance(&mut *session) - .await - .map_err(|e| internal_driver_error("Failed to advance live id cursor", e))? - { - let doc = cursor.deserialize_current().map_err(|e| { - internal_error(format!("Failed to deserialize live id: {e}")) - })?; - if let Ok(id) = doc.get_str("id") { - all.insert(id.to_string()); - } + .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 all.len() > super::search_impl::MAX_RESULT_ID_SET { - return Err(StorageError::Search(SearchError::TooManyResults { - count: all.len(), - max: super::search_impl::MAX_RESULT_ID_SET, - })); + if batch.len() < BATCH_SIZE { + continue; } - all.retain(|id| !present.contains(id)); - Ok(all) } + 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. diff --git a/crates/persistence/tests/mongodb/missing_index.rs b/crates/persistence/tests/mongodb/missing_index.rs index 3b73edb010..847c9c43c1 100644 --- a/crates/persistence/tests/mongodb/missing_index.rs +++ b/crates/persistence/tests/mongodb/missing_index.rs @@ -2792,3 +2792,133 @@ async fn mongodb_missing_conditional_matches_same_bundle_write_on_empty_tenant() 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 45ff80d77b..e27538a3d6 100644 --- a/docs/mongodb/search-indexes.md +++ b/docs/mongodb/search-indexes.md @@ -101,9 +101,13 @@ 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 typed complement otherwise. `ifNoneExist` sees -resources written earlier in the same bundle; URL-borne criteria are resolved -before any write, as the R4 overlap rule requires. +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