Merge pull request #8165 from onflow/peter/fix-access-reindex

[Access] Allow reindexing last block's protocol data

Author: peterargue

module/state_synchronization/indexer/indexer_core.go

@@ -145,8 +145,9 @@ func (c *IndexerCore) IndexBlockData(data *execution_data.BlockExecutionDataEnti
 		return fmt.Errorf("must index block data with the next height %d, but got %d", latest+1, header.Height)
 	}
 
-	// allow rerunning the indexer for same height since we are fetching height from register storage, but there are other storages
-	// for indexing resources which might fail to update the values, so this enables rerunning and reindexing those resources
+	// Data for the block is stored into both the protocol and registers databases. This creates a
+	// race condition where it's possible only one completes if the node crashes at an inopportune time.
+	// In this case, allow reindexing the last block. Both databases should treat this as a no-op.
 	if header.Height == latest {
 		lg.Warn().Msg("reindexing block data")
 		c.metrics.BlockReindexed()
@@ -183,7 +184,6 @@ func (c *IndexerCore) IndexBlockData(data *execution_data.BlockExecutionDataEnti
 						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)
@@ -201,6 +201,10 @@ func (c *IndexerCore) IndexBlockData(data *execution_data.BlockExecutionDataEnti
 			})
 
 		if err != nil {
+			if errors.Is(err, storage.ErrAlreadyExists) {
+				// Since reindexing is a no-op, return early without an error
+				return nil
+			}
 			return fmt.Errorf("could not commit block data: %w", err)
 		}
 

module/state_synchronization/indexer/indexer_core_test.go

