[main] Make migration export scale to large sizes (#2875)

* [main] Make migration export scale to large sizes

forward port from #2863

fixes #2858

---------

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

Author: moritzkiefer-da

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

@@ -628,7 +628,7 @@ class DisasterRecoveryIntegrationTest
 
     val fullDumpFile = migrationDumpFilePath(validator.name)
     clearOrCreate(fullDumpFile)
-    fullDumpFile.write(dump.asJson.spaces2)
+    fullDumpFile.write(dump.toHttp(outputDirectory = None).asJson.spaces2)
   }
 
   private def writeMigrationDumpFile(
@@ -651,7 +651,7 @@ class DisasterRecoveryIntegrationTest
       dump.participantUsers,
       createdAt = dump.createdAt,
     )
-    fullDumpFile.write(fullDump.asJson.spaces2)
+    fullDumpFile.write(fullDump.toHttp(outputDirectory = None).asJson.spaces2)
   }
 
   private def clearOrCreate(f: File) = {

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

@@ -1120,9 +1120,14 @@ class SvFrontendIntegrationTest
   def changeAction(actionName: String)(implicit webDriver: WebDriverType) = {
     eventually() { find(id("display-actions")) should not be empty }
     val dropDownAction = new Select(webDriver.findElement(By.id("display-actions")))
+    val existingAction: String = dropDownAction.getFirstSelectedOption().getAttribute("value")
     dropDownAction.selectByValue(actionName)
 
-    if (actionName != "SRARC_OffboardSv") {
+    if (actionName != "SRARC_OffboardSv" && existingAction != actionName) {
+      logger.debug(
+        s"Changed action from $existingAction to $actionName, waiting for confirmation dialog"
+      )
+      waitForQuery(id("action-change-dialog-proceed"))
       val proceedButton = webDriver.findElement(By.id("action-change-dialog-proceed"))
       proceedButton.click()
     }

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

@@ -21,7 +21,6 @@ import com.digitalasset.canton.admin.participant.v30.PruningServiceGrpc.PruningS
 import com.digitalasset.canton.admin.participant.v30.{ExportAcsOldResponse, PruningServiceGrpc}
 import com.digitalasset.canton.config.RequireTypes.PositiveInt
 import com.digitalasset.canton.config.{ApiLoggingConfig, ClientConfig, PositiveDurationSeconds}
-import com.digitalasset.canton.discard.Implicits.DiscardOps
 import com.digitalasset.canton.logging.NamedLoggerFactory
 import com.digitalasset.canton.participant.synchronizer.SynchronizerConnectionConfig
 import com.digitalasset.canton.sequencing.{
@@ -63,7 +62,8 @@ import org.lfdecentralizedtrust.splice.environment.TopologyAdminConnection.{
 }
 
 import java.time.Instant
-import scala.concurrent.{ExecutionContext, ExecutionContextExecutor, Future, Promise}
+import scala.concurrent.{ExecutionContext, ExecutionContextExecutor, Future}
+import scala.jdk.CollectionConverters.*
 
 /** Connection to the subset of the Canton admin API that we rely
   * on in our own applications.
@@ -312,13 +312,11 @@ class ParticipantAdminConnection(
       filterSynchronizerId: Option[SynchronizerId] = None,
       timestamp: Option[Instant] = None,
       force: Boolean = false,
-  )(implicit traceContext: TraceContext): Future[ByteString] = {
+  )(implicit traceContext: TraceContext): Future[Seq[ByteString]] = {
     logger.debug(
       show"Downloading ACS snapshot from domain $filterSynchronizerId, for parties $parties at timestamp $timestamp"
     )
-    val requestComplete = Promise[ByteString]()
-    // TODO(DACH-NY/canton-network-node#3298) just concatenate the byteString here. Make it scale to 2M contracts.
-    val observer = new GrpcByteChunksToByteArrayObserver[ExportAcsOldResponse](requestComplete)
+    val observer = new SeqAccumulatingObserver[ExportAcsOldResponse]
     runCmd(
       ParticipantAdminCommands.ParticipantRepairManagement.ExportAcsOld(
         parties = parties,
@@ -328,20 +326,36 @@ class ParticipantAdminConnection(
         observer,
         force,
       )
-    ).discard
-    requestComplete.future
+    ).flatMap(_ => observer.resultFuture).map(_.map(_.chunk))
   }
 
-  def uploadAcsSnapshot(acsBytes: ByteString)(implicit
+  def downloadAcsSnapshotNonChunked(
+      parties: Set[PartyId],
+      filterSynchronizerId: Option[SynchronizerId] = None,
+      timestamp: Option[Instant] = None,
+      force: Boolean = false,
+  )(implicit traceContext: TraceContext): Future[ByteString] =
+    downloadAcsSnapshot(parties, filterSynchronizerId, timestamp, force).map(chunks =>
+      ByteString.copyFrom(chunks.asJava)
+    )
+
+  def uploadAcsSnapshot(acsBytes: Seq[ByteString])(implicit
       traceContext: TraceContext
   ): Future[Unit] = {
+    val chunkedAcsBytes: Seq[ByteString] = acsBytes match {
+      case Seq(bytes) =>
+        // Caller has not chunked the bytes, this is possible for SVs that try to onboard or for validator recovery.
+        // The chuning logic here matches what GrpcStreamingUtils.streamToServer does
+        bytes.toByteArray.grouped(1024 * 1024 * 2).map(ByteString.copyFrom(_)).toSeq
+      case _ => acsBytes
+    }
     retryProvider.retryForClientCalls(
       "import_acs",
       "Imports the acs in the participantl",
       runCmd(
         ParticipantAdminCommands.ParticipantRepairManagement
           .ImportAcsOld(
-            acsBytes,
+            chunkedAcsBytes,
             IMPORT_ACS_WORKFLOW_ID_PREFIX,
             allowContractIdSuffixRecomputation = false,
           ),

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

@@ -0,0 +1,26 @@
+// Copyright (c) 2024 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package org.lfdecentralizedtrust.splice.environment
+
+import scala.collection.mutable.ListBuffer
+import scala.concurrent.{Future, Promise}
+
+class SeqAccumulatingObserver[T] extends io.grpc.stub.StreamObserver[T] {
+  private val promise: Promise[List[T]] = Promise[List[T]]()
+  private val buffer: ListBuffer[T] = ListBuffer.empty[T]
+
+  def resultFuture: Future[Seq[T]] = promise.future
+
+  override def onNext(value: T): Unit = {
+    buffer.append(value)
+  }
+
+  override def onError(t: Throwable): Unit = {
+    promise.failure(t)
+  }
+
+  override def onCompleted(): Unit = {
+    promise.success(buffer.toList)
+  }
+}

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

@@ -69,16 +69,16 @@ class SequencerAdminConnection(
 
   def getGenesisState(timestamp: CantonTimestamp)(implicit
       traceContext: TraceContext
-  ): Future[ByteString] = {
-    val responseObserver = new ByteStringStreamObserver[GenesisStateV2Response](_.chunk)
+  ): Future[Seq[ByteString]] = {
+    val responseObserver = new SeqAccumulatingObserver[GenesisStateV2Response]()
     runCmd(
       TopologyAdminCommands.Read
         .GenesisStateV2(
           timestamp = Some(timestamp),
           synchronizerStore = None,
           observer = responseObserver,
         )
-    ).flatMap(_ => responseObserver.resultBytes)
+    ).flatMap(_ => responseObserver.resultFuture.map(_.map(_.chunk)))
   }
 
   def getOnboardingState(sequencerId: SequencerId)(implicit
@@ -101,7 +101,7 @@ class SequencerAdminConnection(
     topologySnapshot.result.foreach(_.writeDelimitedTo(domainParameters.protocolVersion, builder))
     runCmd(
       SequencerAdminCommands.InitializeFromGenesisStateV2(
-        builder.toByteString,
+        Seq(builder.toByteString),
         domainParameters,
       )
     )
@@ -110,7 +110,7 @@ class SequencerAdminConnection(
   /** This is used for initializing the sequencer after hard domain migrations.
     */
   def initializeFromGenesisState(
-      genesisState: ByteString,
+      genesisState: Seq[ByteString],
       domainParameters: StaticSynchronizerParameters,
   )(implicit traceContext: TraceContext): Future[InitializeSequencerResponse] =
     runCmd(

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

@@ -39,7 +39,7 @@ class AcsExporter(
       parties: PartyId*
   )(implicit
       tc: TraceContext
-  ): Future[ByteString] = {
+  ): Future[Seq[ByteString]] = {
     participantAdminConnection.downloadAcsSnapshot(
       parties = parties.toSet,
       filterSynchronizerId = Some(domain),
@@ -51,7 +51,7 @@ class AcsExporter(
   def safeExportParticipantPartiesAcsFromPausedDomain(domain: SynchronizerId)(implicit
       tc: TraceContext,
       ec: ExecutionContext,
-  ): EitherT[Future, AcsExportFailure, (ByteString, Instant)] = {
+  ): EitherT[Future, AcsExportFailure, (Seq[ByteString], Instant)] = {
     EitherT {
       for {
         participantId <- participantAdminConnection.getId()
@@ -70,7 +70,7 @@ class AcsExporter(
   private def safeExportAcsFromPausedDomain(domain: SynchronizerId, parties: PartyId*)(implicit
       tc: TraceContext,
       ec: ExecutionContext,
-  ): EitherT[Future, AcsExportFailure, (ByteString, Instant)] = {
+  ): EitherT[Future, AcsExportFailure, (Seq[ByteString], Instant)] = {
     for {
       paramsState <- domainStateTopology
         .firstAuthorizedStateForTheLatestSynchronizerParametersState(domain)
@@ -83,7 +83,7 @@ class AcsExporter(
       _ <- EitherT.liftF[Future, AcsExportFailure, Unit](
         waitUntilSynchronizerTime(domain, paramsState.acsExportWaitTimestamp)
       )
-      snapshot <- EitherT.liftF[Future, AcsExportFailure, ByteString](
+      snapshot <- EitherT.liftF[Future, AcsExportFailure, Seq[ByteString]](
         participantAdminConnection.downloadAcsSnapshot(
           parties = parties.toSet,
           filterSynchronizerId = Some(domain),

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

@@ -33,7 +33,7 @@ class DomainDataRestorer(
       synchronizerId: SynchronizerId,
       sequencerConnections: SequencerConnections,
       dars: Seq[Dar],
-      acsSnapshot: ByteString,
+      acsSnapshot: Seq[ByteString],
   )(implicit
       tc: TraceContext
   ): Future[Unit] = {
@@ -93,7 +93,7 @@ class DomainDataRestorer(
       }
   }
 
-  private def importAcs(acs: ByteString)(implicit tc: TraceContext) = {
+  private def importAcs(acs: Seq[ByteString])(implicit tc: TraceContext) = {
     participantAdminConnection.uploadAcsSnapshot(
       acs
     )

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

@@ -0,0 +1,81 @@
+// Copyright (c) 2024 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package org.lfdecentralizedtrust.splice.migration
+
+import com.google.protobuf.ByteString
+import java.io.*
+import java.time.Instant
+import java.util.Base64
+import scala.collection.mutable.ListBuffer
+import scala.jdk.CollectionConverters.*
+import scala.util.Using
+
+object DomainMigrationEncoding {
+  private val base64Decoder = Base64.getDecoder()
+
+  def encode(
+      outputDirectory: Option[String],
+      acsTimestamp: Instant,
+      name: String,
+      content: Seq[ByteString],
+  ): String = {
+    outputDirectory match {
+      case None =>
+        Base64.getEncoder.encodeToString(ByteString.copyFrom(content.asJava).toByteArray)
+      case Some(dir) =>
+        val file = s"$dir/${acsTimestamp}-$name"
+        Using.resource(
+          new DataOutputStream(
+            new BufferedOutputStream(
+              new FileOutputStream(file)
+            )
+          )
+        ) { dos =>
+          writeChunks(dos, content)
+        }
+        file
+    }
+  }
+
+  def decode(separateFiles: Option[Boolean], content: String): Seq[ByteString] = {
+    if (separateFiles.getOrElse(false)) {
+      Using.resource(
+        new DataInputStream(
+          new BufferedInputStream(
+            new FileInputStream(content)
+          )
+        )
+      )(readAllChunks)
+    } else {
+      Seq(ByteString.copyFrom(base64Decoder.decode(content)))
+    }
+  }
+
+  @SuppressWarnings(Array("org.wartremover.warts.Var", "org.wartremover.warts.While"))
+  private def readAllChunks(
+      dis: DataInputStream
+  ): Seq[ByteString] = {
+    val acc = ListBuffer.empty[ByteString]
+    var eof = false
+    while (!eof) {
+      try {
+        val length = dis.readInt()
+        val chunk = new Array[Byte](length)
+        dis.readFully(chunk)
+        acc.addOne(ByteString.copyFrom(chunk))
+      } catch {
+        case _: EOFException =>
+          eof = true
+      }
+    }
+    acc.toSeq
+  }
+
+  private def writeChunks(dos: DataOutputStream, chunks: Seq[ByteString]): Unit = {
+    chunks.foreach { chunk =>
+      dos.writeInt(chunk.size)
+      dos.write(chunk.toByteArray)
+    }
+  }
+}

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

@@ -25,10 +25,11 @@ import java.nio.file.Path
 import java.time.Instant
 import scala.concurrent.{ExecutionContext, Future}
 
-abstract class DomainMigrationTrigger[T: Codec](implicit
+abstract class DomainMigrationTrigger[T](implicit
     ec: ExecutionContext,
     mat: Materializer,
     tracer: Tracer,
+    codec: Codec[T],
 ) extends ScheduledTaskTrigger[DomainMigrationTrigger.Task] {
   protected val participantAdminConnection: ParticipantAdminConnection
   protected val sequencerAdminConnection: Option[SequencerAdminConnection]

apps/common/src/main/scala/org/lfdecentralizedtrust/splice/setup/ParticipantPartyMigrator.scala

@@ -349,7 +349,7 @@ class ParticipantPartyMigrator(
       _ <- MonadUtil.sequentialTraverse(partyIds) { partyId =>
         for {
           acsSnapshot <- getAcsSnapshot(partyId)
-          _ <- participantAdminConnection.uploadAcsSnapshot(acsSnapshot)
+          _ <- participantAdminConnection.uploadAcsSnapshot(Seq(acsSnapshot))
         } yield ()
       }
       _ <- participantAdminConnection.reconnectAllDomains()