JVMCBC-1651: Support accessDeleted reads for subdoc replica reads

Sub-document reads from replicas do not support the accessDeleted
flag until server 8.0 (MB-63241 and MB-66949).

This combination is required for transactions; using them here.

Setting accessDeleted with sub-document reads continues to be
unsupported from the user-facing API, since this is niche
functionality.

Change-Id: I36febfe05f9add2456549e6a57614135bb998a1f
Reviewed-on: https://review.couchbase.org/c/couchbase-jvm-clients/+/228666
Tested-by: Build Bot <[email protected]>
Reviewed-by: David Nault <[email protected]>

Author: programmatix

core-io/src/main/java/com/couchbase/client/core/classic/kv/ClassicCoreKvOps.java

@@ -768,7 +768,8 @@ public Flux<CoreSubdocGetResult> subdocGetAllReplicasReactive(CoreCommonOptions
         retryStrategy,
         common.clientContext(),
         common.parentSpan().orElse(null),
-        readPreference
+        readPreference,
+        (byte) 0
     );
   }
 

core-io/src/main/java/com/couchbase/client/core/config/BucketCapabilities.java

@@ -47,7 +47,10 @@ public enum BucketCapabilities {
     DCP_IGNORE_PURGED_TOMBSTONES("dcp.IgnorePurgedTombstones"),
     RANGE_SCAN("rangeScan"),
     SUBDOC_READ_REPLICA("subdoc.ReplicaRead"),
-    NON_DEDUPED_HISTORY("nonDedupedHistory");
+    NON_DEDUPED_HISTORY("nonDedupedHistory"),
+    SUBDOC_BINARY_XATTR("subdoc.BinaryXattr"),
+    // Added in 8.0 (MB-66949)
+    SUBDOC_ACCESS_DELETED("subdoc.AccessDeleted");
 
     private final String raw;
 

core-io/src/main/java/com/couchbase/client/core/service/kv/ReplicaHelper.java

@@ -153,14 +153,15 @@ public static Flux<CoreSubdocGetResult> lookupInAllReplicasReactive(
     final RetryStrategy retryStrategy,
     Map<String, Object> clientContext,
     RequestSpan parentSpan,
-    CoreReadPreference readPreference
+    CoreReadPreference readPreference,
+    byte flags
   ) {
     notNullOrEmpty(documentId, "Id", () -> ReducedKeyValueErrorContext.create(documentId, collectionIdentifier));
 
     CoreEnvironment env = core.context().environment();
     RequestSpan getAllSpan = core.context().coreResources().requestTracer().requestSpan(TracingIdentifiers.SPAN_LOOKUP_IN_ALL_REPLICAS, parentSpan);
     return Reactor
-      .toMono(() -> lookupInAllReplicasRequests(core, collectionIdentifier, documentId, commands, clientContext, retryStrategy, timeout, getAllSpan, readPreference))
+      .toMono(() -> lookupInAllReplicasRequests(core, collectionIdentifier, documentId, commands, clientContext, retryStrategy, timeout, getAllSpan, readPreference, flags))
       .flux()
       .flatMap(Flux::fromStream)
       .flatMap(request -> Reactor
@@ -239,11 +240,12 @@ public static <R> CompletableFuture<List<CompletableFuture<R>>> lookupInAllRepli
     final Map<String, Object> clientContext,
     final RequestSpan parentSpan,
     final CoreReadPreference readPreference,
+    final byte flags,
     final Function<CoreSubdocGetResult, R> responseMapper
   ) {
     RequestSpan getAllSpan = core.context().coreResources().requestTracer().requestSpan(TracingIdentifiers.SPAN_GET_ALL_REPLICAS, parentSpan);
 
-    return lookupInAllReplicasRequests(core, collectionIdentifier, documentId, commands, clientContext, retryStrategy, timeout, getAllSpan, readPreference)
+    return lookupInAllReplicasRequests(core, collectionIdentifier, documentId, commands, clientContext, retryStrategy, timeout, getAllSpan, readPreference, flags)
       .thenApply(stream ->
         stream.map(request ->
           get(core, request)
@@ -312,13 +314,14 @@ public static <R> CompletableFuture<R> lookupInAnyReplicaAsync(
     final Map<String, Object> clientContext,
     final RequestSpan parentSpan,
     final CoreReadPreference readPreference,
+    final byte flags,
     final Function<CoreSubdocGetResult, R> responseMapper) {
 
     RequestSpan getAnySpan = core.context().coreResources().requestTracer()
       .requestSpan(TracingIdentifiers.SPAN_LOOKUP_IN_ANY_REPLICA, parentSpan);
 
     CompletableFuture<List<CompletableFuture<R>>> listOfFutures = lookupInAllReplicasAsync(
-      core, collectionIdentifier, documentId, commands, timeout, retryStrategy, clientContext, getAnySpan, readPreference, responseMapper
+      core, collectionIdentifier, documentId, commands, timeout, retryStrategy, clientContext, getAnySpan, readPreference, flags, responseMapper
     );
 
     // Aggregating the futures here will discard the individual errors, which we don't need
@@ -476,7 +479,8 @@ public static CompletableFuture<Stream<SubdocGetRequest>> lookupInAllReplicasReq
     final RetryStrategy retryStrategy,
     final Duration timeout,
     final RequestSpan parent,
-    final CoreReadPreference readPreference
+    final CoreReadPreference readPreference,
+    final byte flags
   ) {
     notNullOrEmpty(documentId, "Id");
 
@@ -490,7 +494,7 @@ public static CompletableFuture<Stream<SubdocGetRequest>> lookupInAllReplicasReq
         return failedFuture(FeatureNotAvailableException.subdocReadReplica());
       }
 
-      List<SubdocGetRequest> requests = lookupInAllReplicasRequestsWithFallback(core, collectionIdentifier, documentId, commands, clientContext, retryStrategy, timeout, parent, readPreference, topology);
+      List<SubdocGetRequest> requests = lookupInAllReplicasRequestsWithFallback(core, collectionIdentifier, documentId, commands, clientContext, retryStrategy, timeout, parent, readPreference, topology, flags);
       if (requests.isEmpty()) {
         return failedFuture(DocumentUnretrievableException.noReplicasSuitable());
       }
@@ -501,7 +505,7 @@ public static CompletableFuture<Stream<SubdocGetRequest>> lookupInAllReplicasReq
       final Duration retryDelay = Duration.ofMillis(100);
       final CompletableFuture<Stream<SubdocGetRequest>> future = new CompletableFuture<>();
       coreContext.environment().timer().schedule(() -> {
-        lookupInAllReplicasRequests(core, collectionIdentifier, documentId, commands, clientContext, retryStrategy, timeout.minus(retryDelay), parent, readPreference).whenComplete((getRequestStream, throwable) -> {
+        lookupInAllReplicasRequests(core, collectionIdentifier, documentId, commands, clientContext, retryStrategy, timeout.minus(retryDelay), parent, readPreference, flags).whenComplete((getRequestStream, throwable) -> {
           if (throwable != null) {
             future.completeExceptionally(throwable);
           } else {
@@ -524,15 +528,16 @@ private static List<SubdocGetRequest> lookupInAllReplicasRequestsWithFallback(Co
                                                                                 Duration timeout,
                                                                                 RequestSpan parent,
                                                                                 CoreReadPreference readPreference,
-                                                                                CouchbaseBucketConfig topology) {
+                                                                                CouchbaseBucketConfig topology,
+                                                                                byte flags) {
     CoreContext coreContext = core.context();
 
     int numReplicas = topology.numberOfReplicas();
     List<SubdocGetRequest> requests = new ArrayList<>(numReplicas + 1);
     NodeIndexCalculator allowedNodeIndexes = new NodeIndexCalculator(readPreference, topology, coreContext);
     if (allowedNodeIndexes.canUseNodeForActive(documentId)) {
       RequestSpan span = coreContext.coreResources().requestTracer().requestSpan(TracingIdentifiers.SPAN_REQUEST_KV_LOOKUP_IN, parent);
-      SubdocGetRequest activeRequest = SubdocGetRequest.create(timeout, coreContext, collectionIdentifier, retryStrategy, documentId, (byte) 0, commands, span);
+      SubdocGetRequest activeRequest = SubdocGetRequest.create(timeout, coreContext, collectionIdentifier, retryStrategy, documentId, flags, commands, span);
       activeRequest.context().clientContext(clientContext);
       requests.add(activeRequest);
     }
@@ -541,15 +546,15 @@ private static List<SubdocGetRequest> lookupInAllReplicasRequestsWithFallback(Co
       if (allowedNodeIndexes.canUseNodeForReplica(documentId, replica - 1)) {
         RequestSpan replicaSpan = coreContext.coreResources().requestTracer().requestSpan(TracingIdentifiers.SPAN_LOOKUP_IN_ALL_REPLICAS, parent);
         ReplicaSubdocGetRequest replicaRequest = ReplicaSubdocGetRequest.create(
-          timeout, coreContext, collectionIdentifier, retryStrategy, documentId, (byte) 0, commands, replica, replicaSpan
+          timeout, coreContext, collectionIdentifier, retryStrategy, documentId, flags, commands, replica, replicaSpan
         );
         replicaRequest.context().clientContext(clientContext);
         requests.add(replicaRequest);
       }
     }
     if (requests.isEmpty() && readPreference == CoreReadPreference.PREFERRED_SERVER_GROUP_OR_ALL_AVAILABLE) {
       logger.trace("Unable to find suitable replicas with PREFERRED_SERVER_GROUP_WITH_FALLBACK, falling back to NO_PREFERENCE");
-      return lookupInAllReplicasRequestsWithFallback(core, collectionIdentifier, documentId, commands, clientContext, retryStrategy, timeout, parent, CoreReadPreference.NO_PREFERENCE, topology);
+      return lookupInAllReplicasRequestsWithFallback(core, collectionIdentifier, documentId, commands, clientContext, retryStrategy, timeout, parent, CoreReadPreference.NO_PREFERENCE, topology, flags);
     }
     return requests;
   }

core-io/src/main/java/com/couchbase/client/core/topology/BucketCapability.java

@@ -41,6 +41,9 @@ public enum BucketCapability {
   RANGE_SCAN("rangeScan"),
   SUBDOC_READ_REPLICA("subdoc.ReplicaRead"),
   NON_DEDUPED_HISTORY("nonDedupedHistory"),
+  SUBDOC_BINARY_XATTR("subdoc.BinaryXattr"),
+  // Added in 8.0 (MB-66949)
+  SUBDOC_ACCESS_DELETED("subdoc.AccessDeleted")
   ;
 
   private final String wireName;

core-io/src/main/java/com/couchbase/client/core/transaction/util/TransactionKVHandler.java

@@ -53,6 +53,7 @@
 
 import static com.couchbase.client.core.error.DefaultErrorUtil.keyValueStatusToException;
 import static com.couchbase.client.core.msg.kv.SubdocGetRequest.convertCommandsToCore;
+import static com.couchbase.client.core.topology.BucketCapability.SUBDOC_ACCESS_DELETED;
 
 /**
  * Transactions does a lot of KV work from core-io.  This logic is essentially a mini version of java-client, providing
@@ -153,8 +154,20 @@ public static Mono<CoreSubdocGetResult> lookupIn(final Core core,
 
 
             if (preferredReplicaMode) {
-                CompletableFuture<CoreSubdocGetResult> replicas = ReplicaHelper.lookupInAnyReplicaAsync(core, collectionIdentifier, id, convertCommandsToCore(commands), timeout, BestEffortRetryStrategy.INSTANCE,
-                        clientContext, pspan == null ? null : pspan.span(), CoreReadPreference.PREFERRED_SERVER_GROUP_OR_ALL_AVAILABLE, (r) -> r);
+                CompletableFuture<CoreSubdocGetResult> replicas =
+                        BucketConfigUtil.waitForBucketTopology(core, collectionIdentifier.bucket(), timeout).toFuture()
+                                .thenCompose(bucketConfig -> {
+                                    byte flags = 0;
+                                    if (accessDeleted) {
+                                        // We can only accessDeleted when the server supports it (8.0+).
+                                        // Otherwise we will proceed with the operation though it is sub-optimal.
+                                        if (bucketConfig.bucket().capabilities().contains(SUBDOC_ACCESS_DELETED)) {
+                                            flags = SubdocMutateRequest.SUBDOC_DOC_FLAG_ACCESS_DELETED;
+                                        }
+                                    }
+                                    return ReplicaHelper.lookupInAnyReplicaAsync(core, collectionIdentifier, id, convertCommandsToCore(commands), timeout, BestEffortRetryStrategy.INSTANCE,
+                                            clientContext, pspan == null ? null : pspan.span(), CoreReadPreference.PREFERRED_SERVER_GROUP_OR_ALL_AVAILABLE, flags, (r) -> r);
+                                });
 
                 return Reactor.wrap(replicas, () -> {})
                         .switchIfEmpty(Mono.error(new DocumentUnretrievableException(ReducedKeyValueErrorContext.create(id, collectionIdentifier))))

core-io/src/main/java/com/couchbase/client/core/util/BucketConfigUtil.java

@@ -22,6 +22,8 @@
 import com.couchbase.client.core.retry.reactor.Backoff;
 import com.couchbase.client.core.retry.reactor.Retry;
 import com.couchbase.client.core.retry.reactor.RetryExhaustedException;
+import com.couchbase.client.core.topology.ClusterTopologyWithBucket;
+import reactor.core.Exceptions;
 import reactor.core.publisher.Mono;
 
 import java.time.Duration;
@@ -39,6 +41,7 @@ private BucketConfigUtil() {}
      * A bucket config can be null while the bucket has not been opened.  This method allows easily for a config to be
      * available.
      */
+    @Deprecated
     public static Mono<BucketConfig> waitForBucketConfig(final Core core,
                                                          final String bucketName,
                                                          final Duration timeout) {
@@ -60,4 +63,26 @@ public static Mono<BucketConfig> waitForBucketConfig(final Core core,
                     }
                 });
     }
+
+  public static Mono<ClusterTopologyWithBucket> waitForBucketTopology(final Core core,
+                                                                      final String bucketName,
+                                                                      final Duration timeout) {
+    return Mono.fromCallable(() -> {
+        final ClusterTopologyWithBucket bucketConfig = core.clusterConfig().bucketTopology(bucketName);
+        if (bucketConfig == null) {
+          throw new NullPointerException();
+        }
+        return bucketConfig;
+      }).retryWhen(Retry.anyOf(NullPointerException.class)
+        .timeout(timeout)
+        .backoff(Backoff.fixed(retryDelay))
+        .toReactorRetry())
+      .onErrorResume(err -> {
+        if (Exceptions.isRetryExhausted(err)) {
+          return Mono.error(new UnambiguousTimeoutException("Timed out while waiting for bucket config", null));
+        } else {
+          return Mono.error(err);
+        }
+      });
+  }
 }

java-client/src/main/java/com/couchbase/client/java/AsyncCollection.java

@@ -712,6 +712,7 @@ public CompletableFuture<List<CompletableFuture<LookupInReplicaResult>>> lookupI
       opts.clientContext(),
       opts.parentSpan().orElse(null),
       opts.readPreference(),
+      (byte) 0,
       response -> LookupInReplicaResult.from(response, serializer));
   }
 
@@ -752,6 +753,7 @@ public CompletableFuture<LookupInReplicaResult> lookupInAnyReplica(final String
       opts.clientContext(),
       opts.parentSpan().orElse(null),
       opts.readPreference(),
+      (byte) 0,
       response -> LookupInReplicaResult.from(response, serializer));
   }