fix: a stalled upload no longer wedges the bridge and kills idle-stop

internal/pipeproxy/abandonbody_test.go

@@ -0,0 +1,120 @@
+package pipeproxy_test
+
+import (
+	"bufio"
+	"io"
+	"net"
+	"testing"
+	"time"
+
+	"github.com/wslkit/skrog/internal/pipeproxy"
+)
+
+// A client that stalls mid-upload must not wedge the bridge (#435).
+//
+// TestEarlyErrorOnAbandonedUploadIsSalvaged already covers the other half of
+// this: the engine answers early and stops reading, so the body writer is
+// blocked WRITING, and engine.Close() frees it. That case always worked.
+//
+// This is the case that did not. The engine keeps draining, so the writer is
+// blocked READING a client that has gone quiet — and engine.Close() cannot
+// touch a read. The teardown then sat on `<-bodySent` forever, which meant
+// rewriteBinds never returned, Server.handle never ran s.clients.Add(-1),
+// ActiveConns never dropped, and idle-stop was dead for the life of the
+// process.
+//
+// The assertion is deliberately just "it returns". That is the whole bug.
+func TestStalledUploadDoesNotWedgeTheConnection(t *testing.T) {
+	client, bridgeClient := net.Pipe()
+	engineSide, bridgeEngine := net.Pipe()
+	defer client.Close()
+	defer engineSide.Close()
+
+	done := make(chan error, 1)
+	go func() { done <- pipeproxy.RewriteBinds(bridgeClient, bridgeEngine) }()
+
+	// Engine: read the head, answer at once, then KEEP DRAINING. The draining
+	// is the point — it guarantees the body writer can always write, so the
+	// only thing it can be stuck on is the read.
+	go func() {
+		br := bufio.NewReader(engineSide)
+		for {
+			line, err := br.ReadString('\n')
+			if err != nil {
+				return
+			}
+			if line == "\r\n" {
+				break
+			}
+		}
+		engineSide.Write([]byte("HTTP/1.1 400 Bad Request\r\nContent-Length: 3\r\n\r\nbad"))
+		io.Copy(io.Discard, engineSide)
+	}()
+
+	// Client: promise a megabyte, send a handful of bytes, then go silent —
+	// a sleeping laptop, a dropped VPN, a killed CLI.
+	go func() {
+		client.Write([]byte("POST /build HTTP/1.1\r\nHost: d\r\nContent-Length: 1048576\r\n\r\n"))
+		client.Write(make([]byte, 16))
+	}()
+	// Drain the response, or relaying it would block on the unbuffered pipe
+	// and the test would stall for a reason that is not the one under test.
+	go io.Copy(io.Discard, client)
+
+	select {
+	case <-done:
+		// Returned. Whether it returned an error does not matter: the
+		// connection is written off either way, and the caller's deferred
+		// bookkeeping is what had to run.
+	case <-time.After(30 * time.Second):
+		t.Fatal("RewriteBinds never returned on a stalled upload: the connection is wedged, " +
+			"so ActiveConns never drops and idle-stop is disabled for the life of the process (#435)")
+	}
+}
+
+// The bounded wait must not fire in the ordinary case. If abandonBody always
+// waited out its grace, every abandoned upload would add seconds to a
+// connection's teardown, which would turn a correctness fix into a
+// latency bug nobody attributed to it.
+func TestAbandonedUploadTearsDownPromptly(t *testing.T) {
+	client, bridgeClient := net.Pipe()
+	engineSide, bridgeEngine := net.Pipe()
+	defer client.Close()
+	defer engineSide.Close()
+
+	done := make(chan error, 1)
+	go func() { done <- pipeproxy.RewriteBinds(bridgeClient, bridgeEngine) }()
+
+	go func() {
+		br := bufio.NewReader(engineSide)
+		for {
+			line, err := br.ReadString('\n')
+			if err != nil {
+				return
+			}
+			if line == "\r\n" {
+				break
+			}
+		}
+		engineSide.Write([]byte("HTTP/1.1 400 Bad Request\r\nContent-Length: 3\r\n\r\nbad"))
+		io.Copy(io.Discard, engineSide)
+	}()
+	go func() {
+		client.Write([]byte("POST /build HTTP/1.1\r\nHost: d\r\nContent-Length: 1048576\r\n\r\n"))
+		client.Write(make([]byte, 16))
+	}()
+	go io.Copy(io.Discard, client)
+
+	start := time.Now()
+	select {
+	case <-done:
+	case <-time.After(30 * time.Second):
+		t.Fatal("RewriteBinds never returned (#435)")
+	}
+	// 250ms grace + teardown. The 5s bound in abandonBody is the failsafe for
+	// a writer neither close reached, not the expected path.
+	if took := time.Since(start); took > 4*time.Second {
+		t.Errorf("teardown took %v; the bounded wait is being waited out rather than "+
+			"the writer being unblocked", took)
+	}
+}

