fix(network): close idle peer connections, back off on already connected, lower maxbackoff

docs/pages/deployment/server_options_didnuts.rst

@@ -81,8 +81,11 @@
     * - 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
+      - 1h0m0s
       - Maximum between outbound connections attempts to unresponsive nodes (in Golang duration format, e.g. '1h', '30m').
     * - network.nodedid
       -

docs/pages/release_notes.rst

@@ -13,6 +13,11 @@ 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 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
 * #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

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 ('<host>:<port>') 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.")

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).
@@ -64,7 +67,8 @@ 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,
 	}

network/config_test.go

@@ -20,13 +20,16 @@ package network
 
 import (
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 )
 
 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) {

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)
 			}),

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.

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,20 +87,32 @@ 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)
 	return result
 }
 
 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()
+}

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
 }

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