fix(spanner): prevent fastpath tablet routing flaps

Ignore group updates with older generations so stale tablet addresses cannot overwrite newer skip state. At equal generation, retain skip-and-empty tablets unless the incoming tablet incarnation is lexicographically newer, which permits legitimate recovery while rejecting frontend-local availability flaps.

Gate endpoint recreation on active finder membership and hold the reconciliation lock through insertion so concurrent removal cannot resurrect an inactive address.

Author: rahul2393

java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/EndpointLifecycleManager.java

@@ -621,6 +621,9 @@ private void evictEndpoint(String address, EvictionReason reason) {
    * Requests that an evicted endpoint be recreated. The endpoint is created in the background and
    * probing starts immediately. The endpoint will only become eligible for location-aware routing
    * once it reaches READY state.
+   *
+   * <p>Recreation is refused when no finder currently lists the address as active. Route lookups
+   * can race with cache updates that have already removed a tablet.
    */
   void requestEndpointRecreation(String address) {
     if (isShutdown.get() || address == null || address.isEmpty()) {
@@ -635,9 +638,32 @@ void requestEndpointRecreation(String address) {
       return;
     }
 
-    logger.log(Level.FINE, "Recreating previously evicted endpoint for address: {0}", address);
-    EndpointState state = new EndpointState(address, clock.instant());
-    if (endpoints.putIfAbsent(address, state) == null) {
+    boolean stillActive = false;
+    boolean recreated = false;
+    // Check membership and insert under the reconciliation lock so active-set removal cannot
+    // complete between them and leave an inactive endpoint behind.
+    synchronized (activeAddressLock) {
+      for (Set<String> addresses : activeAddressesPerFinder.values()) {
+        if (addresses.contains(address)) {
+          stillActive = true;
+          break;
+        }
+      }
+      if (stillActive) {
+        EndpointState state = new EndpointState(address, clock.instant());
+        recreated = endpoints.putIfAbsent(address, state) == null;
+      }
+    }
+    if (!stillActive) {
+      logger.log(
+          Level.FINE,
+          "Skipping endpoint recreation for {0}: address is not in any finder's active set",
+          address);
+      return;
+    }
+
+    if (recreated) {
+      logger.log(Level.FINE, "Recreating previously evicted endpoint for address: {0}", address);
       // Schedule after putIfAbsent returns so the entry is visible to the scheduler thread.
       scheduler.submit(() -> createAndStartProbing(address));
     }

java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/KeyRangeCache.java

@@ -710,9 +710,16 @@ private class CachedGroup {
 
     void update(Group groupIn) {
       GroupSnapshot current = snapshot;
+      int generationCmp = compare(groupIn.getGeneration(), current.generation);
+      // Response-carried cache updates may arrive out of order. Older group information must not
+      // overwrite newer tablet membership and reintroduce an unavailable tablet.
+      if (generationCmp < 0) {
+        return;
+      }
+
       ByteString generation = current.generation;
       int leaderIndex = current.leaderIndex;
-      if (compare(groupIn.getGeneration(), generation) > 0) {
+      if (generationCmp > 0) {
         generation = groupIn.getGeneration();
         if (groupIn.getLeaderIndex() >= 0 && groupIn.getLeaderIndex() < groupIn.getTabletsCount()) {
           leaderIndex = groupIn.getLeaderIndex();
@@ -721,11 +728,48 @@ void update(Group groupIn) {
         }
       }
 
+      List<TabletSnapshot> tablets;
+      if (generationCmp == 0 && !current.tablets.isEmpty()) {
+        // Frontend-local location caches can disagree about availability at the same Paxos
+        // generation. Tablet incarnation is the authoritative per-tablet freshness signal.
+        tablets = mergeEqualGenerationTablets(current.tablets, groupIn);
+      } else {
+        tablets = new ArrayList<>(groupIn.getTabletsCount());
+        for (int t = 0; t < groupIn.getTabletsCount(); t++) {
+          tablets.add(new TabletSnapshot(groupIn.getTablets(t)));
+        }
+      }
+      snapshot = new GroupSnapshot(generation, leaderIndex, tablets);
+    }
+
+    /**
+     * Keeps a skipped tablet latched at an equal group generation unless fresher tablet metadata
+     * restores its address.
+     */
+    private List<TabletSnapshot> mergeEqualGenerationTablets(
+        List<TabletSnapshot> currentTablets, Group groupIn) {
+      Map<Long, TabletSnapshot> currentByUid = new HashMap<>();
+      for (TabletSnapshot tablet : currentTablets) {
+        currentByUid.put(tablet.tabletUid, tablet);
+      }
+
       List<TabletSnapshot> tablets = new ArrayList<>(groupIn.getTabletsCount());
       for (int t = 0; t < groupIn.getTabletsCount(); t++) {
-        tablets.add(new TabletSnapshot(groupIn.getTablets(t)));
+        TabletSnapshot incoming = new TabletSnapshot(groupIn.getTablets(t));
+        TabletSnapshot previous = currentByUid.get(incoming.tabletUid);
+        boolean restoresSkippedTablet =
+            previous != null
+                && previous.skip
+                && previous.serverAddress.isEmpty()
+                && !incoming.skip
+                && !incoming.serverAddress.isEmpty();
+        boolean hasNewerIncarnation =
+            restoresSkippedTablet
+                && !incoming.incarnation.isEmpty()
+                && compare(incoming.incarnation, previous.incarnation) > 0;
+        tablets.add(restoresSkippedTablet && !hasNewerIncarnation ? previous : incoming);
       }
-      snapshot = new GroupSnapshot(generation, leaderIndex, tablets);
+      return tablets;
     }
 
     RouteLookupResult lookupRoutingHint(

java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/spi/v1/EndpointLifecycleManagerTest.java

@@ -16,6 +16,7 @@
 
 package com.google.cloud.spanner.spi.v1;
 
+import static org.awaitility.Awaitility.await;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
@@ -29,7 +30,9 @@
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.Set;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.locks.LockSupport;
 import java.util.function.BooleanSupplier;
@@ -418,6 +421,80 @@ public void staleEndpointEvictedWhenNoLongerActive() throws Exception {
     assertEquals(1, manager.managedEndpointCount());
   }
 
+  @Test
+  public void requestEndpointRecreationSkippedWhenAddressNotActive() throws Exception {
+    KeyRangeCacheTest.FakeEndpointCache cache = new KeyRangeCacheTest.FakeEndpointCache();
+    manager =
+        new EndpointLifecycleManager(
+            cache, /* probeIntervalSeconds= */ 60, Duration.ofMinutes(30), Clock.systemUTC());
+
+    String finder1 = registerAddresses(manager, "server1", "server2");
+    awaitCondition(
+        "endpoints should be created",
+        () -> cache.getIfPresent("server1") != null && cache.getIfPresent("server2") != null);
+
+    // Remove server1 from the active set (stale-evict).
+    manager.updateActiveAddresses(finder1, Collections.singleton("server2"));
+    assertFalse(manager.isManaged("server1"));
+    assertNull(cache.getIfPresent("server1"));
+
+    // Route-lookup recreation must not revive a non-active address.
+    manager.requestEndpointRecreation("server1");
+    await()
+        .atMost(Duration.ofSeconds(5))
+        .until(
+            () -> !manager.isManaged("server1") && cache.getIfPresent("server1") == null);
+  }
+
+  @Test
+  public void requestEndpointRecreationDoesNotRaceActiveAddressRemoval() throws Exception {
+    KeyRangeCacheTest.FakeEndpointCache cache = new KeyRangeCacheTest.FakeEndpointCache();
+    BlockingClock clock = new BlockingClock(Instant.now());
+    manager =
+        new EndpointLifecycleManager(cache, /* probeIntervalSeconds= */ 60, Duration.ZERO, clock);
+
+    String finder = registerAddresses(manager, "server1");
+    awaitCondition(
+        "endpoint should be created in background", () -> cache.getIfPresent("server1") != null);
+    clock.advance(Duration.ofSeconds(1));
+    manager.checkIdleEviction();
+    assertFalse(manager.isManaged("server1"));
+
+    clock.blockThread("endpoint-recreation");
+    Thread recreation =
+        new Thread(() -> manager.requestEndpointRecreation("server1"), "endpoint-recreation");
+    recreation.start();
+    assertTrue("recreation should reach endpoint insertion", clock.awaitBlocked());
+
+    AtomicBoolean removalDone = new AtomicBoolean();
+    Thread removal =
+        new Thread(
+            () -> {
+              manager.updateActiveAddresses(finder, Collections.emptySet());
+              removalDone.set(true);
+            },
+            "active-address-removal");
+    removal.start();
+
+    long deadlineNanos = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+    while (!removalDone.get()
+        && removal.getState() != Thread.State.BLOCKED
+        && System.nanoTime() < deadlineNanos) {
+      LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(10));
+    }
+    assertTrue(
+        "removal should either finish or block on the active-address lock",
+        removalDone.get() || removal.getState() == Thread.State.BLOCKED);
+
+    clock.releaseBlockedThread();
+    recreation.join(TimeUnit.SECONDS.toMillis(5));
+    removal.join(TimeUnit.SECONDS.toMillis(5));
+
+    assertFalse("recreation thread should finish", recreation.isAlive());
+    assertFalse("removal thread should finish", removal.isAlive());
+    assertFalse("inactive address must not be resurrected", manager.isManaged("server1"));
+  }
+
   @Test
   public void endpointKeptIfReferencedByAnotherFinder() throws Exception {
     KeyRangeCacheTest.FakeEndpointCache cache = new KeyRangeCacheTest.FakeEndpointCache();
@@ -485,4 +562,58 @@ public Clock withZone(ZoneId zone) {
       return this;
     }
   }
+
+  /** Clock that can pause one named thread inside endpoint-state construction. */
+  private static final class BlockingClock extends Clock {
+    private Instant now;
+    private final CountDownLatch blocked = new CountDownLatch(1);
+    private final CountDownLatch release = new CountDownLatch(1);
+    private volatile String blockedThreadName;
+
+    BlockingClock(Instant now) {
+      this.now = now;
+    }
+
+    void blockThread(String threadName) {
+      this.blockedThreadName = threadName;
+    }
+
+    void advance(Duration duration) {
+      now = now.plus(duration);
+    }
+
+    boolean awaitBlocked() throws InterruptedException {
+      return blocked.await(5, TimeUnit.SECONDS);
+    }
+
+    void releaseBlockedThread() {
+      release.countDown();
+    }
+
+    @Override
+    public Instant instant() {
+      if (Thread.currentThread().getName().equals(blockedThreadName)) {
+        blocked.countDown();
+        try {
+          if (!release.await(5, TimeUnit.SECONDS)) {
+            throw new AssertionError("timed out waiting to release blocked clock");
+          }
+        } catch (InterruptedException e) {
+          Thread.currentThread().interrupt();
+          throw new AssertionError(e);
+        }
+      }
+      return now;
+    }
+
+    @Override
+    public ZoneId getZone() {
+      return ZoneId.of("UTC");
+    }
+
+    @Override
+    public Clock withZone(ZoneId zone) {
+      return this;
+    }
+  }
 }