fix(rpc): add gRPC server keepalive so the node accepts client pings and drops a connection that stopped answering

controller/rpc/keepalive/keepalive_test.go

@@ -0,0 +1,255 @@
+package keepalive_test
+
+import (
+	"context"
+	"crypto/ecdsa"
+	"crypto/elliptic"
+	"crypto/rand"
+	"crypto/tls"
+	"crypto/x509"
+	"crypto/x509/pkix"
+	"encoding/pem"
+	"errors"
+	"fmt"
+	"io"
+	"math/big"
+	"net"
+	"os"
+	"path/filepath"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/connectivity"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/keepalive"
+	"google.golang.org/grpc/metadata"
+
+	"github.com/pasarguard/node/common"
+	"github.com/pasarguard/node/config"
+	"github.com/pasarguard/node/controller/rpc"
+	"github.com/pasarguard/node/pkg/tlsutil"
+)
+
+func selfSignedPair(t *testing.T) (string, string) {
+	t.Helper()
+	key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
+	if err != nil {
+		t.Fatal(err)
+	}
+	tmpl := &x509.Certificate{
+		SerialNumber: big.NewInt(1),
+		Subject:      pkix.Name{CommonName: "keepalive-node"},
+		NotBefore:    time.Now().Add(-time.Hour),
+		NotAfter:     time.Now().Add(time.Hour),
+		DNSNames:     []string{"localhost"},
+		IPAddresses:  []net.IP{net.ParseIP("127.0.0.1")},
+		KeyUsage:     x509.KeyUsageDigitalSignature,
+		ExtKeyUsage:  []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
+	}
+	der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key)
+	if err != nil {
+		t.Fatal(err)
+	}
+	keyDER, err := x509.MarshalECPrivateKey(key)
+	if err != nil {
+		t.Fatal(err)
+	}
+	dir := t.TempDir()
+	certPath := filepath.Join(dir, "cert.pem")
+	keyPath := filepath.Join(dir, "key.pem")
+	if err := os.WriteFile(certPath, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(keyPath, pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: keyDER}), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	return certPath, keyPath
+}
+
+func unusedAddr(t *testing.T) string {
+	t.Helper()
+	l, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer l.Close()
+	return l.Addr().String()
+}
+
+func startNode(t *testing.T) (addr, certPath string, key uuid.UUID) {
+	t.Helper()
+	certPath, keyPath := selfSignedPair(t)
+	tlsConfig, err := tlsutil.LoadTLSCredentials(certPath, keyPath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	key = uuid.New()
+	cfg := config.NewTestConfig(t.TempDir(), key)
+	addr = unusedAddr(t)
+	shutdown, service, err := rpc.StartGRPCListener(tlsConfig, addr, cfg)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		service.Disconnect()
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		_ = shutdown(ctx)
+	})
+	return addr, certPath, key
+}
+
+func dialNode(t *testing.T, addr, certPath string, opts ...grpc.DialOption) *grpc.ClientConn {
+	t.Helper()
+	pool, err := tlsutil.LoadClientPool(certPath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	creds := credentials.NewTLS(&tls.Config{RootCAs: pool, ServerName: "localhost"})
+	conn, err := grpc.NewClient(addr, append([]grpc.DialOption{grpc.WithTransportCredentials(creds)}, opts...)...)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { conn.Close() })
+	return conn
+}
+
+func baseInfo(t *testing.T, conn *grpc.ClientConn, key uuid.UUID) {
+	t.Helper()
+	ctx, cancel := context.WithTimeout(metadata.NewOutgoingContext(context.Background(), metadata.Pairs("x-api-key", key.String())), 5*time.Second)
+	defer cancel()
+	if _, err := common.NewNodeServiceClient(conn).GetBaseInfo(ctx, &common.Empty{}); err != nil {
+		t.Fatalf("GetBaseInfo: %v", err)
+	}
+}
+
+func TestNodeKeepsAnIdlePanelConnectionThatPingsEveryTenSeconds(t *testing.T) {
+	t.Parallel()
+	addr, certPath, key := startNode(t)
+	conn := dialNode(t, addr, certPath, grpc.WithKeepaliveParams(keepalive.ClientParameters{
+		Time:                10 * time.Second,
+		Timeout:             5 * time.Second,
+		PermitWithoutStream: true,
+	}))
+	baseInfo(t, conn, key)
+	if state := conn.GetState(); state != connectivity.Ready {
+		t.Fatalf("connection not ready after the first call: %v", state)
+	}
+
+	started := time.Now()
+	ctx, cancel := context.WithTimeout(context.Background(), 45*time.Second)
+	defer cancel()
+	if conn.WaitForStateChange(ctx, connectivity.Ready) {
+		t.Fatalf("the node dropped an idle panel connection that pings every 10 s after %.0fs; state now %v", time.Since(started).Seconds(), conn.GetState())
+	}
+}
+
+type freezableProxy struct {
+	listener     net.Listener
+	target       string
+	frozen       atomic.Bool
+	serverClosed chan struct{}
+	failed       chan error
+	once         sync.Once
+	failOnce     sync.Once
+}
+
+func startProxy(t *testing.T, target string) *freezableProxy {
+	t.Helper()
+	l, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	p := &freezableProxy{
+		listener:     l,
+		target:       target,
+		serverClosed: make(chan struct{}),
+		failed:       make(chan error, 1),
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(func() {
+		cancel()
+		l.Close()
+	})
+	go p.serve(ctx)
+	return p
+}
+
+func (p *freezableProxy) fail(err error) {
+	p.failOnce.Do(func() { p.failed <- err })
+}
+
+func (p *freezableProxy) serve(ctx context.Context) {
+	var dialer net.Dialer
+	for {
+		client, err := p.listener.Accept()
+		if err != nil {
+			return
+		}
+		server, err := dialer.DialContext(ctx, "tcp", p.target)
+		if err != nil {
+			p.fail(fmt.Errorf("the proxy could not reach the node at %s: %w", p.target, err))
+			client.Close()
+			continue
+		}
+		go func() {
+			buf := make([]byte, 32*1024)
+			for {
+				n, err := client.Read(buf)
+				if n > 0 && !p.frozen.Load() {
+					if _, werr := server.Write(buf[:n]); werr != nil {
+						return
+					}
+				}
+				if err != nil {
+					return
+				}
+			}
+		}()
+		go func() {
+			buf := make([]byte, 32*1024)
+			for {
+				n, err := server.Read(buf)
+				if n > 0 && !p.frozen.Load() {
+					_, _ = client.Write(buf[:n])
+				}
+				if err != nil {
+					if errors.Is(err, io.EOF) || p.frozen.Load() {
+						p.once.Do(func() { close(p.serverClosed) })
+					} else {
+						p.fail(fmt.Errorf("the proxy lost its connection to the node before the freeze: %w", err))
+					}
+					client.Close()
+					return
+				}
+			}
+		}()
+	}
+}
+
+func TestNodeClosesAConnectionWhosePanelStoppedAnswering(t *testing.T) {
+	t.Parallel()
+	addr, certPath, key := startNode(t)
+	proxy := startProxy(t, addr)
+	conn := dialNode(t, proxy.listener.Addr().String(), certPath)
+	baseInfo(t, conn, key)
+
+	proxy.frozen.Store(true)
+	started := time.Now()
+	select {
+	case <-proxy.serverClosed:
+		elapsed := time.Since(started)
+		if elapsed < 30*time.Second {
+			t.Fatalf("the node closed a merely idle connection after %.0fs, before its keepalive ping was due", elapsed.Seconds())
+		}
+		t.Logf("the node closed the connection %.1fs after the panel stopped answering", elapsed.Seconds())
+	case err := <-proxy.failed:
+		t.Fatalf("the proxy's own link to the node broke, so this run proves nothing about the node: %v", err)
+	case <-time.After(70 * time.Second):
+		t.Fatal("the node kept a connection to a panel that stopped answering for 70 s; it must ping an idle panel and close the connection when the ping is not acknowledged")
+	}
+}

controller/rpc/keepalive/tcp_user_timeout_linux_test.go

@@ -0,0 +1,77 @@
+//go:build linux
+
+package keepalive_test
+
+import (
+	"context"
+	"net"
+	"os"
+	"strconv"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"golang.org/x/sys/unix"
+	"google.golang.org/grpc"
+)
+
+func sockaddrIs(sa unix.Sockaddr, want *net.TCPAddr) bool {
+	in4, ok := sa.(*unix.SockaddrInet4)
+	if !ok {
+		return false
+	}
+	return in4.Port == want.Port && net.IP(in4.Addr[:]).Equal(want.IP)
+}
+
+func acceptedSocketUserTimeout(t *testing.T, node, client *net.TCPAddr) time.Duration {
+	t.Helper()
+	entries, err := os.ReadDir("/proc/self/fd")
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, entry := range entries {
+		fd, err := strconv.Atoi(entry.Name())
+		if err != nil {
+			continue
+		}
+		local, err := unix.Getsockname(fd)
+		if err != nil || !sockaddrIs(local, node) {
+			continue
+		}
+		peer, err := unix.Getpeername(fd)
+		if err != nil || !sockaddrIs(peer, client) {
+			continue
+		}
+		ms, err := unix.GetsockoptInt(fd, unix.IPPROTO_TCP, unix.TCP_USER_TIMEOUT)
+		if err != nil {
+			t.Fatalf("getsockopt TCP_USER_TIMEOUT: %v", err)
+		}
+		return time.Duration(ms) * time.Millisecond
+	}
+	t.Fatalf("no socket in this process was accepted on %v from %v", node, client)
+	return 0
+}
+
+func TestNodeKeepsGrpcDefaultTCPUserTimeoutOnPanelConnections(t *testing.T) {
+	t.Parallel()
+	addr, certPath, key := startNode(t)
+	var clientAddr atomic.Pointer[net.TCPAddr]
+	conn := dialNode(t, addr, certPath, grpc.WithContextDialer(func(ctx context.Context, target string) (net.Conn, error) {
+		c, err := (&net.Dialer{}).DialContext(ctx, "tcp", target)
+		if err == nil {
+			clientAddr.Store(c.LocalAddr().(*net.TCPAddr))
+		}
+		return c, err
+	}))
+	baseInfo(t, conn, key)
+
+	nodeAddr, err := net.ResolveTCPAddr("tcp", addr)
+	if err != nil {
+		t.Fatal(err)
+	}
+	got := acceptedSocketUserTimeout(t, nodeAddr, clientAddr.Load())
+	t.Logf("TCP_USER_TIMEOUT on the accepted panel connection: %s", got)
+	if got != 20*time.Second {
+		t.Fatalf("TCP_USER_TIMEOUT on the accepted panel connection is %s; grpc-go's default keepalive Timeout gives 20s, and a lower value aborts a stalling slow link sooner", got)
+	}
+}

controller/rpc/service.go

@@ -6,12 +6,19 @@ import (
 	"fmt"
 	"log"
 	"net"
+	"time"
 
 	"github.com/pasarguard/node/common"
 	"github.com/pasarguard/node/config"
 	"github.com/pasarguard/node/controller"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/keepalive"
+)
+
+var (
+	panelKeepaliveEnforcement = keepalive.EnforcementPolicy{MinTime: 5 * time.Second, PermitWithoutStream: true}
+	panelKeepalive            = keepalive.ServerParameters{Time: 30 * time.Second, Timeout: 20 * time.Second}
 )
 
 type Service struct {
@@ -37,6 +44,8 @@ func StartGRPCListener(tlsConfig *tls.Config, addr string, cfg *config.Config) (
 		grpc.Creds(creds),
 		grpc.MaxRecvMsgSize(maxMsgSize),
 		grpc.MaxSendMsgSize(maxMsgSize),
+		grpc.KeepaliveEnforcementPolicy(panelKeepaliveEnforcement),
+		grpc.KeepaliveParams(panelKeepalive),
 		grpc.UnaryInterceptor(ConditionalMiddleware(s)),
 		grpc.StreamInterceptor(ConditionalStreamMiddleware(s)),
 	)