Merge pull request #119 from stellar/fix-events-metric

In the scan operation there were some cases where we were not recording metrics properly. Specifically, if a scan terminated early without any errors we did not call m.eventsDurationMetric.With(prometheus.Labels{"operation": "scan"}).Observe(time.Since(startTime).Seconds())

cmd/soroban-rpc/internal/events/events.go

@@ -99,18 +99,25 @@ type ScanFunction func(xdr.DiagnosticEvent, Cursor, int64, *xdr.Hash) bool
 // remaining events in the range). Note that a read lock is held for the
 // entire duration of the Scan function so f should be written in a way
 // to minimize latency.
-func (m *MemoryStore) Scan(eventRange Range, f ScanFunction) (uint32, error) {
+func (m *MemoryStore) Scan(eventRange Range, f ScanFunction) (lastLedgerInWindow uint32, err error) {
 	startTime := time.Now()
+	defer func() {
+		if err == nil {
+			m.eventsDurationMetric.With(prometheus.Labels{"operation": "scan"}).
+				Observe(time.Since(startTime).Seconds())
+		}
+	}()
+
 	m.lock.RLock()
 	defer m.lock.RUnlock()
 
-	if err := m.validateRange(&eventRange); err != nil {
-		return 0, err
+	if err = m.validateRange(&eventRange); err != nil {
+		return
 	}
 
 	firstLedgerInRange := eventRange.Start.Ledger
 	firstLedgerInWindow := m.eventsByLedger.Get(0).LedgerSeq
-	lastLedgerInWindow := firstLedgerInWindow + (m.eventsByLedger.Len() - 1)
+	lastLedgerInWindow = firstLedgerInWindow + (m.eventsByLedger.Len() - 1)
 	for i := firstLedgerInRange - firstLedgerInWindow; i < m.eventsByLedger.Len(); i++ {
 		bucket := m.eventsByLedger.Get(i)
 		events := bucket.BucketContent
@@ -122,21 +129,19 @@ func (m *MemoryStore) Scan(eventRange Range, f ScanFunction) (uint32, error) {
 		for _, event := range events {
 			cur := event.cursor(bucket.LedgerSeq)
 			if eventRange.End.Cmp(cur) <= 0 {
-				return lastLedgerInWindow, nil
+				return
 			}
 			var diagnosticEvent xdr.DiagnosticEvent
-			err := xdr.SafeUnmarshal(event.diagnosticEventXDR, &diagnosticEvent)
+			err = xdr.SafeUnmarshal(event.diagnosticEventXDR, &diagnosticEvent)
 			if err != nil {
-				return 0, err
+				return
 			}
 			if !f(diagnosticEvent, cur, timestamp, event.txHash) {
-				return lastLedgerInWindow, nil
+				return
 			}
 		}
 	}
-	m.eventsDurationMetric.With(prometheus.Labels{"operation": "scan"}).
-		Observe(time.Since(startTime).Seconds())
-	return lastLedgerInWindow, nil
+	return
 }
 
 // validateRange checks if the range falls within the bounds

cmd/soroban-rpc/internal/events/events_test.go

@@ -4,6 +4,8 @@ import (
 	"bytes"
 	"testing"
 
+	"github.com/prometheus/client_golang/prometheus"
+	dto "github.com/prometheus/client_model/go"
 	"github.com/stellar/go/xdr"
 	"github.com/stretchr/testify/require"
 
@@ -240,9 +242,16 @@ func concat(slices ...[]event) []event {
 	return result
 }
 
-func TestScan(t *testing.T) {
-	m := createStore(t)
+func getMetricValue(metric prometheus.Metric) *dto.Metric {
+	value := &dto.Metric{}
+	err := metric.Write(value)
+	if err != nil {
+		panic(err)
+	}
+	return value
+}
 
+func TestScan(t *testing.T) {
 	genEquivalentInputs := func(input Range) []Range {
 		results := []Range{input}
 		if !input.ClampStart {
@@ -360,6 +369,7 @@ func TestScan(t *testing.T) {
 		},
 	} {
 		for _, input := range genEquivalentInputs(testCase.input) {
+			m := createStore(t)
 			var events []event
 			iterateAll := true
 			f := func(contractEvent xdr.DiagnosticEvent, cursor Cursor, ledgerCloseTimestamp int64, hash *xdr.Hash) bool {
@@ -378,11 +388,17 @@ func TestScan(t *testing.T) {
 			require.NoError(t, err)
 			require.Equal(t, uint32(8), latest)
 			eventsAreEqual(t, testCase.expected, events)
+			metric, err := m.eventsDurationMetric.MetricVec.GetMetricWith(prometheus.Labels{
+				"operation": "scan",
+			})
+			require.NoError(t, err)
+			require.Equal(t, uint64(1), getMetricValue(metric).GetSummary().GetSampleCount())
 			if len(events) > 0 {
 				events = nil
 				iterateAll = false
 				latest, err := m.Scan(input, f)
 				require.NoError(t, err)
+				require.Equal(t, uint64(2), getMetricValue(metric).GetSummary().GetSampleCount())
 				require.Equal(t, uint32(8), latest)
 				eventsAreEqual(t, []event{testCase.expected[0]}, events)
 			}