From 9c79777f2d846e2129f79a7ed88dad0bc6346d7c Mon Sep 17 00:00:00 2001 From: Free-Guy-IR Date: Wed, 4 Nov 2026 02:43:22 +0330 Subject: [PATCH 1/3] fix(rpc): accept panel keepalive pings and detect a panel that stopped answering --- controller/rpc/keepalive/keepalive_test.go | 230 +++++++++++++++++++++ controller/rpc/service.go | 9 + 2 files changed, 239 insertions(+) create mode 100644 controller/rpc/keepalive/keepalive_test.go diff --git a/controller/rpc/keepalive/keepalive_test.go b/controller/rpc/keepalive/keepalive_test.go new file mode 100644 index 00000000..21264a10 --- /dev/null +++ b/controller/rpc/keepalive/keepalive_test.go @@ -0,0 +1,230 @@ +package keepalive_test + +import ( + "context" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "encoding/pem" + "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{} + once 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{})} + t.Cleanup(func() { l.Close() }) + go p.serve() + return p +} + +func (p *freezableProxy) serve() { + for { + client, err := p.listener.Accept() + if err != nil { + return + } + server, err := net.Dial("tcp", p.target) + if err != nil { + 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 err == io.EOF || p.frozen.Load() { + p.once.Do(func() { close(p.serverClosed) }) + } + 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: + if elapsed := time.Since(started); elapsed < 30*time.Second { + t.Fatalf("the node closed a merely idle connection after %.0fs, before its keepalive ping was due", elapsed.Seconds()) + } + case <-time.After(55 * time.Second): + t.Fatal("the node kept a connection to a panel that stopped answering for 55 s; it must ping an idle panel and close the connection when the ping is not acknowledged") + } +} diff --git a/controller/rpc/service.go b/controller/rpc/service.go index d633c412..f5a9d302 100644 --- a/controller/rpc/service.go +++ b/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: 10 * 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)), ) From f538721f6077255ba074c8cb7c270532c95b9209 Mon Sep 17 00:00:00 2001 From: Free-Guy-IR Date: Thu, 5 Nov 2026 00:33:43 +0330 Subject: [PATCH 2/3] fix(rpc): keep grpc-go's default 20 s keepalive timeout so TCP_USER_TIMEOUT is unchanged grpc-go also applies the server keepalive Timeout as the Linux TCP_USER_TIMEOUT of every accepted connection. With Timeout 10 s a stalling slow link was aborted after 10 s instead of 20 s today. Timeout 20 s keeps that value and still closes a connection whose panel stopped answering about 50 s after its last frame. Add a Linux-only test that reads TCP_USER_TIMEOUT from the accepted panel socket, and raise the dead-panel test bound to 70 s. --- controller/rpc/keepalive/keepalive_test.go | 8 +- .../keepalive/tcp_user_timeout_linux_test.go | 77 +++++++++++++++++++ controller/rpc/service.go | 2 +- 3 files changed, 83 insertions(+), 4 deletions(-) create mode 100644 controller/rpc/keepalive/tcp_user_timeout_linux_test.go diff --git a/controller/rpc/keepalive/keepalive_test.go b/controller/rpc/keepalive/keepalive_test.go index 21264a10..c22708d3 100644 --- a/controller/rpc/keepalive/keepalive_test.go +++ b/controller/rpc/keepalive/keepalive_test.go @@ -221,10 +221,12 @@ func TestNodeClosesAConnectionWhosePanelStoppedAnswering(t *testing.T) { started := time.Now() select { case <-proxy.serverClosed: - if elapsed := time.Since(started); elapsed < 30*time.Second { + 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()) } - case <-time.After(55 * time.Second): - t.Fatal("the node kept a connection to a panel that stopped answering for 55 s; it must ping an idle panel and close the connection when the ping is not acknowledged") + t.Logf("the node closed the connection %.1fs after the panel stopped answering", elapsed.Seconds()) + 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") } } diff --git a/controller/rpc/keepalive/tcp_user_timeout_linux_test.go b/controller/rpc/keepalive/tcp_user_timeout_linux_test.go new file mode 100644 index 00000000..3d067563 --- /dev/null +++ b/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) + } +} diff --git a/controller/rpc/service.go b/controller/rpc/service.go index f5a9d302..fef923e9 100644 --- a/controller/rpc/service.go +++ b/controller/rpc/service.go @@ -18,7 +18,7 @@ import ( var ( panelKeepaliveEnforcement = keepalive.EnforcementPolicy{MinTime: 5 * time.Second, PermitWithoutStream: true} - panelKeepalive = keepalive.ServerParameters{Time: 30 * time.Second, Timeout: 10 * time.Second} + panelKeepalive = keepalive.ServerParameters{Time: 30 * time.Second, Timeout: 20 * time.Second} ) type Service struct { From 6b3c30cb976037af6f68460ae664be175ee975ca Mon Sep 17 00:00:00 2001 From: Free-Guy-IR Date: Tue, 10 Nov 2026 04:23:43 +0330 Subject: [PATCH 3/3] test(rpc): surface a broken proxy link instead of waiting out the keepalive window The freezable proxy only signalled serverClosed on io.EOF or once the freeze was set. A non-EOF read error before the freeze left the test to wait the full 70 s and then fail with the wrong conclusion, blaming the node for keeping a connection the proxy had already lost. The proxy now reports such an error on its own channel and the test fails immediately with the real reason, and the dial goes through a net.Dialer with the test's context so it is cancelled with the test. --- controller/rpc/keepalive/keepalive_test.go | 35 ++++++++++++++++++---- 1 file changed, 29 insertions(+), 6 deletions(-) diff --git a/controller/rpc/keepalive/keepalive_test.go b/controller/rpc/keepalive/keepalive_test.go index c22708d3..a6021830 100644 --- a/controller/rpc/keepalive/keepalive_test.go +++ b/controller/rpc/keepalive/keepalive_test.go @@ -9,6 +9,8 @@ import ( "crypto/x509" "crypto/x509/pkix" "encoding/pem" + "errors" + "fmt" "io" "math/big" "net" @@ -151,7 +153,9 @@ type freezableProxy struct { 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 { @@ -160,20 +164,35 @@ func startProxy(t *testing.T, target string) *freezableProxy { if err != nil { t.Fatal(err) } - p := &freezableProxy{listener: l, target: target, serverClosed: make(chan struct{})} - t.Cleanup(func() { l.Close() }) - go p.serve() + 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) serve() { +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 := net.Dial("tcp", p.target) + 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 } @@ -199,8 +218,10 @@ func (p *freezableProxy) serve() { _, _ = client.Write(buf[:n]) } if err != nil { - if err == io.EOF || p.frozen.Load() { + 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 @@ -226,6 +247,8 @@ func TestNodeClosesAConnectionWhosePanelStoppedAnswering(t *testing.T) { 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") }