impl connection recovery

client/client.go

@@ -25,7 +25,9 @@ type Client struct {
 	mappings map[string]Mapping // normalized pattern -> mapping
 	pool     *WorkerPool
 
-	streamWnd int // our advertised per-stream receive window (bytes)
+	streamWnd int     // our advertised per-stream receive window (bytes)
+	shaper    *Shaper // caps aggregate egress to the hub; nil when unlimited
+	chunk     int     // DATA payload cap; shrinks below DataChunkSize at low rates
 
 	mu   sync.Mutex
 	ctrl *wire.FramedConn
@@ -48,6 +50,19 @@ func New(cfg *Config) *Client {
 		c.mappings[NormalizeAddress(m.Pattern)] = m
 	}
 	c.streamWnd = clampWindow(cfg.StreamWindowBytes)
+	// Parsed here rather than in LoadConfig because a Config may also be built
+	// directly (tests). LoadConfig has already rejected a malformed value on the
+	// file path, so a failure here can only come from a hand-built Config.
+	bps, err := parseBandwidth(cfg.MaxBandwidth)
+	if err != nil {
+		log.Printf("client: %v; continuing without a bandwidth limit", err)
+	}
+	c.shaper = NewShaper(bps)
+	c.chunk = c.shaper.chunkSize()
+	if c.shaper != nil {
+		log.Printf("egress shaped to %d B/s (burst %d B, chunk %d B)",
+			bps, int64(c.shaper.burst), c.shaper.chunk)
+	}
 	c.pool = newWorkerPool(c, cfg.MaxConn)
 	return c
 }
@@ -71,6 +86,10 @@ type session struct {
 	fc        *wire.FramedConn
 	peerWnd   int  // hub's advertised per-stream receive window
 	heartbeat bool // hub accepted mux-level PING/PONG on worker conns
+	resume    bool // hub accepted stream resumption (§7.5)
+	// hubGrace is how long the hub will hang a parked player, as advertised in
+	// SessionReady. Zero when resumption was not negotiated.
+	hubGrace time.Duration
 }
 
 // dialSession opens a TCP connection, performs the Intent-17 handshake, the
@@ -120,6 +139,9 @@ func (c *Client) dialSession(magic byte) (sess *session, err error) {
 	}
 	ts := time.Now().UnixMilli()
 	offered := FlagStreamFC | FlagWorkerHeartbeat
