Add a Time To Seal metric to access node to track time it takes to seal a transaction

cmd/access/node_builder/access_node_builder.go

@@ -150,6 +150,7 @@ type AccessNodeConfig struct {
 	logTxTimeToFinalized                 bool
 	logTxTimeToExecuted                  bool
 	logTxTimeToFinalizedExecuted         bool
+	logTxTimeToSealed                    bool
 	retryEnabled                         bool
 	rpcMetricsEnabled                    bool
 	executionDataSyncEnabled             bool
@@ -241,6 +242,7 @@ func DefaultAccessNodeConfig() *AccessNodeConfig {
 		logTxTimeToFinalized:         false,
 		logTxTimeToExecuted:          false,
 		logTxTimeToFinalizedExecuted: false,
+		logTxTimeToSealed:            false,
 		pingEnabled:                  false,
 		retryEnabled:                 false,
 		rpcMetricsEnabled:            false,
@@ -1234,6 +1236,10 @@ func (builder *FlowAccessNodeBuilder) extraFlags() {
 			"log-tx-time-to-finalized-executed",
 			defaultConfig.logTxTimeToFinalizedExecuted,
 			"log transaction time to finalized and executed")
+		flags.BoolVar(&builder.logTxTimeToSealed,
+			"log-tx-time-to-sealed",
+			defaultConfig.logTxTimeToSealed,
+			"log transaction time to sealed")
 		flags.BoolVar(&builder.pingEnabled,
 			"ping-enabled",
 			defaultConfig.pingEnabled,
@@ -1688,6 +1694,7 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 				builder.logTxTimeToFinalized,
 				builder.logTxTimeToExecuted,
 				builder.logTxTimeToFinalizedExecuted,
+				builder.logTxTimeToSealed,
 			)
 			return nil
 		}).

cmd/observer/node_builder/observer_builder.go

