diff --git a/.claude/skills/run-hfs-server/SKILL.md b/.claude/skills/run-hfs-server/SKILL.md index 48247c96e..f5db7751d 100644 --- a/.claude/skills/run-hfs-server/SKILL.md +++ b/.claude/skills/run-hfs-server/SKILL.md @@ -35,6 +35,7 @@ HFS_SERVER_PORT=3000 HFS_LOG_LEVEL=debug cargo run --bin hfs | `HFS_LOG_LEVEL` | `info` | Log level: error, warn, info, debug, trace | | `HFS_BASE_URL` | `http://localhost:8080` | Base URL for Location headers and Bundle links | | `HFS_DATA_DIR` | `./data` | FHIR data directory, including search parameters | +| `HFS_SEARCH_PARAM_CACHE_TTL` | `3600` | Seconds between refreshes of the in-memory SearchParameter registry from storage; a param POSTed to one cluster node becomes visible to others within this interval. `0` disables the refresh. | ## Limits diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f1c5e5309..621450286 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -63,13 +63,19 @@ jobs: rm -f ~/.cargo/config.toml echo '[target.aarch64-apple-darwin]' >> ~/.cargo/config.toml echo 'linker = "clang"' >> ~/.cargo/config.toml - # NOTE: do NOT pass `-undefined dynamic_lookup` here. That is a - # loadable-bundle linker mode, not valid for executables/test - # binaries. On Apple Silicon it yields Mach-O binaries dyld cannot - # map against the shared cache ("syscall to map cache into shared - # region failed" -> SystemConfiguration.framework fails to load -> - # SIGABRT). Match the Linux side: only bump the main-thread stack. - echo 'rustflags = ["-C", "link-arg=-Wl,-stack_size,0x800000"]' >> ~/.cargo/config.toml + # No custom rustflags on macOS. Two flags that were tried here both + # broke the build: + # * `-undefined dynamic_lookup` is a loadable-bundle linker mode, + # invalid for executables/test binaries. On Apple Silicon it + # yields Mach-O binaries dyld cannot map against the shared cache + # ("syscall to map cache into shared region failed" -> + # SystemConfiguration.framework fails to load -> SIGABRT). + # * `-stack_size` is rejected by ld64 on any non-main-executable + # link ("-stack_size option can only be used when linking a main + # executable"), and cargo applies rustflags to proc-macro/cdylib + # links too (e.g. serde_derive), so it fails the build outright. + # The main-thread stack already defaults to 8 MB on macOS, matching + # the Linux `-zstack-size=8388608`, so no override is needed. - name: Run tests with all features run: | @@ -168,13 +174,19 @@ jobs: rm -f ~/.cargo/config.toml echo '[target.aarch64-apple-darwin]' >> ~/.cargo/config.toml echo 'linker = "clang"' >> ~/.cargo/config.toml - # NOTE: do NOT pass `-undefined dynamic_lookup` here. That is a - # loadable-bundle linker mode, not valid for executables/test - # binaries. On Apple Silicon it yields Mach-O binaries dyld cannot - # map against the shared cache ("syscall to map cache into shared - # region failed" -> SystemConfiguration.framework fails to load -> - # SIGABRT). Match the Linux side: only bump the main-thread stack. - echo 'rustflags = ["-C", "link-arg=-Wl,-stack_size,0x800000"]' >> ~/.cargo/config.toml + # No custom rustflags on macOS. Two flags that were tried here both + # broke the build: + # * `-undefined dynamic_lookup` is a loadable-bundle linker mode, + # invalid for executables/test binaries. On Apple Silicon it + # yields Mach-O binaries dyld cannot map against the shared cache + # ("syscall to map cache into shared region failed" -> + # SystemConfiguration.framework fails to load -> SIGABRT). + # * `-stack_size` is rejected by ld64 on any non-main-executable + # link ("-stack_size option can only be used when linking a main + # executable"), and cargo applies rustflags to proc-macro/cdylib + # links too (e.g. serde_derive), so it fails the build outright. + # The main-thread stack already defaults to 8 MB on macOS, matching + # the Linux `-zstack-size=8388608`, so no override is needed. - name: Build pysof wheel working-directory: crates/pysof @@ -226,13 +238,19 @@ jobs: rm -f ~/.cargo/config.toml echo '[target.aarch64-apple-darwin]' >> ~/.cargo/config.toml echo 'linker = "clang"' >> ~/.cargo/config.toml - # NOTE: do NOT pass `-undefined dynamic_lookup` here. That is a - # loadable-bundle linker mode, not valid for executables/test - # binaries. On Apple Silicon it yields Mach-O binaries dyld cannot - # map against the shared cache ("syscall to map cache into shared - # region failed" -> SystemConfiguration.framework fails to load -> - # SIGABRT). Match the Linux side: only bump the main-thread stack. - echo 'rustflags = ["-C", "link-arg=-Wl,-stack_size,0x800000"]' >> ~/.cargo/config.toml + # No custom rustflags on macOS. Two flags that were tried here both + # broke the build: + # * `-undefined dynamic_lookup` is a loadable-bundle linker mode, + # invalid for executables/test binaries. On Apple Silicon it + # yields Mach-O binaries dyld cannot map against the shared cache + # ("syscall to map cache into shared region failed" -> + # SystemConfiguration.framework fails to load -> SIGABRT). + # * `-stack_size` is rejected by ld64 on any non-main-executable + # link ("-stack_size option can only be used when linking a main + # executable"), and cargo applies rustflags to proc-macro/cdylib + # links too (e.g. serde_derive), so it fails the build outright. + # The main-thread stack already defaults to 8 MB on macOS, matching + # the Linux `-zstack-size=8388608`, so no override is needed. - name: Check formatting run: cargo fmt --all -- --check @@ -302,13 +320,19 @@ jobs: rm -f ~/.cargo/config.toml echo '[target.aarch64-apple-darwin]' >> ~/.cargo/config.toml echo 'linker = "clang"' >> ~/.cargo/config.toml - # NOTE: do NOT pass `-undefined dynamic_lookup` here. That is a - # loadable-bundle linker mode, not valid for executables/test - # binaries. On Apple Silicon it yields Mach-O binaries dyld cannot - # map against the shared cache ("syscall to map cache into shared - # region failed" -> SystemConfiguration.framework fails to load -> - # SIGABRT). Match the Linux side: only bump the main-thread stack. - echo 'rustflags = ["-C", "link-arg=-Wl,-stack_size,0x800000"]' >> ~/.cargo/config.toml + # No custom rustflags on macOS. Two flags that were tried here both + # broke the build: + # * `-undefined dynamic_lookup` is a loadable-bundle linker mode, + # invalid for executables/test binaries. On Apple Silicon it + # yields Mach-O binaries dyld cannot map against the shared cache + # ("syscall to map cache into shared region failed" -> + # SystemConfiguration.framework fails to load -> SIGABRT). + # * `-stack_size` is rejected by ld64 on any non-main-executable + # link ("-stack_size option can only be used when linking a main + # executable"), and cargo applies rustflags to proc-macro/cdylib + # links too (e.g. serde_derive), so it fails the build outright. + # The main-thread stack already defaults to 8 MB on macOS, matching + # the Linux `-zstack-size=8388608`, so no override is needed. - name: Install cargo-audit run: cargo install cargo-audit @@ -402,13 +426,19 @@ jobs: rm -f ~/.cargo/config.toml echo '[target.aarch64-apple-darwin]' >> ~/.cargo/config.toml echo 'linker = "clang"' >> ~/.cargo/config.toml - # NOTE: do NOT pass `-undefined dynamic_lookup` here. That is a - # loadable-bundle linker mode, not valid for executables/test - # binaries. On Apple Silicon it yields Mach-O binaries dyld cannot - # map against the shared cache ("syscall to map cache into shared - # region failed" -> SystemConfiguration.framework fails to load -> - # SIGABRT). Match the Linux side: only bump the main-thread stack. - echo 'rustflags = ["-C", "link-arg=-Wl,-stack_size,0x800000"]' >> ~/.cargo/config.toml + # No custom rustflags on macOS. Two flags that were tried here both + # broke the build: + # * `-undefined dynamic_lookup` is a loadable-bundle linker mode, + # invalid for executables/test binaries. On Apple Silicon it + # yields Mach-O binaries dyld cannot map against the shared cache + # ("syscall to map cache into shared region failed" -> + # SystemConfiguration.framework fails to load -> SIGABRT). + # * `-stack_size` is rejected by ld64 on any non-main-executable + # link ("-stack_size option can only be used when linking a main + # executable"), and cargo applies rustflags to proc-macro/cdylib + # links too (e.g. serde_derive), so it fails the build outright. + # The main-thread stack already defaults to 8 MB on macOS, matching + # the Linux `-zstack-size=8388608`, so no override is needed. - name: Install cargo-llvm-cov run: cargo install cargo-llvm-cov @@ -782,13 +812,19 @@ jobs: rm -f ~/.cargo/config.toml echo '[target.aarch64-apple-darwin]' >> ~/.cargo/config.toml echo 'linker = "clang"' >> ~/.cargo/config.toml - # NOTE: do NOT pass `-undefined dynamic_lookup` here. That is a - # loadable-bundle linker mode, not valid for executables/test - # binaries. On Apple Silicon it yields Mach-O binaries dyld cannot - # map against the shared cache ("syscall to map cache into shared - # region failed" -> SystemConfiguration.framework fails to load -> - # SIGABRT). Match the Linux side: only bump the main-thread stack. - echo 'rustflags = ["-C", "link-arg=-Wl,-stack_size,0x800000"]' >> ~/.cargo/config.toml + # No custom rustflags on macOS. Two flags that were tried here both + # broke the build: + # * `-undefined dynamic_lookup` is a loadable-bundle linker mode, + # invalid for executables/test binaries. On Apple Silicon it + # yields Mach-O binaries dyld cannot map against the shared cache + # ("syscall to map cache into shared region failed" -> + # SystemConfiguration.framework fails to load -> SIGABRT). + # * `-stack_size` is rejected by ld64 on any non-main-executable + # link ("-stack_size option can only be used when linking a main + # executable"), and cargo applies rustflags to proc-macro/cdylib + # links too (e.g. serde_derive), so it fails the build outright. + # The main-thread stack already defaults to 8 MB on macOS, matching + # the Linux `-zstack-size=8388608`, so no override is needed. - name: Build Release (Windows) if: matrix.os == 'Windows' diff --git a/README.md b/README.md index 8eb1a581f..f8b15266e 100644 --- a/README.md +++ b/README.md @@ -236,6 +236,7 @@ AWS_REGION=us-east-1 \ | `HFS_BASE_URL` | `http://localhost:8080` | Base URL used in `Location` headers and Bundle links | | `HFS_DATABASE_URL` | `fhir.db` | Database URL (SQLite path or PostgreSQL connection string) | | `HFS_DATA_DIR` | `./data` | Directory containing FHIR data files (search parameters) | +| `HFS_SEARCH_PARAM_CACHE_TTL` | `3600` | Seconds between refreshes of the in-memory SearchParameter registry from storage; a param POSTed to one cluster node becomes visible to others within this interval. `0` disables the refresh. | | `HFS_DEFAULT_FHIR_VERSION` | `R4` | FHIR version (R4, R4B, R5, R6) | | `HFS_LOG_LEVEL` | `info` | Log level (error, warn, info, debug, trace) | diff --git a/book/src/configuration/environment-variables.md b/book/src/configuration/environment-variables.md index d50543d91..1bc28ad34 100644 --- a/book/src/configuration/environment-variables.md +++ b/book/src/configuration/environment-variables.md @@ -11,6 +11,7 @@ All server behavior is controlled through environment variables. No configuratio | `HFS_LOG_LEVEL` | `info` | Log level: `error`, `warn`, `info`, `debug`, `trace` | | `HFS_BASE_URL` | `http://localhost:8080` | Base URL for Location headers and Bundle links | | `HFS_DATA_DIR` | `./data` | Path to FHIR data directory (search parameters) | +| `HFS_SEARCH_PARAM_CACHE_TTL` | `3600` | Seconds between refreshes of the in-memory SearchParameter registry from storage. In a cluster, a SearchParameter POSTed to one node becomes visible to the others within this interval. `0` disables the periodic refresh. | ## Limits diff --git a/crates/fhir/src/search/loader.rs b/crates/fhir/src/search/loader.rs index 13a8c6c76..419e36f0f 100644 --- a/crates/fhir/src/search/loader.rs +++ b/crates/fhir/src/search/loader.rs @@ -159,6 +159,75 @@ impl SearchParameterLoader { Ok(params) } + /// Returns the raw SearchParameter resources from the FHIR spec bundle, + /// as stored-shape JSON. + /// + /// This is the seeding companion to + /// [`load_from_spec_file`](Self::load_from_spec_file): the registry wants + /// parsed definitions, but seeding storage wants the original resource + /// JSON (ids, metadata, and fields the definition model drops). Only + /// entries that also parse as definitions are returned, so storage never + /// holds a parameter the registry would reject. + pub fn load_spec_resources(&self, data_dir: &Path) -> Result, LoaderError> { + let path = data_dir.join(self.spec_filename()); + let content = + std::fs::read_to_string(&path).map_err(|e| LoaderError::ConfigLoadFailed { + path: path.display().to_string(), + message: e.to_string(), + })?; + let json: Value = + serde_json::from_str(&content).map_err(|e| LoaderError::ConfigLoadFailed { + path: path.display().to_string(), + message: format!("Invalid JSON: {}", e), + })?; + + let mut resources = Vec::new(); + if let Some(entries) = json.get("entry").and_then(|e| e.as_array()) { + for entry in entries { + if let Some(resource) = entry.get("resource") + && resource.get("resourceType").and_then(|t| t.as_str()) + == Some("SearchParameter") + && self.parse_resource(resource).is_ok() + { + resources.push(resource.clone()); + } + } + } + Ok(resources) + } + + /// Renders a definition as a minimal FHIR SearchParameter resource. + /// + /// Used to seed storage with the embedded fallback parameters, which are + /// constructed programmatically and have no source JSON. The resource id + /// is the last segment of the canonical URL (`.../Resource-id` → id + /// `Resource-id`), matching the spec bundle's convention. + pub fn definition_to_fhir_resource(def: &SearchParameterDefinition) -> Value { + let id = def.url.rsplit('/').next().unwrap_or(&def.code); + let mut resource = serde_json::json!({ + "resourceType": "SearchParameter", + "id": id, + "url": def.url, + "name": def.name.clone().unwrap_or_else(|| def.code.clone()), + "status": def.status.to_fhir_status(), + "code": def.code, + "base": def.base, + "type": def.param_type.to_string(), + "expression": def.expression, + }); + let fields = resource.as_object_mut().expect("literal object"); + if let Some(description) = &def.description { + fields.insert("description".into(), Value::String(description.clone())); + } + if let Some(target) = &def.target { + fields.insert( + "target".into(), + Value::Array(target.iter().cloned().map(Value::String).collect()), + ); + } + resource + } + /// Loads custom SearchParameter files from the data directory. /// /// Scans the data directory for JSON files that are not the standard @@ -692,6 +761,112 @@ mod tests { assert!(matches!(result, Err(LoaderError::MissingField { field, .. }) if field == "url")); } + /// The workspace data directory holding `search-parameters-r4.json`. + fn workspace_data_dir() -> std::path::PathBuf { + std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../../data") + } + + #[test] + fn test_load_spec_resources_returns_stored_shape_json() { + let loader = SearchParameterLoader::new(FhirVersion::R4); + let resources = loader + .load_spec_resources(&workspace_data_dir()) + .expect("load spec resources"); + + assert!( + resources.len() > 1300, + "expected the full R4 spec set, got {}", + resources.len() + ); + // Every entry is a SearchParameter carrying its bundle id — that id is + // what makes seeding idempotent (a duplicate `create` fails). + for resource in &resources { + assert_eq!( + resource.get("resourceType").and_then(|t| t.as_str()), + Some("SearchParameter") + ); + assert!( + resource.get("id").and_then(|id| id.as_str()).is_some(), + "spec resource is missing an id: {resource}" + ); + } + } + + #[test] + fn test_load_spec_resources_missing_dir_errors() { + let loader = SearchParameterLoader::new(FhirVersion::R4); + let result = loader.load_spec_resources(Path::new("/nonexistent/data/dir")); + assert!(matches!(result, Err(LoaderError::ConfigLoadFailed { .. }))); + } + + #[test] + fn test_definition_to_fhir_resource() { + // A reference parameter with description and targets exercises both + // optional-field branches. + let mut def = SearchParameterDefinition::new( + "http://example.org/fhir/SearchParameter/Patient-organization", + "organization", + SearchParamType::Reference, + "Patient.managingOrganization", + ) + .with_base(vec!["Patient"]) + .with_targets(vec!["Organization"]); + def.description = Some("The organization managing the patient".to_string()); + + let resource = SearchParameterLoader::definition_to_fhir_resource(&def); + assert_eq!( + resource.get("resourceType").and_then(|t| t.as_str()), + Some("SearchParameter") + ); + // Id is the last URL segment, matching the spec bundle convention. + assert_eq!( + resource.get("id").and_then(|i| i.as_str()), + Some("Patient-organization") + ); + assert_eq!( + resource.get("code").and_then(|c| c.as_str()), + Some("organization") + ); + assert_eq!( + resource.get("type").and_then(|t| t.as_str()), + Some("reference") + ); + assert_eq!( + resource.get("description").and_then(|d| d.as_str()), + Some("The organization managing the patient") + ); + assert_eq!( + resource.get("target").and_then(|t| t.as_array()), + Some(&vec![Value::String("Organization".to_string())]) + ); + // The rendered resource round-trips back into a parseable definition. + assert!(loader_can_parse(&resource)); + + // A parameter with no description/target omits those fields entirely. + let minimal = SearchParameterLoader::definition_to_fhir_resource( + &SearchParameterDefinition::new( + "http://hl7.org/fhir/SearchParameter/Resource-id", + "_id", + SearchParamType::Token, + "id", + ) + .with_base(vec!["Resource"]), + ); + assert_eq!( + minimal.get("id").and_then(|i| i.as_str()), + Some("Resource-id") + ); + assert!(minimal.get("description").is_none()); + assert!(minimal.get("target").is_none()); + assert!(loader_can_parse(&minimal)); + } + + fn loader_can_parse(resource: &Value) -> bool { + SearchParameterLoader::new(FhirVersion::R4) + .parse_resource(resource) + .is_ok() + } + #[test] fn test_transform_as_to_oftype() { assert_eq!( diff --git a/crates/fhir/src/search/registry.rs b/crates/fhir/src/search/registry.rs index e27f43303..b0fc71e0e 100644 --- a/crates/fhir/src/search/registry.rs +++ b/crates/fhir/src/search/registry.rs @@ -496,6 +496,29 @@ impl SearchParameterRegistry { Ok(()) } + /// Removes every parameter registered from `source`, returning how many + /// were removed. + /// + /// This is the rebuild primitive for a storage-backed refresh: drop all + /// `Stored` definitions, then re-register what storage currently holds. + /// Slots shared with other sources keep their remaining candidates, so a + /// spec parameter shadowed by a since-deleted stored override resumes + /// resolving searches. + pub fn unregister_source(&mut self, source: SearchParameterSource) -> usize { + let urls: Vec = self + .params_by_url + .iter() + .filter(|(_, p)| p.source == source) + .map(|(url, _)| url.clone()) + .collect(); + for url in &urls { + // The URL came straight out of the index; a NotFound here would + // mean the two indexes disagree, which unregister() prevents. + let _ = self.unregister(url); + } + urls.len() + } + /// Returns all resource types that have registered parameters. pub fn resource_types(&self) -> Vec { self.params_by_type.keys().cloned().collect() @@ -930,6 +953,40 @@ mod tests { ); } + /// `unregister_source` is the storage-refresh rebuild primitive (#235): + /// dropping all `Stored` params must promote any spec param they were + /// shadowing, and leave other sources untouched. + #[test] + fn unregister_source_removes_only_that_source_and_promotes_shadowed() { + let mut registry = SearchParameterRegistry::new(); + registry.register(spec_code_param()).unwrap(); + let stored_override = SearchParameterDefinition::new( + "http://acme.health/fhir/SearchParameter/observation-code-stored", + "code", + SearchParamType::Token, + "Observation.code", + ) + .with_base(vec!["Observation"]) + .with_source(SearchParameterSource::Stored); + registry.register(stored_override.clone()).unwrap(); + assert_eq!( + registry.get_param("Observation", "code").unwrap().url, + stored_override.url + ); + + let removed = registry.unregister_source(SearchParameterSource::Stored); + + assert_eq!(removed, 1); + assert_eq!(registry.len(), 1); + // The spec param the stored override was shadowing resolves again. + assert_eq!( + registry.get_param("Observation", "code").unwrap().url, + SPEC_URL + ); + // Removing a source with no registrations is a no-op. + assert_eq!(registry.unregister_source(SearchParameterSource::Stored), 0); + } + #[test] fn update_status_still_applies_to_the_winner() { let mut registry = SearchParameterRegistry::new(); diff --git a/crates/hfs/README.md b/crates/hfs/README.md index 0b2860962..cb45fa0a1 100644 --- a/crates/hfs/README.md +++ b/crates/hfs/README.md @@ -82,6 +82,7 @@ Options: | `HFS_LOG_LEVEL` | info | Log level (error, warn, info, debug, trace) | | `DATABASE_URL` | fhir.db | Database connection string | | `HFS_DATA_DIR` | ./data | Path to FHIR data directory (search parameters) | +| `HFS_SEARCH_PARAM_CACHE_TTL` | 3600 | Seconds between refreshes of the in-memory SearchParameter registry from storage; a param POSTed to one cluster node becomes visible to others within this interval. `0` disables the refresh. | | `HFS_MAX_BODY_SIZE` | 10485760 | Max request body size (bytes; applies to the decompressed body for compressed requests) | | `HFS_REQUEST_TIMEOUT` | 30 | Request timeout (seconds) | | `HFS_ENABLE_CORS` | true | Enable CORS | diff --git a/crates/hfs/src/main.rs b/crates/hfs/src/main.rs index 9166e8b1f..fdf3fdde1 100644 --- a/crates/hfs/src/main.rs +++ b/crates/hfs/src/main.rs @@ -510,6 +510,8 @@ async fn start_mongodb( backend.init_schema().await?; let backend = Arc::new(backend); + seed_search_parameters(&*backend, &config).await; + spawn_mongodb_search_param_refresh(backend.clone(), &config); let serve_audit_state = audit_state.clone(); // MongoDB is a full standalone primary, so it also hosts the per-user @@ -841,6 +843,108 @@ async fn main() -> anyhow::Result<()> { Ok(()) } +/// Seeds storage with the spec SearchParameters under the default tenant +/// (#235), making storage the source of truth the registry caches over. A +/// failed seed logs and boots anyway: the in-memory registry still resolves +/// searches; only API discovery of the spec parameters is degraded. +async fn seed_search_parameters(backend: &S, config: &ServerConfig) +where + S: helios_persistence::core::ResourceStorage, +{ + let data_dir = config + .data_dir + .clone() + .unwrap_or_else(|| std::path::PathBuf::from("./data")); + if let Err(e) = helios_persistence::search::seed_spec_search_parameters( + backend, + config.default_fhir_version, + &data_dir, + &config.default_tenant, + ) + .await + { + tracing::warn!("SearchParameter seeding failed: {e}"); + } +} + +/// Spawns the periodic registry refresh from storage for the SQLite backend +/// (#235). `HFS_SEARCH_PARAM_CACHE_TTL=0` disables it. A failed pass keeps +/// serving the stale cache; the next tick retries. +#[cfg(feature = "sqlite")] +fn spawn_sqlite_search_param_refresh(backend: Arc, config: &ServerConfig) { + let ttl = config.search_param_cache_ttl; + if ttl == 0 { + return; + } + let interval = std::time::Duration::from_secs(ttl); + tokio::spawn(async move { + loop { + tokio::time::sleep(interval).await; + let refresh = backend.clone(); + match tokio::task::spawn_blocking(move || refresh.refresh_stored_search_parameters()) + .await + { + Ok(Ok(stored)) => { + tracing::debug!(stored, "SearchParameter registry refreshed from storage") + } + Ok(Err(e)) => tracing::warn!( + "SearchParameter registry refresh failed; serving the stale cache: {e}" + ), + Err(e) => tracing::warn!("SearchParameter registry refresh task failed: {e}"), + } + } + }); +} + +/// Postgres flavor of the periodic registry refresh (#235). +#[cfg(feature = "postgres")] +fn spawn_postgres_search_param_refresh( + backend: Arc, + config: &ServerConfig, +) { + let ttl = config.search_param_cache_ttl; + if ttl == 0 { + return; + } + let interval = std::time::Duration::from_secs(ttl); + tokio::spawn(async move { + loop { + tokio::time::sleep(interval).await; + match backend.refresh_stored_search_parameters().await { + Ok(stored) => { + tracing::debug!(stored, "SearchParameter registry refreshed from storage") + } + Err(e) => tracing::warn!( + "SearchParameter registry refresh failed; serving the stale cache: {e}" + ), + } + } + }); +} + +/// MongoDB flavor of the periodic registry refresh (#235). +#[cfg(feature = "mongodb")] +fn spawn_mongodb_search_param_refresh(backend: Arc, config: &ServerConfig) { + let ttl = config.search_param_cache_ttl; + if ttl == 0 { + return; + } + let interval = std::time::Duration::from_secs(ttl); + tokio::spawn(async move { + loop { + tokio::time::sleep(interval).await; + match backend.refresh_stored_search_parameters().await { + Ok(stored) => { + tracing::debug!(stored, "SearchParameter registry refreshed from storage") + } + Err(e) => tracing::warn!( + "SearchParameter registry refresh failed; serving the stale cache: {e}" + ), + } + } + }); +} + /// Starts the server with SQLite-only backend. #[cfg(feature = "sqlite")] async fn start_sqlite( @@ -851,6 +955,8 @@ async fn start_sqlite( ) -> anyhow::Result<()> { let serve_audit_state = audit_state.clone(); let backend = Arc::new(create_sqlite_backend(&config)?); + seed_search_parameters(&*backend, &config).await; + spawn_sqlite_search_param_refresh(backend.clone(), &config); // Second handle to the same backend for the web UI's tenant-maintenance // read/write path (the FHIR app keeps its own). Cheap: the SQLite backend // shares one connection pool behind the Arc. @@ -1380,6 +1486,9 @@ async fn start_sqlite_elasticsearch( sqlite.set_search_offloaded(true); let sqlite = Arc::new(sqlite); info!("SQLite search indexing disabled (offloaded to Elasticsearch)"); + // Seed/refresh on the primary; the ES backend shares its registry Arc. + seed_search_parameters(&*sqlite, &config).await; + spawn_sqlite_search_param_refresh(sqlite.clone(), &config); // Build Elasticsearch configuration from server config let es_nodes: Vec = config @@ -1545,6 +1654,8 @@ async fn start_postgres( backend.init_schema().await?; let backend = Arc::new(backend); + seed_search_parameters(&*backend, &config).await; + spawn_postgres_search_param_refresh(backend.clone(), &config); let serve_audit_state = audit_state.clone(); // The PostgreSQL backend also hosts the per-user settings store, so it always @@ -1626,6 +1737,9 @@ async fn start_postgres_elasticsearch( backend.set_search_offloaded(true); let pg = Arc::new(backend); info!("PostgreSQL search indexing disabled (offloaded to Elasticsearch)"); + // Seed/refresh on the primary; the ES backend shares its registry Arc. + seed_search_parameters(&*pg, &config).await; + spawn_postgres_search_param_refresh(pg.clone(), &config); // Build Elasticsearch configuration from server config let es_nodes: Vec = config @@ -1793,6 +1907,9 @@ async fn start_mongodb_elasticsearch( // Offload search to Elasticsearch let mongo = Arc::new(backend); info!("MongoDB search indexing disabled (offloaded to Elasticsearch)"); + // Seed/refresh on the primary; the ES backend shares its registry Arc. + seed_search_parameters(&*mongo, &config).await; + spawn_mongodb_search_param_refresh(mongo.clone(), &config); // Build Elasticsearch configuration from server config let es_nodes: Vec = config diff --git a/crates/persistence/src/backends/mongodb/backend.rs b/crates/persistence/src/backends/mongodb/backend.rs index ac9b8b879..649481639 100644 --- a/crates/persistence/src/backends/mongodb/backend.rs +++ b/crates/persistence/src/backends/mongodb/backend.rs @@ -345,7 +345,94 @@ impl MongoBackend { /// Initializes the MongoDB schema/index bootstrap for this backend. pub async fn init_schema(&self) -> StorageResult<()> { let db = self.get_database().await?; - schema::initialize_schema_async(&db).await + schema::initialize_schema_async(&db).await?; + + // Restore stored (POSTed) SearchParameters into the registry, as the + // SQLite and Postgres backends do; previously they only re-entered + // the registry through the write hooks, i.e. never after a restart. + let stored_count = self.refresh_stored_search_parameters().await?; + if stored_count > 0 { + tracing::info!( + "Loaded {} stored SearchParameters from database", + stored_count + ); + } + Ok(()) + } + + /// Rebuilds the registry's `Stored` parameters from what the database + /// currently holds, returning how many are registered afterwards. + /// + /// TTL-cache refresh (#235): documents are fetched and parsed before the + /// (sync) write lock is taken, and on any read error the registry is left + /// untouched — stale-serve rather than losing search resolution. + pub async fn refresh_stored_search_parameters(&self) -> StorageResult { + use crate::search::registry::{SearchParameterSource, SearchParameterStatus}; + use mongodb::bson::{Document, doc}; + + let db = self.get_database().await?; + let collection = db.collection::(Self::RESOURCES_COLLECTION); + let mut cursor = collection + .find(doc! { "resource_type": "SearchParameter", "is_deleted": false }) + .await + .map_err(|e| { + crate::error::StorageError::Backend(BackendError::Internal { + backend_name: "mongodb".to_string(), + message: format!("Failed to query SearchParameters: {}", e), + source: None, + }) + })?; + + let loader = SearchParameterLoader::new(self.config.fhir_version); + let mut definitions = Vec::new(); + while cursor.advance().await.map_err(|e| { + crate::error::StorageError::Backend(BackendError::Internal { + backend_name: "mongodb".to_string(), + message: format!("SearchParameter cursor advance: {}", e), + source: None, + }) + })? { + let document = match cursor.deserialize_current() { + Ok(document) => document, + Err(e) => { + tracing::warn!("Failed to read SearchParameter document: {}", e); + continue; + } + }; + let Ok(payload) = document.get_document("data") else { + tracing::warn!("Stored SearchParameter document has no data payload"); + continue; + }; + let json = match super::storage::document_to_value(payload) { + Ok(json) => json, + Err(e) => { + tracing::warn!("Failed to convert SearchParameter payload: {}", e); + continue; + } + }; + match loader.parse_resource(&json) { + Ok(mut def) => { + if def.status == SearchParameterStatus::Active { + def.source = SearchParameterSource::Stored; + definitions.push(def); + } + } + Err(e) => { + tracing::warn!("Failed to parse stored SearchParameter: {}", e); + } + } + } + + let mut registry = self.search_registry.write(); + registry.unregister_source(SearchParameterSource::Stored); + let mut count = 0; + for def in definitions { + match registry.register(def) { + Ok(()) => count += 1, + Err(e) => tracing::warn!("Stored SearchParameter not registered: {}", e), + } + } + Ok(count) } /// Creates a MongoDB client from backend configuration. diff --git a/crates/persistence/src/backends/mongodb/storage.rs b/crates/persistence/src/backends/mongodb/storage.rs index 00e8cfc63..3620c9d2f 100644 --- a/crates/persistence/src/backends/mongodb/storage.rs +++ b/crates/persistence/src/backends/mongodb/storage.rs @@ -75,7 +75,7 @@ fn value_to_document(value: &Value) -> StorageResult { } } -fn document_to_value(doc: &Document) -> StorageResult { +pub(super) fn document_to_value(doc: &Document) -> StorageResult { bson::from_bson::(Bson::Document(doc.clone())) .map_err(|e| serialization_error(format!("Failed to deserialize resource: {}", e))) } diff --git a/crates/persistence/src/backends/postgres/backend.rs b/crates/persistence/src/backends/postgres/backend.rs index 3c96380d6..8f49f7f3d 100644 --- a/crates/persistence/src/backends/postgres/backend.rs +++ b/crates/persistence/src/backends/postgres/backend.rs @@ -417,6 +417,16 @@ impl PostgresBackend { /// Loads SearchParameter resources stored in the database into the registry. async fn load_stored_search_parameters(&self) -> StorageResult { + self.refresh_stored_search_parameters().await + } + + /// Rebuilds the registry's `Stored` parameters from what the database + /// currently holds, returning how many are registered afterwards. + /// + /// TTL-cache refresh (#235): rows are fetched and parsed before the + /// (sync) write lock is taken, and on any read error the registry is left + /// untouched — stale-serve rather than losing search resolution. + pub async fn refresh_stored_search_parameters(&self) -> StorageResult { use crate::search::registry::{SearchParameterSource, SearchParameterStatus}; let client = self.get_client().await?; @@ -435,18 +445,14 @@ impl PostgresBackend { })?; let loader = SearchParameterLoader::new(self.config.fhir_version); - let mut registry = self.search_registry.write(); - let mut count = 0; - + let mut definitions = Vec::new(); for row in rows { let data: serde_json::Value = row.get(0); match loader.parse_resource(&data) { Ok(mut def) => { if def.status == SearchParameterStatus::Active { def.source = SearchParameterSource::Stored; - if registry.register(def).is_ok() { - count += 1; - } + definitions.push(def); } } Err(e) => { @@ -455,6 +461,15 @@ impl PostgresBackend { } } + let mut registry = self.search_registry.write(); + registry.unregister_source(SearchParameterSource::Stored); + let mut count = 0; + for def in definitions { + match registry.register(def) { + Ok(()) => count += 1, + Err(e) => tracing::warn!("Stored SearchParameter not registered: {}", e), + } + } Ok(count) } diff --git a/crates/persistence/src/backends/sqlite/backend.rs b/crates/persistence/src/backends/sqlite/backend.rs index ece4e74b4..afc26c2c0 100644 --- a/crates/persistence/src/backends/sqlite/backend.rs +++ b/crates/persistence/src/backends/sqlite/backend.rs @@ -338,6 +338,19 @@ impl SqliteBackend { /// This is called during schema initialization to restore any custom /// SearchParameters that were POSTed to the server. fn load_stored_search_parameters(&self) -> StorageResult { + self.refresh_stored_search_parameters() + } + + /// Rebuilds the registry's `Stored` parameters from what the database + /// currently holds, returning how many are registered afterwards. + /// + /// This is the TTL-cache refresh (#235): storage is the source of truth, + /// so a SearchParameter POSTed to a cluster-mate appears here on the next + /// pass. All rows are read and parsed **before** the write lock is taken + /// (the lock is sync and shared with request paths), and on any read + /// error the registry is left untouched — a node serves its stale cache + /// rather than losing search resolution. + pub fn refresh_stored_search_parameters(&self) -> StorageResult { use crate::search::registry::{SearchParameterSource, SearchParameterStatus}; let conn = self.get_connection()?; @@ -353,10 +366,6 @@ impl SqliteBackend { }) })?; - let loader = SearchParameterLoader::new(self.config.fhir_version); - let mut registry = self.search_registry.write(); - let mut count = 0; - let rows = stmt .query_map([], |row| row.get::<_, Vec>(0)) .map_err(|e| { @@ -367,6 +376,8 @@ impl SqliteBackend { }) })?; + let loader = SearchParameterLoader::new(self.config.fhir_version); + let mut definitions = Vec::new(); for row in rows { let data = match row { Ok(data) => data, @@ -375,7 +386,6 @@ impl SqliteBackend { continue; } }; - let json: serde_json::Value = match serde_json::from_slice(&data) { Ok(json) => json, Err(e) => { @@ -383,15 +393,12 @@ impl SqliteBackend { continue; } }; - match loader.parse_resource(&json) { Ok(mut def) => { // Only register active parameters if def.status == SearchParameterStatus::Active { def.source = SearchParameterSource::Stored; - if registry.register(def).is_ok() { - count += 1; - } + definitions.push(def); } } Err(e) => { @@ -400,6 +407,17 @@ impl SqliteBackend { } } + // Synchronous rebuild under the existing lock; the Arc identity must + // be preserved (extractor/ES/chain builders hold clones). + let mut registry = self.search_registry.write(); + registry.unregister_source(SearchParameterSource::Stored); + let mut count = 0; + for def in definitions { + match registry.register(def) { + Ok(()) => count += 1, + Err(e) => tracing::warn!("Stored SearchParameter not registered: {}", e), + } + } Ok(count) } diff --git a/crates/persistence/src/search/mod.rs b/crates/persistence/src/search/mod.rs index 97da51468..fdd5635bc 100644 --- a/crates/persistence/src/search/mod.rs +++ b/crates/persistence/src/search/mod.rs @@ -70,6 +70,7 @@ pub mod loader; pub mod range; pub mod registry; pub mod reindex; +pub mod seeder; pub mod text_fold; pub mod writer; @@ -89,5 +90,6 @@ pub use reindex::{ ReindexOperation, ReindexProgress, ReindexRequest, ReindexSource, ReindexStatus, ReindexTarget, ReindexableStorage, ResourcePage, }; +pub use seeder::{SeedOutcome, seed_spec_search_parameters}; pub use text_fold::fold_text; pub use writer::SearchIndexWriter; diff --git a/crates/persistence/src/search/seeder.rs b/crates/persistence/src/search/seeder.rs new file mode 100644 index 000000000..015e1201e --- /dev/null +++ b/crates/persistence/src/search/seeder.rs @@ -0,0 +1,143 @@ +//! Seeds storage with the FHIR spec's SearchParameter resources (#235). +//! +//! Storage — not the in-memory registry — is the source of truth for +//! SearchParameters. On startup each primary backend seeds its store from the +//! same spec bundle the registry loads, so `GET /SearchParameter` discovers +//! the parameters the server actually resolves searches against, and any node +//! in a cluster boots into the same set. +//! +//! Seeding is idempotent and safe under concurrent multi-node boots: every +//! spec resource carries its bundle id (`Patient-name`, `Resource-id`, …), so +//! a second writer's `create` fails with `AlreadyExists` and is treated as +//! "already seeded". Existing resources are never updated or clobbered. +//! +//! Resources are seeded verbatim — including the spec's `status: draft`, which +//! the registry deliberately promotes to active only when loading (see +//! `SearchParameterLoader::load_from_spec_file`). The registry keeps loading +//! spec definitions from the bundled file as `Embedded`; the stored copies +//! exist for API discovery and cluster-wide consistency, not as a second +//! registration path — the stored-parameter refresh skips them (draft, and +//! their canonical URLs are already registered). + +use std::path::Path; + +use helios_fhir::FhirVersion; +use serde_json::Value; + +use crate::core::ResourceStorage; +use crate::error::{ResourceError, StorageError, StorageResult}; +use crate::search::loader::SearchParameterLoader; +use crate::tenant::{TenantContext, TenantId, TenantPermissions}; + +/// Outcome of a seeding pass. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct SeedOutcome { + /// Resources newly written by this pass. + pub created: usize, + /// Resources already present (same id), left untouched. + pub existing: usize, + /// Resources that failed to write and were skipped (logged). + pub failed: usize, +} + +/// Seeds `storage` with the spec SearchParameter bundle for `fhir_version`, +/// plus the embedded fallback parameters, under `tenant_id`. +/// +/// The tenant is the server's default tenant: searches are tenant-scoped, so +/// seeding anywhere else would leave `GET /SearchParameter` empty for the +/// common single-tenant deployment. Non-default tenants do not see the seeded +/// resources via the API (the in-memory registry still resolves searches for +/// every tenant); revisit if shared-resource search lands. +/// +/// Fast path: when the tenant already holds at least as many SearchParameters +/// as the spec set, the pass is skipped entirely — one `count` per boot. A +/// partial set (interrupted seed, or user-POSTed parameters predating this +/// feature) is completed resource-by-resource, skipping whatever exists. +pub async fn seed_spec_search_parameters( + storage: &S, + fhir_version: FhirVersion, + data_dir: &Path, + tenant_id: &str, +) -> StorageResult +where + S: ResourceStorage + ?Sized, +{ + let tenant = TenantContext::new(TenantId::new(tenant_id), TenantPermissions::full_access()); + let loader = SearchParameterLoader::new(fhir_version); + + let mut resources: Vec = match loader.load_spec_resources(data_dir) { + Ok(resources) => resources, + Err(e) => { + // No spec file is a supported minimal deployment (the registry + // falls back the same way); seed only the embedded fallbacks. + tracing::warn!("SearchParameter seeding: no spec bundle loaded: {e}"); + Vec::new() + } + }; + if let Ok(fallbacks) = loader.load_embedded() { + resources.extend( + fallbacks + .iter() + .map(SearchParameterLoader::definition_to_fhir_resource), + ); + } + // Drop entries whose id duplicates one already in the set. Several embedded + // fallbacks (`Resource-id`, `Library-url`, …) share an id with a spec + // bundle entry, so their `create` always fails `AlreadyExists` and never + // adds a row. Left in, they inflate `resources.len()` above the count the + // store can ever reach, so `present >= resources.len()` below would never + // hold and every boot would re-run the full create scan instead of taking + // the single-`count` fast path. + let mut seen_ids = std::collections::HashSet::new(); + resources.retain(|resource| { + resource + .get("id") + .and_then(|id| id.as_str()) + .is_none_or(|id| seen_ids.insert(id.to_string())) + }); + if resources.is_empty() { + return Ok(SeedOutcome { + created: 0, + existing: 0, + failed: 0, + }); + } + + let present = storage.count(&tenant, Some("SearchParameter")).await?; + if present as usize >= resources.len() { + return Ok(SeedOutcome { + created: 0, + existing: resources.len(), + failed: 0, + }); + } + + let mut outcome = SeedOutcome { + created: 0, + existing: 0, + failed: 0, + }; + for resource in resources { + match storage + .create(&tenant, "SearchParameter", resource, fhir_version) + .await + { + Ok(_) => outcome.created += 1, + Err(StorageError::Resource(ResourceError::AlreadyExists { .. })) => { + outcome.existing += 1; + } + Err(e) => { + outcome.failed += 1; + tracing::warn!("SearchParameter seeding: create failed: {e}"); + } + } + } + tracing::info!( + created = outcome.created, + existing = outcome.existing, + failed = outcome.failed, + tenant = %tenant_id, + "Seeded spec SearchParameters into storage" + ); + Ok(outcome) +} diff --git a/crates/persistence/tests/mongodb_tests.rs b/crates/persistence/tests/mongodb_tests.rs index 50f9efb70..22225cbee 100644 --- a/crates/persistence/tests/mongodb_tests.rs +++ b/crates/persistence/tests/mongodb_tests.rs @@ -2455,6 +2455,92 @@ async fn mongodb_integration_search_parameter_create_registers_active() { assert_eq!(param.status, SearchParameterStatus::Active); } +/// The TTL-cache refresh (#235) rebuilds the registry's stored parameters from +/// what the database currently holds: a parameter written by a cluster-mate +/// enters resolution on the next refresh, and a deleted one leaves it. This is +/// the SQLite integration test's contract exercised over the Mongo cursor path. +#[tokio::test] +async fn mongodb_integration_refresh_rebuilds_stored_parameters() { + use helios_persistence::search::registry::SearchParameterSource; + + let Some(backend) = create_backend("search_param_refresh").await else { + eprintln!( + "Skipping mongodb_integration_refresh_rebuilds_stored_parameters (requires Docker or HFS_TEST_MONGODB_URL)" + ); + return; + }; + + let tenant = create_tenant("tenant-search-param-refresh"); + + backend + .create( + &tenant, + "SearchParameter", + json!({ + "resourceType": "SearchParameter", + "id": "mongo-refresh-nickname", + "url": "http://example.org/fhir/SearchParameter/mongo-refresh-nickname", + "name": "MongoRefreshNickname", + "status": "active", + "code": "mongo-refresh-nickname", + "base": ["Patient"], + "type": "string", + "expression": "Patient.name.where(use='nickname').given" + }), + FhirVersion::default(), + ) + .await + .unwrap(); + + // Simulate a freshly booted cluster-mate: storage holds the parameter but + // its registry has never seen it (drop the write-hook registration). + backend + .search_registry() + .write() + .unregister_source(SearchParameterSource::Stored); + assert!( + backend + .search_registry() + .read() + .get_param("Patient", "mongo-refresh-nickname") + .is_none(), + "not visible before the refresh" + ); + + let stored = backend + .refresh_stored_search_parameters() + .await + .expect("refresh from storage"); + assert_eq!(stored, 1); + assert!( + backend + .search_registry() + .read() + .get_param("Patient", "mongo-refresh-nickname") + .is_some(), + "visible after the refresh" + ); + + // Delete it from storage; the next refresh drops it from resolution. + backend + .delete(&tenant, "SearchParameter", "mongo-refresh-nickname") + .await + .unwrap(); + let stored = backend + .refresh_stored_search_parameters() + .await + .expect("refresh after delete"); + assert_eq!(stored, 0); + assert!( + backend + .search_registry() + .read() + .get_param("Patient", "mongo-refresh-nickname") + .is_none(), + "gone after the refresh" + ); +} + #[tokio::test] async fn mongodb_integration_search_parameter_create_draft_not_registered() { let Some(backend) = create_backend("search_param_create_draft").await else { diff --git a/crates/persistence/tests/search_param_seeding.rs b/crates/persistence/tests/search_param_seeding.rs new file mode 100644 index 000000000..c765ce9ce --- /dev/null +++ b/crates/persistence/tests/search_param_seeding.rs @@ -0,0 +1,253 @@ +//! Integration tests for storage-backed SearchParameters (#235): seeding the +//! store from the spec bundle, and refreshing the in-memory registry from +//! storage so cluster-mates' writes become visible. + +#![cfg(feature = "sqlite")] + +use std::path::PathBuf; + +use helios_fhir::FhirVersion; +use helios_persistence::backends::sqlite::{SqliteBackend, SqliteBackendConfig}; +use helios_persistence::core::ResourceStorage; +use helios_persistence::search::seed_spec_search_parameters; +use helios_persistence::tenant::{TenantContext, TenantId, TenantPermissions}; +use serde_json::json; + +/// The workspace data directory holding `search-parameters-r4.json`. +fn workspace_data_dir() -> PathBuf { + PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../../data") +} + +fn create_backend() -> SqliteBackend { + let backend = SqliteBackend::with_config( + ":memory:", + SqliteBackendConfig { + data_dir: Some(workspace_data_dir()), + ..Default::default() + }, + ) + .expect("create in-memory SQLite backend"); + backend.init_schema().expect("init schema"); + backend +} + +fn tenant(id: &str) -> TenantContext { + TenantContext::new(TenantId::new(id), TenantPermissions::full_access()) +} + +#[tokio::test] +async fn seeding_is_idempotent_and_discoverable() { + let backend = create_backend(); + let data_dir = workspace_data_dir(); + + let first = seed_spec_search_parameters(&backend, FhirVersion::R4, &data_dir, "default") + .await + .expect("first seed"); + assert!( + first.created > 1300, + "expected the R4 spec set to seed, created only {}", + first.created + ); + assert_eq!(first.failed, 0); + + // The point of the issue: the spec parameters are now discoverable via a + // storage read (what GET /SearchParameter executes) in the default tenant. + let stored = backend + .count(&tenant("default"), Some("SearchParameter")) + .await + .expect("count seeded"); + assert_eq!(stored as usize, first.created); + + // A second boot takes the fast path and writes nothing. + let second = seed_spec_search_parameters(&backend, FhirVersion::R4, &data_dir, "default") + .await + .expect("second seed"); + assert_eq!(second.created, 0); + assert_eq!(second.failed, 0); + + // Other tenants are unaffected (seeding is default-tenant only). + let other = backend + .count(&tenant("acme"), Some("SearchParameter")) + .await + .expect("count other tenant"); + assert_eq!(other, 0); +} + +/// A partial set (e.g. a user-POSTed parameter predating the seeding feature) +/// is completed without clobbering what exists. +#[tokio::test] +async fn seeding_completes_a_partial_set_without_clobbering() { + let backend = create_backend(); + let data_dir = workspace_data_dir(); + + let custom = json!({ + "resourceType": "SearchParameter", + "id": "acme-preexisting", + "url": "http://acme.health/fhir/SearchParameter/preexisting", + "name": "preexisting", + "status": "active", + "code": "preexisting", + "base": ["Patient"], + "type": "string", + "expression": "Patient.name.given" + }); + backend + .create( + &tenant("default"), + "SearchParameter", + custom, + FhirVersion::R4, + ) + .await + .expect("pre-existing custom parameter"); + + let outcome = seed_spec_search_parameters(&backend, FhirVersion::R4, &data_dir, "default") + .await + .expect("seed over partial set"); + assert!(outcome.created > 1300); + assert_eq!(outcome.failed, 0); + + // The custom parameter survived untouched. + let read = backend + .read(&tenant("default"), "SearchParameter", "acme-preexisting") + .await + .expect("read custom") + .expect("custom parameter still present"); + assert_eq!( + read.content()["url"], + json!("http://acme.health/fhir/SearchParameter/preexisting") + ); +} + +/// A resource already stored under a spec bundle id is reported `existing`, +/// not re-created or clobbered — the idempotency the seeder relies on. +#[tokio::test] +async fn seeding_reports_existing_for_present_spec_ids() { + let backend = create_backend(); + let data_dir = workspace_data_dir(); + + // Pre-create a resource under a spec bundle id (`Patient-name`) but with a + // distinct body, so the seed pass hits the `AlreadyExists` branch for it. + let preexisting = json!({ + "resourceType": "SearchParameter", + "id": "Patient-name", + "url": "http://acme.health/fhir/SearchParameter/preexisting-patient-name", + "name": "preexisting", + "status": "active", + "code": "name", + "base": ["Patient"], + "type": "string", + "expression": "Patient.name" + }); + backend + .create( + &tenant("default"), + "SearchParameter", + preexisting, + FhirVersion::R4, + ) + .await + .expect("pre-create under a spec id"); + + let outcome = seed_spec_search_parameters(&backend, FhirVersion::R4, &data_dir, "default") + .await + .expect("seed over a present spec id"); + assert!( + outcome.existing >= 1, + "the pre-existing spec id should be reported existing, got {outcome:?}" + ); + assert_eq!(outcome.failed, 0); + + // The pre-existing resource was left untouched (create never clobbers). + let read = backend + .read(&tenant("default"), "SearchParameter", "Patient-name") + .await + .expect("read pre-existing") + .expect("still present"); + assert_eq!( + read.content()["url"], + json!("http://acme.health/fhir/SearchParameter/preexisting-patient-name") + ); +} + +/// With no spec bundle in the data directory, seeding still writes the embedded +/// fallback parameters (a supported minimal deployment). +#[tokio::test] +async fn seeding_with_missing_spec_bundle_seeds_only_fallbacks() { + let backend = create_backend(); + let empty_dir = tempfile::tempdir().expect("temp data dir"); + + let outcome = + seed_spec_search_parameters(&backend, FhirVersion::R4, empty_dir.path(), "default") + .await + .expect("seed fallbacks"); + // Only the handful of embedded fallbacks are seeded — no spec set. + assert!( + outcome.created > 0 && outcome.created <= 9, + "expected only embedded fallbacks, got {outcome:?}" + ); + assert_eq!(outcome.failed, 0); +} + +/// The TTL-cache contract: a SearchParameter present in storage (written by +/// this node or a cluster-mate) enters search resolution on a refresh, and a +/// deleted one leaves it. +#[tokio::test] +async fn refresh_rebuilds_stored_parameters_from_storage() { + use helios_persistence::search::registry::SearchParameterSource; + + let backend = create_backend(); + let registry = backend.search_registry(); + + let nickname = json!({ + "resourceType": "SearchParameter", + "id": "acme-nickname", + "url": "http://acme.health/fhir/SearchParameter/patient-nickname", + "name": "nickname", + "status": "active", + "code": "nickname", + "base": ["Patient"], + "type": "string", + "expression": "Patient.name.where(use='nickname').given" + }); + backend + .create( + &tenant("default"), + "SearchParameter", + nickname, + FhirVersion::R4, + ) + .await + .expect("store the parameter"); + + // Simulate a freshly booted cluster-mate: its storage holds the parameter + // but its registry has never seen it (drop the write-hook registration). + registry + .write() + .unregister_source(SearchParameterSource::Stored); + assert!( + registry.read().get_param("Patient", "nickname").is_none(), + "not visible before the refresh" + ); + + let stored = backend.refresh_stored_search_parameters().expect("refresh"); + assert_eq!(stored, 1); + assert!( + registry.read().get_param("Patient", "nickname").is_some(), + "visible after the refresh" + ); + + // Delete it from storage; the next refresh drops it from resolution. + backend + .delete(&tenant("default"), "SearchParameter", "acme-nickname") + .await + .expect("delete the parameter"); + let stored = backend + .refresh_stored_search_parameters() + .expect("refresh after delete"); + assert_eq!(stored, 0); + assert!( + registry.read().get_param("Patient", "nickname").is_none(), + "gone after the refresh" + ); +} diff --git a/crates/rest/src/config.rs b/crates/rest/src/config.rs index 1fca49a6f..8e4e94e6f 100644 --- a/crates/rest/src/config.rs +++ b/crates/rest/src/config.rs @@ -726,6 +726,14 @@ pub struct ServerConfig { #[arg(long, env = "HFS_DATA_DIR")] pub data_dir: Option, + /// Seconds between refreshes of the in-memory SearchParameter registry + /// from storage (#235). In a cluster, a SearchParameter POSTed to one + /// node becomes visible to the others within this interval: deliberate + /// eventual consistency, trading propagation latency for not hitting the + /// database on every search. `0` disables the periodic refresh. + #[arg(long, env = "HFS_SEARCH_PARAM_CACHE_TTL", default_value = "3600")] + pub search_param_cache_ttl: u64, + /// Default page size for search results. #[arg(long, env = "HFS_DEFAULT_PAGE_SIZE", default_value = "20")] pub default_page_size: usize, @@ -891,6 +899,7 @@ impl Default for ServerConfig { require_if_match: false, default_fhir_version: FhirVersion::default_enabled(), data_dir: None, + search_param_cache_ttl: 3600, default_page_size: 20, max_page_size: 1000, storage_backend: "sqlite".to_string(), @@ -1011,6 +1020,7 @@ impl ServerConfig { require_if_match: false, default_fhir_version: FhirVersion::default_enabled(), data_dir: None, + search_param_cache_ttl: 3600, default_page_size: 10, max_page_size: 100, storage_backend: "sqlite".to_string(),