mirror of
https://github.com/fatedier/frp.git
synced 2026-08-30 20:55:57 +08:00
vnet: fix route cleanup and reconnect backoff (#5492)
This commit is contained in:
@@ -48,6 +48,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)
|
||||
|
||||
@@ -152,8 +157,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<<uint(p.consecutiveErrors-1)), 300*time.Second)
|
||||
reconnectDelay = virtualNetReconnectDelay(p.consecutiveErrors)
|
||||
} else {
|
||||
// Reset consecutive errors on successful connection
|
||||
if p.consecutiveErrors > 0 {
|
||||
@@ -182,6 +186,18 @@ func (p *VirtualNetPlugin) run() {
|
||||
}
|
||||
}
|
||||
|
||||
// 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<<uint(consecutiveErrors-1))
|
||||
}
|
||||
|
||||
// cleanupControllerConn closes the current controllerConn (if it exists) under lock.
|
||||
func (p *VirtualNetPlugin) cleanupControllerConn(xl *xlog.Logger) {
|
||||
p.mu.Lock()
|
||||
@@ -202,9 +218,12 @@ func (p *VirtualNetPlugin) Close() error {
|
||||
// Signal the run loop goroutine to stop.
|
||||
p.cancel()
|
||||
|
||||
// Unregister the route from the controller.
|
||||
if p.pluginCtx.VnetController != nil {
|
||||
p.pluginCtx.VnetController.UnregisterClientRoute(p.pluginCtx.Name)
|
||||
// Unregister the route only if it is still owned by this plugin instance.
|
||||
p.mu.Lock()
|
||||
controllerConn := p.controllerConn
|
||||
p.mu.Unlock()
|
||||
if p.pluginCtx.VnetController != nil && controllerConn != nil &&
|
||||
p.pluginCtx.VnetController.UnregisterClientRoute(p.pluginCtx.Name, controllerConn) {
|
||||
xl.Infof("unregistered client route for visitor [%s]", p.pluginCtx.Name)
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,47 @@
|
||||
// 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.
|
||||
|
||||
//go:build !frps
|
||||
|
||||
package visitor
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// TestVirtualNetReconnectDelay verifies the documented exponential backoff and
|
||||
// ensures large error counts remain capped instead of overflowing to zero.
|
||||
func TestVirtualNetReconnectDelay(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
consecutiveErrors int
|
||||
want time.Duration
|
||||
}{
|
||||
{name: "first error", consecutiveErrors: 1, want: 60 * time.Second},
|
||||
{name: "second error", consecutiveErrors: 2, want: 120 * time.Second},
|
||||
{name: "third error", consecutiveErrors: 3, want: 240 * time.Second},
|
||||
{name: "fourth error", consecutiveErrors: 4, want: 300 * time.Second},
|
||||
{name: "shift width boundary", consecutiveErrors: 64, want: 300 * time.Second},
|
||||
{name: "observed retry storm", consecutiveErrors: 329769, want: 300 * time.Second},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
require.Equal(t, tt.want, virtualNetReconnectDelay(tt.consecutiveErrors))
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -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) {
|
||||
|
||||
@@ -0,0 +1,57 @@
|
||||
// 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)
|
||||
|
||||
require.True(controller.UnregisterClientRoute("vnet-visitor", replacementConn))
|
||||
_, err = controller.clientRouter.findConn(net.ParseIP("10.1.0.1"))
|
||||
require.Error(err)
|
||||
}
|
||||
Reference in New Issue
Block a user