[Storage] Refactor indexing transaction error message #8021
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change | ||||||||||||||||
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
|
|
@@ -55,6 +55,7 @@ type TxErrorMessagesEngineSuite struct { | |||||||||||||||||
| reporter *syncmock.IndexReporter | ||||||||||||||||||
| indexReporter *index.Reporter | ||||||||||||||||||
| txResultsIndex *index.TransactionResultsIndex | ||||||||||||||||||
| lockManager storage.LockManager | ||||||||||||||||||
|
|
||||||||||||||||||
| enNodeIDs flow.IdentityList | ||||||||||||||||||
| execClient *accessmock.ExecutionAPIClient | ||||||||||||||||||
|
|
@@ -106,6 +107,9 @@ func (s *TxErrorMessagesEngineSuite) SetupTest() { | |||||||||||||||||
| s.Require().NoError(err) | ||||||||||||||||||
| s.txResultsIndex = index.NewTransactionResultsIndex(s.indexReporter, s.lightTxResults) | ||||||||||||||||||
|
|
||||||||||||||||||
| // Initialize lock manager for tests | ||||||||||||||||||
| s.lockManager = storage.NewTestingLockManager() | ||||||||||||||||||
|
|
||||||||||||||||||
| blockCount := 5 | ||||||||||||||||||
| s.blockMap = make(map[uint64]*flow.Block, blockCount) | ||||||||||||||||||
| s.rootBlock = unittest.Block.Genesis(flow.Emulator) | ||||||||||||||||||
|
|
@@ -177,6 +181,7 @@ func (s *TxErrorMessagesEngineSuite) initEngine(ctx irrecoverable.SignalerContex | |||||||||||||||||
| errorMessageProvider, | ||||||||||||||||||
| s.txErrorMessages, | ||||||||||||||||||
| execNodeIdentitiesProvider, | ||||||||||||||||||
| s.lockManager, | ||||||||||||||||||
| ) | ||||||||||||||||||
|
|
||||||||||||||||||
| eng, err := New( | ||||||||||||||||||
|
|
@@ -245,7 +250,7 @@ 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", mock.Anything, blockID, expectedStoreTxErrorMessages).Return(nil). | ||||||||||||||||||
| Run(func(args mock.Arguments) { | ||||||||||||||||||
| // Ensure the test does not complete its work faster than necessary | ||||||||||||||||||
| wg.Done() | ||||||||||||||||||
|
||||||||||||||||||
| Run(func(args mock.Arguments) { | |
| // Ensure the test does not complete its work faster than necessary | |
| wg.Done() | |
| Run(func(args mock.Arguments) { | |
| 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 |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -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, c.workingData.lockManager), | ||
| stores.NewLatestSealedResultStore(c.workingData.latestPersistedSealedResult, c.executionResult.ID(), c.block.Height), | ||
| } | ||
| blockPersister := persisters.NewBlockPersister( | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -446,9 +446,9 @@ 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, mock.Anything).Return(nil, nil) | ||
| c.persistentResults.On("BatchStore", mock.Anything, mock.Anything, blockID, indexerData.Results).Return(nil) | ||
| c.persistentTxResultErrMsg.On("BatchStore", mock.Anything, 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() | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -60,20 +60,19 @@ func (p *BlockPersister) Persist() error { | |
| p.log.Debug().Msg("started to persist execution data") | ||
| start := time.Now() | ||
|
|
||
| err := storage.WithLocks(p.lockManager, []string{ | ||
| storage.LockInsertCollection, | ||
| storage.LockInsertLightTransactionResult, | ||
|
There was a problem hiding this comment. I don't like this usage here, that the block persister has to know what locks needs to be acquired. The block persister is supposed to be ignorant about what db operation the underlying individual persisters running, therefore doesn't know about the locks to acquire, it supposed to only create a batch object, and ensure it's committed. Maybe we can revisit this, when working on #7910. My idea is that we could let the BatchStore functor to return a required lock ID and the functor, so that the locker doesn't need to remember what lock to acquire. There was a problem hiding this comment. I agree it's slightly awkward, but I think returning the expected lock alongside the functor could cause other problems:
Fundamentally, the layer at which locks are acquired does need to know something about what locks are being acquired. I think just having the upper layer explicitly acquire the needed locks is the simplest way to deal with this pattern. There was a problem hiding this comment.
my 10 cents on this conversation:
|
||
| 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 | ||
| }) | ||
| }) | ||
|
|
||
| if err != nil { | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -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, lockManager), | ||
| stores.NewLatestSealedResultStore(latestPersistedSealedResult, p.executionResult.ID(), p.header.Height), | ||
| }, | ||
| ) | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,6 +1,7 @@ | ||
| package stores | ||
|
|
||
| import ( | ||
| "errors" | ||
| "fmt" | ||
|
|
||
| "github.com/jordanschalm/lockctx" | ||
|
|
@@ -34,8 +35,14 @@ func NewEventsStore( | |
| // | ||
| // No error returns are expected during normal operations | ||
| func (e *EventsStore) Persist(_ lockctx.Proof, batch storage.ReaderBatchWriter) error { | ||
| err := e.persistedEvents.BatchStore(e.blockID, []flow.EventsList{e.data}, batch) | ||
| if err != nil { | ||
| if errors.Is(err, storage.ErrAlreadyExists) { | ||
|
There was a problem hiding this comment. ❓Question I assume this is anticipating future changes? If I am not mistaken, at the moment, In addition, the storage API is lacking error documentation, so added the following todo: Lines 29 to 31 in 9a21aea Overall, I think we need to come back to the events and add overwrite protection checks. After adding these checks, the code in the There was a problem hiding this comment. Yes, the refactor of the implementation has been addressed in this PR: So the error handling is added here first. |
||
| // we don't overwrite, it's ideompotent | ||
|
||
| return nil | ||
| } | ||
| return fmt.Errorf("could not add events to batch: %w", err) | ||
| } | ||
|
|
||
| return nil | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change | ||||||||||||||||||||||||||||||||||
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
|
|
@@ -31,10 +31,10 @@ func NewResultsStore( | |||||||||||||||||||||||||||||||||||
| } | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| // Persist adds results to the batch. | ||||||||||||||||||||||||||||||||||||
| // requires [storage.LockInsertLightTransactionResult] to be hold | ||||||||||||||||||||||||||||||||||||
| // No error returns are expected during normal operations | ||||||||||||||||||||||||||||||||||||
| func (r *ResultsStore) Persist(lctx lockctx.Proof, rw storage.ReaderBatchWriter) error { | ||||||||||||||||||||||||||||||||||||
| if err := r.persistedResults.BatchStore(lctx, rw, r.blockID, r.data); err != nil { | ||||||||||||||||||||||||||||||||||||
| return fmt.Errorf("could not add transaction results to batch: %w", err) | ||||||||||||||||||||||||||||||||||||
|
Comment on lines
37
to
47
There was a problem hiding this comment.
|
||||||||||||||||||||||||||||||||||||
| // No error returns are expected during normal operations | |
| func (r *ResultsStore) Persist(_ lockctx.Proof, batch storage.ReaderBatchWriter) error { | |
| if err := r.persistedResults.BatchStore(r.blockID, r.data, batch); err != nil { | |
| func (r *ResultsStore) Persist(lctx lockctx.Proof, rw storage.ReaderBatchWriter) error { | |
| if err := r.persistedResults.BatchStore(lctx, rw, r.blockID, r.data); err != nil { | |
| return fmt.Errorf("could not add transaction results to batch: %w", err) | |
| // No error returns are expected during normal operations | |
| func (r *ResultsStore) Persist(lctx lockctx.Proof, rw storage.ReaderBatchWriter) error { | |
| // 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. | |
| err := r.persistedResults.BatchStore(lctx, rw, r.blockID, r.data); | |
| if err != nil { | |
| if errors.Is(err, storage.ErrAlreadyExists) { | |
| return nil | |
| } | |
| return fmt.Errorf("could not add transaction results to batch: %w", err) |
Please make sure to keep implementation, interface and all callers in sync (not using different conventions).
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,6 +1,7 @@ | ||
| package stores | ||
|
|
||
| import ( | ||
| "errors" | ||
| "fmt" | ||
|
|
||
| "github.com/jordanschalm/lockctx" | ||
|
|
@@ -16,25 +17,32 @@ type TxResultErrMsgStore struct { | |
| data []flow.TransactionResultErrorMessage | ||
| persistedTxResultErrMsg storage.TransactionResultErrorMessages | ||
| blockID flow.Identifier | ||
| lockManager storage.LockManager | ||
| } | ||
|
|
||
| func NewTxResultErrMsgStore( | ||
| data []flow.TransactionResultErrorMessage, | ||
| persistedTxResultErrMsg storage.TransactionResultErrorMessages, | ||
| blockID flow.Identifier, | ||
| lockManager storage.LockManager, | ||
| ) *TxResultErrMsgStore { | ||
| return &TxResultErrMsgStore{ | ||
| data: data, | ||
| persistedTxResultErrMsg: persistedTxResultErrMsg, | ||
| blockID: blockID, | ||
| lockManager: lockManager, | ||
| } | ||
| } | ||
|
|
||
| // Persist adds transaction result error messages to the batch. | ||
| // | ||
| // No error returns are expected during normal operations | ||
| func (t *TxResultErrMsgStore) Persist(lctx lockctx.Proof, rw storage.ReaderBatchWriter) error { | ||
| err := t.persistedTxResultErrMsg.BatchStore(lctx, rw, t.blockID, t.data) | ||
| if err != nil { | ||
| if errors.Is(err, storage.ErrAlreadyExists) { | ||
| return nil | ||
| } | ||
| return fmt.Errorf("could not add transaction result error messages to batch: %w", err) | ||
| } | ||
| return nil | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -148,17 +148,20 @@ func (c *IndexerCore) IndexBlockData(data *execution_data.BlockExecutionDataEnti | |
| results = append(results, chunk.TransactionResults...) | ||
| } | ||
|
|
||
| err = storage.WithLock(c.lockManager, storage.LockInsertLightTransactionResult, func(lctx lockctx.Context) error { | ||
| return c.protocolDB.WithReaderBatchWriter(func(rw storage.ReaderBatchWriter) error { | ||
| err := c.events.BatchStore(data.BlockID, []flow.EventsList{events}, rw) | ||
| if err != nil { | ||
| return fmt.Errorf("could not index events at height %d: %w", header.Height, err) | ||
| } | ||
|
|
||
| // requires the [storage.LockInsertLightTransactionResult] lock | ||
| err = c.results.BatchStore(lctx, rw, data.BlockID, results) | ||
| if err != nil { | ||
| return fmt.Errorf("could not index transaction results at height %d: %w", header.Height, err) | ||
| } | ||
| return nil | ||
| }) | ||
| }) | ||
|
|
||
| if err != nil { | ||
|
|
||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
should check that the required lock is held here for any
TransactionResultErrorMessages.Store