From 617dcb5b60b6f8134ddec227a54cebf56a09e57b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mauricio=20Pe=C3=B1aranda?= Date: Wed, 21 Oct 2026 10:58:21 -0400 Subject: [PATCH 1/2] perf(mongodb): HFS_MONGODB_MISSING_METADATA switch and missing_params cost bench (#2006) Adds `missing_metadata` (default on, `HFS_MONGODB_MISSING_METADATA`). Off, writes record no `missing_params` (an update removes the field), `idx_resources_missing_params` is neither built nor required, and `:missing=true` takes the value-index route, which returns the same results. This is both the A/B lever for #2006 and the opt-out the issue asks to evaluate. Adds the #[ignore] `missing_params_cost_bench`: loads the same Synthea mix with metadata on and off and reports create/update/$bulk-submit throughput and p50/p99/p99.9, collection and index sizes, WiredTiger cache deltas and the per-type absent-name profile. --- README.md | 1 + .../src/backends/mongodb/backend.rs | 71 ++- .../src/backends/mongodb/storage.rs | 32 +- .../mongodb/missing_params_cost_bench.rs | 536 ++++++++++++++++++ crates/persistence/tests/mongodb_tests.rs | 4 + 5 files changed, 628 insertions(+), 16 deletions(-) create mode 100644 crates/persistence/tests/mongodb/missing_params_cost_bench.rs 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 50f482d88b..ed2113c1b7 100644 --- a/crates/persistence/src/backends/mongodb/backend.rs +++ b/crates/persistence/src/backends/mongodb/backend.rs @@ -212,6 +212,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")] @@ -592,6 +600,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() { @@ -764,6 +780,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 } @@ -786,6 +806,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, @@ -814,6 +835,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( @@ -859,6 +881,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(), @@ -1284,7 +1307,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 { @@ -2295,6 +2327,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/storage.rs b/crates/persistence/src/backends/mongodb/storage.rs index a614311c3a..8cd0b1187b 100644 --- a/crates/persistence/src/backends/mongodb/storage.rs +++ b/crates/persistence/src/backends/mongodb/storage.rs @@ -3543,23 +3543,25 @@ 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), + ) + .collect::>() + }); ( own, - Some(missing), + missing, (!extraction.failures.is_empty()).then(|| extraction.failures.join("; ")), ) } 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 c127ae1554..d001771c15 100644 --- a/crates/persistence/tests/mongodb_tests.rs +++ b/crates/persistence/tests/mongodb_tests.rs @@ -24095,6 +24095,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; + #[path = "mongodb/reindex_failures.rs"] mod reindex_failures; From 429aed25fb01556a2abf8ad60068e06aaa9339f2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mauricio=20Pe=C3=B1aranda?= Date: Wed, 21 Oct 2026 11:42:34 -0400 Subject: [PATCH 2/2] perf(mongodb): stop recording _id/_lastUpdated in missing_params (#2006) `:missing` on `_id` or `_lastUpdated` never reads `missing_params` (`is_missing_resource_parameter` excludes them; both are columns of every resource document), yet every resource without `meta.lastUpdated` in its body -- all of the Synthea corpus -- recorded `_lastUpdated` as absent: one array entry and one multikey index key per resource that nothing reads. Both checks now share `records_missing_metadata`. Tests: the recorded set omits both columns but keeps other absent meta parameters; and the `missing_metadata: false` switch clears what an enabled writer left, skips the index build, and answers `:missing=true` and `:missing=false` exactly as with metadata. --- .../src/backends/mongodb/search_impl.rs | 10 +- .../src/backends/mongodb/storage.rs | 1 + .../tests/mongodb/missing_index.rs | 156 ++++++++++++++++++ 3 files changed, 166 insertions(+), 1 deletion(-) diff --git a/crates/persistence/src/backends/mongodb/search_impl.rs b/crates/persistence/src/backends/mongodb/search_impl.rs index 25a3ed2acf..76e73b5f60 100644 --- a/crates/persistence/src/backends/mongodb/search_impl.rs +++ b/crates/persistence/src/backends/mongodb/search_impl.rs @@ -1626,6 +1626,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, @@ -1663,7 +1671,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 8cd0b1187b..b8f49cd7ef 100644 --- a/crates/persistence/src/backends/mongodb/storage.rs +++ b/crates/persistence/src/backends/mongodb/storage.rs @@ -3557,6 +3557,7 @@ impl MongoBackend { }) .map(|parameter| parameter.name), ) + .filter(|name| super::search_impl::records_missing_metadata(name)) .collect::>() }); ( diff --git a/crates/persistence/tests/mongodb/missing_index.rs b/crates/persistence/tests/mongodb/missing_index.rs index bb3b338e1b..0be1be0ddc 100644 --- a/crates/persistence/tests/mongodb/missing_index.rs +++ b/crates/persistence/tests/mongodb/missing_index.rs @@ -1904,3 +1904,159 @@ async fn mongodb_missing_index_follows_index_build_mode() { .await; } } + +/// `_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; + } + } + } +}