cloud/redapricot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: harden resume/shutdown paths, tighten Intent-18 and PSK handshake handling

Browse Source 
Client (Go) — resume correctness
- C1: completeResume now re-parks the stream when replay fails mid-conn-loss.
  parked was cleared before the replay loop, so the dying conn's teardown
  would start a second resumeLoop and the two loops could strand the stream
  with neither alive. Resume stats are counted only after the replay lands.
- C2: RST(ALREADY_BOUND) is retryable instead of terminating the loop. With
  C1 fixed there is never a genuine second attempt, so "already bound" means
  the hub still holds the stream on a half-open conn; the retry waits out
  that bind (bounded by the grace deadline, teardown on expiry) instead of
  returning and leaving the destination socket hung forever.

Client (Go) — shutdown semantics
- C3: Close() sets a closing flag and cancels an internal context; dialSession
  takes a ctx (DialContext + AfterFunc so shutdown aborts in-flight
  handshakes); the worker pool refuses new conns after closeAll (Allocate,
  background growth, cond waiters); serveControl's reconnect loop is gated by
  closing so Close works even when the caller's Start context is not
  cancelled; conn-loss teardown closes streams outright during shutdown
  instead of parking them for a reattach that will never come.

Client (Go) — hygiene
- C4: pingInterval() clamps at the single point a duration is derived, so a
  hand-built Config with PingIntervalMs <= 0 can no longer panic
  time.NewTicker (added DefaultPingIntervalMs).
- E6: shaperStall is sampled right after shaper.Acquire, before the socket
  write, so a hub that is not reading is no longer charged to the bandwidth
  cap in the stats.
- E7: stream log lines now carry conn%d/sid%d (leg.String()), making streams
  traceable across reattaches.
- P5: mirror constants IntentReserved/RegisterOk/RegisterErrPattern added;
  RegisterAck dispatch logs rejection reasons via the named codes.

Hub (Java) + PROTOCOL.md
- P3: Intent 18 replies with a Minecraft status-response packet
  ([Len: VarInt][0x00][JSON: String]) and closes (socket.end, so the write
  always lands) instead of closing silently; documented in PROTOCOL.md §2.
- P4: PSK address check is strict equality with the lowercase hex address;
  an uppercase/case-folded variant is now rejected per PROTOCOL.md §2.
- P7: PROTOCOL.md §5 SessionReady row lists its real fields
  (Flags/RecvWindow/ResumeGraceMs) instead of "(none)".

