From bbe4efbe1699fd1c2edb5f09fefd9cd6df747c41 Mon Sep 17 00:00:00 2001 From: iceBear67 Date: Wed, 15 Jul 2026 14:59:32 +0000 Subject: [PATCH] add credit based mux window --- PROTOCOL.md | 61 +++++- README.md | 14 +- client/client.go | 59 +++-- client/config.example.json | 1 + client/config.go | 25 ++- client/worker.go | 206 ++++++++++++++---- docs/architecture.md | 51 +++-- e2e/harness_test.go | 12 +- e2e/slowstream_test.go | 131 +++++++++++ server/config.example.json | 3 +- .../java/io/icybear/redapricot/Config.java | 9 +- .../io/icybear/redapricot/HubConnection.java | 32 ++- .../java/io/icybear/redapricot/Protocol.java | 10 + .../io/icybear/redapricot/WorkerConn.java | 174 +++++++++++---- .../icybear/redapricot/CryptoCodecTest.java | 2 +- 15 files changed, 651 insertions(+), 139 deletions(-) create mode 100644 e2e/slowstream_test.go diff --git a/PROTOCOL.md b/PROTOCOL.md index 9e16a55..e4a9da4 100644 --- a/PROTOCOL.md +++ b/PROTOCOL.md @@ -140,8 +140,16 @@ Magic : u8 # 0x01 = control session, 0x02 = worker conn RandLen : VarInt # 8 ≤ RandLen ≤ 64 Rand : Bytes[RandLen] # cryptographically random Timestamp : I64 # client's epoch milliseconds +Flags : VarInt # feature flags; bit 0x01 (STREAM_FC) MUST be set +RecvWindow: VarInt # client's per-stream receive window, bytes (§7.3) ``` +`Flags` is a bitfield of features. Bit `0x01` (STREAM_FC) declares +**per-stream flow control** (§7.3) and is mandatory: `RecvWindow` advertises +the client's per-stream receive window in bytes and must be positive. The hub +closes the connection if the flag is missing, `RecvWindow` is absent or +non-positive, or the fields are malformed. + The hub: 1. Decrypts frame 1 with Phase A. @@ -158,9 +166,13 @@ frame (both directions) is Phase B, counters reset to 0. The hub then sends one Phase-B frame to confirm success: ``` -SessionReady : payload = [ 0x00 ] +SessionReady : payload = [ 0x00, Flags: VarInt, RecvWindow: VarInt ] ``` +The hub echoes the accepted flags (STREAM_FC set) followed by its own +per-stream receive window. A client must reject a SessionReady without the +STREAM_FC flag or without a positive window (an unsupported hub). + A hub that rejects the session simply closes the TCP connection (optionally after a Phase-B `Error` frame, §6). After `SessionReady`: @@ -258,6 +270,7 @@ Data : Bytes[...] # remainder of the frame payload | `0x01` | DATA | both | raw tunneled bytes for the stream. | | `0x02` | FIN | both | *(empty)* — graceful close of the stream (both directions). This is the "disconnect" the hub sends when the player leaves. | | `0x03` | RST | both | *(optional 1 byte reason)* — abnormal close (e.g. CID unknown/expired, destination dial failed). | +| `0x04` | WND | both | `Delta: VarInt` — flow-control credit grant (§7.3). | There is no explicit SYN-ACK: success is implied by the hub forwarding the buffered Handshake as the stream's first `DATA`; failure is an `RST`. @@ -294,11 +307,33 @@ configurable, `1..8`). To place a new stream: closes the destination. When the destination closes, the client sends `FIN`; the hub closes the player socket. `RST` is treated the same way (hard close). -Data on a worker conn is subject to that TCP connection's back-pressure. Each -stream has a bounded outbound queue on the receiving side; overflow resets the -stream (`RST`). (This is a deliberate simplification — no per-stream credit -windows — acceptable for the interactive, low-throughput Minecraft handshake + -gameplay traffic pattern.) +Data on a worker conn is subject to that TCP connection's back-pressure for +its **aggregate** bandwidth; *per-stream* fairness is governed by the credit +windows of §7.3. + +### 7.3 Per-stream flow control + +Every stream carries an independent credit window per direction: + +* Each side advertised its **receive window** W (bytes) at session setup. A + sender may have at most W un-credited DATA bytes outstanding per stream; the + initial budget is W, spent as DATA is sent (`Data` length only — SYN/FIN/RST + frames are free) starting with the very first DATA on the stream (including + the hub's forwarded handshake). +* The receiver returns credit with `WND(Delta)` once bytes are **delivered to + the terminal socket** (written to the player / destination connection), not + when they are merely buffered. Receivers should batch grants (the reference + implementations send one `WND` per W/2 bytes consumed). +* A sender whose window is exhausted pauses reading **that stream's source + socket only**; the shared worker conn is never paused because of a single + stream. A receiver that observes more than W un-credited bytes on a stream + may reset it (`RST`) as a protocol violation. +* Senders should also cap individual DATA payloads (the reference + implementations use 32 KiB) so one stream cannot monopolize the link for a + full 1-MiB frame. + +Both windows may differ (each side enforces the one its peer advertised). +`Delta` must be positive; a `WND` for an unknown stream id is ignored. ## 8. HAProxy protocol v2 (optional) @@ -327,10 +362,14 @@ big-endian. "listen": "0.0.0.0:25565", "psk": "change-me", "timestampWindowMs": 30000, - "pendingTimeoutMs": 10000 + "pendingTimeoutMs": 10000, + "streamWindowBytes": 262144 } ``` +`streamWindowBytes` (optional, default 262144, clamped to [32768, 8388608]) is +the hub's advertised per-stream receive window (§7.3). + ### 9.2 Client — JSON ```json @@ -339,12 +378,16 @@ big-endian. "psk": "change-me", "maxConn": 4, "pingIntervalMs": 20000, + "streamWindowBytes": 262144, "mappings": [ { "pattern": "mc\\.example\\.com", "destination": "127.0.0.1:25566", "proxyProtocol": true } ] } ``` +`streamWindowBytes` (optional, default 262144, clamped to [32768, 8388608]) is +the client's advertised per-stream receive window (§7.3). + Each `pattern` is a regular expression (§5.1) matched against the whole normalized player hostname, case-insensitively. Escape literal dots (`mc\.example\.com`, which is `mc\\.example\\.com` in JSON); an unescaped `.` matches any character. @@ -369,4 +412,8 @@ subdomain or `(alpha|beta)\.mc\.net` for a fixed set. | max frame payload | 1 MiB | | saturation threshold | active streams `> 8` | | max worker conns | `max_conn ∈ [1,8]` | +| feature flag: per-stream flow control | `0x01` | +| stream window default / bounds | 256 KiB, clamped to [32 KiB, 8 MiB] | +| WND grant batching (reference) | one grant per window/2 consumed | +| DATA chunk cap (reference) | 32 KiB | ``` diff --git a/README.md b/README.md index 98facf1..aecee12 100644 --- a/README.md +++ b/README.md @@ -202,7 +202,9 @@ forwarding and case-insensitive matching, regex wildcard pattern routing, multi-megabyte transfers, concurrent streams spreading across multiple worker connections, HAProxy v2 source-address propagation, player- and destination-initiated disconnect propagation, wrong-PSK -rejection, and dropping of unmatched hostnames. The Go and Java crypto layers are +rejection, dropping of unmatched hostnames, stream isolation under a slow +player and under a slow destination (no head-of-line blocking), and rejection +of pre-flow-control peers. The Go and Java crypto layers are independently pinned to the same SHA3-224 test vector so they cannot silently drift apart. @@ -212,10 +214,12 @@ drift apart. ChaCha20-encrypted (no AEAD tag) to minimize overhead. This protects against casual sniffing, not a determined active attacker (see the note at the top of [PROTOCOL.md](PROTOCOL.md) and [docs/architecture.md](docs/architecture.md) §8). -* **No per-stream flow control.** Multiplexing relies on TCP back-pressure per - worker connection, so one very slow stream can head-of-line-block others on - the same connection. Raising `maxConn` spreads load. Fine for interactive - Minecraft traffic; not a general-purpose high-throughput mux. +* **Per-stream flow control.** Each stream has credit-based windows in both + directions (windows exchanged at session setup, default 256 KiB), so a slow + player or slow destination jams only its own stream at a bounded buffer — no + application-level head-of-line blocking between streams. What remains is + TCP-level HOL (packet loss stalls a whole worker connection briefly); + raising `maxConn` spreads that. * **Single hub event loop.** The hub deploys one Vert.x verticle, so all state is confined to one event loop (no locking). Throughput is bounded by one core; ample for hundreds of players, not designed for tens of thousands. diff --git a/client/client.go b/client/client.go index 52c3525..42ebbe3 100644 --- a/client/client.go +++ b/client/client.go @@ -24,6 +24,8 @@ type Client struct { mappings map[string]Mapping // normalized pattern -> mapping pool *WorkerPool + streamWnd int // our advertised per-stream receive window (bytes) + mu sync.Mutex ctrl *wire.FramedConn } @@ -44,16 +46,32 @@ func New(cfg *Config) *Client { for _, m := range cfg.Mappings { c.mappings[NormalizeAddress(m.Pattern)] = m } + c.streamWnd = clampWindow(cfg.StreamWindowBytes) c.pool = newWorkerPool(c, cfg.MaxConn) return c } +func clampWindow(w int) int { + if w <= 0 { + return DefaultStreamWindow + } + if w < MinStreamWindow { + return MinStreamWindow + } + if w > MaxStreamWindow { + return MaxStreamWindow + } + return w +} + // dialSession opens a TCP connection, performs the Intent-17 handshake, the -// Phase-A rekey, and reads SessionReady, returning an established frame conn. -func (c *Client) dialSession(magic byte) (*wire.FramedConn, error) { +// Phase-A rekey, and reads SessionReady, returning an established frame conn +// and the hub's advertised per-stream receive window. Per-stream flow control +// is mandatory: a hub that does not echo the STREAM_FC flag is rejected. +func (c *Client) dialSession(magic byte) (fc *wire.FramedConn, peerWnd int, err error) { conn, err := net.DialTimeout("tcp", c.cfg.Server, 10*time.Second) if err != nil { - return nil, err + return nil, 0, err } if tcp, ok := conn.(*net.TCPConn); ok { _ = tcp.SetNoDelay(true) @@ -68,24 +86,26 @@ func (c *Client) dialSession(magic byte) (*wire.FramedConn, error) { // 1. plaintext Minecraft Handshake, Intent 17, address = hex(SHA3-224(PSK)). hs := wire.BuildHandshake(ProtocolVersion, c.pskAddr, c.serverPort, IntentRedapricot) if _, err := conn.Write(hs); err != nil { - return nil, err + return nil, 0, err } // 2. Phase-A ciphers derived from the PSK. - fc := wire.NewFramedConn(conn, + fc = wire.NewFramedConn(conn, wire.CipherFor(c.pskBytes, wire.DirS2C), // in: server -> client wire.CipherFor(c.pskBytes, wire.DirC2S), // out: client -> server ) - // 3. Rekey frame (Phase A). + // 3. Rekey frame (Phase A), including the mandatory feature flags and our + // per-stream receive window. rnd := make([]byte, 16) if _, err := crand.Read(rnd); err != nil { - return nil, err + return nil, 0, err } ts := time.Now().UnixMilli() - rekeyMsg := wire.NewWriter().U8(magic).VarInt(len(rnd)).Bytes(rnd).I64(ts).Out() + rekeyMsg := wire.NewWriter().U8(magic).VarInt(len(rnd)).Bytes(rnd).I64(ts). + VarInt(FlagStreamFC).VarInt(c.streamWnd).Out() if err := fc.WriteFrame(rekeyMsg); err != nil { - return nil, err + return nil, 0, err } // 4. Switch to Phase-B ciphers: REKEY = Rand || Timestamp(I64 BE). @@ -99,16 +119,26 @@ func (c *Client) dialSession(magic byte) (*wire.FramedConn, error) { wire.CipherFor(rekey, wire.DirC2S), ) - // 5. SessionReady. + // 5. SessionReady: the type byte followed by the hub's accepted flags and + // its per-stream receive window. Both are required. payload, err := fc.ReadFrame() if err != nil { - return nil, err + return nil, 0, err } if len(payload) < 1 || payload[0] != CtlSessionReady { - return nil, fmt.Errorf("expected SessionReady, got %v", payload) + return nil, 0, fmt.Errorf("expected SessionReady, got %v", payload) + } + r := wire.NewReader(payload[1:]) + flags, ferr := r.VarInt() + hubWnd, werr := r.VarInt() + if ferr != nil || werr != nil || flags&FlagStreamFC == 0 || hubWnd <= 0 { + return nil, 0, fmt.Errorf("hub did not accept per-stream flow control (unsupported hub version?)") + } + if hubWnd > MaxStreamWindow { + hubWnd = MaxStreamWindow } ok = true - return fc, nil + return fc, hubWnd, nil } // Start establishes the control session and registers all patterns. It returns @@ -119,7 +149,7 @@ func (c *Client) Start(ctx context.Context) error { } func (c *Client) connectControl(ctx context.Context) error { - fc, err := c.dialSession(MagicControl) + fc, _, err := c.dialSession(MagicControl) if err != nil { return fmt.Errorf("control connect: %w", err) } @@ -233,6 +263,7 @@ func (c *Client) handleControlRequest(cid []byte, pattern, ip string, port int, st := newStream(wc, sid, cid, mapping, ip, port) wc.registerStream(sid, st) wc.sendSyn(sid, cid) + go st.writeLoop() go st.run() } diff --git a/client/config.example.json b/client/config.example.json index ad385ed..8d873ee 100644 --- a/client/config.example.json +++ b/client/config.example.json @@ -3,6 +3,7 @@ "psk": "change-me-to-a-long-random-passphrase", "maxConn": 4, "pingIntervalMs": 20000, + "streamWindowBytes": 262144, "mappings": [ { "pattern": "mc\\.example\\.com", diff --git a/client/config.go b/client/config.go index a9975b5..4f43b3e 100644 --- a/client/config.go +++ b/client/config.go @@ -29,10 +29,24 @@ const ( MuxData = 0x01 MuxFin = 0x02 MuxRst = 0x03 + MuxWnd = 0x04 FrameError = 0x7F SaturationThreshold = 8 + + // Session-establishment feature flags (trailing VarInt on the Rekey message). + FlagStreamFC = 0x01 + + // 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. + DefaultStreamWindow = 256 * 1024 + MinStreamWindow = 32 * 1024 + MaxStreamWindow = 8 << 20 + + // DataChunkSize caps a single DATA frame's payload so no stream monopolizes + // the shared worker connection for long. + DataChunkSize = 32 * 1024 ) // Mapping routes a registered pattern to a real destination. @@ -44,11 +58,12 @@ 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"` - 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 + Mappings []Mapping `json:"mappings"` } // LoadConfig reads and validates a JSON config file. diff --git a/client/worker.go b/client/worker.go index 12fbfc1..66e4754 100644 --- a/client/worker.go +++ b/client/worker.go @@ -55,18 +55,21 @@ func (p *WorkerPool) Allocate() (*WorkerConn, int, error) { } func (p *WorkerPool) dialWorker() (*WorkerConn, error) { - fc, err := p.client.dialSession(MagicWorker) + fc, peerWnd, err := p.client.dialSession(MagicWorker) if err != nil { return nil, err } wc := &WorkerConn{ - pool: p, - fc: fc, - streams: make(map[int]*Stream), - nextSid: 1, + pool: p, + fc: fc, + sendWndInit: peerWnd, + recvWndInit: p.client.streamWnd, + streams: make(map[int]*Stream), + nextSid: 1, } go wc.readLoop() - log.Printf("opened worker conn (#%d in pool)", len(p.conns)+1) + log.Printf("opened worker conn (#%d in pool, send window %d, recv window %d)", + len(p.conns)+1, wc.sendWndInit, wc.recvWndInit) return wc, nil } @@ -101,6 +104,9 @@ 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) + mu sync.Mutex streams map[int]*Stream nextSid int @@ -140,6 +146,9 @@ func (wc *WorkerConn) removeStream(sid int) *Stream { return st } +// readLoop dispatches inbound mux frames. It must never block on a stream's +// destination: DATA is only enqueued (the per-stream writeLoop does the actual +// destination writes), so one slow destination cannot stall other streams. func (wc *WorkerConn) readLoop() { for { payload, err := wc.fc.ReadFrame() @@ -160,9 +169,20 @@ func (wc *WorkerConn) readLoop() { if st := wc.getStream(sid); st != nil { st.deliverFromHub(r.Remaining()) } - case MuxFin, MuxRst: + case MuxWnd: + if delta, err := r.VarInt(); err == nil && delta > 0 { + if st := wc.getStream(sid); st != nil { + st.grantSendWnd(delta) + } + } + case MuxFin: + // Graceful: drain what is already queued to the destination first. if st := wc.removeStream(sid); st != nil { - st.shutdown(false) + st.gracefulFin() + } + case MuxRst: + if st := wc.removeStream(sid); st != nil { + st.teardown(false) } default: log.Printf("worker: unknown mux type %d", ftype) @@ -178,7 +198,7 @@ func (wc *WorkerConn) readLoop() { wc.streams = make(map[int]*Stream) wc.mu.Unlock() for _, st := range streams { - st.shutdown(false) + st.teardown(false) } } @@ -198,7 +218,16 @@ func (wc *WorkerConn) sendRst(sid int) { _ = wc.fc.WriteFrame(wire.NewWriter().U8(MuxRst).VarInt(sid).Out()) } +func (wc *WorkerConn) sendWndUpdate(sid, delta int) { + _ = wc.fc.WriteFrame(wire.NewWriter().U8(MuxWnd).VarInt(sid).VarInt(delta).Out()) +} + // 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 +// writeLoop goroutine. The queue is bounded by the advertised receive window — +// 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 @@ -207,25 +236,33 @@ type Stream struct { srcIP string srcPort int - mu sync.Mutex - dest net.Conn - connected bool - preBuf []byte - closed bool + mu sync.Mutex + cond *sync.Cond + dest net.Conn + connected bool + closed bool + finPending bool // hub sent FIN; close the destination once the queue drains + q [][]byte // hub -> destination, waiting for writeLoop + qBytes int + sendWnd int // flow control: budget for destination -> hub DATA + consumed int // flow control: drained bytes not yet credited back to the hub } func newStream(wc *WorkerConn, sid int, cid []byte, m Mapping, ip string, port int) *Stream { - return &Stream{wc: wc, sid: sid, cid: cid, mapping: m, srcIP: ip, srcPort: port} + s := &Stream{wc: wc, sid: sid, cid: cid, mapping: m, srcIP: ip, srcPort: port, sendWnd: wc.sendWndInit} + s.cond = sync.NewCond(&s.mu) + return s } -// run dials the destination, optionally writes the PROXY v2 header, flushes any -// buffered hub bytes, then pumps destination -> hub. +// 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) + s.teardown(false) return } if tcp, ok := dest.(*net.TCPConn); ok { @@ -240,7 +277,6 @@ func (s *Stream) run() { } } - // Atomically flush pre-connect buffer and enable direct writes. s.mu.Lock() if s.closed { s.mu.Unlock() @@ -248,18 +284,18 @@ func (s *Stream) run() { return } s.dest = dest - if len(s.preBuf) > 0 { - _, _ = dest.Write(s.preBuf) - s.preBuf = nil - } s.connected = true + s.cond.Broadcast() // wake writeLoop: queued hub bytes can flow now s.mu.Unlock() // destination -> hub - buf := make([]byte, 32*1024) + buf := make([]byte, DataChunkSize) for { n, err := dest.Read(buf) if n > 0 { + if !s.acquireSendWnd(n) { + break + } if werr := s.wc.sendData(s.sid, buf[:n]); werr != nil { break } @@ -268,7 +304,99 @@ func (s *Stream) run() { break } } - s.shutdown(true) + s.teardown(true) +} + +// 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. +func (s *Stream) writeLoop() { + for { + s.mu.Lock() + for !s.closed && (!s.connected || (len(s.q) == 0 && !s.finPending)) { + s.cond.Wait() + } + if s.closed { + s.mu.Unlock() + return + } + if len(s.q) == 0 { // finPending and fully drained + s.mu.Unlock() + s.teardown(false) + return + } + data := s.q[0] + s.q = s.q[1:] + s.qBytes -= len(data) + dest := s.dest + s.mu.Unlock() + + if _, err := dest.Write(data); err != nil { + s.teardown(true) + return + } + s.credit(len(data)) + } +} + +// deliverFromHub enqueues hub bytes for the destination. Called from the worker +// readLoop; it never blocks — a peer that exceeds the advertised window is a +// protocol violator and gets the stream reset. +func (s *Stream) deliverFromHub(data []byte) { + s.mu.Lock() + if s.closed || s.finPending { + s.mu.Unlock() + return + } + if s.qBytes+len(data) > s.wc.recvWndInit { + 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) + s.teardown(false) + return + } + s.q = append(s.q, data) + s.qBytes += len(data) + s.cond.Broadcast() + s.mu.Unlock() +} + +// acquireSendWnd blocks until the stream may send n more bytes to the hub. +// Returns false if the stream closed while waiting. +func (s *Stream) acquireSendWnd(n int) bool { + s.mu.Lock() + defer s.mu.Unlock() + for !s.closed && s.sendWnd < n { + s.cond.Wait() + } + if s.closed { + return false + } + s.sendWnd -= n + return true +} + +func (s *Stream) grantSendWnd(delta int) { + s.mu.Lock() + s.sendWnd += delta + s.cond.Broadcast() + s.mu.Unlock() +} + +// credit accounts bytes drained to the destination and grants the hub more +// window once half of our receive window has been consumed. +func (s *Stream) credit(n int) { + s.mu.Lock() + s.consumed += n + if s.closed || s.consumed*2 < s.wc.recvWndInit { + s.mu.Unlock() + return + } + delta := s.consumed + s.consumed = 0 + s.mu.Unlock() + s.wc.sendWndUpdate(s.sid, delta) } func (s *Stream) buildProxyHeader(dest net.Conn) []byte { @@ -283,28 +411,29 @@ func (s *Stream) buildProxyHeader(dest net.Conn) []byte { return BuildProxyV2(srcIP, s.srcPort, dstTCP.IP, dstTCP.Port) } -// deliverFromHub writes bytes coming from the hub to the destination, buffering -// until the destination connection is established. -func (s *Stream) deliverFromHub(data []byte) { +// finDrainTimeout bounds how long a FIN'd stream may keep draining its queue +// into the destination, so a dead destination cannot hold the stream forever. +const finDrainTimeout = 10 * time.Second + +// gracefulFin marks the hub-side close; writeLoop finishes the queue and then +// tears the stream down. +func (s *Stream) gracefulFin() { s.mu.Lock() - if s.closed { + if s.closed || s.finPending { s.mu.Unlock() return } - if !s.connected { - s.preBuf = append(s.preBuf, data...) - s.mu.Unlock() - return + s.finPending = true + if s.dest != nil { + _ = s.dest.SetWriteDeadline(time.Now().Add(finDrainTimeout)) } - dest := s.dest + s.cond.Broadcast() s.mu.Unlock() - if _, err := dest.Write(data); err != nil { - s.shutdown(true) - } } -// shutdown closes the stream; notifyHub sends a FIN to the hub when true. -func (s *Stream) shutdown(notifyHub bool) { +// 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) { s.mu.Lock() if s.closed { s.mu.Unlock() @@ -312,6 +441,7 @@ func (s *Stream) shutdown(notifyHub bool) { } s.closed = true dest := s.dest + s.cond.Broadcast() s.mu.Unlock() if dest != nil { diff --git a/docs/architecture.md b/docs/architecture.md index 82a303a..41da6d6 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -154,25 +154,43 @@ connections (9 + 9 + 2), confirming the algorithm. simplicity and correctness. * **Client:** goroutine-per-concern. One goroutine reads each connection (control or worker); `WriteFrame` is mutex-serialized so many stream goroutines - can share a worker connection safely. A per-stream mutex guards the small - "buffer until the destination is connected, then write directly" handoff so - the forwarded handshake never races ahead of later bytes. + can share a worker connection safely. Each stream has two goroutines: `run` + pumps destination → hub, and `writeLoop` is the only writer to the + destination, draining a per-stream queue fed by the worker readLoop. The + readLoop itself never writes to a destination, so a stalled destination can + never block frame dispatch for other streams. -## 6. Back-pressure +## 6. Back-pressure & flow control -There is no per-stream credit window. Flow is governed by TCP back-pressure on -each worker connection: +Two mechanisms operate at different granularities: -* Hub → player: if a player socket's write queue fills, the hub pauses the - worker connection socket and resumes on drain. -* Destination → hub: the client's `WriteFrame` blocks when the worker socket is - congested, which naturally stops the client reading the destination. +* **Per-stream credit windows** (PROTOCOL.md §7.3; the windows are exchanged + at session establishment): each stream direction has an independent byte + budget equal to the receiver's advertised window (default 256 KiB). A sender + that exhausts a + stream's window pauses *only that stream's source* — the hub pauses the one + player socket, the client parks the one destination-reader goroutine. Credit + is granted back (`WND` frames, batched at half-window) as bytes are actually + written to the terminal socket. The result: a slow player or slow destination + jams its own stream at a bounded buffer size and nothing else. This is what + eliminates head-of-line blocking between streams. +* **Aggregate TCP back-pressure** on each worker connection: when the shared + socket itself is congested (total bandwidth, not one stream), the hub parks + all sending players until it drains, and the client's `WriteFrame` blocks. + This is fair — when the pipe is genuinely full, everyone should slow down. -The consequence is head-of-line blocking *within* a worker connection: one very -slow player can stall other streams sharing that connection. `maxConn` spreads -streams across connections to mitigate this. For interactive Minecraft traffic -(small client→server packets, bursty server→client chunk data) this is a good -trade for a near-zero-overhead mux. +The window also bounds memory: a stream can hold at most one window of +undelivered data per direction (the client's pre-connect handshake buffer is +covered by the same bound). + +Per-stream flow control is mandatory: the hub rejects a session whose Rekey +lacks the STREAM_FC flag, and the client rejects a hub that does not echo it — +peers that predate the mechanism cannot connect at all. + +What remains (by design) is TCP-level head-of-line blocking: a lost packet on +a worker connection stalls all its streams for one retransmit. That is inherent +to mux-over-TCP; the connection pool is the mitigation, and a datagram +transport (QUIC) would be the escape hatch if it ever matters. ## 7. Failure & recovery @@ -190,7 +208,8 @@ trade for a near-zero-overhead mux. ## 8. Known limitations 1. No AEAD — payload integrity/authenticity is not cryptographically guaranteed. -2. No per-stream flow control (see §6). +2. TCP-level head-of-line blocking within a worker connection (lost packets; + see §6) — per-stream flow control removes the application-level variant only. 3. Single-event-loop hub (see §5) bounds throughput to one core. 4. `Intent 18` is reserved but only stubbed (the hub logs and closes). 5. Pattern ownership is last-writer-wins; two clients registering the identical diff --git a/e2e/harness_test.go b/e2e/harness_test.go index 578f8c9..755f78a 100644 --- a/e2e/harness_test.go +++ b/e2e/harness_test.go @@ -129,6 +129,7 @@ type destMode int const ( modeEcho destMode = iota // echo every post-handshake byte modeEchoOnceClose // echo one read, then close the connection + modeBlackhole // accept but never read: immediate write back-pressure ) type proxyInfo struct { @@ -154,6 +155,7 @@ type mockDest struct { mode destMode events chan destEvent connClosed chan struct{} + done chan struct{} } func newMockDest(t *testing.T, mode destMode) *mockDest { @@ -168,8 +170,12 @@ func newMockDest(t *testing.T, mode destMode) *mockDest { mode: mode, events: make(chan destEvent, 128), connClosed: make(chan struct{}, 128), + done: make(chan struct{}), } - t.Cleanup(func() { _ = ln.Close() }) + t.Cleanup(func() { + _ = ln.Close() + close(d.done) + }) go d.serve() return d } @@ -189,6 +195,10 @@ func (d *mockDest) handle(conn net.Conn) { _ = conn.Close() d.connClosed <- struct{}{} }() + if d.mode == modeBlackhole { + <-d.done // hold the connection open without ever reading + return + } br := bufio.NewReader(conn) var ev destEvent diff --git a/e2e/slowstream_test.go b/e2e/slowstream_test.go new file mode 100644 index 0000000..013c331 --- /dev/null +++ b/e2e/slowstream_test.go @@ -0,0 +1,131 @@ +package e2e + +import ( + crand "crypto/rand" + "fmt" + "net" + "testing" + "time" + + "github.com/iceBear67/redapricot/client" + "github.com/iceBear67/redapricot/client/wire" +) + +// echoRounds pushes payload through the tunnel `rounds` times on one player +// connection, failing the test if any round stalls. +func echoRounds(t *testing.T, hubAddr string, rounds, size int) { + t.Helper() + fast := dialPlayer(t, hubAddr, "mc.local") + defer fast.Close() + payload := make([]byte, size) + for i := range payload { + payload[i] = byte(i*13 + 5) + } + for i := 0; i < rounds; i++ { + playerEcho(t, fast, payload) + } +} + +// TestSlowPlayerDoesNotStallOthers: a player that stops reading while megabytes +// are echoed back to it must not stall another stream on the same worker +// connection (maxConn=1 forces sharing). Before per-stream flow control, the +// hub paused the whole worker socket once that player's write queue filled, +// freezing every other stream's downstream data. +func TestSlowPlayerDoesNotStallOthers(t *testing.T) { + const psk = "e2e-slowplayer" + port := freePort(t) + hubAddr := fmt.Sprintf("127.0.0.1:%d", port) + startHub(t, port, psk) + dest := newMockDest(t, modeEcho) + startClient(t, hubAddr, psk, 1, []client.Mapping{ + {Pattern: "mc.local", Destination: dest.addr}, + }) + + // Slow player: 4 MiB goes out, gets echoed back, and is never read. The + // write blocks once buffers fill; the error on test-end close is expected. + slow := dialPlayer(t, hubAddr, "mc.local") + defer slow.Close() + go func() { + _, _ = slow.Write(make([]byte, 4*1024*1024)) + }() + + // Let the slow stream jam: its flow-control window fills and stays full. + time.Sleep(1 * time.Second) + + // The fast player shares the single worker conn and must still round-trip. + echoRounds(t, hubAddr, 10, 8*1024) +} + +// TestSlowDestinationDoesNotStallOthers: a destination that never reads must +// only stall its own stream. Before the async delivery queue, the client wrote +// to destinations inline in the worker readLoop, so one blocked destination +// froze every stream on the connection. +func TestSlowDestinationDoesNotStallOthers(t *testing.T) { + const psk = "e2e-slowdest" + port := freePort(t) + hubAddr := fmt.Sprintf("127.0.0.1:%d", port) + startHub(t, port, psk) + dest := newMockDest(t, modeEcho) + hole := newMockDest(t, modeBlackhole) + startClient(t, hubAddr, psk, 1, []client.Mapping{ + {Pattern: "mc.local", Destination: dest.addr}, + {Pattern: "hole.local", Destination: hole.addr}, + }) + + // This player's destination never reads: the client-side write jams after + // kernel buffers fill, with the stream's queue bounded by its window. + stuck := dialPlayer(t, hubAddr, "hole.local") + defer stuck.Close() + go func() { + _, _ = stuck.Write(make([]byte, 2*1024*1024)) + }() + + time.Sleep(1 * time.Second) + + echoRounds(t, hubAddr, 10, 8*1024) +} + +// TestLegacyClientRejected: per-stream flow control is mandatory. A client that +// performs the old session establishment — a Rekey message without the trailing +// feature flags — must be closed by the hub before SessionReady. +func TestLegacyClientRejected(t *testing.T) { + const psk = "e2e-legacyreject" + port := freePort(t) + hubAddr := fmt.Sprintf("127.0.0.1:%d", port) + startHub(t, port, psk) + + conn, err := net.DialTimeout("tcp", hubAddr, 5*time.Second) + if err != nil { + t.Fatalf("dial hub: %v", err) + } + defer conn.Close() + + pskBytes := []byte(psk) + hs := wire.BuildHandshake(767, wire.PSKAddress(pskBytes), 25565, 17) + if _, err := conn.Write(hs); err != nil { + t.Fatalf("handshake: %v", err) + } + fc := wire.NewFramedConn(conn, + wire.CipherFor(pskBytes, wire.DirS2C), + wire.CipherFor(pskBytes, wire.DirC2S), + ) + rnd := make([]byte, 16) + if _, err := crand.Read(rnd); err != nil { + t.Fatal(err) + } + // Pre-flow-control Rekey: Magic|RandLen|Rand|Timestamp with no flags. + legacyRekey := wire.NewWriter(). + U8(0x01). // control-session magic + VarInt(len(rnd)). + Bytes(rnd). + I64(time.Now().UnixMilli()). + Out() + if err := fc.WriteFrame(legacyRekey); err != nil { + t.Fatalf("rekey: %v", err) + } + + _ = conn.SetReadDeadline(time.Now().Add(10 * time.Second)) + if payload, err := fc.ReadFrame(); err == nil { + t.Fatalf("expected the hub to close a legacy session, got frame %v", payload) + } +} diff --git a/server/config.example.json b/server/config.example.json index 803c98d..cf3c22b 100644 --- a/server/config.example.json +++ b/server/config.example.json @@ -2,5 +2,6 @@ "listen": "0.0.0.0:25565", "psk": "change-me-to-a-long-random-passphrase", "timestampWindowMs": 30000, - "pendingTimeoutMs": 10000 + "pendingTimeoutMs": 10000, + "streamWindowBytes": 262144 } diff --git a/server/src/main/java/io/icybear/redapricot/Config.java b/server/src/main/java/io/icybear/redapricot/Config.java index a3d5e8a..3894145 100644 --- a/server/src/main/java/io/icybear/redapricot/Config.java +++ b/server/src/main/java/io/icybear/redapricot/Config.java @@ -11,7 +11,8 @@ public record Config( int port, String psk, long timestampWindowMs, - long pendingTimeoutMs + long pendingTimeoutMs, + int streamWindowBytes ) { public static Config load(Path file) throws Exception { JsonObject json = new JsonObject(Files.readString(file)); @@ -25,11 +26,15 @@ public record Config( String psk = json.getString("psk"); if (psk == null || psk.isEmpty()) throw new IllegalArgumentException("psk is required"); + int window = json.getInteger("streamWindowBytes", Protocol.DEFAULT_STREAM_WINDOW); + window = Math.max(Protocol.MIN_STREAM_WINDOW, Math.min(Protocol.MAX_STREAM_WINDOW, window)); + return new Config( host, port, psk, json.getLong("timestampWindowMs", 30_000L), - json.getLong("pendingTimeoutMs", 10_000L)); + json.getLong("pendingTimeoutMs", 10_000L), + window); } } diff --git a/server/src/main/java/io/icybear/redapricot/HubConnection.java b/server/src/main/java/io/icybear/redapricot/HubConnection.java index 41f0d4c..abe71d2 100644 --- a/server/src/main/java/io/icybear/redapricot/HubConnection.java +++ b/server/src/main/java/io/icybear/redapricot/HubConnection.java @@ -4,6 +4,7 @@ import io.icybear.redapricot.crypto.Crypto; import io.icybear.redapricot.net.EncryptedFrames; import io.icybear.redapricot.util.Hex; import io.icybear.redapricot.util.ProtoReader; +import io.icybear.redapricot.util.ProtoWriter; import io.icybear.redapricot.util.VarInt; import io.vertx.core.buffer.Buffer; import io.vertx.core.net.NetSocket; @@ -145,6 +146,26 @@ public final class HubConnection { return; } + // Mandatory trailing feature flags: the client must offer per-stream flow + // control and advertise its receive window. Anything else is an + // unsupported (pre-flow-control) peer and is rejected. + int flags; + int peerWindow; + try { + flags = r.readVarInt(); + peerWindow = r.readVarInt(); + } catch (RuntimeException e) { + LOG.warn("{} rekey without feature flags (unsupported client version); closing", id); + frames.close(); + return; + } + if ((flags & Protocol.FLAG_STREAM_FC) == 0 || peerWindow <= 0) { + LOG.warn("{} client did not offer per-stream flow control; closing", id); + frames.close(); + return; + } + peerWindow = Math.min(peerWindow, Protocol.MAX_STREAM_WINDOW); + // REKEY = Rand || Timestamp(I64 big-endian). Magic is excluded. byte[] rekey = new byte[randLen + 8]; System.arraycopy(rand, 0, rekey, 0, randLen); @@ -157,7 +178,12 @@ public final class HubConnection { frames.switchCiphers( Crypto.decryptCipher(rekey, Crypto.DIR_C2S), Crypto.encryptCipher(rekey, Crypto.DIR_S2C)); - frames.send(new byte[]{(byte) Protocol.CTL_SESSION_READY}); + // Echo the accepted flags plus our own receive window. + frames.send(new ProtoWriter() + .u8(Protocol.CTL_SESSION_READY) + .varInt(Protocol.FLAG_STREAM_FC) + .varInt(hub.config.streamWindowBytes()) + .toBytes()); if (magic == Protocol.MAGIC_CONTROL) { ControlSession session = new ControlSession(hub, frames, id); @@ -165,10 +191,10 @@ public final class HubConnection { closeCleanup = session::onClose; LOG.info("{} control session established", id); } else if (magic == Protocol.MAGIC_WORKER) { - WorkerConn worker = new WorkerConn(hub, frames, id); + WorkerConn worker = new WorkerConn(hub, frames, id, peerWindow, hub.config.streamWindowBytes()); frames.setHandler(worker::onFrame); closeCleanup = worker::onClose; - LOG.info("{} worker conn established", id); + LOG.info("{} worker conn established (peer window {})", id, peerWindow); } else { LOG.warn("{} bad magic {}; closing", id, magic); frames.close(); diff --git a/server/src/main/java/io/icybear/redapricot/Protocol.java b/server/src/main/java/io/icybear/redapricot/Protocol.java index 791304b..7374e92 100644 --- a/server/src/main/java/io/icybear/redapricot/Protocol.java +++ b/server/src/main/java/io/icybear/redapricot/Protocol.java @@ -31,6 +31,16 @@ public final class Protocol { public static final int MUX_DATA = 0x01; public static final int MUX_FIN = 0x02; public static final int MUX_RST = 0x03; + public static final int MUX_WND = 0x04; // per-stream flow-control credit grant + + // Session-establishment feature flags (trailing VarInt on the Rekey message, + // echoed after the SessionReady type byte when accepted). + public static final int FLAG_STREAM_FC = 0x01; + + // Per-stream flow-control window bounds (bytes). + public static final int DEFAULT_STREAM_WINDOW = 256 * 1024; + public static final int MIN_STREAM_WINDOW = 32 * 1024; + public static final int MAX_STREAM_WINDOW = 8 << 20; // Any redapricot connection public static final int FRAME_ERROR = 0x7F; diff --git a/server/src/main/java/io/icybear/redapricot/WorkerConn.java b/server/src/main/java/io/icybear/redapricot/WorkerConn.java index 28a229b..67bd191 100644 --- a/server/src/main/java/io/icybear/redapricot/WorkerConn.java +++ b/server/src/main/java/io/icybear/redapricot/WorkerConn.java @@ -17,20 +17,44 @@ import java.util.Set; /** * An authenticated worker connection (Magic 0x02). Multiplexes many player * streams; the client opens streams via SYN(CID) to take over pending players. + * + *

