package e2e import ( "fmt" "io" "net" "sync" "sync/atomic" "testing" "time" "github.com/iceBear67/redapricot/client" "github.com/iceBear67/redapricot/client/wire" ) // blackholeRelay forwards TCP to the hub until it is switched off, after which // already-established pairs silently stop carrying bytes while their sockets // stay open. That is what a stateful middlebox looks like when it forgets a // flow: conntrack expiry, a firewall reload, or a cloud LB idle timeout. No FIN // and no RST ever reach either end, so nothing below the application layer can // notice. // Only flows that already existed when the switch is thrown go dark; new // connections are carried normally, exactly as when a middlebox forgets // established state but keeps forwarding fresh traffic. type blackholeRelay struct { addr string backend string gen atomic.Uint64 down atomic.Bool mu sync.Mutex held []net.Conn } func newBlackholeRelay(t *testing.T, backend string) *blackholeRelay { t.Helper() ln, err := net.Listen("tcp", "127.0.0.1:0") if err != nil { t.Fatalf("relay listen: %v", err) } r := &blackholeRelay{addr: ln.Addr().String(), backend: backend} t.Cleanup(func() { _ = ln.Close() r.mu.Lock() for _, c := range r.held { _ = c.Close() } r.mu.Unlock() }) go func() { for { cli, err := ln.Accept() if err != nil { return } go r.handle(cli) } }() return r } func (r *blackholeRelay) handle(cli net.Conn) { born := r.gen.Load() if r.down.Load() { _ = cli.Close() return } up, err := net.Dial("tcp", r.backend) if err != nil { _ = cli.Close() return } r.mu.Lock() r.held = append(r.held, cli, up) r.mu.Unlock() pipe := func(dst, src net.Conn) { buf := make([]byte, 32*1024) for { n, err := src.Read(buf) if n > 0 && r.gen.Load() == born { if _, werr := dst.Write(buf[:n]); werr != nil { return } } if err != nil { return } } } go pipe(up, cli) go pipe(cli, up) } // blackhole strands every currently-established pair. Later connections are // unaffected. func (r *blackholeRelay) blackhole() { r.gen.Add(1) } // dropAll hard-resets every currently-established pair: a real FIN/RST reaches // both ends immediately, as when a middlebox is restarted or a route flaps, // rather than the silent stranding blackhole models. Later connections are // carried normally. // // stop takes the relay out of service: new connections are refused rather than // carried, so the tunnel stays down until restore is called. Models an outage // the client cannot immediately reconnect through. func (r *blackholeRelay) stop() { r.down.Store(true) } // restore puts the relay back in service. Flows stranded before it are still // dead — only fresh connections are carried, which is what a client gets after // a middlebox or upstream link comes back. func (r *blackholeRelay) restore() { r.down.Store(false) } // This is the fast path into stream resumption: the client learns the conn is // gone at once instead of waiting out a heartbeat timeout. func (r *blackholeRelay) dropAll() { r.gen.Add(1) r.mu.Lock() held := r.held r.held = nil r.mu.Unlock() for _, c := range held { _ = c.Close() } } // TestBlackholedPathRecovers is the end-to-end regression guard for the // stability bug this hardening was written for: with the tunnel's path silently // dropped, the client used to notice nothing at all. Its read loops parked // forever, the dead worker conn stayed in the pool, the hub kept routing players // to a control session nobody was reading, and no player could connect again // until the client process was restarted. // // The heartbeats must now detect the silence, drop both sessions, and let the // existing reconnect path restore service on its own. func TestBlackholedPathRecovers(t *testing.T) { const psk = "e2e-blackhole" hubPort := freePort(t) hubAddr := fmt.Sprintf("127.0.0.1:%d", hubPort) startHub(t, hubPort, psk) dest := newMockDest(t, modeEcho) // The client reaches the hub only through the relay; players connect to the // hub directly, as they would from the internet. relay := newBlackholeRelay(t, hubAddr) const pingMs = 400 // heartbeat timeout is 3x this c := startClientWithPing(t, relay.addr, psk, 4, pingMs, []client.Mapping{ {Pattern: "mc.local", Destination: dest.addr}, }) play := func(what string) error { pc, err := net.DialTimeout("tcp", hubAddr, 5*time.Second) if err != nil { return err } defer pc.Close() if _, err := pc.Write(wire.BuildHandshake(767, "mc.local", 25565, 2)); err != nil { return err } msg := []byte(what) if _, err := pc.Write(msg); err != nil { return err } got := make([]byte, len(msg)) _ = pc.SetReadDeadline(time.Now().Add(15 * time.Second)) _, err = io.ReadFull(pc, got) return err } if err := play("before"); err != nil { t.Fatalf("baseline round-trip failed: %v", err) } relay.blackhole() t.Log("path blackholed: no FIN, no RST, sockets held open") // Wait for the heartbeats to fire, the sessions to be dropped, and the // control session to reconnect through a fresh relay pair. deadline := time.Now().Add(45 * time.Second) var lastErr error for time.Now().Before(deadline) { if lastErr = play("after"); lastErr == nil { t.Logf("recovered on its own; worker conns now %d", c.WorkerConnCount()) return } time.Sleep(500 * time.Millisecond) } t.Fatalf("client never recovered from the blackholed path (last error: %v)", lastErr) }