From 255e14e9512bb874a75194704e7f54b7103c85a0 Mon Sep 17 00:00:00 2001 From: Tim Emiola Date: Mon, 20 Jul 2026 17:34:46 +0900 Subject: [PATCH 1/7] [ci] Allow startedIngestingAt to be null Signed-off-by: Tim Emiola --- ...ord_meta_nullable_started_ingesting_at.sql | 5 + .../ScanVerdictIngestionService.scala | 1 + .../store/db/DbAppActivityRecordStore.scala | 47 +++++-- .../scan/store/db/DbScanVerdictStore.scala | 10 +- .../store/DbAppActivityRecordStoreTest.scala | 130 ++++++++++-------- 5 files changed, 119 insertions(+), 74 deletions(-) create mode 100644 apps/common/src/main/resources/db/migration/canton-network/postgres/stable/V073__app_activity_record_meta_nullable_started_ingesting_at.sql diff --git a/apps/common/src/main/resources/db/migration/canton-network/postgres/stable/V073__app_activity_record_meta_nullable_started_ingesting_at.sql b/apps/common/src/main/resources/db/migration/canton-network/postgres/stable/V073__app_activity_record_meta_nullable_started_ingesting_at.sql new file mode 100644 index 0000000000..7db98255dc --- /dev/null +++ b/apps/common/src/main/resources/db/migration/canton-network/postgres/stable/V073__app_activity_record_meta_nullable_started_ingesting_at.sql @@ -0,0 +1,5 @@ +-- Allow started_ingesting_at to be NULL so the firstSV bootstrap can +-- create the meta row (for round completeness) before the sequencer +-- starts serving traffic. The column is populated when the first +-- verdict batch with traffic summaries arrives. +alter table app_activity_record_meta alter column started_ingesting_at drop not null; diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/automation/ScanVerdictIngestionService.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/automation/ScanVerdictIngestionService.scala index 302df6e015..196cb1d69f 100644 --- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/automation/ScanVerdictIngestionService.scala +++ b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/automation/ScanVerdictIngestionService.scala @@ -267,6 +267,7 @@ class ScanVerdictIngestionService( items, appActivityRecords, lastArchivedRoundO, + hasTrafficSummaries = summaryByTime.nonEmpty, ) } yield { val lastRecordTime = verdicts.lastOption diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbAppActivityRecordStore.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbAppActivityRecordStore.scala index a60594dc25..aa2b61271d 100644 --- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbAppActivityRecordStore.scala +++ b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbAppActivityRecordStore.scala @@ -46,7 +46,8 @@ object DbAppActivityRecordStore { * @param codeVersion code version of the ingestion logic * @param userVersion operator-configured version from ScanAppConfig * @param startedIngestingAt record time (microseconds since epoch) of the first - * verdict in the first batch with activity records + * verdict batch with traffic summaries, or None during + * bootstrap before the sequencer serves traffic * @param earliestIngestedRound the earliest round number in the first batch with activity records * @param lastArchivedRound the highest round number which has been archived as of * the max record_time of the ingested verdicts @@ -55,7 +56,7 @@ object DbAppActivityRecordStore { historyId: Long, codeVersion: Int, userVersion: Int, - startedIngestingAt: Long, + startedIngestingAt: Option[Long], earliestIngestedRound: Long, lastArchivedRound: Option[Long], ) @@ -112,13 +113,13 @@ class DbAppActivityRecordStore( private val startedIngestingAtRef = new AtomicReference[Option[Long]](None) - /** The record time of the first activity record in the store. */ + /** The record time of the first verdict batch with traffic summaries. */ def startedIngestingAt(implicit tc: TraceContext): Future[Option[Long]] = startedIngestingAtRef.get() match { case some @ Some(_) => Future.successful(some) case None => lookupActivityRecordMeta(ingestionVersions.code, ingestionVersions.user).map { metaO => - val tsO = metaO.map(_.startedIngestingAt) + val tsO = metaO.flatMap(_.startedIngestingAt) tsO.foreach(ts => startedIngestingAtRef.compareAndSet(None, Some(ts))) tsO } @@ -303,11 +304,16 @@ class DbAppActivityRecordStore( * activity records exist (e.g., no featured app providers). * On a fresh firstSV with no archived rounds, bootstraps round 0 * as complete. + * + * @param hasTrafficSummaries true when the current verdict batch + * contains traffic summaries from the sequencer. Controls + * whether `started_ingesting_at` is populated in the meta row. */ def insertAppActivityRecordsDBIO( items: Seq[AppActivityRecordT], firstRecordTimeMicros: Long, lastArchivedRoundO: Option[Long] = None, + hasTrafficSummaries: Boolean = false, )(implicit tc: TraceContext): DBIO[Unit] = { val insertRecords = if (items.isEmpty) DBIO.successful(()) @@ -334,16 +340,22 @@ class DbAppActivityRecordStore( val lastArchived = lastArchivedRoundO .orElse(if (isFirstSv) Some(0L) else None) + val startedIngestingAtO = + if (hasTrafficSummaries) Some(firstRecordTimeMicros) else None + for { _ <- insertRecords ensureResult <- earliestRound match { case Some(earliest) => - ensureMetaDBIO((firstRecordTimeMicros, earliest), lastArchived) + ensureMetaDBIO((startedIngestingAtO, earliest), lastArchived) case None => // No archived rounds and not firstSV — skip meta creation. // A later verdict batch will create it. DBIO.successful(Resume: MetaCheckResult) } + _ <- + if (hasTrafficSummaries) updateStartedIngestingAtIfNullDBIO(firstRecordTimeMicros) + else DBIO.successful(0) _ <- (ensureResult, lastArchivedRoundO) match { // We already have meta row, so do the update in place. case (Resume, Some(round)) => updateLastArchivedRoundDBIO(round) @@ -381,7 +393,7 @@ class DbAppActivityRecordStore( historyId = prs.<<[Long], codeVersion = prs.<<[Int], userVersion = prs.<<[Int], - startedIngestingAt = prs.<<[Long], + startedIngestingAt = prs.<<[Option[Long]], earliestIngestedRound = prs.<<[Long], lastArchivedRound = prs.<<[Option[Long]], ) @@ -406,7 +418,7 @@ class DbAppActivityRecordStore( def insertActivityRecordMetaDBIO( codeVersion: Int, userVersion: Int, - startedIngestingAt: Long, + startedIngestingAt: Option[Long], earliestIngestedRound: Long, lastArchivedRound: Option[Long], ) = @@ -418,6 +430,15 @@ class DbAppActivityRecordStore( $earliestIngestedRound, $lastArchivedRound) """.asUpdate + private def updateStartedIngestingAtIfNullDBIO(ts: Long) = + sql"""update #${Tables.activityRecordMeta} + set started_ingesting_at = $ts + where history_id = $historyId + and activity_ingestion_code_version = ${ingestionVersions.code} + and activity_ingestion_user_version = ${ingestionVersions.user} + and started_ingesting_at is null + """.asUpdate + private def updateLastArchivedRoundDBIO(round: Long) = sql"""update #${Tables.activityRecordMeta} set last_archived_round = $round, @@ -431,7 +452,7 @@ class DbAppActivityRecordStore( def insertActivityRecordMetaForTesting( codeVersion: Int, userVersion: Int, - startedIngestingAt: Long, + startedIngestingAt: Option[Long], earliestIngestedRound: Long, lastArchivedRound: Option[Long], )(implicit tc: TraceContext): Future[Unit] = @@ -452,19 +473,19 @@ class DbAppActivityRecordStore( /** DBIO action that checks/inserts the meta row. * - * @param ingestionStart `Some((firstRecordTimeMicros, earliestRound))` when - * the batch has activity records, `None` otherwise. + * @param ingestionStart `(startedIngestingAtO, earliestRound)` — the timestamp + * is None during bootstrap (before traffic summaries are available) * @param lastArchivedRoundO the last archived round to store when inserting * a new meta row */ def ensureMetaDBIO( - ingestionStart: (Long, Long), + ingestionStart: (Option[Long], Long), lastArchivedRoundO: Option[Long] = None, exitOnDowngrade: Boolean = true, )(implicit tc: TraceContext): DBIO[MetaCheckResult] = { val codeVersion = ingestionVersions.code val userVersion = ingestionVersions.user - val (firstRecordTimeMicros, earliestRound) = ingestionStart + val (startedIngestingAtO, earliestRound) = ingestionStart if (metaChecked.get()) DBIO.successful(Resume) else { for { @@ -484,7 +505,7 @@ class DbAppActivityRecordStore( insertActivityRecordMetaDBIO( codeVersion, userVersion, - firstRecordTimeMicros, + startedIngestingAtO, earliestRound, lastArchivedRoundO, ).map { _ => diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbScanVerdictStore.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbScanVerdictStore.scala index 3e5cdf0646..2360aa574a 100644 --- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbScanVerdictStore.scala +++ b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbScanVerdictStore.scala @@ -486,6 +486,7 @@ class DbScanVerdictStore( items: NonEmptyList[(VerdictT, Long => Seq[TransactionViewT])], appActivityRecords: Seq[(CantonTimestamp, AppActivityRecordT)], lastArchivedRoundO: Option[Long] = None, + hasTrafficSummaries: Boolean = false, )(implicit tc: TraceContext): Future[Unit] = { import profile.api.jdbcActionExtensionMethods @@ -499,6 +500,7 @@ class DbScanVerdictStore( resolvedAppActivityRecords, items.head._1.recordTime.toMicros, lastArchivedRoundO, + hasTrafficSummaries, ) } yield () @@ -546,11 +548,17 @@ class DbScanVerdictStore( items: Seq[AppActivityRecordT], firstRecordTimeMicros: Long, lastArchivedRoundO: Option[Long], + hasTrafficSummaries: Boolean, )(implicit tc: TraceContext): DBIO[Unit] = appActivityRecordStoreO match { case None => DBIO.successful(()) case Some(s) => - s.insertAppActivityRecordsDBIO(items, firstRecordTimeMicros, lastArchivedRoundO) + s.insertAppActivityRecordsDBIO( + items, + firstRecordTimeMicros, + lastArchivedRoundO, + hasTrafficSummaries, + ) } private def afterFilters( diff --git a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/DbAppActivityRecordStoreTest.scala b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/DbAppActivityRecordStoreTest.scala index b4b57c17c0..13d94d33ba 100644 --- a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/DbAppActivityRecordStoreTest.scala +++ b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/DbAppActivityRecordStoreTest.scala @@ -67,7 +67,13 @@ class DbAppActivityRecordStoreTest } _ <- store.insertAppActivityRecordsForTesting(records) - _ <- store.insertActivityRecordMetaForTesting(1, 0, baseTs.toMicros, roundNumber, None) + _ <- store.insertActivityRecordMetaForTesting( + 1, + 0, + Some(baseTs.toMicros), + roundNumber, + None, + ) // Spot-check first, last and a middle record via row decoders first <- store.getRecordByVerdictRowIdForTesting(verdictRowIds(0)) middle <- store.getRecordByVerdictRowIdForTesting(verdictRowIds(25)) @@ -102,7 +108,7 @@ class DbAppActivityRecordStoreTest (store2, historyId2) <- newStore() baseTs = CantonTimestamp.now() Seq(rowId2) <- insertRecordsForRounds(store2, historyId2, baseTs, ("other", 10L)) - _ <- store2.insertActivityRecordMetaForTesting(1, 0, baseTs.toMicros, 10L, None) + _ <- store2.insertActivityRecordMetaForTesting(1, 0, Some(baseTs.toMicros), 10L, None) Seq(rowId1) <- insertRecordsForRounds( store1, historyId1, @@ -112,7 +118,7 @@ class DbAppActivityRecordStoreTest _ <- store1.insertActivityRecordMetaForTesting( 1, 0, - baseTs.plusSeconds(1L).toMicros, + Some(baseTs.plusSeconds(1L).toMicros), 20L, None, ) @@ -170,6 +176,7 @@ class DbAppActivityRecordStoreTest NonEmptyList.of(verdict1 -> noViews, verdict2 -> noViews), appActivityRecords, lastArchivedRoundO = Some(9L), + hasTrafficSummaries = true, ) // Verify verdicts were inserted @@ -194,7 +201,7 @@ class DbAppActivityRecordStoreTest r2.value.appProviderParties shouldBe Seq("app2::provider") meta shouldBe defined - meta.value.startedIngestingAt shouldBe baseTs.toMicros + meta.value.startedIngestingAt shouldBe Some(baseTs.toMicros) meta.value.earliestIngestedRound shouldBe 10L meta.value.lastArchivedRound shouldBe Some(9L) } @@ -210,6 +217,7 @@ class DbAppActivityRecordStoreTest NonEmptyList.of(mkVerdict(verdictStore, "update-mono-1", baseTs) -> noViews), Seq(baseTs -> mkRecord(0L, 10L, Seq("app1::provider"), Seq(100L))), lastArchivedRoundO = Some(9L), + hasTrafficSummaries = true, ) // A later batch without activity records still advances the round _ <- verdictStore.insertVerdictsWithAppActivityRecords( @@ -218,6 +226,7 @@ class DbAppActivityRecordStoreTest ), Seq.empty, lastArchivedRoundO = Some(10L), + hasTrafficSummaries = true, ) meta <- appStore.lookupActivityRecordMeta(1, 0) } yield { @@ -234,6 +243,7 @@ class DbAppActivityRecordStoreTest NonEmptyList.of(mkVerdict(verdictStore, "update-no-meta", baseTs) -> noViews), Seq.empty, lastArchivedRoundO = Some(7L), + hasTrafficSummaries = true, ) meta <- appStore.lookupActivityRecordMeta(1, 0) } yield { @@ -362,7 +372,7 @@ class DbAppActivityRecordStoreTest "return None when last_archived_round is not set" in { for { (store, _) <- newStore() - _ <- store.insertActivityRecordMetaForTesting(1, 0, 0L, 0L, None) + _ <- store.insertActivityRecordMetaForTesting(1, 0, Some(0L), 0L, None) result <- store.earliestRoundWithCompleteAppActivity() } yield { result shouldBe None @@ -373,7 +383,7 @@ class DbAppActivityRecordStoreTest for { (store, _) <- newStore() baseTs = CantonTimestamp.now() - _ <- store.insertActivityRecordMetaForTesting(1, 0, baseTs.toMicros, 42L, Some(42L)) + _ <- store.insertActivityRecordMetaForTesting(1, 0, Some(baseTs.toMicros), 42L, Some(42L)) result <- store.earliestRoundWithCompleteAppActivity() } yield { result shouldBe None @@ -384,7 +394,7 @@ class DbAppActivityRecordStoreTest for { (store, _) <- newStore() baseTs = CantonTimestamp.now() - _ <- store.insertActivityRecordMetaForTesting(1, 0, baseTs.toMicros, 42L, Some(43L)) + _ <- store.insertActivityRecordMetaForTesting(1, 0, Some(baseTs.toMicros), 42L, Some(43L)) result <- store.earliestRoundWithCompleteAppActivity() } yield { result.value shouldBe 43L @@ -398,7 +408,7 @@ class DbAppActivityRecordStoreTest _ <- insertRecordsForRounds(store, historyId, baseTs, ("gap-10", 10L)) // Only round 10 has activity; rounds 11 and 12 have zero records but // their archival makes them complete. - _ <- store.insertActivityRecordMetaForTesting(1, 0, baseTs.toMicros, 10L, Some(12L)) + _ <- store.insertActivityRecordMetaForTesting(1, 0, Some(baseTs.toMicros), 10L, Some(12L)) result <- store.earliestRoundWithCompleteAppActivity() } yield { result.value shouldBe 11L @@ -410,12 +420,12 @@ class DbAppActivityRecordStoreTest (store, _) <- newStore() baseTs = CantonTimestamp.now() // Old meta row from a previous ingestion run starting at round 10 - _ <- store.insertActivityRecordMetaForTesting(0, 0, baseTs.toMicros, 10L, Some(15L)) + _ <- store.insertActivityRecordMetaForTesting(0, 0, Some(baseTs.toMicros), 10L, Some(15L)) // Current meta row starting at round 20 _ <- store.insertActivityRecordMetaForTesting( 1, 0, - baseTs.plusSeconds(10L).toMicros, + Some(baseTs.plusSeconds(10L).toMicros), 20L, Some(25L), ) @@ -431,12 +441,12 @@ class DbAppActivityRecordStoreTest (store1, _) <- newStore() (store2, _) <- newStore() baseTs = CantonTimestamp.now() - _ <- store2.insertActivityRecordMetaForTesting(1, 0, baseTs.toMicros, 10L, Some(11L)) + _ <- store2.insertActivityRecordMetaForTesting(1, 0, Some(baseTs.toMicros), 10L, Some(11L)) // store1's meta has no archived round yet _ <- store1.insertActivityRecordMetaForTesting( 1, 0, - baseTs.plusSeconds(2L).toMicros, + Some(baseTs.plusSeconds(2L).toMicros), 50L, None, ) @@ -453,7 +463,7 @@ class DbAppActivityRecordStoreTest (store, historyId) <- newStore() baseTs = CantonTimestamp.now() // earliest_ingested_round = -1 so that round 0 = -1 + 1 is considered complete - _ <- store.insertActivityRecordMetaForTesting(1, 0, baseTs.toMicros, -1L, Some(1L)) + _ <- store.insertActivityRecordMetaForTesting(1, 0, Some(baseTs.toMicros), -1L, Some(1L)) _ <- insertRecordsForRounds( store, historyId, @@ -472,7 +482,7 @@ class DbAppActivityRecordStoreTest for { (store, historyId) <- newStore() baseTs = CantonTimestamp.now() - _ <- store.insertActivityRecordMetaForTesting(1, 0, baseTs.toMicros, -1L, None) + _ <- store.insertActivityRecordMetaForTesting(1, 0, Some(baseTs.toMicros), -1L, None) _ <- insertRecordsForRounds(store, historyId, baseTs, ("round-0", 0L)) result <- store.earliestRoundWithCompleteAppActivity() } yield { @@ -489,11 +499,11 @@ class DbAppActivityRecordStoreTest versions = DbAppActivityRecordStore.IngestionVersions(2, 0) ) baseTs = CantonTimestamp.now() - _ <- store.insertActivityRecordMetaForTesting(1, 0, baseTs.toMicros, 0L, Some(1L)) + _ <- store.insertActivityRecordMetaForTesting(1, 0, Some(baseTs.toMicros), 0L, Some(1L)) _ <- store.insertActivityRecordMetaForTesting( 2, 0, - baseTs.plusSeconds(10L).toMicros, + Some(baseTs.plusSeconds(10L).toMicros), 10L, Some(11L), ) @@ -517,9 +527,9 @@ class DbAppActivityRecordStoreTest ) baseTs = CantonTimestamp.now() // Version 1 meta row exists from prior run - _ <- store.insertActivityRecordMetaForTesting(1, 0, baseTs.toMicros, 0L, Some(1L)) + _ <- store.insertActivityRecordMetaForTesting(1, 0, Some(baseTs.toMicros), 0L, Some(1L)) // Version 2 meta row added by ensureMeta - _ <- runEnsureMeta(store, (baseTs.toMicros + 1000000L, 10L), Some(11L)) + _ <- runEnsureMeta(store, (Some(baseTs.toMicros + 1000000L), 10L), Some(11L)) _ <- insertRecordsForRounds( store, historyId, @@ -554,7 +564,7 @@ class DbAppActivityRecordStoreTest baseTs = CantonTimestamp.now() // Unlike earliestRoundWithCompleteAppActivity, this does not require any // archival to have happened yet. - _ <- store.insertActivityRecordMetaForTesting(1, 0, baseTs.toMicros, 42L, None) + _ <- store.insertActivityRecordMetaForTesting(1, 0, Some(baseTs.toMicros), 42L, None) result <- store.earliestIngestedRound() } yield { result.value shouldBe 42L @@ -565,11 +575,11 @@ class DbAppActivityRecordStoreTest for { (store, _) <- newStore() baseTs = CantonTimestamp.now() - _ <- store.insertActivityRecordMetaForTesting(0, 0, baseTs.toMicros, 10L, Some(15L)) + _ <- store.insertActivityRecordMetaForTesting(0, 0, Some(baseTs.toMicros), 10L, Some(15L)) _ <- store.insertActivityRecordMetaForTesting( 1, 0, - baseTs.plusSeconds(10L).toMicros, + Some(baseTs.plusSeconds(10L).toMicros), 20L, None, ) @@ -584,7 +594,7 @@ class DbAppActivityRecordStoreTest (store1, _) <- newStore() (store2, _) <- newStore() baseTs = CantonTimestamp.now() - _ <- store2.insertActivityRecordMetaForTesting(1, 0, baseTs.toMicros, 10L, Some(11L)) + _ <- store2.insertActivityRecordMetaForTesting(1, 0, Some(baseTs.toMicros), 10L, Some(11L)) result <- store1.earliestIngestedRound() } yield { result shouldBe None @@ -609,7 +619,7 @@ class DbAppActivityRecordStoreTest _ <- store.insertActivityRecordMetaForTesting( codeVersion = 1, userVersion = 0, - startedIngestingAt = 1000000L, + startedIngestingAt = Some(1000000L), earliestIngestedRound = 0L, lastArchivedRound = None, ) @@ -618,7 +628,7 @@ class DbAppActivityRecordStoreTest result shouldBe defined result.value.codeVersion shouldBe 1 result.value.userVersion shouldBe 0 - result.value.startedIngestingAt shouldBe 1000000L + result.value.startedIngestingAt shouldBe Some(1000000L) result.value.lastArchivedRound shouldBe None } } @@ -629,14 +639,14 @@ class DbAppActivityRecordStoreTest _ <- store.insertActivityRecordMetaForTesting( codeVersion = 1, userVersion = 0, - startedIngestingAt = 1000000L, + startedIngestingAt = Some(1000000L), earliestIngestedRound = 0L, lastArchivedRound = None, ) _ <- store.insertActivityRecordMetaForTesting( codeVersion = 2, userVersion = 1, - startedIngestingAt = 2000000L, + startedIngestingAt = Some(2000000L), earliestIngestedRound = 5L, lastArchivedRound = Some(7L), ) @@ -644,9 +654,9 @@ class DbAppActivityRecordStoreTest result2 <- store.lookupActivityRecordMeta(2, 1) resultMissing <- store.lookupActivityRecordMeta(3, 0) } yield { - result1.value.startedIngestingAt shouldBe 1000000L + result1.value.startedIngestingAt shouldBe Some(1000000L) result1.value.lastArchivedRound shouldBe None - result2.value.startedIngestingAt shouldBe 2000000L + result2.value.startedIngestingAt shouldBe Some(2000000L) result2.value.earliestIngestedRound shouldBe 5L result2.value.lastArchivedRound shouldBe Some(7L) resultMissing shouldBe None @@ -660,22 +670,22 @@ class DbAppActivityRecordStoreTest _ <- store1.insertActivityRecordMetaForTesting( codeVersion = 1, userVersion = 0, - startedIngestingAt = 1000000L, + startedIngestingAt = Some(1000000L), earliestIngestedRound = 0L, lastArchivedRound = None, ) _ <- store2.insertActivityRecordMetaForTesting( codeVersion = 1, userVersion = 0, - startedIngestingAt = 9000000L, + startedIngestingAt = Some(9000000L), earliestIngestedRound = 0L, lastArchivedRound = None, ) result1 <- store1.lookupActivityRecordMeta(1, 0) result2 <- store2.lookupActivityRecordMeta(1, 0) } yield { - result1.value.startedIngestingAt shouldBe 1000000L - result2.value.startedIngestingAt shouldBe 9000000L + result1.value.startedIngestingAt shouldBe Some(1000000L) + result2.value.startedIngestingAt shouldBe Some(9000000L) } } @@ -686,21 +696,21 @@ class DbAppActivityRecordStoreTest _ <- store1.insertActivityRecordMetaForTesting( codeVersion = 1, userVersion = 0, - startedIngestingAt = 1000000L, + startedIngestingAt = Some(1000000L), earliestIngestedRound = 0L, lastArchivedRound = None, ) _ <- store2.insertActivityRecordMetaForTesting( codeVersion = 1, userVersion = 0, - startedIngestingAt = 1000000L, + startedIngestingAt = Some(1000000L), earliestIngestedRound = 0L, lastArchivedRound = None, ) _ <- store1.insertActivityRecordMetaForTesting( codeVersion = 99, userVersion = 99, - startedIngestingAt = 9999999L, + startedIngestingAt = Some(9999999L), earliestIngestedRound = 0L, lastArchivedRound = None, ) @@ -708,7 +718,7 @@ class DbAppActivityRecordStoreTest } yield { result2.value.codeVersion shouldBe 1 result2.value.userVersion shouldBe 0 - result2.value.startedIngestingAt shouldBe 1000000L + result2.value.startedIngestingAt shouldBe Some(1000000L) } } } @@ -718,13 +728,13 @@ class DbAppActivityRecordStoreTest "insert meta on first call and resume on second" in { for { (store, _) <- newStore() - r1 <- runEnsureMeta(store, (1000000L, 10L), Some(9L)) - r2 <- runEnsureMeta(store, (1000000L, 10L)) + r1 <- runEnsureMeta(store, (Some(1000000L), 10L), Some(9L)) + r2 <- runEnsureMeta(store, (Some(1000000L), 10L)) meta <- store.lookupActivityRecordMeta(1, 0) } yield { r1 shouldBe InsertMeta r2 shouldBe Resume - meta.value.startedIngestingAt shouldBe 1000000L + meta.value.startedIngestingAt shouldBe Some(1000000L) meta.value.earliestIngestedRound shouldBe 10L meta.value.lastArchivedRound shouldBe Some(9L) } @@ -735,7 +745,7 @@ class DbAppActivityRecordStoreTest (store, _) <- newStore() // Before ensure, no meta row — startedIngestingAt returns None beforeO <- store.startedIngestingAt - _ <- runEnsureMeta(store, (1000000L, 10L)) + _ <- runEnsureMeta(store, (Some(1000000L), 10L)) // After ensure, the read path should load from DB afterO <- store.startedIngestingAt } yield { @@ -747,12 +757,12 @@ class DbAppActivityRecordStoreTest "return Resume when versions match existing meta" in { for { (store, _) <- newStore() - _ <- store.insertActivityRecordMetaForTesting(1, 0, 1000000L, 10L, None) - result <- runEnsureMeta(store, (2000000L, 20L)) + _ <- store.insertActivityRecordMetaForTesting(1, 0, Some(1000000L), 10L, None) + result <- runEnsureMeta(store, (Some(2000000L), 20L)) meta <- store.lookupActivityRecordMeta(1, 0) } yield { result shouldBe Resume - meta.value.startedIngestingAt shouldBe 1000000L + meta.value.startedIngestingAt shouldBe Some(1000000L) meta.value.earliestIngestedRound shouldBe 10L } } @@ -760,14 +770,14 @@ class DbAppActivityRecordStoreTest "insert new row and return InsertMeta on version bump" in { for { (store, _) <- newStore(DbAppActivityRecordStore.IngestionVersions(2, 0)) - _ <- store.insertActivityRecordMetaForTesting(1, 0, 1000000L, 10L, Some(15L)) - result <- runEnsureMeta(store, (2000000L, 20L), Some(19L)) + _ <- store.insertActivityRecordMetaForTesting(1, 0, Some(1000000L), 10L, Some(15L)) + result <- runEnsureMeta(store, (Some(2000000L), 20L), Some(19L)) meta <- store.lookupActivityRecordMeta(2, 0) oldMeta <- store.lookupActivityRecordMeta(1, 0) } yield { result shouldBe InsertMeta meta.value.codeVersion shouldBe 2 - meta.value.startedIngestingAt shouldBe 2000000L + meta.value.startedIngestingAt shouldBe Some(2000000L) meta.value.earliestIngestedRound shouldBe 20L meta.value.lastArchivedRound shouldBe Some(19L) oldMeta.value.lastArchivedRound shouldBe Some(15L) @@ -777,9 +787,9 @@ class DbAppActivityRecordStoreTest "return DowngradeDetected without modifying the row" in { for { (store, _) <- newStore() - _ <- store.insertActivityRecordMetaForTesting(2, 0, 1000000L, 10L, None) + _ <- store.insertActivityRecordMetaForTesting(2, 0, Some(1000000L), 10L, None) result <- loggerFactory.assertLogs( - runEnsureMeta(store, (2000000L, 20L)), + runEnsureMeta(store, (Some(2000000L), 20L)), _.errorMessage should include( "App activity ingestion version downgrade detected" ), @@ -788,7 +798,7 @@ class DbAppActivityRecordStoreTest } yield { result shouldBe DowngradeDetected(1, 0, 2, 0) meta.value.codeVersion shouldBe 2 - meta.value.startedIngestingAt shouldBe 1000000L + meta.value.startedIngestingAt shouldBe Some(1000000L) meta.value.earliestIngestedRound shouldBe 10L } } @@ -796,7 +806,7 @@ class DbAppActivityRecordStoreTest "use earliestRound as provided by caller" in { for { (store, _) <- newStore() - r1 <- runEnsureMeta(store, (1000000L, 10L)) + r1 <- runEnsureMeta(store, (Some(1000000L), 10L)) meta <- store.lookupActivityRecordMeta(1, 0) } yield { r1 shouldBe InsertMeta @@ -809,8 +819,8 @@ class DbAppActivityRecordStoreTest (store, _) <- newStore( versions = DbAppActivityRecordStore.IngestionVersions(2, 0) ) - _ <- store.insertActivityRecordMetaForTesting(1, 0, 1000000L, 5L, None) - r1 <- runEnsureMeta(store, (2000000L, 10L)) + _ <- store.insertActivityRecordMetaForTesting(1, 0, Some(1000000L), 5L, None) + r1 <- runEnsureMeta(store, (Some(2000000L), 10L)) meta <- store.lookupActivityRecordMeta(2, 0) } yield { r1 shouldBe InsertMeta @@ -826,7 +836,7 @@ class DbAppActivityRecordStoreTest (store, _) <- newStore() // Simulate: meta created with earliestRound = 5 (from lastArchivedRound), // then lastArchivedRound advances to 7 - _ <- store.insertActivityRecordMetaForTesting(1, 0, 1000000L, 5L, Some(7L)) + _ <- store.insertActivityRecordMetaForTesting(1, 0, Some(1000000L), 5L, Some(7L)) result <- store.earliestRoundWithCompleteAppActivity() } yield { result.value shouldBe 6L @@ -838,7 +848,7 @@ class DbAppActivityRecordStoreTest (store, _) <- newStore() // Fresh network: earliestRound = -1 (no lastArchivedRound at meta creation), // then lastArchivedRound advances to 0 after first round closes - _ <- store.insertActivityRecordMetaForTesting(1, 0, 1000000L, -1L, Some(0L)) + _ <- store.insertActivityRecordMetaForTesting(1, 0, Some(1000000L), -1L, Some(0L)) result <- store.earliestRoundWithCompleteAppActivity() } yield { result.value shouldBe 0L @@ -881,7 +891,7 @@ class DbAppActivityRecordStoreTest _ <- store.insertActivityRecordMetaForTesting( 1, 0, - CantonTimestamp.now().toMicros, + Some(CantonTimestamp.now().toMicros), 42L, None, ) @@ -896,7 +906,7 @@ class DbAppActivityRecordStoreTest (store, historyId) <- newStore() baseTs = CantonTimestamp.now() _ <- insertRecordsForRounds(store, historyId, baseTs, ("gap-10", 10L)) - _ <- store.insertActivityRecordMetaForTesting(1, 0, baseTs.toMicros, 10L, Some(12L)) + _ <- store.insertActivityRecordMetaForTesting(1, 0, Some(baseTs.toMicros), 10L, Some(12L)) result <- store.latestRoundWithCompleteAppActivity() } yield { result.value shouldBe 12L @@ -908,7 +918,7 @@ class DbAppActivityRecordStoreTest (store1, _) <- newStore() (store2, _) <- newStore() baseTs = CantonTimestamp.now() - _ <- store2.insertActivityRecordMetaForTesting(1, 0, baseTs.toMicros, 10L, Some(11L)) + _ <- store2.insertActivityRecordMetaForTesting(1, 0, Some(baseTs.toMicros), 10L, Some(11L)) result <- store1.latestRoundWithCompleteAppActivity() } yield { result shouldBe None @@ -933,7 +943,7 @@ class DbAppActivityRecordStoreTest _ <- store.insertActivityRecordMetaForTesting( 1, 0, - CantonTimestamp.now().toMicros, + Some(CantonTimestamp.now().toMicros), 10L, None, ) @@ -949,7 +959,7 @@ class DbAppActivityRecordStoreTest _ <- store.insertActivityRecordMetaForTesting( 1, 0, - CantonTimestamp.now().toMicros, + Some(CantonTimestamp.now().toMicros), 10L, Some(12L), ) @@ -1000,7 +1010,7 @@ class DbAppActivityRecordStoreTest private def runEnsureMeta( store: DbAppActivityRecordStore, - ingestionStart: (Long, Long), + ingestionStart: (Option[Long], Long), lastArchivedRoundO: Option[Long] = None, ): Future[MetaCheckResult] = futureUnlessShutdownToFuture( From 45497d87d57b457de0aa1b69b8ccb41ffaa0f429 Mon Sep 17 00:00:00 2001 From: Tim Emiola Date: Wed, 22 Jul 2026 13:22:48 +0900 Subject: [PATCH 2/7] [ci] Make earlierRound always -1 for the firstSV Signed-off-by: Tim Emiola --- .../store/db/DbAppActivityRecordStore.scala | 15 ++++--- .../store/DbAppActivityRecordStoreTest.scala | 44 ++++++++++++++++++- 2 files changed, 51 insertions(+), 8 deletions(-) diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbAppActivityRecordStore.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbAppActivityRecordStore.scala index aa2b61271d..232623d334 100644 --- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbAppActivityRecordStore.scala +++ b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbAppActivityRecordStore.scala @@ -323,14 +323,17 @@ class DbAppActivityRecordStore( } // earliestRound: the lowest round covered by this ingestion batch. + // - Always -1 on firstSV (present since genesis, round 0 is complete) // - From activity records when present // - From lastArchivedRound when no featured apps produced records - // - From bootstrap (-1) on a fresh firstSV with no archived rounds - val earliestRound = items - .map(_.roundNumber) - .minOption - .orElse(lastArchivedRoundO) - .orElse(if (isFirstSv) Some(-1L) else None) + val earliestRound = if (isFirstSv) { + Some(-1L) + } else { + items + .map(_.roundNumber) + .minOption + .orElse(lastArchivedRoundO) + } // lastArchived: the highest round archived as of this verdict batch. // - From the caller when available diff --git a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/DbAppActivityRecordStoreTest.scala b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/DbAppActivityRecordStoreTest.scala index 13d94d33ba..78e1d8e780 100644 --- a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/DbAppActivityRecordStoreTest.scala +++ b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/DbAppActivityRecordStoreTest.scala @@ -356,6 +356,44 @@ class DbAppActivityRecordStoreTest countAfter shouldBe 0L } } + + "set earliestIngestedRound to -1 on firstSV even when lastArchivedRoundO is present" in { + for { + (appStore, verdictStore) <- newStores(isFirstSv = true) + baseTs = CantonTimestamp.now() + + _ <- verdictStore.insertVerdictsWithAppActivityRecords( + NonEmptyList.of(mkVerdict(verdictStore, "update-firstsv-lagged", baseTs) -> noViews), + Seq.empty, + lastArchivedRoundO = Some(1L), + hasTrafficSummaries = true, + ) + meta <- appStore.lookupActivityRecordMeta(1, 0) + } yield { + meta shouldBe defined + meta.value.earliestIngestedRound shouldBe -1L + meta.value.lastArchivedRound shouldBe Some(1L) + } + } + + "set earliestIngestedRound to -1 on firstSV even when activity records exist" in { + for { + (appStore, verdictStore) <- newStores(isFirstSv = true) + baseTs = CantonTimestamp.now() + + _ <- verdictStore.insertVerdictsWithAppActivityRecords( + NonEmptyList.of(mkVerdict(verdictStore, "update-firstsv-records", baseTs) -> noViews), + Seq(baseTs -> mkRecord(0L, 0L, Seq("app1::provider"), Seq(100L))), + lastArchivedRoundO = Some(1L), + hasTrafficSummaries = true, + ) + meta <- appStore.lookupActivityRecordMeta(1, 0) + } yield { + meta shouldBe defined + meta.value.earliestIngestedRound shouldBe -1L + meta.value.lastArchivedRound shouldBe Some(1L) + } + } } "earliestRoundWithCompleteAppActivity" should { @@ -1061,7 +1099,9 @@ class DbAppActivityRecordStoreTest /** Creates both an app activity record store and a verdict store backed by * the same UpdateHistory, for testing insertVerdictsWithAppActivityRecords. */ - private def newStores(): Future[(DbAppActivityRecordStore, DbScanVerdictStore)] = { + private def newStores( + isFirstSv: Boolean = false + ): Future[(DbAppActivityRecordStore, DbScanVerdictStore)] = { val participantId = mkParticipantId("activity-test") val updateHistory = new UpdateHistory( storage.underlying, @@ -1080,7 +1120,7 @@ class DbAppActivityRecordStoreTest storage.underlying, updateHistory, DbAppActivityRecordStore.IngestionVersions(1, 0), - isFirstSv = false, + isFirstSv = isFirstSv, loggerFactory, ) val verdictStore = new DbScanVerdictStore( From 47b8ce3a6ec072ba2aa790cc8c8cba79ebf607a3 Mon Sep 17 00:00:00 2001 From: Tim Emiola Date: Wed, 22 Jul 2026 13:33:02 +0900 Subject: [PATCH 3/7] [ci] Add ensureMetaDBIO test for startedIngestingAtO = None Signed-off-by: Tim Emiola --- .../scan/store/DbAppActivityRecordStoreTest.scala | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/DbAppActivityRecordStoreTest.scala b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/DbAppActivityRecordStoreTest.scala index 78e1d8e780..5dd28c8384 100644 --- a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/DbAppActivityRecordStoreTest.scala +++ b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/DbAppActivityRecordStoreTest.scala @@ -852,6 +852,21 @@ class DbAppActivityRecordStoreTest } } + "insert meta with started_ingesting_at = NULL when startedIngestingAtO is None" in { + for { + (store, _) <- newStore() + r1 <- runEnsureMeta(store, (None, -1L), Some(0L)) + meta <- store.lookupActivityRecordMeta(1, 0) + startedAt <- store.startedIngestingAt + } yield { + r1 shouldBe InsertMeta + meta.value.startedIngestingAt shouldBe None + meta.value.earliestIngestedRound shouldBe -1L + meta.value.lastArchivedRound shouldBe Some(0L) + startedAt shouldBe None + } + } + "use actual earliest_ingested_round on version bump" in { for { (store, _) <- newStore( From fcf6f332d34b25a0246969a73d9e38e8cad837ac Mon Sep 17 00:00:00 2001 From: Tim Emiola Date: Wed, 22 Jul 2026 13:22:03 +0900 Subject: [PATCH 4/7] [ci] Fix stale scaladoc on AppActivityStore.startedIngestingAt Signed-off-by: Tim Emiola --- .../splice/scan/store/AppActivityStore.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AppActivityStore.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AppActivityStore.scala index 2b72c75e71..3cf9009078 100644 --- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AppActivityStore.scala +++ b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AppActivityStore.scala @@ -36,6 +36,6 @@ trait AppActivityStore { tc: TraceContext ): Future[Option[Long]] - /** The record time of the first activity record in the store. */ + /** The record time of the first verdict batch with traffic summaries. */ def startedIngestingAt(implicit tc: TraceContext): Future[Option[Long]] } From a85efbe8da2863e162c7c2d37d6d72ee5fa6e73b Mon Sep 17 00:00:00 2001 From: Tim Emiola Date: Wed, 22 Jul 2026 14:05:15 +0900 Subject: [PATCH 5/7] [ci] confirm that startedIngestingAt gets populated later Signed-off-by: Tim Emiola --- .../store/DbAppActivityRecordStoreTest.scala | 33 +++++++++++++++++++ 1 file changed, 33 insertions(+) diff --git a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/DbAppActivityRecordStoreTest.scala b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/DbAppActivityRecordStoreTest.scala index 5dd28c8384..8f958bb5ac 100644 --- a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/DbAppActivityRecordStoreTest.scala +++ b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/DbAppActivityRecordStoreTest.scala @@ -394,6 +394,39 @@ class DbAppActivityRecordStoreTest meta.value.lastArchivedRound shouldBe Some(1L) } } + + "populate startedIngestingAt on second batch when first batch has no traffic summaries" in { + for { + (appStore, verdictStore) <- newStores(isFirstSv = true) + baseTs = CantonTimestamp.now() + secondTs = baseTs.plusSeconds(10) + + // First batch: no traffic summaries — meta row created with started_ingesting_at = NULL + _ <- verdictStore.insertVerdictsWithAppActivityRecords( + NonEmptyList.of(mkVerdict(verdictStore, "update-bootstrap", baseTs) -> noViews), + Seq.empty, + lastArchivedRoundO = Some(0L), + hasTrafficSummaries = false, + ) + meta1 <- appStore.lookupActivityRecordMeta(1, 0) + startedAt1 <- appStore.startedIngestingAt + + // Second batch: traffic summaries arrive — updateStartedIngestingAtIfNullDBIO fills in the value + _ <- verdictStore.insertVerdictsWithAppActivityRecords( + NonEmptyList.of(mkVerdict(verdictStore, "update-with-summaries", secondTs) -> noViews), + Seq.empty, + lastArchivedRoundO = Some(0L), + hasTrafficSummaries = true, + ) + startedAt2 <- appStore.startedIngestingAt + } yield { + meta1 shouldBe defined + meta1.value.earliestIngestedRound shouldBe -1L + meta1.value.startedIngestingAt shouldBe None + startedAt1 shouldBe None + startedAt2 shouldBe Some(secondTs.toMicros) + } + } } "earliestRoundWithCompleteAppActivity" should { From e35051b396fdd0a57fb9d31fba682697646b8ecb Mon Sep 17 00:00:00 2001 From: Tim Emiola Date: Wed, 22 Jul 2026 14:38:56 +0900 Subject: [PATCH 6/7] [ci] Clarify why missing round data cannot occur after ingestion starts Signed-off-by: Tim Emiola --- .../splice/scan/rewards/AppActivityComputation.scala | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/rewards/AppActivityComputation.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/rewards/AppActivityComputation.scala index 30b4b9ede5..71a7ab7a24 100644 --- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/rewards/AppActivityComputation.scala +++ b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/rewards/AppActivityComputation.scala @@ -125,8 +125,10 @@ class AppActivityComputation( } case None => // Skip activity record computation as we don't have the necessary round data ingested. - // This can happen for freshly onboarded SVs, but is not - // expected to happen once the first activity record has been computed. + // This can happen for freshly onboarded SVs whose round history + // doesn't cover the verdict's sequencing time. It cannot happen + // after ingestion starts because lookupActiveOpenMiningRounds blocks + // until the reference store has caught up to the verdict batch. logger.debug( s"No round data found for sequencingTime=${summary.sequencingTime}, skipping activity record computation" ) From 5e792a1bee69652fe491619cccb638ebd0e0d2a8 Mon Sep 17 00:00:00 2001 From: Tim Emiola Date: Thu, 23 Jul 2026 11:51:56 +0900 Subject: [PATCH 7/7] [ci] Defer non-firstSV meta row creation until traffic summaries are available Signed-off-by: Tim Emiola --- .../store/db/DbAppActivityRecordStore.scala | 3 +- .../store/DbAppActivityRecordStoreTest.scala | 41 +++++++++++++++++++ 2 files changed, 43 insertions(+), 1 deletion(-) diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbAppActivityRecordStore.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbAppActivityRecordStore.scala index 232623d334..40d7ded4c4 100644 --- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbAppActivityRecordStore.scala +++ b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbAppActivityRecordStore.scala @@ -326,13 +326,14 @@ class DbAppActivityRecordStore( // - Always -1 on firstSV (present since genesis, round 0 is complete) // - From activity records when present // - From lastArchivedRound when no featured apps produced records + // but traffic summaries are available val earliestRound = if (isFirstSv) { Some(-1L) } else { items .map(_.roundNumber) .minOption - .orElse(lastArchivedRoundO) + .orElse(if (hasTrafficSummaries) lastArchivedRoundO else None) } // lastArchived: the highest round archived as of this verdict batch. diff --git a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/DbAppActivityRecordStoreTest.scala b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/DbAppActivityRecordStoreTest.scala index 8f958bb5ac..07fb0c55c1 100644 --- a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/DbAppActivityRecordStoreTest.scala +++ b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/DbAppActivityRecordStoreTest.scala @@ -427,6 +427,47 @@ class DbAppActivityRecordStoreTest startedAt2 shouldBe Some(secondTs.toMicros) } } + + "defer non-firstSV meta row creation until traffic summaries are available" in { + for { + (appStore, verdictStore) <- newStores(isFirstSv = false) + baseTs = CantonTimestamp.now() + secondTs = baseTs.plusSeconds(10) + + // First batch: no traffic summaries — meta row should NOT be created + _ <- verdictStore.insertVerdictsWithAppActivityRecords( + NonEmptyList.of(mkVerdict(verdictStore, "update-bootstrap", baseTs) -> noViews), + Seq.empty, + lastArchivedRoundO = Some(5L), + hasTrafficSummaries = false, + ) + meta1 <- appStore.lookupActivityRecordMeta(1, 0) + startedAt1 <- appStore.startedIngestingAt + earliest1 <- appStore.earliestRoundWithCompleteAppActivity() + + // Second batch: traffic summaries arrive — meta row created + _ <- verdictStore.insertVerdictsWithAppActivityRecords( + NonEmptyList.of(mkVerdict(verdictStore, "update-with-summaries", secondTs) -> noViews), + Seq.empty, + lastArchivedRoundO = Some(5L), + hasTrafficSummaries = true, + ) + meta2 <- appStore.lookupActivityRecordMeta(1, 0) + startedAt2 <- appStore.startedIngestingAt + } yield { + // No meta row after first batch + meta1 shouldBe None + startedAt1 shouldBe None + earliest1 shouldBe None + + // Meta row created on second batch with proper values + meta2 shouldBe defined + meta2.value.earliestIngestedRound shouldBe 5L + meta2.value.lastArchivedRound shouldBe Some(5L) + meta2.value.startedIngestingAt shouldBe Some(secondTs.toMicros) + startedAt2 shouldBe Some(secondTs.toMicros) + } + } } "earliestRoundWithCompleteAppActivity" should {