chore: new bidi appendable channel bootstrappable

* Add FlushPolicy.MinFlushSizeFlushPolicy.maxPendingBytes
* Add new contructor to ChannelSession to allow the resultFuture to be passed in.
* Create temporary BlobAppendableUploadImpl.AppendableUnbufferedWritableByteChannel to allow both old and new implementations to be passed to BlobAppendableUploadImpl -- this interface will be removed in cleanup
* Add new implementation using streaming buffer management as a sibling to the existing implementation. The existing implementation will be removed entirely in cleanup.
* Refactored tests to follow

Author: BenWhitehead

google-cloud-storage/src/main/java/com/google/cloud/storage/BidiAppendableUnbufferedWritableByteChannel.java

@@ -17,8 +17,8 @@
 package com.google.cloud.storage;
 
 import com.google.cloud.BaseServiceException;
+import com.google.cloud.storage.BlobAppendableUploadImpl.AppendableUnbufferedWritableByteChannel;
 import com.google.cloud.storage.ChunkSegmenter.ChunkSegment;
-import com.google.cloud.storage.UnbufferedWritableByteChannelSession.UnbufferedWritableByteChannel;
 import java.io.IOException;
 import java.io.InterruptedIOException;
 import java.nio.ByteBuffer;
@@ -27,7 +27,8 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
-final class BidiAppendableUnbufferedWritableByteChannel implements UnbufferedWritableByteChannel {
+final class BidiAppendableUnbufferedWritableByteChannel
+    implements AppendableUnbufferedWritableByteChannel {
 
   private final BidiUploadStreamingStream stream;
   private final ChunkSegmenter chunkSegmenter;

google-cloud-storage/src/main/java/com/google/cloud/storage/BlobAppendableUploadConfig.java

@@ -19,21 +19,28 @@
 import static com.google.cloud.storage.ByteSizeConstants._256KiB;
 import static java.util.Objects.requireNonNull;
 
+import com.google.api.core.ApiFuture;
 import com.google.api.core.ApiFutures;
 import com.google.api.core.BetaApi;
 import com.google.api.core.InternalApi;
+import com.google.api.core.SettableApiFuture;
 import com.google.api.gax.retrying.BasicResultRetryAlgorithm;
 import com.google.api.gax.rpc.AbortedException;
 import com.google.api.gax.rpc.ApiException;
+import com.google.cloud.storage.BidiUploadState.AppendableUploadState;
+import com.google.cloud.storage.BidiUploadState.TakeoverAppendableUploadState;
 import com.google.cloud.storage.BlobAppendableUpload.AppendableUploadWriteableByteChannel;
 import com.google.cloud.storage.BlobAppendableUploadImpl.AppendableObjectBufferedWritableByteChannel;
 import com.google.cloud.storage.Storage.BlobWriteOption;
 import com.google.cloud.storage.TransportCompatibility.Transport;
 import com.google.cloud.storage.UnifiedOpts.ObjectTargetOpt;
 import com.google.cloud.storage.UnifiedOpts.Opts;
+import com.google.common.base.MoreObjects;
 import com.google.storage.v2.BidiWriteObjectRequest;
 import com.google.storage.v2.BidiWriteObjectResponse;
 import com.google.storage.v2.Object;
+import com.google.storage.v2.ServiceConstants.Values;
+import java.util.function.BiFunction;
 import javax.annotation.concurrent.Immutable;
 
 /**
@@ -53,17 +60,20 @@ public final class BlobAppendableUploadConfig {
       new BlobAppendableUploadConfig(
           FlushPolicy.minFlushSize(_256KiB),
           Hasher.enabled(),
-          CloseAction.CLOSE_WITHOUT_FINALIZING);
+          CloseAction.CLOSE_WITHOUT_FINALIZING,
+          false);
 
   private final FlushPolicy flushPolicy;
   private final Hasher hasher;
   private final CloseAction closeAction;
+  private final boolean newImpl;
 
   private BlobAppendableUploadConfig(
-      FlushPolicy flushPolicy, Hasher hasher, CloseAction closeAction) {
+      FlushPolicy flushPolicy, Hasher hasher, CloseAction closeAction, boolean newImpl) {
     this.flushPolicy = flushPolicy;
     this.hasher = hasher;
     this.closeAction = closeAction;
+    this.newImpl = newImpl;
   }
 
   /**
@@ -94,7 +104,7 @@ public BlobAppendableUploadConfig withFlushPolicy(FlushPolicy flushPolicy) {
     if (this.flushPolicy.equals(flushPolicy)) {
       return this;
     }
-    return new BlobAppendableUploadConfig(flushPolicy, hasher, closeAction);
+    return new BlobAppendableUploadConfig(flushPolicy, hasher, closeAction, newImpl);
   }
 
   /**
@@ -124,7 +134,7 @@ public BlobAppendableUploadConfig withCloseAction(CloseAction closeAction) {
     if (this.closeAction == closeAction) {
       return this;
     }
-    return new BlobAppendableUploadConfig(flushPolicy, hasher, closeAction);
+    return new BlobAppendableUploadConfig(flushPolicy, hasher, closeAction, newImpl);
   }
 
   /**
@@ -156,7 +166,7 @@ BlobAppendableUploadConfig withCrc32cValidationEnabled(boolean enabled) {
       return this;
     }
     return new BlobAppendableUploadConfig(
-        flushPolicy, enabled ? Hasher.enabled() : Hasher.noop(), closeAction);
+        flushPolicy, enabled ? Hasher.enabled() : Hasher.noop(), closeAction, newImpl);
   }
 
   /** Never to be made public until {@link Hasher} is public */
@@ -165,6 +175,18 @@ Hasher getHasher() {
     return hasher;
   }
 
+  BlobAppendableUploadConfig useNewImpl() {
+    return new BlobAppendableUploadConfig(flushPolicy, hasher, closeAction, true);
+  }
+
+  @Override
+  public String toString() {
+    return MoreObjects.toStringHelper(this)
+        .add("closeAction", closeAction)
+        .add("flushPolicy", flushPolicy)
+        .toString();
+  }
+
   /**
    * Default instance factory method.
    *
@@ -217,55 +239,123 @@ public enum CloseAction {
   }
 
   BlobAppendableUpload create(GrpcStorageImpl storage, BlobInfo info, Opts<ObjectTargetOpt> opts) {
-    boolean takeOver = info.getGeneration() != null;
-    BidiWriteObjectRequest req =
-        takeOver
-            ? storage.getBidiWriteObjectRequestForTakeover(info, opts)
-            : storage.getBidiWriteObjectRequest(info, opts);
-
-    BidiAppendableWrite baw = new BidiAppendableWrite(req, takeOver);
+    if (newImpl) {
+      // TODO: make configurable
+      int maxRedirectsAllowed = 3;
 
+    long maxPendingBytes = this.getFlushPolicy().getMaxPendingBytes();
+    AppendableUploadState state = storage.getAppendableState(info, opts, maxPendingBytes);
     WritableByteChannelSession<AppendableObjectBufferedWritableByteChannel, BidiWriteObjectResponse>
         build =
-            ResumableMedia.gapic()
-                .write()
-                .bidiByteChannel(storage.storageClient.bidiWriteObjectCallable())
-                .setHasher(this.getHasher())
-                .setByteStringStrategy(ByteStringStrategy.copy())
-                .appendable()
-                .withRetryConfig(
-                    storage.retrier.withAlg(
-                        new BasicResultRetryAlgorithm<Object>() {
-                          @Override
-                          public boolean shouldRetry(
-                              Throwable previousThrowable, Object previousResponse) {
-                            // TODO: remove this later once the redirects are not handled by the
-                            // retry loop
-                            ApiException apiEx = null;
-                            if (previousThrowable instanceof StorageException) {
-                              StorageException se = (StorageException) previousThrowable;
-                              Throwable cause = se.getCause();
-                              if (cause instanceof ApiException) {
-                                apiEx = (ApiException) cause;
+            new AppendableSession(
+                ApiFutures.immediateFuture(state),
+                (start, resultFuture) -> {
+                  BidiUploadStreamingStream stream =
+                      new BidiUploadStreamingStream(
+                          start,
+                          storage.storageDataClient.executor,
+                          storage.storageClient.bidiWriteObjectCallable(),
+                          maxRedirectsAllowed,
+                          storage.storageDataClient.retryContextProvider.create());
+                  ChunkSegmenter chunkSegmenter =
+                      new ChunkSegmenter(
+                          Hasher.enabled(),
+                          ByteStringStrategy.copy(),
+                          Math.min(
+                              Values.MAX_WRITE_CHUNK_BYTES_VALUE, Math.toIntExact(maxPendingBytes)),
+                          /* blockSize= */ 1);
+                  BidiAppendableUnbufferedWritableByteChannel c;
+                  if (state instanceof TakeoverAppendableUploadState) {
+                    // start the takeover reconciliation
+                    stream.awaitTakeoverStateReconciliation();
+                    c =
+                        new BidiAppendableUnbufferedWritableByteChannel(
+                            stream, chunkSegmenter, state.getConfirmedBytes());
+                  } else {
+                    c = new BidiAppendableUnbufferedWritableByteChannel(stream, chunkSegmenter, 0);
+                  }
+                  return new AppendableObjectBufferedWritableByteChannel(
+                      flushPolicy.createBufferedChannel(c),
+                      c,
+                      this.closeAction == CloseAction.FINALIZE_WHEN_CLOSING,
+                      newImpl);
+                },
+                state.getResultFuture());
+
+      return new BlobAppendableUploadImpl(
+          new DefaultBlobWriteSessionConfig.DecoratedWritableByteChannelSession<>(
+              build, BidiBlobWriteSessionConfig.Factory.WRITE_OBJECT_RESPONSE_BLOB_INFO_DECODER));
+    } else {
+      boolean takeOver = info.getGeneration() != null;
+      BidiWriteObjectRequest req =
+          takeOver
+              ? storage.getBidiWriteObjectRequestForTakeover(info, opts)
+              : storage.getBidiWriteObjectRequest(info, opts);
+
+      BidiAppendableWrite baw = new BidiAppendableWrite(req, takeOver);
+
+      WritableByteChannelSession<
+              AppendableObjectBufferedWritableByteChannel, BidiWriteObjectResponse>
+          build =
+              ResumableMedia.gapic()
+                  .write()
+                  .bidiByteChannel(storage.storageClient.bidiWriteObjectCallable())
+                  .setHasher(this.getHasher())
+                  .setByteStringStrategy(ByteStringStrategy.copy())
+                  .appendable()
+                  .withRetryConfig(
+                      storage.retrier.withAlg(
+                          new BasicResultRetryAlgorithm<Object>() {
+                            @Override
+                            public boolean shouldRetry(
+                                Throwable previousThrowable, Object previousResponse) {
+                              // TODO: remove this later once the redirects are not handled by the
+                              // retry loop
+                              ApiException apiEx = null;
+                              if (previousThrowable instanceof StorageException) {
+                                StorageException se = (StorageException) previousThrowable;
+                                Throwable cause = se.getCause();
+                                if (cause instanceof ApiException) {
+                                  apiEx = (ApiException) cause;
+                                }
                               }
+                              if (apiEx instanceof AbortedException) {
+                                return true;
+                              }
+                              return storage
+                                  .retryAlgorithmManager
+                                  .idempotent()
+                                  .shouldRetry(previousThrowable, null);
                             }
-                            if (apiEx instanceof AbortedException) {
-                              return true;
-                            }
-                            return storage
-                                .retryAlgorithmManager
-                                .idempotent()
-                                .shouldRetry(previousThrowable, null);
-                          }
-                        }))
-                .buffered(this.getFlushPolicy())
-                .setStartAsync(ApiFutures.immediateFuture(baw))
-                .setGetCallable(storage.storageClient.getObjectCallable())
-                .setFinalizeOnClose(this.closeAction == CloseAction.FINALIZE_WHEN_CLOSING)
-                .build();
+                          }))
+                  .buffered(this.getFlushPolicy())
+                  .setStartAsync(ApiFutures.immediateFuture(baw))
+                  .setGetCallable(storage.storageClient.getObjectCallable())
+                  .setFinalizeOnClose(this.closeAction == CloseAction.FINALIZE_WHEN_CLOSING)
+                  .build();
 
