From 3595592b9662a42e82b5729fede834dcd8106659 Mon Sep 17 00:00:00 2001 From: Steven van der Vegt Date: Wed, 2 Sep 2026 10:24:09 +0200 Subject: [PATCH 1/3] fix(network): close idle peer connections and back off on already connected Connections to peers on which no message is received for a configurable period (network.idletimeout, default 2 minutes) are now closed, after which the regular reconnect logic takes over. Peers send gossip and diagnostics messages every few seconds, so a silent connection is dead: a half-open TCP connection or a reverse proxy that kept the stream open after the other side went away. Such connections previously lingered until the proxy or node was restarted, and the peer holding the stale connection rejected every new connection attempt with "already connected". The idle check pauses while a message is being handled, so long-running handling (e.g. a large transaction list during sync) is not mistaken for silence. An "already connected" rejection was treated as a clean disconnect and retried every 1 to 5 seconds, bypassing the backoff. It now goes through the exponential backoff like an authentication failure. To make that decision reliable, openOutboundStreams closes the gRPC client connection and waits for the receive loops to exit before reading the close status, which was previously racing with the stream context cancellation. Assisted-by: AI --- .../deployment/server_options_didnuts.rst | 3 + docs/pages/release_notes.rst | 4 + network/cmd/cmd.go | 1 + network/config.go | 4 + network/network.go | 1 + network/transport/grpc/config.go | 16 +++ network/transport/grpc/connection.go | 112 ++++++++++++--- network/transport/grpc/connection_list.go | 5 +- network/transport/grpc/connection_manager.go | 16 ++- .../transport/grpc/connection_manager_test.go | 57 ++++++-- network/transport/grpc/connection_mock.go | 12 ++ network/transport/grpc/connection_test.go | 132 +++++++++++++++++- network/transport/grpc/test.go | 4 + 13 files changed, 326 insertions(+), 41 deletions(-) diff --git a/docs/pages/deployment/server_options_didnuts.rst b/docs/pages/deployment/server_options_didnuts.rst index 584970a1ec..b7d5721de2 100755 --- a/docs/pages/deployment/server_options_didnuts.rst +++ b/docs/pages/deployment/server_options_didnuts.rst @@ -81,6 +81,9 @@ * - network.grpcaddr - \:5555 - Local address for gRPC to listen on. If empty the gRPC server won't be started and other nodes will not be able to connect to this node (outbound connections can still be made). + * - network.idletimeout + - 2m0s + - Period without any received message after which a connection to a peer is closed and re-established (in Golang duration format, e.g. '2m'). Specify 0 to disable. * - network.maxbackoff - 24h0m0s - Maximum between outbound connections attempts to unresponsive nodes (in Golang duration format, e.g. '1h', '30m'). diff --git a/docs/pages/release_notes.rst b/docs/pages/release_notes.rst index dc9266cee4..5116de8086 100644 --- a/docs/pages/release_notes.rst +++ b/docs/pages/release_notes.rst @@ -13,6 +13,10 @@ Unreleased * #4078: Expose the experimental two-VP flow on ``POST /internal/auth/v2/{subjectID}/request-service-access-token`` via the optional ``service_provider_subject_id`` body field by @stevenvegt in https://github.com/nuts-foundation/nuts-node/pull/4228 * #4233: ``request-credential`` API gains an optional ``credential_request_params`` JSON object overlaid on top of the OpenID4VCI Credential Request body sent to the issuer. Lets the wallet talk to issuers that accept additional fields, or to override the credential request entirely. +## Minor fixes/changes +* Network: connections on which no message was received for ``network.idletimeout`` (default ``2m``) are now closed and re-established. Peers send gossip and diagnostics messages every few seconds, so a silent connection is a dead one: typically a half-open TCP connection or a reverse proxy that kept the stream open after the other side went away. Previously such connections lingered until the proxy or node was restarted, and the peer holding the stale connection rejected new connections with ``already connected``. Set ``network.idletimeout`` to ``0`` to disable. By @stevenvegt +* Network: a peer that rejects an outbound connection with ``already connected`` is now retried with exponential backoff instead of every 1 to 5 seconds. By @stevenvegt + ## Security * #4441: Inbound HTTP request bodies are now limited to 1MB on both the public and internal interfaces; larger requests are rejected with HTTP 413 (Request Entity Too Large). Previously no limit was enforced, contrary to what the deployment documentation stated. The heaviest legitimate requests (OAuth POSTs carrying Verifiable Presentations) stay well below this limit, and it matches the ``client_max_body_size 1M`` reverse proxy configuration the documentation recommends. By @stevenvegt in https://github.com/nuts-foundation/nuts-node/pull/4441 * #4439: Helm chart (version 0.0.9): default ``verbosity`` changed from ``debug`` to ``info``, matching the node's own default. Debug verbosity produces far more log output than production needs and increases the impact of any log-hygiene issue. Set ``nuts.config.verbosity: debug`` in your own values to restore the old behavior. By @stevenvegt in https://github.com/nuts-foundation/nuts-node/pull/4439 diff --git a/network/cmd/cmd.go b/network/cmd/cmd.go index 82ba685b28..714de7bada 100644 --- a/network/cmd/cmd.go +++ b/network/cmd/cmd.go @@ -46,6 +46,7 @@ func FlagSet() *pflag.FlagSet { "(outbound connections can still be made).") flagSet.Int("network.connectiontimeout", defs.ConnectionTimeout, "Timeout before an outbound connection attempt times out (in milliseconds).") flagSet.Duration("network.maxbackoff", defs.MaxBackoff, "Maximum between outbound connections attempts to unresponsive nodes (in Golang duration format, e.g. '1h', '30m').") + flagSet.Duration("network.idletimeout", defs.IdleTimeout, "Period without any received message after which a connection to a peer is closed and re-established (in Golang duration format, e.g. '2m'). Specify 0 to disable.") flagSet.StringSlice("network.bootstrapnodes", defs.BootstrapNodes, "List of bootstrap nodes (':') which the node initially connect to.") flagSet.Bool("network.enablediscovery", defs.EnableDiscovery, "Whether to enable automatic connecting to other nodes.") flagSet.String("network.nodedid", defs.NodeDID, "Specifies the DID of the party that operates this node. It is used to identify the node on the network. If the DID document does not exist of is deactivated, the node will not start.") diff --git a/network/config.go b/network/config.go index cd56f3888f..63b9e1b506 100644 --- a/network/config.go +++ b/network/config.go @@ -31,6 +31,9 @@ type Config struct { ConnectionTimeout int `koanf:"connectiontimeout"` // MaxBackoff specifies the maximum backoff for outbound connections MaxBackoff time.Duration `koanf:"maxbackoff"` + // IdleTimeout specifies the period without any received message after which a connection to a peer is closed. + // Peers send gossip and diagnostics messages at a fixed interval, so a silent connection is a dead one. Zero disables the check. + IdleTimeout time.Duration `koanf:"idletimeout"` // Public address of this nodes other nodes can use to connect to this node. BootstrapNodes []string `koanf:"bootstrapnodes"` // Protocols is the list of network protocols to enable on the server. They are specified by version (v1, v2). @@ -65,6 +68,7 @@ func DefaultConfig() Config { GrpcAddr: ":5555", ConnectionTimeout: 5000, MaxBackoff: 24 * time.Hour, + IdleTimeout: 2 * time.Minute, ProtocolV2: v2.DefaultConfig(), EnableDiscovery: true, } diff --git a/network/network.go b/network/network.go index 12718c4e11..8a78840fdb 100644 --- a/network/network.go +++ b/network/network.go @@ -262,6 +262,7 @@ func (n *Network) Configure(config core.ServerConfig) error { if n.connectionManager == nil { grpcOpts := []grpc.ConfigOption{ grpc.WithConnectionTimeout(time.Duration(n.config.ConnectionTimeout) * time.Millisecond), + grpc.WithIdleTimeout(n.config.IdleTimeout), grpc.WithBackoff(func() grpc.Backoff { return grpc.BoundedBackoff(time.Second, n.config.MaxBackoff) }), diff --git a/network/transport/grpc/config.go b/network/transport/grpc/config.go index 4f89bec209..bf6bbfedf4 100644 --- a/network/transport/grpc/config.go +++ b/network/transport/grpc/config.go @@ -37,6 +37,10 @@ func tcpListenerCreator(addr string) (net.Listener, error) { } // ConfigOption is used to build Config. +// defaultIdleTimeout is the default period without received messages after which a connection is considered dead. +// Peers send gossip and diagnostics every 5 seconds by default, so this leaves ample room for a slow peer. +const defaultIdleTimeout = 2 * time.Minute + type ConfigOption func(config *Config) error // NewConfig creates a new Config, used for configuring a gRPC ConnectionManager. @@ -47,6 +51,7 @@ func NewConfig(grpcAddress string, peerID networkTypes.PeerID, options ...Config dialer: grpc.DialContext, listener: tcpListenerCreator, connectionTimeout: 5 * time.Second, + idleTimeout: defaultIdleTimeout, backoffCreator: func() Backoff { return BoundedBackoff(time.Second, time.Hour) }, @@ -102,6 +107,15 @@ func WithConnectionTimeout(value time.Duration) ConfigOption { } } +// WithIdleTimeout specifies the period without any received message after which a connection is closed. +// Zero disables the check. +func WithIdleTimeout(value time.Duration) ConfigOption { + return func(config *Config) error { + config.idleTimeout = value + return nil + } +} + func WithBackoff(value func() Backoff) ConfigOption { return func(config *Config) error { config.backoffCreator = value @@ -130,6 +144,8 @@ type Config struct { clientIPHeaderName string // connectionTimeout specifies the time before an outbound connection attempt times out. connectionTimeout time.Duration + // idleTimeout specifies the period without any received message after which a connection is closed. + idleTimeout time.Duration // listener holds a function to create the net.Listener that is used for inbound connections. listener func(string) (net.Listener, error) // dialer holds a function to open connections to remote gRPC services. diff --git a/network/transport/grpc/connection.go b/network/transport/grpc/connection.go index 8f938f5a5b..2f0dbd72d0 100644 --- a/network/transport/grpc/connection.go +++ b/network/transport/grpc/connection.go @@ -27,6 +27,7 @@ import ( "io" "sync" "sync/atomic" + "time" "github.com/nuts-foundation/nuts-node/network/log" "github.com/nuts-foundation/nuts-node/network/transport" @@ -86,12 +87,16 @@ type Connection interface { // closeError returns the status when the connection closed with an error or nil otherwise closeError() *status.Status + // waitForReceivers blocks until all receive loops have exited, which makes closeError() final. + // The underlying streams must be closed first, otherwise the receive loops block forever. + waitForReceivers() } -func createConnection(parentCtx context.Context, peer transport.Peer) Connection { +func createConnection(parentCtx context.Context, peer transport.Peer, idleTimeout time.Duration) Connection { result := &conn{ - streams: make(map[string]Stream), - outboxes: make(map[string]chan interface{}), + streams: make(map[string]Stream), + outboxes: make(map[string]chan interface{}), + idleTimeout: idleTimeout, } result.ctx, result.cancelCtx = context.WithCancel(parentCtx) result.setPeer(peer) @@ -99,7 +104,15 @@ func createConnection(parentCtx context.Context, peer transport.Peer) Connection } type conn struct { - peer atomic.Value + peer atomic.Value + // idleTimeout is the period without any received message after which the connection is closed. Zero disables the check. + idleTimeout time.Duration + // lastReceived holds the time (unix nanoseconds) a message was last received on any of the connection's streams. + lastReceived atomic.Int64 + // receivers tracks the receive loops, so callers can wait for the close status to be final. + receivers sync.WaitGroup + // handling counts the receive loops that are currently handling a message; the idle timeout does not apply while handling. + handling atomic.Int32 ctx context.Context cancelCtx func() status atomic.Pointer[status.Status] @@ -199,6 +212,11 @@ func (mc *conn) registerStream(protocol Protocol, stream Stream) bool { return false } + if len(mc.streams) == 0 && mc.idleTimeout > 0 { + // first stream on this connection: start watching for idleness + mc.lastReceived.Store(time.Now().UnixNano()) + mc.watchIdle() + } mc.streams[methodName] = stream mc.outboxes[methodName] = make(chan interface{}, OutboxHardLimit) @@ -217,35 +235,49 @@ func (mc *conn) registerStream(protocol Protocol, stream Stream) bool { func (mc *conn) startReceiving(protocol Protocol, stream Stream) { peer := mc.Peer() // copy Peer, because it will be nil when logging after disconnecting. atomic.AddInt32(&mc.activeGoroutines, 1) + mc.receivers.Add(1) go func(activeGoroutines *int32) { defer atomic.AddInt32(activeGoroutines, -1) + defer mc.receivers.Done() for { message := protocol.CreateEnvelope() err := stream.RecvMsg(message) // blocking - if mc.ctx.Err() != nil { - // connection has been closed: drop message and stop receiving - return - } if err != nil { errStatus, isStatusError := status.FromError(err) - if errors.Is(err, io.EOF) || (isStatusError && errStatus.Code() == codes.Canceled) { - log.Logger(). - WithField(core.LogFieldProtocolVersion, protocol.Version()). - WithFields(peer.ToFields()). - Info("Peer closed connection") - } else { - log.Logger(). - WithError(err). - WithField(core.LogFieldProtocolVersion, protocol.Version()). - WithFields(peer.ToFields()). - Warn("Peer connection error") + closedByPeer := !errors.Is(err, io.EOF) && !(isStatusError && errStatus.Code() == codes.Canceled) + if mc.ctx.Err() == nil { + // only log when the connection wasn't closed locally + if closedByPeer { + log.Logger(). + WithError(err). + WithField(core.LogFieldProtocolVersion, protocol.Version()). + WithFields(peer.ToFields()). + Warn("Peer connection error") + } else { + log.Logger(). + WithField(core.LogFieldProtocolVersion, protocol.Version()). + WithFields(peer.ToFields()). + Info("Peer closed connection") + } + } + if closedByPeer { + // Record the peer's close status even if the connection was already cancelled (e.g. because the stream's context is done), + // so the caller can decide whether the peer rejected the connection. + mc.status.Store(errStatus) } - mc.status.Store(errStatus) mc.cancelCtx() break } + if mc.ctx.Err() != nil { + // connection has been closed: drop message and stop receiving + return + } + mc.lastReceived.Store(time.Now().UnixNano()) + mc.handling.Add(1) err = protocol.Handle(mc, message) + mc.handling.Add(-1) + mc.lastReceived.Store(time.Now().UnixNano()) // handling a message counts as activity as well if err != nil { log.Logger(). WithError(err). @@ -258,6 +290,42 @@ func (mc *conn) startReceiving(protocol Protocol, stream Stream) { }(&mc.activeGoroutines) } +// watchIdle disconnects the connection when no message has been received within idleTimeout. +// Peers send gossip and diagnostics messages at a fixed interval, so a silent stream is a dead one +// (e.g. a half-open TCP connection or a proxy that kept the stream open after the other side went away). +func (mc *conn) watchIdle() { + peer := mc.Peer() // copy Peer, because it will be reset by disconnect() + atomic.AddInt32(&mc.activeGoroutines, 1) + go func(activeGoroutines *int32) { + defer atomic.AddInt32(activeGoroutines, -1) + timer := time.NewTimer(mc.idleTimeout) + defer timer.Stop() + for { + select { + case <-mc.ctx.Done(): + return + case <-timer.C: + if mc.handling.Load() > 0 { + // still busy handling a message (e.g. a large transaction list during sync), which is not idle + timer.Reset(mc.idleTimeout) + continue + } + idle := time.Since(time.Unix(0, mc.lastReceived.Load())) + if idle < mc.idleTimeout { + timer.Reset(mc.idleTimeout - idle) + continue + } + log.Logger(). + WithFields(peer.ToFields()). + WithField("idle", idle.Round(time.Second)). + Warn("No messages received from peer within idle timeout, disconnecting") + mc.disconnect() + return + } + } + }(&mc.activeGoroutines) +} + func (mc *conn) startSending(protocol Protocol, stream Stream) { outbox := mc.outboxes[protocol.MethodName()] @@ -321,3 +389,7 @@ func (mc *conn) IsAuthenticated() bool { func (mc *conn) closeError() *status.Status { return mc.status.Load() } + +func (mc *conn) waitForReceivers() { + mc.receivers.Wait() +} diff --git a/network/transport/grpc/connection_list.go b/network/transport/grpc/connection_list.go index f037c3bc21..0c4755f4b4 100644 --- a/network/transport/grpc/connection_list.go +++ b/network/transport/grpc/connection_list.go @@ -22,6 +22,7 @@ import ( "context" "errors" "sync" + "time" "github.com/nuts-foundation/nuts-node/core" "github.com/nuts-foundation/nuts-node/network/transport" @@ -44,6 +45,8 @@ type ConnectionList interface { type connectionList struct { mux sync.Mutex list []Connection + // idleTimeout is passed to new connections, see conn.idleTimeout. + idleTimeout time.Duration } func (c *connectionList) Get(query ...Predicate) Connection { @@ -108,7 +111,7 @@ func (c *connectionList) getOrRegister(ctx context.Context, peer transport.Peer, return existing, false } - result := createConnection(ctx, peer) + result := createConnection(ctx, peer, c.idleTimeout) c.list = append(c.list, result) return result, true } diff --git a/network/transport/grpc/connection_manager.go b/network/transport/grpc/connection_manager.go index 213d5da93b..944d16f1f1 100644 --- a/network/transport/grpc/connection_manager.go +++ b/network/transport/grpc/connection_manager.go @@ -108,7 +108,7 @@ func NewGRPCConnectionManager(config Config, connectionStore stoabs.KVStore, nod authenticator: authenticator, config: config, connectionTimeout: config.connectionTimeout, - connections: &connectionList{}, + connections: &connectionList{idleTimeout: config.idleTimeout}, dialer: config.dialer, dialOptions: []grpc.DialOption{ grpc.WithBlock(), // Dial should block until connection succeeded (or time-out expired) @@ -462,9 +462,17 @@ func (s *grpcConnectionManager) openOutboundStreams(connection Connection, grpcC // Function must block until streams are closed or disconnect() is called. connection.waitUntilDisconnected() - if st := connection.closeError(); st != nil && st.Code() == codes.Unauthenticated { - // return error so entire connection will be tried anew. Otherwise, backoff isn't honored - return st.Err() + // Close the gRPC connection so blocked receive loops return, then wait for them: + // only then is the close status (as sent by the peer) final. + _ = grpcConn.Close() + connection.waitForReceivers() + + if st := connection.closeError(); st != nil { + // Peer rejected the connection: return the error so the backoff is honored instead of reconnecting within seconds. + // ErrAlreadyConnected arrives as codes.Unknown (plain error returned by the peer's stream handler), so match on the message. + if st.Code() == codes.Unauthenticated || st.Message() == ErrAlreadyConnected.Error() { + return st.Err() + } } return nil diff --git a/network/transport/grpc/connection_manager_test.go b/network/transport/grpc/connection_manager_test.go index f5aa5e2144..06fbf31b06 100644 --- a/network/transport/grpc/connection_manager_test.go +++ b/network/transport/grpc/connection_manager_test.go @@ -288,7 +288,7 @@ func Test_grpcConnectionManager_dial(t *testing.T) { cont := newContact(peer, backoff) cm, err := NewGRPCConnectionManager(Config{}, createKVStore(t), *nodeDID, dummyAuthenticator{}, &TestProtocol{}) require.NoError(t, err) - cm.connections.list = append(cm.connections.list, createConnection(cm.ctx, peer)) // add existing connection + cm.connections.list = append(cm.connections.list, createConnection(cm.ctx, peer, 0)) // add existing connection cm.connect(cont) @@ -426,6 +426,45 @@ func Test_grpcConnectionManager_dial(t *testing.T) { // connection is removed again assert.Empty(t, client.connections.list) }) + t.Run("already connected", func(t *testing.T) { + // The server rejects the second stream with ErrAlreadyConnected because both clients present the same peer ID. + // The rejected client must back off instead of treating it as a clean disconnect and retrying within seconds. + serverCfg, serverListener := newBufconnConfig("server") + server, err := NewGRPCConnectionManager(serverCfg, nil, did.DID{}, nil, &TestProtocol{}) + require.NoError(t, err) + require.NoError(t, server.Start()) + defer server.Stop() + + clientCfg, _ := newBufconnConfig("client", withBufconnDialer(serverListener)) + + // first client connects and stays connected + client1, err := NewGRPCConnectionManager(clientCfg, nil, did.DID{}, nil, &TestProtocol{}) + require.NoError(t, err) + client1Done := make(chan struct{}) + go func() { + client1.connect(newContact(transport.Peer{Address: "server"}, newTestBackoff())) + close(client1Done) + }() + test.WaitFor(t, func() (bool, error) { + return len(server.Peers()) == 1, nil + }, 2*time.Second, "waiting for first client to connect") + defer func() { + client1.ctxCancel() + <-client1Done + }() + + // second client with the same peer ID is rejected by the server + client2, err := NewGRPCConnectionManager(clientCfg, nil, did.DID{}, nil, &TestProtocol{}) + require.NoError(t, err) + backoff := &trackingBackoff{mux: &sync.Mutex{}} + cont := newContact(transport.Peer{Address: "server"}, backoff) + + client2.connect(cont) // returns when the server closes the stream + + assert.Equal(t, 1, backoff.backoffCount) + assert.Equal(t, 0, backoff.resetCount) + assert.Empty(t, client2.connections.list) + }) t.Run("wrong DID answered call", func(t *testing.T) { serverCfg, serverListener := newBufconnConfig("server") server, err := NewGRPCConnectionManager(serverCfg, nil, *nodeDID, nil, &TestProtocol{}) @@ -775,7 +814,7 @@ func Test_grpcConnectionManager_openOutboundStreams(t *testing.T) { clientCfg, _ := newBufconnConfig("client", withBufconnDialer(serverListener)) client, err := NewGRPCConnectionManager(clientCfg, nil, did.DID{}, nil, &TestProtocol{}) require.NoError(t, err) - c := createConnection(context.Background(), transport.Peer{}).(*conn) + c := createConnection(context.Background(), transport.Peer{}, 0).(*conn) c.status.Store(status.New(codes.Unauthenticated, "unauthenticated")) grpcConn, err := clientCfg.dialer(context.Background(), "server") require.NoError(t, err) @@ -817,7 +856,7 @@ func Test_grpcConnectionManager_openOutboundStreams(t *testing.T) { clientCfg, _ := newBufconnConfig("client", withBufconnDialer(serverListener)) client, err := NewGRPCConnectionManager(clientCfg, nil, did.DID{}, nil, &TestProtocol{}) require.NoError(t, err) - c := createConnection(context.Background(), transport.Peer{}) + c := createConnection(context.Background(), transport.Peer{}, 0) grpcConn, err := clientCfg.dialer(context.Background(), "server") require.NoError(t, err) var capturedPeer atomic.Value @@ -959,7 +998,7 @@ func Test_grpcConnectionManager_openOutboundStream(t *testing.T) { clientCfg, _ := newBufconnConfig("client", withBufconnDialer(serverListener)) client, err := NewGRPCConnectionManager(clientCfg, nil, did.DID{}, nil, &TestProtocol{}) require.NoError(t, err) - c := createConnection(context.Background(), transport.Peer{}) + c := createConnection(context.Background(), transport.Peer{}, 0) grpcConn, err := clientCfg.dialer(context.Background(), "server") require.NoError(t, err) md, _ := client.constructMetadata(false) @@ -980,7 +1019,7 @@ func Test_grpcConnectionManager_openOutboundStream(t *testing.T) { clientCfg, _ := newBufconnConfig("client", withBufconnDialer(serverListener)) client, err := NewGRPCConnectionManager(clientCfg, nil, did.DID{}, nil, &TestProtocol{}) require.NoError(t, err) - c := createConnection(context.Background(), transport.Peer{}) + c := createConnection(context.Background(), transport.Peer{}, 0) grpcConn, err := clientCfg.dialer(context.Background(), "server") require.NoError(t, err) md, _ := client.constructMetadata(false) @@ -1008,7 +1047,7 @@ func Test_grpcConnectionManager_openOutboundStream(t *testing.T) { clientCfg, _ := newBufconnConfig("client", withBufconnDialer(serverListener)) client, err := NewGRPCConnectionManager(clientCfg, nil, did.DID{}, nil, &TestProtocol{}) require.NoError(t, err) - c := createConnection(context.Background(), transport.Peer{}) + c := createConnection(context.Background(), transport.Peer{}, 0) grpcConn, err := clientCfg.dialer(context.Background(), "server") require.NoError(t, err) @@ -1037,7 +1076,7 @@ func Test_grpcConnectionManager_openOutboundStream(t *testing.T) { authenticator.EXPECT().Authenticate(*nodeDID, gomock.Any()).Return(transport.Peer{}, ErrNodeDIDAuthFailed) client, err := NewGRPCConnectionManager(clientCfg, nil, did.DID{}, authenticator, &TestProtocol{}) require.NoError(t, err) - c := createConnection(context.Background(), transport.Peer{NodeDID: *nodeDID}) + c := createConnection(context.Background(), transport.Peer{NodeDID: *nodeDID}, 0) grpcConn, err := clientCfg.dialer(context.Background(), "server") require.NoError(t, err) @@ -1060,7 +1099,7 @@ func Test_grpcConnectionManager_openOutboundStream(t *testing.T) { authenticator := NewMockAuthenticator(ctrl) client, err := NewGRPCConnectionManager(clientCfg, nil, did.DID{}, authenticator, &TestProtocol{}) require.NoError(t, err) - c := createConnection(context.Background(), transport.Peer{NodeDID: did.MustParseDID("did:nuts:remote")}) + c := createConnection(context.Background(), transport.Peer{NodeDID: did.MustParseDID("did:nuts:remote")}, 0) grpcConn, err := clientCfg.dialer(context.Background(), "server") require.NoError(t, err) @@ -1083,7 +1122,7 @@ func Test_grpcConnectionManager_openOutboundStream(t *testing.T) { authenticator := NewMockAuthenticator(ctrl) // is not called client, err := NewGRPCConnectionManager(clientCfg, nil, did.DID{}, authenticator, &TestProtocol{}) require.NoError(t, err) - c := createConnection(context.Background(), transport.Peer{NodeDID: did.MustParseDID("did:nuts:remote")}) + c := createConnection(context.Background(), transport.Peer{NodeDID: did.MustParseDID("did:nuts:remote")}, 0) grpcConn, err := clientCfg.dialer(context.Background(), "server") require.NoError(t, err) diff --git a/network/transport/grpc/connection_mock.go b/network/transport/grpc/connection_mock.go index c428827d6f..44da33d305 100644 --- a/network/transport/grpc/connection_mock.go +++ b/network/transport/grpc/connection_mock.go @@ -163,6 +163,18 @@ func (mr *MockConnectionMockRecorder) verifyOrSetPeerID(id any) *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "verifyOrSetPeerID", reflect.TypeOf((*MockConnection)(nil).verifyOrSetPeerID), id) } +// waitForReceivers mocks base method. +func (m *MockConnection) waitForReceivers() { + m.ctrl.T.Helper() + m.ctrl.Call(m, "waitForReceivers") +} + +// waitForReceivers indicates an expected call of waitForReceivers. +func (mr *MockConnectionMockRecorder) waitForReceivers() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "waitForReceivers", reflect.TypeOf((*MockConnection)(nil).waitForReceivers)) +} + // waitUntilDisconnected mocks base method. func (m *MockConnection) waitUntilDisconnected() { m.ctrl.T.Helper() diff --git a/network/transport/grpc/connection_test.go b/network/transport/grpc/connection_test.go index 1ca7652176..ac3ef929ed 100644 --- a/network/transport/grpc/connection_test.go +++ b/network/transport/grpc/connection_test.go @@ -23,6 +23,7 @@ import ( "github.com/nuts-foundation/nuts-node/test" "github.com/stretchr/testify/require" "google.golang.org/grpc/codes" + "io" "sync" "sync/atomic" "testing" @@ -40,7 +41,7 @@ func Test_conn_disconnect(t *testing.T) { assert.False(t, conn.IsConnected()) }) t.Run("connected", func(t *testing.T) { - conn := createConnection(context.Background(), transport.Peer{}).(*conn) + conn := createConnection(context.Background(), transport.Peer{}, 0).(*conn) conn.streams["stream name"] = &MockStream{} assert.True(t, conn.IsConnected()) conn.disconnect() @@ -57,7 +58,7 @@ func Test_conn_disconnect(t *testing.T) { func Test_conn_waitUntilDisconnected(t *testing.T) { t.Run("never open, should return immediately", func(t *testing.T) { - conn := createConnection(context.Background(), transport.Peer{}) + conn := createConnection(context.Background(), transport.Peer{}, 0) conn.waitUntilDisconnected() }) t.Run("disconnected while waiting, should return almost immediately", func(t *testing.T) { @@ -82,7 +83,7 @@ func Test_conn_waitUntilDisconnected(t *testing.T) { func Test_conn_registerStream(t *testing.T) { t.Run("ok", func(t *testing.T) { - connection := createConnection(context.Background(), transport.Peer{}).(*conn) + connection := createConnection(context.Background(), transport.Peer{}, 0).(*conn) stream := newServerStream("foo", "", nil) defer stream.cancelFunc() @@ -92,7 +93,7 @@ func Test_conn_registerStream(t *testing.T) { assert.True(t, connection.IsConnected()) }) t.Run("already connected (same protocol)", func(t *testing.T) { - connection := createConnection(context.Background(), transport.Peer{}).(*conn) + connection := createConnection(context.Background(), transport.Peer{}, 0).(*conn) stream := newServerStream("foo", "", nil) defer stream.cancelFunc() @@ -106,7 +107,7 @@ func Test_conn_registerStream(t *testing.T) { func Test_conn_startSending(t *testing.T) { t.Run("disconnect does not panic", func(t *testing.T) { - connection := createConnection(context.Background(), transport.Peer{}).(*conn) + connection := createConnection(context.Background(), transport.Peer{}, 0).(*conn) stream := newServerStream("foo", "", nil) defer stream.cancelFunc() @@ -133,7 +134,7 @@ func Test_conn_startSending(t *testing.T) { func TestConn_Send(t *testing.T) { t.Run("buffer overflow softlimit", func(t *testing.T) { - connection := createConnection(context.Background(), transport.Peer{}).(*conn) + connection := createConnection(context.Background(), transport.Peer{}, 0).(*conn) stream := newServerStream("foo", "", nil) protocol := &TestProtocol{} _ = connection.registerStream(protocol, stream) @@ -159,7 +160,7 @@ func TestConn_Send(t *testing.T) { }) t.Run("buffer overflow hardLimit", func(t *testing.T) { - connection := createConnection(context.Background(), transport.Peer{}).(*conn) + connection := createConnection(context.Background(), transport.Peer{}, 0).(*conn) stream := newServerStream("foo", "", nil) protocol := &TestProtocol{} _ = connection.registerStream(protocol, stream) @@ -178,3 +179,120 @@ func TestConn_Send(t *testing.T) { }) }) } + +// noopHandleProtocol is a TestProtocol that accepts received messages instead of panicking. +type noopHandleProtocol struct { + *TestProtocol +} + +func (noopHandleProtocol) Handle(Connection, interface{}) error { + return nil +} + +// tickingStream delivers an (empty) message at every interval until its context is cancelled. +type tickingStream struct { + *stubServerStream + interval time.Duration +} + +func (s tickingStream) RecvMsg(_ interface{}) error { + select { + case <-time.After(s.interval): + return nil + case <-s.ctx.Done(): + return io.EOF + } +} + +// slowHandleProtocol is a TestProtocol whose Handle blocks for the given duration. +type slowHandleProtocol struct { + *TestProtocol + started chan struct{} + duration time.Duration +} + +func (p slowHandleProtocol) Handle(Connection, interface{}) error { + close(p.started) + time.Sleep(p.duration) + return nil +} + +// oneMessageStream delivers a single (empty) message immediately, then blocks until its context is cancelled. +type oneMessageStream struct { + *stubServerStream + delivered atomic.Bool +} + +func (s *oneMessageStream) RecvMsg(_ interface{}) error { + if s.delivered.CompareAndSwap(false, true) { + return nil + } + <-s.ctx.Done() + return io.EOF +} + +func Test_conn_idleTimeout(t *testing.T) { + t.Run("disconnects when no message is received within the idle timeout", func(t *testing.T) { + connection := createConnection(context.Background(), transport.Peer{}, 0).(*conn) + connection.idleTimeout = 50 * time.Millisecond + stream := newServerStream("foo", "", nil) // RecvMsg blocks until the stream is cancelled + defer stream.cancelFunc() + + require.True(t, connection.registerStream(&TestProtocol{}, stream)) + + select { + case <-connection.ctx.Done(): + case <-time.After(2 * time.Second): + t.Fatal("connection was not closed after idle timeout") + } + assert.False(t, connection.IsConnected()) + }) + t.Run("stays connected while messages are received", func(t *testing.T) { + connection := createConnection(context.Background(), transport.Peer{}, 0).(*conn) + connection.idleTimeout = 100 * time.Millisecond + stream := tickingStream{stubServerStream: newServerStream("foo", "", nil), interval: 20 * time.Millisecond} + defer stream.cancelFunc() + + require.True(t, connection.registerStream(noopHandleProtocol{&TestProtocol{}}, stream)) + + select { + case <-connection.ctx.Done(): + t.Fatal("connection was closed although messages were being received") + case <-time.After(400 * time.Millisecond): + } + assert.True(t, connection.IsConnected()) + }) + t.Run("stays connected while a message is being handled", func(t *testing.T) { + // e.g. during initial sync, handling a large transaction list may take longer than the idle timeout + connection := createConnection(context.Background(), transport.Peer{}, 0).(*conn) + connection.idleTimeout = 50 * time.Millisecond + stream := &oneMessageStream{stubServerStream: newServerStream("foo", "", nil)} + defer stream.cancelFunc() + handling := make(chan struct{}) + protocol := slowHandleProtocol{TestProtocol: &TestProtocol{}, started: handling, duration: 400 * time.Millisecond} + + require.True(t, connection.registerStream(protocol, stream)) + + <-handling + select { + case <-connection.ctx.Done(): + t.Fatal("connection was closed while a message was being handled") + case <-time.After(300 * time.Millisecond): + } + assert.True(t, connection.IsConnected()) + }) + t.Run("zero idle timeout disables the check", func(t *testing.T) { + connection := createConnection(context.Background(), transport.Peer{}, 0).(*conn) + stream := newServerStream("foo", "", nil) + defer stream.cancelFunc() + + require.True(t, connection.registerStream(&TestProtocol{}, stream)) + + select { + case <-connection.ctx.Done(): + t.Fatal("connection was closed without an idle timeout configured") + case <-time.After(200 * time.Millisecond): + } + assert.True(t, connection.IsConnected()) + }) +} diff --git a/network/transport/grpc/test.go b/network/transport/grpc/test.go index 8fb624747c..6ea6a26b64 100644 --- a/network/transport/grpc/test.go +++ b/network/transport/grpc/test.go @@ -131,6 +131,10 @@ func (s *StubConnection) closeError() *status.Status { panic("implement me") } +func (s *StubConnection) waitForReceivers() { + // no receive loops +} + func (s *StubConnection) SetErrorStatus(_ *status.Status) { panic("implement me") } From 11fc3d34c4331709c98eac0919f7ea9cd713b759 Mon Sep 17 00:00:00 2001 From: Steven van der Vegt Date: Wed, 2 Sep 2026 10:38:14 +0200 Subject: [PATCH 2/3] fix(network): lower default maxbackoff from 24h to 1h The backoff towards an unreachable peer grows by 1.5x per failed attempt, is persisted across restarts and is only reset when the peer's NutsComm address changes. A peer that was unreachable for a few days could therefore go unattempted for up to a day after it came back. One extra TLS handshake per hour per dead peer is a negligible price for recovering within the hour. Assisted-by: AI --- docs/pages/deployment/server_options_didnuts.rst | 2 +- docs/pages/release_notes.rst | 1 + network/config.go | 2 +- network/config_test.go | 3 +++ 4 files changed, 6 insertions(+), 2 deletions(-) diff --git a/docs/pages/deployment/server_options_didnuts.rst b/docs/pages/deployment/server_options_didnuts.rst index b7d5721de2..f4110591c6 100755 --- a/docs/pages/deployment/server_options_didnuts.rst +++ b/docs/pages/deployment/server_options_didnuts.rst @@ -85,7 +85,7 @@ - 2m0s - Period without any received message after which a connection to a peer is closed and re-established (in Golang duration format, e.g. '2m'). Specify 0 to disable. * - network.maxbackoff - - 24h0m0s + - 1h0m0s - Maximum between outbound connections attempts to unresponsive nodes (in Golang duration format, e.g. '1h', '30m'). * - network.nodedid - diff --git a/docs/pages/release_notes.rst b/docs/pages/release_notes.rst index 5116de8086..5d95811f96 100644 --- a/docs/pages/release_notes.rst +++ b/docs/pages/release_notes.rst @@ -16,6 +16,7 @@ Unreleased ## Minor fixes/changes * Network: connections on which no message was received for ``network.idletimeout`` (default ``2m``) are now closed and re-established. Peers send gossip and diagnostics messages every few seconds, so a silent connection is a dead one: typically a half-open TCP connection or a reverse proxy that kept the stream open after the other side went away. Previously such connections lingered until the proxy or node was restarted, and the peer holding the stale connection rejected new connections with ``already connected``. Set ``network.idletimeout`` to ``0`` to disable. By @stevenvegt * Network: a peer that rejects an outbound connection with ``already connected`` is now retried with exponential backoff instead of every 1 to 5 seconds. By @stevenvegt +* Network: the default of ``network.maxbackoff`` is lowered from ``24h`` to ``1h``. The backoff is persisted across restarts and only reset when a peer's NutsComm address changes, so a peer that was unreachable for a few days could previously go unattempted for up to a day after it came back. By @stevenvegt ## Security * #4441: Inbound HTTP request bodies are now limited to 1MB on both the public and internal interfaces; larger requests are rejected with HTTP 413 (Request Entity Too Large). Previously no limit was enforced, contrary to what the deployment documentation stated. The heaviest legitimate requests (OAuth POSTs carrying Verifiable Presentations) stay well below this limit, and it matches the ``client_max_body_size 1M`` reverse proxy configuration the documentation recommends. By @stevenvegt in https://github.com/nuts-foundation/nuts-node/pull/4441 diff --git a/network/config.go b/network/config.go index 63b9e1b506..586ab143b0 100644 --- a/network/config.go +++ b/network/config.go @@ -67,7 +67,7 @@ func DefaultConfig() Config { return Config{ GrpcAddr: ":5555", ConnectionTimeout: 5000, - MaxBackoff: 24 * time.Hour, + MaxBackoff: time.Hour, IdleTimeout: 2 * time.Minute, ProtocolV2: v2.DefaultConfig(), EnableDiscovery: true, diff --git a/network/config_test.go b/network/config_test.go index f29f4f0d3a..69faf4aff8 100644 --- a/network/config_test.go +++ b/network/config_test.go @@ -20,6 +20,7 @@ package network import ( "testing" + "time" "github.com/stretchr/testify/assert" ) @@ -27,6 +28,8 @@ import ( func TestDefaultConfig(t *testing.T) { defs := DefaultConfig() assert.Equal(t, ":5555", defs.GrpcAddr) + assert.Equal(t, time.Hour, defs.MaxBackoff, "a peer that comes back after a long outage should be retried within the hour") + assert.Equal(t, 2*time.Minute, defs.IdleTimeout) } func TestConfig_IsProtocolEnabled(t *testing.T) { From 19ebb695711ad5105fc020ddf28ff475196e2a82 Mon Sep 17 00:00:00 2001 From: Steven van der Vegt Date: Wed, 2 Sep 2026 10:45:27 +0200 Subject: [PATCH 3/3] docs: link release notes to PR #4467 Assisted-by: AI --- docs/pages/release_notes.rst | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/docs/pages/release_notes.rst b/docs/pages/release_notes.rst index 5d95811f96..3259ba24c0 100644 --- a/docs/pages/release_notes.rst +++ b/docs/pages/release_notes.rst @@ -14,9 +14,9 @@ Unreleased * #4233: ``request-credential`` API gains an optional ``credential_request_params`` JSON object overlaid on top of the OpenID4VCI Credential Request body sent to the issuer. Lets the wallet talk to issuers that accept additional fields, or to override the credential request entirely. ## Minor fixes/changes -* Network: connections on which no message was received for ``network.idletimeout`` (default ``2m``) are now closed and re-established. Peers send gossip and diagnostics messages every few seconds, so a silent connection is a dead one: typically a half-open TCP connection or a reverse proxy that kept the stream open after the other side went away. Previously such connections lingered until the proxy or node was restarted, and the peer holding the stale connection rejected new connections with ``already connected``. Set ``network.idletimeout`` to ``0`` to disable. By @stevenvegt -* Network: a peer that rejects an outbound connection with ``already connected`` is now retried with exponential backoff instead of every 1 to 5 seconds. By @stevenvegt -* Network: the default of ``network.maxbackoff`` is lowered from ``24h`` to ``1h``. The backoff is persisted across restarts and only reset when a peer's NutsComm address changes, so a peer that was unreachable for a few days could previously go unattempted for up to a day after it came back. By @stevenvegt +* Network: connections on which no message was received for ``network.idletimeout`` (default ``2m``) are now closed and re-established. Peers send gossip and diagnostics messages every few seconds, so a silent connection is a dead one: typically a half-open TCP connection or a reverse proxy that kept the stream open after the other side went away. Previously such connections lingered until the proxy or node was restarted, and the peer holding the stale connection rejected new connections with ``already connected``. Set ``network.idletimeout`` to ``0`` to disable. By @stevenvegt in https://github.com/nuts-foundation/nuts-node/pull/4467 +* Network: a peer that rejects an outbound connection with ``already connected`` is now retried with exponential backoff instead of every 1 to 5 seconds. By @stevenvegt in https://github.com/nuts-foundation/nuts-node/pull/4467 +* Network: the default of ``network.maxbackoff`` is lowered from ``24h`` to ``1h``. The backoff is persisted across restarts and only reset when a peer's NutsComm address changes, so a peer that was unreachable for a few days could previously go unattempted for up to a day after it came back. By @stevenvegt in https://github.com/nuts-foundation/nuts-node/pull/4467 ## Security * #4441: Inbound HTTP request bodies are now limited to 1MB on both the public and internal interfaces; larger requests are rejected with HTTP 413 (Request Entity Too Large). Previously no limit was enforced, contrary to what the deployment documentation stated. The heaviest legitimate requests (OAuth POSTs carrying Verifiable Presentations) stay well below this limit, and it matches the ``client_max_body_size 1M`` reverse proxy configuration the documentation recommends. By @stevenvegt in https://github.com/nuts-foundation/nuts-node/pull/4441