Instrument buffer after contract state events stream (#10209)

CHANGELOG_BEGIN CHANGELOG_END
digital-asset · Jul 12, 2021 · 3fd3abf · 3fd3abf
1 parent 643a2de
commit 3fd3abf
Show file tree

Hide file tree

Showing 2 changed files with 19 additions and 8 deletions.
diff --git a/ledger/metrics/src/main/scala/com/daml/metrics/Metrics.scala b/ledger/metrics/src/main/scala/com/daml/metrics/Metrics.scala
@@ -367,6 +367,9 @@ final class Metrics(val registry: MetricRegistry) {
       val transactionLogUpdatesBufferSize: Counter =
         registry.counter(Prefix :+ "transaction_log_updates_buffer_size")
 
+      val contractStateEventsBufferSize: Counter =
+        registry.counter(Prefix :+ "contract_state_events_buffer_size")
+
       // FIXME Name mushing and inconsistencies here, tracked by https://github.com/digital-asset/daml/issues/5926
       object db {
         private val Prefix: MetricName = index.Prefix :+ "db"

diff --git a/...tegration-api/src/main/scala/platform/store/appendonlydao/events/TransactionsReader.scala b/...tegration-api/src/main/scala/platform/store/appendonlydao/events/TransactionsReader.scala
@@ -318,14 +318,16 @@ private[appendonlydao] final class TransactionsReader(
         )
       }
 
+    val transactionLogUpdatesSource = groupContiguous(eventsSource)(by = _.transactionId)
+      .map { v =>
+        val tx = toTransaction(v)
+        (tx.offset, tx.events.last.eventSequentialId) -> tx
+      }
+      .mapMaterializedValue(_ => NotUsed)
+
     InstrumentedSource
       .bufferedSource(
-        original = groupContiguous(eventsSource)(by = _.transactionId)
-          .map { v =>
-            val tx = toTransaction(v)
-            (tx.offset, tx.events.last.eventSequentialId) -> tx
-          }
-          .mapMaterializedValue(_ => NotUsed),
+        original = transactionLogUpdatesSource,
         counter = metrics.daml.index.transactionLogUpdatesBufferSize,
         size = outputStreamBufferSize,
       )
@@ -406,7 +408,7 @@ private[appendonlydao] final class TransactionsReader(
       )
     )
 
-    Source
+    val contractStateEventsSource = Source
       .fromIterator(() =>
         TransactionsReader
           .splitRange(
@@ -445,7 +447,13 @@ private[appendonlydao] final class TransactionsReader(
       }
       .map(event => (event.eventOffset, event.eventSequentialId) -> event)
       .mapMaterializedValue(_ => NotUsed)
-      .buffer(outputStreamBufferSize, OverflowStrategy.backpressure)
+
+    InstrumentedSource
+      .bufferedSource(
+        original = contractStateEventsSource,
+        counter = metrics.daml.index.contractStateEventsBufferSize,
+        size = outputStreamBufferSize,
+      )
       .concat(endMarker)
   }