mirror of
https://github.com/go-gost/x.git
synced 2026-10-19 16:16:32 +00:00
fix(handler/tun): end a p2p dial when the echoes stop
A p2p dial's stream survives the peer session under it being rebuilt, so the link can go silently dead: the far side re-creates its session while this side's stream becomes a zombie — writes succeed into the void and nothing reports an error. firstInboundTimeout was a one-shot, cleared by the first inbound byte and never re-armed, so such a dial stayed "running" forever: on 2026-10-17 an Android spoke sat in a two-way blackhole for 17 minutes after the hub killed and rebuilt its relay session alone (derp peer-gone kill), while both ends reported the path up; only a manual entrypoint restart healed it. The registration refresh is echoed by the hub while the route to this device is real, so silence is measurable. A p2p dial now re-arms the read deadline to p2pInboundTimeout on every inbound byte; when it expires the dial ends and the caller redials, which re-registers and heals a hub-side rebuild — the same recovery a manual restart performs (observed as "peer relay session up … client=true"). Socket dials keep the one-shot semantics: their transport reports the far side gone by itself.
This commit is contained in:
+38
-7
@@ -31,11 +31,26 @@ var (
|
||||
// keepalive period is configured. A hub that never answers — not started,
|
||||
// or silently refusing — would otherwise leave the spoke "running" forever
|
||||
// on a link that will never deliver. Expiring makes the caller redial and
|
||||
// re-register, which is also how a fixed hub recovers on its own. Any
|
||||
// inbound byte clears it: from then on the link has proven itself alive.
|
||||
// re-register, which is also how a fixed hub recovers on its own. On a
|
||||
// socket dial any inbound byte clears it, from then on the link has proven
|
||||
// itself alive; a p2p dial instead turns it into the rolling bound below,
|
||||
// because its link can silently outlive the far side's session.
|
||||
// A variable, not a constant, so tests can shrink the wait.
|
||||
var firstInboundTimeout = 30 * time.Second
|
||||
|
||||
// p2pInboundTimeout bounds the silence on a p2p dial once the first inbound
|
||||
// byte has proved the link alive. The registration handshake repeats every
|
||||
// p2pRegisterInterval and the hub echoes each one back over the same stream,
|
||||
// so ongoing silence means the hub stopped answering this dial: its session
|
||||
// was rebuilt and this stream is a zombie (2026-10-17: 17 minutes of two-way
|
||||
// blackhole while both ends read the path up), or the route was reclaimed.
|
||||
// Nothing else ends such a dial — writes succeed into the void and no
|
||||
// transport reports an error — so a rolling no-inbound bound is the only
|
||||
// signal. Ending it makes the caller redial, and the redial is what
|
||||
// re-registers and heals a hub-side rebuild. Three missed echoes at the
|
||||
// default 10s cadence. A variable, not a constant, so tests can shrink it.
|
||||
var p2pInboundTimeout = 30 * time.Second
|
||||
|
||||
// p2pRegisterInterval is how often a p2p dial re-sends the registration
|
||||
// handshake when no keepalive period is configured. A variable, not a
|
||||
// constant, so tests can shrink the wait (same as firstInboundTimeout).
|
||||
@@ -73,7 +88,10 @@ func (h *tunHandler) handleClient(ctx context.Context, conn net.Conn, network st
|
||||
// the route to this device while the dial still reads as alive —
|
||||
// and then every return packet is dropped for want of it, at both
|
||||
// ends silently. keepalive() therefore refreshes on a p2p dial
|
||||
// even without a period configured (see p2pRegisterInterval).
|
||||
// even without a period configured (see p2pRegisterInterval),
|
||||
// and that refresh cuts both ways: the hub's echo of it is this
|
||||
// side's proof the route is real, so a hub that stopped answering
|
||||
// ends the dial on p2pInboundTimeout instead of leaving it mute.
|
||||
iterCtx, iterCancel := context.WithCancel(ctx)
|
||||
defer iterCancel()
|
||||
|
||||
@@ -83,7 +101,7 @@ func (h *tunHandler) handleClient(ctx context.Context, conn net.Conn, network st
|
||||
go h.probeDevice(iterCtx, conn, ips)
|
||||
}
|
||||
|
||||
return h.transportClient(ctx, conn, cc, log)
|
||||
return h.transportClient(ctx, conn, cc, network == "ip", log)
|
||||
}()
|
||||
if errors.Is(err, ErrTun) {
|
||||
return err
|
||||
@@ -153,7 +171,7 @@ func (h *tunHandler) keepalive(ctx context.Context, conn net.Conn, ips []net.IP,
|
||||
}
|
||||
}
|
||||
|
||||
func (h *tunHandler) transportClient(ctx context.Context, tun io.ReadWriter, conn net.Conn, log logger.Logger) error {
|
||||
func (h *tunHandler) transportClient(ctx context.Context, tun io.ReadWriter, conn net.Conn, p2p bool, log logger.Logger) error {
|
||||
c, cancel := context.WithCancel(ctx)
|
||||
defer cancel()
|
||||
|
||||
@@ -227,8 +245,11 @@ func (h *tunHandler) transportClient(ctx context.Context, tun io.ReadWriter, con
|
||||
// forever is exactly the silent-"running" this guards against. With
|
||||
// a period the window matches the handshake's own 3×period; without
|
||||
// one the fixed one-shot applies until the first inbound clears it.
|
||||
// (keepalive() sets the same 3×period deadline for its path; the
|
||||
// two agree, and this one is what the read loop below relies on.)
|
||||
// On a p2p dial that first byte re-arms the bound instead of
|
||||
// clearing it (p2pInboundTimeout): its stream can silently outlive
|
||||
// the hub-side session under it, so ongoing silence has to end the
|
||||
// dial. (keepalive() sets the same 3×period deadline for its path;
|
||||
// the two agree, and this one is what the read loop below relies on.)
|
||||
oneShot := h.md.keepAlivePeriod <= 0
|
||||
if oneShot {
|
||||
conn.SetReadDeadline(time.Now().Add(firstInboundTimeout))
|
||||
@@ -255,6 +276,16 @@ func (h *tunHandler) transportClient(ctx context.Context, tun io.ReadWriter, con
|
||||
// a healthy tunnel after 3×period.
|
||||
if h.md.keepAlivePeriod > 0 {
|
||||
conn.SetReadDeadline(time.Now().Add(h.md.keepAlivePeriod * 3))
|
||||
} else if p2p {
|
||||
// A p2p dial keeps the bound rolling for its whole life.
|
||||
// The registration refresh is echoed by the hub while the
|
||||
// route to this device is real, so silence past
|
||||
// p2pInboundTimeout means the echo stopped: the hub-side
|
||||
// session under this dial was rebuilt and the stream is a
|
||||
// zombie, or the route was reclaimed. Nothing else ends
|
||||
// such a dial, and ending it is the whole cure — the
|
||||
// caller redials and re-registers (see handleClient).
|
||||
conn.SetReadDeadline(time.Now().Add(p2pInboundTimeout))
|
||||
} else if oneShot {
|
||||
// The one-shot has served: the link proved itself alive,
|
||||
// so clear it and read without a deadline from here on,
|
||||
|
||||
@@ -3,6 +3,7 @@ package tun
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"io"
|
||||
"net"
|
||||
"testing"
|
||||
"time"
|
||||
@@ -111,3 +112,97 @@ func TestSocketDialSendsRegistrationOnce(t *testing.T) {
|
||||
t.Fatalf("socket dial sent %d registration handshake(s), want 1", got)
|
||||
}
|
||||
}
|
||||
|
||||
// nudgeDevice is a device whose reader the test wakes on demand: Read blocks
|
||||
// until fed. The transport's device reader is not context-aware, so a test
|
||||
// that ends the link feeds it once to let the transport unwind — the same
|
||||
// wait collectFirstError documents.
|
||||
type nudgeDevice struct{ ch chan []byte }
|
||||
|
||||
func (d *nudgeDevice) Read(p []byte) (int, error) {
|
||||
b, ok := <-d.ch
|
||||
if !ok {
|
||||
return 0, io.EOF
|
||||
}
|
||||
return copy(p, b), nil
|
||||
}
|
||||
|
||||
func (d *nudgeDevice) Write(p []byte) (int, error) { return len(p), nil }
|
||||
|
||||
func (d *nudgeDevice) Close() error { return nil }
|
||||
func (d *nudgeDevice) LocalAddr() net.Addr { return nil }
|
||||
func (d *nudgeDevice) RemoteAddr() net.Addr { return nil }
|
||||
func (d *nudgeDevice) SetDeadline(time.Time) error { return nil }
|
||||
func (d *nudgeDevice) SetReadDeadline(time.Time) error { return nil }
|
||||
func (d *nudgeDevice) SetWriteDeadline(time.Time) error { return nil }
|
||||
|
||||
// A hub side that rebuilt its session under this dial stops echoing the
|
||||
// registration: on this side the stream is a zombie. Nothing closes the dial,
|
||||
// and once the first echo cleared the one-shot no read deadline is left, so
|
||||
// before the watchdog the spoke stayed mute forever — 2026-10-17, 17 minutes
|
||||
// of blackhole with both ends reporting the path up, healed only by a manual
|
||||
// entrypoint restart. Silence after the first echo must end the dial, so the
|
||||
// caller redials and re-registers: that re-dial is what heals a hub-side
|
||||
// rebuild (observed as `peer relay session up … gen=43 client=true`).
|
||||
func TestP2PDialRedialsWhenEchoesStop(t *testing.T) {
|
||||
oldReg := p2pRegisterInterval
|
||||
p2pRegisterInterval = 20 * time.Millisecond
|
||||
defer func() { p2pRegisterInterval = oldReg }()
|
||||
|
||||
oldInbound := p2pInboundTimeout
|
||||
p2pInboundTimeout = 150 * time.Millisecond
|
||||
defer func() { p2pInboundTimeout = oldInbound }()
|
||||
|
||||
peers := make(chan net.Conn, 4)
|
||||
router := &fakeRouter{dialFunc: func(context.Context, string, string) (net.Conn, error) {
|
||||
local, peer := net.Pipe()
|
||||
peers <- peer
|
||||
return local, nil
|
||||
}}
|
||||
|
||||
device := &nudgeDevice{ch: make(chan []byte, 4)}
|
||||
// Closing the device wakes its reader with EOF, which is how the dial
|
||||
// loop unwinds (ErrTun) and this test leaves no dialing goroutine behind.
|
||||
defer close(device.ch)
|
||||
|
||||
h := newTestHandler(nil, router)
|
||||
h.md.passphrase = testPassphrase
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
go func() {
|
||||
_ = h.handleClient(ctx, device, "ip", testPeerKey, testConfig(), xlogger.Nop())
|
||||
}()
|
||||
|
||||
var peer1 net.Conn
|
||||
select {
|
||||
case peer1 = <-peers:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("no dial")
|
||||
}
|
||||
drained := make(chan struct{})
|
||||
go func() {
|
||||
_, _ = io.Copy(io.Discard, peer1)
|
||||
close(drained)
|
||||
}()
|
||||
|
||||
// One echo proves the link alive; the hub then goes silent.
|
||||
if _, err := peer1.Write(bareEcho()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// Past the silence window, wake the device reader so the transport can
|
||||
// unwind once its read deadline has already fired.
|
||||
time.Sleep(3 * p2pInboundTimeout)
|
||||
device.ch <- realIPv4Packet()
|
||||
|
||||
select {
|
||||
case <-peers:
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("dial did not end (and redial) after the hub stopped answering")
|
||||
}
|
||||
select {
|
||||
case <-drained:
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("the silent dial was not closed")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -65,7 +65,7 @@ func TestTransportClientDropsKeepaliveFrames(t *testing.T) {
|
||||
}()
|
||||
|
||||
errCh := make(chan error, 1)
|
||||
go func() { errCh <- h.transportClient(ctx, device, ccA, xlogger.Nop()) }()
|
||||
go func() { errCh <- h.transportClient(ctx, device, ccA, false, xlogger.Nop()) }()
|
||||
|
||||
// Protocol frames must be swallowed...
|
||||
if _, err := ccB.Write(registrationFrame()); err != nil {
|
||||
@@ -110,7 +110,7 @@ func TestNoKeepaliveClientTimesOutWaitingForFirstInbound(t *testing.T) {
|
||||
defer cancel()
|
||||
|
||||
errCh := make(chan error, 1)
|
||||
go func() { errCh <- h.transportClient(ctx, device, ccA, xlogger.Nop()) }()
|
||||
go func() { errCh <- h.transportClient(ctx, device, ccA, false, xlogger.Nop()) }()
|
||||
|
||||
select {
|
||||
case err := <-errCh:
|
||||
@@ -129,7 +129,9 @@ func TestNoKeepaliveClientTimesOutWaitingForFirstInbound(t *testing.T) {
|
||||
|
||||
// The mirror: one inbound byte — even protocol chatter, which never reaches
|
||||
// the device — proves the link alive and clears the one-shot, so a healthy
|
||||
// but quiet tunnel is never redialed.
|
||||
// but quiet tunnel is never redialed. This is the socket arm; a p2p dial
|
||||
// re-arms the bound after the first byte instead of clearing it, which
|
||||
// TestP2PDialRedialsWhenEchoesStop covers.
|
||||
func TestNoKeepaliveClientFirstInboundClearsTimeout(t *testing.T) {
|
||||
old := firstInboundTimeout
|
||||
firstInboundTimeout = 200 * time.Millisecond
|
||||
@@ -145,7 +147,7 @@ func TestNoKeepaliveClientFirstInboundClearsTimeout(t *testing.T) {
|
||||
defer cancel()
|
||||
|
||||
errCh := make(chan error, 1)
|
||||
go func() { errCh <- h.transportClient(ctx, device, ccA, xlogger.Nop()) }()
|
||||
go func() { errCh <- h.transportClient(ctx, device, ccA, false, xlogger.Nop()) }()
|
||||
|
||||
if _, err := ccB.Write(bareEcho()); err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -162,3 +164,40 @@ func TestNoKeepaliveClientFirstInboundClearsTimeout(t *testing.T) {
|
||||
|
||||
ccB.Close() // unblock the reader; the test proved what it had to
|
||||
}
|
||||
|
||||
// The p2p arm of the bound: after the first inbound byte the rolling
|
||||
// deadline is armed, and silence past p2pInboundTimeout must report a read
|
||||
// timeout — which the dial loop turns into a redial.
|
||||
func TestP2PDialTimesOutAfterSilenceOnceLinkProven(t *testing.T) {
|
||||
old := p2pInboundTimeout
|
||||
p2pInboundTimeout = 100 * time.Millisecond
|
||||
defer func() { p2pInboundTimeout = old }()
|
||||
|
||||
device, kernelSide := net.Pipe()
|
||||
defer device.Close()
|
||||
defer kernelSide.Close()
|
||||
ccA, ccB := net.Pipe()
|
||||
defer ccA.Close()
|
||||
defer ccB.Close()
|
||||
|
||||
h := newTestHandler(nil, nil)
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
errCh := make(chan error, 1)
|
||||
go func() { errCh <- h.transportClient(ctx, device, ccA, true, xlogger.Nop()) }()
|
||||
|
||||
if _, err := ccB.Write(bareEcho()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
select {
|
||||
case err := <-errCh:
|
||||
var nerr net.Error
|
||||
if !errors.As(err, &nerr) || !nerr.Timeout() {
|
||||
t.Fatalf("transportClient = %v, want a read timeout", err)
|
||||
}
|
||||
case <-time.After(15 * time.Second):
|
||||
t.Fatal("a p2p dial survived silence after the link proved itself")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -113,7 +113,7 @@ func TestTransportClientDropsNonIPFrames(t *testing.T) {
|
||||
}()
|
||||
|
||||
errCh := make(chan error, 1)
|
||||
go func() { errCh <- h.transportClient(ctx, device, ccA, xlogger.Nop()) }()
|
||||
go func() { errCh <- h.transportClient(ctx, device, ccA, false, xlogger.Nop()) }()
|
||||
|
||||
bad := make([]byte, 60)
|
||||
for i := range bad {
|
||||
|
||||
@@ -75,7 +75,7 @@ func TestTransportClientSurvivesAnEmptyPacketRead(t *testing.T) {
|
||||
panicked <- r
|
||||
}
|
||||
}()
|
||||
errCh <- h.transportClient(ctx, dev, ccA, xlogger.Nop())
|
||||
errCh <- h.transportClient(ctx, dev, ccA, false, xlogger.Nop())
|
||||
}()
|
||||
|
||||
select {
|
||||
|
||||
@@ -19,11 +19,18 @@ type fakeRouter struct {
|
||||
// dialConn, when set, is what Dial hands back — the test holds the far
|
||||
// end of it, which is how a dial is driven without a network.
|
||||
dialConn net.Conn
|
||||
// dialFunc, when set, takes precedence over dialConn: it lets a test
|
||||
// hand out a fresh link per dial, which one static conn cannot — the
|
||||
// shape a redial loop needs.
|
||||
dialFunc func(ctx context.Context, network, address string) (net.Conn, error)
|
||||
}
|
||||
|
||||
func (r *fakeRouter) Options() *chain.RouterOptions { return r.options }
|
||||
|
||||
func (r *fakeRouter) Dial(ctx context.Context, network, address string, opts ...chain.DialOption) (net.Conn, error) {
|
||||
if r.dialFunc != nil {
|
||||
return r.dialFunc(ctx, network, address)
|
||||
}
|
||||
if r.dialConn != nil {
|
||||
return r.dialConn, nil
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user