Test handling of multiple verdicts with the same update_id

apps/scan/src/main/openapi/scan.yaml

@@ -4060,6 +4060,9 @@ components:
         update_id:
           description: |
             The ID of the transaction update associated with this verdict.
+
+            Verdicts are deduplicated by update_id. Only the first verdict ingested for a given update_id is stored and returned, while a subsequent verdict with the same update_id is rejected.
+            This can happen for example if a sequencer client retries a successful submission. In that case, the retry is rejected as a duplicate, and the events endpoint will only show the successful verdict but not the rejected duplicates.
           type: string
         migration_id:
           description: |

apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/automation/ScanVerdictIngestionService.scala

@@ -11,7 +11,7 @@ import com.digitalasset.base.error.utils.ErrorDetails
 import com.digitalasset.canton.data.CantonTimestamp
 import com.digitalasset.canton.discard.Implicits.DiscardOps
 import com.digitalasset.canton.lifecycle.{AsyncOrSyncCloseable, SyncCloseable}
-import com.digitalasset.canton.logging.NamedLoggerFactory
+import com.digitalasset.canton.logging.{NamedLoggerFactory, TracedLogger}
 import com.digitalasset.canton.mediator.admin.v30
 import com.digitalasset.canton.sequencing.traffic.TrafficControlErrors
 import com.digitalasset.canton.time.Clock
@@ -64,6 +64,40 @@ object ScanVerdictIngestionService {
         .filter(_ >= start)
         .filterNot(summaryTimes.contains)
   }
