add config request to control event (#305)

* add config request to control event

Signed-off-by: Natalie Morad <natalie.morad@ibm.com>

* fix review

Signed-off-by: Natalie Morad <natalie.morad@ibm.com>

---------

Signed-off-by: Natalie Morad <natalie.morad@ibm.com>

Author: moradna

node/consensus/consensus.go

@@ -53,7 +53,7 @@ type SigVerifier interface {
 }
 
 type Arma interface {
-	SimulateStateTransition(prevState *state.State, events [][]byte) (*state.State, [][]arma_types.BatchAttestationFragment)
+	SimulateStateTransition(prevState *state.State, events [][]byte) (*state.State, [][]arma_types.BatchAttestationFragment, *state.ConfigRequest)
 	Commit(events [][]byte)
 }
 
@@ -198,7 +198,7 @@ func (c *Consensus) VerifyProposal(proposal smartbft_types.Proposal) ([]smartbft
 	}
 
 	c.stateLock.Lock()
-	computedState, attestations := c.Arma.SimulateStateTransition(c.State, batch)
+	computedState, attestations, _ := c.Arma.SimulateStateTransition(c.State, batch)
 	c.stateLock.Unlock()
 
 	availableBlocks := make([]state.AvailableBlock, len(attestations))
@@ -413,7 +413,7 @@ func (c *Consensus) SignProposal(proposal smartbft_types.Proposal, _ []byte) *sm
 	}
 
 	c.stateLock.Lock()
-	_, bafs := c.Arma.SimulateStateTransition(c.State, requests)
+	_, bafs, _ := c.Arma.SimulateStateTransition(c.State, requests)
 	c.stateLock.Unlock()
 
 	sigs := make([][]byte, 0, len(bafs)+1)
@@ -451,7 +451,8 @@ func (c *Consensus) SignProposal(proposal smartbft_types.Proposal, _ []byte) *sm
 // (from SmartBFT API)
 func (c *Consensus) AssembleProposal(metadata []byte, requests [][]byte) smartbft_types.Proposal {
 	c.stateLock.Lock()
-	newState, attestations := c.Arma.SimulateStateTransition(c.State, requests)
+	// TODO: config request not handled yet
+	newState, attestations, _ := c.Arma.SimulateStateTransition(c.State, requests)
 	c.stateLock.Unlock()
 
 	lastCommonBlockHeader := &common.BlockHeader{}

node/consensus/consenter.go

@@ -29,7 +29,7 @@ type Consenter struct {
 	State           *state.State
 }
 
-func (c *Consenter) SimulateStateTransition(prevState *state.State, requests [][]byte) (*state.State, [][]types.BatchAttestationFragment) {
+func (c *Consenter) SimulateStateTransition(prevState *state.State, requests [][]byte) (*state.State, [][]types.BatchAttestationFragment, *state.ConfigRequest) {
 	controlEvents, err := requestsToControlEvents(requests, c.BAFDeserializer.Deserialize)
 	if err != nil {
 		panic(err)
@@ -48,17 +48,17 @@ func (c *Consenter) SimulateStateTransition(prevState *state.State, requests [][
 		filteredControlEvents = append(filteredControlEvents, ce)
 	}
 
-	newState, fragments := prevState.Process(c.Logger, filteredControlEvents...)
+	newState, fragments, configRequest := prevState.Process(c.Logger, filteredControlEvents...)
 	batchAttestations := aggregateFragments(fragments)
 
-	return newState, batchAttestations
+	return newState, batchAttestations, configRequest
 }
 
 // Commit indexes BAs and updates the state.
 // Note that this must hold: Commit(controlEvents) with the same controlEvents is idempotent.
 // TODO revise the recovery from failure or shutdown, specifically the order of Commit and Append.
 func (c *Consenter) Commit(events [][]byte) {
-	state, batchAttestations := c.SimulateStateTransition(c.State, events)
+	state, batchAttestations, _ := c.SimulateStateTransition(c.State, events)
 	if len(batchAttestations) > 0 {
 		c.indexAttestationsInDB(batchAttestations)
 	}

node/consensus/consenter_test.go

@@ -9,6 +9,7 @@ package consensus_test
 import (
 	"testing"
 
+	"github.com/hyperledger/fabric-protos-go-apiv2/common"
 	arma_types "github.com/hyperledger/fabric-x-orderer/common/types"
 	"github.com/hyperledger/fabric-x-orderer/node/consensus"
 	"github.com/hyperledger/fabric-x-orderer/node/consensus/mocks"
@@ -39,7 +40,7 @@ func TestConsenter(t *testing.T) {
 
 	// Test with an event that should be filtered out
 	db.ExistsReturns(true)
-	newState, batchAttestations := consenter.SimulateStateTransition(s, events)
+	newState, batchAttestations, _ := consenter.SimulateStateTransition(s, events)
 	assert.Empty(t, batchAttestations)
 	assert.Empty(t, newState.Pending)
 
@@ -49,7 +50,7 @@ func TestConsenter(t *testing.T) {
 
 	// Test a valid event below threshold
 	db.ExistsReturns(false)
-	newState, batchAttestations = consenter.SimulateStateTransition(s, events)
+	newState, batchAttestations, _ = consenter.SimulateStateTransition(s, events)
 	assert.Empty(t, batchAttestations)
 	assert.Len(t, newState.Pending, 1)
 
@@ -62,7 +63,7 @@ func TestConsenter(t *testing.T) {
 	ba2.SetSignature([]byte{1})
 	events = append(events, (&state.ControlEvent{BAF: ba2}).Bytes())
 
-	newState, batchAttestations = consenter.SimulateStateTransition(s, events)
+	newState, batchAttestations, _ = consenter.SimulateStateTransition(s, events)
 	assert.Len(t, batchAttestations[0], 2)
 	assert.Empty(t, newState.Pending)
 
@@ -82,6 +83,19 @@ func TestConsenter(t *testing.T) {
 	consenter.Commit(events)
 	assert.Equal(t, db.PutCallCount(), 1)
 	assert.Len(t, consenter.State.Complaints, 1)
+
+	// Test ConfigRequest is returned by SimulateStateTransition
+	cr := &state.ConfigRequest{
+		Envelope: &common.Envelope{
+			Payload:   []byte("config-payload"),
+			Signature: []byte("config-signature"),
+		},
+	}
+	events = [][]byte{(&state.ControlEvent{ConfigRequest: cr}).Bytes()}
+
+	_, _, configRequest := consenter.SimulateStateTransition(s, events)
+	assert.Equal(t, cr.Envelope.Payload, configRequest.Envelope.Payload)
+	assert.Equal(t, cr.Envelope.Signature, configRequest.Envelope.Signature)
 }
 
 func createConsenter(s *state.State, logger arma_types.Logger) *consensus.Consenter {

node/consensus/state/state.go

@@ -16,7 +16,9 @@ import (
 	"math"
 	"slices"
 
+	"github.com/hyperledger/fabric-protos-go-apiv2/common"
 	"github.com/hyperledger/fabric-x-orderer/common/types"
+	"github.com/hyperledger/fabric/protoutil"
 )
 
 type Rule func(*State, types.Logger, ...ControlEvent)
@@ -302,16 +304,47 @@ func (c *Complaint) String() string {
 	return fmt.Sprintf("Complaint: Signer: %d; Shard: %d; Term %d; Reason: %s", c.Signer, c.Shard, c.Term, c.Reason)
 }
 
+type ConfigRequest struct {
+	Envelope *common.Envelope
+}
+
+func (c *ConfigRequest) Bytes() []byte {
+	bytes, err := protoutil.Marshal(c.Envelope)
+	if err != nil {
+		panic(fmt.Sprintf("failed to marshal envelope: %v", err))
+	}
+
+	return bytes
+}
+
+func (c *ConfigRequest) FromBytes(bytes []byte) error {
+	envelope, err := protoutil.UnmarshalEnvelope(bytes)
+	if err != nil {
+		return fmt.Errorf("failed to unmarshal envelope: %v", err)
+	}
+
+	c.Envelope = envelope
+	return nil
+}
+
+func (c *ConfigRequest) String() string {
+	// TODO: add more info to this string
+	return "Config Request"
+}
+
 type ControlEvent struct {
-	BAF       types.BatchAttestationFragment
-	Complaint *Complaint
+	BAF           types.BatchAttestationFragment
+	Complaint     *Complaint
+	ConfigRequest *ConfigRequest
 }
 
 func (ce *ControlEvent) String() string {
 	if ce.Complaint != nil {
 		return ce.Complaint.String()
 	} else if ce.BAF != nil {
 		return ce.BAF.String()
+	} else if ce.ConfigRequest != nil {
+		return ce.ConfigRequest.String()
 	}
 	return "empty control event"
 }
@@ -330,6 +363,10 @@ func (ce *ControlEvent) ID() string {
 		binary.BigEndian.PutUint16(payloadToHash[18:], uint16(ce.BAF.Primary()))
 		binary.BigEndian.PutUint16(payloadToHash[20:], uint16(ce.BAF.Shard()))
 		copy(payloadToHash[22:], ce.BAF.Digest())
+	} else if ce.ConfigRequest != nil {
+		// TODO: maybe use a different ID for ConfigRequest
+		ce.ConfigRequest.Envelope.Signature = nil
+		payloadToHash = ce.ConfigRequest.Bytes()
 	} else {
 		return string("")
 	}
@@ -344,6 +381,7 @@ func (ce *ControlEvent) SignerID() string {
 		signerID = fmt.Sprintf("%d", ce.Complaint.Signer)
 	} else if ce.BAF != nil {
 		signerID = fmt.Sprintf("%d", ce.BAF.Signer())
+		// TODO: add ConfigRequest SignerID
 	} else {
 		return string("")
 	}
@@ -363,6 +401,11 @@ func (ce *ControlEvent) Bytes() []byte {
 		bytes = make([]byte, len(rawComplaint)+1)
 		bytes[0] = 2
 		copy(bytes[1:], rawComplaint)
+	case ce.ConfigRequest != nil:
+		rawConfig := ce.ConfigRequest.Bytes()
+		bytes = make([]byte, len(rawConfig)+1)
+		bytes[0] = 3
+		copy(bytes[1:], rawConfig)
 	default:
 		panic("empty control event")
 	}
@@ -379,12 +422,15 @@ func (ce *ControlEvent) FromBytes(bytes []byte, fragmentFromBytes func([]byte) (
 	case 2:
 		ce.Complaint = &Complaint{}
 		return ce.Complaint.FromBytes(bytes[1:])
+	case 3:
+		ce.ConfigRequest = &ConfigRequest{}
+		return ce.ConfigRequest.FromBytes(bytes[1:])
 	}
 
 	return fmt.Errorf("unknown prefix (%d)", bytes[0])
 }
 
-func (s *State) Process(l types.Logger, ces ...ControlEvent) (*State, []types.BatchAttestationFragment) {
+func (s *State) Process(l types.Logger, ces ...ControlEvent) (*State, []types.BatchAttestationFragment, *ConfigRequest) {
 	s2 := s.Clone()
 
 	for _, rule := range Rules {
@@ -393,8 +439,8 @@ func (s *State) Process(l types.Logger, ces ...ControlEvent) (*State, []types.Ba
 
 	// After applying rules, extract all batch attestations for which enough fragments have been collected.
 	extracted := ExtractBatchAttestationsFromPending(s2, l)
-
-	return s2, extracted
+	configRequests := ExtractConfigRequest(ces)
+	return s2, extracted, configRequests
 }
 
 func (s *State) Clone() *State {
@@ -658,3 +704,13 @@ func shardExists(shard types.ShardID, shardTerms []ShardTerm) (int, bool) {
 	}
 	return -1, false
 }
+
+func ExtractConfigRequest(ces []ControlEvent) *ConfigRequest {
+	// TODO: decide how to handle multiple config requests
+	for _, ce := range ces {
+		if ce.ConfigRequest != nil {
+			return ce.ConfigRequest
+		}
+	}
+	return nil
+}

node/consensus/state/state_test.go

@@ -10,6 +10,7 @@ import (
 	"math"
 	"testing"
 
+	"github.com/hyperledger/fabric-protos-go-apiv2/common"
 	"github.com/hyperledger/fabric-x-orderer/common/types"
 	consensus_state "github.com/hyperledger/fabric-x-orderer/node/consensus/state"
 	"github.com/hyperledger/fabric-x-orderer/testutil"
@@ -94,7 +95,7 @@ func TestComplaintSerialization(t *testing.T) {
 
 func TestControlEventSerialization(t *testing.T) {
 	// Serialization and deserialization of ControlEvent with Complaint
-	ce := consensus_state.ControlEvent{nil, &complaint}
+	ce := consensus_state.ControlEvent{Complaint: &complaint}
 
 	var ce2 consensus_state.ControlEvent
 
@@ -115,12 +116,28 @@ func TestControlEventSerialization(t *testing.T) {
 	assert.NoError(t, err)
 
 	assert.Equal(t, ce, ce2)
+
+	// Serialization and deserialization of ControlEvent with ConfigRequest
+	cr := &consensus_state.ConfigRequest{
+		Envelope: &common.Envelope{
+			Payload:   []byte("config-payload"),
+			Signature: []byte("config-signature"),
+		},
+	}
+	ce = consensus_state.ControlEvent{ConfigRequest: cr}
+
+	var ce3 consensus_state.ControlEvent
+	err = ce3.FromBytes(ce.Bytes(), bafd.Deserialize)
+	assert.NoError(t, err)
+	assert.NotNil(t, ce3.ConfigRequest)
+	assert.Equal(t, cr.Envelope.Payload, ce3.ConfigRequest.Envelope.Payload)
+	assert.Equal(t, cr.Envelope.Signature, ce3.ConfigRequest.Envelope.Signature)
 }
 
 func TestCollectAndDeduplicateEvents(t *testing.T) {
 	state := initialState
-	ce := consensus_state.ControlEvent{nil, &complaint}
-	ce2 := consensus_state.ControlEvent{nil, &complaint}
+	ce := consensus_state.ControlEvent{Complaint: &complaint}
+	ce2 := consensus_state.ControlEvent{Complaint: &complaint}
 	logger := testutil.CreateLogger(t, 0)
 
 	// Add a valid Complaint and ensure no duplicates are accepted in the same round
@@ -150,7 +167,7 @@ func TestCollectAndDeduplicateEvents(t *testing.T) {
 		Signature: []byte{4},
 	}
 
-	ce = consensus_state.ControlEvent{nil, &c}
+	ce = consensus_state.ControlEvent{Complaint: &c}
 	consensus_state.CollectAndDeduplicateEvents(&state, logger, ce)
 	assert.Equal(t, state, expectedState)
 
@@ -164,7 +181,7 @@ func TestCollectAndDeduplicateEvents(t *testing.T) {
 		Signature: []byte{4},
 	}
 
-	ce = consensus_state.ControlEvent{nil, &c}
+	ce = consensus_state.ControlEvent{Complaint: &c}
 	consensus_state.CollectAndDeduplicateEvents(&state, logger, ce)
 	assert.Equal(t, state, expectedState)