Add node health routing support

README.md

@@ -721,7 +721,9 @@ the client automatically falls back to the next scope in the chain:
 
 For datacenter or rack scoped routing in a multi-datacenter cluster, include a seed host from the
 target datacenter. The client queries configured seeds with the scope filter and only falls back
-after no seed returns nodes for that scope.
+after no seed returns nodes for that scope. Configured seed hosts remain discovery candidates, but
+they are not added to the routing node list unless `/localnodes` returns them for the selected
+scope.
 
 ```java
 // Rack -> Datacenter -> Cluster fallback chain
@@ -886,6 +888,61 @@ The default configuration works well for most use cases. Consider adjusting sett
 - **Long-running connections**: Default settings are optimal; session resumption primarily
   benefits reconnection scenarios
 
+### Node Health
+
+The client tracks active, quarantined, and down nodes while routing requests:
+
+- transport failures count as health failures
+- consecutive health failures mark a node down
+- down nodes are probed in the background with `GET /localnodes`
+- enough consecutive HTTP 200 probe responses move a down node into quarantine
+- quarantined nodes are sampled into routing once every configured number of attempts
+- enough consecutive successful contacts from a quarantined node promote it back to active
+
+For DynamoDB API requests, most HTTP responses mean the node was reachable for health purposes,
+including application-level errors. Authentication failures are detected from the parsed DynamoDB
+error code, such as `InvalidSignatureException`, `MissingAuthenticationTokenException`, or
+`UnrecognizedClientException`, because Alternator usually reports these as HTTP 400 responses.
+Authentication HTTP statuses (`401` or `403`) and transport failures also count as health failures.
+For `/localnodes`, any non-200 HTTP status or transport failure counts as a health failure.
+Traffic and `/localnodes` probe failure streaks are tracked separately, so a successful probe does
+not clear DynamoDB traffic/auth failures, and a successful DynamoDB request does not clear
+`/localnodes` probe failures.
+
+Key route affinity hashes and votes over all known discovered nodes so candidate order is stable
+while health changes. The final request-routing guard normally skips down nodes and admits at most
+one quarantined node when the quarantine sampling policy allows verification traffic. If there are
+no active nodes, routing fails open across all known nodes without changing their stored health
+state.
+
+The defaults are: 10 consecutive health failures mark a node down, 3 consecutive successful
+down-node probes move it into quarantine, 10 consecutive successful quarantined contacts promote it
+to active, down nodes are probed every 30 seconds, and quarantined nodes are sampled once every 10
+routing attempts.
+
+```java
+import com.scylladb.alternator.NodeHealthConfig;
+
+DynamoDbClient client = AlternatorDynamoDbClient.builder()
+    .endpointOverride(URI.create("https://127.0.0.1:8043"))
+    .credentialsProvider(myCredentials)
+    .withNodeHealthConfig(NodeHealthConfig.builder()
+        .withConsecutiveFailureThreshold(5)
+        .withDownNodeRecoverySuccessThreshold(3)
+        .withDownNodeProbePeriodMs(10_000)
+        .withQuarantineTrafficInterval(20)
+        .withQuarantineSuccessThreshold(10)
+        .build())
+    .build();
+```
+
+For custom integrations, `AlternatorLiveNodes` exposes `reportNodeResult(...)`,
+`getActiveNodes()`, `getQuarantinedNodes()`, and `getDownNodes()`. Report routed DynamoDB request
+outcomes with `TRAFFIC_SUCCESS` or `TRAFFIC_FAILURE`, and local node-health probe outcomes with
+`PROBE_SUCCESS` or `PROBE_FAILURE`. Probe success moves a `DOWN` node to quarantine; sampled traffic
+success moves a quarantined node back to active. Node health can be disabled with
+`AlternatorConfig.builder().withNodeHealthDisabled()`.
+
 ### Key Route Affinity (LWT Optimization)
 
 Key route affinity is an optimization for Lightweight Transactions (LWT) that use Paxos
@@ -894,7 +951,7 @@ it reduces Paxos round-trips and improves latency for conditional writes.
 
 **Note:** Synchronous clients automatically discover missing partition-key names via
 `DescribeTable`. Async clients can use key route affinity with pre-configured partition-key
-names; requests for tables without pre-configured metadata fall back to round-robin routing.
+names; requests for tables without pre-configured metadata fall back to random query-plan routing.
 
 #### Quick start
 
@@ -915,7 +972,7 @@ DynamoDbClient client = AlternatorDynamoDbClient.builder()
 
 | Mode | Description |
 |------|-------------|
-| `KeyRouteAffinity.NONE` | Default — standard round-robin load balancing |
+| `KeyRouteAffinity.NONE` | Default — standard random query-plan load balancing |
 | `KeyRouteAffinity.RMW` | Optimize read-before-write operations (conditional updates/puts/deletes with `ConditionExpression`, `Expected`, or non-NONE `ReturnValues`) |
 | `KeyRouteAffinity.ANY_WRITE` | Optimize all write operations (`PutItem`, `UpdateItem`, `DeleteItem`, `BatchWriteItem`) |
 
@@ -949,7 +1006,7 @@ DynamoDbClient client = AlternatorDynamoDbClient.builder()
 4. For `BatchWriteItem`, each usable write votes for its preferred node; voted nodes are tried by
    vote count, then by node URL
 5. Non-qualifying operations and requests without usable partition keys continue to use
-   round-robin load balancing
+   random query-plan load balancing
 
 #### When to use key route affinity
 

src/integration-test/java/com/scylladb/alternator/AlternatorDynamoDbAsyncClientIT.java

@@ -4,6 +4,7 @@
 import static org.junit.Assume.*;
 
 import com.scylladb.alternator.internal.AlternatorLiveNodes;
+import com.scylladb.alternator.internal.LazyQueryPlan;
 import com.scylladb.alternator.routing.ClusterScope;
 import com.scylladb.alternator.routing.DatacenterScope;
 import com.scylladb.alternator.routing.RackScope;
@@ -175,99 +176,123 @@ public void testUpdateLiveNodes() throws Exception {
   }
 
   @Test
-  public void testNodeDiscoveryWithRoundRobin() throws Exception {
+  public void testNodeDiscoveryWithRandomQueryPlan() throws Exception {
     AlternatorDynamoDbAsyncClientWrapper client = buildClient(IntegrationTestConfig.DATACENTER, "");
 
     // Wait for node discovery
     Thread.sleep(1000);
 
-    Set<URI> meetNodes = new HashSet<>();
     List<URI> allNodes = client.getLiveNodes();
+    assertFalse("Should have at least one live node", allNodes.isEmpty());
 
-    // Call nextAsURI more times than there are nodes to verify round-robin
-    for (int i = 0; i < allNodes.size() * 2; i++) {
-      meetNodes.add(client.nextAsURI());
+    Set<URI> metNodes = new HashSet<>();
+    LazyQueryPlan queryPlan = new LazyQueryPlan(client.getAlternatorLiveNodes());
+    while (queryPlan.hasNext()) {
+      metNodes.add(queryPlan.next());
     }
 
-    assertEquals("Should visit all nodes via round-robin", allNodes.size(), meetNodes.size());
+    assertEquals("Should visit all live nodes via random query plan", new HashSet<>(allNodes), metNodes);
 
     client.close();
   }
 
   @Test
   public void testDynamoDBOperations() throws Exception {
-    String tableName = "java_async_client_test_table";
+    String tableName = "java_async_client_test_table_" + seedUri.getScheme();
 
     AlternatorDynamoDbAsyncClientWrapper wrapper = buildClient();
     DynamoDbAsyncClient client = wrapper.getClient();
 
-    // Clean up if table exists
     try {
+      // Clean up if table exists
+      try {
+        client.deleteTable(DeleteTableRequest.builder().tableName(tableName).build()).get();
+        Thread.sleep(500);
+      } catch (Exception e) {
+        // Table doesn't exist, that's fine
+      }
+
+      // Create table
+      client
+          .createTable(
+              CreateTableRequest.builder()
+                  .tableName(tableName)
+                  .keySchema(
+                      KeySchemaElement.builder().attributeName("ID").keyType(KeyType.HASH).build())
+                  .attributeDefinitions(
+                      AttributeDefinition.builder()
+                          .attributeName("ID")
+                          .attributeType(ScalarAttributeType.S)
+                          .build())
+                  .provisionedThroughput(
+                      ProvisionedThroughput.builder()
+                          .readCapacityUnits(1L)
+                          .writeCapacityUnits(1L)
+                          .build())
+                  .build())
+          .get();
+
+      // Put item
+      client
+          .putItem(
+              PutItemRequest.builder()
+                  .tableName(tableName)
+                  .item(
+                      java.util.Map.of(
+                          "ID", AttributeValue.builder().s("123").build(),
+                          "Name", AttributeValue.builder().s("test-value").build()))
+                  .build())
+          .get();
+
+      GetItemResponse getResult = getItemEventually(client, tableName, "123");
+      java.util.Map<String, AttributeValue> item = getResult.item();
+
+      assertNotNull("Should get item back", item);
+      assertTrue("Should get ID attribute back", item.containsKey("ID"));
+      assertTrue("Should get Name attribute back", item.containsKey("Name"));
+      assertEquals("123", item.get("ID").s());
+      assertEquals("test-value", item.get("Name").s());
+
+      // Delete item
+      client
+          .deleteItem(
+              DeleteItemRequest.builder()
+                  .tableName(tableName)
+                  .key(java.util.Map.of("ID", AttributeValue.builder().s("123").build()))
+                  .build())
+          .get();
+
+      // Clean up
       client.deleteTable(DeleteTableRequest.builder().tableName(tableName).build()).get();
-      Thread.sleep(500);
-    } catch (Exception e) {
-      // Table doesn't exist, that's fine
-    }
+    } finally {
+      try {
+        client.deleteTable(DeleteTableRequest.builder().tableName(tableName).build()).get();
+      } catch (Exception e) {
+        // Best effort cleanup
+      }
 
-    // Create table
-    client
-        .createTable(
-            CreateTableRequest.builder()
-                .tableName(tableName)
-                .keySchema(
-                    KeySchemaElement.builder().attributeName("ID").keyType(KeyType.HASH).build())
-                .attributeDefinitions(
-                    AttributeDefinition.builder()
-                        .attributeName("ID")
-                        .attributeType(ScalarAttributeType.S)
-                        .build())
-                .provisionedThroughput(
-                    ProvisionedThroughput.builder()
-                        .readCapacityUnits(1L)
-                        .writeCapacityUnits(1L)
-                        .build())
-                .build())
-        .get();
-
-    // Put item
-    client
-        .putItem(
-            PutItemRequest.builder()
-                .tableName(tableName)
-                .item(
-                    java.util.Map.of(
-                        "ID", AttributeValue.builder().s("123").build(),
-                        "Name", AttributeValue.builder().s("test-value").build()))
-                .build())
-        .get();
-
-    // Get item
-    GetItemResponse getResult =
-        client
-            .getItem(
-                GetItemRequest.builder()
-                    .tableName(tableName)
-                    .key(java.util.Map.of("ID", AttributeValue.builder().s("123").build()))
-                    .build())
-            .get();
-
-    assertNotNull("Should get item back", getResult.item());
-    assertEquals("123", getResult.item().get("ID").s());
-    assertEquals("test-value", getResult.item().get("Name").s());
-
-    // Delete item
-    client
-        .deleteItem(
-            DeleteItemRequest.builder()
-                .tableName(tableName)
-                .key(java.util.Map.of("ID", AttributeValue.builder().s("123").build()))
-                .build())
-        .get();
-
-    // Clean up
-    client.deleteTable(DeleteTableRequest.builder().tableName(tableName).build()).get();
+      wrapper.close();
+    }
+  }
 
-    wrapper.close();
+  private GetItemResponse getItemEventually(
+      DynamoDbAsyncClient client, String tableName, String id) throws Exception {
+    GetItemRequest request =
+        GetItemRequest.builder()
+            .tableName(tableName)
+            .key(java.util.Map.of("ID", AttributeValue.builder().s(id).build()))
+            .consistentRead(true)
+            .build();
+    GetItemResponse response = null;
+    for (int attempt = 0; attempt < 10; attempt++) {
+      response = client.getItem(request).get();
+      java.util.Map<String, AttributeValue> item = response.item();
+      if (item != null && item.containsKey("ID") && item.containsKey("Name")) {
+        return response;
+      }
+      Thread.sleep(200);
+    }
+    return response;
   }
 
   @Test

src/integration-test/java/com/scylladb/alternator/AlternatorDynamoDbClientIT.java

@@ -4,6 +4,7 @@
 import static org.junit.Assume.*;
 
 import com.scylladb.alternator.internal.AlternatorLiveNodes;
+import com.scylladb.alternator.internal.LazyQueryPlan;
 import com.scylladb.alternator.routing.ClusterScope;
 import com.scylladb.alternator.routing.DatacenterScope;
 import com.scylladb.alternator.routing.RackScope;
@@ -175,21 +176,22 @@ public void testUpdateLiveNodes() throws Exception {
   }
 
   @Test
-  public void testNodeDiscoveryWithRoundRobin() throws Exception {
+  public void testNodeDiscoveryWithRandomQueryPlan() throws Exception {
     AlternatorDynamoDbClientWrapper client = buildClient(IntegrationTestConfig.DATACENTER, "");
 
     // Wait for node discovery
     Thread.sleep(1000);
 
-    Set<URI> meetNodes = new HashSet<>();
     List<URI> allNodes = client.getLiveNodes();
+    assertFalse("Should have at least one live node", allNodes.isEmpty());
 
-    // Call nextAsURI more times than there are nodes to verify round-robin
-    for (int i = 0; i < allNodes.size() * 2; i++) {
-      meetNodes.add(client.nextAsURI());
+    Set<URI> metNodes = new HashSet<>();
+    LazyQueryPlan queryPlan = new LazyQueryPlan(client.getAlternatorLiveNodes());
+    while (queryPlan.hasNext()) {
+      metNodes.add(queryPlan.next());
     }
 
-    assertEquals("Should visit all nodes via round-robin", allNodes.size(), meetNodes.size());
+    assertEquals("Should visit all live nodes via random query plan", new HashSet<>(allNodes), metNodes);
 
     client.close();
   }

src/integration-test/java/com/scylladb/alternator/DynamoDbOperationsIT.java

@@ -262,7 +262,7 @@ public void testOperationsAcrossMultipleNodes() throws Exception {
       // Create table
       client.createTable(createTableRequest(tableName));
 
-      // Write many items - requests will be distributed across nodes via round-robin
+      // Write many items - requests will be distributed across nodes via random query-plan routing
       int itemCount = 30;
       for (int i = 0; i < itemCount; i++) {
         Map<String, AttributeValue> item =

src/integration-test/java/com/scylladb/alternator/KeyRouteAffinityAutodiscoveryIT.java

@@ -125,7 +125,7 @@ public void testAutodiscoveryFindsPartitionKey() throws Exception {
       Thread.sleep(2000);
 
       // Second write should now benefit from cached PK info.
-      // If autodiscovery failed, this would still succeed (falls back to round-robin),
+      // If autodiscovery failed, this would still succeed (falls back to random query-plan routing),
       // but we verify the data is correct regardless.
       client.putItem(
           PutItemRequest.builder()