fix: stop Cassandra index reads stalling while a replica is down

deploy/helm/cassandra/helm/values.yaml

@@ -74,6 +74,13 @@ cassandra:
     # streaming connection is rejected as version-incompatible even though
     # gossip and schema exchange succeed.
     storage_compatibility_mode: NONE
+    # Secondary-index (SAI) reads are range reads, which the coordinator never
+    # retries on another replica, so a crashed replica stalls them for the whole
+    # timeout. 5s gives healthy range reads room to finish (normal function
+    # lookups take up to about 2s on a small cluster) and leaves about 5s of the
+    # services' 10s driver request timeout for the driver retry policy to retry
+    # on another coordinator. Larger values leave the retry too little time.
+    range_request_timeout: 5000ms
     # Automatic repair (Cassandra 5.0). Values supplied by the Cassandra SME.
     reject_repair_compaction_threshold: 1024
     repair_disk_headroom_reject_ratio: 0.2

docs/self-managed/high-availability.md

@@ -317,6 +317,28 @@ The control-plane services default to `LOCAL_QUORUM` consistency, and some
   [Zone spread for the quorum services](#zone-spread-for-the-quorum-services))
   and does not hold above three pods.
 
+#### Cassandra read retry policy (`RangeReadRetryPolicy`)
+
+When one Cassandra pod dies, index-backed reads (for example listing function
+versions) can fail for a few seconds until the other replicas stop treating
+the dead node as alive. The default `NextHostRetryPolicy` retries a read
+timeout on the next replica only when no data came back.
+
+The self-managed `nvcf-api` and `api-keys` services (ncp profile) select
+`RangeReadRetryPolicy`, which also retries once, on the next replica, a read
+timeout that has data but too few responses. This is independent of
+`highAvailability.mode`. To return to the default, or to tune resilience
+further, override the driver retry-policy class through the operator env hooks:
+
+```yaml
+api:
+  env:
+    SPRING_CASSANDRA_ADVANCED_RETRY_POLICY_CLASS: com.nvidia.boot.cassandra.retry.NextHostRetryPolicy
+apikeys:
+  env:
+    SPRING_CASSANDRA_ADVANCED_RETRY_POLICY_CLASS: com.nvidia.boot.cassandra.retry.NextHostRetryPolicy
+```
+
 ## Tuning scheduling policy (`global.affinity` / `global.topologySpreadConstraints`)
 
 Affinity and zone topology spread are shared scheduling policy, so you can

src/control-plane-services/api-keys/src/main/java/com/nvidia/apikeys/config/exceptions/MvcExceptionHandler.java

@@ -18,6 +18,7 @@
 package com.nvidia.apikeys.config.exceptions;
 
 import com.nvidia.boot.exceptions.BootResponseException;
+import com.nvidia.boot.exceptions.ServiceUnavailableException;
 import com.nvidia.boot.exceptions.handlers.BootMvcExceptionHandler;
 import io.opentelemetry.api.trace.Span;
 import jakarta.annotation.Nonnull;
@@ -63,7 +64,7 @@ protected ResponseEntity<Object> handleCassandraUnavailable(
             DataAccessException ex,
             WebRequest request) throws Exception {
         return super.handleException(
-                new CassandraUnavailableException("Key storage is temporarily unavailable", ex),
+                new ServiceUnavailableException("Key storage is temporarily unavailable", ex),
                 request);
     }
 

src/control-plane-services/api-keys/src/main/java/com/nvidia/apikeys/persistance/dao/KeysDao.java

@@ -21,7 +21,6 @@
 import com.datastax.oss.driver.api.core.cql.ResultSet;
 import com.nvidia.apikeys.converters.KeyOwnerVoBuilder;
 import com.nvidia.apikeys.config.exceptions.CassandraException;
-import com.nvidia.apikeys.config.exceptions.CassandraUnavailableException;
 import com.nvidia.apikeys.persistance.models.KeyByOwnerAndServiceModel;
 import com.nvidia.apikeys.persistance.models.KeyModel;
 import com.nvidia.apikeys.persistance.repositories.KeyByOwnerAndServiceRepository;
@@ -31,6 +30,7 @@
 import com.nvidia.apikeys.vo.KeyOwnerVo;
 import com.nvidia.apikeys.vo.KeyVo;
 import com.nvidia.apikeys.vo.SavedKeyVo;
+import com.nvidia.boot.exceptions.ServiceUnavailableException;
 import com.nvidia.boot.exceptions.UnprocessableEntityException;
 import java.time.Clock;
 import java.util.List;
@@ -187,13 +187,15 @@ public void deleteKey(KeyByOwnerAndServiceVo key) {
     /**
      * Writes the rows of a logged batch one by one after the batch failed. The batch outcome is
      * unknown: a timed-out batch may still be replayed from the batchlog. The rows use the batch
-     * timestamp, so writing them again leaves the same state as the batch.
+     * timestamp, so writing them again leaves the same state as the batch. A failure here does not
+     * advertise Retry-After: a client that retries a create generates a different key, while the
+     * first one may already be stored.
      */
     private void writeDirectly(Runnable write) {
         try {
             write.run();
         } catch (TransientDataAccessException | DataAccessResourceFailureException e) {
-            throw new CassandraUnavailableException(
+            throw ServiceUnavailableException.withoutRetryAfter(
                     "Key storage is temporarily unavailable, the outcome of the write is unknown", e);
         }
     }

src/control-plane-services/api-keys/src/main/resources/application-ncp.yaml

@@ -18,6 +18,11 @@ spring:
     contact-points: ${CASSANDRA_CONTACT_POINTS:cassandra.cassandra-system.svc.cluster.local}
     keyspace-name: api_keys_api
     local-datacenter: ${CASSANDRA_DATACENTER:ncp}
+    advanced:
+      retry-policy:
+        # Also retry range-read (index lookup) timeouts on the next host. Override with
+        # SPRING_CASSANDRA_ADVANCED_RETRY_POLICY_CLASS to go back to NextHostRetryPolicy.
+        class: com.nvidia.boot.cassandra.retry.RangeReadRetryPolicy
     ssl:
       enabled: false
       bundle: ""

src/control-plane-services/api-keys/src/test/java/com/nvidia/apikeys/config/exceptions/MvcExceptionHandlerTest.java

@@ -25,6 +25,7 @@
 import org.springframework.dao.DataAccessException;
 import org.springframework.data.cassandra.CassandraConnectionFailureException;
 import org.springframework.data.cassandra.CassandraReadTimeoutException;
+import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.ProblemDetail;
 import org.springframework.http.ResponseEntity;
@@ -52,7 +53,11 @@ private void assertServiceUnavailable(DataAccessException ex) throws Exception {
 
         assertThat(response).isNotNull();
         assertThat(response.getStatusCode()).isEqualTo(HttpStatus.SERVICE_UNAVAILABLE);
-        assertThat(response.getBody()).isInstanceOfSatisfying(ProblemDetail.class, body ->
-                assertThat(body.getDetail()).isEqualTo("Key storage is temporarily unavailable"));
+        assertThat(response.getHeaders().getFirst(HttpHeaders.RETRY_AFTER)).isEqualTo("10");
+        assertThat(response.getBody()).isInstanceOfSatisfying(ProblemDetail.class, body -> {
+            assertThat(body.getDetail()).isEqualTo("Key storage is temporarily unavailable");
+            assertThat(body.getType())
+                    .hasToString("urn:nv-boot:problem-details:service-unavailable");
+        });
     }
 }

src/control-plane-services/api-keys/src/test/java/com/nvidia/apikeys/persistance/dao/KeysDaoTest.java

@@ -39,15 +39,16 @@
 
 import com.datastax.oss.driver.api.core.CqlSession;
 import com.nvidia.apikeys.config.exceptions.CassandraException;
-import com.nvidia.apikeys.config.exceptions.CassandraUnavailableException;
 import com.nvidia.apikeys.persistance.models.KeyModel;
 import com.nvidia.apikeys.persistance.repositories.KeyByOwnerAndServiceRepository;
 import com.nvidia.apikeys.persistance.repositories.KeyRepository;
 import com.nvidia.apikeys.vo.SavedKeyVo;
+import com.nvidia.boot.exceptions.ServiceUnavailableException;
 import java.time.Clock;
 import java.util.List;
 import java.util.Optional;
 import org.junit.jupiter.api.Test;
+import org.springframework.http.HttpHeaders;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.Answers;
 import org.mockito.InOrder;
@@ -156,9 +157,10 @@ void save_throwsUnavailableIfDirectWriteFails() {
                                           any(InsertOptions.class)))
                 .thenThrow(writeTimeout("SIMPLE"));
 
-        assertThrowsExceptionWithDetails(
-                CassandraUnavailableException.class, () -> dao.save(KEY_VO_1, KEY_OWNER_VO_1),
+        var ex = assertThrowsExceptionWithDetails(
+                ServiceUnavailableException.class, () -> dao.save(KEY_VO_1, KEY_OWNER_VO_1),
                 "Key storage is temporarily unavailable, the outcome of the write is unknown");
+        assertThat(ex.getHeaders().containsHeader(HttpHeaders.RETRY_AFTER)).isFalse();
         verify(cassandraTemplateMock, never()).insert(eq(modelMock), any(InsertOptions.class));
     }
 
@@ -216,10 +218,11 @@ void deleteKey_throwsUnavailableIfDirectDeleteFails() {
         when(cassandraTemplateMock.delete(eq(expectedDeletedKeyModel), any(DeleteOptions.class)))
                 .thenThrow(writeTimeout("SIMPLE"));
 
-        assertThrowsExceptionWithDetails(
-                CassandraUnavailableException.class,
+        var ex = assertThrowsExceptionWithDetails(
+                ServiceUnavailableException.class,
                 () -> dao.deleteKey(KEY_BY_OWNER_AND_SERVICE_VO_1),
                 "Key storage is temporarily unavailable, the outcome of the write is unknown");
+        assertThat(ex.getHeaders().containsHeader(HttpHeaders.RETRY_AFTER)).isFalse();
     }
 
     private void mockBatch() {

src/control-plane-services/cloud-functions/nvcf-core/BUILD.bazel

@@ -235,6 +235,7 @@ NVCF_CORE_TEST_DEPS = [
     "@nv_third_party_deps//:org_springframework_spring_context",
     "@nv_third_party_deps//:org_springframework_spring_core",
     "@nv_third_party_deps//:org_springframework_spring_test",
+    "@nv_third_party_deps//:org_springframework_spring_tx",
     "@nv_third_party_deps//:org_springframework_spring_web",
     "@nv_third_party_deps//:org_springframework_spring_webflux",
     "@nv_third_party_deps//:org_springframework_spring_webmvc",

src/control-plane-services/cloud-functions/nvcf-core/src/main/java/com/nvidia/nvcf/configuration/GrpcConfiguration.java

@@ -16,16 +16,23 @@
  */
 package com.nvidia.nvcf.configuration;
 
+import static com.nvidia.nvcf.configuration.exceptions.ExceptionTracingExceptionHandler.STORAGE_UNAVAILABLE;
 import static com.nvidia.nvcf.util.NvcfConstants.TAG_NCA_ID;
 
 import com.nvidia.boot.exceptions.ForbiddenException;
+import com.nvidia.boot.exceptions.ServiceUnavailableException;
 import com.nvidia.boot.exceptions.UnauthorizedException;
 import com.nvidia.nvcf.configuration.exceptions.InvalidInvocationException;
 import io.grpc.Metadata;
 import io.grpc.Status;
 import io.grpc.Status.Code;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.dao.DataAccessException;
+import org.springframework.dao.TransientDataAccessResourceException;
+import org.springframework.data.cassandra.CassandraConnectionFailureException;
+import org.springframework.data.cassandra.CassandraInsufficientReplicasAvailableException;
+import org.springframework.data.cassandra.CassandraReadTimeoutException;
 import org.springframework.grpc.server.GlobalServerInterceptor;
 import org.springframework.grpc.server.advice.GrpcAdvice;
 import org.springframework.grpc.server.advice.GrpcExceptionHandler;
@@ -81,6 +88,17 @@ public io.grpc.StatusRuntimeException handleInvalidInvocationException(
         return status.asRuntimeException(metadata);
     }
 
+    // Keep in sync with ExceptionTracingExceptionHandler, which explains the choice.
+    @GrpcExceptionHandler({
+            CassandraReadTimeoutException.class,
+            CassandraInsufficientReplicasAvailableException.class,
+            TransientDataAccessResourceException.class,
+            CassandraConnectionFailureException.class
+    })
+    public Status handleStorageUnavailable(DataAccessException e) {
+        return toGrpcStatus(new ServiceUnavailableException(STORAGE_UNAVAILABLE, e));
+    }
+
     @GrpcExceptionHandler
     public Status handleException(AccessDeniedException e) {
         return handleErrorResponseException(new ForbiddenException(e.getMessage(), e.getCause()));

src/control-plane-services/cloud-functions/nvcf-core/src/main/java/com/nvidia/nvcf/configuration/exceptions/ExceptionTracingExceptionHandler.java

@@ -20,20 +20,28 @@
 import com.nvidia.boot.exceptions.ConflictException;
 import com.nvidia.boot.exceptions.ForbiddenException;
 import com.nvidia.boot.exceptions.NotFoundException;
+import com.nvidia.boot.exceptions.ServiceUnavailableException;
 import com.nvidia.boot.exceptions.TooManyRequestsException;
 import com.nvidia.boot.exceptions.UnauthorizedException;
 import com.nvidia.boot.exceptions.handlers.BootMvcExceptionHandler;
 import com.nvidia.nvcf.util.NvcfUtils;
 import io.micrometer.tracing.Tracer;
 import jakarta.annotation.Nonnull;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.dao.DataAccessException;
+import org.springframework.dao.TransientDataAccessResourceException;
+import org.springframework.data.cassandra.CassandraConnectionFailureException;
+import org.springframework.data.cassandra.CassandraInsufficientReplicasAvailableException;
+import org.springframework.data.cassandra.CassandraReadTimeoutException;
 import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.ExceptionHandler;
 import org.springframework.web.context.request.WebRequest;
 import org.springframework.web.reactive.function.client.WebClientResponseException;
 
 public class ExceptionTracingExceptionHandler extends BootMvcExceptionHandler {
 
+    public static final String STORAGE_UNAVAILABLE = "Storage is temporarily unavailable";
+
     @Autowired
     private Tracer tracer;
 
@@ -131,4 +139,26 @@ protected ResponseEntity<Object> handleException(
                 new BadRequestException(ex.getMessage(), ex.getCause()), request);
     }
 
+    /**
+     * Cassandra failures that are safe to report as retriable.
+     *
+     * <p>Only failures where Cassandra did no work are: a coordinator read timeout, too few replicas
+     * alive, an overloaded or bootstrapping coordinator, or no node reachable. Write timeouts and
+     * driver timeouts are not, because the write may have been applied and a retry of a create would
+     * duplicate it. Read and write failures are not transient. Keep in sync with
+     * {@code GrpcConfiguration}.
+     */
+    @ExceptionHandler({
+            CassandraReadTimeoutException.class,
+            CassandraInsufficientReplicasAvailableException.class,
+            TransientDataAccessResourceException.class,
+            CassandraConnectionFailureException.class
+    })
+    protected ResponseEntity<Object> handleStorageUnavailable(
+            DataAccessException ex,
+            WebRequest request) throws Exception {
+        return super.handleException(
+                new ServiceUnavailableException(STORAGE_UNAVAILABLE, ex), request);
+    }
+
 }

src/control-plane-services/cloud-functions/nvcf-core/src/test/java/com/nvidia/nvcf/configuration/GrpcConfigurationTest.java

@@ -0,0 +1,68 @@
+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.nvidia.nvcf.configuration;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.nvidia.boot.exceptions.ServiceUnavailableException;
+import com.nvidia.nvcf.configuration.exceptions.ExceptionTracingExceptionHandler;
+import io.grpc.Status;
+import org.junit.jupiter.api.Test;
+import org.springframework.dao.DataAccessException;
+import org.springframework.data.cassandra.CassandraReadTimeoutException;
+import org.springframework.grpc.server.advice.GrpcExceptionHandler;
+import org.springframework.web.bind.annotation.ExceptionHandler;
+import org.springframework.web.context.request.WebRequest;
+
+class GrpcConfigurationTest {
+
+    private static final String CQL = "Query; CQL [SELECT * FROM functions_v3]; read timeout";
+
+    private final GrpcConfiguration grpcConfiguration = new GrpcConfiguration();
+
+    @Test
+    void retriableStorageFailureIsUnavailable() {
+        var status = grpcConfiguration.handleStorageUnavailable(
+                new CassandraReadTimeoutException(false, CQL, null));
+
+        assertThat(status.getCode()).isEqualTo(Status.Code.UNAVAILABLE);
+        assertThat(status.getDescription()).isEqualTo("Storage is temporarily unavailable");
+    }
+
+    @Test
+    void serviceUnavailableIsUnavailable() {
+        var status = grpcConfiguration.handleErrorResponseException(
+                new ServiceUnavailableException("Storage is temporarily unavailable"));
+
+        assertThat(status.getCode()).isEqualTo(Status.Code.UNAVAILABLE);
+        assertThat(status.getDescription()).isEqualTo("Storage is temporarily unavailable");
+    }
+
+    @Test
+    void grpcAndMvcTreatTheSameStorageFailuresAsRetriable() throws Exception {
+        var grpcTypes = GrpcConfiguration.class
+                .getMethod("handleStorageUnavailable", DataAccessException.class)
+                .getAnnotation(GrpcExceptionHandler.class).value();
+        var mvcTypes = ExceptionTracingExceptionHandler.class
+                .getDeclaredMethod("handleStorageUnavailable", DataAccessException.class,
+                                   WebRequest.class)
+                .getAnnotation(ExceptionHandler.class).value();
+
+        assertThat(grpcTypes).isNotEmpty().containsExactlyInAnyOrder(mvcTypes);
+    }
+}