From b0faf73c53c68cd5de81bd4d420b24a691d69bf8 Mon Sep 17 00:00:00 2001 From: wyc <1253932803@qq.com> Date: Wed, 8 Jul 2026 16:02:04 +0800 Subject: [PATCH] fix(transport,pty): gRPC keepalive both ends; reset terminal on detach PTY sessions died after 40-60 min idle (client saw "error reading from server: EOF"): with no keepalive anywhere, idle connections were reaped by NAT/firewall middleboxes, the server never noticed the dead client, so the attach handler blocked in Recv() forever and activePTY leaked -- every reconnect failed with "pty session already attached" until the server was restarted. The abrupt disconnect also left the local terminal with the remote app's kitty keyboard protocol still pushed, echoing "9;1:3u" fragments on every key release. - transport.Dial: client keepalive (30s/10s, PermitWithoutStream); covers both daemon and pty connections. Behind Caddy the PING terminates at the proxy, which is exactly the segment that was being idle-reaped; no Caddy config change is needed or allowed. - server: KeepaliveParams (30s/10s) plus KeepaliveEnforcementPolicy (MinTime=10s) -- without relaxing the 5min gRPC default, plaintext direct deployments would GOAWAY the 30s client pings. - ptyattach: emit a conservative terminal reset sequence (kitty keyboard pop/zero, modifyOtherKeys, alt screen, bracketed paste, mouse/focus reporting, cursor) on every session end after a successful attach; skipped when dialing failed. Known leftover: attach takeover semantics. Behind Caddy a client that dies without a FIN reaching the proxy can still leak activePTY; direct deployments are now covered by server-side keepalive within ~40s. Co-Authored-By: Claude Opus 4.8 --- app/server/main.go | 14 +++++++++++++ deploy/tls/Caddyfile.example | 4 ++++ pkg/config/config.go | 12 +++++++++++ pkg/config/config_test.go | 9 +++++++++ pkg/ptyattach/attach.go | 32 +++++++++++++++++++++++++++++ pkg/ptyattach/attach_test.go | 39 ++++++++++++++++++++++++++++++++++++ pkg/transport/client.go | 11 ++++++++++ 7 files changed, 121 insertions(+) diff --git a/app/server/main.go b/app/server/main.go index 59b392e..cc9bc05 100644 --- a/app/server/main.go +++ b/app/server/main.go @@ -12,6 +12,7 @@ import ( "github.com/spf13/cobra" "google.golang.org/grpc" + "google.golang.org/grpc/keepalive" remotefsv1 "flyingEirc/Rclaude/api/proto/remotefs/v1" "flyingEirc/Rclaude/pkg/auth" @@ -166,7 +167,20 @@ func newGRPCServer( return nil, nil, fmt.Errorf("server: listen %q: %w", cfg.Listen, err) } // recovery 拦截器置于最外层，先于 auth 执行，以兜住整条 handler 同步栈的 panic。 + // keepalive：服务端主动 PING 对端，死连接在 Time+Timeout 内触发 stream 报错， + // 走既有 shutdown/UnregisterPTY 清理路径，避免 activePTY 泄漏卡死重连。 + // Caddy 前置时对端是回源连接（通常 loopback），探测客户端主要靠明文直连模式。 + // EnforcementPolicy 必须放宽到 MinTime < 客户端 Time（gRPC 默认 5 分钟会把 + // 30s 一次的客户端 PING 判为滥用并 GOAWAY），配对约束见 pkg/config。 grpcServer := grpc.NewServer( + grpc.KeepaliveParams(keepalive.ServerParameters{ + Time: config.DefaultGRPCKeepaliveTime, + Timeout: config.DefaultGRPCKeepaliveTimeout, + }), + grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{ + MinTime: config.DefaultGRPCKeepaliveMinTime, + PermitWithoutStream: true, + }), grpc.ChainStreamInterceptor( recoveryStreamInterceptor(logger), auth.StreamServerInterceptor(verifier), diff --git a/deploy/tls/Caddyfile.example b/deploy/tls/Caddyfile.example index a6a19cc..f2cf736 100644 --- a/deploy/tls/Caddyfile.example +++ b/deploy/tls/Caddyfile.example @@ -11,6 +11,10 @@ # 回源，gRPC 直接失败。 # - 不要给这条流设 response_header_timeout 或读写超时：RemotePTY/RemoteFS 都是 # 常驻双向流，交互式 PTY 可长时间空闲，任何流级超时都会把它掐断。 +# - 客户端/服务端已启用 gRPC keepalive（HTTP/2 PING，见 docs/reference/ +# grpc-keepalive.md）。PING 逐跳终止：客户端 PING 到 Caddy 即被应答，作用是 +# 保活 client↔Caddy 段的 NAT/防火墙映射并探测该段死亡；Caddy 本身无需也不应 +# 为此增加任何配置。 {$RCLAUDE_DOMAIN} { reverse_proxy h2c://{$RCLAUDE_UPSTREAM} { diff --git a/pkg/config/config.go b/pkg/config/config.go index 202cfce..6b0de85 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -36,6 +36,18 @@ const ( DefaultAuditQueueSize = 256 DefaultStartupMaxRetries = 3 DefaultStartupRetryDelay = time.Second + // gRPC keepalive：客户端与服务端周期性发 HTTP/2 PING，一是保活路径上的 + // NAT/防火墙映射（PTY 可长时间空闲，实测 40~60 分钟会被中间设备掐断）， + // 二是让两端在 Time+Timeout 内探测到死连接并走既有清理路径。 + // PING 逐跳终止：Caddy 前置 TLS 时，客户端 PING 只到 Caddy，服务端 PING + // 只到 Caddy 回源连接；仅明文直连时两端互相探测。详见 + // docs/reference/grpc-keepalive.md。 + DefaultGRPCKeepaliveTime = 30 * time.Second + DefaultGRPCKeepaliveTimeout = 10 * time.Second + // DefaultGRPCKeepaliveMinTime 是服务端 EnforcementPolicy 允许的客户端 + // PING 最小间隔，必须小于 DefaultGRPCKeepaliveTime：gRPC 默认值为 5 分钟， + // 不放宽的话明文直连部署会把 30s 一次的客户端 PING 判为滥用并 GOAWAY。 + DefaultGRPCKeepaliveMinTime = 10 * time.Second ) var ( diff --git a/pkg/config/config_test.go b/pkg/config/config_test.go index 9c42439..a734def 100644 --- a/pkg/config/config_test.go +++ b/pkg/config/config_test.go @@ -557,3 +557,12 @@ func escapeYAML(p string) string { out = append(out, '"') return string(out) } + +func TestGRPCKeepaliveDefaultsPairing(t *testing.T) { + // 服务端 EnforcementPolicy.MinTime 必须小于客户端 PING 间隔，否则明文直连 + // 部署下服务端会把客户端 keepalive 判为滥用并 GOAWAY。 + assert.Less(t, config.DefaultGRPCKeepaliveMinTime, config.DefaultGRPCKeepaliveTime) + // grpc-go 会把小于 10s 的客户端 keepalive Time 强制钳到 10s， + // 常量低于该值会造成"配置与实际行为不一致"。 + assert.GreaterOrEqual(t, config.DefaultGRPCKeepaliveTime, 10*time.Second) +} diff --git a/pkg/ptyattach/attach.go b/pkg/ptyattach/attach.go index f91aba4..9d94764 100644 --- a/pkg/ptyattach/attach.go +++ b/pkg/ptyattach/attach.go @@ -23,6 +23,24 @@ import ( const defaultTerm = "xterm-256color" +// terminalResetSequence 撤销远端程序可能透传给本地终端仿真器、而断开时来不及 +// 自行恢复的模式。透传架构下本地不解析字节流（区别于 mosh 的终端仿真方案， +// 见 docs/reference/mosh.md），只能在会话结束时保守复位；不支持某序列的终端 +// 会按未知 CSI 忽略。实测残留案例：kitty keyboard protocol 未弹栈导致每次按键 +// 回显 "9;1:3u" 之类的 release 事件残片。 +const terminalResetSequence = "\x1b[4;0m" + // xterm modifyOtherKeys 关闭 + "\x1b[?1049l" + // 退出 alternate screen + "\x1b[?2004l" + // bracketed paste 关闭 + "\x1b[?1000l\x1b[?1002l\x1b[?1003l\x1b[?1006l" + // 鼠标上报及 SGR 编码关闭 + "\x1b[?1004l" + // focus 上报关闭 + "\x1b[?1l\x1b>" + // 光标键/小键盘回到普通模式 + "\x1b[?7h" + // 自动换行恢复 + "\x1b[0m" + // SGR 属性清零 + "\x1b[0 q" + // 光标形状恢复默认 + "\x1b[?25h" // 光标恢复可见 + var ( errTTYRequired = errors.New("ptyattach: stdin and stdout must both be interactive terminals") errEmptyServerToken = errors.New("ptyattach: daemon config must include server.token") @@ -93,6 +111,7 @@ type commandRuntime struct { stream ptyclient.Stream closer io.Closer stdin io.ReadCloser + stdout io.Writer stopBridge func() frameMax int predictor ptyclient.Predictor @@ -184,6 +203,7 @@ func prepareCommandRuntime(ctx context.Context, deps commandDeps, cfg loadedConf stream: stream, closer: closer, stdin: stdin, + stdout: deps.stdout, stopBridge: stopBridge, frameMax: clientFrameMax(int64(cfg.FrameMax)), predictor: newPredictor(cfg.Predict, deps.stdout, termSession.InitialSize), @@ -218,11 +238,23 @@ func closeCommandRuntime(runErr *error, runtime commandRuntime) { *runErr = closeErr } } + // 会话建立后的所有结束路径（含远端异常断开）都要复位终端仿真器， + // 写失败只能放弃（终端多半已不可写），仍继续恢复 termios。 + writeTerminalReset(runtime.stdout) if restoreErr := runtime.termSession.Restore(); restoreErr != nil && runErr != nil && *runErr == nil { *runErr = fmt.Errorf("ptyattach: restore terminal: %w", restoreErr) } } +func writeTerminalReset(out io.Writer) { + if out == nil { + return + } + if _, err := io.WriteString(out, terminalResetSequence); err != nil { + return + } +} + func commandTermName(termName string) string { termName = strings.TrimSpace(termName) if termName == "" { diff --git a/pkg/ptyattach/attach_test.go b/pkg/ptyattach/attach_test.go index 5f1cb08..ad24702 100644 --- a/pkg/ptyattach/attach_test.go +++ b/pkg/ptyattach/attach_test.go @@ -3,6 +3,7 @@ package ptyattach import ( "bytes" "context" + "errors" "io" "os" "runtime" @@ -182,6 +183,8 @@ func TestRunCommandReusesDaemonConfigAndBridgesPTY(t *testing.T) { assert.Equal(t, "example.com:9326", gotAddress) assert.Equal(t, "tok-auth", gotToken) assert.Contains(t, stdout.String(), "remote-ready\n") + // 会话结束时必须以终端复位序列收尾，撤销远端透传的终端模式残留。 + assert.True(t, strings.HasSuffix(stdout.String(), terminalResetSequence)) assert.True(t, restoreCalled) assert.False(t, stdin.closed) @@ -194,6 +197,42 @@ func TestRunCommandReusesDaemonConfigAndBridgesPTY(t *testing.T) { assert.Contains(t, sentStdinPayloads(frames), []byte("hello from cli")) } +func TestRunCommandDialFailureSkipsTerminalReset(t *testing.T) { + var stdout bytes.Buffer + restoreCalled := false + + deps := commandDeps{ + stdin: io.NopCloser(bytes.NewReader(nil)), + stdout: &stdout, + loadConfig: func(string) (loadedConfig, error) { + return loadedConfig{Address: "example.com:9326", Token: "tok-auth", FrameMax: 64}, nil + }, + terminal: fakeTerminal{ + tty: true, + session: terminalSession{ + InitialSize: ptyclient.WindowSize{Cols: 80, Rows: 24}, + Resizes: closedResizeCh(), + Restore: func() error { + restoreCalled = true + return nil + }, + }, + }, + dialPTY: func(context.Context, dialConfig) (ptyclient.Stream, io.Closer, error) { + return nil, nil, errors.New("dial failed") + }, + stdinFD: 0, + stdoutFD: 1, + termName: "xterm-256color", + } + + err := runCommand(context.Background(), deps, "daemon.yaml") + require.Error(t, err) + assert.True(t, restoreCalled) + // 会话从未建立，远端不可能污染终端，不应写复位序列。 + assert.Empty(t, stdout.String()) +} + func TestRunCommandMapsServerErrorToExitStatus(t *testing.T) { stream := newFakeStream() diff --git a/pkg/transport/client.go b/pkg/transport/client.go index 9aa9274..89c2c91 100644 --- a/pkg/transport/client.go +++ b/pkg/transport/client.go @@ -12,9 +12,11 @@ import ( "google.golang.org/grpc" "google.golang.org/grpc/credentials" "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/keepalive" remotefsv1 "flyingEirc/Rclaude/api/proto/remotefs/v1" "flyingEirc/Rclaude/pkg/auth" + "flyingEirc/Rclaude/pkg/config" ) // 错误集合：调用方应使用 errors.Is 比较。 @@ -63,8 +65,17 @@ func Dial(_ context.Context, opts DialOptions) (*grpc.ClientConn, error) { if err != nil { return nil, err } + // keepalive 让空闲的 daemon/PTY 长连接持续产生链路流量：既防止 NAT/防火墙 + // 回收空闲映射，也在 Time+Timeout 内探测到死路径让 RPC 尽快报错退出。 + // Caddy 前置 TLS 时 PING 终止于 Caddy，保活与探测覆盖的是 client↔Caddy 段， + // 这正是空闲掐断的发生段；服务端对应参数见 app/server/main.go。 dialOpts := []grpc.DialOption{ grpc.WithTransportCredentials(creds), + grpc.WithKeepaliveParams(keepalive.ClientParameters{ + Time: config.DefaultGRPCKeepaliveTime, + Timeout: config.DefaultGRPCKeepaliveTimeout, + PermitWithoutStream: true, + }), } if opts.Dialer != nil { dialOpts = append(dialOpts, grpc.WithContextDialer(opts.Dialer))