diff --git a/README.md b/README.md index ab9f324f39..b0b8c31564 100644 --- a/README.md +++ b/README.md @@ -337,6 +337,7 @@ compressed when the client sends `Accept-Encoding`. | `HFS_MONGODB_SEARCH_TIMEOUT_MS` | *(unset)* | Soft MongoDB search/count budget (ms), 1–2147483647, shared by admission, planning, pages, totals and inline includes. Unset or blank preserves existing behavior. Checks occur before commands and matching batches, never per ID. Completed search results are returned even if late; unfinished planning, retries and additional batches can still time out. Active commands/cursor reads may overrun; detected expiry returns an error without partial results. New commands receive the remaining maxTimeMS; implicit getMore retains its cursor budget. The HTTP timeout is separate. | | `HFS_MONGODB_PROBE_TIMEOUT_MS` | `10` | Server execution budget for optional cardinality and ordered-page probes (ms), from 1 through 2147483647. Unset or blank preserves the default. A probe’s own timeout falls back to normal matching; the optional whole-search deadline still applies. | | `HFS_MONGODB_ADAPTIVE_OFFSET_PAGING` | `true` | Samples eligible no-total offsets and caches their execution strategy. `false` disables adaptive offset selection; other search plans are unchanged. | +| `HFS_MONGODB_MISSING_METADATA` | `true` | Records absent search parameters on each resource (`missing_params`) and builds `idx_resources_missing_params` so `:missing=true` can use them. `false` stops recording them (each write removes the field), skips the index build, and answers `:missing=true` from the value indexes instead; results are the same. An existing index is not dropped. | | `HFS_MONGODB_CONNECT_TIMEOUT_MS` | `5000` | TCP handshake timeout (ms) | | `HFS_MONGODB_BROAD_SEARCH_CONCURRENCY` | *(unset: no limit)* | How many potentially broad standard searches run at once per process; others wait for a slot. Reads and narrow searches (`_id`, a reference or uri, a date with `eq`/`ap`, `identifier=system\|code`) bypass this admission limit. A value at or above the pool size is accepted with a startup warning; `0` fails startup. | | `HFS_MONGODB_COUNT_BY_TENANT_MAX_TIME_MS` | `25000` | Server execution budget (`maxTimeMS`) for the cross-tenant resource count behind the Tenants UI, `GET /admin/tenants`, the existence check of `DELETE /admin/tenants/{id}` and the console tenant metrics. No other query carries it. Past it the server stops the count and the call fails with a timeout error, never a zero count. A caller outside the HTTP request timeout, such as the Tenants UI, sees that error directly and shows its error state instead of counts. The REST endpoints answer `504` only when this budget is below `HFS_REQUEST_TIMEOUT`, as the defaults are (25 s against 30 s). The request timeout starts when the request arrives, before the count, so with a budget at or above it the request timeout answers `408` first; the budget still stops the count on the server a moment later. Keep it below `HFS_REQUEST_TIMEOUT` when raising either. It bounds server execution only, not server selection, waiting for a pooled connection or network time. Must be 1 to 2147483647; anything else fails startup. The 25 s default is policy (five seconds under the default request timeout), not a large-store measurement. The count is a covered scan of the `idx_resources_live_tenant` index, which boot creates; for its build cost on an existing store see `docs/mongodb/tenant-count-index.md`. | diff --git a/crates/persistence/src/backends/mongodb/backend.rs b/crates/persistence/src/backends/mongodb/backend.rs index 9385c4ee78..d3f93d1f5a 100644 --- a/crates/persistence/src/backends/mongodb/backend.rs +++ b/crates/persistence/src/backends/mongodb/backend.rs @@ -226,6 +226,14 @@ pub struct MongoBackendConfig { #[serde(default = "default_adaptive_offset_paging")] pub adaptive_offset_paging: bool, + /// Records each resource's absent search parameters in + /// `resources.missing_params` and builds `idx_resources_missing_params` + /// (#1984, #2006). `false` stops recording them (a write removes the + /// field), skips the index build and sends every `:missing=true` search + /// through the value-index route, which returns the same results. + #[serde(default = "default_missing_metadata")] + pub missing_metadata: bool, + /// Server execution budget for optional planning probes, in milliseconds. /// Does not bound network latency or resource/count execution. #[serde(default = "default_probe_timeout_ms")] @@ -606,6 +614,14 @@ impl MongoBackendConfig { _ => return Err("HFS_MONGODB_ADAPTIVE_OFFSET_PAGING must be true or false".into()), } } + if let Some(raw) = env("HFS_MONGODB_MISSING_METADATA") { + match raw.trim().to_ascii_lowercase().as_str() { + "" => {} + "true" | "1" | "yes" | "on" => self.missing_metadata = true, + "false" | "0" | "no" | "off" => self.missing_metadata = false, + _ => return Err("HFS_MONGODB_MISSING_METADATA must be true or false".into()), + } + } if let Some(raw) = env("HFS_MONGODB_PROBE_TIMEOUT_MS") { let raw = raw.trim(); if !raw.is_empty() { @@ -778,6 +794,10 @@ fn default_adaptive_offset_paging() -> bool { true } +fn default_missing_metadata() -> bool { + true +} + fn default_bundle_transaction_budget() -> Duration { super::retry::DEFAULT_BUNDLE_TRANSACTION_BUDGET } @@ -800,6 +820,7 @@ impl Debug for MongoBackendConfig { // is classified as secret or not. let Self { adaptive_offset_paging, + missing_metadata, probe_timeout_ms, search_timeout_ms, connection_string, @@ -828,6 +849,7 @@ impl Debug for MongoBackendConfig { } = self; f.debug_struct("MongoBackendConfig") .field("adaptive_offset_paging", adaptive_offset_paging) + .field("missing_metadata", missing_metadata) .field("probe_timeout_ms", probe_timeout_ms) .field("search_timeout_ms", search_timeout_ms) .field( @@ -873,6 +895,7 @@ impl Default for MongoBackendConfig { fn default() -> Self { Self { adaptive_offset_paging: default_adaptive_offset_paging(), + missing_metadata: default_missing_metadata(), probe_timeout_ms: default_probe_timeout_ms(), search_timeout_ms: None, connection_string: default_connection_string(), @@ -1301,7 +1324,16 @@ impl MongoBackend { let db = self.get_database().await?; schema::initialize_schema_async(&db).await?; - let missing_tx = self.missing_index_tx.lock().await.take(); + let mut missing_tx = self.missing_index_tx.lock().await.take(); + if !self.config.missing_metadata + && let Some(tx) = missing_tx.take() + { + // Disabled (#2006): nothing reads or maintains the index, so it is + // neither built nor required. An existing one is left in place. + let _ = tx.send(Some(BuildOutcome::Skipped { + missing: Vec::new(), + })); + } let missing_handle = missing_tx.map(|tx| { let builder = MissingIndexBuilder::new(db.clone(), self.config.index_build); tokio::spawn(async move { @@ -2312,6 +2344,43 @@ mod tests { ); } + #[test] + fn missing_metadata_defaults_on_and_can_be_disabled() { + let default = MongoBackendConfig::default(); + assert!(default.missing_metadata); + let mut serialized = serde_json::to_value(&default).unwrap(); + serialized + .as_object_mut() + .unwrap() + .remove("missing_metadata"); + assert!( + serde_json::from_value::(serialized) + .unwrap() + .missing_metadata + ); + for (raw, enabled) in [ + ("on", true), + (" TRUE ", true), + ("off", false), + ("0", false), + ("", true), + ] { + let mut config = default.clone(); + config + .apply_search_env(|name| { + (name == "HFS_MONGODB_MISSING_METADATA").then(|| raw.to_string()) + }) + .unwrap(); + assert_eq!(config.missing_metadata, enabled, "{raw:?}"); + } + assert!( + MongoBackendConfig::default() + .apply_search_env(|name| (name == "HFS_MONGODB_MISSING_METADATA") + .then(|| "sometimes".to_string())) + .is_err() + ); + } + #[test] fn broad_search_admission_validates_configuration() { let mut config = MongoBackendConfig::default(); diff --git a/crates/persistence/src/backends/mongodb/search_impl.rs b/crates/persistence/src/backends/mongodb/search_impl.rs index 039d44c5d6..dcba6cbe1c 100644 --- a/crates/persistence/src/backends/mongodb/search_impl.rs +++ b/crates/persistence/src/backends/mongodb/search_impl.rs @@ -1632,6 +1632,14 @@ fn contained_composite_value_stages( Ok(first) } +/// Whether `missing_params` records `name` when absent. `_id` and +/// `_lastUpdated` live on every resource document and `:missing` on them +/// never reads the array, so recording them would only add index keys +/// (#2006). +pub(super) fn records_missing_metadata(name: &str) -> bool { + !matches!(name, "_id" | "_lastUpdated") +} + impl MongoBackend { fn supports_indexed_missing( &self, @@ -1669,7 +1677,7 @@ impl MongoBackend { resource_type: &str, param: &SearchParameter, ) -> bool { - !matches!(param.name.as_str(), "_id" | "_lastUpdated") + records_missing_metadata(¶m.name) && matches!(param.modifier, Some(SearchModifier::Missing)) && param .values diff --git a/crates/persistence/src/backends/mongodb/storage.rs b/crates/persistence/src/backends/mongodb/storage.rs index def78b92dd..714698b563 100644 --- a/crates/persistence/src/backends/mongodb/storage.rs +++ b/crates/persistence/src/backends/mongodb/storage.rs @@ -3598,23 +3598,26 @@ impl MongoBackend { // unparseable date, a Period end that is not a date) is as // absent as one that extracted nothing: without this it would // match neither `:missing=true` nor `:missing=false` (#2000). - let missing = extraction - .missing - .into_iter() - .map(|parameter| parameter.name) - .chain( - extraction - .present - .into_iter() - .filter(|parameter| { - !own.iter().any(|row| row_marks_present(row, parameter)) - }) - .map(|parameter| parameter.name), - ) - .collect::>(); + let missing = self.config().missing_metadata.then(|| { + extraction + .missing + .into_iter() + .map(|parameter| parameter.name) + .chain( + extraction + .present + .into_iter() + .filter(|parameter| { + !own.iter().any(|row| row_marks_present(row, parameter)) + }) + .map(|parameter| parameter.name), + ) + .filter(|name| super::search_impl::records_missing_metadata(name)) + .collect::>() + }); ( own, - Some(missing), + missing, (!extraction.failures.is_empty()).then(|| extraction.failures.join("; ")), ) } diff --git a/crates/persistence/tests/mongodb/missing_index.rs b/crates/persistence/tests/mongodb/missing_index.rs index 656053ee86..7c02660630 100644 --- a/crates/persistence/tests/mongodb/missing_index.rs +++ b/crates/persistence/tests/mongodb/missing_index.rs @@ -2492,3 +2492,159 @@ async fn mongodb_missing_metadata_count_drives_from_selective_positive_filter() } db.drop().await.unwrap(); } + +/// `_id` and `_lastUpdated` are columns of every resource document and +/// `:missing` on them never reads `missing_params`, so they are not recorded +/// even though Synthea-style resources carry no `meta.lastUpdated` (#2006). +#[tokio::test] +async fn mongodb_missing_metadata_omits_resource_columns() { + let Some(backend) = create_backend("missing_columns").await else { + return; + }; + let tenant = create_tenant("missing-columns"); + backend + .create( + &tenant, + "Observation", + observation("no-meta", vec![]), + FhirVersion::default(), + ) + .await + .unwrap(); + let db = backend.get_database().await.unwrap(); + let names = missing_parameters(&db, &tenant, "Observation", "no-meta").await; + assert!( + names.iter().any(|name| name == "_tag"), + "other absent meta parameters are still recorded: {names:?}" + ); + for column in ["_id", "_lastUpdated"] { + assert!( + !names.iter().any(|name| name == column), + "{column}: {names:?}" + ); + } +} + +/// `missing_metadata: false` (#2006): writes leave no `missing_params` and +/// clear what an enabled writer left, the index is not built, and +/// `:missing` answers exactly as with metadata. +#[tokio::test] +async fn mongodb_missing_metadata_disabled_keeps_missing_results() { + let Some(connection_string) = shared_mongo::connection_string().await else { + return; + }; + let database_name = build_test_database_name("missing_disabled"); + let config = MongoBackendConfig { + connection_string, + database_name: database_name.clone(), + data_dir: Some(repo_data_dir()), + ..Default::default() + }; + let tenant = create_tenant("missing-disabled"); + + // An enabled writer first, as on a deployment that later opts out. + let Some(enabled) = build_backend(config.clone()).await else { + return; + }; + register_parameters(&enabled, &tenant).await; + enabled + .create( + &tenant, + "Observation", + observation("written-enabled", vec![]), + FhirVersion::default(), + ) + .await + .unwrap(); + let db = enabled.get_database().await.unwrap(); + assert!( + !missing_parameters(&db, &tenant, "Observation", "written-enabled") + .await + .is_empty() + ); + db.collection::("resources") + .drop_index("idx_resources_missing_params") + .await + .unwrap(); + drop(enabled); + + let Some(disabled) = build_backend(MongoBackendConfig { + missing_metadata: false, + ..config + }) + .await + else { + return; + }; + assert!(matches!( + disabled.wait_for_missing_index_build().await, + Some(BuildOutcome::Skipped { .. }) + )); + let listed = db + .collection::("resources") + .list_index_names() + .await + .unwrap(); + assert!( + !listed + .iter() + .any(|name| name == "idx_resources_missing_params"), + "the disabled backend must not build the index: {listed:?}" + ); + + for (id, values) in [ + ("absent", vec![]), + ("null", extensions(true)), + ("present", extensions(false)), + ] { + disabled + .create( + &tenant, + "Observation", + observation(id, values), + FhirVersion::default(), + ) + .await + .unwrap(); + } + let current = disabled + .read(&tenant, "Observation", "written-enabled") + .await + .unwrap() + .unwrap(); + disabled + .update( + &tenant, + ¤t, + observation("written-enabled", extensions(false)), + ) + .await + .unwrap(); + let without_metadata = db + .collection::("resources") + .count_documents(doc! { + "tenant_id": tenant.tenant_id().as_str(), "resource_type": "Observation", + "missing_params": { "$exists": false }, + }) + .await + .unwrap(); + assert_eq!( + without_metadata, 4, + "no Observation may keep missing_params" + ); + + for parameter in &PARAMETERS { + for missing in [true, false] { + let expected: &[&str] = if missing { + &["absent", "null"] + } else { + &["present", "written-enabled"] + }; + let mut query = missing_query(parameter, missing); + for total in [TotalMode::Accurate, TotalMode::Estimate] { + query.total = Some(total); + assert_results(&disabled, &tenant, &query, expected).await; + } + } + } +} diff --git a/crates/persistence/tests/mongodb/missing_params_cost_bench.rs b/crates/persistence/tests/mongodb/missing_params_cost_bench.rs new file mode 100644 index 0000000000..9c8f3bf710 --- /dev/null +++ b/crates/persistence/tests/mongodb/missing_params_cost_bench.rs @@ -0,0 +1,536 @@ +//! #2006 measurement harness: what `resources.missing_params` and +//! `idx_resources_missing_params` (#1984) cost writers, storage and the +//! WiredTiger cache, measured by loading the same Synthea resources with +//! `missing_metadata` on and off (`HFS_MONGODB_MISSING_METADATA`). +//! +//! Each arm gets a fresh database and runs three phases: +//! `create` (concurrent single-resource creates), `update` (concurrent +//! updates of a sample of the created resources, identical content, so the +//! absent set is unchanged) and `bulk-submit` (`process_ndjson_stream` into a +//! second tenant). After the load it prints collection, index and cache +//! figures, and for the metadata arm the per-type length of the array and the +//! names that are absent from nearly every resource of a type. +//! +//! ```text +//! HFS_TEST_MISSING_COST_MONGODB_URL=mongodb://127.0.0.1:27206/?directConnection=true \ +//! HFS_TEST_MISSING_COST_CORPUS=/path/to/synthea/ndjson \ +//! HFS_TEST_MISSING_COST_ROWS=1000000 \ +//! cargo test -p helios-persistence --features mongodb --test mongodb_tests \ +//! missing_params_cost_bench -- --ignored --nocapture +//! ``` +//! +//! Other knobs: `_CONCURRENCY` (writers, default 32), `_UPDATES` (default +//! 50 000), `_BULK_ROWS` (default 200 000), `_ARMS` (default `on,off`; +//! repeat names to interleave rounds, e.g. `on,off,off,on`). + +use super::*; +use futures::StreamExt; +use helios_persistence::core::{ + BulkProcessingOptions, BulkSubmitProvider, StreamingBulkSubmitProvider, SubmissionId, +}; +use helios_persistence::types::StoredResource; +use std::io::BufRead; +use std::time::{Duration, Instant}; + +/// The corpus's resource types and their line counts, which set each type's +/// share of a load so it keeps the corpus's natural mix. +const MIX: &[(&str, u64)] = &[ + ("Observation", 7_699_881), + ("Procedure", 2_177_173), + ("DiagnosticReport", 1_656_931), + ("Claim", 1_605_515), + ("ExplanationOfBenefit", 1_605_515), + ("Encounter", 827_919), + ("DocumentReference", 827_919), + ("MedicationRequest", 777_596), + ("Condition", 476_422), + ("SupplyDelivery", 337_768), + ("Medication", 272_640), + ("MedicationAdministration", 272_640), + ("Immunization", 170_475), + ("Device", 75_076), + ("ImagingStudy", 50_151), + ("CarePlan", 41_419), + ("CareTeam", 41_419), + ("Patient", 11_704), + ("Provenance", 11_704), + ("AllergyIntolerance", 11_453), +]; + +fn env_usize(name: &str, default: usize) -> usize { + std::env::var(name) + .ok() + .and_then(|value| value.parse().ok()) + .unwrap_or(default) +} + +/// How many of `rows` each type contributes, largest remainder first. +fn shares(rows: usize) -> Vec<(&'static str, usize)> { + let total: u64 = MIX.iter().map(|(_, n)| n).sum(); + let mut shares: Vec<_> = MIX + .iter() + .map(|(t, n)| (*t, (rows as u64 * n / total) as usize)) + .collect(); + let short = rows - shares.iter().map(|(_, n)| n).sum::(); + for share in shares.iter_mut().take(short) { + share.1 += 1; + } + shares +} + +/// Streams `rows` corpus lines with the types interleaved in proportion +/// (smooth weighted round robin), from a blocking reader thread. +fn corpus_lines(corpus: &std::path::Path, rows: usize) -> tokio::sync::mpsc::Receiver { + let (tx, rx) = tokio::sync::mpsc::channel(8192); + let corpus = corpus.to_owned(); + std::thread::spawn(move || { + let mut sources: Vec<_> = shares(rows) + .into_iter() + .filter(|(_, n)| *n > 0) + .map(|(t, n)| { + let file = std::fs::File::open(corpus.join(format!("{t}.ndjson"))) + .unwrap_or_else(|e| panic!("open {t}.ndjson: {e}")); + ( + std::io::BufReader::with_capacity(1 << 20, file).lines(), + n, + 0_i64, + ) + }) + .collect(); + let total: i64 = sources.iter().map(|(_, n, _)| *n as i64).sum(); + for _ in 0..rows { + for source in &mut sources { + if source.1 > 0 { + source.2 += source.1 as i64; + } + } + let pick = sources + .iter_mut() + .filter(|source| source.1 > 0) + .max_by_key(|source| source.2) + .expect("a source with lines left"); + pick.2 -= total; + pick.1 -= 1; + let line = pick.0.next().expect("corpus line").expect("read corpus"); + if tx.blocking_send(line).is_err() { + return; + } + } + }); + rx +} + +/// All lines of one type's share, for `process_ndjson_stream`. +fn first_lines(corpus: &std::path::Path, resource_type: &str, n: usize) -> Vec { + let file = std::fs::File::open(corpus.join(format!("{resource_type}.ndjson"))).unwrap(); + let mut out = Vec::new(); + for line in std::io::BufReader::with_capacity(1 << 20, file) + .lines() + .take(n) + { + out.extend_from_slice(line.unwrap().as_bytes()); + out.push(b'\n'); + } + out +} + +struct Latencies(Vec); + +impl Latencies { + fn summary(mut self, elapsed: Duration) -> String { + self.0.sort_unstable(); + let at = |q: f64| { + let i = ((self.0.len() as f64 * q).ceil() as usize).clamp(1, self.0.len()) - 1; + self.0[i].as_secs_f64() * 1000.0 + }; + format!( + "{} | {:.0} | {:.1} | {:.1} | {:.1}", + self.0.len(), + self.0.len() as f64 / elapsed.as_secs_f64(), + at(0.50), + at(0.99), + at(0.999), + ) + } +} + +/// The WiredTiger cache counters this harness reports, read from `serverStatus`. +const CACHE_COUNTERS: &[&str] = &[ + "pages read into cache", + "pages written from cache", + "bytes read into cache", + "bytes written from cache", + "modified pages evicted", + "unmodified pages evicted", + "pages evicted by application threads", +]; + +async fn cache_counters(db: &mongodb::Database) -> Vec<(String, f64)> { + let status = db + .client() + .database("admin") + .run_command(doc! { "serverStatus": 1 }) + .await + .unwrap(); + let cache = status + .get_document("wiredTiger") + .and_then(|wt| wt.get_document("cache")) + .cloned() + .unwrap_or_default(); + let number = |key: &str| match cache.get(key) { + Some(Bson::Int32(v)) => Some(*v as f64), + Some(Bson::Int64(v)) => Some(*v as f64), + Some(Bson::Double(v)) => Some(*v), + _ => None, + }; + let mut out = Vec::new(); + for key in CACHE_COUNTERS + .iter() + .copied() + .chain(["bytes currently in the cache", "maximum bytes configured"]) + { + if let Some(value) = number(key) { + out.push((key.to_owned(), value)); + } + } + out +} + +fn print_cache_delta(phase: &str, before: &[(String, f64)], after: &[(String, f64)]) { + for (key, value) in after { + if !CACHE_COUNTERS.contains(&key.as_str()) { + continue; + } + let start = before + .iter() + .find(|(k, _)| k == key) + .map_or(0.0, |(_, v)| *v); + println!("cache | {phase} | {key} | {:.0}", value - start); + } +} + +fn mib(bytes: &Bson) -> f64 { + let bytes = match bytes { + Bson::Int32(v) => *v as f64, + Bson::Int64(v) => *v as f64, + Bson::Double(v) => *v, + _ => 0.0, + }; + bytes / (1024.0 * 1024.0) +} + +async fn print_storage(arm: &str, db: &mongodb::Database) { + for collection in ["resources", "resource_history", "search_index"] { + let mut cursor = db + .collection::(collection) + .aggregate([doc! { "$collStats": { "storageStats": {} } }]) + .await + .unwrap(); + let Some(stats) = cursor.next().await else { + continue; + }; + let stats = stats.unwrap(); + let storage = stats.get_document("storageStats").unwrap(); + let get = |key: &str| storage.get(key).cloned().unwrap_or(Bson::Int32(0)); + println!( + "storage | {arm} | {collection} | count {} | data {:.1} MiB | avg obj {:.0} B | on disk {:.1} MiB | indexes {:.1} MiB", + get("count"), + mib(&get("size")), + mib(&get("avgObjSize")) * 1024.0 * 1024.0, + mib(&get("storageSize")), + mib(&get("totalIndexSize")), + ); + if collection == "resources" + && let Ok(sizes) = storage.get_document("indexSizes") + { + for (name, size) in sizes { + println!("index | {arm} | resources.{name} | {:.1} MiB", mib(size)); + } + } + } +} + +/// Per type: resources, mean array length, and the names absent from at least +/// 99% of that type's resources with the share of all entries they hold. +async fn print_missing_profile(db: &mongodb::Database, tenant: &str) { + let resources = db.collection::("resources"); + let mut per_type = resources + .aggregate([ + doc! { "$match": { "tenant_id": tenant, "missing_params": { "$exists": true } } }, + doc! { "$group": { + "_id": "$resource_type", + "resources": { "$sum": 1 }, + "entries": { "$sum": { "$size": "$missing_params" } }, + } }, + doc! { "$sort": { "resources": -1 } }, + ]) + .await + .unwrap(); + let mut types = Vec::new(); + while let Some(row) = per_type.next().await { + let row = row.unwrap(); + let resource_type = row.get_str("_id").unwrap().to_owned(); + let count = row + .get_i32("resources") + .map(i64::from) + .unwrap_or_else(|_| row.get_i64("resources").unwrap()); + let entries = row + .get_i32("entries") + .map(i64::from) + .unwrap_or_else(|_| row.get_i64("entries").unwrap()); + types.push((resource_type, count, entries)); + } + let all_resources: i64 = types.iter().map(|t| t.1).sum(); + let all_entries: i64 = types.iter().map(|t| t.2).sum(); + println!( + "profile | all | {all_resources} resources | {:.2} absent names each", + all_entries as f64 / all_resources.max(1) as f64 + ); + + let mut names = resources + .aggregate([ + doc! { "$match": { "tenant_id": tenant, "missing_params": { "$exists": true } } }, + doc! { "$unwind": "$missing_params" }, + doc! { "$group": { + "_id": { "t": "$resource_type", "n": "$missing_params" }, + "count": { "$sum": 1 }, + } }, + ]) + .with_options( + mongodb::options::AggregateOptions::builder() + .allow_disk_use(true) + .build(), + ) + .await + .unwrap(); + let mut by_name: Vec<(String, String, i64)> = Vec::new(); + while let Some(row) = names.next().await { + let row = row.unwrap(); + let id = row.get_document("_id").unwrap(); + let count = row + .get_i32("count") + .map(i64::from) + .unwrap_or_else(|_| row.get_i64("count").unwrap()); + by_name.push(( + id.get_str("t").unwrap().to_owned(), + id.get_str("n").unwrap().to_owned(), + count, + )); + } + let mut universal_entries = 0; + for (resource_type, count, entries) in &types { + let mut universal: Vec<_> = by_name + .iter() + .filter(|(t, _, c)| t == resource_type && *c * 100 >= *count * 99) + .map(|(_, n, _)| n.as_str()) + .collect(); + universal.sort_unstable(); + let held: i64 = by_name + .iter() + .filter(|(t, n, _)| t == resource_type && universal.contains(&n.as_str())) + .map(|(_, _, c)| c) + .sum(); + universal_entries += held; + println!( + "profile | {resource_type} | {count} resources | {:.2} absent names each | {} absent from >=99%: {}", + *entries as f64 / (*count).max(1) as f64, + universal.len(), + universal.join(" "), + ); + } + println!( + "profile | all | names absent from >=99% of their type hold {:.1}% of entries", + universal_entries as f64 * 100.0 / all_entries.max(1) as f64 + ); +} + +async fn run_arm(arm: &str, metadata: bool, round: usize, url: &str, corpus: &std::path::Path) { + let rows = env_usize("HFS_TEST_MISSING_COST_ROWS", 1_000_000); + let concurrency = env_usize("HFS_TEST_MISSING_COST_CONCURRENCY", 32); + let updates = env_usize("HFS_TEST_MISSING_COST_UPDATES", 50_000).min(rows); + let bulk_rows = env_usize("HFS_TEST_MISSING_COST_BULK_ROWS", 200_000); + let database = format!("hfs_bench_2006_{arm}"); + + Client::with_uri_str(url) + .await + .unwrap() + .database(&database) + .drop() + .await + .unwrap(); + // Built directly rather than through `build_backend`, which caps the pool + // at four connections and would make the writer count meaningless. + let backend = MongoBackend::new(MongoBackendConfig { + connection_string: url.to_owned(), + database_name: database.clone(), + data_dir: Some(repo_data_dir()), + missing_metadata: metadata, + max_connections: (concurrency as u32 + 8).max(16), + index_build: IndexBuildMode::Inline, + ..Default::default() + }) + .unwrap(); + backend.initialize().await.unwrap(); + let db = backend.get_database().await.unwrap(); + let tenant = create_tenant("bench-2006"); + let label = format!("{arm}#{round}"); + + // Phase 1: concurrent creates in the corpus's type mix. + let every = (rows / updates.max(1)).max(1); + let cache_before = cache_counters(&db).await; + let started = Instant::now(); + let rx = corpus_lines(corpus, rows); + let lines = futures::stream::unfold(rx, |mut rx| async move { + rx.recv().await.map(|line| (line, rx)) + }); + let results: Vec<(Duration, Option)> = lines + .enumerate() + .map(|(number, line)| { + let backend = &backend; + let tenant = &tenant; + let label = &label; + async move { + let resource: serde_json::Value = serde_json::from_str(&line).unwrap(); + let resource_type = resource["resourceType"].as_str().unwrap().to_owned(); + let op = Instant::now(); + let stored = backend + .create(tenant, &resource_type, resource, FhirVersion::default()) + .await + .unwrap(); + let elapsed = op.elapsed(); + if number % 100_000 == 0 { + eprintln!("{label}: created {number} after {:?}", started.elapsed()); + } + (elapsed, (number % every == 0).then_some(stored)) + } + }) + .buffer_unordered(concurrency) + .collect() + .await; + let create_elapsed = started.elapsed(); + let mut sample = Vec::new(); + let mut latencies = Vec::with_capacity(results.len()); + for (latency, stored) in results { + latencies.push(latency); + sample.extend(stored); + } + println!( + "write | {label} | create | {}", + Latencies(latencies).summary(create_elapsed) + ); + let cache_after = cache_counters(&db).await; + print_cache_delta(&format!("{label} create"), &cache_before, &cache_after); + + // Phase 2: concurrent updates of the sample with unchanged content. + let cache_before = cache_after; + let started = Instant::now(); + let latencies: Vec = futures::stream::iter(sample.iter()) + .map(|current| { + let backend = &backend; + let tenant = &tenant; + async move { + let op = Instant::now(); + backend + .update(tenant, current, current.content().clone()) + .await + .unwrap(); + op.elapsed() + } + }) + .buffer_unordered(concurrency) + .collect() + .await; + println!( + "write | {label} | update | {}", + Latencies(latencies).summary(started.elapsed()) + ); + let cache_after = cache_counters(&db).await; + print_cache_delta(&format!("{label} update"), &cache_before, &cache_after); + drop(sample); + + // Phase 3: `$bulk-submit` ingestion of the same mix into a second tenant, + // one stream per type, as the submit worker drives it. + let bulk_tenant = create_tenant("bench-2006-bulk"); + let submission = SubmissionId::generate("bench-2006"); + backend + .create_submission(&bulk_tenant, &submission, None) + .await + .unwrap(); + let manifest = backend + .add_manifest( + &bulk_tenant, + &submission, + Some("https://provider.example/manifest.json"), + None, + ) + .await + .unwrap() + .manifest_id; + let cache_before = cache_after; + let mut bulk_elapsed = Duration::ZERO; + let mut bulk_count = 0; + for (resource_type, n) in shares(bulk_rows) { + if n == 0 { + continue; + } + let ndjson = first_lines(corpus, resource_type, n); + let reader = Box::new(tokio::io::BufReader::new(std::io::Cursor::new(ndjson))); + let started = Instant::now(); + let result = backend + .process_ndjson_stream( + &bulk_tenant, + &submission, + &manifest, + resource_type, + reader, + &BulkProcessingOptions::new(), + ) + .await + .unwrap(); + bulk_elapsed += started.elapsed(); + assert_eq!( + result.counts.success as usize, n, + "{resource_type}: {result:?}" + ); + bulk_count += n; + } + println!( + "write | {label} | bulk-submit | {bulk_count} | {:.0} | - | - | -", + bulk_count as f64 / bulk_elapsed.as_secs_f64() + ); + let cache_after = cache_counters(&db).await; + print_cache_delta(&format!("{label} bulk-submit"), &cache_before, &cache_after); + for (key, value) in &cache_after { + if !CACHE_COUNTERS.contains(&key.as_str()) { + println!("cache | {label} | {key} | {:.0}", value); + } + } + + print_storage(&label, &db).await; + if metadata { + print_missing_profile(&db, tenant.tenant_id().as_str()).await; + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 16)] +#[ignore = "#2006 measurement harness; run with --ignored"] +async fn missing_params_cost_bench() { + let Ok(url) = std::env::var("HFS_TEST_MISSING_COST_MONGODB_URL") else { + eprintln!("skip: HFS_TEST_MISSING_COST_MONGODB_URL is not set"); + return; + }; + let corpus = PathBuf::from( + std::env::var("HFS_TEST_MISSING_COST_CORPUS") + .expect("HFS_TEST_MISSING_COST_CORPUS: a directory of Synthea .ndjson files"), + ); + let arms = std::env::var("HFS_TEST_MISSING_COST_ARMS").unwrap_or_else(|_| "on,off".into()); + println!("| arm | phase | ops | ops/s | p50 ms | p99 ms | p99.9 ms |"); + for (round, arm) in arms.split(',').map(str::trim).enumerate() { + let metadata = match arm { + "on" => true, + "off" => false, + other => panic!("unknown arm {other:?}; use on or off"), + }; + run_arm(arm, metadata, round, &url, &corpus).await; + } +} diff --git a/crates/persistence/tests/mongodb_tests.rs b/crates/persistence/tests/mongodb_tests.rs index 6cbe667416..8c246a9a8b 100644 --- a/crates/persistence/tests/mongodb_tests.rs +++ b/crates/persistence/tests/mongodb_tests.rs @@ -24130,6 +24130,10 @@ mongodb_bundle_identity_test!( #[path = "mongodb/missing_index.rs"] mod missing_index; +/// #2006: write, storage and cache cost of `missing_params` metadata. +#[path = "mongodb/missing_params_cost_bench.rs"] +mod missing_params_cost_bench; + /// #2004: positive filters combined with metadata `:missing=true` at volume. #[path = "mongodb/missing_positive_bench.rs"] mod missing_positive_bench;