diff --git a/handler/tun/client.go b/handler/tun/client.go index 4c9b0b8..77417ac 100644 --- a/handler/tun/client.go +++ b/handler/tun/client.go @@ -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, diff --git a/handler/tun/client_register_test.go b/handler/tun/client_register_test.go index 0ecf3b9..0921b33 100644 --- a/handler/tun/client_register_test.go +++ b/handler/tun/client_register_test.go @@ -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") + } +} diff --git a/handler/tun/client_test.go b/handler/tun/client_test.go index 35b5dbd..2c11790 100644 --- a/handler/tun/client_test.go +++ b/handler/tun/client_test.go @@ -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") + } +} diff --git a/handler/tun/datagram_frame_test.go b/handler/tun/datagram_frame_test.go index 7870679..8c1ef12 100644 --- a/handler/tun/datagram_frame_test.go +++ b/handler/tun/datagram_frame_test.go @@ -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 { diff --git a/handler/tun/empty_packet_test.go b/handler/tun/empty_packet_test.go index ce4c68b..ab08203 100644 --- a/handler/tun/empty_packet_test.go +++ b/handler/tun/empty_packet_test.go @@ -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 { diff --git a/handler/tun/handler_test.go b/handler/tun/handler_test.go index 4b771a4..af69fb2 100644 --- a/handler/tun/handler_test.go +++ b/handler/tun/handler_test.go @@ -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 }