feat: query skyway query unobserved blocks

chain/evm/client.go

@@ -116,6 +116,7 @@ type PalomaClienter interface {
 	QueryBatchRequestByNonce(ctx context.Context, nonce uint64, contract string) (skywaytypes.OutgoingTxBatch, error)
 	QueryGetLatestPublishedSnapshot(ctx context.Context, chainReferenceID string) (*valset.Snapshot, error)
 	QueryGetSnapshotByID(ctx context.Context, id uint64) (*valset.Snapshot, error)
+	QueryUnobservedBlocksByValidator(ctx context.Context, chainReferenceID string, orchestrator string) ([]uint64, error)
 }
 
 type Client struct {

chain/evm/compass.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"math/big"
+	"slices"
 	"time"
 
 	"cosmossdk.io/math"
@@ -24,6 +25,7 @@ import (
 	"github.com/palomachain/pigeon/internal/ethfilter"
 	"github.com/palomachain/pigeon/internal/liblog"
 	"github.com/palomachain/pigeon/util/slice"
+	"github.com/sirupsen/logrus"
 	log "github.com/sirupsen/logrus"
 )
 
@@ -74,7 +76,7 @@ type compass struct {
 	chainID                 *big.Int
 	CompassID               string
 	ChainReferenceID        string
-	lastObservedBlockHeight int64
+	lastObservedBlockHeight uint64
 	startingBlockHeight     int64
 	smartContractAddr       common.Address
 	feeMgrContractAddr      common.Address
@@ -996,6 +998,58 @@ func (t compass) provideEvidenceForReferenceBlock(ctx context.Context, queueType
 	return nil
 }
 
+func (t *compass) getLogs(
+	ctx context.Context,
+	logger *logrus.Entry,
+	from, to uint64,
+) ([]ethtypes.Log, uint64, error) {
+	var filter ethereum.FilterQuery
+	var err error
+
+	if from == 0 && to == 0 {
+		// If from and to are zero, we default to search the latest blocks
+		filter, err = ethfilter.Factory().
+			WithFromBlockNumberProvider(t.evm.FindCurrentBlockNumber).
+			WithFromBlockNumberSafetyMargin(1).
+			WithTopics([]common.Hash{
+				batchSendEvent,
+				sendToPalomaEvent,
+				nodeSaleEvent,
+			}).
+			WithAddresses(t.smartContractAddr).
+			Filter(ctx)
+		if err != nil {
+			return nil, 0, err
+		}
+
+		currentBlockNumber := filter.FromBlock.Uint64()
+		if t.lastObservedBlockHeight == 0 {
+			t.lastObservedBlockHeight = currentBlockNumber - 10_000
+		}
+
+		filter.FromBlock = big.NewInt(0).SetUint64(t.lastObservedBlockHeight)
+		filter.ToBlock = big.NewInt(0).SetUint64(min(t.lastObservedBlockHeight+10_000, currentBlockNumber))
+	} else {
+		filter = ethereum.FilterQuery{
+			Addresses: []common.Address{t.smartContractAddr},
+			Topics: [][]common.Hash{{
+				batchSendEvent,
+				sendToPalomaEvent,
+				nodeSaleEvent,
+			}},
+			FromBlock: big.NewInt(0).SetUint64(from),
+			ToBlock:   big.NewInt(0).SetUint64(to),
+		}
+	}
+
+	logger.WithField("from", filter.FromBlock).
+		WithField("to", filter.ToBlock).
+		Debug("Filter is ready")
+
+	logs, err := t.evm.GetEthClient().FilterLogs(ctx, filter)
+	return logs, filter.ToBlock.Uint64(), err
+}
+
 func (t *compass) GetSkywayEvents(
 	ctx context.Context,
 	orchestrator string,
@@ -1004,38 +1058,35 @@ func (t *compass) GetSkywayEvents(
 
 	logger.Debug("Querying compass events")
 
-	filter, err := ethfilter.Factory().
-		WithFromBlockNumberProvider(t.evm.FindCurrentBlockNumber).
-		WithFromBlockNumberSafetyMargin(1).
-		WithTopics([]common.Hash{
-			batchSendEvent,
-			sendToPalomaEvent,
-			nodeSaleEvent,
-		}).
-		WithAddresses(t.smartContractAddr).
-		Filter(ctx)
+	blocks, err := t.paloma.QueryUnobservedBlocksByValidator(ctx, t.ChainReferenceID, orchestrator)
 	if err != nil {
+		logger.WithError(err).Warn("Failed to query unobserved blocks")
 		return nil, err
 	}
 
-	currentBlockNumber := filter.FromBlock.Int64()
-	if t.lastObservedBlockHeight == 0 {
-		t.lastObservedBlockHeight = currentBlockNumber - 10_000
-	}
+	var logs []ethtypes.Log
+	var toBlock uint64
 
-	filter.FromBlock = big.NewInt(t.lastObservedBlockHeight)
-	filter.ToBlock = big.NewInt(min(t.lastObservedBlockHeight+10_000, currentBlockNumber))
-
-	var events []chain.SkywayEventer
+	if len(blocks) == 0 {
+		logs, toBlock, err = t.getLogs(ctx, logger, 0, 0)
+		if err != nil {
+			logger.WithError(err).Warn("Failed to filter events")
+			return nil, err
+		}
+	} else {
+		var moreLogs []ethtypes.Log
+		for i := range blocks {
+			logger.WithField("block", blocks[i]).
+				Debug("Getting logs from block")
 
-	logger.WithField("from", filter.FromBlock).
-		WithField("to", filter.ToBlock).
-		Debug("Filter is ready")
+			moreLogs, toBlock, err = t.getLogs(ctx, logger, blocks[i], blocks[i])
+			if err != nil {
+				logger.WithError(err).Warn("Failed to filter events")
+				return nil, err
+			}
 
-	logs, err := t.evm.GetEthClient().FilterLogs(ctx, filter)
-	if err != nil {
-		logger.WithError(err).Warn("Failed to filter events")
-		return nil, err
+			logs = slices.Concat(logs, moreLogs)
+		}
 	}
 
 	lastSkywayNonce, err := t.paloma.QueryLastObservedSkywayNonceByAddr(ctx, t.ChainReferenceID, orchestrator)
@@ -1048,6 +1099,7 @@ func (t *compass) GetSkywayEvents(
 		WithField("logs", len(logs)).
 		Debug("Ready to parse events")
 
+	var events []chain.SkywayEventer
 	var evt chain.SkywayEventer
 
 	for _, ethLog := range logs {
@@ -1081,7 +1133,7 @@ func (t *compass) GetSkywayEvents(
 		events = append(events, evt)
 	}
 
-	t.lastObservedBlockHeight = filter.ToBlock.Int64()
+	t.lastObservedBlockHeight = toBlock
 
 	return events, err
 }

chain/paloma/query.go

@@ -198,6 +198,22 @@ func (c *Client) QueryGetValidatorAliveUntilBlockHeight(ctx context.Context) (in
 	return aliveUntilRes.AliveUntilBlockHeight, nil
 }
 
+func (c *Client) QueryUnobservedBlocksByValidator(
+	ctx context.Context,
+	chainReferenceID string,
+	orchestrator string,
+) ([]uint64, error) {
+	qc := skyway.NewQueryClient(c.GRPCClient)
+	res, err := qc.GetUnobservedBlocksByAddr(ctx, &skyway.QueryUnobservedBlocksByAddrRequest{
+		ChainReferenceId: chainReferenceID,
+		Address:          orchestrator,
+	})
+	if err != nil {
+		return nil, err
+	}
+	return res.Blocks, nil
+}
+
 func queryMessagesForSigning(
 	ctx context.Context,
 	c grpc.ClientConn,

go.mod

@@ -22,7 +22,7 @@ require (
 	github.com/jarcoal/httpmock v1.3.1
 	github.com/onsi/ginkgo/v2 v2.19.0
 	github.com/onsi/gomega v1.33.1
-	github.com/palomachain/paloma v1.3.1-next.0.20240830071542-741048197401
+	github.com/palomachain/paloma v1.3.1-next.0.20240912110829-e6db2923ff78
 	github.com/roodeag/arbitrum v0.0.0-20230627104516-b95e4c8ebec0
 	github.com/rs/xid v1.5.0
 	github.com/sirupsen/logrus v1.9.3

go.sum

@@ -240,6 +240,8 @@ github.com/VictoriaMetrics/fastcache v1.12.1 h1:i0mICQuojGDL3KblA7wUNlY5lOK6a4bw
 github.com/VictoriaMetrics/fastcache v1.12.1/go.mod h1:tX04vaqcNoQeGLD+ra5pU5sWkuxnzWhEzLwhP9w653o=
 github.com/VividCortex/gohistogram v1.0.0 h1:6+hBz+qvs0JOrrNhhmR7lFxo5sINxBCGXrdtl/UvroE=
 github.com/VividCortex/gohistogram v1.0.0/go.mod h1:Pf5mBqqDxYaXu3hDrrU+w6nw50o/4+TcAqDqk/vUH7g=
+github.com/VolumeFi/paloma v1.5.0-next.0.20240912102351-206ba32e60b3 h1:xjwn0FW8Vs4BIbNIcKIB/rkaPfVjqppUMp4M6+I4Jr8=
+github.com/VolumeFi/paloma v1.5.0-next.0.20240912102351-206ba32e60b3/go.mod h1:/Gyvm8s1/4A7+/SLUBJ05tDW4DGYPbIHGGqKVK6Rv+k=
 github.com/VolumeFi/whoops v0.7.2 h1:BMxDRo1N14QPPkruA9tsqivpWFXlujvFR+CtWGeQ0cc=
 github.com/VolumeFi/whoops v0.7.2/go.mod h1:WH5v7h5SzHTOIx28WZ9kJqdMqOX2VQNPBelwlrgg5PQ=
 github.com/adlio/schema v1.3.3 h1:oBJn8I02PyTB466pZO1UZEn1TV5XLlifBSyMrmHl/1I=
@@ -949,8 +951,8 @@ github.com/ory/dockertest v3.3.5+incompatible/go.mod h1:1vX4m9wsvi00u5bseYwXaSnh
 github.com/pact-foundation/pact-go v1.0.4/go.mod h1:uExwJY4kCzNPcHRj+hCR/HBbOOIwwtUjcrb0b5/5kLM=
 github.com/palomachain/arb-geth v0.0.0-20230824112942-8e77a580a936 h1:fmQAgxcdYBxCZYczws/uxTVOYHZd4fNrOaoHp35HZMM=
 github.com/palomachain/arb-geth v0.0.0-20230824112942-8e77a580a936/go.mod h1:B2H2+2I4UiMR4hvAIaGLyYszNfSTYC8fWIw+kgfuFSQ=
-github.com/palomachain/paloma v1.3.1-next.0.20240830071542-741048197401 h1:ZO/l1TCo4Ry4BuF/PUH3H9Y/ipY6fKPO9rjjy+i0zzE=
-github.com/palomachain/paloma v1.3.1-next.0.20240830071542-741048197401/go.mod h1:91O0VRxBof+IXTZxSsSbk/lV2DSaWQIdUaoCBv6HXDo=
+github.com/palomachain/paloma v1.3.1-next.0.20240912110829-e6db2923ff78 h1:1795bjnQjinb2pHTM3/8ANnUQK9UdvrRCkkv8MsmTBU=
+github.com/palomachain/paloma v1.3.1-next.0.20240912110829-e6db2923ff78/go.mod h1:/Gyvm8s1/4A7+/SLUBJ05tDW4DGYPbIHGGqKVK6Rv+k=
 github.com/pascaldekloe/goe v0.0.0-20180627143212-57f6aae5913c/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc=
 github.com/pascaldekloe/goe v0.1.0 h1:cBOtyMzM9HTpWjXfbbunk26uA6nG3a8n06Wieeh0MwY=
 github.com/pascaldekloe/goe v0.1.0/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc=