system stack: self-heal the TCP forwarder accept loop (sing-box-lx SPEC 040)
Upstream acceptLoop treats any Accept error as terminal and silently returns, leaving the stack alive but every new TCP SYN NAT-rewritten onto a dead port (instant RST) until a full restart. When the listener fd is closed out from under the stack (a stray close on a reused fd number from another runtime in the same process), all new TCP dies forever while UDP/QUIC/DNS keep working. - System.Close() now marks a deliberate shutdown first; acceptLoop still exits quietly on it. - Any other Accept error is logged (the errno names the killer path), the listener is recreated on the same address, the forwarder port is republished atomically, and the loop keeps serving. - If the rebind fails, the loop logs an error and gives up - no worse than upstream. - acceptRecoveries counter is kept as telemetry. tcpPort/tcpPort6 become atomic (written by the heal path, read from the tunLoop dispatch/NAT path); listener replacement is serialized against Close() with a mutex.
This commit is contained in:
Leadaxe
parent
da24acaf4d
commit
d31d20ba58
2 changed files with 244 additions and 39 deletions
140
stack_system.go
140
stack_system.go
|
|
@ -7,6 +7,8 @@ import (
|
|||
"net/netip"
|
||||
"os"
|
||||
"slices"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
|
|
@ -45,8 +47,16 @@ type System struct {
|
|||
icmpTimeout time.Duration
|
||||
tcpListener net.Listener
|
||||
tcpListener6 net.Listener
|
||||
tcpPort uint16
|
||||
tcpPort6 uint16
|
||||
// lx/040: ports are written by acceptLoop on self-heal relisten and read
|
||||
// concurrently from the tunLoop path (dispatch filter + NAT rewrite) —
|
||||
// they must be atomic. listenAccess serializes listener replacement
|
||||
// against Close(); closing marks a deliberate shutdown so acceptLoop can
|
||||
// tell it apart from the listener dying out from under the stack.
|
||||
tcpPort atomic.Uint32
|
||||
tcpPort6 atomic.Uint32
|
||||
closing atomic.Bool
|
||||
listenAccess sync.Mutex
|
||||
acceptRecoveries atomic.Uint32
|
||||
tcpNat *TCPNat
|
||||
udpNat *UDPNat
|
||||
udpNATOptions UDPNatOptions
|
||||
|
|
@ -124,10 +134,15 @@ func (s *System) ResetNetwork() {
|
|||
}
|
||||
|
||||
func (s *System) Close() error {
|
||||
// lx/040: mark the deliberate shutdown BEFORE closing the listeners so
|
||||
// acceptLoop exits quietly instead of treating it as a foreign kill.
|
||||
s.closing.Store(true)
|
||||
s.dispatcher.Close()
|
||||
if s.udpNat != nil {
|
||||
s.udpNat.Close()
|
||||
}
|
||||
s.listenAccess.Lock()
|
||||
defer s.listenAccess.Unlock()
|
||||
return common.Close(
|
||||
s.tcpListener,
|
||||
s.tcpListener6,
|
||||
|
|
@ -143,8 +158,10 @@ func (s *System) Start() error {
|
|||
return nil
|
||||
}
|
||||
|
||||
func (s *System) start() error {
|
||||
_ = fixWindowsFirewall()
|
||||
// lx/040: TCP forwarder bind, shared by start() and the acceptLoop self-heal
|
||||
// relisten path. isIPv6 selects the address family; the bind-to-interface
|
||||
// Control and the EADDRNOTAVAIL retry loop match the original start() code.
|
||||
func (s *System) listenTCP(isIPv6 bool) (net.Listener, error) {
|
||||
var listener net.ListenConfig
|
||||
if s.bindInterface {
|
||||
listener.Control = control.Append(listener.Control, func(network, address string, conn syscall.RawConn) error {
|
||||
|
|
@ -155,37 +172,50 @@ func (s *System) start() error {
|
|||
return nil
|
||||
})
|
||||
}
|
||||
var tcpListener net.Listener
|
||||
var err error
|
||||
if s.inet4NextAddress.IsValid() {
|
||||
network := "tcp4"
|
||||
address := s.inet4Address
|
||||
if isIPv6 {
|
||||
network = "tcp6"
|
||||
address = s.inet6Address
|
||||
}
|
||||
var (
|
||||
tcpListener net.Listener
|
||||
err error
|
||||
)
|
||||
for range 3 {
|
||||
tcpListener, err = listenNetworkNamespace(s.ctx, s.netNs, listener, "tcp4", net.JoinHostPort(s.inet4Address.String(), "0"))
|
||||
tcpListener, err = listenNetworkNamespace(s.ctx, s.netNs, listener, network, net.JoinHostPort(address.String(), "0"))
|
||||
if !retryableListenError(err) {
|
||||
break
|
||||
}
|
||||
time.Sleep(time.Second)
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return tcpListener, nil
|
||||
}
|
||||
|
||||
func (s *System) start() error {
|
||||
_ = fixWindowsFirewall()
|
||||
var tcpListener net.Listener
|
||||
var err error
|
||||
if s.inet4NextAddress.IsValid() {
|
||||
tcpListener, err = s.listenTCP(false)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
s.tcpListener = tcpListener
|
||||
s.tcpPort = M.SocksaddrFromNet(tcpListener.Addr()).Port
|
||||
go s.acceptLoop(tcpListener)
|
||||
s.tcpPort.Store(uint32(M.SocksaddrFromNet(tcpListener.Addr()).Port))
|
||||
go s.acceptLoop(tcpListener, false)
|
||||
}
|
||||
if s.inet6NextAddress.IsValid() {
|
||||
for range 3 {
|
||||
tcpListener, err = listenNetworkNamespace(s.ctx, s.netNs, listener, "tcp6", net.JoinHostPort(s.inet6Address.String(), "0"))
|
||||
if !retryableListenError(err) {
|
||||
break
|
||||
}
|
||||
time.Sleep(time.Second)
|
||||
}
|
||||
tcpListener, err = s.listenTCP(true)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
s.tcpListener6 = tcpListener
|
||||
s.tcpPort6 = M.SocksaddrFromNet(tcpListener.Addr()).Port
|
||||
go s.acceptLoop(tcpListener)
|
||||
s.tcpPort6.Store(uint32(M.SocksaddrFromNet(tcpListener.Addr()).Port))
|
||||
go s.acceptLoop(tcpListener, true)
|
||||
}
|
||||
s.tcpNat = NewNat(s.ctx, s.udpTimeout)
|
||||
udpNATOptions := s.udpNATOptions
|
||||
|
|
@ -367,12 +397,29 @@ func (s *System) processPacket(packet []byte) bool {
|
|||
return writeBack
|
||||
}
|
||||
|
||||
func (s *System) acceptLoop(listener net.Listener) {
|
||||
func (s *System) acceptLoop(listener net.Listener, isIPv6 bool) {
|
||||
for {
|
||||
conn, err := listener.Accept()
|
||||
if err != nil {
|
||||
// lx/040 (SPECS/TASKS/040): upstream silently returns on ANY Accept
|
||||
// error, leaving the stack alive but every new TCP SYN NAT-rewritten
|
||||
// onto a dead port (instant RST) until a VPN restart — the LxBox §047
|
||||
// "browser dead, QUIC alive" failure. A deliberate System.Close is the
|
||||
// only quiet exit; anything else means the listener died out from
|
||||
// under us (e.g. a foreign close on a reused fd number from the
|
||||
// Java side of the shared Android process) — log it (the errno names
|
||||
// the killer) and recreate the listener.
|
||||
if s.closing.Load() {
|
||||
return
|
||||
}
|
||||
newListener, healErr := s.healListener(listener, isIPv6, err)
|
||||
if healErr != nil {
|
||||
s.logger.Error("system stack: tcp", ipVersionSuffix(isIPv6), " accept loop died: ", err, "; relisten failed: ", healErr)
|
||||
return
|
||||
}
|
||||
listener = newListener
|
||||
continue
|
||||
}
|
||||
connPort := M.SocksaddrFromNet(conn.RemoteAddr()).Port
|
||||
session := s.tcpNat.LookupBack(connPort)
|
||||
if session == nil {
|
||||
|
|
@ -383,6 +430,47 @@ func (s *System) acceptLoop(listener net.Listener) {
|
|||
}
|
||||
}
|
||||
|
||||
// lx/040: recreate a TCP forwarder listener that died out from under the
|
||||
// stack. Returns the replacement listener after publishing it (listener field
|
||||
// + atomic port) under listenAccess, or an error if the stack is closing or
|
||||
// the bind failed.
|
||||
func (s *System) healListener(dead net.Listener, isIPv6 bool, cause error) (net.Listener, error) {
|
||||
port := &s.tcpPort
|
||||
if isIPv6 {
|
||||
port = &s.tcpPort6
|
||||
}
|
||||
oldPort := port.Load()
|
||||
s.logger.Warn("system stack: tcp", ipVersionSuffix(isIPv6), " listener (port ", oldPort, ") accept failed: ", cause, " — recreating listener")
|
||||
_ = dead.Close() // release netpoll state; harmless if already closed
|
||||
newListener, err := s.listenTCP(isIPv6)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
s.listenAccess.Lock()
|
||||
defer s.listenAccess.Unlock()
|
||||
if s.closing.Load() {
|
||||
_ = newListener.Close()
|
||||
return nil, net.ErrClosed
|
||||
}
|
||||
if isIPv6 {
|
||||
s.tcpListener6 = newListener
|
||||
} else {
|
||||
s.tcpListener = newListener
|
||||
}
|
||||
newPort := uint32(M.SocksaddrFromNet(newListener.Addr()).Port)
|
||||
port.Store(newPort)
|
||||
recoveries := s.acceptRecoveries.Add(1)
|
||||
s.logger.Warn("system stack: tcp", ipVersionSuffix(isIPv6), " listener recreated (port ", oldPort, " → ", newPort, ", recoveries: ", recoveries, ")")
|
||||
return newListener, nil
|
||||
}
|
||||
|
||||
func ipVersionSuffix(isIPv6 bool) string {
|
||||
if isIPv6 {
|
||||
return "6"
|
||||
}
|
||||
return "4"
|
||||
}
|
||||
|
||||
func (s *System) dispatchIPv4(ipHdr header.IPv4, destination netip.Addr) bool {
|
||||
switch ipHdr.TransportProtocol() {
|
||||
case header.TCPProtocolNumber:
|
||||
|
|
@ -392,7 +480,7 @@ func (s *System) dispatchIPv4(ipHdr header.IPv4, destination netip.Addr) bool {
|
|||
if ipHdr.SourceAddr() == s.inet4Address &&
|
||||
ipHdr.FragmentOffset() == 0 &&
|
||||
len(ipHdr.Payload()) >= header.TCPMinimumSize &&
|
||||
header.TCP(ipHdr.Payload()).SourcePort() == s.tcpPort {
|
||||
header.TCP(ipHdr.Payload()).SourcePort() == uint16(s.tcpPort.Load()) {
|
||||
return false
|
||||
}
|
||||
case header.ICMPv4ProtocolNumber:
|
||||
|
|
@ -411,7 +499,7 @@ func (s *System) dispatchIPv6(ipHdr header.IPv6, destination netip.Addr) bool {
|
|||
}
|
||||
if ipHdr.SourceAddr() == s.inet6Address &&
|
||||
len(ipHdr.Payload()) >= header.TCPMinimumSize &&
|
||||
header.TCP(ipHdr.Payload()).SourcePort() == s.tcpPort6 {
|
||||
header.TCP(ipHdr.Payload()).SourcePort() == uint16(s.tcpPort6.Load()) {
|
||||
return false
|
||||
}
|
||||
case header.ICMPv6ProtocolNumber:
|
||||
|
|
@ -475,7 +563,7 @@ func (s *System) processIPv4TCP(ipHdr header.IPv4, tcpHdr header.TCP) (bool, err
|
|||
destination := netip.AddrPortFrom(ipHdr.DestinationAddr(), tcpHdr.DestinationPort())
|
||||
if !destination.Addr().IsGlobalUnicast() {
|
||||
return false, nil
|
||||
} else if source.Addr() == s.inet4Address && source.Port() == s.tcpPort {
|
||||
} else if source.Addr() == s.inet4Address && source.Port() == uint16(s.tcpPort.Load()) {
|
||||
session := s.tcpNat.LookupBack(destination.Port())
|
||||
if session == nil {
|
||||
return false, E.New("ipv4: tcp: session not found: ", destination.Port())
|
||||
|
|
@ -501,7 +589,7 @@ func (s *System) processIPv4TCP(ipHdr header.IPv4, tcpHdr header.TCP) (bool, err
|
|||
}
|
||||
rewriteIPv4TCP(ipHdr, tcpHdr, s.txChecksumOffload,
|
||||
s.inet4NextAddress, natPort, true,
|
||||
s.inet4Address, s.tcpPort, true)
|
||||
s.inet4Address, uint16(s.tcpPort.Load()), true)
|
||||
}
|
||||
}
|
||||
return true, nil
|
||||
|
|
@ -512,7 +600,7 @@ func (s *System) processIPv6TCP(ipHdr header.IPv6, tcpHdr header.TCP) (bool, err
|
|||
destination := netip.AddrPortFrom(ipHdr.DestinationAddr(), tcpHdr.DestinationPort())
|
||||
if !destination.Addr().IsGlobalUnicast() {
|
||||
return false, nil
|
||||
} else if source.Addr() == s.inet6Address && source.Port() == s.tcpPort6 {
|
||||
} else if source.Addr() == s.inet6Address && source.Port() == uint16(s.tcpPort6.Load()) {
|
||||
session := s.tcpNat.LookupBack(destination.Port())
|
||||
if session == nil {
|
||||
return false, E.New("ipv6: tcp: session not found: ", destination.Port())
|
||||
|
|
@ -538,7 +626,7 @@ func (s *System) processIPv6TCP(ipHdr header.IPv6, tcpHdr header.TCP) (bool, err
|
|||
}
|
||||
rewriteIPv6TCP(ipHdr, tcpHdr, s.txChecksumOffload,
|
||||
s.inet6NextAddress, natPort, true,
|
||||
s.inet6Address, s.tcpPort6, true)
|
||||
s.inet6Address, uint16(s.tcpPort6.Load()), true)
|
||||
}
|
||||
}
|
||||
return true, nil
|
||||
|
|
|
|||
117
stack_system_selfheal_test.go
Normal file
117
stack_system_selfheal_test.go
Normal file
|
|
@ -0,0 +1,117 @@
|
|||
package tun
|
||||
|
||||
// lx/040 (SPECS/TASKS/040-SINGTUN_ACCEPTLOOP_SELFHEAL): acceptLoop self-heal.
|
||||
//
|
||||
// Red/green против апстрима 2d9b8aed5fe2: там acceptLoop(listener) при любой
|
||||
// ошибке Accept молча выходит навсегда — восстановления нет, порт не меняется,
|
||||
// новый connect вечно бьётся в мёртвый сокет. Для red-прогона на чистом
|
||||
// апстрим-чекауте достаточно адаптировать хелперы ниже (currentTCPPort →
|
||||
// s.tcpPort, spawnAcceptLoop → go s.acceptLoop(ln)): тест упадёт по таймауту
|
||||
// ожидания восстановления.
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net"
|
||||
"net/netip"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/sagernet/sing/common/logger"
|
||||
)
|
||||
|
||||
func newSelfHealTestSystem(t *testing.T) *System {
|
||||
t.Helper()
|
||||
s := &System{
|
||||
ctx: context.Background(),
|
||||
logger: logger.NOP(),
|
||||
inet4Address: netip.MustParseAddr("127.0.0.1"),
|
||||
udpTimeout: time.Minute,
|
||||
}
|
||||
s.tcpNat = NewNat(s.ctx, s.udpTimeout)
|
||||
ln, err := s.listenTCP(false)
|
||||
if err != nil {
|
||||
t.Fatalf("listenTCP: %v", err)
|
||||
}
|
||||
s.tcpListener = ln
|
||||
s.tcpPort.Store(uint32(ln.Addr().(*net.TCPAddr).Port))
|
||||
spawnAcceptLoop(s, ln)
|
||||
return s
|
||||
}
|
||||
|
||||
func currentTCPPort(s *System) uint32 {
|
||||
return s.tcpPort.Load()
|
||||
}
|
||||
|
||||
func spawnAcceptLoop(s *System, ln net.Listener) {
|
||||
go s.acceptLoop(ln, false)
|
||||
}
|
||||
|
||||
func dialForwarder(t *testing.T, port uint32) error {
|
||||
t.Helper()
|
||||
conn, err := net.DialTimeout("tcp4", fmt.Sprintf("127.0.0.1:%d", port), time.Second)
|
||||
if err == nil {
|
||||
_ = conn.Close()
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
// Убийство listener'а мимо System.Close (эмуляция чужого close по
|
||||
// переиспользованному fd-номеру) должно приводить к пересозданию listener'а
|
||||
// и продолжению приёма TCP, а не к вечной смерти петли.
|
||||
func TestSystemAcceptLoopSelfHeal(t *testing.T) {
|
||||
s := newSelfHealTestSystem(t)
|
||||
oldPort := currentTCPPort(s)
|
||||
|
||||
if err := dialForwarder(t, oldPort); err != nil {
|
||||
t.Fatalf("healthy listener refused connect: %v", err)
|
||||
}
|
||||
|
||||
// Убить listener из-под стека: closing НЕ выставлен.
|
||||
_ = s.tcpListener.Close()
|
||||
|
||||
deadline := time.Now().Add(5 * time.Second)
|
||||
healed := false
|
||||
for time.Now().Before(deadline) {
|
||||
if s.acceptRecoveries.Load() > 0 {
|
||||
healed = true
|
||||
break
|
||||
}
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
if !healed {
|
||||
t.Fatalf("acceptLoop did not recover within 5s (upstream behavior: silent permanent death)")
|
||||
}
|
||||
|
||||
newPort := currentTCPPort(s)
|
||||
if newPort == oldPort {
|
||||
t.Fatalf("recovered port equals dead port %d — relisten did not publish a new port", oldPort)
|
||||
}
|
||||
if err := dialForwarder(t, newPort); err != nil {
|
||||
t.Fatalf("connect to recreated listener (port %d) failed: %v", newPort, err)
|
||||
}
|
||||
if got := s.acceptRecoveries.Load(); got != 1 {
|
||||
t.Fatalf("acceptRecoveries = %d, want 1", got)
|
||||
}
|
||||
|
||||
s.closing.Store(true)
|
||||
_ = s.tcpListener.Close()
|
||||
}
|
||||
|
||||
// Штатное закрытие (closing выставлен, как это делает System.Close) обязано
|
||||
// оставаться тихим: без пересозданий и без роста счётчика.
|
||||
func TestSystemAcceptLoopQuietOnClose(t *testing.T) {
|
||||
s := newSelfHealTestSystem(t)
|
||||
oldPort := currentTCPPort(s)
|
||||
|
||||
s.closing.Store(true)
|
||||
_ = s.tcpListener.Close()
|
||||
|
||||
time.Sleep(300 * time.Millisecond)
|
||||
if got := s.acceptRecoveries.Load(); got != 0 {
|
||||
t.Fatalf("deliberate close triggered %d recoveries, want 0", got)
|
||||
}
|
||||
if port := currentTCPPort(s); port != oldPort {
|
||||
t.Fatalf("deliberate close changed port %d → %d", oldPort, port)
|
||||
}
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue