Streaming of logs

api/models.go

@@ -11,13 +11,17 @@ import (
 )
 
 var (
-	errFilterNotFound  = errors.New("filter not found")
-	errExceedMaxTopics = errors.New("exceed max topics")
+	errFilterNotFound     = errors.New("filter not found")
+	errExceedMaxTopics    = errors.New("exceed max topics")
+	errExceedMaxAddresses = errors.New("exceed max addresses")
 )
 
 // The maximum number of topic criteria allowed, vm.LOG4 - vm.LOG0
 const maxTopics = 4
 
+// The maximum number of addresses allowed
+const maxAddresses = 6
+
 // TransactionArgs represents the arguments to construct a new transaction
 // or a message call.
 type TransactionArgs struct {

api/stream.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"github.com/ethereum/go-ethereum/eth/filters"
+	"github.com/onflow/flow-evm-gateway/services/logs"
 
 	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/ethereum/go-ethereum/rpc"
@@ -29,6 +31,7 @@ type StreamAPI struct {
 	receipts                storage.ReceiptIndexer
 	blocksBroadcaster       *engine.Broadcaster
 	transactionsBroadcaster *engine.Broadcaster
+	logsBroadcaster         *engine.Broadcaster
 }
 
 func NewStreamAPI(
@@ -37,19 +40,19 @@ func NewStreamAPI(
 	blocks storage.BlockIndexer,
 	transactions storage.TransactionIndexer,
 	receipts storage.ReceiptIndexer,
-	accounts storage.AccountIndexer,
 	blocksBroadcaster *engine.Broadcaster,
 	transactionsBroadcaster *engine.Broadcaster,
+	logsBroadcaster *engine.Broadcaster,
 ) *StreamAPI {
 	return &StreamAPI{
 		logger:                  logger,
 		config:                  config,
 		blocks:                  blocks,
 		transactions:            transactions,
 		receipts:                receipts,
-		accounts:                accounts,
 		blocksBroadcaster:       blocksBroadcaster,
 		transactionsBroadcaster: transactionsBroadcaster,
+		logsBroadcaster:         logsBroadcaster,
 	}
 }
 
@@ -85,7 +88,7 @@ func (s *StreamAPI) newSubscription(
 		s.config.StreamTimeout,
 		s.config.StreamLimit,
 		sub,
-	).Stream(context.Background())
+	).Stream(context.Background()) // todo investigate why the passed in context is canceled so quickly
 
 	go func() {
 		for {
@@ -177,3 +180,36 @@ func (s *StreamAPI) NewPendingTransactions(ctx context.Context, fullTx *bool) (*
 		},
 	)
 }
+
+// Logs creates a subscription that fires for all new log that match the given filter criteria.
+func (s *StreamAPI) Logs(ctx context.Context, criteria filters.FilterCriteria) (*rpc.Subscription, error) {
+	if len(criteria.Topics) > maxTopics {
+		return nil, errExceedMaxTopics
+	}
+	if len(criteria.Addresses) > maxAddresses {
+		return nil, errExceedMaxAddresses
+	}
+
+	return s.newSubscription(
+		ctx,
+		s.logsBroadcaster,
+		func(ctx context.Context, height uint64) (interface{}, error) {
+			block, err := s.blocks.GetByHeight(height)
+			if err != nil {
+				return nil, fmt.Errorf("failed to get block at height: %d: %w", height, err)
+			}
+
+			id, err := block.Hash()
+			if err != nil {
+				return nil, err
+			}
+
+			// convert from the API type
+			f := logs.FilterCriteria{
+				Addresses: criteria.Addresses,
+				Topics:    criteria.Topics,
+			}
+			return logs.NewIDFilter(id, f, s.blocks, s.receipts).Match()
+		},
+	)
+}

bootstrap/bootstrap.go

@@ -36,6 +36,7 @@ func Start(ctx context.Context, cfg *config.Config) error {
 
 	blocksBroadcaster := broadcast.NewBroadcaster()
 	transactionsBroadcaster := broadcast.NewBroadcaster()
+	logsBroadcaster := broadcast.NewBroadcaster()
 
 	// if database is not initialized require init height
 	if _, err := blocks.LatestCadenceHeight(); errors.Is(err, storageErrs.ErrNotInitialized) {
@@ -55,6 +56,7 @@ func Start(ctx context.Context, cfg *config.Config) error {
 			accounts,
 			blocksBroadcaster,
 			transactionsBroadcaster,
+			logsBroadcaster,
 			logger,
 		)
 		if err != nil {
@@ -72,6 +74,7 @@ func Start(ctx context.Context, cfg *config.Config) error {
 		accounts,
 		blocksBroadcaster,
 		transactionsBroadcaster,
+		logsBroadcaster,
 		logger,
 	)
 	if err != nil {
@@ -90,6 +93,7 @@ func startIngestion(
 	accounts storage.AccountIndexer,
 	blocksBroadcaster *broadcast.Broadcaster,
 	transactionsBroadcaster *broadcast.Broadcaster,
+	logsBroadcaster *broadcast.Broadcaster,
 	logger zerolog.Logger,
 ) error {
 	logger.Info().Msg("starting up event ingestion")
@@ -130,6 +134,7 @@ func startIngestion(
 		accounts,
 		blocksBroadcaster,
 		transactionsBroadcaster,
+		logsBroadcaster,
 		logger,
 	)
 	const retries = 15
@@ -158,6 +163,7 @@ func startServer(
 	accounts storage.AccountIndexer,
 	blocksBroadcaster *broadcast.Broadcaster,
 	transactionsBroadcaster *broadcast.Broadcaster,
+	logsBroadcaster *broadcast.Broadcaster,
 	logger zerolog.Logger,
 ) error {
 	l := logger.With().Str("component", "API").Logger()
@@ -214,6 +220,7 @@ func startServer(
 		accounts,
 		blocksBroadcaster,
 		transactionsBroadcaster,
+		logsBroadcaster,
 	)
 
 	pullAPI := api.NewPullAPI(

integration/e2e_test.go

@@ -946,7 +946,7 @@ func TestE2E_Streaming(t *testing.T) {
 		COAAddress:         gwAddress,
 		COAKey:             gwKey,
 		CreateCOAResource:  true,
-		GasPrice:           new(big.Int).SetUint64(1),
+		GasPrice:           new(big.Int).SetUint64(0),
 		LogWriter:          os.Stdout,
 		LogLevel:           zerolog.DebugLevel,
 		StreamLimit:        5,
@@ -976,7 +976,7 @@ func TestE2E_Streaming(t *testing.T) {
 	// connect and subscribe to new blocks
 	blkWrite, blkRead, err := rpcTester.wsConnect()
 	require.NoError(t, err)
-	err = blkWrite(newHeadsRequest())
+	err = blkWrite(newHeadsSubscription())
 	require.NoError(t, err)
 
 	// first block stream response is for successful subscription
@@ -986,7 +986,7 @@ func TestE2E_Streaming(t *testing.T) {
 	// connect and subscribe to new transactions
 	txWrite, txRead, err := rpcTester.wsConnect()
 	require.NoError(t, err)
-	err = txWrite(newTransactionsRequest())
+	err = txWrite(newTransactionsSubscription())
 	require.NoError(t, err)
 
 	// first block stream response is for successful subscription
@@ -1017,15 +1017,15 @@ func TestE2E_Streaming(t *testing.T) {
 		event, err := blkRead()
 		require.NoError(t, err)
 
-		var msg streamMsg
-		require.NoError(t, json.Unmarshal(event, &msg))
+		var res map[string]any
+		require.NoError(t, json.Unmarshal(event.Params.Result, &res))
 
 		// this makes sure we receive the events in correct order
-		h, err := hexutil.DecodeUint64(msg.Params.Result["number"].(string))
+		h, err := hexutil.DecodeUint64(res["number"].(string))
 		require.NoError(t, err)
 		assert.Equal(t, currentHeight, int(h))
 		currentHeight++
-		blkSubID = msg.Params.Subscription
+		blkSubID = event.Params.Subscription
 	}
 
 	// iterate over all transactions and make sure all were received
@@ -1035,30 +1035,138 @@ func TestE2E_Streaming(t *testing.T) {
 		event, err := txRead()
 		require.NoError(t, err)
 
-		var msg streamMsg
-		require.NoError(t, json.Unmarshal(event, &msg))
-
+		var res map[string]string
+		require.NoError(t, json.Unmarshal(event.Params.Result, &res))
 		// this makes sure we received txs in correct order
-		h, err := hexutil.DecodeUint64(msg.Params.Result["blockNumber"].(string))
+		h, err := hexutil.DecodeUint64(res["blockNumber"])
 		require.NoError(t, err)
 		assert.Equal(t, currentHeight, int(h))
-		require.Equal(t, transferEOAAdress.Hex(), msg.Params.Result["to"].(string))
+		require.Equal(t, transferEOAAdress.Hex(), res["to"])
 		currentHeight++
-		txSubID = msg.Params.Subscription
+		txSubID = event.Params.Subscription
 	}
 
 	unsubscribe(t, blkWrite, blkRead, blkSubID)
 	unsubscribe(t, txWrite, txRead, txSubID)
+
+	// deploy contract for logs/events stream filtering
+	nonce, err := rpcTester.getNonce(fundEOAAddress)
+	require.NoError(t, err)
+
+	gasLimit := uint64(4700000) // arbitrarily big
+	eoaKey, err := crypto.HexToECDSA(fundEOARawKey)
+	require.NoError(t, err)
+
+	deployData, err := hex.DecodeString(testContractBinary)
+	require.NoError(t, err)
+
+	signed, _, err := evmSign(nil, gasLimit, eoaKey, nonce, nil, deployData)
+	hash, err := rpcTester.sendRawTx(signed)
+	require.NoError(t, err)
+	require.NotNil(t, hash)
+
+	time.Sleep(300 * time.Millisecond) // todo replace all sleeps with checking for receipt
+
+	rcp, err := rpcTester.getReceipt(hash.Hex())
+	require.NoError(t, err)
+	contractAddress := rcp.ContractAddress
+
+	storeContract, err := newContract(testContractBinary, testContractABI)
+	require.NoError(t, err)
+
+	// different subscription to logs with different filters
+	allLogsWrite, allLogsRead, err := rpcTester.wsConnect()
+	require.NoError(t, err)
+	err = allLogsWrite(newLogsSubscription(contractAddress.String(), ``))
+	require.NoError(t, err)
+
+	singleLogWrite, singleLogRead, err := rpcTester.wsConnect()
+	require.NoError(t, err)
+	topic4 := common.HexToHash("0x3")
+	err = singleLogWrite(newLogsSubscription(
+		contractAddress.String(),
+		fmt.Sprintf(`null, null, null, "%s"`, topic4),
+	))
+	require.NoError(t, err)
+
+	// ignore successful subscription result
+	_, err = allLogsRead()
+	require.NoError(t, err)
+	_, err = singleLogRead()
+	require.NoError(t, err)
+	// ignore current block
+	_, err = allLogsRead()
+	require.NoError(t, err)
+	_, err = singleLogRead()
+	require.NoError(t, err)
+
+	// example event
+	// [{"address":"0x35c2cd9bee2ca40f8b91c924188c5018df2984e2","topics":["0x76efea95e5da1fa661f235b2921ae1d89b99e457ec73fb88e34a1d150f95c64b","0x000000000000000000000000facf71692421039876a5bb4f10ef7a439d8ef61e","0x0000000000000000000000000000000000000000000000000000000000000005","0x0000000000000000000000000000000000000000000000000000000000000001"]
+
+	// submit transactions that emit logs
+	logCount := 5
+	sumA := big.NewInt(5)
+	for i := 0; i < logCount; i++ {
+		sumB := big.NewInt(int64(i))
+		callSum, err := storeContract.call("sum", sumA, sumB)
+		require.NoError(t, err)
+
+		nonce++
+		res, _, err := evmSignAndRun(emu, nil, gasLimit, eoaKey, nonce, &contractAddress, callSum)
+		require.NoError(t, err)
+		require.NoError(t, res.Error)
+	}
+
+	time.Sleep(500 * time.Millisecond)
+
+	currentHeight = startHeight + logCount + 1 // + 1 height for deploy of contract
+	for i := 0; i < logCount; i++ {
+		event, err := allLogsRead()
+		require.NoError(t, err)
+
+		var l []gethTypes.Log
+		assert.NoError(t, json.Unmarshal(event.Params.Result, &l))
+
+		require.Len(t, l, 1)
+		log := l[0]
+		// this makes sure we received logs in correct order
+		assert.Equal(t, uint64(currentHeight), log.BlockNumber)
+		assert.Equal(t, contractAddress.Hex(), log.Address.Hex())
+		assert.Len(t, log.Topics, 4)
+		assert.Equal(t, common.BigToHash(sumA), log.Topics[2])
+		sumB := big.NewInt(int64(i))
+		assert.Equal(t, common.BigToHash(sumB), log.Topics[3])
+
+		currentHeight++
+	}
+
+	// todo remove this once the change in broadcaster is made to ignore null results
+	singleLogRead()
+	singleLogRead()
+	singleLogRead()
+
+	event, err := singleLogRead()
+	require.NoError(t, err)
+
+	var l []gethTypes.Log
+	assert.NoError(t, json.Unmarshal(event.Params.Result, &l))
+	require.Len(t, l, 1)
+	log := l[0]
+	fmt.Println(log)
+
+	assert.Equal(t, uint64(startHeight+logCount+1+3), log.BlockNumber)
+	assert.Equal(t, contractAddress.Hex(), log.Address.Hex())
+	assert.Len(t, log.Topics, 4)
+	assert.Equal(t, common.BigToHash(sumA), log.Topics[2])
+	assert.Equal(t, topic4, log.Topics[3])
 }
 
-func unsubscribe(t *testing.T, write func(string) error, read func() ([]byte, error), id string) {
+func unsubscribe(t *testing.T, write func(string) error, read func() (*streamMsg, error), id string) {
 	require.NotEmpty(t, id)
 	require.NoError(t, write(unsubscribeRequest(id)))
-	res, err := read()
+	event, err := read()
 	require.NoError(t, err)
-	var u map[string]any
-	require.NoError(t, json.Unmarshal(res, &u))
-	require.True(t, u["result"].(bool)) // successfully unsubscribed
+	require.True(t, event.Result.(bool)) // successfully unsubscribed
 }
 
 func TestE2E_Pull(t *testing.T) {