feat: gossiping the latest valset in P2P layer in case they get pruned

cmd/blobstream/deploy/cmd.go

@@ -57,6 +57,7 @@ func Command() *cobra.Command {
 
 			vs, err := getStartingValset(cmd.Context(), querier, 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")
 				return errors.Wrap(
 					err,
 					"cannot initialize the Blobstream contract without having a valset request: %s",

evm/evm_client.go

@@ -216,6 +216,14 @@ func (ec *Client) StateLastEventNonce(opts *bind.CallOpts) (uint64, error) {
 	return nonce.Uint64(), nil
 }
 
+func (ec *Client) StateLastValidatorSetCheckpoint(opts *bind.CallOpts) ([32]byte, error) {
+	checkpoint, err := ec.Wrapper.StateLastValidatorSetCheckpoint(opts)
+	if err != nil {
+		return [32]byte{}, err
+	}
+	return checkpoint, nil
+}
+
 func (ec *Client) WaitForTransaction(
 	ctx context.Context,
 	backend bind.DeployBackend,

orchestrator/broadcaster.go

@@ -29,3 +29,10 @@ func (b Broadcaster) ProvideValsetConfirm(ctx context.Context, nonce uint64, con
 	}
 	return b.BlobstreamDHT.PutValsetConfirm(ctx, p2p.GetValsetConfirmKey(nonce, confirm.EthAddress, signBytes), confirm)
 }
+
+func (b Broadcaster) ProvideLatestValset(ctx context.Context, latestValset types.LatestValset) error {
+	if len(b.BlobstreamDHT.RoutingTable().ListPeers()) == 0 {
+		return ErrEmptyPeersTable
+	}
+	return b.BlobstreamDHT.PutLatestValset(ctx, latestValset)
+}

orchestrator/broadcaster_test.go

@@ -5,6 +5,9 @@ import (
 	"encoding/hex"
 	"math/big"
 	"testing"
+	"time"
+
+	celestiatypes "github.com/celestiaorg/celestia-app/x/qgb/types"
 
 	"github.com/celestiaorg/orchestrator-relayer/evm"
 	"github.com/ethereum/go-ethereum/accounts/keystore"
@@ -97,6 +100,39 @@ func TestBroadcastValsetConfirm(t *testing.T) {
 	assert.Equal(t, *expectedConfirm, actualConfirm)
 }
 
+func TestBroadcastLatestValset(t *testing.T) {
+	network := blobstreamtesting.NewDHTNetwork(context.Background(), 4)
+	defer network.Stop()
+
+	// create a test Valset
+	expectedValset := celestiatypes.Valset{
+		Time:   time.UnixMicro(10),
+		Height: 5,
+		Members: []celestiatypes.BridgeValidator{
+			{
+				Power:      100,
+				EvmAddress: "evm_addr1",
+			},
+			{
+				Power:      200,
+				EvmAddress: "evm_addr2",
+			},
+		},
+	}
+
+	// Broadcast the valset
+	broadcaster := orchestrator.NewBroadcaster(network.DHTs[1])
+	err := broadcaster.ProvideLatestValset(context.Background(), *types.ToLatestValset(expectedValset))
+	assert.NoError(t, err)
+
+	// try to get the valset from another peer
+	actualValset, err := network.DHTs[3].GetLatestValset(context.Background())
+	assert.NoError(t, err)
+	assert.NotNil(t, actualValset)
+
+	assert.True(t, types.IsValsetEqualToLatestValset(expectedValset, actualValset))
+}
+
 // TestEmptyPeersTable tests that values are not broadcasted if the DHT peers
 // table is empty.
 func TestEmptyPeersTable(t *testing.T) {

orchestrator/orchestrator.go

@@ -284,6 +284,12 @@ func (orch Orchestrator) Process(ctx context.Context, nonce uint64) error {
 			orch.Logger.Debug("validator not part of valset. won't sign", "nonce", nonce)
 			return nil
 		}
+
+		if err == nil && previousValset != nil {
+			// add the valset to the p2p network
+			// it's alright if this fails, we can expect other nodes to do it successfully
+			_ = orch.Broadcaster.ProvideLatestValset(ctx, *types.ToLatestValset(*previousValset))
+		}
 	}
 
 	switch castedAtt := att.(type) {
@@ -341,6 +347,10 @@ func (orch Orchestrator) Process(ctx context.Context, nonce uint64) error {
 }
 
 func (orch Orchestrator) ProcessValsetEvent(ctx context.Context, valset celestiatypes.Valset) error {
+	// add the valset to the p2p network
+	// it's alright if this fails, we can expect other nodes to do it successfully
+	_ = orch.Broadcaster.ProvideLatestValset(ctx, *types.ToLatestValset(valset))
+
 	signBytes, err := valset.SignBytes()
 	if err != nil {
 		return err

orchestrator/orchestrator_test.go

@@ -79,6 +79,32 @@ func (s *OrchestratorTestSuite) TestProcessValsetEvent() {
 	assert.Equal(t, s.Orchestrator.EvmAccount.Address.Hex(), confirm.EthAddress)
 }
 
+func (s *OrchestratorTestSuite) TestProcessValsetEventAndProvideValset() {
+	t := s.T()
+	_, err := s.Node.CelestiaNetwork.WaitForHeight(50)
+	require.NoError(t, err)
+
+	vs, err := celestiatypes.NewValset(
+		2,
+		10,
+		[]*celestiatypes.InternalBridgeValidator{{
+			Power:      10,
+			EVMAddress: s.Orchestrator.EvmAccount.Address,
+		}},
+		time.Now(),
+	)
+	require.NoError(t, err)
+
+	// signing and submitting the signature and also providing the valset event
+	err = s.Orchestrator.ProcessValsetEvent(s.Node.Context, *vs)
+	require.NoError(t, err)
+
+	// retrieving the valset
+	actualVs, err := s.Node.DHTNetwork.DHTs[0].GetLatestValset(s.Node.Context)
+	require.NoError(t, err)
+	assert.Equal(t, vs.Nonce, actualVs.Nonce)
+}
+
 func TestValidatorPartOfValset(t *testing.T) {
 	tests := []struct {
 		name           string

p2p/dht.go

@@ -4,10 +4,11 @@ import (
 	"context"
 	"time"
 
+	"github.com/libp2p/go-libp2p-kad-dht/providers"
+
 	"github.com/celestiaorg/orchestrator-relayer/types"
 	ds "github.com/ipfs/go-datastore"
 	dht "github.com/libp2p/go-libp2p-kad-dht"
-	"github.com/libp2p/go-libp2p-kad-dht/providers"
 	"github.com/libp2p/go-libp2p/core/host"
 	"github.com/libp2p/go-libp2p/core/peer"
 	tmlog "github.com/tendermint/tendermint/libs/log"
@@ -17,6 +18,7 @@ const (
 	ProtocolPrefix                 = "/blobstream/0.1.0" // TODO "/blobstream/<version>" ?
 	DataCommitmentConfirmNamespace = "dcc"
 	ValsetConfirmNamespace         = "vc"
+	LatestValsetNamespace          = "lv"
 )
 
 // BlobstreamDHT wrapper around the `IpfsDHT` implementation.
@@ -29,9 +31,9 @@ type BlobstreamDHT struct {
 // NewBlobstreamDHT create a new IPFS DHT using a suitable configuration for the Blobstream.
 // If nil is passed for bootstrappers, the DHT will not try to connect to any existing peer.
 func NewBlobstreamDHT(ctx context.Context, h host.Host, store ds.Batching, bootstrappers []peer.AddrInfo, logger tmlog.Logger) (*BlobstreamDHT, error) {
-	// this value is set to 23 days, which is the unbonding period.
-	// we want to have the signatures available for this whole period.
-	providers.ProvideValidity = time.Hour * 24 * 23
+	// this values is set to a year, so that even in super-stable networks, we have at least
+	// one valset in store for a year.
+	providers.ProvideValidity = time.Hour * 24 * 365
 
 	router, err := dht.New(
 		ctx,
@@ -41,6 +43,7 @@ func NewBlobstreamDHT(ctx context.Context, h host.Host, store ds.Batching, boots
 		dht.ProtocolPrefix(ProtocolPrefix),
 		dht.NamespacedValidator(DataCommitmentConfirmNamespace, DataCommitmentConfirmValidator{}),
 		dht.NamespacedValidator(ValsetConfirmNamespace, ValsetConfirmValidator{}),
+		dht.NamespacedValidator(LatestValsetNamespace, LatestValsetValidator{}),
 		dht.BootstrapPeers(bootstrappers...),
 		dht.DisableProviders(),
 	)
@@ -157,3 +160,34 @@ func (q BlobstreamDHT) GetValsetConfirm(ctx context.Context, key string) (types.
 	}
 	return confirm, nil
 }
+
+// PutLatestValset encodes a valset then puts its value to the DHT.
+// The key will be returned by the `GetValsetKey` method.
+// If the valset is not the latest, it will fail.
+// Returns an error if it fails.
+func (q BlobstreamDHT) PutLatestValset(ctx context.Context, v types.LatestValset) error {
+	encodedData, err := types.MarshalLatestValset(v)
+	if err != nil {
+		return err
+	}
+	err = q.PutValue(ctx, GetLatestValsetKey(), encodedData)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+// GetLatestValset looks for the latest valset in the DHT.
+// The key will be returned by the `GetValsetKey` method.
+// Returns an error if it fails.
+func (q BlobstreamDHT) GetLatestValset(ctx context.Context) (types.LatestValset, error) {
+	encoded, err := q.GetValue(ctx, GetLatestValsetKey()) // this is a blocking call, we should probably use timeout and channel
+	if err != nil {
+		return types.LatestValset{}, err
+	}
+	valset, err := types.UnmarshalLatestValset(encoded)
+	if err != nil {
+		return types.LatestValset{}, err
+	}
+	return valset, nil
+}

p2p/dht_test.go

@@ -7,6 +7,8 @@ import (
 	"testing"
 	"time"
 
+	celestiatypes "github.com/celestiaorg/celestia-app/x/qgb/types"
+
 	"github.com/celestiaorg/orchestrator-relayer/evm"
 	"github.com/ethereum/go-ethereum/accounts/keystore"
 
@@ -95,6 +97,109 @@ func TestPutDataCommitmentConfirm(t *testing.T) {
 	assert.Equal(t, expectedConfirm, actualConfirm)
 }
 
+func TestPutLatestValset(t *testing.T) {
+	network := blobstreamtesting.NewDHTNetwork(context.Background(), 2)
+	defer network.Stop()
+
+	// create a test Valset
+	expectedValset := celestiatypes.Valset{
+		Nonce:  10,
+		Time:   time.UnixMicro(10),
+		Height: 5,
+		Members: []celestiatypes.BridgeValidator{
+			{
+				Power:      100,
+				EvmAddress: "evm_addr1",
+			},
+			{
+				Power:      200,
+				EvmAddress: "evm_addr2",
+			},
+		},
+	}
+
+	// put the test Valset in the DHT
+	err := network.DHTs[0].PutLatestValset(context.Background(), *types.ToLatestValset(expectedValset))
+	assert.NoError(t, err)
+
+	// try to get the latest valset from the same peer
+	actualValset, err := network.DHTs[0].GetLatestValset(context.Background())
+	assert.NoError(t, err)
+	assert.NotNil(t, actualValset)
+
+	assert.True(t, types.IsValsetEqualToLatestValset(expectedValset, actualValset))
+}
+
+func TestPutMultipleLatestValset(t *testing.T) {
+	network := blobstreamtesting.NewDHTNetwork(context.Background(), 3)
+	defer network.Stop()
+
+	// create test Valsets
+	valset1 := celestiatypes.Valset{
+		Nonce:  10,
+		Time:   time.UnixMicro(10),
+		Height: 5,
+		Members: []celestiatypes.BridgeValidator{
+			{
+				Power:      100,
+				EvmAddress: "evm_addr1",
+			},
+			{
+				Power:      200,
+				EvmAddress: "evm_addr2",
+			},
+		},
+	}
+	valset2 := celestiatypes.Valset{
+		Nonce:  11,
+		Time:   time.UnixMicro(10),
+		Height: 5,
+		Members: []celestiatypes.BridgeValidator{
+			{
+				Power:      100,
+				EvmAddress: "evm_addr1",
+			},
+			{
+				Power:      200,
+				EvmAddress: "evm_addr2",
+			},
+		},
+	}
+	valset3 := celestiatypes.Valset{
+		Nonce:  9,
+		Time:   time.UnixMicro(10),
+		Height: 5,
+		Members: []celestiatypes.BridgeValidator{
+			{
+				Power:      100,
+				EvmAddress: "evm_addr1",
+			},
+			{
+				Power:      200,
+				EvmAddress: "evm_addr2",
+			},
+		},
+	}
+
+	// put the valsets in the DHT
+	err := network.DHTs[0].PutLatestValset(context.Background(), *types.ToLatestValset(valset1))
+	assert.NoError(t, err)
+
+	err = network.DHTs[1].PutLatestValset(context.Background(), *types.ToLatestValset(valset2))
+	assert.NoError(t, err)
+
+	// this one should fail since it puts an older valset than ones in store
+	err = network.DHTs[2].PutLatestValset(context.Background(), *types.ToLatestValset(valset3))
+	assert.Error(t, err)
+
+	// try to get the valset from the same peer
+	actualValset, err := network.DHTs[0].GetLatestValset(context.Background())
+	assert.NoError(t, err)
+	assert.NotNil(t, actualValset)
+
+	assert.True(t, types.IsValsetEqualToLatestValset(valset2, actualValset))
+}
+
 func TestNetworkPutDataCommitmentConfirm(t *testing.T) {
 	network := blobstreamtesting.NewDHTNetwork(context.Background(), 10)
 	defer network.Stop()

p2p/errors.go

@@ -1,6 +1,8 @@
 package p2p
 
-import "errors"
+import (
+	"errors"
+)
 
 var (
 	ErrPeersTimeout                    = errors.New("timeout while waiting for peers")
@@ -17,4 +19,6 @@ var (
 	ErrEmptyNamespace                  = errors.New("empty namespace")
 	ErrEmptyEVMAddr                    = errors.New("empty evm address")
 	ErrEmptyDigest                     = errors.New("empty digest")
+	ErrEmptyValset                     = errors.New("empty valset")
+	ErrInvalidLatestValsetKey          = errors.New("invalid latest valset key")
 )

p2p/keys.go

@@ -32,6 +32,11 @@ func GetValsetConfirmKey(nonce uint64, evmAddr string, signBytes string) string
 		evmAddr + ":" + signBytes
 }
 
+// GetLatestValsetKey creates the latest valset key.
+func GetLatestValsetKey() string {
+	return "/" + LatestValsetNamespace + "/latest"
+}
+
 // ParseKey parses a key and returns its fields.
 // Will return an error if the key is missing some fields, some fields are empty, or otherwise invalid.
 func ParseKey(key string) (namespace string, nonce uint64, evmAddr string, digest string, err error) {

p2p/querier.go

@@ -318,3 +318,14 @@ func (q Querier) QueryValsetConfirms(ctx context.Context, nonce uint64, valset c
 	}
 	return confirms, nil
 }
+
+// QueryLatestValset get the latest valset from the p2p network.
+func (q Querier) QueryLatestValset(
+	ctx context.Context,
+) (*types.LatestValset, error) {
+	latestValset, err := q.BlobstreamDHT.GetLatestValset(ctx)
+	if err != nil {
+		return nil, err
+	}
+	return &latestValset, nil
+}