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

@@ -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)

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):
 		}
 	}

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

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.

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):
 		}
 	}