[coordinator] restrict batch size

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

Signed-off-by: Senthil Nathan N <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,72 @@ func (sv *signatureVerifier) sendTransactionsToSVService(
 			}
 		}
 
+		// If a batch exceeds the gRPC message size limit, stream.Send() will fail with a
+		// RESOURCE_EXHAUSTED error. This terminates the stream, and our connection lifecycle
+		// management will trigger a full retry of the operation. On the subsequent attempt (the retry),
+		// we assume the initial failure was due to an oversized first batch. To mitigate this,
+		// we split only the first batch of that retry into smaller chunks. All subsequent batches
+		// in the same attempt are sent without splitting.
+		// Hitting this limit is not expected during normal operation. It should only occur
+		// in rare scenarios, like the failure of many services or when numerous transactions
+		// depend on a single transaction. In those cases, we rely on the connection retry mechanism.
+
+		if firstBatch {
+			if err := splitAndSendToVerifier(stream, request); err != nil {
+				return errors.Wrap(err, "send to stream ended with error")
+			}
+			firstBatch = false
+			continue
+		}
+
 		if err := stream.Send(request); err != nil {
 			return errors.Wrap(err, "send to stream ended with error")
 		}
 		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 err
+		}
+	}
+
+	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 := createInputTxsNodeForSVM(t, blkNum, numTxs, 0)
 	channel.NewWriter(t.Context(), e.inputTxBatch).Write(txBatch)
 	return expectedValidatedTxs
 }
