config ack sender and receiver

common/configack/config_ack_receiver.go

@@ -0,0 +1,133 @@
+/*
+Copyright IBM Corp. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package configack
+
+import (
+	"context"
+	"fmt"
+	"sync"
+
+	"github.com/hyperledger/fabric-x-orderer/common/types"
+
+	"github.com/hyperledger/fabric-lib-go/common/flogging"
+	protos "github.com/hyperledger/fabric-x-orderer/node/protos/comm"
+)
+
+type ClientID struct {
+	NodeType protos.NodeType
+	Shard    types.ShardID
+}
+
+// ConfigAckHandler handle ConfigAck messages from party members (router, batchers, assembler)
+type ConfigAckHandler struct {
+	acks   map[ClientID]uint64
+	logger *flogging.FabricLogger
+	lock   sync.Mutex
+
+	signalChan chan struct{}
+	ctx        context.Context
+	cancel     context.CancelFunc
+}
+
+// NewConfigAckHandler creates a new config ack handler that collects config acknowledgments.
+func NewConfigAckHandler(logger *flogging.FabricLogger, shards []types.ShardID) *ConfigAckHandler {
+	ctx, cancel := context.WithCancel(context.Background())
+
+	handler := &ConfigAckHandler{
+		acks:       make(map[ClientID]uint64),
+		logger:     logger,
+		ctx:        ctx,
+		cancel:     cancel,
+		signalChan: make(chan struct{}, 100),
+	}
+
+	// fill the map
+	handler.initClients(shards)
+
+	return handler
+}
+
+func (ca *ConfigAckHandler) initClients(shards []types.ShardID) {
+	// Router
+	ca.acks[ClientID{
+		NodeType: protos.NodeType_ROUTER,
+		Shard:    0,
+	}] = 0
+
+	// Assembler
+	ca.acks[ClientID{
+		NodeType: protos.NodeType_ASSEMBLER,
+		Shard:    0,
+	}] = 0
+
+	// Batchers
+	for _, shardID := range shards {
+		ca.acks[ClientID{
+			NodeType: protos.NodeType_BATCHER,
+			Shard:    shardID,
+		}] = 0
+	}
+}
+
+func (ca *ConfigAckHandler) Stop() {
+	ca.logger.Infof("config ack handler is stopping")
+	ca.cancel()
+}
+
+// AddAck records an acknowledgment from a party member.
+func (ca *ConfigAckHandler) AddAck(request *protos.ConfigAck) error {
+	clientID := ClientID{
+		NodeType: request.NodeType,
+		Shard:    types.ShardID(request.Shard),
+	}
+
+	ca.lock.Lock()
+	if uint64(request.ConfigSeq) != ca.acks[clientID]+1 {
+		ca.lock.Unlock()
+		ca.logger.Warnf("config ack has been received on sequence %d but the last acknowledged sequence is %d", request.ConfigSeq, ca.acks[clientID])
+		return fmt.Errorf("config ack has been received on sequence %d but the last acknowledged sequence is %d", request.ConfigSeq, ca.acks[clientID])
+	}
+
+	ca.acks[clientID] = uint64(request.ConfigSeq)
+	ca.logger.Infof("config ack has been received on sequence %d", request.ConfigSeq)
+	ca.lock.Unlock()
+
+	ca.signalChan <- struct{}{}
+	return nil
+}
+
+func (ca *ConfigAckHandler) ExpectAck(configSeq uint64) bool {
+	for {
+		select {
+		case <-ca.ctx.Done():
+			ca.logger.Infof("config ack handler is stopped")
+			return false
+		case <-ca.signalChan:
+			ca.lock.Lock()
+			res := ca.allAcksAtSeq(configSeq)
+			ca.lock.Unlock()
+
+			if res {
+				return res
+			}
+		}
+	}
+}
+
+func (ca *ConfigAckHandler) allAcksAtSeq(seq uint64) bool {
+	if len(ca.acks) == 0 {
+		return false
+	}
+
+	for _, ackSeq := range ca.acks {
+		if ackSeq != seq {
+			return false
+		}
+	}
+
+	return true
+}

common/configack/config_ack_receiver_test.go

@@ -0,0 +1,161 @@
+/*
+Copyright IBM Corp. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package configack
+
+import (
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/hyperledger/fabric-x-orderer/common/types"
+	protos "github.com/hyperledger/fabric-x-orderer/node/protos/comm"
+	"github.com/hyperledger/fabric-x-orderer/testutil"
+	"github.com/stretchr/testify/require"
+)
+
+// TestAddAck_Success verifies that a valid sequential ack is accepted.
+func TestAddAck_Success(t *testing.T) {
+	shards := []types.ShardID{1, 2}
+	h := NewConfigAckHandler(testutil.CreateLogger(t, 0), shards)
+	defer h.Stop()
+
+	err := h.AddAck(&protos.ConfigAck{
+		ConfigSeq: 1,
+		NodeType:  protos.NodeType_ROUTER,
+	})
+	require.NoError(t, err)
+}
+
+// TestAddAck_DuplicateRejected verifies that re-sending the same sequence number is rejected
+func TestAddAck_DuplicateRejected(t *testing.T) {
+	shards := []types.ShardID{1, 2}
+	h := NewConfigAckHandler(testutil.CreateLogger(t, 0), shards)
+	defer h.Stop()
+
+	err := h.AddAck(&protos.ConfigAck{
+		ConfigSeq: 1,
+		NodeType:  protos.NodeType_ROUTER,
+	})
+	require.NoError(t, err)
+
+	err = h.AddAck(&protos.ConfigAck{
+		ConfigSeq: 1,
+		NodeType:  protos.NodeType_ROUTER,
+	})
+	require.Error(t, err)
+	require.ErrorContains(t, err, "config ack has been received on sequence 1 but the last acknowledged sequence is 1")
+}
+
+// TestAddAck_OutOfOrderRejected verifies that ack with out-of-sequence number is rejected
+func TestAddAck_OutOfOrderRejected(t *testing.T) {
+	shards := []types.ShardID{1, 2}
+	h := NewConfigAckHandler(testutil.CreateLogger(t, 0), shards)
+	defer h.Stop()
+
+	err := h.AddAck(&protos.ConfigAck{
+		ConfigSeq: 1,
+		NodeType:  protos.NodeType_ROUTER,
+	})
+	require.NoError(t, err)
+
+	err = h.AddAck(&protos.ConfigAck{
+		ConfigSeq: 3,
+		NodeType:  protos.NodeType_ROUTER,
+	})
+	require.Error(t, err)
+	require.ErrorContains(t, err, "config ack has been received on sequence 3 but the last acknowledged sequence is 1")
+}
+
+// TestAddAck_Concurrent verifies that concurrent AddAck calls from different
+// clients do not cause races.
+func TestAddAck_Concurrent(t *testing.T) {
+	shards := []types.ShardID{1, 2}
+	h := NewConfigAckHandler(testutil.CreateLogger(t, 0), shards)
+	defer h.Stop()
+
+	var wg sync.WaitGroup
+
+	sendAck := func(nodeType protos.NodeType, shard uint32) {
+		defer wg.Done()
+		_ = h.AddAck(&protos.ConfigAck{
+			ConfigSeq: 1,
+			NodeType:  nodeType,
+			Shard:     shard,
+		})
+	}
+
+	wg.Add(4)
+	go sendAck(protos.NodeType_ROUTER, 0)
+	go sendAck(protos.NodeType_ASSEMBLER, 0)
+	go sendAck(protos.NodeType_BATCHER, 1)
+	go sendAck(protos.NodeType_BATCHER, 2)
+	wg.Wait()
+}
+
+// TestExpectAck_AllAcknowledged verifies that ExpectAck returns true when all
+// clients have already acked the requested sequence before the call.
+func TestExpectAck_AllAcknowledged(t *testing.T) {
+	shards := []types.ShardID{1, 2}
+	h := NewConfigAckHandler(testutil.CreateLogger(t, 0), shards)
+	defer h.Stop()
+
+	// send ack from all clients (Router, Assembler, Batchers) at seq=1.
+	err := h.AddAck(&protos.ConfigAck{
+		ConfigSeq: 1,
+		NodeType:  protos.NodeType_ROUTER,
+	})
+	require.NoError(t, err)
+
+	err = h.AddAck(&protos.ConfigAck{
+		ConfigSeq: 1,
+		NodeType:  protos.NodeType_BATCHER,
+		Shard:     1,
+	})
+	require.NoError(t, err)
+
+	err = h.AddAck(&protos.ConfigAck{
+		ConfigSeq: 1,
+		NodeType:  protos.NodeType_BATCHER,
+		Shard:     2,
+	})
+	require.NoError(t, err)
+
+	err = h.AddAck(&protos.ConfigAck{
+		ConfigSeq: 1,
+		NodeType:  protos.NodeType_ASSEMBLER,
+	})
+	require.NoError(t, err)
+
+	res := h.ExpectAck(1)
+	require.True(t, res)
+}
+
+// TestExpectAck_StopReturnsFalse verifies that calling Stop() while ExpectAck
+// is blocked causes it to return false.
+func TestExpectAck_StopReturnsFalse(t *testing.T) {
+	shards := []types.ShardID{1, 2}
+	h := NewConfigAckHandler(testutil.CreateLogger(t, 0), shards)
+	defer h.Stop()
+
+	result := make(chan bool, 1)
+	go func() { result <- h.ExpectAck(1) }()
+
+	err := h.AddAck(&protos.ConfigAck{
+		ConfigSeq: 1,
+		NodeType:  protos.NodeType_ROUTER,
+	})
+	require.NoError(t, err)
+
+	h.Stop()
+
+	select {
+	case res := <-result:
+		require.False(t, res)
+	case <-time.After(60 * time.Second):
+		t.Fatal("ExpectAck did not return after Stop")
+	}
+}