[Divulgence pruning] Prune immediate divulgence [DPP-513]

ledger/ledger-api-test-tool/src/main/scala/com/daml/ledger/api/testtool/suites/ParticipantPruningIT.scala

@@ -551,7 +551,11 @@ class ParticipantPruningIT extends LedgerTestSuite {
         divulgence.exerciseCanFetch(_, contract),
       )
 
-      _ <- pruneAtCurrentOffset(participant, bob, pruneAllDivulgedContracts = false)
+      _ <- pruneAtCurrentOffset(
+        participant,
+        bob,
+        pruneAllDivulgedContracts = false,
+      )
 
       // Bob can still see the divulged contract
       _ <- participant.exerciseAndGetContract[Dummy](
@@ -562,7 +566,11 @@ class ParticipantPruningIT extends LedgerTestSuite {
       // Archive the divulged contract
       _ <- participant.exercise(alice, contract.exerciseArchive)
 
-      _ <- pruneAtCurrentOffset(participant, bob, pruneAllDivulgedContracts = false)
+      _ <- pruneAtCurrentOffset(
+        participant,
+        bob,
+        pruneAllDivulgedContracts = false,
+      )
 
       _ <- participant
         .exerciseAndGetContract[Dummy](
@@ -595,15 +603,7 @@ class ParticipantPruningIT extends LedgerTestSuite {
         divulgence.exerciseDivulge(_, contract),
       )
 
-      _ <- divulgencePruneAndCheck(
-        alice,
-        bob,
-        alpha,
-        beta,
-        contract,
-        divulgence,
-        disclosureVisibility = false,
-      )
+      _ <- divulgencePruneAndCheck(alice, bob, alpha, beta, contract, divulgence)
     } yield ()
   })
 
@@ -624,15 +624,7 @@ class ParticipantPruningIT extends LedgerTestSuite {
         divulgeNotDiscloseTemplate.exerciseDivulgeNoDisclose(_, divulgence),
       )
 
-      _ <- divulgencePruneAndCheck(
-        alice,
-        bob,
-        alpha,
-        beta,
-        contract,
-        divulgence,
-        disclosureVisibility = false,
-      )
+      _ <- divulgencePruneAndCheck(alice, bob, alpha, beta, contract, divulgence)
     } yield ()
   })
 
@@ -649,15 +641,8 @@ class ParticipantPruningIT extends LedgerTestSuite {
         alice,
         divulgence.exerciseCreateAndDisclose,
       )
-      _ <- divulgencePruneAndCheck(
-        alice,
-        bob,
-        alpha,
-        beta,
-        contract,
-        divulgence,
-        disclosureVisibility = true,
-      )
+
+      _ <- divulgencePruneAndCheck(alice, bob, alpha, beta, contract, divulgence)
     } yield ()
   })
 
@@ -679,8 +664,6 @@ class ParticipantPruningIT extends LedgerTestSuite {
       beta: ParticipantTestContext,
       contract: Primitive.ContractId[Contract],
       divulgence: binding.Primitive.ContractId[Divulgence],
-      // TODO Divulgence pruning: Remove when immediate divulgence pruning is implemented
-      disclosureVisibility: Boolean,
   )(implicit ec: ExecutionContext) =
     for {
       // Check that Bob can fetch the contract
@@ -715,24 +698,12 @@ class ParticipantPruningIT extends LedgerTestSuite {
 
       _ <- pruneAtCurrentOffset(beta, bob, pruneAllDivulgedContracts = true)
 
-      // TODO Divulgence pruning: Check ACS equality before and after pruning
-      // TODO Divulgence pruning: Remove the true-clause of the if-statement below
-      //                          when disclosure pruning is implemented.
-      _ <-
-        if (disclosureVisibility) {
-          beta
-            .exerciseAndGetContract[Dummy](
-              bob,
-              divulgence.exerciseCanFetch(_, contract),
-            )
-        } else {
-          beta
-            .exerciseAndGetContract[Dummy](
-              bob,
-              divulgence.exerciseCanFetch(_, contract),
-            )
-            .mustFail("Bob cannot access the divulged contract after the second pruning")
-        }
+      _ <- beta
+        .exerciseAndGetContract[Dummy](
+          bob,
+          divulgence.exerciseCanFetch(_, contract),
+        )
+        .mustFail("Bob cannot access the divulged contract after the second pruning")
     } yield ()
 
   private def populateLedgerAndGetOffsets(participant: ParticipantTestContext, submitter: Party)(
@@ -783,13 +754,22 @@ class ParticipantPruningIT extends LedgerTestSuite {
 
   private def pruneAtCurrentOffset(
       participant: ParticipantTestContext,
-      party: Party,
+      localParty: Party,
       pruneAllDivulgedContracts: Boolean,
   )(implicit ec: ExecutionContext): Future[Unit] =
     for {
       offset <- participant.currentEnd()
       // Dummy needed to prune at this offset
-      _ <- participant.create(party, Dummy(party))
+      _ <- participant.create(localParty, Dummy(localParty))
+
+      acsBeforePruning <- participant.activeContracts(localParty)
       _ <- participant.prune(offset, pruneAllDivulgedContracts = pruneAllDivulgedContracts)
-    } yield ()
+      acsAfterPruning <- participant.activeContracts(localParty)
+
+    } yield {
+      assert(
+        acsBeforePruning == acsAfterPruning,
+        s"Active contract set comparison before and after pruning failed: $acsBeforePruning vs $acsAfterPruning",
+      )
+    }
 }

ledger/ledger-on-memory/BUILD.bazel

@@ -226,7 +226,7 @@ conformance_test(
         "--verbose",
         "--include=ParticipantPruningIT",
         # Disable tests targeting only append-only schema functionality
-        "--exclude=ParticipantPruningIT:PRLocalAndNonLocalRetroactiveDivulgences,ParticipantPruningIT:PRRetroactiveDivulgences",
+        "--exclude=ParticipantPruningIT:PRLocalAndNonLocalRetroactiveDivulgences,ParticipantPruningIT:PRRetroactiveDivulgences,ParticipantPruningIT:PRDisclosureAndRetroactiveDivulgence",
     ],
 )
 