@@ -146,6 +147,45 @@ 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)
+	expectedValidatedTxs = make(dependencygraph.TxNodeBatch, 0, totalBlocks*txPerBlock)
+	for i := range totalBlocks {
+		//nolint:gosec // int -> uint64
+		txBatch, expectedValidatedTxsBatch := createInputTxsNodeForSVM(t, uint64(i+1), txPerBlock, 1024*1024)
+		txBatches[i] = txBatch
+		expectedValidatedTxs = append(expectedValidatedTxs, expectedValidatedTxsBatch...)
+	}
+
+	txsBatch := make(dependencygraph.TxNodeBatch, 0, totalBlocks*txPerBlock)
+	for _, b := range txBatches {
+		txsBatch = append(txsBatch, b...)
+	}
+
+	channel.NewWriter(t.Context(), env.inputTxBatch).Write(txsBatch)
+
+	actualValidatedTxs := make(dependencygraph.TxNodeBatch, 0, totalBlocks*txPerBlock)
+	for range totalBlocks {
+		txs := <-env.outputValidatedTxs
+		actualValidatedTxs = append(actualValidatedTxs, txs...)
+	}
+	require.Len(t, actualValidatedTxs, totalBlocks*txPerBlock)
+	require.ElementsMatch(t, expectedValidatedTxs, actualValidatedTxs)
+
+	test.EventuallyIntMetric(
+		t, totalBlocks*txPerBlock+1+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 +246,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,
+func createInputTxsNodeForSVM(
+	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 +292,7 @@ func createTxNodeBatchForTest(
 				Tx: &protovcservice.Transaction{
 					BlockNumber:           txNode.Tx.BlockNumber,
 					TxNum:                 txNode.Tx.TxNum,
+					Namespaces:            txNode.Tx.Namespaces,
 					PrelimInvalidTxStatus: sigInvalidTxStatus,
 				},
 			}

service/coordinator/validator_committer_manager.go

@@ -247,6 +247,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 +261,54 @@ func (vc *validatorCommitter) sendTransactionsToVCService(
 			txBatch[i] = txNode.Tx
 		}
 
-		err := stream.Send(&protovcservice.TransactionBatch{
+		// NOTE: This following logic is duplicated from the signature-verifier-manager.
+		//       This could be resolved by introducing interfaces or passing functions as arguments,
+		//       but that approach seems like overkill for now. We can refactor this into a more
+		//       general solution if the need arises in other services.
+
+		if firstBatch {
+			if err := splitAndSendToVC(stream, txBatch); err != nil {
+				return errors.Wrap(err, "send to stream ended with error")
+			}
+			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, "send to stream ended with error")
 		}
 		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 err
+		}
+	}
+
+	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"
@@ -111,7 +112,7 @@ func TestValidatorCommitterManager(t *testing.T) {
 	t.Run("Send tx batch to use any vcservice", func(t *testing.T) {
 		t.Parallel()
 		env := newVcMgrTestEnv(t, 2)
-		txBatch, expectedTxsStatus := createInputTxsNodeForTest(5, 1, 1)
+		txBatch, expectedTxsStatus := createInputTxsNodeForVCM(t, 1, 5, 0)
 		env.inputTxs <- txBatch
 
 		outTxs := <-env.outputTxs
@@ -125,14 +126,16 @@ func TestValidatorCommitterManager(t *testing.T) {
 			t, 5, 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 := createInputTxsNodeForVCM(t, 2, 5, 0)
+		txBatch2, expectedTxsStatus2 := createInputTxsNodeForVCM(t, 3, 5, 0)
 
 		require.Eventually(t, func() bool {
 			env.inputTxs <- txBatch1
@@ -281,7 +284,7 @@ func TestValidatorCommitterManagerRecovery(t *testing.T) {
 	env.requireRetriedTxsTotal(t, 0)
 
 	numTxs := 10
-	txBatch, expectedTxsStatus := createInputTxsNodeForTest(numTxs, 0, 0)
+	txBatch, expectedTxsStatus := createInputTxsNodeForVCM(t, 0, numTxs, 0)
 	env.inputTxs <- txBatch
 
 	require.Eventually(t, func() bool {
@@ -323,21 +326,84 @@ func TestValidatorCommitterManagerRecovery(t *testing.T) {
 	}, 2*time.Second, 1*time.Second)
 }
 
-func createInputTxsNodeForTest(numTxs, startIndex int, blkNum uint64) (
-	[]*dependencygraph.TransactionNode, *protoblocktx.TransactionsStatus,
+func TestValidatorCommitterManagerWithLargeSize(t *testing.T) {
+	t.Parallel()
+
+	env := newVcMgrTestEnv(t, 2)
+	txBatch, expectedTxsStatus := createInputTxsNodeForVCM(t, 1, 5, 0)
+	env.inputTxs <- txBatch
+
+	outTxs := <-env.outputTxs
+	require.ElementsMatch(t, txBatch, outTxs)
+
+	outTxsStatus := <-env.outputTxsStatus
+	require.Equal(t, expectedTxsStatus.Status, outTxsStatus.Status)
+
+	totalBlocks := 3
+	txPerBlock := 50
+	txBatches := make(dependencygraph.TxNodeBatch, 0, totalBlocks*txPerBlock)
+	expectedTxsStatus = &protoblocktx.TransactionsStatus{Status: make(map[string]*protoblocktx.StatusWithHeight)}
+
+	for i := range totalBlocks {
+		//nolint:gosec // int -> int64
+		txBatch, txStatus := createInputTxsNodeForVCM(t, uint64(i+2), txPerBlock, 1024*1024)
+		txBatches = append(txBatches, txBatch...)
+		maps.Copy(expectedTxsStatus.Status, txStatus.Status)
+	}
+
+	env.inputTxs <- txBatches
+
+	// txBatch would be split into three parts, one per block.
+	outTxs = nil
+	for range 3 {
+		outTxs = append(outTxs, <-env.outputTxs...)
+	}
+	require.ElementsMatch(t, txBatches, outTxs)
+
+	outTxsStatus.Status = make(map[string]*protoblocktx.StatusWithHeight)
+	for range totalBlocks {
+		s := <-env.outputTxsStatus
+		maps.Copy(outTxsStatus.Status, s.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,
+	)
+}
+
+func createInputTxsNodeForVCM(t *testing.T, blkNum uint64, numTxs, valueSize int) (
+	dependencygraph.TxNodeBatch, *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)