From ccf415bfaf920a7de8a7cfa2612c59c996dab330 Mon Sep 17 00:00:00 2001 From: JYC11 Date: Tue, 25 Aug 2026 04:59:06 +0000 Subject: [PATCH] exit scan app when acs and tx user versions are downgraded Signed-off-by: JYC11 --- .../store/db/StoreDescriptorStore.scala | 43 ++++++++++ .../splice/scan/ScanApp.scala | 53 +++++++++++- .../splice/scan/store/CachingScanStore.scala | 5 ++ .../splice/scan/store/ScanStore.scala | 11 ++- .../splice/scan/store/db/DbScanStore.scala | 59 ++++++++----- .../splice/store/db/ScanStoreTest.scala | 82 ++++++++++++++++++- 6 files changed, 228 insertions(+), 25 deletions(-) diff --git a/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/store/db/StoreDescriptorStore.scala b/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/store/db/StoreDescriptorStore.scala index ed4633e4cc..acdd0edbf2 100644 --- a/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/store/db/StoreDescriptorStore.scala +++ b/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) + } } diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/ScanApp.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/ScanApp.scala index 0999b6cae0..de330c888d 100644 --- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/ScanApp.scala +++ b/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 { diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/CachingScanStore.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/CachingScanStore.scala index dec98aef00..0a6f7468cd 100644 --- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/CachingScanStore.scala +++ b/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 } diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/ScanStore.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/ScanStore.scala index 60c9f7e649..b03e6ec07d 100644 --- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/ScanStore.scala +++ b/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 diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbScanStore.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbScanStore.scala index 871c2633b6..8f36f3036c 100644 --- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbScanStore.scala +++ b/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) } diff --git a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/store/db/ScanStoreTest.scala b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/store/db/ScanStoreTest.scala index 3d2271e786..aa1aa654aa 100644 --- a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/store/db/ScanStoreTest.scala +++ b/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 + } + } + } }