From 152f091490481f2f8fa6af57f9171bf194c728aa Mon Sep 17 00:00:00 2001 From: neuronori Date: Thu, 25 Jun 2026 12:02:40 +0000 Subject: [PATCH] fix(client,jitsi): bound peer wait, guard reconnect SYN race, drop pre-latch broadcasts - bound WaitForPeer in bringUpLink/tryReopenSession to handshake timeout so a missing peer fails fast instead of hanging before the SOCKS listener - call waitForPeer on the reconnect path too: resetLinkPeer clears the peer epoch, so without it the post-reconnect SYN re-races the server bridge - restore strict requireTargetedPeer guard: the server always replies targeted (it latches the client epoch from the SYN before smux replies), so untargeted broadcasts before latch must be dropped - remove per-frame debug logging on the bridge send/recv hot path - extract buildSmuxClient / establishPeerSession helpers and wrap interface errors to satisfy cyclop/nestif/wrapcheck --- internal/client/client.go | 108 +++++++++++++------- internal/engine/jitsi/jitsi.go | 18 ++-- internal/server/server.go | 84 ++++++++------- internal/transport/datachannel/transport.go | 5 +- internal/transport/vp8channel/transport.go | 2 +- 5 files changed, 133 insertions(+), 84 deletions(-) diff --git a/internal/client/client.go b/internal/client/client.go index df35b49..c845383 100644 --- a/internal/client/client.go +++ b/internal/client/client.go @@ -224,34 +224,20 @@ func (c *Client) bringUpLink( return fmt.Errorf("failed to connect link: %w", err) } - if waiter, ok := ln.(transport.PeerReadyTransport); ok { - if err := waiter.WaitForPeer(ctx); err != nil { - return fmt.Errorf("wait for peer: %w", err) - } + if err := waitForPeer(ctx, ln); err != nil { + return err } c.conn = muxconn.New(ln, c.cipher) c.controlConn = muxconn.NewControl(ln, c.cipher) - sess, err := smux.Client(c.conn, smuxConfig(linkMaxPayload(ln))) + sess, controlSess, err := buildSmuxClient(ln, c.conn, c.controlConn) if err != nil { - return fmt.Errorf("smux client: %w", err) - } - - // If the transport has an isolated control plane, open the handshake/ - // control smux session over it instead of the bulk data session. - var controlSess *smux.Session - if c.controlConn != nil { - var cerr error - controlSess, cerr = smux.Client(c.controlConn, controlSmuxConfig(linkMaxPayload(ln))) - if cerr != nil { - _ = sess.Close() - _ = c.conn.Close() + _ = c.conn.Close() + if c.controlConn != nil { _ = c.controlConn.Close() - return fmt.Errorf("control smux client: %w", cerr) } - } else { - controlSess = sess + return err } control, sid, err := openControlStream(ctx, controlSess, c.deviceID, c.claims) @@ -281,6 +267,56 @@ func (c *Client) bringUpLink( return nil } +// peerWaitTimeout bounds how long bringUpLink/tryReopenSession will block +// waiting for the remote peer to appear before giving up. Without a bound a +// missing peer (server offline, wrong room, never joins) would hang the +// caller indefinitely — before the SOCKS listener is even created — instead +// of surfacing a failure. We reuse the handshake timeout so a missing peer +// fails on the same ~15s budget as a wedged handshake would. +const peerWaitTimeout = handshake.DefaultTimeout + +// waitForPeer blocks until the transport reports a remote peer is ready, the +// peer-wait deadline elapses, or ctx is cancelled. Transports that don't +// implement PeerReadyTransport return immediately. +func waitForPeer(ctx context.Context, ln transport.Transport) error { + waiter, ok := ln.(transport.PeerReadyTransport) + if !ok { + return nil + } + waitCtx, cancel := context.WithTimeout(ctx, peerWaitTimeout) + defer cancel() + if err := waiter.WaitForPeer(waitCtx); err != nil { + return fmt.Errorf("wait for peer: %w", err) + } + return nil +} + +// buildSmuxClient creates the bulk-data smux session over conn and, when the +// transport exposes an isolated control plane (controlConn != nil), a separate +// smux session over controlConn for handshake/control traffic. When there is +// no control plane the bulk session doubles as the control session. On error +// any session opened here is closed; the caller owns conn/controlConn. +func buildSmuxClient( + ln transport.Transport, + conn, controlConn *muxconn.Conn, +) (*smux.Session, *smux.Session, error) { + sess, err := smux.Client(conn, smuxConfig(linkMaxPayload(ln))) + if err != nil { + return nil, nil, fmt.Errorf("smux client: %w", err) + } + if controlConn == nil { + return sess, sess, nil + } + // Separate smux session for the control stream only: small buffers, no + // smux keepalive (our own control.Run ping/pong handles liveness). + controlSess, err := smux.Client(controlConn, controlSmuxConfig(linkMaxPayload(ln))) + if err != nil { + _ = sess.Close() + return nil, nil, fmt.Errorf("control smux client: %w", err) + } + return sess, controlSess, nil +} + // openControlStream opens stream #1 on sess and performs the handshake. // The stream stays open for the lifetime of the smux session and carries // post-handshake control messages. @@ -516,31 +552,25 @@ func (c *Client) tryReopenSession( _ = oldCtrl.Close() } - // When we have a dedicated control conn, open the handshake/control smux - // session over it. Otherwise fall back to the data conn (legacy transports). - controlSmuxConn := conn - if controlConn != nil { - controlSmuxConn = controlConn - } - - sess, err := smux.Client(conn, smuxConfig(linkMaxPayload(c.ln))) + sess, controlSess, err := buildSmuxClient(c.ln, conn, controlConn) if err != nil { logger.Warnf("smux re-init failed (attempt %d): %v", attempt, err) return false } - var controlSess *smux.Session - if controlConn != nil { - // Separate smux session for the control stream only. We use a minimal - // config: small buffers, no keepalive (liveness is our own ping/pong). - controlSess, err = smux.Client(controlSmuxConn, controlSmuxConfig(linkMaxPayload(c.ln))) - if err != nil { - logger.Warnf("control smux re-init failed (attempt %d): %v", attempt, err) - _ = sess.Close() - return false + // Wait for the peer to re-announce before opening the control stream. + // resetLinkPeer cleared the peer epoch on reconnect, so without this the + // client's first SYN can race ahead of the server's bridge being open + // again — the same handshake-ordering race WaitForPeer guards against on + // the initial connect. The retry/backoff loop would eventually recover, + // but only after a full handshake timeout per attempt. + if err := waitForPeer(ctx, c.ln); err != nil { + logger.Warnf("wait for peer on reconnect failed (attempt %d): %v", attempt, err) + _ = sess.Close() + if controlSess != sess { + _ = controlSess.Close() } - } else { - controlSess = sess + return false } ctrlStream, sid, err := openControlStreamTimeout(ctx, controlSess, c.deviceID, c.claims, handshake.DefaultTimeout) diff --git a/internal/engine/jitsi/jitsi.go b/internal/engine/jitsi/jitsi.go index d29bb76..9ed98f6 100644 --- a/internal/engine/jitsi/jitsi.go +++ b/internal/engine/jitsi/jitsi.go @@ -1094,7 +1094,6 @@ func (s *Session) sendLoop() { if !ok { return } - logger.Debugf("jitsi: bridge send broadcast len=%d", len(data)) s.sendBridgeFrame("", data) case frame, ok := <-s.peerSendQueue: if !ok { @@ -1171,7 +1170,6 @@ func (s *Session) recvLoop() { case <-s.done: return case msg, ok := <-msgs: - logger.Debugf("jitsi: recvLoop got msg ok=%v class=%q from=%q", ok, msg.Class, msg.From) if !s.deliverBridgeMessage(msg, ok) { return } @@ -1189,7 +1187,6 @@ func (s *Session) deliverBridgeMessage(msg j.BridgeMessage, ok bool) bool { } return false } - logger.Debugf("jitsi: bridge msg class=%q from=%q len=%d", msg.Class, msg.From, len(msg.RawJSON)) payload, valid := bridgePayload(msg) if !valid { return true @@ -1273,12 +1270,17 @@ func (s *Session) acceptEpochFrame(payload []byte) ([]byte, bool) { receiverEpoch, s.localEpoch.Load()) return nil, false } - // Drop untargeted (broadcast) frames from unknown or third-party senders. - // Broadcasts from our latched peer are always accepted — the server may - // send welcome frames as broadcasts before it learns our localEpoch. + // Drop untargeted (broadcast) frames unless they come from the peer we + // have already latched onto. The server's first reply to a client is + // always targeted (it latches the client's localEpoch from the client's + // initial SYN frame before smux ever emits a reply), so a legitimate + // welcome carries receiverEpoch == localEpoch and never reaches this + // branch. Untargeted frames here are therefore either a third-party + // olcrtc instance broadcasting before our peer does, or post-latch + // broadcasts from our own peer (which we keep accepting). if s.requireTargetedPeer && s.onPeerData == nil && receiverEpoch != s.localEpoch.Load() { knownPeerEpoch := s.peerEpoch.Load() - if knownPeerEpoch != 0 && senderEpoch != knownPeerEpoch { + if knownPeerEpoch == 0 || senderEpoch != knownPeerEpoch { logger.Debugf("jitsi: drop untargeted bridge frame senderEpoch=0x%08x localEpoch=0x%08x", senderEpoch, s.localEpoch.Load()) return nil, false @@ -1842,7 +1844,7 @@ func (s *Session) WaitForPeer(ctx context.Context) error { } select { case <-ctx.Done(): - return ctx.Err() + return fmt.Errorf("wait for peer: %w", ctx.Err()) case <-time.After(pollInterval): } } diff --git a/internal/server/server.go b/internal/server/server.go index 2e86148..d654fbf 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -861,41 +861,8 @@ func (s *Server) acceptHandshake(ctx context.Context, sess *smux.Session) bool { } func (s *Server) servePeer(ps *peerSession) { - if ps.sessionID == "" { - s.sessMu.RLock() - hasControl := s.controlConn != nil - s.sessMu.RUnlock() - if !hasControl { - // No isolated control plane (e.g. datachannel in peer-routing mode): - // drive the handshake inline on this peer's smux session, mirroring - // the legacy path in waitHandshake/serveSingle. - if !s.acceptHandshake(s.baseCtx, ps.session) { - s.removePeerSession(ps.peerID, "handshake failed") - return - } - s.sessMu.RLock() - ps.sessionID = s.sessionID - ps.deviceID = s.deviceID - s.sessMu.RUnlock() - } else { - // Isolated control plane: spin-wait until acceptHandshake completes. - for { - if s.stopping() { - s.removePeerSession(ps.peerID, "closed") - return - } - s.sessMu.RLock() - sid := s.sessionID - did := s.deviceID - s.sessMu.RUnlock() - if sid != "" { - ps.sessionID = sid - ps.deviceID = did - break - } - time.Sleep(10 * time.Millisecond) - } - } + if ps.sessionID == "" && !s.establishPeerSession(ps) { + return } for { if s.stopping() { @@ -918,6 +885,53 @@ func (s *Server) servePeer(ps *peerSession) { } } +// establishPeerSession completes the handshake for a freshly accepted peer +// session and populates ps.sessionID/deviceID. It returns false if the peer +// should be dropped (handshake failed or the server is shutting down). +func (s *Server) establishPeerSession(ps *peerSession) bool { + s.sessMu.RLock() + hasControl := s.controlConn != nil + s.sessMu.RUnlock() + if !hasControl { + // No isolated control plane (e.g. datachannel in peer-routing mode): + // drive the handshake inline on this peer's smux session, mirroring + // the legacy path in waitHandshake/serveSingle. + if !s.acceptHandshake(s.baseCtx, ps.session) { + s.removePeerSession(ps.peerID, "handshake failed") + return false + } + s.sessMu.RLock() + ps.sessionID = s.sessionID + ps.deviceID = s.deviceID + s.sessMu.RUnlock() + return true + } + // Isolated control plane: spin-wait until acceptHandshake completes. + return s.waitPeerHandshake(ps) +} + +// waitPeerHandshake blocks until the isolated control plane has completed the +// handshake and published a session ID, copying it onto ps. Returns false if +// the server stops before the handshake lands. +func (s *Server) waitPeerHandshake(ps *peerSession) bool { + for { + if s.stopping() { + s.removePeerSession(ps.peerID, "closed") + return false + } + s.sessMu.RLock() + sid := s.sessionID + did := s.deviceID + s.sessMu.RUnlock() + if sid != "" { + ps.sessionID = sid + ps.deviceID = did + return true + } + time.Sleep(10 * time.Millisecond) + } +} + func (s *Server) resetLinkPeer() { s.sessMu.RLock() ln := s.ln diff --git a/internal/transport/datachannel/transport.go b/internal/transport/datachannel/transport.go index 0dde74c..2ed8c55 100644 --- a/internal/transport/datachannel/transport.go +++ b/internal/transport/datachannel/transport.go @@ -136,7 +136,10 @@ func (p *streamTransport) WaitForPeer(ctx context.Context) error { if !ok { return nil } - return waiter.WaitForPeer(ctx) + if err := waiter.WaitForPeer(ctx); err != nil { + return fmt.Errorf("wait for peer: %w", err) + } + return nil } // Features describes the current datachannel transport semantics. diff --git a/internal/transport/vp8channel/transport.go b/internal/transport/vp8channel/transport.go index dad0a10..a9b315e 100644 --- a/internal/transport/vp8channel/transport.go +++ b/internal/transport/vp8channel/transport.go @@ -584,7 +584,7 @@ func (p *streamTransport) WaitForPeer(ctx context.Context) error { } select { case <-ctx.Done(): - return ctx.Err() + return fmt.Errorf("wait for peer: %w", ctx.Err()) case <-time.After(pollInterval): } }