Add list per prefix to fix productions topology snapshots (#3369)

Signed-off-by: Julien Tinguely <[email protected]>

Author: julientinguely-da

apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/PeriodicTopologySnapshotIntegrationTest.scala

@@ -27,7 +27,7 @@ class PeriodicTopologySnapshotIntegrationTest[T <: BackupDumpConfig] extends Int
     BackupDumpConfig.Gcp(GcpBucketConfig.inferForTesting(TopologySnapshotTest), None)
 
   private def listDump(utcDate: String): Seq[Blob] =
-    bucket.list(startOffset = s"topology_snapshot_$utcDate", endOffset = "")
+    bucket.listBlobsByPrefix(prefix = s"topology_snapshot_$utcDate")
 
   override def environmentDefinition: SpliceEnvironmentDefinition =
     EnvironmentDefinition

apps/app/src/test/scala/org/lfdecentralizedtrust/splice/util/DataExportTestUtil.scala

@@ -25,7 +25,7 @@ trait DataExportTestUtil extends TestCommon {
       s"$cluster/$namespace/${getFileName(instant)}"
     val now = Instant.now
     // Query everything within the last 20min and check that we have at least one.
-    val blobs = bucket.list(name(now.plus(-20, ChronoUnit.MINUTES)), name(now))
+    val blobs = bucket.listBlobsByOffset(name(now.plus(-20, ChronoUnit.MINUTES)), name(now))
     blobs should not be empty
     forAll(blobs) { blob =>
       val dump = bucket.readStringFromBucket(Paths.get(blob.getName))

apps/common/src/main/scala/org/lfdecentralizedtrust/splice/util/BackupDump.scala

@@ -96,12 +96,7 @@ object BackupDump {
           case Some(p) => s"$p/"
           case None => ""
         }
-        val blobs = gcpBucket.list(s"$pref$offset", "")
-        loggerFactory
-          .getLogger(this.getClass)
-          .info(
-            s"all matched blobs found: ${blobs.map(_.getName)}, using offset $offset, and prefix $pref, bucket ${bucketConfig.bucketName}, project: ${bucketConfig.projectId}"
-          )
+        val blobs = gcpBucket.listBlobsByPrefix(prefix = s"$pref$offset")
         blobs.nonEmpty
       case _ =>
         throw Status.UNIMPLEMENTED

apps/common/src/main/scala/org/lfdecentralizedtrust/splice/util/GcpBucket.scala

@@ -41,7 +41,7 @@ class GcpBucket(config: GcpBucketConfig, override val loggerFactory: NamedLogger
   }
 
   @SuppressWarnings(Array("org.wartremover.warts.Var", "org.wartremover.warts.While"))
-  def list(startOffset: String, endOffset: String): Seq[Blob] = {
+  def listBlobsByOffset(startOffset: String, endOffset: String): Seq[Blob] = {
     val blobs = Seq.newBuilder[Blob]
     var page = storage.list(
       config.bucketName,
@@ -56,6 +56,21 @@ class GcpBucket(config: GcpBucketConfig, override val loggerFactory: NamedLogger
     blobs.result()
   }
 
+  @SuppressWarnings(Array("org.wartremover.warts.Var", "org.wartremover.warts.While"))
+  def listBlobsByPrefix(prefix: String): Seq[Blob] = {
+    val blobs = Seq.newBuilder[Blob]
+    var page = storage.list(
+      config.bucketName,
+      Storage.BlobListOption.prefix(prefix),
+    )
+    blobs ++= page.getValues().asScala
+    while (page.hasNextPage) {
+      page = page.getNextPage
+      blobs ++= page.getValues().asScala
+    }
+    blobs.result()
+  }
+
   def readBytesFromBucket(fileName: String): Array[Byte] = {
     val blobId = BlobId.of(config.bucketName, fileName)
     val blob = storage.get(blobId)