+
+  /** Groups a batch of verdicts by update id and returns only the update ids that
+    * appear more than once within the batch.
+    */
+  def findDuplicateUpdateIds(batch: Seq[v30.Verdict]): Map[String, Seq[(v30.Verdict, Int)]] =
+    batch.zipWithIndex.groupBy(_._1.updateId).filter(_._2.size > 1)
+
+  /** True if any verdict in the duplicate groups has an accept after another verdict. */
+  def duplicatesContainSubsequentAccept(duplicates: Map[String, Seq[(v30.Verdict, Int)]]): Boolean =
+    duplicates.exists { case (_, group) =>
+      group.drop(1).exists(_._1.verdict == v30.VerdictResult.VERDICT_RESULT_ACCEPTED)
+    }
+
+  /** Finds and logs when duplicate verdicts exist in a batch.
+    * Logs at WARN level when any duplicate is an accept, otherwise at INFO level.
+    */
+  def logDuplicateUpdateIds(batch: Seq[v30.Verdict], logger: TracedLogger)(implicit
+      tc: TraceContext
+  ): Unit = {
+    val duplicates = findDuplicateUpdateIds(batch)
+    if (duplicates.nonEmpty) {
+      val message = s"Received multiple verdicts with the same update id in the same batch. " +
+        s"Batch: ${batch.size} verdicts with record times ${batch.map(_.getRecordTime).map(CantonTimestamp.tryFromProtoTimestamp).mkString("[", ",", "]")}. " +
+        s"Duplicate verdicts: ${duplicates.values.flatten
+            .map { case (verdict, index) =>
+              s"${index} => ${verdict}"
+            }
+            .mkString("[\n", ",\n", "\n]")}"
+
+      if (duplicatesContainSubsequentAccept(duplicates))
+        logger.warn(s"$message Duplicate verdicts contains a subsequent accept.")
+      else logger.info(message)
+    }
+  }
 }
 
 class ScanVerdictIngestionService(
@@ -387,22 +421,7 @@ class ScanVerdictIngestionService(
       .batch(math.max(1, config.mediatorVerdictIngestion.batchSize.toLong), Vector(_))(_ :+ _)
       // TODO(DACH-NY/cn-test-failures#8281): Remove once we have figured out why we're getting duplicate data.
       .map(batch => {
-        val duplicates = batch.zipWithIndex
-          .groupBy(_._1.updateId)
-          .filter(_._2.size > 1)
-
-        if (duplicates.nonEmpty) {
-          logger.info(
-            s"Received multiple verdicts with the same update id in the same batch. " +
-              s"Batch: ${batch.size} verdicts with record times ${batch.map(_.getRecordTime).map(CantonTimestamp.tryFromProtoTimestamp).mkString("[", ",", "]")}. " +
-              s"Duplicate verdicts: ${duplicates.values.flatten
-                  .map { case (verdict, index) =>
-                    s"${index} => ${verdict}"
-                  }
-                  .mkString("[\n", ",\n", "\n]")}"
-          )
-        }
-
+        ScanVerdictIngestionService.logDuplicateUpdateIds(batch, logger)
         batch
       })
 

apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbScanVerdictStore.scala

@@ -421,10 +421,22 @@ class DbScanVerdictStore(
 
       for {
         alreadyExisting <- checkExist.map(_.toSet)
-        nonExisting = items.filter(item => !alreadyExisting.contains(item._1.updateId))
-        _ = logger.info(
-          s"Already ingested verdicts: $alreadyExisting. Non-existing: ${nonExisting.map(_._1.updateId)}."
-        )
+        (dropped, nonExisting) = items.partition(item => alreadyExisting.contains(item._1.updateId))
+        droppedAccepts =
+          dropped.filter(_._1.verdictResult == DbScanVerdictStore.VerdictResultDbValue.Accepted)
+        nonExistingMessage = s"Non-existing: ${nonExisting.map(_._1.updateId)}."
+        _ =
+          if (droppedAccepts.nonEmpty)
+            logger.warn(
+              s"Dropping duplicate accepted verdicts: ${droppedAccepts.map(_._1.updateId)}. " +
+                s"All dropped verdicts: ${dropped.map(_._1.updateId)}. $nonExistingMessage"
+            )
+          else if (dropped.nonEmpty)
+            logger.info(
+              s"Dropping duplicate verdicts: ${dropped.map(_._1.updateId)}. $nonExistingMessage"
+            )
+          else
+            logger.info(s"Already ingested verdicts: $alreadyExisting. $nonExistingMessage")
         rowIdMap <-
           if (nonExisting.nonEmpty) {
             DBIO

apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/automation/ScanVerdictIngestionServiceTest.scala

@@ -1,13 +1,24 @@
 package org.lfdecentralizedtrust.splice.scan.automation
 
+import com.digitalasset.canton.BaseTest
 import com.digitalasset.canton.data.CantonTimestamp
-import org.scalatest.matchers.should.Matchers
+import com.digitalasset.canton.logging.SuppressionRule
+import com.digitalasset.canton.mediator.admin.v30
 import org.scalatest.wordspec.AnyWordSpec
+import org.slf4j.event.Level.INFO
 
-class ScanVerdictIngestionServiceTest extends AnyWordSpec with Matchers {
+class ScanVerdictIngestionServiceTest extends AnyWordSpec with BaseTest {
 
   private def ts(micros: Long) = CantonTimestamp.ofEpochMicro(micros)
 
+  private def mkVerdict(updateId: String, accepted: Boolean): v30.Verdict =
+    v30.Verdict.defaultInstance.copy(
+      updateId = updateId,
+      verdict =
+        if (accepted) v30.VerdictResult.VERDICT_RESULT_ACCEPTED
+        else v30.VerdictResult.VERDICT_RESULT_REJECTED,
+    )
+
   "findMissingTrafficSummaries" should {
 
     "return empty when ingestion hasn't started" in {
@@ -50,4 +61,96 @@ class ScanVerdictIngestionServiceTest extends AnyWordSpec with Matchers {
       ) shouldBe empty
     }
   }
+
+  "findDuplicateUpdateIds" should {
+
+    "return empty when all update ids are distinct" in {
+      ScanVerdictIngestionService.findDuplicateUpdateIds(
+        Seq(mkVerdict("a", true), mkVerdict("b", false))
+      ) shouldBe empty
+    }
+
+    "return only the update ids that appear more than once" in {
+      val result = ScanVerdictIngestionService.findDuplicateUpdateIds(
+        Seq(mkVerdict("a", true), mkVerdict("b", false), mkVerdict("a", false))
+      )
+      result.keySet shouldBe Set("a")
+      result("a").map(_._2) shouldBe Seq(0, 2)
+    }
+  }
+
+  "duplicatesContainSubsequentAccept" should {
+
+    "return false when no duplicate is an accept" in {
+      val duplicates = ScanVerdictIngestionService.findDuplicateUpdateIds(
+        Seq(mkVerdict("a", false), mkVerdict("a", false))
+      )
+      ScanVerdictIngestionService.duplicatesContainSubsequentAccept(duplicates) shouldBe false
+    }
+
+    "return true when a duplicate group contains an accept after another verdict" in {
+      val duplicates = ScanVerdictIngestionService.findDuplicateUpdateIds(
+        Seq(mkVerdict("a", false), mkVerdict("a", true))
+      )
+      ScanVerdictIngestionService.duplicatesContainSubsequentAccept(duplicates) shouldBe true
+    }
+
+    "return false when a duplicate group contains an accept before another verdict" in {
+      val duplicates = ScanVerdictIngestionService.findDuplicateUpdateIds(
+        Seq(mkVerdict("a", true), mkVerdict("a", false))
+      )
+      ScanVerdictIngestionService.duplicatesContainSubsequentAccept(duplicates) shouldBe false
+    }
+
+    "ignore an accept that is not part of a duplicate group" in {
+      // "a" is accepted but unique; only "b" is duplicated (both rejected).
+      val duplicates = ScanVerdictIngestionService.findDuplicateUpdateIds(
+        Seq(mkVerdict("a", true), mkVerdict("b", false), mkVerdict("b", false))
+      )
+      ScanVerdictIngestionService.duplicatesContainSubsequentAccept(duplicates) shouldBe false
+    }
+  }
+
+  "logDuplicateUpdateIds" should {
+
+    "not log when there are no duplicates" in {
+      loggerFactory.assertLogsSeq(SuppressionRule.LevelAndAbove(INFO))(
+        ScanVerdictIngestionService.logDuplicateUpdateIds(
+          Seq(mkVerdict("a", true), mkVerdict("b", true)),
+          logger,
+        ),
+        _ shouldBe empty,
+      )
+    }
+
+    "log at info level when no duplicate is an accept" in {
+      loggerFactory.assertLogs(SuppressionRule.LevelAndAbove(INFO))(
+        ScanVerdictIngestionService.logDuplicateUpdateIds(
+          Seq(mkVerdict("a", false), mkVerdict("a", false)),
+          logger,
+        ),
+        _.infoMessage should include("Duplicate verdicts:"),
+      )
+    }
+
+    "log at warning level when a duplicate is a subsequent accept" in {
+      loggerFactory.assertLogs(
+        ScanVerdictIngestionService.logDuplicateUpdateIds(
+          Seq(mkVerdict("a", false), mkVerdict("a", true)),
+          logger,
+        ),
+        _.warningMessage should endWith("Duplicate verdicts contains a subsequent accept."),
+      )
+    }
+
+    "log at info level when a duplicate is not a subsequent accept" in {
+      loggerFactory.assertLogs(SuppressionRule.LevelAndAbove(INFO))(
+        ScanVerdictIngestionService.logDuplicateUpdateIds(
+          Seq(mkVerdict("a", true), mkVerdict("a", false)),
+          logger,
+        ),
+        _.infoMessage should include("Duplicate verdicts:"),
+      )
+    }
+  }
 }

apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/DbAppActivityRecordStoreTest.scala

@@ -412,6 +412,101 @@ class DbAppActivityRecordStoreTest
         countAfter shouldBe 0L
       }
     }
+
+    "drop a rejected verdict that's a duplicate of a prior accept" in {
+      val updateId = "update-dupe-reject-after-accept"
+      val ts1 = CantonTimestamp.now()
+      val ts2 = ts1.plusSeconds(1L)
+      for {
+        (appStore, verdictStore) <- newStores()
+        accepted = mkVerdict(
+          verdictStore,
+          updateId,
+          ts1,
+          DbScanVerdictStore.VerdictResultDbValue.Accepted,
+        )
+        rejected = mkVerdict(
+          verdictStore,
+          updateId,
+          ts2,
+          DbScanVerdictStore.VerdictResultDbValue.Rejected,
+        )
+        // First batch with the accepted verdict and its activity record
+        _ <- verdictStore.insertVerdictsWithAppActivityRecords(
+          NonEmptyList.of(accepted -> noViews),
+          Seq(ts1 -> mkRecord(0L, 10L, Seq("app1::provider"), Seq(100L))),
+          hasTrafficSummaries = true,
+          firstActiveRoundO = Some(10L),
+          lastArchivedRoundO = Some(9L),
+        )
+        countAfterBatch1 <- countRecords()
+        // A later batch with a rejection for the same update_id
+        _ <- verdictStore.insertVerdictsWithAppActivityRecords(
+          NonEmptyList.of(rejected -> noViews),
+          Seq(ts2 -> mkRecord(0L, 11L, Seq("app2::provider"), Seq(200L))),
+          hasTrafficSummaries = true,
+          firstActiveRoundO = Some(11L),
+          lastArchivedRoundO = Some(10L),
+        )
+        v <- verdictStore.getVerdictByUpdateId(updateId)
+        countAfterBatch2 <- countRecords()
+      } yield {
+        v shouldBe defined
+        v.value.verdictResult shouldBe DbScanVerdictStore.VerdictResultDbValue.Accepted
+        v.value.recordTime shouldBe ts1
+        // The rejection's activity record was dropped along with the verdict
+        countAfterBatch2 shouldBe countAfterBatch1
+      }
+    }
+
+    "drop and warn of an accepted verdict that's a duplicate of a prior rejection" in {
+      val updateId = "update-dupe-accept-after-reject"
+      val ts1 = CantonTimestamp.now()
+      val ts2 = ts1.plusSeconds(1L)
+      for {
+        (appStore, verdictStore) <- newStores()
+        rejected = mkVerdict(
+          verdictStore,
+          updateId,
+          ts1,
+          DbScanVerdictStore.VerdictResultDbValue.Rejected,
+        )
+        accepted = mkVerdict(
+          verdictStore,
+          updateId,
+          ts2,
+          DbScanVerdictStore.VerdictResultDbValue.Accepted,
+        )
+        // First batch with the rejection and its activity record
+        _ <- verdictStore.insertVerdictsWithAppActivityRecords(
+          NonEmptyList.of(rejected -> noViews),
+          Seq(ts1 -> mkRecord(0L, 10L, Seq("app1::provider"), Seq(100L))),
+          hasTrafficSummaries = true,
+          firstActiveRoundO = Some(10L),
+          lastArchivedRoundO = Some(9L),
+        )
+        countAfterBatch1 <- countRecords()
+        // A later batch with an accept for the same update_id
+        _ <- loggerFactory.assertLogs(
+          verdictStore.insertVerdictsWithAppActivityRecords(
+            NonEmptyList.of(accepted -> noViews),
+            Seq(ts2 -> mkRecord(0L, 11L, Seq("app2::provider"), Seq(200L))),
+            hasTrafficSummaries = true,
+            firstActiveRoundO = Some(11L),
+            lastArchivedRoundO = Some(10L),
+          ),
+          _.warningMessage should startWith("Dropping duplicate accepted verdicts"),
+        )
+        v <- verdictStore.getVerdictByUpdateId(updateId)
+        countAfterBatch2 <- countRecords()
+      } yield {
+        v shouldBe defined
+        v.value.verdictResult shouldBe DbScanVerdictStore.VerdictResultDbValue.Rejected
+        v.value.recordTime shouldBe ts1
+        // The accept's activity record was dropped along with the verdict
+        countAfterBatch2 shouldBe countAfterBatch1
+      }
+    }
   }
 
   "earliestRoundWithCompleteAppActivity" should {
@@ -1202,6 +1297,7 @@ class DbAppActivityRecordStoreTest
       verdictStore: DbScanVerdictStore,
       updateId: String,
       recordTs: CantonTimestamp,
+      verdictResult: Short = DbScanVerdictStore.VerdictResultDbValue.Accepted,
   ): verdictStore.VerdictT =
     new verdictStore.VerdictT(
       rowId = 0L,
@@ -1210,7 +1306,7 @@ class DbAppActivityRecordStoreTest
       recordTime = recordTs,
       finalizationTime = recordTs,
       submittingParticipantUid = "participant1",
-      verdictResult = DbScanVerdictStore.VerdictResultDbValue.Accepted,
+      verdictResult = verdictResult,
       mediatorGroup = 0,
       updateId = updateId,
       submittingParties = Seq.empty,

apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/ScanEventStoreTest.scala

@@ -86,17 +86,20 @@ class ScanEventStoreTest extends StoreTestBase with HasExecutionContext with Spl
       for {
         ctx <- newEventStore()
         _ <- insertVerdict(ctx.verdictStore, "update1", CantonTimestamp.MinValue.plusSeconds(1L))
-        _ <- ctx.verdictStore.insertVerdictAndTransactionViews(
-          Seq(
-            // won't be reinserted
-            mkVerdict(ctx.verdictStore, "update1", CantonTimestamp.MinValue.plusSeconds(1L)) -> (
-              (_: Long) => Seq.empty
-            ),
-            // will be inserted
-            mkVerdict(ctx.verdictStore, "update2", CantonTimestamp.MinValue.plusSeconds(2L)) -> (
-              (_: Long) => Seq.empty
-            ),
-          )
+        _ <- loggerFactory.assertLogs(
+          ctx.verdictStore.insertVerdictAndTransactionViews(
+            Seq(
+              // won't be reinserted
+              mkVerdict(ctx.verdictStore, "update1", CantonTimestamp.MinValue.plusSeconds(1L)) -> (
+                (_: Long) => Seq.empty
+              ),
+              // will be inserted
+              mkVerdict(ctx.verdictStore, "update2", CantonTimestamp.MinValue.plusSeconds(2L)) -> (
+                (_: Long) => Seq.empty
+              ),
+            )
+          ),
+          _.warningMessage should startWith("Dropping duplicate accepted verdicts"),
         )
         result <- ctx.verdictStore.listVerdicts(None, includeImportUpdates = false, 10)
       } yield {