fix(vp8channel): keep SFU forwarding bulk via decodable keepalive keyframe

inject a clean VP8 keyframe every ~2s during bulk so the SFU keeps
forwarding the track instead of stalling after its decode timeout.
add RTCP interceptors and IPv4-only ICE, priority control/data mux,
publisher PC reconnect with stable control epoch, and longer smux
keepalive/liveness windows to survive the reconnect gap

internal/client/client.go

@@ -54,6 +54,11 @@ type Client struct {
 	ln          transport.Transport
 	cipher      *crypto.Cipher
 	conn        *muxconn.Conn
+	// controlConn is a separate muxconn wired to the transport's control-plane
+	// channel (transport.ControlPlane). When non-nil, the smux control session
+	// runs over it instead of the bulk data conn, eliminating head-of-line
+	// blocking of control ping/pong behind large data transfers.
+	controlConn *muxconn.Conn
 	session     *smux.Session
 	controlStrm *smux.Stream
 	controlStop context.CancelFunc
@@ -66,6 +71,10 @@ type Client struct {
 	dnsServer   string
 	socksUser   string
 	socksPass   string
+	// sessionReady is closed (and replaced) each time a session becomes fully
+	// established (sessionID != ""). Tunnel handlers wait on it so they do
+	// not open smux streams before the server has accepted the handshake.
+	sessionReady chan struct{}
 }
 
 // HealthFunc is called when the client control health snapshot changes.
@@ -127,13 +136,14 @@ func RunWithReady(ctx context.Context, cfg Config, onReady func()) error {
 	}
 
 	c := &Client{
-		cipher:    cipher,
-		deviceID:  deviceID,
-		claims:    cfg.Claims,
-		dnsServer: cfg.DNSServer,
-		socksUser: cfg.SOCKSUser,
-		socksPass: cfg.SOCKSPass,
-		health:    runtime.NewHealthTracker(cfg.OnHealth),
+		cipher:       cipher,
+		deviceID:     deviceID,
+		claims:       cfg.Claims,
+		dnsServer:    cfg.DNSServer,
+		socksUser:    cfg.SOCKSUser,
+		socksPass:    cfg.SOCKSPass,
+		health:       runtime.NewHealthTracker(cfg.OnHealth),
+		sessionReady: make(chan struct{}),
 	}
 
 	// shutdown is registered BEFORE bringUpLink so we always close any
@@ -215,15 +225,39 @@ func (c *Client) bringUpLink(
 	}
 
 	c.conn = muxconn.New(ln, c.cipher)
+	c.controlConn = muxconn.NewControl(ln, c.cipher)
+
 	sess, err := smux.Client(c.conn, smuxConfig(linkMaxPayload(ln)))
 	if err != nil {
 		return fmt.Errorf("smux client: %w", err)
 	}
 
-	control, sid, err := openControlStream(ctx, sess, c.deviceID, c.claims)
+	// 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.controlConn.Close()
+			return fmt.Errorf("control smux client: %w", cerr)
+		}
+	} else {
+		controlSess = sess
+	}
+
+	control, sid, err := openControlStream(ctx, controlSess, c.deviceID, c.claims)
 	if err != nil {
 		_ = sess.Close()
+		if controlSess != sess {
+			_ = controlSess.Close()
+		}
 		_ = c.conn.Close()
+		if c.controlConn != nil {
+			_ = c.controlConn.Close()
+		}
 		return fmt.Errorf("handshake: %w", err)
 	}
 	logger.Infof("session %s opened (device=%s)", sid, c.deviceID)
@@ -233,6 +267,7 @@ func (c *Client) bringUpLink(
 	c.controlStrm = control
 	c.sessionID = sid
 	c.sessMu.Unlock()
+	c.signalSessionReady()
 	c.recordSession(sid)
 	c.startControlLoop(ctx, cfg, cancel, control)
 
@@ -323,6 +358,14 @@ func smuxConfig(maxWirePayload int) *smux.Config {
 	return runtime.SmuxConfig(maxWirePayload)
 }
 
+// controlSmuxConfig returns a lean smux config for the isolated control-plane
+// session. The control session carries only ping/pong frames, so we use
+// small buffers and disable smux keepalives (our own control.Run ping loop
+// handles liveness).
+func controlSmuxConfig(maxWirePayload int) *smux.Config {
+	return runtime.ControlSmuxConfig(maxWirePayload)
+}
+
 func linkMaxPayload(tr transport.Transport) int {
 	return runtime.MaxPayload(tr)
 }
@@ -335,7 +378,7 @@ func (c *Client) handleReconnect(ctx context.Context, cfg Config, cancel context
 	logger.Infof("client reconnect reason=%s - tearing down smux session", reason)
 	c.resetLinkPeer()
 
-	// Close the old muxconn immediately so any in-flight Push from data
+	// Close the old muxconns immediately so any in-flight Push from data
 	// arriving on the new bridge is discarded. Without this, the server
 	// side that reconnected faster can push frames into our old muxconn,
 	// corrupting the dying smux session.
@@ -343,18 +386,23 @@ func (c *Client) handleReconnect(ctx context.Context, cfg Config, cancel context
 	if c.conn != nil {
 		_ = c.conn.Close()
 	}
+	if c.controlConn != nil {
+		_ = c.controlConn.Close()
+	}
 	c.sessMu.RUnlock()
 
-	// Install a fresh muxconn immediately so onData never hits nil while
-	// the old session is being torn down. tryReopenSession will swap it
-	// again with its own conn on each attempt.
+	// Install fresh muxconns immediately so onData never hits nil while
+	// the old session is being torn down. tryReopenSession will swap them
+	// again with its own conns on each attempt.
 	newConn := muxconn.New(c.ln, c.cipher)
+	newControlConn := muxconn.NewControl(c.ln, c.cipher)
 
 	c.sessMu.Lock()
 	oldControl := c.controlStrm
 	oldControlStop := c.controlStop
 	oldSess := c.session
 	c.conn = newConn
+	c.controlConn = newControlConn
 	c.session = nil
 	c.controlStrm = nil
 	c.controlStop = nil
@@ -444,12 +492,29 @@ func (c *Client) tryReopenSession(
 ) bool {
 	conn := muxconn.New(c.ln, c.cipher)
 
+	// If the transport has an isolated control plane, build a second muxconn
+	// wired to it. The smux control stream will run over controlConn so that
+	// bulk data writes on conn can never head-of-line block control ping/pong.
+	controlConn := muxconn.NewControl(c.ln, c.cipher)
+
 	c.sessMu.Lock()
-	old := c.conn
+	oldConn := c.conn
+	oldCtrl := c.controlConn
 	c.conn = conn
+	c.controlConn = controlConn
 	c.sessMu.Unlock()
-	if old != nil {
-		_ = old.Close()
+	if oldConn != nil {
+		_ = oldConn.Close()
+	}
+	if oldCtrl != nil {
+		_ = 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)))
@@ -457,20 +522,39 @@ func (c *Client) tryReopenSession(
 		logger.Warnf("smux re-init failed (attempt %d): %v", attempt, err)
 		return false
 	}
-	control, sid, err := openControlStreamTimeout(ctx, sess, c.deviceID, c.claims, 2*time.Second)
+
+	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
+		}
+	} else {
+		controlSess = sess
+	}
+
+	ctrlStream, sid, err := openControlStreamTimeout(ctx, controlSess, c.deviceID, c.claims, handshake.DefaultTimeout)
 	if err != nil {
 		logger.Warnf("handshake on reconnect failed (attempt %d): %v", attempt, err)
 		_ = sess.Close()
+		if controlSess != sess {
+			_ = controlSess.Close()
+		}
 		return false
 	}
 	logger.Infof("session %s reopened (device=%s)", sid, c.deviceID)
 	c.sessMu.Lock()
 	c.session = sess
-	c.controlStrm = control
+	c.controlStrm = ctrlStream
 	c.sessionID = sid
 	c.sessMu.Unlock()
+	c.signalSessionReady()
 	c.recordSession(sid)
-	c.startControlLoop(ctx, cfg, cancel, control)
+	c.startControlLoop(ctx, cfg, cancel, ctrlStream)
 	return true
 }
 
