mirror of
https://github.com/MHSanaei/3x-ui.git
synced 2026-11-11 07:34:59 +03:00
fix(amneziawg): keep a port forward accepting after a failed accept
A TCP port forward's accept loop returned on any accept error, leaving the socket bound and the forward registered as open, so nothing served the port again until a panel restart; briefly pushing the panel to its open-file limit was enough. Forward and egress accept loops now log, wait 100 ms (or until closed) and keep accepting.
This commit is contained in:
@@ -218,6 +218,9 @@ func (s *socks5EgressServer) acceptLoop(ln net.Listener, closing chan struct{})
|
|||||||
default:
|
default:
|
||||||
}
|
}
|
||||||
logger.Warningf("amneziawgnet: egress accept: %v", err)
|
logger.Warningf("amneziawgnet: egress accept: %v", err)
|
||||||
|
if !waitAcceptRetry(closing) {
|
||||||
|
return
|
||||||
|
}
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
select {
|
select {
|
||||||
|
|||||||
@@ -270,6 +270,21 @@ func openPortForwardListener(gstack *stack.Stack, inboundID int, key portForward
|
|||||||
|
|
||||||
const portForwardDialTimeout = 10 * time.Second
|
const portForwardDialTimeout = 10 * time.Second
|
||||||
|
|
||||||
|
// acceptRetryDelay paces an accept loop's retries, so an error that lasts
|
||||||
|
// (EMFILE, until descriptors free up) neither spins it nor floods the log.
|
||||||
|
const acceptRetryDelay = 100 * time.Millisecond
|
||||||
|
|
||||||
|
// waitAcceptRetry sleeps acceptRetryDelay after a failed Accept; false means
|
||||||
|
// closing fired meanwhile and the loop should return instead of retrying.
|
||||||
|
func waitAcceptRetry(closing <-chan struct{}) bool {
|
||||||
|
select {
|
||||||
|
case <-closing:
|
||||||
|
return false
|
||||||
|
case <-time.After(acceptRetryDelay):
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// tunnelNetwork returns the gVisor network protocol number matching addr's
|
// tunnelNetwork returns the gVisor network protocol number matching addr's
|
||||||
// address family, for dialing toward it inside the embedded stack.
|
// address family, for dialing toward it inside the embedded stack.
|
||||||
func tunnelNetwork(addr netip.Addr) tcpip.NetworkProtocolNumber {
|
func tunnelNetwork(addr netip.Addr) tcpip.NetworkProtocolNumber {
|
||||||
@@ -325,7 +340,12 @@ func (l *tcpForwardListener) acceptLoop(gstack *stack.Stack, inboundID int, key
|
|||||||
default:
|
default:
|
||||||
}
|
}
|
||||||
logger.Warningf("amneziawgnet: port-forward: inbound %d peer %q: accept tcp :%d: %v", inboundID, key.email, key.port, err)
|
logger.Warningf("amneziawgnet: port-forward: inbound %d peer %q: accept tcp :%d: %v", inboundID, key.email, key.port, err)
|
||||||
return
|
// Keep accepting: returning would strand a bound socket nobody
|
||||||
|
// serves, under a key Reconcile still sees as open and never reopens.
|
||||||
|
if !waitAcceptRetry(l.closing) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
continue
|
||||||
}
|
}
|
||||||
go relayTCPForward(gstack, conn, inboundID, key, target)
|
go relayTCPForward(gstack, conn, inboundID, key, target)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -2,11 +2,14 @@ package amneziawgnet
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"io"
|
"io"
|
||||||
"net"
|
"net"
|
||||||
"net/netip"
|
"net/netip"
|
||||||
"sync"
|
"sync"
|
||||||
|
"sync/atomic"
|
||||||
|
"syscall"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
@@ -222,6 +225,43 @@ func TestPortForwardSetReconcileSurvivesPreBoundPort(t *testing.T) {
|
|||||||
set.Close()
|
set.Close()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// failFirstAccept fails the first Accept the way the kernel does once the
|
||||||
|
// process is at RLIMIT_NOFILE, then behaves like the listener it wraps.
|
||||||
|
type failFirstAccept struct {
|
||||||
|
net.Listener
|
||||||
|
failed atomic.Bool
|
||||||
|
}
|
||||||
|
|
||||||
|
func (l *failFirstAccept) Accept() (net.Conn, error) {
|
||||||
|
if l.failed.CompareAndSwap(false, true) {
|
||||||
|
return nil, &net.OpError{Op: "accept", Net: "tcp", Addr: l.Addr(), Err: syscall.EMFILE}
|
||||||
|
}
|
||||||
|
return l.Listener.Accept()
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestTCPForwardAcceptLoopSurvivesTransientError(t *testing.T) {
|
||||||
|
host, err := net.Listen("tcp", "127.0.0.1:0")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("listen: %v", err)
|
||||||
|
}
|
||||||
|
l := &tcpForwardListener{ln: &failFirstAccept{Listener: host}, closing: make(chan struct{})}
|
||||||
|
defer l.Close()
|
||||||
|
noTarget := func(string) (netip.Addr, bool) { return netip.Addr{}, false }
|
||||||
|
go l.acceptLoop(newTestStack(t, "10.211.4.1"), 505, portForwardKey{email: "a@x", proto: tcpForward}, noTarget)
|
||||||
|
|
||||||
|
conn, err := net.DialTimeout("tcp", host.Addr().String(), time.Second)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("dial: %v", err)
|
||||||
|
}
|
||||||
|
defer conn.Close()
|
||||||
|
// relayTCPForward closes a connection it has no target for, so EOF proves
|
||||||
|
// the loop accepted it; a dead loop leaves it queued in the backlog.
|
||||||
|
_ = conn.SetReadDeadline(time.Now().Add(3 * time.Second))
|
||||||
|
if _, err := conn.Read(make([]byte, 1)); !errors.Is(err, io.EOF) {
|
||||||
|
t.Fatalf("read after a transient accept error = %v, want io.EOF from an accepted connection", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// --- Real round trip: a genuine amneziawg-go client handshakes against a
|
// --- Real round trip: a genuine amneziawg-go client handshakes against a
|
||||||
// real server Device, PortForwardSet opens a real host listener, and a real
|
// real server Device, PortForwardSet opens a real host listener, and a real
|
||||||
// external-side dial (this test's own process) round-trips bytes through
|
// external-side dial (this test's own process) round-trips bytes through
|
||||||
|
|||||||
Reference in New Issue
Block a user