recovery from key - store the parties being migrated in the local db for recovery in case of failure

apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/ValidatorReonboardingIntegrationTest.scala

@@ -1,14 +1,19 @@
 package org.lfdecentralizedtrust.splice.integration.tests
 
+import com.digitalasset.canton.HasExecutionContext
+import com.digitalasset.canton.config.CantonRequireTypes.InstanceName
+import com.digitalasset.canton.config.RequireTypes.Port
+import com.digitalasset.canton.config.{DbConfig, FullClientConfig}
+import com.digitalasset.canton.data.CantonTimestamp
+import com.digitalasset.canton.lifecycle.{CloseContext, FlagCloseable}
+import com.digitalasset.canton.logging.{SuppressingLogger, SuppressionRule}
+import com.digitalasset.canton.metrics.CommonMockMetrics
+import com.digitalasset.canton.resource.DbStorageSingle
+import com.digitalasset.canton.topology.{ForceFlag, ParticipantId, PartyId}
+import com.typesafe.config.ConfigValueFactory
+import org.apache.pekko.http.scaladsl.model.Uri
+import org.lfdecentralizedtrust.splice.config.*
 import org.lfdecentralizedtrust.splice.config.ConfigTransforms.bumpUrl
-import org.lfdecentralizedtrust.splice.config.{
-  AuthTokenSourceConfig,
-  NetworkAppClientConfig,
-  ParticipantBootstrapDumpConfig,
-  ParticipantClientConfig,
-  SpliceConfig,
-}
-import com.digitalasset.canton.logging.SuppressionRule
 import org.lfdecentralizedtrust.splice.environment.RetryFor
 import org.lfdecentralizedtrust.splice.identities.NodeIdentitiesDump
 import org.lfdecentralizedtrust.splice.integration.EnvironmentDefinition
@@ -22,17 +27,16 @@ import org.lfdecentralizedtrust.splice.validator.config.{
   MigrateValidatorPartyConfig,
   ValidatorCantonIdentifierConfig,
 }
-import com.digitalasset.canton.config.CantonRequireTypes.InstanceName
-import com.digitalasset.canton.config.{DbConfig, FullClientConfig}
-import com.digitalasset.canton.config.RequireTypes.Port
-import com.digitalasset.canton.data.CantonTimestamp
-import com.digitalasset.canton.topology.{ForceFlag, ParticipantId, PartyId}
-import com.typesafe.config.ConfigValueFactory
-import org.apache.pekko.http.scaladsl.model.Uri
+import org.lfdecentralizedtrust.splice.validator.migration.ParticipantPartyMigrator
+import org.lfdecentralizedtrust.splice.validator.store.{
+  ValidatorConfigProvider,
+  ValidatorInternalStore,
+}
 import org.scalatest.time.{Minute, Span}
 import org.slf4j.event.Level
 
 import java.nio.file.{Files, Path, Paths}
+import scala.concurrent.duration.DurationInt
 
 trait ValidatorReonboardingIntegrationTestBase
     extends IntegrationTest
@@ -316,7 +320,11 @@ class ValidatorReonboardingIntegrationTest extends ValidatorReonboardingIntegrat
       "EXTRA_PARTICIPANT_ADMIN_USER" -> aliceValidatorLocalBackend.config.ledgerApiUser,
       "EXTRA_PARTICIPANT_DB" -> newParticipantDb,
     ) {
-      loggerFactory.assertLogsSeq(SuppressionRule.LevelAndAbove(Level.WARN))(
+      loggerFactory.assertLogsSeq(
+        SuppressionRule.forLogger[ParticipantPartyMigrator] || SuppressionRule
+          .forLogger[ValidatorConfigProvider] || SuppressionRule
+          .LevelAndAbove(Level.WARN)
+      )(
         {
           aliceValidatorLocalBackend.startSync()
         },
@@ -327,6 +335,16 @@ class ValidatorReonboardingIntegrationTest extends ValidatorReonboardingIntegrat
               "not be able to migrate due to an unsupported namespace"
             )
           }
+          forExactly(1, entries) {
+            _.message should include(
+              "Storing all the hosted parties (4) in the database to recover in case of failures"
+            )
+          }
+          forExactly(1, entries) {
+            _.message should include(
+              "Clearing parties that were migrated"
+            )
+          }
         },
       )
 
