Retry fetch logs

.goreleaser.yml

@@ -53,6 +53,8 @@ docker_manifests:
     image_templates:
     - 'avaplatform/awm-relayer:{{ .Tag }}-amd64'
     - 'avaplatform/awm-relayer:{{ .Tag }}-arm64'
+    # If tag is an rc, do not push the latest tag
+    skip_push: auto
 release:
   # Repo in which the release will be created.
   # Default is extracted from the origin remote URL or empty if its private hosted.

types/types.go

@@ -6,7 +6,8 @@ package types
 import (
 	"context"
 	"errors"
-	"math/big"
+	"fmt"
+	"time"
 
 	avalancheWarp "github.com/ava-labs/avalanchego/vms/platformvm/warp"
 	"github.com/ava-labs/subnet-evm/core/types"
@@ -19,6 +20,11 @@ import (
 var WarpPrecompileLogFilter = warp.WarpABI.Events["SendWarpMessage"].ID
 var ErrInvalidLog = errors.New("invalid warp message log")
 
+const (
+	filterLogsRetries = 5
+	retryInterval     = 1 * time.Second
+)
+
 // WarpBlockInfo describes the block height and logs needed to process Warp messages.
 // WarpBlockInfo instances are populated by the subscriber, and forwared to the
 // Listener to process
@@ -44,12 +50,7 @@ func NewWarpBlockInfo(header *types.Header, ethClient ethclient.Client) (*WarpBl
 	)
 	// Check if the block contains warp logs, and fetch them from the client if it does
 	if header.Bloom.Test(WarpPrecompileLogFilter[:]) {
-		logs, err = ethClient.FilterLogs(context.Background(), interfaces.FilterQuery{
-			Topics:    [][]common.Hash{{WarpPrecompileLogFilter}},
-			Addresses: []common.Address{warp.ContractAddress},
-			FromBlock: big.NewInt(int64(header.Number.Uint64())),
-			ToBlock:   big.NewInt(int64(header.Number.Uint64())),
-		})
+		logs, err = fetchWarpLogsWithRetries(ethClient, header, filterLogsRetries, retryInterval)
 		if err != nil {
 			return nil, err
 		}
@@ -101,3 +102,28 @@ func UnpackWarpMessage(unsignedMsgBytes []byte) (*avalancheWarp.UnsignedMessage,
 	}
 	return unsignedMsg, nil
 }
+
+// The node serving the filter logs request may be behind the node serving the block header request,
+// so we retry a few times to ensure we get the logs
+func fetchWarpLogsWithRetries(ethClient ethclient.Client, header *types.Header, numRetries int, retryInterval time.Duration) ([]types.Log, error) {
+	var (
+		logs []types.Log
+		err  error
+	)
+
+	for i := 0; i < numRetries; i++ {
+		logs, err = ethClient.FilterLogs(context.Background(), interfaces.FilterQuery{
+			Topics:    [][]common.Hash{{WarpPrecompileLogFilter}},
+			Addresses: []common.Address{warp.ContractAddress},
+			FromBlock: header.Number,
+			ToBlock:   header.Number,
+		})
+		if err == nil {
+			return logs, nil
+		}
+		if i != numRetries-1 {
+			time.Sleep(retryInterval)
+		}
+	}
+	return nil, fmt.Errorf("failed to fetch warp logs for block %d after %d retries: %w", header.Number.Uint64(), filterLogsRetries, err)
+}