WIP: Raft consensus for lotus nodes in a cluster

node/impl/full.go

@@ -5,6 +5,7 @@ import (
 	"time"
 
 	logging "github.com/ipfs/go-log/v2"
+	consensus "github.com/libp2p/go-libp2p-consensus"
 	"github.com/libp2p/go-libp2p/core/peer"
 
 	"github.com/filecoin-project/lotus/api"
@@ -117,4 +118,8 @@ func (n *FullNodeAPI) NodeStatus(ctx context.Context, inclChainStatus bool) (sta
 	return status, nil
 }
 
+func (n *FullNodeAPI) RaftState(ctx context.Context) (consensus.State, error) {
+	return n.MpoolAPI.GetRaftState(ctx)
+}
+
 var _ api.FullNode = &FullNodeAPI{}

node/impl/full/mpool.go

@@ -5,6 +5,8 @@ import (
 	"encoding/json"
 
 	"github.com/ipfs/go-cid"
+	consensus "github.com/libp2p/go-libp2p-consensus"
+	"github.com/libp2p/go-libp2p/core/peer"
 	"go.uber.org/fx"
 	"golang.org/x/xerrors"
 
@@ -41,9 +43,11 @@ type MpoolAPI struct {
 	MpoolModuleAPI
 
 	WalletAPI
+
 	GasAPI
 
-	MessageSigner *messagesigner.MessageSigner
+	MessageSigner messagesigner.MsgSigner
+	//	MessageSigner *messagesigner.MessageSigner
 
 	PushLocks *dtypes.MpoolLocker
 }
@@ -142,6 +146,20 @@ func (a *MpoolAPI) MpoolPushMessage(ctx context.Context, msg *types.Message, spe
 	msg = &cp
 	inMsg := *msg
 
+	// Redirect to leader if current node is not leader. A single non raft based node is always the leader
+	if !a.MessageSigner.IsLeader(ctx) {
+		var signedMsg types.SignedMessage
+		redirected, err := a.MessageSigner.RedirectToLeader(ctx, "MpoolPushMessage", api.MpoolMessageWhole{msg, spec}, &signedMsg)
+		if err != nil {
+			return nil, err
+		}
+		// It's possible that the current node became the leader between the check and the redirect
+		// In that case, continue with rest of execution and only return signedMsg if something was redirected
+		if redirected {
+			return &signedMsg, nil
+		}
+	}
+
 	// Check if this uuid has already been processed
 	if spec != nil {
 		signedMessage, err := a.MessageSigner.GetSignedMessage(ctx, spec.MsgUuid)
@@ -195,7 +213,7 @@ func (a *MpoolAPI) MpoolPushMessage(ctx context.Context, msg *types.Message, spe
 	}
 
 	// Sign and push the message
-	signedMsg, err := a.MessageSigner.SignMessage(ctx, msg, func(smsg *types.SignedMessage) error {
+	signedMsg, err := a.MessageSigner.SignMessage(ctx, msg, spec, func(smsg *types.SignedMessage) error {
 		if _, err := a.MpoolModuleAPI.MpoolPush(ctx, smsg); err != nil {
 			return xerrors.Errorf("mpool push: failed to push message: %w", err)
 		}
@@ -271,3 +289,11 @@ func (a *MpoolAPI) MpoolGetNonce(ctx context.Context, addr address.Address) (uin
 func (a *MpoolAPI) MpoolSub(ctx context.Context) (<-chan api.MpoolUpdate, error) {
 	return a.Mpool.Updates(ctx)
 }
+
+func (a *MpoolAPI) GetRaftState(ctx context.Context) (consensus.State, error) {
+	return a.MessageSigner.GetRaftState(ctx)
+}
+
+func (a *MpoolAPI) RaftLeader(ctx context.Context) (peer.ID, error) {
+	return a.MessageSigner.RaftLeader(ctx)
+}