fix: update ApiaryUnbufferedWritableByteChannel to be graceful of non…

…-quantum aligned write calls (#2493)

Despite GCS only allowing incremental updates to a resumable session at 256KiB byte boundaries, we have observed an extremely rare case of an incremental write being non-quantum aligned. The change in this PR makes ApiaryUnbufferedWritableByteChannel graceful to this possibility, and will only set the finalization header when close is invoked.

If a write is not quantum aligned, it will either:
1) not be consumed at all, at which point write can be called again with the still enqueued bytes
2) partially consumed, with the position of the provided ByteBuffers updated to reflect how much of their bytes were consumed, matching up with the number of bytes actually consumed returned from `write()`

Add new integration test to intentionally perform non-quantum aligned `write()` calls to ApiaryUnbufferedWritableByteChannel.

Separate change to gRPC affected code path to come in a later PR.

b/330550326

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

@@ -96,30 +96,32 @@ private long internalWrite(ByteBuffer[] srcs, int offset, int length, boolean fi
     }
     RewindableContent content = RewindableContent.of(Utils.subArray(srcs, offset, length));
     long available = content.getLength();
+    // as long as request has at least 256KiB GCS will accept bytes in 256KiB increments,
+    // however if a request is smaller than 256KiB it MUST be the finalization request.
+    if (!finalize && available < ByteSizeConstants._256KiB) {
+      return 0;
+    }
     long newFinalByteOffset = cumulativeByteCount + available;
     final HttpContentRange header;
     ByteRangeSpec rangeSpec = ByteRangeSpec.explicit(cumulativeByteCount, newFinalByteOffset);