@@ -418,7 +436,7 @@ class ValidatorReonboardingWithPartiesToMigrateIntegrationTest
   // (which will fail unless we migrate all parties) despite migrating the operator party.
   // This can only work if there is no collision between the participant admin party and
   // the validator operator party.
-  override val aliceValidatorParticipantNameHint = s"${aliceValidatorPartyHint}-participant";
+  override val aliceValidatorParticipantNameHint = s"$aliceValidatorPartyHint-participant";
 
   // we bootstrap from dump so we can predict the party IDs
   val testDumpDir: Path = Paths.get("apps/app/src/test/resources/dumps")
@@ -459,7 +477,7 @@ class ValidatorReonboardingWithPartiesToMigrateIntegrationTest
         testResourcesPath / "standalone-participant-extra-no-auth.conf",
       ),
       Seq.empty,
-      "alice-participant",
+      "alice-reonboard-from-key-database",
       "EXTRA_PARTICIPANT_ADMIN_USER" -> aliceValidatorLocalBackend.config.ledgerApiUser,
       "EXTRA_PARTICIPANT_DB" -> oldParticipantDb,
     ) {
@@ -569,3 +587,132 @@ class ValidatorReonboardingWithPartiesToMigrateIntegrationTest
     }
   }
 }
+
+class ValidatorReonboardingWithPartiesToMigrateFromDbIntegrationTest
+    extends ValidatorReonboardingIntegrationTestBase
+    with HasExecutionContext {
+
+  // avoid db collisions; ptm := partiesToMigrate
+  override def dbsSuffix = "validator_reonboard_ptm_db"
+  override def oldParticipantDb = "participant_alice_validator_ptm_db"
+  override def newParticipantDb = "participant_alice_validator_reonboard_new_ptm_db"
+  override def appsDb = "splice_apps_reonboard_ptm_db"
+
+  // For this test we want to test that we won't try to revoke the domain trust certificate
+  // (which will fail unless we migrate all parties) despite migrating the operator party.
+  // This can only work if there is no collision between the participant admin party and
+  // the validator operator party.
+  override val aliceValidatorParticipantNameHint = s"${aliceValidatorPartyHint}-participant2";
+
+  "re-onboard validator with partiesToMigrate" in { implicit env =>
+    initDsoWithSv1Only()
+    // We need a standalone instance to avoid breaking the long-running nodes.
+    val (dump, aliceValidatorWalletParty, aliceParty, charlieParty) = withCanton(
+      Seq(
+        testResourcesPath / "standalone-participant-extra.conf",
+        testResourcesPath / "standalone-participant-extra-no-auth.conf",
+      ),
+      Seq.empty,
+      "alice-participant-rdb",
+      "EXTRA_PARTICIPANT_ADMIN_USER" -> aliceValidatorLocalBackend.config.ledgerApiUser,
+      "EXTRA_PARTICIPANT_DB" -> oldParticipantDb,
+    ) {
+      aliceValidatorBackend.startSync()
+      val aliceValidatorWalletParty =
+        PartyId.tryFromProtoPrimitive(aliceValidatorWalletClient.userStatus().party)
+      aliceValidatorWalletClient.tap(100)
+
+      val aliceParty = onboardWalletUser(aliceWalletClient, aliceValidatorBackend)
+      aliceWalletClient.tap(100)
+      val charlieParty = onboardWalletUser(charlieWalletClient, aliceValidatorBackend)
+      charlieWalletClient.tap(100)
+
+      val dump = aliceValidatorBackend.dumpParticipantIdentities()
+
+      aliceValidatorBackend.stop()
+      (dump, aliceValidatorWalletParty, aliceParty, charlieParty)
+    }
+
+    better.files
+      .File(dumpPath)
+      .overwrite(
+        dump.toJson.noSpaces
+      )
+
+    withCanton(
+      Seq(
+        testResourcesPath / "standalone-participant-extra.conf",
+        testResourcesPath / "standalone-participant-extra-no-auth.conf",
+      ),
+      Seq(),
+      "alice-reonboard-participant-stored-in-db",
+      "EXTRA_PARTICIPANT_ADMIN_USER" -> aliceValidatorLocalBackend.config.ledgerApiUser,
+      "EXTRA_PARTICIPANT_DB" -> newParticipantDb,
+    ) {
+
+      loggerFactory.assertLogsSeq(
+        SuppressionRule.forLogger[ParticipantPartyMigrator] || SuppressionRule
+          .forLogger[ValidatorConfigProvider] || SuppressionRule
+          .LevelAndAbove(Level.WARN)
+      )(
+        {
+          aliceValidatorLocalBackend.start()
+          clue("setting migrating parties in the local db") {
+            val loggerFactory = SuppressingLogger(getClass)
+            val closeable = FlagCloseable.withCloseContext(logger, timeouts)
+            implicit val closeContext: CloseContext = closeable.closeContext
+            val provider = new ValidatorConfigProvider(
+              ValidatorInternalStore(
+                DbStorageSingle.tryCreate(
+                  aliceValidatorLocalBackend.config.storage,
+                  wallClock,
+                  None,
+                  connectionPoolForParticipant = false,
+                  None,
+                  CommonMockMetrics.dbStorage,
+                  timeouts,
+                  loggerFactory,
+                ),
+                loggerFactory,
+              ),
+              loggerFactory,
+            )
+            eventuallySucceeds(30.seconds, 20.millis, suppressErrors = false) {
+              provider
+                .setPartiesToMigrate(
+                  Set(aliceParty, aliceValidatorWalletParty)
+                )
+                .futureValue
+            }
+            closeable.close()
+          }
+          aliceValidatorLocalBackend.waitForInitialization()
+        },
+        entries => {
+          forExactly(1, entries) {
+            _.message should include(
+              "hosted parties in the local database, this indicates a retry of the migration process"
+            )
+          }
+        },
+      )
+
+      aliceValidatorLocalBackend.appState.configProvider
+        .getPartiesToMigrate()
+        .value
+        .futureValue shouldBe None
+
+      Seq(
+        aliceValidatorWalletParty,
+        aliceParty,
+      ).foreach { partyId =>
+        clue(s"check mapping and amulet balance of $partyId") {
+          assertMapping(
+            partyId,
+            aliceValidatorLocalBackend.participantClient.id,
+          )
+        }
+      }
+    }
+  }
+}

