fix(network): close peer connections that go idle (backport of #4562)

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

docs/pages/release_notes.rst

@@ -11,6 +11,7 @@ Unreleased
 - Network: a peer that rejects an outbound connection with ``already connected`` is now retried with exponential backoff instead of every 1 to 5 seconds. Previously such a peer was retried every 1 to 5 seconds indefinitely, and never connected. 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.
 - Network: failed connection attempts are now logged at debug level instead of warning level. A warning is logged once, on the attempt that reaches ``network.maxbackoff``, so an unreachable peer no longer repeats the same warning on every retry.
+- 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/4562
 
 ## Security
 

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).
@@ -65,6 +68,7 @@ func DefaultConfig() Config {
 		GrpcAddr:          ":5555",
 		ConnectionTimeout: 5000,
 		MaxBackoff:        time.Hour,
+		IdleTimeout:       2 * time.Minute,
 		ProtocolV2:        v2.DefaultConfig(),
 		EnableDiscovery:   true,
 	}

network/config_test.go

@@ -29,6 +29,7 @@ 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

@@ -92,10 +92,11 @@ type Connection interface {
 	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)
@@ -104,16 +105,32 @@ func createConnection(parentCtx context.Context, peer transport.Peer) Connection
 
 type conn struct {
 	peer atomic.Value
-	// receivers tracks the receive loops, so callers can wait for the close status to be final.
-	// It deliberately covers less than activeGoroutines below, which also counts the send loops.
+	// 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 a message was last received on any of the connection's streams.
+	lastReceived atomic.Value
+	// handling counts the receive loops that are currently handling a message; the idle timeout does not apply while handling.
+	handling atomic.Int32
+	// receivers tracks the receive loops only, so waitForReceivers can block until the close status is final.
+	// A WaitGroup is needed because only the receive loops write the close status, and a connection can have
+	// several of them, one per protocol stream.
 	receivers sync.WaitGroup
 	ctx       context.Context
 	cancelCtx func()
 	status    atomic.Pointer[status.Status]
 	mux       sync.RWMutex
 	streams   map[string]Stream
 	outboxes  map[string]chan interface{}
-	// activeGoroutines counts every goroutine started for this connection: the receive loops and the send loops.
+	// activeGoroutines counts every goroutine started for this connection: the receive loops, the send loops and
+	// the idle watcher. It exists for the goroutine leak check in the tests, which asserts on the number.
+	//
+	// It cannot replace receivers above and receivers cannot replace it: a WaitGroup can be waited on but its
+	// value cannot be read, an atomic counter can be read but cannot be waited on. Neither is a substitute for
+	// the other, which is why the connection keeps both. Waiting on all goroutines instead of the receive loops
+	// would also be wrong: startSending can block in SendMsg for as long as the gRPC transport takes to give up
+	// on a dead connection, which has nothing to do with the close status.
+	//
+	// Both are maintained by startGoroutine and startReceiveLoop, so no caller has to remember to do it.
 	activeGoroutines int32
 }
 
@@ -207,6 +224,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())
+		mc.watchIdle()
+	}
 	mc.streams[methodName] = stream
 	mc.outboxes[methodName] = make(chan interface{}, OutboxHardLimit)
 
@@ -222,13 +244,34 @@ func (mc *conn) registerStream(protocol Protocol, stream Stream) bool {
 	return true
 }
 
-func (mc *conn) startReceiving(protocol Protocol, stream Stream) {
-	peer := mc.Peer() // copy Peer, because it will be nil when logging after disconnecting.
+// startGoroutine runs fn in a goroutine that is counted in activeGoroutines for as long as it runs.
+func (mc *conn) startGoroutine(fn func()) {
 	atomic.AddInt32(&mc.activeGoroutines, 1)
+	go func() {
+		defer atomic.AddInt32(&mc.activeGoroutines, -1)
+		fn()
+	}()
+}
+
+// startReceiveLoop runs a receive loop. On top of the counting done by startGoroutine it registers the loop with
+// receivers, so waitForReceivers blocks until it has exited. Receive loops must be started through this function
+// and not through startGoroutine, otherwise the close status can be read before it is written.
+func (mc *conn) startReceiveLoop(fn func()) {
 	mc.receivers.Add(1)
-	go func(activeGoroutines *int32) {
-		defer atomic.AddInt32(activeGoroutines, -1)
+	mc.startGoroutine(func() {
 		defer mc.receivers.Done()
+		fn()
+	})
+}
+
+// goroutineCount returns how many goroutines are currently running for this connection.
+func (mc *conn) goroutineCount() int32 {
+	return atomic.LoadInt32(&mc.activeGoroutines)
+}
+
+func (mc *conn) startReceiving(protocol Protocol, stream Stream) {
+	peer := mc.Peer() // copy Peer, because it will be nil when logging after disconnecting.
+	mc.startReceiveLoop(func() {
 		for {
 			message := protocol.CreateEnvelope()
 			err := stream.RecvMsg(message) // blocking
@@ -271,8 +314,12 @@ func (mc *conn) startReceiving(protocol Protocol, stream Stream) {
 				// connection has been closed: drop message and stop receiving
 				return
 			}
+			mc.lastReceived.Store(time.Now())
 
+			mc.handling.Add(1)
 			err = protocol.Handle(mc, message)
+			mc.handling.Add(-1)
+			mc.lastReceived.Store(time.Now()) // handling a message counts as activity as well
 			if err != nil {
 				log.Logger().
 					WithError(err).
@@ -282,15 +329,48 @@ func (mc *conn) startReceiving(protocol Protocol, stream Stream) {
 					Warn("Error handling message")
 			}
 		}
-	}(&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()
+	mc.startGoroutine(func() {
+		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
+				}
+				lastReceived, _ := mc.lastReceived.Load().(time.Time)
+				idle := time.Since(lastReceived)
+				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
+			}
+		}
+	})
 }
 
 func (mc *conn) startSending(protocol Protocol, stream Stream) {
 	outbox := mc.outboxes[protocol.MethodName()]
 
-	atomic.AddInt32(&mc.activeGoroutines, 1)
-	go func(activeGoroutines *int32) {
-		defer atomic.AddInt32(activeGoroutines, -1)
+	mc.startGoroutine(func() {
 	loop:
 		for {
 			select {
@@ -331,7 +411,7 @@ func (mc *conn) startSending(protocol Protocol, stream Stream) {
 					Warn("Error while closing client for gRPC stream")
 			}
 		}
-	}(&mc.activeGoroutines)
+	})
 }
 
 func (mc *conn) IsConnected() bool {

network/transport/grpc/connection_list.go

@@ -22,6 +22,7 @@ import (
 	"context"
 	"errors"
 	"sync"
+	"time"
 
 	"github.com/nuts-foundation/nuts-node/v6/core"
 	"github.com/nuts-foundation/nuts-node/v6/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

@@ -120,7 +120,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)