Switch to new ACS export/import endpoints

[ci]

This breaks cross-version compatibility in a few ways:

1. Sponsor SV needs to be on same version as the candidate.
2. For recover from key scan needs to be on the same version as the
validator.
3. A HSM export needs to happen on the same version as the import.

1 and 2 don't seem problematic to me. 3 is a bit more problematic. In
particular, we need to avoid merging this while we might still want to
cut a new 3.x release that we upgrade testnet or mainnet to directly
and even afterwards there may be stragglers. We can either tell people
they need to upgrade to 0.4.1 first to complete the HDM which sounds
fairly reasonable or we opt for the safer option of waiting for 30
days after the HDM is complete because then they would be pruned away
anyway.

Signed-off-by: Moritz Kiefer <moritz.kiefer@purelyfunctional.org>

Author: cocreature

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

@@ -18,7 +18,12 @@ import com.digitalasset.canton.admin.api.client.data.{
   ParticipantStatus,
   PruningSchedule,
 }
-import com.digitalasset.canton.admin.participant.v30.{ExportAcsOldResponse, PruningServiceGrpc}
+import com.digitalasset.canton.participant.admin.data.ContractIdImportMode
+import com.digitalasset.canton.admin.participant.v30.{
+  ExportAcsResponse,
+  ExportAcsAtTimestampResponse,
+  PruningServiceGrpc,
+}
 import com.digitalasset.canton.admin.participant.v30.PruningServiceGrpc.PruningServiceStub
 import com.digitalasset.canton.config.RequireTypes.PositiveInt
 import com.digitalasset.canton.config.{ApiLoggingConfig, ClientConfig, PositiveDurationSeconds}
