Issue: Potential Blocking in BroadcastConsensus May Compromise BFT Liveness

[zhanghefan123]

The current implementation of BroadcastConsensus in the Controller may block when sending messages to other nodes, potentially compromising the BFT protocol's liveness. This occurs because:

c.Comm.SendConsensus(node, m) can block when gRPC's getTransport function has no available connections

The synchronous nature of the broadcast means a single blocked send can delay all subsequent messages

This blocking behavior could prevent timely dissemination of consensus messages, violating the protocol's liveness guarantees

func (c *Controller) BroadcastConsensus(m *protos.Message) {
    for _, node := range c.NodesList {
        if c.ID == node {
            continue
        }
        c.Comm.SendConsensus(node, m) // Synchronous call that may block
    }
    if m.GetPrePrepare() != nil || m.GetPrepare() != nil || m.GetCommit() != nil {
        if leader, _ := c.iAmTheLeader(); leader {
            c.LeaderMonitor.HeartbeatWasSent()
        }
    }
}

the proposed solution is as follows:

func (c *Controller) BroadcastConsensus(m *protos.Message) {
    for _, node := range c.NodesList {
        if c.ID == node {
            continue
        }
        go c.Comm.SendConsensus(node, m) // Asynchronous non-blocking call
    }
    if m.GetPrePrepare() != nil || m.GetPrepare() != nil || m.GetCommit() != nil {
        if leader, _ := c.iAmTheLeader(); leader {
            c.LeaderMonitor.HeartbeatWasSent()
        }
    }
}

[pfi79]

What makes you think SendConsensus is blocking? Have you encountered such a thing in fabric?
I haven't.

SendConsensus provides an external application like fabric. There is no blocking there and everything is fine.

[HagarMeir]

Maybe we can update the documentation to be clear it should not be blocking.
Here: https://github.com/hyperledger-labs/SmartBFT/blob/main/pkg/api/dependencies.go

[zhanghefan123]

In our experiment, we deployed four Fabric ordering nodes using veth pairs and Docker containers to simulate a network environment. We then disabled fabric orderer node 4’s network interface to test fault tolerance.

Under normal conditions, the transactions per second (TPS) should remain stable, as the BFT-SMaRT protocol is designed to tolerate f faulty nodes in a 3f+1 configuration. However, we observed an immediate drop in TPS, contrary to expectations.

We examine the code and use

func (c *Controller) BroadcastConsensus(m *protos.Message) {
	defer func(startTime time.Time) {
		timeDuration := time.Since(startTime)
		if timeDuration > time.Second {
			fmt.Printf("broadcast consensus takes %v\n to finish\n", timeDuration)
		}
	}(time.Now())
	if c.EnableRoutine {
		for _, node := range c.NodesList {
			if c.ID == node {
				continue
			}
			go c.Comm.SendConsensus(node, m)
		}
	}
if m.GetPrePrepare() != nil || m.GetPrepare() != nil || m.GetCommit() != nil {
		if leader, _ := c.iAmTheLeader(); leader {
			c.LeaderMonitor.HeartbeatWasSent()
		}
	}
}

Upon examining the logs, we discovered that the broadcast consensus process took over 20 seconds to complete, which leads to the observed drop in TPS.

Through further analysis, we identified that the getTransport function blocks the sendConsensus process. We suspect this occurs when consensus nodes fail to find available subconnections, ultimately causing the process to terminate upon timeout expiration.

op := func(a *csAttempt) error {
		if err := a.getTransport(); err != nil {
			return err
		}
		if err := a.newStream(); err != nil {
			return err
		}
		// Because this operation is always called either here (while creating
		// the clientStream) or by the retry code while locked when replaying
		// the operation, it is safe to access cs.attempt directly.
		cs.attempt = a
		return nil
	}

[pfi79]

Show your SendConsensus implementation

[zhanghefan123]

I haven't made any modifications to sendConsensus. The issue actually lies with the getOrCreateStream function, which is causing the block.

// SendConsensus passes the given ConsensusRequest message to the raft.Node instance.
func (s *RPC) SendConsensus(destination uint64, msg *orderer.ConsensusRequest) error {
	if s.Logger.IsEnabledFor(zapcore.DebugLevel) {
		defer s.consensusSent(time.Now(), destination, msg)
	}

	stream, err := s.getOrCreateStream(destination, ConsensusOperation)
	if err != nil {
		return err
	}

	req := &orderer.StepRequest{
		Payload: &orderer.StepRequest_ConsensusRequest{
			ConsensusRequest: msg,
		},
	}

	s.consensusLock.Lock()
	defer s.consensusLock.Unlock()

	err = stream.Send(req)
	if err != nil {
		s.unMapStream(destination, ConsensusOperation, stream.ID)
	}

	return err
}

[zhanghefan123]

getOrCreateStream ->
stream, err = stub.NewStream(s.Timeout) ->
rc.GetStreamFunc(ctx) ->
clusterClient.Step(ctx) ->
c.cc.NewStream(ctx, &ClusterNodeService_ServiceDesc.Streams[0], ClusterNodeService_Step_FullMethodName, opts...)
->return newClientStream(ctx, desc, cc, method, opts...)
-> newClientStreamWithParams(ctx, desc, cc, method, mc, onCommit, done, opts...)
-> getTransport() (blocks here)

[pfi79]

So you've been testing the work fabric.

I think the problem is in the fabric's messaging functions?

[zhanghefan123]

Your analysis accurately identifies the root cause. The observed blocking behavior indeed stems from gRPC's implementation characteristics.

A critical distinction exists between graceful node shutdown and abrupt interface termination. Unlike proper node termination that follows TCP's four-way handshake protocol, interface disruption prevents this orderly connection teardown. Consequently, peer nodes interpret this scenario as transient network packet loss rather than permanent node failure.

This misclassification leads to persistent retry attempts to communicate with the affected orderer (Orderer4), resulting in both wasted network resources and system performance degradation. The gRPC layer's handling of such edge cases appears to be a contributing factor to this suboptimal behavior.

getOrCreateStream ->
stream, err = stub.NewStream(s.Timeout) ->
rc.GetStreamFunc(ctx) ->
clusterClient.Step(ctx) ->
c.cc.NewStream(ctx, &ClusterNodeService_ServiceDesc.Streams[0], ClusterNodeService_Step_FullMethodName, opts...)
->return newClientStream(ctx, desc, cc, method, opts...)
-> newClientStreamWithParams(ctx, desc, cc, method, mc, onCommit, done, opts...)
-> getTransport() (blocks here)
-> cs.cc.getTransport
-> cc.pickerWrapper.pick

// pick returns the transport that will be used for the RPC.
// It may block in the following cases:
// - there's no picker
// - the current picker returns ErrNoSubConnAvailable
// - the current picker returns other errors and failfast is false.
// - the subConn returned by the current picker is not READY
// When one of these situations happens, pick blocks until the picker gets updated.
func (pw *pickerWrapper) pick(ctx context.Context, failfast bool, info balancer.PickInfo) (transport.ClientTransport, balancer.PickResult, error) {
...
}

[zhanghefan123]

👋👋👋

[yacovm]

I think if we change this, to:

if connState != connectivity.Ready {

it will fix it, but testing needs to be done.

@zhanghefan123 do you want to test this?

[zhanghefan123]

Yes, I can take care of the testing. I'll verify the behavior with the updated code if connState != connectivity.Ready { and report my observations.