Raft consensus for lotus nodes in a cluster

.circleci/config.yml

@@ -924,6 +924,11 @@ workflows:
           suite: itest-mempool
           target: "./itests/mempool_test.go"
 
+      - test:
+          name: test-itest-mpool_push_with_uuid
+          suite: itest-mpool_push_with_uuid
+          target: "./itests/mpool_push_with_uuid_test.go"
+
       - test:
           name: test-itest-multisig
           suite: itest-multisig
@@ -959,6 +964,11 @@ workflows:
           suite: itest-paych_cli
           target: "./itests/paych_cli_test.go"
 
+      - test:
+          name: test-itest-raft_messagesigner
+          suite: itest-raft_messagesigner
+          target: "./itests/raft_messagesigner_test.go"
+
       - test:
           name: test-itest-sdr_upgrade
           suite: itest-sdr_upgrade

api/api_full.go

@@ -8,6 +8,7 @@ import (
 
 	blocks "github.com/ipfs/go-block-format"
 	"github.com/ipfs/go-cid"
+	consensus "github.com/libp2p/go-libp2p-consensus"
 	"github.com/libp2p/go-libp2p/core/peer"
 
 	"github.com/filecoin-project/go-address"
@@ -751,6 +752,9 @@ type FullNode interface {
 	// LOTUS_BACKUP_BASE_PATH environment variable set to some path, and that
 	// the path specified when calling CreateBackup is within the base path
 	CreateBackup(ctx context.Context, fpath string) error //perm:admin
+
+	RaftState(ctx context.Context) (consensus.State, error) //perm:read
+	RaftLeader(ctx context.Context) (peer.ID, error)        //perm:read
 }
 
 type StorageAsk struct {

api/types.go

@@ -58,6 +58,11 @@ type MessageSendSpec struct {
 	MsgUuid uuid.UUID
 }
 
+type MpoolMessageWhole struct {
+	Msg  *types.Message
+	Spec *MessageSendSpec
+}
+
 // GraphSyncDataTransfer provides diagnostics on a data transfer happening over graphsync
 type GraphSyncDataTransfer struct {
 	// GraphSync request id for this transfer

chain/messagesigner/messagesigner.go

@@ -9,6 +9,8 @@ import (
 	"github.com/ipfs/go-datastore"
 	"github.com/ipfs/go-datastore/namespace"
 	logging "github.com/ipfs/go-log/v2"
+	consensus "github.com/libp2p/go-libp2p-consensus"
+	"github.com/libp2p/go-libp2p/core/peer"
 	cbg "github.com/whyrusleeping/cbor-gen"
 	"golang.org/x/xerrors"
 
@@ -29,6 +31,19 @@ type MpoolNonceAPI interface {
 	GetActor(context.Context, address.Address, types.TipSetKey) (*types.Actor, error)
 }
 
+type MsgSigner interface {
+	SignMessage(ctx context.Context, msg *types.Message, spec *api.MessageSendSpec, cb func(*types.SignedMessage) error) (*types.SignedMessage, error)
+	GetSignedMessage(ctx context.Context, uuid uuid.UUID) (*types.SignedMessage, error)
+	StoreSignedMessage(ctx context.Context, uuid uuid.UUID, message *types.SignedMessage) error
+	NextNonce(ctx context.Context, addr address.Address) (uint64, error)
+	SaveNonce(ctx context.Context, addr address.Address, nonce uint64) error
+	DstoreKey(addr address.Address) datastore.Key
+	IsLeader(ctx context.Context) bool
+	RaftLeader(ctx context.Context) (peer.ID, error)
+	RedirectToLeader(ctx context.Context, method string, arg interface{}, ret interface{}) (bool, error)
+	GetRaftState(ctx context.Context) (consensus.State, error)
+}
+
 // MessageSigner keeps track of nonces per address, and increments the nonce
 // when signing a message
 type MessageSigner struct {
@@ -38,6 +53,8 @@ type MessageSigner struct {
 	ds     datastore.Batching
 }
 
+//var _ full.MsgSigner = &MessageSigner{}
+
 func NewMessageSigner(wallet api.Wallet, mpool MpoolNonceAPI, ds dtypes.MetadataDS) *MessageSigner {
 	ds = namespace.Wrap(ds, datastore.NewKey("/message-signer/"))
 	return &MessageSigner{
@@ -49,12 +66,12 @@ func NewMessageSigner(wallet api.Wallet, mpool MpoolNonceAPI, ds dtypes.Metadata
 
 // SignMessage increments the nonce for the message From address, and signs
 // the message
-func (ms *MessageSigner) SignMessage(ctx context.Context, msg *types.Message, cb func(*types.SignedMessage) error) (*types.SignedMessage, error) {
+func (ms *MessageSigner) SignMessage(ctx context.Context, msg *types.Message, spec *api.MessageSendSpec, cb func(*types.SignedMessage) error) (*types.SignedMessage, error) {
 	ms.lk.Lock()
 	defer ms.lk.Unlock()
 
 	// Get the next message nonce
-	nonce, err := ms.nextNonce(ctx, msg.From)
+	nonce, err := ms.NextNonce(ctx, msg.From)
 	if err != nil {
 		return nil, xerrors.Errorf("failed to create nonce: %w", err)
 	}
@@ -72,7 +89,7 @@ func (ms *MessageSigner) SignMessage(ctx context.Context, msg *types.Message, cb
 		Extra: mb.RawData(),
 	})
 	if err != nil {
-		return nil, xerrors.Errorf("failed to sign message: %w", err)
+		return nil, xerrors.Errorf("failed to sign message: %w, addr=%s", err, msg.From)
 	}
 
 	// Callback with the signed message
@@ -86,7 +103,7 @@ func (ms *MessageSigner) SignMessage(ctx context.Context, msg *types.Message, cb
 	}
 
 	// If the callback executed successfully, write the nonce to the datastore
-	if err := ms.saveNonce(ctx, msg.From, nonce); err != nil {
+	if err := ms.SaveNonce(ctx, msg.From, nonce); err != nil {
 		return nil, xerrors.Errorf("failed to save nonce: %w", err)
 	}
 
@@ -113,9 +130,9 @@ func (ms *MessageSigner) StoreSignedMessage(ctx context.Context, uuid uuid.UUID,
 	return ms.ds.Put(ctx, key, serializedMsg)
 }
 
-// nextNonce gets the next nonce for the given address.
+// NextNonce gets the next nonce for the given address.
 // If there is no nonce in the datastore, gets the nonce from the message pool.
-func (ms *MessageSigner) nextNonce(ctx context.Context, addr address.Address) (uint64, error) {
+func (ms *MessageSigner) NextNonce(ctx context.Context, addr address.Address) (uint64, error) {
 	// Nonces used to be created by the mempool and we need to support nodes
 	// that have mempool nonces, so first check the mempool for a nonce for
 	// this address. Note that the mempool returns the actor state's nonce
@@ -126,7 +143,7 @@ func (ms *MessageSigner) nextNonce(ctx context.Context, addr address.Address) (u
 	}
 
 	// Get the next nonce for this address from the datastore
-	addrNonceKey := ms.dstoreKey(addr)
+	addrNonceKey := ms.DstoreKey(addr)
 	dsNonceBytes, err := ms.ds.Get(ctx, addrNonceKey)
 
 	switch {
@@ -159,14 +176,14 @@ func (ms *MessageSigner) nextNonce(ctx context.Context, addr address.Address) (u
 	}
 }
 
-// saveNonce increments the nonce for this address and writes it to the
+// SaveNonce increments the nonce for this address and writes it to the
 // datastore
-func (ms *MessageSigner) saveNonce(ctx context.Context, addr address.Address, nonce uint64) error {
+func (ms *MessageSigner) SaveNonce(ctx context.Context, addr address.Address, nonce uint64) error {
 	// Increment the nonce
 	nonce++
 
 	// Write the nonce to the datastore
-	addrNonceKey := ms.dstoreKey(addr)
+	addrNonceKey := ms.DstoreKey(addr)
 	buf := bytes.Buffer{}
 	_, err := buf.Write(cbg.CborEncodeMajorType(cbg.MajUnsignedInt, nonce))
 	if err != nil {
@@ -179,6 +196,22 @@ func (ms *MessageSigner) saveNonce(ctx context.Context, addr address.Address, no
 	return nil
 }
 
-func (ms *MessageSigner) dstoreKey(addr address.Address) datastore.Key {
+func (ms *MessageSigner) DstoreKey(addr address.Address) datastore.Key {
 	return datastore.KeyWithNamespaces([]string{dsKeyActorNonce, addr.String()})
 }
+
+func (ms *MessageSigner) IsLeader(ctx context.Context) bool {
+	return true
+}
+
+func (ms *MessageSigner) RedirectToLeader(ctx context.Context, method string, arg interface{}, ret interface{}) (bool, error) {
+	return false, xerrors.Errorf("Single node shouldn't have any redirects")
+}
+
+func (ms *MessageSigner) GetRaftState(ctx context.Context) (consensus.State, error) {
+	return nil, xerrors.Errorf("This is a non raft consensus message signer")
+}
+
+func (ms *MessageSigner) RaftLeader(ctx context.Context) (peer.ID, error) {
+	return "", xerrors.Errorf("No leaders in non raft message signer")
+}

chain/messagesigner/messagesigner_consensus.go

@@ -0,0 +1,99 @@
+package messagesigner
+
+import (
+	"context"
+
+	"github.com/google/uuid"
+	"github.com/ipfs/go-datastore"
+	"github.com/ipfs/go-datastore/namespace"
+	libp2pconsensus "github.com/libp2p/go-libp2p-consensus"
+	"github.com/libp2p/go-libp2p/core/peer"
+	"golang.org/x/xerrors"
+
+	"github.com/filecoin-project/lotus/api"
+	"github.com/filecoin-project/lotus/chain/types"
+	consensus "github.com/filecoin-project/lotus/lib/consensus/raft"
+	"github.com/filecoin-project/lotus/node/modules/dtypes"
+)
+
+type MessageSignerConsensus struct {
+	MsgSigner
+	consensus *consensus.Consensus
+}
+
+func NewMessageSignerConsensus(
+	wallet api.Wallet,
+	mpool MpoolNonceAPI,
+	ds dtypes.MetadataDS,
+	consensus *consensus.Consensus) *MessageSignerConsensus {
+
+	ds = namespace.Wrap(ds, datastore.NewKey("/message-signer-consensus/"))
+	return &MessageSignerConsensus{
+		MsgSigner: &MessageSigner{
+			wallet: wallet,
+			mpool:  mpool,
+			ds:     ds,
+		},
+		consensus: consensus,
+	}
+}
+
+func (ms *MessageSignerConsensus) IsLeader(ctx context.Context) bool {
+	return ms.consensus.IsLeader(ctx)
+}
+
+func (ms *MessageSignerConsensus) RedirectToLeader(ctx context.Context, method string, arg interface{}, ret interface{}) (bool, error) {
+	ok, err := ms.consensus.RedirectToLeader(method, arg, ret.(*types.SignedMessage))
+	if err != nil {
+		return ok, err
+	}
+	return ok, nil
+}
+
+func (ms *MessageSignerConsensus) SignMessage(
+	ctx context.Context,
+	msg *types.Message,
+	spec *api.MessageSendSpec,
+	cb func(*types.SignedMessage) error) (*types.SignedMessage, error) {
+
+	signedMsg, err := ms.MsgSigner.SignMessage(ctx, msg, spec, cb)
+	if err != nil {
+		return nil, err
+	}
+
+	// We can't have an empty/default uuid as part of the consensus state so generate a new uuid if spec is empty
+	u := uuid.New()
+	if spec != nil {
+		u = spec.MsgUuid
+	}
+
+	op := &consensus.ConsensusOp{signedMsg.Message.Nonce, u, signedMsg.Message.From, signedMsg}
+	err = ms.consensus.Commit(ctx, op)
+	if err != nil {
+		return nil, err
+	}
+
+	return signedMsg, nil
+}
+
+func (ms *MessageSignerConsensus) GetSignedMessage(ctx context.Context, uuid uuid.UUID) (*types.SignedMessage, error) {
+	state, err := ms.consensus.State(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	cstate := state.(consensus.RaftState)
+	msg, ok := cstate.MsgUuids[uuid]
+	if !ok {
+		return nil, xerrors.Errorf("Msg with Uuid %s not available", uuid)
+	}
+	return msg, nil
+}
+
+func (ms *MessageSignerConsensus) GetRaftState(ctx context.Context) (libp2pconsensus.State, error) {
+	return ms.consensus.State(ctx)
+}
+
+func (ms *MessageSignerConsensus) RaftLeader(ctx context.Context) (peer.ID, error) {
+	return ms.consensus.Leader(ctx)
+}

chain/messagesigner/messagesigner_test.go

@@ -187,7 +187,7 @@ func TestMessageSignerSignMessage(t *testing.T) {
 					mpool.setNonce(m.msg.From, m.mpoolNonce[0])
 				}
 				merr := m.cbErr
-				smsg, err := ms.SignMessage(ctx, m.msg, func(message *types.SignedMessage) error {
+				smsg, err := ms.SignMessage(ctx, m.msg, nil, func(message *types.SignedMessage) error {
 					return merr
 				})