@@ -10,7 +10,6 @@ import (
 	"github.com/rs/zerolog"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
-	mocks "github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 
 	collectionsmock "github.com/onflow/flow-go/engine/access/ingestion/collections/mock"
@@ -31,6 +30,7 @@ import (
 	pebbleStorage "github.com/onflow/flow-go/storage/pebble"
 	"github.com/onflow/flow-go/utils/unittest"
 	"github.com/onflow/flow-go/utils/unittest/fixtures"
+	"github.com/onflow/flow-go/utils/unittest/mocks"
 )
 
 type indexCoreTest struct {
@@ -80,7 +80,7 @@ func newIndexCoreTest(
 
 func (i *indexCoreTest) useDefaultBlockByHeight() *indexCoreTest {
 	i.headers.
-		On("BlockIDByHeight", mocks.AnythingOfType("uint64")).
+		On("BlockIDByHeight", mock.AnythingOfType("uint64")).
 		Return(func(height uint64) (flow.Identifier, error) {
 			for _, b := range i.blocks {
 				if b.Height == height {
@@ -91,7 +91,7 @@ func (i *indexCoreTest) useDefaultBlockByHeight() *indexCoreTest {
 		})
 
 	i.headers.
-		On("ByHeight", mocks.AnythingOfType("uint64")).
+		On("ByHeight", mock.AnythingOfType("uint64")).
 		Return(func(height uint64) (*flow.Header, error) {
 			for _, b := range i.blocks {
 				if b.Height == height {
@@ -268,17 +268,15 @@ func TestExecutionState_IndexBlockData(t *testing.T) {
 	t.Run("Index AllTheThings", func(t *testing.T) {
 		test := newIndexCoreTest(t, g, blocks, tf.ExecutionDataEntity()).initIndexer()
 
-		test.events.On("BatchStore",
-			mock.MatchedBy(func(lctx lockctx.Proof) bool { return lctx.HoldsLock(storage.LockInsertEvent) }),
-			blockID, []flow.EventsList{tf.ExpectedEvents}, mock.Anything).
+		test.events.
+			On("BatchStore", mocks.MatchLock(storage.LockInsertEvent), blockID, []flow.EventsList{tf.ExpectedEvents}, mock.Anything).
 			Return(func(lctx lockctx.Proof, blockID flow.Identifier, events []flow.EventsList, batch storage.ReaderBatchWriter) error {
 				require.True(t, lctx.HoldsLock(storage.LockInsertEvent))
 				require.NotNil(t, batch)
 				return nil
 			})
-		test.results.On("BatchStore",
-			mock.MatchedBy(func(lctx lockctx.Proof) bool { return lctx.HoldsLock(storage.LockInsertLightTransactionResult) }),
-			mock.Anything, blockID, tf.ExpectedResults).
+		test.results.
+			On("BatchStore", mocks.MatchLock(storage.LockInsertLightTransactionResult), mock.Anything, blockID, tf.ExpectedResults).
 			Return(func(lctx lockctx.Proof, batch storage.ReaderBatchWriter, blockID flow.Identifier, results []flow.LightTransactionResult) error {
 				require.True(t, lctx.HoldsLock(storage.LockInsertLightTransactionResult))
 				require.NotNil(t, batch)
@@ -294,9 +292,8 @@ func TestExecutionState_IndexBlockData(t *testing.T) {
 			Return(nil)
 		test.collectionIndexer.On("IndexCollections", tf.ExpectedCollections).Return(nil).Once()
 		for txID, scheduledTxID := range tf.ExpectedScheduledTransactions {
-			test.scheduledTransactions.On("BatchIndex",
-				mock.MatchedBy(func(lctx lockctx.Proof) bool { return lctx.HoldsLock(storage.LockIndexScheduledTransaction) }),
-				blockID, txID, scheduledTxID, mock.Anything).
+			test.scheduledTransactions.
+				On("BatchIndex", mocks.MatchLock(storage.LockIndexScheduledTransaction), blockID, txID, scheduledTxID, mock.Anything).
 				Return(func(lctx lockctx.Proof, blockID flow.Identifier, txID flow.Identifier, scheduledTxID uint64, batch storage.ReaderBatchWriter) error {
 					require.True(t, lctx.HoldsLock(storage.LockIndexScheduledTransaction))
 					require.NotNil(t, batch)
@@ -343,6 +340,39 @@ func TestExecutionState_IndexBlockData(t *testing.T) {
 		assert.ErrorIs(t, err, storage.ErrNotFound)
 	})
 
+	// test that reindexing the last block does not return an error and does not write any data
+	t.Run("Reindexing last block", func(t *testing.T) {
+		test := newIndexCoreTest(t, g, blocks, tf.ExecutionDataEntity()).
+			initIndexer().
+			setLastHeight(func(t *testing.T) uint64 {
+				return tf.Block.Height
+			})
+
+		// reset all mocks to avoid false positives
+		test.events.On("BatchStore", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Unset()
+		test.results.On("BatchStore", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Unset()
+		test.scheduledTransactions.On("BatchIndex", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Unset()
+		test.registers.On("Store", mock.Anything, mock.Anything).Unset()
+		test.collectionIndexer.On("IndexCollections", mock.Anything).Unset()
+
+		// setup mocks to behave as they would if the block was already indexed.
+		// tx results and scheduled transactions will not be called since events returned an error.
+		test.events.
+			On("BatchStore", mocks.MatchLock(storage.LockInsertEvent), blockID, []flow.EventsList{tf.ExpectedEvents}, mock.Anything).
+			Return(storage.ErrAlreadyExists).
+			Once()
+		test.collectionIndexer.
+			On("IndexCollections", tf.ExpectedCollections).
+			Return(nil).
+			Once()
+		test.registers.
+			On("Store", mock.Anything, tf.Block.Height).
+			Return(nil).
+			Once()
+
+		err := test.indexer.IndexBlockData(tf.ExecutionDataEntity())
+		assert.NoError(t, err)
+	})
 }
 
 func TestExecutionState_RegisterValues(t *testing.T) {

storage/light_transaction_results.go

@@ -22,6 +22,7 @@ type LightTransactionResultsReader interface {
 
 	// ByBlockID gets all transaction results for a block, ordered by transaction index
 	// CAUTION: this function returns the empty list in case for block IDs without known results.
+	//
 	// No error returns are expected during normal operations.
 	ByBlockID(id flow.Identifier) ([]flow.LightTransactionResult, error)
 }
@@ -33,6 +34,8 @@ type LightTransactionResults interface {
 	// BatchStore persists and indexes all transaction results (light representation) for the given blockID
 	// as part of the provided batch. The caller must acquire [storage.LockInsertLightTransactionResult] and
 	// hold it until the write batch has been committed.
-	// It returns [storage.ErrAlreadyExists] if light transaction results for the block already exist.
+	//
+	// Expected error returns during normal operation:
+	//   - [storage.ErrAlreadyExists] if light transaction results for the block already exist.
 	BatchStore(lctx lockctx.Proof, rw ReaderBatchWriter, blockID flow.Identifier, transactionResults []flow.LightTransactionResult) error
 }

utils/unittest/mocks/matchers.go

@@ -0,0 +1,22 @@
+package mocks
+
+import (
+	"github.com/jordanschalm/lockctx"
+	"github.com/stretchr/testify/mock"
+)
+
+// MatchLock returns an argument matcher that checks if the argument is a `lockctx.Proof` that holds
+// the provided lock.
+//
+// Example:
+//
+//	events.
+//		On("BatchStore", mocks.MatchLock(storage.LockInsertEvent), blockID, expectedEvents, mock.Anything).
+//		Return(func(lctx lockctx.Proof, blockID flow.Identifier, events []flow.EventsList, batch storage.ReaderBatchWriter) error {
+//			require.NotNil(t, batch)
+//			return nil
+//		}).
+//		Once()
+func MatchLock(lock string) interface{} {
+	return mock.MatchedBy(func(lctx lockctx.Proof) bool { return lctx.HoldsLock(lock) })
+}