From d50e5d682c4c80381e52c621ce243956edc237f1 Mon Sep 17 00:00:00 2001 From: "May.Buzaglo" Date: Tue, 30 Jun 2026 14:32:16 +0300 Subject: [PATCH 1/2] config ack sender and receiver Signed-off-by: May.Buzaglo --- common/configack/config_ack_receiver.go | 133 +++++++++++++ common/configack/config_ack_receiver_test.go | 161 ++++++++++++++++ common/configack/config_ack_sender.go | 191 +++++++++++++++++++ common/configack/config_ack_sender_test.go | 115 +++++++++++ node/consensus/consensus.go | 4 +- 5 files changed, 602 insertions(+), 2 deletions(-) create mode 100644 common/configack/config_ack_receiver.go create mode 100644 common/configack/config_ack_receiver_test.go create mode 100644 common/configack/config_ack_sender.go create mode 100644 common/configack/config_ack_sender_test.go diff --git a/common/configack/config_ack_receiver.go b/common/configack/config_ack_receiver.go new file mode 100644 index 00000000..c5e33b97 --- /dev/null +++ b/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 { + ca.lock.Lock() + defer ca.lock.Unlock() + + clientID := ClientID{ + NodeType: request.NodeType, + Shard: types.ShardID(request.Shard), + } + + if uint64(request.ConfigSeq) != ca.acks[clientID]+1 { + 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.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 +} diff --git a/common/configack/config_ack_receiver_test.go b/common/configack/config_ack_receiver_test.go new file mode 100644 index 00000000..3685dc00 --- /dev/null +++ b/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_DuplicateRejected 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") + } +} diff --git a/common/configack/config_ack_sender.go b/common/configack/config_ack_sender.go new file mode 100644 index 00000000..0a67c955 --- /dev/null +++ b/common/configack/config_ack_sender.go @@ -0,0 +1,191 @@ +/* +Copyright IBM Corp. All Rights Reserved. + +SPDX-License-Identifier: Apache-2.0 +*/ + +package configack + +import ( + "context" + "fmt" + "time" + + "github.com/hyperledger/fabric-lib-go/common/flogging" + "github.com/hyperledger/fabric-x-orderer/common/types" + "github.com/hyperledger/fabric-x-orderer/node/comm" + protos "github.com/hyperledger/fabric-x-orderer/node/protos/comm" + "google.golang.org/grpc" +) + +// TODO: take minRetryInterval + maxRetryInterval + DialTimeout from config +const ( + minRetryInterval = 50 * time.Millisecond + maxRetryInterval = 10 * time.Second +) + +type ConfigAcker interface { + Stop() + SubmitConfigAck(configSeq uint32) error +} + +type configAcker struct { + consensusEndpoint string + consensusRootCAs [][]byte + tlsCert []byte + tlsKey []byte + logger *flogging.FabricLogger + partyID types.PartyID + nodeType protos.NodeType + shard types.ShardID + + ctx context.Context + cancel context.CancelFunc +} + +type ConnectionInfo struct { + TLSCert []byte + TLSKey []byte + ConsensusEndpoint string + ConsensusRootCAs [][]byte + PartyID types.PartyID + NodeType protos.NodeType + Shard types.ShardID +} + +func NewConfigAcker(connInfo *ConnectionInfo, logger *flogging.FabricLogger) *configAcker { + ctx, cancel := context.WithCancel(context.Background()) + + ca := &configAcker{ + consensusEndpoint: connInfo.ConsensusEndpoint, + consensusRootCAs: connInfo.ConsensusRootCAs, + tlsCert: connInfo.TLSCert, + tlsKey: connInfo.TLSKey, + logger: logger, + partyID: connInfo.PartyID, + nodeType: connInfo.NodeType, + shard: connInfo.Shard, + ctx: ctx, + cancel: cancel, + } + return ca +} + +func (ca *configAcker) SubmitConfigAck(configSeq uint32) error { + return ca.handleConfigAck(configSeq) +} + +func (ca *configAcker) Stop() { + ca.logger.Infof("config acker is stopping") + ca.cancel() +} + +// handleConfigAck attempts to deliver a ConfigAck for the given configSeq to the consenter. +// +// The whole delivery process is bounded by a single timeout context derived from ca.ctx. +// This timeout covers all retry attempts together. Therefore, once the timeout expires, +// handleConfigAck stops retrying and returns an error. +// +// Each retry attempt consists of: +// - connect to the consenter +// - send the ConfigAck +// +// If either the connection or the send fails, the function waits before retrying the +// entire flow again. The wait interval uses exponential backoff, starting from +// minRetryInterval and doubling after each failed attempt, up to maxRetryInterval. +// +// The function returns nil once the ConfigAck is successfully sent. It returns an error +// if the total timeout expires or if ca.ctx is cancelled, for example when the +// configAcker is stopped. +func (ca *configAcker) handleConfigAck(configSeq uint32) error { + ctx, cancel := context.WithTimeout(ca.ctx, 60*time.Second) + defer cancel() + + err := ca.tryToHandleConfigAck(ctx, configSeq) + if err == nil { + return nil + } + + interval := minRetryInterval + numOfRetries := 1 + + for { + select { + case <-ctx.Done(): + return fmt.Errorf("sending config ack to consensus aborted, because context is done and configAcker stopped") + case <-time.After(interval): + ca.logger.Debugf("Retry attempt #%d", numOfRetries) + numOfRetries++ + + err := ca.tryToHandleConfigAck(ctx, configSeq) + if err != nil { + interval = min(interval*2, maxRetryInterval) + ca.logger.Errorf("Sending config ack to consensus failed: %v, trying again in: %s", err, interval) + continue + } + + ca.logger.Infof("config ack was successfully sent to consensus") + return nil + } + } +} + +func (ca *configAcker) tryToHandleConfigAck(ctx context.Context, configSeq uint32) error { + conn, err := ca.connectToConsenter() + if err != nil { + ca.logger.Errorf("failed to connect to consensus, err: %v\n", err) + return err + } + defer conn.Close() + + err = ca.sendConfigAckToConsensus(ctx, conn, configSeq) + if err != nil { + ca.logger.Errorf("failed to send config ack to consensus, err: %v\n", err) + return err + } + + return nil +} + +func (ca *configAcker) connectToConsenter() (*grpc.ClientConn, error) { + cc := comm.ClientConfig{ + AsyncConnect: false, + KaOpts: comm.KeepaliveOptions{ + ClientInterval: 30 * time.Second, + ClientTimeout: 30 * time.Second, + }, + SecOpts: comm.SecureOptions{ + UseTLS: true, + ServerRootCAs: ca.consensusRootCAs, + Key: ca.tlsKey, + Certificate: ca.tlsCert, + RequireClientCert: true, + }, + DialTimeout: time.Second * 5, + } + + return cc.Dial(ca.consensusEndpoint) +} + +// sendConfigAckToConsensus sends a single ConfigAck RPC attempt over the given connection. +// The RPC is bounded by a timeout derived from ca.ctx. If ca.ctx is cancelled, the RPC returns with an error. +// If the consenter returns an error in the ConfigAckResponse, it is logged for +// visibility. The response error is not returned to the caller because the client +// does not take any recovery action based on this response; only the RPC error is +// used to decide whether the ConfigAck should be retried. +func (ca *configAcker) sendConfigAckToConsensus(ctx context.Context, conn *grpc.ClientConn, configSeq uint32) error { + client := protos.NewConsensusClient(conn) + + configAckReq := &protos.ConfigAck{ + ConfigSeq: configSeq, + NodeType: ca.nodeType, + Shard: uint32(ca.shard), + } + + ca.logger.Infof("Sending ConfigAck for config sequence %d to consenter", configSeq) + resp, err := client.AckConfig(ctx, configAckReq) + if resp != nil && resp.GetError() != "" { + ca.logger.Warnf("Received bad response from consensus on config ack: %v\n", resp.Error) + } + return err +} diff --git a/common/configack/config_ack_sender_test.go b/common/configack/config_ack_sender_test.go new file mode 100644 index 00000000..7d0763f4 --- /dev/null +++ b/common/configack/config_ack_sender_test.go @@ -0,0 +1,115 @@ +/* +Copyright IBM Corp. All Rights Reserved. + +SPDX-License-Identifier: Apache-2.0 +*/ + +package configack + +import ( + "fmt" + "sync/atomic" + "testing" + "time" + + "github.com/hyperledger/fabric-x-orderer/common/types" + "github.com/hyperledger/fabric-x-orderer/node/comm/tlsgen" + protos "github.com/hyperledger/fabric-x-orderer/node/protos/comm" + "github.com/hyperledger/fabric-x-orderer/testutil" + "github.com/hyperledger/fabric-x-orderer/testutil/stub" + "github.com/stretchr/testify/require" +) + +func createTestSetupForConfigAcker(t *testing.T, nodeType protos.NodeType, shard types.ShardID) (stub.StubConsenter, *configAcker) { + logger := testutil.CreateLogger(t, 0) + ca, err := tlsgen.NewCA() + require.NoError(t, err) + + clientPair, err := ca.NewClientCertKeyPair() + require.NoError(t, err) + + sc := stub.NewStubConsenter(t, ca, types.PartyID(1)) + + return sc, NewConfigAcker(&ConnectionInfo{ + TLSCert: clientPair.Cert, + TLSKey: clientPair.Key, + ConsensusEndpoint: sc.GetConsenterEndpoint(), + ConsensusRootCAs: [][]byte{ca.CertBytes()}, + PartyID: types.PartyID(1), + NodeType: nodeType, + Shard: shard, + }, logger) +} + +// TestSubmitConfigAck_Success verifies that a call to SubmitConfigAck returns nil +// when the consenter accepts the RPC on the first attempt. +func TestSubmitConfigAck_Success(t *testing.T) { + stubConsenter, configAcker := createTestSetupForConfigAcker(t, protos.NodeType_ROUTER, 0) + + stubConsenter.Start() + defer stubConsenter.Stop() + + require.NoError(t, configAcker.SubmitConfigAck(1)) + configAcker.Stop() +} + +// TestSubmitConfigAck_RetriesOnSendFailure verifies that when the RPC attempt fails because of an error in consensus the acker retries and eventually succeeds. +func TestSubmitConfigAck_RetriesOnSendFailure(t *testing.T) { + stubConsenter, configAcker := createTestSetupForConfigAcker(t, protos.NodeType_ROUTER, 0) + + var callCount int32 + stubConsenter.AckConfigHandler = func(req *protos.ConfigAck) (*protos.ConfigAckResponse, error) { + if atomic.AddInt32(&callCount, 1) <= 10 { + return nil, fmt.Errorf("temporary failure") + } + return &protos.ConfigAckResponse{}, nil + } + + stubConsenter.Start() + defer stubConsenter.Stop() + + require.NoError(t, configAcker.SubmitConfigAck(1)) + require.EqualValues(t, 11, atomic.LoadInt32(&callCount)) + configAcker.Stop() +} + +// TestSubmitConfigAck_RetriesOnConnectFailure verifies that the acker keeps +// retrying when the server is initially down and succeeds once it comes back up. +func TestSubmitConfigAck_RetriesOnConnectFailure(t *testing.T) { + stubConsenter, configAcker := createTestSetupForConfigAcker(t, protos.NodeType_ROUTER, 0) + + // Start the server after a short delay + go func() { + time.AfterFunc(20*time.Second, func() { + stubConsenter.Start() + }) + }() + + defer stubConsenter.Stop() + + require.NoError(t, configAcker.SubmitConfigAck(1)) + configAcker.Stop() +} + +// TestSubmitConfigAck_StopCancelsBlockedCall verifies that Stop() unblocks a +// SubmitConfigAck that is stuck retrying against an unavailable consenter. +func TestSubmitConfigAck_StopCancelsRetries(t *testing.T) { + _, configAcker := createTestSetupForConfigAcker(t, protos.NodeType_ROUTER, 0) + + done := make(chan error, 1) + go func() { + done <- configAcker.SubmitConfigAck(5) + }() + + // Give the acker time to attempt at least one connection. + time.Sleep(10 * time.Second) + configAcker.Stop() + + select { + case err := <-done: + require.Error(t, err) + require.ErrorContains(t, err, "sending config ack to consensus aborted, because context is done and configAcker stopped") + case <-time.After(5 * time.Second): + t.Fatal("SubmitConfigAck did not return after Stop was called") + } +} diff --git a/node/consensus/consensus.go b/node/consensus/consensus.go index c0af1ec8..4e562cd2 100644 --- a/node/consensus/consensus.go +++ b/node/consensus/consensus.go @@ -1275,7 +1275,7 @@ func (c *Consensus) UpdateStateAndRuntimeConfig(block *common.Block) smartbft_ty } // AckConfig handles ConfigAck RPC calls from party members (router, batchers, assembler). -// TODO: implement +// TODO: implement the following: check the client cert, add the ack to the ack receiver handler and send a response back func (c *Consensus) AckConfig(ctx context.Context, req *protos.ConfigAck) (*protos.ConfigAckResponse, error) { - return nil, fmt.Errorf("not implemented") + return &protos.ConfigAckResponse{}, nil } From 12d299716b1416771c20457663030a68efad34ba Mon Sep 17 00:00:00 2001 From: "May.Buzaglo" Date: Tue, 7 Jul 2026 17:24:38 +0300 Subject: [PATCH 2/2] address copilot review comments Signed-off-by: May.Buzaglo --- common/configack/config_ack_receiver.go | 6 ++--- common/configack/config_ack_receiver_test.go | 2 +- common/configack/config_ack_sender.go | 23 ++++++++++---------- common/configack/config_ack_sender_test.go | 8 +++++-- 4 files changed, 22 insertions(+), 17 deletions(-) diff --git a/common/configack/config_ack_receiver.go b/common/configack/config_ack_receiver.go index c5e33b97..dbf8042b 100644 --- a/common/configack/config_ack_receiver.go +++ b/common/configack/config_ack_receiver.go @@ -80,21 +80,21 @@ func (ca *ConfigAckHandler) Stop() { // AddAck records an acknowledgment from a party member. func (ca *ConfigAckHandler) AddAck(request *protos.ConfigAck) error { - ca.lock.Lock() - defer ca.lock.Unlock() - 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 diff --git a/common/configack/config_ack_receiver_test.go b/common/configack/config_ack_receiver_test.go index 3685dc00..dc700e77 100644 --- a/common/configack/config_ack_receiver_test.go +++ b/common/configack/config_ack_receiver_test.go @@ -50,7 +50,7 @@ func TestAddAck_DuplicateRejected(t *testing.T) { require.ErrorContains(t, err, "config ack has been received on sequence 1 but the last acknowledged sequence is 1") } -// TestAddAck_DuplicateRejected verifies that ack with out-of-sequence number is rejected +// 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) diff --git a/common/configack/config_ack_sender.go b/common/configack/config_ack_sender.go index 0a67c955..cb69a41b 100644 --- a/common/configack/config_ack_sender.go +++ b/common/configack/config_ack_sender.go @@ -18,12 +18,6 @@ import ( "google.golang.org/grpc" ) -// TODO: take minRetryInterval + maxRetryInterval + DialTimeout from config -const ( - minRetryInterval = 50 * time.Millisecond - maxRetryInterval = 10 * time.Second -) - type ConfigAcker interface { Stop() SubmitConfigAck(configSeq uint32) error @@ -41,6 +35,10 @@ type configAcker struct { ctx context.Context cancel context.CancelFunc + + minRetryInterval time.Duration + maxRetryInterval time.Duration + DialTimeout time.Duration } type ConnectionInfo struct { @@ -67,6 +65,9 @@ func NewConfigAcker(connInfo *ConnectionInfo, logger *flogging.FabricLogger) *co shard: connInfo.Shard, ctx: ctx, cancel: cancel, + minRetryInterval: 50 * time.Millisecond, // TODO: take from config + maxRetryInterval: 10 * time.Second, // TODO: take from config + DialTimeout: 5 * time.Second, // TODO: take from config } return ca } @@ -106,20 +107,20 @@ func (ca *configAcker) handleConfigAck(configSeq uint32) error { return nil } - interval := minRetryInterval + interval := ca.minRetryInterval numOfRetries := 1 for { select { case <-ctx.Done(): - return fmt.Errorf("sending config ack to consensus aborted, because context is done and configAcker stopped") + return fmt.Errorf("sending config ack to consensus aborted: %w", ctx.Err()) case <-time.After(interval): ca.logger.Debugf("Retry attempt #%d", numOfRetries) numOfRetries++ err := ca.tryToHandleConfigAck(ctx, configSeq) if err != nil { - interval = min(interval*2, maxRetryInterval) + interval = min(interval*2, ca.maxRetryInterval) ca.logger.Errorf("Sending config ack to consensus failed: %v, trying again in: %s", err, interval) continue } @@ -161,7 +162,7 @@ func (ca *configAcker) connectToConsenter() (*grpc.ClientConn, error) { Certificate: ca.tlsCert, RequireClientCert: true, }, - DialTimeout: time.Second * 5, + DialTimeout: ca.DialTimeout, } return cc.Dial(ca.consensusEndpoint) @@ -185,7 +186,7 @@ func (ca *configAcker) sendConfigAckToConsensus(ctx context.Context, conn *grpc. ca.logger.Infof("Sending ConfigAck for config sequence %d to consenter", configSeq) resp, err := client.AckConfig(ctx, configAckReq) if resp != nil && resp.GetError() != "" { - ca.logger.Warnf("Received bad response from consensus on config ack: %v\n", resp.Error) + ca.logger.Warnf("Received bad response from consensus on config ack: %s", resp.GetError()) } return err } diff --git a/common/configack/config_ack_sender_test.go b/common/configack/config_ack_sender_test.go index 7d0763f4..a961b2dc 100644 --- a/common/configack/config_ack_sender_test.go +++ b/common/configack/config_ack_sender_test.go @@ -78,9 +78,13 @@ func TestSubmitConfigAck_RetriesOnSendFailure(t *testing.T) { func TestSubmitConfigAck_RetriesOnConnectFailure(t *testing.T) { stubConsenter, configAcker := createTestSetupForConfigAcker(t, protos.NodeType_ROUTER, 0) + configAcker.DialTimeout = 2 * time.Second + configAcker.minRetryInterval = 2 * time.Millisecond + configAcker.maxRetryInterval = 2 * time.Second + // Start the server after a short delay go func() { - time.AfterFunc(20*time.Second, func() { + time.AfterFunc(10*time.Second, func() { stubConsenter.Start() }) }() @@ -108,7 +112,7 @@ func TestSubmitConfigAck_StopCancelsRetries(t *testing.T) { select { case err := <-done: require.Error(t, err) - require.ErrorContains(t, err, "sending config ack to consensus aborted, because context is done and configAcker stopped") + require.ErrorContains(t, err, "sending config ack to consensus aborted") case <-time.After(5 * time.Second): t.Fatal("SubmitConfigAck did not return after Stop was called") }