-    return new BlobAppendableUploadImpl(
-        new DefaultBlobWriteSessionConfig.DecoratedWritableByteChannelSession<>(
-            build, BidiBlobWriteSessionConfig.Factory.WRITE_OBJECT_RESPONSE_BLOB_INFO_DECODER));
+      return new BlobAppendableUploadImpl(
+          new DefaultBlobWriteSessionConfig.DecoratedWritableByteChannelSession<>(
+              build, BidiBlobWriteSessionConfig.Factory.WRITE_OBJECT_RESPONSE_BLOB_INFO_DECODER));
+    }
+  }
+
+  private static final class AppendableSession
+      extends ChannelSession<
+          AppendableUploadState,
+          BidiWriteObjectResponse,
+          AppendableObjectBufferedWritableByteChannel>
+      implements WritableByteChannelSession<
+          AppendableObjectBufferedWritableByteChannel, BidiWriteObjectResponse> {
+    private AppendableSession(
+        ApiFuture<AppendableUploadState> startFuture,
+        BiFunction<
+                AppendableUploadState,
+                SettableApiFuture<BidiWriteObjectResponse>,
+                AppendableObjectBufferedWritableByteChannel>
+            f,
+        SettableApiFuture<BidiWriteObjectResponse> resultFuture) {
+      super(startFuture, f, resultFuture);
+    }
   }
 }

