From 6f02e59e3f17ced7ae7d2551df9c095ac68e11d3 Mon Sep 17 00:00:00 2001 From: Steven van der Vegt Date: Fri, 18 Sep 2026 15:28:33 +0000 Subject: [PATCH] fix(network): close peer connections that go idle (#4562) * fix(network): close peer connections that go idle Peers send gossip and diagnostics messages every few seconds, so a stream without any incoming message is a dead one: a half-open TCP connection, or a reverse proxy that kept the stream open after the other side went away. Such a connection lingered until the proxy or node was restarted, and the peer holding the stale stream rejected every new connection with "already connected". Connections on which nothing was received for network.idletimeout (default 2m) are now closed, after which the regular reconnect logic takes over. Time spent handling a message does not count as idle, so a large transaction list during sync does not trigger a disconnect. Set network.idletimeout to 0 to disable the check. This works through reverse proxies, unlike gRPC keepalive pings, which proxies answer themselves. Assisted-by: AI * refactor(network): start connection goroutines through one helper A connection keeps two pieces of bookkeeping for its goroutines: an atomic counter that the tests assert on, and a WaitGroup that waitForReceivers blocks on. Neither can replace the other. A WaitGroup can be waited on but its value cannot be read; an atomic counter can be read but cannot be waited on. Waiting on all goroutines instead of the receive loops would be wrong as well, because startSending can block in SendMsg for as long as the gRPC transport takes to give up on a dead connection. Both were maintained by hand at each of the three places that start a goroutine, so adding a receive loop meant remembering two separate registrations. They now go through startGoroutine and startReceiveLoop, which do the bookkeeping, and the counter and the WaitGroup are no longer touched anywhere else. The tests read the counter through goroutineCount. No behaviour change. Assisted-by: AI (cherry picked from commit 13187b8b237da6592d5f98300ad161b82d0d7e90) --- docs/pages/deployment/cli-reference.rst | 3 +- docs/pages/deployment/server_options.rst | 3 +- docs/pages/release_notes.rst | 1 + network/cmd/cmd.go | 1 + network/config.go | 4 + network/config_test.go | 1 + network/network.go | 1 + network/transport/grpc/config.go | 16 +++ network/transport/grpc/connection.go | 110 ++++++++++++-- network/transport/grpc/connection_list.go | 5 +- network/transport/grpc/connection_manager.go | 2 +- .../transport/grpc/connection_manager_test.go | 18 +-- network/transport/grpc/connection_test.go | 136 ++++++++++++++++-- 13 files changed, 264 insertions(+), 37 deletions(-) diff --git a/docs/pages/deployment/cli-reference.rst b/docs/pages/deployment/cli-reference.rst index 4d441a3e9..904290b8e 100755 --- a/docs/pages/deployment/cli-reference.rst +++ b/docs/pages/deployment/cli-reference.rst @@ -46,7 +46,7 @@ The following options apply to the server commands below: --http.default.log string What to log about HTTP requests. Options are 'nothing', 'metadata' (log request method, URI, IP and response code), and 'metadata-and-body' (log the request and response body, in addition to the metadata). (default "metadata") --http.default.tls string Whether to enable TLS for the default interface, options are 'disabled', 'server', 'server-client'. Leaving it empty is synonymous to 'disabled', --internalratelimiter When set, expensive internal calls are rate-limited to protect the network. Always enabled in strict mode. (default true) - --jsonld.contexts.localmapping stringToString This setting allows mapping external URLs to local files for e.g. preventing external dependencies. These mappings have precedence over those in remoteallowlist. (default [https://nuts.nl/credentials/v1=assets/contexts/nuts.ldjson,https://www.w3.org/2018/credentials/v1=assets/contexts/w3c-credentials-v1.ldjson,https://w3c-ccg.github.io/lds-jws2020/contexts/lds-jws2020-v1.json=assets/contexts/lds-jws2020-v1.ldjson,https://schema.org=assets/contexts/schema-org-v13.ldjson]) + --jsonld.contexts.localmapping stringToString This setting allows mapping external URLs to local files for e.g. preventing external dependencies. These mappings have precedence over those in remoteallowlist. (default [https://schema.org=assets/contexts/schema-org-v13.ldjson,https://nuts.nl/credentials/v1=assets/contexts/nuts.ldjson,https://www.w3.org/2018/credentials/v1=assets/contexts/w3c-credentials-v1.ldjson,https://w3c-ccg.github.io/lds-jws2020/contexts/lds-jws2020-v1.json=assets/contexts/lds-jws2020-v1.ldjson]) --jsonld.contexts.remoteallowlist strings In strict mode, fetching external JSON-LD contexts is not allowed except for context-URLs listed here. (default [https://schema.org,https://www.w3.org/2018/credentials/v1,https://w3c-ccg.github.io/lds-jws2020/contexts/lds-jws2020-v1.json]) --loggerformat string Log format (text, json) (default "text") --network.bootstrapnodes strings List of bootstrap nodes (':') which the node initially connect to. @@ -54,6 +54,7 @@ The following options apply to the server commands below: --network.enablediscovery Whether to enable automatic connecting to other nodes. (default true) --network.enabletls Whether to enable TLS for gRPC connections, which can be disabled for demo/development purposes. It is NOT meant for TLS offloading (see 'tls.offload'). Disabling TLS is not allowed in strict-mode. (default true) --network.grpcaddr string 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). (default ":5555") + --network.idletimeout duration 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. (default 2m0s) --network.maxbackoff duration Maximum between outbound connections attempts to unresponsive nodes (in Golang duration format, e.g. '1h', '30m'). (default 1h0m0s) --network.nodedid string Specifies the DID of the organization that operates this node, typically a vendor for EPD software. 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.protocols ints Specifies the list of network protocols to enable on the server. They are specified by version (1, 2). If not set, all protocols are enabled. diff --git a/docs/pages/deployment/server_options.rst b/docs/pages/deployment/server_options.rst index 628a4daba..c79159e70 100755 --- a/docs/pages/deployment/server_options.rst +++ b/docs/pages/deployment/server_options.rst @@ -52,7 +52,7 @@ http.default.auth.type Whether to enable authentication for the default interface, specify 'token_v2' for bearer token mode or 'token' for legacy bearer token mode. http.default.cors.origin [] When set, enables CORS from the specified origins on the default HTTP interface. **JSONLD** - jsonld.contexts.localmapping [https://nuts.nl/credentials/v1=assets/contexts/nuts.ldjson,https://www.w3.org/2018/credentials/v1=assets/contexts/w3c-credentials-v1.ldjson,https://w3c-ccg.github.io/lds-jws2020/contexts/lds-jws2020-v1.json=assets/contexts/lds-jws2020-v1.ldjson,https://schema.org=assets/contexts/schema-org-v13.ldjson] This setting allows mapping external URLs to local files for e.g. preventing external dependencies. These mappings have precedence over those in remoteallowlist. + jsonld.contexts.localmapping [https://www.w3.org/2018/credentials/v1=assets/contexts/w3c-credentials-v1.ldjson,https://w3c-ccg.github.io/lds-jws2020/contexts/lds-jws2020-v1.json=assets/contexts/lds-jws2020-v1.ldjson,https://schema.org=assets/contexts/schema-org-v13.ldjson,https://nuts.nl/credentials/v1=assets/contexts/nuts.ldjson] This setting allows mapping external URLs to local files for e.g. preventing external dependencies. These mappings have precedence over those in remoteallowlist. jsonld.contexts.remoteallowlist [https://schema.org,https://www.w3.org/2018/credentials/v1,https://w3c-ccg.github.io/lds-jws2020/contexts/lds-jws2020-v1.json] In strict mode, fetching external JSON-LD contexts is not allowed except for context-URLs listed here. **Network** network.bootstrapnodes [] List of bootstrap nodes (':') which the node initially connect to. @@ -60,6 +60,7 @@ network.enablediscovery true Whether to enable automatic connecting to other nodes. network.enabletls true Whether to enable TLS for gRPC connections, which can be disabled for demo/development purposes. It is NOT meant for TLS offloading (see 'tls.offload'). Disabling TLS is not allowed in strict-mode. 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'). network.nodedid Specifies the DID of the organization that operates this node, typically a vendor for EPD software. 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.protocols [] Specifies the list of network protocols to enable on the server. They are specified by version (1, 2). If not set, all protocols are enabled. diff --git a/docs/pages/release_notes.rst b/docs/pages/release_notes.rst index 7edd8c671..8cb1d9880 100644 --- a/docs/pages/release_notes.rst +++ b/docs/pages/release_notes.rst @@ -12,6 +12,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 diff --git a/network/cmd/cmd.go b/network/cmd/cmd.go index dd93f58ad..26a644606 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 organization that operates this node, typically a vendor for EPD software. 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 80c90e7ba..f4b242b2d 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: time.Hour, + IdleTimeout: 2 * time.Minute, ProtocolV2: v2.DefaultConfig(), EnableDiscovery: true, } diff --git a/network/config_test.go b/network/config_test.go index b6ffc74d6..69faf4aff 100644 --- a/network/config_test.go +++ b/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) { diff --git a/network/network.go b/network/network.go index 457946e4b..86aebe2a6 100644 --- a/network/network.go +++ b/network/network.go @@ -241,6 +241,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 9640fa8d8..b8c0c84c9 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) }, @@ -97,6 +102,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 @@ -123,6 +137,8 @@ type Config struct { clientCertHeaderName 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 b69a560bc..d96a6ed2c 100644 --- a/network/transport/grpc/connection.go +++ b/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,8 +105,15 @@ 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() @@ -113,7 +121,16 @@ type conn struct { 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 { diff --git a/network/transport/grpc/connection_list.go b/network/transport/grpc/connection_list.go index 57464e602..b8900553b 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/v5/core" "github.com/nuts-foundation/nuts-node/v5/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 3d1e2a636..22d0523f4 100644 --- a/network/transport/grpc/connection_manager.go +++ b/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) diff --git a/network/transport/grpc/connection_manager_test.go b/network/transport/grpc/connection_manager_test.go index f59a3bcd5..9110706cf 100644 --- a/network/transport/grpc/connection_manager_test.go +++ b/network/transport/grpc/connection_manager_test.go @@ -291,7 +291,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) @@ -818,7 +818,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) @@ -860,7 +860,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 @@ -1005,7 +1005,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) @@ -1026,7 +1026,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) @@ -1054,7 +1054,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) @@ -1083,7 +1083,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) @@ -1106,7 +1106,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) @@ -1129,7 +1129,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_test.go b/network/transport/grpc/connection_test.go index b260527a5..890cd7579 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/v5/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() @@ -114,7 +115,7 @@ func Test_conn_startSending(t *testing.T) { p := &TestProtocol{} _ = connection.registerStream(p, stream) - assert.Equal(t, int32(2), connection.activeGoroutines) // startSending and startReceiving + assert.Equal(t, int32(2), connection.goroutineCount()) // startSending and startReceiving // Disconnect before cancelling the stream: this guarantees the connection context is // cancelled before RecvMsg returns, so the receive loop drops the message instead of @@ -123,7 +124,7 @@ func Test_conn_startSending(t *testing.T) { stream.cancelFunc() test.WaitFor(t, func() (bool, error) { - return atomic.LoadInt32(&connection.activeGoroutines) == 0, nil + return connection.goroutineCount() == 0, nil }, 5*time.Second, "waiting for all goroutines to exit") // A deliberate local disconnect must not record a close error. Default value is OK. @@ -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()) + }) +}