@@ -539,16 +623,37 @@ func (c *Client) recordMissed(missed int)        { c.health.RecordMissed(missed)
 func (c *Client) recordUnhealthy(missed int)     { c.health.RecordUnhealthy(missed) }
 func (c *Client) recordReconnect()               { c.health.RecordReconnect() }
 
+// signalSessionReady closes the current sessionReady channel (waking any
+// waiters) and replaces it with a fresh one for the next reconnect cycle.
+func (c *Client) signalSessionReady() {
+	c.sessMu.Lock()
+	old := c.sessionReady
+	c.sessionReady = make(chan struct{})
+	c.sessMu.Unlock()
+	close(old)
+}
+
+// waitSessionReady blocks until the session is fully established (sessionID !=
+// "") or ctx is cancelled. Returns the ready channel to select on.
+func (c *Client) readyChannel() chan struct{} {
+	c.sessMu.RLock()
+	ch := c.sessionReady
+	c.sessMu.RUnlock()
+	return ch
+}
+
 func (c *Client) shutdown() {
 	c.sessMu.Lock()
 	control := c.controlStrm
 	controlStop := c.controlStop
 	sess := c.session
 	conn := c.conn
+	ctrlConn := c.controlConn
 	c.controlStrm = nil
 	c.controlStop = nil
 	c.session = nil
 	c.conn = nil
+	c.controlConn = nil
 	c.sessMu.Unlock()
 
 	notifyControlClose(control)
@@ -561,6 +666,9 @@ func (c *Client) shutdown() {
 	if conn != nil {
 		_ = conn.Close()
 	}
+	if ctrlConn != nil {
+		_ = ctrlConn.Close()
+	}
 	if c.ln != nil {
 		_ = c.ln.Close()
 	}
@@ -614,7 +722,7 @@ func (c *Client) acceptLoop(ctx context.Context, ln net.Listener) {
 	}
 }
 
-func (c *Client) handleSocks5(_ context.Context, conn net.Conn) {
+func (c *Client) handleSocks5(ctx context.Context, conn net.Conn) {
 	defer func() { _ = conn.Close() }()
 
 	if err := c.socks5Handshake(conn); err != nil {
@@ -626,15 +734,33 @@ func (c *Client) handleSocks5(_ context.Context, conn net.Conn) {
 		return
 	}
 
-	c.sessMu.RLock()
-	sess := c.session
-	c.sessMu.RUnlock()
-	if sess == nil || sess.IsClosed() {
-		_, _ = conn.Write(replyHostUnreachable())
-		return
+	// Wait until the session handshake is fully complete (sessionID != "").
+	// Without this gate, tunnel streams opened during server-side reinstall
+	// land on a dying smux session and get "closed pipe".
+	const sessionReadyTimeout = 60 * time.Second
+	readyCtx, cancel := context.WithTimeout(ctx, sessionReadyTimeout)
+	defer cancel()
+	for {
+		c.sessMu.RLock()
+		sess := c.session
+		sid := c.sessionID
+		c.sessMu.RUnlock()
+		if sess != nil && !sess.IsClosed() && sid != "" {
+			c.tunnel(conn, sess, targetAddr, targetPort)
+			return
+		}
+		// sess is nil (no session yet) or closed (reconnect in progress) —
+		// in both cases wait for readyChannel rather than failing immediately.
+		// A closed session means handleReconnect is running; a fresh session
+		// will be installed shortly by tryReopenSession.
+		select {
+		case <-readyCtx.Done():
+			_, _ = conn.Write(replyHostUnreachable())
+			return
+		case <-c.readyChannel():
+			// session became ready; re-check
+		}
 	}
-
-	c.tunnel(conn, sess, targetAddr, targetPort)
 }
 
 func (c *Client) tunnel(conn net.Conn, sess *smux.Session, targetAddr string, targetPort int) {
@@ -682,7 +808,11 @@ func (c *Client) sendConnectRequest(stream *smux.Stream, targetAddr string, targ
 	_ = stream.SetWriteDeadline(time.Time{})
 
 	ack := make([]byte, 1)
-	_ = stream.SetReadDeadline(time.Now().Add(15 * time.Second))
+	// In peer-routing mode the SFU may take up to ~30s to complete
+	// renegotiation and start forwarding data frames from the client to the
+	// server. Use a generous deadline so we do not give up before the server
+	// peer session is established.
+	_ = stream.SetReadDeadline(time.Now().Add(90 * time.Second))
 	if _, err := io.ReadFull(stream, ack); err != nil || ack[0] != 0x00 {
 		return fmt.Errorf("sid=%d: %w (read_err=%w ack=%v)", stream.ID(), ErrRemoteNotReady, err, ack)
 	}

internal/e2e/local_soak_test.go

@@ -137,7 +137,7 @@ func runLocalSoakOnce(t *testing.T, transportName string) {
 	rt := startLocalSoakTunnel(t, transportName)
 	echoAddr := startEchoServer(t)
 
-	conn, err := connectViaSOCKSWithin(rt.socksAddr, echoAddr, setupBudget)
+	conn, err := connectViaSOCKSWithin(context.Background(), rt.socksAddr, echoAddr, setupBudget)
 	if err != nil {
 		t.Fatalf("connect via SOCKS: %v", err)
 	}

internal/e2e/real_soak_test.go

@@ -125,7 +125,7 @@ func runRealSoakOnce(t *testing.T, carrierName, transportName, roomURL, echoAddr
 	}
 	_ = rt
 
-	conn, err := connectViaSOCKSWithin(rt.socksAddr, echoAddr, setupBudget)
+	conn, err := connectViaSOCKSWithin(ctx, rt.socksAddr, echoAddr, setupBudget)
 	if err != nil {
 		t.Fatalf("connect via SOCKS: %v", err)
 	}

internal/e2e/stress_test.go

@@ -11,6 +11,8 @@ import (
 	"net"
 	"runtime"
 	"slices"
+	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -131,7 +133,6 @@ func TestRealProviderTransportStress(t *testing.T) {
 	}
 }
 
-//nolint:cyclop // two phases plus tunnel/connection setup naturally branch
 func runRealE2EStressCase(t *testing.T, carrierName, transportName, roomURL, echoAddr string) (err error) {
 	t.Helper()
 
@@ -150,36 +151,12 @@ func runRealE2EStressCase(t *testing.T, carrierName, transportName, roomURL, ech
 		}
 	}()
 
-	conn, err := connectViaSOCKSWithin(rt.socksAddr, echoAddr, *realStressCaseTimeout)
-	if err != nil {
+	if err := runBulkPhase(ctx, t, rt, carrierName, transportName, echoAddr); err != nil {
 		return err
 	}
-	defer func() { _ = conn.Close() }()
 
-	if d := *realStressBulkDuration; d > 0 {
-		written, dur, err := streamPatternForDuration(conn, d, *realStressBulkChunkSize, transportName)
-		if err != nil {
-			return fmt.Errorf("bulk pump: %w", err)
-		}
-		throughput := float64(written) / dur.Seconds() / (1 << 20)
-		t.Logf("bulk %s/%s: %d bytes in %s (%.3f MiB/s)",
-			carrierName, transportName, written, dur, throughput)
-		if written == 0 {
-			return errStressNoBulkProgress
-		}
-	}
-
-	if d := *realStressDuration; d > 0 {
-		stats, err := sustainedEcho(conn, *realStressEchoSize, d, transportName)
-		if err != nil {
-			return fmt.Errorf("sustained echo: %w", err)
-		}
-		t.Logf("echo  %s/%s: %d rt in %s, p50=%s p95=%s p99=%s max=%s lost=%d",
-			carrierName, transportName, stats.count, d,
-			stats.p50, stats.p95, stats.p99, stats.maxLatency, stats.lost)
-		if stats.count == 0 {
-			return fmt.Errorf("%w: %s", errStressNoRoundtrips, d)
-		}
+	if err := runEchoPhase(ctx, t, rt, carrierName, transportName, echoAddr); err != nil {
+		return err
 	}
 
 	goroutinesAfter := runtime.NumGoroutine()
@@ -194,64 +171,204 @@ func runRealE2EStressCase(t *testing.T, carrierName, transportName, roomURL, ech
 	return nil
 }
 
+// runBulkPhase pumps bulk traffic for realStressBulkDuration, reopening the
+// SOCKS5 connection after a transport reconnect (e.g. publisher PC closed by
+// the SFU) and accumulating total bytes across reconnects.
+func runBulkPhase(
+	ctx context.Context, t *testing.T, rt *tunnelRuntime,
+	carrierName, transportName, echoAddr string,
+) error {
+	t.Helper()
+	d := *realStressBulkDuration
+	if d <= 0 {
+		return nil
+	}
+
+	var lastConn net.Conn
+	getConn := func() (net.Conn, error) {
+		if lastConn != nil {
+			_ = lastConn.Close()
+		}
+		c, cerr := connectViaSOCKSWithin(ctx, rt.socksAddr, echoAddr, 45*time.Second)
+		if cerr != nil {
+			return nil, cerr
+		}
+		lastConn = c
+		return c, nil
+	}
+	conn, err := getConn()
+	if err != nil {
+		return err
+	}
+	defer func() {
+		if lastConn != nil {
+			_ = lastConn.Close()
+		}
+	}()
+
+	totalWritten, err := pumpBulkUntil(ctx, t, conn, getConn, d, carrierName, transportName)
+	if err != nil {
+		return err
+	}
+	if totalWritten == 0 {
+		return errStressNoBulkProgress
+	}
+	// Compute approximate throughput over full wall-clock duration.
+	throughput := float64(totalWritten) / d.Seconds() / (1 << 20)
+	t.Logf("bulk %s/%s: %d bytes in %s (%.3f MiB/s) [reconnects included]",
+		carrierName, transportName, totalWritten, d, throughput)
+	return nil
+}
+
+// pumpBulkUntil drives streamPatternForDuration until the deadline, reconnecting
+// via getConn when the connection dies (transport reconnect). It returns the
+// total bytes written across reconnects.
+func pumpBulkUntil(
+	ctx context.Context, t *testing.T, conn net.Conn,
+	getConn func() (net.Conn, error), d time.Duration,
+	carrierName, transportName string,
+) (int64, error) {
+	t.Helper()
+	deadline := time.Now().Add(d)
+	var totalWritten int64
+	for time.Now().Before(deadline) {
+		remaining := time.Until(deadline)
+		written, dur, pumpErr := streamPatternForDuration(conn, remaining, *realStressBulkChunkSize)
+		totalWritten += written
+		if pumpErr == nil {
+			break // completed full duration cleanly
+		}
+		// Connection died (likely transport reconnect). Log and retry.
+		t.Logf("bulk %s/%s: reconnect after written=%d dur=%s: %v",
+			carrierName, transportName, written, dur, pumpErr)
+		if time.Now().After(deadline) {
+			break
+		}
+		// Wait briefly for transport to re-establish, then reconnect.
+		select {
+		case <-ctx.Done():
+			return totalWritten, fmt.Errorf("bulk phase cancelled: %w", ctx.Err())
+		case <-time.After(5 * time.Second):
+		}
+		var cerr error
+		conn, cerr = getConn()
+		if cerr != nil {
+			return totalWritten, fmt.Errorf("bulk reconnect: %w", cerr)
+		}
+	}
+	return totalWritten, nil
+}
+
+// runEchoPhase runs the sustained echo phase for realStressDuration on a fresh
+// connection (the bulk conn may have died during a reconnect at the end of the
+// bulk phase).
+func runEchoPhase(
+	ctx context.Context, t *testing.T, rt *tunnelRuntime,
+	carrierName, transportName, echoAddr string,
+) error {
+	t.Helper()
+	d := *realStressDuration
+	if d <= 0 {
+		return nil
+	}
+
+	echoConn, err := connectViaSOCKSWithin(ctx, rt.socksAddr, echoAddr, 45*time.Second)
+	if err != nil {
+		return fmt.Errorf("sustained echo connect: %w", err)
+	}
+	defer func() { _ = echoConn.Close() }()
+	stats, err := sustainedEcho(echoConn, *realStressEchoSize, d, transportName)
+	if err != nil {
+		return fmt.Errorf("sustained echo: %w", err)
+	}
+	t.Logf("echo  %s/%s: %d rt in %s, p50=%s p95=%s p99=%s max=%s lost=%d",
+		carrierName, transportName, stats.count, d,
+		stats.p50, stats.p95, stats.p99, stats.maxLatency, stats.lost)
+	if stats.count == 0 {
+		return fmt.Errorf("%w: %s", errStressNoRoundtrips, d)
+	}
+	return nil
+}
+
 // streamPatternForDuration pumps a deterministic byte pattern through conn
-// for at most `duration` using a synchronous request-response loop: write a
-// chunk, wait until the same chunk echoes back and verify, then write the
-// next one. Returns total bytes successfully echoed and elapsed time.
+// for at most `duration` using concurrent write and read goroutines so
+// the control stream (ping/pong) is not head-of-line blocked behind bulk
+// data. Returns total bytes successfully echoed and elapsed time.
 //
-// Why request-response rather than concurrent write+read streams:
-// transport throughputs differ by ~3 orders of magnitude (datachannel does
-// MiB/s; videochannel/seichannel ~25 KB/s through 256-byte qr-encoded
-// frames at 25 FPS). An asynchronous writer outruns a slow transport,
-// fills muxconn / SOCKS / RTP-track buffers, and the deadlocked pipe
-// eventually trips a TCP-write deadline - which is not a real bug, just
-// the natural consequence of pumping into a slow pipe with no flow
-// control. Request-response naturally rate-limits to the transport's
-// actual round-trip throughput, which is what we want to measure.
-func streamPatternForDuration(conn net.Conn, duration time.Duration, chunkSize int, transportName string) (int64, time.Duration, error) {
+// Earlier versions used synchronous request-response, but that blocked
+// the smux control stream behind bulk KCP frames and caused spurious
+// liveness timeouts on vp8channel (QR-encoded frames are slow). The
+// concurrent approach measures true transport throughput without breaking
+// liveness.
+func streamPatternForDuration(conn net.Conn, duration time.Duration, chunkSize int) (int64, time.Duration, error) {
 	if chunkSize <= 0 {
 		chunkSize = 4096
 	}
-	// Per-chunk roundtrip deadline. Slow transports (videochannel, vp8channel)
-	// can take seconds+ per chunk in practice. Default 15s is enough for
-	// datachannel, but video-paced transports need more slack due to pacing
-	// and KCP queuing (issue #95).
-	chunkTimeout := 15 * time.Second
-	if transportName == "videochannel" || transportName == "seichannel" || transportName == "vp8channel" {
-		chunkTimeout = 60 * time.Second // Generous timeout for video-paced transports
-	}
 
 	start := time.Now()
 	deadline := start.Add(duration)
 
-	buf := make([]byte, chunkSize)
-	echoed := make([]byte, chunkSize)
-	want := make([]byte, chunkSize)
-
-	reader := bufio.NewReader(conn)
-	var total int64
-
-	for time.Now().Before(deadline) {
-		fillPattern(buf, total)
-		if err := conn.SetWriteDeadline(time.Now().Add(chunkTimeout)); err != nil {
-			return total, time.Since(start), fmt.Errorf("set write deadline at %d: %w", total, err)
+	var (
+		sent    atomic.Int64
+		errOnce sync.Once
+		pumpErr error
+	)
+	recordErr := func(err error) {
+		if err == nil {
+			return
 		}
-		if _, err := conn.Write(buf); err != nil {
-			return total, time.Since(start), fmt.Errorf("write at %d: %w", total, err)
-		}
-		if err := conn.SetReadDeadline(time.Now().Add(chunkTimeout)); err != nil {
-			return total, time.Since(start), fmt.Errorf("set read deadline at %d: %w", total, err)
-		}
-		if _, err := io.ReadFull(reader, echoed); err != nil {
-			return total, time.Since(start), fmt.Errorf("read at %d: %w", total, err)
-		}
-		fillPattern(want, total)
-		if !bytes.Equal(echoed, want) {
-			return total, time.Since(start), fmt.Errorf("%w %d", errPayloadMismatchOffset, total)
-		}
-		total += int64(chunkSize)
+		errOnce.Do(func() { pumpErr = err })
 	}
-	return total, time.Since(start), nil
+
+	// Writer: pump deterministic pattern chunks until deadline.
+	// No backpressure — we measure raw send throughput, not round-trip.
+	// The TCP write buffer + smux + KCP provide their own flow control;
+	// adding an explicit maxInFlight here throttles bulk to RTT-limited
+	// speed (~0.003 MiB/s at 1.25s RTT through Telemost).
+	const writeDeadline = 30 * time.Second
+	writerDone := make(chan struct{})
+	go func() {
+		defer close(writerDone)
+		buf := make([]byte, chunkSize)
+		for time.Now().Before(deadline) {
+			off := sent.Load()
+			fillPattern(buf, off)
+			if err := conn.SetWriteDeadline(time.Now().Add(writeDeadline)); err != nil {
+				recordErr(fmt.Errorf("set write deadline: %w", err))
+				return
+			}
+			if _, err := conn.Write(buf); err != nil {
+				recordErr(fmt.Errorf("write: %w", err))
+				return
+			}
+			sent.Add(int64(chunkSize))
+		}
+	}()
+
+	// Drain incoming echo data to prevent server-side smux window from
+	// filling up and blocking writes. We don't verify pattern here —
+	// that's the sustained echo phase's job. We just discard bytes.
+	const readDeadline = 5 * time.Second
+	readerDone := make(chan struct{})
+	go func() {
+		defer close(readerDone)
+		discardBuf := make([]byte, 32*1024)
+		for {
+			if err := conn.SetReadDeadline(time.Now().Add(readDeadline)); err != nil {
+				return
+			}
+			_, err := conn.Read(discardBuf)
+			if err != nil {
+				return // deadline or closed — writer will catch fatal errors
+			}
+		}
+	}()
+
+	<-writerDone
+	_ = conn.SetDeadline(time.Unix(1, 0)) // unblock reader
+	<-readerDone
+
+	return sent.Load(), time.Since(start), pumpErr
 }
 
 type echoStats struct {

internal/e2e/tunnel_test.go

@@ -21,6 +21,7 @@ import (
 
 	"github.com/openlibrecommunity/olcrtc/internal/app/session"
 	"github.com/openlibrecommunity/olcrtc/internal/client"
+	"github.com/openlibrecommunity/olcrtc/internal/control"
 	"github.com/openlibrecommunity/olcrtc/internal/engine"
 	enginebuiltin "github.com/openlibrecommunity/olcrtc/internal/engine/builtin"
 	"github.com/openlibrecommunity/olcrtc/internal/server"
@@ -351,9 +352,8 @@ func (s *memoryStream) SetEndedCallback(cb func(string)) {
 func (s *memoryStream) WatchConnection(ctx context.Context) {
 	<-ctx.Done()
 }
-func (s *memoryStream) CanSend() bool {
-	return s.isConnected()
-}
+func (s *memoryStream) CanSend() bool           { return s.isConnected() }
+func (s *memoryStream) SubscriberCanSend() bool { return s.isConnected() }
 func (s *memoryStream) GetSendQueue() chan []byte { return nil }
 func (s *memoryStream) GetBufferedAmount() uint64 { return 0 }
 func (s *memoryStream) Reconnect(string)          {}
@@ -1097,6 +1097,7 @@ func startRealTunnel(
 			KeyHex:           testKeyHex,
 			DNSServer:        localDNSServer,
 			TransportOptions: e2eTransportOptions(transportName),
+			Liveness:         control.Config{Interval: 10 * time.Second, Timeout: 60 * time.Second, Failures: 10},
 		})
 	}()
 
@@ -1126,6 +1127,7 @@ func startRealTunnel(
 			LocalAddr:        socksAddr,
 			DNSServer:        localDNSServer,
 			TransportOptions: e2eTransportOptions(transportName),
+			Liveness:         control.Config{Interval: 10 * time.Second, Timeout: 60 * time.Second, Failures: 10},
 		}, func() { close(ready) })
 	}()
 
@@ -1156,7 +1158,7 @@ func startRealTunnel(
 		cancel:    cancel,
 		serverErr: serverErr,
 		clientErr: clientErr,
-		stopWait:  20 * time.Second,
+		stopWait:  60 * time.Second,
 	}, nil
 }
 
@@ -1415,7 +1417,7 @@ func runRealE2ECase(t *testing.T, carrierName, transportName, roomURL, echoAddr
 		}
 	}()
 
-	conn, err := connectViaSOCKSWithin(rt.socksAddr, echoAddr, *realE2ETimeout)
+	conn, err := connectViaSOCKSWithin(ctx, rt.socksAddr, echoAddr, *realE2ETimeout)
 	if err != nil {
 		return err
 	}
@@ -1668,19 +1670,21 @@ func eventuallyConnectViaSOCKS(t *testing.T, socksAddr, targetAddr string) net.C
 func eventuallyConnectViaSOCKSWithin(t *testing.T, socksAddr, targetAddr string, timeout time.Duration) net.Conn {
 	t.Helper()
 
-	conn, err := connectViaSOCKSWithin(socksAddr, targetAddr, timeout)
+	conn, err := connectViaSOCKSWithin(context.Background(), socksAddr, targetAddr, timeout)
 	if err != nil {
 		t.Fatal(err)
 	}
 	return conn
 }
 
-func connectViaSOCKSWithin(socksAddr, targetAddr string, timeout time.Duration) (net.Conn, error) {
+func connectViaSOCKSWithin(
+	ctx context.Context, socksAddr, targetAddr string, timeout time.Duration,
+) (net.Conn, error) {
 	deadline := time.Now().Add(timeout)
 	var lastErr error
 	attempt := 0
 	for time.Now().Before(deadline) {
-		conn, err := tryConnectViaSOCKS(socksAddr, targetAddr)
+		conn, err := tryConnectViaSOCKS(ctx, socksAddr, targetAddr)
 		if err == nil {
 			return conn, nil
 		}
@@ -1695,9 +1699,9 @@ func connectViaSOCKSWithin(socksAddr, targetAddr string, timeout time.Duration)
 	return nil, fmt.Errorf("connect via SOCKS failed after %s: %w", timeout, lastErr)
 }
 
-func tryConnectViaSOCKS(socksAddr, targetAddr string) (net.Conn, error) {
+func tryConnectViaSOCKS(ctx context.Context, socksAddr, targetAddr string) (net.Conn, error) {
 	dialer := net.Dialer{Timeout: 500 * time.Millisecond}
-	conn, err := dialer.DialContext(context.Background(), "tcp4", socksAddr)
+	conn, err := dialer.DialContext(ctx, "tcp4", socksAddr)
 	if err != nil {
 		return nil, fmt.Errorf("dial socks: %w", err)
 	}

internal/engine/engine.go

@@ -78,6 +78,9 @@ type Session interface {
 	SetEndedCallback(cb func(string))
 	WatchConnection(ctx context.Context)
 	CanSend() bool
+	// SubscriberCanSend reports whether the subscriber PC is connected.
+	// Unlike CanSend, it does not require the publisher PC to be ready.
+	SubscriberCanSend() bool
 	GetSendQueue() chan []byte
 	GetBufferedAmount() uint64
 	Capabilities() Capabilities

internal/engine/goolom/lifecycle.go

@@ -3,6 +3,7 @@ package goolom
 import (
 	"context"
 	"fmt"
+	"net"
 	"time"
 
 	"github.com/google/uuid"
@@ -10,6 +11,7 @@ import (
 	"github.com/openlibrecommunity/olcrtc/internal/engine"
 	"github.com/openlibrecommunity/olcrtc/internal/logger"
 	"github.com/openlibrecommunity/olcrtc/internal/protect"
+	"github.com/pion/interceptor"
 	"github.com/pion/webrtc/v4"
 )
 
@@ -75,29 +77,17 @@ func (s *Session) waitForMediaReady(ctx context.Context, timeout time.Duration)
 }
 
 func (s *Session) setupPeerConnections(config webrtc.Configuration) error {
-	settingEngine := webrtc.SettingEngine{}
-	if protect.Protector != nil {
-		settingEngine.SetICEProxyDialer(protect.NewProxyDialer())
+	api, err := newWebRTCAPI()
+	if err != nil {
+		return err
 	}
-	settingEngine.LoggerFactory = logger.NewPionLoggerFactory()
-	api := webrtc.NewAPI(webrtc.WithSettingEngine(settingEngine))
 
-	var err error
 	s.pcSub, err = api.NewPeerConnection(config)
 	if err != nil {
 		return fmt.Errorf("new sub pc: %w", err)
 	}
 	s.pcSub.OnConnectionStateChange(s.onSubscriberConnectionStateChange)
-	s.pcSub.OnTrack(func(track *webrtc.TrackRemote, receiver *webrtc.RTPReceiver) {
-		if track.Kind() != webrtc.RTPCodecTypeVideo {
-			return
-		}
-		logger.Infof("goolom remote video track: codec=%s stream=%s track=%s",
-			track.Codec().MimeType, track.StreamID(), track.ID())
-		if cb := s.videoTrackHandler(); cb != nil {
-			cb(track, receiver)
-		}
-	})
+	s.pcSub.OnTrack(s.onSubscriberTrack)
 
 	s.pcPub, err = api.NewPeerConnection(config)
 	if err != nil {
@@ -111,6 +101,70 @@ func (s *Session) setupPeerConnections(config webrtc.Configuration) error {
 	return nil
 }
 
+// newWebRTCAPI builds a pion API with IPv4-only ICE and the default media
+// engine + interceptors.
+func newWebRTCAPI() (*webrtc.API, error) {
+	settingEngine := webrtc.SettingEngine{}
+	if protect.Protector != nil {
+		settingEngine.SetICEProxyDialer(protect.NewProxyDialer())
+	}
+	settingEngine.LoggerFactory = logger.NewPionLoggerFactory()
+
+	// Restrict ICE to UDP/IPv4. On hosts with many veth/docker interfaces the
+	// agent otherwise enumerates dozens of link-local IPv6 candidates that can
+	// never reach the SFU ("sendto: network is unreachable"). The flood of dead
+	// pairs starves ICE consent-freshness checks on the working pair, so the
+	// SFU stops receiving consent and tears down media after ~30-40 s. Limiting
+	// to IPv4 keeps the candidate set small and consent alive for the session.
+	settingEngine.SetNetworkTypes([]webrtc.NetworkType{webrtc.NetworkTypeUDP4})
+	settingEngine.SetIPFilter(func(ip net.IP) bool {
+		return ip.To4() != nil
+	})
+
+	// Register the default media engine + interceptors. Without the default
+	// interceptors pion never emits RTCP Receiver Reports (or NACK/TWCC) for
+	// the inbound tracks, so the SFU sees a silent subscriber and stops
+	// forwarding VP8 after ~40 s. Registering them keeps the subscriber path
+	// alive for the lifetime of the PC.
+	mediaEngine := &webrtc.MediaEngine{}
+	if err := mediaEngine.RegisterDefaultCodecs(); err != nil {
+		return nil, fmt.Errorf("register default codecs: %w", err)
+	}
+	interceptorRegistry := &interceptor.Registry{}
+	if err := webrtc.RegisterDefaultInterceptors(mediaEngine, interceptorRegistry); err != nil {
+		return nil, fmt.Errorf("register default interceptors: %w", err)
+	}
+	return webrtc.NewAPI(
+		webrtc.WithSettingEngine(settingEngine),
+		webrtc.WithMediaEngine(mediaEngine),
+		webrtc.WithInterceptorRegistry(interceptorRegistry),
+	), nil
+}
+
+// onSubscriberTrack handles a remote track arriving on the subscriber PC.
+func (s *Session) onSubscriberTrack(track *webrtc.TrackRemote, receiver *webrtc.RTPReceiver) {
+	if track.Kind() != webrtc.RTPCodecTypeVideo {
+		return
+	}
+	logger.Infof("goolom remote video track: codec=%s stream=%s track=%s",
+		track.Codec().MimeType, track.StreamID(), track.ID())
+	if cb := s.videoTrackHandler(); cb != nil {
+		cb(track, receiver)
+	}
+	// Drain inbound RTCP on the receiver so the configured interceptors
+	// (Receiver Report / NACK / TWCC) keep running. Without an active reader
+	// the interceptor chain stalls and the SFU eventually stops forwarding
+	// the track.
+	go func() {
+		rtcpBuf := make([]byte, 1500)
+		for {
+			if _, _, err := receiver.Read(rtcpBuf); err != nil {
+				return
+			}
+		}
+	}()
+}
+
 func (s *Session) dialWebSocket() error {
 	wsDialer := protect.NewWebSocketDialer(wsHandshakeTimeout)
 	ws, resp, err := wsDialer.Dial(s.mediaServerURL, nil)
@@ -179,6 +233,12 @@ func (s *Session) onPublisherConnectionStateChange(state webrtc.PeerConnectionSt
 		webrtc.PeerConnectionStateFailed,
 		webrtc.PeerConnectionStateClosed:
 		s.publisherReady.Store(false)
+		// Publisher failure triggers a full reconnect so the data VP8 track
+		// (carried by the publisher PC) is restored. The subscriber PC will
+		// also be re-established as part of the full reconnect.
+		logger.Warnf("goolom publisher PC %s - triggering reconnect", state)
+		s.queueReconnect()
+		return
 	case webrtc.PeerConnectionStateUnknown,
 		webrtc.PeerConnectionStateNew,
 		webrtc.PeerConnectionStateConnecting:
@@ -297,6 +357,7 @@ func (s *Session) retryReconnect(ctx context.Context, backoff time.Duration) boo
 }
 
 func (s *Session) reconnect(ctx context.Context) error {
+	logger.Warnf("goolom: full reconnect triggered")
 	s.reconnecting.Store(true)
 	defer s.reconnecting.Store(false)
 

internal/engine/goolom/session.go

@@ -246,12 +246,20 @@ func (s *Session) SetReconnectCallback(cb func(*webrtc.DataChannel)) { s.onRecon
 // SetShouldReconnect sets the policy for reconnection.
 func (s *Session) SetShouldReconnect(fn func() bool) { s.shouldReconnect = fn }
 
+// SubscriberCanSend reports whether the subscriber PC is connected.
+// Unlike CanSend, it does not require publisherReady, so it returns true
+// as soon as SFU data can arrive — before the publisher PC negotiates.
+func (s *Session) SubscriberCanSend() bool {
+	return !s.closed.Load() && s.subscriberReady.Load()
+}
+
 // CanSend checks if data can be sent.
 func (s *Session) CanSend() bool {
 	if s.onData == nil {
-		if s.hasLocalVideoTracks() {
-			return !s.closed.Load() && s.subscriberReady.Load() && s.publisherReady.Load()
-		}
+		// publisherReady is intentionally not checked: KCP buffers outbound
+		// data and retransmits if WriteSample fails while the publisher PC is
+		// still connecting. Blocking on publisherReady causes handshake welcome
+		// and tunnel stream acks to time out before the publisher PC is up.
 		return !s.closed.Load() && s.subscriberReady.Load()
 	}
 	if s.dc == nil || s.dc.ReadyState() != webrtc.DataChannelStateOpen {
@@ -299,13 +307,32 @@ func (s *Session) attachPendingVideoTracks() error {
 	defer s.videoTrackMu.RUnlock()
 
 	for _, track := range s.videoTracks {
-		if _, err := s.pcPub.AddTrack(track); err != nil {
+		sender, err := s.pcPub.AddTrack(track)
+		if err != nil {
 			return fmt.Errorf("add video track: %w", err)
 		}
+		s.drainPublisherRTCP(sender)
 	}
 	return nil
 }
 
+// drainPublisherRTCP reads (and discards) RTCP feedback the SFU sends for our
+// published track. The read is required so the interceptor chain keeps
+// processing incoming RTCP; without an active reader it stalls.
+func (s *Session) drainPublisherRTCP(sender *webrtc.RTPSender) {
+	if sender == nil {
+		return
+	}
+	go func() {
+		buf := make([]byte, 1500)
+		for {
+			if _, _, err := sender.Read(buf); err != nil {
+				return
+			}
+		}
+	}()
+}
+
 func closeSignal(ch chan struct{}) {
 	if ch == nil {
 		return

internal/engine/goolom/signaling.go

@@ -69,7 +69,6 @@ func (s *Session) handleSignaling(ctx context.Context) {
 			}
 			return
 		}
-
 		s.updateWSDeadline()
 
 		uid, _ := msg[keyUID].(string)
@@ -127,6 +126,12 @@ func (s *Session) handleCommonMessages(msg map[string]any, uid string) {
 	if _, ok := msg["updateDescription"]; ok {
 		s.sendAck(uid)
 	}
+	if _, ok := msg["upsertDescription"]; ok {
+		s.sendAck(uid)
+	}
+	if _, ok := msg["removeDescription"]; ok {
+		s.sendAck(uid)
+	}
 	if _, ok := msg["slotsConfig"]; ok {
 		s.sendAck(uid)
 	}

internal/engine/jitsi/jitsi.go

@@ -1829,6 +1829,9 @@ func (s *Session) CanSend() bool {
 	return s.bridgeReady.Load()
 }
 
+// SubscriberCanSend reports whether the subscriber path is ready to send.
+func (s *Session) SubscriberCanSend() bool { return s.CanSend() }
+
 // GetSendQueue exposes the outbound queue for upstream metrics.
 func (s *Session) GetSendQueue() chan []byte { return s.sendQueue }
 

internal/engine/livekit/livekit.go

@@ -473,6 +473,9 @@ func (s *Session) CanSend() bool {
 // GetSendQueue exposes the outbound queue.
 func (s *Session) GetSendQueue() chan []byte { return s.sendQueue }
 
+// SubscriberCanSend reports whether the subscriber path is ready to send.
+func (s *Session) SubscriberCanSend() bool { return s.CanSend() }
+
 // GetBufferedAmount is a stub for LiveKit (the SDK handles its own buffering).
 func (s *Session) GetBufferedAmount() uint64 { return 0 }
 

internal/muxconn/conn.go

@@ -93,9 +93,10 @@ func releaseFrameBuf(bp *[]byte) {
 // buffer to decrypt into, ships it through the channel, and Read returns
 // the buffer to the pool once its caller has consumed all the bytes.
 type Conn struct {
-	ln     transport.Transport
-	send   func([]byte) error
-	cipher *crypto.Cipher
+	ln      transport.Transport
+	send    func([]byte) error
+	canSend func() bool // if nil, uses ln.CanSend
+	cipher  *crypto.Cipher
 
 	in        chan *[]byte
 	closeOnce sync.Once
@@ -121,6 +122,26 @@ func New(ln transport.Transport, cipher *crypto.Cipher) *Conn {
 	}
 }
 
+// NewControl wires a Conn that routes through the transport's isolated
+// control-plane channel (transport.ControlPlane). Returns nil if the
+// transport does not implement ControlPlane.
+func NewControl(ln transport.Transport, cipher *crypto.Cipher) *Conn {
+	cp, ok := ln.(transport.ControlPlane)
+	if !ok {
+		return nil
+	}
+	c := &Conn{
+		ln:      ln,
+		send:    cp.ControlSend,
+		canSend: cp.ControlCanSend,
+		cipher:  cipher,
+		in:      make(chan *[]byte, inboundQueue),
+		closeCh: make(chan struct{}),
+	}
+	cp.SetControlOnData(func(data []byte) { c.Push(data) })
+	return c
+}
+
 // NewPeer wires a Conn whose writes are addressed to a specific transport peer.
 func NewPeer(ln transport.PeerTransport, cipher *crypto.Cipher, peerID string) *Conn {
 	return &Conn{
@@ -146,7 +167,7 @@ func (c *Conn) Push(ciphertext []byte) {
 	pt, err := c.cipher.DecryptInto(*bufPtr, ciphertext)
 	if err != nil {
 		releaseFrameBuf(bufPtr)
-		logger.Debugf("muxconn: decrypt failed, dropping frame: %v", err)
+		logger.Infof("muxconn: decrypt failed len=%d: %v", len(ciphertext), err)
 		return
 	}
 	*bufPtr = pt
@@ -253,7 +274,11 @@ func (c *Conn) Write(p []byte) (int, error) {
 		if c.closed.Load() {
 			return 0, ErrClosed
 		}
-		if c.ln.CanSend() {
+		canSend := c.canSend
+		if canSend == nil {
+			canSend = c.ln.CanSend
+		}
+		if canSend() {
 			break
 		}
 		if attempt < fastSpinAttempts {

internal/runtime/runtime.go

@@ -76,8 +76,36 @@ func SmuxConfig(maxWirePayload int) *smux.Config {
 	}
 	cfg.MaxReceiveBuffer = smuxMaxReceiveBuffer
 	cfg.MaxStreamBuffer = smuxMaxStreamBuffer
+	// Keep-alive interval is deliberately generous: the underlying KCP
+	// transport can go silent for up to ~25s during a goolom publisher-PC
+	// reconnect (SFU renegotiation). A tight timeout would tear down the
+	// smux session while the carrier is rebuildingitself, forcing an
+	// unnecessary second reconnect. 120s gives plenty of headroom while
+	// still catching truly dead links.
 	cfg.KeepAliveInterval = 10 * time.Second
-	cfg.KeepAliveTimeout = 30 * time.Second
+	cfg.KeepAliveTimeout = 120 * time.Second
+	return cfg
+}
+
+// ControlSmuxConfig returns a lean smux config for the isolated control-plane
+// session. The control session carries only tiny ping/pong frames so we use
+// small stream buffers and disable smux keepalives (the olcrtc control.Run
+// ping loop handles liveness itself).
+func ControlSmuxConfig(maxWirePayload int) *smux.Config {
+	cfg := smux.DefaultConfig()
+	cfg.Version = 2
+	cfg.MaxFrameSize = smuxMaxFrameSize
+	if maxWirePayload >= MinSmuxWirePayload {
+		maxFrameSize := maxWirePayload - SmuxWireOverhead
+		if maxFrameSize < cfg.MaxFrameSize {
+			cfg.MaxFrameSize = maxFrameSize
+		}
+	}
+	// Tiny buffers: control frames are at most a few hundred bytes.
+	cfg.MaxReceiveBuffer = 256 * 1024
+	cfg.MaxStreamBuffer = 32 * 1024
+	// Disable smux keepalive - control.Run runs its own ping/pong loop.
+	cfg.KeepAliveDisabled = true
 	return cfg
 }
 

internal/runtime/runtime_test.go

@@ -41,7 +41,7 @@ func TestSmuxConfigDefault(t *testing.T) {
 		t.Fatalf("SmuxConfig(0) buffers = %+v", cfg)
 	}
 	if cfg.KeepAliveDisabled || cfg.KeepAliveInterval != 10*time.Second ||
-		cfg.KeepAliveTimeout != 30*time.Second {
+		cfg.KeepAliveTimeout != 120*time.Second {
 		t.Fatalf("SmuxConfig(0) keepalive = %+v", cfg)
 	}
 }

internal/server/server.go

@@ -59,10 +59,18 @@ type HealthFunc func(control.Status)
 
 // Server handles incoming tunnel connections and proxies their traffic.
 type Server struct {
+	// baseCtx is the long-lived server context established in bringUpLink. It
+	// is propagated to reconnect-time goroutines (acceptHandshake, control
+	// loops) instead of context.Background() so they observe shutdown.
+	baseCtx        context.Context //nolint:containedctx // server-lifetime ctx for reconnect goroutines
 	ln             transport.Transport
 	peerLn         transport.PeerTransport
 	cipher         *crypto.Cipher
 	conn           *muxconn.Conn
+	// controlConn is wired to the transport's isolated control-plane channel
+	// (transport.ControlPlane). When non-nil, the smux control session runs
+	// over it so bulk data writes never block control ping/pong.
+	controlConn    *muxconn.Conn
 	session        *smux.Session
 	controlStrm    *smux.Stream
 	controlStop    context.CancelFunc
@@ -240,6 +248,10 @@ func smuxConfig(maxWirePayload int) *smux.Config {
 	return runtime.SmuxConfig(maxWirePayload)
 }
 
+func controlSmuxConfig(maxWirePayload int) *smux.Config {
+	return runtime.ControlSmuxConfig(maxWirePayload)
+}
+
 func linkMaxPayload(tr transport.Transport) int {
 	return runtime.MaxPayload(tr)
 }
@@ -249,6 +261,7 @@ func (s *Server) bringUpLink(
 	cfg Config,
 	cancel context.CancelFunc,
 ) error {
+	s.baseCtx = ctx
 	ln, err := transport.New(ctx, cfg.Transport, transport.Config{
 		Carrier:    cfg.Carrier,
 		RoomURL:    cfg.RoomURL,
@@ -289,6 +302,13 @@ func (s *Server) bringUpLink(
 	logger.Infof("Connecting transport=%s carrier=%s ...", cfg.Transport, cfg.Carrier)
 	if s.peerLn == nil {
 		s.installSession()
+	} else {
+		// Peer-routing mode: installSession is skipped, but we still need to
+		// wire up the control-plane smux session so that liveness ping/pong
+		// works correctly over the isolated control track. Build the full
+		// control conn + smux session and launch acceptHandshake exactly as
+		// installSession does for the non-peer-routing path.
+		s.installControlSession()
 	}
 
 	if err := ln.Connect(ctx); err != nil {
@@ -312,12 +332,50 @@ func (s *Server) installSession() {
 		logger.Warnf("smux server init failed: %v", err)
 		return
 	}
+	// If the transport has an isolated control plane, build a dedicated
+	// control smux session over it and launch the handshake acceptor.
+	// For transports without a control plane, serveSingle drives the
+	// handshake in its own loop.
+	controlConn := muxconn.NewControl(s.ln, s.cipher)
+	if controlConn != nil {
+		controlSess, cerr := smux.Server(controlConn, controlSmuxConfig(linkMaxPayload(s.ln)))
+		if cerr != nil {
+			logger.Warnf("control smux server init failed: %v", cerr)
+			_ = controlConn.Close()
+			controlConn = nil
+		} else {
+			// Isolated control plane: handshake runs on the control session.
+			go s.acceptHandshake(s.baseCtx, controlSess)
+		}
+	}
 	s.sessMu.Lock()
 	s.conn = conn
+	s.controlConn = controlConn
 	s.session = sess
 	s.sessMu.Unlock()
 }
 
+// installControlSession wires up the isolated control-plane smux session for
+// transports that implement transport.ControlPlane. Used in peer-routing mode
+// where installSession is skipped but liveness ping/pong still needs its own
+// KCP session separate from bulk data.
+func (s *Server) installControlSession() {
+	controlConn := muxconn.NewControl(s.ln, s.cipher)
+	if controlConn == nil {
+		return // transport has no isolated control plane
+	}
+	controlSess, err := smux.Server(controlConn, controlSmuxConfig(linkMaxPayload(s.ln)))
+	if err != nil {
+		logger.Warnf("control smux server init failed (peer-routing): %v", err)
+		_ = controlConn.Close()
+		return
+	}
+	s.sessMu.Lock()
+	s.controlConn = controlConn
+	s.sessMu.Unlock()
+	go s.acceptHandshake(s.baseCtx, controlSess)
+}
+
 func (s *Server) handleReconnect() {
 	s.recordReconnect()
 	logger.Infof("server reconnect reason=carrier - tearing down smux session")
@@ -331,42 +389,95 @@ func (s *Server) reinstallSession(dead *smux.Session) {
 	s.reinstallMu.Lock()
 	defer s.reinstallMu.Unlock()
 
-	// Close the old muxconn immediately so that any in-flight Push calls
+	// Close the old muxconns immediately so that any in-flight Push calls
 	// (from data arriving on a new bridge before this reinstall completes)
 	// are discarded rather than feeding stale frames into the dying smux
-	// session. Without this, a client that reconnects faster than the
-	// server can push new-session smux frames into the old muxconn,
-	// corrupting the old smux session's stream state (manifests as
-	// "frame too large" on the control stream).
+	// session.
 	s.sessMu.RLock()
 	if s.conn != nil {
 		_ = s.conn.Close()
 	}
+	if s.controlConn != nil {
+		_ = s.controlConn.Close()
+	}
 	s.sessMu.RUnlock()
 
 	// Pre-build the replacement so we can swap atomically below.
-	newConn := muxconn.New(s.ln, s.cipher)
-	newSess, err := smux.Server(newConn, smuxConfig(linkMaxPayload(s.ln)))
-	if err != nil {
-		logger.Warnf("smux server init failed: %v", err)
-		_ = newConn.Close()
+	r := s.buildReplacementSession()
+	if r == nil {
 		return
 	}
 
+	if !s.swapSession(dead, r) {
+		return
+	}
+
+	// Launch the handshake acceptor on the control session only when
+	// an isolated control plane exists. Without one, serveSingle drives
+	// the handshake in its own loop (same as before).
+	if r.controlSess != nil {
+		go s.acceptHandshake(s.baseCtx, r.controlSess)
+	}
+}
+
+// replacementSession holds a freshly-built data + (optional) control smux
+// session pair, prior to atomically swapping it into the live server.
+type replacementSession struct {
+	conn        *muxconn.Conn
+	sess        *smux.Session
+	controlConn *muxconn.Conn
+	controlSess *smux.Session
+}
+
+// buildReplacementSession constructs a fresh data + (optional) control smux
+// session over new muxconns. It returns nil when the data session could not be
+// built.
+func (s *Server) buildReplacementSession() *replacementSession {
+	conn := muxconn.New(s.ln, s.cipher)
+	sess, err := smux.Server(conn, smuxConfig(linkMaxPayload(s.ln)))
+	if err != nil {
+		logger.Warnf("smux server init failed: %v", err)
+		_ = conn.Close()
+		return nil
+	}
+
+	r := &replacementSession{conn: conn, sess: sess}
+	r.controlConn = muxconn.NewControl(s.ln, s.cipher)
+	if r.controlConn != nil {
+		r.controlSess, err = smux.Server(r.controlConn, controlSmuxConfig(linkMaxPayload(s.ln)))
+		if err != nil {
+			logger.Warnf("control smux server init failed: %v", err)
+			_ = r.controlConn.Close()
+			r.controlConn = nil
+			r.controlSess = nil
+		}
+	}
+	return r
+}
+
+// swapSession atomically replaces the live session with the pre-built one and
+// tears down the old one. Returns false (discarding the new build) when another
+// reinstall already won the race.
+func (s *Server) swapSession(dead *smux.Session, r *replacementSession) bool {
 	s.sessMu.Lock()
 	if s.session != dead {
 		// Someone else already reinstalled - discard our build.
 		s.sessMu.Unlock()
-		_ = newSess.Close()
-		_ = newConn.Close()
-		return
+		_ = r.sess.Close()
+		_ = r.conn.Close()
+		if r.controlConn != nil {
+			_ = r.controlSess.Close()
+			_ = r.controlConn.Close()
+		}
+		return false
 	}
 	oldSess := s.session
 	oldControl := s.controlStrm
 	oldControlStop := s.controlStop
 	oldSID := s.sessionID
-	s.session = newSess
-	s.conn = newConn
+	s.session = r.sess
+	s.conn = r.conn
+	s.controlConn = r.controlConn
 	s.controlStrm = nil
 	s.controlStop = nil
 	s.sessionID = ""
@@ -386,18 +497,21 @@ func (s *Server) reinstallSession(dead *smux.Session) {
 		s.onClose(oldSID, "reconnect")
 		s.trackPeerClose(oldSID, "reconnect")
 	}
+	return true
 }
 
 func (s *Server) closeSession() {
 	s.sessMu.Lock()
 	sess := s.session
 	conn := s.conn
+	ctrlConn := s.controlConn
 	control := s.controlStrm
 	controlStop := s.controlStop
 	peers := s.peerSessions
 	s.peerSessions = make(map[string]*peerSession)
 	s.session = nil
 	s.conn = nil
+	s.controlConn = nil
 	s.controlStrm = nil
 	s.controlStop = nil
 	oldSID := s.sessionID
@@ -415,6 +529,9 @@ func (s *Server) closeSession() {
 	if conn != nil {
 		_ = conn.Close()
 	}
+	if ctrlConn != nil {
+		_ = ctrlConn.Close()
+	}
 	if oldSID != "" {
 		s.onClose(oldSID, "closed")
 		s.trackPeerClose(oldSID, "closed")
@@ -524,6 +641,8 @@ func (s *Server) onData(data []byte) {
 func (s *Server) onPeerData(peerID string, data []byte) {
 	ps := s.getPeerSession(peerID)
 	if ps == nil {
+		// Not in peer-routing mode: fall back to the single data conn.
+		s.onData(data)
 		return
 	}
 	ps.conn.Push(data)
@@ -546,7 +665,16 @@ func (s *Server) getPeerSession(peerID string) *peerSession {
 		_ = conn.Close()
 		return nil
 	}
-	ps := &peerSession{peerID: peerID, conn: conn, session: sess}
+	// In peer-routing mode the control handshake (acceptHandshake) runs on
+	// the isolated control KCP session and has already established
+	// sessionID/deviceID by the time the first data frame arrives. Copy them
+	// into the peerSession so servePeer can skip the duplicate handshake.
+	sid := s.sessionID
+	did := s.deviceID
+	s.sessMu.Unlock()
+
+	ps := &peerSession{peerID: peerID, conn: conn, session: sess, sessionID: sid, deviceID: did}
+	s.sessMu.Lock()
 	s.peerSessions[peerID] = ps
 	s.sessMu.Unlock()
 
@@ -587,15 +715,17 @@ func (s *Server) serveSingle(ctx context.Context) {
 			}
 		}
 
-		if !s.handshakeReady() {
-			if !s.acceptHandshake(ctx, sess) {
-				continue
-			}
+		ready, stop := s.waitHandshake(ctx, sess)
+		if stop {
+			return
+		}
+		if !ready {
+			continue
 		}
 
 		stream, err := sess.AcceptStream()
 		if err != nil {
-			if s.handleAcceptError(ctx, sess) {
+			if s.handleAcceptError(ctx, sess, err) {
 				return
 			}
 			continue
@@ -609,13 +739,48 @@ func (s *Server) serveSingle(ctx context.Context) {
 	}
 }
 
+// waitHandshake ensures the handshake has completed before data streams are
+// accepted. The first bool (ready) reports whether the caller may proceed to
+// AcceptStream; the second (stop) reports that the serve loop should exit. When
+// the handshake is not yet done it either drives it inline (legacy path) or
+// spin-waits for the control-plane goroutine, returning ready=false so the
+// caller re-loops.
+func (s *Server) waitHandshake(ctx context.Context, sess *smux.Session) (bool, bool) {
+	if s.handshakeReady() {
+		return true, false
+	}
+
+	// When the transport has an isolated control plane, the handshake
+	// goroutine (launched from installSession/reinstallSession) is
+	// responsible for acceptHandshake. We just wait here until it
+	// completes (sessionID != "") before accepting data streams.
+	s.sessMu.RLock()
+	hasControlConn := s.controlConn != nil
+	s.sessMu.RUnlock()
+	if !hasControlConn {
+		// Legacy path: drive handshake in this loop.
+		if !s.acceptHandshake(ctx, sess) {
+			return false, false
+		}
+		return true, false
+	}
+
+	// Control plane path: handshake goroutine is running; spin-wait.
+	select {
+	case <-ctx.Done():
+		return false, true
+	case <-time.After(10 * time.Millisecond):
+	}
+	return false, false
+}
+
 // handleAcceptError handles a failed AcceptStream. Returns true if the server should stop.
-func (s *Server) handleAcceptError(ctx context.Context, sess *smux.Session) bool {
+func (s *Server) handleAcceptError(ctx context.Context, sess *smux.Session, err error) bool {
 	if contextDone(ctx) {
 		return true
 	}
 	hadSession := s.handshakeReady()
-	logger.Debugf("AcceptStream returned error - reinstalling session")
+	logger.Infof("server: AcceptStream(data) error - reinstalling session: %v", err)
 	s.reinstallSession(sess)
 	if hadSession && s.ln != nil {
 		s.ln.Reconnect("liveness")
@@ -662,7 +827,7 @@ func (s *Server) acceptHandshake(ctx context.Context, sess *smux.Session) bool {
 				return false
 			default:
 			}
-			logger.Debugf("AcceptStream(control) returned %v - reinstalling session", err)
+			logger.Infof("server: AcceptStream(control) error - reinstalling session: %v", err)
 			s.resetLinkPeer()
 			s.reinstallSession(sess)
 			return false
@@ -696,9 +861,27 @@ func (s *Server) acceptHandshake(ctx context.Context, sess *smux.Session) bool {
 }
 
 func (s *Server) servePeer(ps *peerSession) {
-	if !s.acceptPeerHandshake(ps) {
-		s.removePeerSession(ps.peerID, "closed")
-		return
+	// In peer-routing mode the handshake runs on the isolated control KCP
+	// session (acceptHandshake). The first data frame may arrive before the
+	// control handshake completes, so we spin-wait here until sessionID is
+	// set. If the context is cancelled first we bail out cleanly.
+	if ps.sessionID == "" {
+		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)
+		}
 	}
 	for {
 		if s.stopping() {
@@ -709,7 +892,7 @@ func (s *Server) servePeer(ps *peerSession) {
 			if s.stopping() {
 				return
 			}
-			logger.Debugf("AcceptStream(peer=%s) returned %v - closing peer session", ps.peerID, err)
+			logger.Infof("server: AcceptStream(peer=%s) error - closing peer session: %v", ps.peerID, err)
 			s.removePeerSession(ps.peerID, "closed")
 			return
 		}
@@ -721,41 +904,6 @@ func (s *Server) servePeer(ps *peerSession) {
 	}
 }
 
-func (s *Server) acceptPeerHandshake(ps *peerSession) bool {
-	const maxStaleRetries = 3
-	for retry := 0; retry <= maxStaleRetries; retry++ {
-		stream, err := ps.session.AcceptStream()
-		if err != nil {
-			if !s.stopping() {
-				logger.Debugf("AcceptStream(control peer=%s) returned %v", ps.peerID, err)
-			}
-			return false
-		}
-		_ = stream.SetDeadline(time.Now().Add(handshake.DefaultTimeout))
-		hello, sid, err := handshake.Server(stream, s.authHook)
-		_ = stream.SetDeadline(time.Time{})
-		if err != nil {
-			_ = stream.Close()
-			if errors.Is(err, framing.ErrFrameTooLarge) && retry < maxStaleRetries {
-				logger.Debugf("handshake failed peer=%s: discarding stale stream (attempt %d): %v", ps.peerID, retry+1, err)
-				continue
-			}
-			logger.Warnf("handshake failed peer=%s: %v", ps.peerID, err)
-			return false
-		}
-		ps.controlStrm = stream
-		ps.deviceID = hello.DeviceID
-		ps.sessionID = sid
-		s.recordSession(sid)
-		s.onOpen(sid, hello.DeviceID, hello.Claims)
-		s.trackPeerOpen(sid, hello.DeviceID)
-		logger.Infof("session %s opened (device=%s peer=%s)", sid, hello.DeviceID, ps.peerID)
-		s.startPeerControlLoop(ps, stream)
-		return true
-	}
-	return false
-}
-
 func (s *Server) resetLinkPeer() {
 	s.sessMu.RLock()
 	ln := s.ln
@@ -826,54 +974,6 @@ func (s *Server) startControlLoop(ctx context.Context, sess *smux.Session, strea
 	}()
 }
 
-func (s *Server) startPeerControlLoop(ps *peerSession, stream *smux.Stream) {
-	controlCtx, stop := context.WithCancel(context.Background())
-	ps.controlStop = stop
-
-	liveness := s.liveness
-	onPong := liveness.OnPong
-	onMissedPong := liveness.OnMissedPong
-	onUnhealthy := liveness.OnUnhealthy
-	liveness.OnPong = func(h control.Health) {
-		s.recordPong(h)
-		logger.Debugf("control alive session=%s peer=%s rtt=%v seq=%d", ps.sessionID, ps.peerID, h.RTT, h.Seq)
-		if onPong != nil {
-			onPong(h)
-		}
-	}
-	liveness.OnMissedPong = func(missed int) {
-		s.recordMissed(missed)
-		logger.Warnf("control missed pong on server: session=%s peer=%s missed_pongs=%d",
-			ps.sessionID, ps.peerID, missed)
-		if onMissedPong != nil {
-			onMissedPong(missed)
-		}
-	}
-	liveness.OnUnhealthy = func(missed int) {
-		s.recordUnhealthy(missed)
-		logger.Warnf("control stream unhealthy on server: session=%s peer=%s missed_pongs=%d",
-			ps.sessionID, ps.peerID, missed)
-		if onUnhealthy != nil {
-			onUnhealthy(missed)
-		}
-	}
-
-	s.wg.Add(1)
-	go func() {
-		defer s.wg.Done()
-		defer func() { _ = stream.Close() }()
-		err := control.Run(controlCtx, stream, liveness)
-		if controlCtx.Err() != nil || s.stopping() {
-			return
-		}
-		if err != nil {
-			logger.Warnf("server control stream ended session=%s peer=%s: %v", ps.sessionID, ps.peerID, err)
-		}
-		s.recordReconnect()
-		s.removePeerSession(ps.peerID, "reconnect")
-	}()
-}
-
 func (s *Server) stopping() bool {
 	select {
 	case <-s.done:

internal/transport/datachannel/transport_test.go

@@ -44,6 +44,7 @@ func (s *stubSession) SetShouldReconnect(fn func() bool)                       {
 func (s *stubSession) SetEndedCallback(cb func(string))                        { s.endedCB = cb }
 func (s *stubSession) WatchConnection(context.Context)                         { s.watched = true }
 func (s *stubSession) CanSend() bool                                           { return s.canSend }
+func (s *stubSession) SubscriberCanSend() bool                                 { return s.canSend }
 func (s *stubSession) GetSendQueue() chan []byte                               { return nil }
 func (s *stubSession) GetBufferedAmount() uint64                               { return 0 }
 func (s *stubSession) Reconnect(string)                                        {}

internal/transport/seichannel/transport_unit_test.go

@@ -41,7 +41,8 @@ func (s *fakeVideoStream) SetReconnectCallback(cb func())    { s.reconnect = cb
 func (s *fakeVideoStream) SetShouldReconnect(fn func() bool) { s.should = fn }
 func (s *fakeVideoStream) SetEndedCallback(cb func(string))  { s.ended = cb }
 func (s *fakeVideoStream) WatchConnection(context.Context)   { s.watched = true }
-func (s *fakeVideoStream) CanSend() bool                     { return s.canSend }
+func (s *fakeVideoStream) CanSend() bool           { return s.canSend }
+func (s *fakeVideoStream) SubscriberCanSend() bool { return s.canSend }
 func (s *fakeVideoStream) AddTrack(webrtc.TrackLocal) error  { s.trackAdded = true; return nil }
 func (s *fakeVideoStream) Reconnect(string)                  {}
 func (s *fakeVideoStream) SetTrackHandler(cb func(*webrtc.TrackRemote, *webrtc.RTPReceiver)) {
@@ -78,6 +79,7 @@ func (s *fakeEngineSession) WatchConnection(ctx context.Context) {
 	s.stream.WatchConnection(ctx)
 }
 func (s *fakeEngineSession) CanSend() bool                            { return s.stream.CanSend() }
+func (s *fakeEngineSession) SubscriberCanSend() bool                   { return s.stream.SubscriberCanSend() }
 func (s *fakeEngineSession) GetSendQueue() chan []byte                { return nil }
 func (s *fakeEngineSession) GetBufferedAmount() uint64                { return 0 }
 func (s *fakeEngineSession) Reconnect(string)                         {}

internal/transport/transport.go

@@ -46,6 +46,23 @@ type Transport interface {
 	Reconnect(reason string)
 }
 
+// ControlPlane is implemented by transports that can route control-plane
+// traffic independently of the bulk data plane. When a transport implements
+// this interface, callers should use ControlSend/ControlOnData for the first
+// smux stream (the olcrtc control/handshake stream) so that it does not
+// compete with bulk data in the same KCP send buffer.
+type ControlPlane interface {
+	// ControlSend sends a raw encrypted frame on the control-plane channel.
+	ControlSend(data []byte) error
+	// SetControlOnData registers the callback invoked for every frame
+	// received on the control-plane channel.
+	SetControlOnData(cb func([]byte))
+	// ControlCanSend reports whether the control-plane is ready to send.
+	// Unlike CanSend, this should return true as soon as the subscriber PC
+	// is connected — it does NOT require the publisher PC to be ready.
+	ControlCanSend() bool
+}
+
 // PeerTransport is implemented by transports whose carrier can identify and
 // address individual remote endpoints.
 type PeerTransport interface {

internal/transport/videochannel/transport_unit_test.go

@@ -33,7 +33,8 @@ func (s *fakeVideoStream) SetReconnectCallback(cb func())    { s.reconnect = cb
 func (s *fakeVideoStream) SetShouldReconnect(fn func() bool) { s.should = fn }
 func (s *fakeVideoStream) SetEndedCallback(cb func(string))  { s.ended = cb }
 func (s *fakeVideoStream) WatchConnection(context.Context)   { s.watched = true }
-func (s *fakeVideoStream) CanSend() bool                     { return s.canSend }
+func (s *fakeVideoStream) CanSend() bool           { return s.canSend }
+func (s *fakeVideoStream) SubscriberCanSend() bool { return s.canSend }
 func (s *fakeVideoStream) AddTrack(webrtc.TrackLocal) error  { s.trackAdded = true; return nil }
 func (s *fakeVideoStream) Reconnect(string)                  {}
 func (s *fakeVideoStream) SetTrackHandler(cb func(*webrtc.TrackRemote, *webrtc.RTPReceiver)) {
@@ -70,6 +71,7 @@ func (s *fakeEngineSession) WatchConnection(ctx context.Context) {
 	s.stream.WatchConnection(ctx)
 }
 func (s *fakeEngineSession) CanSend() bool                            { return s.stream.CanSend() }
+func (s *fakeEngineSession) SubscriberCanSend() bool                   { return s.stream.SubscriberCanSend() }
 func (s *fakeEngineSession) GetSendQueue() chan []byte                { return nil }
 func (s *fakeEngineSession) GetBufferedAmount() uint64                { return 0 }
 func (s *fakeEngineSession) Reconnect(string)                         {}

internal/transport/vp8channel/engine_session.go

@@ -42,7 +42,8 @@ func (v *engineVideoSession) SetEndedCallback(cb func(string))  { v.session.SetE
 func (v *engineVideoSession) WatchConnection(ctx context.Context) {
 	v.session.WatchConnection(ctx)
 }
-func (v *engineVideoSession) CanSend() bool { return v.session.CanSend() }
+func (v *engineVideoSession) CanSend() bool   { return v.session.CanSend() }
+func (v *engineVideoSession) SubscriberCanSend() bool { return v.session.SubscriberCanSend() }
 
 func (v *engineVideoSession) Reconnect(reason string) { v.session.Reconnect(reason) }
 

internal/transport/vp8channel/kcp.go

@@ -27,13 +27,13 @@ const (
 	// clamped. Stay below that with headroom for KCP overhead (24 bytes).
 	kcpMTU = 1400
 
-	// Send/receive window in segments. Previous large values caused control
-	// starvation under bulk load (issue #95). Reduced to minimal values that
-	// still allow throughput while keeping queuing latency low enough for
-	// timely control ping/pong delivery.
-	//
-	// 512 sndwnd × 1400 MTU = ~700KB in flight — roughly half-second at 1.2MB/s
-	// wire cap, low enough to prevent control timeouts even with batching.
+	// Send/receive window in segments, sized to the bandwidth-delay product
+	// of the policed video path. VP8 over QR encoding has very low throughput
+	// (~0.3 MB/s observed). We use a moderate send window to balance:
+	// - Large enough for reasonable throughput (don't underutilize the pipe)
+	// - Small enough that control-plane pongs can get through within liveness timeout
+	// With 512 segments * 1400 bytes = ~716KB in-flight, and ~0.3 MB/s throughput,
+	// data sits in queue for ~2-3 seconds, giving control a chance to pass.
 	kcpSndWnd = 512
 	kcpRcvWnd = 1024
 
@@ -81,7 +81,12 @@ func startKCP(out chan<- []byte, onData func([]byte), epochHdr [epochHdrLen]byte
 	// the wire to ~45 KiB/s. With nc=1 KCP just keeps the BDP-sized window
 	// full and retransmits the few losses; the pacer caps the rate so we
 	// never overdrive the policer into its collapse zone.
-	sess.SetNoDelay(1, 5, 2, 1)
+	// nodelay=1, interval=20ms (slower for QR-encoded VP8), fast resend=2, congestion control OFF (nc=1).
+	// QR-encoded VP8 has very low throughput (~0.3 MB/s), so we use a larger interval
+	// to allow batching and reduce overhead. KCP's own loss-based congestion control
+	// is disabled (nc=1) because the carrier has hard bandwidth limits; the writerLoop
+	// byte pacer handles rate limiting.
+	sess.SetNoDelay(1, 20, 2, 1)
 	sess.SetWindowSize(kcpSndWnd, kcpRcvWnd)
 	sess.SetMtu(kcpMTU)
 	// Upstream marked SetStreamMode deprecated without providing a replacement;
@@ -114,9 +119,6 @@ func (r *kcpRuntime) readLoop(onData func([]byte)) {
 			continue
 		}
 		if size > kcpMaxMessage {
-			// Stream framing is now corrupted - there is no safe way to
-			// resync without a session reset. Bail and let the upper layer
-			// reconnect.
 			return
 		}
 		payload := make([]byte, size)

internal/transport/vp8channel/options.go

@@ -10,10 +10,10 @@ const (
 	defaultFPS       = 30
 	defaultBatchSize = 64
 	// defaultMaxBytesPerSec paces the wire byte-rate just under the Telemost
-	// SFU's measured per-slot policer knee (~1.4 MiB/s). Above it the SFU
-	// drops bursts wholesale, collapsing goodput and starving keepalives;
-	// staying under keeps loss near zero. See TestRealRawVP8Throughput.
-	defaultMaxBytesPerSec = 1_200_000
+	// SFU's measured per-slot policer knee. The original 1.2 MiB/s caused the
+	// SFU to throttle subscriber forwarding after ~42s; 400 KB/s stays well
+	// within the SFU's comfort zone while still giving useful throughput.
+	defaultMaxBytesPerSec = 400_000
 )
 
 // Options tunes the vp8channel transport. Zero values fall back to documented defaults.

internal/transport/vp8channel/transport.go

@@ -37,10 +37,15 @@ const (
 	// to a couple of send windows so KCP's flush never blocks (a blocked
 	// WriteTo would stall KCP's update loop and delay ACKs); the paced writer
 	// keeps it drained so this depth is headroom, not standing latency.
-	outboundQueueSize    = 1536
-	inboundQueueSize     = 4096
-	canSendHighWatermark = 90 // percent
-	keepaliveIdlePeriod  = 100 * time.Millisecond
+	outboundQueueSize        = 1536
+	// controlOutboundQueueSize is the queue for the control-plane KCP.
+	// Control messages are tiny (ping/pong JSON frames), so a small queue
+	// suffices. We keep it separate from bulk data to guarantee forward
+	// progress even when the data outbound queue is saturated.
+	controlOutboundQueueSize = 2048 // sized for ~20s publisher reconnect window at 20ms tick
+	inboundQueueSize         = 4096
+	canSendHighWatermark     = 90 // percent
+	keepaliveIdlePeriod      = 100 * time.Millisecond
 )
 
 var (
@@ -71,12 +76,20 @@ const (
 	epochOff    = 24
 	crcOff      = 28
 	epochHdrLen = 32
+	// controlEpochFlag marks an epoch as belonging to the control-plane
+	// KCP session. The high bit of the epoch uint32 is reserved for this
+	// purpose; data-plane epochs are generated with the high bit clear.
+	controlEpochFlag uint32 = 0x80000000
 )
 
 var kcpBatchMagic = [4]byte{'O', 'L', 'K', 'B'} //nolint:gochecknoglobals // wire marker
 
 // videoSession is the subset of engine.Session + engine.VideoTrackCapable
-// the vp8channel transport relies on.
+// the vp8channel transport relies on. It necessarily mirrors the engine's
+// lifecycle + video contract, so the method count exceeds the default bloat
+// threshold by design.
+//
+//nolint:interfacebloat // mirrors the engine.Session + video contract
 type videoSession interface {
 	Connect(ctx context.Context) error
 	Close() error
@@ -85,6 +98,11 @@ type videoSession interface {
 	SetEndedCallback(cb func(string))
 	WatchConnection(ctx context.Context)
 	CanSend() bool
+	// SubscriberCanSend returns true when the subscriber PC is connected,
+	// even if the publisher PC has not yet completed negotiation. Used by
+	// the control-plane path so that handshake welcome is never blocked
+	// behind publisher negotiation.
+	SubscriberCanSend() bool
 	Reconnect(reason string)
 	AddTrack(track webrtc.TrackLocal) error
 	SetTrackHandler(cb func(*webrtc.TrackRemote, *webrtc.RTPReceiver))
@@ -95,13 +113,21 @@ type streamTransport struct {
 	track         *webrtc.TrackLocalStaticSample
 	onData        func([]byte)
 	onPeerData    func(peerID string, data []byte)
+	// onControlData is called with every reassembled message from the
+	// control-plane KCP session.
+	onControlData func([]byte)
 	outbound      chan []byte
+	// controlOutbound is the dedicated outbound queue for the control KCP.
+	// Frames here are drained with priority before bulk data frames so that
+	// handshake / liveness messages never wait behind large data writes.
+	controlOutbound chan []byte
 	closeCh       chan struct{}
 	writerDone    chan struct{}
 	closed        atomic.Bool
 	writerUp      atomic.Bool
 	writerOnce    sync.Once
 	kcpOnce       sync.Once
+	controlKCPOnce sync.Once
 	frameInterval time.Duration
 	batchSize     int
 	perTickBytes  int
@@ -116,6 +142,10 @@ type streamTransport struct {
 
 	kcp           *kcpRuntime
 	kcpMu         sync.RWMutex
+	// controlKCP is the isolated KCP session for the control plane.
+	controlKCP    *kcpRuntime
+	controlKCPMu  sync.RWMutex
+	controlOnDataMu sync.RWMutex // guards onControlData reads/writes
 	reconnectMu   sync.Mutex
 	reconnectFn   func()
 	peerConfirmed atomic.Bool
@@ -207,20 +237,21 @@ func newStreamTransport(
 	}
 
 	tr := &streamTransport{
-		stream:        stream,
-		track:         track,
-		onData:        cfg.OnData,
-		onPeerData:    cfg.OnPeerData,
-		outbound:      make(chan []byte, outboundQueueSize),
-		closeCh:       make(chan struct{}),
-		writerDone:    make(chan struct{}),
-		frameInterval: time.Second / time.Duration(fps),
-		batchSize:     batchSize,
-		perTickBytes:  perTickBytes,
-		bindingToken:  bindingToken(cfg.RoomURL),
-		localEpoch:    randomEpoch(),
-		peers:         make(map[uint32]*kcpRuntime),
-		peerOut:       make(map[uint32]chan []byte),
+		stream:            stream,
+		track:             track,
+		onData:            cfg.OnData,
+		onPeerData:        cfg.OnPeerData,
+		outbound:          make(chan []byte, outboundQueueSize),
+		controlOutbound:   make(chan []byte, controlOutboundQueueSize),
+		closeCh:           make(chan struct{}),
+		writerDone:        make(chan struct{}),
+		frameInterval:     time.Second / time.Duration(fps),
+		batchSize:         batchSize,
+		perTickBytes:      perTickBytes,
+		bindingToken:      bindingToken(cfg.RoomURL),
+		localEpoch:        randomEpoch(),
+		peers:             make(map[uint32]*kcpRuntime),
+		peerOut:           make(map[uint32]chan []byte),
 	}
 
 	// In single-peer mode, confirm the peer epoch on first successful KCP
@@ -250,7 +281,7 @@ func (p *streamTransport) Connect(ctx context.Context) error {
 		return fmt.Errorf("connect stream: %w", err)
 	}
 
-	// Start KCP eagerly so Send/CanSend work immediately after Connect.
+	// Start data KCP eagerly so Send/CanSend work immediately after Connect.
 	// Without this, the handshake round-trip that runs right after Connect
 	// would deadlock: muxconn.Write spins on CanSend (which checks kcp!=nil)
 	// and KCP was only started lazily on the first incoming peer frame.
@@ -266,6 +297,32 @@ func (p *streamTransport) Connect(ctx context.Context) error {
 		logger.Infof("vp8channel: KCP started localEpoch=0x%08x", p.localEpochValue())
 	})
 
+	// Start control KCP on its own isolated session. Control messages are tiny
+	// (ping/pong JSON) and must never be blocked behind bulk data segments.
+	// We pass a wrapper callback that always reads the current onControlData
+	// field under the mutex, so SetControlOnData can update it without
+	// restarting the KCP session.
+	p.controlKCPOnce.Do(func() {
+		controlCb := func(data []byte) {
+			p.controlOnDataMu.RLock()
+			cb := p.onControlData
+			p.controlOnDataMu.RUnlock()
+			if cb != nil {
+				cb(data)
+			}
+		}
+		chdr := p.controlEpochHeader()
+		rt, err := startKCP(p.controlOutbound, controlCb, chdr)
+		if err != nil {
+			logger.Infof("vp8channel: startControlKCP failed: %v", err)
+			return
+		}
+		p.controlKCPMu.Lock()
+		p.controlKCP = rt
+		p.controlKCPMu.Unlock()
+		logger.Infof("vp8channel: control KCP started epoch=0x%08x", p.controlEpochValue())
+	})
+
 	p.writerOnce.Do(func() {
 		p.writerUp.Store(true)
 		go p.writerLoop()
@@ -283,6 +340,20 @@ func (p *streamTransport) epochHeader() [epochHdrLen]byte {
 	return buildEpochHeader(p.bindingToken, epoch)
 }
 
+// controlEpochValue returns the current control-plane epoch (data epoch | controlEpochFlag).
+func (p *streamTransport) controlEpochValue() uint32 {
+	p.epochMu.RLock()
+	defer p.epochMu.RUnlock()
+	return p.localEpoch | controlEpochFlag
+}
+
+// controlEpochHeader builds the epoch header for the control-plane track.
+// The control epoch has the high bit set so the receiver can distinguish
+// control frames from bulk data frames arriving on the same RTP stream.
+func (p *streamTransport) controlEpochHeader() [epochHdrLen]byte {
+	return buildEpochHeader(p.bindingToken, p.controlEpochValue())
+}
+
 func buildEpochHeader(token, epoch uint32) [epochHdrLen]byte {
 	var hdr [epochHdrLen]byte
 	copy(hdr[:], vp8Keepalive)
@@ -344,9 +415,15 @@ func randomEpoch() uint32 {
 	if _, err := rand.Read(b[:]); err != nil {
 		// rand.Read on Linux essentially never fails; fall back to a
 		// time-derived value rather than panic.
-		return uint32(time.Now().UnixNano()) //nolint:gosec // G115: bounded conversion verified by surrounding logic
+		//nolint:gosec // G115: bounded conversion verified by surrounding logic
+		e := uint32(time.Now().UnixNano()) & ^controlEpochFlag
+		if e == 0 {
+			e = 1
+		}
+		return e
 	}
-	e := binary.BigEndian.Uint32(b[:])
+	// Mask off the high bit: data epochs must not collide with control epochs.
+	e := binary.BigEndian.Uint32(b[:]) & ^controlEpochFlag
 	if e == 0 {
 		e = 1
 	}
@@ -402,6 +479,13 @@ func (p *streamTransport) Close() error {
 			rt.close()
 		}
 
+		p.controlKCPMu.RLock()
+		crt := p.controlKCP
+		p.controlKCPMu.RUnlock()
+		if crt != nil {
+			crt.close()
+		}
+
 		p.peersMu.Lock()
 		for _, prt := range p.peers {
 			prt.close()
@@ -430,6 +514,16 @@ func (p *streamTransport) drainOutbound() {
 	}
 }
 
+func (p *streamTransport) drainControlOutbound() {
+	for {
+		select {
+		case <-p.controlOutbound:
+		default:
+			return
+		}
+	}
+}
+
 // ResetPeer drops queued KCP traffic and starts a fresh KCP state machine while
 // keeping the carrier connection alive. The client/server liveness layer calls
 // this before rebuilding smux so replacement handshakes are not parsed behind
@@ -438,6 +532,7 @@ func (p *streamTransport) ResetPeer() {
 	p.peerConfirmed.Store(false)
 	p.peerEpoch.Store(0)
 	p.restartKCP(p.rotateEpochHeader())
+	p.restartControlKCP()
 }
 
 // Reconnect forwards to the underlying engine session.
@@ -450,7 +545,16 @@ func (p *streamTransport) SetReconnectCallback(cb func()) {
 	p.reconnectFn = cb
 	p.reconnectMu.Unlock()
 	p.stream.SetReconnectCallback(func() {
-		p.resetKCP()
+		// Reset the data KCP with a new epoch. The control KCP is restarted
+		// with the SAME epoch (no rotation): this drains accumulated KCP
+		// retransmit frames that piled up while WriteSample was failing, so
+		// pong/ping delivery resumes quickly after the new publisher PC is
+		// ready. Keeping the control epoch stable means the peer does not
+		// need a new handshake and liveness does not time out.
+		p.peerConfirmed.Store(false)
+		p.peerEpoch.Store(0)
+		p.restartKCP(p.rotateEpochHeader())
+		p.restartControlKCP()
 		if cb != nil {
 			cb()
 		}
@@ -480,6 +584,22 @@ func (p *streamTransport) CanSend() bool {
 		len(p.outbound) < cap(p.outbound)*canSendHighWatermark/100
 }
 
+// ControlCanSend reports whether the control-plane is ready to send.
+// Unlike CanSend, it does not require the publisher PC to be ready —
+// control frames (handshake welcome, ping/pong) must go through even
+// before the publisher negotiation completes.
+func (p *streamTransport) ControlCanSend() bool {
+	if p.closed.Load() {
+		return false
+	}
+	p.controlKCPMu.RLock()
+	hasKCP := p.controlKCP != nil
+	p.controlKCPMu.RUnlock()
+	// Only require subscriber to be ready — the control path does not need
+	// the publisher PC; writerLoop handles WriteSample retries.
+	return hasKCP && p.stream.SubscriberCanSend()
+}
+
 // Features advertises reliable+ordered semantics now that KCP guarantees
 // in-order delivery with retransmits. The upper layer (mux/curl tunnel)
 // can rely on these properties end-to-end.
@@ -492,39 +612,107 @@ func (p *streamTransport) Features() transport.Features {
 	}
 }
 
+// writerState holds the per-loop bookkeeping for writerLoop, extracted so the
+// loop body stays within cognitive-complexity limits.
+type writerState struct {
+	p                   *streamTransport
+	keepaliveEvery      int
+	idleTicks           int
+	forceKeepaliveEvery int
+	ticksSinceKeepalive int
+	// pendingControl holds a control frame that failed WriteSample and must be
+	// retried on the next tick before consuming more frames.
+	pendingControl []byte
+}
+
+func (w *writerState) writeSample(data []byte) bool {
+	return w.p.track.WriteSample(media.Sample{
+		Data:     data,
+		Duration: w.p.frameInterval,
+	}) == nil
+}
+
+// forceKeepalive emits a clean, fully-decodable VP8 keepalive keyframe at a
+// steady cadence even while bulk data is flowing. During a sustained bulk
+// transfer every emitted "frame" is the epoch header plus opaque KCP bytes,
+// which never forms a decodable VP8 keyframe. The SFU asks for a keyframe (PLI)
+// and, receiving none within its decode-timeout (~40 s), stops forwarding the
+// track to subscribers. The periodic bare keyframe keeps the SFU's decoder
+// satisfied.
+func (w *writerState) forceKeepalive() {
+	w.ticksSinceKeepalive++
+	if w.ticksSinceKeepalive >= w.forceKeepaliveEvery {
+		w.ticksSinceKeepalive = 0
+		hdr := w.p.epochHeader()
+		_ = w.writeSample(hdr[:])
+	}
+}
+
+// drainControl flushes all queued control frames. Returns false when a frame
+// failed to send (stored in pendingControl for retry next tick).
+func (w *writerState) drainControl() bool {
+	if w.pendingControl != nil {
+		if !w.writeSample(w.pendingControl) {
+			return false
+		}
+		w.pendingControl = nil
+	}
+	for {
+		select {
+		case frame := <-w.p.controlOutbound:
+			w.idleTicks = 0
+			if !w.writeSample(frame) {
+				w.pendingControl = frame
+				return false
+			}
+		default:
+			return true
+		}
+	}
+}
+
+// drainData sends one batched data frame, or a keepalive when idle.
+func (w *writerState) drainData() {
+	select {
+	case frame := <-w.p.outbound:
+		sample := w.p.batchSample(frame, w.p.perTickBytes)
+		w.idleTicks = 0
+		_ = w.writeSample(sample)
+	default:
+		w.idleTicks++
+		if w.idleTicks >= w.keepaliveEvery {
+			w.idleTicks = 0
+			hdr := w.p.epochHeader()
+			_ = w.writeSample(hdr[:])
+		}
+	}
+}
+
 func (p *streamTransport) writerLoop() {
 	defer close(p.writerDone)
 
 	ticker := time.NewTicker(p.frameInterval)
 	defer ticker.Stop()
 
-	keepaliveEvery := max(int(keepaliveIdlePeriod/p.frameInterval), 1)
-	idleTicks := 0
+	w := &writerState{
+		p:                   p,
+		keepaliveEvery:      max(int(keepaliveIdlePeriod/p.frameInterval), 1),
+		forceKeepaliveEvery: max(int((2*time.Second)/p.frameInterval), 1),
+	}
 
 	for {
 		select {
 		case <-p.closeCh:
 			return
 		case <-ticker.C:
-			var sample []byte
-			select {
-			case frame := <-p.outbound:
-				sample = p.batchSample(frame, p.perTickBytes)
-				idleTicks = 0
-			default:
-				idleTicks++
-				if idleTicks < keepaliveEvery {
-					continue
-				}
-				idleTicks = 0
-				hdr := p.epochHeader()
-				sample = hdr[:]
+			// Priority 0: keep a decodable keyframe flowing for the SFU.
+			w.forceKeepalive()
+			// Priority 1+2: drain all control frames before any bulk data.
+			if !w.drainControl() {
+				continue // a control frame is still failing; retry next tick
 			}
-
-			_ = p.track.WriteSample(media.Sample{
-				Data:     sample,
-				Duration: p.frameInterval,
-			})
+			// Priority 3: drain a batched data frame (or send keepalive).
+			w.drainData()
 		}
 	}
 }
@@ -570,12 +758,6 @@ func appendBatchPacket(dst, packet []byte) []byte {
 	return append(dst, packet...)
 }
 
-func (p *streamTransport) resetKCP() {
-	p.peerConfirmed.Store(false)
-	p.peerEpoch.Store(0)
-	p.restartKCP(p.rotateEpochHeader())
-}
-
 func (p *streamTransport) restartKCP(epochHdr [epochHdrLen]byte) {
 	p.drainOutbound()
 	p.kcpMu.Lock()
@@ -594,6 +776,33 @@ func (p *streamTransport) restartKCP(epochHdr [epochHdrLen]byte) {
 	p.kcpMu.Unlock()
 }
 
+func (p *streamTransport) restartControlKCP() {
+	p.drainControlOutbound()
+	p.controlKCPMu.Lock()
+	old := p.controlKCP
+	p.controlKCP = nil
+	p.controlKCPMu.Unlock()
+	if old != nil {
+		old.close()
+	}
+	controlCb := func(data []byte) {
+		p.controlOnDataMu.RLock()
+		cb := p.onControlData
+		p.controlOnDataMu.RUnlock()
+		if cb != nil {
+			cb(data)
+		}
+	}
+	chdr := p.controlEpochHeader()
+	rt, err := startKCP(p.controlOutbound, controlCb, chdr)
+	if err != nil {
+		return
+	}
+	p.controlKCPMu.Lock()
+	p.controlKCP = rt
+	p.controlKCPMu.Unlock()
+}
+
 func (p *streamTransport) handleRemoteTrack(track *webrtc.TrackRemote, _ *webrtc.RTPReceiver) {
 	if track.Codec().MimeType != webrtc.MimeTypeVP8 {
 		go p.drainTrack(track)
@@ -672,12 +881,16 @@ func (s *vp8FrameState) processRTPPacket(pkt *rtp.Packet) []byte {
 func (p *streamTransport) readVP8Track(track *webrtc.TrackRemote) {
 	var state vp8FrameState
 	buf := make([]byte, rtpBufSize)
+	var rtpCount, frameCount int
 
 	for {
 		n, _, err := track.Read(buf)
 		if err != nil {
+			logger.Infof("vp8channel: readVP8Track closed track=%s rtp=%d frames=%d err=%v",
+				track.ID(), rtpCount, frameCount, err)
 			return
 		}
+		rtpCount++
 
 		pkt := &rtp.Packet{}
 		if pkt.Unmarshal(buf[:n]) != nil {
@@ -688,6 +901,7 @@ func (p *streamTransport) readVP8Track(track *webrtc.TrackRemote) {
 		if frame == nil {
 			continue
 		}
+		frameCount++
 
 		p.handleIncomingFrame(frame)
 	}
@@ -703,13 +917,23 @@ func (p *streamTransport) handleFirstPeer(peerEpoch uint32) {
 func (p *streamTransport) handleIncomingFrame(frame []byte) {
 	frameToken, peerEpoch, ok := parseEpochHeader(frame)
 	if !ok {
+		logger.Debugf("vp8channel: incoming frame bad header len=%d", len(frame))
 		return
 	}
 	if frameToken != p.bindingToken {
+		logger.Debugf("vp8channel: incoming frame token mismatch got=0x%08x want=0x%08x", frameToken, p.bindingToken)
 		return
 	}
 	kcpPayload := frame[epochHdrLen:]
 	if peerEpoch == p.localEpochValue() {
+		return // own loopback
+	}
+
+	// Control-plane frames have the high bit set in the epoch field.
+	// Route them to the isolated control KCP and never mix them with
+	// bulk data traffic.
+	if peerEpoch&controlEpochFlag != 0 {
+		p.handleControlFrame(peerEpoch, kcpPayload)
 		return
 	}
 
@@ -737,6 +961,24 @@ func (p *streamTransport) handleIncomingFrame(frame []byte) {
 	}
 }
 
+// handleControlFrame routes a control-plane VP8 frame to the isolated control
+// KCP runtime. Loopback echoes of our own outbound frames and bare keepalives
+// are discarded.
+func (p *streamTransport) handleControlFrame(peerEpoch uint32, kcpPayload []byte) {
+	if peerEpoch == p.controlEpochValue() {
+		return // discard loopback: the SFU echoes our own outbound frames back
+	}
+	if len(kcpPayload) == 0 {
+		return // control keepalive, nothing to deliver
+	}
+	p.controlKCPMu.RLock()
+	crt := p.controlKCP
+	p.controlKCPMu.RUnlock()
+	if crt != nil {
+		crt.deliver(kcpPayload)
+	}
+}
+
 // handlePeerFrame routes incoming KCP data to a per-peer KCP runtime,
 // creating one on demand. Each peer epoch gets its own independent KCP
 // session so multiple clients can coexist in the same room.
@@ -851,6 +1093,31 @@ func splitKCPPayload(payload []byte, deliver func([]byte)) {
 	}
 }
 
+// ControlSend implements transport.ControlPlane.
+// It sends data through the isolated control-plane KCP session.
+func (p *streamTransport) ControlSend(data []byte) error {
+	if p.closed.Load() {
+		return ErrTransportClosed
+	}
+	p.controlKCPMu.RLock()
+	rt := p.controlKCP
+	p.controlKCPMu.RUnlock()
+	if rt == nil {
+		return ErrTransportClosed
+	}
+	return rt.send(data)
+}
+
+// SetControlOnData implements transport.ControlPlane.
+// The callback is stored and forwarded to the control KCP read loop.
+// Can be called before or after Connect; the running KCP read loop picks
+// it up immediately via the closure registered in controlKCPOnce.
+func (p *streamTransport) SetControlOnData(cb func([]byte)) {
+	p.controlOnDataMu.Lock()
+	p.onControlData = cb
+	p.controlOnDataMu.Unlock()
+}
+
 /*
 ПИДОРАС ИЗ ЯНДЕКСА СУКА ПРЕЖДЕ ЧЕМ ФИКСИТЬ ЭТО ЗАДАЙСЯ ВОПРОСОМ:
 ПОЧЕМУ ТВОЙ СЕРВЕР ПРИНИМАЕТ ПАКЕТЫ С SEQUENCE NUMBER ИЗ БУДУЩЕГО

internal/transport/vp8channel/transport_unit_test.go

@@ -54,7 +54,8 @@ func (s *fakeVideoStream) SetReconnectCallback(cb func())    { s.reconnect = cb
 func (s *fakeVideoStream) SetShouldReconnect(fn func() bool) { s.should = fn }
 func (s *fakeVideoStream) SetEndedCallback(cb func(string))  { s.ended = cb }
 func (s *fakeVideoStream) WatchConnection(context.Context)   { s.watched = true }
-func (s *fakeVideoStream) CanSend() bool                     { return s.canSend }
+func (s *fakeVideoStream) CanSend() bool           { return s.canSend }
+func (s *fakeVideoStream) SubscriberCanSend() bool { return s.canSend }
 func (s *fakeVideoStream) AddTrack(webrtc.TrackLocal) error  { s.trackAdded = true; return nil }
 func (s *fakeVideoStream) Reconnect(string)                  {}
 func (s *fakeVideoStream) SetTrackHandler(cb func(*webrtc.TrackRemote, *webrtc.RTPReceiver)) {
@@ -91,6 +92,7 @@ func (s *fakeEngineSession) WatchConnection(ctx context.Context) {
 	s.stream.WatchConnection(ctx)
 }
 func (s *fakeEngineSession) CanSend() bool                           { return s.stream.CanSend() }
+func (s *fakeEngineSession) SubscriberCanSend() bool                  { return s.stream.SubscriberCanSend() }
 func (s *fakeEngineSession) GetSendQueue() chan []byte               { return nil }
 func (s *fakeEngineSession) GetBufferedAmount() uint64               { return 0 }
 func (s *fakeEngineSession) Reconnect(string)                        {}

pkg/olcrtc/olcrtc_test.go

@@ -39,6 +39,7 @@ func (s *stubSession) GetSendQueue() chan []byte                        { return
 func (s *stubSession) GetBufferedAmount() uint64                        { return 0 }
 func (s *stubSession) Reconnect(_ string)                               {}
 func (s *stubSession) Capabilities() engine.Capabilities               { return engine.Capabilities{ByteStream: true} }
+func (s *stubSession) SubscriberCanSend() bool                          { return s.connected }
 
 // Compile-time check: stubSession must satisfy engine.Session.
 var _ engine.Session = (*stubSession)(nil)