internal/pipeproxy/rewrite.go

@@ -100,6 +100,65 @@ func RewriteBindsFor(t SourceTranslator, sink AuditSink, gate Gate) func(net.Con
 	return func(c net.Conn, e io.ReadWriteCloser) error { return rewriteBinds(c, e, sink, gate, t) }
 }
 
+const (
+	// bodyGrace is how long a final response waits for the request body to
+	// finish forwarding before the connection is written off.
+	bodyGrace = 250 * time.Millisecond
+
+	// abandonGrace bounds the wait for the body writer to notice it has been
+	// cut off. See abandonBody: the point is that this wait ENDS.
+	abandonGrace = 5 * time.Second
+)
+
+// abandonBody tears down a request body that is still streaming after the
+// engine has already answered, and waits — bounded — for its writer to exit.
+//
+// The writer is `bodySent <- req.Write(engine)`, and req.Write does two things
+// that can block: it READS req.Body, which reads the client, and it WRITES to
+// the engine. This used to close only the engine and then wait forever:
+//
+//	engine.Close()
+//	<-bodySent
+//
+// which unblocks a writer stuck on the write and does nothing at all for one
+// stuck on the read. A client that stalls mid-upload — a laptop that sleeps
+// during `docker build`, a dropped VPN, a killed CLI — parks req.Write in a
+// Read that never returns, so <-bodySent never returns, so rewriteBinds never
+// returns, so Server.handle never runs `s.clients.Add(-1)`.
+//
+// The cost of that is out of all proportion to the cause: ActiveConns stays
+// above zero for the life of the process, maybeIdleStop vetoes on "open client
+// connections" forever, and the idle-timeout feature is silently dead with
+// nothing in the log to say so. Serve's wg.Wait() never completes either, so
+// shutdown hangs holding the single-instance lock — which the comment above
+// Serve says must not happen.
+//
+// So: cut BOTH sides, and bound the wait.
+//
+// A read deadline in the past is what reaches the read. It makes the in-flight
+// Read return immediately and every later one fail, which is exactly right
+// here — the caller has already set resp.Close, so this connection is finished
+// either way.
+//
+// The bounded select is the belt to that pair of braces. If the writer is
+// wedged on something neither close reached, leaking one goroutine is strictly
+// better than not returning: a leaked goroutine costs a little memory, while
+// not returning disables idle-stop for every user of this process.
+func abandonBody(client net.Conn, engine io.ReadWriteCloser, bodySent <-chan error) {
+	engine.Close()
+	if client != nil {
+		// Errors are not actionable: the deadline is best-effort on a
+		// connection already being discarded, and a transport that does not
+		// support deadlines still gets the engine close and the bound below.
+		_ = client.SetReadDeadline(time.Now())
+	}
+	select {
+	case <-bodySent:
+	case <-time.After(abandonGrace):
+		trace("REQ body writer did not exit within %s; abandoning it", abandonGrace)
+	}
+}
+
 func rewriteBinds(client net.Conn, engine io.ReadWriteCloser, audit AuditSink, gate Gate, translate SourceTranslator) error {
 	clientR := bufio.NewReader(client)
 	engineR := bufio.NewReader(engine)
@@ -221,16 +280,13 @@ func rewriteBinds(client net.Conn, engine io.ReadWriteCloser, audit AuditSink, g
 				trace("REQ body aborted by early response (%d): %v", resp.StatusCode, werr)
 				resp.Close = true // unsendable remainder: never reuse this connection
 			}
-		case <-time.After(250 * time.Millisecond):
+		case <-time.After(bodyGrace):
 			trace("REQ body still streaming after early response (%d); abandoning the connection", resp.StatusCode)
 			resp.Close = true
 			// The engine side is torn down AFTER the response is relayed to
 			// the client below; deferring the close here keeps the salvaged
 			// body readable. Mark it so.
-			defer func() {
-				engine.Close()
-				<-bodySent
-			}()
+			defer abandonBody(client, engine, bodySent)
 		}
 
 		// A real response is in hand: the engine is answering, so later EOFs on