diff --git a/internal/pipeproxy/abandonbody_test.go b/internal/pipeproxy/abandonbody_test.go new file mode 100644 index 0000000..490eaf1 --- /dev/null +++ b/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) + } +} diff --git a/internal/pipeproxy/rewrite.go b/internal/pipeproxy/rewrite.go index 23dde5c..991f92e 100644 --- a/internal/pipeproxy/rewrite.go +++ b/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