-    boolean quantumAligned = available % ByteSizeConstants._256KiB == 0;
-    if (quantumAligned && finalize) {
+    if (finalize) {
       header = HttpContentRange.of(rangeSpec, newFinalByteOffset);
       finished = true;
-    } else if (quantumAligned) {
+    } else {
       header = HttpContentRange.of(rangeSpec);
-    } else { // not quantum aligned, have to finalize
-      header = HttpContentRange.of(rangeSpec, newFinalByteOffset);
-      finished = true;
     }
     try {
       ResumableOperationResult<@Nullable StorageObject> operationResult =
           session.put(content, header);
       long persistedSize = operationResult.getPersistedSize();
       committedBytesCallback.accept(persistedSize);
+      long written = persistedSize - cumulativeByteCount;
       this.cumulativeByteCount = persistedSize;
       if (finished) {
         StorageObject storageObject = operationResult.getObject();
         result.set(storageObject);
       }
-      return available;
+      return written;
     } catch (Exception e) {
       result.setException(e);
       throw StorageException.coalesce(e);

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

@@ -64,8 +64,7 @@ enum JsonResumableSessionFailureScenario {
   SCENARIO_7(
       BaseServiceException.UNKNOWN_CODE,
       "dataLoss",
-      "Client side data loss detected. Bytes acked is more than client sent."),
-  SCENARIO_9(503, "backendNotConnected", "Ack less than bytes sent");
+      "Client side data loss detected. Bytes acked is more than client sent.");
 
   private static final String PREFIX_I = "\t|< ";
   private static final String PREFIX_O = "\t|> ";

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

@@ -109,10 +109,9 @@ public void rewindTo(long offset) {
           success = true;
           return ResumableOperationResult.incremental(ackRange.endOffset());
         } else if (ackRange.endOffset() < effectiveEnd) {
-          StorageException se =
-              JsonResumableSessionFailureScenario.SCENARIO_9.toStorageException(uploadId, response);
-          span.setStatus(Status.UNKNOWN.withDescription(se.getMessage()));
-          throw se;
+          rewindTo(ackRange.endOffset());
+          success = true;
+          return ResumableOperationResult.incremental(ackRange.endOffset());
         } else {
           StorageException se =
               JsonResumableSessionFailureScenario.SCENARIO_7.toStorageException(uploadId, response);

google-cloud-storage/src/test/java/com/google/cloud/storage/ITJsonResumableSessionPutTaskTest.java

@@ -110,51 +110,6 @@ public void emptyObjectHappyPath() throws Exception {
     }
   }
 
-  /**
-   *
-   *
-   * <h4>S.9</h4>
-   *
-   * Partial successful append to session
-   *
-   * <p>The client has sent N bytes, the server confirmed N bytes as committed. The client sends K
-   * bytes starting at offset N. The server responds with only N + L with 0 &lt;= L &lt; K bytes as
-   * committed.
-   */
-  @Test
-  public void scenario9() throws Exception {
-
-    HttpRequestHandler handler =
-        req -> {
-          String contentRangeString = req.headers().get(CONTENT_RANGE);
-          HttpContentRange parse = HttpContentRange.parse(contentRangeString);
-          long endInclusive = ((HttpContentRange.HasRange<?>) parse).range().endOffsetInclusive();
-          FullHttpResponse resp =
-              new DefaultFullHttpResponse(req.protocolVersion(), RESUME_INCOMPLETE);
-          ByteRangeSpec range = ByteRangeSpec.explicitClosed(0L, endInclusive - 1);
-          resp.headers().set(HttpHeaderNames.RANGE, range.getHttpRangeHeader());
-          return resp;
-        };
-
-    try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler)) {
-      URI endpoint = fakeHttpServer.getEndpoint();
-      String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID());
-
-      AtomicLong confirmedBytes = new AtomicLong(-1L);
-
-      JsonResumableSessionPutTask task =
-          new JsonResumableSessionPutTask(
-              httpClientContext,
-              uploadUrl,
-              RewindableContent.empty(),
-              HttpContentRange.of(ByteRangeSpec.explicitClosed(0L, 10L)));
-
-      StorageException se = assertThrows(StorageException.class, task::call);
-      assertThat(se.getCode()).isEqualTo(503);
-      assertThat(confirmedBytes.get()).isEqualTo(-1L);
-    }
-  }
-
   /**
    *
    *

google-cloud-storage/src/test/java/com/google/cloud/storage/ITJsonResumableSessionTest.java

@@ -98,7 +98,7 @@ public void rewindWillQueryStatusOnlyWhenDirty() throws Exception {
           DefaultFullHttpResponse resp =
               new DefaultFullHttpResponse(req.protocolVersion(), RESUME_INCOMPLETE);
           if (range1.getHeaderValue().equals(contentRange)) {
-            resp.headers().set(RANGE, ByteRangeSpec.explicit(0L, _256KiBL).getHttpRangeHeader());
+            return new DefaultFullHttpResponse(req.protocolVersion(), SERVICE_UNAVAILABLE);
           } else if (range2.getHeaderValue().equals(contentRange)) {
             resp.headers().set(RANGE, ByteRangeSpec.explicit(0L, _256KiBL).getHttpRangeHeader());
           } else {

google-cloud-storage/src/test/java/com/google/cloud/storage/ITUnbufferedResumableUploadTest.java

@@ -0,0 +1,117 @@
+/*
+ * Copyright 2024 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.storage;
+
+import static com.google.common.truth.Truth.assertThat;
+
+import com.google.api.core.ApiFutures;
+import com.google.api.services.storage.model.StorageObject;
+import com.google.cloud.storage.ITUnbufferedResumableUploadTest.ObjectSizes;
+import com.google.cloud.storage.TransportCompatibility.Transport;
+import com.google.cloud.storage.UnbufferedWritableByteChannelSession.UnbufferedWritableByteChannel;
+import com.google.cloud.storage.UnifiedOpts.ObjectTargetOpt;
+import com.google.cloud.storage.UnifiedOpts.Opts;
+import com.google.cloud.storage.it.runner.StorageITRunner;
+import com.google.cloud.storage.it.runner.annotations.Backend;
+import com.google.cloud.storage.it.runner.annotations.CrossRun;
+import com.google.cloud.storage.it.runner.annotations.CrossRun.Exclude;
+import com.google.cloud.storage.it.runner.annotations.Inject;
+import com.google.cloud.storage.it.runner.annotations.Parameterized;
+import com.google.cloud.storage.it.runner.annotations.Parameterized.Parameter;
+import com.google.cloud.storage.it.runner.annotations.Parameterized.ParametersProvider;
+import com.google.cloud.storage.it.runner.registry.Generator;
+import com.google.cloud.storage.spi.v1.StorageRpc;
+import com.google.common.collect.ImmutableList;
+import java.io.IOException;
+import java.math.BigInteger;
+import java.nio.ByteBuffer;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.function.Supplier;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+@RunWith(StorageITRunner.class)
+@CrossRun(
+    backends = {Backend.PROD},
+    transports = {Transport.HTTP, Transport.GRPC})
+@Parameterized(ObjectSizes.class)
+public final class ITUnbufferedResumableUploadTest {
+
+  @Inject public Storage storage;
+  @Inject public BucketInfo bucket;
+  @Inject public Generator generator;
+
+  @Parameter public int objectSize;
+
+  public static final class ObjectSizes implements ParametersProvider {
+
+    @Override
+    public ImmutableList<Integer> parameters() {
+      return ImmutableList.of(256 * 1024, 2 * 1024 * 1024);
+    }
+  }
+
+  @Test
+  @Exclude(transports = Transport.GRPC)
+  public void json()
+      throws IOException, ExecutionException, InterruptedException, TimeoutException {
+    BlobInfo blobInfo = BlobInfo.newBuilder(bucket, generator.randomObjectName()).build();
+    Opts<ObjectTargetOpt> opts = Opts.empty();
+    final Map<StorageRpc.Option, ?> optionsMap = opts.getRpcOptions();
+    BlobInfo.Builder builder = blobInfo.toBuilder().setMd5(null).setCrc32c(null);
+    BlobInfo updated = opts.blobInfoMapper().apply(builder).build();
+
+    StorageObject encode = Conversions.json().blobInfo().encode(updated);
+    HttpStorageOptions options = (HttpStorageOptions) storage.getOptions();
+    Supplier<String> uploadIdSupplier =
+        ResumableMedia.startUploadForBlobInfo(
+            options,
+            updated,
+            optionsMap,
+            StorageRetryStrategy.getUniformStorageRetryStrategy().getIdempotentHandler());
+    JsonResumableWrite jsonResumableWrite =
+        JsonResumableWrite.of(encode, optionsMap, uploadIdSupplier.get(), 0);
+
+    UnbufferedWritableByteChannelSession<StorageObject> session =
+        ResumableMedia.http()
+            .write()
+            .byteChannel(HttpClientContext.from(options.getStorageRpcV1()))
+            .resumable()
+            .unbuffered()
+            .setStartAsync(ApiFutures.immediateFuture(jsonResumableWrite))
+            .build();
+
+    int additional = 13;
+    long size = objectSize + additional;
+    ByteBuffer b = DataGenerator.base64Characters().genByteBuffer(size);
+
+    UnbufferedWritableByteChannel open = session.open();
+    int written = open.write(b);
+    assertThat(written).isEqualTo(objectSize);
+    assertThat(b.remaining()).isEqualTo(additional);
+
+    int writtenAndClose = open.writeAndClose(b);
+    assertThat(writtenAndClose).isEqualTo(additional);
+    open.close();
+
+    StorageObject storageObject = session.getResult().get(2, TimeUnit.SECONDS);
+    assertThat(storageObject.getSize()).isEqualTo(BigInteger.valueOf(size));
+  }
+}