feat: use archive nodes to query attestations

cmd/blobstream/deploy/cmd.go

@@ -43,21 +43,33 @@ func Command() *cobra.Command {
 
 			encCfg := encoding.MakeConfig(app.ModuleEncodingRegisters...)
 
-			querier := rpc.NewAppQuerier(logger, config.coreGRPC, encCfg)
-			err = querier.Start()
+			appQuerier := rpc.NewAppQuerier(logger, config.coreGRPC, encCfg)
+			err = appQuerier.Start()
 			if err != nil {
 				return err
 			}
 			defer func() {
-				err := querier.Stop()
+				err := appQuerier.Stop()
 				if err != nil {
 					logger.Error(err.Error())
 				}
 			}()
 
-			vs, err := getStartingValset(cmd.Context(), querier, config.startingNonce)
+			tmQuerier := rpc.NewTmQuerier(config.coreRPC, logger)
+			err = tmQuerier.Start()
 			if err != nil {
-				logger.Error("couldn't get valset from state (probably pruned). connect to an archive node to be able to deploy the contract")
+				return err
+			}
+			defer func(tmQuerier *rpc.TmQuerier) {
+				err := tmQuerier.Stop()
+				if err != nil {
+					logger.Error(err.Error())
+				}
+			}(tmQuerier)
+
+			vs, err := getStartingValset(cmd.Context(), *tmQuerier, appQuerier, config.startingNonce)
+			if err != nil {
+				logger.Error("couldn't get valset from state (probably pruned). connect to an archive node to be able to deploy the contract", "err", err.Error())
 				return errors.Wrap(
 					err,
 					"cannot initialize the Blobstream contract without having a valset request: %s",
@@ -130,15 +142,29 @@ func Command() *cobra.Command {
 }
 
 // getStartingValset get the valset that will be used to init the bridge contract.
-func getStartingValset(ctx context.Context, querier *rpc.AppQuerier, startingNonce string) (*types.Valset, error) {
+func getStartingValset(ctx context.Context, tmQuerier rpc.TmQuerier, appQuerier *rpc.AppQuerier, startingNonce string) (*types.Valset, error) {
 	switch startingNonce {
 	case "latest":
-		return querier.QueryLatestValset(ctx)
+		vs, err := appQuerier.QueryLatestValset(ctx)
+		if err != nil {
+			appQuerier.Logger.Debug("couldn't get the attestation from node state. trying with historical data if the target node is archival", "nonce", 1, "err", err.Error())
+			currentHeight, err := tmQuerier.QueryHeight(ctx)
+			if err != nil {
+				return nil, err
+			}
+			return appQuerier.QueryRecursiveLatestValset(ctx, uint64(currentHeight))
+		}
+		return vs, nil
 	case "earliest":
 		// TODO make the first nonce 1 a const
-		att, err := querier.QueryAttestationByNonce(ctx, 1)
+		att, err := appQuerier.QueryAttestationByNonce(ctx, 1)
 		if err != nil {
-			return nil, err
+			appQuerier.Logger.Debug("couldn't get the attestation from node state. trying with historical data if the target node is archival", "nonce", 1, "err", err.Error())
+			historicalAtt, err := appQuerier.QueryHistoricalAttestationByNonce(ctx, 1, 1)
+			if err != nil {
+				return nil, err
+			}
+			att = historicalAtt
 		}
 		vs, ok := att.(*types.Valset)
 		if !ok {
@@ -150,17 +176,23 @@ func getStartingValset(ctx context.Context, querier *rpc.AppQuerier, startingNon
 		if err != nil {
 			return nil, err
 		}
-		attestation, err := querier.QueryAttestationByNonce(ctx, nonce)
+		currentHeight, err := tmQuerier.QueryHeight(ctx)
+		if err != nil {
+			return nil, err
+		}
+		attestation, err := appQuerier.QueryRecursiveHistoricalAttestationByNonce(ctx, nonce, uint64(currentHeight))
 		if err != nil {
 			return nil, err
 		}
 		if attestation == nil {
 			return nil, types.ErrNilAttestation
 		}
-		value, ok := attestation.(*types.Valset)
-		if ok {
+		switch value := attestation.(type) {
+		case *types.Valset:
 			return value, nil
+		case *types.DataCommitment:
+			return appQuerier.QueryRecursiveHistoricalLastValsetBeforeNonce(ctx, nonce, value.EndBlock)
 		}
-		return querier.QueryLastValsetBeforeNonce(ctx, nonce)
 	}
+	return nil, ErrNotFound
 }

cmd/blobstream/deploy/config.go

@@ -17,6 +17,8 @@ const (
 	FlagEVMGasLimit     = "evm.gas-limit"
 	FlagCoreGRPCHost    = "core.grpc.host"
 	FlagCoreGRPCPort    = "core.grpc.port"
+	FlagCoreRPCHost     = "core.rpc.host"
+	FlagCoreRPCPort     = "core.rpc.port"
 	FlagStartingNonce   = "starting-nonce"
 	ServiceNameDeployer = "deployer"
 )
@@ -26,6 +28,8 @@ func addDeployFlags(cmd *cobra.Command) *cobra.Command {
 	cmd.Flags().Uint64(FlagEVMChainID, 5, "Specify the evm chain id")
 	cmd.Flags().String(FlagCoreGRPCHost, "localhost", "Specify the grpc address host")
 	cmd.Flags().Uint(FlagCoreGRPCPort, 9090, "Specify the grpc address port")
+	cmd.Flags().String(FlagCoreRPCHost, "localhost", "Specify the rpc address host")
+	cmd.Flags().Uint(FlagCoreRPCPort, 26657, "Specify the rpc address port")
 	cmd.Flags().String(FlagEVMRPC, "http://localhost:8545", "Specify the ethereum rpc address")
 	cmd.Flags().String(
 		FlagStartingNonce,
@@ -48,11 +52,12 @@ func addDeployFlags(cmd *cobra.Command) *cobra.Command {
 
 type deployConfig struct {
 	*base.Config
-	evmRPC, coreGRPC string
-	evmChainID       uint64
-	evmAccAddress    string
-	startingNonce    string
-	evmGasLimit      uint64
+	evmRPC            string
+	coreRPC, coreGRPC string
+	evmChainID        uint64
+	evmAccAddress     string
+	startingNonce     string
+	evmGasLimit       uint64
 }
 
 func parseDeployFlags(cmd *cobra.Command) (deployConfig, error) {
@@ -75,6 +80,14 @@ func parseDeployFlags(cmd *cobra.Command) (deployConfig, error) {
 	if err != nil {
 		return deployConfig{}, err
 	}
+	coreRPCHost, err := cmd.Flags().GetString(FlagCoreRPCHost)
+	if err != nil {
+		return deployConfig{}, err
+	}
+	coreRPCPort, err := cmd.Flags().GetUint(FlagCoreRPCPort)
+	if err != nil {
+		return deployConfig{}, err
+	}
 	evmRPC, err := cmd.Flags().GetString(FlagEVMRPC)
 	if err != nil {
 		return deployConfig{}, err
@@ -107,6 +120,7 @@ func parseDeployFlags(cmd *cobra.Command) (deployConfig, error) {
 		evmAccAddress: evmAccAddr,
 		evmChainID:    evmChainID,
 		coreGRPC:      fmt.Sprintf("%s:%d", coreGRPCHost, coreGRPCPort),
+		coreRPC:       fmt.Sprintf("tcp://%s:%d", coreRPCHost, coreRPCPort),
 		evmRPC:        evmRPC,
 		startingNonce: startingNonce,
 		evmGasLimit:   evmGasLimit,

cmd/blobstream/deploy/errors.go

@@ -2,4 +2,7 @@ package deploy
 
 import "errors"
 
-var ErrUnmarshallValset = errors.New("couldn't unmarsall valset")
+var (
+	ErrUnmarshallValset = errors.New("couldn't unmarshall valset")
+	ErrNotFound         = errors.New("not found")
+)

e2e/scripts/deploy_blobstream_contract.sh

@@ -68,6 +68,8 @@ echo "deploying Blobstream contract..."
   --evm.account "${EVM_ACCOUNT}" \
   --core.grpc.host "${CORE_GRPC_HOST}" \
   --core.grpc.port "${CORE_GRPC_PORT}" \
+  --core.rpc.host "${CORE_RPC_HOST}" \
+  --core.rpc.port "${CORE_RPC_PORT}" \
   --starting-nonce "${STARTING_NONCE}" \
   --evm.rpc "${EVM_ENDPOINT}" \
   --evm.passphrase=123 > /opt/output

orchestrator/orchestrator_test.go

@@ -206,7 +206,9 @@ func TestProcessWithoutValsetInStore(t *testing.T) {
 				testnode.ImmediateProposals(codec),
 				qgbtesting.SetDataCommitmentWindowParams(codec, celestiatypes.Params{DataCommitmentWindow: 101}),
 			},
-			TimeIotaMs: 6048000, // to have enough time to sign attestations after they're pruned
+			TimeIotaMs:    6048000, // to have enough time to sign attestations after they're pruned
+			Pruning:       "default",
+			TimeoutCommit: 5 * time.Millisecond,
 		},
 	)
 	_, err := node.CelestiaNetwork.WaitForHeight(400)

orchestrator/suite_test.go

@@ -3,6 +3,7 @@ package orchestrator_test
 import (
 	"context"
 	"testing"
+	"time"
 
 	"github.com/celestiaorg/celestia-app/app"
 	"github.com/celestiaorg/celestia-app/app/encoding"
@@ -31,7 +32,9 @@ func (s *OrchestratorTestSuite) SetupSuite() {
 				testnode.ImmediateProposals(codec),
 				blobstreamtesting.SetDataCommitmentWindowParams(codec, types.Params{DataCommitmentWindow: 101}),
 			},
-			TimeIotaMs: 1,
+			TimeIotaMs:    1,
+			Pruning:       "default",
+			TimeoutCommit: 5 * time.Millisecond,
 		},
 	)
 	s.Orchestrator = blobstreamtesting.NewOrchestrator(t, s.Node)

relayer/relayer.go

@@ -148,7 +148,15 @@ func (r *Relayer) ProcessAttestation(ctx context.Context, opts *bind.TransactOpt
 		r.logger.Error("failed to query the last valset before nonce (probably pruned). recovering via falling back to the P2P network", "err", err.Error())
 		previousValset, err = r.QueryValsetFromP2PNetworkAndValidateIt(ctx)
 		if err != nil {
-			return nil, err
+			r.logger.Error("failed to query the last valset before nonce from p2p network. trying using an archive node", "err", err.Error())
+			currentHeight, err := r.TmQuerier.QueryHeight(ctx)
+			if err != nil {
+				return nil, err
+			}
+			previousValset, err = r.AppQuerier.QueryRecursiveHistoricalLastValsetBeforeNonce(ctx, attI.GetNonce(), uint64(currentHeight))
+			if err != nil {
+				return nil, err
+			}
 		}
 	}
 	switch att := attI.(type) {

relayer/relayer_test.go

@@ -68,7 +68,9 @@ func TestUseValsetFromP2P(t *testing.T) {
 				testnode.ImmediateProposals(codec),
 				qgbtesting.SetDataCommitmentWindowParams(codec, types.Params{DataCommitmentWindow: 101}),
 			},
-			TimeIotaMs: 2000000, // so attestations are pruned after they're queried
+			TimeIotaMs:    2000000, // so attestations are pruned after they're queried
+			Pruning:       "default",
+			TimeoutCommit: 5 * time.Millisecond,
 		},
 	)
 

rpc/app_historic_querier_test.go

@@ -0,0 +1,80 @@
+package rpc_test
+
+import (
+	"context"
+)
+
+func (s *HistoricQuerierTestSuite) TestQueryHistoricAttestationByNonce() {
+	appQuerier := s.setupAppQuerier()
+
+	// this one should fail because the attestation is deleted from the state
+	_, err := appQuerier.QueryAttestationByNonce(context.Background(), 1)
+	s.Error(err)
+
+	att, err := appQuerier.QueryHistoricalAttestationByNonce(context.Background(), 1, 10)
+	s.NoError(err)
+	s.NotNil(att)
+	s.Equal(uint64(1), att.GetNonce())
+}
+
+func (s *HistoricQuerierTestSuite) TestQueryRecursiveHistoricAttestationByNonce() {
+	appQuerier := s.setupAppQuerier()
+
+	// this one should fail because the attestation is deleted from the state
+	_, err := appQuerier.QueryAttestationByNonce(context.Background(), 1)
+	s.Error(err)
+
+	height, err := s.Network.LatestHeight()
+	s.Require().NoError(err)
+	att, err := appQuerier.QueryRecursiveHistoricalAttestationByNonce(context.Background(), 1, uint64(height))
+	s.Require().NoError(err)
+	s.NotNil(att)
+	s.Equal(uint64(1), att.GetNonce())
+}
+
+func (s *HistoricQuerierTestSuite) TestQueryHistoricalLatestAttestationNonce() {
+	appQuerier := s.setupAppQuerier()
+
+	nonce, err := appQuerier.QueryHistoricalLatestAttestationNonce(context.Background(), 2)
+	s.Require().NoError(err)
+	s.Equal(uint64(1), nonce)
+}
+
+func (s *HistoricQuerierTestSuite) TestQueryHistoricalValsetByNonce() {
+	appQuerier := s.setupAppQuerier()
+
+	// this one should fail because the attestation is deleted from the state
+	_, err := appQuerier.QueryValsetByNonce(context.Background(), 1)
+	s.Error(err)
+
+	att, err := appQuerier.QueryHistoricalValsetByNonce(context.Background(), 1, 10)
+	s.Require().NoError(err)
+	s.NotNil(att)
+	s.Equal(uint64(1), att.GetNonce())
+}
+
+func (s *HistoricQuerierTestSuite) TestQueryHistoricalLastValsetBeforeNonce() {
+	appQuerier := s.setupAppQuerier()
+
+	// this one should fail because the attestation is deleted from the state
+	_, err := appQuerier.QueryLastValsetBeforeNonce(context.Background(), 2)
+	s.Error(err)
+
+	att, err := appQuerier.QueryHistoricalLastValsetBeforeNonce(context.Background(), 2, 102)
+	s.Require().NoError(err)
+	s.NotNil(att)
+	s.Equal(uint64(1), att.GetNonce())
+}
+
+func (s *HistoricQuerierTestSuite) TestQueryRecursiveHistoricalLastValsetBeforeNonce() {
+	appQuerier := s.setupAppQuerier()
+
+	// this one should fail because the attestation is deleted from the state
+	_, err := appQuerier.QueryLastValsetBeforeNonce(context.Background(), 2)
+	s.Error(err)
+
+	att, err := appQuerier.QueryRecursiveHistoricalLastValsetBeforeNonce(context.Background(), 2, 201)
+	s.Require().NoError(err)
+	s.NotNil(att)
+	s.Equal(uint64(1), att.GetNonce())
+}