Support batched transactions

go.mod

@@ -7,7 +7,7 @@ require (
 	github.com/goccy/go-json v0.10.2
 	github.com/hashicorp/golang-lru/v2 v2.0.7
 	github.com/onflow/cadence v1.0.0-preview.23
-	github.com/onflow/flow-go v0.34.0-crescendo-preview.16
+	github.com/onflow/flow-go v0.34.0-crescendo-preview.17
 	github.com/onflow/flow-go-sdk v1.0.0-preview.22
 	github.com/onflow/go-ethereum v1.13.4
 	github.com/rs/cors v1.8.0

go.sum

@@ -1942,8 +1942,8 @@ github.com/onflow/flow-ft/lib/go/contracts v0.7.1-0.20240424211859-3ff4c0fe2a1e
 github.com/onflow/flow-ft/lib/go/contracts v0.7.1-0.20240424211859-3ff4c0fe2a1e/go.mod h1:PwsL8fC81cjnUnTfmyL/HOIyHnyaw/JA474Wfj2tl6A=
 github.com/onflow/flow-ft/lib/go/templates v0.7.1-0.20240424211859-3ff4c0fe2a1e h1:jl7SYAui/gYRmBofrY//Ln8ixRJwvLzvwLstNfRKmWY=
 github.com/onflow/flow-ft/lib/go/templates v0.7.1-0.20240424211859-3ff4c0fe2a1e/go.mod h1:uQ8XFqmMK2jxyBSVrmyuwdWjTEb+6zGjRYotfDJ5pAE=
-github.com/onflow/flow-go v0.34.0-crescendo-preview.16 h1:GTOLB2FTG4En91WAd51ReNM5zGJ56KJY9k7AN/ZsCMw=
-github.com/onflow/flow-go v0.34.0-crescendo-preview.16/go.mod h1:WbsDXtDBGRil+pQevdxTlP2r3a67/Aq5Y7+ab/P0aFw=
+github.com/onflow/flow-go v0.34.0-crescendo-preview.17 h1:HnISCj+nZkfwDHmVF1VmbmyuVbOH76GNGKFTMbthD7c=
+github.com/onflow/flow-go v0.34.0-crescendo-preview.17/go.mod h1:WbsDXtDBGRil+pQevdxTlP2r3a67/Aq5Y7+ab/P0aFw=
 github.com/onflow/flow-go-sdk v1.0.0-M1/go.mod h1:TDW0MNuCs4SvqYRUzkbRnRmHQL1h4X8wURsCw9P9beo=
 github.com/onflow/flow-go-sdk v1.0.0-preview.22 h1:ahHlppdDd4TjHMfnE73SD15AR16KTgbczdhFjGjAtFM=
 github.com/onflow/flow-go-sdk v1.0.0-preview.22/go.mod h1:0hJfpIajLtBvaAUfJAdvWx6WBVp5hS0DJidc0NJLgE4=

services/ingestion/engine.go

@@ -234,6 +234,7 @@ func (e *Engine) indexTransaction(tx models.Transaction, receipt *gethTypes.Rece
 		Str("contract-address", receipt.ContractAddress.String()).
 		Int("log-count", len(receipt.Logs)).
 		Uint64("evm-height", receipt.BlockNumber.Uint64()).
+		Uint("tx-index", receipt.TransactionIndex).
 		Str("receipt-tx-hash", receipt.TxHash.String()).
 		Str("tx-hash", txHash.String()).
 		Msg("ingesting new transaction executed event")

services/logs/filter.go

@@ -65,14 +65,16 @@ func (r *RangeFilter) Match() ([]*gethTypes.Log, error) {
 		}
 
 		// todo do this concurrently
-		receipt, err := r.receipts.GetByBlockHeight(heights[i])
+		receipts, err := r.receipts.GetByBlockHeight(heights[i])
 		if err != nil {
 			return nil, err
 		}
 
-		for _, log := range receipt.Logs {
-			if exactMatch(log, r.criteria) {
-				logs = append(logs, log)
+		for _, receipt := range receipts {
+			for _, log := range receipt.Logs {
+				if exactMatch(log, r.criteria) {
+					logs = append(logs, log)
+				}
 			}
 		}
 	}
@@ -109,15 +111,17 @@ func (i *IDFilter) Match() ([]*gethTypes.Log, error) {
 		return nil, err
 	}
 
-	receipt, err := i.receipts.GetByBlockHeight(big.NewInt(int64(blk.Height)))
+	receipts, err := i.receipts.GetByBlockHeight(big.NewInt(int64(blk.Height)))
 	if err != nil {
 		return nil, err
 	}
 
 	logs := make([]*gethTypes.Log, 0)
-	for _, log := range receipt.Logs {
-		if exactMatch(log, i.criteria) {
-			logs = append(logs, log)
+	for _, receipt := range receipts {
+		for _, log := range receipt.Logs {
+			if exactMatch(log, i.criteria) {
+				logs = append(logs, log)
+			}
 		}
 	}
 

services/logs/filter_test.go

@@ -110,13 +110,19 @@ func receiptStorage() storage.ReceiptIndexer {
 	receiptStorage := &mocks.ReceiptIndexer{}
 	receiptStorage.
 		On("GetByBlockHeight", mock.AnythingOfType("*big.Int")).
-		Return(func(height *big.Int) (*gethTypes.Receipt, error) {
+		Return(func(height *big.Int) ([]*gethTypes.Receipt, error) {
+			rcps := make([]*gethTypes.Receipt, 0)
 			for _, r := range receipts {
 				if r.BlockNumber.Cmp(height) == 0 {
-					return r, nil
+					rcps = append(rcps, r)
 				}
 			}
-			return nil, errors.ErrNotFound
+
+			if len(rcps) == 0 {
+				return nil, errors.ErrNotFound
+			}
+
+			return rcps, nil
 		})
 
 	receiptStorage.

storage/index.go

@@ -64,8 +64,7 @@ type ReceiptIndexer interface {
 	// GetByBlockHeight returns the receipt for the block height.
 	// Expected errors:
 	// - errors.NotFound if the receipt is not found
-	// TODO right now one transaction per block, but this might change in future so the API needs to be updated.
-	GetByBlockHeight(height *big.Int) (*gethTypes.Receipt, error)
+	GetByBlockHeight(height *big.Int) ([]*gethTypes.Receipt, error)
 
 	// BloomsForBlockRange returns slice of bloom values and a slice of block heights
 	// corresponding to each item in the bloom slice. It only matches the blooms between

storage/index_testsuite.go

@@ -128,12 +128,33 @@ type ReceiptTestSuite struct {
 }
 
 func (s *ReceiptTestSuite) TestStoreReceipt() {
-	receipt := mocks.NewReceipt(1, common.HexToHash("0xf1"))
 
 	s.Run("store receipt successfully", func() {
+		receipt := mocks.NewReceipt(1, common.HexToHash("0xf1"))
 		err := s.ReceiptIndexer.Store(receipt)
 		s.Require().NoError(err)
 	})
+
+	s.Run("store multiple receipts at same height", func() {
+		const height = 5
+		receipts := []*types.Receipt{
+			mocks.NewReceipt(height, common.HexToHash("0x1")),
+			mocks.NewReceipt(height, common.HexToHash("0x2")),
+			mocks.NewReceipt(height, common.HexToHash("0x3")),
+		}
+
+		for _, r := range receipts {
+			err := s.ReceiptIndexer.Store(r)
+			s.Require().NoError(err)
+		}
+
+		storeReceipts, err := s.ReceiptIndexer.GetByBlockHeight(big.NewInt(height))
+		s.Require().NoError(err)
+
+		for i, sr := range storeReceipts {
+			s.compareReceipts(receipts[i], sr)
+		}
+	})
 }
 
 func (s *ReceiptTestSuite) TestGetReceiptByTransactionID() {
@@ -155,15 +176,17 @@ func (s *ReceiptTestSuite) TestGetReceiptByTransactionID() {
 	})
 }
 
-func (s *ReceiptTestSuite) TestGetReceiptByBlockID() {
-	s.Run("existing block ID", func() {
+func (s *ReceiptTestSuite) TestGetReceiptByBlockHeight() {
+	s.Run("existing block height", func() {
 		receipt := mocks.NewReceipt(3, common.HexToHash("0x1"))
 		err := s.ReceiptIndexer.Store(receipt)
 		s.Require().NoError(err)
+		// add one more receipt that shouldn't be retrieved
+		s.Require().NoError(s.ReceiptIndexer.Store(mocks.NewReceipt(4, common.HexToHash("0x2"))))
 
-		retReceipt, err := s.ReceiptIndexer.GetByBlockHeight(receipt.BlockNumber)
+		retReceipts, err := s.ReceiptIndexer.GetByBlockHeight(receipt.BlockNumber)
 		s.Require().NoError(err)
-		s.compareReceipts(receipt, retReceipt)
+		s.compareReceipts(receipt, retReceipts[0])
 	})
 
 	s.Run("non-existing block height", func() {
@@ -193,7 +216,33 @@ func (s *ReceiptTestSuite) TestBloomsForBlockRange() {
 		s.Require().Len(heights, len(testBlooms))
 		s.Require().Equal(testBlooms, blooms)
 
-		// todo smaller block range
+		blooms, heights, err = s.ReceiptIndexer.BloomsForBlockRange(start, big.NewInt(13))
+		s.Require().NoError(err)
+		s.Require().Len(blooms, 4)
+		s.Require().Len(heights, 4)
+		s.Require().Equal(testBlooms[0:4], blooms)
+	})
+
+	s.Run("valid block range with multiple receipts per block", func() {
+		start := big.NewInt(15)
+		end := big.NewInt(20)
+		testBlooms := make([]*types.Bloom, 0)
+		testHeights := make([]*big.Int, 0)
+
+		for i := start.Uint64(); i < end.Uint64(); i++ {
+			r1 := mocks.NewReceipt(i, common.HexToHash(fmt.Sprintf("0x%d", i)))
+			r2 := mocks.NewReceipt(i, common.HexToHash(fmt.Sprintf("0x%d", i)))
+			s.Require().NoError(s.ReceiptIndexer.Store(r1))
+			s.Require().NoError(s.ReceiptIndexer.Store(r2))
+			testBlooms = append(testBlooms, &r1.Bloom, &r2.Bloom)
+			testHeights = append(testHeights, big.NewInt(int64(i)))
+		}
+
+		blooms, heights, err := s.ReceiptIndexer.BloomsForBlockRange(start, end)
+		s.Require().NoError(err)
+		s.Require().Len(blooms, len(testBlooms))
+		s.Require().Len(heights, len(testHeights))
+		s.Require().Equal(testBlooms, blooms)
 	})
 
 	s.Run("invalid block range", func() {