From 6422ea563b146717a026a346eb6a96b9e0c2d1c9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BC=A0=E4=BB=A5=E7=90=B3?= Date: Fri, 18 Sep 2026 17:12:44 +0800 Subject: [PATCH 1/7] fix(transport): acknowledge down-poll chunks so a slow push cannot lose bytes The HTTP transport carries an end-to-end TLS stream over finite request / response pairs. /h/down dequeued bytes from the session's side queue and then wrote them into the response, so anything that stopped the reader from receiving that response whole destroyed those bytes: the stream got a hole in the middle and TLS reported it as "tls: bad record MAC" several minutes into a push. httpconn made this silent by discarding the error from io.ReadAll on the response body and handing the partial body on as if it were complete. Two things conspired on a slow link. The relay coalesced everything queued into one response, unbounded, so a reader that fell behind was handed megabytes at once; and http.Client.Timeout bounded the whole exchange, so it fired while that response body was still downloading and cut it in half. The down direction is now acknowledged. A data-bearing 200 carries X-Wanctl-Down-Seq, the next poll reports the last fully received sequence in ack=, and the relay holds a delivered chunk until it is acked, re-sending it byte for byte otherwise. httpconn no longer swallows a truncated body: it does not advance the ack and polls again, so the relay re-sends. Coalescing is capped at the controller's write batch (256 KiB), which also bounds the memory one unacked chunk holds, and the client bounds the wait for response headers instead of the whole exchange. A poll that arrives without ack= is served the old way, so a mixed-version fleet keeps working. transport=ws does not have this bug: a session there is one TCP connection piped byte for byte, so a carrier failure tears the session down instead of leaving a gap. The hybrid ws/http path reads the same queues in process, where the hand-off is a function return rather than a response that can half arrive. Fixes #57 Co-Authored-By: Claude Fable 5.1 --- internal/httpconn/httpconn.go | 96 +++++++- internal/relay/http.go | 77 ++++++- internal/relay/slowlink_test.go | 379 ++++++++++++++++++++++++++++++++ 3 files changed, 541 insertions(+), 11 deletions(-) create mode 100644 internal/relay/slowlink_test.go diff --git a/internal/httpconn/httpconn.go b/internal/httpconn/httpconn.go index 59693c5..a9f369b 100644 --- a/internal/httpconn/httpconn.go +++ b/internal/httpconn/httpconn.go @@ -9,9 +9,17 @@ // forwards them promptly: // // up: POST /h/up?session=&role= — one request per Write, body = bytes. -// down: GET /h/down?session=&role= — long-poll: returns available bytes +// down: GET /h/down?session=&role=&ack= — long-poll: returns available bytes // (200), 204 if none within the poll window (client re-polls), or 410 when // the session is closed (-> io.EOF). +// +// The down direction is acknowledged. Every data-bearing 200 carries a +// monotonic X-Wanctl-Down-Seq, and the next poll reports the highest sequence +// the reader has fully received in ack=. The relay holds a delivered chunk +// until it is acked and re-sends it otherwise, because an HTTP response that +// only half arrives would otherwise punch a hole in the middle of the +// end-to-end TLS stream — which the TLS layer reports, several minutes into a +// slow push, as "tls: bad record MAC" (issue #57). package httpconn import ( @@ -22,6 +30,7 @@ import ( "net" "net/http" "net/url" + "strconv" "sync" "time" @@ -39,6 +48,7 @@ type conn struct { readM sync.Mutex leftover []byte eof bool + ackSeq uint64 // highest down-poll sequence fully received writeM sync.Mutex pending []byte @@ -51,25 +61,60 @@ type conn struct { const ( writeBatchBytes = 256 << 10 writeFlushDelay = 5 * time.Millisecond + + // DownSeqHeader carries the sequence number of a data-bearing down-poll + // response; DownAckParam is the query parameter the next poll reports the + // last fully received sequence in. A poll without DownAckParam is a + // pre-acknowledgement client and the relay serves it the old way. + DownSeqHeader = "X-Wanctl-Down-Seq" + DownAckParam = "ack" + + // downPollAttempts bounds how many times one Read retries a down poll that + // the carrier failed to deliver. The relay still holds the chunk, so a + // retry is a re-send rather than a hole; the bound is what stops a + // permanently broken link from spinning forever. + downPollAttempts = 6 + downRetryDelay = 250 * time.Millisecond ) // Dial constructs a net.Conn for a session/role. base is the relay's HTTP origin // (http:// or https://, or ws(s):// which is normalized). No network I/O happens // here; the first Read long-polls the down channel. func Dial(ctx context.Context, base, session, role, token string) (net.Conn, error) { + return DialWith(ctx, base, session, role, token, nil) +} + +// DialWith is Dial with an explicit *http.Client for the up/down requests. A nil +// client uses the package default. Tests use it to inject a carrier that drops, +// delays or truncates responses. +func DialWith(ctx context.Context, base, session, role, token string, hc *http.Client) (net.Conn, error) { httpBase, err := config.RelayHTTPOrigin(base) if err != nil { return nil, err } + if hc == nil { + hc = defaultClient() + } return &conn{ base: httpBase, session: session, role: role, token: token, - hc: &http.Client{Timeout: 60 * time.Second}, + hc: hc, }, nil } +func defaultClient() *http.Client { + tr := http.DefaultTransport.(*http.Transport).Clone() + // A down poll parks on the relay for its whole poll window before + // answering, so only the wait for response *headers* can be bounded + // tightly. The previous blanket http.Client.Timeout bounded the entire + // exchange, so on a slow link it fired while a response body was still + // downloading and cut the body in half (issue #57). + tr.ResponseHeaderTimeout = 45 * time.Second + return &http.Client{Transport: tr, Timeout: 5 * time.Minute} +} + func (c *conn) Read(p []byte) (int, error) { c.readM.Lock() defer c.readM.Unlock() @@ -84,13 +129,28 @@ func (c *conn) Read(p []byte) (int, error) { if c.eof { return 0, io.EOF } - q := url.Values{"session": {c.session}, "role": {c.role}} - downURL := c.base + "/h/down?" + q.Encode() + failures := 0 + // retry reports whether a carrier failure is worth another poll. The relay + // keeps an unacked chunk, so polling again re-sends the same bytes instead + // of leaving a gap in the stream. + retry := func(err error) error { + failures++ + if failures >= downPollAttempts { + return fmt.Errorf("down poll failed %d times in a row: %w", failures, err) + } + time.Sleep(downRetryDelay) + return nil + } for { if c.isClosed() { return 0, io.EOF } - req, err := http.NewRequest("GET", downURL, nil) + q := url.Values{ + "session": {c.session}, + "role": {c.role}, + DownAckParam: {strconv.FormatUint(c.ackSeq, 10)}, + } + req, err := http.NewRequest("GET", c.base+"/h/down?"+q.Encode(), nil) if err != nil { return 0, err } @@ -100,15 +160,37 @@ func (c *conn) Read(p []byte) (int, error) { if c.isClosed() { return 0, io.EOF } - return 0, err + if giveUp := retry(err); giveUp != nil { + return 0, giveUp + } + continue } switch resp.StatusCode { case http.StatusNoContent: resp.Body.Close() + failures = 0 continue // no data this round; poll again case http.StatusOK: - body, _ := io.ReadAll(resp.Body) + body, readErr := io.ReadAll(resp.Body) resp.Body.Close() + if readErr != nil { + // The body was cut short. Do not advance the ack and do not + // hand the partial body on: the relay re-sends the whole + // chunk on the next poll. Accepting a truncated body here is + // what made a multi-minute push die with "tls: bad record + // MAC" (issue #57). + if giveUp := retry(readErr); giveUp != nil { + return 0, giveUp + } + continue + } + failures = 0 + if seq, convErr := strconv.ParseUint(resp.Header.Get(DownSeqHeader), 10, 64); convErr == nil && seq > 0 { + if seq <= c.ackSeq { + continue // a re-send of a chunk already consumed + } + c.ackSeq = seq + } if len(body) == 0 { continue } diff --git a/internal/relay/http.go b/internal/relay/http.go index 33ca9fc..af9c2c2 100644 --- a/internal/relay/http.go +++ b/internal/relay/http.go @@ -5,10 +5,12 @@ import ( "errors" "io" "net/http" + "strconv" "sync" "time" "wanctl/internal/delegation" + "wanctl/internal/httpconn" "wanctl/internal/limits" "wanctl/internal/sessionauth" ) @@ -33,6 +35,15 @@ type sideQueue struct { ch chan []byte done chan struct{} once sync.Once + + // A drained chunk is removed from ch before it is written to an HTTP + // response, so if that response is not delivered whole the bytes are gone + // and the end-to-end TLS stream has a hole in it. Readers that speak the + // acknowledged down protocol therefore get the chunk held here until they + // report having received it (issue #57). + ackMu sync.Mutex + seq uint64 + unacked []byte } func newSideQueue() *sideQueue { @@ -52,12 +63,50 @@ func (q *sideQueue) push(b []byte) bool { func (q *sideQueue) close() { q.once.Do(func() { close(q.done) }) } -// drain returns bytes available within timeout, coalescing any queued chunks. -// closed is true only when the queue is closed and no more bytes remain. +// take is drain for a reader that acknowledges what it received. ack is the +// highest sequence the reader has fully received. While an older chunk is still +// outstanding it is returned again, byte for byte, under its original sequence; +// only an ack that covers it lets the relay move on. The returned seq is 0 when +// there is no data. +func (q *sideQueue) take(ack uint64, timeout time.Duration) (data []byte, seq uint64, closed bool) { + q.ackMu.Lock() + if q.unacked != nil { + if ack < q.seq { + data, seq = q.unacked, q.seq + q.ackMu.Unlock() + return data, seq, false + } + q.unacked = nil + } + q.ackMu.Unlock() + + data, closed = q.drain(timeout) + if len(data) == 0 { + return nil, 0, closed + } + q.ackMu.Lock() + q.seq++ + seq = q.seq + q.unacked = data + q.ackMu.Unlock() + return data, seq, false +} + +// maxDrainBytes caps how much one drain coalesces into a single down-poll +// response. Without a cap a reader that fell behind gets handed everything the +// writer queued meanwhile — megabytes in one response — and on a slow link a +// response that large cannot be downloaded inside any sane timeout. Capping it +// at the controller's write batch makes the down direction mirror the up one +// and bounds the memory an unacked chunk holds. +const maxDrainBytes = 256 << 10 + +// drain returns bytes available within timeout, coalescing queued chunks up to +// maxDrainBytes. closed is true only when the queue is closed and no more bytes +// remain. func (q *sideQueue) drain(timeout time.Duration) (data []byte, closed bool) { collect := func(first []byte) []byte { out := append([]byte{}, first...) - for { + for len(out) < maxDrainBytes { select { case b := <-q.ch: out = append(out, b...) @@ -65,6 +114,7 @@ func (q *sideQueue) drain(timeout time.Duration) (data []byte, closed bool) { return out } } + return out } select { case b := <-q.ch: @@ -410,7 +460,23 @@ func (r *Relay) handleHDown(w http.ResponseWriter, req *http.Request) { if req.URL.Query().Get("role") == "agent" { src = s.toAgent } - data, closed := src.drain(downPollWait) + var ( + data []byte + seq uint64 + closed bool + ) + if ackParam := req.URL.Query().Get(httpconn.DownAckParam); ackParam != "" { + ack, err := strconv.ParseUint(ackParam, 10, 64) + if err != nil { + http.Error(w, "bad ack", http.StatusBadRequest) + return + } + data, seq, closed = src.take(ack, downPollWait) + } else { + // Pre-acknowledgement client: serve it the old fire-and-forget way so + // a mixed-version fleet keeps working. + data, closed = src.drain(downPollWait) + } if s.lease != nil && !s.lease.credentialValid() { r.closeHTTPSession(req.URL.Query().Get("session"), s) http.Error(w, "session closed", http.StatusGone) @@ -425,6 +491,9 @@ func (r *Relay) handleHDown(w http.ResponseWriter, req *http.Request) { return } w.Header().Set("Content-Type", "application/octet-stream") + if seq > 0 { + w.Header().Set(httpconn.DownSeqHeader, strconv.FormatUint(seq, 10)) + } w.WriteHeader(http.StatusOK) w.Write(data) } diff --git a/internal/relay/slowlink_test.go b/internal/relay/slowlink_test.go new file mode 100644 index 0000000..bc8c65e --- /dev/null +++ b/internal/relay/slowlink_test.go @@ -0,0 +1,379 @@ +package relay + +import ( + "crypto/ed25519" + "crypto/rand" + "crypto/sha256" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "encoding/hex" + "errors" + "fmt" + "io" + "math/big" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "wanctl/internal/delegation" + "wanctl/internal/httpconn" + "wanctl/internal/sessionauth" +) + +// The HTTP transport carries an end-to-end TLS stream over finite request / +// response pairs. A down-poll response that the controller or agent only +// receives part of therefore punches a hole in the middle of that stream, and +// TLS reports the hole as "bad record MAC" (issue #57). On a slow link the +// truncation comes from the client-side timeout firing while the body is still +// downloading; here it is injected deterministically. + +// truncatedBody hands back the first `remaining` bytes of a response and then +// fails, exactly the way a body read aborted by http.Client.Timeout does. +type truncatedBody struct { + inner io.ReadCloser + remaining int + spent bool +} + +var errTruncated = errors.New("carrier: response body truncated mid-download") + +func (b *truncatedBody) Read(p []byte) (int, error) { + if b.spent { + return 0, errTruncated + } + if len(p) > b.remaining { + p = p[:b.remaining] + } + if len(p) == 0 { + b.spent = true + return 0, errTruncated + } + n, err := b.inner.Read(p) + b.remaining -= n + if b.remaining == 0 { + b.spent = true + } + return n, err +} + +func (b *truncatedBody) Close() error { return b.inner.Close() } + +// faultyCarrier injects faults into the Nth /h/down response that actually +// carries bytes (status 200): a body cut short, or a connection that dies +// before the response is delivered at all. +type faultyCarrier struct { + base http.RoundTripper + truncateOn map[int]int // nth data-bearing down poll -> bytes to deliver first + dropOn map[int]bool // nth data-bearing down poll -> fail the request + + mu sync.Mutex + downs int + truncated int + dropped int +} + +func (f *faultyCarrier) RoundTrip(req *http.Request) (*http.Response, error) { + down := strings.HasPrefix(req.URL.Path, "/h/down") + resp, err := f.base.RoundTrip(req) + if err != nil || !down || resp.StatusCode != http.StatusOK { + return resp, err + } + f.mu.Lock() + f.downs++ + n := f.downs + cut, truncate := f.truncateOn[n] + drop := f.dropOn[n] + if truncate { + f.truncated++ + } + if drop { + f.dropped++ + } + f.mu.Unlock() + if drop { + resp.Body.Close() + return nil, fmt.Errorf("carrier: connection reset before response was read") + } + if truncate { + resp.Body = &truncatedBody{inner: resp.Body, remaining: cut} + } + return resp, nil +} + +func (f *faultyCarrier) counts() (downs, truncated, dropped int) { + f.mu.Lock() + defer f.mu.Unlock() + return f.downs, f.truncated, f.dropped +} + +func testTLSCert(t *testing.T) tls.Certificate { + t.Helper() + pub, priv, err := ed25519.GenerateKey(rand.Reader) + if err != nil { + t.Fatal(err) + } + tmpl := &x509.Certificate{ + SerialNumber: big.NewInt(1), + Subject: pkix.Name{CommonName: "wanctl-slowlink-test"}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(time.Hour), + KeyUsage: x509.KeyUsageDigitalSignature | x509.KeyUsageCertSign, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth, x509.ExtKeyUsageClientAuth}, + BasicConstraintsValid: true, + IsCA: true, + DNSNames: []string{"wanctl-slowlink-test"}, + } + der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, pub, priv) + if err != nil { + t.Fatal(err) + } + return tls.Certificate{Certificate: [][]byte{der}, PrivateKey: priv} +} + +// newHTTPTunnelSession registers a session on the relay directly, which is what +// /h/dial does once the target agent has been found. +func newHTTPTunnelSession(t *testing.T, r *Relay, sid, ns, device string) { + t.Helper() + auth := sessionauth.Open{ + Session: sid, + Device: device, + CallerNamespace: ns, + OwnerNamespace: ns, + } + r.newHTTPSession(sid, auth, delegation.Access{Namespace: ns}, "tok-alice") +} + +// pushOverFaultyCarrier runs a payload-sized push from a controller to an agent +// over the relay's HTTP transport, with faults injected into the agent's down +// polls. It returns the SHA-256 the agent computed over what it received. +func pushOverFaultyCarrier(t *testing.T, payload []byte, carrier *faultyCarrier) (sum string, err error) { + t.Helper() + r := New(EnvTokenStore("tok-alice:alice")) + srv := httptest.NewServer(r.Handler()) + defer srv.Close() + + const sid = "sess-slowlink" + newHTTPTunnelSession(t, r, sid, "alice", "home-pc") + + cert := testTLSCert(t) + serverCfg := &tls.Config{Certificates: []tls.Certificate{cert}, MinVersion: tls.VersionTLS13} + clientCfg := &tls.Config{InsecureSkipVerify: true, MinVersion: tls.VersionTLS13} + + agentRaw, err := httpconn.DialWith(t.Context(), srv.URL, sid, "agent", "tok-alice", &http.Client{Transport: carrier}) + if err != nil { + t.Fatal(err) + } + clientRaw, err := httpconn.Dial(t.Context(), srv.URL, sid, "client", "tok-alice") + if err != nil { + t.Fatal(err) + } + + type agentResult struct { + sum string + err error + } + done := make(chan agentResult, 1) + go func() { + defer agentRaw.Close() + tlsAgent := tls.Server(agentRaw, serverCfg) + h := sha256.New() + if _, copyErr := io.Copy(h, tlsAgent); copyErr != nil { + done <- agentResult{err: fmt.Errorf("agent read: %w", copyErr)} + // Still answer so the controller is not left blocked. + fmt.Fprintf(tlsAgent, "%x\n", h.Sum(nil)) + return + } + digest := fmt.Sprintf("%x", h.Sum(nil)) + fmt.Fprintf(tlsAgent, "%s\n", digest) + tlsAgent.CloseWrite() + done <- agentResult{sum: digest} + }() + + tlsClient := tls.Client(clientRaw, clientCfg) + defer clientRaw.Close() + var ctrlErr error + if _, writeErr := tlsClient.Write(payload); writeErr != nil { + ctrlErr = fmt.Errorf("controller write: %w", writeErr) + } + if ctrlErr == nil { + if closeErr := tlsClient.CloseWrite(); closeErr != nil { + ctrlErr = fmt.Errorf("controller close-write: %w", closeErr) + } + } + reply := make([]byte, 96) + var n int + if ctrlErr == nil { + var readErr error + n, readErr = io.ReadFull(tlsClient, reply[:65]) + if readErr != nil { + ctrlErr = fmt.Errorf("controller read back: %w", readErr) + } + } + select { + case res := <-done: + // The agent's TLS error is the interesting one: it is what the user + // sees as "tls: bad record MAC". The controller usually only learns + // that the session went away. + if res.err != nil { + if ctrlErr != nil { + return "", fmt.Errorf("%w (controller saw: %v)", res.err, ctrlErr) + } + return "", res.err + } + if ctrlErr != nil { + return "", ctrlErr + } + return strings.TrimSpace(string(reply[:n])), nil + case <-time.After(60 * time.Second): + if ctrlErr != nil { + return "", fmt.Errorf("timed out waiting for the agent (controller saw: %v)", ctrlErr) + } + return "", errors.New("timed out waiting for the agent") + } +} + +// TestPushSurvivesTruncatedDownPoll is the reproduction for issue #57: on a slow +// link a down-poll response is cut short, and before the fix the receiving side +// swallowed the read error and fed the truncated bytes straight into TLS. +func TestPushSurvivesTruncatedDownPoll(t *testing.T) { + payload := make([]byte, 20<<20) + if _, err := rand.Read(payload); err != nil { + t.Fatal(err) + } + want := sha256.Sum256(payload) + + carrier := &faultyCarrier{ + base: http.DefaultTransport, + truncateOn: map[int]int{3: 1024, 11: 64 << 10}, + } + got, err := pushOverFaultyCarrier(t, payload, carrier) + downs, truncated, dropped := carrier.counts() + if err != nil { + t.Fatalf("push failed after %d data-bearing down polls (%d truncated, %d dropped): %v", + downs, truncated, dropped, err) + } + if truncated != len(carrier.truncateOn) { + t.Fatalf("carrier truncated %d responses, want %d (the fault never fired)", truncated, len(carrier.truncateOn)) + } + if got != hex.EncodeToString(want[:]) { + t.Fatalf("agent received a different %d-byte stream: sha256 %s, want %s", len(payload), got, hex.EncodeToString(want[:])) + } +} + +// TestPushSurvivesDroppedDownPoll covers the other half of a flaky carrier: the +// down poll itself fails after the relay has already dequeued the bytes. +func TestPushSurvivesDroppedDownPoll(t *testing.T) { + payload := make([]byte, 8<<20) + if _, err := rand.Read(payload); err != nil { + t.Fatal(err) + } + want := sha256.Sum256(payload) + + carrier := &faultyCarrier{ + base: http.DefaultTransport, + dropOn: map[int]bool{2: true, 7: true}, + } + got, err := pushOverFaultyCarrier(t, payload, carrier) + downs, truncated, dropped := carrier.counts() + if err != nil { + t.Fatalf("push failed after %d data-bearing down polls (%d truncated, %d dropped): %v", + downs, truncated, dropped, err) + } + if dropped != len(carrier.dropOn) { + t.Fatalf("carrier dropped %d responses, want %d (the fault never fired)", dropped, len(carrier.dropOn)) + } + if got != hex.EncodeToString(want[:]) { + t.Fatalf("agent received a different %d-byte stream: sha256 %s, want %s", len(payload), got, hex.EncodeToString(want[:])) + } +} + +func TestSideQueueResendsUntilAcked(t *testing.T) { + q := newSideQueue() + q.push([]byte("first")) + q.push([]byte("second")) + + data, seq, closed := q.take(0, time.Second) + if string(data) != "firstsecond" || seq != 1 || closed { + t.Fatalf("first take = %q seq %d closed %v, want %q seq 1", data, seq, closed, "firstsecond") + } + // An unacked reader is served the same bytes again rather than the queue + // moving on without it. + q.push([]byte("third")) + again, againSeq, _ := q.take(0, time.Second) + if string(again) != "firstsecond" || againSeq != 1 { + t.Fatalf("re-send = %q seq %d, want %q seq 1", again, againSeq, "firstsecond") + } + next, nextSeq, _ := q.take(1, time.Second) + if string(next) != "third" || nextSeq != 2 { + t.Fatalf("after ack = %q seq %d, want %q seq 2", next, nextSeq, "third") + } + q.close() + tail, tailSeq, tailClosed := q.take(2, 50*time.Millisecond) + if len(tail) != 0 || tailSeq != 0 || !tailClosed { + t.Fatalf("drained closed queue = %q seq %d closed %v, want empty and closed", tail, tailSeq, tailClosed) + } +} + +func TestDrainCoalescingIsBounded(t *testing.T) { + q := newSideQueue() + chunk := make([]byte, 64<<10) + for range 16 { // 1 MiB queued, four times the cap + q.push(chunk) + } + data, _ := q.drain(time.Second) + if len(data) > maxDrainBytes { + t.Fatalf("one drain returned %d bytes, want at most %d", len(data), maxDrainBytes) + } + if len(data) != maxDrainBytes { + t.Fatalf("one drain returned %d bytes, want it to fill the %d-byte cap", len(data), maxDrainBytes) + } +} + +// A controller or agent built before the acknowledged down protocol sends no +// ack parameter. It must keep working against an updated relay. +func TestDownPollWithoutAckIsFireAndForget(t *testing.T) { + r := New(EnvTokenStore("tok-alice:alice")) + srv := httptest.NewServer(r.Handler()) + defer srv.Close() + + const sid = "sess-legacy" + newHTTPTunnelSession(t, r, sid, "alice", "home-pc") + r.session(sid).toClient.push([]byte("hello")) + + get := func(query string) *http.Response { + t.Helper() + req, err := http.NewRequest("GET", srv.URL+"/h/down?"+query, nil) + if err != nil { + t.Fatal(err) + } + req.Header.Set("Authorization", "Bearer tok-alice") + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + return resp + } + + resp := get("session=" + sid + "&role=client") + body, _ := io.ReadAll(resp.Body) + resp.Body.Close() + if resp.StatusCode != http.StatusOK || string(body) != "hello" { + t.Fatalf("legacy down poll = %d %q, want 200 %q", resp.StatusCode, body, "hello") + } + if got := resp.Header.Get(httpconn.DownSeqHeader); got != "" { + t.Fatalf("legacy down poll carried a sequence header %q, want none", got) + } + // Without an ack the relay must not hold the chunk, or the legacy reader + // would be served the same bytes forever. + q := r.session(sid).toClient + q.ackMu.Lock() + held := len(q.unacked) + q.ackMu.Unlock() + if held != 0 { + t.Fatalf("relay held %d bytes for a client that cannot ack them", held) + } +} From dbc20fec48031aa25ded40efcf473b5534295bd9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BC=A0=E4=BB=A5=E7=90=B3?= Date: Fri, 18 Sep 2026 18:01:35 +0800 Subject: [PATCH 2/7] fix(transport): serialize down polls, split at the cap, drain a closed session MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review of the acknowledged down protocol found four ways bytes could still go missing. Each is closed here with a test that fails without the fix. Overlapping polls on one direction each took a chunk. A reader whose request was abandoned while it was parked on an empty queue leaves that poll running inside drain; the retry it sends next enters drain too, and whichever finishes second overwrites the first's unacked chunk and loses it. Checking the ack, draining, numbering the chunk and storing it is now one operation per direction, admitted one poll at a time. A poll that is abandoned after it has drained still records its chunk, so the bytes wait for the next poll instead of disappearing with the response nobody received. Retrying a poll the carrier failed to deliver is safe only against a relay that holds the chunk until it is acked. Against an older one, which dequeues before it answers, the retry resumes at the chunk after the lost bytes and skips them without a word — the opposite of what the retry was added for. The relay now marks every /h/down answer with X-Wanctl-Down-Ack, and a reader that has not seen that on this session fails the read the way main does rather than retrying. A peer closing gracefully used to delete the session, which with a drain cap in place throws away everything still queued behind the first response: the far side read one cap's worth and the rest 404ed. A graceful close now stops new bytes and reports EOF once the queues run dry, but leaves the session reachable until the reader has taken all of it. Credential revocation is unchanged and still tears the session down at once. The cap was tested before appending rather than bounding the append, so a single chunk larger than it, or two chunks straddling it, came back whole. A chunk that would overshoot is now split, its tail served first next time so the stream keeps its order, and the comment states the memory each direction can actually hold. The cap itself moves from 256 KiB to 2 MiB, sized against the link in the issue: 2 MiB at 60 KB/s downloads in about 34 s, a ninth of the client's 5-minute per-request bound, while serial polling can now carry 2 MiB per round trip instead of 256 KiB. A measured 32 MiB at a simulated 100 ms RTT goes from 2.46 MiB/s to 19.24 MiB/s. TestDownPollThroughputAtRTT keeps that measurable. Fixes #57 Co-Authored-By: Claude Opus 5 (1M context) --- internal/httpconn/httpconn.go | 41 ++- internal/relay/bridge.go | 24 +- internal/relay/http.go | 204 +++++++++++--- internal/relay/slowlink_test.go | 476 +++++++++++++++++++++++++++++--- 4 files changed, 661 insertions(+), 84 deletions(-) diff --git a/internal/httpconn/httpconn.go b/internal/httpconn/httpconn.go index a9f369b..039fd7b 100644 --- a/internal/httpconn/httpconn.go +++ b/internal/httpconn/httpconn.go @@ -20,6 +20,13 @@ // only half arrives would otherwise punch a hole in the middle of the // end-to-end TLS stream — which the TLS layer reports, several minutes into a // slow push, as "tls: bad record MAC" (issue #57). +// +// Re-polling after a response the carrier failed to deliver is safe only +// against a relay that does hold that chunk. A relay from before this protocol +// dequeues before it answers, so the same retry would resume at the chunk after +// the lost bytes and skip them silently. The relay therefore marks every +// /h/down answer with X-Wanctl-Down-Ack, and a reader that has not seen it on +// this session fails the read instead of retrying. package httpconn import ( @@ -49,6 +56,7 @@ type conn struct { leftover []byte eof bool ackSeq uint64 // highest down-poll sequence fully received + ackable bool // the relay has answered this session with the ack protocol writeM sync.Mutex pending []byte @@ -69,6 +77,13 @@ const ( DownSeqHeader = "X-Wanctl-Down-Seq" DownAckParam = "ack" + // DownAckCapabilityHeader is how a relay says it holds a delivered chunk + // until it is acked. A relay without it dequeues before it answers, so a + // poll it failed to deliver is bytes that no longer exist anywhere and + // re-polling would resume at the chunk after them. Retrying is therefore + // gated on having seen this (or a sequence header) on this session. + DownAckCapabilityHeader = "X-Wanctl-Down-Ack" + // downPollAttempts bounds how many times one Read retries a down poll that // the carrier failed to deliver. The relay still holds the chunk, so a // retry is a re-send rather than a hole; the bound is what stops a @@ -112,6 +127,9 @@ func defaultClient() *http.Client { // exchange, so on a slow link it fired while a response body was still // downloading and cut the body in half (issue #57). tr.ResponseHeaderTimeout = 45 * time.Second + // The whole request still has a bound, generous enough that a full + // maxDrainBytes response downloads well inside it on the 60 KB/s link from + // issue #57 (about 34 s) even after the poll parked on the relay first. return &http.Client{Transport: tr, Timeout: 5 * time.Minute} } @@ -130,10 +148,16 @@ func (c *conn) Read(p []byte) (int, error) { return 0, io.EOF } failures := 0 - // retry reports whether a carrier failure is worth another poll. The relay - // keeps an unacked chunk, so polling again re-sends the same bytes instead - // of leaving a gap in the stream. + // retry reports whether a carrier failure is worth another poll. It is + // worth one only against a relay that has shown it holds the chunk until + // it is acked; against any other, polling again resumes after bytes that + // are already gone, which is a silent hole in the stream rather than the + // loud failure the caller needs. Pre-acknowledgement relays therefore keep + // the old behaviour: an undelivered poll fails the read. retry := func(err error) error { + if !c.ackable { + return fmt.Errorf("down poll failed and cannot be retried: %w (this relay has not answered with %s, so it does not hold undelivered bytes)", err, DownAckCapabilityHeader) + } failures++ if failures >= downPollAttempts { return fmt.Errorf("down poll failed %d times in a row: %w", failures, err) @@ -165,6 +189,15 @@ func (c *conn) Read(p []byte) (int, error) { } continue } + // Headers arrive before the body, so a response whose body is cut short + // still proves what the relay speaks. + if resp.Header.Get(DownAckCapabilityHeader) == "1" { + c.ackable = true + } + seq, seqErr := strconv.ParseUint(resp.Header.Get(DownSeqHeader), 10, 64) + if seqErr == nil && seq > 0 { + c.ackable = true + } switch resp.StatusCode { case http.StatusNoContent: resp.Body.Close() @@ -185,7 +218,7 @@ func (c *conn) Read(p []byte) (int, error) { continue } failures = 0 - if seq, convErr := strconv.ParseUint(resp.Header.Get(DownSeqHeader), 10, 64); convErr == nil && seq > 0 { + if seqErr == nil && seq > 0 { if seq <= c.ackSeq { continue // a re-send of a chunk already consumed } diff --git a/internal/relay/bridge.go b/internal/relay/bridge.go index 79f9786..cdb3a11 100644 --- a/internal/relay/bridge.go +++ b/internal/relay/bridge.go @@ -1,6 +1,7 @@ package relay import ( + "context" "io" "net/http" "sort" @@ -36,7 +37,7 @@ func (c *httpSessionConn) Read(p []byte) (int, error) { c.readMu.Lock() defer c.readMu.Unlock() for len(c.readBuf) == 0 { - data, closed := c.readQ.drain(downPollWait) + data, closed, _ := c.readQ.pollDrain(context.Background(), downPollWait) if len(data) > 0 { c.readBuf = data break @@ -89,6 +90,22 @@ func (r *Relay) newHTTPSession(sid string, auth sessionauth.Open, access delegat return s } +// closeHTTPSessionDrainable ends a session the gentle way: the queues stop +// accepting bytes and report EOF once empty, but the session stays registered +// so whichever peer is still reading collects the backlog instead of watching +// it 404. See handleHClose. +func (r *Relay) closeHTTPSessionDrainable(sid string, s *httpSession) { + if s == nil { + return + } + r.hmu.Lock() + if r.hsess[sid] == s && s.closedAt.IsZero() { + s.closedAt = time.Now() + } + r.hmu.Unlock() + s.close() +} + func (r *Relay) closeHTTPSession(sid string, s *httpSession) { if s == nil { return @@ -112,7 +129,10 @@ func (r *Relay) httpSessionConn(sid string, s *httpSession, role string) io.Read return &httpSessionConn{ readQ: readQ, writeQ: writeQ, - close: func() { r.closeHTTPSession(sid, s) }, + // The WebSocket leg ending is an ordinary end of session, so the HTTP + // peer keeps its queued bytes until it has read them, exactly as it + // does when that peer posts /h/close itself. + close: func() { r.closeHTTPSessionDrainable(sid, s) }, } } diff --git a/internal/relay/http.go b/internal/relay/http.go index af9c2c2..f844751 100644 --- a/internal/relay/http.go +++ b/internal/relay/http.go @@ -1,6 +1,7 @@ package relay import ( + "context" "encoding/json" "errors" "io" @@ -36,6 +37,14 @@ type sideQueue struct { done chan struct{} once sync.Once + // turn admits one poll at a time. Checking the ack, draining, assigning the + // sequence and storing the unacked chunk have to be a single operation: + // two polls that overlap — a reader whose request was cancelled while it + // was parked on an empty queue, plus the retry it sent afterwards — would + // otherwise each take a chunk, and the second would overwrite the first's + // unacked chunk and lose it for good. + turn chan struct{} + // A drained chunk is removed from ch before it is written to an HTTP // response, so if that response is not delivered whole the bytes are gone // and the end-to-end TLS stream has a hole in it. Readers that speak the @@ -44,10 +53,13 @@ type sideQueue struct { ackMu sync.Mutex seq uint64 unacked []byte + // head is the tail of a chunk that was split at the drain cap. It is served + // before anything still in ch, so splitting never reorders the stream. + head []byte } func newSideQueue() *sideQueue { - return &sideQueue{ch: make(chan []byte, 256), done: make(chan struct{})} + return &sideQueue{ch: make(chan []byte, 256), done: make(chan struct{}), turn: make(chan struct{}, 1)} } func (q *sideQueue) push(b []byte) bool { @@ -63,73 +75,163 @@ func (q *sideQueue) push(b []byte) bool { func (q *sideQueue) close() { q.once.Do(func() { close(q.done) }) } +// acquire admits this poll, waiting for any poll already in flight on this +// direction to finish. It reports false when the caller's request went away +// first, in which case nothing was taken from the queue. +func (q *sideQueue) acquire(ctx context.Context) bool { + // A free turn is always taken, even by a request that has already been + // abandoned: it still has to record whatever it drains as unacked rather + // than leave the queue to a poll that would renumber it. + select { + case q.turn <- struct{}{}: + return true + default: + } + select { + case q.turn <- struct{}{}: + return true + case <-ctx.Done(): + return false + } +} + +func (q *sideQueue) release() { <-q.turn } + // take is drain for a reader that acknowledges what it received. ack is the // highest sequence the reader has fully received. While an older chunk is still // outstanding it is returned again, byte for byte, under its original sequence; // only an ack that covers it lets the relay move on. The returned seq is 0 when -// there is no data. -func (q *sideQueue) take(ack uint64, timeout time.Duration) (data []byte, seq uint64, closed bool) { +// there is no data, and ok is false when the request was abandoned before this +// poll got its turn. +// +// A chunk is recorded as unacked before take returns, so a request whose +// context is cancelled after the drain — the response never reaching the reader +// — leaves the bytes to be re-served to the next poll rather than dropping them. +func (q *sideQueue) take(ctx context.Context, ack uint64, timeout time.Duration) (data []byte, seq uint64, closed, ok bool) { + if !q.acquire(ctx) { + return nil, 0, false, false + } + defer q.release() + q.ackMu.Lock() if q.unacked != nil { if ack < q.seq { data, seq = q.unacked, q.seq q.ackMu.Unlock() - return data, seq, false + return data, seq, false, true } q.unacked = nil } q.ackMu.Unlock() - data, closed = q.drain(timeout) + data, closed = q.drain(ctx, timeout) if len(data) == 0 { - return nil, 0, closed + return nil, 0, closed, true } q.ackMu.Lock() q.seq++ seq = q.seq q.unacked = data q.ackMu.Unlock() - return data, seq, false + return data, seq, false, true } -// maxDrainBytes caps how much one drain coalesces into a single down-poll -// response. Without a cap a reader that fell behind gets handed everything the -// writer queued meanwhile — megabytes in one response — and on a slow link a -// response that large cannot be downloaded inside any sane timeout. Capping it -// at the controller's write batch makes the down direction mirror the up one -// and bounds the memory an unacked chunk holds. -const maxDrainBytes = 256 << 10 +// pollDrain serves a reader that cannot acknowledge: a pre-acknowledgement +// client, or the in-process bridge, where the hand-off is a function return +// rather than a response that can half arrive. It takes the same turn as take +// so two readers never drain the same direction at once. +func (q *sideQueue) pollDrain(ctx context.Context, timeout time.Duration) (data []byte, closed, ok bool) { + if !q.acquire(ctx) { + return nil, false, false + } + defer q.release() + data, closed = q.drain(ctx, timeout) + return data, closed, true +} + +// maxDrainBytes is a hard cap on how much one drain coalesces into a single +// down-poll response. Uncapped, a reader that fell behind is handed everything +// the writer queued meanwhile — megabytes in one response — and on issue #57's +// 60 KB/s link a response that large could not finish downloading inside the +// client's request bound, so it was cut in half and the TLS stream lost a chunk +// out of its middle. +// +// The size is chosen against that link. 2 MiB at 60 KB/s takes about 34 s to +// download, and the client bounds one request at 5 minutes, so a full response +// finishes in roughly a ninth of its budget even after the 20 s the poll may +// have parked on the relay first. Larger buys nothing there and only makes the +// re-send after a truncated body more expensive; smaller costs throughput +// everywhere else, because serial polling cannot carry more than maxDrainBytes +// per round trip (2 MiB / 100 ms RTT = 20 MiB/s, against 256 KiB / 100 ms RTT +// = 2.5 MiB/s). +// +// It bounds the response because a chunk that would overshoot is split and its +// tail served first next time, not appended whole. The memory one direction +// holds is therefore maxDrainBytes for the unacked chunk, plus the tail of at +// most one split chunk, plus the 256-slot queue itself — whose chunks are +// bounded by limits.RelayHTTPUploadBytes on the /h/up path but not on the +// in-process bridge, which is why the split has to exist at all. +const maxDrainBytes = 2 << 20 // drain returns bytes available within timeout, coalescing queued chunks up to // maxDrainBytes. closed is true only when the queue is closed and no more bytes -// remain. -func (q *sideQueue) drain(timeout time.Duration) (data []byte, closed bool) { - collect := func(first []byte) []byte { - out := append([]byte{}, first...) +// remain. Callers hold the queue's turn. +func (q *sideQueue) drain(ctx context.Context, timeout time.Duration) (data []byte, closed bool) { + var out []byte + // appendCapped takes as much of b as still fits and parks the rest in head. + // out is grown by hand so that neither its length nor the memory behind it + // can pass the cap, and an idle poll that never sees a byte allocates none. + appendCapped := func(b []byte) { + if room := maxDrainBytes - len(out); len(b) > room { + q.ackMu.Lock() + q.head = b[room:] + q.ackMu.Unlock() + b = b[:room] + } + if need := len(out) + len(b); need > cap(out) { + grown := make([]byte, len(out), max(need, min(2*cap(out), maxDrainBytes))) + copy(grown, out) + out = grown + } + out = append(out, b...) + } + // fill drains what is already queued, without waiting. + fill := func() { for len(out) < maxDrainBytes { select { case b := <-q.ch: - out = append(out, b...) + appendCapped(b) default: - return out + return } } - return out } - select { - case b := <-q.ch: - return collect(b), false - default: + q.ackMu.Lock() + head := q.head + q.head = nil + q.ackMu.Unlock() + if len(head) > 0 { + appendCapped(head) + } + fill() + if len(out) > 0 { + return out, false } select { case b := <-q.ch: - return collect(b), false + appendCapped(b) + fill() + return out, false case <-time.After(timeout): return nil, false + case <-ctx.Done(): + return nil, false case <-q.done: select { case b := <-q.ch: - return collect(b), false + appendCapped(b) + fill() + return out, false default: return nil, true } @@ -151,6 +253,11 @@ type httpSession struct { lease *accessLease ownerNS string lastActive time.Time + // closedAt is set when a peer closed the session gracefully. The session + // then stops accepting new bytes but stays in the registry, so the peer + // still reading it collects what is already queued instead of having the + // remainder 404 out from under it. Guarded by hmu. + closedAt time.Time } func (s *httpSession) close() { @@ -451,6 +558,10 @@ func (r *Relay) handleHDown(w http.ResponseWriter, req *http.Request) { http.Error(w, "unauthorized", http.StatusUnauthorized) return } + // Tell the reader this relay honours ack=. A reader may only retry a poll + // the carrier failed to deliver once it has seen this, because a relay + // without it has already dequeued the bytes and a retry would skip them. + w.Header().Set(httpconn.DownAckCapabilityHeader, "1") s := r.sessionForAccess(req.URL.Query().Get("session"), access, req.URL.Query().Get("role")) if s == nil { http.Error(w, "no such session", http.StatusNotFound) @@ -464,6 +575,7 @@ func (r *Relay) handleHDown(w http.ResponseWriter, req *http.Request) { data []byte seq uint64 closed bool + served bool ) if ackParam := req.URL.Query().Get(httpconn.DownAckParam); ackParam != "" { ack, err := strconv.ParseUint(ackParam, 10, 64) @@ -471,11 +583,14 @@ func (r *Relay) handleHDown(w http.ResponseWriter, req *http.Request) { http.Error(w, "bad ack", http.StatusBadRequest) return } - data, seq, closed = src.take(ack, downPollWait) + data, seq, closed, served = src.take(req.Context(), ack, downPollWait) } else { // Pre-acknowledgement client: serve it the old fire-and-forget way so // a mixed-version fleet keeps working. - data, closed = src.drain(downPollWait) + data, closed, served = src.pollDrain(req.Context(), downPollWait) + } + if !served { + return // the reader gave up before this poll got its turn } if s.lease != nil && !s.lease.credentialValid() { r.closeHTTPSession(req.URL.Query().Get("session"), s) @@ -483,6 +598,9 @@ func (r *Relay) handleHDown(w http.ResponseWriter, req *http.Request) { return } if closed && len(data) == 0 { + // This direction is closed, empty and fully acked, so a gracefully + // closed session has nothing left to hand anyone and can go now. + r.releaseDrainedSession(req.URL.Query().Get("session"), s) http.Error(w, "session closed", http.StatusGone) return } @@ -512,13 +630,31 @@ func (r *Relay) handleHClose(w http.ResponseWriter, req *http.Request) { http.Error(w, "no such session", http.StatusNotFound) return } - delete(r.hsess, sid) + s.closedAt = time.Now() r.hmu.Unlock() + // Closing the queues stops new bytes and makes the far side see EOF once + // they run dry, but the session stays registered: with the drain cap a + // backlog needs several more polls to come out, and deleting it here would + // 404 them away. It leaves on the first poll that finds this direction + // drained, or when the sweeper finds nobody polling it any more. s.close() - if s.lease != nil { + w.WriteHeader(http.StatusOK) +} + +// releaseDrainedSession retires a gracefully closed session once a reader has +// taken everything it held. A session torn down any other way (credential +// revocation, dial failure, the idle sweeper) is already gone from the registry +// and this is a no-op. +func (r *Relay) releaseDrainedSession(sid string, s *httpSession) { + r.hmu.Lock() + retire := !s.closedAt.IsZero() && r.hsess[sid] == s + if retire { + delete(r.hsess, sid) + } + r.hmu.Unlock() + if retire && s.lease != nil { s.lease.close() } - w.WriteHeader(http.StatusOK) } func (r *Relay) session(sid string) *httpSession { @@ -577,7 +713,9 @@ func (r *Relay) startHTTPReaper() { // reapHTTP drops HTTP-registry entries whose agent stopped polling and // sessions both parties abandoned. A live agent refreshes lastSeen every poll // cycle and a live session is polled every downPollWait, so neither is at -// risk. Exported timing via the constants keeps the test honest. +// risk. This is also the bounded idle time that retires a gracefully closed +// session nobody came back to drain. Exported timing via the constants keeps +// the test honest. func (r *Relay) reapHTTP(now time.Time) { var dead []*httpSession var offline []*httpAgent diff --git a/internal/relay/slowlink_test.go b/internal/relay/slowlink_test.go index bc8c65e..7073fc3 100644 --- a/internal/relay/slowlink_test.go +++ b/internal/relay/slowlink_test.go @@ -1,6 +1,8 @@ package relay import ( + "bytes" + "context" "crypto/ed25519" "crypto/rand" "crypto/sha256" @@ -12,8 +14,10 @@ import ( "fmt" "io" "math/big" + "net" "net/http" "net/http/httptest" + "os" "strings" "sync" "testing" @@ -62,6 +66,95 @@ func (b *truncatedBody) Read(p []byte) (int, error) { func (b *truncatedBody) Close() error { return b.inner.Close() } +// These tests carry real HTTP — the same net/http client and server the product +// uses, with only the TCP listen and dial replaced by net.Pipe — so they need no +// listening socket and run in a sandbox that forbids one. +type memoryListener struct { + conns chan net.Conn + done chan struct{} + once sync.Once +} + +func (l *memoryListener) Accept() (net.Conn, error) { + select { + case c := <-l.conns: + return c, nil + case <-l.done: + return nil, net.ErrClosed + } +} + +func (l *memoryListener) Close() error { l.once.Do(func() { close(l.done) }); return nil } +func (l *memoryListener) Addr() net.Addr { return &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 80} } + +type memoryServer struct { + URL string + Transport *http.Transport + srv *http.Server + listener *memoryListener +} + +func newMemoryServer(t *testing.T, h http.Handler) *memoryServer { + t.Helper() + l := &memoryListener{conns: make(chan net.Conn), done: make(chan struct{})} + tr := &http.Transport{DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) { + a, b := net.Pipe() + select { + case l.conns <- b: + return a, nil + case <-ctx.Done(): + a.Close() + b.Close() + return nil, ctx.Err() + case <-l.done: + a.Close() + b.Close() + return nil, net.ErrClosed + } + }} + srv := &http.Server{Handler: h} + go srv.Serve(l) + s := &memoryServer{URL: "http://memory.test", Transport: tr, srv: srv, listener: l} + t.Cleanup(func() { s.Transport.CloseIdleConnections(); s.srv.Close() }) + return s +} + +type roundTripFunc func(*http.Request) (*http.Response, error) + +func (f roundTripFunc) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) } + +// legacyRelay is a relay from before the acknowledged down protocol: it +// dequeues a chunk before it answers and advertises nothing. Dropping ack= +// takes the current handler down exactly that path, and stripping the two +// response headers hides the capability the way an older build would. +func legacyRelay(h http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + q := req.URL.Query() + q.Del(httpconn.DownAckParam) + req.URL.RawQuery = q.Encode() + h.ServeHTTP(&legacyWriter{ResponseWriter: w}, req) + }) +} + +type legacyWriter struct{ http.ResponseWriter } + +func (w *legacyWriter) WriteHeader(code int) { + w.Header().Del(httpconn.DownAckCapabilityHeader) + w.Header().Del(httpconn.DownSeqHeader) + w.ResponseWriter.WriteHeader(code) +} + +// tunnelSession registers a session on a fresh relay and returns both. +func tunnelSession(t *testing.T) (*Relay, *httpSession, string) { + t.Helper() + const sid = "sess-slowlink" + r := New(EnvTokenStore("tok-alice:alice")) + newHTTPTunnelSession(t, r, sid, "alice", "home-pc") + s := r.session(sid) + t.Cleanup(func() { r.closeHTTPSession(sid, s) }) + return r, s, sid +} + // faultyCarrier injects faults into the Nth /h/down response that actually // carries bytes (status 200): a body cut short, or a connection that dies // before the response is delivered at all. @@ -153,11 +246,11 @@ func newHTTPTunnelSession(t *testing.T, r *Relay, sid, ns, device string) { func pushOverFaultyCarrier(t *testing.T, payload []byte, carrier *faultyCarrier) (sum string, err error) { t.Helper() r := New(EnvTokenStore("tok-alice:alice")) - srv := httptest.NewServer(r.Handler()) - defer srv.Close() + srv := newMemoryServer(t, r.Handler()) const sid = "sess-slowlink" newHTTPTunnelSession(t, r, sid, "alice", "home-pc") + carrier.base = srv.Transport cert := testTLSCert(t) serverCfg := &tls.Config{Certificates: []tls.Certificate{cert}, MinVersion: tls.VersionTLS13} @@ -167,7 +260,7 @@ func pushOverFaultyCarrier(t *testing.T, payload []byte, carrier *faultyCarrier) if err != nil { t.Fatal(err) } - clientRaw, err := httpconn.Dial(t.Context(), srv.URL, sid, "client", "tok-alice") + clientRaw, err := httpconn.DialWith(t.Context(), srv.URL, sid, "client", "tok-alice", &http.Client{Transport: srv.Transport}) if err != nil { t.Fatal(err) } @@ -246,10 +339,7 @@ func TestPushSurvivesTruncatedDownPoll(t *testing.T) { } want := sha256.Sum256(payload) - carrier := &faultyCarrier{ - base: http.DefaultTransport, - truncateOn: map[int]int{3: 1024, 11: 64 << 10}, - } + carrier := &faultyCarrier{truncateOn: map[int]int{2: 1024, 5: 64 << 10}} got, err := pushOverFaultyCarrier(t, payload, carrier) downs, truncated, dropped := carrier.counts() if err != nil { @@ -273,10 +363,7 @@ func TestPushSurvivesDroppedDownPoll(t *testing.T) { } want := sha256.Sum256(payload) - carrier := &faultyCarrier{ - base: http.DefaultTransport, - dropOn: map[int]bool{2: true, 7: true}, - } + carrier := &faultyCarrier{dropOn: map[int]bool{2: true, 5: true}} got, err := pushOverFaultyCarrier(t, payload, carrier) downs, truncated, dropped := carrier.counts() if err != nil { @@ -296,69 +383,331 @@ func TestSideQueueResendsUntilAcked(t *testing.T) { q.push([]byte("first")) q.push([]byte("second")) - data, seq, closed := q.take(0, time.Second) - if string(data) != "firstsecond" || seq != 1 || closed { - t.Fatalf("first take = %q seq %d closed %v, want %q seq 1", data, seq, closed, "firstsecond") + data, seq, closed, ok := q.take(context.Background(), 0, time.Second) + if string(data) != "firstsecond" || seq != 1 || closed || !ok { + t.Fatalf("first take = %q seq %d closed %v ok %v, want %q seq 1", data, seq, closed, ok, "firstsecond") } // An unacked reader is served the same bytes again rather than the queue // moving on without it. q.push([]byte("third")) - again, againSeq, _ := q.take(0, time.Second) + again, againSeq, _, _ := q.take(context.Background(), 0, time.Second) if string(again) != "firstsecond" || againSeq != 1 { t.Fatalf("re-send = %q seq %d, want %q seq 1", again, againSeq, "firstsecond") } - next, nextSeq, _ := q.take(1, time.Second) + next, nextSeq, _, _ := q.take(context.Background(), 1, time.Second) if string(next) != "third" || nextSeq != 2 { t.Fatalf("after ack = %q seq %d, want %q seq 2", next, nextSeq, "third") } q.close() - tail, tailSeq, tailClosed := q.take(2, 50*time.Millisecond) + tail, tailSeq, tailClosed, _ := q.take(context.Background(), 2, 50*time.Millisecond) if len(tail) != 0 || tailSeq != 0 || !tailClosed { t.Fatalf("drained closed queue = %q seq %d closed %v, want empty and closed", tail, tailSeq, tailClosed) } } +// Overlapping polls on one direction — a reader whose request was abandoned +// while parked on an empty queue, plus the retry it sent afterwards — used to +// each take a chunk, and the second overwrote the first's unacked chunk. Every +// poll here reports the same ack, so every poll that gets data must get the +// same data: anything else means the queue advanced past bytes nobody has +// confirmed receiving. +func TestOverlappingPollsCannotTakeDifferentChunks(t *testing.T) { + q := newSideQueue() + const pollers = 8 + got := make(chan []byte, pollers) + var wg sync.WaitGroup + for range pollers { + wg.Add(1) + go func() { + defer wg.Done() + data, _, _, _ := q.take(context.Background(), 0, 2*time.Second) + got <- data + }() + } + // Let every poller reach the queue, then feed distinct chunks slowly + // enough that they do not coalesce into one drain. + time.Sleep(50 * time.Millisecond) + for i := range pollers { + q.push(bytes.Repeat([]byte{byte('a' + i)}, 64)) + time.Sleep(5 * time.Millisecond) + } + wg.Wait() + close(got) + + var first []byte + for data := range got { + if len(data) == 0 { + continue + } + if first == nil { + first = data + continue + } + if !bytes.Equal(data, first) { + t.Fatalf("two polls that had acknowledged nothing were served different chunks, %q and %q: the queue advanced past an unacknowledged chunk", + first[:8], data[:8]) + } + } + if first == nil { + t.Fatal("no poll was served any data") + } + q.ackMu.Lock() + seq := q.seq + q.ackMu.Unlock() + if seq != 1 { + t.Fatalf("queue reached sequence %d while nothing had been acknowledged, want 1", seq) + } +} + +// A poll the reader abandoned after the relay had already drained it must leave +// the bytes as the unacked chunk, not drop them: the next poll re-serves them +// under the same sequence. +func TestAbandonedPollKeepsItsChunkForTheNextPoll(t *testing.T) { + r, s, sid := tunnelSession(t) + s.toAgent.push([]byte("FINAL")) + + poll := func(ctx context.Context) *httptest.ResponseRecorder { + req := httptest.NewRequest("GET", "/h/down?session="+sid+"&role=agent&ack=0", nil) + req.Header.Set("Authorization", "Bearer tok-alice") + rec := httptest.NewRecorder() + r.Handler().ServeHTTP(rec, req.WithContext(ctx)) + return rec + } + + gone, cancel := context.WithCancel(context.Background()) + cancel() // the reader is already gone by the time the relay answers + first := poll(gone) + if first.Code != http.StatusOK || first.Body.String() != "FINAL" { + t.Fatalf("abandoned poll = %d %q, want 200 %q", first.Code, first.Body.String(), "FINAL") + } + s.toAgent.ackMu.Lock() + held := string(s.toAgent.unacked) + s.toAgent.ackMu.Unlock() + if held != "FINAL" { + t.Fatalf("relay held %q after the poll was abandoned, want %q", held, "FINAL") + } + second := poll(context.Background()) + if second.Code != http.StatusOK || second.Body.String() != "FINAL" { + t.Fatalf("next poll = %d %q, want the same 200 %q", second.Code, second.Body.String(), "FINAL") + } + if first.Header().Get(httpconn.DownSeqHeader) != second.Header().Get(httpconn.DownSeqHeader) { + t.Fatalf("re-send was renumbered: %q then %q", + first.Header().Get(httpconn.DownSeqHeader), second.Header().Get(httpconn.DownSeqHeader)) + } +} + +// A poll may only be retried against a relay that holds the chunk until it is +// acknowledged. An older relay dequeues before it answers, so retrying there +// resumes at the chunk after the bytes that were lost — a silent hole in the +// stream. The read has to fail instead. +func TestDownPollRetriesOnlyAgainstAnAcknowledgingRelay(t *testing.T) { + // The fault lands on the second data-bearing poll, after one has already + // been answered: mid-stream, which is where issue #57 bites. + faults := map[string]*faultyCarrier{ + "response dropped": {dropOn: map[int]bool{2: true}}, + "body cut short": {truncateOn: map[int]int{2: 2}}, + } + for name, fault := range faults { + for _, legacy := range []bool{true, false} { + relayKind := "current relay" + if legacy { + relayKind = "pre-acknowledgement relay" + } + t.Run(name+"/"+relayKind, func(t *testing.T) { + r, s, sid := tunnelSession(t) + handler := r.Handler() + if legacy { + handler = legacyRelay(handler) + } + srv := newMemoryServer(t, handler) + carrier := &faultyCarrier{base: srv.Transport, dropOn: fault.dropOn, truncateOn: fault.truncateOn} + c, err := httpconn.DialWith(t.Context(), srv.URL, sid, "agent", "tok-alice", + &http.Client{Transport: carrier, Timeout: 5 * time.Second}) + if err != nil { + t.Fatal(err) + } + defer c.Close() + + // One clean exchange first. Against the current relay this is + // where the reader learns the relay holds unacknowledged + // chunks; against the older one there is nothing to learn. + s.toAgent.push([]byte("WARMUP")) + buf := make([]byte, 32) + n, err := c.Read(buf) + if err != nil || string(buf[:n]) != "WARMUP" { + t.Fatalf("warm-up read = %q %v, want %q", buf[:n], err, "WARMUP") + } + + s.toAgent.push([]byte("FIRST")) + go func() { + time.Sleep(100 * time.Millisecond) + s.toAgent.push([]byte("SECOND")) + }() + + n, readErr := c.Read(buf) + if legacy { + if readErr == nil { + t.Fatalf("an unrecoverable failure against a relay that cannot re-send was retried and the read continued with %q", buf[:n]) + } + return + } + if readErr != nil { + t.Fatalf("read against an acknowledging relay = %v, want the chunk re-sent", readErr) + } + if string(buf[:n]) != "FIRST" { + t.Fatalf("read = %q, want the re-sent %q", buf[:n], "FIRST") + } + }) + } + } +} + +// A peer closing gracefully means "no more bytes", not "discard what is +// queued". With a drain cap a backlog needs several polls to come out, so the +// session has to stay reachable until the reader has taken all of it. +func TestGracefulCloseStaysDrainable(t *testing.T) { + r, s, sid := tunnelSession(t) + srv := newMemoryServer(t, r.Handler()) + + // The peer closes while the first response is in flight, with most of the + // backlog still queued behind the drain cap. + var once sync.Once + carrier := roundTripFunc(func(req *http.Request) (*http.Response, error) { + resp, err := srv.Transport.RoundTrip(req) + if err == nil && req.URL.Path == "/h/down" && resp.StatusCode == http.StatusOK { + once.Do(func() { + closeReq := httptest.NewRequest("POST", "/h/close?session="+sid, nil) + closeReq.Header.Set("Authorization", "Bearer tok-alice") + rec := httptest.NewRecorder() + r.Handler().ServeHTTP(rec, closeReq) + if rec.Code != http.StatusOK { + t.Errorf("close = %d, want 200", rec.Code) + } + }) + } + return resp, err + }) + c, err := httpconn.DialWith(t.Context(), srv.URL, sid, "agent", "tok-alice", &http.Client{Transport: carrier}) + if err != nil { + t.Fatal(err) + } + defer c.Close() + + const chunks = 4 + want := make([]byte, 0, chunks*maxDrainBytes) + for i := range chunks { + chunk := bytes.Repeat([]byte{byte('A' + i)}, maxDrainBytes) + s.toAgent.push(chunk) + want = append(want, chunk...) + } + + got, err := io.ReadAll(c) + if err != nil { + t.Fatalf("read after a graceful close: %v", err) + } + if !bytes.Equal(got, want) { + t.Fatalf("read %d bytes after a graceful close, want all %d that were queued", len(got), len(want)) + } + if r.session(sid) != nil { + t.Fatal("a fully drained closed session was left in the registry") + } +} + +// Revoking the controller's credential is not a graceful close: it tears the +// session down at once and the queued bytes go with it. +func TestRevokedCredentialTearsDownImmediately(t *testing.T) { + a := testTransportGrant() + store := &transportGrantStore{grants: map[string]delegation.Access{"delegate": a, "owner": {Namespace: "alice"}}} + r := New(store) + s := r.newHTTPSession("revoked", sessionauth.Open{ + Session: "revoked", Device: "allowed", CallerNamespace: "alice", OwnerNamespace: "alice", + }, a, "delegate") + defer r.closeHTTPSession("revoked", s) + s.toAgent.push([]byte("SECRET")) + h := r.Handler() + if first := grantRequest(h, "GET", "/h/down?session=revoked&role=agent&ack=0", "owner"); first.Code != http.StatusOK { + t.Fatalf("first poll = %d, want 200", first.Code) + } + store.revoke("delegate") + second := grantRequest(h, "GET", "/h/down?session=revoked&role=agent&ack=0", "owner") + if second.Code != http.StatusGone || strings.Contains(second.Body.String(), "SECRET") { + t.Fatalf("poll after revocation = %d %q, want 410 with nothing held", second.Code, second.Body.String()) + } + if r.session("revoked") != nil { + t.Fatal("a revoked session stayed drainable") + } +} + +// The cap is a bound on the response, not a threshold it is tested against +// before appending: a chunk that would overshoot is split, and its tail is +// served first next time so the stream keeps its order. +func TestDrainCapSplitsRatherThanOvershoots(t *testing.T) { + cases := map[string][]int{ + "one chunk over the cap": {maxDrainBytes + 4096}, + "two chunks straddling it": {maxDrainBytes - 1, maxDrainBytes}, + "irregular sizes": {1, 7777, maxDrainBytes - 3, 999, maxDrainBytes * 2}, + "a full queue of big ones": {maxDrainBytes, maxDrainBytes, maxDrainBytes}, + } + for name, sizes := range cases { + t.Run(name, func(t *testing.T) { + q := newSideQueue() + var want []byte + for i, n := range sizes { + chunk := bytes.Repeat([]byte{byte('a' + i)}, n) + q.push(chunk) + want = append(want, chunk...) + } + var got []byte + for ack := uint64(0); len(got) < len(want); { + data, seq, _, ok := q.take(context.Background(), ack, 250*time.Millisecond) + if !ok || len(data) == 0 { + t.Fatalf("queue ran dry after %d of %d bytes", len(got), len(want)) + } + if len(data) > maxDrainBytes { + t.Fatalf("one response carried %d bytes, %d over the %d-byte cap", len(data), len(data)-maxDrainBytes, maxDrainBytes) + } + if cap(data) > maxDrainBytes { + t.Fatalf("one response was backed by %d bytes of memory, over the %d-byte cap", cap(data), maxDrainBytes) + } + got = append(got, data...) + ack = seq + } + if !bytes.Equal(got, want) { + t.Fatalf("the split stream did not come back in order: %d bytes read, %d queued", len(got), len(want)) + } + }) + } +} + func TestDrainCoalescingIsBounded(t *testing.T) { q := newSideQueue() chunk := make([]byte, 64<<10) - for range 16 { // 1 MiB queued, four times the cap + for range 4 * (maxDrainBytes / len(chunk)) { // four times the cap, queued q.push(chunk) } - data, _ := q.drain(time.Second) - if len(data) > maxDrainBytes { - t.Fatalf("one drain returned %d bytes, want at most %d", len(data), maxDrainBytes) - } + data, _ := q.drain(context.Background(), time.Second) if len(data) != maxDrainBytes { - t.Fatalf("one drain returned %d bytes, want it to fill the %d-byte cap", len(data), maxDrainBytes) + t.Fatalf("one drain returned %d bytes, want exactly the %d-byte cap", len(data), maxDrainBytes) } } // A controller or agent built before the acknowledged down protocol sends no // ack parameter. It must keep working against an updated relay. func TestDownPollWithoutAckIsFireAndForget(t *testing.T) { - r := New(EnvTokenStore("tok-alice:alice")) - srv := httptest.NewServer(r.Handler()) - defer srv.Close() + r, s, sid := tunnelSession(t) + s.toClient.push([]byte("hello")) + srv := newMemoryServer(t, r.Handler()) + client := &http.Client{Transport: srv.Transport} - const sid = "sess-legacy" - newHTTPTunnelSession(t, r, sid, "alice", "home-pc") - r.session(sid).toClient.push([]byte("hello")) - - get := func(query string) *http.Response { - t.Helper() - req, err := http.NewRequest("GET", srv.URL+"/h/down?"+query, nil) - if err != nil { - t.Fatal(err) - } - req.Header.Set("Authorization", "Bearer tok-alice") - resp, err := http.DefaultClient.Do(req) - if err != nil { - t.Fatal(err) - } - return resp + req, err := http.NewRequest("GET", srv.URL+"/h/down?session="+sid+"&role=client", nil) + if err != nil { + t.Fatal(err) + } + req.Header.Set("Authorization", "Bearer tok-alice") + resp, err := client.Do(req) + if err != nil { + t.Fatal(err) } - - resp := get("session=" + sid + "&role=client") body, _ := io.ReadAll(resp.Body) resp.Body.Close() if resp.StatusCode != http.StatusOK || string(body) != "hello" { @@ -369,7 +718,7 @@ func TestDownPollWithoutAckIsFireAndForget(t *testing.T) { } // Without an ack the relay must not hold the chunk, or the legacy reader // would be served the same bytes forever. - q := r.session(sid).toClient + q := s.toClient q.ackMu.Lock() held := len(q.unacked) q.ackMu.Unlock() @@ -377,3 +726,40 @@ func TestDownPollWithoutAckIsFireAndForget(t *testing.T) { t.Fatalf("relay held %d bytes for a client that cannot ack them", held) } } + +// TestDownPollThroughputAtRTT is a measuring stick, not an assertion: serial +// polling cannot carry more than maxDrainBytes per round trip, so the cap sets +// the ceiling on a high-latency link. Run it with +// WANCTL_TRANSPORT_THROUGHPUT=1 go test ./internal/relay -run ThroughputAtRTT -v +func TestDownPollThroughputAtRTT(t *testing.T) { + if os.Getenv("WANCTL_TRANSPORT_THROUGHPUT") == "" { + t.Skip("set WANCTL_TRANSPORT_THROUGHPUT=1 to measure down-poll throughput") + } + r, s, sid := tunnelSession(t) + srv := newMemoryServer(t, r.Handler()) + polls := 0 + carrier := roundTripFunc(func(req *http.Request) (*http.Response, error) { + if req.URL.Path == "/h/down" { + polls++ + time.Sleep(100 * time.Millisecond) // a 100 ms round trip + } + return srv.Transport.RoundTrip(req) + }) + c, err := httpconn.DialWith(t.Context(), srv.URL, sid, "agent", "tok-alice", &http.Client{Transport: carrier}) + if err != nil { + t.Fatal(err) + } + defer c.Close() + + const total = 32 << 20 + for range total / (256 << 10) { + s.toAgent.push(make([]byte, 256<<10)) + } + start := time.Now() + if _, err := io.ReadFull(c, make([]byte, total)); err != nil { + t.Fatal(err) + } + elapsed := time.Since(start) + t.Logf("%d MiB at a 100 ms per-poll round trip: polls=%d elapsed=%s throughput=%.2f MiB/s (cap %d KiB)", + total>>20, polls, elapsed, float64(total>>20)/elapsed.Seconds(), maxDrainBytes>>10) +} From 2194f790db9fd07905ec5509222137ad53622e68 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BC=A0=E4=BB=A5=E7=90=B3?= Date: Fri, 18 Sep 2026 18:10:57 +0800 Subject: [PATCH 3/7] test(transport): overlap two polls on the wire, and check the older relay is one Two gaps the author of the acknowledgement change pointed out. The fault carrier injects one fault per data-bearing poll and never leaves two polls running at once, so nothing in the suite reached the overlap through real HTTP: the queue-level test covered the primitive, not the handler. The new carrier fails a poll at the client while deliberately letting that same request keep running on the relay, detached from the client's context, so the retry arrives with the abandoned poll still parked on the queue. Feeding two chunks one at a time then reproduces the reported behaviour exactly: without the serialized turn the reader's first chunk comes back as the second one. The pre-acknowledgement relay in the retry-gating test was taken on trust. It now has to prove itself: the carrier records what the relay advertised, and each case asserts that the older relay offered neither the capability nor a sequence header while the current one offered both. That is the new client against an old relay, which had no coverage at all before this branch. Co-Authored-By: Claude Opus 5 (1M context) --- internal/relay/slowlink_test.go | 183 +++++++++++++++++++++++++++++++- 1 file changed, 179 insertions(+), 4 deletions(-) diff --git a/internal/relay/slowlink_test.go b/internal/relay/slowlink_test.go index 7073fc3..2913451 100644 --- a/internal/relay/slowlink_test.go +++ b/internal/relay/slowlink_test.go @@ -18,8 +18,10 @@ import ( "net/http" "net/http/httptest" "os" + "runtime" "strings" "sync" + "sync/atomic" "testing" "time" @@ -163,10 +165,11 @@ type faultyCarrier struct { truncateOn map[int]int // nth data-bearing down poll -> bytes to deliver first dropOn map[int]bool // nth data-bearing down poll -> fail the request - mu sync.Mutex - downs int - truncated int - dropped int + mu sync.Mutex + downs int + truncated int + dropped int + lastHeader http.Header } func (f *faultyCarrier) RoundTrip(req *http.Request) (*http.Response, error) { @@ -176,6 +179,7 @@ func (f *faultyCarrier) RoundTrip(req *http.Request) (*http.Response, error) { return resp, err } f.mu.Lock() + f.lastHeader = resp.Header.Clone() f.downs++ n := f.downs cut, truncate := f.truncateOn[n] @@ -203,6 +207,12 @@ func (f *faultyCarrier) counts() (downs, truncated, dropped int) { return f.downs, f.truncated, f.dropped } +func (f *faultyCarrier) advertised() http.Header { + f.mu.Lock() + defer f.mu.Unlock() + return f.lastHeader +} + func testTLSCert(t *testing.T) tls.Certificate { t.Helper() pub, priv, err := ed25519.GenerateKey(rand.Reader) @@ -459,6 +469,157 @@ func TestOverlappingPollsCannotTakeDifferentChunks(t *testing.T) { } } +// pendingDrains counts how many goroutines are parked inside sideQueue.drain. +// A poll the carrier abandoned keeps running on the relay, and that is the +// state this has to observe from the outside. +func pendingDrains() int { + buf := make([]byte, 1<<20) + return bytes.Count(buf[:runtime.Stack(buf, true)], []byte("(*sideQueue).drain(")) +} + +func awaitPendingDrains(n int) bool { + deadline := time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + if pendingDrains() >= n { + return true + } + time.Sleep(time.Millisecond) + } + return false +} + +// TestOverlappingDownPollsOnTheWireDoNotLoseAChunk is the overlap from the far +// side of the wire, which the per-poll fault carrier cannot reach: it fails a +// poll at the client while deliberately leaving that same request running on +// the relay, so the retry arrives with the abandoned poll still parked on the +// queue. Two chunks are then fed one at a time. Before the fix the abandoned +// poll took the first and the retry took the second, overwriting the first's +// unacknowledged chunk, and the reader silently resumed at the second. +func TestOverlappingDownPollsOnTheWireDoNotLoseAChunk(t *testing.T) { + r, s, sid := tunnelSession(t) + var entered atomic.Int32 + counted := http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + if req.URL.Path == "/h/down" { + entered.Add(1) + } + r.Handler().ServeHTTP(w, req) + }) + srv := newMemoryServer(t, counted) + + var abandonOnce sync.Once + carrier := roundTripFunc(func(req *http.Request) (*http.Response, error) { + if req.URL.Path != "/h/down" || req.URL.Query().Get(httpconn.DownAckParam) == "0" { + return srv.Transport.RoundTrip(req) // the warm-up poll goes through + } + abandoned := false + abandonOnce.Do(func() { + abandoned = true + // Detached from the client's context on purpose: the relay never + // learns this reader went away and keeps the poll parked. + detached := req.Clone(context.Background()) + go func() { + if resp, err := srv.Transport.RoundTrip(detached); err == nil { + resp.Body.Close() + } + }() + awaitPendingDrains(1) // it is on the queue before the retry is sent + }) + if abandoned { + return nil, errors.New("carrier: connection reset before the response was read") + } + return srv.Transport.RoundTrip(req) + }) + + c, err := httpconn.DialWith(t.Context(), srv.URL, sid, "agent", "tok-alice", &http.Client{Transport: carrier}) + if err != nil { + t.Fatal(err) + } + defer c.Close() + + // One clean exchange, so the reader knows the relay holds unacked chunks. + s.toAgent.push([]byte("WARMUP")) + buf := make([]byte, 64) + if n, err := c.Read(buf); err != nil || string(buf[:n]) != "WARMUP" { + t.Fatalf("warm-up read = %q %v, want %q", buf[:n], err, "WARMUP") + } + + first := bytes.Repeat([]byte("A"), 4096) + second := bytes.Repeat([]byte("B"), 4096) + // The reader hands back each chunk as it arrives, so a stream that resumed + // at the wrong one is reported as that rather than as a stall. + arrived := make(chan []byte, 2) + read := make(chan error, 1) + go func() { + for range 2 { + chunk := make([]byte, 4096) + if _, err := io.ReadFull(c, chunk); err != nil { + read <- err + return + } + arrived <- chunk + } + read <- nil + }() + nextChunk := func(want []byte, which string) { + t.Helper() + select { + case got := <-arrived: + if !bytes.Equal(got, want) { + t.Fatalf("the %s chunk read back as %q, want %q: a chunk the reader never acknowledged was dropped", + which, got[:8], want[:8]) + } + case err := <-read: + t.Fatalf("reading the %s chunk after an abandoned poll overlapped its retry: %v", which, err) + case <-time.After(15 * time.Second): + t.Fatalf("timed out on the %s chunk: the stream stalled after an abandoned poll overlapped its retry", which) + } + } + + if !awaitPendingDrains(1) { + t.Fatal("the abandoned poll never reached the queue") + } + // Wait for the retry to reach the relay as well: parked in drain before the + // fix, waiting its turn after it. + deadline := time.Now().Add(3 * time.Second) + for entered.Load() < 3 && time.Now().Before(deadline) { + time.Sleep(time.Millisecond) + } + if entered.Load() < 3 { + t.Fatalf("only %d down polls reached the relay, want the retry as well", entered.Load()) + } + time.Sleep(20 * time.Millisecond) // let the retry settle wherever it waits + + // Feed the chunks one at a time, so they cannot coalesce into one drain. + seqAfter := func(n uint64) bool { + for time.Now().Before(deadline) { + s.toAgent.ackMu.Lock() + seq := s.toAgent.seq + s.toAgent.ackMu.Unlock() + if seq >= n { + return true + } + time.Sleep(time.Millisecond) + } + return false + } + s.toAgent.push(first) + if !seqAfter(2) { + t.Fatal("no poll took the first chunk") + } + s.toAgent.push(second) + + nextChunk(first, "first") + nextChunk(second, "second") + select { + case err := <-read: + if err != nil { + t.Fatalf("read after an abandoned poll overlapped its retry: %v", err) + } + case <-time.After(15 * time.Second): + t.Fatal("timed out reading after an abandoned poll overlapped its retry") + } +} + // A poll the reader abandoned after the relay had already drained it must leave // the bytes as the unacked chunk, not drop them: the next poll re-serves them // under the same sequence. @@ -537,6 +698,20 @@ func TestDownPollRetriesOnlyAgainstAnAcknowledgingRelay(t *testing.T) { if err != nil || string(buf[:n]) != "WARMUP" { t.Fatalf("warm-up read = %q %v, want %q", buf[:n], err, "WARMUP") } + // Check the relay under test really is the one this case + // claims, rather than trusting the wrapper to have hidden the + // protocol. + advertised := carrier.advertised() + gotCap := advertised.Get(httpconn.DownAckCapabilityHeader) + gotSeq := advertised.Get(httpconn.DownSeqHeader) + if legacy && (gotCap != "" || gotSeq != "") { + t.Fatalf("the pre-acknowledgement relay advertised %s=%q %s=%q, want neither", + httpconn.DownAckCapabilityHeader, gotCap, httpconn.DownSeqHeader, gotSeq) + } + if !legacy && (gotCap != "1" || gotSeq == "") { + t.Fatalf("the current relay advertised %s=%q %s=%q, want both", + httpconn.DownAckCapabilityHeader, gotCap, httpconn.DownSeqHeader, gotSeq) + } s.toAgent.push([]byte("FIRST")) go func() { From f9b7b658937101f373e3399e92f13cd6437b73e6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BC=A0=E4=BB=A5=E7=90=B3?= Date: Fri, 18 Sep 2026 18:18:01 +0800 Subject: [PATCH 4/7] fix(transport): retire a closed session only when both directions are empty MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two things the round-two review found in the drainable close. Reaching the end of one direction says nothing about the other. A controller that has read everything it was sent hits EOF on its queue, and the session was retired on that alone — while the agent still held a delivered chunk it had not acknowledged and the tail of a chunk split at the cap. Its next poll got a 404 for bytes the relay had promised to keep. Retirement now waits until neither direction has a queued chunk, a split tail or an unacknowledged chunk. When the far side never comes back for its half, nothing fires and the idle sweeper retires the session instead, as it does for any abandoned one. Both ways a session ends gracefully are affected and both are covered. Keeping a closed session reachable also left /h/up open on it, and push chose between a writable queue and a closed one by whichever the runtime picked, so roughly half the uploads sent after the close returned 200 and queued bytes onto a session nobody would read. That broke the premise the drain check rests on. An upload to a closed session is now refused before it reaches the queue, including an empty one, which never reached the queue to be refused by it; and close and enqueue take the same lock, so once close has returned no later push can succeed. A push already waiting for room on a full queue when the close lands is genuinely concurrent with it and may still go either way, but its answer and the queue always agree. Co-Authored-By: Claude Opus 5 (1M context) --- internal/relay/http.go | 79 ++++++++++++-- internal/relay/slowlink_test.go | 177 ++++++++++++++++++++++++++++++++ 2 files changed, 248 insertions(+), 8 deletions(-) diff --git a/internal/relay/http.go b/internal/relay/http.go index f844751..0515b5b 100644 --- a/internal/relay/http.go +++ b/internal/relay/http.go @@ -62,9 +62,29 @@ func newSideQueue() *sideQueue { return &sideQueue{ch: make(chan []byte, 256), done: make(chan struct{}), turn: make(chan struct{}, 1)} } +// push enqueues a copy of b, or reports false once the queue is closed. The +// decision is taken under ackMu, which close also takes, so "is it closed" and +// "enqueue it" cannot both look true to a writer racing a close: once close has +// returned, every later push is refused. Only a push that finds the queue full +// waits outside the lock, and one that was already waiting there when the close +// landed is genuinely concurrent with it, so either answer is honest. func (q *sideQueue) push(b []byte) bool { cp := make([]byte, len(b)) copy(cp, b) + q.ackMu.Lock() + select { + case <-q.done: + q.ackMu.Unlock() + return false + default: + } + select { + case q.ch <- cp: + q.ackMu.Unlock() + return true + default: + } + q.ackMu.Unlock() select { case q.ch <- cp: return true @@ -73,7 +93,21 @@ func (q *sideQueue) push(b []byte) bool { } } -func (q *sideQueue) close() { q.once.Do(func() { close(q.done) }) } +func (q *sideQueue) close() { + q.once.Do(func() { + q.ackMu.Lock() + close(q.done) + q.ackMu.Unlock() + }) +} + +// drained reports that this direction has nothing left for anyone: no chunk +// waiting to be acknowledged, no tail left over from a split, nothing queued. +func (q *sideQueue) drained() bool { + q.ackMu.Lock() + defer q.ackMu.Unlock() + return q.unacked == nil && q.head == nil && len(q.ch) == 0 +} // acquire admits this poll, waiting for any poll already in flight on this // direction to finish. It reports false when the caller's request went away @@ -541,6 +575,14 @@ func (r *Relay) handleHUp(w http.ResponseWriter, req *http.Request) { http.Error(w, "session closed", http.StatusGone) return } + // A gracefully closed session is kept reachable so the far side can finish + // reading it, not so it can be written to again. Refusing here is what the + // queue would say anyway; saying it before the push keeps the answer the + // same for an empty body, which never reaches the queue at all. + if r.gracefullyClosed(s) { + http.Error(w, "session closed", http.StatusGone) + return + } dst := s.toAgent // role=client writes toward the agent if req.URL.Query().Get("role") == "agent" { dst = s.toClient @@ -558,9 +600,12 @@ func (r *Relay) handleHDown(w http.ResponseWriter, req *http.Request) { http.Error(w, "unauthorized", http.StatusUnauthorized) return } - // Tell the reader this relay honours ack=. A reader may only retry a poll - // the carrier failed to deliver once it has seen this, because a relay - // without it has already dequeued the bytes and a retry would skip them. + // Tell the reader this relay honours ack=. It rides on every answer past + // this point — the data-bearing 200, the empty 204, and the 400/404/410 + // refusals — but not on the 401 above, which is answered before the relay + // knows who is asking. A reader may only retry a poll the carrier failed to + // deliver once it has seen this, because a relay without it has already + // dequeued the bytes and a retry would skip them. w.Header().Set(httpconn.DownAckCapabilityHeader, "1") s := r.sessionForAccess(req.URL.Query().Get("session"), access, req.URL.Query().Get("role")) if s == nil { @@ -641,11 +686,21 @@ func (r *Relay) handleHClose(w http.ResponseWriter, req *http.Request) { w.WriteHeader(http.StatusOK) } -// releaseDrainedSession retires a gracefully closed session once a reader has -// taken everything it held. A session torn down any other way (credential -// revocation, dial failure, the idle sweeper) is already gone from the registry -// and this is a no-op. +// releaseDrainedSession retires a gracefully closed session once *both* +// directions have been taken. Reaching the end of one direction says nothing +// about the other: a controller that has read everything it was sent still +// leaves the agent holding an unacknowledged chunk and a split tail, and +// retiring the session on the first EOF would 404 those away. A session torn +// down any other way (credential revocation, dial failure, the idle sweeper) is +// already gone from the registry and this is a no-op. +// +// When the far side never comes back to drain its direction, nothing here +// fires and the session is retired by the idle sweeper instead: reapHTTP scans +// every httpAgentTTL and drops a session no one has polled for httpSessionIdle. func (r *Relay) releaseDrainedSession(sid string, s *httpSession) { + if !s.toClient.drained() || !s.toAgent.drained() { + return + } r.hmu.Lock() retire := !s.closedAt.IsZero() && r.hsess[sid] == s if retire { @@ -657,6 +712,14 @@ func (r *Relay) releaseDrainedSession(sid string, s *httpSession) { } } +// gracefullyClosed reports whether a peer has ended this session. Like +// closedAt itself it is guarded by hmu. +func (r *Relay) gracefullyClosed(s *httpSession) bool { + r.hmu.Lock() + defer r.hmu.Unlock() + return !s.closedAt.IsZero() +} + func (r *Relay) session(sid string) *httpSession { r.hmu.Lock() defer r.hmu.Unlock() diff --git a/internal/relay/slowlink_test.go b/internal/relay/slowlink_test.go index 2913451..b789cdb 100644 --- a/internal/relay/slowlink_test.go +++ b/internal/relay/slowlink_test.go @@ -19,6 +19,7 @@ import ( "net/http/httptest" "os" "runtime" + "strconv" "strings" "sync" "sync/atomic" @@ -788,6 +789,182 @@ func TestGracefulCloseStaysDrainable(t *testing.T) { } } +// downPoll drives one /h/down straight through the handler, with no server or +// carrier in the way, so a test can interleave the two directions by hand. +func downPoll(t *testing.T, r *Relay, sid, role string, ack uint64) *httptest.ResponseRecorder { + t.Helper() + req := httptest.NewRequest("GET", fmt.Sprintf("/h/down?session=%s&role=%s&ack=%d", sid, role, ack), nil) + req.Header.Set("Authorization", "Bearer tok-alice") + rec := httptest.NewRecorder() + r.Handler().ServeHTTP(rec, req) + return rec +} + +func upPost(t *testing.T, r *Relay, sid, role string, body []byte) *httptest.ResponseRecorder { + t.Helper() + req := httptest.NewRequest("POST", "/h/up?session="+sid+"&role="+role, bytes.NewReader(body)) + req.Header.Set("Authorization", "Bearer tok-alice") + rec := httptest.NewRecorder() + r.Handler().ServeHTTP(rec, req) + return rec +} + +// Reaching the end of one direction says nothing about the other. The reader +// that finished still leaves its peer holding an unacknowledged chunk and the +// tail of a split one, and retiring the session on that first EOF takes both +// away: the peer's next poll gets a 404 for bytes the relay had promised to +// keep. Covered for both ways a session ends gracefully, and for a poll that +// was already parked on the empty direction when the close landed. +func TestGracefulCloseWaitsForBothDirections(t *testing.T) { + closers := map[string]func(t *testing.T, r *Relay, s *httpSession, sid string){ + "peer posts /h/close": func(t *testing.T, r *Relay, s *httpSession, sid string) { + req := httptest.NewRequest("POST", "/h/close?session="+sid, nil) + req.Header.Set("Authorization", "Bearer tok-alice") + rec := httptest.NewRecorder() + r.Handler().ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("close = %d, want 200", rec.Code) + } + }, + "websocket leg ends": func(t *testing.T, r *Relay, s *httpSession, sid string) { + if err := r.httpSessionConn(sid, s, "client").(io.Closer).Close(); err != nil { + t.Fatalf("bridge close: %v", err) + } + }, + } + for name, closeSession := range closers { + for _, parked := range []bool{false, true} { + label := name + if parked { + label += "/with a poll already parked" + } + t.Run(label, func(t *testing.T) { + r, s, sid := tunnelSession(t) + + // The agent is mid-stream: one chunk delivered but not yet + // acknowledged, and a tail left over from splitting at the cap. + const tail = 5 + payload := bytes.Repeat([]byte("A"), maxDrainBytes+tail) + s.toAgent.push(payload) + first := downPoll(t, r, sid, "agent", 0) + if first.Code != http.StatusOK || first.Body.Len() != maxDrainBytes { + t.Fatalf("first agent poll = %d with %d bytes, want 200 with %d", first.Code, first.Body.Len(), maxDrainBytes) + } + + // The controller has read everything it was sent and parks on + // an empty queue, either before or after the close. + empty := make(chan *httptest.ResponseRecorder, 1) + if parked { + go func() { empty <- downPoll(t, r, sid, "client", 0) }() + if !awaitPendingDrains(1) { + t.Fatal("the controller poll never reached the queue") + } + } + closeSession(t, r, s, sid) + if !parked { + empty <- downPoll(t, r, sid, "client", 0) + } + select { + case rec := <-empty: + if rec.Code != http.StatusGone { + t.Fatalf("poll on the drained direction = %d, want 410", rec.Code) + } + case <-time.After(30 * time.Second): + t.Fatal("the parked controller poll never woke") + } + + // None of that is the agent's business: its chunk and the tail + // behind it must still be there. + resend := downPoll(t, r, sid, "agent", 0) + if resend.Code != http.StatusOK { + t.Fatalf("agent re-poll = %d, want the unacknowledged chunk re-sent; the other direction draining retired the session", + resend.Code) + } + if !bytes.Equal(resend.Body.Bytes(), payload[:maxDrainBytes]) { + t.Fatalf("agent re-poll returned %d bytes, want the same %d", resend.Body.Len(), maxDrainBytes) + } + seq, err := strconv.ParseUint(resend.Header().Get(httpconn.DownSeqHeader), 10, 64) + if err != nil { + t.Fatalf("re-send carried no sequence: %v", err) + } + rest := downPoll(t, r, sid, "agent", seq) + if rest.Code != http.StatusOK || !bytes.Equal(rest.Body.Bytes(), payload[maxDrainBytes:]) { + t.Fatalf("split tail = %d %q, want 200 with the last %d bytes", rest.Code, rest.Body.Bytes(), tail) + } + tailSeq, err := strconv.ParseUint(rest.Header().Get(httpconn.DownSeqHeader), 10, 64) + if err != nil { + t.Fatalf("tail carried no sequence: %v", err) + } + // Now both directions really are empty, so the session goes. + if done := downPoll(t, r, sid, "agent", tailSeq); done.Code != http.StatusGone { + t.Fatalf("poll after the last byte = %d, want 410", done.Code) + } + if r.session(sid) != nil { + t.Fatal("a session both sides had drained was left in the registry") + } + }) + } + } +} + +// A gracefully closed session stays reachable so the far side can finish +// reading it. That is not permission to write to it again: keeping the session +// registered used to leave /h/up wide open, and push picking between a writable +// queue and a closed one let some of those uploads land. +func TestUploadsAfterGracefulCloseAreRejected(t *testing.T) { + r, s, sid := tunnelSession(t) + closeReq := httptest.NewRequest("POST", "/h/close?session="+sid, nil) + closeReq.Header.Set("Authorization", "Bearer tok-alice") + rec := httptest.NewRecorder() + r.Handler().ServeHTTP(rec, closeReq) + if rec.Code != http.StatusOK { + t.Fatalf("close = %d, want 200", rec.Code) + } + accepted := 0 + for range 64 { + if upPost(t, r, sid, "client", []byte("late")).Code == http.StatusOK { + accepted++ + } + } + if accepted != 0 { + t.Fatalf("%d of 64 uploads were accepted after the session was closed", accepted) + } + if queued := len(s.toAgent.ch); queued != 0 { + t.Fatalf("%d chunks were queued onto a closed session", queued) + } + // An empty body never reaches the queue, so it has to be refused by the + // same check rather than falling through to a 200. + if code := upPost(t, r, sid, "client", nil).Code; code != http.StatusGone { + t.Fatalf("empty upload after close = %d, want 410", code) + } +} + +// An upload genuinely racing the close may land or be refused, but the answer +// and the queue have to agree: a 200 means those bytes are there to be read. +func TestUploadRacingGracefulCloseIsAllOrNothing(t *testing.T) { + for i := range 64 { + r, s, sid := tunnelSession(t) + chunk := []byte("racing") + start := make(chan struct{}) + done := make(chan struct{}) + go func() { + <-start + r.closeHTTPSessionDrainable(sid, s) + close(done) + }() + close(start) + code := upPost(t, r, sid, "client", chunk).Code + <-done + queued := len(s.toAgent.ch) + switch { + case code == http.StatusOK && queued != 1: + t.Fatalf("run %d: upload answered 200 but %d chunks are queued", i, queued) + case code != http.StatusOK && queued != 0: + t.Fatalf("run %d: upload answered %d but %d chunks were queued anyway", i, code, queued) + } + } +} + // Revoking the controller's credential is not a graceful close: it tears the // session down at once and the queued bytes go with it. func TestRevokedCredentialTearsDownImmediately(t *testing.T) { From efc3a18e2a19c65a92c3fa5790491335b2da6843 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BC=A0=E4=BB=A5=E7=90=B3?= Date: Fri, 18 Sep 2026 18:31:37 +0800 Subject: [PATCH 5/7] fix(transport): count a poll holding a chunk as data still in the queue The drain check read the fields a chunk ends up in, but a chunk spends a moment in neither. take holds the direction's turn while it receives from the channel and copies, and only afterwards takes ackMu to record what it has; in between, the channel is empty and unacked is still nil. A poll on the other direction that checked both sides during that window found the session finished and retired it, and the chunk the first poll went on to record could never be re-sent: its retry got a 404, with the response it was meant to replace already lost. A take now marks the direction from the moment it takes the turn until after it has published its chunk, in the same critical section unacked is written in, and the check treats a marked direction as non-empty. The check also requires the direction to be closed, which is what stops its answer going stale: once that holds nothing can be pushed, so only a take could move anything, and a take can only find what the check just established is not there. Both ways a session ends gracefully are affected and both are covered. The regression provokes the window rather than injecting one, by polling the empty direction in a loop while the other works through two megabytes; on the unfixed code it reproduces within the first few attempts on every run. The same interleaving driven deterministically, with the poll held between draining and recording, was used to confirm the fix separately. Fixes #57 Co-Authored-By: Claude Opus 5 (1M context) --- internal/relay/http.go | 50 ++++++++++++++++++++++---- internal/relay/slowlink_test.go | 63 +++++++++++++++++++++++++++++++++ 2 files changed, 107 insertions(+), 6 deletions(-) diff --git a/internal/relay/http.go b/internal/relay/http.go index 0515b5b..80729f1 100644 --- a/internal/relay/http.go +++ b/internal/relay/http.go @@ -56,6 +56,11 @@ type sideQueue struct { // head is the tail of a chunk that was split at the drain cap. It is served // before anything still in ch, so splitting never reorders the stream. head []byte + // inflight marks a poll that holds the turn and may be part way through + // taking bytes out. Between the receive from ch and the store into unacked + // those bytes are in no field at all, so without this the queue looks empty + // while a whole chunk is in a poll's hands. + inflight bool } func newSideQueue() *sideQueue { @@ -101,12 +106,38 @@ func (q *sideQueue) close() { }) } -// drained reports that this direction has nothing left for anyone: no chunk -// waiting to be acknowledged, no tail left over from a split, nothing queued. -func (q *sideQueue) drained() bool { +// beginTake and endTake bracket a poll's hold on the queue, so that a chunk +// which has left ch but not yet reached unacked still counts as being here. +// They are the same critical section unacked is published in, which is what +// makes settled's answer good until the queue is next touched. +func (q *sideQueue) beginTake() { + q.ackMu.Lock() + q.inflight = true + q.ackMu.Unlock() +} + +func (q *sideQueue) endTake() { + q.ackMu.Lock() + q.inflight = false + q.ackMu.Unlock() +} + +// settled reports that this direction can never hand anyone another byte: it is +// closed to new ones, no poll is part way through taking any out, and none are +// left queued, held as a split tail, or waiting to be acknowledged. +// +// The closed check is what keeps the answer from going stale. Once it holds, +// push refuses and only a take could move anything — and a take can only find +// what the other three checks just said is not there. +func (q *sideQueue) settled() bool { + select { + case <-q.done: + default: + return false + } q.ackMu.Lock() defer q.ackMu.Unlock() - return q.unacked == nil && q.head == nil && len(q.ch) == 0 + return !q.inflight && q.unacked == nil && q.head == nil && len(q.ch) == 0 } // acquire admits this poll, waiting for any poll already in flight on this @@ -146,6 +177,8 @@ func (q *sideQueue) take(ctx context.Context, ack uint64, timeout time.Duration) return nil, 0, false, false } defer q.release() + q.beginTake() + defer q.endTake() // runs after unacked is stored, before the turn is freed q.ackMu.Lock() if q.unacked != nil { @@ -179,6 +212,8 @@ func (q *sideQueue) pollDrain(ctx context.Context, timeout time.Duration) (data return nil, false, false } defer q.release() + q.beginTake() + defer q.endTake() data, closed = q.drain(ctx, timeout) return data, closed, true } @@ -690,7 +725,10 @@ func (r *Relay) handleHClose(w http.ResponseWriter, req *http.Request) { // directions have been taken. Reaching the end of one direction says nothing // about the other: a controller that has read everything it was sent still // leaves the agent holding an unacknowledged chunk and a split tail, and -// retiring the session on the first EOF would 404 those away. A session torn +// retiring the session on the first EOF would 404 those away — including a +// chunk a poll has already pulled out of the queue but not yet recorded, which +// is why settled covers a take in flight and not just the fields it writes. A +// session torn // down any other way (credential revocation, dial failure, the idle sweeper) is // already gone from the registry and this is a no-op. // @@ -698,7 +736,7 @@ func (r *Relay) handleHClose(w http.ResponseWriter, req *http.Request) { // fires and the session is retired by the idle sweeper instead: reapHTTP scans // every httpAgentTTL and drops a session no one has polled for httpSessionIdle. func (r *Relay) releaseDrainedSession(sid string, s *httpSession) { - if !s.toClient.drained() || !s.toAgent.drained() { + if !s.toClient.settled() || !s.toAgent.settled() { return } r.hmu.Lock() diff --git a/internal/relay/slowlink_test.go b/internal/relay/slowlink_test.go index b789cdb..b8a7f10 100644 --- a/internal/relay/slowlink_test.go +++ b/internal/relay/slowlink_test.go @@ -907,6 +907,69 @@ func TestGracefulCloseWaitsForBothDirections(t *testing.T) { } } +// The queue looks empty for as long as a poll has a chunk in its hands but has +// not yet recorded it as unacknowledged: it has left the channel and reached no +// field. A poll on the other direction that checks both sides during that +// window used to find the session finished and retire it, and the chunk the +// first poll then recorded could never be re-sent — its retry got a 404. +// +// This is a logical race, not a data race, so it is provoked rather than +// injected: the empty direction is polled in a tight loop while the other one +// works through two megabytes. The reviewer's probe hit it on the first +// attempt for both ways a session ends. +func TestRetirementWaitsForAPollHoldingAChunk(t *testing.T) { + defer runtime.GOMAXPROCS(runtime.GOMAXPROCS(4)) + for _, viaBridge := range []bool{false, true} { + name := "peer posts /h/close" + if viaBridge { + name = "websocket leg ends" + } + t.Run(name, func(t *testing.T) { + for attempt := 1; attempt <= 100; attempt++ { + r, s, sid := tunnelSession(t) + s.toAgent.push(bytes.Repeat([]byte("A"), 1<<20)) + s.toAgent.push(bytes.Repeat([]byte("B"), 1<<20)) + if viaBridge { + r.httpSessionConn(sid, s, "client").(io.Closer).Close() + } else { + closeReq := httptest.NewRequest("POST", "/h/close?session="+sid, nil) + closeReq.Header.Set("Authorization", "Bearer tok-alice") + r.Handler().ServeHTTP(httptest.NewRecorder(), closeReq) + } + + // The agent takes the backlog while the controller, which has + // nothing left to read, keeps asking. + agentPoll := make(chan *httptest.ResponseRecorder, 1) + go func() { agentPoll <- downPoll(t, r, sid, "agent", 0) }() + var first *httptest.ResponseRecorder + for first == nil { + select { + case first = <-agentPoll: + default: + downPoll(t, r, sid, "client", 0) + } + } + + // Treat the agent's response as lost: it was never acknowledged, + // so the relay owes it again. + if r.session(sid) == nil { + s.toAgent.ackMu.Lock() + held, seq := len(s.toAgent.unacked), s.toAgent.seq + s.toAgent.ackMu.Unlock() + t.Fatalf("attempt %d: the session was retired while a poll held a chunk (its answer was %d with %d bytes, seq %d, now recorded as %d unacknowledged bytes that can never be re-sent)", + attempt, first.Code, first.Body.Len(), seq, held) + } + retry := downPoll(t, r, sid, "agent", 0) + if retry.Code != http.StatusOK || retry.Body.Len() != first.Body.Len() { + t.Fatalf("attempt %d: re-poll for the unacknowledged chunk = %d with %d bytes, want 200 with %d", + attempt, retry.Code, retry.Body.Len(), first.Body.Len()) + } + r.closeHTTPSession(sid, s) + } + }) + } +} + // A gracefully closed session stays reachable so the far side can finish // reading it. That is not permission to write to it again: keeping the session // registered used to leave /h/up wide open, and push picking between a writable From ccada5b7f06e7587409e0c498cf96db75cc9d2b0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BC=A0=E4=BB=A5=E7=90=B3?= Date: Fri, 18 Sep 2026 18:50:42 +0800 Subject: [PATCH 6/7] fix(transport): let every reader retire the session it was the last to hold MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CI caught a session that outlived both its peers. A reader parked on one direction is still holding it while the close walks the two queues, so whoever looks first looks too early: a poll that wakes on the first queue finds the second still open, and a poll that wakes on the second finds the first still held. Retirement was attempted from exactly one place, the poll that reaches EOF, and a check that came too early was never retried — so nothing retired the session and it sat in the registry until the idle sweeper a minute later. The bridge test waits two seconds for that and failed. The in-process bridge reader is the case that cannot come right on its own: it holds a direction exactly as an HTTP poll does, but it never attempted retirement at all, so whenever it was the last one holding, no one was left to try. With one processor it reproduces on the first attempt. Retirement is now attempted by every site that can complete the last of the conditions it waits on: every poll on the way out rather than only the one that sees EOF, every read on the in-process bridge, and the close itself, which is what shuts the second queue. Whichever of them finishes last finds the session finished and retires it. A closed session with nothing queued is now retired as the close returns, so a later request finds no session and is answered 404 rather than 410 — what the code did before any of this, and either way the client treats it as the end. The upload-after-close regression moves to the case that actually exercises the check: a close with a backlog still to come out, where the session is kept on purpose and /h/up can still reach it. Co-Authored-By: Claude Opus 5 (1M context) --- internal/relay/bridge.go | 11 +++- internal/relay/http.go | 28 +++++++--- internal/relay/slowlink_test.go | 94 +++++++++++++++++++++++++++++++-- 3 files changed, 121 insertions(+), 12 deletions(-) diff --git a/internal/relay/bridge.go b/internal/relay/bridge.go index cdb3a11..443ad13 100644 --- a/internal/relay/bridge.go +++ b/internal/relay/bridge.go @@ -24,6 +24,10 @@ type httpSessionConn struct { readQ *sideQueue writeQ *sideQueue close func() + // settle runs after a read lets go of the queue. This reader is as much a + // holder of a direction as an HTTP poll is, and if it is the last one to + // let go it has to be the one that retires the session. + settle func() readMu sync.Mutex readBuf []byte @@ -36,6 +40,9 @@ func (c *httpSessionConn) Read(p []byte) (int, error) { } c.readMu.Lock() defer c.readMu.Unlock() + if c.settle != nil { + defer c.settle() + } for len(c.readBuf) == 0 { data, closed, _ := c.readQ.pollDrain(context.Background(), downPollWait) if len(data) > 0 { @@ -104,6 +111,7 @@ func (r *Relay) closeHTTPSessionDrainable(sid string, s *httpSession) { } r.hmu.Unlock() s.close() + r.releaseDrainedSession(sid, s) } func (r *Relay) closeHTTPSession(sid string, s *httpSession) { @@ -132,7 +140,8 @@ func (r *Relay) httpSessionConn(sid string, s *httpSession, role string) io.Read // The WebSocket leg ending is an ordinary end of session, so the HTTP // peer keeps its queued bytes until it has read them, exactly as it // does when that peer posts /h/close itself. - close: func() { r.closeHTTPSessionDrainable(sid, s) }, + close: func() { r.closeHTTPSessionDrainable(sid, s) }, + settle: func() { r.releaseDrainedSession(sid, s) }, } } diff --git a/internal/relay/http.go b/internal/relay/http.go index 80729f1..d0e551e 100644 --- a/internal/relay/http.go +++ b/internal/relay/http.go @@ -651,6 +651,11 @@ func (r *Relay) handleHDown(w http.ResponseWriter, req *http.Request) { if req.URL.Query().Get("role") == "agent" { src = s.toAgent } + // Whatever this poll does, it may be the one that empties the last + // direction or releases the last hold on it, so it checks on the way out. + // Reaching EOF is not the only way a session becomes finished, and a check + // that ran too early is never retried unless every site retries it. + defer r.releaseDrainedSession(req.URL.Query().Get("session"), s) var ( data []byte seq uint64 @@ -678,9 +683,6 @@ func (r *Relay) handleHDown(w http.ResponseWriter, req *http.Request) { return } if closed && len(data) == 0 { - // This direction is closed, empty and fully acked, so a gracefully - // closed session has nothing left to hand anyone and can go now. - r.releaseDrainedSession(req.URL.Query().Get("session"), s) http.Error(w, "session closed", http.StatusGone) return } @@ -715,9 +717,12 @@ func (r *Relay) handleHClose(w http.ResponseWriter, req *http.Request) { // Closing the queues stops new bytes and makes the far side see EOF once // they run dry, but the session stays registered: with the drain cap a // backlog needs several more polls to come out, and deleting it here would - // 404 them away. It leaves on the first poll that finds this direction - // drained, or when the sweeper finds nobody polling it any more. + // 404 them away. It leaves once both directions have been taken, or when + // the sweeper finds nobody polling it any more. s.close() + // Closing the second queue can be the last thing a finished session was + // waiting for, and a poll that woke on the first one has already looked. + r.releaseDrainedSession(sid, s) w.WriteHeader(http.StatusOK) } @@ -732,9 +737,16 @@ func (r *Relay) handleHClose(w http.ResponseWriter, req *http.Request) { // down any other way (credential revocation, dial failure, the idle sweeper) is // already gone from the registry and this is a no-op. // -// When the far side never comes back to drain its direction, nothing here -// fires and the session is retired by the idle sweeper instead: reapHTTP scans -// every httpAgentTTL and drops a session no one has polled for httpSessionIdle. +// Every site that can make the last of those conditions true calls this +// afterwards — each poll, each read on the in-process bridge, and the close +// itself, which is what shuts the second queue. One call alone would not do: +// whoever looks first may look while another reader still has a chunk in hand, +// and a check that came too early is only harmless if someone checks again. +// +// When the far side never comes back to drain its direction, none of them can +// succeed and the session is retired by the idle sweeper instead: reapHTTP +// scans every httpAgentTTL and drops a session no one has polled for +// httpSessionIdle. func (r *Relay) releaseDrainedSession(sid string, s *httpSession) { if !s.toClient.settled() || !s.toAgent.settled() { return diff --git a/internal/relay/slowlink_test.go b/internal/relay/slowlink_test.go index b8a7f10..08d2997 100644 --- a/internal/relay/slowlink_test.go +++ b/internal/relay/slowlink_test.go @@ -907,6 +907,78 @@ func TestGracefulCloseWaitsForBothDirections(t *testing.T) { } } +// awaitInflight waits until a reader is inside the queue, holding it. +func awaitInflight(t *testing.T, q *sideQueue) { + t.Helper() + deadline := time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + q.ackMu.Lock() + held := q.inflight + q.ackMu.Unlock() + if held { + return + } + runtime.Gosched() + } + t.Fatal("no reader ever took hold of the queue") +} + +// A reader that was already parked when the close arrived is still holding its +// direction while the close walks the two queues, so whoever looks first looks +// too early: a poll that wakes on the first queue finds the second still open, +// and a poll that wakes on the second finds the first still held. Retirement +// has to be attempted by every site that could have completed the last +// condition, because a check that came too early is only harmless if someone +// checks again. Otherwise nothing retires the session and it lingers until the +// idle sweeper, a minute later. The in-process bridge reader is one of those +// sites and used to attempt nothing at all. +func TestGracefulCloseRetiresASessionAReaderWasParkedOn(t *testing.T) { + // One processor is what CI had when this surfaced: a parked reader does not + // get to run again before the close and the polls after it do. + defer runtime.GOMAXPROCS(runtime.GOMAXPROCS(1)) + cases := map[string]func(t *testing.T, r *Relay, s *httpSession, sid string){ + "peer posts /h/close with both directions parked": func(t *testing.T, r *Relay, s *httpSession, sid string) { + go downPoll(t, r, sid, "client", 0) + go downPoll(t, r, sid, "agent", 0) + awaitInflight(t, s.toClient) + awaitInflight(t, s.toAgent) + req := httptest.NewRequest("POST", "/h/close?session="+sid, nil) + req.Header.Set("Authorization", "Bearer tok-alice") + r.Handler().ServeHTTP(httptest.NewRecorder(), req) + }, + "websocket leg ends with the bridge read parked": func(t *testing.T, r *Relay, s *httpSession, sid string) { + go r.httpSessionConn(sid, s, "client").Read(make([]byte, 64)) + awaitInflight(t, s.toClient) + if err := r.httpSessionConn(sid, s, "client").(io.Closer).Close(); err != nil { + t.Fatalf("bridge close: %v", err) + } + if code := downPoll(t, r, sid, "agent", 0).Code; code != http.StatusGone { + t.Fatalf("poll after the close = %d, want 410", code) + } + }, + } + for name, run := range cases { + t.Run(name, func(t *testing.T) { + for attempt := 1; attempt <= 30; attempt++ { + r, s, sid := tunnelSession(t) + run(t, r, s, sid) + gone := false + for deadline := time.Now().Add(2 * time.Second); time.Now().Before(deadline); { + if r.session(sid) == nil { + gone = true + break + } + runtime.Gosched() + } + if !gone { + t.Fatalf("attempt %d: the session was still registered two seconds after both sides were done with it; nothing retried the check that ran while a parked reader still held a direction", + attempt) + } + } + }) + } +} + // The queue looks empty for as long as a poll has a chunk in its hands but has // not yet recorded it as unacknowledged: it has left the channel and reached no // field. A poll on the other direction that checks both sides during that @@ -975,7 +1047,10 @@ func TestRetirementWaitsForAPollHoldingAChunk(t *testing.T) { // registered used to leave /h/up wide open, and push picking between a writable // queue and a closed one let some of those uploads land. func TestUploadsAfterGracefulCloseAreRejected(t *testing.T) { + // With a backlog still to come out, the session is deliberately kept in the + // registry, which is exactly when /h/up can still reach it. r, s, sid := tunnelSession(t) + s.toAgent.push([]byte("backlog")) closeReq := httptest.NewRequest("POST", "/h/close?session="+sid, nil) closeReq.Header.Set("Authorization", "Bearer tok-alice") rec := httptest.NewRecorder() @@ -983,6 +1058,9 @@ func TestUploadsAfterGracefulCloseAreRejected(t *testing.T) { if rec.Code != http.StatusOK { t.Fatalf("close = %d, want 200", rec.Code) } + if r.session(sid) == nil { + t.Fatal("a session with a backlog was retired at the close, so /h/up was never reachable to test") + } accepted := 0 for range 64 { if upPost(t, r, sid, "client", []byte("late")).Code == http.StatusOK { @@ -992,14 +1070,24 @@ func TestUploadsAfterGracefulCloseAreRejected(t *testing.T) { if accepted != 0 { t.Fatalf("%d of 64 uploads were accepted after the session was closed", accepted) } - if queued := len(s.toAgent.ch); queued != 0 { - t.Fatalf("%d chunks were queued onto a closed session", queued) - } // An empty body never reaches the queue, so it has to be refused by the // same check rather than falling through to a 200. if code := upPost(t, r, sid, "client", nil).Code; code != http.StatusGone { t.Fatalf("empty upload after close = %d, want 410", code) } + if queued := len(s.toAgent.ch); queued != 1 { + t.Fatalf("the queue holds %d chunks, want only the one chunk from before the close", queued) + } + // Once the backlog is out the session goes, and a late upload finds nothing. + if code := downPoll(t, r, sid, "agent", 0).Code; code != http.StatusOK { + t.Fatal("the backlog did not come out") + } + if code := downPoll(t, r, sid, "agent", 1).Code; code != http.StatusGone { + t.Fatal("the drained direction did not report EOF") + } + if code := upPost(t, r, sid, "client", []byte("later still")).Code; code == http.StatusOK { + t.Fatalf("an upload was accepted after the session was retired") + } } // An upload genuinely racing the close may land or be refused, but the answer From 435eb37c9f183763a5e46d8c4074edc85acaeaa8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BC=A0=E4=BB=A5=E7=90=B3?= Date: Fri, 18 Sep 2026 18:51:42 +0800 Subject: [PATCH 7/7] docs(transport): rewrap the retirement comment Co-Authored-By: Claude Opus 5 (1M context) --- internal/relay/http.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/internal/relay/http.go b/internal/relay/http.go index d0e551e..f0bf504 100644 --- a/internal/relay/http.go +++ b/internal/relay/http.go @@ -732,10 +732,10 @@ func (r *Relay) handleHClose(w http.ResponseWriter, req *http.Request) { // leaves the agent holding an unacknowledged chunk and a split tail, and // retiring the session on the first EOF would 404 those away — including a // chunk a poll has already pulled out of the queue but not yet recorded, which -// is why settled covers a take in flight and not just the fields it writes. A -// session torn -// down any other way (credential revocation, dial failure, the idle sweeper) is -// already gone from the registry and this is a no-op. +// is why settled covers a take in flight and not just the fields it writes. +// +// A session torn down any other way (credential revocation, dial failure, the +// idle sweeper) is already gone from the registry and this is a no-op. // // Every site that can make the last of those conditions true calls this // afterwards — each poll, each read on the in-process bridge, and the close