apps/common/src/main/scala/org/lfdecentralizedtrust/splice/environment/TopologyAdminConnection.scala

@@ -1384,8 +1384,8 @@ abstract class TopologyAdminConnection(
     runCmd(VaultAdminCommands.ImportKeyPair(ByteString.copyFrom(keyPair), name, password = None))
   }
 
-  private def listSynchronizerTrustCertificate(synchronizerId: SynchronizerId, member: Member)(
-      implicit tc: TraceContext
+  def listSynchronizerTrustCertificate(synchronizerId: SynchronizerId, member: Member)(implicit
+      tc: TraceContext
   ): Future[Seq[TopologyResult[SynchronizerTrustCertificate]]] =
     runCmd(
       TopologyAdminCommands.Read.ListSynchronizerTrustCertificate(

apps/common/src/main/scala/org/lfdecentralizedtrust/splice/util/FutureUnlessShutdownUtil.scala

@@ -22,6 +22,7 @@ object FutureUnlessShutdownUtil {
   )(implicit ec: ExecutionContext): OptionT[Future, A] = {
     OptionT(futureUnlessShutdownToFuture(f.value))
   }
+
   implicit class FutureUnlessShutdownOps[A](val f: FutureUnlessShutdown[A]) extends AnyVal {
     def toFuture(implicit ec: ExecutionContext): Future[A] =
       f.failOnShutdownToAbortException("Splice unsafe shutdown future")

apps/common/src/test/scala/org/lfdecentralizedtrust/splice/util/SpliceRateLimiterTest.scala

@@ -10,7 +10,6 @@ import org.apache.pekko.stream.Materializer
 import org.apache.pekko.stream.scaladsl.{Sink, Source}
 import org.apache.pekko.stream.testkit.StreamSpec
 import org.lfdecentralizedtrust.splice.admin.api.client.commands.HttpCommandException
-import org.lfdecentralizedtrust.splice.util.{SpliceRateLimitMetrics, SpliceRateLimiter}
 import org.lfdecentralizedtrust.splice.util.SpliceRateLimiterTest.runRateLimited
 
 import scala.concurrent.Future