Raft consensus for lotus nodes in a cluster

.circleci/config.yml

@@ -915,6 +915,11 @@ workflows:
           suite: itest-mpool_msg_uuid
           target: "./itests/mpool_msg_uuid_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
@@ -955,6 +960,11 @@ workflows:
           suite: itest-pending_deal_allocation
           target: "./itests/pending_deal_allocation_test.go"
 
+      - test:
+          name: test-itest-raft_messagesigner
+          suite: itest-raft_messagesigner
+          target: "./itests/raft_messagesigner_test.go"
+
       - test:
           name: test-itest-remove_verifreg_datacap
           suite: itest-remove_verifreg_datacap

api/api_full.go

@@ -764,6 +764,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) (*RaftStateData, error) //perm:read
+	RaftLeader(ctx context.Context) (peer.ID, error)       //perm:read
 }
 
 type StorageAsk struct {

api/docgen/docgen.go

@@ -349,6 +349,10 @@ func init() {
 	addExample(map[string]bitfield.BitField{
 		"": bitfield.NewFromSet([]uint64{5, 6, 7, 10}),
 	})
+	addExample(&api.RaftStateData{
+		NonceMap: make(map[address.Address]uint64),
+		MsgUuids: make(map[uuid.UUID]*types.SignedMessage),
+	})
 
 	addExample(http.Header{
 		"Authorization": []string{"Bearer ey.."},

api/types.go

@@ -59,6 +59,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
@@ -334,3 +339,61 @@ type ForkUpgradeParams struct {
 	UpgradeSkyrHeight          abi.ChainEpoch
 	UpgradeSharkHeight         abi.ChainEpoch
 }
+
+type NonceMapType map[address.Address]uint64
+type MsgUuidMapType map[uuid.UUID]*types.SignedMessage
+
+type RaftStateData struct {
+	NonceMap NonceMapType
+	MsgUuids MsgUuidMapType
+}
+
+func (n *NonceMapType) MarshalJSON() ([]byte, error) {
+	marshalled := make(map[string]uint64)
+	for a, n := range *n {
+		marshalled[a.String()] = n
+	}
+	return json.Marshal(marshalled)
+}
+
+func (n *NonceMapType) UnmarshalJSON(b []byte) error {
+	unmarshalled := make(map[string]uint64)
+	err := json.Unmarshal(b, &unmarshalled)
+	if err != nil {
+		return err
+	}
+	*n = make(map[address.Address]uint64)
+	for saddr, nonce := range unmarshalled {
+		a, err := address.NewFromString(saddr)
+		if err != nil {
+			return err
+		}
+		(*n)[a] = nonce
+	}
+	return nil
+}
+
+func (m *MsgUuidMapType) MarshalJSON() ([]byte, error) {
+	marshalled := make(map[string]*types.SignedMessage)
+	for u, msg := range *m {
+		marshalled[u.String()] = msg
+	}
+	return json.Marshal(marshalled)
+}
+
+func (m *MsgUuidMapType) UnmarshalJSON(b []byte) error {
+	unmarshalled := make(map[string]*types.SignedMessage)
+	err := json.Unmarshal(b, &unmarshalled)
+	if err != nil {
+		return err
+	}
+	*m = make(map[uuid.UUID]*types.SignedMessage)
+	for suid, msg := range unmarshalled {
+		u, err := uuid.Parse(suid)
+		if err != nil {
+			return err
+		}
+		(*m)[u] = msg
+	}
+	return nil
+}

chain/messagepool/messagepool.go

@@ -82,6 +82,7 @@ var (
 	ErrRBFTooLowPremium       = errors.New("replace by fee has too low GasPremium")
 	ErrTooManyPendingMessages = errors.New("too many pending messages for actor")
 	ErrNonceGap               = errors.New("unfulfilled nonce gap")
+	ErrExistingNonce          = errors.New("message with nonce already exists")
 )
 
 const (
@@ -276,7 +277,7 @@ func (ms *msgSet) add(m *types.SignedMessage, mp *MessagePool, strict, untrusted
 			}
 		} else {
 			return false, xerrors.Errorf("message from %s with nonce %d already in mpool: %w",
-				m.Message.From, m.Message.Nonce, ErrSoftValidationFailure)
+				m.Message.From, m.Message.Nonce, ErrExistingNonce)
 		}
 
 		ms.requiredFunds.Sub(ms.requiredFunds, exms.Message.RequiredFunds().Int)
@@ -667,7 +668,7 @@ func (mp *MessagePool) verifyMsgBeforeAdd(ctx context.Context, m *types.SignedMe
 	return publish, nil
 }
 
-func (mp *MessagePool) Push(ctx context.Context, m *types.SignedMessage) (cid.Cid, error) {
+func (mp *MessagePool) Push(ctx context.Context, m *types.SignedMessage, publish bool) (cid.Cid, error) {
 	done := metrics.Timer(ctx, metrics.MpoolPushDuration)
 	defer done()
 
@@ -683,14 +684,14 @@ func (mp *MessagePool) Push(ctx context.Context, m *types.SignedMessage) (cid.Ci
 	}()
 
 	mp.curTsLk.Lock()
-	publish, err := mp.addTs(ctx, m, mp.curTs, true, false)
+	ok, err := mp.addTs(ctx, m, mp.curTs, true, false)
 	if err != nil {
 		mp.curTsLk.Unlock()
 		return cid.Undef, err
 	}
 	mp.curTsLk.Unlock()
 
-	if publish {
+	if ok && publish {
 		msgb, err := m.Serialize()
 		if err != nil {
 			return cid.Undef, xerrors.Errorf("error serializing message: %w", err)
@@ -1583,3 +1584,8 @@ func getBaseFeeLowerBound(baseFee, factor types.BigInt) types.BigInt {
 
 	return baseFeeLowerBound
 }
+
+type MpoolNonceAPI interface {
+	GetNonce(context.Context, address.Address, types.TipSetKey) (uint64, error)
+	GetActor(context.Context, address.Address, types.TipSetKey) (*types.Actor, error)
+}

chain/messagepool/messagepool_test.go

@@ -545,7 +545,7 @@ func TestLoadLocal(t *testing.T) {
 	for i := 0; i < 10; i++ {
 		m := makeTestMessage(w1, a1, a2, uint64(i), gasLimit, uint64(i+1))
 		//stm: @CHAIN_MEMPOOL_PUSH_001
-		cid, err := mp.Push(context.TODO(), m)
+		cid, err := mp.Push(context.TODO(), m, true)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -618,7 +618,7 @@ func TestClearAll(t *testing.T) {
 	for i := 0; i < 10; i++ {
 		m := makeTestMessage(w1, a1, a2, uint64(i), gasLimit, uint64(i+1))
 		//stm: @CHAIN_MEMPOOL_PUSH_001
-		_, err := mp.Push(context.TODO(), m)
+		_, err := mp.Push(context.TODO(), m, true)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -676,7 +676,7 @@ func TestClearNonLocal(t *testing.T) {
 	for i := 0; i < 10; i++ {
 		m := makeTestMessage(w1, a1, a2, uint64(i), gasLimit, uint64(i+1))
 		//stm: @CHAIN_MEMPOOL_PUSH_001
-		_, err := mp.Push(context.TODO(), m)
+		_, err := mp.Push(context.TODO(), m, true)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -749,7 +749,7 @@ func TestUpdates(t *testing.T) {
 	for i := 0; i < 10; i++ {
 		m := makeTestMessage(w1, a1, a2, uint64(i), gasLimit, uint64(i+1))
 		//stm: @CHAIN_MEMPOOL_PUSH_001
-		_, err := mp.Push(context.TODO(), m)
+		_, err := mp.Push(context.TODO(), m, true)
 		if err != nil {
 			t.Fatal(err)
 		}

chain/messagepool/provider.go

@@ -12,7 +12,6 @@ import (
 	"github.com/filecoin-project/go-state-types/abi"
 	"github.com/filecoin-project/go-state-types/network"
 
-	"github.com/filecoin-project/lotus/chain/messagesigner"
 	"github.com/filecoin-project/lotus/chain/stmgr"
 	"github.com/filecoin-project/lotus/chain/store"
 	"github.com/filecoin-project/lotus/chain/types"
@@ -42,7 +41,7 @@ type mpoolProvider struct {
 	sm *stmgr.StateManager
 	ps *pubsub.PubSub
 
-	lite messagesigner.MpoolNonceAPI
+	lite MpoolNonceAPI
 }
 
 var _ Provider = (*mpoolProvider)(nil)
@@ -51,7 +50,7 @@ func NewProvider(sm *stmgr.StateManager, ps *pubsub.PubSub) Provider {
 	return &mpoolProvider{sm: sm, ps: ps}
 }
 
-func NewProviderLite(sm *stmgr.StateManager, ps *pubsub.PubSub, noncer messagesigner.MpoolNonceAPI) Provider {
+func NewProviderLite(sm *stmgr.StateManager, ps *pubsub.PubSub, noncer MpoolNonceAPI) Provider {
 	return &mpoolProvider{sm: sm, ps: ps, lite: noncer}
 }
 

chain/messagepool/repub_test.go

@@ -60,7 +60,7 @@ func TestRepubMessages(t *testing.T) {
 	for i := 0; i < 10; i++ {
 		m := makeTestMessage(w1, a1, a2, uint64(i), gasLimit, uint64(i+1))
 		//stm: @CHAIN_MEMPOOL_PUSH_001
-		_, err := mp.Push(context.TODO(), m)
+		_, err := mp.Push(context.TODO(), m, true)
 		if err != nil {
 			t.Fatal(err)
 		}