@@ -285,25 +290,46 @@ class ParticipantAdminConnection(
 
   def downloadAcsSnapshot(
       parties: Set[PartyId],
-      filterSynchronizerId: Option[SynchronizerId] = None,
-      timestamp: Option[Instant] = None,
-      force: Boolean = false,
+      filterSynchronizerId: SynchronizerId,
+      timestamp: Instant,
   )(implicit traceContext: TraceContext): Future[ByteString] = {
     logger.debug(
       show"Downloading ACS snapshot from domain $filterSynchronizerId, for parties $parties at timestamp $timestamp"
     )
     val requestComplete = Promise[ByteString]()
     // TODO(#3298) just concatenate the byteString here. Make it scale to 2M contracts.
-    val observer = new GrpcByteChunksToByteArrayObserver[ExportAcsOldResponse](requestComplete)
+    val observer =
+      new GrpcByteChunksToByteArrayObserver[ExportAcsAtTimestampResponse](requestComplete)
     runCmd(
-      ParticipantAdminCommands.ParticipantRepairManagement.ExportAcsOld(
+      ParticipantAdminCommands.PartyManagement.ExportAcsAtTimestamp(
         parties = parties,
-        partiesOffboarding = false,
         filterSynchronizerId,
         timestamp,
         observer,
-        Map.empty,
-        force,
+      )
+    ).discard
+    requestComplete.future
+  }
+
+  def downloadAcsSnapshotAtOffset(
+      parties: Set[PartyId],
+      filterSynchronizerId: SynchronizerId,
+      offset: Long,
+  )(implicit traceContext: TraceContext): Future[ByteString] = {
+    logger.debug(
+      show"Downloading ACS snapshot from domain $filterSynchronizerId, for parties $parties at offset $offset"
+    )
+    val requestComplete = Promise[ByteString]()
+    // TODO(#3298) just concatenate the byteString here. Make it scale to 2M contracts.
+    val observer =
+      new GrpcByteChunksToByteArrayObserver[ExportAcsResponse](requestComplete)
+    runCmd(
+      ParticipantAdminCommands.PartyManagement.ExportAcs(
+        parties = parties,
+        Some(filterSynchronizerId),
+        offset,
+        observer,
+        contractSynchronizerRenames = Map.empty,
       )
     ).discard
     requestComplete.future
@@ -317,10 +343,10 @@ class ParticipantAdminConnection(
       "Imports the acs in the participantl",
       runCmd(
         ParticipantAdminCommands.ParticipantRepairManagement
-          .ImportAcsOld(
+          .ImportAcs(
             acsBytes,
-            IMPORT_ACS_WORKFLOW_ID_PREFIX,
-            allowContractIdSuffixRecomputation = false,
+            workflowIdPrefix = IMPORT_ACS_WORKFLOW_ID_PREFIX,
+            contractIdImportMode = ContractIdImportMode.Validation,
           )
       ).map(_ => ()),
       logger,

apps/common/src/main/scala/org/lfdecentralizedtrust/splice/migration/AcsExporter.scala

@@ -39,16 +39,14 @@ class AcsExporter(
   def exportAcsAtTimestamp(
       domain: SynchronizerId,
       timestamp: Instant,
-      force: Boolean,
       parties: PartyId*
   )(implicit
       tc: TraceContext
   ): Future[ByteString] = {
     participantAdminConnection.downloadAcsSnapshot(
       parties = parties.toSet,
-      filterSynchronizerId = Some(domain),
-      timestamp = Some(timestamp),
-      force = force,
+      filterSynchronizerId = domain,
+      timestamp = timestamp,
     )
   }
 
@@ -92,9 +90,8 @@ class AcsExporter(
       snapshot <- EitherT.liftF[Future, AcsExportFailure, ByteString](
         participantAdminConnection.downloadAcsSnapshot(
           parties = parties.toSet,
-          filterSynchronizerId = Some(domain),
-          timestamp = Some(acsSnapshotTimestamp),
-          force = true,
+          filterSynchronizerId = domain,
+          timestamp = acsSnapshotTimestamp,
         )
       )
     } yield {

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

@@ -247,7 +247,7 @@ class ScanApp(
         config.spliceInstanceNames,
         participantAdminConnection,
         sequencerAdminConnection,
-        store,
+        automation,
         acsSnapshotStore,
         dsoAnsResolver,
         config.miningRoundsCacheTimeToLiveOverride,

apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/http/HttpScanHandler.scala

@@ -41,6 +41,7 @@ import org.lfdecentralizedtrust.splice.http.v0.definitions.{
 import org.lfdecentralizedtrust.splice.http.v0.scan.ScanResource
 import org.lfdecentralizedtrust.splice.http.v0.{definitions, scan as v0}
 import org.lfdecentralizedtrust.splice.scan.store.{AcsSnapshotStore, ScanStore, TxLogEntry}
+import org.lfdecentralizedtrust.splice.store.AppStoreWithIngestion
 import org.lfdecentralizedtrust.splice.util.{
   Codec,
   Contract,
@@ -50,9 +51,10 @@ import org.lfdecentralizedtrust.splice.util.{
 }
 import org.lfdecentralizedtrust.splice.util.PrettyInstances.*
 import com.digitalasset.canton.logging.NamedLoggerFactory
-import com.digitalasset.canton.participant.admin.data.ActiveContractOld as ActiveContract
+import com.digitalasset.canton.participant.admin.data.ActiveContract
 import com.digitalasset.canton.topology.{Member, PartyId, SynchronizerId}
 import com.digitalasset.canton.tracing.TraceContext
+import com.digitalasset.canton.util.{ByteStringUtil, GrpcStreamingUtils, ResourceUtil}
 import com.digitalasset.canton.util.ShowUtil.*
 import com.google.protobuf.ByteString
 import io.grpc.Status
@@ -62,6 +64,7 @@ import scala.concurrent.{ExecutionContextExecutor, Future}
 import scala.jdk.CollectionConverters.*
 import scala.jdk.OptionConverters.*
 import scala.util.{Try, Using}
+import java.io.ByteArrayInputStream
 import java.util.Base64
 import java.util.zip.GZIPOutputStream
 import java.time.{Instant, OffsetDateTime, ZoneOffset}
@@ -93,7 +96,7 @@ class HttpScanHandler(
     spliceInstanceNames: SpliceInstanceNamesConfig,
     participantAdminConnection: ParticipantAdminConnection,
     sequencerAdminConnection: SequencerAdminConnection,
-    protected val store: ScanStore,
+    protected val storeWithIngestion: AppStoreWithIngestion[ScanStore],
     snapshotStore: AcsSnapshotStore,
     dsoAnsResolver: DsoAnsResolver,
     miningRoundsCacheTimeToLiveOverride: Option[NonNegativeFiniteDuration],
@@ -110,6 +113,7 @@ class HttpScanHandler(
     with HttpValidatorLicensesHandler
     with HttpFeatureSupportHandler {
 
+  private val store = storeWithIngestion.store
   override protected val workflowId: String = this.getClass.getSimpleName
   override protected val votesStore: VotesStore = store
   override protected val validatorLicensesStore: AppStore = store
@@ -1100,25 +1104,45 @@ class HttpScanHandler(
   /** Filter the given ACS snapshot to contracts the given party is a stakeholder on */
   // TODO(#9340) Move this logic inside a Canton gRPC API.
   private def filterAcsSnapshot(input: ByteString, stakeholder: PartyId): ByteString = {
-    val contracts = ActiveContract
-      .loadFromByteString(input)
-      .valueOr(error =>
-        throw Status.INTERNAL
-          .withDescription(s"Failed to read ACS snapshot: ${error}")
-          .asRuntimeException()
-      )
+    val decompressedBytes =
+      ByteStringUtil
+        .decompressGzip(input, None)
+        .valueOr(err =>
+          throw Status.INVALID_ARGUMENT
+            .withDescription(s"Failed to decompress bytes: $err")
+            .asRuntimeException
+        )
+    val contracts = ResourceUtil.withResource(
+      new ByteArrayInputStream(decompressedBytes.toByteArray)
+    ) { inputSource =>
+      GrpcStreamingUtils
+        .parseDelimitedFromTrusted[ActiveContract](
+          inputSource,
+          ActiveContract,
+        )
+        .valueOr(err =>
+          throw Status.INVALID_ARGUMENT
+            .withDescription(s"Failed to parse contracts in acs snapshot: $err")
+            .asRuntimeException
+        )
+    }
     val output = ByteString.newOutput
     Using.resource(new GZIPOutputStream(output)) { outputStream =>
-      contracts.filter(c => c.contract.metadata.stakeholders.contains(stakeholder.toLf)).foreach {
-        c =>
+      contracts
+        .filter(c =>
+          c.contract.getCreatedEvent.signatories.contains(
+            stakeholder.toLf
+          ) || c.contract.getCreatedEvent.observers.contains(stakeholder.toLf)
+        )
+        .foreach { c =>
           c.writeDelimitedTo(outputStream) match {
             case Left(error) =>
               throw Status.INTERNAL
                 .withDescription(s"Failed to write ACS snapshot: ${error}")
                 .asRuntimeException()
             case Right(_) => outputStream.flush()
           }
-      }
+        }
     }
     output.toByteString
   }
@@ -1133,6 +1157,20 @@ class HttpScanHandler(
     withSpan(s"$workflowId.getAcsSnapshot") { _ => _ =>
       val partyId = PartyId.tryFromProtoPrimitive(party)
       for {
+        synchronizerId <- store
+          .lookupAmuletRules()
+          .map(
+            _.getOrElse(
+              throw io.grpc.Status.FAILED_PRECONDITION
+                .withDescription("No amulet rules.")
+                .asRuntimeException()
+            ).state.fold(
+              identity,
+              throw io.grpc.Status.FAILED_PRECONDITION
+                .withDescription("Amulet rules are in flight.")
+                .asRuntimeException(),
+            )
+          )
         // The DSO party is a stakeholder on all "important" contracts, in particular, all amulet holdings and ANS entries.
         // This means the SV participants ingest data for that party and we can take a snapshot for that party.
         // To make sure the snapshot is the same regardless of which SV is queried, we filter it down to
@@ -1141,10 +1179,22 @@ class HttpScanHandler(
         // that users backup their own ACS.
         // As the DSO party is hosted on all SVs, an arbitrary scan instance can be chosen for the ACS snapshot.
         // BFT reads are usually not required since ACS commitments act as a check that the ACS was correct.
-        acsSnapshot <- participantAdminConnection.downloadAcsSnapshot(
-          Set(partyId),
-          timestamp = recordTime.map(_.toInstant),
-        )
+        acsSnapshot <- recordTime match {
+          case None =>
+            storeWithIngestion.connection.ledgerEnd().flatMap { offset =>
+              participantAdminConnection.downloadAcsSnapshotAtOffset(
+                Set(partyId),
+                offset = offset,
+                filterSynchronizerId = synchronizerId,
+              )
+            }
+          case Some(time) =>
+            participantAdminConnection.downloadAcsSnapshot(
+              Set(partyId),
+              timestamp = time.toInstant,
+              filterSynchronizerId = synchronizerId,
+            )
+        }
       } yield {
         val filteredAcsSnapshot =
           filterAcsSnapshot(acsSnapshot, store.key.dsoParty)

apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/admin/http/HttpSvAdminHandler.scala

@@ -560,7 +560,6 @@ class HttpSvAdminHandler(
         .getDomainDataSnapshot(
           Instant.parse(timestamp),
           partyId.map(Codec.tryDecode(Codec.Party)(_)),
-          force.getOrElse(false),
         )
         .map { response =>
           val responseHttp = response.toHttp

apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/migration/DomainDataSnapshotGenerator.scala

@@ -45,7 +45,6 @@ class DomainDataSnapshotGenerator(
   def getDomainDataSnapshot(
       timestamp: Instant,
       partyId: Option[PartyId],
-      force: Boolean,
   )(implicit
       ec: ExecutionContext,
       tc: TraceContext,
@@ -57,7 +56,6 @@ class DomainDataSnapshotGenerator(
       .exportAcsAtTimestamp(
         decentralizedSynchronizer,
         timestamp,
-        force,
         partyId.fold(Seq(dsoStore.key.dsoParty, dsoStore.key.svParty))(Seq(_))*
       )
     dars <- darExporter.exportAllDars()

apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/onboarding/sponsor/DsoPartyMigration.scala

@@ -101,8 +101,8 @@ class DsoPartyMigration(
           participantAdminConnection
             .downloadAcsSnapshot(
               Set(dsoParty),
-              filterSynchronizerId = Some(decentralizedSynchronizer),
-              timestamp = Some(authorizedAt),
+              filterSynchronizerId = decentralizedSynchronizer,
+              timestamp = authorizedAt,
             )
             .recoverWith { case ex: StatusRuntimeException =>
               val errorDetails = ErrorDetails.from(ex: StatusRuntimeException)

apps/validator/src/main/scala/org/lfdecentralizedtrust/splice/validator/admin/http/HttpValidatorAdminHandler.scala

@@ -176,7 +176,6 @@ class HttpValidatorAdminHandler(
             synchronizerId,
             // TODO(#9731): get migration id from scan instead of configuring here
             migrationId getOrElse (config.domainMigrationId + 1),
-            force.getOrElse(false),
           )
           .map { response =>
             v0.ValidatorAdminResource.GetValidatorDomainDataSnapshotResponse.OK(

apps/validator/src/main/scala/org/lfdecentralizedtrust/splice/validator/migration/DomainMigrationDumpGenerator.scala

@@ -79,7 +79,6 @@ class DomainMigrationDumpGenerator(
       timestamp: Instant,
       domain: SynchronizerId,
       migrationId: Long,
-      force: Boolean,
   )(implicit
       ec: ExecutionContext,
       tc: TraceContext,
@@ -97,7 +96,6 @@ class DomainMigrationDumpGenerator(
       acsSnapshot <- acsExporter.exportAcsAtTimestamp(
         domain,
         timestamp,
-        force,
         parties*
       )
       dars <- darExporter.exportAllDars()