Skip to content

fix(spanner): prevent fastpath tablet routing flaps #13803

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
rahul2393 wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix(spanner): prevent fastpath tablet routing flaps #13803

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 29 additions & 3 deletions ...cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/EndpointLifecycleManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()) {
Expand All @@ -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;
}
}
Comment on lines +641 to +656

@gemini-code-assist gemini-code-assist Bot Jul 17, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Calling clock.instant() inside the synchronized (activeAddressLock) block violates the open call design principle (calling an overridable/external method while holding a lock). Since Clock is an overridable dependency, its implementation could block or acquire other locks, potentially leading to deadlocks or scalability bottlenecks.

Moving clock.instant() outside of the synchronized block avoids holding the lock during this call. This also makes the concurrent test more robust and deterministic, as the removal thread can finish cleanly without relying on thread blocking states.

    Instant now = clock.instant();
    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, now);
        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));
}
Expand Down
50 changes: 47 additions & 3 deletions ...ner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/KeyRangeCache.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand All @@ -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++) {

@samyakpjgoogle samyakpjgoogle Jul 17, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In this logic I think we will are allowing a skip=true to come after a skip=False and after that we will again allow skip=False becuase it will have incarnation number and cache wont

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(
Expand Down
131 changes: 131 additions & 0 deletions ...d-spanner/src/test/java/com/google/cloud/spanner/spi/v1/EndpointLifecycleManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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();
Expand Down Expand Up @@ -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;
}
}
}
Loading
Loading