fix(artifacts): do not derive a save's version number from a failed listing

core/src/main/java/com/google/adk/artifacts/GcsArtifactService.java

@@ -254,25 +254,10 @@ public Single<ImmutableList<Integer>> listVersions(
       String appName, String userId, String sessionId, String filename) {
     return Single.fromCallable(
         () -> {
-          String prefix = getBlobPrefix(appName, userId, sessionId, filename);
           try {
-            return Streams.stream(
-                    storageClient.list(bucketName, BlobListOption.prefix(prefix)).iterateAll())
-                .map(Blob::getName)
-                .map(
-                    name -> {
-                      int versionDelimiterIndex = name.lastIndexOf('/');
-                      return versionDelimiterIndex != -1
-                              && versionDelimiterIndex < name.length() - 1
-                          ? Optional.of(name.substring(versionDelimiterIndex + 1))
-                          : Optional.<String>empty();
-                    })
-                .flatMap(Optional::stream)
-                .map(Integer::parseInt)
-                .sorted()
-                .collect(ImmutableList.toImmutableList());
+            return readVersions(appName, userId, sessionId, filename);
           } catch (StorageException e) {
-            return ImmutableList.of();
+            return ImmutableList.<Integer>of();
           }
         });
   }
@@ -314,7 +299,7 @@ static SaveResult create(Blob blob, int version) {
 
   private Single<SaveResult> saveArtifactAndReturnBlob(
       String appName, String userId, String sessionId, String filename, Part artifact) {
-    return listVersions(appName, userId, sessionId, filename)
+    return Single.fromCallable(() -> versionsBeforeSaving(appName, userId, sessionId, filename))
         .map(versions -> versions.isEmpty() ? 0 : max(versions) + 1)
         .flatMap(
             nextVersion ->
@@ -351,4 +336,64 @@ private Single<SaveResult> saveArtifactAndReturnBlob(
                       }
                     }));
   }
+
+  /**
+   * Reads the versions stored for an artifact, letting a storage failure propagate.
+   *
+   * <p>An empty result from this method therefore means the artifact has no versions, never that
+   * the listing could not be performed. Each caller decides what a failure means for it: {@link
+   * #listVersions} reports it as "no versions", while {@link #saveArtifactAndReturnBlob} must not,
+   * because it derives the next version number from the result and would otherwise write over an
+   * object that already exists.
+   *
+   * @param appName Application name.
+   * @param userId User ID.
+   * @param sessionId Session ID.
+   * @param filename Artifact filename.
+   * @return sorted version numbers found under the artifact's prefix.
+   * @throws StorageException if the listing could not be performed.
+   */
+  private ImmutableList<Integer> readVersions(
+      String appName, String userId, String sessionId, String filename) {
+    String prefix = getBlobPrefix(appName, userId, sessionId, filename);
+    return Streams.stream(
+            storageClient.list(bucketName, BlobListOption.prefix(prefix)).iterateAll())
+        .map(Blob::getName)
+        .map(
+            name -> {
+              int versionDelimiterIndex = name.lastIndexOf('/');
+              return versionDelimiterIndex != -1 && versionDelimiterIndex < name.length() - 1
+                  ? Optional.of(name.substring(versionDelimiterIndex + 1))
+                  : Optional.<String>empty();
+            })
+        .flatMap(Optional::stream)
+        .map(Integer::parseInt)
+        .sorted()
+        .collect(ImmutableList.toImmutableList());
+  }
+
+  /**
+   * Reads the versions that already exist, for a save that is about to derive the next version
+   * number from them.
+   *
+   * <p>The write that follows carries no precondition, so treating a failed listing as "no
+   * versions" would compute version 0 and replace whatever is already stored under that name, while
+   * reporting success to the caller. Surfacing the failure instead is the same choice {@link
+   * #listArtifactKeys} and {@link #deleteArtifact} already make for the same operation.
+   *
+   * @param appName Application name.
+   * @param userId User ID.
+   * @param sessionId Session ID.
+   * @param filename Artifact filename.
+   * @return sorted version numbers already stored for the artifact.
+   * @throws VerifyException if the versions could not be listed.
+   */
+  private ImmutableList<Integer> versionsBeforeSaving(
+      String appName, String userId, String sessionId, String filename) {
+    try {
+      return readVersions(appName, userId, sessionId, filename);
+    } catch (StorageException e) {
+      throw new VerifyException("Failed to list artifact versions from GCS", e);
+    }
+  }
 }

core/src/test/java/com/google/adk/artifacts/GcsArtifactServiceTest.java

@@ -20,6 +20,7 @@
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
@@ -452,6 +453,7 @@ public void delete_storageException_throwsVerifyException() {
   @Test
   public void listVersions_storageException_returnsEmptyList() {
     String prefix = String.format("%s/%s/%s/%s/", APP_NAME, USER_ID, SESSION_ID, FILENAME);
+
     when(mockStorage.list(BUCKET_NAME, BlobListOption.prefix(prefix)))
         .thenThrow(new StorageException(500, "Induced error"));
 
@@ -461,6 +463,46 @@ public void listVersions_storageException_returnsEmptyList() {
     assertThat(versions).isEmpty();
   }
 
+  @Test
+  public void save_listStorageException_propagates() {
+    Part artifact = Part.fromBytes(new byte[] {4, 5}, "image/png");
+    when(mockStorage.list(eq(BUCKET_NAME), any(BlobListOption.class)))
+        .thenThrow(new StorageException(500, "Induced error"));
+
+    VerifyException thrown =
+        assertThrows(
+            VerifyException.class,
+            () ->
+                service
+                    .saveArtifact(APP_NAME, USER_ID, SESSION_ID, FILENAME, artifact)
+                    .blockingGet());
+
+    // The cause is the whole point: an operator has to be able to see which call failed and why.
+    assertThat(thrown).hasCauseThat().isInstanceOf(StorageException.class);
+    verify(mockStorage).list(eq(BUCKET_NAME), any(BlobListOption.class));
+  }
+
+  /**
+   * The listing failing is not itself the harm. Deriving a version number from the empty list it
+   * used to return is: the save computed version 0 and replaced whatever was already stored under
+   * that name. This asserts no write is attempted at all.
+   */
+  @Test
+  public void save_listStorageException_doesNotWrite() {
+    Part artifact = Part.fromBytes(new byte[] {4, 5}, "image/png");
+    when(mockStorage.list(eq(BUCKET_NAME), any(BlobListOption.class)))
+        .thenThrow(new StorageException(500, "Induced error"));
+
+    assertThrows(
+        VerifyException.class,
+        () ->
+            service.saveArtifact(APP_NAME, USER_ID, SESSION_ID, FILENAME, artifact).blockingGet());
+
+    // Asserts the listing was reached, so this cannot pass by failing earlier for another reason.
+    verify(mockStorage).list(eq(BUCKET_NAME), any(BlobListOption.class));
+    verify(mockStorage, never()).create(any(BlobInfo.class), any(byte[].class));
+  }
+
   @Test
   public void saveAndReload_noContentTypeAnywhere_defaultsToOctetStream() {
     // Artifact with no mime type