feat(relayer): persist cursors for attestation stream (#2425)

Persisting cursors for each stream which is used during relayer
bootstrap.

We add a storage for cursors and monitor the stored cursors and confirm
them once the submissions are finalized on the network. Confirmed
cursors are also trimmed from the database.

The confirmation rules are:
- Confirm an empty cursor if a previous cursor is confirmed
- Confirm a non empty cursor if a finalized submission from the network
contains equal or higher attestation offset

Issue: #2412

relayer/app/app.go

@@ -13,17 +13,25 @@ import (
 	"github.com/omni-network/omni/lib/netconf"
 	"github.com/omni-network/omni/lib/xchain"
 	xprovider "github.com/omni-network/omni/lib/xchain/provider"
+	"github.com/omni-network/omni/relayer/app/cursor"
 
 	"github.com/cometbft/cometbft/rpc/client"
 	"github.com/cometbft/cometbft/rpc/client/http"
 
 	"github.com/ethereum/go-ethereum/common"
 	ethcrypto "github.com/ethereum/go-ethereum/crypto"
+
+	dbm "github.com/cosmos/cosmos-db"
 )
 
 func Run(ctx context.Context, cfg Config) error {
 	log.Info(ctx, "Starting relayer")
 
+	db, err := initializeDB(ctx, cfg)
+	if err != nil {
+		return err
+	}
+
 	buildinfo.Instrument(ctx)
 
 	// Start metrics first, so app is "up"
@@ -60,6 +68,12 @@ func Run(ctx context.Context, cfg Config) error {
 	pricer := newTokenPricer(ctx)
 	pnl := newPnlLogger(network.ID, pricer)
 
+	cursors, err := cursor.New(db, xprov.GetSubmittedCursor, network)
+	if err != nil {
+		return err
+	}
+	cursors.StartLoops(ctx)
+
 	for _, destChain := range network.EVMChains() {
 		// Setup send provider
 		sendProvider := func() (SendAsync, error) {
@@ -93,7 +107,9 @@ func Run(ctx context.Context, cfg Config) error {
 			xprov,
 			CreateSubmissions,
 			sendProvider,
-			awaitValSet)
+			awaitValSet,
+			cursors,
+		)
 
 		go worker.Run(ctx)
 	}
@@ -133,6 +149,21 @@ func initializeRPCClients(chains []netconf.Chain, endpoints xchain.RPCEndpoints)
 	return rpcClientPerChain, nil
 }
 
+func initializeDB(ctx context.Context, cfg Config) (dbm.DB, error) {
+	var db dbm.DB
+	if cfg.DBDir == "" {
+		log.Warn(ctx, "No --db-dir provided, using in-memory DB", nil)
+		return dbm.NewMemDB(), nil
+	}
+	var err error
+	db, err = dbm.NewGoLevelDB("indexer", cfg.DBDir, nil)
+	if err != nil {
+		return nil, errors.Wrap(err, "new golevel db")
+	}
+
+	return db, nil
+}
+
 func makePortalRegistry(network netconf.ID, endpoints xchain.RPCEndpoints) (*bindings.PortalRegistry, error) {
 	meta := netconf.MetadataByID(network, network.Static().OmniExecutionChainID)
 	rpc, err := endpoints.ByNameOrID(meta.Name, meta.ChainID)

relayer/app/config.go

@@ -3,6 +3,7 @@ package relayer
 import (
 	"bytes"
 	"text/template"
+	"time"
 
 	"github.com/omni-network/omni/lib/buildinfo"
 	"github.com/omni-network/omni/lib/errors"
@@ -16,19 +17,23 @@ import (
 )
 
 type Config struct {
-	RPCEndpoints   xchain.RPCEndpoints
-	PrivateKey     string
-	HaloURL        string
-	Network        netconf.ID
-	MonitoringAddr string
+	RPCEndpoints    xchain.RPCEndpoints
+	PrivateKey      string
+	HaloURL         string
+	Network         netconf.ID
+	MonitoringAddr  string
+	DBDir           string
+	ConfirmInterval time.Duration
 }
 
 func DefaultConfig() Config {
 	return Config{
-		PrivateKey:     "relayer.key",
-		HaloURL:        "localhost:26657",
-		Network:        "",
-		MonitoringAddr: ":26660",
+		PrivateKey:      "relayer.key",
+		HaloURL:         "localhost:26657",
+		Network:         "",
+		MonitoringAddr:  ":26660",
+		DBDir:           "./db",
+		ConfirmInterval: 30 * time.Second,
 	}
 }