test: make StateChangeRequest broadcast-able (#305)

Make `test.StateChangeRequest` behaves like tx
on ethereum:
- Can be broadcasted and cached in a pool.
- Uniquely indexed, and be removed after
  applied.

Changes:
- Make cloneDKGx functions in test.State as
  utilities.
- Add hash and timestamp fields to
  test.StateChangeRequest.
- Add two methods to test.State:
   - PackOwnRequests would pack all pending
     change requests owned by this instance as
     byte slice, and move them to global pending
     requests pool.
   - AddRequestsFromOthers would add pending
     change requests from others to global
     pending requests pool.
- The method State.PackRequests now would pack
  requests in global pending requests pool.
- The method State.Apply would remove
  corresponding StateChangeRequest by hash.

5 files changed, 445 insertions, 278 deletions

diff --git a/core/test/state-change-request.go b/core/test/state-change-request.go
new file mode 100644
index 0000000..0f49db0
--- /dev/null
+++ b/core/test/state-change-request.go
@@ -0,0 +1,135 @@
+// Copyright 2018 The dexon-consensus Authors
+// This file is part of the dexon-consensus library.
+//
+// The dexon-consensus library is free software: you can redistribute it
+// and/or modify it under the terms of the GNU Lesser General Public License as
+// published by the Free Software Foundation, either version 3 of the License,
+// or (at your option) any later version.
+//
+// The dexon-consensus library is distributed in the hope that it will be
+// useful, but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser
+// General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the dexon-consensus library. If not, see
+// <http://www.gnu.org/licenses/>.
+
+package test
+
+import (
+	"time"
+
+	"github.com/dexon-foundation/dexon-consensus/common"
+	"github.com/dexon-foundation/dexon-consensus/core/crypto"
+	typesDKG "github.com/dexon-foundation/dexon-consensus/core/types/dkg"
+	"github.com/dexon-foundation/dexon/rlp"
+)
+
+// StateChangeType is the type of state change request.
+type StateChangeType uint8
+
+// Types of state change.
+const (
+	StateChangeNothing StateChangeType = iota
+	// DKG & CRS
+	StateAddCRS
+	StateAddDKGComplaint
+	StateAddDKGMasterPublicKey
+	StateAddDKGFinal
+	// Configuration related.
+	StateChangeNumChains
+	StateChangeLambdaBA
+	StateChangeLambdaDKG
+	StateChangeRoundInterval
+	StateChangeMinBlockInterval
+	StateChangeMaxBlockInterval
+	StateChangeK
+	StateChangePhiRatio
+	StateChangeNotarySetSize
+	StateChangeDKGSetSize
+	// Node set related.
+	StateAddNode
+)
+
+// StateChangeRequest carries information of state change request.
+type StateChangeRequest struct {
+	Type    StateChangeType `json:"type"`
+	Payload interface{}     `json:"payload"`
+	// The purpose of these fields are aiming to provide an unique ID for each
+	// change request.
+	Hash      common.Hash
+	Timestamp uint64
+}
+
+// this structure is mainly for marshalling for StateChangeRequest.
+type rawStateChangeRequest struct {
+	Type      StateChangeType
+	Payload   rlp.RawValue
+	Hash      common.Hash
+	Timestamp uint64
+}
+
+// NewStateChangeRequest constructs an StateChangeRequest instance.
+func NewStateChangeRequest(
+	t StateChangeType, payload interface{}) *StateChangeRequest {
+	now := uint64(time.Now().UTC().UnixNano())
+	b, err := rlp.EncodeToBytes(struct {
+		Type      StateChangeType
+		Payload   interface{}
+		Timestamp uint64
+	}{t, payload, now})
+	if err != nil {
+		panic(err)
+	}
+	return &StateChangeRequest{
+		Hash:      crypto.Keccak256Hash(b),
+		Type:      t,
+		Payload:   payload,
+		Timestamp: now,
+	}
+}
+
+// Clone a StateChangeRequest instance.
+func (req *StateChangeRequest) Clone() (copied *StateChangeRequest) {
+	copied = &StateChangeRequest{
+		Type:      req.Type,
+		Hash:      req.Hash,
+		Timestamp: req.Timestamp,
+	}
+	// NOTE: The cloned DKGx structs would be different from sources in binary
+	//       level, thus would produce different hash from the source.
+	//       I don't want different hash for source/copied requests thus would
+	//       copy the hash from source directly.
+	switch req.Type {
+	case StateAddNode:
+		srcBytes := req.Payload.([]byte)
+		copiedBytes := make([]byte, len(srcBytes))
+		copy(copiedBytes, srcBytes)
+		req.Payload = copiedBytes
+	case StateAddCRS:
+		crsReq := req.Payload.(*crsAdditionRequest)
+		copied.Payload = &crsAdditionRequest{
+			Round: crsReq.Round,
+			CRS:   crsReq.CRS,
+		}
+	case StateAddDKGFinal:
+		copied.Payload = cloneDKGFinalize(req.Payload.(*typesDKG.Finalize))
+	case StateAddDKGMasterPublicKey:
+		copied.Payload = cloneDKGMasterPublicKey(
+			req.Payload.(*typesDKG.MasterPublicKey))
+	case StateAddDKGComplaint:
+		copied.Payload = cloneDKGComplaint(req.Payload.(*typesDKG.Complaint))
+	default:
+		copied.Payload = req.Payload
+	}
+	return
+}
+
+// Equal checks equality between two StateChangeRequest.
+func (req *StateChangeRequest) Equal(other *StateChangeRequest) error {
+	if req.Hash == other.Hash {
+		return nil
+	}
+	return ErrStatePendingChangesNotEqual
+}
diff --git a/core/test/state-change-request_test.go b/core/test/state-change-request_test.go
new file mode 100644
index 0000000..658a9fb
--- /dev/null
+++ b/core/test/state-change-request_test.go
@@ -0,0 +1,54 @@
+// Copyright 2018 The dexon-consensus Authors
+// This file is part of the dexon-consensus library.
+//
+// The dexon-consensus library is free software: you can redistribute it
+// and/or modify it under the terms of the GNU Lesser General Public License as
+// published by the Free Software Foundation, either version 3 of the License,
+// or (at your option) any later version.
+//
+// The dexon-consensus library is distributed in the hope that it will be
+// useful, but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser
+// General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the dexon-consensus library. If not, see
+// <http://www.gnu.org/licenses/>.
+
+package test
+
+import (
+	"testing"
+
+	typesDKG "github.com/dexon-foundation/dexon-consensus/core/types/dkg"
+	"github.com/stretchr/testify/suite"
+)
+
+type StateChangeRequestTestSuite struct {
+	suite.Suite
+}
+
+func (s *StateChangeRequestTestSuite) TestEqual() {
+	// Basically, only the cloned one would be equal.
+	st00 := NewStateChangeRequest(StateChangeNumChains, uint32(4))
+	st01 := NewStateChangeRequest(StateChangeNumChains, uint32(4))
+	s.Error(ErrStatePendingChangesNotEqual, st00.Equal(st01))
+	// Even with identical payload, they would be different.
+	mKey := typesDKG.NewMasterPublicKey()
+	st10 := NewStateChangeRequest(StateAddDKGMasterPublicKey, mKey)
+	st11 := NewStateChangeRequest(StateAddDKGMasterPublicKey, mKey)
+	s.Error(ErrStatePendingChangesNotEqual, st10.Equal(st11))
+}
+
+func (s *StateChangeRequestTestSuite) TestClone() {
+	// The cloned one should be no error when compared with 'Equal' method.
+	st00 := NewStateChangeRequest(StateChangeNumChains, uint32(7))
+	s.NoError(st00.Equal(st00.Clone()))
+	st10 := NewStateChangeRequest(
+		StateAddDKGMasterPublicKey, typesDKG.NewMasterPublicKey())
+	s.NoError(st10.Equal(st10.Clone()))
+}
+
+func TestStateChangeRequest(t *testing.T) {
+	suite.Run(t, new(StateChangeRequestTestSuite))
+}
diff --git a/core/test/state.go b/core/test/state.go
index 05394ed..2bb04e5 100644
--- a/core/test/state.go
+++ b/core/test/state.go
@@ -21,7 +21,6 @@ import (
 	"bytes"
 	"errors"
 	"math"
-	"reflect"
 	"sync"
 	"time"
 
@@ -33,9 +32,6 @@ import (
 	"github.com/dexon-foundation/dexon/rlp"
 )
 
-// StateChangeType is the type of state change request.
-type StateChangeType uint8
-
 var (
 	// ErrDuplicatedChange means the change request is already applied.
 	ErrDuplicatedChange = errors.New("duplicated change")
@@ -70,29 +66,10 @@ var (
 	// ErrStatePendingChangesNotEqual means pending change requests of two
 	// states are not equal.
 	ErrStatePendingChangesNotEqual = errors.New("pending changes not equal")
-)
-
-// Types of state change.
-const (
-	StateChangeNothing StateChangeType = iota
-	// DKG & CRS
-	StateAddCRS
-	StateAddDKGComplaint
-	StateAddDKGMasterPublicKey
-	StateAddDKGFinal
-	// Configuration related.
-	StateChangeNumChains
-	StateChangeLambdaBA
-	StateChangeLambdaDKG
-	StateChangeRoundInterval
-	StateChangeMinBlockInterval
-	StateChangeMaxBlockInterval
-	StateChangeK
-	StateChangePhiRatio
-	StateChangeNotarySetSize
-	StateChangeDKGSetSize
-	// Node set related.
-	StateAddNode
+	// ErrNotInRemoteMode means callers attempts to call functions for remote
+	// mode when the State instance is still in local mode.
+	ErrNotInRemoteMode = errors.New(
+		"attempting to use remote functions in local mode")
 )
 
 type crsAdditionRequest struct {
@@ -100,17 +77,6 @@ type crsAdditionRequest struct {
 	CRS   common.Hash `json:"crs"`
 }
 
-// StateChangeRequest carries information of state change request.
-type StateChangeRequest struct {
-	Type    StateChangeType `json:"type"`
-	Payload interface{}     `json:"payload"`
-}
-
-type rawStateChangeRequest struct {
-	Type    StateChangeType
-	Payload rlp.RawValue
-}
-
 // State emulates what the global state in governace contract on a fullnode.
 type State struct {
 	// Configuration related.
@@ -134,16 +100,9 @@ type State struct {
 	// Other stuffs
 	local bool
 	lock  sync.RWMutex
-	// ChangeRequest(s) are organized as map, indexed by type of state change.
-	// For each time to apply state change, only the last request would be
-	// applied.
-	pendingChangedConfigs      map[StateChangeType]interface{}
-	pendingNodes               [][]byte
-	pendingDKGComplaints       []*typesDKG.Complaint
-	pendingDKGFinals           []*typesDKG.Finalize
-	pendingDKGMasterPublicKeys []*typesDKG.MasterPublicKey
-	pendingCRS                 []*crsAdditionRequest
-	pendingChangesLock         sync.Mutex
+	// Pending change requests.
+	ownRequests    map[common.Hash]*StateChangeRequest
+	globalRequests map[common.Hash]*StateChangeRequest
 }
 
 // NewState constructs an State instance with genesis information, including:
@@ -157,20 +116,21 @@ func NewState(
 	}
 	genesisCRS := crypto.Keccak256Hash([]byte("__ DEXON"))
 	return &State{
-		local:                 local,
-		numChains:             uint32(len(nodes)),
-		lambdaBA:              lambda,
-		lambdaDKG:             lambda * 10,
-		roundInterval:         lambda * 10000,
-		minBlockInterval:      time.Millisecond * 1,
-		maxBlockInterval:      lambda * 8,
-		crs:                   []common.Hash{genesisCRS},
-		nodes:                 nodes,
-		phiRatio:              0.667,
-		k:                     0,
-		notarySetSize:         uint32(len(nodes)),
-		dkgSetSize:            uint32(len(nodes)),
-		pendingChangedConfigs: make(map[StateChangeType]interface{}),
+		local:            local,
+		numChains:        uint32(len(nodes)),
+		lambdaBA:         lambda,
+		lambdaDKG:        lambda * 10,
+		roundInterval:    lambda * 10000,
+		minBlockInterval: time.Millisecond * 1,
+		maxBlockInterval: lambda * 8,
+		crs:              []common.Hash{genesisCRS},
+		nodes:            nodes,
+		phiRatio:         0.667,
+		k:                0,
+		notarySetSize:    uint32(len(nodes)),
+		dkgSetSize:       uint32(len(nodes)),
+		ownRequests:      make(map[common.Hash]*StateChangeRequest),
+		globalRequests:   make(map[common.Hash]*StateChangeRequest),
 		dkgFinals: make(
 			map[uint64]map[types.NodeID]*typesDKG.Finalize),
 		dkgComplaints: make(
@@ -181,7 +141,8 @@ func NewState(
 }
 
 // SwitchToRemoteMode turn this State instance into remote mode: all changes
-// are pending, and need to be packed/unpacked to apply.
+// are pending, and need to be packed/unpacked to apply. Once this state switch
+// to remote mode, there would be no way to switch back to local mode.
 func (s *State) SwitchToRemoteMode() {
 	s.lock.Lock()
 	defer s.lock.Unlock()
@@ -278,41 +239,24 @@ func (s *State) unpackPayload(
 	return
 }
 
-func (s *State) cloneDKGComplaint(
-	comp *typesDKG.Complaint) (copied *typesDKG.Complaint) {
-	b, err := rlp.EncodeToBytes(comp)
-	if err != nil {
-		panic(err)
-	}
-	copied = &typesDKG.Complaint{}
-	if err = rlp.DecodeBytes(b, copied); err != nil {
-		panic(err)
-	}
-	return
-}
-
-func (s *State) cloneDKGMasterPublicKey(mpk *typesDKG.MasterPublicKey) (
-	copied *typesDKG.MasterPublicKey) {
-	b, err := rlp.EncodeToBytes(mpk)
-	if err != nil {
-		panic(err)
-	}
-	copied = typesDKG.NewMasterPublicKey()
-	if err = rlp.DecodeBytes(b, copied); err != nil {
-		panic(err)
-	}
-	return
-}
-
-func (s *State) cloneDKGFinalize(final *typesDKG.Finalize) (
-	copied *typesDKG.Finalize) {
-	b, err := rlp.EncodeToBytes(final)
-	if err != nil {
-		panic(err)
+func (s *State) unpackRequests(
+	b []byte) (reqs []*StateChangeRequest, err error) {
+	// Try to unmarshal this byte stream into []*StateChangeRequest.
+	rawReqs := []*rawStateChangeRequest{}
+	if err = rlp.DecodeBytes(b, &rawReqs); err != nil {
+		return
 	}
-	copied = &typesDKG.Finalize{}
-	if err = rlp.DecodeBytes(b, copied); err != nil {
-		panic(err)
+	for _, r := range rawReqs {
+		var payload interface{}
+		if payload, err = s.unpackPayload(r); err != nil {
+			return
+		}
+		reqs = append(reqs, &StateChangeRequest{
+			Type:      r.Type,
+			Payload:   payload,
+			Hash:      r.Hash,
+			Timestamp: r.Timestamp,
+		})
 	}
 	return
 }
@@ -433,43 +377,27 @@ func (s *State) Equal(other *State) error {
 		}
 	}
 	// Check pending changes.
-	if !reflect.DeepEqual(
-		s.pendingChangedConfigs, other.pendingChangedConfigs) {
-		return ErrStatePendingChangesNotEqual
-	}
-	if !reflect.DeepEqual(s.pendingCRS, other.pendingCRS) {
-		return ErrStatePendingChangesNotEqual
-	}
-	if !reflect.DeepEqual(s.pendingNodes, other.pendingNodes) {
-		return ErrStatePendingChangesNotEqual
-	}
-	// Check pending DKG complaints.
-	if len(s.pendingDKGComplaints) != len(other.pendingDKGComplaints) {
-		return ErrStatePendingChangesNotEqual
-	}
-	for idx, comp := range s.pendingDKGComplaints {
-		if !comp.Equal(other.pendingDKGComplaints[idx]) {
+	checkPending := func(
+		src, target map[common.Hash]*StateChangeRequest) error {
+		if len(src) != len(target) {
 			return ErrStatePendingChangesNotEqual
 		}
-	}
-	// Check pending DKG finals.
-	if len(s.pendingDKGFinals) != len(other.pendingDKGFinals) {
-		return ErrStatePendingChangesNotEqual
-	}
-	for idx, final := range s.pendingDKGFinals {
-		if !final.Equal(other.pendingDKGFinals[idx]) {
-			return ErrStatePendingChangesNotEqual
+		for k, v := range src {
+			otherV, exists := target[k]
+			if !exists {
+				return ErrStatePendingChangesNotEqual
+			}
+			if err := v.Equal(otherV); err != nil {
+				return err
+			}
 		}
+		return nil
 	}
-	// Check pending DKG Master public keys.
-	if len(s.pendingDKGMasterPublicKeys) !=
-		len(other.pendingDKGMasterPublicKeys) {
-		return ErrStatePendingChangesNotEqual
+	if err := checkPending(s.ownRequests, other.ownRequests); err != nil {
+		return err
 	}
-	for idx, mKey := range s.pendingDKGMasterPublicKeys {
-		if !mKey.Equal(other.pendingDKGMasterPublicKeys[idx]) {
-			return ErrStatePendingChangesNotEqual
-		}
+	if err := checkPending(s.globalRequests, other.globalRequests); err != nil {
+		return err
 	}
 	return nil
 }
@@ -494,8 +422,7 @@ func (s *State) Clone() (copied *State) {
 			map[uint64]map[types.NodeID][]*typesDKG.Complaint),
 		dkgMasterPublicKeys: make(
 			map[uint64]map[types.NodeID]*typesDKG.MasterPublicKey),
-		dkgFinals:             make(map[uint64]map[types.NodeID]*typesDKG.Finalize),
-		pendingChangedConfigs: make(map[StateChangeType]interface{}),
+		dkgFinals: make(map[uint64]map[types.NodeID]*typesDKG.Finalize),
 	}
 	// Nodes
 	for nID, key := range s.nodes {
@@ -508,7 +435,7 @@ func (s *State) Clone() (copied *State) {
 		for nID, comps := range complaintsForRound {
 			tmpComps := []*typesDKG.Complaint{}
 			for _, comp := range comps {
-				tmpComps = append(tmpComps, s.cloneDKGComplaint(comp))
+				tmpComps = append(tmpComps, cloneDKGComplaint(comp))
 			}
 			copied.dkgComplaints[round][nID] = tmpComps
 		}
@@ -518,44 +445,26 @@ func (s *State) Clone() (copied *State) {
 			make(map[types.NodeID]*typesDKG.MasterPublicKey)
 		for nID, mKey := range mKeysForRound {
 			copied.dkgMasterPublicKeys[round][nID] =
-				s.cloneDKGMasterPublicKey(mKey)
+				cloneDKGMasterPublicKey(mKey)
 		}
 	}
 	for round, finalsForRound := range s.dkgFinals {
 		copied.dkgFinals[round] = make(map[types.NodeID]*typesDKG.Finalize)
 		for nID, final := range finalsForRound {
-			copied.dkgFinals[round][nID] = s.cloneDKGFinalize(final)
+			copied.dkgFinals[round][nID] = cloneDKGFinalize(final)
 		}
 	}
 	for _, crs := range s.crs {
 		copied.crs = append(copied.crs, crs)
 	}
 	// Pending Changes
-	for t, v := range s.pendingChangedConfigs {
-		copied.pendingChangedConfigs[t] = v
-	}
-	for _, bs := range s.pendingNodes {
-		tmpBytes := make([]byte, len(bs))
-		copy(tmpBytes, bs)
-		copied.pendingNodes = append(copied.pendingNodes, tmpBytes)
-	}
-	for _, comp := range s.pendingDKGComplaints {
-		copied.pendingDKGComplaints = append(
-			copied.pendingDKGComplaints, s.cloneDKGComplaint(comp))
-	}
-	for _, final := range s.pendingDKGFinals {
-		copied.pendingDKGFinals = append(
-			copied.pendingDKGFinals, s.cloneDKGFinalize(final))
-	}
-	for _, mKey := range s.pendingDKGMasterPublicKeys {
-		copied.pendingDKGMasterPublicKeys = append(
-			copied.pendingDKGMasterPublicKeys, s.cloneDKGMasterPublicKey(mKey))
-	}
-	for _, req := range s.pendingCRS {
-		copied.pendingCRS = append(copied.pendingCRS, &crsAdditionRequest{
-			Round: req.Round,
-			CRS:   req.CRS,
-		})
+	copied.ownRequests = make(map[common.Hash]*StateChangeRequest)
+	for k, req := range s.ownRequests {
+		copied.ownRequests[k] = req.Clone()
+	}
+	copied.globalRequests = make(map[common.Hash]*StateChangeRequest)
+	for k, req := range s.globalRequests {
+		copied.globalRequests[k] = req.Clone()
 	}
 	return
 }
@@ -564,24 +473,16 @@ func (s *State) Clone() (copied *State) {
 // be called when we extract these request from delivered blocks.
 func (s *State) Apply(reqsAsBytes []byte) (err error) {
 	// Try to unmarshal this byte stream into []*StateChangeRequest.
-	rawReqs := []*rawStateChangeRequest{}
-	if err = rlp.DecodeBytes(reqsAsBytes, &rawReqs); err != nil {
+	reqs, err := s.unpackRequests(reqsAsBytes)
+	if err != nil {
 		return
 	}
-	var reqs []*StateChangeRequest
-	for _, r := range rawReqs {
-		var payload interface{}
-		if payload, err = s.unpackPayload(r); err != nil {
-			return
-		}
-		reqs = append(reqs, &StateChangeRequest{
-			Type:    r.Type,
-			Payload: payload,
-		})
-	}
 	s.lock.Lock()
 	defer s.lock.Unlock()
 	for _, req := range reqs {
+		// Remove this request from pending set once it's about to apply.
+		delete(s.globalRequests, req.Hash)
+		delete(s.ownRequests, req.Hash)
 		if err = s.isValidRequest(req); err != nil {
 			if err == ErrDuplicatedChange {
 				err = nil
@@ -596,53 +497,57 @@ func (s *State) Apply(reqsAsBytes []byte) (err error) {
 	return
 }
 
-// PackRequests pack current pending requests as byte slice, which
-// could be sent as blocks' payload and unmarshall back to apply.
-func (s *State) PackRequests() (b []byte, err error) {
-	packed := []*StateChangeRequest{}
-	s.pendingChangesLock.Lock()
-	defer s.pendingChangesLock.Unlock()
-	// Pack simple configuration changes first. There should be no
-	// validity problems for those changes.
-	for k, v := range s.pendingChangedConfigs {
-		packed = append(packed, &StateChangeRequest{
-			Type:    k,
-			Payload: v,
-		})
+// AddRequestsFromOthers add requests from others, they won't be packed by
+// 'PackOwnRequests'.
+func (s *State) AddRequestsFromOthers(reqsAsBytes []byte) (err error) {
+	if s.local {
+		err = ErrNotInRemoteMode
+		return
 	}
-	s.pendingChangedConfigs = make(map[StateChangeType]interface{})
-	// For other changes, we need to check their validity.
-	s.lock.RLock()
-	defer s.lock.RUnlock()
-	for _, bytesOfKey := range s.pendingNodes {
-		packed = append(packed, &StateChangeRequest{
-			Type:    StateAddNode,
-			Payload: bytesOfKey,
-		})
+	reqs, err := s.unpackRequests(reqsAsBytes)
+	if err != nil {
+		return
 	}
-	for _, comp := range s.pendingDKGComplaints {
-		packed = append(packed, &StateChangeRequest{
-			Type:    StateAddDKGComplaint,
-			Payload: comp,
-		})
+	s.lock.Lock()
+	defer s.lock.Unlock()
+	for _, req := range reqs {
+		s.globalRequests[req.Hash] = req
 	}
-	for _, final := range s.pendingDKGFinals {
-		packed = append(packed, &StateChangeRequest{
-			Type:    StateAddDKGFinal,
-			Payload: final,
-		})
+	return
+}
+
+// PackRequests pack all current pending requests, include those from others.
+func (s *State) PackRequests() (b []byte, err error) {
+	// Convert own requests to global one for packing.
+	if _, err = s.PackOwnRequests(); err != nil {
+		return
 	}
-	for _, masterPubKey := range s.pendingDKGMasterPublicKeys {
-		packed = append(packed, &StateChangeRequest{
-			Type:    StateAddDKGMasterPublicKey,
-			Payload: masterPubKey,
-		})
+	// Pack requests in global pool.
+	packed := []*StateChangeRequest{}
+	s.lock.Lock()
+	defer s.lock.Unlock()
+	for _, v := range s.globalRequests {
+		packed = append(packed, v)
 	}
-	for _, crs := range s.pendingCRS {
-		packed = append(packed, &StateChangeRequest{
-			Type:    StateAddCRS,
-			Payload: crs,
-		})
+	return rlp.EncodeToBytes(packed)
+}
+
+// PackOwnRequests pack current pending requests as byte slice, which
+// could be sent as blocks' payload and unmarshall back to apply.
+//
+// Once a request is packed as own request, it would be turned into a normal
+// pending request and won't be packed by this function. This would ensure
+// each request broadcasted(gossip) once.
+//
+// This function is not required to call in local mode.
+func (s *State) PackOwnRequests() (b []byte, err error) {
+	packed := []*StateChangeRequest{}
+	s.lock.Lock()
+	defer s.lock.Unlock()
+	for k, v := range s.ownRequests {
+		delete(s.ownRequests, k)
+		s.globalRequests[k] = v
+		packed = append(packed, v)
 	}
 	if b, err = rlp.EncodeToBytes(packed); err != nil {
 		return
@@ -782,45 +687,29 @@ func (s *State) RequestChange(
 		payload = uint64(payload.(int))
 	case StateChangePhiRatio:
 		payload = math.Float32bits(payload.(float32))
-	}
-	req := &StateChangeRequest{
-		Type:    t,
-		Payload: payload,
-	}
-	if s.local {
-		err = func() error {
-			s.lock.Lock()
-			defer s.lock.Unlock()
-			if err := s.isValidRequest(req); err != nil {
-				return err
-			}
-			return s.applyRequest(req)
-		}()
-		return
-	}
-	s.lock.RLock()
-	defer s.lock.RUnlock()
-	if err = s.isValidRequest(req); err != nil {
-		return
-	}
-	s.pendingChangesLock.Lock()
-	defer s.pendingChangesLock.Unlock()
-	switch t {
-	case StateAddNode:
-		s.pendingNodes = append(s.pendingNodes, payload.([]byte))
+	// These cases for for type assertion, make sure callers pass expected types.
 	case StateAddCRS:
-		s.pendingCRS = append(s.pendingCRS, payload.(*crsAdditionRequest))
-	case StateAddDKGComplaint:
-		s.pendingDKGComplaints = append(
-			s.pendingDKGComplaints, payload.(*typesDKG.Complaint))
-	case StateAddDKGMasterPublicKey:
-		s.pendingDKGMasterPublicKeys = append(
-			s.pendingDKGMasterPublicKeys, payload.(*typesDKG.MasterPublicKey))
+		payload = payload.(*crsAdditionRequest)
 	case StateAddDKGFinal:
-		s.pendingDKGFinals = append(
-			s.pendingDKGFinals, payload.(*typesDKG.Finalize))
-	default:
-		s.pendingChangedConfigs[t] = payload
+		payload = payload.(*typesDKG.Finalize)
+	case StateAddDKGMasterPublicKey:
+		payload = payload.(*typesDKG.MasterPublicKey)
+	case StateAddDKGComplaint:
+		payload = payload.(*typesDKG.Complaint)
+	}
+	req := NewStateChangeRequest(t, payload)
+	s.lock.Lock()
+	defer s.lock.Unlock()
+	if s.local {
+		if err = s.isValidRequest(req); err != nil {
+			return
+		}
+		err = s.applyRequest(req)
+	} else {
+		if err = s.isValidRequest(req); err != nil {
+			return
+		}
+		s.ownRequests[req.Hash] = req
 	}
 	return
 }
@@ -847,7 +736,7 @@ func (s *State) DKGComplaints(round uint64) []*typesDKG.Complaint {
 	tmpComps := make([]*typesDKG.Complaint, 0, len(comps))
 	for _, compProp := range comps {
 		for _, comp := range compProp {
-			tmpComps = append(tmpComps, s.cloneDKGComplaint(comp))
+			tmpComps = append(tmpComps, cloneDKGComplaint(comp))
 		}
 	}
 	return tmpComps
@@ -865,7 +754,7 @@ func (s *State) DKGMasterPublicKeys(round uint64) []*typesDKG.MasterPublicKey {
 	}
 	mpks := make([]*typesDKG.MasterPublicKey, 0, len(masterPublicKeys))
 	for _, mpk := range masterPublicKeys {
-		mpks = append(mpks, s.cloneDKGMasterPublicKey(mpk))
+		mpks = append(mpks, cloneDKGMasterPublicKey(mpk))
 	}
 	return mpks
 }
diff --git a/core/test/state_test.go b/core/test/state_test.go
index fc43ab7..6c5f882 100644
--- a/core/test/state_test.go
+++ b/core/test/state_test.go
@@ -194,6 +194,18 @@ func (s *StateTestSuite) TestEqual() {
 	req.NoError(st.Equal(st5))
 	delete(st5.dkgFinals, uint64(2))
 	req.Equal(st.Equal(st5), ErrStateDKGFinalsNotEqual)
+	// Switch to remote mode.
+	st.SwitchToRemoteMode()
+	// Make some change.
+	req.NoError(st.RequestChange(StateChangeK, int(5)))
+	st6 := st.Clone()
+	req.NoError(st.Equal(st6))
+	// Remove the pending change, should not be equal.
+	req.Len(st6.ownRequests, 1)
+	for k := range st6.ownRequests {
+		delete(st6.ownRequests, k)
+	}
+	req.Error(ErrStatePendingChangesNotEqual, st.Equal(st6))
 }
 
 func (s *StateTestSuite) TestPendingChangesEqual() {
@@ -214,31 +226,6 @@ func (s *StateTestSuite) TestPendingChangesEqual() {
 	comp := s.newDKGComplaint(2)
 	final := s.newDKGFinal(2)
 	s.makeDKGChanges(st, masterPubKey, comp, final)
-	// Remove pending config changes.
-	st1 := st.Clone()
-	req.NoError(st.Equal(st1))
-	st1.pendingChangedConfigs = make(map[StateChangeType]interface{})
-	req.Equal(st.Equal(st1), ErrStatePendingChangesNotEqual)
-	// Remove pending crs changes.
-	st2 := st.Clone()
-	req.NoError(st.Equal(st2))
-	st2.pendingCRS = []*crsAdditionRequest{}
-	req.Equal(st.Equal(st2), ErrStatePendingChangesNotEqual)
-	// Remove pending dkg complaints changes.
-	st3 := st.Clone()
-	req.NoError(st.Equal(st3))
-	st3.pendingDKGComplaints = []*typesDKG.Complaint{}
-	req.Equal(st.Equal(st3), ErrStatePendingChangesNotEqual)
-	// Remove pending dkg master public key changes.
-	st4 := st.Clone()
-	req.NoError(st.Equal(st4))
-	st4.pendingDKGMasterPublicKeys = []*typesDKG.MasterPublicKey{}
-	req.Equal(st.Equal(st4), ErrStatePendingChangesNotEqual)
-	// Remove pending dkg finalize changes.
-	st5 := st.Clone()
-	req.NoError(st.Equal(st5))
-	st5.pendingDKGFinals = []*typesDKG.Finalize{}
-	req.Equal(st.Equal(st5), ErrStatePendingChangesNotEqual)
 }
 
 func (s *StateTestSuite) TestLocalMode() {
@@ -353,6 +340,67 @@ func (s *StateTestSuite) TestPacking() {
 	req.True(st.IsDKGFinal(2, 0))
 }
 
+func (s *StateTestSuite) TestRequestBroadcastAndPack() {
+	// This test case aims to demonstrate this scenario:
+	// - a change request is pending at one node.
+	// - that request can be packed by PackOwnRequests and sent to other nodes.
+	// - when some other node allow to propose a block, it will pack all those
+	//   'own' requests from others into the block's payload.
+	// - when all nodes receive that block, all pending requests (including
+	//   those 'own' requests) would be cleaned.
+	var (
+		req    = s.Require()
+		lambda = 250 * time.Millisecond
+	)
+	_, genesisNodes, err := NewKeys(20)
+	req.NoError(err)
+	st := NewState(genesisNodes, lambda, false)
+	st1 := NewState(genesisNodes, lambda, false)
+	req.NoError(st.Equal(st1))
+	// Make configuration changes.
+	s.makeConfigChanges(st)
+	// Add new CRS.
+	crs := common.NewRandomHash()
+	req.NoError(st.ProposeCRS(1, crs))
+	// Add new node.
+	prvKey, err := ecdsa.NewPrivateKey()
+	req.NoError(err)
+	pubKey := prvKey.PublicKey()
+	st.RequestChange(StateAddNode, pubKey)
+	// Add DKG stuffs.
+	masterPubKey := s.newDKGMasterPublicKey(2)
+	comp := s.newDKGComplaint(2)
+	final := s.newDKGFinal(2)
+	s.makeDKGChanges(st, masterPubKey, comp, final)
+	// Pack those changes into a byte stream, and pass it to other State
+	// instance.
+	packed, err := st.PackOwnRequests()
+	req.NoError(err)
+	req.NotEmpty(packed)
+	// The second attempt to pack would get empty result.
+	emptyPackedAsByte, err := st.PackOwnRequests()
+	req.NoError(err)
+	emptyPacked, err := st.unpackRequests(emptyPackedAsByte)
+	req.NoError(err)
+	req.Empty(emptyPacked)
+	// Pass it to others.
+	req.NoError(st1.AddRequestsFromOthers(packed))
+	// These two instance are equal now, because their pending change requests
+	// are synced.
+	req.NoError(st.Equal(st1))
+	// Make them apply those pending changes.
+	applyChangesForRemoteState := func(s *State) {
+		p, err := s.PackRequests()
+		req.NoError(err)
+		req.NotEmpty(p)
+		req.NoError(s.Apply(p))
+	}
+	applyChangesForRemoteState(st)
+	applyChangesForRemoteState(st1)
+	// They should be equal after applying those changes.
+	req.NoError(st.Equal(st1))
+}
+
 func TestState(t *testing.T) {
 	suite.Run(t, new(StateTestSuite))
 }
diff --git a/core/test/utils.go b/core/test/utils.go
index da76860..f5f4c36 100644
--- a/core/test/utils.go
+++ b/core/test/utils.go
@@ -27,6 +27,8 @@ import (
 	"github.com/dexon-foundation/dexon-consensus/core/crypto"
 	"github.com/dexon-foundation/dexon-consensus/core/crypto/ecdsa"
 	"github.com/dexon-foundation/dexon-consensus/core/types"
+	typesDKG "github.com/dexon-foundation/dexon-consensus/core/types/dkg"
+	"github.com/dexon-foundation/dexon/rlp"
 )
 
 func stableRandomHash(block *types.Block) (common.Hash, error) {
@@ -116,3 +118,42 @@ func NewKeys(count int) (
 	}
 	return
 }
+
+func cloneDKGComplaint(
+	comp *typesDKG.Complaint) (copied *typesDKG.Complaint) {
+	b, err := rlp.EncodeToBytes(comp)
+	if err != nil {
+		panic(err)
+	}
+	copied = &typesDKG.Complaint{}
+	if err = rlp.DecodeBytes(b, copied); err != nil {
+		panic(err)
+	}
+	return
+}
+
+func cloneDKGMasterPublicKey(mpk *typesDKG.MasterPublicKey) (
+	copied *typesDKG.MasterPublicKey) {
+	b, err := rlp.EncodeToBytes(mpk)
+	if err != nil {
+		panic(err)
+	}
+	copied = typesDKG.NewMasterPublicKey()
+	if err = rlp.DecodeBytes(b, copied); err != nil {
+		panic(err)
+	}
+	return
+}
+
+func cloneDKGFinalize(final *typesDKG.Finalize) (
+	copied *typesDKG.Finalize) {
+	b, err := rlp.EncodeToBytes(final)
+	if err != nil {
+		panic(err)
+	}
+	copied = &typesDKG.Finalize{}
+	if err = rlp.DecodeBytes(b, copied); err != nil {
+		panic(err)
+	}
+	return
+}