ledger/ledger-on-sql/BUILD.bazel

@@ -280,7 +280,7 @@ da_scala_test_suite(
                 "--verbose",
                 "--include=ParticipantPruningIT",
                 # Disable tests targeting only append-only schema functionality
-                "--exclude=ParticipantPruningIT:PRLocalAndNonLocalRetroactiveDivulgences,ParticipantPruningIT:PRRetroactiveDivulgences",
+                "--exclude=ParticipantPruningIT:PRLocalAndNonLocalRetroactiveDivulgences,ParticipantPruningIT:PRRetroactiveDivulgences,ParticipantPruningIT:PRDisclosureAndRetroactiveDivulgence",
             ],
         ),
         conformance_test(
@@ -441,6 +441,8 @@ conformance_test(
     test_tool_args = [
         "--verbose",
         "--include=ParticipantPruningIT",
+        # Disable tests targeting only multi-participant setups
+        "--exclude=ParticipantPruningIT:PRDisclosureAndRetroactiveDivulgence",
     ],
 )
 
@@ -459,6 +461,8 @@ conformance_test(
     test_tool_args = [
         "--verbose",
         "--include=ParticipantPruningIT",
+        # Disable tests targeting only multi-participant setups
+        "--exclude=ParticipantPruningIT:PRDisclosureAndRetroactiveDivulgence",
     ],
 )
 
@@ -477,6 +481,8 @@ conformance_test(
     test_tool_args = [
         "--verbose",
         "--include=ParticipantPruningIT",
+        # Disable tests targeting only multi-participant setups
+        "--exclude=ParticipantPruningIT:PRDisclosureAndRetroactiveDivulgence",
     ],
 )
 
@@ -493,7 +499,7 @@ conformance_test(
         "--verbose",
         "--include=ParticipantPruningIT",
         # Disable tests targeting only append-only schema functionality
-        "--exclude=ParticipantPruningIT:PRLocalAndNonLocalRetroactiveDivulgences,ParticipantPruningIT:PRRetroactiveDivulgences",
+        "--exclude=ParticipantPruningIT:PRLocalAndNonLocalRetroactiveDivulgences,ParticipantPruningIT:PRRetroactiveDivulgences,ParticipantPruningIT:PRDisclosureAndRetroactiveDivulgence",
     ],
 )
 

ledger/participant-integration-api/src/main/scala/platform/store/backend/common/CommonStorageBackend.scala

