Exit CN applications when a store user version downgrade is detected

apps/common/src/main/scala/org/lfdecentralizedtrust/splice/store/db/StoreDescriptorStore.scala

@@ -104,4 +104,47 @@ object StoreDescriptorStore extends StoreErrors {
 
     } yield newStoreId
   }
+
+  /** True if the running config's userVersion is strictly lower than the
+    * highest userVersion that actually has committed data. `None` sorts lowest,
+    * so a removed config field counts as a downgrade against any stored version.
+    */
+  def isUserVersionDowngrade(
+      configured: Option[Long],
+      maxStoredWithData: Option[Long],
+  ): Boolean = Ordering[Option[Long]].lt(configured, maxStoredWithData)
+
+  def maxStoredUserVersionWithData(
+      storage: DbStorage,
+      dataTableName: String,
+      expectedDescriptor: StoreDescriptor,
+  )(implicit
+      traceContext: TraceContext,
+      executionContext: scala.concurrent.ExecutionContext,
+      closeContext: CloseContext,
+  ): FutureUnlessShutdown[Option[Long]] = {
+    val expectedNoUserVersion =
+      String256M.tryCreate(expectedDescriptor.copy(userVersion = None).toJson.noSpacesSortKeys)
+    storage
+      .query(
+        sql"""
+        with recursive used as (
+          (select store_id from #$dataTableName order by store_id limit 1)
+          union all
+          select (
+            select store_id from #$dataTableName
+            where store_id > used.store_id order by store_id limit 1
+          )
+          from used where used.store_id is not null
+        )
+        select max((d.descriptor ->> 'userVersion')::bigint)
+        from store_descriptors d
+        join used u on d.id = u.store_id
+        where u.store_id is not null
+          and (d.descriptor - 'userVersion') = ${expectedNoUserVersion}::jsonb
+      """.as[Option[Long]].headOption,
+        "maxStoredUserVersionWithData",
+      )
+      .map(_.flatten)
+  }
 }

apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/ScanApp.scala

@@ -81,8 +81,9 @@ import org.lfdecentralizedtrust.splice.scan.store.db.{
   DbAppActivityRecordStore,
   DbScanAppRewardsStore,
   DbScanVerdictStore,
+  ScanTables,
 }
