From e6e2bb65dc6e3eec69d3f1e18dd1eb9ba4731b2e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mauricio=20Pe=C3=B1aranda?= Date: Mon, 14 Sep 2026 09:41:11 -0400 Subject: [PATCH 1/5] fix(elasticsearch): raise the nested-object limit so large resources stay searchable (#1050) Elasticsearch rejects a whole document once it holds more nested objects than `index.mapping.nested_objects.limit`, summed across every nested search-parameter field. Its default of 10000 is below real data: 458 of the 11,704 Synthea Provenance resources carry more than 10000 `target` references (the largest, 28,192), so they were stored in the primary but never searchable, and `$reindex` failed on them the same way. Add `ElasticsearchConfig::nested_objects_limit` (default 50000) and `HFS_ELASTICSEARCH_NESTED_OBJECTS_LIMIT`. The limit is written into the index template for new indices, and `Backend::initialize` raises it on existing `{prefix}_*` indices that are below it. The setting is dynamic, so resources that already indexed need no reindex. The startup pass only ever raises, sends index names in chunks, and is non-fatal, so a missing `manage` privilege cannot stop the server. Tests: the mapping carries the setting; the startup parser picks exactly the indices below the target; config and serde defaults; and, against a real Elasticsearch, a 12,000-target Provenance indexes through the `$reindex` page writer, while an index created at 10000 rejects it until a backend started with the raised limit raises that existing index. --- .claude/skills/run-hfs-server/SKILL.md | 1 + README.md | 1 + .../configuration/environment-variables.md | 1 + crates/hfs/README.md | 1 + crates/hfs/src/main.rs | 4 + crates/persistence/README.md | 17 ++ .../src/backends/elasticsearch/backend.rs | 41 ++++- .../src/backends/elasticsearch/schema.rs | 162 +++++++++++++++++- .../persistence/tests/elasticsearch_tests.rs | 149 ++++++++++++++++ crates/rest/src/config.rs | 19 ++ 10 files changed, 394 insertions(+), 2 deletions(-) diff --git a/.claude/skills/run-hfs-server/SKILL.md b/.claude/skills/run-hfs-server/SKILL.md index ce97cca3c0..f8078c1c14 100644 --- a/.claude/skills/run-hfs-server/SKILL.md +++ b/.claude/skills/run-hfs-server/SKILL.md @@ -75,6 +75,7 @@ HFS_SERVER_PORT=3000 HFS_LOG_LEVEL=debug cargo run --bin hfs | `HFS_ELASTICSEARCH_PASSWORD` | none | Elasticsearch basic auth password | | `HFS_ELASTICSEARCH_REFRESH_INTERVAL` | `1s` | Index `refresh_interval` applied when an index is created (`-1` disables periodic refresh) | | `HFS_ELASTICSEARCH_WRITE_REFRESH` | `false` | `refresh` parameter on index/delete writes: `false`, `wait_for`, or `true` | +| `HFS_ELASTICSEARCH_NESTED_OBJECTS_LIMIT` | `50000` | Index `mapping.nested_objects.limit`: max nested objects per document across all nested search-parameter fields. Set on new indices; raised at startup on existing indices below it | | `HFS_COMPOSITE_SYNC_MODE` | `asynchronous` | ES-backed composite write sync mode: asynchronous, synchronous, or hybrid | Use `HFS_COMPOSITE_SYNC_MODE=synchronous` **and** `HFS_ELASTICSEARCH_WRITE_REFRESH=wait_for` when callers need read-your-write search semantics, such as integration tests or bulk loads that immediately search. Either alone still leaves a window: synchronous mode only guarantees the document reached Elasticsearch, and it is not searchable until the next index refresh. See `crates/persistence/README.md` (Search visibility on Elasticsearch-backed composites). diff --git a/README.md b/README.md index 497ff812b6..0d3d0e9d52 100644 --- a/README.md +++ b/README.md @@ -301,6 +301,7 @@ compressed when the client sends `Accept-Encoding`. | `HFS_ELASTICSEARCH_PASSWORD` | *(none)* | ES basic auth password | | `HFS_ELASTICSEARCH_REFRESH_INTERVAL` | `1s` | Index `refresh_interval` applied when an index is created (`-1` disables periodic refresh) | | `HFS_ELASTICSEARCH_WRITE_REFRESH` | `false` | `refresh` parameter on index/delete writes: `false`, `wait_for`, or `true` | +| `HFS_ELASTICSEARCH_NESTED_OBJECTS_LIMIT` | `50000` | Index `mapping.nested_objects.limit`: max nested objects per document across all nested search-parameter fields. Set on new indices; raised at startup on existing indices below it | **PostgreSQL** (used to assemble a connection when `HFS_DATABASE_URL` is not set) diff --git a/book/src/configuration/environment-variables.md b/book/src/configuration/environment-variables.md index 3c11f5d3b7..6c2c70bfeb 100644 --- a/book/src/configuration/environment-variables.md +++ b/book/src/configuration/environment-variables.md @@ -57,6 +57,7 @@ are compressed when the client sends `Accept-Encoding`. | `HFS_ELASTICSEARCH_PASSWORD` | *(none)* | Elasticsearch basic auth password | | `HFS_ELASTICSEARCH_REFRESH_INTERVAL` | `1s` | Index `refresh_interval` applied when an index is created (`-1` disables periodic refresh) | | `HFS_ELASTICSEARCH_WRITE_REFRESH` | `false` | `refresh` parameter on index/delete writes: `false`, `wait_for`, or `true` | +| `HFS_ELASTICSEARCH_NESTED_OBJECTS_LIMIT` | `50000` | Index `mapping.nested_objects.limit`: max nested objects per document across all nested search-parameter fields. Set on new indices; raised at startup on existing indices below it | | `HFS_S3_BUCKET` | `hfs` | S3 bucket name | | `HFS_S3_REGION` | *(AWS chain)* | AWS region override | | `HFS_S3_PREFIX` | *(none)* | Optional key prefix for all S3 object keys | diff --git a/crates/hfs/README.md b/crates/hfs/README.md index fc468bc0a3..5a30a99b74 100644 --- a/crates/hfs/README.md +++ b/crates/hfs/README.md @@ -99,6 +99,7 @@ Options: | `HFS_COMPOSITE_SYNC_MODE` | `asynchronous` | Composite-store write sync mode for ES-backed backends (`sqlite-elasticsearch`, `postgres-elasticsearch`, `mongodb-elasticsearch`, `s3-elasticsearch`). One of `asynchronous`, `synchronous`, `hybrid`. With `asynchronous` (default) the write returns as soon as the primary commits and the search backend is updated on a background worker — lowest latency, but a follow-up search can race the indexing. Use `synchronous` when callers need read-your-write semantics (e.g. integration tests, bulk-load flows that immediately search). Ignored when the storage backend has no search secondary. | | `HFS_ELASTICSEARCH_REFRESH_INTERVAL` | `1s` | Elasticsearch index `refresh_interval` for ES-backed backends. Controls how quickly indexed documents become searchable when no per-write refresh is requested. `-1` disables periodic refresh entirely. Applied when an index is created; indices that already exist keep their current setting. | | `HFS_ELASTICSEARCH_WRITE_REFRESH` | `false` | The `refresh` parameter applied to Elasticsearch index/delete operations. One of `false` (no per-write refresh), `wait_for` (block each write until the affected shards refresh), or `true` (force a refresh per write; expensive, low-volume deployments only). | +| `HFS_ELASTICSEARCH_NESTED_OBJECTS_LIMIT` | `50000` | Elasticsearch `index.mapping.nested_objects.limit`: the maximum nested objects a single document may contain, summed across every nested search-parameter field. Elasticsearch's own default of 10000 rejects larger documents, so a resource with very many indexed values (for example a Synthea `Provenance` with 13,554 `target` references) is stored but never searchable. Applied to new indices through the index template, and raised at startup on existing indices that are below it; the setting is dynamic, so no reindex is needed for resources that already indexed. | Read-after-write search on an ES-backed composite needs **both** `HFS_COMPOSITE_SYNC_MODE=synchronous` and diff --git a/crates/hfs/src/main.rs b/crates/hfs/src/main.rs index a58dc2f996..d11f689527 100644 --- a/crates/hfs/src/main.rs +++ b/crates/hfs/src/main.rs @@ -2191,6 +2191,7 @@ async fn start_sqlite_elasticsearch( fhir_version: config.default_fhir_version, refresh_interval: config.elasticsearch_refresh_interval.clone(), write_refresh: es_write_refresh_from_config(&config)?, + nested_objects_limit: config.elasticsearch_nested_objects_limit, ..Default::default() }; @@ -2485,6 +2486,7 @@ async fn start_postgres_elasticsearch( fhir_version: config.default_fhir_version, refresh_interval: config.elasticsearch_refresh_interval.clone(), write_refresh: es_write_refresh_from_config(&config)?, + nested_objects_limit: config.elasticsearch_nested_objects_limit, ..Default::default() }; @@ -2696,6 +2698,7 @@ async fn start_mongodb_elasticsearch( fhir_version: config.default_fhir_version, refresh_interval: config.elasticsearch_refresh_interval.clone(), write_refresh: es_write_refresh_from_config(&config)?, + nested_objects_limit: config.elasticsearch_nested_objects_limit, ..Default::default() }; @@ -3108,6 +3111,7 @@ async fn start_s3_elasticsearch( fhir_version: config.default_fhir_version, refresh_interval: config.elasticsearch_refresh_interval.clone(), write_refresh: es_write_refresh_from_config(&config)?, + nested_objects_limit: config.elasticsearch_nested_objects_limit, ..Default::default() }; diff --git a/crates/persistence/README.md b/crates/persistence/README.md index 7f85fda2a5..f977d45d6c 100644 --- a/crates/persistence/README.md +++ b/crates/persistence/README.md @@ -1407,6 +1407,23 @@ immediate follow-up search misses the write. The `hfs` binary exposes these as `HFS_ELASTICSEARCH_WRITE_REFRESH` (see the [hfs README](../hfs/README.md#environment-variables)). +#### Very large resources on Elasticsearch-backed composites + +Every indexed search-parameter value is a nested object in the resource's +Elasticsearch document, and Elasticsearch rejects the **whole document** once it +holds more than `index.mapping.nested_objects.limit` of them. The resource stays +stored in the primary and readable by id, but is absent from every search. +Elasticsearch's own default of 10000 is exceeded by real data: 458 of the 11,704 +Synthea `Provenance` resources carry more than 10000 `target` references (the +largest, 28,192). + +`ElasticsearchConfig::nested_objects_limit` (default 50000) is written into the +index template, so new indices get it, and is raised during backend +initialization on existing indices that are below it. The setting is dynamic, +so resources that already indexed need no reindex; resources rejected before the +raise are repaired with `POST /{type}/$reindex`. The `hfs` binary exposes it as +`HFS_ELASTICSEARCH_NESTED_OBJECTS_LIMIT`. + ### Cost-Based Optimization The cost estimator uses benchmark-derived costs to make routing decisions: diff --git a/crates/persistence/src/backends/elasticsearch/backend.rs b/crates/persistence/src/backends/elasticsearch/backend.rs index 8b71dd2088..6430a1dad2 100644 --- a/crates/persistence/src/backends/elasticsearch/backend.rs +++ b/crates/persistence/src/backends/elasticsearch/backend.rs @@ -120,6 +120,18 @@ pub struct ElasticsearchConfig { #[serde(default = "default_max_result_window")] pub max_result_window: u32, + /// Maximum nested objects one document may contain, summed across every + /// `nested` search-parameter field (default: 50000). + /// + /// Elasticsearch's own default of 10000 rejects the whole document, so a + /// resource with more indexed values than that — a Synthea `Provenance` + /// whose `target` array alone holds 13,554 references — is stored but + /// never searchable (#1050). The setting is dynamic: new indices take it + /// from the index template, existing ones are raised in + /// [`Backend::initialize`]. + #[serde(default = "default_nested_objects_limit")] + pub nested_objects_limit: u32, + /// Request timeout in milliseconds (default: 30000). #[serde(default = "default_request_timeout_ms")] pub request_timeout_ms: u64, @@ -158,6 +170,10 @@ fn default_max_result_window() -> u32 { 10000 } +fn default_nested_objects_limit() -> u32 { + 50_000 +} + fn default_request_timeout_ms() -> u64 { 30000 } @@ -172,6 +188,7 @@ impl Default for ElasticsearchConfig { refresh_interval: default_refresh_interval(), write_refresh: WriteRefreshPolicy::default(), max_result_window: default_max_result_window(), + nested_objects_limit: default_nested_objects_limit(), request_timeout_ms: default_request_timeout_ms(), auth: None, disable_certificate_validation: false, @@ -540,7 +557,29 @@ impl Backend for ElasticsearchBackend { backend_name: "elasticsearch".to_string(), message: format!("Failed to create index template: {}", e), source: None, - }) + })?; + + // The template only reaches indices created from now on. Indices that + // already exist keep Elasticsearch's 10000 nested-object limit until + // raised here (#1050). Not fatal: a missing `manage` privilege must not + // stop the server, and every resource under the old limit still + // indexes. + let limit = self.config().nested_objects_limit; + match super::schema::raise_nested_objects_limit(self).await { + Ok(0) => {} + Ok(raised) => tracing::info!( + indices = raised, + limit, + "raised the Elasticsearch nested-object limit on existing indices" + ), + Err(e) => tracing::warn!( + error = %e, + limit, + "could not raise the Elasticsearch nested-object limit on existing indices; \ + resources with more nested values than an index's current limit stay unsearchable" + ), + } + Ok(()) } async fn migrate(&self) -> Result<(), BackendError> { diff --git a/crates/persistence/src/backends/elasticsearch/schema.rs b/crates/persistence/src/backends/elasticsearch/schema.rs index 4eb4078179..8f88514b5c 100644 --- a/crates/persistence/src/backends/elasticsearch/schema.rs +++ b/crates/persistence/src/backends/elasticsearch/schema.rs @@ -3,7 +3,10 @@ //! Defines the index structure for FHIR resources in Elasticsearch. //! Uses nested objects for search parameters to ensure correct multi-value matching. -use elasticsearch::indices::{IndicesCreateParts, IndicesExistsParts, IndicesPutTemplateParts}; +use elasticsearch::indices::{ + IndicesCreateParts, IndicesExistsParts, IndicesGetSettingsParts, IndicesPutSettingsParts, + IndicesPutTemplateParts, +}; use serde_json::json; use crate::error::{BackendError, StorageResult}; @@ -24,6 +27,7 @@ pub fn create_index_mapping(config: &super::backend::ElasticsearchConfig) -> ser "number_of_shards": config.number_of_shards, "number_of_replicas": config.number_of_replicas, "index.max_result_window": config.max_result_window, + "index.mapping.nested_objects.limit": config.nested_objects_limit, "refresh_interval": config.refresh_interval, "analysis": { "normalizer": { @@ -259,6 +263,121 @@ pub async fn create_index_template(backend: &ElasticsearchBackend) -> StorageRes } /// Ensures an index exists for the given tenant and resource type, creating it if necessary. +/// The index setting capping how many nested objects one document may hold. +const NESTED_OBJECTS_LIMIT_SETTING: &str = "index.mapping.nested_objects.limit"; + +/// Raises the nested-object limit on every existing index under the configured +/// prefix whose current limit is below the configured one, and returns how many +/// indices were raised. +/// +/// New indices take the limit from the index template, but a template only +/// applies when an index is created. A deployment that indexed data before +/// #1050 keeps Elasticsearch's default of 10000 on those indices, and keeps +/// silently dropping large resources from search on the next write or +/// `$reindex`. The setting is dynamic, so it changes on a live index without +/// closing or reindexing it. +/// +/// The limit is only ever raised: an index an operator already set higher by +/// hand is left alone. +pub async fn raise_nested_objects_limit(backend: &ElasticsearchBackend) -> StorageResult { + /// Index names per update request, keeping the request URL short. + const INDICES_PER_REQUEST: usize = 50; + + let target = u64::from(backend.config().nested_objects_limit); + let pattern = format!("{}_*", backend.config().index_prefix); + + let response = backend + .client() + .indices() + .get_settings(IndicesGetSettingsParts::IndexName( + &[&pattern], + &[NESTED_OBJECTS_LIMIT_SETTING], + )) + .include_defaults(true) + .flat_settings(true) + .allow_no_indices(true) + .ignore_unavailable(true) + .send() + .await + .map_err(|e| settings_error(format!("Failed to read index settings: {e}")))?; + let status = response.status_code(); + let body: serde_json::Value = response + .json() + .await + .map_err(|e| settings_error(format!("Failed to parse index settings: {e}")))?; + if !status.is_success() { + return Err(settings_error(format!( + "Reading index settings failed (status {status}): {body}" + ))); + } + + let below = indices_below_nested_limit(&body, target); + for chunk in below.chunks(INDICES_PER_REQUEST) { + let names: Vec<&str> = chunk.iter().map(String::as_str).collect(); + let mut settings = serde_json::Map::new(); + settings.insert(NESTED_OBJECTS_LIMIT_SETTING.to_string(), json!(target)); + let response = backend + .client() + .indices() + .put_settings(IndicesPutSettingsParts::Index(&names)) + .body(serde_json::Value::Object(settings)) + .send() + .await + .map_err(|e| { + settings_error(format!( + "Failed to raise {NESTED_OBJECTS_LIMIT_SETTING}: {e}" + )) + })?; + let status = response.status_code(); + if !status.is_success() { + let text = response.text().await.unwrap_or_default(); + return Err(settings_error(format!( + "Raising {NESTED_OBJECTS_LIMIT_SETTING} failed (status {status}): {text}" + ))); + } + } + Ok(below.len()) +} + +/// The indices in a `GET _settings?include_defaults=true&flat_settings=true` +/// response whose nested-object limit is below `target`, sorted by name. +/// +/// An explicit per-index value wins over the cluster default. An index that +/// reports the setting in neither place is left alone rather than guessed at. +fn indices_below_nested_limit(body: &serde_json::Value, target: u64) -> Vec { + let Some(indices) = body.as_object() else { + return Vec::new(); + }; + let mut below: Vec = indices + .iter() + .filter_map(|(name, entry)| { + let value = entry + .get("settings") + .and_then(|s| s.get(NESTED_OBJECTS_LIMIT_SETTING)) + .or_else(|| { + entry + .get("defaults") + .and_then(|d| d.get(NESTED_OBJECTS_LIMIT_SETTING)) + })?; + let current = value + .as_str() + .and_then(|s| s.parse::().ok()) + .or_else(|| value.as_u64())?; + (current < target).then(|| name.clone()) + }) + .collect(); + below.sort(); + below +} + +fn settings_error(message: String) -> crate::error::StorageError { + crate::error::StorageError::Backend(BackendError::Internal { + backend_name: "elasticsearch".to_string(), + message, + source: None, + }) +} + pub async fn ensure_index( backend: &ElasticsearchBackend, tenant_id: &str, @@ -468,6 +587,12 @@ mod tests { // Verify settings assert_eq!(mapping["settings"]["number_of_shards"], 1); assert_eq!(mapping["settings"]["number_of_replicas"], 1); + // #1050: raised above Elasticsearch's 10000 default, which silently + // drops resources with many indexed values from search. + assert_eq!( + mapping["settings"]["index.mapping.nested_objects.limit"], + 50_000 + ); // Verify mappings exist let props = &mapping["mappings"]["properties"]; @@ -489,4 +614,39 @@ mod tests { // Verify normalizer assert!(mapping["settings"]["analysis"]["normalizer"]["lowercase_normalizer"].is_object()); } + + /// #1050: which existing indices the startup pass raises. An explicit + /// value beats the cluster default, an index already above the target is + /// never lowered, and an index that reports no value is not guessed at. + #[test] + fn test_indices_below_nested_limit() { + let body = json!({ + "hfs_t_provenance": { + "settings": {}, + "defaults": { "index.mapping.nested_objects.limit": "10000" } + }, + "hfs_t_observation": { + "settings": { "index.mapping.nested_objects.limit": "10000" }, + "defaults": {} + }, + "hfs_t_patient": { + "settings": { "index.mapping.nested_objects.limit": "80000" }, + "defaults": { "index.mapping.nested_objects.limit": "10000" } + }, + "hfs_t_encounter": { + "settings": { "index.mapping.nested_objects.limit": "50000" } + }, + "hfs_t_unknown": { "settings": {}, "defaults": {} } + }); + + assert_eq!( + indices_below_nested_limit(&body, 50_000), + vec![ + "hfs_t_observation".to_string(), + "hfs_t_provenance".to_string() + ] + ); + assert!(indices_below_nested_limit(&json!({}), 50_000).is_empty()); + assert!(indices_below_nested_limit(&json!("not an object"), 50_000).is_empty()); + } } diff --git a/crates/persistence/tests/elasticsearch_tests.rs b/crates/persistence/tests/elasticsearch_tests.rs index 1d348ac5a4..0d73ee5445 100644 --- a/crates/persistence/tests/elasticsearch_tests.rs +++ b/crates/persistence/tests/elasticsearch_tests.rs @@ -25,6 +25,7 @@ fn test_elasticsearch_config_defaults() { assert_eq!(config.number_of_shards, 1); assert_eq!(config.number_of_replicas, 1); assert!(config.auth.is_none()); + assert_eq!(config.nested_objects_limit, 50_000); } #[test] @@ -53,6 +54,17 @@ fn test_write_refresh_policy_default_is_false() { assert_eq!(config.write_refresh, WriteRefreshPolicy::False); } +/// #1050: a config that omits the field must not fall back to +/// Elasticsearch's own limit of 10000, which drops large resources from search. +#[test] +fn test_nested_objects_limit_defaults_above_elasticsearch_default() { + assert_eq!(ElasticsearchConfig::default().nested_objects_limit, 50_000); + + let json = r#"{"nodes": ["http://localhost:9200"]}"#; + let config: ElasticsearchConfig = serde_json::from_str(json).unwrap(); + assert_eq!(config.nested_objects_limit, 50_000); +} + #[test] fn test_write_refresh_policy_parsing() { assert_eq!( @@ -854,6 +866,60 @@ mod es_integration { TenantContext::new(TenantId::new(id), TenantPermissions::full_access()) } + /// A backend on `index_prefix` with the given nested-object limit, so the + /// #1050 tests can put two backends on the same indices. + async fn create_backend_with_nested_limit( + index_prefix: &str, + nested_objects_limit: u32, + ) -> ElasticsearchBackend { + let es = shared_es().await; + let config = ElasticsearchConfig { + nodes: vec![format!("http://{}:{}", es.host, es.port)], + index_prefix: index_prefix.to_string(), + number_of_replicas: 0, + refresh_interval: "1ms".to_string(), + nested_objects_limit, + ..Default::default() + }; + let backend = ElasticsearchBackend::with_shared_registry(config, build_search_registry()) + .expect("Failed to create ElasticsearchBackend"); + backend + .initialize() + .await + .expect("Failed to initialize ES backend"); + backend + } + + /// A Synthea-shaped `Provenance` whose `target` array alone holds more + /// nested reference values than Elasticsearch's default limit of 10000 — + /// the shape of the 458 resources #1050 found stored but unsearchable. + fn oversized_provenance( + tenant: &TenantContext, + id: &str, + targets: usize, + ) -> helios_persistence::types::StoredResource { + let target: Vec = (0..targets) + .map(|n| json!({ "reference": format!("Observation/obs-{n}") })) + .collect(); + helios_persistence::types::StoredResource::from_storage( + "Provenance", + id.to_string(), + "1", + tenant.tenant_id().clone(), + json!({ + "resourceType": "Provenance", + "id": id, + "target": target, + "recorded": "2009-02-28T07:56:45.469-05:00", + "agent": [{ "who": { "reference": "Practitioner/p1" } }] + }), + chrono::Utc::now(), + chrono::Utc::now(), + None, + FhirVersion::default(), + ) + } + /// #519: the #456 boundary table over the real ES search path. #[tokio::test] async fn es_day_precision_date_boundaries() { @@ -886,6 +952,89 @@ mod es_integration { assert_eq!(created.version_id(), "1"); } + /// #1050: Elasticsearch's default nested-object limit of 10000 rejects the + /// whole document, so a resource with more indexed values than that was + /// stored but never searchable. With the raised default it indexes through + /// the `$reindex` page writer. + #[tokio::test] + async fn es_integration_resource_over_elasticsearch_default_nested_limit_indexes() { + use helios_persistence::search::ReindexTarget; + + let backend = create_backend().await; + let tenant = create_tenant("nested-limit-default"); + let provenance = oversized_provenance(&tenant, "oversized", 12_000); + + let outcomes = backend + .write_search_entries_page(&tenant, std::slice::from_ref(&provenance)) + .await; + assert!( + outcomes[0].is_ok(), + "a Provenance with 12000 targets must index: {:?}", + outcomes[0] + ); + assert!( + backend + .read(&tenant, "Provenance", "oversized") + .await + .unwrap() + .is_some(), + "the document must be in the index" + ); + } + + /// #1050: the index template only reaches indices created after it, so an + /// index that already existed at Elasticsearch's limit of 10000 kept + /// rejecting large resources. Starting a backend with the raised limit must + /// raise it on that existing index. `ensure_index` never touches an index + /// that exists, so only the startup pass can explain the second write + /// succeeding. + #[tokio::test] + async fn es_integration_startup_raises_nested_limit_on_existing_index() { + use helios_persistence::search::ReindexTarget; + + let prefix = format!("hfs_{}", uuid::Uuid::new_v4().simple()); + let tenant = create_tenant("nested-limit-existing"); + let provenance = oversized_provenance(&tenant, "oversized", 12_000); + + // An index created under Elasticsearch's own limit rejects it. + let before = create_backend_with_nested_limit(&prefix, 10_000).await; + let rejected = before + .write_search_entries_page(&tenant, std::slice::from_ref(&provenance)) + .await; + let error = rejected[0] + .as_ref() + .expect_err("a limit of 10000 must reject 12000 nested objects"); + assert!( + error.to_string().contains("nested"), + "the rejection must be the nested-object limit, got: {error}" + ); + assert!( + before + .read(&tenant, "Provenance", "oversized") + .await + .unwrap() + .is_none() + ); + + // A backend started with the raised limit fixes the existing index. + let after = create_backend_with_nested_limit(&prefix, 50_000).await; + let outcomes = after + .write_search_entries_page(&tenant, std::slice::from_ref(&provenance)) + .await; + assert!( + outcomes[0].is_ok(), + "the raised index must accept it: {:?}", + outcomes[0] + ); + assert!( + after + .read(&tenant, "Provenance", "oversized") + .await + .unwrap() + .is_some() + ); + } + /// `$reindex` walks a page at a time and, before #1021, Elasticsearch used /// the default trait implementation: one HTTP round trip per resource. This /// pins the batched override — every resource of the page indexed, its own diff --git a/crates/rest/src/config.rs b/crates/rest/src/config.rs index e97a493153..504dd25ffd 100644 --- a/crates/rest/src/config.rs +++ b/crates/rest/src/config.rs @@ -1130,6 +1130,19 @@ pub struct ServerConfig { #[arg(long, env = "HFS_ELASTICSEARCH_WRITE_REFRESH", default_value = "false")] pub elasticsearch_write_refresh: String, + /// Maximum nested objects one Elasticsearch document may contain, summed + /// across every nested search-parameter field. Elasticsearch's default of + /// 10000 rejects larger documents outright, leaving those resources stored + /// but unsearchable (#1050). New indices take the value from the index + /// template; existing indices below it are raised at startup (the setting + /// is dynamic, so no reindex of already-indexed resources is needed). + #[arg( + long, + env = "HFS_ELASTICSEARCH_NESTED_OBJECTS_LIMIT", + default_value = "50000" + )] + pub elasticsearch_nested_objects_limit: u32, + /// Enable SQL-on-FHIR operations ($sql-run, $sql-export). /// When enabled, the configured storage backend MUST provide an in-DB /// SOF runner (sqlite or postgres) — there is no in-process fallback. @@ -1397,6 +1410,7 @@ impl Default for ServerConfig { elasticsearch_password: None, elasticsearch_refresh_interval: "1s".to_string(), elasticsearch_write_refresh: "false".to_string(), + elasticsearch_nested_objects_limit: 50_000, sof_enabled: true, ui_enabled: true, dashboard_reconcile_interval_secs: 30, @@ -1535,6 +1549,10 @@ impl ServerConfig { errors.push("Batch max concurrency cannot be 0".to_string()); } + if self.elasticsearch_nested_objects_limit == 0 { + errors.push("Elasticsearch nested objects limit cannot be 0".to_string()); + } + if self.dashboard_reconcile_interval_secs == 0 { errors.push("Dashboard reconcile interval cannot be 0".to_string()); } @@ -1640,6 +1658,7 @@ impl ServerConfig { elasticsearch_password: None, elasticsearch_refresh_interval: "1s".to_string(), elasticsearch_write_refresh: "false".to_string(), + elasticsearch_nested_objects_limit: 50_000, sof_enabled: true, ui_enabled: true, dashboard_reconcile_interval_secs: 30, From 6890dc000c1603eee29467a92cca59fc020414c3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mauricio=20Pe=C3=B1aranda?= Date: Mon, 14 Sep 2026 09:53:04 -0400 Subject: [PATCH 2/5] fix(reindex): report failing resources and stop retrying permanent rejections (#1050) A reindex that finished with per-resource errors exposed only `errorCount` in `$reindex-status`, so the 458 Provenance records Elasticsearch rejected in #1050 could only be found by reading server logs. And the automatic deferred reindex (#1087) retried any such completion once, which for a document the index rejects outright just repeats the same failure over the same resources. - Each recorded error now carries `retryable`: true for the conditions REST answers with 503/504 (Unavailable, ConnectionFailed, PoolExhausted, Timeout), false otherwise. Unclassified errors deserialize as retryable, keeping the old behaviour. - The Elasticsearch `_bulk` path classifies failures: a request that never arrived, a 429 or a 5xx is transient and surfaces from `write_search_entries_page` as Unavailable; a 4xx item rejection (the nested-object limit) stays Internal. `create_many` keeps its existing error mapping. - `$reindex-status` lists the first 100 failing resources as `error` parameters (resourceType, resourceId, message, retryable), plus `errorsOmitted` and, for a failed job, `errorMessage`. `errorCount` is unchanged. - An automatic generation that completes with only permanent errors is not retried; it logs the count, the first failing Type/ids and the first error. Any transient error, a failed job or a panic still gets the single retry. --- .agents/skills/bulk-data-submit/SKILL.md | 4 +- .claude/skills/bulk-data-submit/SKILL.md | 2 +- .../src/backends/elasticsearch/storage.rs | 71 +++- crates/persistence/src/search/reindex.rs | 384 ++++++++++++++++-- .../persistence/tests/elasticsearch_tests.rs | 10 + crates/rest/README.md | 8 + 6 files changed, 428 insertions(+), 51 deletions(-) diff --git a/.agents/skills/bulk-data-submit/SKILL.md b/.agents/skills/bulk-data-submit/SKILL.md index 8a8729fe15..1ac5665aca 100644 --- a/.agents/skills/bulk-data-submit/SKILL.md +++ b/.agents/skills/bulk-data-submit/SKILL.md @@ -112,12 +112,12 @@ The backend capability splits into `BulkSubmitIngest` (the synchronous `BulkSubm - With `HFS_BULK_SUBMIT_DEFER_INDEXING=true` (bulk fast-load, #903, the default since #946), ingestion skips search-index and FTS writes. The worker requests an automatic full-type reindex after the manifest becomes terminal, so `$bulk-submit-status` can answer `200` while search is still incomplete. - Automatic deferred reindex requests share the coordinator owned by their `ReindexOperation` (#1087). One tenant has one active generation plus one pending, deduplicated type set. The process runs at most `W` automatic generations and retains at most `2W` tenant entries, where `W` is the existing `HFS_BULK_SUBMIT_WORKER_CONCURRENCY` value. Admission applies backpressure before it adds another tenant. HFS has no separate bulk-submit reindex-concurrency variable. - The coordinator releases and reacquires its execution permit between generations so another admitted tenant can progress. Separate `ReindexOperation` instances remain independent because they can have different writer and registry sets. Explicit `$reindex` jobs bypass this coordinator and can overlap automatic work. -- A clean automatic generation ends `Completed` with no resource errors. Failure, panic, or an errorful completion gets one retry with the active and pending types. A second consecutive failure abandons that generation and logs the manual `$reindex` repair. Independently queued work that arrived during the retry still runs as a new generation with its own retry budget. Cancellation does not retry the cancelled active types, but independently queued pending types still run after the cancelled task has stopped writing. +- A clean automatic generation ends `Completed` with no resource errors. Failure, panic, or a completion with *transient* resource errors (a backend that was unavailable, timed out, or answered Elasticsearch `429`/`5xx`) gets one retry with the active and pending types. A completion whose resource errors are all *permanent* — a document the search backend rejects outright, such as one over Elasticsearch's nested-object limit (#1050) — is not retried, because a rerun fails identically; it logs the error count and the first failing `Type/id`s instead. A second consecutive failure abandons that generation and logs the manual `$reindex` repair. Independently queued work that arrived during the retry still runs as a new generation with its own retry budget. Cancellation does not retry the cancelled active types, but independently queued pending types still run after the cancelled task has stopped writing. - Coordination and reindex job state are in memory and local to one HFS process. A restart loses pending work, and separate processes do not coordinate. Full-type scans remain in use, so a finite burst can still cause one active scan and one accumulated follow-up. Limiting work to successful manifest IDs was evaluated and deferred because the generic path lacks bounded receipt deduplication, current-resource handling for missing or deleted IDs, and consistent semantics for every composite target. - The coordination logic is common to standalone SQLite, PostgreSQL, and MongoDB plus the Elasticsearch composites that wire reindex. Current performance evidence is PostgreSQL-only; do not claim equivalent latency or database-work improvements for the other backends without measuring them. See `docs/deferred-reindex-coordination-benchmark.md`. - MongoDB ingests a batch, not an entry (#1000): one `find` resolves which ids already exist, then one `insert`/`update` command per collection. The per-entry path it replaced cost ~9 round trips per resource and ran at ~60–76 resources/s with the server two-thirds idle; batched it reaches ~720, and ~3 100 with indexing deferred. The flush is ordered commands, not one transaction — the per-entry path was not atomic across a batch either, and a batch-wide transaction would widen #1001 from one lost entry to a whole batch. - Composite deployments (primary + Elasticsearch, including the `mongo-es`/`s3-es` modes) must wrap the primary's job store with `composite_submit_jobs(...)`: ingestion runs on the primary, whose own indexing is offloaded, so without the wrapper a completed import is readable by id and invisible to every search (#882, and #1021 for the modes that were missed). Guard: `crates/hfs/tests/bulk_submit/run_composite_es_index_check.sh`. -- On those composite deployments (#1007), the worker copies every manifest's ingested resources into the secondary search index *before* writing the manifest's receipt (not at `finish_manifest`, which no longer syncs — a manifest already terminal is never re-synced by a restart). A resource the secondary still rejects after its retries gets an entry result of `processing-error` in the receipt, with an OperationOutcome (`incomplete`) naming the `Type/id`, the rejecting backend, and `POST /{type}/$reindex` as the repair; the resource itself stays stored and readable by id, and `failed_entries` on the status counts it. +- On those composite deployments (#1007), the worker copies every manifest's ingested resources into the secondary search index *before* writing the manifest's receipt (not at `finish_manifest`, which no longer syncs — a manifest already terminal is never re-synced by a restart). A resource the secondary still rejects after its retries gets an entry result of `processing-error` in the receipt, with an OperationOutcome (`incomplete`) naming the `Type/id`, the rejecting backend, and `POST /{type}/$reindex` as the repair; the resource itself stays stored and readable by id, and `failed_entries` on the status counts it. If the rejection is Elasticsearch's nested-object limit (`The number of nested documents has exceeded the allowed limit`), `$reindex` fails the same way until `HFS_ELASTICSEARCH_NESTED_OBJECTS_LIMIT` (default 50000, raised on existing indices at startup) is above that resource's nested value count (#1050). Each failed resource is listed, with its error and whether it is retryable, in `GET /$reindex-status/{job_id}`. - After that copy, the worker compares the primary's tenant-wide count against each secondary's for every resource type the manifest ingested. A mismatch is recorded as a `warning` OperationOutcome (also `incomplete`, naming both counts) in the manifest's `error` artifact and logged. This only runs when `HFS_COMPOSITE_SYNC_MODE` is `synchronous` or `hybrid`; under the default `asynchronous` mode the secondary's count reflects whatever had already drained from its queue rather than this manifest's sync, so the check is skipped and the receipt then only guarantees the resources committed on the primary. - Without `HFS_ELASTICSEARCH_WRITE_REFRESH=wait_for`, a small count difference can be a write not yet visible rather than a real gap; reconfirm with `GET /{type}?_summary=count` before treating it as drift. See "Verifying and repairing search drift" below. - `HFS_BULK_SUBMIT_DEFER_INDEXING` is read once at startup, not per submission. Repair a stale or missed index with `$reindex`. diff --git a/.claude/skills/bulk-data-submit/SKILL.md b/.claude/skills/bulk-data-submit/SKILL.md index 7cb102ccce..4c7b938c0b 100644 --- a/.claude/skills/bulk-data-submit/SKILL.md +++ b/.claude/skills/bulk-data-submit/SKILL.md @@ -119,7 +119,7 @@ The backend capability splits into `BulkSubmitIngest` (the synchronous `BulkSubm - With `HFS_BULK_SUBMIT_DEFER_INDEXING=true` (bulk fast-load, #903 — **the default since #946**) ingestion skips the search-index and FTS writes and an automatic per-type reindex rebuilds them when each manifest finishes. Reads and history are complete throughout; search sees a manifest's resources once its reindex lands. That rebuild is started *after* the manifest is already terminal and is fire-and-forget (`bulk_submit_worker.rs` → `reindex.rs`, `tokio::spawn`), so `$bulk-submit-status` answers `200` while search is still incomplete, and the job lives only in an in-memory map — no column on `bulk_manifests` records that indexing is outstanding and nothing re-fires it at startup. A restart in that window is not recoverable on its own. - MongoDB ingests a batch, not an entry: one `find` resolves which of the batch's ids already exist, then one `insert` or `update` command per collection writes the whole batch (`backends/mongodb/bulk_ingest.rs`). Before #1000 each entry cost ~9 round trips of its own — a `read`, `create`'s second existence probe, the resource and history inserts, a search-index delete and insert, a transaction commit, the rollback record and the receipt — which pinned ingest at ~60–76 resources/s with `mongod` two-thirds idle. The batch flush is a sequence of commands rather than one transaction. Every command is retried on a transient driver error (`RetryableError`/`RetryableWriteError` label, I/O error, cleared pool — not a server-selection timeout) with 100 ms doubling backoff capped at 1 s over six attempts, checking the submission's cancel token before each sleep; a retry never duplicates what an earlier attempt landed (resources are re-read and matched on version + the batch's own `last_updated` + content, history and rollback rows dedupe on their unique keys, the search index is cleared before re-insert). When a stage outlives its retries the batch's entries get `processing-error` receipts with issue code `transient` and the file continues with its next batch, so `max_errors`/`continue_on_error` govern backend failures too (#1001); re-submitting the file converges. Only a receipt write that itself fails after retries still aborts the file. The manifest counters are a `$inc` and may over-count one batch if a retried attempt had actually landed — the receipts are authoritative. - **On a composite deployment (primary + Elasticsearch), the ingest engine does not reach the secondary by itself.** Ingestion runs on the *primary's* engine, and the primary deliberately skips its own indexing when search is offloaded — so `main.rs` wraps the primary's job store in `CompositeSubmitJobs`, which syncs each manifest's ingested resources into the secondary. Every composite mode must call `composite_submit_jobs(...)`; `mongo-es` and `s3-es` did not, and a completed import there was readable by id and invisible to every search — 15.27M of 15.28M resources on the reported deployment, with `GET` by id passing every smoke test (#1021). `crates/hfs/tests/bulk_submit/run_composite_es_index_check.sh` asserts the searchable count, not just readability, and is the guard against a fourth composite backend repeating it. -- The sync itself runs *before* the manifest's receipt is written (#1007), as an explicit worker step — not at `finish_manifest`, which no longer syncs by itself, so a manifest that already reached a terminal state is never re-synced by a restart; repair it with `$reindex`. A resource the secondary still rejects after its retries gets an entry result of `processing-error` in the receipt, carrying an OperationOutcome (`incomplete`) that names the `Type/id`, the rejecting backend, and `POST /{type}/$reindex` as the repair; the resource itself stays stored and readable by id, and the status's `failed_entries` counts it. +- The sync itself runs *before* the manifest's receipt is written (#1007), as an explicit worker step — not at `finish_manifest`, which no longer syncs by itself, so a manifest that already reached a terminal state is never re-synced by a restart; repair it with `$reindex`. A resource the secondary still rejects after its retries gets an entry result of `processing-error` in the receipt, carrying an OperationOutcome (`incomplete`) that names the `Type/id`, the rejecting backend, and `POST /{type}/$reindex` as the repair; the resource itself stays stored and readable by id, and the status's `failed_entries` counts it. If the rejection is Elasticsearch's nested-object limit (`The number of nested documents has exceeded the allowed limit`), `$reindex` fails the same way until `HFS_ELASTICSEARCH_NESTED_OBJECTS_LIMIT` (default 50000, raised on existing indices at startup) is above that resource's nested value count (#1050). Each failed resource is listed, with its error and whether it is retryable, in `GET /$reindex-status/{job_id}`. - After that copy, for every resource type the manifest ingested, the worker compares the primary's tenant-wide resource count against each secondary's. A mismatch is recorded as a `warning` OperationOutcome (also `incomplete`, naming both counts) in the manifest's `error` artifact and logged on the server. This check only runs when `HFS_COMPOSITE_SYNC_MODE` is `synchronous` or `hybrid`; under the default `asynchronous` mode the secondary's count reflects whatever had already drained from its queue rather than this manifest's own sync, so the check is skipped and the receipt then guarantees only that the resources committed on the primary. - Without `HFS_ELASTICSEARCH_WRITE_REFRESH=wait_for`, a small count difference can be a write that has not become visible yet rather than a real gap; reconfirm with `GET /{type}?_summary=count` before treating it as drift. See "Verifying and repairing search drift" below. - **Under deferred indexing the rebuild is the import.** On SQLite the fast-load diff --git a/crates/persistence/src/backends/elasticsearch/storage.rs b/crates/persistence/src/backends/elasticsearch/storage.rs index 50644f0dcf..d087fb5f42 100644 --- a/crates/persistence/src/backends/elasticsearch/storage.rs +++ b/crates/persistence/src/backends/elasticsearch/storage.rs @@ -29,6 +29,22 @@ use super::schema; /// making a load pay one refresh wait per handful of documents. const BULK_OPS_PER_REQUEST: usize = 500; +/// Why a resource's documents did not index in a `_bulk` request. +struct BulkFailure { + message: String, + /// The cluster never judged the document — the request did not arrive, or + /// the cluster pushed back (`429`) or failed (`5xx`) — so a rerun may + /// succeed. A `4xx` rejection of the document itself (for example the + /// nested-object limit, #1050) is permanent. + transient: bool, +} + +/// Whether a `_bulk` request or item status asks for a retry rather than +/// rejecting the document. +fn is_transient_bulk_status(status: u64) -> bool { + status == 429 || (500..600).contains(&status) +} + fn internal_error(message: String) -> StorageError { StorageError::Backend(BackendError::Internal { backend_name: "elasticsearch".to_string(), @@ -732,7 +748,7 @@ impl ResourceStorage for ElasticsearchBackend { .into_iter() .zip(failures) .map(|(p, failure)| match failure { - Some(message) => Err(internal_error(message)), + Some(failure) => Err(internal_error(failure.message)), None => Ok(StoredResource::from_storage( resource_type, &p.id, @@ -1337,15 +1353,19 @@ impl ElasticsearchBackend { &self, ops: &[(usize, &str, &str, &Value)], owners: usize, - ) -> Vec> { - let mut failures: Vec> = vec![None; owners]; + ) -> Vec> { + let mut failures: Vec> = (0..owners).map(|_| None).collect(); fn fail_chunk( - failures: &mut [Option], + failures: &mut [Option], chunk: &[(usize, &str, &str, &Value)], message: String, + transient: bool, ) { for (i, ..) in chunk { - failures[*i].get_or_insert_with(|| message.clone()); + failures[*i].get_or_insert_with(|| BulkFailure { + message: message.clone(), + transient, + }); } } for chunk in ops.chunks(BULK_OPS_PER_REQUEST) { @@ -1369,6 +1389,7 @@ impl ElasticsearchBackend { &mut failures, chunk, format!("Failed to send bulk index request: {e}"), + true, ); continue; } @@ -1381,6 +1402,7 @@ impl ElasticsearchBackend { &mut failures, chunk, format!("Bulk index request failed (status {status}): {payload}"), + is_transient_bulk_status(u64::from(status.as_u16())), ); continue; } @@ -1389,6 +1411,7 @@ impl ElasticsearchBackend { &mut failures, chunk, format!("Failed to read bulk index response: {e}"), + true, ); continue; } @@ -1409,8 +1432,13 @@ impl ElasticsearchBackend { .and_then(|v| v.get("error")) .map(|v| v.to_string()) .unwrap_or_else(|| "no item in bulk response".to_string()); - failures[*i].get_or_insert_with(|| { - format!("Failed to index document (status {item_status}): {error}") + // A missing item (status 0) means the response did not + // account for the document, not that it was rejected. + failures[*i].get_or_insert_with(|| BulkFailure { + message: format!( + "Failed to index document (status {item_status}): {error}" + ), + transient: item_status == 0 || is_transient_bulk_status(item_status), }); } } @@ -1560,9 +1588,15 @@ impl ReindexTarget for ElasticsearchBackend { prepared .into_iter() .zip(failures) - .map(|(p, failure)| match p.failure.or(failure) { - Some(message) => Err(internal_error(message)), - None => Ok(p.values), + .map(|(p, failure)| match (p.failure, failure) { + (Some(message), _) => Err(internal_error(message)), + // `$reindex` retries a run only for transient failures, so a + // rejection must not pass for an outage or the reverse. + (None, Some(failure)) if failure.transient => { + Err(unavailable_error(failure.message)) + } + (None, Some(failure)) => Err(internal_error(failure.message)), + (None, None) => Ok(p.values), }) .collect() } @@ -1837,3 +1871,20 @@ async fn delete_by_query_scoped( let payload: Value = response.json().await.unwrap_or_default(); Ok(payload.get("deleted").and_then(|d| d.as_u64()).unwrap_or(0)) } + +#[cfg(test)] +mod tests { + use super::is_transient_bulk_status; + + #[test] + fn bulk_statuses_that_ask_for_a_retry_are_transient() { + for status in [429, 500, 502, 503, 504] { + assert!(is_transient_bulk_status(status), "{status}"); + } + // 400 is how Elasticsearch rejects a document over the nested-object + // limit (#1050); rerunning it changes nothing. + for status in [200, 201, 400, 404, 409, 413] { + assert!(!is_transient_bulk_status(status), "{status}"); + } + } +} diff --git a/crates/persistence/src/search/reindex.rs b/crates/persistence/src/search/reindex.rs index a5cb2f2ec0..d72c20b9e2 100644 --- a/crates/persistence/src/search/reindex.rs +++ b/crates/persistence/src/search/reindex.rs @@ -395,6 +395,42 @@ pub struct ReindexProgressError { pub resource_id: String, /// Error message. pub error: String, + /// Whether the failure was transient — the writer was unavailable, timed + /// out, or asked to back off — so running the same work again may succeed. + /// A permanent failure (a document the search backend rejects outright, + /// such as one over Elasticsearch's nested-object limit, #1050) fails the + /// same way on every rerun. + #[serde(default = "retryable_by_default")] + pub retryable: bool, +} + +/// An error recorded without a classification keeps the retry it always had. +fn retryable_by_default() -> bool { + true +} + +/// Per-resource errors listed in [`ReindexProgress::to_parameters`]. The +/// status response stays bounded however many resources fail; `errorCount` +/// is always the full total. +const MAX_REPORTED_RESOURCE_ERRORS: usize = 100; + +/// Failing `Type/id`s named in the log line of an automatic generation that +/// ended with permanent errors. +const MAX_LOGGED_RESOURCE_ERRORS: usize = 5; + +/// Whether a writer's error is transient: the conditions the REST layer answers +/// with `503` or `504`, where the backend never judged the resource itself. +fn is_transient_error(error: &crate::error::StorageError) -> bool { + use crate::error::{BackendError, StorageError}; + matches!( + error, + StorageError::Backend( + BackendError::Unavailable { .. } + | BackendError::ConnectionFailed { .. } + | BackendError::PoolExhausted { .. } + | BackendError::Timeout { .. } + ) + ) } impl ReindexProgress { @@ -428,20 +464,51 @@ impl ReindexProgress { !self.errors.is_empty() || self.error_message.is_some() } + /// Returns true if the job recorded per-resource errors, every one of them + /// permanent, and did not fail as a whole. + pub fn has_only_permanent_errors(&self) -> bool { + self.error_message.is_none() + && !self.errors.is_empty() + && self.errors.iter().all(|error| !error.retryable) + } + /// Converts to FHIR Parameters resource. + /// + /// Besides the counters, lists the job's failure message and the first + /// [`MAX_REPORTED_RESOURCE_ERRORS`] failing resources, each with its error + /// and whether it is retryable, so an operator can find the resources that + /// are stored but not searchable without reading server logs. pub fn to_parameters(&self) -> serde_json::Value { - serde_json::json!({ - "resourceType": "Parameters", - "parameter": [ - {"name": "jobId", "valueString": self.job_id}, - {"name": "status", "valueCode": format!("{:?}", self.status).to_lowercase()}, - {"name": "total", "valueInteger": self.total_resources}, - {"name": "processed", "valueInteger": self.processed_resources}, - {"name": "entriesCreated", "valueInteger": self.entries_created}, - {"name": "errorCount", "valueInteger": self.errors.len()}, - {"name": "percentage", "valueDecimal": self.percentage()} - ] - }) + let mut parameter = vec![ + serde_json::json!({"name": "jobId", "valueString": self.job_id}), + serde_json::json!({"name": "status", "valueCode": format!("{:?}", self.status).to_lowercase()}), + serde_json::json!({"name": "total", "valueInteger": self.total_resources}), + serde_json::json!({"name": "processed", "valueInteger": self.processed_resources}), + serde_json::json!({"name": "entriesCreated", "valueInteger": self.entries_created}), + serde_json::json!({"name": "errorCount", "valueInteger": self.errors.len()}), + serde_json::json!({"name": "percentage", "valueDecimal": self.percentage()}), + ]; + if let Some(message) = &self.error_message { + parameter.push(serde_json::json!({"name": "errorMessage", "valueString": message})); + } + for error in self.errors.iter().take(MAX_REPORTED_RESOURCE_ERRORS) { + parameter.push(serde_json::json!({ + "name": "error", + "part": [ + {"name": "resourceType", "valueCode": error.resource_type}, + {"name": "resourceId", "valueId": error.resource_id}, + {"name": "message", "valueString": error.error}, + {"name": "retryable", "valueBoolean": error.retryable} + ] + })); + } + if self.errors.len() > MAX_REPORTED_RESOURCE_ERRORS { + parameter.push(serde_json::json!({ + "name": "errorsOmitted", + "valueInteger": self.errors.len() - MAX_REPORTED_RESOURCE_ERRORS + })); + } + serde_json::json!({"resourceType": "Parameters", "parameter": parameter}) } } @@ -946,9 +1013,57 @@ impl std::fmt::Debug for ReindexOperation { enum AutomaticGenerationOutcome { Clean, Cancelled, + /// The job completed, but every resource error it recorded is permanent. + /// A rerun would be rejected the same way, so it is reported, not retried. + PermanentErrors { + count: usize, + resources: String, + first_error: String, + }, Failed(String), } +/// Decides what an automatic generation's finished job means for retry policy. +fn automatic_outcome(progress: Option) -> AutomaticGenerationOutcome { + match progress { + Some(progress) if progress.status == ReindexStatus::Completed && !progress.has_errors() => { + AutomaticGenerationOutcome::Clean + } + Some(progress) if progress.status == ReindexStatus::Cancelled => { + AutomaticGenerationOutcome::Cancelled + } + Some(progress) + if progress.status == ReindexStatus::Completed + && progress.has_only_permanent_errors() => + { + AutomaticGenerationOutcome::PermanentErrors { + count: progress.errors.len(), + resources: progress + .errors + .iter() + .take(MAX_LOGGED_RESOURCE_ERRORS) + .map(|error| format!("{}/{}", error.resource_type, error.resource_id)) + .collect::>() + .join(", "), + first_error: progress.errors[0].error.clone(), + } + } + Some(progress) => AutomaticGenerationOutcome::Failed(format!( + "status {:?}, {} resource errors{}", + progress.status, + progress.errors.len(), + progress + .error_message + .as_deref() + .map(|message| format!(", error: {message}")) + .unwrap_or_default() + )), + None => { + AutomaticGenerationOutcome::Failed("job status disappeared after task exit".to_string()) + } + } +} + impl AutomaticReindexCoordinator { fn limits(&self, requested: usize) -> AutomaticReindexLimits { let requested = requested.max(1); @@ -1120,30 +1235,7 @@ impl AutomaticReindexCoordinator { "deferred reindex generation started" ); let _ = task_exit.await; - let outcome = match op.get_progress(&job_id).await { - Some(progress) - if progress.status == ReindexStatus::Completed - && !progress.has_errors() => - { - AutomaticGenerationOutcome::Clean - } - Some(progress) if progress.status == ReindexStatus::Cancelled => { - AutomaticGenerationOutcome::Cancelled - } - Some(progress) => AutomaticGenerationOutcome::Failed(format!( - "status {:?}, {} resource errors{}", - progress.status, - progress.errors.len(), - progress - .error_message - .as_deref() - .map(|message| format!(", error: {message}")) - .unwrap_or_default() - )), - None => AutomaticGenerationOutcome::Failed( - "job status disappeared after task exit".to_string(), - ), - }; + let outcome = automatic_outcome(op.get_progress(&job_id).await); (Some(job_id), outcome) } Err(error) => ( @@ -1168,6 +1260,9 @@ impl AutomaticReindexCoordinator { AutomaticGenerationOutcome::Cancelled => { state.consecutive_failures = 0; } + AutomaticGenerationOutcome::PermanentErrors { .. } => { + state.consecutive_failures = 0; + } AutomaticGenerationOutcome::Failed(_) => { state.consecutive_failures += 1; if state.consecutive_failures == 1 { @@ -1217,6 +1312,20 @@ impl AutomaticReindexCoordinator { types = ?resource_types, "deferred reindex generation was cancelled" ), + AutomaticGenerationOutcome::PermanentErrors { + count, + resources, + first_error, + } => tracing::error!( + tenant = %tenant_id, + generation, + job_id = ?job_id, + types = ?resource_types, + errors = count, + resources = %resources, + first_error = %first_error, + "deferred reindex completed, but resources were rejected permanently and are stored but not searchable; not retrying because a rerun fails the same way (every failure is listed by $reindex-status for this job)" + ), AutomaticGenerationOutcome::Failed(error) if retry => tracing::warn!( tenant = %tenant_id, generation, @@ -1279,6 +1388,7 @@ fn push_error( resource_type: &str, resource_id: &str, error: String, + retryable: bool, ) { let mut jobs_guard = jobs.write(); if let Some(progress) = jobs_guard.get_mut(job_id) { @@ -1286,6 +1396,7 @@ fn push_error( resource_type: resource_type.to_string(), resource_id: resource_id.to_string(), error, + retryable, }); } } @@ -1465,6 +1576,7 @@ async fn run_reindex( resource_type, page.resources[i].id(), format!("Failed to rebuild index entries: {e}"), + is_transient_error(&e), ), } } @@ -1715,6 +1827,7 @@ mod tests { active_writes: std::sync::atomic::AtomicUsize, max_active_writes: std::sync::atomic::AtomicUsize, failing_writes: std::sync::atomic::AtomicUsize, + permanent_failing_writes: std::sync::atomic::AtomicUsize, } impl ControlledBackend { @@ -1736,6 +1849,7 @@ mod tests { active_writes: std::sync::atomic::AtomicUsize::new(0), max_active_writes: std::sync::atomic::AtomicUsize::new(0), failing_writes: std::sync::atomic::AtomicUsize::new(failing_writes), + permanent_failing_writes: std::sync::atomic::AtomicUsize::new(0), }), receiver, ) @@ -1852,6 +1966,20 @@ mod tests { } .into()); } + if self + .permanent_failing_writes + .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |remaining| { + remaining.checked_sub(1) + }) + .is_ok() + { + return Err(crate::error::BackendError::Internal { + backend_name: "controlled".into(), + message: "injected permanent write failure".into(), + source: None, + } + .into()); + } Ok(1) } @@ -2618,8 +2746,8 @@ mod tests { assert_eq!(backend.count_calls.load(Ordering::SeqCst), 2); } - // Per-resource write errors leave the physical job Completed but are - // still failures for automatic retry policy. + // Per-resource transient write errors leave the physical job Completed + // but are still failures for automatic retry policy. let (backend, mut events) = ControlledBackend::new(Vec::new(), 2); let op = controlled_operation(backend.clone()); let hook = ReindexOnFinish::new(op.clone()); @@ -2637,6 +2765,121 @@ mod tests { assert_eq!(backend.write_calls.load(Ordering::SeqCst), 2); } + #[tokio::test] + async fn automatic_reindex_does_not_retry_a_completion_with_only_permanent_errors() { + let (backend, mut events) = ControlledBackend::new(Vec::new(), 0); + backend.permanent_failing_writes.store(1, Ordering::SeqCst); + let op = controlled_operation(backend.clone()); + let hook = ReindexOnFinish::new(op.clone()); + + hook.reindex_types( + &named_tenant("permanent-errors"), + vec!["Patient".to_string()], + ) + .await; + await_controlled_write(&mut events, "permanent-errors", "Patient").await; + // Enough permits for a retry, so a wrongly retried job shows up as a + // second count and write instead of a hang. + backend.write_gate.add_permits(2); + await_automatic_idle(&op).await; + + assert_eq!(backend.count_calls.load(Ordering::SeqCst), 1); + assert_eq!(backend.write_calls.load(Ordering::SeqCst), 1); + let jobs = op.list_jobs(); + assert_eq!(jobs.len(), 1); + assert_eq!(jobs[0].status, ReindexStatus::Completed); + assert_eq!(jobs[0].errors.len(), 1); + assert!(!jobs[0].errors[0].retryable); + assert_eq!(jobs[0].errors[0].resource_id, "controlled-1"); + } + + #[test] + fn automatic_outcome_retries_unless_every_resource_error_is_permanent() { + let error = |id: &str, retryable| ReindexProgressError { + resource_type: "Provenance".to_string(), + resource_id: id.to_string(), + error: format!("rejected {id}"), + retryable, + }; + let mut progress = ReindexProgress::new("job"); + progress.status = ReindexStatus::Completed; + assert!(matches!( + automatic_outcome(Some(progress.clone())), + AutomaticGenerationOutcome::Clean + )); + + progress.errors.push(error("big", false)); + match automatic_outcome(Some(progress.clone())) { + AutomaticGenerationOutcome::PermanentErrors { + count, + resources, + first_error, + } => { + assert_eq!(count, 1); + assert_eq!(resources, "Provenance/big"); + assert_eq!(first_error, "rejected big"); + } + other => panic!("expected permanent errors, got {other:?}"), + } + + // One transient error among permanent ones earns the retry. + progress.errors.push(error("flaky", true)); + assert!(matches!( + automatic_outcome(Some(progress.clone())), + AutomaticGenerationOutcome::Failed(_) + )); + + // A job that failed as a whole is retried whatever its resource errors. + progress.errors.truncate(1); + progress.status = ReindexStatus::Failed; + progress.error_message = Some("Failed to fetch resources".to_string()); + assert!(matches!( + automatic_outcome(Some(progress)), + AutomaticGenerationOutcome::Failed(_) + )); + assert!(matches!( + automatic_outcome(None), + AutomaticGenerationOutcome::Failed(_) + )); + } + + #[test] + fn transient_errors_are_the_ones_rest_answers_with_503_or_504() { + use crate::error::BackendError; + let backend_name = || "test".to_string(); + let message = || "detail".to_string(); + for transient in [ + BackendError::Unavailable { + backend_name: backend_name(), + message: message(), + }, + BackendError::ConnectionFailed { + backend_name: backend_name(), + message: message(), + }, + BackendError::PoolExhausted { + backend_name: backend_name(), + }, + BackendError::Timeout { + backend_name: backend_name(), + message: message(), + }, + ] { + assert!(is_transient_error(&transient.into())); + } + for permanent in [ + BackendError::Internal { + backend_name: backend_name(), + message: message(), + source: None, + }, + BackendError::QueryError { message: message() }, + BackendError::SerializationError { message: message() }, + ] { + assert!(!is_transient_error(&permanent.into())); + } + } + #[tokio::test] async fn automatic_reindex_keeps_new_work_after_a_retry_is_exhausted() { let (backend, mut events) = ControlledBackend::new(Vec::new(), 2); @@ -2749,9 +2992,20 @@ mod tests { resource_type: "Patient".to_string(), resource_id: "1".to_string(), error: "test error".to_string(), + retryable: true, }); assert!(progress.has_errors()); + assert!(!progress.has_only_permanent_errors()); + + // Errors recorded before classification existed keep their retry. + let unclassified: ReindexProgressError = serde_json::from_value(serde_json::json!({ + "resource_type": "Patient", + "resource_id": "1", + "error": "test error" + })) + .unwrap(); + assert!(unclassified.retryable); } #[test] @@ -2762,4 +3016,58 @@ mod tests { assert_eq!(params["resourceType"], "Parameters"); assert!(params["parameter"].is_array()); } + + #[test] + fn test_progress_to_parameters_lists_bounded_resource_errors() { + let named = |params: &serde_json::Value, name: &str| -> Vec { + params["parameter"] + .as_array() + .unwrap() + .iter() + .filter(|parameter| parameter["name"] == name) + .cloned() + .collect() + }; + let mut progress = ReindexProgress::new("job-123"); + progress.status = ReindexStatus::Completed; + progress.errors.push(ReindexProgressError { + resource_type: "Provenance".to_string(), + resource_id: "oversized".to_string(), + error: "nested documents exceeded the allowed limit".to_string(), + retryable: false, + }); + let params = progress.to_parameters(); + assert!(named(¶ms, "errorMessage").is_empty()); + assert!(named(¶ms, "errorsOmitted").is_empty()); + let errors = named(¶ms, "error"); + assert_eq!(errors.len(), 1); + assert_eq!( + errors[0]["part"], + serde_json::json!([ + {"name": "resourceType", "valueCode": "Provenance"}, + {"name": "resourceId", "valueId": "oversized"}, + {"name": "message", "valueString": "nested documents exceeded the allowed limit"}, + {"name": "retryable", "valueBoolean": false} + ]) + ); + + for n in 0..MAX_REPORTED_RESOURCE_ERRORS + 4 { + progress.errors.push(ReindexProgressError { + resource_type: "Patient".to_string(), + resource_id: format!("p{n}"), + error: "unavailable".to_string(), + retryable: true, + }); + } + progress.error_message = Some("Failed to fetch resources".to_string()); + let params = progress.to_parameters(); + let total = MAX_REPORTED_RESOURCE_ERRORS + 5; + assert_eq!(named(¶ms, "errorCount")[0]["valueInteger"], total); + assert_eq!(named(¶ms, "error").len(), MAX_REPORTED_RESOURCE_ERRORS); + assert_eq!(named(¶ms, "errorsOmitted")[0]["valueInteger"], 5); + assert_eq!( + named(¶ms, "errorMessage")[0]["valueString"], + "Failed to fetch resources" + ); + } } diff --git a/crates/persistence/tests/elasticsearch_tests.rs b/crates/persistence/tests/elasticsearch_tests.rs index 0d73ee5445..e9404ec580 100644 --- a/crates/persistence/tests/elasticsearch_tests.rs +++ b/crates/persistence/tests/elasticsearch_tests.rs @@ -1008,6 +1008,16 @@ mod es_integration { error.to_string().contains("nested"), "the rejection must be the nested-object limit, got: {error}" ); + // A rejection, not an outage: `$reindex` must not retry it (#1050). + assert!( + matches!( + error, + helios_persistence::error::StorageError::Backend( + helios_persistence::error::BackendError::Internal { .. } + ) + ), + "the nested-object rejection must be permanent, got: {error:?}" + ); assert!( before .read(&tenant, "Provenance", "oversized") diff --git a/crates/rest/README.md b/crates/rest/README.md index d8cd72e33d..acb2bb7b40 100644 --- a/crates/rest/README.md +++ b/crates/rest/README.md @@ -137,6 +137,14 @@ the background and is polled via `/$reindex-status/[job_id]`. per minute. Under high job volume, the count limit can evict a status earlier; an evicted status returns `404`. Tasks still executing, including cancellation in progress, are protected. Cancellation channels are released when tasks exit. +- A resource a search index rejects does not fail the job: it is counted in + `errorCount`, stays stored and readable by id, and is listed as an `error` + parameter (`resourceType`, `resourceId`, `message`, `retryable`) for the first + 100 failures, with `errorsOmitted` counting the rest. `retryable` is `true` + when the index was unavailable, timed out or pushed back, and `false` when it + rejected the document itself — for Elasticsearch, a resource over + `HFS_ELASTICSEARCH_NESTED_OBJECTS_LIMIT` (#1050) — which a rerun cannot fix. + A job that fails as a whole also carries `errorMessage`. - The `s3` backend standalone has no search index of any kind, so `$reindex` there returns `501`. Every other backend and composite supports it. - The same applies after a **server upgrade that adds a parameter to the From b60eb609328b60a1cacf85a5075331db9e8842c6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mauricio=20Pe=C3=B1aranda?= Date: Mon, 14 Sep 2026 10:10:03 -0400 Subject: [PATCH 3/5] perf(mongodb): page bulk-submit receipts by keyset in index order (#1046) The composite secondary sync read `bulk_entry_results` with `skip` and a `{line_number, file_url}` sort. Neither receipt index ends in that order, so every page examined every receipt key in the manifest and sorted in memory: on the #941 corpus (18,955,865 receipts) one page took 39 s, putting the sync of a manifest #1024 ingests in 55 minutes at roughly 95 hours. - `get_entry_results_page` now continues strictly after the last stored `(file_url, line_number)` with an `$or` range, sorted in that order, and returns `EntryResultContinuation::Keyset` with every receipt's stored identity, like SQLite and PostgreSQL (#986). An `Offset` continuation is rejected. - A receipt missing `file_url` or `line_number`, or with a negative line, is an error instead of a default, since a cursor built from an invented identity would skip or repeat receipts. - Schema v10 replaces `idx_bulk_entry_results_outcome` with `idx_bulk_entry_results_outcome_line`, which adds `(file_url, line_number)` so the sync's `outcome = success` pages walk it too. Its prefix still serves outcome counts. - The shared receipt paging contract now runs on MongoDB (`exact_sql_pages` becomes `exact_keyset_pages`), and a profiler guard asserts filtered pages have no blocking sort and no `skip`. Measured with explain on the #941 corpus: a page past receipt 9,000,000 examines 1,001 keys in 9 ms (was 18,955,865 keys, 39,096 ms). On 2M synthetic receipts with the v10 index, the filtered sync page examines 1,001 keys in 3 ms (was 1,900,000 keys, 3,692 ms). --- .../src/backends/mongodb/bulk_submit.rs | 100 +++++--- .../src/backends/mongodb/schema.rs | 15 +- .../src/backends/sqlite/bulk_submit.rs | 8 +- crates/persistence/src/core/bulk_submit.rs | 13 +- .../tests/bulk_submit/paging_contract.rs | 7 +- crates/persistence/tests/mongodb_tests.rs | 234 ++++++++++++++++-- crates/persistence/tests/postgres_tests.rs | 2 +- 7 files changed, 316 insertions(+), 63 deletions(-) diff --git a/crates/persistence/src/backends/mongodb/bulk_submit.rs b/crates/persistence/src/backends/mongodb/bulk_submit.rs index 90a2cd1364..30b5f1afa5 100644 --- a/crates/persistence/src/backends/mongodb/bulk_submit.rs +++ b/crates/persistence/src/backends/mongodb/bulk_submit.rs @@ -47,10 +47,10 @@ use crate::core::bulk_export_worker::{LeaseError, WorkerId}; use crate::core::bulk_submit::{ BulkEntryOutcome, BulkEntryResult, BulkProcessingOptions, BulkSubmitProvider, BulkSubmitRollbackProvider, CANCELLED_ABORT_REASON, ChangeType, EntryCountSummary, - EntryResultContinuation, EntryResultPage, ManifestPhase, ManifestStatus, NdjsonEntry, - PagedEntryResult, StreamProcessingResult, StreamingBulkSubmitProvider, SubmissionChange, - SubmissionId, SubmissionManifest, SubmissionStatus, SubmissionSummary, UnindexedEntry, - invalid_entry_result_page, + EntryResultContinuation, EntryResultCursor, EntryResultPage, ManifestPhase, ManifestStatus, + NdjsonEntry, PagedEntryResult, StreamProcessingResult, StreamingBulkSubmitProvider, + SubmissionChange, SubmissionId, SubmissionManifest, SubmissionStatus, SubmissionSummary, + UnindexedEntry, invalid_entry_result_page, }; use crate::core::bulk_submit_publication::{ ManifestPublicationResult, ManifestPublicationStatus, canonical_publication_files, @@ -239,6 +239,34 @@ fn decode_entry_result(doc: &Document) -> BulkEntryResult { } } +/// Decodes a receipt together with its stored `(file_url, line_number)` +/// identity, the key receipt pages continue from. +/// +/// Every receipt has carried both fields since MongoDB hosted `$bulk-submit` +/// (#521), so one without them — or with a negative line — is corrupt. It is an +/// error rather than a default: a keyset cursor built from an invented identity +/// would silently skip or repeat receipts. +fn decode_paged_entry_result(doc: &Document) -> StorageResult { + let file_url = doc + .get_str("file_url") + .map_err(|e| internal_error(format!("entry result missing file_url: {e}")))? + .to_string(); + let line = doc + .get_i64("line_number") + .map_err(|e| internal_error(format!("entry result missing line_number: {e}")))?; + let line_number = u64::try_from(line) + .map_err(|_| internal_error("Negative stored receipt line number".to_string()))?; + let mut result = decode_entry_result(doc); + result.line_number = line_number; + Ok(PagedEntryResult { + result, + stored_identity: Some(EntryResultCursor { + file_url, + line_number, + }), + }) +} + fn decode_change(doc: &Document) -> SubmissionChange { let change_type: ChangeType = doc .get_str("change_type") @@ -953,12 +981,17 @@ impl BulkSubmitProvider for MongoBackend { "Receipt page limit must be greater than zero", )); } - let offset = match continuation { - None => 0, - Some(EntryResultContinuation::Offset(offset)) => *offset, - Some(EntryResultContinuation::Keyset(_)) => { + let after = match continuation { + None => None, + Some(EntryResultContinuation::Keyset(cursor)) => Some(( + cursor.file_url.as_str(), + i64::try_from(cursor.line_number).map_err(|_| { + invalid_entry_result_page("Receipt cursor line exceeds MongoDB int64 range") + })?, + )), + Some(EntryResultContinuation::Offset(_)) => { return Err(invalid_entry_result_page( - "mongodb receipt pages require an offset continuation", + "MongoDB receipt pages require a keyset continuation", )); } }; @@ -966,10 +999,25 @@ impl BulkSubmitProvider for MongoBackend { if let Some(outcome) = outcome_filter { filter.insert("outcome", outcome.to_string()); } + // Strictly after the last stored identity. Each branch is a bounded + // range on an index ending in `(file_url, line_number)`, so a page costs + // the same at any depth — unlike `skip`, which rescanned every earlier + // receipt on every page (#1046). + if let Some((file_url, line)) = after { + filter.insert( + "$or", + vec![ + doc! { "file_url": { "$gt": file_url } }, + doc! { "file_url": file_url, "line_number": { "$gt": line } }, + ], + ); + } + // The key order both receipt indexes end in, so no page needs a blocking + // in-memory sort. The old `{line_number, file_url}` order could not use + // either index. let options = FindOptions::builder() - .sort(doc! { "line_number": 1_i32, "file_url": 1_i32 }) - .skip(Some(offset as u64)) - .limit(Some(limit as i64)) + .sort(doc! { "file_url": 1_i32, "line_number": 1_i32 }) + .limit(Some(i64::from(limit))) .build(); let cursor = self .entry_results() @@ -978,30 +1026,20 @@ impl BulkSubmitProvider for MongoBackend { .with_options(options) .await .map_err(|e| internal_error(format!("query entry results: {e}")))?; - let results: Vec<_> = collect(cursor) + let entries = collect(cursor) .await? .iter() - .map(decode_entry_result) - .collect(); - let next = if results.len() == limit as usize { - Some(EntryResultContinuation::Offset( - offset - .checked_add(limit) - .ok_or_else(|| invalid_entry_result_page("Receipt offset exceeds u32 range"))?, - )) + .map(decode_paged_entry_result) + .collect::>>()?; + let next = if entries.len() == limit as usize { + entries + .last() + .and_then(|entry| entry.stored_identity.clone()) + .map(EntryResultContinuation::Keyset) } else { None }; - Ok(EntryResultPage { - entries: results - .into_iter() - .map(|result| PagedEntryResult { - result, - stored_identity: None, - }) - .collect(), - next, - }) + Ok(EntryResultPage { entries, next }) } async fn get_entry_counts( diff --git a/crates/persistence/src/backends/mongodb/schema.rs b/crates/persistence/src/backends/mongodb/schema.rs index a624cd3ab0..1a52d76f73 100644 --- a/crates/persistence/src/backends/mongodb/schema.rs +++ b/crates/persistence/src/backends/mongodb/schema.rs @@ -16,8 +16,11 @@ use super::backend::MongoBackendConfig; /// v7 adds the Bulk Data Submit collections and their indexes. v8 moves the /// artifact identity under its owning manifest. v9 replaces /// `idx_resources_type_deleted` with a longer index that also carries the -/// `$reindex` page order (#1021). -pub const SCHEMA_VERSION: i32 = 9; +/// `$reindex` page order (#1021). v10 replaces `idx_bulk_entry_results_outcome` +/// with `idx_bulk_entry_results_outcome_line`, which also carries the receipt +/// keyset order, so outcome-filtered receipt pages need no in-memory sort +/// (#1046). +pub const SCHEMA_VERSION: i32 = 10; /// Initialize MongoDB collections/indexes required by the backend. /// @@ -443,16 +446,22 @@ async fn ensure_bulk_submit_indexes(database: &Database) -> StorageResult<()> { true, ) .await?; + // Receipt pages walk `(file_url, line_number)` after an optional outcome + // filter. `idx_bulk_entry_results_line` serves the unfiltered walk; this + // serves the filtered one, and its prefix still serves outcome counts. let mut outcome_key = submission_key.clone(); outcome_key.insert("manifest_id", 1_i32); outcome_key.insert("outcome", 1_i32); + outcome_key.insert("file_url", 1_i32); + outcome_key.insert("line_number", 1_i32); create_index( &entry_results, outcome_key, - "idx_bulk_entry_results_outcome", + "idx_bulk_entry_results_outcome_line", false, ) .await?; + drop_index_if_present(&entry_results, "idx_bulk_entry_results_outcome").await?; let changes = database.collection::(CHANGES_COLLECTION); let mut change_key = submission_key.clone(); diff --git a/crates/persistence/src/backends/sqlite/bulk_submit.rs b/crates/persistence/src/backends/sqlite/bulk_submit.rs index 19cf684fba..c5370aa0a1 100644 --- a/crates/persistence/src/backends/sqlite/bulk_submit.rs +++ b/crates/persistence/src/backends/sqlite/bulk_submit.rs @@ -4394,8 +4394,12 @@ mod tests { #[tokio::test] async fn bulk_submit_exact_keyset_pages() { - paging_contract::exact_sql_pages(&create_test_backend(), &create_test_tenant(), i64::MAX) - .await; + paging_contract::exact_keyset_pages( + &create_test_backend(), + &create_test_tenant(), + i64::MAX, + ) + .await; } #[tokio::test] diff --git a/crates/persistence/src/core/bulk_submit.rs b/crates/persistence/src/core/bulk_submit.rs index edb363222b..7fae3e1209 100644 --- a/crates/persistence/src/core/bulk_submit.rs +++ b/crates/persistence/src/core/bulk_submit.rs @@ -559,10 +559,11 @@ pub struct EntryResultCursor { /// to the same provider with the same scope, outcome filter and page limit. #[derive(Debug, Clone, PartialEq, Eq)] pub enum EntryResultContinuation { - /// PostgreSQL and SQLite continue strictly after this stored identity. + /// PostgreSQL, SQLite and MongoDB continue strictly after this stored + /// identity. Keyset(EntryResultCursor), - /// MongoDB and S3 retain their existing offset-based traversal internally. - /// SQL providers reject this variant, including offset zero. + /// S3 retains its existing offset-based traversal internally. Keyset + /// providers reject this variant, including offset zero. Offset(u32), } @@ -571,7 +572,7 @@ pub enum EntryResultContinuation { pub struct PagedEntryResult { /// The unchanged ingestion result, also used by persisted S3 objects. pub result: BulkEntryResult, - /// Always present for PostgreSQL and SQLite. Other adapters may omit this: + /// Always present for PostgreSQL, SQLite and MongoDB. S3 may omit this: /// old S3 receipt objects do not retain a recoverable original file URL. pub stored_identity: Option, } @@ -1523,8 +1524,8 @@ pub trait BulkSubmitProvider: ResourceStorage { /// nonzero limit unchanged throughout that traversal. Filtering happens /// before limiting; a full last page may require a final empty request. /// - /// SQL providers use native keyset pagination and return every stored - /// identity. MongoDB/S3 encapsulate their existing offset mechanism. A + /// PostgreSQL, SQLite and MongoDB use native keyset pagination and return + /// every stored identity. S3 encapsulates its existing offset mechanism. A /// continuation of the wrong kind is an error, never a fallback request. /// /// This replaces the former `get_entry_results` offset method and is a diff --git a/crates/persistence/tests/bulk_submit/paging_contract.rs b/crates/persistence/tests/bulk_submit/paging_contract.rs index 80a4e98f1c..3db5628137 100644 --- a/crates/persistence/tests/bulk_submit/paging_contract.rs +++ b/crates/persistence/tests/bulk_submit/paging_contract.rs @@ -1,4 +1,5 @@ -// Included by the SQLite unit tests and PostgreSQL integration tests. +// Included by the SQLite unit tests and the PostgreSQL and MongoDB integration +// tests: every backend with a stored receipt identity pages by keyset. // The including module supplies `persistence` as an alias for the library. use persistence::core::bulk_submit::{ BulkEntryOutcome, BulkSubmitProvider, EntryResultContinuation, EntryResultCursor, @@ -45,7 +46,7 @@ async fn collect_pages( assert!(page.entries.len() <= limit as usize); if let Some(token) = &page.next { let EntryResultContinuation::Keyset(cursor) = token else { - panic!("SQL returned an OFFSET continuation"); + panic!("backend returned an OFFSET continuation"); }; assert_eq!( Some(cursor), @@ -68,7 +69,7 @@ async fn collect_pages( panic!("receipt traversal did not terminate"); } -pub async fn exact_sql_pages( +pub async fn exact_keyset_pages( backend: &B, tenant: &TenantContext, max_line: i64, diff --git a/crates/persistence/tests/mongodb_tests.rs b/crates/persistence/tests/mongodb_tests.rs index 1423d3849f..116ecfe2bc 100644 --- a/crates/persistence/tests/mongodb_tests.rs +++ b/crates/persistence/tests/mongodb_tests.rs @@ -8005,17 +8005,27 @@ mod bulk_submit { counts.total, 2, "both files' line 1 must survive as separate results" ); + let line_one = |file: &str| helios_persistence::core::EntryResultCursor { + file_url: file.to_string(), + line_number: 1, + }; let mut next = None; let mut ids = Vec::new(); - for page_index in 0..3 { + for expected_file in [Some("a.ndjson"), Some("b.ndjson"), None] { let page = backend .get_entry_results_page(&tenant, &id, &manifest_id, None, 1, next.as_ref()) .await .unwrap(); - assert!( + assert_eq!( page.entries .iter() - .all(|entry| entry.stored_identity.is_none()) + .map(|entry| entry.stored_identity.clone()) + .collect::>(), + expected_file + .map(|file| Some(line_one(file))) + .into_iter() + .collect::>(), + "each receipt carries its stored identity" ); ids.extend( page.entries @@ -8023,15 +8033,14 @@ mod bulk_submit { .map(|entry| entry.result.resource_id.unwrap()), ); next = page.next; - if page_index < 2 { - assert_eq!( + match expected_file { + Some(file) => assert_eq!( next, - Some(helios_persistence::core::EntryResultContinuation::Offset( - page_index + 1 + Some(helios_persistence::core::EntryResultContinuation::Keyset( + line_one(file) )) - ); - } else { - assert!(next.is_none(), "exact multiple must terminate"); + ), + None => assert!(next.is_none(), "exact multiple must terminate"), } } assert_eq!(ids, ["pa", "pb"]); @@ -8049,11 +8058,8 @@ mod bulk_submit { &manifest_id, None, 1, - Some(&helios_persistence::core::EntryResultContinuation::Keyset( - helios_persistence::core::EntryResultCursor { - file_url: String::new(), - line_number: 0, - } + Some(&helios_persistence::core::EntryResultContinuation::Offset( + 0 )) ) .await @@ -8080,14 +8086,208 @@ mod bulk_submit { .get_entry_results_page(&tenant, &id, &manifest_id, None, 1, None) .await .unwrap(); - assert!(delegated.entries[0].stored_identity.is_none()); + assert_eq!( + delegated.entries[0].stored_identity, + Some(line_one("a.ndjson")) + ); assert_eq!( delegated.next, - Some(helios_persistence::core::EntryResultContinuation::Offset(1)) + Some(helios_persistence::core::EntryResultContinuation::Keyset( + line_one("a.ndjson") + )) ); eprintln!("Verified MongoDB bulk-submit receipt pagination and Composite delegation"); } + mod receipt_paging_contract { + use helios_persistence as persistence; + include!(concat!( + env!("CARGO_MANIFEST_DIR"), + "/tests/bulk_submit/paging_contract.rs" + )); + } + + #[async_trait::async_trait] + impl receipt_paging_contract::ReceiptFixture for MongoBackend { + async fn seed_receipts( + &self, + tenant: &TenantContext, + submission: &SubmissionId, + manifest: &str, + rows: &[receipt_paging_contract::ReceiptRow], + ) { + // The fields `entry_result_statement` writes, minus `created`: an + // absent flag must keep meaning "not created". + let documents: Vec = rows + .iter() + .map(|row| { + doc! { + "tenant_id": tenant.tenant_id().as_str(), + "submitter": &submission.submitter, + "submission_id": &submission.submission_id, + "manifest_id": manifest, + "file_url": &row.file, + "line_number": row.line, + "resource_type": "Patient", + "resource_id": &row.id, + "outcome": row.outcome, + } + }) + .collect(); + self.get_database() + .await + .unwrap() + .collection::("bulk_entry_results") + .insert_many(documents) + .await + .unwrap(); + } + } + + /// The receipt paging contract SQLite and PostgreSQL already run: exact + /// keyset traversal in `(file_url, line_number)` order under every outcome + /// filter and page size, scope isolation, and range errors. Before #1046 + /// MongoDB paged with `skip` in `{line_number, file_url}` order, which no + /// index could serve. + #[tokio::test] + async fn test_receipt_paging_contract() { + let Some(backend) = create_backend("submit_receipt_paging").await else { + return; + }; + receipt_paging_contract::exact_keyset_pages( + &backend, + &create_tenant("receipt-paging"), + i64::MAX, + ) + .await; + eprintln!("Verified MongoDB receipt keyset paging contract"); + } + + /// #1046's acceptance criterion as a plan guard: the composite sync's + /// outcome-filtered receipt pages — the first, and one past a keyset + /// cursor — are index walks with no blocking in-memory sort and no `skip`. + /// The old `{line_number, file_url}` order could not use any index, so + /// every page sorted the whole manifest in memory. + #[tokio::test] + async fn test_receipt_pages_are_index_walks_without_a_blocking_sort() { + use receipt_paging_contract::{ReceiptFixture, ReceiptRow}; + + let Some(backend) = create_backend("submit_receipt_plan").await else { + eprintln!( + "Skipping test_receipt_pages_are_index_walks_without_a_blocking_sort (requires Docker or HFS_TEST_MONGODB_URL)" + ); + return; + }; + let tenant = create_tenant("receipt-plan"); + let submission = SubmissionId::generate("receipt-plan"); + let rows: Vec<_> = ["a.ndjson", "b.ndjson"] + .into_iter() + .flat_map(|file| { + (0..50).map(move |line| ReceiptRow { + file: file.to_string(), + line, + id: format!("{file}-{line}"), + outcome: if line % 10 == 0 { + "validation-error" + } else { + "success" + }, + }) + }) + .collect(); + backend + .seed_receipts(&tenant, &submission, "manifest", &rows) + .await; + + let database = backend.get_database().await.unwrap(); + if let Err(e) = database.run_command(doc! { "profile": 2_i32 }).await { + eprintln!( + "Skipping test_receipt_pages_are_index_walks_without_a_blocking_sort plan assertions: \ + {{profile: 2}} was refused ({e})" + ); + return; + } + let first = backend + .get_entry_results_page( + &tenant, + &submission, + "manifest", + Some(BulkEntryOutcome::Success), + 10, + None, + ) + .await + .unwrap(); + assert!(first.next.is_some(), "the first page must continue"); + backend + .get_entry_results_page( + &tenant, + &submission, + "manifest", + Some(BulkEntryOutcome::Success), + 10, + first.next.as_ref(), + ) + .await + .unwrap(); + let _ = database.run_command(doc! { "profile": 0_i32 }).await; + + let options = mongodb::options::FindOptions::builder() + .sort(doc! { "ts": 1_i32 }) + .build(); + let mut cursor = database + .collection::("system.profile") + .find(doc! { + "ns": format!("{}.bulk_entry_results", backend.config().database_name), + "op": "query", + "command.find": "bulk_entry_results", + }) + .with_options(options) + .await + .expect("failed to query system.profile"); + let mut entries = Vec::new(); + while cursor + .advance() + .await + .expect("failed to advance profile cursor") + { + entries.push( + cursor + .deserialize_current() + .expect("failed to deserialize profile entry"), + ); + } + assert_eq!( + entries.len(), + 2, + "expected both receipt pages to be profiled" + ); + for entry in &entries { + // Absent, not false, when there is no sort stage. + assert!( + !entry.get_bool("hasSortStage").unwrap_or(false), + "receipt page must not sort in memory: {entry}" + ); + let plan = entry.get_str("planSummary").unwrap_or_default(); + assert!( + plan.contains("IXSCAN"), + "expected an index walk, got {plan}" + ); + let command = entry + .get_document("command") + .expect("profile entry missing command"); + assert!( + command.get("skip").is_none(), + "receipt pages must not skip: {command}" + ); + assert_eq!( + command.get_document("sort").ok(), + Some(&doc! { "file_url": 1_i32, "line_number": 1_i32 }) + ); + } + eprintln!("Verified MongoDB receipt pages walk an index without a blocking sort"); + } + /// The manifest counters are cumulative across every run of a manifest, so /// both writers into them — the worker's `add_manifest_progress` and the /// ingestion engine's per-batch bookkeeping — must add rather than assign. diff --git a/crates/persistence/tests/postgres_tests.rs b/crates/persistence/tests/postgres_tests.rs index 4bdfcd00ab..1c7b78d1f3 100644 --- a/crates/persistence/tests/postgres_tests.rs +++ b/crates/persistence/tests/postgres_tests.rs @@ -1890,7 +1890,7 @@ mod postgres_integration { #[tokio::test] async fn postgres_bulk_submit_exact_keyset_pages() { - receipt_paging_contract::exact_sql_pages( + receipt_paging_contract::exact_keyset_pages( &create_backend().await, &create_tenant("receipt-pages"), i64::from(i32::MAX), From 08a3c03048a8da4834ba56b25809067b063caaf2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mauricio=20Pe=C3=B1aranda?= Date: Mon, 14 Sep 2026 10:28:53 -0400 Subject: [PATCH 4/5] feat(ui): say when the search index is rebuilding (#1065) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit While a `$reindex` runs, stored resources stay readable by id but searches can miss them, and nothing on the page said so. #1082 stopped the rail showing placeholder zeros in that state and named the other half as follow-up: a `reindex_jobs_active`-style snapshot field, fed from the reindex job registry, driving a "search index rebuilding" banner. This is that follow-up. - `ReindexProgress` records the tenant it rebuilds. - `DashboardSnapshot::reindex_active` carries a `ReindexActivity` (running jobs, processed, total) for the tenant, `None` when nothing is rebuilding or no `$reindex` operation is wired. - The REST dashboard provider fills it from `ReindexOperation`'s in-memory job registry, never storage, so page loads stay constant-time (#1078). - Home shows a warning line in the live region with the percentage ("Search index rebuilding — 42% (8,000 of 19,000 resources)"), or without one while the rebuild is still counting, never "0%". A running rebuild keeps the region on the fast refresh, and the line is announced once, not on every percentage tick. - Resources shows the same line under its page head. - Strings in en, es and de. --- crates/observability/src/dashboard.rs | 41 +++++ crates/persistence/src/search/reindex.rs | 12 +- crates/rest/src/dashboard.rs | 77 ++++++++- crates/rest/src/lib.rs | 3 +- crates/ui/src/lib.rs | 60 ++++++- crates/ui/templates/pages/index.html | 8 + crates/ui/templates/pages/resources.html | 4 + crates/ui/tests/dashboard_pending_http.rs | 1 + .../ui/tests/dashboard_view_all_types_http.rs | 1 + crates/ui/tests/rebuild_banner_http.rs | 158 ++++++++++++++++++ locales/de/main.ftl | 2 + locales/en/main.ftl | 6 + locales/es/main.ftl | 2 + 13 files changed, 370 insertions(+), 5 deletions(-) create mode 100644 crates/ui/tests/rebuild_banner_http.rs diff --git a/crates/observability/src/dashboard.rs b/crates/observability/src/dashboard.rs index b1712f1f2c..a8b74e70d2 100644 --- a/crates/observability/src/dashboard.rs +++ b/crates/observability/src/dashboard.rs @@ -162,6 +162,28 @@ pub struct ExportJobCounts { pub queued: u64, } +/// A search-index rebuild (`$reindex`) in progress for one tenant (#1065). +/// +/// Carried by [`DashboardSnapshot::reindex_active`]. While it runs, stored +/// resources stay readable by id but searches can miss them, so the UI says so +/// rather than letting an empty result read as lost data. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash)] +pub struct ReindexActivity { + /// Rebuild jobs running (queued or in progress). + pub jobs: u64, + /// Resources processed so far, across those jobs. + pub processed: u64, + /// Resources to process, across those jobs; `0` while still being counted. + pub total: u64, +} + +impl ReindexActivity { + /// Whole percent done, `None` while the total is still being counted. + pub fn percent(&self) -> Option { + (self.total > 0).then(|| self.processed.min(self.total) * 100 / self.total) + } +} + /// A snapshot of the figures the dashboard renders. Plain data — no storage or /// FHIR types — so this crate stays dependency-light. #[derive(Clone, Debug, Default)] @@ -195,6 +217,10 @@ pub struct DashboardSnapshot { /// Non-terminal bulk-submit (import) jobs for the tenant. `None` under the /// same conditions as [`Self::export_jobs`]. pub import_jobs_active: Option, + /// Search-index rebuilds running for the tenant (#1065). `None` when none + /// is running or the deployment has no `$reindex` operation: the rebuild + /// banner is then simply absent, never a fabricated "0%". + pub reindex_active: Option, /// Where the figures come from and how far they can be trusted (#1078). /// Only [`Figures::Exact`] and [`Figures::Approximate`] carry figures; the /// other variants leave totals, `available` and `series` empty, and those @@ -709,6 +735,20 @@ mod tests { cond() } + #[test] + fn reindex_activity_percent_is_whole_and_unknown_until_counted() { + let activity = |processed, total| ReindexActivity { + jobs: 1, + processed, + total, + }; + assert_eq!(activity(0, 0).percent(), None); + assert_eq!(activity(1, 3).percent(), Some(33)); + assert_eq!(activity(18_957_456, 18_957_914).percent(), Some(99)); + // A counter that overshoots its total never reads above 100%. + assert_eq!(activity(12, 10).percent(), Some(100)); + } + #[tokio::test] async fn cold_load_fills_the_cache_and_fresh_hits_reuse_it() { let cache = SnapCache::default(); @@ -935,6 +975,7 @@ mod tests { available: Vec::new(), export_jobs: None, import_jobs_active: None, + reindex_active: None, figures: Figures::Exact { read_at: DateTime::from_timestamp(FIXED_READ_AT, 0).unwrap(), }, diff --git a/crates/persistence/src/search/reindex.rs b/crates/persistence/src/search/reindex.rs index d72c20b9e2..bcf44f2832 100644 --- a/crates/persistence/src/search/reindex.rs +++ b/crates/persistence/src/search/reindex.rs @@ -358,6 +358,13 @@ pub struct ReindexProgress { /// Unique job identifier. pub job_id: String, + /// The tenant the job rebuilds, so a per-tenant view (the dashboard's + /// rebuild banner, #1065) can pick out its own jobs. `None` only on + /// progress built without a tenant, such as one deserialized from before + /// the field existed. + #[serde(default)] + pub tenant_id: Option, + /// Current status. pub status: ReindexStatus, @@ -438,6 +445,7 @@ impl ReindexProgress { pub fn new(job_id: impl Into) -> Self { Self { job_id: job_id.into(), + tenant_id: None, status: ReindexStatus::Queued, total_resources: 0, processed_resources: 0, @@ -837,7 +845,8 @@ impl ReindexOperation { self.ensure_cleanup_task(); self.cleanup_old_jobs(REINDEX_STATUS_RETENTION_SECONDS); let job_id = Uuid::new_v4().to_string(); - let progress = ReindexProgress::new(&job_id); + let mut progress = ReindexProgress::new(&job_id); + progress.tenant_id = Some(tenant.tenant_id().as_str().to_string()); // Store the job self.jobs.write().insert(job_id.clone(), progress); @@ -2791,6 +2800,7 @@ mod tests { assert_eq!(jobs[0].errors.len(), 1); assert!(!jobs[0].errors[0].retryable); assert_eq!(jobs[0].errors[0].resource_id, "controlled-1"); + assert_eq!(jobs[0].tenant_id.as_deref(), Some("permanent-errors")); } #[test] diff --git a/crates/rest/src/dashboard.rs b/crates/rest/src/dashboard.rs index e510a252ce..3afba133e0 100644 --- a/crates/rest/src/dashboard.rs +++ b/crates/rest/src/dashboard.rs @@ -126,7 +126,7 @@ use async_trait::async_trait; use chrono::{DateTime, Duration, Utc}; use helios_observability::dashboard::{ DashboardPoint, DashboardProvider, DashboardSeries, DashboardSnapshot, DashboardWindow, - ExportJobCounts, Figures, TypeCount, + ExportJobCounts, Figures, ReindexActivity, TypeCount, }; use helios_observability::dashboard_counters::{ CountersSeries, DashboardCounters, ReconcileOutcome, StorageMarker, @@ -612,6 +612,23 @@ fn lock(mutex: &Mutex) -> MutexGuard<'_, T> { mutex.lock().unwrap_or_else(|e| e.into_inner()) } +/// Sums `tenant`'s queued and in-progress rebuilds among `jobs`, or `None` +/// when it has none running. +fn reindex_activity_of( + jobs: &[helios_persistence::search::ReindexProgress], + tenant: &str, +) -> Option { + let running: Vec<_> = jobs + .iter() + .filter(|job| job.status.is_running() && job.tenant_id.as_deref() == Some(tenant)) + .collect(); + (!running.is_empty()).then(|| ReindexActivity { + jobs: running.len() as u64, + processed: running.iter().map(|job| job.processed_resources).sum(), + total: running.iter().map(|job| job.total_resources).sum(), + }) +} + /// A tenant's last read job counts (see [`JOB_COUNTS_TTL`]). #[derive(Clone, Copy, Default)] struct JobCountsEntry { @@ -927,6 +944,9 @@ pub(crate) struct StorageDashboardProvider { export_jobs: Option>, /// Bulk-submit job store, when the active backend provides one. submit_jobs: Option>, + /// The server's `$reindex` operation, when wired: its in-memory job + /// registry is where a running search-index rebuild shows up (#1065). + reindex: Option>, /// The live write counters every figure is served from (#1078): the set /// the server's write observer feeds, injected by `build_app`. counters: Arc, @@ -960,6 +980,7 @@ impl StorageDashboardProvider { ), export_jobs: None, submit_jobs: None, + reindex: None, counters: Arc::new(DashboardCounters::new()), job_counts: Mutex::new(HashMap::new()), viewed: Mutex::new(HashMap::new()), @@ -982,6 +1003,24 @@ impl StorageDashboardProvider { self } + /// Attaches the `$reindex` operation, whose running jobs become each + /// snapshot's [`DashboardSnapshot::reindex_active`] (#1065). `None` leaves + /// that field `None`, so no rebuild banner is ever shown. + pub(crate) fn with_reindex( + mut self, + reindex: Option>, + ) -> Self { + self.reindex = reindex; + self + } + + /// The tenant's running search-index rebuilds. Reads the operation's + /// in-memory job registry, never storage, so a page load stays + /// constant-time (#1078). + fn reindex_activity(&self, tenant: &str) -> Option { + reindex_activity_of(&self.reindex.as_ref()?.list_jobs(), tenant) + } + /// Replaces the write counters this provider reads and seeds — the set the /// server's post-commit write observer records into. pub(crate) fn with_counters(mut self, counters: Arc) -> Self { @@ -1467,6 +1506,7 @@ where available, export_jobs, import_jobs_active, + reindex_active: self.reindex_activity(tenant_key), figures, }) } @@ -1508,6 +1548,7 @@ where available: Vec::new(), export_jobs, import_jobs_active, + reindex_active: self.reindex_activity(tenant_key), figures, } } @@ -2218,6 +2259,40 @@ mod tests { use serde_json::Value; use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; + #[test] + fn reindex_activity_sums_only_the_tenants_running_jobs() { + use helios_persistence::search::{ReindexProgress, ReindexStatus}; + + let job = |tenant: &str, status, processed, total| { + let mut job = ReindexProgress::new(format!("{tenant}-{processed}-{total}")); + job.tenant_id = Some(tenant.to_string()); + job.status = status; + job.processed_resources = processed; + job.total_resources = total; + job + }; + let jobs = [ + job("acme", ReindexStatus::InProgress, 250, 1_000), + job("acme", ReindexStatus::Queued, 0, 0), + job("acme", ReindexStatus::Completed, 500, 500), + job("other", ReindexStatus::InProgress, 9, 10), + ]; + assert_eq!( + reindex_activity_of(&jobs, "acme"), + Some(ReindexActivity { + jobs: 2, + processed: 250, + total: 1_000, + }) + ); + assert_eq!(reindex_activity_of(&jobs, "idle"), None); + assert_eq!( + reindex_activity_of(&jobs[2..3], "acme"), + None, + "a finished rebuild is not running" + ); + } + /// A private counter set per test. fn isolated_counters() -> Arc { Arc::new(DashboardCounters::new()) diff --git a/crates/rest/src/lib.rs b/crates/rest/src/lib.rs index b5801383ba..ad0ad0b5ca 100644 --- a/crates/rest/src/lib.rs +++ b/crates/rest/src/lib.rs @@ -616,7 +616,8 @@ where .with_job_stores( bulk_export.as_ref().map(|b| Arc::clone(&b.jobs)), bulk_submit.as_ref().map(|b| Arc::clone(&b.jobs)), - ), + ) + .with_reindex(ops_reindex.clone()), ); helios_observability::dashboard::set_provider(dashboard_provider.clone()); // The provider never awaits storage on a page load (#1078): it serves diff --git a/crates/ui/src/lib.rs b/crates/ui/src/lib.rs index e19c514893..ba6c959446 100644 --- a/crates/ui/src/lib.rs +++ b/crates/ui/src/lib.rs @@ -96,7 +96,7 @@ use axum_htmx::{AutoVaryLayer, HxHistoryRestoreRequest, HxRequest, HxTarget}; use chrono::{DateTime, Datelike, Duration, Utc}; use helios_observability::dashboard::{ DashboardPoint, DashboardSeries, DashboardSnapshot, DashboardWindow, ExportJobCounts, Figures, - SnapshotState, TypeCount, + ReindexActivity, SnapshotState, TypeCount, }; use helios_persistence::core::{BulkProviderStore, ResourceStorage, SettingsStore}; use rust_embed::RustEmbed; @@ -809,6 +809,26 @@ fn dashboard_notice(state: &SnapshotState, now: DateTime) -> NoticeLine { } } +/// `data-dash-notice` slug of the rebuild line (#1065). +const REBUILD_NOTICE: &str = "rebuilding"; + +/// The "search index rebuilding" sentence for a running rebuild (#1065): +/// with its percentage once the rebuild has counted its resources, without +/// one before, so it never shows a fabricated "0%". +fn rebuild_text(i18n: &I18n, activity: &ReindexActivity) -> String { + match activity.percent() { + Some(percent) => i18n.t_args( + "search-index-rebuilding", + &std::collections::BTreeMap::from([ + ("percent".to_string(), percent.to_string()), + ("processed".to_string(), grouped(activity.processed)), + ("total".to_string(), grouped(activity.total)), + ]), + ), + None => i18n.t("search-index-rebuilding-counting"), + } +} + /// The landing page. `dash_live` (`#dash-live`) and `chart_card` /// (`#dash-chart`) are also rendered alone, as the htmx fragments [`index`] /// answers a request targeting either region with. @@ -834,6 +854,9 @@ struct IndexPage { /// Whether the page renders its waiting state: no figure is known yet /// (#1078). chart_waiting: bool, + /// A search-index rebuild running for the tenant (#1065), rendered as a + /// warning line inside the live region so each refresh keeps it current. + rebuild: Option, /// Whether the storage backend cannot count at all /// ([`Figures::Unsupported`]): the chart area says so /// instead of waiting or charting, and nothing polls. @@ -898,6 +921,21 @@ struct IndexPage { } impl IndexPage { + /// The rebuild line's wording (see [`rebuild_text`]). + fn rebuild_text(&self, activity: &ReindexActivity) -> String { + rebuild_text(&self.i18n, activity) + } + + /// The rebuild line's `aria-live`: announced when it appears, then quiet + /// on the refreshes that only move its percentage. + fn rebuild_aria_live(&self) -> &'static str { + if self.quiet_notices.iter().any(|seen| seen == REBUILD_NOTICE) { + "off" + } else { + "polite" + } + } + /// The `aria-live` politeness of a notice line of `kind` (#1078). fn notice_aria_live(&self, kind: &DashboardNotice) -> &'static str { if self.quiet_notices.iter().any(|seen| seen == kind.slug()) { @@ -972,6 +1010,9 @@ struct ResourcesPage { create_advertised_types: String, create_schema_types: String, create_metadata_available: bool, + /// A search-index rebuild running for the tenant (#1065): results may miss + /// stored resources until it finishes, so the page head says so. + rebuild: Option, /// The search-builder partial's save controls are the Saved Queries page's /// job, not this one's. show_save: bool, @@ -993,6 +1034,13 @@ struct ResourcesPage { builder_url: Option, } +impl ResourcesPage { + /// The rebuild line's wording (see [`rebuild_text`]). + fn rebuild_text(&self, activity: &ReindexActivity) -> String { + rebuild_text(&self.i18n, activity) + } +} + /// Explains how to configure terminology navigation, or why the configured /// value cannot be used (#611). #[derive(Template)] @@ -2536,6 +2584,7 @@ async fn resources( .map(capability::CreateTargets::schema_resources_csv) .unwrap_or_default(), create_metadata_available: targets.is_some(), + rebuild: live.as_ref().and_then(|snapshot| snapshot.reindex_active), show_save: false, rail_counts_approximate, rail_entries, @@ -7946,6 +7995,7 @@ fn dash_state( .map(|jobs| (jobs.running, jobs.queued)) .hash(&mut hasher); snapshot.import_jobs_active.hash(&mut hasher); + snapshot.reindex_active.hash(&mut hasher); std::mem::discriminant(&snapshot.figures).hash(&mut hasher); (now.timestamp() / 60).hash(&mut hasher); format!("{:016x}", hasher.finish()) @@ -8070,7 +8120,8 @@ async fn build_index_page( let live_refresh = ready && !chart_waiting && !unsupported; let refresh_moving = live_refresh && (snapshot.figures.is_approximate() - || snapshot.import_jobs_active.is_some_and(|n| n > 0)); + || snapshot.import_jobs_active.is_some_and(|n| n > 0) + || snapshot.reindex_active.is_some()); let refresh_href = if slow_watch { Some(format!("{retry_base}&retry={DASH_PENDING_RETRIES}")) } else { @@ -8089,6 +8140,7 @@ async fn build_index_page( all_types: dash.all_types, all_types_href: dash.all_types_href, notice, + rebuild: snapshot.reindex_active, chart_waiting, chart_unsupported: unsupported, figures_unknown_key: if unsupported { @@ -8680,6 +8732,7 @@ fn sample_snapshot(window: DashboardWindow) -> DashboardSnapshot { available, export_jobs: None, import_jobs_active: None, + reindex_active: None, // Invented figures take the same rendering path as measured ones, so // they carry figures. Nothing reads this time: a no-provider render's // notice is the undated sample-data line, which says the figures are @@ -8864,6 +8917,7 @@ mod tests { None, ); IndexPage { + rebuild: None, status: Status { version, checked_at, @@ -9522,6 +9576,7 @@ mod tests { available: Vec::new(), export_jobs: None, import_jobs_active: None, + reindex_active: None, figures: Figures::Exact { read_at: DateTime::from_timestamp(1_752_451_200, 0).expect("valid instant"), }, @@ -9571,6 +9626,7 @@ mod tests { }], export_jobs: None, import_jobs_active: None, + reindex_active: None, figures: Figures::Exact { read_at: DateTime::from_timestamp(1_752_451_200, 0).expect("valid instant"), }, diff --git a/crates/ui/templates/pages/index.html b/crates/ui/templates/pages/index.html index 2c9b48b7e9..e697fa4abe 100644 --- a/crates/ui/templates/pages/index.html +++ b/crates/ui/templates/pages/index.html @@ -220,6 +220,14 @@

{{ i18n.t("nav-home") }}

{{ i18n.t("chart-pending-retry") }} {% endif %}

+ {# + #1065: a search-index rebuild is running. Stored resources stay readable + by id, but searches can miss them until it finishes. Inside the live + region, so the refresh a running rebuild keeps going updates it. + #} + {% if let Some(activity) = rebuild %} +

{{ self.rebuild_text(activity) }}

+ {% endif %} {% if chart_unsupported %} {# #1078: this backend cannot count, so there is nothing to wait for and diff --git a/crates/ui/templates/pages/resources.html b/crates/ui/templates/pages/resources.html index da6a647e5b..a2cd7d9eb3 100644 --- a/crates/ui/templates/pages/resources.html +++ b/crates/ui/templates/pages/resources.html @@ -19,6 +19,10 @@

{{ i18n.t("resources-heading") }}

{{ i18n.t("resources-lede") }}

+ {# #1065: results may miss stored resources while the search index rebuilds. #} + {% if let Some(activity) = rebuild %} +

{{ self.rebuild_text(activity) }}

+ {% endif %}