@@ -37,7 +37,6 @@ import scalaz.syntax.tag._
 private[backend] trait CommonStorageBackend[DB_BATCH] extends StorageBackend[DB_BATCH] {
 
   private val logger: ContextualizedLogger = ContextualizedLogger.get(this.getClass)
-
   // Ingestion
 
   private val SQL_DELETE_OVERSPILL_ENTRIES: List[SqlQuery] =
@@ -55,6 +54,8 @@ private[backend] trait CommonStorageBackend[DB_BATCH] extends StorageBackend[DB_
       SQL("DELETE FROM party_entries WHERE ledger_offset > {ledger_offset}"),
     )
 
+  def queryStrategy: QueryStrategy
+
   override def initializeIngestion(
       connection: Connection
   ): Option[ParameterStorageBackend.LedgerEnd] = {
@@ -479,6 +480,7 @@ private[backend] trait CommonStorageBackend[DB_BATCH] extends StorageBackend[DB_
 
   // Events
 
+  // TODO Cleanup: Extract to [[EventStorageBackendTemplate]]
   def pruneEvents(
       pruneUpToInclusive: Offset,
       pruneAllDivulgedContracts: Boolean,
@@ -524,6 +526,35 @@ private[backend] trait CommonStorageBackend[DB_BATCH] extends StorageBackend[DB_
             )"""
     }(connection, loggingContext)
 
+    if (pruneAllDivulgedContracts) {
+      val pruneAfterClause = {
+        // We need to distinguish between the two cases since lexicographical comparison
+        // in Oracle doesn't work with '' (empty strings are treated as NULLs) as one of the operands
+        participantAllDivulgedContractsPrunedUpToInclusive(connection) match {
+          case Some(pruneAfter) => cSQL"and event_offset > $pruneAfter"
+          case None => cSQL""
+        }
+      }
+
+      pruneWithLogging(queryDescription = "Immediate divulgence events pruning") {
+        SQL"""
+            -- Immediate divulgence pruning
+            delete from participant_events_create c
+            where event_offset <= $pruneUpToInclusive
+            -- Only prune create events which did not have a locally hosted party before their creation offset
+            and not exists (
+              select 1
+              from party_entries p
+              where p.typ = 'accept'
+              and p.ledger_offset <= c.event_offset
+              and #${queryStrategy.isTrue("p.is_local")}
+              and #${queryStrategy.arrayContains("c.flat_event_witnesses", "p.party")}
+            )
+            $pruneAfterClause
+         """
+      }(connection, loggingContext)
+    }
+
     pruneWithLogging(queryDescription = "Exercise (consuming) events pruning") {
       SQL"""
           -- Exercise events (consuming)
@@ -541,6 +572,14 @@ private[backend] trait CommonStorageBackend[DB_BATCH] extends StorageBackend[DB_
     }(connection, loggingContext)
   }
 
+  private def participantAllDivulgedContractsPrunedUpToInclusive(
+      connection: Connection
+  ): Option[Offset] =
+    SQL"select participant_all_divulged_contracts_pruned_up_to_inclusive from parameters"
+      .as(offset("participant_all_divulged_contracts_pruned_up_to_inclusive").?.single)(
+        connection
+      )
+
   private def pruneWithLogging(queryDescription: String)(query: SimpleSql[Row])(
       connection: Connection,
       loggingContext: LoggingContext,

ledger/participant-integration-api/src/main/scala/platform/store/backend/common/QueryStrategy.scala

@@ -47,4 +47,12 @@ trait QueryStrategy {
     * @return the function name
     */
   def booleanOrAggregationFunction: String = "bool_or"
+
+  /** Predicate which tests if the element referenced by the `elementColumnName`
+    * is in the array from column `arrayColumnName`
+    */
+  def arrayContains(arrayColumnName: String, elementColumnName: String): String
+
+  /** Boolean predicate */
+  def isTrue(booleanColumnName: String): String
 }

ledger/participant-integration-api/src/main/scala/platform/store/backend/h2/H2StorageBackend.scala

@@ -127,6 +127,10 @@ private[backend] object H2StorageBackend
           .map(p => cSQL"array_contains(#$columnName, '#${p.toString}')")
           .mkComposite("(", " or ", ")")
 
+    override def arrayContains(arrayColumnName: String, elementColumnName: String): String =
+      s"array_contains($arrayColumnName, $elementColumnName)"
+
+    override def isTrue(booleanColumnName: String): String = booleanColumnName
   }
 
   override def queryStrategy: QueryStrategy = H2QueryStrategy

ledger/participant-integration-api/src/main/scala/platform/store/backend/oracle/OracleStorageBackend.scala

@@ -113,6 +113,11 @@ private[backend] object OracleStorageBackend
       s"""case when ($column = $value) then 1 else 0 end"""
 
     override def booleanOrAggregationFunction: String = "max"
+
+    override def arrayContains(arrayColumnName: String, elementColumnName: String): String =
+      s"EXISTS (SELECT 1 FROM JSON_TABLE($arrayColumnName, '$$[*]' columns (value PATH '$$')) WHERE value = $elementColumnName)"
+
+    override def isTrue(booleanColumnName: String): String = s"$booleanColumnName = 1"
   }
 
   override def queryStrategy: QueryStrategy = OracleQueryStrategy

ledger/participant-integration-api/src/main/scala/platform/store/backend/postgresql/PostgresStorageBackend.scala

@@ -188,6 +188,10 @@ private[backend] object PostgresStorageBackend
       cSQL"#$columnName::text[] && $partiesArray::text[]"
     }
 
+    override def arrayContains(arrayColumnName: String, elementColumnName: String): String =
+      s"$elementColumnName = any($arrayColumnName)"
+
+    override def isTrue(booleanColumnName: String): String = booleanColumnName
   }
 
   override def queryStrategy: QueryStrategy = PostgresQueryStrategy

ledger/participant-integration-api/src/test/lib/scala/platform/store/backend/StorageBackendTestValues.scala

@@ -70,6 +70,7 @@ private[backend] object StorageBackendTestValues {
   def dtoPartyEntry(
       offset: Offset,
       party: String = someParty,
+      isLocal: Boolean = true,
   ): DbDto.PartyEntry = DbDto.PartyEntry(
     ledger_offset = offset.toHexString,
     recorded_at = someTime,
@@ -78,7 +79,7 @@ private[backend] object StorageBackendTestValues {
     display_name = Some(party),
     typ = JdbcLedgerDao.acceptType,
     rejection_reason = None,
-    is_local = Some(true),
+    is_local = Some(isLocal),
   )
 
   def dtoPackage(offset: Offset): DbDto.Package = DbDto.Package(