Traffic-Based App Rewards: Resolve bootstrap errors

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;

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

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"
                 )

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]]
 }

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(())
@@ -317,14 +323,18 @@ 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)
+    //     but traffic summaries are available
+    val earliestRound = if (isFirstSv) {
+      Some(-1L)
+    } else {
+      items
+        .map(_.roundNumber)
+        .minOption
+        .orElse(if (hasTrafficSummaries) lastArchivedRoundO else None)
+    }
 
     // lastArchived: the highest round archived as of this verdict batch.
     //   - From the caller when available
@@ -334,16 +344,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 +397,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 +422,7 @@ class DbAppActivityRecordStore(
   def insertActivityRecordMetaDBIO(
       codeVersion: Int,
       userVersion: Int,
-      startedIngestingAt: Long,
+      startedIngestingAt: Option[Long],
       earliestIngestedRound: Long,
       lastArchivedRound: Option[Long],
   ) =
@@ -418,6 +434,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 +456,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 +477,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 +509,7 @@ class DbAppActivityRecordStore(
             insertActivityRecordMetaDBIO(
               codeVersion,
               userVersion,
-              firstRecordTimeMicros,
+              startedIngestingAtO,
               earliestRound,
               lastArchivedRoundO,
             ).map { _ =>

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(