Verified: go vet, go test -race ./client/..., gradle test, full e2e suite
(twice), resume e2e 3x, plus live probes of the hub with the real client
codec (Intent-18 status reply, strict-lowercase PSK acceptance/rejection).%
This commit is contained in:
iceBear67
2026-08-15 17:47:39 +08:00
parent 7bd84af48d
commit da17140583
7 changed files with 504 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/client.go
+51 -9
View File
@@ -25,6 +25,18 @@ type Client struct {
mappings map[string]Mapping // normalized pattern -> mapping
pool *WorkerPool
// ctx/cancel own every pool dial: Close cancels it so an in-flight dial
// aborts instead of holding a goroutine for the whole handshake timeout.
// The control path uses the caller's context from Start, which is the same
// shutdown signal by convention.
ctx context.Context
cancel context.CancelFunc
// closing is set by Close; a conn-loss teardown checks it and closes
// streams outright rather than parking them for a reattach that is never
// coming. Allocate also consults it through the pool's own flag.
closing atomic.Bool
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
@@ -35,11 +47,14 @@ type Client struct {
// New builds a client from config.
func New(cfg *Config) *Client {
ctx, cancel := context.WithCancel(context.Background())
c := &Client{
cfg: cfg,
pskBytes: []byte(cfg.PSK),
pskAddr: wire.PSKAddress([]byte(cfg.PSK)),
mappings: make(map[string]Mapping),
ctx: ctx,
cancel: cancel,
}
if _, portStr, err := net.SplitHostPort(cfg.Server); err == nil {
if p, err := net.LookupPort("tcp", portStr); err == nil {
@@ -99,11 +114,22 @@ type session struct {
// The whole exchange is bounded by HandshakeTimeout. A hub that accepts the
// socket but never answers (wedged event loop, a load balancer accepting on its
// behalf) must fail fast rather than park the caller forever.
func (c *Client) dialSession(magic byte) (sess *session, err error) {
conn, err := net.DialTimeout("tcp", c.cfg.Server, HandshakeTimeout)
//
// ctx bounds the dial: the control path passes the caller's context so a
// shutdown mid-handshake aborts the attempt, and the pool passes the client's
// own context so Close cancels in-flight dials. After the dial, a cancelled
// ctx keeps aborting by closing the conn underneath the deadline-bounded
// handshake I/O.
func (c *Client) dialSession(ctx context.Context, magic byte) (sess *session, err error) {
d := &net.Dialer{Timeout: HandshakeTimeout}
conn, err := d.DialContext(ctx, "tcp", c.cfg.Server)
if err != nil {
return nil, err
}
// When ctx ends (shutdown), close the conn so the handshake below fails
// immediately instead of waiting out its deadline.
stop := context.AfterFunc(ctx, func() { _ = conn.Close() })
defer stop()
if tcp, ok := conn.(*net.TCPConn); ok {
_ = tcp.SetNoDelay(true)
_ = tcp.SetKeepAlive(true)
@@ -227,7 +253,7 @@ func (c *Client) Start(ctx context.Context) error {
}
func (c *Client) connectControl(ctx context.Context) error {
sess, err := c.dialSession(MagicControl)
sess, err := c.dialSession(ctx, MagicControl)
if err != nil {
return fmt.Errorf("control connect: %w", err)
}
@@ -271,7 +297,7 @@ func (c *Client) serveControl(ctx context.Context, ctrl *ctrlSession) {
c.dispatchControl(ctrl, payload)
}
_ = ctrl.fc.Close()
if ctx.Err() != nil {
if ctx.Err() != nil || c.closing.Load() {
return
}
// Reconnect with backoff, but try immediately first. While the control
@@ -281,8 +307,10 @@ func (c *Client) serveControl(ctx context.Context, ctrl *ctrlSession) {
// 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; {
// backoff that has grown to the cap. Close() is authoritative on its own:
// it gates the loop directly because a caller may shut the client down by
// calling Close() without cancelling the context it passed to Start.
for backoff := time.Duration(0); ctx.Err() == nil && !c.closing.Load(); {
if backoff > 0 {
select {
case <-ctx.Done():
@@ -290,6 +318,9 @@ func (c *Client) serveControl(ctx context.Context, ctrl *ctrlSession) {
case <-time.After(backoff):
}
}
if c.closing.Load() {
return
}
if err := c.connectControl(ctx); err == nil {
return
} else {
@@ -316,7 +347,14 @@ func (c *Client) dispatchControl(ctrl *ctrlSession, payload []byte) {
case CtlRegisterAck:
pattern, _ := r.String()
status, _ := r.U8()
log.Printf("register ack %q status=%d", pattern, status)
switch status {
case RegisterOk:
log.Printf("pattern %q registered", pattern)
case RegisterErrPattern:
log.Printf("pattern %q rejected: not a valid regular expression", pattern)
default:
log.Printf("pattern %q rejected: status=%d", pattern, status)
}
case CtlControlRequest:
cid, err := r.Bytes(CIDLen)
if err != nil {
@@ -400,7 +438,7 @@ func (c *Client) handleControlRequest(cid []byte, pattern, ip string, port int)
go st.writeLoop()
go st.run()
if err := lg.wc.sendSyn(lg.sid, cid); err != nil {
log.Printf("stream %d: SYN failed: %v", lg.sid, err)
log.Printf("stream %s: SYN failed: %v", lg, err)
st.teardown(false)
}
}
@@ -409,8 +447,12 @@ func (c *Client) handleControlRequest(cid []byte, pattern, ip string, port int)
// (exposed for tests/observability).
func (c *Client) WorkerConnCount() int { return c.pool.count() }
// Close tears down the control session and all worker connections.
// Close tears down the control session and all worker connections. Idempotent:
// a second call (or a Close racing a reconnect) only re-closes what is still
// open.
func (c *Client) Close() {
c.closing.Store(true)
c.cancel() // aborts in-flight pool dials
c.mu.Lock()
fc := c.ctrl
c.mu.Unlock()
client/config.go
+33 -7
View File
@@ -12,7 +12,12 @@ import (
// Protocol constants (mirror of the Java Protocol class; see PROTOCOL.md).
const (
IntentRedapricot = 17
ProtocolVersion = 767 // arbitrary; the hub ignores it
// IntentReserved is the handshake intent reserved for redapricot
// management/status (mirror of Protocol.INTENT_RESERVED). The hub never
// pattern-matches it; it replies with a Minecraft status line and closes,
// so an operator can probe the port without joining the protocol.
IntentReserved = 18
ProtocolVersion = 767 // arbitrary; the hub ignores it
MagicControl = 0x01
MagicWorker = 0x02
@@ -27,6 +32,10 @@ const (
CtlPing = 0x05
CtlPong = 0x06
// RegisterAck status codes (mirror of Protocol.REGISTER_OK/_ERR_PATTERN).
RegisterOk = 0x00
RegisterErrPattern = 0x01 // pattern is not a valid regular expression
MuxSyn = 0x00
MuxData = 0x01
MuxFin = 0x02
@@ -45,8 +54,8 @@ const (
// 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.
// while "already bound" means the hub has the stream on another conn — the
// reattach retries until that bind dies and the hub re-parks the stream.
RstUnspecified = 0x00
RstUnknownStream = 0x01
RstAlreadyBound = 0x02
@@ -129,9 +138,15 @@ const (
// a session is declared dead and dropped.
MissedHeartbeats = 3
// DefaultPingIntervalMs is used when a config carries no (or a
// non-positive) interval. LoadConfig applies the same default.
DefaultPingIntervalMs = 20000
// MinPingIntervalMs floors the configured ping interval so the derived
// heartbeat timeout can never be short enough to cause spurious drops.
// Applied in LoadConfig, i.e. to configs that come from disk.
// Applied wherever the interval is read, not just on the file path: a
// hand-built Config (tests) carrying 0 would otherwise panic
// time.NewTicker at the call site.
MinPingIntervalMs = 1000
// DefaultResumeGraceMs is how long a parked stream keeps trying to reattach
@@ -167,9 +182,20 @@ func (c *Config) heartbeatTimeout() time.Duration {
return c.pingInterval() * MissedHeartbeats
}
// pingInterval is the configured heartbeat period.
// pingInterval is the configured heartbeat period, clamped at the single point
// where a duration is derived. LoadConfig also clamps on the file path; this
// covers Configs built directly (tests), where a 0 or negative PingIntervalMs
// would panic time.NewTicker — a panic, not a log line, because the interval
// feeds the heartbeat timeout too.
func (c *Config) pingInterval() time.Duration {
return time.Duration(c.PingIntervalMs) * time.Millisecond
ms := c.PingIntervalMs
if ms <= 0 {
ms = DefaultPingIntervalMs
}
if ms < MinPingIntervalMs {
ms = MinPingIntervalMs
}
return time.Duration(ms) * time.Millisecond
}
// Mapping routes a registered pattern to a real destination.
@@ -291,7 +317,7 @@ func LoadConfig(path string) (*Config, error) {
c.MaxConn = 8
}
if c.PingIntervalMs <= 0 {
c.PingIntervalMs = 20000
c.PingIntervalMs = DefaultPingIntervalMs
}
if c.PingIntervalMs < MinPingIntervalMs {
c.PingIntervalMs = MinPingIntervalMs
client/resume.go
+33 -12
View File
@@ -24,7 +24,7 @@ import (
var (
errResumeUnknown = errors.New("hub does not know this stream")
errResumeRaced = errors.New("another reattach already bound this stream")
errResumeRaced = errors.New("hub has this stream bound to another conn")
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")
@@ -103,11 +103,15 @@ func (s *Stream) resumeLoop(grace time.Duration) {
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
}
// errResumeRaced falls through to the retry below. The hub has this
// stream bound to a conn that is not ours — a half-open conn whose
// death the hub has not yet learned, or a bind left behind by a racing
// attempt on a now-dead conn. There is no other live attempt: park is
// the only resumeLoop starter and it refuses to double-start. Retrying
// is safe precisely because nothing else owns the stream — the foreign
// bind dies with its conn, the hub re-parks, and a later RESUME lands.
// The grace deadline bounds the loop and expiry tears the stream down,
// so a hub that never re-parks cannot hang us forever.
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).
@@ -255,11 +259,6 @@ func (s *Stream) completeResume(wc *WorkerConn, sid int, res resumeResult) error
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()
@@ -269,6 +268,17 @@ func (s *Stream) completeResume(wc *WorkerConn, sid int, res resumeResult) error
n = s.client.chunk
}
if err := wc.sendData(sid, replay[:n]); err != nil {
// The conn died mid-replay. The stream is still resumable, but not
// from this leg — put it back in the parked state before returning
// so the conn's teardown takes park()'s already-branch instead of
// starting a second resumeLoop. The loop we came from keeps
// retrying with the fresh CID, which the hub re-parked alongside
// the stream when this conn died. Without this re-park the flag
// cleared above would let two loops race one stream, and a racing
// RST would then strand it with neither loop alive.
s.mu.Lock()
s.parked = true
s.mu.Unlock()
s.sendMu.Unlock()
return err
}
@@ -276,6 +286,17 @@ func (s *Stream) completeResume(wc *WorkerConn, sid int, res resumeResult) error
}
s.sendMu.Unlock()
// Counted only once the replay actually landed: a failed reattach above
// returns before this, so a conn dying mid-replay does not inflate the
// resume count with an attempt that never completed.
if s.stats != nil {
s.mu.Lock()
s.stats.resumes++
s.stats.hung += time.Since(s.parkedAt)
s.stats.replayBytes += replayed
s.mu.Unlock()
}
// A destination that closed while we were parked owed the hub a FIN that had
// nowhere to go at the time.
if owedFin {
@@ -283,7 +304,7 @@ func (s *Stream) completeResume(wc *WorkerConn, sid int, res resumeResult) error
s.teardown(false)
return nil
}
log.Printf("stream %d resumed (%d bytes replayed, %d outstanding)", sid, replayed, outstanding)
log.Printf("stream conn%d/sid%d resumed (%d bytes replayed, %d outstanding)", wc.id, sid, replayed, outstanding)
return nil
}
client/worker.go
+56 -5
View File
@@ -1,6 +1,8 @@
package client
import (
"errors"
"fmt"
"log"
"net"
"sync"
@@ -10,6 +12,11 @@ import (
"github.com/iceBear67/redapricot/client/wire"
)
// errPoolClosed is returned by Allocate after Close: the pool is shutting down
// and must not start new dials, so a caller (handleControlRequest,
// allocateForResume) gives up rather than wait on a cond no one will satisfy.
var errPoolClosed = errors.New("worker pool closed")
// StreamsBeforeGrowing is how many streams a worker conn may carry before the
// pool starts opening another one. Set to 1 so the pool fans out to maxConn
// under load *before* stacking streams: concentrating every player on a single
@@ -37,6 +44,7 @@ type WorkerPool struct {
dialing int // dials currently in flight (foreground + background)
dialGen uint64
dialErr error // most recent dial failure
closed bool // closeAll ran; no new conns may join the pool
}
func newWorkerPool(c *Client, maxConn int) *WorkerPool {
@@ -56,6 +64,13 @@ func newWorkerPool(c *Client, maxConn int) *WorkerPool {
func (p *WorkerPool) Allocate() (*WorkerConn, int, error) {
p.mu.Lock()
for {
if p.closed {
// Close won. No new conn may join the pool, so no stream may be
// placed; waiting on the cond could only be satisfied by a dial we
// must not start.
p.mu.Unlock()
return nil, 0, errPoolClosed
}
best, bestCount := p.leastLoadedLocked()
if best != nil {
p.maybeGrowLocked(bestCount)
@@ -86,6 +101,15 @@ func (p *WorkerPool) Allocate() (*WorkerConn, int, error) {
p.mu.Unlock()
return nil, 0, err
}
if p.closed {
// Close raced this dial: the conn must not enter the pool. Closing
// it here, under p.mu, is a raw socket close — fine, and it makes
// the shutdown atomic with the pool state.
p.cond.Broadcast()
p.mu.Unlock()
_ = wc.fc.Close()
return nil, 0, errPoolClosed
}
p.conns = append(p.conns, wc)
p.cond.Broadcast()
}
@@ -113,6 +137,9 @@ func (p *WorkerPool) maybeGrowLocked(bestCount int) {
if bestCount < StreamsBeforeGrowing {
return
}
if p.closed {
return // shutdown; do not start dials nobody will join the pool
}
if len(p.conns)+p.dialing >= p.maxConn {
if bestCount > SaturationThreshold {
log.Printf("worker pool at maxConn=%d with %d streams on the least-loaded conn", p.maxConn, bestCount)
@@ -129,6 +156,8 @@ func (p *WorkerPool) maybeGrowLocked(bestCount int) {
p.dialErr = err
switch {
case err != nil:
case p.closed:
surplus = wc // Close raced this background dial
case len(p.conns) < p.maxConn:
p.conns = append(p.conns, wc)
default:
@@ -146,8 +175,9 @@ func (p *WorkerPool) maybeGrowLocked(bestCount int) {
}
// dialWorker establishes one worker conn. It must be called without p.mu held.
// The dial runs on the client's context so Close aborts it mid-handshake.
func (p *WorkerPool) dialWorker() (*WorkerConn, error) {
sess, err := p.client.dialSession(MagicWorker)
sess, err := p.client.dialSession(p.client.ctx, MagicWorker)
if err != nil {
return nil, err
}
@@ -200,7 +230,11 @@ func (p *WorkerPool) remove(wc *WorkerConn) {
func (p *WorkerPool) closeAll() {
p.mu.Lock()
p.closed = true
conns := append([]*WorkerConn(nil), p.conns...)
// Wake waiters parked in Allocate: the closed flag they re-check is the
// only thing that can release them now that no conn will ever join.
p.cond.Broadcast()
p.mu.Unlock()
for _, wc := range conns {
_ = wc.fc.Close()
@@ -401,6 +435,14 @@ func (wc *WorkerConn) readLoop() {
// 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.
// During Close there is no reattach to come: a stream that parked now would
// hold its destination socket open past shutdown, so teardown instead.
if wc.pool.client.closing.Load() {
for _, st := range streams {
st.teardown(false)
}
return
}
for _, st := range streams {
if !st.park(wc.grace) {
st.teardown(false)
@@ -445,6 +487,11 @@ type leg struct {
sid int
}
// String formats one (conn, sid) snapshot for log correlation. Stream ids
// restart at 1 per conn, so a bare sid cannot be traced across a reattach —
// the conn id is what ties the log lines together.
func (lg *leg) String() string { return fmt.Sprintf("conn%d/sid%d", lg.wc.id, lg.sid) }
// 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
@@ -556,7 +603,7 @@ func (s *Stream) run() {
dest, err := net.DialTimeout("tcp", s.mapping.Destination, 10*time.Second)
if err != nil {
lg := s.conn()
log.Printf("stream %d: dial %s failed: %v", lg.sid, s.mapping.Destination, err)
log.Printf("stream %s: dial %s failed: %v", lg, s.mapping.Destination, err)
lg.wc.removeStream(lg.sid)
lg.wc.sendRst(lg.sid)
s.teardown(false)
@@ -569,7 +616,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.conn().sid, err)
log.Printf("stream %s: proxy header write: %v", s.conn(), err)
}
}
}
@@ -637,6 +684,10 @@ func (s *Stream) sendToHub(data []byte) bool {
if !s.client.shaper.Acquire(&s.share, n, s.done) {
return false
}
// Sampled here, before the socket write: emit's write to the hub can
// block when the hub is not reading, and charging that time to the
// bandwidth cap would blame maxBandwidth for a hub that is not draining.
waited := shaperStall.elapsed()
if !s.emit(data[:n]) {
return false
}
@@ -645,7 +696,7 @@ func (s *Stream) sendToHub(data []byte) bool {
// 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.shaperStall += waited
s.stats.bytesUp += int64(n)
s.mu.Unlock()
}
@@ -754,7 +805,7 @@ func (s *Stream) deliverFromHub(data []byte) {
if s.qBytes+len(data) > s.client.streamWnd {
s.mu.Unlock()
lg := s.conn()
log.Printf("stream %d: peer exceeded flow-control window; resetting", lg.sid)
log.Printf("stream %s: peer exceeded flow-control window; resetting", lg)
lg.wc.removeStream(lg.sid)
lg.wc.sendRst(lg.sid)
s.teardown(false)