Fix ACS snapshots on short-lived migrations (#2781)

[ci]

---------

Signed-off-by: Oriol Muñoz <oriol.munoz@digitalasset.com>
Signed-off-by: Moritz Kiefer <moritz.kiefer@purelyfunctional.org>

Author: OriolMunoz-da

apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/automation/AcsSnapshotTrigger.scala

@@ -167,6 +167,26 @@ class AcsSnapshotTrigger(
   ): Future[Option[AcsSnapshotTrigger.Task]] = {
     for {
       migrationRecordTimeRange <- updateHistory.getRecordTimeRange(migrationIdToBackfill)
+      maxTime = migrationRecordTimeRange
+        .map(_._2.max)
+        .maxOption
+        .getOrElse(
+          throw new IllegalStateException(
+            s"SynchronizerId with no data in $migrationRecordTimeRange"
+          )
+        )
+      minTime = migrationRecordTimeRange
+        .map(_._2.min)
+        .minOption
+        .getOrElse(
+          throw new IllegalStateException(
+            s"SynchronizerId with no data in $migrationRecordTimeRange"
+          )
+        )
+      firstSnapshotTime = computeFirstSnapshotTime(minTime)
+      migrationLastedLongEnough = firstSnapshotTime
+        .plus(Duration.ofHours(snapshotPeriodHours.toLong))
+        .isBefore(maxTime)
       latestSnapshot <- store
         .lookupSnapshotBefore(migrationIdToBackfill, CantonTimestamp.MaxValue)
       task <- latestSnapshot match {
@@ -176,14 +196,7 @@ class AcsSnapshotTrigger(
         case Some(snapshot)
             if snapshot.snapshotRecordTime.plus(
               Duration.ofHours(snapshotPeriodHours.toLong)
-            ) > migrationRecordTimeRange
-              .map(_._2.max)
-              .maxOption
-              .getOrElse(
-                throw new IllegalStateException(
-                  s"SynchronizerId with no data in $migrationRecordTimeRange"
-                )
-              ) =>
+            ) > maxTime =>
           logger.info(
             s"Backfilling of migration id $migrationIdToBackfill is complete. Trying with next oldest."
           )
@@ -196,6 +209,12 @@ class AcsSnapshotTrigger(
                 .Task(nextSnapshotTime(snapshot), migrationIdToBackfill, Some(snapshot))
             )
           )
+        case None if !migrationLastedLongEnough =>
+          logger.info(
+            s"Migration id $migrationIdToBackfill didn't last more than $snapshotPeriodHours hours (from $minTime to $maxTime), so it won't have any snapshots."
+          )
+          lastCompleteBackfilledMigrationId.set(Right(migrationIdToBackfill))
+          taskToContinueBackfillingACSSnapshots().value
         case None =>
           firstSnapshotForMigrationIdTask(migrationIdToBackfill)
       }
@@ -249,20 +268,27 @@ class AcsSnapshotTrigger(
           logger.info(s"No updates other than ACS imports found. Retrying snapshot creation later.")
           None
         case Some(firstNonAcsImport) =>
-          val firstNonAcsImportRecordTime =
-            firstNonAcsImport.update.update.recordTime.toInstant.atOffset(ZoneOffset.UTC)
-          val (hourForSnapshot, plusDays) = timesToDoSnapshot
-            .find(_ > firstNonAcsImportRecordTime.get(ChronoField.HOUR_OF_DAY)) match {
-            case Some(hour) => hour -> 0 // current day at hour
-            case None => 0 -> 1 // next day at 00:00
-          }
-          val until = firstNonAcsImportRecordTime.toLocalDate
-            .plusDays(plusDays.toLong)
-            .atTime(hourForSnapshot, 0)
-            .toInstant(ZoneOffset.UTC)
-          Some(AcsSnapshotTrigger.Task(CantonTimestamp.assertFromInstant(until), migrationId, None))
+          val firstNonAcsImportRecordTime = firstNonAcsImport.update.update.recordTime
+          Some(
+            AcsSnapshotTrigger
+              .Task(computeFirstSnapshotTime(firstNonAcsImportRecordTime), migrationId, None)
+          )
       }
   }
+
+  private def computeFirstSnapshotTime(firstNonAcsImportRecordTime: CantonTimestamp) = {
+    val firstUpdateUTCTime = firstNonAcsImportRecordTime.toInstant.atOffset(ZoneOffset.UTC)
+    val (hourForSnapshot, plusDays) = timesToDoSnapshot
+      .find(_ > firstUpdateUTCTime.get(ChronoField.HOUR_OF_DAY)) match {
+      case Some(hour) => hour -> 0 // current day at hour
+      case None => 0 -> 1 // next day at 00:00
+    }
+    val until = firstUpdateUTCTime.toLocalDate
+      .plusDays(plusDays.toLong)
+      .atTime(hourForSnapshot, 0)
+      .toInstant(ZoneOffset.UTC)
+    CantonTimestamp.assertFromInstant(until)
+  }
 }
 
 object AcsSnapshotTrigger {

apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/automation/AcsSnapshotTriggerTest.scala

@@ -346,6 +346,31 @@ class AcsSnapshotTriggerTest
             trigger.isDoneBackfillingAcsSnapshots should be(false)
           }
 
+          "return no task if the previous migration id is backfilled but didn't last long enough" in new AcsSnapshotTriggerTestScope(
+            true,
+            currentMigrationId = 1L,
+          ) {
+            // no snapshot needed for the current migration id
+            historyBackfilled(currentMigrationId, complete = true)
+            previousSnapshot(now.minusSeconds(60L), currentMigrationId)
+
+            // therefore we attempt to backfill the previous migration id,
+            // which is backfilled
+            val previousMigrationId = currentMigrationId - 1
+            historyBackfilled(previousMigrationId, complete = true)
+            // last update happened shortly after the end
+            recordTimeRange(
+              previousMigrationId,
+              // didn't last enough
+              min = CantonTimestamp.MinValue.plusSeconds(3600L - 1),
+              max = CantonTimestamp.MinValue.plusSeconds(3600L + 600),
+            )
+            noPreviousSnapshot(previousMigrationId)
+
+            trigger.retrieveTasks().futureValue should be(Seq.empty)
+            trigger.isDoneBackfillingAcsSnapshots should be(true)
+          }
+
           "return the first task to backfill the previous migration id" in new AcsSnapshotTriggerTestScope(
             true
           ) {
@@ -707,11 +732,15 @@ class AcsSnapshotTriggerTest
         )
     }
 
-    def recordTimeRange(migrationId: Long, max: CantonTimestamp): Unit = {
+    def recordTimeRange(
+        migrationId: Long,
+        max: CantonTimestamp,
+        min: CantonTimestamp = CantonTimestamp.MinValue,
+    ): Unit = {
       when(updateHistory.getRecordTimeRange(eqTo(migrationId))(any[TraceContext]))
         .thenReturn(
           Future.successful(
-            Map(dummyDomain -> DomainRecordTimeRange(CantonTimestamp.MinValue, max))
+            Map(dummyDomain -> DomainRecordTimeRange(min, max))
           )
         )
     }