Fix #4162: limit concurrent outgoing TransactionListQuery conversations

When many peers connect at startup and we're out of sync, sendTransactionListQuery
can be called for many peers simultaneously. Each peer responds with TransactionList
messages that cause BBolt writes, starving read operations (e.g. notifier startup).

Add a counting semaphore (default 3, configurable via network.v2.maxtransactionlistqueries)
that limits concurrent outgoing TransactionListQuery conversations. Queries that
exceed the limit are dropped; the gossip/state-comparison mechanism will retry them.
The slot is released when the last TransactionList message of the conversation arrives.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: reinkrul

network/transport/v2/protocol.go

@@ -54,18 +54,24 @@ type Config struct {
 	GossipInterval int `koanf:"gossipinterval"`
 	// DiagnosticsInterval specifies how often (in milliseconds) the node should broadcast its diagnostics message.
 	DiagnosticsInterval int `koanf:"diagnosticsinterval"`
+	// MaxConcurrentTransactionListQueries limits the number of outgoing TransactionListQuery conversations active at
+	// the same time. Limiting this reduces BBolt write lock contention during peer synchronization at startup.
+	// See https://github.com/nuts-foundation/nuts-node/issues/4162
+	MaxConcurrentTransactionListQueries int `koanf:"maxtransactionlistqueries"`
 }
 
 const defaultPayloadRetryDelay = 5 * time.Second
 const defaultGossipInterval = 5000
 const defaultDiagnosticsInterval = 5000
+const defaultMaxConcurrentTransactionListQueries = 3
 
 // DefaultConfig returns the default config for protocol v2
 func DefaultConfig() Config {
 	return Config{
-		PayloadRetryDelay:   defaultPayloadRetryDelay,
-		GossipInterval:      defaultGossipInterval,
-		DiagnosticsInterval: defaultDiagnosticsInterval,
+		PayloadRetryDelay:                   defaultPayloadRetryDelay,
+		GossipInterval:                      defaultGossipInterval,
+		DiagnosticsInterval:                 defaultDiagnosticsInterval,
+		MaxConcurrentTransactionListQueries: defaultMaxConcurrentTransactionListQueries,
 	}
 }
 
