@@ -32,13 +52,6 @@ an API that records Zoom, Google Meet, Microsoft Teams, in-person meetings, and
-
-
- 
-
- The complete IDE crafted for professional Go developers
-
-
## What is frp?
diff --git a/README_zh.md b/README_zh.md
index 600cd4fd..e4d28410 100644
--- a/README_zh.md
+++ b/README_zh.md
@@ -14,6 +14,26 @@ frp 是一个完全开源的项目，我们的开发工作完全依靠赞助者
Gold Sponsors
+
+
+ 
+
+ High-performance residential and ISP proxies for developers
+
+
+ 90M+ residential IPs worldwide. Rotating IPs, sticky sessions, and traffic that never expires.
+
+ From $0.55/GB. Use RAPID10 for 10% off. Try it for free.
+
+
+
+
+
+
+ The complete IDE crafted for professional Go developers
+
+
+
@@ -34,13 +54,6 @@ an API that records Zoom, Google Meet, Microsoft Teams, in-person meetings, and
-
-
-
-
- The complete IDE crafted for professional Go developers
-
-
## 为什么使用 frp ？
diff --git a/Release.md b/Release.md
index 33fc0aea..35d0442c 100644
--- a/Release.md
+++ b/Release.md
@@ -1,9 +1,13 @@
-## Features
+## Improvements
-* UDP packet payloads for ordinary UDP proxies and SUDP now use a dedicated binary codec when frpc and frps successfully negotiate the capability under wire protocol v2, using a more compact wire representation. Wire protocol v1 remains JSON; wire protocol v2 falls back to JSON `UDPPacket` when the peer does not support or did not negotiate the capability.
+* Building frp from source now requires Go 1.26.0 or later.
## Fixes
-* Fixed a server panic and remote denial of service caused by a client sending a negative `pool_count`. Negative values are now rejected before work-connection pool resources are allocated.
-* Fixed `frpc verify` ignoring configured `featureGates`, which caused VirtualNet configurations to be rejected even when the feature was enabled.
-* Fixed a case-insensitive validation bypass that allowed `customDomains` under the configured `subDomainHost` to be registered using mixed-case domain names.
+* Updated dependencies to fix SSH tunnel gateway channel deadlocks and panics caused by malformed challenges from NTLM-authenticated HTTP proxies.
+* Fixed VirtualNet routing issues during reconnection and shutdown, and excessive retries after repeated connection failures.
+* Health checks now reset the failure count after success, so `healthCheck.maxFailed` applies to consecutive failures.
+* Fixed control connections getting stuck after message write failures, which could block reconnection.
+* Fixed incorrect daily traffic statistics around daylight-saving transitions and skipped local midnights.
+* Fixed early-data loss in HTTP virtual-host CONNECT tunnels. Requests with a positive content length or transfer encoding now return HTTP 400.
+* Fixed work connections remaining open when server-side bandwidth limiting is enabled for TCP, HTTP, and UDP proxies.
diff --git a/client/connector.go b/client/connector.go
index 34b24f0d..41332378 100644
--- a/client/connector.go
+++ b/client/connector.go
@@ -24,7 +24,7 @@ import (
"time"
libnet "github.com/fatedier/golib/net"
- fmux "github.com/hashicorp/yamux"
+ fmux "github.com/fatedier/yamux"
quic "github.com/quic-go/quic-go"
"github.com/samber/lo"
diff --git a/client/health/health.go b/client/health/health.go
index d298e615..cedbba37 100644
--- a/client/health/health.go
+++ b/client/health/health.go
@@ -30,6 +30,11 @@ import (
var ErrHealthCheckType = errors.New("error health check type")
+func newHealthTimer(interval time.Duration) (<-chan time.Time, func()) {
+ timer := time.NewTimer(interval)
+ return timer.C, func() { timer.Stop() }
+}
+
type Monitor struct {
checkType string
interval time.Duration
@@ -49,6 +54,9 @@ type Monitor struct {
ctx context.Context
cancel context.CancelFunc
+ doneCh chan struct{}
+
+ timerFactory func(time.Duration) (<-chan time.Time, func())
}
func NewMonitor(ctx context.Context, cfg v1.HealthCheckConfig, addr string,
@@ -91,6 +99,8 @@ func NewMonitor(ctx context.Context, cfg v1.HealthCheckConfig, addr string,
statusFailedFn: statusFailedFn,
ctx: newctx,
cancel: cancel,
+ doneCh: make(chan struct{}),
+ timerFactory: newHealthTimer,
}
}
@@ -102,39 +112,66 @@ func (monitor *Monitor) Stop() {
monitor.cancel()
}
+// Done is closed when the worker launched by Start has exited.
+func (monitor *Monitor) Done() <-chan struct{} {
+ return monitor.doneCh
+}
+
func (monitor *Monitor) checkWorker() {
- xl := xlog.FromContextSafe(monitor.ctx)
+ defer close(monitor.doneCh)
+
for {
+ if monitor.ctx.Err() != nil {
+ return
+ }
+
doCtx, cancel := context.WithDeadline(monitor.ctx, time.Now().Add(monitor.timeout))
err := monitor.doCheck(doCtx)
+ cancel()
// check if this monitor has been closed
- select {
- case <-monitor.ctx.Done():
- cancel()
+ if monitor.ctx.Err() != nil {
return
- default:
- cancel()
}
+ monitor.handleCheckResult(err)
- if err == nil {
- xl.Tracef("do one health check success")
- if !monitor.statusOK && monitor.statusNormalFn != nil {
- xl.Infof("health check status change to success")
- monitor.statusOK = true
- monitor.statusNormalFn()
- }
- } else {
- xl.Warnf("do one health check failed: %v", err)
- monitor.failedTimes++
- if monitor.statusOK && int(monitor.failedTimes) >= monitor.maxFailedTimes && monitor.statusFailedFn != nil {
- xl.Warnf("health check status change to failed")
- monitor.statusOK = false
- monitor.statusFailedFn()
- }
+ if !monitor.waitForNextCheck() {
+ return
}
+ }
+}
- time.Sleep(monitor.interval)
+func (monitor *Monitor) handleCheckResult(err error) {
+ xl := xlog.FromContextSafe(monitor.ctx)
+ if err == nil {
+ xl.Tracef("do one health check success")
+ monitor.failedTimes = 0
+ if !monitor.statusOK && monitor.statusNormalFn != nil {
+ xl.Infof("health check status change to success")
+ monitor.statusOK = true
+ monitor.statusNormalFn()
+ }
+ return
+ }
+
+ xl.Warnf("do one health check failed: %v", err)
+ monitor.failedTimes++
+ if monitor.statusOK && int(monitor.failedTimes) >= monitor.maxFailedTimes && monitor.statusFailedFn != nil {
+ xl.Warnf("health check status change to failed")
+ monitor.statusOK = false
+ monitor.statusFailedFn()
+ }
+}
+
+func (monitor *Monitor) waitForNextCheck() bool {
+ timerC, stopTimer := monitor.timerFactory(monitor.interval)
+ defer stopTimer()
+
+ select {
+ case <-monitor.ctx.Done():
+ return false
+ case <-timerC:
+ return monitor.ctx.Err() == nil
}
}
diff --git a/client/health/health_test.go b/client/health/health_test.go
new file mode 100644
index 00000000..80705384
--- /dev/null
+++ b/client/health/health_test.go
@@ -0,0 +1,575 @@
+// Copyright 2026 The frp Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package health
+
+import (
+ "context"
+ "errors"
+ "net"
+ "net/http"
+ "net/http/httptest"
+ "strings"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/require"
+
+ v1 "github.com/fatedier/frp/pkg/config/v1"
+)
+
+type tcpHealthBackend struct {
+ listener net.Listener
+ accepted chan struct{}
+ done chan struct{}
+}
+
+func newTCPHealthBackend(t *testing.T, addr string, accepted chan struct{}) *tcpHealthBackend {
+ listener, err := net.Listen("tcp", addr)
+ require.NoError(t, err)
+
+ backend := &tcpHealthBackend{
+ listener: listener,
+ accepted: accepted,
+ done: make(chan struct{}),
+ }
+ go func() {
+ defer close(backend.done)
+ for {
+ conn, err := backend.listener.Accept()
+ if err != nil {
+ return
+ }
+ _ = conn.Close()
+ select {
+ case backend.accepted <- struct{}{}:
+ default:
+ }
+ }
+ }()
+ return backend
+}
+
+func (backend *tcpHealthBackend) Close() {
+ _ = backend.listener.Close()
+ <-backend.done
+}
+
+func TestMonitorConsecutiveFailureWindows(t *testing.T) {
+ tests := []struct {
+ name string
+ setup func(*testing.T, func(), func()) (*Monitor, func(bool))
+ }{
+ {
+ name: "HTTP",
+ setup: func(t *testing.T, normalFn, failedFn func()) (*Monitor, func(bool)) {
+ var healthy atomic.Bool
+ healthy.Store(true)
+ server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+ if !healthy.Load() {
+ w.WriteHeader(http.StatusServiceUnavailable)
+ }
+ }))
+ t.Cleanup(server.Close)
+
+ monitor := NewMonitor(
+ context.Background(),
+ v1.HealthCheckConfig{
+ Type: "http",
+ Path: "/health",
+ TimeoutSeconds: 1,
+ IntervalSeconds: 1,
+ MaxFailed: 3,
+ },
+ strings.TrimPrefix(server.URL, "http://"),
+ normalFn,
+ failedFn,
+ )
+ return monitor, healthy.Store
+ },
+ },
+ {
+ name: "TCP",
+ setup: func(t *testing.T, normalFn, failedFn func()) (*Monitor, func(bool)) {
+ listener, err := net.Listen("tcp", "127.0.0.1:0")
+ require.NoError(t, err)
+ t.Cleanup(func() { _ = listener.Close() })
+ go func() {
+ for {
+ conn, err := listener.Accept()
+ if err != nil {
+ return
+ }
+ _ = conn.Close()
+ }
+ }()
+
+ monitor := NewMonitor(
+ context.Background(),
+ v1.HealthCheckConfig{
+ Type: "tcp",
+ TimeoutSeconds: 1,
+ IntervalSeconds: 1,
+ MaxFailed: 3,
+ },
+ listener.Addr().String(),
+ normalFn,
+ failedFn,
+ )
+ healthyAddr := monitor.addr
+ return monitor, func(healthy bool) {
+ if healthy {
+ monitor.addr = healthyAddr
+ } else {
+ monitor.addr = "127.0.0.1:0"
+ }
+ }
+ },
+ },
+ }
+
+ for _, test := range tests {
+ t.Run(test.name, func(t *testing.T) {
+ var events []string
+ monitor, setHealthy := test.setup(
+ t,
+ func() { events = append(events, "normal") },
+ func() { events = append(events, "failed") },
+ )
+ t.Cleanup(monitor.Stop)
+
+ runCheck := func(healthy bool) {
+ t.Helper()
+ setHealthy(healthy)
+ ctx, cancel := context.WithTimeout(monitor.ctx, time.Second)
+ err := monitor.doCheck(ctx)
+ cancel()
+ if healthy {
+ require.NoError(t, err)
+ } else {
+ require.Error(t, err)
+ }
+ monitor.handleCheckResult(err)
+ }
+
+ runCheck(true)
+ require.True(t, monitor.statusOK)
+ require.Zero(t, monitor.failedTimes)
+ require.Equal(t, []string{"normal"}, events)
+
+ runCheck(false)
+ runCheck(false)
+ require.True(t, monitor.statusOK)
+ require.Equal(t, uint64(2), monitor.failedTimes)
+ require.Equal(t, []string{"normal"}, events)
+
+ runCheck(true)
+ require.True(t, monitor.statusOK)
+ require.Zero(t, monitor.failedTimes)
+ require.Equal(t, []string{"normal"}, events)
+
+ runCheck(false)
+ require.True(t, monitor.statusOK)
+ require.Equal(t, uint64(1), monitor.failedTimes)
+ require.Equal(t, []string{"normal"}, events)
+
+ runCheck(false)
+ require.True(t, monitor.statusOK)
+ require.Equal(t, uint64(2), monitor.failedTimes)
+
+ runCheck(false)
+ require.False(t, monitor.statusOK)
+ require.Equal(t, uint64(3), monitor.failedTimes)
+ require.Equal(t, []string{"normal", "failed"}, events)
+
+ runCheck(true)
+ require.True(t, monitor.statusOK)
+ require.Zero(t, monitor.failedTimes)
+ require.Equal(t, []string{"normal", "failed", "normal"}, events)
+ })
+ }
+}
+
+type workerStatus struct {
+ failedTimes uint64
+ statusOK bool
+}
+
+type workerInterval struct {
+ status workerStatus
+ timer chan time.Time
+}
+
+func observeWorkerIntervals(monitor *Monitor) <-chan workerInterval {
+ intervals := make(chan workerInterval, 1)
+ monitor.timerFactory = func(time.Duration) (<-chan time.Time, func()) {
+ timer := make(chan time.Time, 1)
+ // The worker takes this snapshot after handleCheckResult. Sending it
+ // publishes the state to the test; no test goroutine reads live fields.
+ interval := workerInterval{
+ status: workerStatus{failedTimes: monitor.failedTimes, statusOK: monitor.statusOK},
+ timer: timer,
+ }
+ select {
+ case intervals <- interval:
+ case <-monitor.ctx.Done():
+ }
+ return timer, func() {}
+ }
+ return intervals
+}
+
+func awaitWorkerInterval(t *testing.T, intervals <-chan workerInterval, want workerStatus) chan time.Time {
+ t.Helper()
+ select {
+ case interval := <-intervals:
+ require.Equal(t, want, interval.status)
+ return interval.timer
+ case <-time.After(time.Second):
+ t.Fatal("health worker did not reach the interval barrier")
+ return nil
+ }
+}
+
+func stopMonitorWorker(t *testing.T, monitor *Monitor) {
+ t.Helper()
+ monitor.Stop()
+ select {
+ case <-monitor.Done():
+ case <-time.After(time.Second):
+ t.Error("health worker did not exit after Stop")
+ }
+}
+
+func TestMonitorWorkerProcessesResults(t *testing.T) {
+ requestReady := make(chan struct{}, 1)
+ responses := make(chan int)
+ requestCanceled := make(chan struct{})
+ var cancelOnce sync.Once
+ server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ select {
+ case requestReady <- struct{}{}:
+ case <-r.Context().Done():
+ return
+ }
+ select {
+ case code := <-responses:
+ w.WriteHeader(code)
+ case <-r.Context().Done():
+ cancelOnce.Do(func() { close(requestCanceled) })
+ }
+ }))
+ t.Cleanup(server.Close)
+
+ events := make(chan string, 3)
+ monitor := NewMonitor(
+ context.Background(),
+ v1.HealthCheckConfig{
+ Type: "http",
+ Path: "/health",
+ TimeoutSeconds: 5,
+ MaxFailed: 3,
+ },
+ strings.TrimPrefix(server.URL, "http://"),
+ func() { events <- "normal" },
+ func() { events <- "failed" },
+ )
+ intervals := observeWorkerIntervals(monitor)
+ t.Cleanup(func() { stopMonitorWorker(t, monitor) })
+
+ awaitRequest := func() {
+ t.Helper()
+ select {
+ case <-requestReady:
+ case <-time.After(time.Second):
+ t.Fatal("health check request did not start")
+ }
+ }
+ awaitEvent := func(want string) {
+ t.Helper()
+ select {
+ case got := <-events:
+ require.Equal(t, want, got)
+ case <-time.After(time.Second):
+ t.Fatalf("health check callback %q was not called", want)
+ }
+ }
+ checkResult := func(code int, want workerStatus, event string) {
+ t.Helper()
+ awaitRequest()
+ select {
+ case responses <- code:
+ case <-time.After(time.Second):
+ t.Fatal("health check handler did not accept the response")
+ }
+ timer := awaitWorkerInterval(t, intervals, want)
+ if event != "" {
+ awaitEvent(event)
+ }
+ // The worker has finished this result and cannot emit another
+ // callback until the test releases the next check.
+ select {
+ case got := <-events:
+ t.Fatalf("unexpected health check callback %q", got)
+ default:
+ }
+ timer <- time.Now()
+ }
+
+ monitor.Start()
+ checkResult(http.StatusOK, workerStatus{statusOK: true}, "normal")
+ checkResult(http.StatusServiceUnavailable, workerStatus{failedTimes: 1, statusOK: true}, "")
+ checkResult(http.StatusServiceUnavailable, workerStatus{failedTimes: 2, statusOK: true}, "")
+ checkResult(http.StatusOK, workerStatus{statusOK: true}, "")
+ checkResult(http.StatusServiceUnavailable, workerStatus{failedTimes: 1, statusOK: true}, "")
+ checkResult(http.StatusServiceUnavailable, workerStatus{failedTimes: 2, statusOK: true}, "")
+ checkResult(http.StatusServiceUnavailable, workerStatus{failedTimes: 3}, "failed")
+ checkResult(http.StatusOK, workerStatus{statusOK: true}, "normal")
+
+ awaitRequest()
+ stopMonitorWorker(t, monitor)
+ select {
+ case <-requestCanceled:
+ case <-time.After(time.Second):
+ t.Fatal("health check request context was not canceled")
+ }
+}
+
+func TestMonitorTCPWorkerProcessesResults(t *testing.T) {
+ initialAccepted := make(chan struct{}, 1)
+ initialBackend := newTCPHealthBackend(t, "127.0.0.1:0", initialAccepted)
+ addr := initialBackend.listener.Addr().String()
+ recoveryAccepted := make(chan struct{}, 1)
+
+ normalCallbacks := make(chan workerStatus, 2)
+ failedCallbacks := make(chan workerStatus, 2)
+ var monitor *Monitor
+ monitor = NewMonitor(
+ context.Background(),
+ v1.HealthCheckConfig{
+ Type: "tcp",
+ TimeoutSeconds: 1,
+ MaxFailed: 3,
+ },
+ addr,
+ func() {
+ normalCallbacks <- workerStatus{failedTimes: monitor.failedTimes, statusOK: monitor.statusOK}
+ },
+ func() {
+ failedCallbacks <- workerStatus{failedTimes: monitor.failedTimes, statusOK: monitor.statusOK}
+ },
+ )
+ intervals := observeWorkerIntervals(monitor)
+ recoveryBackend := (*tcpHealthBackend)(nil)
+ t.Cleanup(func() {
+ stopMonitorWorker(t, monitor)
+ initialBackend.Close()
+ if recoveryBackend != nil {
+ recoveryBackend.Close()
+ }
+ })
+
+ awaitStatus := func(ch <-chan workerStatus, want workerStatus, message string) {
+ t.Helper()
+ select {
+ case got := <-ch:
+ require.Equal(t, want, got)
+ case <-time.After(time.Second):
+ t.Fatal(message)
+ }
+ }
+
+ monitor.Start()
+ select {
+ case <-initialAccepted:
+ case <-time.After(time.Second):
+ t.Fatal("TCP health check did not reach the initial backend")
+ }
+ awaitStatus(normalCallbacks, workerStatus{failedTimes: 0, statusOK: true}, "TCP worker did not report the initial success")
+
+ initialTimer := awaitWorkerInterval(t, intervals, workerStatus{statusOK: true})
+ initialBackend.Close()
+ initialTimer <- time.Now()
+
+ firstFailureTimer := awaitWorkerInterval(t, intervals, workerStatus{failedTimes: 1, statusOK: true})
+ firstFailureTimer <- time.Now()
+
+ secondFailureTimer := awaitWorkerInterval(t, intervals, workerStatus{failedTimes: 2, statusOK: true})
+ secondFailureTimer <- time.Now()
+
+ awaitStatus(failedCallbacks, workerStatus{failedTimes: 3, statusOK: false}, "TCP worker did not report the third failed health check")
+ thirdFailureTimer := awaitWorkerInterval(t, intervals, workerStatus{failedTimes: 3})
+
+ recoveryBackend = newTCPHealthBackend(t, addr, recoveryAccepted)
+ thirdFailureTimer <- time.Now()
+ select {
+ case <-recoveryAccepted:
+ case <-time.After(time.Second):
+ t.Fatal("TCP health check did not reach the recovery backend")
+ }
+ awaitStatus(normalCallbacks, workerStatus{failedTimes: 0, statusOK: true}, "TCP worker did not report recovery")
+
+ recoveryTimer := awaitWorkerInterval(t, intervals, workerStatus{statusOK: true})
+ recoveryBackend.Close()
+ recoveryTimer <- time.Now()
+
+ firstRecoveryFailureTimer := awaitWorkerInterval(t, intervals, workerStatus{failedTimes: 1, statusOK: true})
+ firstRecoveryFailureTimer <- time.Now()
+
+ secondRecoveryFailureTimer := awaitWorkerInterval(t, intervals, workerStatus{failedTimes: 2, statusOK: true})
+ secondRecoveryFailureTimer <- time.Now()
+
+ awaitStatus(failedCallbacks, workerStatus{failedTimes: 3, statusOK: false}, "TCP worker did not report the third failed health check after recovery")
+ stopMonitorWorker(t, monitor)
+}
+
+func TestMonitorMaxFailedOne(t *testing.T) {
+ var events []string
+ monitor := NewMonitor(
+ context.Background(),
+ v1.HealthCheckConfig{Type: "tcp", MaxFailed: 1},
+ "",
+ func() { events = append(events, "normal") },
+ func() { events = append(events, "failed") },
+ )
+ t.Cleanup(monitor.Stop)
+
+ checkErr := errors.New("health check failed")
+ monitor.handleCheckResult(nil)
+ monitor.handleCheckResult(checkErr)
+ require.Equal(t, []string{"normal", "failed"}, events)
+ require.False(t, monitor.statusOK)
+ require.Equal(t, uint64(1), monitor.failedTimes)
+
+ monitor.handleCheckResult(checkErr)
+ require.Equal(t, []string{"normal", "failed"}, events)
+
+ monitor.handleCheckResult(nil)
+ monitor.handleCheckResult(checkErr)
+ require.Equal(t, []string{"normal", "failed", "normal", "failed"}, events)
+}
+
+func TestMonitorStopCancelsWork(t *testing.T) {
+ t.Run("in-flight check", func(t *testing.T) {
+ requestStarted := make(chan struct{})
+ requestCanceled := make(chan struct{})
+ server := httptest.NewServer(http.HandlerFunc(func(_ http.ResponseWriter, r *http.Request) {
+ close(requestStarted)
+ <-r.Context().Done()
+ close(requestCanceled)
+ }))
+ t.Cleanup(server.Close)
+
+ monitor := NewMonitor(
+ context.Background(),
+ v1.HealthCheckConfig{Type: "http", Path: "/health"},
+ strings.TrimPrefix(server.URL, "http://"),
+ func() {},
+ func() {},
+ )
+ t.Cleanup(func() { stopMonitorWorker(t, monitor) })
+ monitor.Start()
+
+ select {
+ case <-requestStarted:
+ case <-time.After(time.Second):
+ t.Fatal("health check request did not start")
+ }
+ stopMonitorWorker(t, monitor)
+ select {
+ case <-requestCanceled:
+ case <-time.After(time.Second):
+ t.Fatal("health check request context was not canceled")
+ }
+ })
+
+ t.Run("interval wait", func(t *testing.T) {
+ monitor := NewMonitor(context.Background(), v1.HealthCheckConfig{Type: "tcp"}, "", nil, nil)
+ intervals := observeWorkerIntervals(monitor)
+ t.Cleanup(func() { stopMonitorWorker(t, monitor) })
+ monitor.Start()
+
+ // A real worker has completed its first check and installed a timer
+ // that will never fire. Stop must release the wait and exit the loop.
+ awaitWorkerInterval(t, intervals, workerStatus{})
+ stopMonitorWorker(t, monitor)
+ })
+}
+
+type gatedErrContext struct {
+ context.Context
+ errCalled chan struct{}
+ releaseErr chan struct{}
+}
+
+func (ctx *gatedErrContext) Err() error {
+ select {
+ case ctx.errCalled <- struct{}{}:
+ default:
+ }
+ <-ctx.releaseErr
+ return ctx.Context.Err()
+}
+
+func TestMonitorTimerCancellationAfterTimerFires(t *testing.T) {
+ baseCtx, cancel := context.WithCancel(context.Background())
+ ctx := &gatedErrContext{
+ Context: baseCtx,
+ errCalled: make(chan struct{}, 1),
+ releaseErr: make(chan struct{}),
+ }
+ monitor := NewMonitor(
+ context.Background(),
+ v1.HealthCheckConfig{Type: "tcp"},
+ "",
+ nil,
+ nil,
+ )
+ monitor.ctx = ctx
+ monitor.cancel = cancel
+
+ timerReady := make(chan time.Time, 1)
+ timerReady <- time.Now()
+ var timerStopped atomic.Bool
+ monitor.timerFactory = func(time.Duration) (<-chan time.Time, func()) {
+ return timerReady, func() { timerStopped.Store(true) }
+ }
+
+ waitResult := make(chan bool, 1)
+ go func() {
+ waitResult <- monitor.waitForNextCheck()
+ }()
+
+ // ctx.Done is not ready when select runs, so receiving timerReady is the
+ // only possible branch. Err signals after that receive and blocks until the
+ // test cancels the context, deterministically exercising the cancellation
+ // re-check in the timer branch.
+ select {
+ case <-ctx.errCalled:
+ case <-time.After(time.Second):
+ t.Fatal("timer branch did not re-check the monitor context")
+ }
+ cancel()
+ close(ctx.releaseErr)
+
+ select {
+ case shouldContinue := <-waitResult:
+ require.False(t, shouldContinue)
+ case <-time.After(time.Second):
+ t.Fatal("timer wait did not observe context cancellation")
+ }
+ require.True(t, timerStopped.Load())
+}
diff --git a/client/proxy/proxy_wrapper.go b/client/proxy/proxy_wrapper.go
index 37a71277..30ea5e33 100644
--- a/client/proxy/proxy_wrapper.go
+++ b/client/proxy/proxy_wrapper.go
@@ -267,8 +267,9 @@ func (pw *Wrapper) InWorkConn(workConn net.Conn, m *msg.StartWorkConn) {
xl := pw.xl
pw.mu.RLock()
pxy := pw.pxy
+ running := pw.Phase == ProxyPhaseRunning
pw.mu.RUnlock()
- if pxy != nil && pw.Phase == ProxyPhaseRunning {
+ if pxy != nil && running {
xl.Debugf("start a new work connection, localAddr: %s remoteAddr: %s", workConn.LocalAddr().String(), workConn.RemoteAddr().String())
go pxy.InWorkConn(workConn, m)
} else {
diff --git a/client/proxy/proxy_wrapper_test.go b/client/proxy/proxy_wrapper_test.go
new file mode 100644
index 00000000..639f3cf1
--- /dev/null
+++ b/client/proxy/proxy_wrapper_test.go
@@ -0,0 +1,76 @@
+//go:build !frps
+
+package proxy
+
+import (
+ "io"
+ "net"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/require"
+
+ "github.com/fatedier/frp/pkg/msg"
+ "github.com/fatedier/frp/pkg/util/xlog"
+)
+
+type closingWorkProxy struct {
+ Proxy
+ accepted atomic.Int64
+}
+
+func (p *closingWorkProxy) InWorkConn(conn net.Conn, _ *msg.StartWorkConn) {
+ p.accepted.Add(1)
+ _ = conn.Close()
+}
+
+func assertWrapperWorkConnClosed(t *testing.T, pw *Wrapper) {
+ t.Helper()
+ conn, peer := net.Pipe()
+ defer conn.Close()
+ defer peer.Close()
+ require.NoError(t, peer.SetReadDeadline(time.Now().Add(time.Second)))
+ pw.InWorkConn(conn, &msg.StartWorkConn{})
+ _, err := peer.Read(make([]byte, 1))
+ require.ErrorIs(t, err, io.EOF)
+}
+
+func TestWrapperInWorkConnPhaseSnapshot(t *testing.T) {
+ pxy := &closingWorkProxy{}
+ pw := &Wrapper{pxy: pxy, xl: xlog.New()}
+ for _, phase := range []string{ProxyPhaseNew, ProxyPhaseWaitStart, ProxyPhaseStartErr, ProxyPhaseCheckFailed, ProxyPhaseClosed} {
+ pw.Phase = phase
+ assertWrapperWorkConnClosed(t, pw)
+ }
+ require.Zero(t, pxy.accepted.Load(), "inactive proxy received a work connection")
+ pw.Phase = ProxyPhaseRunning
+ pw.pxy = nil
+ assertWrapperWorkConnClosed(t, pw)
+ pw.pxy = pxy
+ assertWrapperWorkConnClosed(t, pw)
+ require.Equal(t, int64(1), pxy.accepted.Load())
+
+ start, done := make(chan struct{}), make(chan struct{})
+ go func() {
+ <-start
+ for range 20000 {
+ pw.mu.Lock()
+ pw.Phase = ProxyPhaseRunning
+ pw.mu.Unlock()
+ pw.mu.Lock()
+ pw.Phase = ProxyPhaseClosed
+ pw.mu.Unlock()
+ }
+ close(done)
+ }()
+ close(start)
+ for range 10000 {
+ assertWrapperWorkConnClosed(t, pw)
+ }
+ select {
+ case <-done:
+ case <-time.After(3 * time.Second):
+ t.Fatal("phase writer did not finish")
+ }
+}
diff --git a/client/proxy/xtcp.go b/client/proxy/xtcp.go
index aef66780..fcd1487f 100644
--- a/client/proxy/xtcp.go
+++ b/client/proxy/xtcp.go
@@ -22,7 +22,7 @@ import (
"reflect"
"time"
- fmux "github.com/hashicorp/yamux"
+ fmux "github.com/fatedier/yamux"
"github.com/quic-go/quic-go"
v1 "github.com/fatedier/frp/pkg/config/v1"
diff --git a/client/service.go b/client/service.go
index 7964050f..055de40d 100644
--- a/client/service.go
+++ b/client/service.go
@@ -155,6 +155,9 @@ type Service struct {
ctx context.Context
// call cancel to stop service
cancel context.CancelCauseFunc
+ // Protect cancel publication and shutdown before the first Run.
+ lifecycleMu sync.Mutex
+ closed bool
connectorCreator func(context.Context, *v1.ClientCommonConfig) Connector
handleWorkConnCb func(*v1.ProxyBaseConfig, net.Conn, *msg.StartWorkConn) bool
@@ -225,8 +228,20 @@ func NewService(options ServiceOptions) (*Service, error) {
func (svr *Service) Run(ctx context.Context) error {
ctx, cancel := context.WithCancelCause(ctx)
+ svr.lifecycleMu.Lock()
+ if svr.closed {
+ svr.lifecycleMu.Unlock()
+ cancel(nil)
+ return ctx.Err()
+ }
svr.ctx = xlog.NewContext(ctx, xlog.FromContextSafe(ctx))
svr.cancel = cancel
+ svr.lifecycleMu.Unlock()
+ defer cancel(nil)
+ defer svr.stop()
+ if err := ctx.Err(); err != nil {
+ return err
+ }
// set custom DNSServer
if svr.common.DNSServer != "" {
@@ -237,7 +252,6 @@ func (svr *Service) Run(ctx context.Context) error {
vnetController := svr.vnetController
if err := svr.vnetController.Init(); err != nil {
log.Errorf("init virtual network controller error: %v", err)
- svr.stop()
return err
}
go func() {
@@ -263,19 +277,25 @@ func (svr *Service) Run(ctx context.Context) error {
if svr.ctl == nil {
cancelCause := cancelErr{}
_ = errors.As(context.Cause(svr.ctx), &cancelCause)
- svr.stop()
return fmt.Errorf("login to the server failed: %v. With loginFailExit enabled, no additional retries will be attempted", cancelCause.Err)
}
go svr.keepControllerWorking()
<-svr.ctx.Done()
- svr.stop()
return nil
}
func (svr *Service) keepControllerWorking() {
- <-svr.ctl.Done()
+ // ctl is swapped under ctlMu in loopLoginUntilSuccess / stop; take a
+ // snapshot before waiting so stop's nil write does not race this read.
+ svr.ctlMu.RLock()
+ ctl := svr.ctl
+ svr.ctlMu.RUnlock()
+ if ctl == nil {
+ return
+ }
+ <-ctl.Done()
// There is a situation where the login is successful but due to certain reasons,
// the control immediately exits. It is necessary to limit the frequency of reconnection in this case.
@@ -285,8 +305,11 @@ func (svr *Service) keepControllerWorking() {
// loopLoginUntilSuccess is another layer of loop that will continuously attempt to
// login to the server until successful.
svr.loopLoginUntilSuccess(20*time.Second, false)
- if svr.ctl != nil {
- <-svr.ctl.Done()
+ svr.ctlMu.RLock()
+ ctl := svr.ctl
+ svr.ctlMu.RUnlock()
+ if ctl != nil {
+ <-ctl.Done()
return false, errors.New("control is closed and try another loop")
}
// If the control is nil, it means that the login failed and the service is also closed.
@@ -377,7 +400,7 @@ func (svr *Service) UpdateAllConfigurer(proxyCfgs []v1.ProxyConfigurer, visitorC
svr.ctlMu.RUnlock()
if ctl != nil {
- return svr.ctl.UpdateAllConfigurer(proxyCfgs, visitorCfgs)
+ return ctl.UpdateAllConfigurer(proxyCfgs, visitorCfgs)
}
return nil
}
@@ -417,7 +440,20 @@ func (svr *Service) Close() {
func (svr *Service) GracefulClose(d time.Duration) {
svr.gracefulShutdownDuration.Store(int64(d))
- svr.cancel(nil)
+ svr.lifecycleMu.Lock()
+ if svr.closed {
+ svr.lifecycleMu.Unlock()
+ return
+ }
+ svr.closed = true
+ cancel := svr.cancel
+ svr.lifecycleMu.Unlock()
+ if cancel != nil {
+ // Once Run has claimed startup, it also owns cleanup.
+ cancel(nil)
+ } else {
+ svr.stop()
+ }
}
func (svr *Service) stop() {
diff --git a/client/service_lifecycle_test.go b/client/service_lifecycle_test.go
new file mode 100644
index 00000000..13969506
--- /dev/null
+++ b/client/service_lifecycle_test.go
@@ -0,0 +1,186 @@
+package client
+
+import (
+ "context"
+ "io"
+ "net"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/require"
+
+ "github.com/fatedier/frp/pkg/config/source"
+ v1 "github.com/fatedier/frp/pkg/config/v1"
+ "github.com/fatedier/frp/pkg/config/v1/validation"
+ "github.com/fatedier/frp/pkg/msg"
+)
+
+func newLifecycleTestService(t *testing.T, creator func(context.Context, *v1.ClientCommonConfig) Connector) (*Service, string) {
+ t.Helper()
+ svr, err := NewService(ServiceOptions{
+ Common: &v1.ClientCommonConfig{
+ LoginFailExit: new(false),
+ WebServer: v1.WebServerConfig{Addr: "127.0.0.1", Port: getFreeTCPPort(t)},
+ },
+ ConfigSourceAggregator: source.NewAggregator(source.NewConfigSource()),
+ ConnectorCreator: creator,
+ })
+ require.NoError(t, err)
+ _, err = validation.NewConfigValidator(nil).ValidateClientCommonConfig(svr.common)
+ require.NoError(t, err)
+ address := svr.webServer.Address()
+ assertLifecycleAdminPort(t, address, false)
+ return svr, address
+}
+
+func assertLifecycleAdminPort(t *testing.T, address string, released bool) {
+ t.Helper()
+ ln, err := net.Listen("tcp", address)
+ if ln != nil {
+ _ = ln.Close()
+ }
+ if released {
+ require.NoError(t, err, "admin listener was not released")
+ } else {
+ require.Error(t, err, "admin listener was not bound")
+ }
+}
+
+func waitLifecycleSignal(t *testing.T, done <-chan struct{}) {
+ t.Helper()
+ select {
+ case <-done:
+ case <-time.After(3 * time.Second):
+ t.Fatal("service lifecycle operation did not finish")
+ }
+}
+
+func TestServiceRunAfterEarlyShutdown(t *testing.T) {
+ for _, mode := range []string{"close before Run", "parent already cancelled"} {
+ t.Run(mode, func(t *testing.T) {
+ var attempts atomic.Int64
+ svr, address := newLifecycleTestService(t, func(context.Context, *v1.ClientCommonConfig) Connector {
+ attempts.Add(1)
+ return &failingConnector{err: net.ErrClosed}
+ })
+ // Detect startup's global DNS side effect without making DNS requests.
+ resolver := net.DefaultResolver
+ t.Cleanup(func() { net.DefaultResolver = resolver })
+ svr.common.DNSServer = "127.0.0.1:53"
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ if mode == "close before Run" {
+ svr.Close()
+ assertLifecycleAdminPort(t, address, true)
+ } else {
+ cancel()
+ }
+ done := make(chan struct{})
+ var runErr error
+ go func() {
+ runErr = svr.Run(ctx)
+ close(done)
+ }()
+ waitLifecycleSignal(t, done)
+ require.ErrorIs(t, runErr, context.Canceled)
+ require.Zero(t, attempts.Load(), "closed service attempted login")
+ require.Same(t, resolver, net.DefaultResolver, "closed service changed DNS")
+ assertLifecycleAdminPort(t, address, true)
+ })
+ }
+}
+
+func TestServiceStartupShutdownLeavesCleanupToRun(t *testing.T) {
+ for _, mode := range []string{"Close", "parent cancellation"} {
+ t.Run(mode, func(t *testing.T) {
+ entered, resume := make(chan struct{}), make(chan struct{})
+ svr, address := newLifecycleTestService(t, func(ctx context.Context, _ *v1.ClientCommonConfig) Connector {
+ close(entered)
+ <-resume
+ return &failingConnector{err: ctx.Err()}
+ })
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ defer close(resume)
+ runDone := make(chan struct{})
+ go func() { _ = svr.Run(ctx); close(runDone) }()
+ waitLifecycleSignal(t, entered)
+ closeDone := make(chan struct{})
+ go func() {
+ if mode == "Close" {
+ svr.Close()
+ } else {
+ cancel()
+ }
+ close(closeDone)
+ }()
+ waitLifecycleSignal(t, closeDone)
+ select {
+ case <-runDone:
+ t.Fatal("Run exited while startup still owned its resources")
+ default:
+ }
+ assertLifecycleAdminPort(t, address, false)
+ resume <- struct{}{}
+ waitLifecycleSignal(t, runDone)
+ assertLifecycleAdminPort(t, address, true)
+ require.ErrorIs(t, svr.ctx.Err(), context.Canceled)
+ })
+ }
+}
+
+func TestServiceShutdownAfterLogin(t *testing.T) {
+ for _, mode := range []string{"Close", "parent cancellation"} {
+ t.Run(mode, func(t *testing.T) {
+ clientConn, serverConn := net.Pipe()
+ defer clientConn.Close()
+ defer serverConn.Close()
+ connector := &testConnector{conn: clientConn}
+ svr, address := newLifecycleTestService(t, func(context.Context, *v1.ClientCommonConfig) Connector {
+ return connector
+ })
+ defer svr.Close()
+ peerDone := make(chan error, 1)
+ go func() {
+ rw := msg.NewV1ReadWriter(serverConn)
+ var login msg.Login
+ if err := rw.ReadMsgInto(&login); err != nil {
+ peerDone <- err
+ return
+ }
+ if err := rw.WriteMsg(&msg.LoginResp{RunID: "lifecycle-test"}); err != nil {
+ peerDone <- err
+ return
+ }
+ _, err := io.Copy(io.Discard, serverConn)
+ peerDone <- err
+ }()
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ runDone := make(chan struct{})
+ var runErr error
+ go func() { runErr = svr.Run(ctx); close(runDone) }()
+ require.Eventually(t, func() bool {
+ svr.ctlMu.RLock()
+ defer svr.ctlMu.RUnlock()
+ return svr.ctl != nil
+ }, 3*time.Second, time.Millisecond, "login did not establish control")
+ if mode == "Close" {
+ svr.Close()
+ } else {
+ cancel()
+ }
+ waitLifecycleSignal(t, runDone)
+ require.NoError(t, runErr)
+ require.True(t, connector.closed.Load())
+ select {
+ case err := <-peerDone:
+ require.NoError(t, err)
+ case <-time.After(3 * time.Second):
+ t.Fatal("control connection was not released")
+ }
+ assertLifecycleAdminPort(t, address, true)
+ })
+ }
+}
diff --git a/client/service_test.go b/client/service_test.go
index 29f141a1..cae26da3 100644
--- a/client/service_test.go
+++ b/client/service_test.go
@@ -9,8 +9,6 @@ import (
"strings"
"testing"
- "github.com/samber/lo"
-
"github.com/fatedier/frp/pkg/config/source"
v1 "github.com/fatedier/frp/pkg/config/v1"
)
@@ -49,7 +47,7 @@ func TestRunStopsStartedComponentsOnInitialLoginFailure(t *testing.T) {
svr, err := NewService(ServiceOptions{
Common: &v1.ClientCommonConfig{
- LoginFailExit: lo.ToPtr(true),
+ LoginFailExit: new(true),
WebServer: v1.WebServerConfig{
Addr: "127.0.0.1",
Port: port,
diff --git a/client/visitor/xtcp.go b/client/visitor/xtcp.go
index e7a60895..114f47be 100644
--- a/client/visitor/xtcp.go
+++ b/client/visitor/xtcp.go
@@ -25,7 +25,7 @@ import (
"time"
libio "github.com/fatedier/golib/io"
- fmux "github.com/hashicorp/yamux"
+ fmux "github.com/fatedier/yamux"
quic "github.com/quic-go/quic-go"
"golang.org/x/time/rate"
diff --git a/doc/pic/sponsor_rapidproxy.png b/doc/pic/sponsor_rapidproxy.png
new file mode 100644
index 00000000..11ec7c3a
Binary files /dev/null and b/doc/pic/sponsor_rapidproxy.png differ
diff --git a/dockerfiles/Dockerfile-for-frpc b/dockerfiles/Dockerfile-for-frpc
index 6760bb16..e83b10f0 100644
--- a/dockerfiles/Dockerfile-for-frpc
+++ b/dockerfiles/Dockerfile-for-frpc
@@ -8,7 +8,7 @@ RUN npm install
WORKDIR /web/frpc
RUN npm run build
-FROM golang:1.25 AS building
+FROM golang:1.26 AS building
COPY . /building
COPY --from=web-builder /web/frpc/dist /building/web/frpc/dist
diff --git a/dockerfiles/Dockerfile-for-frps b/dockerfiles/Dockerfile-for-frps
index 8ef7dff4..c13004d3 100644
--- a/dockerfiles/Dockerfile-for-frps
+++ b/dockerfiles/Dockerfile-for-frps
@@ -8,7 +8,7 @@ RUN npm install
WORKDIR /web/frps
RUN npm run build
-FROM golang:1.25 AS building
+FROM golang:1.26 AS building
COPY . /building
COPY --from=web-builder /web/frps/dist /building/web/frps/dist
diff --git a/go.mod b/go.mod
index 9e02c333..838ec8d1 100644
--- a/go.mod
+++ b/go.mod
@@ -1,15 +1,15 @@
module github.com/fatedier/frp
-go 1.25.0
+go 1.26.0
require (
github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5
github.com/coreos/go-oidc/v3 v3.18.0
github.com/fatedier/golib v0.8.2
+ github.com/fatedier/yamux v0.2.0
github.com/google/uuid v1.6.0
github.com/gorilla/mux v1.8.1
github.com/gorilla/websocket v1.5.0
- github.com/hashicorp/yamux v0.1.1
github.com/onsi/ginkgo/v2 v2.23.4
github.com/onsi/gomega v1.36.3
github.com/pelletier/go-toml/v2 v2.2.0
@@ -25,8 +25,8 @@ require (
github.com/tidwall/gjson v1.17.1
github.com/vishvananda/netlink v1.3.0
github.com/xtaci/kcp-go/v5 v5.6.13
- golang.org/x/crypto v0.54.0
- golang.org/x/net v0.56.0
+ golang.org/x/crypto v0.56.0
+ golang.org/x/net v0.57.0
golang.org/x/oauth2 v0.36.0
golang.org/x/sync v0.22.0
golang.org/x/sys v0.47.0
@@ -39,7 +39,7 @@ require (
)
require (
- github.com/Azure/go-ntlmssp v0.1.0 // indirect
+ github.com/Azure/go-ntlmssp v0.1.1 // indirect
github.com/beorn7/perks v1.0.1 // indirect
github.com/cespare/xxhash/v2 v2.2.0 // indirect
github.com/davecgh/go-spew v1.1.1 // indirect
@@ -64,8 +64,8 @@ require (
github.com/tjfoc/gmsm v1.4.1 // indirect
github.com/vishvananda/netns v0.0.4 // indirect
go.uber.org/automaxprocs v1.6.0 // indirect
- golang.org/x/text v0.40.0 // indirect
- golang.org/x/tools v0.47.0 // indirect
+ golang.org/x/text v0.41.0 // indirect
+ golang.org/x/tools v0.48.0 // indirect
golang.zx2c4.com/wintun v0.0.0-20230126152724-0fa3db229ce2 // indirect
google.golang.org/protobuf v1.36.5 // indirect
gopkg.in/yaml.v2 v2.4.0 // indirect
@@ -73,6 +73,3 @@ require (
sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd // indirect
sigs.k8s.io/yaml v1.3.0 // indirect
)
-
-// TODO(fatedier): Temporary use the modified version, update to the official version after merging into the official repository.
-replace github.com/hashicorp/yamux => github.com/fatedier/yamux v0.0.0-20250825093530-d0154be01cd6
diff --git a/go.sum b/go.sum
index 30641d78..e5b5ba7e 100644
--- a/go.sum
+++ b/go.sum
@@ -1,6 +1,6 @@
cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
-github.com/Azure/go-ntlmssp v0.1.0 h1:DjFo6YtWzNqNvQdrwEyr/e4nhU3vRiwenz5QX7sFz+A=
-github.com/Azure/go-ntlmssp v0.1.0/go.mod h1:NYqdhxd/8aAct/s4qSYZEerdPuH1liG2/X9DiVTbhpk=
+github.com/Azure/go-ntlmssp v0.1.1 h1:l+FM/EEMb0U9QZE7mKNEDw5Mu3mFiaa2GKOoTSsNDPw=
+github.com/Azure/go-ntlmssp v0.1.1/go.mod h1:NYqdhxd/8aAct/s4qSYZEerdPuH1liG2/X9DiVTbhpk=
github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5 h1:0CwZNZbxp69SHPdPJAN/hZIm0C4OItdklCFmMRWYpio=
github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5/go.mod h1:wHh0iHkYZB8zMSxRWpUBQtwG5a7fFgvEO+odwuTv2gs=
@@ -22,8 +22,8 @@ github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1m
github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
github.com/fatedier/golib v0.8.2 h1:02n2Dg7KJ7rR7p7n4/6hBUjaLQf2J7EiHYZQsgGTvww=
github.com/fatedier/golib v0.8.2/go.mod h1:ArUGvPg2cOw/py2RAuBt46nNZH2VQ5Z70p109MAZpJw=
-github.com/fatedier/yamux v0.0.0-20250825093530-d0154be01cd6 h1:u92UUy6FURPmNsMBUuongRWC0rBqN6gd01Dzu+D21NE=
-github.com/fatedier/yamux v0.0.0-20250825093530-d0154be01cd6/go.mod h1:c5/tk6G0dSpXGzJN7Wk1OEie8grdSJAmeawId9Zvd34=
+github.com/fatedier/yamux v0.2.0 h1:H+2A9iBVh7aJlEOc1Ws1FXWOaecBf2nRv9zpFMPUWg8=
+github.com/fatedier/yamux v0.2.0/go.mod h1:d4FtRDrC9sHvRpiDL6J5EnfjLhqzZZplHe5yToSn2Ac=
github.com/go-jose/go-jose/v4 v4.1.4 h1:moDMcTHmvE6Groj34emNPLs/qtYXRVcd6S7NHbHz3kA=
github.com/go-jose/go-jose/v4 v4.1.4/go.mod h1:x4oUasVrzR7071A4TnHLGSPpNOm2a21K9Kf04k1rs08=
github.com/go-logr/logr v1.4.2 h1:6pFjapn8bFcIbiKo3XT4j/BhANplGihG6tvd+8rYgrY=
@@ -151,8 +151,8 @@ go.uber.org/mock v0.5.2/go.mod h1:wLlUxC2vVTPTaE3UD51E0BGOAElKrILxhVSDYQLld5o=
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
golang.org/x/crypto v0.0.0-20201012173705-84dcc777aaee/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
-golang.org/x/crypto v0.54.0 h1:YLIA59K4fiNzHzjnZt2tUJQjQtUWfWbeHBqKtk3eScw=
-golang.org/x/crypto v0.54.0/go.mod h1:KWL8ny2AZdGR2cWmzeHrp2azQPGogOv+HeQaVEXC2dk=
+golang.org/x/crypto v0.56.0 h1:GUh5Ii4J5jtcseSMiRqr1jXCNHoxjeV9Fmekc2oLy6Y=
+golang.org/x/crypto v0.56.0/go.mod h1:OMW5y6CY9l38uPLmxU6l6pwcXp1obtLo3e6gT7gQR2I=
golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=
golang.org/x/lint v0.0.0-20190227174305-5b3e6a55c961/go.mod h1:wehouNa3lNwaWXcvxsM5YxQ5yQlVC4a0KAMCusXpPoU=
@@ -163,8 +163,8 @@ golang.org/x/net v0.0.0-20190213061140-3a22650c66bd/go.mod h1:mL1N/T3taQHkDXs73r
golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
golang.org/x/net v0.0.0-20201010224723-4f7140c49acb/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
-golang.org/x/net v0.56.0 h1:Rw8j/hFzGvJUZwNBXnAtf5sVDVt+65SK2C7IxCxZt5o=
-golang.org/x/net v0.56.0/go.mod h1:D3Ku6r+V6JROoZK144D2XfMHFcMq/0zSfLelVTCFKec=
+golang.org/x/net v0.57.0 h1:K5+3DljvIuDG9/Jv9rvyMywYNFCQ9RSUY6OOTTkT+tE=
+golang.org/x/net v0.57.0/go.mod h1:KpXc8iv+r3XplLAG/f7Jsf9RPszJzdR0f58q9vGOuEU=
golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
golang.org/x/oauth2 v0.36.0 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs=
golang.org/x/oauth2 v0.36.0/go.mod h1:YDBUJMTkDnJS+A4BP4eZBjCqtokkg1hODuPjwiGPO7Q=
@@ -186,8 +186,8 @@ golang.org/x/term v0.45.0 h1:NwWyBmoJCbfTHpxrWoZ9C6/VxOf7ic219I8xZZFdrf0=
golang.org/x/term v0.45.0/go.mod h1:9aqxs0blBcrm/n0L9QW0aRVD+ktan8ssZromtqJC43w=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
-golang.org/x/text v0.40.0 h1:Ub2Z6/xjgF1WrYQz2nuITOEegKFtiIy+rieRJ5lHZKs=
-golang.org/x/text v0.40.0/go.mod h1:hpnzDAfGV753zIKo+wk3u1bVKCGPbrnF7+7LBF/UHVY=
+golang.org/x/text v0.41.0 h1:vz/seA0lnX87Othu2f/0L24RcgrXD9/YFTSuGjj3rH8=
+golang.org/x/text v0.41.0/go.mod h1:jvf1O8ajNzZqhSrQBPbutR/EB83Cc0CFrezNQIwbb5M=
golang.org/x/time v0.10.0 h1:3usCWA8tQn0L8+hFJQNgzpWbd89begxN66o1Ojdn5L4=
golang.org/x/time v0.10.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM=
golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
@@ -195,8 +195,8 @@ golang.org/x/tools v0.0.0-20190114222345-bf090417da8b/go.mod h1:n7NCudcB/nEzxVGm
golang.org/x/tools v0.0.0-20190226205152-f727befe758c/go.mod h1:9Yl7xja0Znq3iFh3HoIrodX9oNMXvdceNzlUR8zjMvY=
golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
golang.org/x/tools v0.0.0-20190524140312-2c0ae7006135/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q=
-golang.org/x/tools v0.47.0 h1:7Kn5x/d1svx/PzryTsqeoZN4TZwqeH5pGWjefhLi/1Q=
-golang.org/x/tools v0.47.0/go.mod h1:dFHnyTvFWY212G+h7ZY4Vsp/K3U4/7W9TyVaAul8uCA=
+golang.org/x/tools v0.48.0 h1:3+hClM1aLL5mjMKm5ovokw9epgRXPuu2tILgismM6RE=
+golang.org/x/tools v0.48.0/go.mod h1:08xX0orndb/F7jJxGDicx061tyd5pcMto75YMAXr6lk=
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
golang.zx2c4.com/wintun v0.0.0-20230126152724-0fa3db229ce2 h1:B82qJJgjvYKsXS9jeunTOisW56dUokqW/FOteYJJ/yg=
golang.zx2c4.com/wintun v0.0.0-20230126152724-0fa3db229ce2/go.mod h1:deeaetjYA+DHMHg+sMSMI58GrEteJUUzzw7en6TJQcI=
diff --git a/pkg/config/legacy/conversion.go b/pkg/config/legacy/conversion.go
index ec8d0698..f394fea4 100644
--- a/pkg/config/legacy/conversion.go
+++ b/pkg/config/legacy/conversion.go
@@ -17,8 +17,6 @@ package legacy
import (
"strings"
- "github.com/samber/lo"
-
"github.com/fatedier/frp/pkg/config/types"
v1 "github.com/fatedier/frp/pkg/config/v1"
)
@@ -49,7 +47,7 @@ func Convert_ClientCommonConf_To_v1(conf *ClientCommonConf) *v1.ClientCommonConf
out.Transport.ConnectServerLocalIP = conf.ConnectServerLocalIP
out.Transport.ProxyURL = conf.HTTPProxy
out.Transport.PoolCount = conf.PoolCount
- out.Transport.TCPMux = lo.ToPtr(conf.TCPMux)
+ out.Transport.TCPMux = new(conf.TCPMux)
out.Transport.TCPMuxKeepaliveInterval = conf.TCPMuxKeepaliveInterval
out.Transport.Protocol = conf.Protocol
out.Transport.HeartbeatInterval = conf.HeartbeatInterval
@@ -57,8 +55,8 @@ func Convert_ClientCommonConf_To_v1(conf *ClientCommonConf) *v1.ClientCommonConf
out.Transport.QUIC.KeepalivePeriod = conf.QUICKeepalivePeriod
out.Transport.QUIC.MaxIdleTimeout = conf.QUICMaxIdleTimeout
out.Transport.QUIC.MaxIncomingStreams = conf.QUICMaxIncomingStreams
- out.Transport.TLS.Enable = lo.ToPtr(conf.TLSEnable)
- out.Transport.TLS.DisableCustomTLSFirstByte = lo.ToPtr(conf.DisableCustomTLSFirstByte)
+ out.Transport.TLS.Enable = new(conf.TLSEnable)
+ out.Transport.TLS.DisableCustomTLSFirstByte = new(conf.DisableCustomTLSFirstByte)
out.Transport.TLS.CertFile = conf.TLSCertFile
out.Transport.TLS.KeyFile = conf.TLSKeyFile
out.Transport.TLS.TrustedCaFile = conf.TLSTrustedCaFile
@@ -77,7 +75,7 @@ func Convert_ClientCommonConf_To_v1(conf *ClientCommonConf) *v1.ClientCommonConf
out.WebServer.PprofEnable = conf.PprofEnable
out.DNSServer = conf.DNSServer
- out.LoginFailExit = lo.ToPtr(conf.LoginFailExit)
+ out.LoginFailExit = new(conf.LoginFailExit)
out.Start = conf.Start
out.UDPPacketSize = conf.UDPPacketSize
out.Metadatas = conf.Metas
@@ -134,14 +132,14 @@ func Convert_ServerCommonConf_To_v1(conf *ServerCommonConf) *v1.ServerConfig {
out.Log.MaxDays = conf.LogMaxDays
out.Log.DisablePrintColor = conf.DisableLogColor
- out.DetailedErrorsToClient = lo.ToPtr(conf.DetailedErrorsToClient)
+ out.DetailedErrorsToClient = new(conf.DetailedErrorsToClient)
out.SubDomainHost = conf.SubDomainHost
out.Custom404Page = conf.Custom404Page
out.UserConnTimeout = conf.UserConnTimeout
out.UDPPacketSize = conf.UDPPacketSize
out.NatHoleAnalysisDataReserveHours = conf.NatHoleAnalysisDataReserveHours
- out.Transport.TCPMux = lo.ToPtr(conf.TCPMux)
+ out.Transport.TCPMux = new(conf.TCPMux)
out.Transport.TCPMuxKeepaliveInterval = conf.TCPMuxKeepaliveInterval
out.Transport.TCPKeepAlive = conf.TCPKeepAlive
out.Transport.MaxPoolCount = conf.MaxPoolCount
diff --git a/pkg/config/v1/client.go b/pkg/config/v1/client.go
index 05d02c94..6736030a 100644
--- a/pkg/config/v1/client.go
+++ b/pkg/config/v1/client.go
@@ -85,7 +85,7 @@ type ClientCommonConfig struct {
func (c *ClientCommonConfig) Complete() error {
c.ServerAddr = util.EmptyOr(c.ServerAddr, "0.0.0.0")
c.ServerPort = util.EmptyOr(c.ServerPort, 7000)
- c.LoginFailExit = util.EmptyOr(c.LoginFailExit, lo.ToPtr(true))
+ c.LoginFailExit = util.EmptyOr(c.LoginFailExit, new(true))
c.NatHoleSTUNServer = util.EmptyOr(c.NatHoleSTUNServer, "stun.easyvoip.com:3478")
if err := c.Auth.Complete(); err != nil {
@@ -151,7 +151,7 @@ func (c *ClientTransportConfig) Complete() {
c.DialServerKeepAlive = util.EmptyOr(c.DialServerKeepAlive, 7200)
c.ProxyURL = util.EmptyOr(c.ProxyURL, os.Getenv("http_proxy"))
c.PoolCount = util.EmptyOr(c.PoolCount, 1)
- c.TCPMux = util.EmptyOr(c.TCPMux, lo.ToPtr(true))
+ c.TCPMux = util.EmptyOr(c.TCPMux, new(true))
c.TCPMuxKeepaliveInterval = util.EmptyOr(c.TCPMuxKeepaliveInterval, 30)
if lo.FromPtr(c.TCPMux) {
// If TCPMux is enabled, heartbeat of application layer is unnecessary because we can rely on heartbeat in tcpmux.
@@ -180,8 +180,8 @@ type TLSClientConfig struct {
}
func (c *TLSClientConfig) Complete() {
- c.Enable = util.EmptyOr(c.Enable, lo.ToPtr(true))
- c.DisableCustomTLSFirstByte = util.EmptyOr(c.DisableCustomTLSFirstByte, lo.ToPtr(true))
+ c.Enable = util.EmptyOr(c.Enable, new(true))
+ c.DisableCustomTLSFirstByte = util.EmptyOr(c.DisableCustomTLSFirstByte, new(true))
}
type AuthClientConfig struct {
diff --git a/pkg/config/v1/proxy_plugin.go b/pkg/config/v1/proxy_plugin.go
index 0d6ca3d0..7c63c2f9 100644
--- a/pkg/config/v1/proxy_plugin.go
+++ b/pkg/config/v1/proxy_plugin.go
@@ -17,8 +17,6 @@ package v1
import (
"reflect"
- "github.com/samber/lo"
-
"github.com/fatedier/frp/pkg/util/jsonx"
"github.com/fatedier/frp/pkg/util/util"
)
@@ -125,7 +123,7 @@ type HTTPS2HTTPPluginOptions struct {
}
func (o *HTTPS2HTTPPluginOptions) Complete() {
- o.EnableHTTP2 = util.EmptyOr(o.EnableHTTP2, lo.ToPtr(true))
+ o.EnableHTTP2 = util.EmptyOr(o.EnableHTTP2, new(true))
}
func (o *HTTPS2HTTPPluginOptions) Clone() ClientPluginOptions {
@@ -149,7 +147,7 @@ type HTTPS2HTTPSPluginOptions struct {
}
func (o *HTTPS2HTTPSPluginOptions) Complete() {
- o.EnableHTTP2 = util.EmptyOr(o.EnableHTTP2, lo.ToPtr(true))
+ o.EnableHTTP2 = util.EmptyOr(o.EnableHTTP2, new(true))
}
func (o *HTTPS2HTTPSPluginOptions) Clone() ClientPluginOptions {
diff --git a/pkg/config/v1/server.go b/pkg/config/v1/server.go
index 67c1d84b..6c1c75c4 100644
--- a/pkg/config/v1/server.go
+++ b/pkg/config/v1/server.go
@@ -118,7 +118,7 @@ func (c *ServerConfig) Complete() error {
}
c.VhostHTTPTimeout = util.EmptyOr(c.VhostHTTPTimeout, 60)
- c.DetailedErrorsToClient = util.EmptyOr(c.DetailedErrorsToClient, lo.ToPtr(true))
+ c.DetailedErrorsToClient = util.EmptyOr(c.DetailedErrorsToClient, new(true))
c.UserConnTimeout = util.EmptyOr(c.UserConnTimeout, 10)
c.UDPPacketSize = util.EmptyOr(c.UDPPacketSize, 1500)
c.NatHoleAnalysisDataReserveHours = util.EmptyOr(c.NatHoleAnalysisDataReserveHours, 7*24)
@@ -180,7 +180,7 @@ type ServerTransportConfig struct {
}
func (c *ServerTransportConfig) Complete() {
- c.TCPMux = util.EmptyOr(c.TCPMux, lo.ToPtr(true))
+ c.TCPMux = util.EmptyOr(c.TCPMux, new(true))
c.TCPMuxKeepaliveInterval = util.EmptyOr(c.TCPMuxKeepaliveInterval, 30)
c.TCPKeepAlive = util.EmptyOr(c.TCPKeepAlive, 7200)
c.MaxPoolCount = util.EmptyOr(c.MaxPoolCount, 5)
diff --git a/pkg/msg/handler.go b/pkg/msg/handler.go
index 1f2775c9..b0ac187a 100644
--- a/pkg/msg/handler.go
+++ b/pkg/msg/handler.go
@@ -19,6 +19,8 @@ import (
"io"
"net"
"reflect"
+ "sync"
+ "time"
"github.com/fatedier/frp/pkg/proto/wire"
)
@@ -111,14 +113,22 @@ type Dispatcher struct {
sendCh chan Message
doneCh chan struct{}
+ stopCh chan struct{}
+ stopOnce sync.Once
msgHandlers map[reflect.Type]func(Message)
}
+// NewDispatcher uses SetReadDeadline, when supported, and Close, when rw
+// implements io.Closer, to interrupt ReadMsg after a write failure. Without a
+// working read deadline, Close or the caller must arrange for ReadMsg to return.
+// These methods must return in bounded time and must not wait for Done: Done
+// closes only after ReadMsg, any synchronous handler, and stop have returned.
func NewDispatcher(rw ReadWriter) *Dispatcher {
return &Dispatcher{
rw: rw,
sendCh: make(chan Message, 100),
doneCh: make(chan struct{}),
+ stopCh: make(chan struct{}),
msgHandlers: make(map[reflect.Type]func(Message)),
}
}
@@ -129,37 +139,78 @@ func (d *Dispatcher) Run() {
go d.readLoop()
}
+// stop rejects new sends and interrupts ReadMsg when supported by rw. Done remains
+// owned by readLoop so the control owner cannot clean up while a handler is running.
+func (d *Dispatcher) stop() {
+ d.stopOnce.Do(func() {
+ close(d.stopCh)
+ if conn, ok := d.rw.(interface{ SetReadDeadline(time.Time) error }); ok {
+ _ = conn.SetReadDeadline(time.Now())
+ }
+ if closer, ok := d.rw.(io.Closer); ok {
+ _ = closer.Close()
+ }
+ })
+}
+
func (d *Dispatcher) sendLoop() {
for {
select {
- case <-d.doneCh:
+ case <-d.stopCh:
return
case m := <-d.sendCh:
- _ = d.rw.WriteMsg(m)
+ if err := d.rw.WriteMsg(m); err != nil {
+ d.stop()
+ return
+ }
}
}
}
func (d *Dispatcher) readLoop() {
+ defer close(d.doneCh)
+ defer d.stop()
for {
m, err := d.rw.ReadMsg()
if err != nil {
- close(d.doneCh)
return
}
+ select {
+ case <-d.stopCh:
+ return
+ default:
+ }
+
if handler, ok := d.msgHandlers[reflect.TypeOf(m)]; ok {
handler(m)
}
}
}
+// Send queues m for writing. A nil error does not guarantee delivery.
func (d *Dispatcher) Send(m Message) error {
select {
- case <-d.doneCh:
+ case <-d.stopCh:
+ return io.EOF
+ default:
+ }
+ return d.enqueue(m)
+}
+
+func (d *Dispatcher) enqueue(m Message) error {
+ select {
+ case <-d.stopCh:
return io.EOF
case d.sendCh <- m:
- return nil
+ // Both cases can be ready if shutdown races with enqueueing. Do not
+ // report success when shutdown is already observable after the send.
+ select {
+ case <-d.stopCh:
+ return io.EOF
+ default:
+ return nil
+ }
}
}
diff --git a/pkg/msg/handler_shutdown_test.go b/pkg/msg/handler_shutdown_test.go
new file mode 100644
index 00000000..3f74c06d
--- /dev/null
+++ b/pkg/msg/handler_shutdown_test.go
@@ -0,0 +1,82 @@
+// Copyright 2026 The frp Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package msg
+
+import (
+ "bytes"
+ "errors"
+ "io"
+ "testing"
+)
+
+// Preserve the original exported constructor function type.
+var _ func(ReadWriter) *Dispatcher = NewDispatcher
+
+func TestDispatcherEnqueueRejectsStoppedDispatcher(t *testing.T) {
+ c := newRegressionControl(t)
+ d := NewDispatcher(c)
+ d.stop()
+
+ // Model shutdown after Send's initial check: the stop and buffered send
+ // cases are both ready. A send selected here must still return EOF.
+ if err := d.enqueue(&ReqWorkConn{}); !errors.Is(err, io.EOF) {
+ t.Fatalf("enqueue racing with shutdown: %v, want EOF", err)
+ }
+ // This single call checks that a stopped dispatcher's enqueue returns EOF; it
+ // does not guarantee exercising the send case and post-enqueue check on every run.
+ if len(d.sendCh) > 0 {
+ t.Log("selected the send case; post-enqueue check returned EOF")
+ }
+}
+
+func TestDispatcherAcceptsReadWriterWithoutClose(t *testing.T) {
+ t.Run("codec factory", func(t *testing.T) {
+ d := NewDispatcher(NewV1ReadWriter(&bytes.Buffer{}))
+ d.Run()
+ regressionAwait(t, d.Done(), "non-closable codec EOF")
+ })
+
+ t.Run("write failure", func(t *testing.T) {
+ c := newRegressionControl(t)
+ regressionWarm(t, c)
+ // Hide Close to model an existing caller implementing only ReadWriter.
+ d := NewDispatcher(struct{ ReadWriter }{c})
+ d.Run()
+ t.Cleanup(func() {
+ c.transport.close()
+ regressionAwait(t, d.Done(), "externally interrupted reader")
+ })
+
+ c.transport.arm()
+ if err := d.Send(&ReqWorkConn{}); err != nil && !errors.Is(err, io.EOF) {
+ t.Fatal(err)
+ }
+ if err := regressionReceiveWrite(t, c); !errors.Is(err, errRegressionWriteTimeout) {
+ t.Fatalf("injected write failure: %v", err)
+ }
+ regressionAwait(t, d.stopCh, "write failure stops sends")
+ if err := d.Send(&ReqWorkConn{}); !errors.Is(err, io.EOF) {
+ t.Fatalf("Send after write failure: %v, want EOF", err)
+ }
+ if c.closeCalls.Load() != 0 {
+ t.Fatal("dispatcher closed a connection hidden by ReadWriter")
+ }
+ select {
+ case <-d.Done():
+ t.Fatal("Done closed before the non-closable reader returned")
+ default:
+ }
+ })
+}
diff --git a/pkg/msg/handler_write_error_test.go b/pkg/msg/handler_write_error_test.go
new file mode 100644
index 00000000..17142297
--- /dev/null
+++ b/pkg/msg/handler_write_error_test.go
@@ -0,0 +1,383 @@
+// Copyright 2026 The frp Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package msg
+
+import (
+ "bytes"
+ "errors"
+ "io"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "testing/synctest"
+ "time"
+
+ libcrypto "github.com/fatedier/golib/crypto"
+)
+
+var (
+ errRegressionWriteTimeout = errors.New("injected recoverable connection write timeout")
+ regressionKey = []byte("isolated-public-test-key")
+)
+
+// regressionTransport fails exactly one Write, then accepts later writes. This
+// models an individual mux stream write timing out while its session survives.
+type regressionTransport struct {
+ mu sync.Mutex
+ buffer bytes.Buffer
+ calls int
+ failures int
+ failNext bool
+ closed chan struct{}
+ once sync.Once
+}
+
+func (w *regressionTransport) Write(p []byte) (int, error) {
+ w.mu.Lock()
+ defer w.mu.Unlock()
+ w.calls++
+ if w.failNext {
+ w.failNext = false
+ w.failures++
+ return 0, errRegressionWriteTimeout
+ }
+ select {
+ case <-w.closed:
+ return 0, io.ErrClosedPipe
+ default:
+ }
+ return w.buffer.Write(p)
+}
+
+func (w *regressionTransport) arm() {
+ w.mu.Lock()
+ w.failNext = true
+ w.mu.Unlock()
+}
+
+func (w *regressionTransport) snapshot() (calls, failures, size int) {
+ w.mu.Lock()
+ defer w.mu.Unlock()
+ return w.calls, w.failures, w.buffer.Len()
+}
+
+func (w *regressionTransport) close() {
+ w.once.Do(func() { close(w.closed) })
+}
+
+// regressionControl uses the production FRP message encoder and production
+// golib crypto.Writer. Only the final transport and incoming messages are fake.
+type regressionControl struct {
+ w *libcrypto.Writer
+ transport *regressionTransport
+ writes chan error
+ incoming chan Message
+ beforeRead func() error
+ beforeWrite func()
+ closeCalls atomic.Int32
+}
+
+func newRegressionControl(t *testing.T) *regressionControl {
+ t.Helper()
+ transport := ®ressionTransport{closed: make(chan struct{})}
+ w, err := libcrypto.NewWriter(transport, regressionKey)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return ®ressionControl{
+ w: w, transport: transport, writes: make(chan error, 100), incoming: make(chan Message, 1),
+ }
+}
+
+func (c *regressionControl) ReadMsg() (Message, error) {
+ if c.beforeRead != nil {
+ if err := c.beforeRead(); err != nil {
+ return nil, err
+ }
+ }
+ select {
+ case m := <-c.incoming:
+ return m, nil
+ case <-c.transport.closed:
+ return nil, io.EOF
+ }
+}
+
+func (c *regressionControl) ReadMsgInto(Message) error {
+ _, err := c.ReadMsg()
+ return err
+}
+
+func (c *regressionControl) WriteMsg(m Message) error {
+ if c.beforeWrite != nil {
+ c.beforeWrite()
+ }
+ err := WriteMsg(c.w, m)
+ c.writes <- err
+ return err
+}
+
+func (c *regressionControl) Close() error {
+ c.closeCalls.Add(1)
+ c.transport.close()
+ return nil
+}
+
+func regressionReceiveWrite(t *testing.T, c *regressionControl) error {
+ t.Helper()
+ select {
+ case err := <-c.writes:
+ return err
+ case <-time.After(time.Second):
+ t.Fatal("dispatcher did not attempt WriteMsg")
+ return nil
+ }
+}
+
+func regressionAwait(t *testing.T, ch <-chan struct{}, what string) {
+ t.Helper()
+ select {
+ case <-ch:
+ case <-time.After(time.Second):
+ t.Fatalf("timed out waiting for %s", what)
+ }
+}
+
+func regressionWarm(t *testing.T, c *regressionControl) {
+ t.Helper()
+ if err := c.WriteMsg(&ReqWorkConn{}); err != nil {
+ t.Fatal(err)
+ }
+ <-c.writes
+ c.transport.mu.Lock()
+ raw := append([]byte(nil), c.transport.buffer.Bytes()...)
+ c.transport.mu.Unlock()
+ m, err := ReadMsg(libcrypto.NewReader(bytes.NewReader(raw), regressionKey))
+ if err != nil {
+ t.Fatalf("encrypted control roundtrip: %v", err)
+ }
+ if _, ok := m.(*ReqWorkConn); !ok {
+ t.Fatalf("decoded %T", m)
+ }
+}
+
+func TestControlCryptoWriterKeepsErrorAfterTransportRecovery(t *testing.T) {
+ c := newRegressionControl(t)
+ regressionWarm(t, c)
+ defer c.transport.close()
+ c.transport.arm()
+ if err := c.WriteMsg(&ReqWorkConn{}); !errors.Is(err, errRegressionWriteTimeout) {
+ t.Fatalf("want timeout, got %v", err)
+ }
+ <-c.writes
+ beforeCalls, failures, beforeBytes := c.transport.snapshot()
+ for range 5 {
+ if err := c.WriteMsg(&ReqWorkConn{}); !errors.Is(err, errRegressionWriteTimeout) {
+ t.Fatalf("sticky error missing: %v", err)
+ }
+ <-c.writes
+ }
+ afterCalls, _, afterBytes := c.transport.snapshot()
+ if beforeCalls != afterCalls || beforeBytes != afterBytes || failures != 1 {
+ t.Fatalf("unexpected retry: calls %d -> %d, failures %d, bytes %d -> %d",
+ beforeCalls, afterCalls, failures, beforeBytes, afterBytes)
+ }
+ if n, err := c.transport.Write([]byte("healthy underlying transport")); err != nil || n == 0 {
+ t.Fatalf("underlying transport did not recover: %v", err)
+ }
+ t.Log("one recoverable write error permanently invalidates this crypto writer; recovered underlying transport is never retried")
+}
+
+func TestDispatcherWriteErrorClosesControlAndFreshSessionWorks(t *testing.T) {
+ c := newRegressionControl(t)
+ regressionWarm(t, c)
+ defer c.transport.close()
+ d := NewDispatcher(c)
+ d.Run()
+ c.transport.arm()
+ if err := d.Send(&ReqWorkConn{}); err != nil && !errors.Is(err, io.EOF) {
+ t.Fatal(err)
+ }
+ if err := regressionReceiveWrite(t, c); !errors.Is(err, errRegressionWriteTimeout) {
+ t.Fatal(err)
+ }
+ regressionAwait(t, c.transport.closed, "failed control connection close")
+ regressionAwait(t, d.Done(), "failed dispatcher completion")
+ if err := d.Send(&ReqWorkConn{}); !errors.Is(err, io.EOF) {
+ t.Fatalf("Send after completed shutdown: %v", err)
+ }
+ if c.closeCalls.Load() != 1 {
+ t.Fatalf("connection closed %d times", c.closeCalls.Load())
+ }
+
+ fresh := newRegressionControl(t)
+ defer fresh.transport.close()
+ d2 := NewDispatcher(fresh)
+ d2.Run()
+ for range 5 {
+ if err := d2.Send(&ReqWorkConn{}); err != nil {
+ t.Fatal(err)
+ }
+ if err := regressionReceiveWrite(t, fresh); err != nil {
+ t.Fatalf("fresh session write failed: %v", err)
+ }
+ }
+ fresh.transport.mu.Lock()
+ raw := append([]byte(nil), fresh.transport.buffer.Bytes()...)
+ fresh.transport.mu.Unlock()
+ r := libcrypto.NewReader(bytes.NewReader(raw), regressionKey)
+ for i := range 5 {
+ m, err := ReadMsg(r)
+ if err != nil {
+ t.Fatalf("new stream decode %d: %v", i, err)
+ }
+ if _, ok := m.(*ReqWorkConn); !ok {
+ t.Fatalf("decoded %T", m)
+ }
+ }
+ fresh.transport.close()
+ regressionAwait(t, d2.Done(), "new-session cleanup")
+ t.Log("write error closes failed control; new session sends and decrypts five valid ReqWorkConn messages")
+}
+
+func TestDispatcherWriteErrorWaitsForSynchronousHandler(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ c := newRegressionControl(t)
+ regressionWarm(t, c)
+ defer c.transport.close()
+ d := NewDispatcher(c)
+ entered, release := make(chan struct{}), make(chan struct{})
+ var releaseOnce sync.Once
+ defer releaseOnce.Do(func() { close(release) })
+ d.RegisterHandler(&ReqWorkConn{}, func(Message) {
+ close(entered)
+ <-release
+ })
+ d.Run()
+ c.incoming <- &ReqWorkConn{}
+ regressionAwait(t, entered, "synchronous handler entry")
+ c.transport.arm()
+ if err := d.Send(&ReqWorkConn{}); err != nil && !errors.Is(err, io.EOF) {
+ t.Fatal(err)
+ }
+ if err := regressionReceiveWrite(t, c); !errors.Is(err, errRegressionWriteTimeout) {
+ t.Fatal(err)
+ }
+ regressionAwait(t, c.transport.closed, "write failure closes control during handler")
+ if err := d.Send(&ReqWorkConn{}); !errors.Is(err, io.EOF) {
+ t.Fatalf("Send after stop while handler is active: %v", err)
+ }
+ // Let the writer finish stopping while the handler remains blocked.
+ synctest.Wait()
+ select {
+ case <-d.Done():
+ t.Fatal("Done closed before synchronous handler returned; owner cleanup could race proxy registration")
+ default:
+ }
+ releaseOnce.Do(func() { close(release) })
+ regressionAwait(t, d.Done(), "completion after handler returned")
+ })
+}
+
+func TestDispatcherWriteErrorUnblocksSenderWithFullQueue(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ c := newRegressionControl(t)
+ regressionWarm(t, c)
+ defer c.transport.close()
+ entered, release := make(chan struct{}), make(chan struct{})
+ var releaseOnce sync.Once
+ defer releaseOnce.Do(func() { close(release) })
+ c.beforeWrite = func() { close(entered); <-release }
+ c.transport.arm()
+ d := NewDispatcher(c)
+ if err := d.Send(&ReqWorkConn{}); err != nil {
+ t.Fatal(err)
+ }
+ d.Run()
+ regressionAwait(t, entered, "blocked transport write")
+ for i := 0; i < cap(d.sendCh); i++ {
+ if err := d.Send(&ReqWorkConn{}); err != nil {
+ t.Fatal(err)
+ }
+ }
+ sent := make(chan error, 1)
+ go func() { sent <- d.Send(&ReqWorkConn{}) }()
+ // Wait until the sender has reached its full-queue channel wait.
+ synctest.Wait()
+ select {
+ case err := <-sent:
+ t.Fatalf("Send did not block on full queue: %v", err)
+ default:
+ }
+ releaseOnce.Do(func() { close(release) })
+ select {
+ case err := <-sent:
+ if !errors.Is(err, io.EOF) {
+ t.Fatalf("blocked Send after write failure: %v", err)
+ }
+ case <-time.After(time.Second):
+ t.Fatal("write failure did not unblock full-queue sender")
+ }
+ regressionAwait(t, d.Done(), "full-queue dispatcher cleanup")
+ })
+}
+
+func TestDispatcherConcurrentReadWriteFailureClosesOnce(t *testing.T) {
+ c := newRegressionControl(t)
+ regressionWarm(t, c)
+ defer c.transport.close()
+ readEntered, writeEntered := make(chan struct{}), make(chan struct{})
+ release := make(chan struct{})
+ var releaseOnce sync.Once
+ defer releaseOnce.Do(func() { close(release) })
+ c.beforeRead = func() error {
+ close(readEntered)
+ <-release
+ return io.EOF
+ }
+ c.beforeWrite = func() {
+ close(writeEntered)
+ <-release
+ }
+ d := NewDispatcher(c)
+ c.transport.arm()
+ if err := d.Send(&ReqWorkConn{}); err != nil {
+ t.Fatal(err)
+ }
+ readExited, writeExited := make(chan struct{}), make(chan struct{})
+ go func() {
+ defer close(readExited)
+ d.readLoop()
+ }()
+ go func() {
+ defer close(writeExited)
+ d.sendLoop()
+ }()
+ regressionAwait(t, readEntered, "read path before injected failure")
+ regressionAwait(t, writeEntered, "write path before injected failure")
+ releaseOnce.Do(func() { close(release) })
+ regressionAwait(t, readExited, "failed read loop exit")
+ regressionAwait(t, writeExited, "failed write loop exit")
+ regressionAwait(t, d.Done(), "concurrent read/write shutdown")
+ if err := regressionReceiveWrite(t, c); !errors.Is(err, errRegressionWriteTimeout) {
+ t.Fatalf("concurrent WriteMsg failure: %v", err)
+ }
+ if _, failures, _ := c.transport.snapshot(); failures != 1 {
+ t.Fatalf("injected transport failed %d times", failures)
+ }
+ if c.closeCalls.Load() != 1 {
+ t.Fatalf("connection closed %d times", c.closeCalls.Load())
+ }
+}
diff --git a/pkg/msg/handler_yamux_test.go b/pkg/msg/handler_yamux_test.go
new file mode 100644
index 00000000..25d91a9e
--- /dev/null
+++ b/pkg/msg/handler_yamux_test.go
@@ -0,0 +1,225 @@
+// Copyright 2026 The frp Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package msg
+
+import (
+ "errors"
+ "io"
+ "net"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ fmux "github.com/fatedier/yamux"
+
+ "github.com/fatedier/frp/pkg/proto/wire"
+ netpkg "github.com/fatedier/frp/pkg/util/net"
+)
+
+// Delay one real TCP write without injecting an error. The stream's write
+// failure must come from yamux's ConnectionWriteTimeout.
+type delayedTCPConn struct {
+ net.Conn
+ armed atomic.Bool
+ entered chan struct{}
+ release chan struct{}
+}
+
+func (c *delayedTCPConn) Write(p []byte) (int, error) {
+ if c.armed.Swap(false) {
+ close(c.entered)
+ <-c.release
+ }
+ return c.Conn.Write(p)
+}
+
+type observedYamuxConn struct {
+ net.Conn
+ writes chan error
+ readEntered chan struct{}
+ readOnce sync.Once
+ closed chan struct{}
+ closeCalls atomic.Int32
+}
+
+func (c *observedYamuxConn) Read(p []byte) (int, error) {
+ c.readOnce.Do(func() { close(c.readEntered) })
+ return c.Conn.Read(p)
+}
+
+func (c *observedYamuxConn) Write(p []byte) (int, error) {
+ n, err := c.Conn.Write(p)
+ c.writes <- err
+ return n, err
+}
+
+func (c *observedYamuxConn) Close() error {
+ c.closeCalls.Add(1)
+ err := c.Conn.Close()
+ close(c.closed)
+ return err
+}
+
+func newLoopbackYamuxPair(t *testing.T) (*fmux.Session, *fmux.Session, *delayedTCPConn) {
+ t.Helper()
+ listener, err := net.Listen("tcp", "127.0.0.1:0")
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer listener.Close()
+ clientTCP, err := net.Dial("tcp", listener.Addr().String())
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { clientTCP.Close() })
+ serverTCP, err := listener.Accept()
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { serverTCP.Close() })
+ gate := &delayedTCPConn{Conn: clientTCP, entered: make(chan struct{}), release: make(chan struct{})}
+ config := fmux.DefaultConfig()
+ config.ConnectionWriteTimeout = 80 * time.Millisecond
+ config.LogOutput = io.Discard
+ client, err := fmux.Client(gate, config)
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { client.Close() })
+ server, err := fmux.Server(serverTCP, config)
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { server.Close() })
+ return client, server, gate
+}
+
+func loopbackCryptoConn(t *testing.T, conn net.Conn, protocol string, role netpkg.AEADCryptoRole) *Conn {
+ t.Helper()
+ var rw io.ReadWriter
+ var err error
+ if protocol == wire.ProtocolV2 {
+ rw, err = netpkg.NewAEADCryptoReadWriter(
+ conn, regressionKey, role, wire.AEADAlgorithmAES256GCM, []byte("isolated-test-transcript"))
+ } else {
+ rw, err = netpkg.NewCryptoReadWriter(conn, regressionKey)
+ }
+ if err != nil {
+ t.Fatal(err)
+ }
+ return NewConn(conn, NewReadWriter(rw, protocol))
+}
+
+func loopbackControlPair(t *testing.T, client, server *fmux.Session, protocol string) (*Conn, *Conn, *observedYamuxConn) {
+ t.Helper()
+ local, err := client.OpenStream()
+ if err != nil {
+ t.Fatal(err)
+ }
+ remote, err := server.AcceptStream()
+ if err != nil {
+ t.Fatal(err)
+ }
+ observed := &observedYamuxConn{
+ Conn: local, writes: make(chan error, 16), readEntered: make(chan struct{}), closed: make(chan struct{}),
+ }
+ return loopbackCryptoConn(t, observed, protocol, netpkg.AEADCryptoRoleClient),
+ loopbackCryptoConn(t, remote, protocol, netpkg.AEADCryptoRoleServer), observed
+}
+
+func loopbackReadReq(t *testing.T, conn *Conn) {
+ t.Helper()
+ if err := conn.SetReadDeadline(time.Now().Add(time.Second)); err != nil {
+ t.Fatal(err)
+ }
+ defer conn.SetReadDeadline(time.Time{})
+ m, err := conn.ReadMsg()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, ok := m.(*ReqWorkConn); !ok {
+ t.Fatalf("decoded %T, want ReqWorkConn", m)
+ }
+}
+
+func TestDispatcherYamuxWriteTimeoutInterruptsRead(t *testing.T) {
+ for _, protocol := range []string{wire.ProtocolV1, wire.ProtocolV2} {
+ t.Run(protocol, func(t *testing.T) {
+ client, server, gate := newLoopbackYamuxPair(t)
+ var releaseOnce sync.Once
+ defer releaseOnce.Do(func() { close(gate.release) })
+ control, peer, observed := loopbackControlPair(t, client, server, protocol)
+ if err := control.WriteMsg(&ReqWorkConn{}); err != nil {
+ t.Fatal(err)
+ }
+ loopbackReadReq(t, peer)
+ // Warm-up has completed, including all crypto header writes.
+ for len(observed.writes) > 0 {
+ <-observed.writes
+ }
+ d := NewDispatcher(control)
+ readExited := make(chan struct{})
+ go func() {
+ defer close(readExited)
+ d.readLoop()
+ }()
+ regressionAwait(t, observed.readEntered, "control reader entry")
+ gate.armed.Store(true)
+ // Queue before starting the writer so this Send cannot race its failure.
+ if err := d.Send(&ReqWorkConn{}); err != nil {
+ t.Fatal(err)
+ }
+ go d.sendLoop()
+ regressionAwait(t, gate.entered, "TCP write delay barrier")
+ select {
+ case err := <-observed.writes:
+ if !errors.Is(err, fmux.ErrConnectionWriteTimeout) {
+ t.Fatalf("stream write error: %v, want yamux connection write timeout", err)
+ }
+ t.Logf("production yamux write failed: %v", err)
+ case <-time.After(time.Second):
+ t.Fatal("yamux write did not time out")
+ }
+ regressionAwait(t, d.stopCh, "write failure stops sends")
+ if err := d.Send(&ReqWorkConn{}); !errors.Is(err, io.EOF) {
+ t.Fatalf("Send after stop: %v, want EOF", err)
+ }
+ releaseOnce.Do(func() { close(gate.release) })
+ regressionAwait(t, observed.closed, "control Close returns")
+ // peer never closes its write half: local yamux Close alone cannot
+ // interrupt Read. Neither a peer FIN nor session cleanup may help.
+ regressionAwait(t, d.Done(), "Done without peer FIN")
+ regressionAwait(t, readExited, "control readLoop exit without peer FIN")
+ if observed.closeCalls.Load() != 1 {
+ t.Fatalf("control closed %d times, want once", observed.closeCalls.Load())
+ }
+ if client.IsClosed() || server.IsClosed() {
+ t.Fatal("yamux session unexpectedly closed")
+ }
+ fresh, freshPeer, freshObserved := loopbackControlPair(t, client, server, protocol)
+ for range 5 {
+ if err := fresh.WriteMsg(&ReqWorkConn{}); err != nil {
+ t.Fatal(err)
+ }
+ loopbackReadReq(t, freshPeer)
+ for len(freshObserved.writes) > 0 {
+ <-freshObserved.writes
+ }
+ }
+ t.Log("Done/readLoop exited without peer FIN; Close once; same yamux session decrypted five fresh messages")
+ })
+ }
+}
diff --git a/pkg/plugin/visitor/virtual_net.go b/pkg/plugin/visitor/virtual_net.go
index fa8c700e..f95856c2 100644
--- a/pkg/plugin/visitor/virtual_net.go
+++ b/pkg/plugin/visitor/virtual_net.go
@@ -20,6 +20,7 @@ import (
"context"
"errors"
"fmt"
+ "io"
"net"
"sync"
"time"
@@ -33,10 +34,16 @@ func init() {
Register(v1.VisitorPluginVirtualNet, NewVirtualNetPlugin)
}
+type clientRouteController interface {
+ RegisterClientRoute(context.Context, string, []net.IPNet, io.ReadWriteCloser)
+ UnregisterClientRoute(string, io.Writer) bool
+}
+
type VirtualNetPlugin struct {
pluginCtx PluginContext
- routes []net.IPNet
+ routeController clientRouteController
+ routes []net.IPNet
mu sync.Mutex
controllerConn net.Conn
@@ -48,6 +55,11 @@ type VirtualNetPlugin struct {
cancel context.CancelFunc
}
+const (
+ virtualNetReconnectBaseDelay = 60 * time.Second
+ virtualNetReconnectMaxDelay = 300 * time.Second
+)
+
func NewVirtualNetPlugin(pluginCtx PluginContext, options v1.VisitorPluginOptions) (Plugin, error) {
opts := options.(*v1.VirtualNetVisitorPluginOptions)
@@ -55,6 +67,9 @@ func NewVirtualNetPlugin(pluginCtx PluginContext, options v1.VisitorPluginOption
pluginCtx: pluginCtx,
routes: make([]net.IPNet, 0),
}
+ if pluginCtx.VnetController != nil {
+ p.routeController = pluginCtx.VnetController
+ }
p.ctx, p.cancel = context.WithCancel(pluginCtx.Ctx)
@@ -85,7 +100,7 @@ func (p *VirtualNetPlugin) Name() string {
func (p *VirtualNetPlugin) Start() {
xl := xlog.FromContextSafe(p.pluginCtx.Ctx)
- if p.pluginCtx.VnetController == nil {
+ if p.routeController == nil {
return
}
@@ -111,16 +126,17 @@ func (p *VirtualNetPlugin) run() {
select {
case <-p.ctx.Done():
xl.Infof("VirtualNetPlugin run loop for visitor [%s] stopping (context cancelled before pipe creation).", p.pluginCtx.Name)
- p.cleanupControllerConn(xl)
+ p.cleanupCurrentControllerConn(xl)
return
default:
}
controllerConn, pluginConn := net.Pipe()
-
- p.mu.Lock()
- p.controllerConn = controllerConn
- p.mu.Unlock()
+ xl.Infof("attempting to register client route for visitor [%s]", p.pluginCtx.Name)
+ if !p.registerControllerConn(controllerConn, pluginConn) {
+ xl.Infof("VirtualNetPlugin run loop for visitor [%s] stopping (context cancelled before route registration).", p.pluginCtx.Name)
+ return
+ }
// Wrap with CloseNotifyConn which supports both close notification and error recording
var closeErr error
@@ -129,8 +145,6 @@ func (p *VirtualNetPlugin) run() {
close(currentCloseSignal) // Signal the run loop on close.
})
- xl.Infof("attempting to register client route for visitor [%s]", p.pluginCtx.Name)
- p.pluginCtx.VnetController.RegisterClientRoute(p.ctx, p.pluginCtx.Name, p.routes, controllerConn)
xl.Infof("successfully registered client route for visitor [%s]. Starting connection handler with CloseNotifyConn.", p.pluginCtx.Name)
// Pass the CloseNotifyConn to the visitor for handling.
@@ -141,7 +155,7 @@ func (p *VirtualNetPlugin) run() {
select {
case <-p.ctx.Done():
xl.Infof("VirtualNetPlugin run loop stopping for visitor [%s] (context cancelled while waiting).", p.pluginCtx.Name)
- p.cleanupControllerConn(xl)
+ p.cleanupControllerConn(xl, controllerConn)
return
case <-currentCloseSignal:
// Determine reconnect delay based on error with exponential backoff
@@ -152,8 +166,7 @@ func (p *VirtualNetPlugin) run() {
p.pluginCtx.Name, p.consecutiveErrors, closeErr)
// Exponential backoff: 60s, 120s, 240s, 300s (capped)
- baseDelay := 60 * time.Second
- reconnectDelay = min(baseDelay*time.Duration(1< 0 {
@@ -167,7 +180,7 @@ func (p *VirtualNetPlugin) run() {
}
// The visitor closed the plugin side. Close the controller side.
- p.cleanupControllerConn(xl)
+ p.cleanupControllerConn(xl, controllerConn)
xl.Infof("waiting %v before attempting reconnection for visitor [%s]...", reconnectDelay, p.pluginCtx.Name)
select {
@@ -182,16 +195,66 @@ func (p *VirtualNetPlugin) run() {
}
}
-// cleanupControllerConn closes the current controllerConn (if it exists) under lock.
-func (p *VirtualNetPlugin) cleanupControllerConn(xl *xlog.Logger) {
+// registerControllerConn publishes and registers controllerConn atomically with
+// respect to Close. A canceled plugin cannot register a new route.
+func (p *VirtualNetPlugin) registerControllerConn(controllerConn, pluginConn net.Conn) bool {
+ p.mu.Lock()
+ if p.ctx.Err() != nil || p.routeController == nil {
+ p.mu.Unlock()
+ _ = controllerConn.Close()
+ _ = pluginConn.Close()
+ return false
+ }
+
+ p.controllerConn = controllerConn
+ p.routeController.RegisterClientRoute(p.ctx, p.pluginCtx.Name, p.routes, controllerConn)
+ p.mu.Unlock()
+ return true
+}
+
+// virtualNetReconnectDelay returns a bounded reconnect delay without allowing
+// the exponential shift to overflow for large consecutive error counts.
+func virtualNetReconnectDelay(consecutiveErrors int) time.Duration {
+ if consecutiveErrors <= 1 {
+ return virtualNetReconnectBaseDelay
+ }
+ if consecutiveErrors >= 4 {
+ return virtualNetReconnectMaxDelay
+ }
+ return virtualNetReconnectBaseDelay * time.Duration(1< 0 || len(req.TransferEncoding) > 0 {
+ http.Error(rw, http.StatusText(http.StatusBadRequest), http.StatusBadRequest)
+ return
+ }
+
hj, ok := rw.(http.Hijacker)
if !ok {
rw.WriteHeader(http.StatusInternalServerError)
return
}
- client, _, err := hj.Hijack()
+ // Forward only CONNECT headers; the original Body must not be used after Hijack.
+ outreq := req.Clone(req.Context())
+ outreq.Body = nil
+ outreq.GetBody = nil
+ outreq.ContentLength = 0
+ outreq.TransferEncoding = nil
+ outreq.Trailer = nil
+ outreq.Header.Del("Content-Length")
+ outreq.Header.Del("Transfer-Encoding")
+ outreq.Header.Del("Trailer")
+
+ client, buffered, err := hj.Hijack()
if err != nil {
rw.WriteHeader(http.StatusInternalServerError)
return
@@ -224,8 +240,12 @@ func (rp *HTTPReverseProxy) connectHandler(rw http.ResponseWriter, req *http.Req
client.Close()
return
}
- _ = req.Write(remote)
- go libio.Join(remote, client)
+ if err := outreq.Write(remote); err != nil {
+ _ = remote.Close()
+ _ = client.Close()
+ return
+ }
+ go libio.Join(remote, libio.WrapReadWriteCloser(buffered.Reader, client, client.Close))
}
func getRequestRouteUser(req *http.Request) string {
diff --git a/pkg/util/vhost/http_test.go b/pkg/util/vhost/http_test.go
index a6493b08..e661ca08 100644
--- a/pkg/util/vhost/http_test.go
+++ b/pkg/util/vhost/http_test.go
@@ -3,17 +3,266 @@ package vhost
import (
"bufio"
"fmt"
+ "io"
"net"
"net/http"
"net/http/httptest"
+ "strings"
+ "sync/atomic"
"testing"
"time"
"github.com/stretchr/testify/require"
httppkg "github.com/fatedier/frp/pkg/util/http"
+ netpkg "github.com/fatedier/frp/pkg/util/net"
)
+func TestHTTPConnectTunnel(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ framing string
+ pipelined bool
+ }{
+ {name: "plain"},
+ {name: "plain-early", pipelined: true},
+ {name: "cl0", framing: "Content-Length: 0\r\n"},
+ {name: "cl0-early", framing: "Content-Length: 0\r\n", pipelined: true},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ remote, backend := net.Pipe()
+ defer remote.Close()
+ defer backend.Close()
+ require.NoError(t, backend.SetDeadline(time.Now().Add(5*time.Second)))
+
+ rp := NewHTTPReverseProxy(HTTPReverseProxyOptions{}, NewRouters())
+ require.NoError(t, rp.Register(RouteConfig{
+ Domain: "target.example.com",
+ Username: "alice",
+ Password: "secret",
+ CreateConnFn: func(string) (net.Conn, error) {
+ return remote, nil
+ },
+ }))
+ listener := netpkg.NewInternalListener()
+ server := &http.Server{Handler: rp, ReadHeaderTimeout: time.Second}
+ serveErr := make(chan error, 1)
+ go func() {
+ serveErr <- server.Serve(listener)
+ }()
+ defer func() {
+ require.NoError(t, server.Close())
+ require.ErrorIs(t, <-serveErr, http.ErrServerClosed)
+ }()
+
+ client, incoming := net.Pipe()
+ defer client.Close()
+ defer incoming.Close()
+ require.NoError(t, client.SetDeadline(time.Now().Add(5*time.Second)))
+ require.NoError(t, listener.PutConn(incoming))
+
+ proxyAuth := httppkg.BasicAuth("alice", "secret")
+ request := "CONNECT target.example.com:443 HTTP/1.1\r\nHost: target.example.com:443\r\n" +
+ "X-Connect-Test: tunnel\r\nProxy-Authorization: " + proxyAuth + "\r\n" + tc.framing + "\r\n"
+ if tc.pipelined {
+ // One pipe write puts the tunnel data in net/http's request buffer.
+ request += "early data"
+ }
+ _, err := io.WriteString(client, request)
+ require.NoError(t, err)
+ backendReader := bufio.NewReader(backend)
+ req, err := http.ReadRequest(backendReader)
+ require.NoError(t, err)
+ defer req.Body.Close()
+ require.Equal(t, http.MethodConnect, req.Method)
+ require.Equal(t, "target.example.com:443", req.Host)
+ require.Equal(t, "target.example.com:443", req.URL.Host)
+ require.Equal(t, "tunnel", req.Header.Get("X-Connect-Test"))
+ require.Equal(t, proxyAuth, req.Header.Get("Proxy-Authorization"))
+ require.Zero(t, req.ContentLength)
+ require.Empty(t, req.TransferEncoding)
+ if tc.pipelined {
+ data := make([]byte, len("early data"))
+ _, err = io.ReadFull(backendReader, data)
+ require.NoError(t, err)
+ require.Equal(t, "early data", string(data))
+ }
+
+ writeErr := make(chan error, 1)
+ go func() {
+ _, err := io.WriteString(backend, "HTTP/1.1 200 Connection Established\r\n\r\nreply")
+ writeErr <- err
+ }()
+ clientReader := bufio.NewReader(client)
+ response, err := http.ReadResponse(clientReader, req)
+ require.NoError(t, err)
+ defer response.Body.Close()
+ require.Equal(t, http.StatusOK, response.StatusCode)
+ reply := make([]byte, len("reply"))
+ _, err = io.ReadFull(clientReader, reply)
+ require.NoError(t, err)
+ require.Equal(t, "reply", string(reply))
+ require.NoError(t, <-writeErr)
+
+ _, err = io.WriteString(client, "later data")
+ require.NoError(t, err)
+ data := make([]byte, len("later data"))
+ _, err = io.ReadFull(backendReader, data)
+ require.NoError(t, err)
+ require.Equal(t, "later data", string(data))
+ })
+ }
+}
+
+func TestHTTPConnectRejectsContent(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ framing string
+ body string
+ }{
+ {name: "content-length", framing: "Content-Length: 1\r\n", body: "X"},
+ {name: "chunked", framing: "Transfer-Encoding: chunked\r\n", body: "1\r\nX\r\n0\r\n\r\n"},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ backendCalls := 0
+ rp := NewHTTPReverseProxy(HTTPReverseProxyOptions{}, NewRouters())
+ require.NoError(t, rp.Register(RouteConfig{
+ Domain: "target.example.com",
+ CreateConnFn: func(string) (net.Conn, error) {
+ backendCalls++
+ return nil, io.ErrClosedPipe
+ },
+ }))
+ req, err := http.ReadRequest(bufio.NewReader(strings.NewReader(
+ "CONNECT target.example.com:443 HTTP/1.1\r\nHost: target.example.com:443\r\n" + tc.framing + "\r\n" + tc.body)))
+ require.NoError(t, err)
+ defer req.Body.Close()
+ rw := &connectTestResponseWriter{ResponseRecorder: httptest.NewRecorder()}
+
+ rp.ServeHTTP(rw, req)
+
+ require.Equal(t, http.StatusBadRequest, rw.Code)
+ require.Zero(t, rw.hijackCalls, "reject content before Hijack")
+ require.Zero(t, backendCalls, "reject content before creating a backend")
+ })
+ }
+}
+
+func TestHTTPConnectHeaderWriteFailure(t *testing.T) {
+ for _, originalBody := range []bool{false, true} {
+ t.Run(fmt.Sprintf("originalBody=%v", originalBody), func(t *testing.T) {
+ client, incoming := net.Pipe()
+ defer client.Close()
+ defer incoming.Close()
+ remote, backend := net.Pipe()
+ defer remote.Close()
+ defer backend.Close()
+ clientConn := &connectTestConn{Conn: incoming}
+ remoteConn := &connectTestConn{Conn: remote, failWrites: true}
+ rp := NewHTTPReverseProxy(HTTPReverseProxyOptions{}, NewRouters())
+ require.NoError(t, rp.Register(RouteConfig{
+ Domain: "target.example.com",
+ CreateConnFn: func(string) (net.Conn, error) {
+ return remoteConn, nil
+ },
+ }))
+ req := rp.injectRequestInfoToCtx(httptest.NewRequest(http.MethodConnect, "target.example.com:443", nil))
+ require.Equal(t, "target.example.com:443", req.Host)
+ body := &connectTestBody{}
+ if originalBody {
+ req.Body = body
+ }
+ requestBody := req.Body
+ getBodyCalls := 0
+ req.GetBody = func() (io.ReadCloser, error) {
+ getBodyCalls++
+ return body, nil
+ }
+ req.Header.Set("Content-Length", "0")
+ req.Header.Set("Trailer", "X-Connect-Trailer")
+ req.Trailer = http.Header{"X-Connect-Trailer": {"value"}}
+ originalHeader, originalTrailer := req.Header.Clone(), req.Trailer.Clone()
+ rw := &connectTestResponseWriter{
+ ResponseRecorder: httptest.NewRecorder(),
+ conn: clientConn,
+ buffered: bufio.NewReadWriter(bufio.NewReader(clientConn), bufio.NewWriter(clientConn)),
+ }
+
+ rp.connectHandler(rw, req)
+
+ require.Equal(t, 1, rw.hijackCalls)
+ require.Zero(t, body.reads, "do not read the original Body after Hijack")
+ require.Zero(t, body.closes, "do not close the original Body after Hijack")
+ require.Positive(t, remoteConn.writes.Load(), "exercise CONNECT header write failure")
+ require.True(t, remoteConn.closed.Load(), "close backend on CONNECT header write failure")
+ require.True(t, clientConn.closed.Load(), "close client on CONNECT header write failure")
+ require.Zero(t, remoteConn.reads.Load(), "do not start Join after header write failure")
+ require.Zero(t, clientConn.reads.Load(), "do not start Join after header write failure")
+ require.Zero(t, getBodyCalls)
+ require.Equal(t, requestBody, req.Body)
+ require.NotNil(t, req.GetBody)
+ require.Equal(t, originalHeader, req.Header)
+ require.Equal(t, originalTrailer, req.Trailer)
+ })
+ }
+}
+
+type connectTestResponseWriter struct {
+ *httptest.ResponseRecorder
+ conn net.Conn
+ buffered *bufio.ReadWriter
+ hijackCalls int
+}
+
+func (rw *connectTestResponseWriter) Hijack() (net.Conn, *bufio.ReadWriter, error) {
+ rw.hijackCalls++
+ if rw.conn == nil {
+ return nil, nil, io.ErrClosedPipe
+ }
+ return rw.conn, rw.buffered, nil
+}
+
+type connectTestConn struct {
+ net.Conn
+ failWrites bool
+ closed atomic.Bool
+ reads atomic.Int32
+ writes atomic.Int32
+}
+
+func (c *connectTestConn) Read(p []byte) (int, error) {
+ c.reads.Add(1)
+ return c.Conn.Read(p)
+}
+
+func (c *connectTestConn) Write(p []byte) (int, error) {
+ c.writes.Add(1)
+ if c.failWrites {
+ return 0, io.ErrClosedPipe
+ }
+ return c.Conn.Write(p)
+}
+
+func (c *connectTestConn) Close() error {
+ c.closed.Store(true)
+ return c.Conn.Close()
+}
+
+type connectTestBody struct {
+ reads int
+ closes int
+}
+
+func (b *connectTestBody) Read([]byte) (int, error) {
+ b.reads++
+ return 0, io.ErrUnexpectedEOF
+}
+
+func (b *connectTestBody) Close() error {
+ b.closes++
+ return nil
+}
+
func TestHTTPServerProtocols(t *testing.T) {
rp := NewHTTPReverseProxy(HTTPReverseProxyOptions{}, NewRouters())
protocols := new(http.Protocols)
diff --git a/pkg/vnet/controller.go b/pkg/vnet/controller.go
index 348d50d2..99c817b7 100644
--- a/pkg/vnet/controller.go
+++ b/pkg/vnet/controller.go
@@ -246,9 +246,9 @@ func (c *Controller) RegisterClientRoute(ctx context.Context, name string, route
go c.readLoopClient(ctx, conn)
}
-// UnregisterClientRoute Remove client route from routing table
-func (c *Controller) UnregisterClientRoute(name string) {
- c.clientRouter.delRoute(name)
+// UnregisterClientRoute removes a client route only when it is still owned by conn.
+func (c *Controller) UnregisterClientRoute(name string, conn io.Writer) bool {
+ return c.clientRouter.delRoute(name, conn)
}
// StartServerConnReadLoop starts the read loop for a server connection
@@ -304,10 +304,15 @@ func (r *clientRouter) findConn(dst net.IP) (io.Writer, error) {
return nil, fmt.Errorf("no route found for destination %s", dst)
}
-func (r *clientRouter) delRoute(name string) {
+func (r *clientRouter) delRoute(name string, conn io.Writer) bool {
r.mu.Lock()
defer r.mu.Unlock()
+ re, ok := r.routes[name]
+ if !ok || re.conn != conn {
+ return false
+ }
delete(r.routes, name)
+ return true
}
func (r *clientRouter) removeConnRoute(conn io.Writer) {
diff --git a/pkg/vnet/controller_test.go b/pkg/vnet/controller_test.go
new file mode 100644
index 00000000..d3b1c068
--- /dev/null
+++ b/pkg/vnet/controller_test.go
@@ -0,0 +1,64 @@
+// Copyright 2026 The frp Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package vnet
+
+import (
+ "net"
+ "testing"
+
+ "github.com/stretchr/testify/require"
+
+ v1 "github.com/fatedier/frp/pkg/config/v1"
+)
+
+// TestClientRouterDeleteRouteRequiresMatchingConnection verifies that a stale
+// visitor cannot remove a replacement route registered under the same name.
+func TestClientRouterDeleteRouteRequiresMatchingConnection(t *testing.T) {
+ require := require.New(t)
+ controller := NewController(v1.VirtualNetConfig{})
+
+ _, route, err := net.ParseCIDR("10.1.0.1/32")
+ require.NoError(err)
+
+ oldConn, oldPeer := net.Pipe()
+ t.Cleanup(func() {
+ _ = oldConn.Close()
+ _ = oldPeer.Close()
+ })
+ replacementConn, replacementPeer := net.Pipe()
+ t.Cleanup(func() {
+ _ = replacementConn.Close()
+ _ = replacementPeer.Close()
+ })
+
+ controller.clientRouter.addRoute("vnet-visitor", []net.IPNet{*route}, oldConn)
+ controller.clientRouter.addRoute("vnet-visitor", []net.IPNet{*route}, replacementConn)
+
+ require.False(controller.UnregisterClientRoute("vnet-visitor", oldConn))
+ got, err := controller.clientRouter.findConn(net.ParseIP("10.1.0.1"))
+ require.NoError(err)
+ require.Same(replacementConn, got)
+
+ // The read loop for an old connection can exit after a replacement route
+ // has already been registered. Its deferred cleanup must keep the new owner.
+ controller.clientRouter.removeConnRoute(oldConn)
+ got, err = controller.clientRouter.findConn(net.ParseIP("10.1.0.1"))
+ require.NoError(err)
+ require.Same(replacementConn, got)
+
+ require.True(controller.UnregisterClientRoute("vnet-visitor", replacementConn))
+ _, err = controller.clientRouter.findConn(net.ParseIP("10.1.0.1"))
+ require.Error(err)
+}
diff --git a/server/http/controller_v2_test.go b/server/http/controller_v2_test.go
index 042cedbc..6a233a36 100644
--- a/server/http/controller_v2_test.go
+++ b/server/http/controller_v2_test.go
@@ -615,7 +615,7 @@ func TestMatchV2ProxyQueryMatchesSpecFields(t *testing.T) {
name: "tcp remote port",
item: model.V2ProxyResp{Name: "tcp-proxy", Spec: model.V2ProxySpec{
Type: "tcp",
- TCP: &model.V2TCPProxySpec{RemotePort: v2TestIntPtr(6000)},
+ TCP: &model.V2TCPProxySpec{RemotePort: new(6000)},
}},
q: "6000",
want: true,
@@ -624,7 +624,7 @@ func TestMatchV2ProxyQueryMatchesSpecFields(t *testing.T) {
name: "udp remote port",
item: model.V2ProxyResp{Name: "udp-proxy", Spec: model.V2ProxySpec{
Type: "udp",
- UDP: &model.V2UDPProxySpec{RemotePort: v2TestIntPtr(7000)},
+ UDP: &model.V2UDPProxySpec{RemotePort: new(7000)},
}},
q: "7000",
want: true,
@@ -633,7 +633,7 @@ func TestMatchV2ProxyQueryMatchesSpecFields(t *testing.T) {
name: "remote port does not match colon form",
item: model.V2ProxyResp{Name: "tcp-proxy", Spec: model.V2ProxySpec{
Type: "tcp",
- TCP: &model.V2TCPProxySpec{RemotePort: v2TestIntPtr(6000)},
+ TCP: &model.V2TCPProxySpec{RemotePort: new(6000)},
}},
q: ":6000",
want: false,
@@ -767,10 +767,6 @@ func TestLegacyAPIResponsesRemainBare(t *testing.T) {
}
}
-func v2TestIntPtr(value int) *int {
- return &value
-}
-
func newV2TestController(t *testing.T) *Controller {
t.Helper()
diff --git a/server/proxy/http.go b/server/proxy/http.go
index 05afc2e9..c3da8383 100644
--- a/server/proxy/http.go
+++ b/server/proxy/http.go
@@ -139,9 +139,7 @@ func (pxy *HTTPProxy) GetRealConn(remoteAddr string) (workConn net.Conn, err err
}
if pxy.GetLimiter() != nil {
- rwc = libio.WrapReadWriteCloser(limit.NewReader(rwc, pxy.GetLimiter()), limit.NewWriter(rwc, pxy.GetLimiter()), func() error {
- return rwc.Close()
- })
+ rwc = libio.WrapReadWriteCloser(limit.NewReader(rwc, pxy.GetLimiter()), limit.NewWriter(rwc, pxy.GetLimiter()), rwc.Close)
}
workConn = netpkg.WrapReadWriteCloserToConn(rwc, tmpConn)
diff --git a/server/proxy/proxy.go b/server/proxy/proxy.go
index f4e75f0a..705aa9da 100644
--- a/server/proxy/proxy.go
+++ b/server/proxy/proxy.go
@@ -308,9 +308,7 @@ func (pxy *BaseProxy) handleUserTCPConnection(userConn net.Conn) {
}
if pxy.GetLimiter() != nil {
- local = libio.WrapReadWriteCloser(limit.NewReader(local, pxy.GetLimiter()), limit.NewWriter(local, pxy.GetLimiter()), func() error {
- return local.Close()
- })
+ local = libio.WrapReadWriteCloser(limit.NewReader(local, pxy.GetLimiter()), limit.NewWriter(local, pxy.GetLimiter()), local.Close)
}
xl.Debugf("join connections, workConn(l[%s] r[%s]) userConn(l[%s] r[%s])", workConn.LocalAddr().String(),
diff --git a/server/proxy/udp.go b/server/proxy/udp.go
index 47bdfc8c..8a69c360 100644
--- a/server/proxy/udp.go
+++ b/server/proxy/udp.go
@@ -217,9 +217,7 @@ func (pxy *UDPProxy) Run() (remoteAddr string, err error) {
}
if pxy.GetLimiter() != nil {
- rwc = libio.WrapReadWriteCloser(limit.NewReader(rwc, pxy.GetLimiter()), limit.NewWriter(rwc, pxy.GetLimiter()), func() error {
- return rwc.Close()
- })
+ rwc = libio.WrapReadWriteCloser(limit.NewReader(rwc, pxy.GetLimiter()), limit.NewWriter(rwc, pxy.GetLimiter()), rwc.Close)
}
pxy.workConn = netpkg.WrapReadWriteCloserToConn(rwc, workConn)
diff --git a/server/service.go b/server/service.go
index 82883bb2..4556358b 100644
--- a/server/service.go
+++ b/server/service.go
@@ -25,11 +25,12 @@ import (
"net/http"
"os"
"strconv"
+ "sync"
"time"
"github.com/fatedier/golib/crypto"
"github.com/fatedier/golib/net/mux"
- fmux "github.com/hashicorp/yamux"
+ fmux "github.com/fatedier/yamux"
quic "github.com/quic-go/quic-go"
"github.com/samber/lo"
@@ -120,6 +121,9 @@ type Service struct {
// web server for dashboard UI and apis
webServer *httppkg.Server
+ // HTTP vhost resources are started and owned by NewService.
+ httpVhostServer *http.Server
+ httpVhostListener net.Listener
sshTunnelGateway *ssh.Gateway
@@ -133,7 +137,10 @@ type Service struct {
// service context
ctx context.Context
// call cancel to stop service
- cancel context.CancelFunc
+ cancel context.CancelFunc
+ lifecycleMu sync.Mutex
+ closed bool
+ closeOnce sync.Once
}
func NewService(cfg *v1.ServerConfig) (*Service, error) {
@@ -319,6 +326,8 @@ func NewService(cfg *v1.ServerConfig) (*Service, error) {
return nil, fmt.Errorf("create vhost http listener error, %v", err)
}
}
+ svr.httpVhostServer = server
+ svr.httpVhostListener = l
go func() {
_ = server.Serve(l)
}()
@@ -365,8 +374,22 @@ func NewService(cfg *v1.ServerConfig) (*Service, error) {
func (svr *Service) Run(ctx context.Context) {
ctx, cancel := context.WithCancel(ctx)
+ svr.lifecycleMu.Lock()
+ if svr.closed {
+ svr.lifecycleMu.Unlock()
+ cancel()
+ return
+ }
svr.ctx = ctx
svr.cancel = cancel
+ svr.lifecycleMu.Unlock()
+ defer svr.Close()
+ if ctx.Err() != nil {
+ return
+ }
+ // Cancellation must unblock the synchronous main listener's Accept.
+ stopClose := context.AfterFunc(ctx, func() { _ = svr.Close() })
+ defer stopClose()
// run dashboard web server.
if svr.webServer != nil {
@@ -400,13 +423,21 @@ func (svr *Service) Run(ctx context.Context) {
svr.HandleListener(svr.listener, false)
<-svr.ctx.Done()
- // service context may not be canceled by svr.Close(), we should call it here to release resources
- if svr.listener != nil {
- svr.Close()
- }
}
func (svr *Service) Close() error {
+ svr.lifecycleMu.Lock()
+ svr.closed = true
+ cancel := svr.cancel
+ svr.lifecycleMu.Unlock()
+ if cancel != nil {
+ cancel()
+ }
+ svr.closeOnce.Do(svr.closeResources)
+ return nil
+}
+
+func (svr *Service) closeResources() {
if svr.kcpListener != nil {
svr.kcpListener.Close()
}
@@ -428,16 +459,17 @@ func (svr *Service) Close() error {
if svr.webServer != nil {
svr.webServer.Close()
}
+ if svr.httpVhostServer != nil {
+ svr.httpVhostServer.Close()
+ // Close even if Serve has not registered its listener yet.
+ svr.httpVhostListener.Close()
+ }
if svr.sshTunnelGateway != nil {
svr.sshTunnelGateway.Close()
}
svr.rc.Close()
svr.muxer.Close()
svr.ctlManager.Close()
- if svr.cancel != nil {
- svr.cancel()
- }
- return nil
}
func (svr *Service) handleConnection(ctx context.Context, conn net.Conn, internal bool) {
diff --git a/server/service_lifecycle_test.go b/server/service_lifecycle_test.go
new file mode 100644
index 00000000..61481b32
--- /dev/null
+++ b/server/service_lifecycle_test.go
@@ -0,0 +1,151 @@
+package server
+
+import (
+ "context"
+ "fmt"
+ "net"
+ "sync"
+ "sync/atomic"
+ "testing"
+
+ "github.com/stretchr/testify/require"
+
+ v1 "github.com/fatedier/frp/pkg/config/v1"
+ "github.com/fatedier/frp/pkg/config/v1/validation"
+)
+
+type lifecycleListener struct {
+ net.Listener
+ accepted, closeEntered, closeResume chan struct{}
+ acceptOnce, closeOnce sync.Once
+ accepts, closes atomic.Int64
+}
+
+func (l *lifecycleListener) Accept() (net.Conn, error) {
+ l.accepts.Add(1)
+ l.acceptOnce.Do(func() { close(l.accepted) })
+ return l.Listener.Accept()
+}
+
+func (l *lifecycleListener) Close() error {
+ l.closes.Add(1)
+ err := l.Listener.Close()
+ if l.closeEntered != nil {
+ l.closeOnce.Do(func() { close(l.closeEntered) })
+ <-l.closeResume
+ }
+ return err
+}
+
+func freeLifecycleTCPPort(t *testing.T) int {
+ t.Helper()
+ l, err := net.Listen("tcp", "127.0.0.1:0")
+ require.NoError(t, err)
+ defer l.Close()
+ return l.Addr().(*net.TCPAddr).Port
+}
+
+func newLifecycleTestService(t *testing.T, web bool) (*Service, *lifecycleListener, []string) {
+ t.Helper()
+ cfg := &v1.ServerConfig{
+ BindAddr: "127.0.0.1", ProxyBindAddr: "127.0.0.1",
+ }
+ if web {
+ cfg.WebServer = v1.WebServerConfig{Addr: "127.0.0.1", Port: freeLifecycleTCPPort(t)}
+ cfg.VhostHTTPPort = freeLifecycleTCPPort(t)
+ }
+ require.NoError(t, cfg.Complete())
+ cfg.BindPort = 0
+ _, err := validation.NewConfigValidator(nil).ValidateServerConfig(cfg)
+ require.NoError(t, err)
+ svr, err := NewService(cfg)
+ require.NoError(t, err)
+ l := &lifecycleListener{Listener: svr.listener, accepted: make(chan struct{})}
+ svr.listener = l
+ addresses := []string{l.Addr().String()}
+ if web {
+ addresses = append(addresses, svr.webServer.Address())
+ addresses = append(addresses, net.JoinHostPort(cfg.ProxyBindAddr, fmt.Sprint(cfg.VhostHTTPPort)))
+ }
+ assertLifecyclePorts(t, addresses, false)
+ return svr, l, addresses
+}
+
+func assertLifecyclePorts(t *testing.T, addresses []string, released bool) {
+ t.Helper()
+ for _, address := range addresses {
+ ln, err := net.Listen("tcp", address)
+ if ln != nil {
+ _ = ln.Close()
+ }
+ if released {
+ require.NoError(t, err, "listener %s was not released", address)
+ } else {
+ require.Error(t, err, "listener %s was not bound", address)
+ }
+ }
+}
+
+func TestServiceRunAfterEarlyShutdown(t *testing.T) {
+ for _, mode := range []string{"close before Run", "close before Run with web", "parent already cancelled"} {
+ t.Run(mode, func(t *testing.T) {
+ svr, listener, addresses := newLifecycleTestService(t, mode == "close before Run with web")
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ defer svr.Close()
+ if mode != "parent already cancelled" {
+ require.NoError(t, svr.Close())
+ assertLifecyclePorts(t, addresses, true)
+ } else {
+ cancel()
+ }
+ done := make(chan struct{})
+ go func() { svr.Run(ctx); close(done) }()
+ waitForSignal(t, done, "Run after early shutdown")
+ require.Zero(t, listener.accepts.Load(), "closed service started accepting")
+ require.Equal(t, int64(1), listener.closes.Load())
+ assertLifecyclePorts(t, addresses, true)
+ })
+ }
+}
+
+func TestServiceCloseMarksClosedBeforeCleanup(t *testing.T) {
+ svr, listener, addresses := newLifecycleTestService(t, false)
+ listener.closeEntered, listener.closeResume = make(chan struct{}), make(chan struct{})
+ var releaseOnce sync.Once
+ release := func() { releaseOnce.Do(func() { close(listener.closeResume) }) }
+ defer release()
+ closeDone := make(chan struct{})
+ go func() { _ = svr.Close(); close(closeDone) }()
+ waitForSignal(t, listener.closeEntered, "Close cleanup barrier")
+ runDone := make(chan struct{})
+ go func() { svr.Run(context.Background()); close(runDone) }()
+ waitForSignal(t, runDone, "Run to observe closed before cleanup finishes")
+ require.Zero(t, listener.accepts.Load(), "Run started during cleanup")
+ assertLifecyclePorts(t, addresses, false)
+ release()
+ waitForSignal(t, closeDone, "Close cleanup to finish")
+ assertLifecyclePorts(t, addresses, true)
+}
+
+func TestServiceParentCancellationAndCloseShareCleanup(t *testing.T) {
+ svr, listener, addresses := newLifecycleTestService(t, false)
+ listener.closeEntered, listener.closeResume = make(chan struct{}), make(chan struct{})
+ var releaseOnce sync.Once
+ release := func() { releaseOnce.Do(func() { close(listener.closeResume) }) }
+ defer release()
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ runDone := make(chan struct{})
+ go func() { svr.Run(ctx); close(runDone) }()
+ waitForSignal(t, listener.accepted, "main listener Accept")
+ cancel()
+ waitForSignal(t, listener.closeEntered, "parent cancellation to close main listener")
+ closeDone := make(chan struct{})
+ go func() { _ = svr.Close(); close(closeDone) }()
+ release()
+ waitForSignal(t, closeDone, "external Close to finish")
+ waitForSignal(t, runDone, "Run cleanup to finish")
+ require.Equal(t, int64(1), listener.closes.Load(), "cleanup ran more than once")
+ assertLifecyclePorts(t, addresses, true)
+}
diff --git a/test/e2e/compatibility/compatibility_test.go b/test/e2e/compatibility/compatibility_test.go
index ad505510..a643f06e 100644
--- a/test/e2e/compatibility/compatibility_test.go
+++ b/test/e2e/compatibility/compatibility_test.go
@@ -192,19 +192,52 @@ transport.wireProtocol = "v2"
})
})
-var _ = ginkgo.Describe("[Compatibility: BinaryUDPPacket]", func() {
+var _ = ginkgo.Describe("[Compatibility: UDPDefault]", func() {
+ f := framework.NewDefaultFramework()
+
+ ginkgo.It("current frps exchanges UDP with baseline frpc using the default wire protocol", func() {
+ portName := port.GenName("CompatDefaultUDPBaselineFRPC")
+ f.RunProcessesWithBinaries(
+ compatCtx.CurrentFRPSPath,
+ compatCtx.BaselineFRPCPath,
+ consts.DefaultServerConfig,
+ []string{udpClientConfig(portName, "")},
+ )
+ framework.NewRequestExpect(f).Protocol("udp").PortName(portName).Ensure()
+ })
+
+ ginkgo.It("baseline frps exchanges UDP with current frpc defaulting to v1", func() {
+ portName := port.GenName("CompatDefaultUDPBaselineFRPS")
+ f.RunProcessesWithBinaries(
+ compatCtx.BaselineFRPSPath,
+ compatCtx.CurrentFRPCPath,
+ consts.DefaultServerConfig,
+ []string{udpClientConfig(portName, "")},
+ )
+ framework.NewRequestExpect(f).Protocol("udp").PortName(portName).Ensure()
+ })
+})
+
+var _ = ginkgo.Describe("[Compatibility: UDPWireV2]", func() {
f := framework.NewDefaultFramework()
ginkgo.BeforeEach(func() {
- supportsV2, knownVersion := baselineSupportsControlWireProtocolV2(compatCtx.BaselineVersion)
- if !knownVersion || !supportsV2 {
- ginkgo.Skip(fmt.Sprintf("baseline version %q does not have known wire protocol v2 support", compatCtx.BaselineVersion))
+ supportsV2, knownVersion := baselineSupportsUDPWireProtocolV2(compatCtx.BaselineVersion)
+ if !knownVersion {
+ ginkgo.Skip(fmt.Sprintf("baseline version %q has unknown ordinary UDP wire v2 framing support", compatCtx.BaselineVersion))
+ }
+ if !supportsV2 {
+ // v0.69.0 supports control v2 but still uses legacy ordinary UDP framing.
+ // Mixing that early v2 implementation with the v0.69.1 framing was not supported.
+ ginkgo.Skip(fmt.Sprintf("baseline %q predates ordinary UDP wire v2 framing (v0.69.1); default UDP is tested separately",
+ compatCtx.BaselineVersion))
}
})
- ginkgo.It("current frps falls back to JSON for baseline frpc", func() {
- portName := port.GenName("CompatBinaryUDPBaselineFRPC")
- clientConf := udpClientConfig("udp", portName, `transport.wireProtocol = "v2"`)
+ // The negotiated payload codec may be JSON or binary-v1; both use v2 framing.
+ ginkgo.It("current frps exchanges v2-framed UDP with baseline frpc", func() {
+ portName := port.GenName("CompatV2UDPBaselineFRPC")
+ clientConf := udpClientConfig(portName, `transport.wireProtocol = "v2"`)
f.RunProcessesWithBinaries(
compatCtx.CurrentFRPSPath,
compatCtx.BaselineFRPCPath,
@@ -214,9 +247,9 @@ var _ = ginkgo.Describe("[Compatibility: BinaryUDPPacket]", func() {
framework.NewRequestExpect(f).Protocol("udp").PortName(portName).Ensure()
})
- ginkgo.It("current frpc falls back to JSON for baseline frps", func() {
- portName := port.GenName("CompatBinaryUDPBaselineFRPS")
- clientConf := udpClientConfig("udp", portName, `transport.wireProtocol = "v2"`)
+ ginkgo.It("current frpc exchanges v2-framed UDP with baseline frps", func() {
+ portName := port.GenName("CompatV2UDPBaselineFRPS")
+ clientConf := udpClientConfig(portName, `transport.wireProtocol = "v2"`)
f.RunProcessesWithBinaries(
compatCtx.BaselineFRPSPath,
compatCtx.CurrentFRPCPath,
@@ -243,7 +276,7 @@ remotePort = {{ .%s }}
`, consts.PortServerName, extra, proxyName, framework.TCPEchoServerPort, remotePortName)
}
-func udpClientConfig(proxyName string, remotePortName string, extra string) string {
+func udpClientConfig(remotePortName string, extra string) string {
return fmt.Sprintf(`
serverAddr = "127.0.0.1"
serverPort = {{ .%s }}
@@ -252,11 +285,11 @@ log.level = "trace"
%s
[[proxies]]
-name = "%s"
+name = "udp"
type = "udp"
localPort = {{ .%s }}
remotePort = {{ .%s }}
-`, consts.PortServerName, extra, proxyName, framework.UDPEchoServerPort, remotePortName)
+`, consts.PortServerName, extra, framework.UDPEchoServerPort, remotePortName)
}
func expectProcessExit(p *process.Process, timeout time.Duration) {
@@ -268,6 +301,14 @@ func expectProcessExit(p *process.Process, timeout time.Duration) {
}
func baselineSupportsControlWireProtocolV2(version string) (supports bool, known bool) {
+ return baselineVersionAtLeast(version, [3]int{0, 69, 0})
+}
+
+func baselineSupportsUDPWireProtocolV2(version string) (supports bool, known bool) {
+ return baselineVersionAtLeast(version, [3]int{0, 69, 1})
+}
+
+func baselineVersionAtLeast(version string, minimum [3]int) (supports bool, known bool) {
version = strings.TrimPrefix(version, "v")
parts := strings.Split(version, ".")
if len(parts) != 3 {
@@ -287,7 +328,39 @@ func baselineSupportsControlWireProtocolV2(version string) (supports bool, known
return false, false
}
- return compareSemanticVersion(major, minor, patch, 0, 69, 0) >= 0, true
+ return compareSemanticVersion(major, minor, patch, minimum[0], minimum[1], minimum[2]) >= 0, true
+}
+
+func TestBaselineWireProtocolSupport(t *testing.T) {
+ for _, tc := range []struct {
+ version string
+ known bool
+ controlV2 bool
+ udpV2 bool
+ }{
+ {version: "0.61.0", known: true},
+ {version: "0.68.1", known: true},
+ {version: "0.69.0", known: true, controlV2: true},
+ {version: "v0.69.0", known: true, controlV2: true},
+ {version: "0.69.1", known: true, controlV2: true, udpV2: true},
+ {version: "v0.69.1", known: true, controlV2: true, udpV2: true},
+ {version: "0.70.0", known: true, controlV2: true, udpV2: true},
+ {version: "0.71.0", known: true, controlV2: true, udpV2: true},
+ {version: "1.0.0", known: true, controlV2: true, udpV2: true},
+ {version: "custom"},
+ {version: "0.69"},
+ {version: "0.69.x"},
+ {version: "0.69.1-rc.1"},
+ } {
+ t.Run(tc.version, func(t *testing.T) {
+ if supports, known := baselineSupportsControlWireProtocolV2(tc.version); supports != tc.controlV2 || known != tc.known {
+ t.Errorf("control v2 support = (%v, %v), want (%v, %v)", supports, known, tc.controlV2, tc.known)
+ }
+ if supports, known := baselineSupportsUDPWireProtocolV2(tc.version); supports != tc.udpV2 || known != tc.known {
+ t.Errorf("UDP v2 framing support = (%v, %v), want (%v, %v)", supports, known, tc.udpV2, tc.known)
+ }
+ })
+ }
}
func compareSemanticVersion(major, minor, patch int, baseMajor, baseMinor, basePatch int) int {
diff --git a/test/e2e/v1/features/bandwidth_limit.go b/test/e2e/v1/features/bandwidth_limit.go
index 11503adb..a975bfe5 100644
--- a/test/e2e/v1/features/bandwidth_limit.go
+++ b/test/e2e/v1/features/bandwidth_limit.go
@@ -2,6 +2,9 @@ package features
import (
"fmt"
+ "io"
+ "net"
+ "strconv"
"strings"
"time"
@@ -105,4 +108,82 @@ var _ = ginkgo.Describe("[Feature: Bandwidth Limit]", func() {
framework.ExpectTrue(duration.Seconds() > 8, "100Kb with 10KB limit, want > 8 seconds, but got %s", duration.String())
})
+
+ ginkgo.It("Server Bandwidth Limit closes work connection when user disconnects", func() {
+ newFunc := func() *plugin.Request {
+ var r plugin.Request
+ r.Content = &plugin.NewProxyContent{}
+ return &r
+ }
+ pluginPort := f.AllocPort()
+ handler := func(req *plugin.Request) *plugin.Response {
+ var ret plugin.Response
+ content := req.Content.(*plugin.NewProxyContent)
+ content.BandwidthLimit = "10MB"
+ content.BandwidthLimitMode = "server"
+ ret.Content = content
+ return &ret
+ }
+ pluginServer := pluginpkg.NewHTTPPluginServer(pluginPort, newFunc, handler, nil)
+
+ f.RunServer("", pluginServer)
+
+ serverConf := consts.DefaultServerConfig + fmt.Sprintf(`
+ [[httpPlugins]]
+ name = "test"
+ addr = "127.0.0.1:%d"
+ path = "/handler"
+ ops = ["NewProxy"]
+ `, pluginPort)
+ clientConf := consts.DefaultClientConfig
+
+ // The local server echoes the first message, then stays idle until the
+ // connection from frpc is closed.
+ localClosed := make(chan struct{})
+ localPort := f.AllocPort()
+ localServer := streamserver.New(streamserver.TCP, streamserver.WithBindPort(localPort),
+ streamserver.WithCustomHandler(func(c net.Conn) {
+ defer c.Close()
+ buf := make([]byte, 1024)
+ n, err := c.Read(buf)
+ if err != nil {
+ return
+ }
+ if _, err := c.Write(buf[:n]); err != nil {
+ return
+ }
+ _, _ = io.Copy(io.Discard, c)
+ close(localClosed)
+ }))
+ f.RunServer("", localServer)
+
+ remotePort := f.AllocPort()
+ clientConf += fmt.Sprintf(`
+ [[proxies]]
+ name = "tcp"
+ type = "tcp"
+ localPort = %d
+ remotePort = %d
+ `, localPort, remotePort)
+
+ f.RunProcesses(serverConf, []string{clientConf})
+
+ conn, err := net.Dial("tcp", net.JoinHostPort("127.0.0.1", strconv.Itoa(remotePort)))
+ framework.ExpectNoError(err)
+ _ = conn.SetDeadline(time.Now().Add(5 * time.Second))
+ _, err = conn.Write([]byte("ping"))
+ framework.ExpectNoError(err)
+ buf := make([]byte, 4)
+ _, err = io.ReadFull(conn, buf)
+ framework.ExpectNoError(err)
+ framework.ExpectEqual(string(buf), "ping")
+
+ conn.Close()
+
+ select {
+ case <-localClosed:
+ case <-time.After(5 * time.Second):
+ framework.Failf("local connection is not closed after user connection is closed")
+ }
+ })
})
diff --git a/test/e2e/v1/features/group.go b/test/e2e/v1/features/group.go
index 85b938a9..87270e7a 100644
--- a/test/e2e/v1/features/group.go
+++ b/test/e2e/v1/features/group.go
@@ -1,14 +1,20 @@
package features
import (
+ "context"
"crypto/tls"
+ "encoding/json"
"fmt"
+ "net/http"
+ "net/url"
"strconv"
"sync"
"time"
"github.com/onsi/ginkgo/v2"
+ "github.com/onsi/gomega"
+ "github.com/fatedier/frp/pkg/sdk/client"
"github.com/fatedier/frp/pkg/transport"
"github.com/fatedier/frp/test/e2e/framework"
"github.com/fatedier/frp/test/e2e/framework/consts"
@@ -17,6 +23,80 @@ import (
"github.com/fatedier/frp/test/e2e/pkg/request"
)
+func waitForProxyStatus(proxyClient *client.Client, proxyName, want string) error {
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+
+ ticker := time.NewTicker(25 * time.Millisecond)
+ defer ticker.Stop()
+
+ var lastStatus string
+ var lastErr error
+ for {
+ status, err := proxyClient.GetProxyStatus(ctx, proxyName)
+ if err == nil {
+ lastErr = nil
+ lastStatus = status.Status
+ if status.Status == want {
+ return nil
+ }
+ } else {
+ lastErr = err
+ }
+
+ select {
+ case <-ctx.Done():
+ if lastErr != nil {
+ return fmt.Errorf("timeout waiting for proxy %q status %q: last error: %w", proxyName, want, lastErr)
+ }
+ return fmt.Errorf("timeout waiting for proxy %q status %q: last status %q", proxyName, want, lastStatus)
+ case <-ticker.C:
+ }
+ }
+}
+
+func waitForSignal(signal <-chan struct{}) error {
+ timer := time.NewTimer(5 * time.Second)
+ defer timer.Stop()
+
+ select {
+ case <-signal:
+ return nil
+ case <-timer.C:
+ return fmt.Errorf("timeout waiting for backend health-check signal")
+ }
+}
+
+// Keep this longer than client/proxy.statusCheckInterval (3s). The wrapper's
+// health notification is deliberately non-blocking, so the E2E assertion must
+// also cover the fallback poll plus scheduling margin while recovery is gated.
+const proxyFallbackObservationWindow = 4 * time.Second
+
+func waitForServerProxyStatus(port int, proxyName, want string, timeout time.Duration) error {
+ return waitForLifecycleCondition(timeout, func() error {
+ body, err := getLifecycleEndpoint(port, "/api/proxies/"+url.PathEscape(proxyName))
+ if err != nil {
+ return err
+ }
+ var status struct {
+ Status string `json:"status"`
+ }
+ if err := json.Unmarshal([]byte(body), &status); err != nil {
+ return err
+ }
+ if status.Status != want {
+ return fmt.Errorf("frps proxy %q status %q, want %q", proxyName, status.Status, want)
+ }
+ return nil
+ })
+}
+
+type httpHealthStage struct {
+ healthy bool
+ signal chan<- struct{}
+ release <-chan struct{}
+}
+
var _ = ginkgo.Describe("[Feature: Group]", func() {
f := framework.NewDefaultFramework()
@@ -252,7 +332,10 @@ var _ = ginkgo.Describe("[Feature: Group]", func() {
ginkgo.Describe("Health Check", func() {
ginkgo.It("TCP", func() {
- serverConf := consts.DefaultServerConfig
+ dashboardPort := f.AllocPort()
+ serverConf := consts.DefaultServerConfig + fmt.Sprintf(`
+ webServer.port = %d
+ `, dashboardPort)
clientConf := consts.DefaultClientConfig
fooPort := f.AllocPort()
@@ -260,11 +343,17 @@ var _ = ginkgo.Describe("[Feature: Group]", func() {
f.RunServer("", fooServer)
barPort := f.AllocPort()
- barServer := streamserver.New(streamserver.TCP, streamserver.WithBindPort(barPort), streamserver.WithRespContent([]byte("bar")))
+ newBarServer := func() *streamserver.Server {
+ return streamserver.New(streamserver.TCP, streamserver.WithBindPort(barPort), streamserver.WithRespContent([]byte("bar")))
+ }
+ barServer := newBarServer()
f.RunServer("", barServer)
remotePort := f.AllocPort()
+ adminPort := f.AllocPort()
clientConf += fmt.Sprintf(`
+ webServer.port = %d
+
[[proxies]]
name = "foo"
type = "tcp"
@@ -284,11 +373,16 @@ var _ = ginkgo.Describe("[Feature: Group]", func() {
loadBalancer.groupKey = "123"
healthCheck.type = "tcp"
healthCheck.intervalSeconds = 1
- `, fooPort, remotePort, barPort, remotePort)
+ healthCheck.maxFailed = 3
+ `, adminPort, fooPort, remotePort, barPort, remotePort)
- _, clientProcesses := f.RunProcesses(serverConf, []string{clientConf})
+ f.RunProcesses(serverConf, []string{clientConf})
+ proxyClient := f.APIClientForFrpc(adminPort)
+ framework.ExpectNoError(waitForProxyStatus(proxyClient, "foo", "running"))
+ framework.ExpectNoError(waitForProxyStatus(proxyClient, "bar", "running"))
- // check foo and bar is ok
+ // Both requests traverse frps, the load-balancing group, frpc, and the
+ // corresponding local backend.
results := []string{}
for range 10 {
framework.NewRequestExpect(f).Port(remotePort).Ensure(validateFooBarResponse, func(resp *request.Response) bool {
@@ -298,33 +392,38 @@ var _ = ginkgo.Describe("[Feature: Group]", func() {
}
framework.ExpectContainElements(results, []string{"foo", "bar"})
- // close bar server, check foo is ok
- failedCount := clientProcesses[0].CountOutput("[bar] health check failed")
- barServer.Close()
- framework.ExpectNoError(clientProcesses[0].WaitForOutput("[bar] health check failed", failedCount+1, 5*time.Second))
- for range 10 {
- framework.NewRequestExpect(f).Port(remotePort).ExpectResp([]byte("foo")).Ensure()
- }
+ // frps removes the group listener before deleting the proxy from
+ // its manager. Its offline status is the removal barrier; frpc's
+ // local check-failed status alone does not acknowledge that work.
+ for range 2 {
+ framework.ExpectNoError(barServer.Close())
+ framework.ExpectNoError(waitForProxyStatus(proxyClient, "bar", "check failed"))
+ framework.ExpectNoError(waitForServerProxyStatus(dashboardPort, "bar", "offline", 5*time.Second))
+ for range 10 {
+ framework.NewRequestExpect(f).Port(remotePort).ExpectResp([]byte("foo")).Ensure()
+ }
- // resume bar server, check foo and bar is ok
- successCount := clientProcesses[0].CountOutput("[bar] health check success")
- f.RunServer("", barServer)
- framework.ExpectNoError(clientProcesses[0].WaitForOutput("[bar] health check success", successCount+1, 5*time.Second))
- results = []string{}
- for range 10 {
- framework.NewRequestExpect(f).Port(remotePort).Ensure(validateFooBarResponse, func(resp *request.Response) bool {
- results = append(results, string(resp.Content))
- return true
- })
+ barServer = newBarServer()
+ f.RunServer("", barServer)
+ framework.ExpectNoError(waitForProxyStatus(proxyClient, "bar", "running"))
+ results = []string{}
+ for range 10 {
+ framework.NewRequestExpect(f).Port(remotePort).Ensure(validateFooBarResponse, func(resp *request.Response) bool {
+ results = append(results, string(resp.Content))
+ return true
+ })
+ }
+ framework.ExpectContainElements(results, []string{"foo", "bar"})
}
- framework.ExpectContainElements(results, []string{"foo", "bar"})
})
ginkgo.It("HTTP", func() {
vhostPort := f.AllocPort()
+ dashboardPort := f.AllocPort()
serverConf := consts.DefaultServerConfig + fmt.Sprintf(`
vhostHTTPPort = %d
- `, vhostPort)
+ webServer.port = %d
+ `, vhostPort, dashboardPort)
clientConf := consts.DefaultClientConfig
fooPort := f.AllocPort()
@@ -332,10 +431,41 @@ var _ = ginkgo.Describe("[Feature: Group]", func() {
f.RunServer("", fooServer)
barPort := f.AllocPort()
- barServer := newHTTPServer(barPort, "bar")
+ var stageMu sync.RWMutex
+ stage := &httpHealthStage{healthy: true, signal: make(chan struct{}, 1)}
+ barServer := httpserver.New(
+ httpserver.WithBindPort(barPort),
+ httpserver.WithHandler(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ stageMu.RLock()
+ currentStage := stage
+ stageMu.RUnlock()
+
+ if r.URL.Path == "/healthz" {
+ if !currentStage.healthy {
+ w.WriteHeader(http.StatusServiceUnavailable)
+ }
+ select {
+ case currentStage.signal <- struct{}{}:
+ default:
+ }
+ if currentStage.release != nil {
+ select {
+ case <-currentStage.release:
+ case <-r.Context().Done():
+ return
+ }
+ }
+ return
+ }
+ _, _ = w.Write([]byte("bar"))
+ })),
+ )
f.RunServer("", barServer)
+ adminPort := f.AllocPort()
clientConf += fmt.Sprintf(`
+ webServer.port = %d
+
[[proxies]]
name = "foo"
type = "http"
@@ -356,12 +486,17 @@ var _ = ginkgo.Describe("[Feature: Group]", func() {
loadBalancer.groupKey = "123"
healthCheck.type = "http"
healthCheck.intervalSeconds = 1
+ healthCheck.maxFailed = 3
+ healthCheck.timeoutSeconds = 10
healthCheck.path = "/healthz"
- `, fooPort, barPort)
+ `, adminPort, fooPort, barPort)
- _, clientProcesses := f.RunProcesses(serverConf, []string{clientConf})
+ f.RunProcesses(serverConf, []string{clientConf})
+ proxyClient := f.APIClientForFrpc(adminPort)
+ framework.ExpectNoError(waitForProxyStatus(proxyClient, "foo", "running"))
+ framework.ExpectNoError(waitForProxyStatus(proxyClient, "bar", "running"))
- // send first HTTP request
+ // Ordinary requests traverse the real HTTP proxy path.
var contents []string
framework.NewRequestExpect(f).Port(vhostPort).
RequestModify(func(r *request.Request) {
@@ -371,8 +506,6 @@ var _ = ginkgo.Describe("[Feature: Group]", func() {
contents = append(contents, string(resp.Content))
return true
})
-
- // send second HTTP request, should be forwarded to another service
framework.NewRequestExpect(f).Port(vhostPort).
RequestModify(func(r *request.Request) {
r.HTTP().HTTPHost("example.com")
@@ -381,25 +514,83 @@ var _ = ginkgo.Describe("[Feature: Group]", func() {
contents = append(contents, string(resp.Content))
return true
})
-
framework.ExpectContainElements(contents, []string{"foo", "bar"})
- // check foo and bar is ok
+ // Every failed health response is gated by the test. This gives each
+ // stage a request-level barrier and avoids cumulative process logs.
+ runFailureRecovery := func() {
+ failureSignals := make(chan struct{}, 1)
+ release := make(chan struct{}, 1)
+ stageMu.Lock()
+ stage = &httpHealthStage{healthy: false, signal: failureSignals, release: release}
+ stageMu.Unlock()
+
+ framework.ExpectNoError(waitForSignal(failureSignals))
+ release <- struct{}{}
+ framework.ExpectNoError(waitForSignal(failureSignals))
+
+ recoverySignals := make(chan struct{}, 1)
+ recoveryRelease := make(chan struct{})
+ stageMu.Lock()
+ stage = &httpHealthStage{healthy: true, signal: recoverySignals, release: recoveryRelease}
+ stageMu.Unlock()
+ release <- struct{}{}
+
+ // The next request can only start after the worker consumed the
+ // second failed response. Hold this success response until the
+ // assertion finishes, so premature removal cannot be hidden by
+ // recovery. Consistently also gives the proxy worker time to act
+ // on an erroneous failure notification without retrying it away.
+ framework.ExpectNoError(waitForSignal(recoverySignals))
+ gomega.Consistently(func() string {
+ ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+ defer cancel()
+ status, err := proxyClient.GetProxyStatus(ctx, "bar")
+ if err != nil {
+ return err.Error()
+ }
+ return status.Status
+ }, proxyFallbackObservationWindow, 25*time.Millisecond).Should(gomega.Equal("running"))
+ close(recoveryRelease)
+ }
+ runFailureRecovery()
results := doFooBarHTTPRequest(vhostPort, "example.com")
framework.ExpectContainElements(results, []string{"foo", "bar"})
- // close bar server, check foo is ok
- failedCount := clientProcesses[0].CountOutput("[bar] health check failed")
- barServer.Close()
- framework.ExpectNoError(clientProcesses[0].WaitForOutput("[bar] health check failed", failedCount+1, 5*time.Second))
+ // Repeat the two-failure window to verify that recovery cleared the
+ // counter before another failure sequence began.
+ runFailureRecovery()
+ results = doFooBarHTTPRequest(vhostPort, "example.com")
+ framework.ExpectContainElements(results, []string{"foo", "bar"})
+
+ // Three failed responses reach MaxFailed and remove bar from the
+ // group; a healthy response then re-registers it.
+ failureSignals := make(chan struct{}, 1)
+ release := make(chan struct{}, 1)
+ stageMu.Lock()
+ stage = &httpHealthStage{healthy: false, signal: failureSignals, release: release}
+ stageMu.Unlock()
+ for range 3 {
+ framework.ExpectNoError(waitForSignal(failureSignals))
+ release <- struct{}{}
+ }
+ framework.ExpectNoError(waitForProxyStatus(proxyClient, "bar", "check failed"))
+ // HTTPProxy.Close unregisters the group route before frps reports
+ // offline. Only then assert that fresh data-plane requests use foo.
+ framework.ExpectNoError(waitForServerProxyStatus(dashboardPort, "bar", "offline", 5*time.Second))
results = doFooBarHTTPRequest(vhostPort, "example.com")
framework.ExpectContainElements(results, []string{"foo"})
framework.ExpectNotContainElements(results, []string{"bar"})
- // resume bar server, check foo and bar is ok
- successCount := clientProcesses[0].CountOutput("[bar] health check success")
- f.RunServer("", barServer)
- framework.ExpectNoError(clientProcesses[0].WaitForOutput("[bar] health check success", successCount+1, 5*time.Second))
+ recoverySignals := make(chan struct{}, 1)
+ stageMu.Lock()
+ stage = &httpHealthStage{healthy: true, signal: recoverySignals}
+ stageMu.Unlock()
+ // Release any additional failed request already waiting while the
+ // removal and data-plane assertions were in progress.
+ close(release)
+ framework.ExpectNoError(waitForSignal(recoverySignals))
+ framework.ExpectNoError(waitForProxyStatus(proxyClient, "bar", "running"))
results = doFooBarHTTPRequest(vhostPort, "example.com")
framework.ExpectContainElements(results, []string{"foo", "bar"})
})