-import org.lfdecentralizedtrust.splice.store.db.DbAppStore
+import org.lfdecentralizedtrust.splice.store.db.{DbAppStore, StoreDescriptorStore}
 import org.lfdecentralizedtrust.splice.store.{
   ChoiceContextContractFetcher,
   PageLimit,
@@ -94,8 +95,8 @@ import org.lfdecentralizedtrust.splice.util.HasHealth
 
 import scala.concurrent.{ExecutionContextExecutor, Future}
 import cats.implicits.*
-
 import org.apache.pekko.stream.Materializer
+import org.lfdecentralizedtrust.splice.util.FutureUnlessShutdownUtil.FutureUnlessShutdownOps
 
 /** Class representing a Scan app instance.
   *
@@ -217,6 +218,37 @@ class ScanApp(
         config.acsStoreDescriptorUserVersion,
         config.txLogStoreDescriptorUserVersion,
       )
+      _ <- appInitStep("Check store user-version downgrade") {
+        for {
+          acsMax <- StoreDescriptorStore
+            .maxStoredUserVersionWithData(
+              storage,
+              ScanTables.acsTableName,
+              store.acsStoreDescriptor,
+            )
+            .toFuture
+          txLogMax <- StoreDescriptorStore
+            .maxStoredUserVersionWithData(
+              storage,
+              ScanTables.txLogTableName,
+              store.txLogStoreDescriptor,
+            )
+            .toFuture
+        } yield {
+          exitIfDowngrade(
+            ScanTables.acsTableName,
+            "acs-store-descriptor-user-version",
+            config.acsStoreDescriptorUserVersion,
+            acsMax,
+          )
+          exitIfDowngrade(
+            ScanTables.txLogTableName,
+            "tx-log-store-descriptor-user-version",
+            config.txLogStoreDescriptorUserVersion,
+            txLogMax,
+          )
+        }
+      }
       updateHistory = new UpdateHistory(
         storage,
         domainMigrationId,
@@ -599,6 +631,23 @@ class ScanApp(
 
   protected[this] override def automationServices(st: ScanApp.State) =
     Seq(st.automation, st.verdictAutomation)
+
+  private def exitIfDowngrade(
+      tableName: String,
+      configFieldName: String,
+      configured: Option[Long],
+      maxStoredWithData: Option[Long],
+      exitOnDowngrade: Boolean = true,
+  )(implicit tc: TraceContext): Unit =
+    if (StoreDescriptorStore.isUserVersionDowngrade(configured, maxStoredWithData)) {
+      logger.error(
+        s"Store user-version downgrade detected for table '$tableName': " +
+          s"configured=$configured, highest with committed data=$maxStoredWithData. " +
+          s"You likely removed or lowered the '$configFieldName' field in the scan app config. " +
+          s"Shutting down to prevent silently orphaning data."
+      )
+      if (exitOnDowngrade) sys.exit(1)
+    }
 }
 
 object ScanApp {

apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/CachingScanStore.scala

@@ -31,6 +31,7 @@ import org.lfdecentralizedtrust.splice.codegen.java.splice.validatorlicense.Vali
 import org.lfdecentralizedtrust.splice.environment.RetryProvider
 import org.lfdecentralizedtrust.splice.scan.config.{CacheConfig, ScanCacheConfig}
 import org.lfdecentralizedtrust.splice.scan.store.db.DbScanStoreMetrics
+import org.lfdecentralizedtrust.splice.store.db.StoreDescriptor
 import org.lfdecentralizedtrust.splice.store.{
   Limit,
   MiningRoundsStore,
@@ -319,4 +320,8 @@ class CachingScanStore(
   }
 
   def defaultLimit: Limit = store.defaultLimit
+
+  override def acsStoreDescriptor: StoreDescriptor = store.acsStoreDescriptor
+
+  override def txLogStoreDescriptor: StoreDescriptor = store.txLogStoreDescriptor
 }

apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/ScanStore.scala

@@ -22,12 +22,16 @@ import org.lfdecentralizedtrust.splice.scan.config.ScanCacheConfig
 import org.lfdecentralizedtrust.splice.scan.store.db.ScanTables.ScanAcsStoreRowData
 import org.lfdecentralizedtrust.splice.scan.store.db.{DbScanStore, DbScanStoreMetrics}
 import org.lfdecentralizedtrust.splice.store.MultiDomainAcsStore.ContractCompanion
-import org.lfdecentralizedtrust.splice.store.db.{AcsInterfaceViewRowData, AcsJdbcTypes}
+import org.lfdecentralizedtrust.splice.store.db.{
+  AcsInterfaceViewRowData,
+  AcsJdbcTypes,
+  StoreDescriptor,
+}
 import org.lfdecentralizedtrust.splice.store.{
   AppStore,
   DsoRulesStore,
-  Limit,
   ExternalPartyConfigStateStore,
+  Limit,
   MiningRoundsStore,
   MultiDomainAcsStore,
   TxLogAppStore,
@@ -51,6 +55,9 @@ trait ScanStore
     with VotesStore
     with ExternalPartyConfigStateStore {
 
+  def acsStoreDescriptor: StoreDescriptor
+  def txLogStoreDescriptor: StoreDescriptor
+
   override def dsoPartyId = key.dsoParty
 
   def key: ScanStore.Key

apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbScanStore.scala

@@ -98,6 +98,36 @@ class DbScanTxLogStoreConfig(loggerFactory: NamedLoggerFactory)
 object DbScanStore {
   type CacheKey = java.lang.Long // caffeine metrics function demands AnyRefs
   type CacheValue = BigDecimal
+
+  def acsStoreDescriptor(
+      participantId: ParticipantId,
+      dsoParty: PartyId,
+      userVersion: Option[Long],
+  ): StoreDescriptor = StoreDescriptor(
+    version = 3,
+    name = "DbScanStore",
+    party = dsoParty,
+    participant = participantId,
+    key = Map(
+      "dsoParty" -> dsoParty.toProtoPrimitive
+    ),
+    userVersion = userVersion,
+  )
+
+  def txLogStoreDescriptor(
+      participantId: ParticipantId,
+      dsoParty: PartyId,
+      userVersion: Option[Long],
+  ): StoreDescriptor = StoreDescriptor(
+    version = 1,
+    name = "DbScanStore",
+    party = dsoParty,
+    participant = participantId,
+    key = Map(
+      "dsoParty" -> dsoParty.toProtoPrimitive
+    ),
+    userVersion = userVersion,
+  )
 }
 class DbScanStore(
     override val key: ScanStore.Key,
@@ -128,26 +158,10 @@ class DbScanStore(
       // Do not modify any part of the store descriptor unless you are sure that the resulting downtime is acceptable.
       // If you do modify it, make sure to very clearly document in the release notes that there will be planned downtime,
       // and notify the person coordinating the deployment.
-      acsStoreDescriptor = StoreDescriptor(
-        version = 3,
-        name = "DbScanStore",
-        party = key.dsoParty,
-        participant = participantId,
-        key = Map(
-          "dsoParty" -> key.dsoParty.toProtoPrimitive
-        ),
-        userVersion = acsStoreDescriptorUserVersion,
-      ),
-      txLogStoreDescriptor = StoreDescriptor(
-        version = 1,
-        name = "DbScanStore",
-        party = key.dsoParty,
-        participant = participantId,
-        key = Map(
-          "dsoParty" -> key.dsoParty.toProtoPrimitive
-        ),
-        userVersion = txLogStoreDescriptorUserVersion,
-      ),
+      acsStoreDescriptor =
+        DbScanStore.acsStoreDescriptor(participantId, key.dsoParty, acsStoreDescriptorUserVersion),
+      txLogStoreDescriptor = DbScanStore
+        .txLogStoreDescriptor(participantId, key.dsoParty, txLogStoreDescriptorUserVersion),
       domainMigrationId,
       ingestionConfig,
     )
@@ -764,4 +778,9 @@ class DbScanStore(
       row.map(contractFromEvent(companion)(_))
     }
   }
+
+  override def acsStoreDescriptor: StoreDescriptor =
+    DbScanStore.acsStoreDescriptor(participantId, key.dsoParty, acsStoreDescriptorUserVersion)
+  override def txLogStoreDescriptor: StoreDescriptor =
+    DbScanStore.txLogStoreDescriptor(participantId, key.dsoParty, txLogStoreDescriptorUserVersion)
 }

apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/store/db/ScanStoreTest.scala

@@ -41,7 +41,7 @@ import org.lfdecentralizedtrust.splice.codegen.java.splice.{
 }
 import org.lfdecentralizedtrust.splice.environment.{DarResources, RetryProvider}
 import org.lfdecentralizedtrust.splice.history.*
-import org.lfdecentralizedtrust.splice.scan.store.db.{DbScanStore, DbScanStoreMetrics}
+import org.lfdecentralizedtrust.splice.scan.store.db.{DbScanStore, DbScanStoreMetrics, ScanTables}
 import org.lfdecentralizedtrust.splice.scan.store.*
 import org.lfdecentralizedtrust.splice.store.MultiDomainAcsStore.ContractState.Assigned
 import org.lfdecentralizedtrust.splice.store.UpdateHistory.BackfillingRequirement
@@ -1894,4 +1894,84 @@ class DbScanStoreTest
       }
     }
   }
+
+  "isUserVersionDowngrade" should {
+    "treat None as the lowest version and detect only strict decreases" in {
+      // (configured, maxStoredWithData, expectedIsDowngrade)
+      val cases = Seq(
+        (None, None, false),
+        (None, Some(0L), true), // removed config field vs any stored version
+        (Some(0L), None, false), // no data on disk -> never a downgrade
+        (Some(0L), Some(0L), false), // 0 is a legal value distinct from None
+        (None, Some(5L), true),
+        (Some(3L), Some(5L), true),
+        (Some(5L), Some(5L), false),
+        (Some(5L), Some(3L), false), // upgrade
+        (Some(5L), None, false),
+      )
+      Future.successful {
+        cases.foreach { case (configured, maxStored, expected) =>
+          withClue(s"configured=$configured maxStored=$maxStored: ") {
+            StoreDescriptorStore.isUserVersionDowngrade(configured, maxStored) shouldBe expected
+          }
+        }
+        succeed
+      }
+    }
+  }
+
+  "maxStoredUserVersionWithData" should {
+    val alice = userParty(443)
+    val aliceLicense =
+      validatorLicense(alice, dsoParty, Some(new FaucetState(new Round(0), new Round(1000), 0L)))
+
+    def maxAcsUserVersion(expected: StoreDescriptor): Option[Long] =
+      StoreDescriptorStore
+        .maxStoredUserVersionWithData(storage, ScanTables.acsTableName, expected)
+        .failOnShutdown("test doesn't shutdown")
+        .futureValue
+
+    "return the highest userVersion among descriptors that have committed data" in {
+      for {
+        store <- mkStore()
+        _ <- dummyDomain.create(aliceLicense)(store.multiDomainAcsStore)
+        storeV5 <- mkStore(dsoParty = dsoParty, acsStoreDescriptorUserVersion = Some(5L))
+        _ <- dummyDomain.create(aliceLicense)(storeV5.multiDomainAcsStore)
+      } yield {
+        maxAcsUserVersion(store.acsStoreDescriptor) shouldBe Some(5L)
+      }
+    }
+
+    "ignore a matching descriptor that has no committed data" in {
+      for {
+        store <- mkStore(dsoParty = dsoParty, acsStoreDescriptorUserVersion = Some(3L))
+        _ <- dummyDomain.create(aliceLicense)(store.multiDomainAcsStore)
+        _ <- mkStore(dsoParty = dsoParty, acsStoreDescriptorUserVersion = Some(9L))
+      } yield {
+        maxAcsUserVersion(store.acsStoreDescriptor) shouldBe Some(3L)
+      }
+    }
+
+    "return None when the store has no committed data" in {
+      for {
+        store <- mkStore(dsoParty = dsoParty, acsStoreDescriptorUserVersion = Some(1L))
+      } yield {
+        maxAcsUserVersion(store.acsStoreDescriptor) shouldBe None
+      }
+    }
+
+    "exclude committed data whose descriptor differs in a non-userVersion field" in {
+      for {
+        store <- mkStore()
+        _ <- dummyDomain.create(aliceLicense)(store.multiDomainAcsStore)
+      } yield {
+        maxAcsUserVersion(
+          store.acsStoreDescriptor.copy(participant = mkParticipantId("other-participant"))
+        ) shouldBe None
+        maxAcsUserVersion(
+          store.acsStoreDescriptor.copy(party = userParty(999))
+        ) shouldBe None
+      }
+    }
+  }
 }