Support configuring pruning for DABFT

apps/app/src/main/scala/org/lfdecentralizedtrust/splice/console/SvAppReference.scala

@@ -430,6 +430,11 @@ class SvAppBackendReference(
   lazy val sequencerClient: SequencerClientReference =
     sequencerClientFor(_.current)
 
+  lazy val sequencerClientSuccessor: SequencerClientReference =
+    sequencerClientFor(
+      _.successor.getOrElse(throw new IllegalStateException("successor not configured"))
+    )
+
   def sequencerClientFor(
       node: SvSynchronizerNodesConfig => SvSynchronizerNodeConfig
   ): SequencerClientReference = {

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

@@ -4,9 +4,10 @@ import better.files.File.apply
 import cats.implicits.catsSyntaxOptionId
 import com.digitalasset.canton.{HasExecutionContext, SynchronizerAlias}
 import com.digitalasset.canton.admin.api.client.data
+import com.digitalasset.canton.admin.api.client.data.PruningSchedule
 import com.digitalasset.canton.concurrent.Threading
 import com.digitalasset.canton.config.CantonRequireTypes.InstanceName
-import com.digitalasset.canton.config.NonNegativeFiniteDuration
+import com.digitalasset.canton.config.{NonNegativeFiniteDuration, PositiveDurationSeconds}
 import com.digitalasset.canton.config.RequireTypes.NonNegativeInt
 import com.digitalasset.canton.data.CantonTimestamp
 import com.digitalasset.canton.topology.admin.grpc.TopologyStoreId.Synchronizer
@@ -377,6 +378,16 @@ class RollForwardLsuIntegrationTest
         aliceValidatorWalletClient.tap(100.0)
       }
 
+      clue("SV1's DABFT node has pruning config set") {
+        sv1LocalBackend.sequencerClient.bft.pruning.get_schedule() shouldBe Some(
+          PruningSchedule(
+            "0 /10 * * * ?",
+            PositiveDurationSeconds.ofMinutes(5),
+            PositiveDurationSeconds.ofDays(30),
+          )
+        )
+      }
+
       clue("stop apps manually to prevent errors from the synchronizer being force stopped") {
         stopAllAsync(allNodes*).futureValue
         allSvLocalBackends.par.foreach(

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

@@ -9,6 +9,7 @@ import com.daml.grpc.adapter.ExecutionSequencerFactory
 import com.daml.grpc.adapter.client.pekko.ClientAdapter
 import com.digitalasset.canton.admin.api.client.commands.{
   GrpcAdminCommand,
+  PruningSchedulerCommands,
   SequencerAdminCommands,
   TopologyAdminCommands,
 }
@@ -25,6 +26,7 @@ import com.digitalasset.canton.protocol.StaticSynchronizerParameters
 import com.digitalasset.canton.sequencer.admin.v30.{
   OnboardingStateV2Request,
   OnboardingStateV2Response,
+  SequencerBftPruningAdministrationServiceGrpc,
 }
 import com.digitalasset.canton.sequencing.protocol
 import com.digitalasset.canton.synchronizer.sequencer.SequencerPruningStatus
@@ -85,12 +87,27 @@ class SequencerAdminConnection(
       retryProvider,
     )
     with StatusAdminConnection
-    with SequencerBftAdminConnection {
+    with SequencerBftAdminConnection
+    with PruningAdminConnection {
 
   override val serviceName = "Canton Sequencer Admin API"
 
   override type Status = SequencerStatus
 
+  override val pruningCommands: PruningSchedulerCommands[
+    SequencerBftPruningAdministrationServiceGrpc.SequencerBftPruningAdministrationServiceStub
+  ] = new PruningSchedulerCommands[
+    SequencerBftPruningAdministrationServiceGrpc.SequencerBftPruningAdministrationServiceStub
+  ](
+    SequencerBftPruningAdministrationServiceGrpc.stub,
+    _.setSchedule(_),
+    _.clearSchedule(_),
+    _.setCron(_),
+    _.setMaxDuration(_),
+    _.setRetention(_),
+    _.getSchedule(_),
+  )
+
   override protected def getStatusRequest: GrpcAdminCommand[?, ?, NodeStatus[SequencerStatus]] =
     SequencerAdminCommands.Health.SequencerStatusCommand()
 

apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/LocalSynchronizerNode.scala

@@ -529,6 +529,14 @@ class LocalSynchronizerNode(
       config.mediator.pruning
     )
 
+  def ensureDABFTPruningSchedule()(implicit tc: TraceContext): Future[Unit] = {
+    if (config.sequencer.isBftSequencer) {
+      sequencerAdminConnection.ensurePruningSchedule(
+        config.sequencer.dabftPruning
+      )
+    } else Future.unit
+  }
+
   override protected def onClosed(): Unit = {
     LifeCycle.close(sequencerAdminConnection, mediatorAdminConnection)(logger)
     super.onClosed()

apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/SvApp.scala

@@ -484,6 +484,7 @@ class SvApp(
               ) {
                 node.ensureMediatorPruningSchedule()
               }
+              _ <- node.ensureDABFTPruningSchedule()
             } yield ()
           } else {
             logger.info(

apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/config/SvAppConfig.scala

@@ -489,6 +489,13 @@ final case class SvSequencerConfig(
     sequencerAvailabilityDelay: NonNegativeFiniteDuration = NonNegativeFiniteDuration.ofSeconds(60),
     pruning: Option[SequencerPruningConfig] = None,
     isBftSequencer: Boolean = false,
+    dabftPruning: Option[PruningConfig] = Some(
+      PruningConfig(
+        cron = "0 /10 * * * ?", // Run every 10min,
+        maxDuration = PositiveDurationSeconds.ofMinutes(5),
+        retention = PositiveDurationSeconds.ofDays(30),
+      )
+    ),
 ) {
   def toCantonConfig: RemoteSequencerConfig = RemoteSequencerConfig(
     adminApi,