fix: CHWBL scheduler should use own load balancing when initial replica is not a candidate

When the initial replica from consistent hashing is not in the
candidate list, the scheduler now continues walking the ring with
CHWBL load-aware logic instead of returning all candidates to Ray's
default scheduling.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: Levi080513

cluster-image-builder/serve/_replica_scheduler/chwbl_scheduler.py

@@ -99,9 +99,6 @@ async def choose_replicas(
         replica_hash, replica_idx = self._search(payload_hash)
         initial_replica_id = self._hash_to_replica_id[replica_hash]
 
-        if initial_replica_id not in candidate_map:
-            return [candidate_replicas]
-
         logger.debug(
             f"CHWBL: Initial lookup for payload hash {payload_hash} -> "
             f"replica {initial_replica_id}"
@@ -112,6 +109,7 @@ async def choose_replicas(
 
         # Start from the initial replica and check load constraints
         checked_replica_ids: Set[ReplicaID] = set()
+        default_replica_id = None
         current_idx = replica_idx
         while len(checked_replica_ids) < len(candidate_map):
             current_hash = self._sorted_hashes[current_idx]
@@ -124,6 +122,9 @@ async def choose_replicas(
 
             checked_replica_ids.add(current_replica_id)
 
+            if default_replica_id is None:
+                default_replica_id = current_replica_id
+
             # Check if this replica meets the load constraints using snapshot
             if self._check_load_with_snapshot(current_replica_id, load_snapshot):
                 logger.info(
@@ -136,13 +137,14 @@ async def choose_replicas(
             # Move to next replica
             current_idx = (current_idx + 1) % len(self._sorted_hashes)
 
-        # All replicas overloaded, fallback to primary to preserve affinity
+        # All replicas overloaded, fallback to first candidate on ring to preserve affinity
+        fallback_id = default_replica_id
         logger.warning(
-            f"CHWBL: Using primary replica {initial_replica_id} "
+            f"CHWBL: Using fallback replica {fallback_id} "
             f"as no replica met load factor for payload hash {payload_hash}, "
-            f"snapshot_load={load_snapshot.get(initial_replica_id, 0)}"
+            f"snapshot_load={load_snapshot.get(fallback_id, 0)}"
         )
-        return [[candidate_map[initial_replica_id]]]
+        return [[candidate_map[fallback_id]]]
 
     def _check_load_with_snapshot(
         self, replica_id: ReplicaID, load_snapshot: Dict[ReplicaID, int]