[ci] logging and docs

Signed-off-by: Itai Segall <itai.segall@digitalasset.com>

Author: isegall-da

apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/AcsSnapshotBulkStorage.scala

@@ -28,7 +28,8 @@ class AcsSnapshotBulkStorage(
     extends NamedLogging {
 
   // TODO(#3429): persist progress (or conclude it from the S3 storage), and start from latest successfully dumped snapshot upon restart
-  private def getStartTimestamp: Future[Option[(Long, CantonTimestamp)]] = kvProvider.getLatestAcsSnapshotInBulkStorage().value
+  private def getStartTimestamp: Future[Option[(Long, CantonTimestamp)]] =
+    kvProvider.getLatestAcsSnapshotInBulkStorage().value
 
   // When new snapshot is not yet available, how long to wait for a new one.
   // TODO(#3429): make it longer for prod (so consider making it configurable/overridable for tests)
@@ -64,37 +65,47 @@ class AcsSnapshotBulkStorage(
       .collect { case Some((migrationId, timestamp)) => (migrationId, timestamp) }
   }
 
-  /** This is the main implementation of the pipeline. It is a Pekko Source that gets a `start` timestamp, and starts dumping to S3
-    *  all snapshots (strictly) after `start`. It is an infinite source that should never complete.
+  /**  This is the main implementation of the pipeline. It is a Pekko Source that reads a `start` timestamp
+    *   from the DB, and starts dumping to S3 all snapshots (strictly) after `start`. After every snapshot that
+    *   is successfully dumped, it persists to the DB its timestamp.
+    *   It is an infinite source that should never complete.
     */
-  private def mksrc(): Source[(Long, CantonTimestamp), (KillSwitch, Future[Done])] =  {
-      logger.debug("Starting ACS snapshot dump source")
-      val base =
-        Source.single[Unit](())
-          .mapAsync(1){_ => getStartTimestamp}
-          .flatMapConcat {
-            case Some((startMigrationId, startAfterTimestamp)) => getAcsSnapshotTimestampsAfter(startMigrationId, startAfterTimestamp)
-            case None => getAcsSnapshotTimestampsAfter(0, CantonTimestamp.MinValue)
-          }
-          .via(
-            SingleAcsSnapshotBulkStorage.asFlow(
-              config,
-              acsSnapshotStore,
-              s3Connection,
-              loggerFactory,
+  private def mksrc(): Source[(Long, CantonTimestamp), (KillSwitch, Future[Done])] = {
+    val base =
+      Source
+        .single[Unit](())
+        .mapAsync(1) { _ => getStartTimestamp }
+        .flatMapConcat {
+          case Some((startMigrationId, startAfterTimestamp)) =>
+            logger.info(
+              s"Latest dumped snapshot was from migration $startMigrationId, timestamp $startAfterTimestamp"
             )
+            getAcsSnapshotTimestampsAfter(startMigrationId, startAfterTimestamp)
+          case None =>
+            logger.info("Not dumped snapshots yet, starting from genesis")
+            getAcsSnapshotTimestampsAfter(0, CantonTimestamp.MinValue)
+        }
+        .via(
+          SingleAcsSnapshotBulkStorage.asFlow(
+            config,
+            acsSnapshotStore,
+            s3Connection,
+            loggerFactory,
           )
-          .mapAsync(1){case (migrationId, timestamp) =>
-            for {
-              _ <- kvProvider.setLatestAcsSnapshotsInBulkStorage(migrationId, timestamp)
-            } yield {
-              (migrationId, timestamp)
-            }
+        )
+        .mapAsync(1) { case (migrationId, timestamp) =>
+          for {
+            _ <- kvProvider.setLatestAcsSnapshotsInBulkStorage(migrationId, timestamp)
+          } yield {
+            logger.info(
+              s"Successfully completed dumping snapshots from migration $migrationId, timestamp $timestamp"
+            )
+            (migrationId, timestamp)
           }
-
+        }
 
     val withKs = base.viaMat(KillSwitches.single)(Keep.right)
-      withKs.watchTermination() { case (ks, done) => (ks: KillSwitch, done) }
+    withKs.watchTermination() { case (ks, done) => (ks: KillSwitch, done) }
   }
 
   /**  wraps mksrc (where the main pipeline logic is implemented) in a retry loop, to retry upon failures.

apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/ScanKeyValueProviderTest.scala

@@ -13,7 +13,7 @@ import org.lfdecentralizedtrust.splice.store.db.SplicePostgresTest
 import org.scalatest.matchers.should.Matchers
 
 import scala.concurrent.Future
-import scala.concurrent.duration._
+import scala.concurrent.duration.*
 
 class ScanKeyValueProviderTest
     extends StoreTest

apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/AcsSnapshotBulkStorageTest.scala

@@ -122,7 +122,7 @@ class AcsSnapshotBulkStorageTest
 
           _ = clue("Initially, a single snapshot is dumped") {
             probe.request(2)
-            probe.expectNext(2.minutes).value shouldBe(0, ts1)
+            probe.expectNext(2.minutes).value shouldBe (0, ts1)
             probe.expectNoMessage(10.seconds)
           }
           persistedTs1 <- kvProvider.getLatestAcsSnapshotInBulkStorage().value
@@ -131,7 +131,7 @@ class AcsSnapshotBulkStorageTest
           _ = clue("Add another snapshot to the store, it is also dumped") {
             store.addSnapshot(CantonTimestamp.tryFromInstant(Instant.ofEpochSecond(20)))
             val next = probe.expectNext(2.minutes)
-            next.value shouldBe(0, ts2)
+            next.value shouldBe (0, ts2)
             probe.expectNoMessage(10.seconds)
             next.killSwitch.shutdown()
           }