[Storage] Refactor indexing transaction error message

admin/commands/storage/backfill_tx_error_messages_test.go

@@ -77,6 +77,7 @@ func TestBackfillTxErrorMessages(t *testing.T) {
 func (suite *BackfillTxErrorMessagesSuite) SetupTest() {
 	suite.log = zerolog.New(os.Stderr)
 
+	lockManager := storage.NewTestingLockManager()
 	suite.state = new(protocolmock.State)
 	suite.headers = new(storagemock.Headers)
 	suite.receipts = new(storagemock.ExecutionReceipts)
@@ -160,6 +161,7 @@ func (suite *BackfillTxErrorMessagesSuite) SetupTest() {
 		errorMessageProvider,
 		suite.txErrorMessages,
 		executionNodeIdentitiesProvider,
+		lockManager,
 	)
 
 	suite.command = NewBackfillTxErrorMessagesCommand(
@@ -532,7 +534,7 @@ func (suite *BackfillTxErrorMessagesSuite) mockStoreTxErrorMessages(
 		}
 	}
 
-	suite.txErrorMessages.On("Store", blockID, txErrorMessages).Return(nil).Once()
+	suite.txErrorMessages.On("Store", mock.Anything, blockID, txErrorMessages).Return(nil).Once()
 }
 
 // assertAllExpectations asserts that all the expectations set on various mocks are met,

cmd/access/node_builder/access_node_builder.go

@@ -2217,6 +2217,7 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 					notNil(builder.txResultErrorMessageProvider),
 					builder.transactionResultErrorMessages,
 					notNil(builder.ExecNodeIdentitiesProvider),
+					node.StorageLockMgr,
 				)
 			}
 

engine/access/ingestion/tx_error_messages/tx_error_messages_core.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 
+	"github.com/jordanschalm/lockctx"
 	"github.com/rs/zerolog"
 
 	"github.com/onflow/flow-go/engine/access/rpc/backend/transactions/error_messages"
@@ -24,6 +25,7 @@ type TxErrorMessagesCore struct {
 	txErrorMessageProvider         error_messages.Provider
 	transactionResultErrorMessages storage.TransactionResultErrorMessages
 	execNodeIdentitiesProvider     *commonrpc.ExecutionNodeIdentitiesProvider
+	lockManager                    storage.LockManager
 }
 
 // NewTxErrorMessagesCore creates a new instance of TxErrorMessagesCore.
@@ -32,12 +34,14 @@ func NewTxErrorMessagesCore(
 	txErrorMessageProvider error_messages.Provider,
 	transactionResultErrorMessages storage.TransactionResultErrorMessages,
 	execNodeIdentitiesProvider *commonrpc.ExecutionNodeIdentitiesProvider,
+	lockManager storage.LockManager,
 ) *TxErrorMessagesCore {
 	return &TxErrorMessagesCore{
 		log:                            log.With().Str("module", "tx_error_messages_core").Logger(),
 		txErrorMessageProvider:         txErrorMessageProvider,
 		transactionResultErrorMessages: transactionResultErrorMessages,
 		execNodeIdentitiesProvider:     execNodeIdentitiesProvider,
+		lockManager:                    lockManager,
 	}
 }
 
@@ -62,6 +66,15 @@ func (c *TxErrorMessagesCore) FetchErrorMessages(ctx context.Context, blockID fl
 	return c.FetchErrorMessagesByENs(ctx, blockID, execNodes)
 }
 
