fix(elasticsearch): a read of an index still being created waits for its primary shard (#1402)

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::<Value>().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);

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<Option<Value>> {
-    let mut last_failure: Option<RetryableFailure> = 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<u64> = (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::<u64>(), 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() {

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.
 // ---------------------------------------------------------------------------

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.