Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 13 commits into from
Nov 19, 2025
+511 −348
Merged

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
9ea9d74
rollback
nicu-da Nov 13, 2025
e035e30
move
nicu-da Nov 13, 2025
aa041da
[ci] .
nicu-da Nov 13, 2025
c510967
[ci] info logs
nicu-da Nov 14, 2025
3476fd2
[ci] .
nicu-da Nov 14, 2025
299db18
[ci] .
nicu-da Nov 14, 2025
53b7c62
[ci] .
nicu-da Nov 14, 2025
771809f
[ci] add test
nicu-da Nov 14, 2025
28ba32a
[ci] .
nicu-da Nov 18, 2025
2e941e2
[ci] .
nicu-da Nov 18, 2025
5c67b6d
Merge branch 'main' into nicuda/fix/resilience_key_recovery
nicu-da Nov 18, 2025
ccfd727
[ci] .
nicu-da Nov 18, 2025
6560f64
[ci] .
nicu-da Nov 19, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 28 additions & 18 deletions .../lfdecentralizedtrust/splice/integration/tests/ValidatorReonboardingIntegrationTest.scala
View file
Open in desktop
Copy link
Contributor Author

@nicu-da nicu-da Nov 18, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@martinflorian-da can you take another look? Added a test for the new functionality. In theory there's a slight chance that it will be flaky but I am confident enough that the startup is not fast enough for it to flake.

Copy link
Contributor

@martinflorian-da martinflorian-da Nov 18, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ufff IDK, it does sound quite a bit shaky flaky to me... Things change, we'll forget what we did here, then we'll reorder app init order or change something else... and then go debug this.

I guess you run start where you do so it sets up the DB for you? Can't we just do that in the test, before running start?

Yes it seems a bit over the top for this particular PR, but the capability of initialing this store with arbitrary data before we start an app sounds like a handy thing to have also for other tests.

I am confident enough

Not going to veto it but if it was my PR I think I'd lean towards not adding that test, even if the alternative was "I only tested it manually once".

Copy link
Contributor Author

@nicu-da nicu-da Nov 18, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I guess you run start where you do so it sets up the DB for you? Can't we just do that in the test, before running start?

We could but it's not trivial

Not going to veto it but if it was my PR I think I'd lean towards not adding that test, even if the alternative was "I only tested it manually once".

honestly SGTM, will just remove the test, I was on the fence about it but then realized that our startup is real slow but then again that can always change and we shouldn't count on it

Copy link
Contributor

@martinflorian-da martinflorian-da Nov 18, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

👍

...it's also a dangerous precedent... Not everyone's gut feeling is as well calibrated as yours, but they might still end up copying this approach...

Original file line number Diff line number Diff line change
@@ -1,14 +1,15 @@
package org.lfdecentralizedtrust.splice.integration.tests

import org.lfdecentralizedtrust.splice.config.ConfigTransforms.bumpUrl
import org.lfdecentralizedtrust.splice.config.{
AuthTokenSourceConfig,
NetworkAppClientConfig,
ParticipantBootstrapDumpConfig,
ParticipantClientConfig,
SpliceConfig,
}
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.logging.SuppressionRule
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.environment.RetryFor
import org.lfdecentralizedtrust.splice.identities.NodeIdentitiesDump
import org.lfdecentralizedtrust.splice.integration.EnvironmentDefinition
Expand All @@ -22,13 +23,8 @@ 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
import org.scalatest.time.{Minute, Span}
import org.slf4j.event.Level

Expand Down Expand Up @@ -316,7 +312,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()
},
Expand All @@ -327,6 +327,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"
)
}
},
)