@@ -157,6 +157,7 @@ type ObserverServiceConfig struct {
 	logTxTimeToFinalized                 bool
 	logTxTimeToExecuted                  bool
 	logTxTimeToFinalizedExecuted         bool
+	logTxTimeToSealed                    bool
 	executionDataSyncEnabled             bool
 	executionDataIndexingEnabled         bool
 	executionDataDBMode                  string
@@ -234,6 +235,7 @@ func DefaultObserverServiceConfig() *ObserverServiceConfig {
 		logTxTimeToFinalized:                 false,
 		logTxTimeToExecuted:                  false,
 		logTxTimeToFinalizedExecuted:         false,
+		logTxTimeToSealed:                    false,
 		executionDataSyncEnabled:             false,
 		executionDataIndexingEnabled:         false,
 		executionDataDBMode:                  execution_data.ExecutionDataDBModeBadger.String(),
@@ -678,6 +680,10 @@ func (builder *ObserverServiceBuilder) extraFlags() {
 			"log-tx-time-to-finalized-executed",
 			defaultConfig.logTxTimeToFinalizedExecuted,
 			"log transaction time to finalized and executed")
+		flags.BoolVar(&builder.logTxTimeToSealed,
+			"log-tx-time-to-sealed",
+			defaultConfig.logTxTimeToSealed,
+			"log transaction time to sealed")
 		flags.BoolVar(&builder.rpcMetricsEnabled, "rpc-metrics-enabled", defaultConfig.rpcMetricsEnabled, "whether to enable the rpc metrics")
 		flags.BoolVar(&builder.executionDataIndexingEnabled,
 			"execution-data-indexing-enabled",
@@ -1759,6 +1765,7 @@ func (builder *ObserverServiceBuilder) enqueueRPCServer() {
 			builder.logTxTimeToFinalized,
 			builder.logTxTimeToExecuted,
 			builder.logTxTimeToFinalizedExecuted,
+			builder.logTxTimeToSealed,
 		)
 		return nil
 	})

model/flow/transaction_timing.go

@@ -10,6 +10,7 @@ type TransactionTiming struct {
 	Received      time.Time
 	Finalized     time.Time
 	Executed      time.Time
+	Sealed        time.Time
 }
 
 func (t TransactionTiming) ID() Identifier {

module/metrics.go

@@ -1076,6 +1076,10 @@ type TransactionMetrics interface {
 	// works if the transaction was earlier added as received.
 	TransactionFinalized(txID flow.Identifier, when time.Time)
 
+	// TransactionSealed reports the time spent between the transaction being received and sealed. Reporting only
+	// works if the transaction was earlier added as received.
+	TransactionSealed(txID flow.Identifier, when time.Time)
+
 	// TransactionExecuted reports the time spent between the transaction being received and executed. Reporting only
 	// works if the transaction was earlier added as received.
 	TransactionExecuted(txID flow.Identifier, when time.Time)

module/metrics/noop.go

@@ -214,6 +214,7 @@ func (nc *NoopCollector) ScriptExecutionNotIndexed()
 func (nc *NoopCollector) TransactionResultFetched(dur time.Duration, size int)                  {}
 func (nc *NoopCollector) TransactionReceived(txID flow.Identifier, when time.Time)              {}
 func (nc *NoopCollector) TransactionFinalized(txID flow.Identifier, when time.Time)             {}
+func (nc *NoopCollector) TransactionSealed(txID flow.Identifier, when time.Time)                {}
 func (nc *NoopCollector) TransactionExecuted(txID flow.Identifier, when time.Time)              {}
 func (nc *NoopCollector) TransactionExpired(txID flow.Identifier)                               {}
 func (nc *NoopCollector) TransactionValidated()                                                 {}

module/metrics/transaction.go

@@ -18,9 +18,11 @@ type TransactionCollector struct {
 	logTimeToFinalized             bool
 	logTimeToExecuted              bool
 	logTimeToFinalizedExecuted     bool
+	logTimeToSealed                bool
 	timeToFinalized                prometheus.Summary
 	timeToExecuted                 prometheus.Summary
 	timeToFinalizedExecuted        prometheus.Summary
+	timeToSealed                   prometheus.Summary
 	transactionSubmission          *prometheus.CounterVec
 	transactionSize                prometheus.Histogram
 	scriptExecutedDuration         *prometheus.HistogramVec
@@ -40,6 +42,7 @@ func NewTransactionCollector(
 	logTimeToFinalized bool,
 	logTimeToExecuted bool,
 	logTimeToFinalizedExecuted bool,
+	logTimeToSealed bool,
 ) *TransactionCollector {
 
 	tc := &TransactionCollector{
@@ -48,6 +51,7 @@ func NewTransactionCollector(
 		logTimeToFinalized:         logTimeToFinalized,
 		logTimeToExecuted:          logTimeToExecuted,
 		logTimeToFinalizedExecuted: logTimeToFinalizedExecuted,
+		logTimeToSealed:            logTimeToSealed,
 		timeToFinalized: promauto.NewSummary(prometheus.SummaryOpts{
 			Name:      "time_to_finalized_seconds",
 			Namespace: namespaceAccess,
@@ -91,6 +95,20 @@ func NewTransactionCollector(
 			AgeBuckets: 5,
 			BufCap:     500,
 		}),
+		timeToSealed: promauto.NewSummary(prometheus.SummaryOpts{
+			Name:      "time_to_seal_seconds",
+			Namespace: namespaceAccess,
+			Subsystem: subsystemTransactionTiming,
+			Help:      "the duration of how long it took between the transaction was received until it was sealed",
+			Objectives: map[float64]float64{
+				0.01: 0.001,
+				0.5:  0.05,
+				0.99: 0.001,
+			},
+			MaxAge:     10 * time.Minute,
+			AgeBuckets: 5,
+			BufCap:     500,
+		}),
 		transactionSubmission: promauto.NewCounterVec(prometheus.CounterOpts{
 			Name:      "transaction_submission",
 			Namespace: namespaceAccess,
@@ -269,6 +287,27 @@ func (tc *TransactionCollector) TransactionExecuted(txID flow.Identifier, when t
 	}
 }
 
+func (tc *TransactionCollector) TransactionSealed(txID flow.Identifier, when time.Time) {
+	t, updated := tc.transactionTimings.Adjust(txID, func(t *flow.TransactionTiming) *flow.TransactionTiming {
+		t.Sealed = when
+		return t
+	})
+
+	if !updated {
+		tc.log.Debug().
+			Str("transaction_id", txID.String()).
+			Msg("failed to update TransactionSealed metric")
+		return
+	}
+
+	tc.trackTTS(t, tc.logTimeToSealed)
+
+	// remove transaction timing from mempool if sealed
+	if !t.Sealed.IsZero() {
+		tc.transactionTimings.Remove(txID)
+	}
+}
+
 func (tc *TransactionCollector) trackTTF(t *flow.TransactionTiming, log bool) {
 	if t.Received.IsZero() || t.Finalized.IsZero() {
 		return
@@ -317,6 +356,20 @@ func (tc *TransactionCollector) trackTTFE(t *flow.TransactionTiming, log bool) {
 	}
 }
 
+func (tc *TransactionCollector) trackTTS(t *flow.TransactionTiming, log bool) {
+	if t.Received.IsZero() || t.Sealed.IsZero() {
+		return
+	}
+	duration := t.Sealed.Sub(t.Received).Seconds()
+
+	tc.timeToSealed.Observe(duration)
+
+	if log {
+		tc.log.Info().Str("transaction_id", t.TransactionID.String()).Float64("duration", duration).
+			Msg("transaction time to sealed")
+	}
+}
+
 func (tc *TransactionCollector) TransactionSubmissionFailed() {
 	tc.transactionSubmission.WithLabelValues("failed").Inc()
 }

module/state_synchronization/indexer/collection_executed_metric.go

@@ -2,6 +2,7 @@ package indexer
 
 import (
 	"errors"
+	"sync"
 	"time"
 
 	"github.com/rs/zerolog"
@@ -25,6 +26,9 @@ type CollectionExecutedMetricImpl struct {
 
 	collections storage.Collections
 	blocks      storage.Blocks
+
+	blockTransactions map[flow.Identifier][]flow.Identifier // Map to track transactions for each block for sealed metrics
+	mutex             sync.RWMutex
 }
 
 func NewCollectionExecutedMetricImpl(
@@ -44,6 +48,7 @@ func NewCollectionExecutedMetricImpl(
 		blocksToMarkExecuted:       blocksToMarkExecuted,
 		collections:                collections,
 		blocks:                     blocks,
+		blockTransactions:          make(map[flow.Identifier][]flow.Identifier),
 	}, nil
 }
 
@@ -86,9 +91,33 @@ func (c *CollectionExecutedMetricImpl) BlockFinalized(block *flow.Block) {
 			continue
 		}
 
+		c.mutex.Lock()
 		for _, t := range l.Transactions {
+			c.blockTransactions[blockID] = append(c.blockTransactions[blockID], t)
 			c.accessMetrics.TransactionFinalized(t, now)
 		}
+		c.mutex.Unlock()
+	}
+
+	// Process block seals
+	for _, s := range block.Payload.Seals {
+		c.mutex.RLock()
+		transactions, found := c.blockTransactions[s.BlockID]
+		c.mutex.RUnlock() // release the read lock after reading
+
+		if found && len(transactions) != 0 {
+			// Lock for writing since we will modify the transactions list
+			c.mutex.Lock()
+			// Process all transactions
+			for _, t := range transactions {
+				c.accessMetrics.TransactionSealed(t, now)
+			}
+
+			// Remove the entire block of transactions once processed
+			delete(c.blockTransactions, s.BlockID)
+			// Unlock after modifications are done
+			c.mutex.Unlock()
+		}
 	}
 
 	if ti, found := c.blocksToMarkExecuted.ByID(blockID); found {