fix(relay): reap idle half-open forwards instead of pinning them forever

A peer that never closes after receiving the relayed FIN used to pin
the forward's goroutine pair and both TCP connections until SIGKILL —
and with them the drain in main. Force-closing both ends as soon as one
direction finishes would have thrown out TCP half-close (a client that
FINs its request and then reads a long response), so the surviving
direction now runs under an idle grace instead: a read deadline
re-armed before every Read once the other direction is done. Active
streams are never cut — only pairs sitting idle past the grace are
reaped. Locked by two tests: the silent-peer pair is reaped at the
grace, and a response still streaming after the client's half-close
survives well past it.

The demo provider's comment also spells out why serve-demo is neither
Wait()ed nor a zombie: the provider exits within seconds, so init has
long adopted — and reaps — the subprocess when its three minutes are up.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Signed-off-by: Nicolas De Loof <nicolas.deloof@gmail.com>

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

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
 }
 

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")
+	}
+}