Expand Down Expand Up @@ -418,7 +428,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")
Expand Down Expand Up @@ -459,7 +469,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,
) {
Expand Down
4 changes: 2 additions & 2 deletions .../src/main/scala/org/lfdecentralizedtrust/splice/environment/TopologyAdminConnection.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down
1 change: 1 addition & 0 deletions ...common/src/main/scala/org/lfdecentralizedtrust/splice/util/FutureUnlessShutdownUtil.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ object FutureUnlessShutdownUtil {
)(implicit ec: ExecutionContext): OptionT[Future, A] = {
OptionT(futureUnlessShutdownToFuture(f.value))
}

Copy link
Contributor

@martinflorian-da martinflorian-da Nov 14, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Very important to have newlines! 😛

implicit class FutureUnlessShutdownOps[A](val f: FutureUnlessShutdown[A]) extends AnyVal {
def toFuture(implicit ec: ExecutionContext): Future[A] =
f.failOnShutdownToAbortException("Splice unsafe shutdown future")
Expand Down
1 change: 0 additions & 1 deletion apps/common/src/test/scala/org/lfdecentralizedtrust/splice/util/SpliceRateLimiterTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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}
Copy link
Contributor

@martinflorian-da martinflorian-da Nov 14, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Just going to pretend I understand how this passed static checks previously...

import org.lfdecentralizedtrust.splice.util.SpliceRateLimiterTest.runRateLimited

import scala.concurrent.Future
Expand Down
106 changes: 54 additions & 52 deletions apps/validator/src/main/scala/org/lfdecentralizedtrust/splice/validator/ValidatorApp.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,13 +6,36 @@ package org.lfdecentralizedtrust.splice.validator
import cats.implicits.{catsSyntaxApplicativeByValue as _, *}
import com.daml.grpc.adapter.ExecutionSequencerFactory
import com.daml.ledger.javaapi.data.User
import com.digitalasset.canton.SynchronizerAlias
import com.digitalasset.canton.concurrent.FutureSupervisor
import com.digitalasset.canton.config.CantonRequireTypes.InstanceName
import com.digitalasset.canton.config.ProcessingTimeout
import com.digitalasset.canton.config.RequireTypes.NonNegativeLong
import com.digitalasset.canton.data.CantonTimestamp
import com.digitalasset.canton.ledger.api.util.DurationConversion
import com.digitalasset.canton.lifecycle.LifeCycle
import com.digitalasset.canton.logging.{NamedLoggerFactory, TracedLogger}
import com.digitalasset.canton.resource.{DbStorage, Storage}
import com.digitalasset.canton.time.Clock
import com.digitalasset.canton.topology.{PartyId, SynchronizerId}
import com.digitalasset.canton.tracing.{TraceContext, TracerProvider}
import com.digitalasset.canton.util.MonadUtil
import com.google.protobuf.ByteString
import io.grpc.Status
import io.opentelemetry.api.trace.Tracer
import org.apache.pekko.actor.ActorSystem
import org.apache.pekko.http.cors.scaladsl.CorsDirectives.*
import org.apache.pekko.http.cors.scaladsl.settings.CorsSettings
import org.apache.pekko.http.scaladsl.model.HttpMethods
import org.apache.pekko.http.scaladsl.server.Directives.*
import org.apache.pekko.http.scaladsl.server.directives.BasicDirectives
import org.lfdecentralizedtrust.splice.admin.api.TraceContextDirectives.withTraceContext
import org.lfdecentralizedtrust.splice.admin.http.{AdminRoutes, HttpErrorHandler}
import org.lfdecentralizedtrust.splice.auth.*
import org.lfdecentralizedtrust.splice.automation.{
DomainParamsAutomationService,
DomainTimeAutomationService,
}
import org.lfdecentralizedtrust.splice.auth.*
import org.lfdecentralizedtrust.splice.config.{NetworkAppClientConfig, SharedSpliceAppParameters}
import org.lfdecentralizedtrust.splice.environment.*
import org.lfdecentralizedtrust.splice.environment.ledger.api.DedupDuration
Expand All @@ -31,80 +54,48 @@ import org.lfdecentralizedtrust.splice.migration.{
ParticipantUsersDataRestorer,
}
import org.lfdecentralizedtrust.splice.scan.admin.api.client
import org.lfdecentralizedtrust.splice.scan.admin.api.client.BftScanConnection.BftScanClientConfig
import org.lfdecentralizedtrust.splice.scan.admin.api.client.{
BftScanConnection,
MinimalScanConnection,
SingleScanConnection,
}
import org.lfdecentralizedtrust.splice.scan.admin.api.client.BftScanConnection.BftScanClientConfig
import org.lfdecentralizedtrust.splice.scan.config.ScanAppClientConfig
import org.lfdecentralizedtrust.splice.setup.{
NodeInitializer,
ParticipantInitializer,
ParticipantPartyMigrator,
}
import org.lfdecentralizedtrust.splice.store.{AppStoreWithIngestion, HistoryMetrics, UpdateHistory}
import org.lfdecentralizedtrust.splice.setup.{NodeInitializer, ParticipantInitializer}
import org.lfdecentralizedtrust.splice.store.AppStoreWithIngestion.SpliceLedgerConnectionPriority
import org.lfdecentralizedtrust.splice.store.MultiDomainAcsStore.QueryResult
import org.lfdecentralizedtrust.splice.util.{
AmuletConfigSchedule,
BackupDump,
HasHealth,
PackageVetting,
SpliceCircuitBreaker,
}
import org.lfdecentralizedtrust.splice.store.UpdateHistory.BackfillingRequirement
import org.lfdecentralizedtrust.splice.store.{AppStoreWithIngestion, HistoryMetrics, UpdateHistory}
import org.lfdecentralizedtrust.splice.util.*
import org.lfdecentralizedtrust.splice.validator.admin.http.*
import org.lfdecentralizedtrust.splice.validator.automation.{
ValidatorAutomationService,
ValidatorPackageVettingTrigger,
}
import org.lfdecentralizedtrust.splice.validator.config.{
AppInstance,
MigrateValidatorPartyConfig,
ValidatorAppBackendConfig,
ValidatorCantonIdentifierConfig,
ValidatorOnboardingConfig,
}
import org.lfdecentralizedtrust.splice.validator.config.*
import org.lfdecentralizedtrust.splice.validator.domain.DomainConnector
import org.lfdecentralizedtrust.splice.validator.metrics.ValidatorAppMetrics
import org.lfdecentralizedtrust.splice.validator.migration.DomainMigrationDump
import org.lfdecentralizedtrust.splice.validator.store.ValidatorStore
import org.lfdecentralizedtrust.splice.validator.migration.{
DomainMigrationDump,
ParticipantPartyMigrator,
}
import org.lfdecentralizedtrust.splice.validator.store.{
ValidatorConfigProvider,
ValidatorInternalStore,
ValidatorStore,
}
import org.lfdecentralizedtrust.splice.validator.util.ValidatorUtil
import org.lfdecentralizedtrust.splice.wallet.{ExternalPartyWalletManager, UserWalletManager}
import org.lfdecentralizedtrust.splice.wallet.admin.http.{
HttpExternalWalletHandler,
HttpWalletHandler,
}
import org.lfdecentralizedtrust.splice.wallet.automation.UserWalletAutomationService
import org.lfdecentralizedtrust.splice.wallet.util.ValidatorTopupConfig
import org.lfdecentralizedtrust.tokenstandard.metadata.v1.Resource as TokenStandardMetadataResource
import org.lfdecentralizedtrust.tokenstandard.transferinstruction.v1.Resource as TokenStandardTransferInstructionResource
import org.lfdecentralizedtrust.splice.wallet.{ExternalPartyWalletManager, UserWalletManager}
import org.lfdecentralizedtrust.tokenstandard.allocation.v1.Resource as TokenStandardAllocationResource
import org.lfdecentralizedtrust.tokenstandard.allocationinstruction.v1.Resource as TokenStandardAllocationInstructionResource
import com.digitalasset.canton.concurrent.FutureSupervisor
import com.digitalasset.canton.config.CantonRequireTypes.InstanceName
import com.digitalasset.canton.config.ProcessingTimeout
import com.digitalasset.canton.config.RequireTypes.NonNegativeLong
import com.digitalasset.canton.data.CantonTimestamp
import com.digitalasset.canton.ledger.api.util.DurationConversion
import com.digitalasset.canton.lifecycle.LifeCycle
import com.digitalasset.canton.logging.{NamedLoggerFactory, TracedLogger}
import com.digitalasset.canton.resource.{DbStorage, Storage}
import com.digitalasset.canton.time.Clock
import com.digitalasset.canton.topology.{PartyId, SynchronizerId}
import com.digitalasset.canton.tracing.{TraceContext, TracerProvider}
import com.digitalasset.canton.util.MonadUtil
import com.digitalasset.canton.SynchronizerAlias
import io.grpc.Status
import io.opentelemetry.api.trace.Tracer
import org.apache.pekko.actor.ActorSystem
import org.apache.pekko.http.cors.scaladsl.CorsDirectives.*
import org.apache.pekko.http.cors.scaladsl.settings.CorsSettings
import org.apache.pekko.http.scaladsl.model.HttpMethods
import org.apache.pekko.http.scaladsl.server.Directives.*
import org.apache.pekko.http.scaladsl.server.directives.BasicDirectives
import com.google.protobuf.ByteString
import org.lfdecentralizedtrust.splice.store.AppStoreWithIngestion.SpliceLedgerConnectionPriority
import org.lfdecentralizedtrust.splice.store.UpdateHistory.BackfillingRequirement
import org.lfdecentralizedtrust.tokenstandard.metadata.v1.Resource as TokenStandardMetadataResource
import org.lfdecentralizedtrust.tokenstandard.transferinstruction.v1.Resource as TokenStandardTransferInstructionResource

import scala.concurrent.{ExecutionContextExecutor, Future}
import scala.util.{Failure, Success}
Expand Down Expand Up @@ -334,10 +325,15 @@ class ValidatorApp(
.getOrElse(
BaseLedgerConnection.sanitizeUserIdToPartyString(config.ledgerApiUser)
)
val configProvider = new ValidatorConfigProvider(
ValidatorInternalStore(storage, loggerFactory),
loggerFactory,
)
val participantPartyMigrator = new ParticipantPartyMigrator(
connection,
participantAdminConnection,
config.domains.global.alias,
configProvider,
loggerFactory,
)
appInitStep("Migrating party data") {
Expand Down Expand Up @@ -823,6 +819,10 @@ class ValidatorApp(
readOnlyLedgerConnection,
loggerFactory,
)
configProvider = new ValidatorConfigProvider(
ValidatorInternalStore(storage, loggerFactory),
loggerFactory,
)
walletManagerOpt =
if (config.enableWallet) {
val externalPartyWalletManager = new ExternalPartyWalletManager(
Expand Down Expand Up @@ -1213,6 +1213,7 @@ class ValidatorApp(
domainTimeAutomationService,
domainParamsAutomationService,
store,
configProvider,
automation,
walletManagerOpt,
timeouts,
Expand All @@ -1234,6 +1235,7 @@ object ValidatorApp {
domainTimeAutomationService: DomainTimeAutomationService,
domainParamsAutomationService: DomainParamsAutomationService,
store: ValidatorStore,
configProvider: ValidatorConfigProvider,
automation: ValidatorAutomationService,
walletManager: Option[UserWalletManager],
timeouts: ProcessingTimeout,
Expand Down
Loading
Loading