feat(gax): add progress listener models and UploadProgressTracker

sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadProgressListener.java

@@ -0,0 +1,50 @@
+/*
+ * Copyright 2026 Google LLC
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *     * Neither the name of Google LLC nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+package com.google.api.gax.rpc;
+
+import com.google.api.core.BetaApi;
+import org.jspecify.annotations.NullMarked;
+
+/** A callback listener for observing progress and state transitions of a resumable upload. */
+@BetaApi
+@FunctionalInterface
+@NullMarked
+public interface ResumableUploadProgressListener {
+
+  /**
+   * Invoked when upload progress or state changes.
+   *
+   * <p>Cancellation via {@link ResumableUploadFuture#cancel(boolean)} can be invoked safely from
+   * within this callback.
+   *
+   * @param status the current status snapshot of the upload
+   */
+  void onProgress(ResumableUploadStatus status);
+}

sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadStatus.java

@@ -0,0 +1,101 @@
+/*
+ * Copyright 2026 Google LLC
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *     * Neither the name of Google LLC nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+package com.google.api.gax.rpc;
+
+import com.google.api.core.BetaApi;
+import com.google.auto.value.AutoValue;
+import org.jspecify.annotations.NullMarked;
+import org.jspecify.annotations.Nullable;
+
+/** Status snapshot of an ongoing or completed resumable upload session. */
+@BetaApi
+@NullMarked
+@AutoValue
+public abstract class ResumableUploadStatus {
+
+  /** The state of the resumable upload session. */
+  public enum State {
+    /** Session initiation is in progress (acquiring upload session URL). */
+    STARTING,
+
+    /** The session initiation completed successfully. */
+    STARTED,
+
+    /** Transmitting chunk payloads to the server. */
+    UPLOADING,
+
+    /** A recoverable error occurred; querying server status and resynchronizing offset. */
+    RECOVERING,
+
+    /** The server query status succeeded and the committed offset was received. */
+    OFFSET_RECEIVED,
+
+    /** The upload was successfully finalized by the server. */
+    FINALIZED,
+
+    /** The upload failed unrecoverably or was cancelled. */
+    FAILED
+  }
+
+  /**
+   * Returns the negotiated upload session URI, or {@code null} if session initiation is pending.
+   */
+  public abstract @Nullable String getUploadUrl();
+
+  /** Returns the number of bytes confirmed as uploaded to the server so far. */
+  public abstract long getBytesUploaded();
+
+  /** Returns the current state of the upload session. */
+  public abstract State getState();
+
+  /** Returns the exception that triggered recovery or caused failure, if any. */
+  public abstract @Nullable Throwable getException();
+
+  public abstract Builder toBuilder();
+
+  public static Builder newBuilder() {
+    return new AutoValue_ResumableUploadStatus.Builder()
+        .setBytesUploaded(0L)
+        .setState(State.STARTING);
+  }
+
+  @AutoValue.Builder
+  public abstract static class Builder {
+    public abstract Builder setUploadUrl(@Nullable String uploadUrl);
+
+    public abstract Builder setBytesUploaded(long bytesUploaded);
+
+    public abstract Builder setState(State state);
+
+    public abstract Builder setException(@Nullable Throwable exception);
+
+    public abstract ResumableUploadStatus build();
+  }
+}

sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/UploadProgressTracker.java