+	if c.cfg.resumeEnabled() {
+		offered |= FlagStreamResume
+	}
 	rekeyMsg := wire.NewWriter().U8(magic).VarInt(len(rnd)).Bytes(rnd).I64(ts).
 		VarInt(offered).VarInt(c.streamWnd).Out()
 	if err := fc.WriteFrame(rekeyMsg); err != nil {
@@ -155,6 +177,22 @@ func (c *Client) dialSession(magic byte) (sess *session, err error) {
 	if hubWnd > MaxStreamWindow {
 		hubWnd = MaxStreamWindow
 	}
+	// Resumption is negotiated per connection, and the hub's grace period rides
+	// along when it accepts. Our own grace is clamped strictly under the hub's:
+	// the client must always give up first, or the hub drops a hanging player
+	// while we are still mid-reattach. A hub that accepts the flag but omits the
+	// grace is treated as not supporting it at all rather than guessed at.
+	resume := flags&FlagStreamResume != 0
+	var hubGrace time.Duration
+	if resume {
+		graceMs, gerr := r.VarInt()
+		if gerr != nil || graceMs <= 0 {
+			log.Printf("hub accepted stream resume without advertising a grace period; disabling resume")
+			resume = false
+		} else {
+			hubGrace = time.Duration(graceMs) * time.Millisecond
+		}
+	}
 
 	// The session is live: drop the establishment deadline. From here on
 	// liveness is the heartbeat's job (and WriteFrame bounds each write).
@@ -162,14 +200,30 @@ func (c *Client) dialSession(magic byte) (sess *session, err error) {
 		return nil, err
 	}
 	ok = true
-	return &session{fc: fc, peerWnd: hubWnd, heartbeat: flags&FlagWorkerHeartbeat != 0}, nil
+	return &session{
+		fc:        fc,
+		peerWnd:   hubWnd,
+		heartbeat: flags&FlagWorkerHeartbeat != 0,
+		resume:    resume,
+		hubGrace:  hubGrace,
+	}, nil
 }
 
+// statsOn reports whether performance diagnostics are enabled. When off, no
+// counter struct is ever allocated and the instrumentation is a single branch.
+func (c *Client) statsOn() bool { return c.cfg.StatsIntervalMs > 0 }
+
 // Start establishes the control session and registers all patterns. It returns
 // once the initial connection succeeds; subsequent drops are handled in the
 // background with reconnect.
 func (c *Client) Start(ctx context.Context) error {
-	return c.connectControl(ctx)
+	if err := c.connectControl(ctx); err != nil {
+		return err
+	}
+	if c.statsOn() {
+		go c.statsLoop(ctx.Done())
+	}
+	return nil
 }
 
 func (c *Client) connectControl(ctx context.Context) error {
@@ -220,17 +274,33 @@ func (c *Client) serveControl(ctx context.Context, ctrl *ctrlSession) {
 	if ctx.Err() != nil {
 		return
 	}
-	// Reconnect with backoff.
-	for backoff := 500 * time.Millisecond; ctx.Err() == nil; backoff *= 2 {
-		if backoff > 10*time.Second {
-			backoff = 10 * time.Second
+	// Reconnect with backoff, but try immediately first. While the control
+	// session is down the hub has no live route for this client, so every
+	// millisecond of delay is a player arriving to be told there is no such
+	// server — and a session usually dies to a transient blip that the very next
+	// dial would have survived. Sleeping first spent that window unconditionally.
+	//
+	// The wait is on ctx rather than time.Sleep so shutdown is not held up by a
+	// backoff that has grown to the cap.
+	for backoff := time.Duration(0); ctx.Err() == nil; {
+		if backoff > 0 {
+			select {
+			case <-ctx.Done():
+				return
+			case <-time.After(backoff):
+			}
 		}
-		time.Sleep(backoff)
 		if err := c.connectControl(ctx); err == nil {
 			return
 		} else {
 			log.Printf("control reconnect failed: %v", err)
 		}
+		switch {
+		case backoff == 0:
+			backoff = 500 * time.Millisecond
+		case backoff < maxControlBackoff:
+			backoff = min(backoff*2, maxControlBackoff)
+		}
 	}
 }
 
@@ -299,20 +369,38 @@ func (c *Client) handleControlRequest(cid []byte, pattern, ip string, port int)
 		return
 	}
 	log.Printf("player %s:%d joined via pattern %q -> %s", ip, port, pattern, mapping.Destination)
-	wc, sid, err := c.pool.Allocate()
-	if err != nil {
-		log.Printf("worker allocate failed: %v", err)
+
+	// Allocate and publish must agree on a live conn: Allocate hands out a
+	// (conn, sid) pair that can die before we register on it, which would strand
+	// the stream in a map nothing iterates. registerStream reports that, and we
+	// simply pick another conn.
+	var st *Stream
+	var lg *leg
+	for attempt := 0; attempt < allocateAttempts; attempt++ {
+		wc, sid, err := c.pool.Allocate()
+		if err != nil {
+			log.Printf("worker allocate failed: %v", err)
+			return
+		}
+		st = newStream(c, wc, sid, cid, mapping, ip, port)
+		// Register before SYN so inbound DATA can never race ahead of the table,
+		// and start the pumps before the (bounded) SYN write so a failed or slow
+		// SYN cannot strand a stream that nothing would ever tear down.
+		if wc.registerStream(sid, st) {
+			lg = st.conn()
+			break
+		}
+		st = nil
+	}
+	if st == nil {
+		log.Printf("worker allocate failed: no live conn after %d attempts", allocateAttempts)
 		return
 	}
-	st := newStream(wc, sid, cid, mapping, ip, port)
-	// Register before SYN so inbound DATA can never race ahead of the table,
-	// and start the pumps before the (bounded) SYN write so a failed or slow
-	// SYN cannot strand a stream that nothing would ever tear down.
-	wc.registerStream(sid, st)
+
 	go st.writeLoop()
 	go st.run()
-	if err := wc.sendSyn(sid, cid); err != nil {
-		log.Printf("stream %d: SYN failed: %v", sid, err)
+	if err := lg.wc.sendSyn(lg.sid, cid); err != nil {
+		log.Printf("stream %d: SYN failed: %v", lg.sid, err)
 		st.teardown(false)
 	}
 }
@@ -330,4 +418,5 @@ func (c *Client) Close() {
 		_ = fc.Close()
 	}
 	c.pool.closeAll()
+	c.shaper.Stop()
 }

client/config.example.json

@@ -4,6 +4,10 @@
   "maxConn": 4,
   "pingIntervalMs": 20000,
   "streamWindowBytes": 262144,
+  "maxBandwidth": "",
+  "streamResume": true,
+  "resumeGraceMs": 15000,
+  "statsIntervalMs": 0,
   "mappings": [
     {
       "pattern": "mc\\.example\\.com",

client/config.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"os"
+	"strconv"
 	"strings"
 	"time"
 )
@@ -33,11 +34,26 @@ const (
 	MuxWnd  = 0x04
 	MuxPing = 0x05
 	MuxPong = 0x06
+	// MuxResume reattaches a parked stream to this conn (CID + our accepted
+	// offset); MuxResumeAck carries the hub's accepted offset and a fresh CID.
+	MuxResume    = 0x07
+	MuxResumeAck = 0x08
 
 	// MuxCtlSid is the reserved stream id carrying connection-scoped mux frames
 	// (PING/PONG). Real streams are numbered from 1.
 	MuxCtlSid = 0
 
+	// RST reason codes (optional trailing byte; absence means "unspecified").
+	// Distinguishing them matters for resume: an unknown stream is terminal,
+	// while "already bound" means a racing attempt won and this one must leave
+	// the stream alone rather than tear down a player the hub just rebound.
+	RstUnspecified     = 0x00
+	RstUnknownStream   = 0x01
+	RstAlreadyBound    = 0x02
+	RstResumeAbandoned = 0x03
+	RstFlowControl     = 0x04
+	RstDialFailed      = 0x05
+
 	FrameError = 0x7F
 
 	// SaturationThreshold caps how many streams share one worker conn once the
@@ -52,6 +68,12 @@ const (
 	// firewall) is never detected: the read loop parks forever, the dead conn
 	// stays in the pool, and no player can be served until the client restarts.
 	FlagWorkerHeartbeat = 0x02
+	// FlagStreamResume enables stream resumption (PROTOCOL.md §7.5): a worker
+	// conn drop parks its streams instead of killing them, the hub hangs the
+	// player sockets, and the client reattaches each stream byte-exactly over a
+	// fresh conn. Negotiated, so either side may decline and get today's
+	// behaviour (immediate teardown) unchanged.
+	FlagStreamResume = 0x04
 
 	// Per-stream flow-control window bounds (bytes). The advertised window is the
 	// receiver's promise of how much un-credited DATA it will buffer per stream.
@@ -64,6 +86,32 @@ const (
 	DataChunkSize = 32 * 1024
 )
 
+// Egress bandwidth shaping (see shaper.go and docs/architecture.md §6). These
+// are entirely client-local: nothing here appears on the wire.
+const (
+	// MinBandwidth floors a configured cap. Below this the tunnel cannot carry a
+	// Minecraft session at all, so such a value is a unit typo ("20bps" for
+	// "20mbps") and is rejected rather than silently clamped.
+	MinBandwidth = 8 * 1024
+
+	// ShaperBurstSeconds is how much transmission time the token bucket banks
+	// while idle. Big enough to absorb a chunk-load spike; small enough that
+	// releasing it cannot overrun the physical uplink and rebuild the standing
+	// queue the cap exists to prevent.
+	ShaperBurstSeconds = 0.2
+
+	// MinShaperBurst must exceed DataChunkSize: a request larger than the bucket
+	// could never be afforded and would park forever.
+	MinShaperBurst = 64 * 1024
+	MaxShaperBurst = 4 << 20
+
+	// ShaperSliceSeconds bounds how long one stream holds the link before the
+	// scheduler can switch, by sizing the send chunk to that much transmission
+	// time. Above ~13 Mbps this yields DataChunkSize and nothing changes.
+	ShaperSliceSeconds = 0.02
+	MinShaperChunk     = 4 * 1024
+)
+
 // Timeouts. Every tunnel socket is covered by one of these: without them a
 // silently dropped path (no FIN/RST) leaves the client parked forever.
 const (
@@ -85,6 +133,32 @@ const (
 	// heartbeat timeout can never be short enough to cause spurious drops.
 	// Applied in LoadConfig, i.e. to configs that come from disk.
 	MinPingIntervalMs = 1000
+
+	// DefaultResumeGraceMs is how long a parked stream keeps trying to reattach
+	// before giving up and closing the destination.
+	//
+	// Chosen against the backend, not the tunnel: a hung player stops answering
+	// the game server's KeepAlive, and vanilla disconnects a silent client at
+	// 30s. A longer grace would resume sessions the backend then kicks anyway.
+	DefaultResumeGraceMs = 15000
+
+	// MinResumeGraceMs floors the grace so it can always fit at least one dial;
+	// a grace shorter than HandshakeTimeout could never complete an attempt.
+	MinResumeGraceMs = 2000
+
+	// ResumeRetryDelay paces reattach attempts after a failure. Short, because
+	// the player is hanging for the whole grace period.
+	ResumeRetryDelay = 500 * time.Millisecond
+
+	// maxControlBackoff caps the control-session reconnect delay. The hub holds
+	// this client's routes only for its own registration grace, so a backoff that
+	// grew past that would strand players it is hanging on our behalf.
+	maxControlBackoff = 10 * time.Second
+
+	// ResumeAckTimeout bounds the wait for RESUME_ACK on a conn that completed
+	// its handshake but then went quiet, so a wedged hub does not consume the
+	// entire grace budget in one attempt.
+	ResumeAckTimeout = 10 * time.Second
 )
 
 // heartbeatTimeout is how long a session may go without a reply before it is
@@ -111,12 +185,87 @@ type Mapping struct {
 
 // Config is the client configuration (PROTOCOL.md §9.2).
 type Config struct {
-	Server            string    `json:"server"`
-	PSK               string    `json:"psk"`
-	MaxConn           int       `json:"maxConn"`
-	PingIntervalMs    int       `json:"pingIntervalMs"`
-	StreamWindowBytes int       `json:"streamWindowBytes"` // per-stream receive window; 0 = default
-	Mappings          []Mapping `json:"mappings"`
+	Server            string `json:"server"`
+	PSK               string `json:"psk"`
+	MaxConn           int    `json:"maxConn"`
+	PingIntervalMs    int    `json:"pingIntervalMs"`
+	StreamWindowBytes int    `json:"streamWindowBytes"` // per-stream receive window; 0 = default
+	// MaxBandwidth caps what the client sends to the hub, aggregated over every
+	// stream on every worker conn — the direction that carries the game server's
+	// output to the players, and the one a residential uplink runs out of first.
+	// Empty means no limit. See parseBandwidth for the accepted syntax.
+	MaxBandwidth string `json:"maxBandwidth"`
+	// StreamResume enables stream resumption (PROTOCOL.md §7.5). A pointer so an
+	// absent key means "on" while an explicit false disables it: with it off the
+	// client never offers the flag, allocates no retransmit buffers, and behaves
+	// exactly as a pre-resume client.
+	StreamResume *bool `json:"streamResume"`
+	// ResumeGraceMs bounds how long a parked stream keeps trying to reattach.
+	// Clamped below the hub's advertised grace so the client always gives up
+	// first and the hub is never left holding a player nobody will claim.
+	ResumeGraceMs int `json:"resumeGraceMs"`
+	// StatsIntervalMs enables the periodic performance summary; 0 (the default)
+	// disables it and costs nothing.
+	StatsIntervalMs int       `json:"statsIntervalMs"`
+	Mappings        []Mapping `json:"mappings"`
+}
+
+// resumeEnabled reports whether stream resumption is configured on.
+func (c *Config) resumeEnabled() bool { return c.StreamResume == nil || *c.StreamResume }
+
+// resumeGrace is how long a parked stream may keep trying to reattach.
+func (c *Config) resumeGrace() time.Duration {
+	ms := c.ResumeGraceMs
+	if ms <= 0 {
+		ms = DefaultResumeGraceMs
+	}
+	if ms < MinResumeGraceMs {
+		ms = MinResumeGraceMs
+	}
+	return time.Duration(ms) * time.Millisecond
+}
+
+// bandwidthUnits maps a rate suffix to its value in bytes per second. Bit units
+// are decimal because that is what ISPs quote; byte units are binary to match
+// streamWindowBytes. Ordered longest-suffix-first so "kbps" is not read as
+// "bps", nor "gb/s" as "b/s".
+var bandwidthUnits = []struct {
+	suffix string
+	mul    float64
+}{
+	{"gbps", 1e9 / 8}, {"gbit", 1e9 / 8},
+	{"mbps", 1e6 / 8}, {"mbit", 1e6 / 8},
+	{"kbps", 1e3 / 8}, {"kbit", 1e3 / 8},
+	{"gb/s", 1 << 30}, {"mb/s", 1 << 20}, {"kb/s", 1 << 10},
+	{"bps", 1.0 / 8},
+	{"b/s", 1},
+}
+
+// parseBandwidth converts a human-readable rate to bytes per second. The empty
+// string means "no limit" and yields 0.
+//
+//	"20mbps"  20 megabits/s  = 2500000 B/s
+//	"512kbps" 512 kilobits/s =   64000 B/s
+//	"2MB/s"   2 mebibytes/s  = 2097152 B/s
+//	"1500000" a bare number is already bytes per second
+func parseBandwidth(s string) (int64, error) {
+	s = strings.TrimSpace(s)
+	if s == "" {
+		return 0, nil
+	}
+	lower := strings.ToLower(s)
+	num, mul := lower, 1.0
+	for _, u := range bandwidthUnits {
+		if strings.HasSuffix(lower, u.suffix) {
+			num, mul = strings.TrimSpace(lower[:len(lower)-len(u.suffix)]), u.mul
+			break
+		}
+	}
+	v, err := strconv.ParseFloat(num, 64)
+	if err != nil || v <= 0 {
+		return 0, fmt.Errorf(`maxBandwidth: cannot read %q as a rate (try "20mbps", "2MB/s", or bytes per second)`, s)
+	}
+	return int64(v * mul), nil
 }
 
 // LoadConfig reads and validates a JSON config file.
@@ -147,6 +296,14 @@ func LoadConfig(path string) (*Config, error) {
 	if c.PingIntervalMs < MinPingIntervalMs {
 		c.PingIntervalMs = MinPingIntervalMs
 	}
+	// Parsed here only to fail fast on a bad value; New does the real conversion.
+	bps, err := parseBandwidth(c.MaxBandwidth)
+	if err != nil {
+		return nil, err
+	}
+	if bps > 0 && bps < MinBandwidth {
+		return nil, fmt.Errorf("maxBandwidth %q is only %d B/s; the minimum is %d B/s", c.MaxBandwidth, bps, MinBandwidth)
+	}
 	if len(c.Mappings) == 0 {
 		return nil, fmt.Errorf("at least one mapping is required")
 	}

client/pool_test.go

@@ -85,7 +85,9 @@ func TestAllocateSpreadsAcrossConns(t *testing.T) {
 		return &WorkerConn{pool: p, streams: make(map[int]*Stream), nextSid: 1, done: make(chan struct{})}
 	}
 	p.conns = []*WorkerConn{newConn()}
-	p.conns[0].registerStream(1, &Stream{sid: 1})
+	if !p.conns[0].registerStream(1, &Stream{}) {
+		t.Fatal("registerStream refused on a live conn")
+	}
 
 	// One conn holding a stream, pool below maxConn: growth is warranted.
 	_, bestCount := p.leastLoadedLocked()

client/resume.go

@@ -0,0 +1,332 @@
+package client
+
+import (
+	"errors"
+	"log"
+	"time"
+
+	"github.com/iceBear67/redapricot/client/wire"
+)
+
+// Stream resumption (PROTOCOL.md §7.5).
+//
+// A worker conn carries many players but is only the middle leg of each: when
+// it dies, both terminal sockets are usually still perfectly healthy. Tearing
+// the streams down therefore throws away working connections because a
+// replaceable transport failed — one conntrack expiry disconnects everyone on
+// that conn.
+//
+// Instead the stream parks: the destination socket stays open, the hub hangs the
+// player socket, and the client reattaches over a fresh conn. Correctness rests
+// on retransmission being byte-exact. Bytes handed to a dying socket are lost
+// with no notification, and the frame cipher cannot be resynchronized, so each
+// side replays from the offset the other reports it accepted.
+
+var (
+	errResumeUnknown  = errors.New("hub does not know this stream")
+	errResumeRaced    = errors.New("another reattach already bound this stream")
+	errResumeRefused  = errors.New("hub refused the reattach")
+	errResumeTimeout  = errors.New("no RESUME_ACK from the hub")
+	errResumeConnLost = errors.New("the conn carrying the reattach died")
+	errResumeTooOld   = errors.New("hub accepted past what we still hold")
+	errResumeNoResume = errors.New("hub does not support stream resumption")
+)
+
+// resumeGrace is how long a stream parked from a conn may keep trying, clamped
+// under what the hub advertised. The client must always give up first: a hub
+// that drops the player while we are still reattaching would leave us pumping a
+// destination nobody is reading.
+func (c *Client) resumeGrace(hubGrace time.Duration) time.Duration {
+	grace := c.cfg.resumeGrace()
+	if hubGrace > 0 && hubGrace < grace {
+		grace = hubGrace
+	}
+	return grace
+}
+
+// park suspends a stream whose worker conn died instead of destroying it, and
+// starts trying to reattach. Reports false when the stream cannot be parked, in
+// which case the caller tears it down as before.
+//
+// A stream already closing (finPending) is not parked: the hub has said the
+// player is gone, so there is nothing left to preserve.
+func (s *Stream) park(grace time.Duration) bool {
+	if !s.resumable {
+		return false
+	}
+	s.mu.Lock()
+	if s.closed || s.finPending {
+		s.mu.Unlock()
+		return false
+	}
+	already := s.parked
+	s.parked = true
+	if s.stats != nil && !already {
+		s.parkedAt = time.Now()
+	}
+	s.mu.Unlock()
+
+	if already {
+		// A reattach was already in flight and had registered this stream on the
+		// conn that just died — which is how it got here at all. That attempt
+		// still owns the stream, so reporting failure would have the caller tear
+		// down a player that is mid-recovery. Fail its wait immediately rather
+		// than let it sit out the ack timeout: the grace budget is small, and
+		// spending ten seconds of it waiting on a socket that is already gone is
+		// the difference between reattaching and dropping the player.
+		s.deliverResume(resumeResult{err: errResumeConnLost})
+		return true
+	}
+	go s.resumeLoop(grace)
+	return true
+}
+
+// resumeLoop reattaches the stream, retrying until it succeeds or the grace
+// period runs out.
+//
+// Attempts are started right up to the deadline rather than reserving a whole
+// dial's worth of budget for the last one. Reserving it would be self-defeating
+// — the grace and HandshakeTimeout are the same order of magnitude, so the
+// reservation can consume the entire budget and leave no attempt at all — and
+// overshooting is safe: an attempt that lands after the hub has dropped the
+// player is answered with RST(unknown stream) and tears down cleanly.
+func (s *Stream) resumeLoop(grace time.Duration) {
+	deadline := time.Now().Add(grace)
+	for {
+		if s.isClosed() {
+			return
+		}
+		if !time.Now().Before(deadline) {
+			break
+		}
+		err := s.tryResume()
+		if err == nil {
+			return
+		}
+		if errors.Is(err, errResumeRaced) {
+			// Another attempt owns the stream now; leaving it alone is the whole
+			// point — tearing down here would kill a player the hub considers live.
+			return
+		}
+		if errors.Is(err, errResumeUnknown) || errors.Is(err, errResumeTooOld) {
+			// Terminal: the hub has no state for this stream (it restarted, the
+			// grace expired, or a load balancer sent us to a different instance).
+			log.Printf("stream resume abandoned: %v", err)
+			s.teardown(false)
+			return
+		}
+		select {
+		case <-s.done:
+			return
+		case <-time.After(ResumeRetryDelay):
+		}
+	}
+	log.Printf("stream resume gave up after %s; closing destination", grace)
+	// No FIN: the only conns we could send it on are the ones that just failed
+	// us. The hub drops the hanging player when its own grace expires.
+	s.teardown(false)
+}
+
+// tryResume performs one reattach attempt: find a live conn, claim a stream id
+// on it, send RESUME, and replay from wherever the hub says it got to.
+func (s *Stream) tryResume() error {
+	wc, sid, err := s.allocateForResume()
+	if err != nil {
+		return err
+	}
+
+	s.mu.Lock()
+	cid := s.cid
+	accepted := s.acceptedOffset
+	delivered := s.deliveredOffset
+	wait := make(chan resumeResult, 1)
+	s.resumeWait = wait
+	s.mu.Unlock()
+
+	msg := wire.NewWriter().U8(MuxResume).VarInt(sid).Bytes(cid).
+		I64(accepted).I64(delivered).Out()
+	if err := wc.fc.WriteFrame(msg); err != nil {
+		s.abandonAttempt(wc, sid)
+		return err
+	}
+
+	var res resumeResult
+	select {
+	case res = <-wait:
+	case <-s.done:
+		// Torn down while waiting. teardown only deregisters the leg the stream
+		// was bound to, which is not this one, so the claim made above has to be
+		// withdrawn here or it stays in the new conn's table forever.
+		s.abandonAttempt(wc, sid)
+		return errResumeRefused
+	case <-time.After(ResumeAckTimeout):
+		s.abandonAttempt(wc, sid)
+		return errResumeTimeout
+	}
+	if res.err != nil {
+		s.abandonAttempt(wc, sid)
+		return res.err
+	}
+	return s.completeResume(wc, sid, res)
+}
+
+// allocateForResume picks a live conn that will honour a reattach.
+func (s *Stream) allocateForResume() (*WorkerConn, int, error) {
+	for attempt := 0; attempt < allocateAttempts; attempt++ {
+		wc, sid, err := s.client.pool.Allocate()
+		if err != nil {
+			return nil, 0, err
+		}
+		// Re-checked per conn, not assumed from the dead one: this may be a
+		// different or restarted hub. Sending RESUME to a hub that does not know
+		// the frame type would hang the player for the rest of the grace waiting
+		// for an answer that is never coming.
+		if !wc.resume {
+			return nil, 0, errResumeNoResume
+		}
+		if wc.registerStream(sid, s) {
+			return wc, sid, nil
+		}
+	}
+	return nil, 0, errResumeRefused
+}
+
+// abandonAttempt withdraws a failed attempt from the conn it was made on, so a
+// retry can never leave two RESUMEs outstanding for one stream.
+func (s *Stream) abandonAttempt(wc *WorkerConn, sid int) {
+	wc.removeStream(sid)
+	s.mu.Lock()
+	s.resumeWait = nil
+	s.mu.Unlock()
+}
+
+// completeResume rebinds the stream to its new conn and replays what the hub is
+// missing, holding sendMu throughout so live traffic cannot overtake the replay.
+func (s *Stream) completeResume(wc *WorkerConn, sid int, res resumeResult) error {
+	s.sendMu.Lock()
+	// Delivery is a strictly stronger fact than credit — the hub only credits what
+	// it has delivered — so the reported offset can be adopted wholesale. Doing so
+	// also repairs the ledger: the grants destroyed by the outage are exactly the
+	// gap between the two, and without this the retained region would carry that
+	// dead prefix for the rest of the stream's life.
+	s.ackedOffset.Store(res.delivered)
+	s.un.advance(res.delivered)
+	replay := s.un.from(res.accepted)
+	if replay == nil {
+		s.sendMu.Unlock()
+		s.abandonAttempt(wc, sid)
+		return errResumeTooOld
+	}
+	// Three offsets, three jobs, and conflating any two of them breaks something
+	// different.
+	//
+	// What to replay is measured from what the hub *accepted* — the bytes it
+	// never received. What the window should be is measured from what it
+	// *delivered*, because the window is a promise about undelivered bytes.
+	// It cannot be measured from what it *credited*: credit arrives as deltas,
+	// and the grants in flight when the connection died are gone for good, so a
+	// window derived from them would be permanently short — and, when a full
+	// window was outstanding at the drop, permanently zero. That is a deadlock,
+	// not a slowdown: no credit can arrive because nothing can be sent.
+	outstanding := s.un.length()
+	replayed := s.un.end() - res.accepted
+
+	// Publish the new binding before any frame goes out on it, and as one value:
+	// stream ids restart at 1 per conn, so a half-updated pair would address a
+	// different player's stream.
+	s.leg.Store(&leg{wc: wc, sid: sid})
+
+	s.mu.Lock()
+	// Restated, not patched. The window is a delta ledger and the outage tore a
+	// hole in it; deriving it afresh from the delivered offset closes the hole
+	// exactly, whatever was lost.
+	s.sendWnd = wc.sendWndInit - int(outstanding)
+	if s.sendWnd < 0 {
+		s.sendWnd = 0
+	}
+	// Symmetrically, our own pending credit is discarded rather than flushed:
+	// the delivered offset we reported already tells the hub everything those
+	// deltas would have, and sending both would grant the same bytes twice.
+	// Counting resumes from this baseline.
+	s.consumed = 0
+	if len(res.cid) == CIDLen {
+		s.cid = res.cid // fresh capability, so a CID is never reusable twice
+	}
+	s.parked = false
+	s.resumeWait = nil
+	owedFin := s.finToHub
+	if s.stats != nil {
+		s.stats.resumes++
+		s.stats.hung += time.Since(s.parkedAt)
+		s.stats.replayBytes += replayed
+	}
+	s.cond.Broadcast() // release acquireSendWnd and any parked writer
+	s.mu.Unlock()
+
+	for len(replay) > 0 {
+		n := len(replay)
+		if n > s.client.chunk {
+			n = s.client.chunk
+		}
+		if err := wc.sendData(sid, replay[:n]); err != nil {
+			s.sendMu.Unlock()
+			return err
+		}
+		replay = replay[n:]
+	}
+	s.sendMu.Unlock()
+
+	// A destination that closed while we were parked owed the hub a FIN that had
+	// nowhere to go at the time.
+	if owedFin {
+		wc.sendFin(sid)
+		s.teardown(false)
+		return nil
+	}
+	log.Printf("stream %d resumed (%d bytes replayed, %d outstanding)", sid, replayed, outstanding)
+	return nil
+}
+
+// deliverResume hands an answer to a reattach that is waiting for one. Reports
+// false when no attempt was in flight, so the caller can treat the frame as it
+// would on any live stream.
+func (s *Stream) deliverResume(res resumeResult) bool {
+	s.mu.Lock()
+	ch := s.resumeWait
+	s.resumeWait = nil
+	s.mu.Unlock()
+	if ch == nil {
+		return false
+	}
+	ch <- res // buffered, and read at most once per attempt
+	return true
+}
+
+// onRst applies an RST, using the reason to tell a stream that is genuinely gone
+// from one that a racing reattach has taken over.
+func (s *Stream) onRst(reason int) {
+	err := errResumeRefused
+	switch reason {
+	case RstUnknownStream:
+		err = errResumeUnknown
+	case RstAlreadyBound:
+		err = errResumeRaced
+	}
+	if s.deliverResume(resumeResult{err: err}) {
+		return
+	}
+	s.teardown(false)
+}
+
+// noteFinWhileParked records a FIN the stream owes the hub but cannot send,
+// because the only conn it has is the one that just died. Reports false when the
+// stream is not parked and the caller should send it normally.
+func (s *Stream) noteFinWhileParked() bool {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if !s.parked {
+		return false
+	}
+	s.finToHub = true
+	return true
+}

client/resume_bench_test.go

@@ -0,0 +1,104 @@
+package client
+
+import (
+	"testing"
+)
+
+// The retained region is the one real cost stream resumption adds to the send
+// path: a chunk has to survive past the frame write, so it is copied. These
+// pin both halves of that claim — that the copy is the only cost, and that
+// disabling the feature removes it entirely rather than merely shrinking it.
+
+func benchStream(resumable bool) *Stream {
+	s := &Stream{resumable: resumable}
+	return s
+}
+
+// BenchmarkRetainChunk measures what emit adds over a bare frame write: the
+// trim-and-append into the retained region. Compare the two variants; the delta
+// is the per-byte copy the feature costs.
+func BenchmarkRetainChunk(b *testing.B) {
+	chunk := make([]byte, DataChunkSize)
+	window := int64(DefaultStreamWindow)
+
+	b.Run("resume-on", func(b *testing.B) {
+		s := benchStream(true)
+		b.SetBytes(int64(len(chunk)))
+		b.ReportAllocs()
+		b.ResetTimer()
+		for i := 0; i < b.N; i++ {
+			// Model the steady state: credit trails one window behind, so the
+			// buffer trims about as fast as it grows and stays bounded.
+			acked := s.un.end() - window
+			if acked < 0 {
+				acked = 0
+			}
+			s.un.advance(acked)
+			s.un.append(chunk)
+		}
+		if got := int64(s.un.length()); got > window+int64(len(chunk)) {
+			b.Fatalf("retained region grew past one window: %d", got)
+		}
+	})
+
+	b.Run("resume-off", func(b *testing.B) {
+		s := benchStream(false)
+		b.SetBytes(int64(len(chunk)))
+		b.ReportAllocs()
+		b.ResetTimer()
+		for i := 0; i < b.N; i++ {
+			if s.resumable {
+				s.un.advance(0)
+				s.un.append(chunk)
+			}
+		}
+	})
+}
+
+// TestResumeDisabledAllocatesNothing pins the off switch at the level that
+// matters. It is easy for a feature flag to stop the wire behaviour while
+// leaving the bookkeeping running, which would keep the memory cost and the
+// per-byte copy for a user who explicitly turned it off — a partial revert that
+// nobody would notice.
+func TestResumeDisabledAllocatesNothing(t *testing.T) {
+	chunk := make([]byte, DataChunkSize)
+	s := benchStream(false)
+
+	allocs := testing.AllocsPerRun(1000, func() {
+		if s.resumable {
+			s.un.advance(0)
+			s.un.append(chunk)
+		}
+	})
+	if allocs != 0 {
+		t.Fatalf("resume disabled still allocated %.1f times per send", allocs)
+	}
+	if s.un.buf != nil {
+		t.Fatalf("resume disabled still allocated a retained region of %d bytes", cap(s.un.buf))
+	}
+}
+
+// TestRetainedRegionStaysWithinWindow is the memory bound the design rests on:
+// flow control already caps outstanding bytes at one window, so the retained
+// region needs no cap of its own. If that ever stopped holding, a busy stream
+// would grow without limit and the hub would be the first to notice.
+func TestRetainedRegionStaysWithinWindow(t *testing.T) {
+	const window = DefaultStreamWindow
+	chunk := make([]byte, DataChunkSize)
+	var u unackedBuf
+
+	for i := 0; i < 5000; i++ {
+		// A sender may never have more than one window outstanding, which is
+		// exactly what acquireSendWnd enforces before emit is ever reached.
+		if u.length()+len(chunk) > window {
+			u.advance(u.base() + int64(len(chunk)))
+		}
+		u.append(chunk)
+		if u.length() > window {
+			t.Fatalf("round %d: retained %d bytes for a %d-byte window", i, u.length(), window)
+		}
+	}
+	if cap(u.buf) > 4*window {
+		t.Fatalf("backing array grew to %d for a %d-byte window", cap(u.buf), window)
+	}
+}

client/shaper.go

@@ -0,0 +1,268 @@
+package client
+
+import (
+	"sync"
+	"time"
+)
+
+// minShaperWait floors the dispatcher's sleep so floating-point dust in the
+// token arithmetic cannot spin it.
+const minShaperWait = time.Millisecond
+
+// Shaper caps the aggregate rate at which the client writes DATA to the hub and
+// divides that budget across streams.
+//
+// The credit windows of PROTOCOL.md §7.3 bound how many bytes may be *in flight*
+// per stream; they say nothing about bytes per *second*. That is the gap this
+// fills. On a residential uplink one player loading chunks will otherwise
+// saturate the line and push every other player's keepalive past its timeout.
+//
+// Two mechanisms are layered:
+//
+//   - A token bucket sets the long-run rate and the size of the burst that may
+//     be spent after an idle period.
+//   - Start-time fair queueing decides who spends those tokens. A global virtual
+//     clock advances with each grant; every stream remembers the virtual time at
+//     which its last request finished. A request is stamped
+//     max(share.vfinish, vclock) and the lowest stamp is served first, so a
+//     stream that keeps sending pushes its own stamp further out and yields to
+//     quieter streams. The clamp to vclock is what keeps bursts cheap: a stream
+//     returning from idle is pulled back to the head of the clock, so it cannot
+//     hoard credit while it was idle, but it is not punished for the idleness
+//     either. One stream alone gets the whole rate.
+//
+// A nil *Shaper means "no limit"; every method short-circuits, so call sites do
+// not branch.
+type Shaper struct {
+	rate  float64 // bytes per second
+	burst float64 // token bucket capacity, bytes
+	chunk int     // how much a caller should request at a time
+
+	mu      sync.Mutex
+	tokens  float64
+	last    time.Time
+	vclock  float64      // virtual time, in bytes of service granted
+	waiting []*shaperReq // unordered; the dispatcher scans for the lowest vstart
+
+	wake chan struct{} // cap 1, non-blocking: nudges the dispatcher
+	done chan struct{}
+	once sync.Once
+}
+
+// shaperShare is one stream's position in the fair queue. It lives on the
+// Stream and dies with it; a fresh share starts at zero and is clamped up to
+// the current virtual clock on its first request.
+type shaperShare struct{ vfinish float64 }
+
+// shaperReq is one pending Acquire. granted and membership in Shaper.waiting
+// are both guarded by Shaper.mu.
+type shaperReq struct {
+	n       int
+	vstart  float64
+	grant   chan struct{}
+	granted bool
+}
+
+// NewShaper builds a shaper for the given rate. A non-positive rate returns nil,
+// which every method treats as "unlimited".
+func NewShaper(bytesPerSec int64) *Shaper {
+	if bytesPerSec <= 0 {
+		return nil
+	}
+	rate := float64(bytesPerSec)
+
+	burst := rate * ShaperBurstSeconds
+	// The floor is a correctness constraint, not a preference: a request larger
+	// than the bucket could never be afforded and would park forever.
+	if burst < MinShaperBurst {
+		burst = MinShaperBurst
+	}
+	if burst > MaxShaperBurst {
+		burst = MaxShaperBurst
+	}
+
+	chunk := int(rate * ShaperSliceSeconds)
+	if chunk < MinShaperChunk {
+		chunk = MinShaperChunk
+	}
+	if chunk > DataChunkSize {
+		chunk = DataChunkSize
+	}
+
+	sh := &Shaper{
+		rate:   rate,
+		burst:  burst,
+		chunk:  chunk,
+		tokens: burst,
+		last:   time.Now(),
+		wake:   make(chan struct{}, 1),
+		done:   make(chan struct{}),
+	}
+	go sh.dispatch()
+	return sh
+}
+
+// chunkSize is how many bytes a sender should offer per request. It is sized to
+// ShaperSliceSeconds of transmission so no stream holds the link for long before
+// the scheduler can switch: at 1 Mbps a full 32 KiB chunk takes ~256 ms, which is
+// enough dead air to drag other players towards a keepalive timeout.
+func (sh *Shaper) chunkSize() int {
+	if sh == nil {
+		return DataChunkSize
+	}
+	return sh.chunk
+}
+
+// Acquire blocks until n bytes of bandwidth budget are available for the stream
+// owning share. It returns false only when cancel fires first, in which case
+// nothing was charged.
+//
+// cancel is the stream's done channel: a stream torn down while parked here must
+// not keep a goroutine (and its Stream) alive waiting for tokens it will never
+// use.
+func (sh *Shaper) Acquire(share *shaperShare, n int, cancel <-chan struct{}) bool {
+	if sh == nil || n <= 0 {
+		return true
+	}
+	req := &shaperReq{n: n, grant: make(chan struct{})}
+
+	sh.mu.Lock()
+	// Stamp the request and reserve this stream's slot in virtual time up front,
+	// so a stream cannot queue many requests at the same cheap stamp.
+	req.vstart = share.vfinish
+	if req.vstart < sh.vclock {
+		req.vstart = sh.vclock
+	}
+	share.vfinish = req.vstart + float64(n)
+	sh.waiting = append(sh.waiting, req)
+	sh.mu.Unlock()
+	sh.nudge()
+
+	select {
+	case <-req.grant:
+		return true
+	case <-sh.done:
+		// Shaping stopped: let live traffic through rather than stalling it.
+		sh.mu.Lock()
+		sh.removeLocked(req)
+		sh.mu.Unlock()
+		return true
+	case <-cancel:
+		sh.mu.Lock()
+		granted := req.granted
+		if !granted {
+			sh.removeLocked(req)
+		}
+		sh.mu.Unlock()
+		return granted
+	}
+}
+
+// Stop shuts the dispatcher down and releases everyone parked in Acquire.
+func (sh *Shaper) Stop() {
+	if sh == nil {
+		return
+	}
+	sh.once.Do(func() { close(sh.done) })
+}
+
+// dispatch is the single goroutine that hands out tokens. It sleeps exactly as
+// long as the next waiter needs rather than polling on a fixed tick, so an idle
+// shaper costs nothing.
+func (sh *Shaper) dispatch() {
+	for {
+		wait := sh.grantReady()
+		var tick <-chan time.Time
+		var timer *time.Timer
+		if wait > 0 {
+			timer = time.NewTimer(wait)
+			tick = timer.C
+		}
+		select {
+		case <-tick:
+		case <-sh.wake:
+		case <-sh.done:
+			if timer != nil {
+				timer.Stop()
+			}
+			return
+		}
+		if timer != nil {
+			timer.Stop()
+		}
+	}
+}
+
+// grantReady refills the bucket and grants every waiter it can afford, lowest
+// virtual start time first. It returns how long until the next waiter becomes
+// affordable, or 0 when nothing is pending.
+func (sh *Shaper) grantReady() time.Duration {
+	sh.mu.Lock()
+	defer sh.mu.Unlock()
+
+	now := time.Now()
+	if elapsed := now.Sub(sh.last); elapsed > 0 {
+		sh.tokens += sh.rate * elapsed.Seconds()
+		if sh.tokens > sh.burst {
+			sh.tokens = sh.burst
+		}
+		sh.last = now
+	}
+
+	for {
+		req := sh.headLocked()
+		if req == nil {
+			return 0
+		}
+		// Callers stay under chunkSize, which NewShaper keeps below the bucket.
+		// Should a future caller not, wait for a full bucket rather than for a
+		// token count that can never be reached, and let the balance go negative:
+		// the debt is repaid by the next refill, so the long-run rate still holds.
+		need := min(float64(req.n), sh.burst)
+		if need > sh.tokens {
+			wait := time.Duration((need - sh.tokens) / sh.rate * float64(time.Second))
+			if wait < minShaperWait {
+				wait = minShaperWait
+			}
+			return wait
+		}
+		sh.tokens -= float64(req.n)
+		// The clock follows the request being served, never runs ahead of it.
+		if req.vstart > sh.vclock {
+			sh.vclock = req.vstart
+		}
+		req.granted = true
+		sh.removeLocked(req)
+		close(req.grant)
+	}
+}
+
+// headLocked returns the pending request with the lowest virtual start time.
+// A linear scan is deliberate: the queue holds at most one entry per live
+// stream (tens, not thousands), so a heap would cost more in complexity than it
+// saves in comparisons.
+func (sh *Shaper) headLocked() *shaperReq {
+	var best *shaperReq
+	for _, w := range sh.waiting {
+		if best == nil || w.vstart < best.vstart {
+			best = w
+		}
+	}
+	return best
+}
+
+func (sh *Shaper) removeLocked(req *shaperReq) {
+	for i, w := range sh.waiting {
+		if w == req {
+			sh.waiting = append(sh.waiting[:i], sh.waiting[i+1:]...)
+			return
+		}
+	}
+}
+
+func (sh *Shaper) nudge() {
+	select {
+	case sh.wake <- struct{}{}:
+	default:
+	}
+}

client/shaper_test.go

@@ -0,0 +1,231 @@
+package client
+
+import (
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func TestParseBandwidth(t *testing.T) {
+	cases := []struct {
+		in   string
+		want int64
+	}{
+		{"", 0},
+		{"20mbps", 2_500_000},
+		{"20Mbps", 2_500_000},
+		{"20 mbps", 2_500_000},
+		{"1.5mbit", 187_500},
+		{"512kbps", 64_000},
+		{"1gbps", 125_000_000},
+		{"2MB/s", 2 << 20},
+		{"500kb/s", 500 << 10},
+		{"1GB/s", 1 << 30},
+		{"8bps", 1},
+		{"4096b/s", 4096},
+		{"1500000", 1_500_000}, // bare number is already bytes/sec
+	}
+	for _, c := range cases {
+		got, err := parseBandwidth(c.in)
+		if err != nil {
+			t.Errorf("parseBandwidth(%q): unexpected error %v", c.in, err)
+			continue
+		}
+		if got != c.want {
+			t.Errorf("parseBandwidth(%q) = %d, want %d", c.in, got, c.want)
+		}
+	}
+
+	for _, bad := range []string{"fast", "20megabits", "-5mbps", "0", "0mbps", "mbps", "20 mb ps"} {
+		if _, err := parseBandwidth(bad); err == nil {
+			t.Errorf("parseBandwidth(%q): expected an error", bad)
+		}
+	}
+}
+
+// A nil shaper is the "unlimited" case and must be safe on every path, because
+// call sites deliberately do not branch on it.
+func TestNilShaperIsUnlimited(t *testing.T) {
+	var sh *Shaper
+	if sh = NewShaper(0); sh != nil {
+		t.Fatal("NewShaper(0) should return nil")
+	}
+	if got := sh.chunkSize(); got != DataChunkSize {
+		t.Errorf("nil chunkSize = %d, want %d", got, DataChunkSize)
+	}
+	if !sh.Acquire(&shaperShare{}, 1<<20, nil) {
+		t.Error("nil Acquire should always succeed")
+	}
+	sh.Stop() // must not panic
+}
+
+// The virtual-time bookkeeping is what makes the shaper fair, so assert it
+// directly. The rate is high enough that tokens never bind, leaving only the
+// stamping under test — no timing, no flakiness.
+func TestShaperIdleStreamCannotHoardCredit(t *testing.T) {
+	sh := NewShaper(1 << 30)
+	defer sh.Stop()
+
+	var heavy, light shaperShare
+	for i := 0; i < 10; i++ {
+		if !sh.Acquire(&heavy, 1000, nil) {
+			t.Fatal("acquire failed")
+		}
+	}
+	if heavy.vfinish != 10000 {
+		t.Errorf("heavy.vfinish = %v, want 10000", heavy.vfinish)
+	}
+	sh.mu.Lock()
+	vclock := sh.vclock
+	sh.mu.Unlock()
+	if vclock != 9000 {
+		t.Errorf("vclock = %v, want 9000 (the stamp of the last request served)", vclock)
+	}
+
+	// light was idle for all of it. Its stale vfinish of 0 must be clamped up to
+	// the current clock: it may not bank the virtual time it never spent, which
+	// is what would let it starve heavy on return.
+	if !sh.Acquire(&light, 1000, nil) {
+		t.Fatal("acquire failed")
+	}
+	if light.vfinish != vclock+1000 {
+		t.Errorf("light.vfinish = %v, want %v (clamped to the clock, not 1000)", light.vfinish, vclock+1000)
+	}
+}
+
+func TestShaperEnforcesRate(t *testing.T) {
+	const rate = 1 << 20 // 1 MiB/s
+	sh := NewShaper(rate)
+	defer sh.Stop()
+
+	var share shaperShare
+	const total = 512 << 10
+	const chunk = 8 << 10
+
+	start := time.Now()
+	for sent := 0; sent < total; sent += chunk {
+		if !sh.Acquire(&share, chunk, nil) {
+			t.Fatal("acquire failed")
+		}
+	}
+	elapsed := time.Since(start)
+
+	// The bucket starts full, so the burst is free and only the remainder is
+	// paced: (512 KiB - 200 KiB) / 1 MiB/s ≈ 300 ms. Bounds are wide on purpose.
+	if elapsed < 200*time.Millisecond {
+		t.Errorf("sent %d bytes at %d B/s in only %v; the cap is not being enforced", total, rate, elapsed)
+	}
+	if elapsed > time.Second {
+		t.Errorf("took %v, far longer than the ~300ms the rate implies", elapsed)
+	}
+}
+
+// An idle stream must be able to spend the banked burst at once, otherwise a
+// player joining pays for the cap in visible chunk-loading latency.
+func TestShaperAllowsBurst(t *testing.T) {
+	sh := NewShaper(1 << 20)
+	defer sh.Stop()
+
+	var share shaperShare
+	start := time.Now()
+	for i := 0; i < 6; i++ {
+		if !sh.Acquire(&share, 32<<10, nil) { // 192 KiB, inside the 200 KiB bucket
+			t.Fatal("acquire failed")
+		}
+	}
+	if elapsed := time.Since(start); elapsed > 100*time.Millisecond {
+		t.Errorf("burst of 192 KiB took %v; the bucket should have covered it instantly", elapsed)
+	}
+}
+
+// The point of the whole exercise: a stream that never stops asking must not
+// crowd another one out.
+func TestShaperSharesFairlyBetweenStreams(t *testing.T) {
+	sh := NewShaper(1 << 20)
+	defer sh.Stop()
+
+	stop := make(chan struct{})
+	var counts [2]atomic.Int64
+	var wg sync.WaitGroup
+	for i := range counts {
+		wg.Add(1)
+		go func(i int) {
+			defer wg.Done()
+			var share shaperShare
+			for {
+				if !sh.Acquire(&share, 4<<10, stop) {
+					return
+				}
+				counts[i].Add(4 << 10)
+			}
+		}(i)
+	}
+
+	// Spend the token bucket before measuring, the way the idle-credit test
+	// raises the rate so tokens never bind: isolate the property under test.
+	//
+	// While the bucket has tokens there is no queue to arbitrate — every request
+	// is granted the moment it arrives, and being the stream that is owed service
+	// only helps when both are enqueued at the same instant. The burst is
+	// therefore first-come-first-served by construction, and at 0.2s of
+	// transmission it is a quarter of a 600ms window, enough to swamp the result:
+	// a run where one goroutine happened to win the bucket landed at 520192 vs
+	// 315392, which is exactly "one took the whole burst, then the two split the
+	// remainder evenly".
+	//
+	// Fairness here is a steady-state property, and that is what matters in
+	// practice — the bucket is empty whenever the link is actually busy.
+	time.Sleep(250 * time.Millisecond)
+	counts[0].Store(0)
+	counts[1].Store(0)
+
+	time.Sleep(600 * time.Millisecond)
+	close(stop)
+	wg.Wait()
+
+	a, b := counts[0].Load(), counts[1].Load()
+	if a == 0 || b == 0 {
+		t.Fatalf("one stream was starved entirely: %d vs %d", a, b)
+	}
+	lo, hi := min(a, b), max(a, b)
+	t.Logf("steady-state split: %d vs %d bytes (%.3fx)", a, b, float64(hi)/float64(lo))
+	if float64(hi) > 1.35*float64(lo) {
+		t.Errorf("unfair split: %d vs %d bytes (>35%% apart)", a, b)
+	}
+}
+
+// A stream torn down while parked must release immediately and leave no trace
+// in the queue, or its goroutine (and the Stream it closes over) leaks.
+func TestShaperAcquireCancels(t *testing.T) {
+	sh := NewShaper(MinBandwidth) // 8 KiB/s: a parked request would wait seconds
+	defer sh.Stop()
+
+	var share shaperShare
+	if !sh.Acquire(&share, MinShaperBurst, nil) { // drain the bucket
+		t.Fatal("acquire failed")
+	}
+
+	cancel := make(chan struct{})
+	result := make(chan bool, 1)
+	go func() { result <- sh.Acquire(&share, 32<<10, cancel) }()
+
+	time.Sleep(50 * time.Millisecond)
+	close(cancel)
+
+	select {
+	case ok := <-result:
+		if ok {
+			t.Error("Acquire returned true after cancellation")
+		}
+	case <-time.After(time.Second):
+		t.Fatal("Acquire did not return after its cancel channel closed")
+	}
+
+	sh.mu.Lock()
+	n := len(sh.waiting)
+	sh.mu.Unlock()
+	if n != 0 {
+		t.Errorf("%d cancelled request(s) left in the queue", n)
+	}
+}

client/stats.go

@@ -0,0 +1,223 @@
+package client
+
+import (
+	"fmt"
+	"log"
+	"sort"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+)
+
+// Performance diagnostics.
+//
+// The question an operator actually has is "why is this tunnel slow?", and
+// nothing here could answer it before. A stream that is not moving bytes is
+// blocked on exactly one of three things:
+//
+//   - the flow-control window — the peer is not draining to its terminal
+//     socket, so the bottleneck is past the tunnel (a struggling game server, a
+//     player on a bad link);
+//   - the shaper — the configured bandwidth cap is the binding constraint, and
+//     raising it is the fix;
+//   - the peer socket itself — bytes move, but slowly, which points at the path
+//     rather than at either end.
+//
+// Those three are indistinguishable from throughput alone and call for
+// completely different responses, so they are counted apart.
+//
+// Cost. Both structs are nil unless statsIntervalMs is set, so the default is a
+// single predictable branch per event and no allocation at all. When enabled,
+// counters sit under locks the code already holds; only the frame counters use
+// atomics, because the read loop must never queue behind a send. A clock is read
+// only when a goroutine is about to block, never per chunk — if nothing stalls,
+// nothing is timed.
+
+// streamStats accumulates one stream's lifetime. Guarded by Stream.mu.
+type streamStats struct {
+	opened time.Time
+
+	bytesUp   int64 // destination -> hub
+	bytesDown int64 // hub -> destination
+
+	windowStall time.Duration // blocked with no send credit
+	shaperStall time.Duration // blocked on the bandwidth cap
+	qPeak       int           // high-water mark of the receive queue
+
+	resumes     int
+	hung        time.Duration // total time parked awaiting a reattach
+	replayBytes int64
+}
+
+// connStats accumulates one worker conn's lifetime.
+type connStats struct {
+	opened    time.Time
+	framesIn  atomic.Int64
+	framesOut atomic.Int64
+	writeErrs atomic.Int64
+
+	// Round-trip time of the mux heartbeat. The probe already carries a
+	// timestamp that the peer echoes and both sides currently throw away, so
+	// this measures tunnel latency for no added cost — and it is the best signal
+	// available for head-of-line blocking, where one stream's backlog delays
+	// every other stream sharing the connection.
+	mu      sync.Mutex
+	rttLast time.Duration
+	rttMin  time.Duration
+	rttMax  time.Duration
+	rttSum  time.Duration
+	rttN    int64
+}
+
+func (cs *connStats) observeRTT(d time.Duration) {
+	if cs == nil || d < 0 {
+		return // a nonce we cannot read as one of our own timestamps
+	}
+	cs.mu.Lock()
+	defer cs.mu.Unlock()
+	cs.rttLast = d
+	if cs.rttN == 0 || d < cs.rttMin {
+		cs.rttMin = d
+	}
+	if d > cs.rttMax {
+		cs.rttMax = d
+	}
+	cs.rttSum += d
+	cs.rttN++
+}
+
+func (cs *connStats) rtt() (last, min, avg, max time.Duration) {
+	cs.mu.Lock()
+	defer cs.mu.Unlock()
+	if cs.rttN == 0 {
+		return 0, 0, 0, 0
+	}
+	return cs.rttLast, cs.rttMin, cs.rttSum / time.Duration(cs.rttN), cs.rttMax
+}
+
+// stallClock times a block without charging the path that does not block: the
+// clock is read only once a wait is actually about to happen.
+type stallClock struct{ start time.Time }
+
+func (t *stallClock) begin(on bool) {
+	if on && t.start.IsZero() {
+		t.start = time.Now()
+	}
+}
+
+func (t *stallClock) elapsed() time.Duration {
+	if t.start.IsZero() {
+		return 0
+	}
+	return time.Since(t.start)
+}
+
+// statsLoop prints one aggregate line per interval. Never started when
+// statsIntervalMs is 0, which is the default.
+func (c *Client) statsLoop(stop <-chan struct{}) {
+	ticker := time.NewTicker(time.Duration(c.cfg.StatsIntervalMs) * time.Millisecond)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-stop:
+			return
+		case <-ticker.C:
+			log.Print(c.StatsLine())
+		}
+	}
+}
+
+// StatsLine renders the current pool and per-conn state as one greppable line.
+// Exported so tests and embedders can sample it without waiting for the ticker.
+func (c *Client) StatsLine() string {
+	conns := c.pool.snapshot()
+	var b strings.Builder
+	fmt.Fprintf(&b, "stats conns=%d", len(conns))
+
+	live, parked := 0, 0
+	for _, wc := range conns {
+		wc.mu.Lock()
+		streams := make([]*Stream, 0, len(wc.streams))
+		for _, s := range wc.streams {
+			streams = append(streams, s)
+		}
+		wc.mu.Unlock()
+		live += len(streams)
+		for _, s := range streams {
+			s.mu.Lock()
+			if s.parked {
+				parked++
+			}
+			s.mu.Unlock()
+		}
+
+		fmt.Fprintf(&b, " | conn%d streams=%d", wc.id, len(streams))
+		if cs := wc.stats; cs != nil {
+			_, mn, avg, mx := cs.rtt()
+			fmt.Fprintf(&b, " frames=%d/%d rtt=%s/%s/%s",
+				cs.framesIn.Load(), cs.framesOut.Load(), round(mn), round(avg), round(mx))
+			if n := cs.writeErrs.Load(); n > 0 {
+				fmt.Fprintf(&b, " writeErrs=%d", n)
+			}
+		}
+	}
+	fmt.Fprintf(&b, " | streams=%d parked=%d", live, parked)
+	return b.String()
+}
+
+// logSummary reports a stream's lifetime as it closes. This is the artifact that
+// answers a specific complaint after the fact, once the periodic line has
+// scrolled away.
+func (s *Stream) logSummary() {
+	s.mu.Lock()
+	st := s.stats
+	if st == nil {
+		s.mu.Unlock()
+		return
+	}
+	line := fmt.Sprintf("stream closed after %s: up=%s down=%s stalled(window=%s shaper=%s) qPeak=%s",
+		round(time.Since(st.opened)), bytesHuman(st.bytesUp), bytesHuman(st.bytesDown),
+		round(st.windowStall), round(st.shaperStall), bytesHuman(int64(st.qPeak)))
+	if st.resumes > 0 {
+		line += fmt.Sprintf(" resumes=%d hung=%s replayed=%s",
+			st.resumes, round(st.hung), bytesHuman(st.replayBytes))
+	}
+	s.mu.Unlock()
+	log.Print(line)
+}
+
+func (p *WorkerPool) snapshot() []*WorkerConn {
+	p.mu.Lock()
+	conns := append([]*WorkerConn(nil), p.conns...)
+	p.mu.Unlock()
+	sort.Slice(conns, func(i, j int) bool { return conns[i].id < conns[j].id })
+	return conns
+}
+
+// round trims a duration to something readable in a log line.
+func round(d time.Duration) time.Duration {
+	switch {
+	case d <= 0:
+		return 0
+	case d < time.Millisecond:
+		return d.Round(time.Microsecond)
+	case d < time.Second:
+		return d.Round(time.Millisecond)
+	default:
+		return d.Round(10 * time.Millisecond)
+	}
+}
+
+func bytesHuman(n int64) string {
+	const unit = 1024
+	if n < unit {
+		return fmt.Sprintf("%dB", n)
+	}
+	div, exp := int64(unit), 0
+	for v := n / unit; v >= unit; v /= unit {
+		div *= unit
+		exp++
+	}
+	return fmt.Sprintf("%.1f%ciB", float64(n)/float64(div), "KMGT"[exp])
+}

client/unacked.go

@@ -0,0 +1,64 @@
+package client
+
+// unackedBuf holds the bytes a stream has sent but the peer has not yet
+// credited — exactly the region a reattach may have to retransmit
+// (PROTOCOL.md §7.5).
+//
+// It needs no cap of its own: credit is only granted as bytes reach the peer's
+// terminal socket, so flow control already bounds the outstanding region to one
+// window. That is what makes byte-exact resumption affordable at all.
+//
+// A read offset rather than a copy-down on every trim. Credit arrives once per
+// half-window, and copying the live remainder each time would add a second
+// per-byte copy to the whole send path; compacting only once the dead prefix
+// dominates makes it amortized O(1).
+type unackedBuf struct {
+	buf     []byte
+	head    int   // bytes at the front already credited, awaiting reclamation
+	baseOff int64 // stream offset of buf[head]
+}
+
+// length is how many bytes are still outstanding.
+func (u *unackedBuf) length() int { return len(u.buf) - u.head }
+
+// base is the offset of the first byte still held.
+func (u *unackedBuf) base() int64 { return u.baseOff }
+
+// end is the offset one past the last byte sent.
+func (u *unackedBuf) end() int64 { return u.baseOff + int64(u.length()) }
+
+func (u *unackedBuf) append(p []byte) { u.buf = append(u.buf, p...) }
+
+// advance drops everything the peer has credited up to off.
+func (u *unackedBuf) advance(off int64) {
+	drop := int(off - u.baseOff)
+	if drop <= 0 {
+		return
+	}
+	if n := u.length(); drop > n {
+		drop = n // only reachable from a peer crediting bytes it was never sent
+	}
+	u.head += drop
+	u.baseOff += int64(drop)
+	switch {
+	case u.head == len(u.buf):
+		u.buf, u.head = u.buf[:0], 0 // fully drained: restart at the front
+	case u.head > len(u.buf)/2:
+		u.buf = append(u.buf[:0], u.buf[u.head:]...)
+		u.head = 0
+	}
+}
+
+// from returns the outstanding bytes at and after off, or nil when off falls
+// outside what is still held — which means the peer reported an offset we can no
+// longer satisfy, and the stream cannot be resumed.
+func (u *unackedBuf) from(off int64) []byte {
+	skip := off - u.baseOff
+	if skip < 0 || skip > int64(u.length()) {
+		return nil
+	}
+	return u.buf[u.head+int(skip):]
+}
+
+// reset releases the buffer once a stream can no longer be resumed.
+func (u *unackedBuf) reset() { u.buf, u.head = nil, 0 }

client/unacked_test.go

@@ -0,0 +1,112 @@
+package client
+
+import (
+	"bytes"
+	"testing"
+)
+
+// The retained region is what a reattach replays from, so an off-by-one here is
+// not a dropped byte but a spliced stream: the peer resumes mid-packet and the
+// session dies in a way no round-trip test would attribute to this code.
+
+func TestUnackedTracksOffsets(t *testing.T) {
+	var u unackedBuf
+	u.append([]byte("hello"))
+	u.append([]byte("world"))
+
+	if got := u.length(); got != 10 {
+		t.Fatalf("length = %d, want 10", got)
+	}
+	if got := u.end(); got != 10 {
+		t.Fatalf("end = %d, want 10", got)
+	}
+	if got := u.from(0); !bytes.Equal(got, []byte("helloworld")) {
+		t.Fatalf("from(0) = %q", got)
+	}
+	// A reattach replays from wherever the peer got to, which lands anywhere —
+	// including the middle of a chunk boundary.
+	if got := u.from(3); !bytes.Equal(got, []byte("loworld")) {
+		t.Fatalf("from(3) = %q", got)
+	}
+	if got := u.from(10); len(got) != 0 {
+		t.Fatalf("from(end) = %q, want empty", got)
+	}
+}
+
+func TestUnackedAdvanceDropsCreditedBytes(t *testing.T) {
+	var u unackedBuf
+	u.append([]byte("abcdefghij"))
+
+	u.advance(4)
+	if got := u.length(); got != 6 {
+		t.Fatalf("length after advance = %d, want 6", got)
+	}
+	if got := u.end(); got != 10 {
+		t.Fatalf("end must not move when bytes are dropped: got %d, want 10", got)
+	}
+	if got := u.from(4); !bytes.Equal(got, []byte("efghij")) {
+		t.Fatalf("from(4) = %q", got)
+	}
+	// Below the retained region: the peer named an offset we can no longer
+	// satisfy, which must be reported rather than silently clamped — replaying
+	// the wrong range is worse than refusing to replay.
+	if got := u.from(3); got != nil {
+		t.Fatalf("from(3) below base = %q, want nil", got)
+	}
+	if got := u.from(11); got != nil {
+		t.Fatalf("from(11) past end = %q, want nil", got)
+	}
+}
+
+// Interleaving appends and advances is the steady-state pattern: credit arrives
+// every half window while the sender keeps writing. The buffer must stay exact
+// across the compaction that eventually triggers.
+func TestUnackedSurvivesInterleavedAppendAndAdvance(t *testing.T) {
+	var u unackedBuf
+	var sent []byte
+	var acked int64
+
+	for i := 0; i < 200; i++ {
+		chunk := bytes.Repeat([]byte{byte(i)}, 97)
+		sent = append(sent, chunk...)
+		// emit's order: reclaim what has been credited so far, then retain the
+		// new chunk. The base therefore trails the credit that arrived since.
+		base := acked
+		u.advance(base)
+		u.append(chunk)
+
+		if got, want := u.end(), int64(len(sent)); got != want {
+			t.Fatalf("round %d: end = %d, want %d", i, got, want)
+		}
+		if got, want := u.length(), len(sent)-int(base); got != want {
+			t.Fatalf("round %d: length = %d, want %d", i, got, want)
+		}
+		if got, want := u.from(base), sent[base:]; !bytes.Equal(got, want) {
+			t.Fatalf("round %d: retained region diverges from what was sent", i)
+		}
+
+		// The peer can only ever credit bytes it has actually received.
+		if i%3 == 0 {
+			if acked += 61; acked > int64(len(sent)) {
+				acked = int64(len(sent))
+			}
+		}
+	}
+}
+
+// Compaction reuses the backing array, so a stream that runs for hours must not
+// grow one: this is a full window per stream, on both sides.
+func TestUnackedReclaimsBackingArray(t *testing.T) {
+	var u unackedBuf
+	chunk := bytes.Repeat([]byte{7}, 4096)
+	for i := 0; i < 500; i++ {
+		u.advance(u.end()) // fully credited every round
+		u.append(chunk)
+	}
+	if u.length() != len(chunk) {
+		t.Fatalf("length = %d, want %d", u.length(), len(chunk))
+	}
+	if cap(u.buf) > 8*len(chunk) {
+		t.Fatalf("backing array grew to %d bytes for a %d-byte window", cap(u.buf), len(chunk))
+	}
+}

client/worker.go

@@ -17,16 +17,24 @@ import (
 // exactly how a whole server's worth of players used to drop at once.
 const StreamsBeforeGrowing = 1
 
+// allocateAttempts bounds how many times a caller retries Allocate when the
+// conn it was handed dies before the stream could be registered on it. The race
+// is narrow and each retry picks a different conn, so a small bound is enough;
+// an unbounded loop would spin against a hub that is refusing every connection.
+const allocateAttempts = 3
+
 // WorkerPool manages up to maxConn worker connections and allocates streams
 // using the least-loaded strategy (PROTOCOL.md §7.1).
 type WorkerPool struct {
 	client  *Client
 	maxConn int
 
+	connSeq atomic.Int64 // conn ids, for log correlation
+
 	mu      sync.Mutex
 	cond    *sync.Cond
 	conns   []*WorkerConn
-	dialing int   // dials currently in flight (foreground + background)
+	dialing int // dials currently in flight (foreground + background)
 	dialGen uint64
 	dialErr error // most recent dial failure
 }
@@ -147,11 +155,16 @@ func (p *WorkerPool) dialWorker() (*WorkerConn, error) {
 		pool:        p,
 		fc:          sess.fc,
 		sendWndInit: sess.peerWnd,
-		recvWndInit: p.client.streamWnd,
+		resume:      sess.resume,
+		grace:       p.client.resumeGrace(sess.hubGrace),
+		id:          int(p.connSeq.Add(1)),
 		streams:     make(map[int]*Stream),
 		nextSid:     1,
 		done:        make(chan struct{}),
 	}
+	if p.client.statsOn() {
+		wc.stats = &connStats{opened: time.Now()}
+	}
 	wc.lastPong.Store(time.Now().UnixMilli())
 	go wc.readLoop()
 	if sess.heartbeat {
@@ -160,8 +173,8 @@ func (p *WorkerPool) dialWorker() (*WorkerConn, error) {
 		log.Printf("worker conn: hub does not support the mux heartbeat; " +
 			"a silently dropped path will only be caught by TCP keepalive")
 	}
-	log.Printf("opened worker conn (send window %d, recv window %d, heartbeat %v)",
-		wc.sendWndInit, wc.recvWndInit, sess.heartbeat)
+	log.Printf("opened worker conn (send window %d, recv window %d, heartbeat %v, resume %v)",
+		wc.sendWndInit, p.client.streamWnd, sess.heartbeat, wc.resume)
 	return wc, nil
 }
 
@@ -195,19 +208,34 @@ func (p *WorkerPool) closeAll() {
 }
 
 // WorkerConn is one multiplexed worker connection to the hub.
+//
+// Only genuinely per-connection state lives here. Client-wide values (the
+// shaper, the advertised receive window, the DATA chunk cap) belong to Client:
+// reading them through a connection pointer would make every such read a
+// re-parenting hazard once a stream can migrate between conns.
 type WorkerConn struct {
 	pool *WorkerPool
 	fc   *wire.FramedConn
 
 	sendWndInit int // hub's advertised per-stream receive window (our send budget)
-	recvWndInit int // our advertised per-stream receive window (bounds each recv queue)
+
+	// resume is whether this conn negotiated stream resumption, and grace how
+	// long a stream parked from it may keep trying to reattach. Both are
+	// per-conn: a reattach may land on a different (or restarted) hub, so the
+	// flag must be re-checked on the conn that will carry the RESUME.
+	resume bool
+	grace  time.Duration
 
 	done     chan struct{} // closed when readLoop exits
 	lastPong atomic.Int64  // unix ms of the most recent PONG
 
+	id    int        // for log correlation only
+	stats *connStats // nil unless diagnostics are enabled
+
 	mu      sync.Mutex
 	streams map[int]*Stream
 	nextSid int
+	closed  bool // readLoop has exited; registerStream must refuse
 }
 
 // heartbeatLoop proves the worker conn is still carrying frames end to end. TCP
@@ -251,10 +279,23 @@ func (wc *WorkerConn) newSid() int {
 	return sid
 }
 
-func (wc *WorkerConn) registerStream(sid int, st *Stream) {
+// registerStream publishes a stream in the conn's table, or reports false if
+// the conn has already died.
+//
+// The check is not advisory. Allocate hands out a (conn, sid) pair under the
+// pool lock, and the conn's readLoop can exit before the caller gets here — it
+// has then already swapped the stream map, so a blind insert would land in a map
+// nothing iterates and the stream would never be torn down. That normally hides
+// behind a failing SYN, but not on a half-open conn whose readLoop died on a
+// framing error while the socket is still writable. Callers must re-Allocate.
+func (wc *WorkerConn) registerStream(sid int, st *Stream) bool {
 	wc.mu.Lock()
+	defer wc.mu.Unlock()
+	if wc.closed {
+		return false
+	}
 	wc.streams[sid] = st
-	wc.mu.Unlock()
+	return true
 }
 
 func (wc *WorkerConn) getStream(sid int) *Stream {
@@ -280,6 +321,9 @@ func (wc *WorkerConn) readLoop() {
 		if err != nil {
 			break
 		}
+		if wc.stats != nil {
+			wc.stats.framesIn.Add(1)
+		}
 		r := wire.NewReader(payload)
 		ftype, err := r.U8()
 		if err != nil {
@@ -306,14 +350,37 @@ func (wc *WorkerConn) readLoop() {
 				st.gracefulFin()
 			}
 		case MuxRst:
+			// The reason is an optional trailing byte; older peers send none.
+			reason := RstUnspecified
+			if b, err := r.U8(); err == nil {
+				reason = int(b)
+			}
 			if st := wc.removeStream(sid); st != nil {
-				st.teardown(false)
+				st.onRst(reason)
+			}
+		case MuxResumeAck:
+			accepted, aerr := r.I64()
+			delivered, derr := r.I64()
+			cid, cerr := r.Bytes(CIDLen)
+			if aerr != nil || derr != nil || cerr != nil {
+				continue
+			}
+			if st := wc.getStream(sid); st != nil {
+				st.deliverResume(resumeResult{accepted: accepted, delivered: delivered, cid: cid})
 			}
 		case MuxPing:
 			nonce, _ := r.I64()
 			_ = wc.fc.WriteFrame(wire.NewWriter().U8(MuxPong).VarInt(MuxCtlSid).I64(nonce).Out())
 		case MuxPong:
-			wc.lastPong.Store(time.Now().UnixMilli())
+			now := time.Now()
+			wc.lastPong.Store(now.UnixMilli())
+			// The probe's nonce is the timestamp we sent, echoed back, so the
+			// round trip is free to measure and nobody was reading it.
+			if wc.stats != nil {
+				if sent, err := r.I64(); err == nil {
+					wc.stats.observeRTT(now.Sub(time.UnixMilli(sent)))
+				}
+			}
 		default:
 			log.Printf("worker: unknown mux type %d", ftype)
 		}
@@ -322,14 +389,22 @@ func (wc *WorkerConn) readLoop() {
 	close(wc.done)
 	wc.pool.remove(wc)
 	wc.mu.Lock()
+	// Marked before the map is swapped, under the same lock, so a concurrent
+	// registerStream either lands in the map we are about to drain or is refused.
+	wc.closed = true
 	streams := make([]*Stream, 0, len(wc.streams))
 	for _, st := range wc.streams {
 		streams = append(streams, st)
 	}
 	wc.streams = make(map[int]*Stream)
 	wc.mu.Unlock()
+	// Only the tunnel leg died. Where the session negotiated resumption the
+	// destination sockets are kept open and each stream reattaches over a fresh
+	// conn (§7.5); otherwise this is the old, unconditional teardown.
 	for _, st := range streams {
-		st.teardown(false)
+		if !st.park(wc.grace) {
+			st.teardown(false)
+		}
 	}
 }
 
@@ -338,7 +413,14 @@ func (wc *WorkerConn) sendSyn(sid int, cid []byte) error {
 }
 
 func (wc *WorkerConn) sendData(sid int, data []byte) error {
-	return wc.fc.WriteFrame(wire.NewWriter().U8(MuxData).VarInt(sid).Bytes(data).Out())
+	err := wc.fc.WriteFrame(wire.NewWriter().U8(MuxData).VarInt(sid).Bytes(data).Out())
+	if wc.stats != nil {
+		wc.stats.framesOut.Add(1)
+		if err != nil {
+			wc.stats.writeErrs.Add(1)
+		}
+	}
+	return err
 }
 
 func (wc *WorkerConn) sendFin(sid int) {
@@ -353,6 +435,16 @@ func (wc *WorkerConn) sendWndUpdate(sid, delta int) {
 	_ = wc.fc.WriteFrame(wire.NewWriter().U8(MuxWnd).VarInt(sid).VarInt(delta).Out())
 }
 
+// leg binds a stream to one worker conn. The two fields are only meaningful
+// together: stream ids are per-conn and restart at 1, so conn A's sid 3 and
+// conn B's sid 3 belong to different players. A torn read across the two would
+// credit, reset or FIN a stranger's stream, so the pair is swapped as one
+// immutable value rather than as two fields.
+type leg struct {
+	wc  *WorkerConn
+	sid int
+}
+
 // Stream bridges one player (via the hub) to one destination connection.
 //
 // Data from the hub is queued and written to the destination by a dedicated
@@ -360,24 +452,74 @@ func (wc *WorkerConn) sendWndUpdate(sid, delta int) {
 // the hub never sends more un-credited bytes, so overflow is a protocol
 // violation and resets the stream.
 type Stream struct {
-	wc      *WorkerConn
-	sid     int
-	cid     []byte
+	client  *Client
+	leg     atomic.Pointer[leg]
 	mapping Mapping
 	srcIP   string
 	srcPort int
 	vel     *velocityForwarder // non-nil when the mapping sets velocitySecret
 
+	share shaperShare   // this stream's position in the egress fair queue
+	done  chan struct{} // closed on teardown; unparks a shaper wait
+
+	// resumable is fixed at creation from the conn's negotiated flag. With it
+	// false none of the bookkeeping below runs and no buffer is ever allocated,
+	// so a client with resumption disabled pays exactly what it used to.
+	resumable bool
+
+	stats *streamStats // nil unless diagnostics are enabled
+
+	// ackedOffset is the running sum of WND deltas received. Credit is granted
+	// only as bytes reach the peer's terminal socket, so it is a sound lower
+	// bound on what has been delivered. Atomic because the worker readLoop
+	// advances it and must never block behind the send path.
+	ackedOffset atomic.Int64
+
+	// sendMu serializes the send path — buffer the chunk, advance the offset,
+	// write the frame — against a reattach's retransmit, so replayed bytes can
+	// never interleave with live ones.
+	//
+	// Deliberately not s.mu: deliverFromHub takes s.mu from the worker readLoop,
+	// and holding s.mu across a WriteFrame would stall frame dispatch for every
+	// other stream on the connection.
+	sendMu sync.Mutex
+	un     unackedBuf // guarded by sendMu
+
 	mu         sync.Mutex
 	cond       *sync.Cond
+	cid        []byte // takeover capability; re-minted by the hub on each resume
 	dest       net.Conn
 	connected  bool
 	closed     bool
-	finPending bool     // hub sent FIN; close the destination once the queue drains
-	q          []qentry // hub/local -> destination, waiting for writeLoop
-	qBytes     int      // hub bytes only: bounds the peer against its window
-	sendWnd    int      // flow control: budget for destination -> hub DATA
-	consumed   int      // flow control: drained bytes not yet credited back to the hub
+	parked     bool      // worker conn died; awaiting reattach on a fresh one
+	parkedAt   time.Time // when the current hang began; diagnostics only
+	finPending bool      // hub sent FIN; close the destination once the queue drains
+	finToHub   bool      // destination closed while parked; FIN owed once reattached
+	q          []qentry  // hub/local -> destination, waiting for writeLoop
+	qBytes     int       // hub bytes only: bounds the peer against its window
+	// acceptedOffset counts hub bytes enqueued toward the destination. This, not
+	// "bytes written", is what a reattach reports: acceptance is synchronous and
+	// stable at park time, whereas delivery is signalled asynchronously and goes
+	// silent exactly when the connection dies — which would under-report and make
+	// the hub retransmit bytes the player already has.
+	acceptedOffset int64
+	// deliveredOffset counts hub bytes actually written to the destination.
+	// Distinct from acceptedOffset and needed for a different job: a reattach
+	// replays from what the peer *accepted*, but sizes the window from what it
+	// *delivered*, because the window is a promise about undelivered bytes.
+	deliveredOffset int64
+	sendWnd         int // flow control: budget for destination -> hub DATA
+	consumed        int // flow control: drained bytes not yet credited back to the hub
+	resumeWait      chan resumeResult
+}
+
+// resumeResult is the hub's answer to a RESUME: how far it got in both senses,
+// plus a fresh CID — or the reason the reattach was refused.
+type resumeResult struct {
+	accepted  int64
+	delivered int64
+	cid       []byte
+	err       error
 }
 
 // qentry is one queued write towards the destination. Only hub-originated
@@ -389,8 +531,13 @@ type qentry struct {
 	fromHub bool
 }
 
-func newStream(wc *WorkerConn, sid int, cid []byte, m Mapping, ip string, port int) *Stream {
-	s := &Stream{wc: wc, sid: sid, cid: cid, mapping: m, srcIP: ip, srcPort: port, sendWnd: wc.sendWndInit}
+func newStream(c *Client, wc *WorkerConn, sid int, cid []byte, m Mapping, ip string, port int) *Stream {
+	s := &Stream{client: c, cid: cid, mapping: m, srcIP: ip, srcPort: port,
+		resumable: wc.resume, sendWnd: wc.sendWndInit, done: make(chan struct{})}
+	s.leg.Store(&leg{wc: wc, sid: sid})
+	if c.statsOn() {
+		s.stats = &streamStats{opened: time.Now()}
+	}
 	if m.VelocitySecret != "" {
 		s.vel = newVelocityForwarder(m.VelocitySecret, ip)
 	}
@@ -398,14 +545,20 @@ func newStream(wc *WorkerConn, sid int, cid []byte, m Mapping, ip string, port i
 	return s
 }
 
+// conn returns the stream's current binding. Every caller must take exactly one
+// snapshot and use both fields from it; re-loading mid-operation reintroduces
+// the torn-pair hazard the leg exists to prevent.
+func (s *Stream) conn() *leg { return s.leg.Load() }
+
 // run dials the destination, optionally writes the PROXY v2 header, then pumps
 // destination -> hub (respecting the stream send window when negotiated).
 func (s *Stream) run() {
 	dest, err := net.DialTimeout("tcp", s.mapping.Destination, 10*time.Second)
 	if err != nil {
-		log.Printf("stream %d: dial %s failed: %v", s.sid, s.mapping.Destination, err)
-		s.wc.removeStream(s.sid)
-		s.wc.sendRst(s.sid)
+		lg := s.conn()
+		log.Printf("stream %d: dial %s failed: %v", lg.sid, s.mapping.Destination, err)
+		lg.wc.removeStream(lg.sid)
+		lg.wc.sendRst(lg.sid)
 		s.teardown(false)
 		return
 	}
@@ -416,7 +569,7 @@ func (s *Stream) run() {
 	if s.mapping.ProxyProtocol {
 		if hdr := s.buildProxyHeader(dest); hdr != nil {
 			if _, err := dest.Write(hdr); err != nil {
-				log.Printf("stream %d: proxy header write: %v", s.sid, err)
+				log.Printf("stream %d: proxy header write: %v", s.conn().sid, err)
 			}
 		}
 	}
@@ -462,26 +615,88 @@ func (s *Stream) run() {
 	s.teardown(true)
 }
 
-// sendToHub forwards destination bytes to the hub in DATA frames of at most
-// DataChunkSize, honoring the stream send window. Returns false once the
-// stream closed or the worker conn failed.
+// sendToHub forwards destination bytes to the hub in bounded DATA frames,
+// honoring both the stream send window and the client-wide bandwidth cap.
+// Returns false once the stream closed or the worker conn failed.
 func (s *Stream) sendToHub(data []byte) bool {
 	for len(data) > 0 {
 		n := len(data)
-		if n > DataChunkSize {
-			n = DataChunkSize
+		if n > s.client.chunk {
+			n = s.client.chunk
 		}
+		// Credit first, bandwidth second. The reverse order would charge the
+		// budget for bytes still parked on an exhausted window, so the client
+		// would throttle itself below the configured rate. Holding credit while
+		// waiting for tokens is free — credit is per-stream, and the hub returns
+		// it as it drains data to the player, independent of our pacing.
 		if !s.acquireSendWnd(n) {
 			return false
 		}
-		if err := s.wc.sendData(s.sid, data[:n]); err != nil {
+		var shaperStall stallClock
+		shaperStall.begin(s.stats != nil)
+		if !s.client.shaper.Acquire(&s.share, n, s.done) {
 			return false
 		}
+		if !s.emit(data[:n]) {
+			return false
+		}
+		if s.stats != nil {
+			s.mu.Lock()
+			// Separated from the window stall on purpose: this one says the
+			// configured cap is the binding constraint, and raising maxBandwidth
+			// is the fix. The window stall says the opposite.
+			s.stats.shaperStall += shaperStall.elapsed()
+			s.stats.bytesUp += int64(n)
+			s.mu.Unlock()
+		}
 		data = data[n:]
 	}
 	return true
 }
 
+// emit records a chunk for possible retransmission and writes it to the current
+// worker conn. Returns false once the stream is finished with.
+//
+// The record is taken first and unconditionally. A write that fails on a dying
+// connection has already spent window and may have put part of the frame on the
+// wire, so the only trustworthy account of what the peer still owes us is the
+// one taken before the attempt. That also makes a failure survivable: while the
+// stream can still be resumed the bytes are already safe, and the reattach
+// replays them from wherever the hub says it got to.
+//
+// Note the old code let a failed write drop the rest of the chunk on the floor —
+// the caller's slice advance sat after the error return.
+func (s *Stream) emit(chunk []byte) bool {
+	s.sendMu.Lock()
+	if s.resumable {
+		// Reclaim what the hub has credited before growing the buffer, so the
+		// outstanding region stays bounded by one window.
+		s.un.advance(s.ackedOffset.Load())
+		s.un.append(chunk)
+	}
+	lg := s.conn()
+	err := lg.wc.sendData(lg.sid, chunk)
+	s.sendMu.Unlock()
+
+	if err == nil {
+		return true
+	}
+	if !s.resumable {
+		return false
+	}
+	// The conn is gone but the stream is not: readLoop parks it and a reattach
+	// replays the buffer. Keep pumping the destination — acquireSendWnd stops us
+	// once a full window is outstanding, so nothing is lost and nothing grows
+	// without bound.
+	return !s.isClosed()
+}
+
+func (s *Stream) isClosed() bool {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.closed
+}
+
 // writeLoop is the only writer to the destination. It drains the receive queue,
 // credits the hub as bytes land on the destination socket, and performs the
 // deferred graceful close when a FIN arrived with data still queued.
@@ -513,6 +728,12 @@ func (s *Stream) writeLoop() {
 			return
 		}
 		if e.fromHub {
+			s.mu.Lock()
+			s.deliveredOffset += int64(len(e.data))
+			if s.stats != nil {
+				s.stats.bytesDown += int64(len(e.data))
+			}
+			s.mu.Unlock()
 			s.credit(len(e.data))
 		}
 	}
@@ -530,16 +751,26 @@ func (s *Stream) deliverFromHub(data []byte) {
 		s.mu.Unlock()
 		return
 	}
-	if s.qBytes+len(data) > s.wc.recvWndInit {
+	if s.qBytes+len(data) > s.client.streamWnd {
 		s.mu.Unlock()
-		log.Printf("stream %d: peer exceeded flow-control window; resetting", s.sid)
-		s.wc.removeStream(s.sid)
-		s.wc.sendRst(s.sid)
+		lg := s.conn()
+		log.Printf("stream %d: peer exceeded flow-control window; resetting", lg.sid)
+		lg.wc.removeStream(lg.sid)
+		lg.wc.sendRst(lg.sid)
 		s.teardown(false)
 		return
 	}
 	s.q = append(s.q, qentry{data: data, fromHub: true})
 	s.qBytes += len(data)
+	// Accepted, not delivered: from here the bytes are ours to write, and the
+	// only way we fail to is by destroying the stream — which also ends any
+	// prospect of resuming it. That makes this a sound reattach coordinate.
+	s.acceptedOffset += int64(len(data))
+	if s.stats != nil && s.qBytes > s.stats.qPeak {
+		// How close the receive queue came to the advertised window: near it
+		// means the destination is the slow party.
+		s.stats.qPeak = s.qBytes
+	}
 	s.cond.Broadcast()
 	s.mu.Unlock()
 }
@@ -562,9 +793,18 @@ func (s *Stream) injectToDest(data []byte) {
 func (s *Stream) acquireSendWnd(n int) bool {
 	s.mu.Lock()
 	defer s.mu.Unlock()
+	// Timed only when it actually blocks, so a stream that never runs out of
+	// credit never reads the clock. A large windowStall is the signal that the
+	// peer is not draining to its terminal socket — the bottleneck is past the
+	// tunnel, not in it.
+	var stall stallClock
 	for !s.closed && s.sendWnd < n {
+		stall.begin(s.stats != nil)
 		s.cond.Wait()
 	}
+	if s.stats != nil {
+		s.stats.windowStall += stall.elapsed()
+	}
 	if s.closed {
 		return false
 	}
@@ -573,6 +813,11 @@ func (s *Stream) acquireSendWnd(n int) bool {
 }
 
 func (s *Stream) grantSendWnd(delta int) {
+	// The running sum doubles as the acked offset: the hub grants credit exactly
+	// as bytes reach the player socket, so a byte that has been credited can
+	// never need retransmitting. Advanced without a lock so the worker readLoop
+	// never blocks behind a send in progress.
+	s.ackedOffset.Add(int64(delta))
 	s.mu.Lock()
 	s.sendWnd += delta
 	s.cond.Broadcast()
@@ -581,17 +826,22 @@ func (s *Stream) grantSendWnd(delta int) {
 
 // credit accounts bytes drained to the destination and grants the hub more
 // window once half of our receive window has been consumed.
+// While parked the grant is only withheld, never dropped: consumed keeps
+// accumulating and a reattach flushes it on the new conn. Resetting it would
+// destroy up to half a window of credit per outage, and after a few flaps the
+// stream would throttle to a crawl.
 func (s *Stream) credit(n int) {
 	s.mu.Lock()
 	s.consumed += n
-	if s.closed || s.consumed*2 < s.wc.recvWndInit {
+	if s.closed || s.parked || s.consumed*2 < s.client.streamWnd {
 		s.mu.Unlock()
 		return
 	}
 	delta := s.consumed
 	s.consumed = 0
 	s.mu.Unlock()
-	s.wc.sendWndUpdate(s.sid, delta)
+	lg := s.conn()
+	lg.wc.sendWndUpdate(lg.sid, delta)
 }
 
 func (s *Stream) buildProxyHeader(dest net.Conn) []byte {
@@ -629,6 +879,13 @@ func (s *Stream) gracefulFin() {
 // teardown closes the stream immediately; notifyHub sends a FIN when true.
 // Idempotent; wakes every goroutine parked on the stream.
 func (s *Stream) teardown(notifyHub bool) {
+	// A parked stream owes the hub a FIN it cannot send: the only conn it has is
+	// the one that just failed. Keep it alive so the reattach can deliver it and
+	// the player gets a clean disconnect, rather than hanging until the hub's
+	// grace expires.
+	if notifyHub && s.noteFinWhileParked() {
+		return
+	}
 	s.mu.Lock()
 	if s.closed {
 		s.mu.Unlock()
@@ -636,14 +893,23 @@ func (s *Stream) teardown(notifyHub bool) {
 	}
 	s.closed = true
 	dest := s.dest
+	close(s.done) // guarded by the idempotence check above, so exactly once
 	s.cond.Broadcast()
 	s.mu.Unlock()
 
+	// The stream can no longer be resumed, so the retransmit buffer is dead
+	// weight — up to a full window of it per stream.
+	s.sendMu.Lock()
+	s.un.reset()
+	s.sendMu.Unlock()
+
 	if dest != nil {
 		_ = dest.Close()
 	}
-	s.wc.removeStream(s.sid)
+	lg := s.conn()
+	lg.wc.removeStream(lg.sid)
 	if notifyHub {
-		s.wc.sendFin(s.sid)
+		lg.wc.sendFin(lg.sid)
 	}
+	s.logSummary()
 }