[coordinator] restrict batch size (#87)

#### Type of change

- Bug fix
 
#### Description

This commit restrict the batch size sent to verifier and vcservice when
it receives resource exhausted status code on grpc.

#### Related issues

  - Resolves #77

---------

Signed-off-by: Senthil Nathan N <cendhu@gmail.com>
Signed-off-by: senthil <cendhu@gmail.com>

Author: cendhu

service/coordinator/signature_verifier_manager.go

@@ -202,6 +202,7 @@ func (sv *signatureVerifier) sendTransactionsToSVService(
 	inputTxBatch channel.Reader[dependencygraph.TxNodeBatch],
 ) error {
 	var policyVersion uint64
+	firstBatch := true
 	for {
 		txBatch, ctxAlive := inputTxBatch.Read()
 		if !ctxAlive {
@@ -231,13 +232,62 @@ func (sv *signatureVerifier) sendTransactionsToSVService(
 			}
 		}
 
+		if firstBatch {
+			if err := splitAndSendToVerifier(stream, request); err != nil {
+				return err
+			}
+			firstBatch = false
+			continue
+		}
+
 		if err := stream.Send(request); err != nil {
-			return errors.Wrap(err, "send to stream ended with error")
+			return errors.Wrap(err, streamEndErrWrap)
 		}
 		logger.Debugf("Batch contains %d TXs, and was stored in the accumulator and sent to a sv", batchSize)
 	}
 }
 
+func splitAndSendToVerifier(
+	stream protosigverifierservice.Verifier_StartStreamClient,
+	r *protosigverifierservice.RequestBatch,
+) error {
+	// We group transactions by block to ensure our batch sizes do not exceed the gRPC message limit.
+	// This strategy prevents RESOURCE_EXHAUSTED errors because the orderer's maximum block size
+	// will be configured to be safely smaller than the gRPC send/receive limit.
+	// For added safety, we can split each block's transactions into more batches, but this is deferred for now
+	// until the orderer implements all sanity checks on the configuration provided in the config block.
+	// For example, if the orderer can enforce that the maximum block size should be at most half of the
+	// maximum message size in gRPC, one batch would be adequate.
+	blkToBatch := make(map[uint64]*protosigverifierservice.RequestBatch)
+	for _, req := range r.Requests {
+		rBatch, ok := blkToBatch[req.BlockNum]
+		if !ok {
+			rBatch = &protosigverifierservice.RequestBatch{
+				Requests: make([]*protosigverifierservice.Request, 0, len(r.Requests)),
+			}
+			blkToBatch[req.BlockNum] = rBatch
+		}
+
+		rBatch.Requests = append(rBatch.Requests, req)
+	}
+
+	updateSent := false
+	for _, rBatch := range blkToBatch {
+		if !updateSent {
+			rBatch.Update = r.Update
+			updateSent = false
+		}
+
+		if err := stream.Send(rBatch); err != nil {
+			// ResourceExhausted should not occur here, as we have split a block's transactions
+			// into two batches, assuming the block size is less than the maximum gRPC message size.
+			return errors.Wrap(err, streamEndErrWrap)
+		}
+	}
+
+	return nil
+}
+
 func (sv *signatureVerifier) receiveStatusAndForwardToOutput(
 	stream protosigverifierservice.Verifier_StartStreamClient,
 	outputValidatedTxs channel.Writer[dependencygraph.TxNodeBatch],

service/coordinator/signature_verifier_manager_test.go

@@ -8,6 +8,7 @@ package coordinator
 
 import (
 	"context"
+	"crypto/rand"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -87,7 +88,7 @@ func newSvMgrTestEnv(t *testing.T, numSvService int, expectedEndErrorMsg ...byte
 func (e *svMgrTestEnv) submitTxBatch(t *testing.T, numTxs int) dependencygraph.TxNodeBatch {
 	t.Helper()
 	blkNum := e.curBlockNum.Add(1) - 1
-	txBatch, expectedValidatedTxs := createTxNodeBatchForTest(t, blkNum, numTxs)
+	txBatch, expectedValidatedTxs := createTxNodeBatchForTest(t, blkNum, numTxs, 0)
 	channel.NewWriter(t.Context(), e.inputTxBatch).Write(txBatch)
 	return expectedValidatedTxs
 }
@@ -146,6 +147,36 @@ func TestSignatureVerifierManagerWithSingleVerifier(t *testing.T) {
 	)
 }
 
+func TestSignatureVerifierManagerWithLargeSize(t *testing.T) {
+	t.Parallel()
+	env := newSvMgrTestEnv(t, 1)
+
+	expectedValidatedTxs := env.submitTxBatch(t, 1)
+	env.requireTxBatch(t, expectedValidatedTxs)
+
+	totalBlocks := 3
+	txPerBlock := 50
+	txBatches := make([]dependencygraph.TxNodeBatch, totalBlocks)
+	expectedValidatedTxsBatches := make([]dependencygraph.TxNodeBatch, totalBlocks)
+	for i := range 3 {
+		//nolint:gosec // int -> uint64
+		txBatches[i], expectedValidatedTxsBatches[i] = createTxNodeBatchForTest(t, uint64(i+1), txPerBlock, 1024*1024)
+	}
+
+	txsBatch := make(dependencygraph.TxNodeBatch, 0, totalBlocks*txPerBlock)
+	for _, b := range txBatches {
+		txsBatch = append(txsBatch, b...)
+	}
+
+	channel.NewWriter(t.Context(), env.inputTxBatch).Write(txsBatch)
+	// env.requireTxBatch(t, expectedValidatedTxs)
+
+	test.EventuallyIntMetric(
+		t, totalBlocks*txPerBlock+1, env.signVerifierManager.config.metrics.sigverifierTransactionProcessedTotal,
+		30*time.Second, 10*time.Millisecond,
+	)
+}
+
 func TestSignatureVerifierManagerWithMultipleVerifiers(t *testing.T) {
 	t.Parallel()
 	env := newSvMgrTestEnv(t, 2)
@@ -206,19 +237,37 @@ func TestSignatureVerifierWithAllInvalidTxs(t *testing.T) {
 	}
 
 	env := newSvMgrTestEnv(t, 1)
+	env.requireTxBatch(t, env.submitTxBatch(t, 1))
+
 	channel.NewWriter(t.Context(), env.inputTxBatch).Write(txBatch)
 	env.requireTxBatch(t, expectedValidatedTxs)
 }
 
 func createTxNodeBatchForTest(
-	_ *testing.T,
-	blkNum uint64, numTxs int,
+	t *testing.T,
+	blkNum uint64, numTxs, valueSize int,
 ) (inputTxBatch, expectedValidatedTxs dependencygraph.TxNodeBatch) {
+	t.Helper()
+
+	b := make([]byte, valueSize)
+	_, err := rand.Read(b)
+	require.NoError(t, err)
+
+	ns := []*protoblocktx.TxNamespace{
+		{
+			BlindWrites: []*protoblocktx.Write{
+				{
+					Value: b,
+				},
+			},
+		},
+	}
 	for i := range numTxs {
 		txNode := &dependencygraph.TransactionNode{
 			Tx: &protovcservice.Transaction{
 				BlockNumber: blkNum,
 				TxNum:       uint32(i), //nolint:gosec
+				Namespaces:  ns,
 			},
 		}
 
@@ -234,6 +283,7 @@ func createTxNodeBatchForTest(
 				Tx: &protovcservice.Transaction{
 					BlockNumber:           txNode.Tx.BlockNumber,
 					TxNum:                 txNode.Tx.TxNum,
+					Namespaces:            ns,
 					PrelimInvalidTxStatus: sigInvalidTxStatus,
 				},
 			}

service/coordinator/validator_committer_manager.go

@@ -25,6 +25,8 @@ import (
 	"github.com/hyperledger/fabric-x-committer/utils/monitoring/promutil"
 )
 
+const streamEndErrWrap = "sending to stream ended with an error"
+
 type (
 	// validatorCommitterManager is responsible for managing all communication with
 	// all vcservices. It is responsible for:
@@ -247,6 +249,7 @@ func (vc *validatorCommitter) sendTransactionsToVCService(
 	stream protovcservice.ValidationAndCommitService_StartValidateAndCommitStreamClient,
 	inputTxsNode channel.Reader[dependencygraph.TxNodeBatch],
 ) error {
+	firstBatch := true
 	for {
 		txsNode, ok := inputTxsNode.Read()
 		if !ok {
@@ -260,17 +263,49 @@ func (vc *validatorCommitter) sendTransactionsToVCService(
 			txBatch[i] = txNode.Tx
 		}
 
-		err := stream.Send(&protovcservice.TransactionBatch{
+		if firstBatch {
+			if err := splitAndSendToVC(stream, txBatch); err != nil {
+				return err
+			}
+			firstBatch = false
+			continue
+		}
+
+		if err := stream.Send(&protovcservice.TransactionBatch{
 			Transactions: txBatch,
-		})
-		if err != nil {
-			// The stream ended or the VCM was closed.
-			return errors.Wrap(err, "receive from stream ended with error")
+		}); err != nil {
+			return errors.Wrap(err, streamEndErrWrap)
 		}
 		logger.Debugf("TX node contains %d TXs, and was sent to a vcservice", len(txBatch))
 	}
 }
 
+func splitAndSendToVC(
+	stream protovcservice.ValidationAndCommitService_StartValidateAndCommitStreamClient,
+	txBatch []*protovcservice.Transaction,
+) error {
+	blkToBatch := make(map[uint64]*protovcservice.TransactionBatch)
+	for _, tx := range txBatch {
+		rBatch, ok := blkToBatch[tx.BlockNumber]
+		if !ok {
+			rBatch = &protovcservice.TransactionBatch{
+				Transactions: make([]*protovcservice.Transaction, 0, len(txBatch)),
+			}
+			blkToBatch[tx.BlockNumber] = rBatch
+		}
+
+		rBatch.Transactions = append(rBatch.Transactions, tx)
+	}
+
+	for _, rBatch := range blkToBatch {
+		if err := stream.Send(rBatch); err != nil {
+			return errors.Wrap(err, streamEndErrWrap)
+		}
+	}
+
+	return nil
+}
+
 func (vc *validatorCommitter) receiveStatusAndForwardToOutput(
 	stream protovcservice.ValidationAndCommitService_StartValidateAndCommitStreamClient,
 	outputTxsNode channel.Writer[dependencygraph.TxNodeBatch],

service/coordinator/validator_committer_manager_test.go

@@ -8,11 +8,12 @@ package coordinator
 
 import (
 	"context"
+	"crypto/rand"
 	"maps"
-	"strconv"
 	"testing"
 	"time"
 
+	"github.com/google/uuid"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"google.golang.org/protobuf/proto"
@@ -99,7 +100,7 @@ func (e *vcMgrTestEnv) requireRetriedTxsTotal(t *testing.T, expectedRetriedTxsTo
 	)
 }
 
-func TestValidatorCommitterManager(t *testing.T) {
+func TestValidatorCommitterManagerX(t *testing.T) {
 	t.Parallel()
 
 	ensureZeroWaitingTxs := func(env *vcMgrTestEnv) {
@@ -108,10 +109,10 @@ func TestValidatorCommitterManager(t *testing.T) {
 		}
 	}
 
-	t.Run("Send tx batch to use any vcservice", func(t *testing.T) {
+	t.Run("Send tx batch to use any vcservice and send a batch with larger size", func(t *testing.T) {
 		t.Parallel()
 		env := newVcMgrTestEnv(t, 2)
-		txBatch, expectedTxsStatus := createInputTxsNodeForTest(5, 1, 1)
+		txBatch, expectedTxsStatus := createInputTxsNodeForTest(t, 5, 0, 1)
 		env.inputTxs <- txBatch
 
 		outTxs := <-env.outputTxs
@@ -125,14 +126,49 @@ func TestValidatorCommitterManager(t *testing.T) {
 			t, 5, env.validatorCommitterManager.config.metrics.vcserviceTransactionProcessedTotal,
 			2*time.Second, 100*time.Millisecond,
 		)
+
+		totalBlocks := 3
+		txPerBlock := 50
+		txBatches := make(dependencygraph.TxNodeBatch, 0, totalBlocks*txPerBlock)
+		expectedTxsStatus = &protoblocktx.TransactionsStatus{Status: make(map[string]*protoblocktx.StatusWithHeight)}
+
+		for i := range 3 {
+			//nolint:gosec // int -> int64
+			txBatch, txStatus := createInputTxsNodeForTest(t, txPerBlock, 1024*1024, uint64(i+2))
+			txBatches = append(txBatches, txBatch...)
+			maps.Copy(expectedTxsStatus.Status, txStatus.Status)
+		}
+
+		env.inputTxs <- txBatches
+
+		// txBatch would be split into three parts, one per block.
+		outTxs = <-env.outputTxs
+		outTxs = append(outTxs, <-env.outputTxs...)
+		outTxs = append(outTxs, <-env.outputTxs...)
+		require.ElementsMatch(t, txBatches, outTxs)
+
+		outTxsStatus = <-env.outputTxsStatus
+		status := <-env.outputTxsStatus
+		maps.Copy(outTxsStatus.Status, status.Status)
+		status = <-env.outputTxsStatus
+		maps.Copy(outTxsStatus.Status, status.Status)
+		require.Equal(t, expectedTxsStatus.Status, outTxsStatus.Status)
+
+		test.EventuallyIntMetric(
+			t, 5+totalBlocks*txPerBlock,
+			env.validatorCommitterManager.config.metrics.vcserviceTransactionProcessedTotal,
+			2*time.Second, 100*time.Millisecond,
+		)
+
 		ensureZeroWaitingTxs(env)
 	})
 
 	t.Run("send batches to ensure all vcservices are used", func(t *testing.T) {
 		t.Parallel()
 		env := newVcMgrTestEnv(t, 2)
-		txBatch1, expectedTxsStatus1 := createInputTxsNodeForTest(5, 1, 2)
-		txBatch2, expectedTxsStatus2 := createInputTxsNodeForTest(5, 6, 3)
+
+		txBatch1, expectedTxsStatus1 := createInputTxsNodeForTest(t, 5, 0, 2)
+		txBatch2, expectedTxsStatus2 := createInputTxsNodeForTest(t, 5, 0, 3)
 
 		require.Eventually(t, func() bool {
 			env.inputTxs <- txBatch1
@@ -281,7 +317,7 @@ func TestValidatorCommitterManagerRecovery(t *testing.T) {
 	env.requireRetriedTxsTotal(t, 0)
 
 	numTxs := 10
-	txBatch, expectedTxsStatus := createInputTxsNodeForTest(numTxs, 0, 0)
+	txBatch, expectedTxsStatus := createInputTxsNodeForTest(t, numTxs, 0, 0)
 	env.inputTxs <- txBatch
 
 	require.Eventually(t, func() bool {
@@ -323,21 +359,36 @@ func TestValidatorCommitterManagerRecovery(t *testing.T) {
 	}, 2*time.Second, 1*time.Second)
 }
 
-func createInputTxsNodeForTest(numTxs, startIndex int, blkNum uint64) (
+func createInputTxsNodeForTest(t *testing.T, numTxs, valueSize int, blkNum uint64) (
 	[]*dependencygraph.TransactionNode, *protoblocktx.TransactionsStatus,
 ) {
+	t.Helper()
+
 	txsNode := make([]*dependencygraph.TransactionNode, numTxs)
 	expectedTxsStatus := &protoblocktx.TransactionsStatus{
 		Status: make(map[string]*protoblocktx.StatusWithHeight),
 	}
 
+	b := make([]byte, valueSize)
+	_, err := rand.Read(b)
+	require.NoError(t, err)
+
 	for i := range numTxs {
-		id := "tx" + strconv.Itoa(startIndex+i)
+		id := uuid.NewString()
 		txsNode[i] = &dependencygraph.TransactionNode{
 			Tx: &protovcservice.Transaction{
 				ID:          id,
 				BlockNumber: blkNum,
 				TxNum:       uint32(i), //nolint:gosec
+				Namespaces: []*protoblocktx.TxNamespace{
+					{
+						BlindWrites: []*protoblocktx.Write{
+							{
+								Value: b,
+							},
+						},
+					},
+				},
 			},
 		}
 		expectedTxsStatus.Status[id] = types.CreateStatusWithHeight(protoblocktx.Status_COMMITTED, blkNum, i)

utils/connection/client_util.go

@@ -34,8 +34,10 @@ const (
 
 	// defaultGrpcMaxAttempts is set to a high number to allow the timeout to dictate the retry end condition.
 	defaultGrpcMaxAttempts = 1024
-	maxMsgSize             = 100 * 1024 * 1024
-	scResolverSchema       = "sc.connection"
+	// TODO: All services including the orderer must use the same default maximum message size.
+	//       Hence, we need to move this constant to fabric-x-common.
+	maxMsgSize       = 100 * 1024 * 1024
+	scResolverSchema = "sc.connection"
 )
 
 type (