@@ -79,16 +85,20 @@ func New(
 	diagnosticsProvider func() transport.Diagnostics,
 	dagStore stoabs.KVStore,
 ) transport.Protocol {
+	if config.MaxConcurrentTransactionListQueries <= 0 {
+		config.MaxConcurrentTransactionListQueries = defaultMaxConcurrentTransactionListQueries
+	}
 	ctx, cancel := context.WithCancel(context.Background())
 	p := &protocol{
-		cancel:      cancel,
-		config:      config,
-		ctx:         ctx,
-		state:       state,
-		nodeDID:     nodeDID,
-		decrypter:   decrypter,
-		docResolver: docResolver,
-		dagStore:    dagStore,
+		cancel:                  cancel,
+		config:                  config,
+		ctx:                     ctx,
+		state:                   state,
+		nodeDID:                 nodeDID,
+		decrypter:               decrypter,
+		docResolver:             docResolver,
+		dagStore:                dagStore,
+		transactionListQuerySem: make(chan struct{}, config.MaxConcurrentTransactionListQueries),
 	}
 	p.sender = p
 	p.diagnosticsMan = newPeerDiagnosticsManager(diagnosticsProvider, p.sender.broadcastDiagnostics)
@@ -113,6 +123,10 @@ type protocol struct {
 	sender                 messageSender
 	listHandler            *transactionListHandler
 	dagStore               stoabs.KVStore
+	// transactionListQuerySem is a counting semaphore that limits the number of concurrent outgoing
+	// TransactionListQuery conversations. See https://github.com/nuts-foundation/nuts-node/issues/4162
+	transactionListQuerySem   chan struct{}
+	transactionListQueryConvs sync.Map // tracks conversationID strings that hold a semaphore slot
 }
 
 func (p *protocol) CreateClientStream(outgoingContext context.Context, grpcConn grpcLib.ClientConnInterface) (grpcLib.ClientStream, error) {

network/transport/v2/senders.go

@@ -82,6 +82,20 @@ func (p *protocol) sendTransactionListQuery(connection grpc.Connection, refs []h
 		return nil
 	}
 
+	// Limit the number of concurrent outgoing TransactionListQuery conversations to avoid flooding BBolt with
+	// write operations when many peers respond simultaneously. See https://github.com/nuts-foundation/nuts-node/issues/4162
+	select {
+	case p.transactionListQuerySem <- struct{}{}:
+		p.transactionListQueryConvs.Store(conversation.conversationID.String(), struct{}{})
+	default:
+		log.Logger().
+			WithFields(connection.Peer().ToFields()).
+			WithField(core.LogFieldConversationID, conversation.conversationID.String()).
+			Debug("TransactionListQuery limit reached, not requesting TransactionList from peer")
+		p.cMan.done(conversation.conversationID)
+		return nil
+	}
+
 	log.Logger().
 		WithFields(connection.Peer().ToFields()).
 		WithField(core.LogFieldConversationID, conversation.conversationID.String()).

network/transport/v2/senders_test.go

@@ -21,6 +21,7 @@ package v2
 
 import (
 	"errors"
+	"fmt"
 	"github.com/stretchr/testify/require"
 	"math"
 	"testing"
@@ -168,6 +169,29 @@ func TestProtocol_sendTransactionListQuery(t *testing.T) {
 		assert.NotNil(t, actualEnvelope.GetTransactionListQuery().GetConversationID())
 	})
 
+	t.Run("limit reached - query is dropped", func(t *testing.T) {
+		proto, mocks := newTestProtocol(t, nil)
+		// Fill the semaphore to the limit (default 3), using different peers so the per-peer
+		// conversation guard doesn't block us.
+		for i := range proto.config.MaxConcurrentTransactionListQueries {
+			mockConn := grpc.NewMockConnection(mocks.Controller)
+			mockConn.EXPECT().Peer().AnyTimes().Return(transport.Peer{ID: transport.PeerID(fmt.Sprintf("peer-%d", i))})
+			mockConn.EXPECT().Send(proto, gomock.Any(), false).Return(nil)
+			_ = proto.sendTransactionListQuery(mockConn, []hash.SHA256Hash{hash.FromSlice([]byte("list query"))})
+		}
+		require.Equal(t, proto.config.MaxConcurrentTransactionListQueries, len(proto.transactionListQuerySem))
+
+		// Next query on a new peer should be dropped (semaphore full).
+		overLimitConn := grpc.NewMockConnection(mocks.Controller)
+		overLimitConn.EXPECT().Peer().AnyTimes().Return(transport.Peer{ID: "over-limit"})
+		// No Send expected: query must be dropped.
+
+		err := proto.sendTransactionListQuery(overLimitConn, []hash.SHA256Hash{hash.FromSlice([]byte("list query"))})
+
+		assert.NoError(t, err)
+		assert.Equal(t, proto.config.MaxConcurrentTransactionListQueries, len(proto.transactionListQuerySem))
+	})
+
 	performMultipleConversationsTest(t, peer, func(c grpc.Connection, p *protocol, mocks protocolMocks) error {
 		return p.sendTransactionListQuery(c, []hash.SHA256Hash{hash.FromSlice([]byte("list query"))})
 	})

network/transport/v2/transactionlist_handler.go

@@ -120,6 +120,10 @@ func (p *protocol) handleTransactionList(ctx context.Context, connection grpc.Co
 
 	if msg.MessageNumber >= msg.TotalMessages {
 		p.cMan.done(cid)
+		// Release the semaphore slot if this conversation held one (TransactionListQuery, not TransactionRangeQuery).
+		if _, held := p.transactionListQueryConvs.LoadAndDelete(cid.String()); held {
+			<-p.transactionListQuerySem
+		}
 	} else {
 		p.cMan.resetTimeout(cid)
 	}