fix(vhost): close pending connections when a route listener closes (#5592)

When an HTTPS or tcpmux route listener is closed after a connection has been
resolved to it but before the handoff, Muxer.handle clears the deadline and
then sends on the closed accept channel. The recovered panic returns without
closing the connection, so the peer waits until its own read timeout.

Add a done channel and sync.Once to Listener. Close closes done and is
idempotent; handle and Accept select on accept/done, and a pending connection
is closed when done is closed. Connections already handed to Accept keep their
current ownership. Remove the PanicToError-based control flow.

Cover the pending handoff with net.Pipe and real TCP tests, plus an idempotent
Close test.

Fixes #5573
This commit is contained in:
fatedier
2026-11-02 14:56:32 +08:00
committed by GitHub
parent 93b52799bb
commit e6e0aaf9d4
2 changed files with 133 additions and 10 deletions
+17 -10
View File
@@ -17,10 +17,9 @@ import (
"fmt" "fmt"
"net" "net"
"strings" "strings"
"sync"
"time" "time"
"github.com/fatedier/golib/errors"
"github.com/fatedier/frp/pkg/util/log" "github.com/fatedier/frp/pkg/util/log"
netpkg "github.com/fatedier/frp/pkg/util/net" netpkg "github.com/fatedier/frp/pkg/util/net"
"github.com/fatedier/frp/pkg/util/xlog" "github.com/fatedier/frp/pkg/util/xlog"
@@ -138,6 +137,7 @@ func (v *Muxer) Listen(ctx context.Context, cfg *RouteConfig) (l *Listener, err
password: cfg.Password, password: cfg.Password,
mux: v, mux: v,
accept: make(chan net.Conn), accept: make(chan net.Conn),
done: make(chan struct{}),
ctx: ctx, ctx: ctx,
} }
err = v.registryRouter.Add(cfg.Domain, cfg.Location, cfg.RouteByHTTPUser, l) err = v.registryRouter.Add(cfg.Domain, cfg.Location, cfg.RouteByHTTPUser, l)
@@ -215,11 +215,12 @@ func (v *Muxer) handle(c net.Conn) {
c = sConn c = sConn
xl.Debugf("new request host [%s] path [%s] httpUser [%s]", name, path, httpUser) xl.Debugf("new request host [%s] path [%s] httpUser [%s]", name, path, httpUser)
err = errors.PanicToError(func() { select {
l.accept <- c case l.accept <- c:
}) // The connection is now owned by the Accept caller.
if err != nil { case <-l.done:
xl.Warnf("listener is already closed, ignore this request") xl.Warnf("listener is already closed, ignore this request")
_ = c.Close()
} }
} }
@@ -232,13 +233,17 @@ type Listener struct {
password string password string
mux *Muxer // for closing Muxer mux *Muxer // for closing Muxer
accept chan net.Conn accept chan net.Conn
done chan struct{}
closeOnce sync.Once
ctx context.Context ctx context.Context
} }
func (l *Listener) Accept() (net.Conn, error) { func (l *Listener) Accept() (net.Conn, error) {
xl := xlog.FromContextSafe(l.ctx) xl := xlog.FromContextSafe(l.ctx)
conn, ok := <-l.accept var conn net.Conn
if !ok { select {
case conn = <-l.accept:
case <-l.done:
return nil, fmt.Errorf("listener closed") return nil, fmt.Errorf("listener closed")
} }
@@ -258,8 +263,10 @@ func (l *Listener) Accept() (net.Conn, error) {
} }
func (l *Listener) Close() error { func (l *Listener) Close() error {
l.mux.registryRouter.Del(l.name, l.location, l.routeByHTTPUser) l.closeOnce.Do(func() {
close(l.accept) l.mux.registryRouter.Del(l.name, l.location, l.routeByHTTPUser)
close(l.done)
})
return nil return nil
} }
+116
View File
@@ -0,0 +1,116 @@
// 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 vhost
import (
"context"
"net"
"testing"
"time"
)
func newCloseTestMuxer() *Muxer {
return &Muxer{
timeout: time.Second,
registryRouter: NewRouters(),
vhostFunc: func(c net.Conn) (net.Conn, map[string]string, error) {
return c, map[string]string{"Host": "example.com"}, nil
},
}
}
func testPendingHandoffIsClosed(t *testing.T, client, server net.Conn) {
t.Helper()
m := newCloseTestMuxer()
entered := make(chan struct{})
release := make(chan struct{})
m.SetSuccessHookFunc(func(net.Conn, map[string]string) error {
close(entered)
<-release
return nil
})
l, err := m.Listen(context.Background(), &RouteConfig{Domain: "example.com"})
if err != nil {
t.Fatal(err)
}
handled := make(chan struct{})
go func() {
defer close(handled)
m.handle(server)
}()
<-entered
if err := client.SetReadDeadline(time.Now().Add(300 * time.Millisecond)); err != nil {
t.Fatal(err)
}
// Close the route after resolution but before the handoff. The pending
// connection must be closed, not abandoned.
if err := l.Close(); err != nil {
t.Fatal(err)
}
close(release)
<-handled
if _, err := client.Read(make([]byte, 1)); err == nil {
t.Fatal("connection was not closed by the muxer")
} else if ne, ok := err.(net.Error); ok && ne.Timeout() {
t.Fatalf("connection was not closed by the muxer: %v", err)
}
}
func TestListenerCloseReleasesPendingConn(t *testing.T) {
client, server := net.Pipe()
defer client.Close()
testPendingHandoffIsClosed(t, client, server)
}
func TestListenerCloseReleasesPendingTCPConn(t *testing.T) {
ln, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatal(err)
}
defer ln.Close()
client, err := net.Dial("tcp", ln.Addr().String())
if err != nil {
t.Fatal(err)
}
defer client.Close()
server, err := ln.Accept()
if err != nil {
t.Fatal(err)
}
testPendingHandoffIsClosed(t, client, server)
}
func TestListenerCloseIsIdempotent(t *testing.T) {
m := newCloseTestMuxer()
l, err := m.Listen(context.Background(), &RouteConfig{Domain: "example.com"})
if err != nil {
t.Fatal(err)
}
if err := l.Close(); err != nil {
t.Fatal(err)
}
if err := l.Close(); err != nil {
t.Fatal(err)
}
if _, err := l.Accept(); err == nil {
t.Fatal("Accept after Close should return an error")
}
}