google-cloud-storage/src/main/java/com/google/cloud/storage/BlobAppendableUploadImpl.java

@@ -19,6 +19,7 @@
 import com.google.api.core.ApiFuture;
 import com.google.api.core.BetaApi;
 import com.google.cloud.storage.BufferedWritableByteChannelSession.BufferedWritableByteChannel;
+import com.google.cloud.storage.UnbufferedWritableByteChannelSession.UnbufferedWritableByteChannel;
 import com.google.common.base.Preconditions;
 import java.io.IOException;
 import java.nio.ByteBuffer;
@@ -63,17 +64,20 @@ static final class AppendableObjectBufferedWritableByteChannel
       implements BufferedWritableByteChannel,
           BlobAppendableUpload.AppendableUploadWriteableByteChannel {
     private final BufferedWritableByteChannel buffered;
-    private final GapicBidiUnbufferedAppendableWritableByteChannel unbuffered;
+    private final AppendableUnbufferedWritableByteChannel unbuffered;
     private final boolean finalizeOnClose;
     private final ReentrantLock lock;
+    private final boolean newImpl;
 
     AppendableObjectBufferedWritableByteChannel(
         BufferedWritableByteChannel buffered,
-        GapicBidiUnbufferedAppendableWritableByteChannel unbuffered,
-        boolean finalizeOnClose) {
+        AppendableUnbufferedWritableByteChannel unbuffered,
+        boolean finalizeOnClose,
+        boolean newImpl) {
       this.buffered = buffered;
       this.unbuffered = unbuffered;
       this.finalizeOnClose = finalizeOnClose;
+      this.newImpl = newImpl;
       lock = new ReentrantLock();
     }
 
@@ -111,10 +115,17 @@ public boolean isOpen() {
     public void finalizeAndClose() throws IOException {
       lock.lock();
       try {
-        if (buffered.isOpen()) {
-          buffered.flush();
-          unbuffered.finalizeWrite();
-          buffered.close();
+        if (newImpl) {
+          if (buffered.isOpen()) {
+            unbuffered.nextWriteShouldFinalize();
+            buffered.close();
+          }
+        } else {
+          if (buffered.isOpen()) {
+            buffered.flush();
+            unbuffered.finalizeWrite();
+            buffered.close();
+          }
         }
       } finally {
         lock.unlock();
@@ -142,4 +153,11 @@ public void close() throws IOException {
       }
     }
   }
+
+  /** Temporary shim to allow selective implementation */
+  interface AppendableUnbufferedWritableByteChannel extends UnbufferedWritableByteChannel {
+    default void nextWriteShouldFinalize() {}
+
+    default void finalizeWrite() throws IOException {}
+  }
 }

google-cloud-storage/src/main/java/com/google/cloud/storage/ChannelSession.java

@@ -39,9 +39,16 @@ class ChannelSession<StartT, ResultT, ChannelT> {
 
   ChannelSession(
       ApiFuture<StartT> startFuture, BiFunction<StartT, SettableApiFuture<ResultT>, ChannelT> f) {
+    this(startFuture, f, SettableApiFuture.create());
+  }
+
+  ChannelSession(
+      ApiFuture<StartT> startFuture,
+      BiFunction<StartT, SettableApiFuture<ResultT>, ChannelT> f,
+      SettableApiFuture<ResultT> resultFuture) {
     this.startFuture = startFuture;
-    this.resultFuture = SettableApiFuture.create();
-    this.f = (s) -> f.apply(s, resultFuture);
+    this.resultFuture = resultFuture;
+    this.f = (s) -> f.apply(s, this.resultFuture);
   }
 
   public ApiFuture<ChannelT> openAsync() {