diff --git a/docs/examples/provider.go b/docs/examples/provider.go index 192e0c787..89ef92c4a 100644 --- a/docs/examples/provider.go +++ b/docs/examples/provider.go @@ -193,7 +193,10 @@ func up(options options, args []string) { // through it only after up has returned, so reaping it here would // tear the endpoint down before anyone reached it. Its lifetime is // its own: it exits by itself after three minutes (serve-demo), the - // way a real provider's resource outlives the provider CLI run. + // way a real provider's resource outlives the provider CLI run. No + // Wait() and no zombie either: this process exits within seconds, so + // the subprocess is long re-parented to init — which reaps it — when + // its three minutes are up. // // the endpoint is announced as seen from THIS process's host — // the relay translates loopback into the container-visible name diff --git a/relay/main.go b/relay/main.go index 510d00198..8abfc1f41 100644 --- a/relay/main.go +++ b/relay/main.go @@ -40,6 +40,7 @@ import ( "strconv" "strings" "sync" + "sync/atomic" "syscall" "time" ) @@ -149,6 +150,28 @@ func serve(ctx context.Context, listener net.Listener, upstream string, wg *sync } } +// halfCloseIdleTimeout bounds how long the surviving direction may sit IDLE +// once the other one has finished. Half-close semantics stay intact — a peer +// that keeps sending data after the other side's FIN is relayed for as long +// as it takes — but a peer that never closes after our FIN can no longer +// pin the goroutine pair and both connections forever (and with them the +// drain in main). A variable so tests exercise the expiry quickly. +var halfCloseIdleTimeout = 60 * time.Second + +// idleConn re-arms a read deadline before each Read once armed: active +// transfers never expire, idle ones do. +type idleConn struct { + net.Conn + armed atomic.Bool +} + +func (c *idleConn) Read(p []byte) (int, error) { + if c.armed.Load() { + _ = c.Conn.SetReadDeadline(time.Now().Add(halfCloseIdleTimeout)) + } + return c.Conn.Read(p) +} + // forward deliberately takes no context: a connection accepted at the // shutdown boundary (context cancelled, listener not yet closed) must still // be served — that is the drain contract — and a cancelled context would make @@ -163,10 +186,21 @@ func forward(downstream net.Conn, upstream string) { } defer up.Close() + down := &idleConn{Conn: downstream} + upc := &idleConn{Conn: up} done := make(chan struct{}, 2) - go func() { _, _ = io.Copy(up, downstream); closeWrite(up); done <- struct{}{} }() - go func() { _, _ = io.Copy(downstream, up); closeWrite(downstream); done <- struct{}{} }() + go func() { _, _ = io.Copy(up, down); closeWrite(up); done <- struct{}{} }() + go func() { _, _ = io.Copy(downstream, upc); closeWrite(downstream); done <- struct{}{} }() <-done + // One direction is done: from here on the survivor may stream for as + // long as data flows, but no longer sit idle forever. Arm the per-read + // deadline for its future Reads, and set one immediately for a survivor + // already blocked in Read. + down.armed.Store(true) + upc.armed.Store(true) + deadline := time.Now().Add(halfCloseIdleTimeout) + _ = downstream.SetReadDeadline(deadline) + _ = up.SetReadDeadline(deadline) <-done } diff --git a/relay/main_test.go b/relay/main_test.go index bf3adcc94..e8c0f4156 100644 --- a/relay/main_test.go +++ b/relay/main_test.go @@ -17,8 +17,11 @@ package main import ( + "io" + "net" "reflect" "testing" + "time" ) // A provider announces endpoints as seen from its host; the relay, which @@ -48,3 +51,142 @@ func TestParseRoutesRejectsMalformedEntries(t *testing.T) { } } } + +// tcpPair returns a connected (client, server) TCP pair. +func tcpPair(t *testing.T) (net.Conn, net.Conn) { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer ln.Close() + type accepted struct { + conn net.Conn + err error + } + ch := make(chan accepted, 1) + go func() { + conn, err := ln.Accept() + ch <- accepted{conn, err} + }() + client, err := net.Dial("tcp", ln.Addr().String()) + if err != nil { + t.Fatal(err) + } + server := <-ch + if server.err != nil { + t.Fatal(server.err) + } + return client, server.conn +} + +// silentUpstream accepts one connection, drains it, and sits silent with its +// write side open until closed. +func silentUpstream(t *testing.T) net.Listener { + t.Helper() + upstream, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = upstream.Close() }) + go func() { + conn, err := upstream.Accept() + if err != nil { + return + } + defer conn.Close() + _, _ = io.Copy(io.Discard, conn) // read the FIN, never answer, never close + time.Sleep(5 * time.Second) + }() + return upstream +} + +// A peer that never closes after receiving our FIN must not pin the forward +// forever: once one direction is done, the surviving one may stream for as +// long as data flows but is reaped when idle past the grace. +func TestForwardReapsIdleHalfOpenPair(t *testing.T) { + restore := halfCloseIdleTimeout + halfCloseIdleTimeout = 200 * time.Millisecond + t.Cleanup(func() { halfCloseIdleTimeout = restore }) + + upstream := silentUpstream(t) + client, server := tcpPair(t) + t.Cleanup(func() { _ = client.Close() }) + + finished := make(chan struct{}) + go func() { + forward(server, upstream.Addr().String()) + close(finished) + }() + + if _, err := client.Write([]byte("request")); err != nil { + t.Fatal(err) + } + _ = client.Close() // downstream fully gone: down->up finishes, up->down survives + + select { + case <-finished: + // reaped by the idle grace instead of hanging on the silent upstream + case <-time.After(5 * time.Second): + t.Fatal("forward still blocked on a half-open pair after the idle grace") + } +} + +// Half-close semantics survive the reaper: a response still flowing after the +// client half-closed its request side is relayed past the grace, not cut. +func TestForwardKeepsStreamingAfterHalfClose(t *testing.T) { + restore := halfCloseIdleTimeout + halfCloseIdleTimeout = 300 * time.Millisecond + t.Cleanup(func() { halfCloseIdleTimeout = restore }) + + upstream, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = upstream.Close() }) + const chunks = 6 + go func() { + conn, err := upstream.Accept() + if err != nil { + return + } + defer conn.Close() + // stream chunks for well past the idle grace, each within it + for range chunks { + time.Sleep(150 * time.Millisecond) + if _, err := conn.Write([]byte("chunk")); err != nil { + return + } + } + }() + + client, server := tcpPair(t) + t.Cleanup(func() { _ = client.Close() }) + finished := make(chan struct{}) + go func() { + forward(server, upstream.Addr().String()) + close(finished) + }() + + closeWrite(client) // request side done; the response keeps streaming + + got := 0 + buf := make([]byte, 64) + _ = client.SetReadDeadline(time.Now().Add(5 * time.Second)) + for { + n, err := client.Read(buf) + got += n + if err != nil { + break + } + } + if got < chunks*len("chunk") { + t.Fatalf("streaming was cut by the idle reaper: got %d bytes, want %d", got, chunks*len("chunk")) + } + // let forward return before the Cleanup restores the shared grace var + select { + case <-finished: + case <-time.After(5 * time.Second): + t.Fatal("forward did not return after both directions ended") + } +}