diff --git a/crates/persistence/src/backends/elasticsearch/schema.rs b/crates/persistence/src/backends/elasticsearch/schema.rs index af5e624a5..34180d98c 100644 --- a/crates/persistence/src/backends/elasticsearch/schema.rs +++ b/crates/persistence/src/backends/elasticsearch/schema.rs @@ -828,6 +828,22 @@ pub async fn ensure_index( )); } + // The create request waits for the primary shard by itself (the default + // `wait_for_active_shards=1`), so the write that follows cannot race the + // allocation. It gives up after its 30 s `timeout` and says so; the index + // exists all the same, the write waits for the primary again, and a read + // retries an unstarted shard (#1402) — so this is worth a line, not an + // error. + if let Ok(body) = response.json::().await + && body.get("shards_acknowledged") == Some(&Value::Bool(false)) + { + tracing::warn!( + index, + "Elasticsearch created the index but its primary shard had not started when \ + the request timed out; reads of it fail until it does" + ); + } + // Created from the current mapping, marker included. backend.mark_schema_checked(&index); tracing::debug!("Created Elasticsearch index '{}'", index); diff --git a/crates/persistence/src/backends/elasticsearch/search_impl.rs b/crates/persistence/src/backends/elasticsearch/search_impl.rs index 315ae1131..eebcdb0f9 100644 --- a/crates/persistence/src/backends/elasticsearch/search_impl.rs +++ b/crates/persistence/src/backends/elasticsearch/search_impl.rs @@ -74,6 +74,38 @@ pub(super) const MAX_SEARCH_RETRIES: u32 = 2; /// Initial backoff before retrying a transient ES error. Doubled per attempt. pub(super) const RETRY_BASE_DELAY_MS: u64 = 100; +/// The error type of a read that reached an index whose primary shard has no +/// started copy. +const NO_SHARD_AVAILABLE: &str = "no_shard_available_action_exception"; + +/// Retries for a read answered [`NO_SHARD_AVAILABLE`] (in addition to the +/// initial attempt), in place of [`MAX_SEARCH_RETRIES`] (#1402). +/// +/// Unlike the other transient answers this one is *expected*: an index is in +/// the cluster state — so it is not an `index_not_found_exception`, which +/// reads as an empty set — from the moment its creation starts, and its +/// primary shard is started only some time later. The request that creates +/// the index waits for that; a read from anyone else (another request, the +/// composite's asynchronous sync worker indexing behind a write, another HFS +/// instance) does not, and gets a `503` for the whole window. The window is +/// ~100 ms on an idle single node and was seen to pass the general budget's +/// ~300 ms on a loaded one, so the first search after the first write of a +/// resource type failed. +/// +/// With [`NO_SHARD_RETRY_MAX_DELAY_MS`] the waits are 100, 200, 400, 800 ms +/// and then four of 1 s: at most [`NO_SHARD_RETRY_BUDGET_MS`] of waiting +/// before the failure is reported. A shard that is genuinely lost (a red +/// index) therefore costs a read that long instead of ~300 ms; the answer is +/// an error either way. +const MAX_NO_SHARD_RETRIES: u32 = 8; + +/// Cap on the doubling backoff between [`MAX_NO_SHARD_RETRIES`] attempts. +const NO_SHARD_RETRY_MAX_DELAY_MS: u64 = 1_000; + +/// Total waiting the [`NO_SHARD_AVAILABLE`] schedule can add to one read. +#[cfg(test)] +const NO_SHARD_RETRY_BUDGET_MS: u64 = 5_500; + /// How a non-success Elasticsearch response is handled (#1294). #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(super) enum EsFailureClass { @@ -98,7 +130,7 @@ pub(super) enum EsFailureClass { const RETRYABLE_ES_ERROR_TYPES: &[&str] = &[ "es_rejected_execution_exception", "circuit_breaking_exception", - "no_shard_available_action_exception", + NO_SHARD_AVAILABLE, ]; /// Error types under which Elasticsearch 7.17 reports a query *value* it could @@ -393,6 +425,26 @@ enum RetryableFailure { Transient { status: u16, body: String }, } +impl RetryableFailure { + /// How many times a read that failed this way is sent again, and how long + /// to wait before resend number `attempt` (0-based): the general schedule, + /// or the longer one for [`NO_SHARD_AVAILABLE`]. + fn retry_schedule(&self, attempt: u32) -> (u32, u64) { + let delay_ms = RETRY_BASE_DELAY_MS << attempt.min(16); + match self { + RetryableFailure::Transient { body, .. } + if es_error_types(body).iter().any(|t| t == NO_SHARD_AVAILABLE) => + { + ( + MAX_NO_SHARD_RETRIES, + delay_ms.min(NO_SHARD_RETRY_MAX_DELAY_MS), + ) + } + _ => (MAX_SEARCH_RETRIES, delay_ms), + } + } +} + /// The result of searching an index that does not exist. /// /// Indices are created lazily on the first write of a resource type, so a type @@ -441,9 +493,8 @@ pub(super) async fn send_read_with_retry( index: &str, body: Value, ) -> StorageResult> { - let mut last_failure: Option = None; - - for attempt in 0..=MAX_SEARCH_RETRIES { + let mut attempt: u32 = 0; + let (attempts, last_failure) = loop { let failure = match send_search_once(backend, op, index, body.clone()).await { SearchAttempt::Body(v) => return Ok(Some(v)), SearchAttempt::EmptyIndex => return Ok(None), @@ -454,33 +505,34 @@ pub(super) async fn send_read_with_retry( } }; - if attempt < MAX_SEARCH_RETRIES { - let delay_ms = RETRY_BASE_DELAY_MS << attempt; - tracing::warn!( - attempt = attempt + 1, - max = MAX_SEARCH_RETRIES + 1, - delay_ms, - index, - "Retryable ES {} failure, retrying", - op.name() - ); - sleep(Duration::from_millis(delay_ms)).await; + // The budget is that of the failure just seen, so a read that meets + // an unstarted shard and then some other transient answer stops as + // soon as it is past the general budget. + let (max_retries, delay_ms) = failure.retry_schedule(attempt); + if attempt >= max_retries { + break (attempt + 1, failure); } - last_failure = Some(failure); - } + tracing::warn!( + attempt = attempt + 1, + max = max_retries + 1, + delay_ms, + index, + "Retryable ES {} failure, retrying", + op.name() + ); + sleep(Duration::from_millis(delay_ms)).await; + attempt += 1; + }; - let attempts = MAX_SEARCH_RETRIES + 1; - Err( - match last_failure.expect("a retryable branch always sets last_failure") { - RetryableFailure::Unreachable(message) => unavailable_error(format!( - "Elasticsearch unreachable after {attempts} attempts: {message}" - )), - RetryableFailure::Transient { status, body } => internal_error(format!( - "{} failed after {attempts} attempts (status {status}): {body}", - op.title() - )), - }, - ) + Err(match last_failure { + RetryableFailure::Unreachable(message) => unavailable_error(format!( + "Elasticsearch unreachable after {attempts} attempts: {message}" + )), + RetryableFailure::Transient { status, body } => internal_error(format!( + "{} failed after {attempts} attempts (status {status}): {body}", + op.title() + )), + }) } /// Converts an Elasticsearch hit's `sort` array into cursor values, dropping @@ -1337,6 +1389,47 @@ mod tests { .to_string() } + /// #1402: only a shard with no started copy gets the longer schedule, and + /// what that schedule can cost a read is the documented bound. + #[test] + fn only_an_unstarted_shard_gets_the_longer_retry_schedule() { + const SPEE: &str = "search_phase_execution_exception"; + let transient = |status: u16, body: String| RetryableFailure::Transient { status, body }; + + for status in [500, 503] { + let no_shard = transient(status, es_error_body(SPEE, NO_SHARD_AVAILABLE, None)); + let waits: Vec = (0..MAX_NO_SHARD_RETRIES) + .map(|attempt| { + let (max_retries, delay_ms) = no_shard.retry_schedule(attempt); + assert_eq!(max_retries, MAX_NO_SHARD_RETRIES); + delay_ms + }) + .collect(); + assert_eq!(waits, [100, 200, 400, 800, 1_000, 1_000, 1_000, 1_000]); + assert_eq!(waits.iter().sum::(), NO_SHARD_RETRY_BUDGET_MS); + } + + for other in [ + transient(503, String::new()), + transient( + 503, + es_error_body(SPEE, "node_disconnected_exception", None), + ), + transient( + 429, + es_error_body( + "es_rejected_execution_exception", + "es_rejected_execution_exception", + None, + ), + ), + RetryableFailure::Unreachable("connection refused".to_string()), + ] { + assert_eq!(other.retry_schedule(0), (MAX_SEARCH_RETRIES, 100)); + assert_eq!(other.retry_schedule(1), (MAX_SEARCH_RETRIES, 200)); + } + } + /// #1294: status × error type → retry / client error / server error. #[test] fn es_failure_classification_table() { diff --git a/crates/persistence/tests/elasticsearch_search_wiremock.rs b/crates/persistence/tests/elasticsearch_search_wiremock.rs index 307c48df1..7550ed37b 100644 --- a/crates/persistence/tests/elasticsearch_search_wiremock.rs +++ b/crates/persistence/tests/elasticsearch_search_wiremock.rs @@ -284,6 +284,122 @@ async fn a_bad_query_that_mentions_index_not_found_is_still_a_bad_query() { } } +// --------------------------------------------------------------------------- +// A read of an index that is still being created (#1402). +// --------------------------------------------------------------------------- + +/// Attempts a read answered `no_shard_available_action_exception` gets, the +/// first included (mirrors the private `MAX_NO_SHARD_RETRIES + 1`). +const NO_SHARD_READ_ATTEMPTS: usize = 9; + +/// A real 7.17.29 answer for a search (and, identically, a count) of an index +/// whose creation is still in flight: the index is in the cluster state, its +/// primary shard is not started. +const NO_SHARD_AVAILABLE: &str = r#"{"error":{"root_cause":[{"type":"no_shard_available_action_exception","reason":"[b25ac7417c31][172.17.0.2:9300][indices:data/read/search[phase/query]]","index_uuid":"EmpvuGepQaG0jYAwhcHI1A","shard":"0","index":"hfs_read-stub_patient"}],"type":"search_phase_execution_exception","reason":"all shards failed","phase":"query","grouped":true,"failed_shards":[{"shard":0,"index":"hfs_read-stub_patient","node":"O31KMMFTRTe1bsMsfGLEvg","reason":{"type":"no_shard_available_action_exception","reason":"[b25ac7417c31][172.17.0.2:9300][indices:data/read/search[phase/query]]","index_uuid":"EmpvuGepQaG0jYAwhcHI1A","shard":"0","index":"hfs_read-stub_patient"}}]},"status":503}"#; + +fn no_shard_available() -> ResponseTemplate { + ResponseTemplate::new(503).set_body_raw(NO_SHARD_AVAILABLE, "application/json") +} + +/// The issue's sequence: someone else is creating the index, and a read meets +/// its unstarted primary for longer than the general retry budget lasts. The +/// read waits the creation out instead of failing. +#[tokio::test] +async fn a_read_outlasts_an_index_creation_longer_than_the_general_budget() { + const UNSTARTED_ANSWERS: usize = READ_ATTEMPTS + 1; + + let server = MockServer::start().await; + for url_path in [SEARCH_PATH, COUNT_PATH] { + let calls = Arc::new(AtomicUsize::new(0)); + on_post(&server, url_path, move |_| { + if calls.fetch_add(1, Ordering::SeqCst) < UNSTARTED_ANSWERS { + return no_shard_available(); + } + ResponseTemplate::new(200).set_body_json(json!({ + "count": 0, + "hits": { "total": { "value": 0, "relation": "eq" }, "hits": [] } + })) + }) + .await; + } + let es = backend(&server); + let query = SearchQuery::new("Patient"); + + let result = es + .search(&tenant(), &query) + .await + .expect("an index still being created is waited for"); + assert!(result.resources.items.is_empty()); + assert_eq!( + requests_to(&server, "POST", SEARCH_PATH).await, + UNSTARTED_ANSWERS + 1 + ); + + assert_eq!(es.search_count(&tenant(), &query).await.unwrap(), 0); + assert_eq!( + requests_to(&server, "POST", COUNT_PATH).await, + UNSTARTED_ANSWERS + 1 + ); +} + +/// The wait is bounded: a shard that never starts is an error after the +/// longer budget, never an empty result. +#[tokio::test] +async fn a_shard_that_never_starts_is_an_error_after_a_bounded_wait() { + let server = MockServer::start().await; + on_post(&server, SEARCH_PATH, |_| no_shard_available()).await; + + let error = backend(&server) + .search(&tenant(), &SearchQuery::new("Patient")) + .await + .map(|_| ()) + .expect_err("a lost shard is not an empty result"); + + assert!( + matches!(error, StorageError::Backend(BackendError::Internal { .. })), + "{error:?}" + ); + assert!( + error + .to_string() + .contains(&format!("after {NO_SHARD_READ_ATTEMPTS} attempts")), + "{error}" + ); + assert_eq!( + requests_to(&server, "POST", SEARCH_PATH).await, + NO_SHARD_READ_ATTEMPTS + ); +} + +/// The longer budget belongs to the unstarted shard alone: once the cluster +/// answers with any other transient failure, a read that is already past the +/// general budget stops. +#[tokio::test] +async fn the_longer_budget_does_not_carry_over_to_other_transient_failures() { + let server = MockServer::start().await; + let calls = Arc::new(AtomicUsize::new(0)); + on_post(&server, SEARCH_PATH, move |_| { + if calls.fetch_add(1, Ordering::SeqCst) < READ_ATTEMPTS { + no_shard_available() + } else { + error_body(503, "stubbed_exception") + } + }) + .await; + + let error = backend(&server) + .search(&tenant(), &SearchQuery::new("Patient")) + .await + .map(|_| ()) + .expect_err("an overloaded cluster is still an error"); + + assert!(error.to_string().contains("stubbed_exception"), "{error}"); + assert_eq!( + requests_to(&server, "POST", SEARCH_PATH).await, + READ_ATTEMPTS + 1 + ); +} + // --------------------------------------------------------------------------- // Startup mapping reconcile: which requests it sends. // --------------------------------------------------------------------------- diff --git a/crates/persistence/tests/elasticsearch_storage_write_wiremock.rs b/crates/persistence/tests/elasticsearch_storage_write_wiremock.rs index c29b6b575..52be78da1 100644 --- a/crates/persistence/tests/elasticsearch_storage_write_wiremock.rs +++ b/crates/persistence/tests/elasticsearch_storage_write_wiremock.rs @@ -310,6 +310,32 @@ async fn index_writes_retry_transient_failures_then_succeed() { } } +/// #1402: a create-index request that timed out waiting for the primary shard +/// (`shards_acknowledged: false`) still created the index. The write goes +/// ahead — Elasticsearch makes the index request wait for the primary — and +/// the index is not created a second time. +#[tokio::test] +async fn a_created_index_whose_primary_has_not_started_does_not_fail_the_write() { + let server = MockServer::start().await; + on(&server, "HEAD", INDEX_PATH, |_| ResponseTemplate::new(404)).await; + on(&server, "PUT", INDEX_PATH, |_| { + ResponseTemplate::new(200).set_body_json(json!({ + "acknowledged": true, + "shards_acknowledged": false, + "index": "hfs_write-stub_patient" + })) + }) + .await; + on(&server, "POST", TENANT_DBQ_PATH, |_| swept(0)).await; + on(&server, "POST", DOC_PATH, |_| indexed()).await; + + run_index_write(&backend(&server), "create") + .await + .expect("an unstarted primary must not fail the write"); + assert_eq!(requests_to(&server, "PUT", INDEX_PATH).await, 1); + assert_eq!(requests_to(&server, "POST", DOC_PATH).await, 1); +} + /// Every retryable answer is retried, and an exhausted retry is an /// *unavailable* error: the document was never judged, so a later attempt (the /// composite's own retry, a `$reindex`) may well succeed.