+// FetchErrorMessagesByENs requests the transaction result error messages for the specified block ID from
+// any of the given execution nodes and persists them once retrieved. This function blocks until ingesting
+// the tx error messages is completed or failed.
+//
+// Note that transaction error messages are auxiliary data provided by the Execution Nodes on a goodwill basis and
+// not protected by the protocol. Execution Error messages might be non-deterministic, i.e. potentially different
+// for different execution nodes. Hence, we also persist which execution node (`execNode) provided the error message.
+//
+// It returns [storage.ErrAlreadyExists] if tx result error messages for the block already exist.
 func (c *TxErrorMessagesCore) FetchErrorMessagesByENs(
 	ctx context.Context,
 	blockID flow.Identifier,
@@ -71,7 +84,6 @@ func (c *TxErrorMessagesCore) FetchErrorMessagesByENs(
 	if err != nil {
 		return fmt.Errorf("could not check existance of transaction result error messages: %w", err)
 	}
-
 	if exists {
 		return nil
 	}
@@ -100,14 +112,10 @@ func (c *TxErrorMessagesCore) FetchErrorMessagesByENs(
 	return nil
 }
 
-// storeTransactionResultErrorMessages stores the transaction result error messages for a given block ID.
+// storeTransactionResultErrorMessages persists and indexes all transaction result error messages for the given blockID.
+// The caller must acquire [storage.LockInsertTransactionResultErrMessage] and hold it until the write batch has been committed.
 //
-// Parameters:
-// - blockID: The identifier of the block for which the error messages are to be stored.
-// - errorMessagesResponses: A slice of responses containing the error messages to be stored.
-// - execNode: The execution node associated with the error messages.
-//
-// No errors are expected during normal operation.
+// It returns [storage.ErrAlreadyExists] if tx result error messages for the block already exist.
 func (c *TxErrorMessagesCore) storeTransactionResultErrorMessages(
 	blockID flow.Identifier,
 	errorMessagesResponses []*execproto.GetTransactionErrorMessagesResponse_Result,
@@ -124,7 +132,9 @@ func (c *TxErrorMessagesCore) storeTransactionResultErrorMessages(
 		errorMessages = append(errorMessages, errorMessage)
 	}
 
-	err := c.transactionResultErrorMessages.Store(blockID, errorMessages)
+	err := storage.WithLock(c.lockManager, storage.LockInsertTransactionResultErrMessage, func(lctx lockctx.Context) error {
+		return c.transactionResultErrorMessages.Store(lctx, blockID, errorMessages)
+	})
 	if err != nil {
 		return fmt.Errorf("failed to store transaction error messages: %w", err)
 	}

engine/access/ingestion/tx_error_messages/tx_error_messages_core_test.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"testing"
 
+	"github.com/jordanschalm/lockctx"
 	execproto "github.com/onflow/flow/protobuf/go/flow/execution"
 	"github.com/rs/zerolog"
 	"github.com/stretchr/testify/mock"
@@ -21,7 +22,8 @@ import (
 	"github.com/onflow/flow-go/module/irrecoverable"
 	syncmock "github.com/onflow/flow-go/module/state_synchronization/mock"
 	protocol "github.com/onflow/flow-go/state/protocol/mock"
-	storage "github.com/onflow/flow-go/storage/mock"
+	"github.com/onflow/flow-go/storage"
+	storagemock "github.com/onflow/flow-go/storage/mock"
 	"github.com/onflow/flow-go/utils/unittest"
 )
 
@@ -37,13 +39,14 @@ type TxErrorMessagesCoreSuite struct {
 		params   *protocol.Params
 	}
 
-	receipts        *storage.ExecutionReceipts
-	txErrorMessages *storage.TransactionResultErrorMessages
-	lightTxResults  *storage.LightTransactionResults
+	receipts        *storagemock.ExecutionReceipts
+	txErrorMessages *storagemock.TransactionResultErrorMessages
+	lightTxResults  *storagemock.LightTransactionResults
 
 	reporter       *syncmock.IndexReporter
 	indexReporter  *index.Reporter
 	txResultsIndex *index.TransactionResultsIndex
+	lockManager    storage.LockManager
 
 	enNodeIDs   flow.IdentityList
 	execClient  *accessmock.ExecutionAPIClient
@@ -79,18 +82,21 @@ func (s *TxErrorMessagesCoreSuite) SetupTest() {
 	s.proto.params = protocol.NewParams(s.T())
 	s.execClient = accessmock.NewExecutionAPIClient(s.T())
 	s.connFactory = connectionmock.NewConnectionFactory(s.T())
-	s.receipts = storage.NewExecutionReceipts(s.T())
-	s.txErrorMessages = storage.NewTransactionResultErrorMessages(s.T())
+	s.receipts = storagemock.NewExecutionReceipts(s.T())
+	s.txErrorMessages = storagemock.NewTransactionResultErrorMessages(s.T())
 	s.rootBlock = unittest.Block.Genesis(flow.Emulator)
 	s.finalizedBlock = unittest.BlockWithParentFixture(s.rootBlock.ToHeader()).ToHeader()
 
-	s.lightTxResults = storage.NewLightTransactionResults(s.T())
+	s.lightTxResults = storagemock.NewLightTransactionResults(s.T())
 	s.reporter = syncmock.NewIndexReporter(s.T())
 	s.indexReporter = index.NewReporter()
 	err := s.indexReporter.Initialize(s.reporter)
 	s.Require().NoError(err)
 	s.txResultsIndex = index.NewTransactionResultsIndex(s.indexReporter, s.lightTxResults)
 
+	// Initialize lock manager for tests
+	s.lockManager = storage.NewTestingLockManager()
+
 	s.proto.state.On("Params").Return(s.proto.params)
 
 	// Mock the finalized root block header with height 0.
@@ -143,8 +149,11 @@ func (s *TxErrorMessagesCoreSuite) TestHandleTransactionResultErrorMessages() {
 	expectedStoreTxErrorMessages := createExpectedTxErrorMessages(resultsByBlockID, s.enNodeIDs.NodeIDs()[0])
 
 	// Mock the storage of the fetched error messages into the protocol database.
-	s.txErrorMessages.On("Store", blockId, expectedStoreTxErrorMessages).
-		Return(nil).Once()
+	s.txErrorMessages.On("Store", mock.Anything, blockId, expectedStoreTxErrorMessages).
+		Return(func(lctx lockctx.Proof, blockID flow.Identifier, transactionResultErrorMessages []flow.TransactionResultErrorMessage) error {
+			require.True(s.T(), lctx.HoldsLock(storage.LockInsertTransactionResultErrMessage))
+			return nil
+		}).Once()
 
 	core := s.initCore()
 	err := core.FetchErrorMessages(irrecoverableCtx, blockId)
@@ -228,8 +237,11 @@ func (s *TxErrorMessagesCoreSuite) TestHandleTransactionResultErrorMessages_Erro
 
 		// Simulate an error when attempting to store the fetched transaction error messages in storage.
 		expectedStoreTxErrorMessages := createExpectedTxErrorMessages(resultsByBlockID, s.enNodeIDs.NodeIDs()[0])
-		s.txErrorMessages.On("Store", blockId, expectedStoreTxErrorMessages).
-			Return(fmt.Errorf("storage error")).Once()
+		s.txErrorMessages.On("Store", mock.Anything, blockId, expectedStoreTxErrorMessages).
+			Return(func(lctx lockctx.Proof, blockID flow.Identifier, transactionResultErrorMessages []flow.TransactionResultErrorMessage) error {
+				require.True(s.T(), lctx.HoldsLock(storage.LockInsertTransactionResultErrMessage))
+				return fmt.Errorf("storage error")
+			}).Once()
 
 		core := s.initCore()
 		err := core.FetchErrorMessages(irrecoverableCtx, blockId)
@@ -268,6 +280,7 @@ func (s *TxErrorMessagesCoreSuite) initCore() *TxErrorMessagesCore {
 		errorMessageProvider,
 		s.txErrorMessages,
 		execNodeIdentitiesProvider,
+		s.lockManager,
 	)
 	return core
 }
@@ -311,7 +324,7 @@ func mockTransactionResultsByBlock(count int) []flow.LightTransactionResult {
 
 // setupReceiptsForBlock sets up mock execution receipts for a block and returns the receipts along
 // with the identities of the execution nodes that processed them.
-func setupReceiptsForBlock(receipts *storage.ExecutionReceipts, block *flow.Block, eNodeID flow.Identifier) {
+func setupReceiptsForBlock(receipts *storagemock.ExecutionReceipts, block *flow.Block, eNodeID flow.Identifier) {
 	receipt1 := unittest.ReceiptForBlockFixture(block)
 	receipt1.ExecutorID = eNodeID
 	receipt2 := unittest.ReceiptForBlockFixture(block)
@@ -328,7 +341,7 @@ func setupReceiptsForBlock(receipts *storage.ExecutionReceipts, block *flow.Bloc
 }
 
 // setupReceiptsForBlockWithResult sets up mock execution receipts for a block with a specific execution result
-func setupReceiptsForBlockWithResult(receipts *storage.ExecutionReceipts, executionResult *flow.ExecutionResult, executorIDs ...flow.Identifier) {
+func setupReceiptsForBlockWithResult(receipts *storagemock.ExecutionReceipts, executionResult *flow.ExecutionResult, executorIDs ...flow.Identifier) {
 	receiptList := make(flow.ExecutionReceiptList, 0, len(executorIDs))
 	for _, enID := range executorIDs {
 		receiptList = append(receiptList, unittest.ExecutionReceiptFixture(

engine/access/ingestion/tx_error_messages/tx_error_messages_engine_test.go

@@ -7,6 +7,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/jordanschalm/lockctx"
 	execproto "github.com/onflow/flow/protobuf/go/flow/execution"
 	"github.com/rs/zerolog"
 	"github.com/stretchr/testify/mock"
@@ -55,6 +56,7 @@ type TxErrorMessagesEngineSuite struct {
 	reporter       *syncmock.IndexReporter
 	indexReporter  *index.Reporter
 	txResultsIndex *index.TransactionResultsIndex
+	lockManager    storage.LockManager
 
 	enNodeIDs   flow.IdentityList
 	execClient  *accessmock.ExecutionAPIClient
@@ -87,9 +89,13 @@ func (s *TxErrorMessagesEngineSuite) SetupTest() {
 	s.log = unittest.Logger()
 	s.metrics = metrics.NewNoopCollector()
 	s.ctx, s.cancel = context.WithCancel(context.Background())
+
+	// Initialize database and lock manager
 	pdb, dbDir := unittest.TempPebbleDB(s.T())
 	s.db = pebbleimpl.ToDB(pdb)
 	s.dbDir = dbDir
+	s.lockManager = storage.NewTestingLockManager()
+
 	// mock out protocol state
 	s.proto.state = protocol.NewFollowerState(s.T())
 	s.proto.snapshot = protocol.NewSnapshot(s.T())
@@ -177,6 +183,7 @@ func (s *TxErrorMessagesEngineSuite) initEngine(ctx irrecoverable.SignalerContex
 		errorMessageProvider,
 		s.txErrorMessages,
 		execNodeIdentitiesProvider,
+		s.lockManager,
 	)
 
 	eng, err := New(
@@ -245,10 +252,12 @@ func (s *TxErrorMessagesEngineSuite) TestOnFinalizedBlockHandleTxErrorMessages()
 		expectedStoreTxErrorMessages := createExpectedTxErrorMessages(resultsByBlockID, s.enNodeIDs.NodeIDs()[0])
 
 		// Mock the storage of the fetched error messages into the protocol database.
-		s.txErrorMessages.On("Store", blockID, expectedStoreTxErrorMessages).Return(nil).
+		s.txErrorMessages.On("Store", mock.Anything, blockID, expectedStoreTxErrorMessages).Return(nil).
 			Run(func(args mock.Arguments) {
-				// Ensure the test does not complete its work faster than necessary
-				wg.Done()
+				lctx, ok := args[0].(lockctx.Proof)
+				require.True(s.T(), ok, "expecting lock proof, but cast failed")
+				require.True(s.T(), lctx.HoldsLock(storage.LockInsertTransactionResultErrMessage))
+				wg.Done() // Ensure the test does not complete its work faster than necessary
 			}).Once()
 	}
 

module/executiondatasync/optimistic_sync/core.go

@@ -335,7 +335,7 @@ func (c *CoreImpl) Persist() error {
 		stores.NewEventsStore(indexerData.Events, c.workingData.persistentEvents, c.executionResult.BlockID),
 		stores.NewResultsStore(indexerData.Results, c.workingData.persistentResults, c.executionResult.BlockID),
 		stores.NewCollectionsStore(indexerData.Collections, c.workingData.persistentCollections),
-		stores.NewTxResultErrMsgStore(c.workingData.txResultErrMsgsData, c.workingData.persistentTxResultErrMsgs, c.executionResult.BlockID),
+		stores.NewTxResultErrMsgStore(c.workingData.txResultErrMsgsData, c.workingData.persistentTxResultErrMsgs, c.executionResult.BlockID, c.workingData.lockManager),
 		stores.NewLatestSealedResultStore(c.workingData.latestPersistedSealedResult, c.executionResult.ID(), c.block.Height),
 	}
 	blockPersister := persisters.NewBlockPersister(

module/executiondatasync/optimistic_sync/core_impl_test.go

@@ -6,6 +6,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/jordanschalm/lockctx"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/suite"
 
@@ -446,9 +447,21 @@ func (c *CoreImplSuite) TestCoreImpl_Persist() {
 		indexerData := core.workingData.indexerData
 		c.persistentRegisters.On("Store", flow.RegisterEntries(indexerData.Registers), tf.block.Height).Return(nil)
 		c.persistentEvents.On("BatchStore", blockID, []flow.EventsList{indexerData.Events}, mock.Anything).Return(nil)
-		c.persistentCollections.On("BatchStoreAndIndexByTransaction", mock.Anything, mock.Anything, mock.Anything).Return(nil, nil)
-		c.persistentResults.On("BatchStore", blockID, indexerData.Results, mock.Anything).Return(nil)
-		c.persistentTxResultErrMsg.On("BatchStore", blockID, core.workingData.txResultErrMsgsData, mock.Anything).Return(nil)
+		c.persistentCollections.On("BatchStoreAndIndexByTransaction",
+			mock.MatchedBy(func(lctx lockctx.Proof) bool {
+				return lctx.HoldsLock(storage.LockInsertCollection)
+			}),
+			mock.Anything, mock.Anything, mock.Anything).Return(nil, nil)
+		c.persistentResults.On("BatchStore",
+			mock.MatchedBy(func(lctx lockctx.Proof) bool {
+				return lctx.HoldsLock(storage.LockInsertLightTransactionResult)
+			}),
+			mock.Anything, blockID, indexerData.Results).Return(nil)
+		c.persistentTxResultErrMsg.On("BatchStore",
+			mock.MatchedBy(func(lctx lockctx.Proof) bool {
+				return lctx.HoldsLock(storage.LockInsertTransactionResultErrMessage)
+			}),
+			mock.Anything, blockID, core.workingData.txResultErrMsgsData).Return(nil)
 		c.latestPersistedSealedResult.On("BatchSet", tf.exeResult.ID(), tf.block.Height, mock.Anything).Return(nil)
 
 		err = core.Persist()

module/executiondatasync/optimistic_sync/persisters/block.go

@@ -60,20 +60,19 @@ func (p *BlockPersister) Persist() error {
 	p.log.Debug().Msg("started to persist execution data")
 	start := time.Now()
 
-	lctx := p.lockManager.NewContext()
-	err := lctx.AcquireLock(storage.LockInsertCollection)
-	if err != nil {
-		return fmt.Errorf("could not acquire lock for inserting light collections: %w", err)
-	}
-	defer lctx.Release()
-
-	err = p.protocolDB.WithReaderBatchWriter(func(batch storage.ReaderBatchWriter) error {
-		for _, persister := range p.persisterStores {
-			if err := persister.Persist(lctx, batch); err != nil {
-				return err
+	err := storage.WithLocks(p.lockManager, []string{
+		storage.LockInsertCollection,
+		storage.LockInsertLightTransactionResult,
+		storage.LockInsertTransactionResultErrMessage,
+	}, func(lctx lockctx.Context) error {
+		return p.protocolDB.WithReaderBatchWriter(func(batch storage.ReaderBatchWriter) error {
+			for _, persister := range p.persisterStores {
+				if err := persister.Persist(lctx, batch); err != nil {
+					return err
+				}
 			}
-		}
-		return nil
+			return nil
+		})
 	})
 
 	if err != nil {

module/executiondatasync/optimistic_sync/persisters/block_test.go

@@ -111,7 +111,7 @@ func (p *PersisterSuite) testWithDatabase() {
 				stores.NewEventsStore(p.indexerData.Events, events, p.executionResult.BlockID),
 				stores.NewResultsStore(p.indexerData.Results, results, p.executionResult.BlockID),
 				stores.NewCollectionsStore(p.indexerData.Collections, collections),
-				stores.NewTxResultErrMsgStore(p.txErrMsgs, txResultErrMsg, p.executionResult.BlockID),
+				stores.NewTxResultErrMsgStore(p.txErrMsgs, txResultErrMsg, p.executionResult.BlockID, lockManager),
 				stores.NewLatestSealedResultStore(latestPersistedSealedResult, p.executionResult.ID(), p.header.Height),
 			},
 		)

module/executiondatasync/optimistic_sync/persisters/stores/events.go

@@ -1,6 +1,7 @@
 package stores
 
 import (
+	"errors"
 	"fmt"
 
 	"github.com/jordanschalm/lockctx"
@@ -34,8 +35,16 @@ func NewEventsStore(
 //
 // No error returns are expected during normal operations
 func (e *EventsStore) Persist(_ lockctx.Proof, batch storage.ReaderBatchWriter) error {
-	if err := e.persistedEvents.BatchStore(e.blockID, []flow.EventsList{e.data}, batch); err != nil {
+	err := e.persistedEvents.BatchStore(e.blockID, []flow.EventsList{e.data}, batch)
+	if err != nil {
+		if errors.Is(err, storage.ErrAlreadyExists) {
+			// CAUTION: here we assume that if something is already stored for our blockID, then the data is identical.
+			// This only holds true for sealed execution results, whose consistency has previously been verified by
+			// comparing the data's hash to commitments in the execution result.
+			return nil
+		}
 		return fmt.Errorf("could not add events to batch: %w", err)
 	}
+
 	return nil
 }