Track database schema generation separately from migrations

app/src/main/resources/explorer-backend-openapi.json

@@ -4325,6 +4325,7 @@
                   "releaseVersion": "1.11.2+17-00593e8e-SNAPSHOT",
                   "commit": "00593e8e8c718d6bd27fe218e7aa438ef56611cc",
                   "migrationsVersion": 1,
+                  "dbVersion": 1,
                   "lastFinalizedInputTime": 1611041396892,
                   "lastHoldersUpdate": 1611041396892
                 }
@@ -9499,6 +9500,7 @@
           "releaseVersion",
           "commit",
           "migrationsVersion",
+          "dbVersion",
           "lastFinalizedInputTime",
           "lastHoldersUpdate"
         ],
@@ -9513,6 +9515,10 @@
             "type": "integer",
             "format": "int32"
           },
+          "dbVersion": {
+            "type": "integer",
+            "format": "int32"
+          },
           "lastFinalizedInputTime": {
             "type": "integer",
             "format": "int64"

app/src/main/scala/org/alephium/explorer/api/EndpointExamples.scala

@@ -307,6 +307,7 @@ object EndpointExamples extends EndpointsExamples {
       releaseVersion = "1.11.2+17-00593e8e-SNAPSHOT",
       commit = "00593e8e8c718d6bd27fe218e7aa438ef56611cc",
       migrationsVersion = 1,
+      dbVersion = 1,
       lastFinalizedInputTime = ts,
       lastHoldersUpdate = ts
     )

app/src/main/scala/org/alephium/explorer/api/model/ExplorerInfo.scala

@@ -11,6 +11,7 @@ final case class ExplorerInfo(
     releaseVersion: String,
     commit: String,
     migrationsVersion: Int,
+    dbVersion: Int,
     lastFinalizedInputTime: TimeStamp,
     lastHoldersUpdate: TimeStamp
 )

app/src/main/scala/org/alephium/explorer/persistence/DBInitializer.scala

@@ -16,7 +16,10 @@ import slick.jdbc.meta.MTable
 import org.alephium.explorer.AnyOps
 import org.alephium.explorer.config.ExplorerConfig
 import org.alephium.explorer.persistence.DBRunner._
+import org.alephium.explorer.persistence.model.AppState.DbVersion
+import org.alephium.explorer.persistence.queries.AppStateQueries
 import org.alephium.explorer.persistence.schema._
+import org.alephium.explorer.persistence.schema.CustomGetResult._
 import org.alephium.util.discard
 
 object DBInitializer extends StrictLogging {
@@ -64,6 +67,7 @@ object DBInitializer extends StrictLogging {
       databaseConfig: DatabaseConfig[PostgresProfile]
   ): Future[Unit] = {
     for {
+      _ <- checkDbVersion()
       _ <- createTables()
       _ <- Migrations.migrate(databaseConfig)
       _ <- createIndexes()
@@ -73,6 +77,19 @@ object DBInitializer extends StrictLogging {
     }
   }
 
+  /** Reject incompatible schemas before creating tables or indexes. */
+  def checkDbVersion()(implicit
+      executionContext: ExecutionContext,
+      databaseConfig: DatabaseConfig[PostgresProfile]
+  ): Future[Unit] =
+    run(MTable.getTables(None, None, None, Some(Seq("TABLE")))).flatMap { tables =>
+      if (!tables.exists(_.name.name === "app_state")) {
+        Future.unit
+      } else {
+        run(AppStateQueries.get(DbVersion).flatMap(Migrations.checkDbVersion))
+      }
+    }
+
   private def createTables()(implicit
       executionContext: ExecutionContext,
       databaseConfig: DatabaseConfig[PostgresProfile]

app/src/main/scala/org/alephium/explorer/persistence/Database.scala

@@ -24,7 +24,7 @@ class Database(bootMode: BootMode)(implicit
   override def startSelfOnce(): Future[Unit] =
     bootMode match {
       case BootMode.ReadOnly =>
-        HealthCheckDao.healthCheck().mapSyncToUnit()
+        DBInitializer.checkDbVersion().flatMap(_ => HealthCheckDao.healthCheck().mapSyncToUnit())
 
       case BootMode.ReadWrite | BootMode.WriteOnly =>
         DBInitializer.initialize().mapSyncToUnit()

app/src/main/scala/org/alephium/explorer/persistence/Migrations.scala

@@ -14,7 +14,7 @@ import slick.jdbc.PostgresProfile.api._
 
 import org.alephium.explorer.config.ExplorerConfig
 import org.alephium.explorer.foldFutures
-import org.alephium.explorer.persistence.model.AppState.MigrationVersion
+import org.alephium.explorer.persistence.model.AppState.{DbVersion, MigrationVersion}
 import org.alephium.explorer.persistence.queries.AppStateQueries
 import org.alephium.explorer.persistence.schema.CustomGetResult._
 import org.alephium.explorer.persistence.schema.CustomSetParameter._
@@ -31,6 +31,21 @@ object Migrations extends StrictLogging {
   // scalastyle:off magic.number
   val latestVersion: MigrationVersion = MigrationVersion(11)
 
+  // Monotonic schema generation. Never reset this when resetting the migration counter.
+  val latestDbVersion: DbVersion = DbVersion(1)
+
+  def checkDbVersion(version: Option[DbVersion]): DBActionAll[Unit] = version match {
+    // Unversioned databases use the existing, supported migration path for initial adoption.
+    case None | Some(DbVersion(latestDbVersion.version)) => DBIOAction.successful(())
+    case Some(DbVersion(current)) =>
+      DBIOAction.failed(
+        new IllegalStateException(
+          s"Incompatible database schema version $current; expected ${latestDbVersion.version}. " +
+            "Use a compatible explorer release or rebuild the database."
+        )
+      )
+  }
+
   def migration1(implicit ec: ExecutionContext): DBActionAll[Unit] = {
     // We retrigger the download of fungible and non-fungible tokens' metadata that have sub-category
     for {
@@ -327,7 +342,7 @@ object Migrations extends StrictLogging {
       case None | Some(MigrationVersion(latestVersion.version)) =>
         logger.info(s"No migrations needed")
         DBIOAction.successful(())
-      case Some(MigrationVersion(current)) if current > latestVersion.version =>
+      case Some(MigrationVersion(current)) if current < 0 || current > latestVersion.version =>
         throw new Exception("Incompatible migration versions, please reset your database")
       case Some(MigrationVersion(current)) =>
         logger.info(s"Applying ${latestVersion.version - current} migrations")
@@ -351,7 +366,7 @@ object Migrations extends StrictLogging {
       case None | Some(MigrationVersion(latestVersion.version)) =>
         logger.info(s"No background migrations needed")
         Future.unit
-      case Some(MigrationVersion(current)) if current > latestVersion.version =>
+      case Some(MigrationVersion(current)) if current < 0 || current > latestVersion.version =>
         throw new Exception("Incompatible migration versions, please reset your database")
       case Some(MigrationVersion(_)) =>
         /*
@@ -409,10 +424,13 @@ object Migrations extends StrictLogging {
     logger.info("Migrating")
     for {
       currentVersion <- DBRunner
-        .run(databaseConfig)(for {
-          version <- getVersion()
-          _       <- migrationsQuery(version)
-        } yield version)
+        .run(databaseConfig)((for {
+          dbVersion <- AppStateQueries.get(DbVersion)
+          _         <- checkDbVersion(dbVersion)
+          version   <- getVersion()
+          _         <- migrationsQuery(version)
+          _         <- AppStateQueries.insertOrUpdate(latestDbVersion)
+        } yield version).transactionally)
       _ <- runMigrateInBackground(currentVersion)(explorerConfig, ec, databaseConfig)
     } yield ()
   }

app/src/main/scala/org/alephium/explorer/persistence/model/AppState.scala

@@ -52,6 +52,7 @@ object AppState {
   def unapply(result: AppState): (AppStateKey[_], ByteString) =
     result match {
       case LastFinalizedInputTime(value) => (LastFinalizedInputTime, serialize(value))
+      case DbVersion(value)              => (DbVersion, serialize(value))
       case MigrationVersion(value)       => (MigrationVersion, serialize(value))
       case LastHoldersUpdate(value)      => (LastHoldersUpdate, serialize(value))
       case FinalizedTxCount(value)       => (FinalizedTxCount, serialize(value))
@@ -66,6 +67,15 @@ object AppState {
 
   final case class LastFinalizedInputTime(time: TimeStamp) extends AppState
 
+  object DbVersion extends AppStateKey[DbVersion] {
+    val key = "db_version"
+
+    def apply(bytes: ByteString): Either[SerdeError, DbVersion] =
+      deserialize[Int](bytes).map(DbVersion(_))
+  }
+
+  final case class DbVersion(version: Int) extends AppState
+
   object MigrationVersion extends AppStateKey[MigrationVersion] {
     val key = "migrations_version"
 
@@ -95,6 +105,6 @@ object AppState {
 
   /** All the keys */
   def keys(): Array[AppStateKey[_]] =
-    Array(LastFinalizedInputTime, MigrationVersion, LastHoldersUpdate)
+    Array(LastFinalizedInputTime, MigrationVersion, DbVersion, LastHoldersUpdate)
 
 }

app/src/main/scala/org/alephium/explorer/persistence/schema/CustomGetResult.scala

@@ -462,6 +462,13 @@ object CustomGetResult {
         hashrate = result.<<
       )
 
+  implicit val dbVersionGetResult: GetResult[AppState.DbVersion] =
+    (result: PositionedResult) =>
+      AppState.DbVersion(ByteString.fromArrayUnsafe(result.nextBytes())) match {
+        case Left(error)  => throw error
+        case Right(value) => value
+      }
+
   implicit val migrationVersionGetResult: GetResult[AppState.MigrationVersion] =
     (result: PositionedResult) =>
       AppState.MigrationVersion(ByteString.fromArrayUnsafe(result.nextBytes())) match {

app/src/main/scala/org/alephium/explorer/web/InfosServer.scala

@@ -94,23 +94,26 @@ class InfosServer(
     DBRunner
       .run(
         for {
+          dbVersionOpt              <- AppStateQueries.get(AppState.DbVersion)
           migrationsVersionOpt      <- AppStateQueries.get(AppState.MigrationVersion)
           lastFinalizedInputTimeOpt <- AppStateQueries.get(AppState.LastFinalizedInputTime)
           lastHoldersUpdateOpt      <- AppStateQueries.get(AppState.LastHoldersUpdate)
         } yield {
+          val dbVersion         = dbVersionOpt.map(_.version).getOrElse(0)
           val migrationsVersion = migrationsVersionOpt.map(_.version).getOrElse(0)
           val lastFinalizedInputTime =
             lastFinalizedInputTimeOpt.map(_.time).getOrElse(TimeStamp.zero)
           val lastHoldersUpdate =
             lastHoldersUpdateOpt.map(_.time).getOrElse(TimeStamp.zero)
-          (migrationsVersion, lastFinalizedInputTime, lastHoldersUpdate)
+          (migrationsVersion, dbVersion, lastFinalizedInputTime, lastHoldersUpdate)
         }
       )
-      .map { case (migrationsVersion, lastFinalizedInputTime, lastHoldersUpdate) =>
+      .map { case (migrationsVersion, dbVersion, lastFinalizedInputTime, lastHoldersUpdate) =>
         ExplorerInfo(
           BuildInfo.releaseVersion,
           BuildInfo.commitId,
           migrationsVersion,
+          dbVersion,
           lastFinalizedInputTime,
           lastHoldersUpdate
         )

app/src/test/scala/org/alephium/explorer/api/model/ApiModelSpec.scala

@@ -404,11 +404,12 @@ class ApiModelSpec() extends AlephiumSpec {
                       |  "releaseVersion": "1.2.3",
                       |  "commit": "b96f64ff",
                       |  "migrationsVersion": 0,
+                      |  "dbVersion": 1,
                       |  "lastFinalizedInputTime": 1234,
                       |  "lastHoldersUpdate": 1234
                       |}""".stripMargin
     check(
-      ExplorerInfo("1.2.3", "b96f64ff", 0, TimeStamp.unsafe(1234), TimeStamp.unsafe(1234)),
+      ExplorerInfo("1.2.3", "b96f64ff", 0, 1, TimeStamp.unsafe(1234), TimeStamp.unsafe(1234)),
       expected
     )
   }

app/src/test/scala/org/alephium/explorer/persistence/DbVersionSpec.scala

@@ -0,0 +1,71 @@
+// Copyright (c) Alephium
+// SPDX-License-Identifier: LGPL-3.0-only
+
+package org.alephium.explorer.persistence
+
+import slick.jdbc.PostgresProfile.api._
+import slick.jdbc.meta.MTable
+
+import org.alephium.explorer.AlephiumFutureSpec
+import org.alephium.explorer.config.BootMode
+import org.alephium.explorer.persistence.model.AppState.{DbVersion, MigrationVersion}
+import org.alephium.explorer.persistence.queries.AppStateQueries
+import org.alephium.explorer.persistence.schema.CustomGetResult._
+import org.alephium.explorer.persistence.schema.TokenHolderSchema
+
+class DbVersionSpec extends AlephiumFutureSpec with DatabaseFixtureForEach with TestDBRunner {
+  "schema version" should {
+    "initialize a fresh database and remain stable across restarts" in {
+      Migrations.migrate(databaseConfig).futureValue
+      exec(AppStateQueries.get(DbVersion)) is Some(Migrations.latestDbVersion)
+      exec(AppStateQueries.get(MigrationVersion)) is Some(Migrations.latestVersion)
+      Migrations.migrate(databaseConfig).futureValue
+      exec(AppStateQueries.get(DbVersion)) is Some(Migrations.latestDbVersion)
+    }
+
+    "adopt a supported database that only has a migration version" in {
+      exec(AppStateQueries.insertOrUpdate(Migrations.latestVersion))
+      Migrations.migrate(databaseConfig).futureValue
+      exec(AppStateQueries.get(DbVersion)) is Some(Migrations.latestDbVersion)
+      exec(AppStateQueries.get(MigrationVersion)) is Some(Migrations.latestVersion)
+    }
+
+    "reject incompatible schema generations even when migration counters match" in {
+      Seq(-1, 0, 2).foreach { version =>
+        exec(AppStateQueries.insertOrUpdate(DbVersion(version)))
+        exec(AppStateQueries.insertOrUpdate(Migrations.latestVersion))
+        DBInitializer.initialize().failed.futureValue is a[IllegalStateException]
+        exec(AppStateQueries.get(DbVersion)) is Some(DbVersion(version))
+        exec(AppStateQueries.get(MigrationVersion)) is Some(Migrations.latestVersion)
+      }
+    }
+
+    "reject a newer schema in read-only mode" in {
+      exec(AppStateQueries.insertOrUpdate(DbVersion(2)))
+      new org.alephium.explorer.persistence.Database(BootMode.ReadOnly)
+        .startSelfOnce()
+        .failed
+        .futureValue is a[IllegalStateException]
+      exec(AppStateQueries.get(DbVersion)) is Some(DbVersion(2))
+    }
+
+    "reject a newer schema before creating missing tables" in {
+      exec(AppStateQueries.insertOrUpdate(DbVersion(2)))
+      exec(TokenHolderSchema.table.schema.drop)
+      try {
+        DBInitializer.initialize().failed.futureValue is a[IllegalStateException]
+        exec(MTable.getTables).exists(_.name.name == "token_holders") is false
+        exec(AppStateQueries.get(DbVersion)) is Some(DbVersion(2))
+      } finally {
+        exec(TokenHolderSchema.table.schema.create)
+      }
+    }
+
+    "not stamp a version when synchronous migration fails" in {
+      exec(AppStateQueries.insertOrUpdate(MigrationVersion(-1)))
+      Migrations.migrate(databaseConfig).failed.futureValue is a[Exception]
+      exec(AppStateQueries.get(DbVersion)) is None
+      exec(AppStateQueries.get(MigrationVersion)) is Some(MigrationVersion(-1))
+    }
+  }
+}

app/src/test/scala/org/alephium/explorer/persistence/queries/AppStateQueriesSpec.scala

@@ -30,6 +30,14 @@ class AppStateQueriesSpec extends AlephiumFutureSpec with DatabaseFixtureForEach
       }
     }
 
+    "write DbVersion independently of the migration counter" in {
+      exec(AppStateQueries.insert(AppState.DbVersion(1))) is 1
+      exec(AppStateQueries.insert(AppState.MigrationVersion(11))) is 1
+      exec(AppStateQueries.insertOrUpdate(AppState.MigrationVersion(0))) is 1
+      exec(AppStateQueries.get(AppState.DbVersion)) is Some(AppState.DbVersion(1))
+      exec(AppStateQueries.get(AppState.MigrationVersion)) is Some(AppState.MigrationVersion(0))
+    }
+
     "write MigrationVersion key-value" in {
       forAll { (int: Int) =>
         exec(AppStateSchema.table.delete)
@@ -61,6 +69,10 @@ class AppStateQueriesSpec extends AlephiumFutureSpec with DatabaseFixtureForEach
       }
     }
 
+    "fail to read corrupted DbVersion" in {
+      checkFailure(AppState.DbVersion)
+    }
+
     "fail to read corrupted MigrationVersion" in {
       checkFailure(AppState.MigrationVersion)
     }

app/src/test/scala/org/alephium/explorer/web/InfosServerSpec.scala

@@ -8,6 +8,7 @@ import scala.concurrent.{ExecutionContext, Future}
 
 import slick.basic.DatabaseConfig
 import slick.jdbc.PostgresProfile
+import slick.jdbc.PostgresProfile.api._
 
 import org.alephium.explorer._
 import org.alephium.explorer.ConfigDefaults._
@@ -16,6 +17,7 @@ import org.alephium.explorer.api.model._
 import org.alephium.explorer.cache.{BlockCache, TestBlockCache, TransactionCache}
 import org.alephium.explorer.config.BootMode
 import org.alephium.explorer.persistence.{Database, DatabaseFixtureForAll, Migrations}
+import org.alephium.explorer.persistence.queries.AppStateQueries
 import org.alephium.explorer.service._
 import org.alephium.protocol.ALPH
 import org.alephium.util.TimeStamp
@@ -95,12 +97,25 @@ class InfosServerSpec()
         BuildInfo.releaseVersion,
         BuildInfo.commitId,
         Migrations.latestVersion.version,
+        Migrations.latestDbVersion.version,
         TimeStamp.zero,
         TimeStamp.zero
       )
     }
   }
 
+  "report zero for an unversioned legacy database" in {
+    databaseConfig.db.run(sqlu"DELETE FROM app_state WHERE key = 'db_version'").futureValue
+    try {
+      Get(s"/infos") check { response =>
+        response.as[ExplorerInfo].dbVersion is 0
+      }
+    } finally {
+      databaseConfig.db.run(AppStateQueries.insertOrUpdate(Migrations.latestDbVersion)).futureValue
+      ()
+    }
+  }
+
   "return chains heights" in {
     Get(s"/infos/heights") check { response =>
       response.as[ArraySeq[PerChainHeight]] is ArraySeq(chainHeight)