diff --git a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/EndpointLifecycleManager.java b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/EndpointLifecycleManager.java index 91407d46a3a0..b0790a5e05b5 100644 --- a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/EndpointLifecycleManager.java +++ b/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. + * + *

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 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)); } diff --git a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/EndpointOverloadCooldownTracker.java b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/EndpointOverloadCooldownTracker.java index de385899d38d..059b3de4d6ec 100644 --- a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/EndpointOverloadCooldownTracker.java +++ b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/EndpointOverloadCooldownTracker.java @@ -17,35 +17,78 @@ package com.google.cloud.spanner.spi.v1; import com.google.common.annotations.VisibleForTesting; +import io.grpc.Status; import java.time.Clock; import java.time.Duration; import java.time.Instant; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.LongUnaryOperator; +import java.util.function.Predicate; +import javax.annotation.Nullable; /** - * Tracks short-lived endpoint cooldowns after routed {@code RESOURCE_EXHAUSTED} failures. + * Tracks short-lived endpoint cooldowns after routed overload or availability failures. * - *

This allows later requests to try a different replica instead of immediately routing back to - * the same overloaded endpoint. + *

Server-hinted overloads and connection failures use separate escalation lanes because a short + * load-shed response must not weaken protection for an unavailable endpoint. */ -final class EndpointOverloadCooldownTracker { +final class EndpointOverloadCooldownTracker implements Predicate { @VisibleForTesting static final Duration DEFAULT_INITIAL_COOLDOWN = Duration.ofSeconds(10); @VisibleForTesting static final Duration DEFAULT_MAX_COOLDOWN = Duration.ofMinutes(1); @VisibleForTesting static final Duration DEFAULT_RESET_AFTER = Duration.ofMinutes(10); + @VisibleForTesting static final int DEFAULT_SUCCESSES_PER_REPAIR = 3; + @VisibleForTesting static final Duration DEFAULT_MIN_HINTED_COOLDOWN = Duration.ofMillis(100); + @VisibleForTesting static final Duration DEFAULT_MAX_HINTED_CLIENT_FLOOR = Duration.ofSeconds(2); + @VisibleForTesting static final Duration DEFAULT_MAX_HINTED_JITTER = Duration.ofMillis(500); + @VisibleForTesting static final int DEFAULT_MAX_TIER = 6; + + private static final Duration MIN_PROBE_RESERVATION = Duration.ofMillis(250); + private static final Duration MAX_PROBE_RESERVATION = Duration.ofMillis(500); @VisibleForTesting static final class CooldownState { - private final int consecutiveFailures; - private final Instant cooldownUntil; - private final Instant lastFailureAt; - - private CooldownState(int consecutiveFailures, Instant cooldownUntil, Instant lastFailureAt) { - this.consecutiveFailures = consecutiveFailures; - this.cooldownUntil = cooldownUntil; - this.lastFailureAt = lastFailureAt; + final int overloadFailures; + final int unavailableFailures; + final int successesTowardRepair; + final Instant overloadUntil; + final Instant unavailableUntil; + final Instant overloadProbeNotBefore; + final Instant probeReservedUntil; + @Nullable final Instant lastOverloadFailureAt; + @Nullable final Instant lastUnavailableFailureAt; + + private CooldownState( + int overloadFailures, + int unavailableFailures, + int successesTowardRepair, + Instant overloadUntil, + Instant unavailableUntil, + Instant overloadProbeNotBefore, + Instant probeReservedUntil, + @Nullable Instant lastOverloadFailureAt, + @Nullable Instant lastUnavailableFailureAt) { + this.overloadFailures = overloadFailures; + this.unavailableFailures = unavailableFailures; + this.successesTowardRepair = successesTowardRepair; + this.overloadUntil = overloadUntil; + this.unavailableUntil = unavailableUntil; + this.overloadProbeNotBefore = overloadProbeNotBefore; + this.probeReservedUntil = probeReservedUntil; + this.lastOverloadFailureAt = lastOverloadFailureAt; + this.lastUnavailableFailureAt = lastUnavailableFailureAt; + } + } + + private static final class HintedCooldown { + final Duration cooldown; + final Duration probeDelay; + + private HintedCooldown(Duration cooldown, Duration probeDelay) { + this.cooldown = cooldown; + this.probeDelay = probeDelay; } } @@ -94,6 +137,11 @@ private CooldownState(int consecutiveFailures, Instant cooldownUntil, Instant la randomLong == null ? bound -> ThreadLocalRandom.current().nextLong(bound) : randomLong; } + @Override + public boolean test(String address) { + return isCoolingDown(address); + } + boolean isCoolingDown(String address) { if (address == null || address.isEmpty()) { return false; @@ -103,33 +151,198 @@ boolean isCoolingDown(String address) { if (state == null) { return false; } - if (state.cooldownUntil.isAfter(now)) { + if (state.overloadUntil.isAfter(now) || state.unavailableUntil.isAfter(now)) { return true; } - if (Duration.between(state.lastFailureAt, now).compareTo(resetAfter) < 0) { + if (!isIdle(state, now)) { return false; } entries.remove(address, state); CooldownState current = entries.get(address); - return current != null && current.cooldownUntil.isAfter(now); + return current != null + && (current.overloadUntil.isAfter(now) || current.unavailableUntil.isAfter(now)); } void recordFailure(String address) { - if (address == null || address.isEmpty()) { + recordFailure(address, Status.Code.RESOURCE_EXHAUSTED, null); + } + + void recordFailure(String address, Status.Code statusCode, @Nullable Duration serverRetryDelay) { + if (address == null + || address.isEmpty() + || (statusCode != Status.Code.RESOURCE_EXHAUSTED + && statusCode != Status.Code.UNAVAILABLE)) { return; } Instant now = clock.instant(); entries.compute( address, (ignored, state) -> { - int consecutiveFailures = 1; - if (state != null - && Duration.between(state.lastFailureAt, now).compareTo(resetAfter) < 0) { - consecutiveFailures = state.consecutiveFailures + 1; + CooldownState current = state == null || isIdle(state, now) ? emptyState(now) : state; + if (statusCode == Status.Code.RESOURCE_EXHAUSTED) { + int failures = + nextFailureTier(current.overloadFailures, current.lastOverloadFailureAt, now); + Duration cooldown; + Duration probeDelay; + if (validRetryDelay(serverRetryDelay)) { + HintedCooldown hinted = hintedOverloadCooldown(failures, serverRetryDelay); + cooldown = hinted.cooldown; + probeDelay = hinted.probeDelay; + } else { + cooldown = cooldownForFailures(failures); + probeDelay = cooldown; + } + return new CooldownState( + failures, + current.unavailableFailures, + 0, + later(current.overloadUntil, now.plus(cooldown)), + current.unavailableUntil, + later(current.overloadProbeNotBefore, now.plus(probeDelay)), + current.probeReservedUntil, + now, + current.lastUnavailableFailureAt); + } + int failures = + nextFailureTier(current.unavailableFailures, current.lastUnavailableFailureAt, now); + return new CooldownState( + current.overloadFailures, + failures, + 0, + current.overloadUntil, + later(current.unavailableUntil, now.plus(cooldownForFailures(failures))), + current.overloadProbeNotBefore, + current.probeReservedUntil, + current.lastOverloadFailureAt, + now); + }); + } + + void recordSuccess(String address) { + if (address == null || address.isEmpty()) { + return; + } + Instant now = clock.instant(); + entries.computeIfPresent( + address, + (ignored, state) -> { + if (isIdle(state, now)) { + return null; + } + int successes = state.successesTowardRepair + 1; + if (successes < DEFAULT_SUCCESSES_PER_REPAIR) { + return copyWithTiersAndSuccesses( + state, state.overloadFailures, state.unavailableFailures, successes); + } + int overloadFailures = Math.max(0, state.overloadFailures - 1); + int unavailableFailures = Math.max(0, state.unavailableFailures - 1); + if (overloadFailures == 0 + && unavailableFailures == 0 + && !state.overloadUntil.isAfter(now) + && !state.unavailableUntil.isAfter(now)) { + return null; + } + return copyWithTiersAndSuccesses(state, overloadFailures, unavailableFailures, 0); + }); + } + + boolean tryReserveProbe(String address) { + if (address == null || address.isEmpty()) { + return false; + } + Instant now = clock.instant(); + AtomicBoolean reserved = new AtomicBoolean(); + entries.computeIfPresent( + address, + (ignored, state) -> { + if (!state.overloadUntil.isAfter(now) + || state.unavailableUntil.isAfter(now) + || state.overloadProbeNotBefore.isAfter(now) + || state.probeReservedUntil.isAfter(now)) { + return state; } - Duration cooldown = cooldownForFailures(consecutiveFailures); - return new CooldownState(consecutiveFailures, now.plus(cooldown), now); + long minMillis = MIN_PROBE_RESERVATION.toMillis(); + long rangeMillis = + MAX_PROBE_RESERVATION.toMillis() - MIN_PROBE_RESERVATION.toMillis() + 1L; + Duration reservation = Duration.ofMillis(minMillis + randomLong.applyAsLong(rangeMillis)); + reserved.set(true); + return new CooldownState( + state.overloadFailures, + state.unavailableFailures, + state.successesTowardRepair, + state.overloadUntil, + state.unavailableUntil, + state.overloadProbeNotBefore, + now.plus(reservation), + state.lastOverloadFailureAt, + state.lastUnavailableFailureAt); }); + return reserved.get(); + } + + @Nullable + Instant getLastOverloadFailureAt(String address) { + CooldownState state = entries.get(address); + return state == null ? null : state.lastOverloadFailureAt; + } + + private CooldownState emptyState(Instant now) { + return new CooldownState(0, 0, 0, now, now, now, now, null, null); + } + + private CooldownState copyWithTiersAndSuccesses( + CooldownState state, int overloadFailures, int unavailableFailures, int successes) { + return new CooldownState( + overloadFailures, + unavailableFailures, + successes, + state.overloadUntil, + state.unavailableUntil, + state.overloadProbeNotBefore, + state.probeReservedUntil, + state.lastOverloadFailureAt, + state.lastUnavailableFailureAt); + } + + private boolean isRecent(@Nullable Instant lastFailureAt, Instant now) { + return lastFailureAt != null && Duration.between(lastFailureAt, now).compareTo(resetAfter) < 0; + } + + private boolean isIdle(CooldownState state, Instant now) { + return !state.overloadUntil.isAfter(now) + && !state.unavailableUntil.isAfter(now) + && !isRecent(state.lastOverloadFailureAt, now) + && !isRecent(state.lastUnavailableFailureAt, now); + } + + private int nextFailureTier(int previousFailures, @Nullable Instant lastFailureAt, Instant now) { + if (previousFailures == 0 || !isRecent(lastFailureAt, now)) { + return 1; + } + return Math.min(DEFAULT_MAX_TIER, previousFailures + 1); + } + + private static boolean validRetryDelay(@Nullable Duration retryDelay) { + // A present zero delay uses the 100 ms floor; an absent hint uses unhinted backoff. + return retryDelay != null && !retryDelay.isNegative(); + } + + private HintedCooldown hintedOverloadCooldown(int failures, Duration serverRetryDelay) { + Duration serverFloor = + serverRetryDelay.compareTo(DEFAULT_MIN_HINTED_COOLDOWN) < 0 + ? DEFAULT_MIN_HINTED_COOLDOWN + : serverRetryDelay; + Duration clientFloor = DEFAULT_MIN_HINTED_COOLDOWN; + for (int i = 1; + i < failures && clientFloor.compareTo(DEFAULT_MAX_HINTED_CLIENT_FLOOR) < 0; + i++) { + clientFloor = min(clientFloor.multipliedBy(2), DEFAULT_MAX_HINTED_CLIENT_FLOOR); + } + Duration base = serverFloor.compareTo(clientFloor) >= 0 ? serverFloor : clientFloor; + Duration jitterLimit = min(base.dividedBy(4), DEFAULT_MAX_HINTED_JITTER); + long jitterRangeMillis = jitterLimit.toMillis() + 1L; + Duration jitter = Duration.ofMillis(randomLong.applyAsLong(jitterRangeMillis)); + return new HintedCooldown(base.plus(jitter), serverFloor.plus(jitter)); } private Duration cooldownForFailures(int failures) { @@ -147,6 +360,14 @@ private Duration cooldownForFailures(int failures) { return Duration.ofMillis(floorMillis + randomLong.applyAsLong(rangeSize)); } + private static Duration min(Duration first, Duration second) { + return first.compareTo(second) <= 0 ? first : second; + } + + private static Instant later(Instant first, Instant second) { + return first.isAfter(second) ? first : second; + } + @VisibleForTesting CooldownState getState(String address) { return entries.get(address); diff --git a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/KeyAwareChannel.java b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/KeyAwareChannel.java index c0dfdb8210fe..608d116cd935 100644 --- a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/KeyAwareChannel.java +++ b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/KeyAwareChannel.java @@ -25,7 +25,11 @@ import com.google.common.base.Ticker; import com.google.common.cache.Cache; import com.google.common.cache.CacheBuilder; +import com.google.protobuf.Any; import com.google.protobuf.ByteString; +import com.google.protobuf.InvalidProtocolBufferException; +import com.google.protobuf.util.Durations; +import com.google.rpc.RetryInfo; import com.google.spanner.v1.BeginTransactionRequest; import com.google.spanner.v1.CommitRequest; import com.google.spanner.v1.CommitResponse; @@ -43,11 +47,14 @@ import io.grpc.ManagedChannel; import io.grpc.Metadata; import io.grpc.MethodDescriptor; +import io.grpc.protobuf.ProtoUtils; +import io.grpc.protobuf.StatusProto; import io.opentelemetry.api.common.Attributes; import io.opentelemetry.api.trace.Span; import java.io.IOException; import java.lang.ref.ReferenceQueue; import java.lang.ref.SoftReference; +import java.time.Duration; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; @@ -81,6 +88,8 @@ final class KeyAwareChannel extends ManagedChannel { "google.spanner.v1.Spanner/BeginTransaction"; private static final String COMMIT_METHOD = "google.spanner.v1.Spanner/Commit"; private static final String ROLLBACK_METHOD = "google.spanner.v1.Spanner/Rollback"; + private static final Metadata.Key RETRY_INFO_KEY = + ProtoUtils.keyForProto(RetryInfo.getDefaultInstance()); private final ManagedChannel defaultChannel; private final ChannelEndpointCache endpointCache; @@ -355,7 +364,10 @@ private void clearAffinity(ByteString transactionId) { transactionAffinities.invalidate(transactionId); } - private void recordEndpointCooldown(@Nullable ChannelEndpoint endpoint) { + private void recordEndpointCooldown( + @Nullable ChannelEndpoint endpoint, + io.grpc.Status.Code statusCode, + @Nullable Metadata trailers) { if (endpoint == null) { return; } @@ -363,7 +375,65 @@ private void recordEndpointCooldown(@Nullable ChannelEndpoint endpoint) { if (defaultEndpointAddress.equals(address)) { return; } - endpointOverloadCooldowns.recordFailure(address); + endpointOverloadCooldowns.recordFailure(address, statusCode, retryDelay(statusCode, trailers)); + } + + private void recordEndpointSuccess(@Nullable ChannelEndpoint endpoint) { + if (endpoint == null) { + return; + } + String address = endpoint.getAddress(); + if (!defaultEndpointAddress.equals(address)) { + endpointOverloadCooldowns.recordSuccess(address); + } + } + + @Nullable + @VisibleForTesting + static Duration retryDelay(io.grpc.Status.Code statusCode, @Nullable Metadata trailers) { + if (trailers == null) { + return null; + } + Duration retryDelay = null; + try { + retryDelay = retryDelay(trailers.get(RETRY_INFO_KEY)); + } catch (IllegalArgumentException ignored) { + // Treat malformed direct details as absent. + } + try { + com.google.rpc.Status richStatus = + StatusProto.fromStatusAndTrailers(io.grpc.Status.fromCode(statusCode), trailers); + if (richStatus != null) { + for (Any detail : richStatus.getDetailsList()) { + if (!detail.is(RetryInfo.class)) { + continue; + } + try { + Duration candidate = retryDelay(detail.unpack(RetryInfo.class)); + if (candidate != null && (retryDelay == null || candidate.compareTo(retryDelay) > 0)) { + retryDelay = candidate; + } + } catch (InvalidProtocolBufferException ignored) { + // Skip malformed details without hiding later valid RetryInfo details. + } + } + } + } catch (IllegalArgumentException ignored) { + // Malformed rich status metadata must not interfere with transport close handling. + } + return retryDelay; + } + + @Nullable + private static Duration retryDelay(@Nullable RetryInfo retryInfo) { + if (retryInfo == null + || !retryInfo.hasRetryDelay() + || !Durations.isValid(retryInfo.getRetryDelay()) + || Durations.isNegative(retryInfo.getRetryDelay())) { + return null; + } + return Duration.ofSeconds( + retryInfo.getRetryDelay().getSeconds(), retryInfo.getRetryDelay().getNanos()); } private void maybeRecordErrorPenalty( @@ -791,7 +861,7 @@ private void maybeTrackReadWriteBegin(TransactionSelector selector) { private Predicate excludedEndpoints() { if (excludedEndpoints == null) { - excludedEndpoints = parentChannel.endpointOverloadCooldowns::isCoolingDown; + excludedEndpoints = parentChannel.endpointOverloadCooldowns; } return excludedEndpoints; } @@ -932,14 +1002,17 @@ public void onMessage(ResponseT message) { @Override public void onClose(io.grpc.Status status, Metadata trailers) { - if (shouldExcludeEndpointOnRetry(status.getCode())) { + if (status.isOk()) { + call.parentChannel.recordEndpointSuccess(call.selectedEndpoint); + } else if (shouldExcludeEndpointOnRetry(status.getCode())) { call.parentChannel.maybeRecordErrorPenalty( call.selectedDatabaseScope, call.selectedEndpoint, status.getCode(), call.selectedOperationUid, call.selectedPreferLeader); - call.parentChannel.recordEndpointCooldown(call.selectedEndpoint); + call.parentChannel.recordEndpointCooldown( + call.selectedEndpoint, status.getCode(), trailers); } if (call.selectedEndpoint != null) { call.selectedEndpoint.decrementActiveRequests(); diff --git a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/KeyRangeCache.java b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/KeyRangeCache.java index 98c26381285d..d630f70339ab 100644 --- a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/KeyRangeCache.java +++ b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/KeyRangeCache.java @@ -28,6 +28,7 @@ import com.google.spanner.v1.RoutingHint; import com.google.spanner.v1.Tablet; import java.time.Duration; +import java.time.Instant; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -678,6 +679,21 @@ String debugString() { } } + private static final class TabletFreshnessBaseline { + final ByteString incarnation; + final String serverAddress; + final boolean requiresNewerIncarnationForFirstAddress; + + private TabletFreshnessBaseline( + ByteString incarnation, + String serverAddress, + boolean requiresNewerIncarnationForFirstAddress) { + this.incarnation = incarnation; + this.serverAddress = serverAddress; + this.requiresNewerIncarnationForFirstAddress = requiresNewerIncarnationForFirstAddress; + } + } + private static final class GroupSnapshot { final ByteString generation; final int leaderIndex; @@ -703,6 +719,7 @@ private class CachedGroup { final long groupUid; volatile GroupSnapshot snapshot = EMPTY_GROUP_SNAPSHOT; int refs = 1; + private final Map freshnessBaselineByTabletUid = new HashMap<>(); CachedGroup(long groupUid) { this.groupUid = groupUid; @@ -710,10 +727,18 @@ 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(); + freshnessBaselineByTabletUid.clear(); if (groupIn.getLeaderIndex() >= 0 && groupIn.getLeaderIndex() < groupIn.getTabletsCount()) { leaderIndex = groupIn.getLeaderIndex(); } else { @@ -721,11 +746,105 @@ void update(Group groupIn) { } } + List tablets; + if (generationCmp == 0) { + // 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++) { + TabletSnapshot tablet = new TabletSnapshot(groupIn.getTablets(t)); + tablets.add(tablet); + rememberFreshnessBaseline(tablet); + } + } + snapshot = new GroupSnapshot(generation, leaderIndex, tablets); + } + + /** Reconciles equal-generation tablets against generation-scoped freshness baselines. */ + private List mergeEqualGenerationTablets( + List currentTablets, Group groupIn) { + Map currentByUid = new HashMap<>(); + for (TabletSnapshot tablet : currentTablets) { + currentByUid.put(tablet.tabletUid, tablet); + } + List 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); + TabletFreshnessBaseline baseline = freshnessBaselineByTabletUid.get(incoming.tabletUid); + boolean makesRoutable = !incoming.skip && !incoming.serverAddress.isEmpty(); + int incarnationCmp = + baseline == null ? 1 : compare(incoming.incarnation, baseline.incarnation); + boolean hasNewerIncarnation = + baseline != null && !incoming.incarnation.isEmpty() && incarnationCmp > 0; + boolean hasOlderFirstAddress = + baseline != null && baseline.serverAddress.isEmpty() && incarnationCmp < 0; + boolean changesProvenAddress = + baseline != null + && !baseline.serverAddress.isEmpty() + && !incoming.serverAddress.equals(baseline.serverAddress); + boolean requiresFreshFirstAddress = + baseline != null + && baseline.serverAddress.isEmpty() + && baseline.requiresNewerIncarnationForFirstAddress; + boolean latchesFirstAddressFreshness = + baseline != null + && baseline.serverAddress.isEmpty() + && incoming.skip + && incoming.serverAddress.isEmpty() + && incarnationCmp >= 0; + // Routability depends only on incoming metadata and the freshness baseline. Intermediate + // skip, missing-address, and omission states cannot relax a proven address change. + boolean refusesStaleAddressChange = + makesRoutable + && (hasOlderFirstAddress + || ((changesProvenAddress || requiresFreshFirstAddress) + && !hasNewerIncarnation)); + if (refusesStaleAddressChange) { + if (previous != null) { + tablets.add(previous); + } + } else { + tablets.add(incoming); + if (baseline == null + || makesRoutable + || hasNewerIncarnation + || latchesFirstAddressFreshness) { + rememberFreshnessBaseline(incoming); + } + } } - snapshot = new GroupSnapshot(generation, leaderIndex, tablets); + return tablets; + } + + private void rememberFreshnessBaseline(TabletSnapshot tablet) { + TabletFreshnessBaseline baseline = freshnessBaselineByTabletUid.get(tablet.tabletUid); + int incarnationCmp = baseline == null ? 1 : compare(tablet.incarnation, baseline.incarnation); + boolean hasNewerIncarnation = incarnationCmp > 0; + ByteString incarnation = hasNewerIncarnation ? tablet.incarnation : baseline.incarnation; + String serverAddress = baseline == null ? "" : baseline.serverAddress; + boolean requiresNewerIncarnationForFirstAddress = + baseline != null && baseline.requiresNewerIncarnationForFirstAddress; + if (!tablet.serverAddress.isEmpty() && incarnationCmp >= 0) { + serverAddress = tablet.serverAddress; + requiresNewerIncarnationForFirstAddress = false; + } else if (hasNewerIncarnation) { + // Missing-address metadata can accept its first address at the same incarnation. A skipped + // empty-address baseline still requires freshness before resurrection. + requiresNewerIncarnationForFirstAddress = tablet.skip; + } else if (serverAddress.isEmpty() + && tablet.skip + && tablet.serverAddress.isEmpty() + && incarnationCmp == 0) { + requiresNewerIncarnationForFirstAddress = true; + } + freshnessBaselineByTabletUid.put( + tablet.tabletUid, + new TabletFreshnessBaseline( + incarnation, serverAddress, requiresNewerIncarnationForFirstAddress)); } RouteLookupResult lookupRoutingHint( @@ -756,7 +875,10 @@ RouteLookupResult lookupRoutingHint( selectionStats); if (selected == null) { RouteFailureReason failureReason = selectionStats.toFailureReason(); - if (failureReason == RouteFailureReason.ALL_EXCLUDED_OR_COOLDOWN) { + // A skipped or addressless tablet can make the aggregate reason NO_ROUTABLE_REPLICA even + // when every otherwise usable replica is cooling. + if (failureReason == RouteFailureReason.ALL_EXCLUDED_OR_COOLDOWN + || selectionStats.sawExcludedReplica) { TabletSnapshot preferredLeader = preferLeader ? localLeaderForScoreBias(snapshot, hasDirectedReadOptions) : null; selected = @@ -765,6 +887,7 @@ RouteLookupResult lookupRoutingHint( preferLeader, directedReadOptions, hintBuilder, + excludedEndpoints, resolvedEndpoints, preferredLeader); if (selected != null) { @@ -1035,12 +1158,16 @@ private TabletSnapshot selectScoreAwareExcludedOrCoolingDownTablet( boolean preferLeader, DirectedReadOptions directedReadOptions, RoutingHint.Builder hintBuilder, + Predicate excludedEndpoints, Map resolvedEndpoints, @javax.annotation.Nullable TabletSnapshot preferredLeader) { long operationUid = hintBuilder.getOperationUid(); List candidates = new ArrayList<>(); for (TabletSnapshot tablet : snapshot.tablets) { - if (!tablet.matches(directedReadOptions) || tablet.skip || tablet.serverAddress.isEmpty()) { + if (!tablet.matches(directedReadOptions) + || tablet.skip + || tablet.serverAddress.isEmpty() + || !excludedEndpoints.test(tablet.serverAddress)) { continue; } ChannelEndpoint endpoint = resolveEndpoint(tablet, resolvedEndpoints); @@ -1056,6 +1183,32 @@ private TabletSnapshot selectScoreAwareExcludedOrCoolingDownTablet( if (candidates.isEmpty()) { return null; } + if (excludedEndpoints instanceof EndpointOverloadCooldownTracker) { + EndpointOverloadCooldownTracker cooldowns = + (EndpointOverloadCooldownTracker) excludedEndpoints; + candidates.sort( + (first, second) -> { + Instant firstFailure = cooldowns.getLastOverloadFailureAt(first.tablet.serverAddress); + Instant secondFailure = + cooldowns.getLastOverloadFailureAt(second.tablet.serverAddress); + if (firstFailure == null) { + return secondFailure == null ? 0 : 1; + } + if (secondFailure == null) { + return -1; + } + int failureComparison = firstFailure.compareTo(secondFailure); + return failureComparison != 0 + ? failureComparison + : Double.compare(first.selectionCost, second.selectionCost); + }); + for (EligibleReplica candidate : candidates) { + if (cooldowns.tryReserveProbe(candidate.tablet.serverAddress)) { + return candidate.tablet; + } + } + return null; + } return selectEligibleReplica(candidates).tablet; } diff --git a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/spi/v1/EndpointLifecycleManagerTest.java b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/spi/v1/EndpointLifecycleManagerTest.java index 341974f17fd6..009013f813fd 100644 --- a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/spi/v1/EndpointLifecycleManagerTest.java +++ b/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; + } + } } diff --git a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/spi/v1/EndpointOverloadCooldownTrackerTest.java b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/spi/v1/EndpointOverloadCooldownTrackerTest.java new file mode 100644 index 000000000000..ebae87cd4b50 --- /dev/null +++ b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/spi/v1/EndpointOverloadCooldownTrackerTest.java @@ -0,0 +1,250 @@ +/* + * Copyright 2026 Google LLC + * + * 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.google.cloud.spanner.spi.v1; + +import static com.google.common.truth.Truth.assertThat; + +import io.grpc.Status; +import java.time.Clock; +import java.time.Duration; +import java.time.Instant; +import java.time.ZoneId; +import java.time.ZoneOffset; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +@RunWith(JUnit4.class) +public class EndpointOverloadCooldownTrackerTest { + + private static final String ADDRESS = "server:1234"; + + private final MutableClock clock = new MutableClock(); + private final EndpointOverloadCooldownTracker tracker = + new EndpointOverloadCooldownTracker( + Duration.ofSeconds(10), + Duration.ofMinutes(1), + Duration.ofMinutes(10), + clock, + ignored -> 0L); + + @Test + public void hintedResourceExhaustedHonorsServerDelay() { + tracker.recordFailure(ADDRESS, Status.Code.RESOURCE_EXHAUSTED, Duration.ofMillis(100)); + + assertThat(tracker.isCoolingDown(ADDRESS)).isTrue(); + clock.advance(Duration.ofMillis(99)); + assertThat(tracker.isCoolingDown(ADDRESS)).isTrue(); + clock.advance(Duration.ofMillis(1)); + assertThat(tracker.isCoolingDown(ADDRESS)).isFalse(); + } + + @Test + public void zeroRetryDelayUsesMinimumHintedCooldown() { + tracker.recordFailure(ADDRESS, Status.Code.RESOURCE_EXHAUSTED, Duration.ZERO); + + clock.advance(Duration.ofMillis(99)); + assertThat(tracker.isCoolingDown(ADDRESS)).isTrue(); + clock.advance(Duration.ofMillis(1)); + assertThat(tracker.isCoolingDown(ADDRESS)).isFalse(); + } + + @Test + public void hintedResourceExhaustedAddsBoundedPositiveJitter() { + EndpointOverloadCooldownTracker jitteredTracker = + new EndpointOverloadCooldownTracker( + Duration.ofSeconds(10), + Duration.ofMinutes(1), + Duration.ofMinutes(10), + clock, + bound -> bound - 1L); + + jitteredTracker.recordFailure(ADDRESS, Status.Code.RESOURCE_EXHAUSTED, Duration.ofMillis(100)); + + clock.advance(Duration.ofMillis(124)); + assertThat(jitteredTracker.isCoolingDown(ADDRESS)).isTrue(); + clock.advance(Duration.ofMillis(1)); + assertThat(jitteredTracker.isCoolingDown(ADDRESS)).isFalse(); + } + + @Test + public void hintedResourceExhaustedEscalatesOnlyToShortClientCap() { + for (int i = 0; i < 20; i++) { + tracker.recordFailure(ADDRESS, Status.Code.RESOURCE_EXHAUSTED, Duration.ofMillis(100)); + } + + assertThat(tracker.getState(ADDRESS).overloadFailures) + .isEqualTo(EndpointOverloadCooldownTracker.DEFAULT_MAX_TIER); + clock.advance(Duration.ofMillis(1999)); + assertThat(tracker.isCoolingDown(ADDRESS)).isTrue(); + clock.advance(Duration.ofMillis(1)); + assertThat(tracker.isCoolingDown(ADDRESS)).isFalse(); + } + + @Test + public void unhintedResourceExhaustedKeepsLongBackoff() { + tracker.recordFailure(ADDRESS, Status.Code.RESOURCE_EXHAUSTED, null); + tracker.recordFailure(ADDRESS, Status.Code.RESOURCE_EXHAUSTED, null); + + clock.advance(Duration.ofMillis(9999)); + assertThat(tracker.isCoolingDown(ADDRESS)).isTrue(); + clock.advance(Duration.ofMillis(1)); + assertThat(tracker.isCoolingDown(ADDRESS)).isFalse(); + } + + @Test + public void threeSuccessesRepairEachLaneWithoutShorteningActiveDeadline() { + tracker.recordFailure(ADDRESS, Status.Code.RESOURCE_EXHAUSTED, Duration.ofMillis(100)); + tracker.recordFailure(ADDRESS, Status.Code.RESOURCE_EXHAUSTED, Duration.ofMillis(100)); + tracker.recordFailure(ADDRESS, Status.Code.UNAVAILABLE, null); + Instant overloadUntil = tracker.getState(ADDRESS).overloadUntil; + Instant unavailableUntil = tracker.getState(ADDRESS).unavailableUntil; + + tracker.recordSuccess(ADDRESS); + tracker.recordSuccess(ADDRESS); + tracker.recordSuccess(ADDRESS); + + EndpointOverloadCooldownTracker.CooldownState repaired = tracker.getState(ADDRESS); + assertThat(repaired.overloadFailures).isEqualTo(1); + assertThat(repaired.unavailableFailures).isEqualTo(0); + assertThat(repaired.overloadUntil).isEqualTo(overloadUntil); + assertThat(repaired.unavailableUntil).isEqualTo(unavailableUntil); + } + + @Test + public void trackedFailureResetsSuccessRepairProgress() { + tracker.recordFailure(ADDRESS, Status.Code.RESOURCE_EXHAUSTED, Duration.ofMillis(100)); + tracker.recordSuccess(ADDRESS); + tracker.recordSuccess(ADDRESS); + + tracker.recordFailure(ADDRESS, Status.Code.RESOURCE_EXHAUSTED, Duration.ofMillis(100)); + + assertThat(tracker.getState(ADDRESS).successesTowardRepair).isEqualTo(0); + } + + @Test + public void unavailableAndResourceExhaustedUseIndependentLanes() { + tracker.recordFailure(ADDRESS, Status.Code.RESOURCE_EXHAUSTED, Duration.ofMillis(100)); + tracker.recordFailure(ADDRESS, Status.Code.UNAVAILABLE, null); + + EndpointOverloadCooldownTracker.CooldownState state = tracker.getState(ADDRESS); + assertThat(state.overloadFailures).isEqualTo(1); + assertThat(state.unavailableFailures).isEqualTo(1); + clock.advance(Duration.ofMillis(100)); + assertThat(tracker.isCoolingDown(ADDRESS)).isTrue(); + clock.advance(Duration.ofMillis(4900)); + assertThat(tracker.isCoolingDown(ADDRESS)).isFalse(); + } + + @Test + public void unavailableIgnoresShortCooldownHints() { + tracker.recordFailure(ADDRESS, Status.Code.UNAVAILABLE, Duration.ofHours(1)); + + clock.advance(Duration.ofMillis(4999)); + assertThat(tracker.isCoolingDown(ADDRESS)).isTrue(); + clock.advance(Duration.ofMillis(1)); + assertThat(tracker.isCoolingDown(ADDRESS)).isFalse(); + } + + @Test + public void idleStateResetsAfterTenMinutes() { + tracker.recordFailure(ADDRESS, Status.Code.RESOURCE_EXHAUSTED, Duration.ofMillis(100)); + tracker.recordFailure(ADDRESS, Status.Code.RESOURCE_EXHAUSTED, Duration.ofMillis(100)); + + clock.advance(Duration.ofMinutes(10)); + assertThat(tracker.isCoolingDown(ADDRESS)).isFalse(); + assertThat(tracker.getState(ADDRESS)).isNull(); + + tracker.recordFailure(ADDRESS, Status.Code.RESOURCE_EXHAUSTED, Duration.ofMillis(100)); + assertThat(tracker.getState(ADDRESS).overloadFailures).isEqualTo(1); + } + + @Test + public void failureAfterIdleWindowResetsBothLanesWithoutPriorLookup() { + tracker.recordFailure(ADDRESS, Status.Code.RESOURCE_EXHAUSTED, Duration.ofMillis(100)); + tracker.recordFailure(ADDRESS, Status.Code.RESOURCE_EXHAUSTED, Duration.ofMillis(100)); + tracker.recordFailure(ADDRESS, Status.Code.UNAVAILABLE, null); + tracker.recordFailure(ADDRESS, Status.Code.UNAVAILABLE, null); + clock.advance(Duration.ofMinutes(10)); + + tracker.recordFailure(ADDRESS, Status.Code.RESOURCE_EXHAUSTED, Duration.ofMillis(100)); + + assertThat(tracker.getState(ADDRESS).overloadFailures).isEqualTo(1); + assertThat(tracker.getState(ADDRESS).unavailableFailures).isEqualTo(0); + } + + @Test + public void hintedOverloadProbeWaitsForServerDelayAndReservesAddress() { + for (int i = 0; i < EndpointOverloadCooldownTracker.DEFAULT_MAX_TIER; i++) { + tracker.recordFailure(ADDRESS, Status.Code.RESOURCE_EXHAUSTED, Duration.ofMillis(100)); + } + + clock.advance(Duration.ofMillis(99)); + assertThat(tracker.tryReserveProbe(ADDRESS)).isFalse(); + clock.advance(Duration.ofMillis(1)); + assertThat(tracker.tryReserveProbe(ADDRESS)).isTrue(); + assertThat(tracker.tryReserveProbe(ADDRESS)).isFalse(); + clock.advance(Duration.ofMillis(249)); + assertThat(tracker.tryReserveProbe(ADDRESS)).isFalse(); + clock.advance(Duration.ofMillis(1)); + assertThat(tracker.tryReserveProbe(ADDRESS)).isTrue(); + } + + @Test + public void unavailableLanePreventsEarlyOverloadProbe() { + for (int i = 0; i < EndpointOverloadCooldownTracker.DEFAULT_MAX_TIER; i++) { + tracker.recordFailure(ADDRESS, Status.Code.RESOURCE_EXHAUSTED, Duration.ofMillis(100)); + } + tracker.recordFailure(ADDRESS, Status.Code.UNAVAILABLE, null); + + clock.advance(Duration.ofMillis(100)); + assertThat(tracker.tryReserveProbe(ADDRESS)).isFalse(); + } + + @Test + public void untrackedStatusAndEmptyAddressDoNothing() { + tracker.recordFailure(ADDRESS, Status.Code.ABORTED, Duration.ofMillis(100)); + tracker.recordFailure("", Status.Code.RESOURCE_EXHAUSTED, Duration.ofMillis(100)); + tracker.recordSuccess(""); + + assertThat(tracker.isCoolingDown(ADDRESS)).isFalse(); + assertThat(tracker.getState(ADDRESS)).isNull(); + } + + private static final class MutableClock extends Clock { + private Instant now = Instant.ofEpochSecond(100); + + @Override + public ZoneId getZone() { + return ZoneOffset.UTC; + } + + @Override + public Clock withZone(ZoneId zone) { + return this; + } + + @Override + public Instant instant() { + return now; + } + + private void advance(Duration duration) { + now = now.plus(duration); + } + } +} diff --git a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/spi/v1/KeyAwareChannelTest.java b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/spi/v1/KeyAwareChannelTest.java index af9366020da9..7d55b286342b 100644 --- a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/spi/v1/KeyAwareChannelTest.java +++ b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/spi/v1/KeyAwareChannelTest.java @@ -26,11 +26,14 @@ import com.google.cloud.spanner.XGoogSpannerRequestId; import com.google.common.base.Ticker; import com.google.common.testing.FakeTicker; +import com.google.protobuf.Any; import com.google.protobuf.ByteString; import com.google.protobuf.Empty; import com.google.protobuf.ListValue; import com.google.protobuf.TextFormat; import com.google.protobuf.Value; +import com.google.protobuf.util.Durations; +import com.google.rpc.RetryInfo; import com.google.spanner.v1.BeginTransactionRequest; import com.google.spanner.v1.CacheUpdate; import com.google.spanner.v1.CommitRequest; @@ -57,6 +60,8 @@ import io.grpc.Metadata; import io.grpc.MethodDescriptor; import io.grpc.Status; +import io.grpc.protobuf.ProtoUtils; +import io.grpc.protobuf.StatusProto; import java.io.IOException; import java.time.Clock; import java.time.Duration; @@ -508,6 +513,186 @@ public void resourceExhaustedRoutedEndpointIsAvoidedOnRetry() throws Exception { assertThat(harness.channel.isCoolingDown("server-a:1234")).isTrue(); } + @Test + public void retryDelayExtractsDirectRetryInfo() { + Metadata trailers = new Metadata(); + trailers.put( + ProtoUtils.keyForProto(RetryInfo.getDefaultInstance()), + RetryInfo.newBuilder().setRetryDelay(Durations.fromMillis(100)).build()); + + assertThat(KeyAwareChannel.retryDelay(Status.Code.RESOURCE_EXHAUSTED, trailers)) + .isEqualTo(Duration.ofMillis(100)); + } + + @Test + public void retryDelayExtractsRichStatusRetryInfo() { + Metadata trailers = + StatusProto.toStatusRuntimeException( + com.google.rpc.Status.newBuilder() + .setCode(Status.Code.RESOURCE_EXHAUSTED.value()) + .addDetails( + Any.pack( + RetryInfo.newBuilder() + .setRetryDelay(Durations.fromMillis(200)) + .build())) + .build()) + .getTrailers(); + + assertThat(trailers).isNotNull(); + assertThat(KeyAwareChannel.retryDelay(Status.Code.RESOURCE_EXHAUSTED, trailers)) + .isEqualTo(Duration.ofMillis(200)); + } + + @Test + public void retryDelayUsesLargestDirectOrRichRetryInfo() { + Metadata trailers = + StatusProto.toStatusRuntimeException( + com.google.rpc.Status.newBuilder() + .setCode(Status.Code.RESOURCE_EXHAUSTED.value()) + .addDetails( + Any.pack( + RetryInfo.newBuilder() + .setRetryDelay(Durations.fromMillis(200)) + .build())) + .build()) + .getTrailers(); + assertThat(trailers).isNotNull(); + trailers.put( + ProtoUtils.keyForProto(RetryInfo.getDefaultInstance()), + RetryInfo.newBuilder().setRetryDelay(Durations.fromMillis(100)).build()); + + assertThat(KeyAwareChannel.retryDelay(Status.Code.RESOURCE_EXHAUSTED, trailers)) + .isEqualTo(Duration.ofMillis(200)); + } + + @Test + public void retryDelaySkipsMalformedRichDetailBeforeValidRetryInfo() { + Metadata trailers = + StatusProto.toStatusRuntimeException( + com.google.rpc.Status.newBuilder() + .setCode(Status.Code.RESOURCE_EXHAUSTED.value()) + .addDetails( + Any.newBuilder() + .setTypeUrl("type.googleapis.com/google.rpc.RetryInfo") + .setValue(ByteString.copyFrom(new byte[] {(byte) 0xff})) + .build()) + .addDetails( + Any.pack( + RetryInfo.newBuilder() + .setRetryDelay(Durations.fromMillis(200)) + .build())) + .build()) + .getTrailers(); + + assertThat(trailers).isNotNull(); + assertThat(KeyAwareChannel.retryDelay(Status.Code.RESOURCE_EXHAUSTED, trailers)) + .isEqualTo(Duration.ofMillis(200)); + } + + @Test + public void retryDelayRetainsValidRetryInfoBeforeMalformedRichDetail() { + Metadata trailers = + StatusProto.toStatusRuntimeException( + com.google.rpc.Status.newBuilder() + .setCode(Status.Code.RESOURCE_EXHAUSTED.value()) + .addDetails( + Any.pack( + RetryInfo.newBuilder() + .setRetryDelay(Durations.fromMillis(200)) + .build())) + .addDetails( + Any.newBuilder() + .setTypeUrl("type.googleapis.com/google.rpc.RetryInfo") + .setValue(ByteString.copyFrom(new byte[] {(byte) 0xff})) + .build()) + .build()) + .getTrailers(); + + assertThat(trailers).isNotNull(); + assertThat(KeyAwareChannel.retryDelay(Status.Code.RESOURCE_EXHAUSTED, trailers)) + .isEqualTo(Duration.ofMillis(200)); + } + + @Test + public void retryDelayTreatsMalformedDirectAndRichDetailsAsAbsent() { + Metadata malformedDirect = new Metadata(); + malformedDirect.put( + Metadata.Key.of( + ProtoUtils.keyForProto(RetryInfo.getDefaultInstance()).name(), + Metadata.BINARY_BYTE_MARSHALLER), + new byte[] {(byte) 0xff}); + Metadata malformedRich = new Metadata(); + malformedRich.put( + Metadata.Key.of("grpc-status-details-bin", Metadata.BINARY_BYTE_MARSHALLER), + new byte[] {(byte) 0xff}); + + assertThat(KeyAwareChannel.retryDelay(Status.Code.RESOURCE_EXHAUSTED, malformedDirect)) + .isNull(); + assertThat(KeyAwareChannel.retryDelay(Status.Code.RESOURCE_EXHAUSTED, malformedRich)).isNull(); + } + + @Test + public void routedFailurePassesRetryInfoToCooldownTracker() throws Exception { + Instant now = Instant.ofEpochSecond(100); + EndpointOverloadCooldownTracker tracker = + new EndpointOverloadCooldownTracker( + Duration.ofSeconds(10), + Duration.ofMinutes(1), + Duration.ofMinutes(10), + Clock.fixed(now, ZoneOffset.UTC), + ignored -> 0L); + TestHarness harness = createHarness(tracker); + seedCache(harness, createLeaderAndReplicaCacheUpdate()); + ClientCall call = + harness.channel.newCall( + SpannerGrpc.getExecuteSqlMethod(), retryCallOptions(retryRequestId(6L))); + call.start(new CapturingListener(), new Metadata()); + call.sendMessage( + ExecuteSqlRequest.newBuilder() + .setSession(SESSION) + .setRoutingHint(RoutingHint.newBuilder().setKey(bytes("b"))) + .build()); + @SuppressWarnings("unchecked") + RecordingClientCall delegate = + (RecordingClientCall) + harness.endpointCache.latestCallForAddress("server-a:1234"); + Metadata trailers = new Metadata(); + trailers.put( + ProtoUtils.keyForProto(RetryInfo.getDefaultInstance()), + RetryInfo.newBuilder().setRetryDelay(Durations.fromMillis(100)).build()); + + delegate.emitOnClose(Status.RESOURCE_EXHAUSTED, trailers); + + EndpointOverloadCooldownTracker.CooldownState state = tracker.getState("server-a:1234"); + assertThat(state.overloadFailures).isEqualTo(1); + assertThat(state.overloadUntil).isEqualTo(now.plusMillis(100)); + } + + @Test + public void successfulRoutedCloseAdvancesCooldownRepair() throws Exception { + EndpointOverloadCooldownTracker tracker = createDeterministicCooldownTracker(); + TestHarness harness = createHarness(tracker); + seedCache(harness, createLeaderAndReplicaCacheUpdate()); + ClientCall call = + harness.channel.newCall( + SpannerGrpc.getExecuteSqlMethod(), retryCallOptions(retryRequestId(7L))); + call.start(new CapturingListener(), new Metadata()); + call.sendMessage( + ExecuteSqlRequest.newBuilder() + .setSession(SESSION) + .setRoutingHint(RoutingHint.newBuilder().setKey(bytes("b"))) + .build()); + @SuppressWarnings("unchecked") + RecordingClientCall delegate = + (RecordingClientCall) + harness.endpointCache.latestCallForAddress("server-a:1234"); + tracker.recordFailure("server-a:1234", Status.Code.RESOURCE_EXHAUSTED, Duration.ofMillis(100)); + + delegate.emitOnClose(Status.OK, new Metadata()); + + assertThat(tracker.getState("server-a:1234").successesTowardRepair).isEqualTo(1); + } + @Test public void resourceExhaustedOrUnavailableRoutedEndpointRecordsErrorPenalty() throws Exception { assertRoutedEndpointErrorPenaltyRecorded(Status.RESOURCE_EXHAUSTED, 101L); @@ -580,7 +765,7 @@ public void resourceExhaustedAffinityEndpointIsAvoidedForSubsequentTransactionRe } @Test - public void resourceExhaustedRoutedEndpointRetriesSameReplicaWhenSingleReplicaIsExcluded() + public void resourceExhaustedRoutedEndpointDoesNotProbeBeforeUnhintedCooldownExpires() throws Exception { TestHarness harness = createHarness(); CallOptions retryCallOptions = retryCallOptions(3L); @@ -610,14 +795,13 @@ public void resourceExhaustedRoutedEndpointRetriesSameReplicaWhenSingleReplicaIs secondCall.start(new CapturingListener(), new Metadata()); secondCall.sendMessage(request); - assertThat(harness.endpointCache.callCountForAddress("server-a:1234")).isEqualTo(2); - assertThat(harness.defaultManagedChannel.callCount()).isEqualTo(1); + assertThat(harness.endpointCache.callCountForAddress("server-a:1234")).isEqualTo(1); + assertThat(harness.defaultManagedChannel.callCount()).isEqualTo(2); } @Test - public void - resourceExhaustedRoutedEndpointRetriesLowestCostExcludedReplicaWhenAllReplicasExcluded() - throws Exception { + public void resourceExhaustedRoutedEndpointsDoNotProbeBeforeUnhintedCooldownExpires() + throws Exception { TestHarness harness = createHarness(); seedCache(harness, createLeaderAndReplicaCacheUpdate()); CallOptions retryCallOptions = retryCallOptions(100L); @@ -660,8 +844,9 @@ public void resourceExhaustedRoutedEndpointRetriesSameReplicaWhenSingleReplicaIs thirdCall.start(new CapturingListener(), new Metadata()); thirdCall.sendMessage(request); - assertThat(harness.endpointCache.callCountForAddress("server-a:1234")).isEqualTo(2); + assertThat(harness.endpointCache.callCountForAddress("server-a:1234")).isEqualTo(1); assertThat(harness.endpointCache.callCountForAddress("server-b:1234")).isEqualTo(1); + assertThat(harness.defaultManagedChannel.callCount()).isEqualTo(2); } @Test diff --git a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/spi/v1/KeyRangeCacheTest.java b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/spi/v1/KeyRangeCacheTest.java index b7645f044a13..4aeeebadc654 100644 --- a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/spi/v1/KeyRangeCacheTest.java +++ b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/spi/v1/KeyRangeCacheTest.java @@ -17,6 +17,7 @@ package com.google.cloud.spanner.spi.v1; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; @@ -33,9 +34,14 @@ import io.grpc.ConnectivityState; import io.grpc.ManagedChannel; import io.grpc.MethodDescriptor; +import java.time.Clock; import java.time.Duration; +import java.time.Instant; +import java.time.ZoneId; +import java.time.ZoneOffset; import java.util.HashMap; import java.util.Map; +import java.util.Set; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import javax.annotation.Nullable; @@ -214,6 +220,91 @@ public void lookupRoutingHintUsesLowestScoreWhenAllCandidatesAreExcludedOrCoolin assertEquals(KeyRangeCache.RouteFailureReason.NONE, result.failureReason); } + @Test + public void lookupRoutingHintRescuesOldestCoolingReplicaWhenAnotherTabletIsSkipped() { + FakeEndpointCache endpointCache = new FakeEndpointCache(); + KeyRangeCache cache = new KeyRangeCache(endpointCache); + cache.addRanges( + CacheUpdate.newBuilder() + .addRange( + Range.newBuilder() + .setStartKey(bytes("a")) + .setLimitKey(bytes("z")) + .setGroupUid(5) + .setSplitId(1) + .setGeneration(bytes("1"))) + .addGroup( + Group.newBuilder() + .setGroupUid(5) + .setGeneration(bytes("1")) + .setLeaderIndex(1) + .addTablets( + Tablet.newBuilder() + .setTabletUid(1) + .setIncarnation(bytes("1")) + .setSkip(true)) + .addTablets( + Tablet.newBuilder() + .setTabletUid(2) + .setServerAddress("server1") + .setIncarnation(bytes("1"))) + .addTablets( + Tablet.newBuilder() + .setTabletUid(3) + .setServerAddress("server2") + .setIncarnation(bytes("1")))) + .build()); + endpointCache.get("server1"); + endpointCache.get("server2"); + MutableClock clock = new MutableClock(); + EndpointOverloadCooldownTracker cooldowns = + new EndpointOverloadCooldownTracker( + Duration.ofSeconds(10), + Duration.ofMinutes(1), + Duration.ofMinutes(10), + clock, + ignored -> 0L); + for (int i = 0; i < 2; i++) { + cooldowns.recordFailure( + "server1", io.grpc.Status.Code.RESOURCE_EXHAUSTED, Duration.ofMillis(100)); + } + clock.advance(Duration.ofMillis(10)); + for (int i = 0; i < 2; i++) { + cooldowns.recordFailure( + "server2", io.grpc.Status.Code.RESOURCE_EXHAUSTED, Duration.ofMillis(100)); + } + clock.advance(Duration.ofMillis(100)); + + KeyRangeCache.RouteLookupResult first = + cache.lookupRoutingHint( + true, + KeyRangeCache.RangeMode.COVERING_SPLIT, + DirectedReadOptions.getDefaultInstance(), + RoutingHint.newBuilder().setKey(bytes("a")), + cooldowns); + KeyRangeCache.RouteLookupResult second = + cache.lookupRoutingHint( + true, + KeyRangeCache.RangeMode.COVERING_SPLIT, + DirectedReadOptions.getDefaultInstance(), + RoutingHint.newBuilder().setKey(bytes("a")), + cooldowns); + KeyRangeCache.RouteLookupResult third = + cache.lookupRoutingHint( + true, + KeyRangeCache.RangeMode.COVERING_SPLIT, + DirectedReadOptions.getDefaultInstance(), + RoutingHint.newBuilder().setKey(bytes("a")), + cooldowns); + + assertNotNull(first.endpoint); + assertEquals("server1", first.endpoint.getAddress()); + assertNotNull(second.endpoint); + assertEquals("server2", second.endpoint.getAddress()); + assertNull(third.endpoint); + assertEquals(KeyRangeCache.RouteFailureReason.NO_ROUTABLE_REPLICA, third.failureReason); + } + @Test public void lookupRoutingHintReportsNoReadyReplica() { FakeEndpointCache endpointCache = new FakeEndpointCache(); @@ -1068,8 +1159,613 @@ boolean wasRecentlyEvictedTransientFailure(String address) { } } + private static final class MutableClock extends Clock { + private Instant now = Instant.ofEpochSecond(100); + + @Override + public ZoneId getZone() { + return ZoneOffset.UTC; + } + + @Override + public Clock withZone(ZoneId zone) { + return this; + } + + @Override + public Instant instant() { + return now; + } + + private void advance(Duration duration) { + now = now.plus(duration); + } + } + + @Test + public void staleGenerationGroupUpdateDoesNotOverwriteTablets() { + FakeEndpointCache endpointCache = new FakeEndpointCache(); + KeyRangeCache cache = new KeyRangeCache(endpointCache); + + // Newer generation: dead tablet skip+empty, leader on survivor. + cache.addRanges( + CacheUpdate.newBuilder() + .addRange( + Range.newBuilder() + .setStartKey(bytes("a")) + .setLimitKey(bytes("z")) + .setGroupUid(5) + .setSplitId(1) + .setGeneration(bytes("2"))) + .addGroup( + Group.newBuilder() + .setGroupUid(5) + .setGeneration(bytes("2")) + .setLeaderIndex(1) + .addTablets(Tablet.newBuilder().setTabletUid(1).setSkip(true).setDistance(0)) + .addTablets( + Tablet.newBuilder() + .setTabletUid(2) + .setServerAddress("survivor") + .setIncarnation(bytes("2")) + .setDistance(0))) + .build()); + + // Older generation still names the isolated host — must not re-inject it. + cache.addRanges( + CacheUpdate.newBuilder() + .addRange( + Range.newBuilder() + .setStartKey(bytes("a")) + .setLimitKey(bytes("z")) + .setGroupUid(5) + .setSplitId(1) + .setGeneration(bytes("1"))) + .addGroup( + Group.newBuilder() + .setGroupUid(5) + .setGeneration(bytes("1")) + .setLeaderIndex(0) + .addTablets( + Tablet.newBuilder() + .setTabletUid(1) + .setServerAddress("isolated-dead") + .setIncarnation(bytes("1")) + .setDistance(0) + .setSkip(false)) + .addTablets( + Tablet.newBuilder() + .setTabletUid(2) + .setServerAddress("survivor") + .setIncarnation(bytes("1")) + .setDistance(0))) + .build()); + + Set active = cache.getActiveAddresses(); + assertTrue("survivor should remain active", active.contains("survivor")); + assertFalse( + "stale generation must not re-inject isolated-dead into active addresses: " + active, + active.contains("isolated-dead")); + } + + @Test + public void equalGenerationWithNewerIncarnationRestoresSkippedEmptyTablet() { + FakeEndpointCache endpointCache = new FakeEndpointCache(); + KeyRangeCache cache = new KeyRangeCache(endpointCache); + + addSingleTabletGroup(cache, "1", true, "", ""); + addSingleTabletGroup(cache, "1", false, "recovered", "1"); + endpointCache.get("recovered"); + + assertTrue(cache.getActiveAddresses().contains("recovered")); + RoutingHint.Builder hint = RoutingHint.newBuilder().setKey(bytes("a")); + ChannelEndpoint selected = + cache.fillRoutingHint( + /* preferLeader= */ true, + KeyRangeCache.RangeMode.COVERING_SPLIT, + DirectedReadOptions.getDefaultInstance(), + hint); + assertNotNull(selected); + assertEquals("recovered", selected.getAddress()); + assertEquals(0, hint.getSkippedTabletUidCount()); + } + + @Test + public void equalGenerationWithSameIncarnationDoesNotRestoreSkippedEmptyTablet() { + FakeEndpointCache endpointCache = new FakeEndpointCache(); + KeyRangeCache cache = new KeyRangeCache(endpointCache); + + addSingleTabletGroup(cache, "1", true, "", "1"); + addSingleTabletGroup(cache, "1", false, "isolated-dead", "1"); + + assertFalse( + "equal-gen unskip with the same incarnation must remain latched", + cache.getActiveAddresses().contains("isolated-dead")); + } + + @Test + public void equalGenerationWithEmptyIncarnationDoesNotRestoreSkippedEmptyTablet() { + FakeEndpointCache endpointCache = new FakeEndpointCache(); + KeyRangeCache cache = new KeyRangeCache(endpointCache); + + addSingleTabletGroup(cache, "1", true, "", ""); + addSingleTabletGroup(cache, "1", false, "isolated-dead", ""); + + assertFalse( + "equal-gen unskip with an empty incarnation must remain latched", + cache.getActiveAddresses().contains("isolated-dead")); + } + + @Test + public void equalGenerationWithOlderIncarnationDoesNotRestoreSkippedEmptyTablet() { + FakeEndpointCache endpointCache = new FakeEndpointCache(); + KeyRangeCache cache = new KeyRangeCache(endpointCache); + + addSingleTabletGroup(cache, "1", true, "", "2"); + addSingleTabletGroup(cache, "1", false, "isolated-dead", "1"); + + assertFalse( + "equal-gen unskip with an older incarnation must remain latched", + cache.getActiveAddresses().contains("isolated-dead")); + } + + @Test + public void equalGenerationTreatsNonEmptyIncarnationAsNewerThanEmpty() { + assertTrue( + restoresSkippedTablet(ByteString.EMPTY, ByteString.copyFrom(new byte[] {(byte) 0x00}))); + } + + @Test + public void equalGenerationTreatsProperPrefixExtensionAsNewerIncarnation() { + assertTrue( + restoresSkippedTablet( + ByteString.copyFrom(new byte[] {(byte) 0x01}), + ByteString.copyFrom(new byte[] {(byte) 0x01, (byte) 0x00}))); + } + + @Test + public void equalGenerationComparesLastIncarnationByteUnsigned() { + assertTrue( + restoresSkippedTablet( + ByteString.copyFrom(new byte[] {(byte) 0x01, (byte) 0x7f}), + ByteString.copyFrom(new byte[] {(byte) 0x01, (byte) 0x80}))); + } + + @Test + public void equalGenerationRejectsLongerLexicographicallySmallerIncarnation() { + assertFalse( + restoresSkippedTablet( + ByteString.copyFrom(new byte[] {(byte) 0x02}), + ByteString.copyFrom(new byte[] {(byte) 0x01, (byte) 0xff}))); + } + + @Test + public void equalGenerationSkipPreservesIncarnationAndRefusesReplay() { + KeyRangeCache cache = new KeyRangeCache(new FakeEndpointCache()); + ByteString incarnation = ByteString.copyFrom(new byte[] {(byte) 0x01}); + + addSingleTabletGroup(cache, "1", false, "healthy", incarnation); + addSingleTabletGroup(cache, "1", true, "", ByteString.EMPTY); + addSingleTabletGroup(cache, "1", false, "replayed", incarnation); + + assertTrue(cache.getActiveAddresses().isEmpty()); + } + + @Test + public void equalGenerationSkipPreservesIncarnationAndAcceptsNewerRecovery() { + KeyRangeCache cache = new KeyRangeCache(new FakeEndpointCache()); + ByteString incarnation = ByteString.copyFrom(new byte[] {(byte) 0x01}); + + addSingleTabletGroup(cache, "1", false, "healthy", incarnation); + addSingleTabletGroup(cache, "1", true, "", ByteString.EMPTY); + addSingleTabletGroup(cache, "1", false, "replayed", incarnation); + assertTrue(cache.getActiveAddresses().isEmpty()); + + addSingleTabletGroup( + cache, "1", false, "recovered", ByteString.copyFrom(new byte[] {(byte) 0x02})); + assertTrue(cache.getActiveAddresses().contains("recovered")); + } + + @Test + public void equalGenerationRepeatedSkipRoundsDoNotLaunderIncarnation() { + KeyRangeCache cache = new KeyRangeCache(new FakeEndpointCache()); + ByteString incarnation = ByteString.copyFrom(new byte[] {(byte) 0x01}); + + addSingleTabletGroup(cache, "1", false, "healthy", incarnation); + for (int i = 0; i < 3; i++) { + addSingleTabletGroup(cache, "1", true, "", ByteString.EMPTY); + addSingleTabletGroup(cache, "1", false, "replayed-" + i, incarnation); + assertTrue("skip round " + i + " must remain latched", cache.getActiveAddresses().isEmpty()); + } + } + + @Test + public void equalGenerationOmissionDoesNotEraseIncarnationHistory() { + KeyRangeCache cache = new KeyRangeCache(new FakeEndpointCache()); + ByteString incarnation = ByteString.copyFrom(new byte[] {(byte) 0x01}); + + addSingleTabletGroup(cache, "1", false, "healthy", incarnation); + addSingleTabletGroup(cache, "1", 2L, false, "other", incarnation); + addSingleTabletGroup(cache, "1", true, "", ByteString.EMPTY); + addSingleTabletGroup(cache, "1", false, "replayed", incarnation); + assertFalse(cache.getActiveAddresses().contains("replayed")); + + addSingleTabletGroup( + cache, "1", false, "recovered", ByteString.copyFrom(new byte[] {(byte) 0x02})); + assertTrue(cache.getActiveAddresses().contains("recovered")); + } + + @Test + public void equalGenerationOmittedHealthyTabletReappearsAtSameAddress() { + KeyRangeCache cache = new KeyRangeCache(new FakeEndpointCache()); + ByteString incarnation = ByteString.copyFrom(new byte[] {(byte) 0x01}); + + addSingleTabletGroup(cache, "1", false, "healthy", incarnation); + addSingleTabletGroup(cache, "1", 2L, false, "other", incarnation); + addSingleTabletGroup(cache, "1", false, "healthy", incarnation); + + assertTrue(cache.getActiveAddresses().contains("healthy")); + } + + @Test + public void equalGenerationOmittedHealthyTabletRequiresNewerIncarnationForNewAddress() { + KeyRangeCache cache = new KeyRangeCache(new FakeEndpointCache()); + ByteString incarnation = ByteString.copyFrom(new byte[] {(byte) 0x01}); + + addSingleTabletGroup(cache, "1", false, "healthy", incarnation); + addSingleTabletGroup(cache, "1", 2L, false, "other", incarnation); + addSingleTabletGroup(cache, "1", false, "replayed", incarnation); + assertFalse(cache.getActiveAddresses().contains("replayed")); + + addSingleTabletGroup( + cache, "1", false, "recovered", ByteString.copyFrom(new byte[] {(byte) 0x02})); + assertTrue(cache.getActiveAddresses().contains("recovered")); + } + + @Test + public void equalGenerationEmptyGroupRefillsAtSameAddress() { + KeyRangeCache cache = new KeyRangeCache(new FakeEndpointCache()); + ByteString incarnation = ByteString.copyFrom(new byte[] {(byte) 0x01}); + + addSingleTabletGroup(cache, "1", false, "healthy", incarnation); + addEmptyTabletGroup(cache, "1"); + addSingleTabletGroup(cache, "1", false, "healthy", incarnation); + + assertTrue(cache.getActiveAddresses().contains("healthy")); + } + + @Test + public void equalGenerationUnfreshAddressedSkipDoesNotAuthorizeNewAddress() { + FakeEndpointCache endpointCache = new FakeEndpointCache(); + KeyRangeCache cache = new KeyRangeCache(endpointCache); + ByteString incarnation = ByteString.copyFrom(new byte[] {(byte) 0x01}); + endpointCache.get("replayed"); + + addSingleTabletGroup(cache, "1", false, "healthy", incarnation); + addSingleTabletGroup(cache, "1", true, "replayed", ByteString.EMPTY); + addSingleTabletGroup(cache, "1", false, "replayed", incarnation); + + assertNull(selectSingleTablet(cache)); + + addSingleTabletGroup( + cache, "1", false, "replayed", ByteString.copyFrom(new byte[] {(byte) 0x02})); + ChannelEndpoint selected = selectSingleTablet(cache); + assertNotNull(selected); + assertEquals("replayed", selected.getAddress()); + } + + @Test + public void equalGenerationOmittedTabletUnfreshSkipDoesNotAuthorizeNewAddress() { + FakeEndpointCache endpointCache = new FakeEndpointCache(); + KeyRangeCache cache = new KeyRangeCache(endpointCache); + ByteString incarnation = ByteString.copyFrom(new byte[] {(byte) 0x01}); + endpointCache.get("replayed"); + + addSingleTabletGroup(cache, "1", false, "healthy", incarnation); + addSingleTabletGroup(cache, "1", 2L, false, "other", incarnation); + addSingleTabletGroup(cache, "1", true, "replayed", ByteString.EMPTY); + addSingleTabletGroup(cache, "1", false, "replayed", incarnation); + + assertNull(selectSingleTablet(cache)); + } + + @Test + public void equalGenerationEmptyAddressStateDoesNotBypassAddressBaseline() { + FakeEndpointCache endpointCache = new FakeEndpointCache(); + KeyRangeCache cache = new KeyRangeCache(endpointCache); + ByteString incarnation = ByteString.copyFrom(new byte[] {(byte) 0x01}); + endpointCache.get("replayed"); + + addSingleTabletGroup(cache, "1", false, "healthy", incarnation); + addSingleTabletGroup(cache, "1", true, "replayed", ByteString.EMPTY); + addSingleTabletGroup(cache, "1", false, "", incarnation); + addSingleTabletGroup(cache, "1", false, "replayed", incarnation); + + assertNull(selectSingleTablet(cache)); + } + + @Test + public void equalGenerationOmittedTabletEmptyAddressStateDoesNotBypassBaseline() { + FakeEndpointCache endpointCache = new FakeEndpointCache(); + KeyRangeCache cache = new KeyRangeCache(endpointCache); + ByteString incarnation = ByteString.copyFrom(new byte[] {(byte) 0x01}); + endpointCache.get("replayed"); + + addSingleTabletGroup(cache, "1", false, "healthy", incarnation); + addSingleTabletGroup(cache, "1", 2L, false, "other", incarnation); + addSingleTabletGroup(cache, "1", true, "replayed", ByteString.EMPTY); + addSingleTabletGroup(cache, "1", false, "", incarnation); + addSingleTabletGroup(cache, "1", false, "replayed", incarnation); + + assertNull(selectSingleTablet(cache)); + } + + @Test + public void equalGenerationFreshAddressMoveAdvancesBaseline() { + FakeEndpointCache endpointCache = new FakeEndpointCache(); + KeyRangeCache cache = new KeyRangeCache(endpointCache); + ByteString firstIncarnation = ByteString.copyFrom(new byte[] {(byte) 0x01}); + ByteString secondIncarnation = ByteString.copyFrom(new byte[] {(byte) 0x02}); + endpointCache.get("healthy"); + endpointCache.get("moved"); + + addSingleTabletGroup(cache, "1", false, "healthy", firstIncarnation); + addSingleTabletGroup(cache, "1", false, "moved", secondIncarnation); + ChannelEndpoint selected = selectSingleTablet(cache); + assertNotNull(selected); + assertEquals("moved", selected.getAddress()); + + addSingleTabletGroup(cache, "1", false, "", secondIncarnation); + addSingleTabletGroup(cache, "1", false, "healthy", firstIncarnation); + assertNull(selectSingleTablet(cache)); + + addSingleTabletGroup(cache, "1", false, "moved", secondIncarnation); + selected = selectSingleTablet(cache); + assertNotNull(selected); + assertEquals("moved", selected.getAddress()); + } + + @Test + public void equalGenerationFirstAddressDiscoveryDoesNotRequireNewerIncarnation() { + FakeEndpointCache endpointCache = new FakeEndpointCache(); + KeyRangeCache cache = new KeyRangeCache(endpointCache); + ByteString incarnation = ByteString.copyFrom(new byte[] {(byte) 0x01}); + endpointCache.get("healthy"); + + addSingleTabletGroup(cache, "1", false, "old", incarnation); + addSingleTabletGroup(cache, "2", false, "", incarnation); + addSingleTabletGroup(cache, "2", false, "healthy", incarnation); + + ChannelEndpoint selected = selectSingleTablet(cache); + assertNotNull(selected); + assertEquals("healthy", selected.getAddress()); + } + + @Test + public void equalGenerationFirstAddressDiscoveryEstablishesAddressBaseline() { + FakeEndpointCache endpointCache = new FakeEndpointCache(); + KeyRangeCache cache = new KeyRangeCache(endpointCache); + ByteString incarnation = ByteString.copyFrom(new byte[] {(byte) 0x01}); + endpointCache.get("healthy"); + endpointCache.get("replayed"); + + addSingleTabletGroup(cache, "1", false, "old", incarnation); + addSingleTabletGroup(cache, "2", false, "", incarnation); + addSingleTabletGroup(cache, "2", false, "healthy", incarnation); + addSingleTabletGroup(cache, "2", false, "replayed", incarnation); + + ChannelEndpoint selected = selectSingleTablet(cache); + assertNotNull(selected); + assertEquals("healthy", selected.getAddress()); + } + + @Test + public void equalGenerationSkippedEmptyStateLatchesFirstAddressFreshness() { + FakeEndpointCache endpointCache = new FakeEndpointCache(); + KeyRangeCache cache = new KeyRangeCache(endpointCache); + ByteString incarnation = ByteString.copyFrom(new byte[] {(byte) 0x01}); + endpointCache.get("replayed"); + + addSingleTabletGroup(cache, "1", false, "", incarnation); + addSingleTabletGroup(cache, "1", true, "", incarnation); + addSingleTabletGroup(cache, "1", false, "replayed", incarnation); + + assertNull(selectSingleTablet(cache)); + } + + @Test + public void equalGenerationOlderIncarnationCannotSupplyFirstAddress() { + FakeEndpointCache endpointCache = new FakeEndpointCache(); + KeyRangeCache cache = new KeyRangeCache(endpointCache); + endpointCache.get("replayed"); + + addSingleTabletGroup(cache, "1", false, "", ByteString.copyFrom(new byte[] {(byte) 0x02})); + addSingleTabletGroup( + cache, "1", false, "replayed", ByteString.copyFrom(new byte[] {(byte) 0x01})); + + assertNull(selectSingleTablet(cache)); + } + + @Test + public void equalGenerationFirstSeenSkippedEmptyTabletRequiresNewerIncarnation() { + FakeEndpointCache endpointCache = new FakeEndpointCache(); + KeyRangeCache cache = new KeyRangeCache(endpointCache); + ByteString incarnation = ByteString.copyFrom(new byte[] {(byte) 0x01}); + endpointCache.get("replayed"); + + addSingleTabletGroup(cache, "1", 2L, false, "other", incarnation); + addSingleTabletGroup(cache, "1", true, "", incarnation); + addSingleTabletGroup(cache, "1", false, "replayed", incarnation); + + assertNull(selectSingleTablet(cache)); + } + + @Test + public void equalGenerationFirstSeenAddressedSkipCannotChangeAddressWithoutNewerIncarnation() { + FakeEndpointCache endpointCache = new FakeEndpointCache(); + KeyRangeCache cache = new KeyRangeCache(endpointCache); + ByteString incarnation = ByteString.copyFrom(new byte[] {(byte) 0x01}); + endpointCache.get("replayed"); + + addSingleTabletGroup(cache, "1", 2L, false, "other", incarnation); + addSingleTabletGroup(cache, "1", true, "skipped", incarnation); + addSingleTabletGroup(cache, "1", false, "replayed", incarnation); + + assertNull(selectSingleTablet(cache)); + } + + @Test + public void equalGenerationFirstSeenRoutableTabletEstablishesAddressBaseline() { + FakeEndpointCache endpointCache = new FakeEndpointCache(); + KeyRangeCache cache = new KeyRangeCache(endpointCache); + ByteString incarnation = ByteString.copyFrom(new byte[] {(byte) 0x01}); + endpointCache.get("healthy"); + endpointCache.get("replayed"); + + addSingleTabletGroup(cache, "1", 2L, false, "other", incarnation); + addSingleTabletGroup(cache, "1", false, "healthy", incarnation); + addSingleTabletGroup(cache, "1", false, "replayed", incarnation); + + ChannelEndpoint selected = selectSingleTablet(cache); + assertNotNull(selected); + assertEquals("healthy", selected.getAddress()); + } + + @Test + public void equalGenerationAddressedSkipRefusesSameIncarnationAddressReplacement() { + KeyRangeCache cache = new KeyRangeCache(new FakeEndpointCache()); + ByteString incarnation = ByteString.copyFrom(new byte[] {(byte) 0x01}); + + addSingleTabletGroup(cache, "1", false, "healthy", incarnation); + addSingleTabletGroup(cache, "1", true, "healthy", ByteString.EMPTY); + addSingleTabletGroup(cache, "1", false, "replayed", incarnation); + + assertFalse(cache.getActiveAddresses().contains("replayed")); + assertTrue(cache.getActiveAddresses().contains("healthy")); + } + + @Test + public void equalGenerationAddressedSkipAllowsSameAddressUnskip() { + FakeEndpointCache endpointCache = new FakeEndpointCache(); + KeyRangeCache cache = new KeyRangeCache(endpointCache); + ByteString incarnation = ByteString.copyFrom(new byte[] {(byte) 0x01}); + + addSingleTabletGroup(cache, "1", false, "healthy", incarnation); + addSingleTabletGroup(cache, "1", true, "healthy", ByteString.EMPTY); + addSingleTabletGroup(cache, "1", false, "healthy", incarnation); + endpointCache.get("healthy"); + + ChannelEndpoint selected = + cache.fillRoutingHint( + /* preferLeader= */ true, + KeyRangeCache.RangeMode.COVERING_SPLIT, + DirectedReadOptions.getDefaultInstance(), + RoutingHint.newBuilder().setKey(bytes("a"))); + assertNotNull(selected); + assertEquals("healthy", selected.getAddress()); + } + + @Test + public void equalGenerationNewerSkippedIncarnationRatchetsHistory() { + KeyRangeCache cache = new KeyRangeCache(new FakeEndpointCache()); + + addSingleTabletGroup( + cache, "1", false, "healthy", ByteString.copyFrom(new byte[] {(byte) 0x01})); + addSingleTabletGroup(cache, "1", true, "", ByteString.copyFrom(new byte[] {(byte) 0x02})); + addSingleTabletGroup( + cache, "1", false, "replayed", ByteString.copyFrom(new byte[] {(byte) 0x02})); + assertTrue(cache.getActiveAddresses().isEmpty()); + + addSingleTabletGroup( + cache, "1", false, "recovered", ByteString.copyFrom(new byte[] {(byte) 0x03})); + assertTrue(cache.getActiveAddresses().contains("recovered")); + } + // --- Helper methods --- + private static boolean restoresSkippedTablet( + ByteString previousIncarnation, ByteString incomingIncarnation) { + KeyRangeCache cache = new KeyRangeCache(new FakeEndpointCache()); + addSingleTabletGroup(cache, "1", true, "", previousIncarnation); + addSingleTabletGroup(cache, "1", false, "recovered", incomingIncarnation); + return cache.getActiveAddresses().contains("recovered"); + } + + private static ChannelEndpoint selectSingleTablet(KeyRangeCache cache) { + return cache.fillRoutingHint( + /* preferLeader= */ true, + KeyRangeCache.RangeMode.COVERING_SPLIT, + DirectedReadOptions.getDefaultInstance(), + RoutingHint.newBuilder().setKey(bytes("a"))); + } + + private static void addSingleTabletGroup( + KeyRangeCache cache, + String generation, + boolean skip, + String serverAddress, + String incarnation) { + addSingleTabletGroup(cache, generation, skip, serverAddress, bytes(incarnation)); + } + + private static void addSingleTabletGroup( + KeyRangeCache cache, + String generation, + boolean skip, + String serverAddress, + ByteString incarnation) { + addSingleTabletGroup(cache, generation, 1L, skip, serverAddress, incarnation); + } + + private static void addSingleTabletGroup( + KeyRangeCache cache, + String generation, + long tabletUid, + boolean skip, + String serverAddress, + ByteString incarnation) { + cache.addRanges( + CacheUpdate.newBuilder() + .addRange( + Range.newBuilder() + .setStartKey(bytes("a")) + .setLimitKey(bytes("z")) + .setGroupUid(5) + .setSplitId(1) + .setGeneration(bytes(generation))) + .addGroup( + Group.newBuilder() + .setGroupUid(5) + .setGeneration(bytes(generation)) + .setLeaderIndex(0) + .addTablets( + Tablet.newBuilder() + .setTabletUid(tabletUid) + .setServerAddress(serverAddress) + .setIncarnation(incarnation) + .setDistance(0) + .setSkip(skip))) + .build()); + } + + private static void addEmptyTabletGroup(KeyRangeCache cache, String generation) { + cache.addRanges( + CacheUpdate.newBuilder() + .addRange( + Range.newBuilder() + .setStartKey(bytes("a")) + .setLimitKey(bytes("z")) + .setGroupUid(5) + .setSplitId(1) + .setGeneration(bytes(generation))) + .addGroup( + Group.newBuilder() + .setGroupUid(5) + .setGeneration(bytes(generation)) + .setLeaderIndex(0)) + .build()); + } + private static CacheUpdate singleReplicaUpdate(String serverAddress) { return CacheUpdate.newBuilder() .addRange(