@@ -0,0 +1,239 @@
+/*
+ * Copyright 2026 Google LLC
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *     * Neither the name of Google LLC nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+package com.google.api.gax.rpc;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.common.util.concurrent.MoreExecutors;
+import com.google.errorprone.annotations.concurrent.GuardedBy;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Executor;
+import org.jspecify.annotations.NullMarked;
+import org.jspecify.annotations.Nullable;
+
+/**
+ * Thread-safe tracker and dispatcher for resumable upload progress and state transitions.
+ *
+ * <p>Enforces monotonic progress reporting, isolates listeners from upload pipeline failures,
+ * serializes callbacks per listener, and manages transition to terminal states.
+ */
+@NullMarked
+class UploadProgressTracker {
+
+  private static final class RegisteredListener {
+    final ResumableUploadProgressListener listener;
+    final Executor sequentialExecutor;
+
+    RegisteredListener(ResumableUploadProgressListener listener, Executor executor) {
+      this.listener = listener;
+      this.sequentialExecutor = MoreExecutors.newSequentialExecutor(executor);
+    }
+  }
+
+  private final Object lock = new Object();
+
+  @GuardedBy("lock")
+  private final List<RegisteredListener> listeners = new ArrayList<>();
+
+  @GuardedBy("lock")
+  private ResumableUploadStatus currentStatus;
+
+  @GuardedBy("lock")
+  private boolean terminal;
+
+  @GuardedBy("lock")
+  private @Nullable String uploadSessionUrl;
+
+  UploadProgressTracker() {
+    this.currentStatus =
+        ResumableUploadStatus.newBuilder()
+            .setState(ResumableUploadStatus.State.STARTING)
+            .setBytesUploaded(0L)
+            .build();
+  }
+
+  void addListener(ResumableUploadProgressListener listener, Executor executor) {
+    checkNotNull(listener, "listener must not be null");
+    checkNotNull(executor, "executor must not be null");
+    RegisteredListener entry = new RegisteredListener(listener, executor);
+    ResumableUploadStatus snapshot;
+    synchronized (lock) {
+      snapshot = this.currentStatus;
+      if (!terminal) {
+        listeners.add(entry);
+      }
+    }
+    entry.sequentialExecutor.execute(() -> dispatchSafely(listener, snapshot));
+  }
+
+  ResumableUploadStatus getStatus() {
+    synchronized (lock) {
+      return currentStatus;
+    }
+  }
+
+  void onStarted(String uploadUrl) {
+    checkNotNull(uploadUrl, "uploadUrl must not be null");
+    List<RegisteredListener> snapshot;
+    ResumableUploadStatus status;
+    synchronized (lock) {
+      this.uploadSessionUrl = uploadUrl;
+      status =
+          currentStatus.toBuilder()
+              .setState(ResumableUploadStatus.State.STARTED)
+              .setUploadUrl(uploadUrl)
+              .build();
+      snapshot = updateStatusLocked(status);
+    }
+    notifyListeners(snapshot, status);
+  }
+
+  void onChunkUploaded(long bytesUploaded) {
+    List<RegisteredListener> snapshot;
+    ResumableUploadStatus status;
+    synchronized (lock) {
+      if (terminal) {
+        return;
+      }
+      long bytes = Math.max(currentStatus.getBytesUploaded(), bytesUploaded);
+      status =
+          currentStatus.toBuilder()
+              .setState(ResumableUploadStatus.State.UPLOADING)
+              .setBytesUploaded(bytes)
+              .setUploadUrl(uploadSessionUrl)
+              .build();
+      snapshot = updateStatusLocked(status);
+    }
+    notifyListeners(snapshot, status);
+  }
+
+  void onRecovering(@Nullable Throwable cause) {
+    List<RegisteredListener> snapshot;
+    ResumableUploadStatus status;
+    synchronized (lock) {
+      if (terminal) {
+        return;
+      }
+      status =
+          currentStatus.toBuilder()
+              .setState(ResumableUploadStatus.State.RECOVERING)
+              .setException(cause)
+              .setUploadUrl(uploadSessionUrl)
+              .build();
+      snapshot = updateStatusLocked(status);
+    }
+    notifyListeners(snapshot, status);
+  }
+
+  void onOffsetReceived(long committedOffset) {
+    List<RegisteredListener> snapshot;
+    ResumableUploadStatus status;
+    synchronized (lock) {
+      if (terminal) {
+        return;
+      }
+      long bytes = Math.max(currentStatus.getBytesUploaded(), committedOffset);
+      status =
+          currentStatus.toBuilder()
+              .setState(ResumableUploadStatus.State.OFFSET_RECEIVED)
+              .setBytesUploaded(bytes)
+              .setUploadUrl(uploadSessionUrl)
+              .build();
+      snapshot = updateStatusLocked(status);
+    }
+    notifyListeners(snapshot, status);
+  }
+
+  void onFinalized(long totalBytes) {
+    List<RegisteredListener> snapshot;
+    ResumableUploadStatus status;
+    synchronized (lock) {
+      if (terminal) {
+        return;
+      }
+      terminal = true;
+      long bytes = Math.max(currentStatus.getBytesUploaded(), totalBytes);
+      status =
+          currentStatus.toBuilder()
+              .setState(ResumableUploadStatus.State.FINALIZED)
+              .setBytesUploaded(bytes)
+              .setUploadUrl(uploadSessionUrl)
+              .build();
+      snapshot = updateStatusLocked(status);
+    }
+    notifyListeners(snapshot, status);
+  }
+
+  void onFailed(@Nullable Throwable error, @Nullable String sessionUrl) {
+    List<RegisteredListener> snapshot;
+    ResumableUploadStatus status;
+    synchronized (lock) {
+      if (terminal) {
+        return;
+      }
+      terminal = true;
+      String url = sessionUrl != null ? sessionUrl : uploadSessionUrl;
+      status =
+          currentStatus.toBuilder()
+              .setState(ResumableUploadStatus.State.FAILED)
+              .setException(error)
+              .setUploadUrl(url)
+              .build();
+      snapshot = updateStatusLocked(status);
+    }
+    notifyListeners(snapshot, status);
+  }
+
+  @GuardedBy("lock")
+  private List<RegisteredListener> updateStatusLocked(ResumableUploadStatus newStatus) {
+    this.currentStatus = newStatus;
+    if (newStatus.getUploadUrl() != null && this.uploadSessionUrl == null) {
+      this.uploadSessionUrl = newStatus.getUploadUrl();
+    }
+    return new ArrayList<>(this.listeners);
+  }
+
+  private void notifyListeners(
+      List<RegisteredListener> targetListeners, ResumableUploadStatus status) {
+    for (RegisteredListener entry : targetListeners) {
+      entry.sequentialExecutor.execute(() -> dispatchSafely(entry.listener, status));
+    }
+  }
+
+  private static void dispatchSafely(
+      ResumableUploadProgressListener listener, ResumableUploadStatus status) {
+    try {
+      listener.onProgress(status);
+    } catch (Throwable ignored) {
+      // Listener exceptions are isolated from the upload pipeline
+    }
+  }
+}