txnsync: Use dynamic maxEncodedTransactionGroups (algorand#3168)

Change constants in the txnsync around message size limits into variables based on maxTxPoolSize. This will allow for support of larger message sizes when dealing with larger load.

Reran existing tests

Author: nicholasguoalgorand

txnsync/encodedgroups_test.go

@@ -30,7 +30,7 @@ import (
 func TestBadBitmask(t *testing.T) {
 	partitiontest.PartitionTest(t)
 
-	txnGroups, genesisID, genesisHash, err := txnGroupsData(96)
+	txnGroups, genesisID, genesisHash, err := txnGroupsData(50)
 	require.NoError(t, err)
 
 	var s syncState
@@ -41,7 +41,7 @@ func TestBadBitmask(t *testing.T) {
 	require.Equal(t, errIndexNotFound, err)
 }
 
-// corrupted bitmask may bcause panic during decoding. This test is to make sure it is an error and not a panic
+// corrupted bitmask may cause panic during decoding. This test is to make sure it is an error and not a panic
 func badEncodeTransactionGroups(t *testing.T, s *syncState, inTxnGroups []pooldata.SignedTxGroup, dataExchangeRate uint64) (packedTransactionGroups, error) {
 	txnCount := 0
 	for _, txGroup := range inTxnGroups {

txnsync/encodedgroupstypes.go

@@ -26,11 +26,12 @@ import (
 	"github.com/algorand/go-algorand/protocol"
 )
 
-const maxEncodedTransactionGroups = 30000
-const maxEncodedTransactionGroupEntries = 30000
-const maxBitmaskSize = (maxEncodedTransactionGroupEntries+7)/8 + 1
-const maxSignatureBytes = maxEncodedTransactionGroupEntries * len(crypto.Signature{})
-const maxAddressBytes = maxEncodedTransactionGroupEntries * crypto.DigestSize
+// set in init() in service.go
+var maxEncodedTransactionGroups int
+var maxEncodedTransactionGroupEntries int
+var maxBitmaskSize int
+var maxSignatureBytes int
+var maxAddressBytes int
 
 var errInvalidTxType = errors.New("invalid txtype")
 

txnsync/exchange.go

@@ -22,10 +22,13 @@ import (
 )
 
 const txnBlockMessageVersion = 1
-const maxBloomFilterSize = 100000
 const maxAcceptedMsgSeq = 64
-const maxEncodedTransactionGroupBytes = 10000000
-const maxProposalSize = 3500000 // 10K * 32 + sizeof(block header)-
+
+// set in init() in service.go
+var maxBloomFilterSize int
+var maxEncodedTransactionGroupBytes int
+
+const maxProposalSize = 3500000 // 10K * 32 + sizeof(block header)
 
 type transactionBlockMessage struct {
 	_struct struct{} `codec:",omitempty,omitemptyarray"` //nolint:structcheck,unused

txnsync/incoming.go

@@ -60,7 +60,7 @@ func (s *syncState) asyncIncomingMessageHandler(networkPeer interface{}, peer *P
 	_, err = incomingMessage.message.UnmarshalMsg(message)
 	if err != nil {
 		// if we received a message that we cannot parse, disconnect.
-		s.log.Infof("received unparsable transaction sync message from peer. disconnecting from peer.")
+		s.log.Infof("received unparsable transaction sync message from peer. disconnecting from peer: %v, bytes: %d", err, len(message))
 		s.incomingMessagesQ.erase(peer, networkPeer)
 		return err
 	}

txnsync/service.go

@@ -51,9 +51,25 @@ func MakeTransactionSyncService(log logging.Logger, conn NodeConnector, isRelay
 	}
 	s.state.service = s
 	s.state.xorBuilder.MaxIterations = 10
+
+	setTransactionSyncVariables(cfg)
 	return s
 }
 
+func setTransactionSyncVariables(cfg config.Local) {
+	if cfg.TxPoolSize < maxEncodedTransactionGroups {
+		return
+	}
+	maxEncodedTransactionGroups = cfg.TxPoolSize
+	maxEncodedTransactionGroupEntries = cfg.TxPoolSize
+	maxBitmaskSize = (maxEncodedTransactionGroupEntries+7)/8 + 1
+	maxSignatureBytes = maxEncodedTransactionGroupEntries * len(crypto.Signature{})
+	maxAddressBytes = maxEncodedTransactionGroupEntries * crypto.DigestSize
+
+	maxBloomFilterSize = cfg.TxPoolSize * 5                  // 32 bit xor uses slightly more than 4 bytes/element.
+	maxEncodedTransactionGroupBytes = cfg.TxPoolSize * 10000 // assume each transaction takes 10KB, as a worst-case-scenario for bounding purposes only.
+}
+
 // Start starts the transaction sync
 func (s *Service) Start() {
 	s.ctx, s.cancelCtx = context.WithCancel(context.Background())
@@ -76,3 +92,7 @@ func (s *Service) Stop() {
 func (s *Service) GetIncomingMessageHandler() IncomingMessageHandler {
 	return s.state.asyncIncomingMessageHandler
 }
+
+func init() {
+	setTransactionSyncVariables(config.GetDefaultLocal())
+}

txnsync/txngroups.go

@@ -152,7 +152,7 @@ func decodeTransactionGroups(ptg packedTransactionGroups, genesisID string, gene
 		return nil, err
 	}
 
-	if stub.TransactionGroupCount > maxEncodedTransactionGroups {
+	if stub.TransactionGroupCount > uint64(maxEncodedTransactionGroups) {
 		return nil, errors.New("invalid TransactionGroupCount")
 	}
 
@@ -191,7 +191,7 @@ func decodeTransactionGroups(ptg packedTransactionGroups, genesisID string, gene
 
 func decompressTransactionGroupsBytes(data []byte, lenDecompressedBytes uint64) (decoded []byte, err error) {
 	compressionRatio := lenDecompressedBytes / uint64(len(data)) // data should have been compressed between 0 and 95%
-	if lenDecompressedBytes > maxEncodedTransactionGroupBytes || compressionRatio <= 0 || compressionRatio >= maxCompressionRatio {
+	if lenDecompressedBytes > uint64(maxEncodedTransactionGroupBytes) || compressionRatio <= 0 || compressionRatio >= maxCompressionRatio {
 		return nil, fmt.Errorf("invalid lenDecompressedBytes: %d, len(data): %d", lenDecompressedBytes, len(data))
 	}
 

txnsync/txngroups_test.go

@@ -232,7 +232,7 @@ func txnGroupsData(numBlocks int) (txnGroups []pooldata.SignedTxGroup, genesisID
 func TestTxnGroupEncodingLarge(t *testing.T) {
 	partitiontest.PartitionTest(t)
 
-	txnGroups, genesisID, genesisHash, err := txnGroupsData(969)
+	txnGroups, genesisID, genesisHash, err := txnGroupsData(500)
 	require.NoError(t, err)
 
 	var s syncState
@@ -268,10 +268,10 @@ func TestTxnGroupEncodingLarge(t *testing.T) {
 		}
 	}
 	require.Equal(t, 2, len(count))
-	require.Equal(t, 18351, count["axfer"])
-	require.Equal(t, 1663, count["pay"])
-	require.Equal(t, 20005, sigs)
-	require.Equal(t, 9, msigs)
+	require.Equal(t, 9834, count["axfer"])
+	require.Equal(t, 850, count["pay"])
+	require.Equal(t, 10678, sigs)
+	require.Equal(t, 6, msigs)
 	require.Equal(t, 0, lsigs)
 }