Every stream has a credit window in both directions (PROTOCOL.md §7.3), + * so one slow player only ever stalls its own stream — the shared worker + * socket is never paused because of a single stream. */ @RequiredArgsConstructor public final class WorkerConn { private static final Logger LOG = LogManager.getLogger("redapricot.worker"); + /** Cap on a single DATA frame so no stream monopolizes the shared link for long. */ + private static final int CHUNK = 32 * 1024; + private final Hub hub; private final EncryptedFrames frames; private final String id; - private final Map streams = new HashMap<>(); + private final int sendWndInit; // client's advertised per-stream receive window (our send budget) + private final int recvWndInit; // our advertised per-stream receive window (basis for credit grants) - // Backpressure for the single shared worker socket, arbitrated across all streams. - private final Set upstreamPaused = new HashSet<>(); // players parked until the worker write queue drains - private final Set downstreamBlocked = new HashSet<>(); // sids whose player write queue is full; non-empty => worker read paused - private boolean workerDrainArmed = false; // whether the worker socket's single drainHandler is set + private final Map streams = new HashMap<>(); + + // Aggregate backpressure for the single shared worker socket: players parked + // until its write queue drains. Per-stream fairness is the credit windows' + // job; this only reacts to the whole pipe being congested. + private final Set upstreamPaused = new HashSet<>(); + private boolean workerDrainArmed = false; // whether the worker socket's single drainHandler is set + + /** Per-stream flow-control bookkeeping. */ + private final class StreamState { + final NetSocket player; + int sendWnd = sendWndInit; // budget for player -> client DATA + Buffer pendingUp; // player bytes awaiting send window (player is paused meanwhile) + boolean pausedForWindow; + int credited; // client -> player bytes flushed but not yet granted back + + StreamState(NetSocket player) { + this.player = player; + } + } public void onFrame(byte[] payload) { ProtoReader r = new ProtoReader(payload); @@ -39,6 +63,7 @@ public final class WorkerConn { switch (type) { case Protocol.MUX_SYN -> handleSyn(sid, r.readBytes(Protocol.CID_LEN)); case Protocol.MUX_DATA -> handleData(sid, r.readBuffer(r.remaining())); + case Protocol.MUX_WND -> handleWnd(sid, r.readVarInt()); case Protocol.MUX_FIN, Protocol.MUX_RST -> closeStream(sid); case Protocol.FRAME_ERROR -> LOG.warn("worker {} error frame", id); default -> LOG.warn("worker {} unknown mux type {}", id, type); @@ -53,72 +78,126 @@ public final class WorkerConn { return; } NetSocket player = p.getSocket(); - streams.put(sid, player); + StreamState st = new StreamState(player); + streams.put(sid, st); // From now on the player socket belongs to this stream. player.handler(buf -> { - sendData(sid, buf.getBytes()); - if (frames.writeQueueFull()) { - player.pause(); - upstreamPaused.add(player); - armWorkerDrain(); - } + sendUpstream(sid, st, buf); + checkAggregate(st); }); - player.closeHandler(v -> onPlayerGone(sid, player)); - player.exceptionHandler(t -> onPlayerGone(sid, player)); + player.closeHandler(v -> onPlayerGone(sid, st)); + player.exceptionHandler(t -> onPlayerGone(sid, st)); // Forward the buffered handshake (and any pipelined bytes), then resume. - sendData(sid, p.getBuffered().getBytes()); - player.resume(); + sendUpstream(sid, st, p.getBuffered()); + if (!st.pausedForWindow) player.resume(); + checkAggregate(st); LOG.info("worker {} stream {} bound to {}", id, sid, p.getPattern()); } - private void handleData(int sid, Buffer data) { - NetSocket player = streams.get(sid); - if (player == null) return; - player.write(data); - // A slow player pauses the shared worker read side; the block set lets us resume only - // once every blocked player has drained (and release it if a player disconnects meanwhile). - if (player.writeQueueFull() && downstreamBlocked.add(sid)) { - if (downstreamBlocked.size() == 1) frames.socket().pause(); - player.drainHandler(v -> unblockDownstream(sid)); + /** + * Send player bytes to the client, chunked and clipped to the stream window; + * the overflow is parked in {@code pendingUp} and the player socket paused + * until the client grants more credit. + */ + private void sendUpstream(int sid, StreamState st, Buffer buf) { + if (st.pendingUp != null) { // still waiting for window; keep ordering + st.pendingUp.appendBuffer(buf); + return; } + int off = drainUpstream(sid, st, buf, 0); + if (off < buf.length()) { + st.pendingUp = buf.getBuffer(off, buf.length()); + if (!st.pausedForWindow) { + st.pausedForWindow = true; + st.player.pause(); + } + } + } + + /** Send from {@code buf[off..]} within the stream window, chunked; returns the new offset. */ + private int drainUpstream(int sid, StreamState st, Buffer buf, int off) { + while (off < buf.length() && st.sendWnd > 0) { + int n = Math.min(Math.min(CHUNK, st.sendWnd), buf.length() - off); + sendData(sid, buf.getBytes(off, off + n)); + st.sendWnd -= n; + off += n; + } + return off; + } + + /** The client granted {@code delta} more bytes of credit on a stream. */ + private void handleWnd(int sid, int delta) { + StreamState st = streams.get(sid); + if (st == null || delta <= 0) return; + st.sendWnd += delta; + if (st.pendingUp != null) { + Buffer pending = st.pendingUp; + int off = drainUpstream(sid, st, pending, 0); + st.pendingUp = off >= pending.length() ? null : pending.getBuffer(off, pending.length()); + } + if (st.pendingUp == null && st.pausedForWindow) { + st.pausedForWindow = false; + if (!upstreamPaused.contains(st)) st.player.resume(); + } + checkAggregate(st); + } + + private void handleData(int sid, Buffer data) { + StreamState st = streams.get(sid); + if (st == null) return; + // Never pause the shared socket: the client bounds what it sends per + // stream to our advertised window, so a slow player only piles up a + // bounded amount in its own write queue; credit is granted back as the + // write completes (i.e. the bytes reached the player socket). + int len = data.length(); + st.player.write(data).onComplete(ar -> { + if (ar.failed() || frames.isClosed() || streams.get(sid) != st) return; + st.credited += len; + if (st.credited * 2 >= recvWndInit) { + int delta = st.credited; + st.credited = 0; + sendWnd(sid, delta); + } + }); } private void closeStream(int sid) { - NetSocket player = streams.remove(sid); - unblockDownstream(sid); - if (player != null) { - upstreamPaused.remove(player); - player.close(); + StreamState st = streams.remove(sid); + if (st != null) { + upstreamPaused.remove(st); + st.player.close(); } } - /** Register (once) the shared worker socket's single drain handler; on drain, wake every parked player. */ + /** Park the player if the shared worker socket's write queue is congested. */ + private void checkAggregate(StreamState st) { + if (frames.writeQueueFull() && upstreamPaused.add(st)) { + st.player.pause(); + armWorkerDrain(); + } + } + + /** Register (once) the shared worker socket's single drain handler; on drain, wake parked players. */ private void armWorkerDrain() { if (workerDrainArmed) return; workerDrainArmed = true; frames.socket().drainHandler(v -> { workerDrainArmed = false; if (upstreamPaused.isEmpty()) return; - NetSocket[] parked = upstreamPaused.toArray(new NetSocket[0]); + StreamState[] parked = upstreamPaused.toArray(new StreamState[0]); upstreamPaused.clear(); - for (NetSocket pl : parked) pl.resume(); + for (StreamState st : parked) { + if (!st.pausedForWindow) st.player.resume(); + } }); } - /** A blocked player drained or vanished: drop its downstream block, resuming the shared worker read side when none remain. */ - private void unblockDownstream(int sid) { - if (downstreamBlocked.remove(sid) && downstreamBlocked.isEmpty()) { - frames.socket().resume(); - } - } - /** The player side of a stream vanished: drop it from every table, release any backpressure it held, and FIN the peer if still live. */ - private void onPlayerGone(int sid, NetSocket player) { - boolean wasLive = streams.remove(sid) != null; - upstreamPaused.remove(player); - unblockDownstream(sid); + private void onPlayerGone(int sid, StreamState st) { + boolean wasLive = streams.remove(sid) == st; + upstreamPaused.remove(st); if (wasLive) sendFin(sid); } @@ -134,10 +213,13 @@ public final class WorkerConn { frames.send(new ProtoWriter().u8(Protocol.MUX_RST).varInt(sid).toBytes()); } + private void sendWnd(int sid, int delta) { + frames.send(new ProtoWriter().u8(Protocol.MUX_WND).varInt(sid).varInt(delta).toBytes()); + } + public void onClose() { - for (NetSocket player : streams.values()) player.close(); + for (StreamState st : streams.values()) st.player.close(); streams.clear(); - downstreamBlocked.clear(); upstreamPaused.clear(); LOG.info("worker {} closed", id); } diff --git a/server/src/test/java/io/icybear/redapricot/CryptoCodecTest.java b/server/src/test/java/io/icybear/redapricot/CryptoCodecTest.java index 657de50..c29f74c 100644 --- a/server/src/test/java/io/icybear/redapricot/CryptoCodecTest.java +++ b/server/src/test/java/io/icybear/redapricot/CryptoCodecTest.java @@ -77,7 +77,7 @@ class CryptoCodecTest { /** A Hub whose event loop is never touched (register/match/normalize use no Vert.x state). */ private static Hub testHub() { - return new Hub(null, new Config("0.0.0.0", 25565, "test-psk", 30_000L, 10_000L)); + return new Hub(null, new Config("0.0.0.0", 25565, "test-psk", 30_000L, 10_000L, Protocol.DEFAULT_STREAM_WINDOW)); } private static ControlSession testSession(Hub hub, String id) {