Merge remote-tracking branch 'origin/master' into fix-handshake

# Conflicts:
#	internal/client/client.go

internal/client/client.go

@@ -300,7 +300,7 @@ func buildSmuxClient(
 	ln transport.Transport,
 	conn, controlConn *muxconn.Conn,
 ) (*smux.Session, *smux.Session, error) {
-	sess, err := smux.Client(conn, smuxConfig(linkMaxPayload(ln)))
+	sess, err := smux.Client(conn, runtime.SmuxConfigFor(ln))
 	if err != nil {
 		return nil, nil, fmt.Errorf("smux client: %w", err)
 	}
@@ -606,6 +606,14 @@ func (c *Client) startControlLoop(
 	c.sessMu.Unlock()
 
 	liveness := cfg.Liveness
+	// Relax the pong timeout only for transports with an isolated control
+	// plane (vp8channel): KCP batching + frame pacing can delay control
+	// packets under load. Conventional carriers (jitsi/datachannel) keep the
+	// conservative default so a dead link is detected promptly. A user-set
+	// timeout larger than the default is left untouched.
+	if runtime.IsControlPlane(c.ln) && liveness.Timeout <= control.DefaultTimeout {
+		liveness.Timeout = runtime.LivenessTimeout(c.ln)
+	}
 	onPong := liveness.OnPong
 	onMissedPong := liveness.OnMissedPong
 	onUnhealthy := liveness.OnUnhealthy
@@ -844,11 +852,11 @@ func (c *Client) sendConnectRequest(stream *smux.Stream, targetAddr string, targ
 	_ = stream.SetWriteDeadline(time.Time{})
 
 	ack := make([]byte, 1)
-	// 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))
+	// ControlPlane transports (vp8channel peer-routing) may take ~30s for the
+	// SFU to complete renegotiation and start forwarding data frames, so they
+	// get a generous deadline. Conventional carriers (jitsi/datachannel) use
+	// the conservative window so a stuck CONNECT fails fast.
+	_ = stream.SetReadDeadline(time.Now().Add(runtime.ConnectAckTimeout(c.ln)))
 	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/control/control.go

@@ -34,15 +34,13 @@ const (
 	// ping byte can be head-of-line blocked behind queued data for several
 	// seconds, which is liveness-OK, not a dead link.
 	//
-	// Increased from 15s to 45s to prevent false-positive disconnects on
-	// vp8channel and other video-paced transports where KCP batching and
-	// frame pacing can delay control packets under load (issue #95).
-	//
-	// 45s is based on empirical stress testing: with vp8channel's pacing at
-	// 60fps and 512-segment KCP window, worst-case head-of-line blocking under
-	// sustained bulk transfer can reach ~25-30s before control packet gets
-	// through. 45s gives sufficient margin for jitter and RTT variation.
-	DefaultTimeout = 45 * time.Second
+	// Conservative default: a conventional carrier (jitsi/datachannel) that
+	// goes silent this long is genuinely dead and should reconnect promptly.
+	// Video-paced transports with isolated control planes (vp8channel) need a
+	// longer window because KCP batching + frame pacing can delay control
+	// packets under load (issue #95); that relaxed window is applied per
+	// transport via runtime.LivenessTimeout, not by widening this default.
+	DefaultTimeout = 15 * time.Second
 	// DefaultFailures is the default number of consecutive missed pongs before
 	// the stream is marked unhealthy.
 	DefaultFailures = 4

internal/e2e/deadlink_test.go

@@ -0,0 +1,120 @@
+package e2e
+
+import (
+	"context"
+	"flag"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/openlibrecommunity/olcrtc/internal/client"
+	"github.com/openlibrecommunity/olcrtc/internal/control"
+	"github.com/openlibrecommunity/olcrtc/internal/server"
+)
+
+// deadLinkEnabled gates TestDeadLinkDetectionWindow. It runs a real
+// in-process client/server tunnel over the memory carrier, then silently
+// black-holes the carrier (link stays "up", frames vanish) and measures how
+// long the client takes to notice the dead link and trigger a reconnect.
+//
+// This is the exact failure mode behind the jitsi/datachannel regression:
+// the peer leaves but the WebRTC PC has not torn down, so the only signal is
+// the control-stream liveness timeout. The bad commit widened that timeout to
+// 120s globally; the fix scopes the long window to ControlPlane transports so
+// datachannel detects a dead link on the conservative ~30s smux keepalive.
+var deadLinkEnabled = flag.Bool( //nolint:gochecknoglobals // package-level state intentional
+	"olcrtc.deadlink",
+	false,
+	"run TestDeadLinkDetectionWindow (measures dead-link detection latency)",
+)
+
+// TestDeadLinkDetectionWindow proves the datachannel dead-link detection
+// window returned to the conservative band after the fix. It fails if the
+// client does not detect the silently-dead carrier and reconnect within the
+// expected window (well under the buggy 120s, comfortably above the
+// conservative 30s keepalive timeout).
+func TestDeadLinkDetectionWindow(t *testing.T) {
+	if !*deadLinkEnabled {
+		t.Skip("dead-link test disabled; pass -olcrtc.deadlink to enable")
+	}
+
+	const (
+		transportName = transportData // datachannel: no isolated control plane
+		// The conservative smux keepalive timeout is 30s (interval 10s). Allow
+		// generous headroom for scheduling + a probe cycle, but stay well below
+		// the buggy 120s so the bad commit fails this bound.
+		detectionBudget = 75 * time.Second
+		setupBudget     = 30 * time.Second
+	)
+
+	carrierName, room := registerMemoryCarrier(t)
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+
+	socksAddr := freeLocalAddr(ctx, t)
+
+	// Count reconnects observed by the client's health hook. A dead-link
+	// detection manifests as a reconnect (control loop ends -> handleReconnect).
+	var reconnects atomic.Uint64
+	detected := make(chan struct{}, 1)
+	onHealth := func(st control.Status) {
+		if st.Reconnects > 0 || st.UnhealthyEvents > 0 {
+			if reconnects.Swap(st.Reconnects) == 0 {
+				select {
+				case detected <- struct{}{}:
+				default:
+				}
+			}
+		}
+	}
+
+	serverErr := make(chan error, 1)
+	go func() {
+		serverErr <- server.Run(ctx, server.Config{
+			Transport: transportName,
+			Carrier:   carrierName,
+			RoomURL:   testRoom,
+			KeyHex:    testKeyHex,
+			DNSServer: localDNSServer,
+		})
+	}()
+	room.waitConnected(t, 1)
+
+	ready := make(chan struct{})
+	clientErr := make(chan error, 1)
+	go func() {
+		clientErr <- client.RunWithReady(ctx, client.Config{
+			Transport: transportName,
+			Carrier:   carrierName,
+			RoomURL:   testRoom,
+			KeyHex:    testKeyHex,
+			DeviceID:  testClientDeviceID,
+			LocalAddr: socksAddr,
+			DNSServer: localDNSServer,
+			OnHealth:  onHealth,
+		}, func() { close(ready) })
+	}()
+	waitForReadyWithin(t, ready, setupBudget)
+
+	// Let the control stream settle into a healthy steady state first.
+	time.Sleep(2 * time.Second)
+
+	// Silently kill the carrier: link stays up, frames are dropped. Only the
+	// control-stream liveness timeout can detect this.
+	t.Logf("[deadlink] enabling blackhole at %s", time.Now().Format("15:04:05"))
+	start := time.Now()
+	room.enableBlackhole()
+
+	select {
+	case <-detected:
+		elapsed := time.Since(start)
+		t.Logf("[deadlink] dead link detected + reconnect in %s", elapsed.Round(time.Millisecond))
+		if elapsed > detectionBudget {
+			t.Fatalf("dead-link detection took %s, want <= %s (regression: liveness window too wide)",
+				elapsed.Round(time.Second), detectionBudget)
+		}
+	case <-time.After(detectionBudget):
+		t.Fatalf("dead link NOT detected within %s (regression: client treats dead carrier as alive)",
+			detectionBudget)
+	}
+}

internal/e2e/tunnel_test.go

@@ -139,6 +139,7 @@ func (r *memoryRoom) connectedCount() int {
 	return count
 }
 
+//nolint:unparam // want is fixed at 1 today but kept for call-site clarity/future multi-peer rooms
 func (r *memoryRoom) waitConnected(t *testing.T, want int) {
 	t.Helper()
 
@@ -184,6 +185,25 @@ func (r *memoryRoom) triggerEnded(reason string) {
 	}
 }
 
+// enableBlackhole flips every stream in the room into the silent-dead-link
+// state: the carrier stays "connected" but no further frame is delivered.
+// Only the control-stream liveness timeout can detect this, so the time from
+// here to the first reconnect attempt measures the dead-link detection window.
+func (r *memoryRoom) enableBlackhole() {
+	r.mu.Lock()
+	streams := make([]*memoryStream, 0, len(r.streams))
+	for stream := range r.streams {
+		streams = append(streams, stream)
+	}
+	r.mu.Unlock()
+
+	for _, stream := range streams {
+		stream.mu.Lock()
+		stream.blackhole = true
+		stream.mu.Unlock()
+	}
+}
+
 // peerOf returns the other stream in a 2-party room or nil if there is
 // no peer (yet). Video loopback relies on a single 1:1 partner so we
 // just pick the first non-self stream we see.
@@ -220,6 +240,13 @@ type memoryStream struct {
 	mu        sync.Mutex
 	connected bool
 	closed    bool
+	// blackhole simulates a silently dead link: the carrier stays
+	// "connected" (Send returns nil, the pipe is not closed) but no frame
+	// is ever delivered to the peer. This is the failure mode a real SFU
+	// produces when a peer leaves but the WebRTC PC has not yet torn down:
+	// only the control-stream liveness timeout can detect it. Used to prove
+	// the dead-link detection window is governed by the liveness timeout.
+	blackhole bool
 	reconnect func()
 	ended     func(string)
 	track     webrtc.TrackLocal
@@ -281,6 +308,12 @@ func (s *memoryStream) Send(data []byte) error {
 		s.mu.Unlock()
 		return io.ErrClosedPipe
 	}
+	// Silently dead link: pretend the send succeeded but deliver nothing.
+	// The carrier stays "up", so only control-stream liveness can detect it.
+	if s.blackhole {
+		s.mu.Unlock()
+		return nil
+	}
 	s.mu.Unlock()
 
 	payload := append([]byte(nil), data...)

internal/runtime/runtime.go

@@ -76,17 +76,66 @@ 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
+	return cfg
+}
+
+// SmuxConfigLong is SmuxConfig with a relaxed keep-alive timeout for
+// transports whose carrier can legitimately go silent for tens of seconds
+// (vp8channel/goolom publisher-PC reconnect + SFU renegotiation). A tight
+// timeout would tear down the smux session while the carrier is rebuilding
+// itself, forcing an unnecessary second reconnect. Only transports that
+// implement transport.ControlPlane use this; conventional carriers
+// (jitsi/datachannel) keep the conservative 30s timeout so a genuinely dead
+// link is detected and reconnected promptly.
+func SmuxConfigLong(maxWirePayload int) *smux.Config {
+	cfg := SmuxConfig(maxWirePayload)
 	cfg.KeepAliveTimeout = 120 * time.Second
 	return cfg
 }
 
+// IsControlPlane reports whether the transport routes control-plane traffic
+// on an isolated channel (transport.ControlPlane). The relaxed liveness/
+// keep-alive windows are scoped to these transports only.
+func IsControlPlane(tr transport.Transport) bool {
+	_, ok := tr.(transport.ControlPlane)
+	return ok
+}
+
+// SmuxConfigFor returns the data-plane smux config appropriate for the
+// transport: relaxed keep-alive for ControlPlane carriers, conservative
+// otherwise.
+func SmuxConfigFor(tr transport.Transport) *smux.Config {
+	maxWirePayload := MaxPayload(tr)
+	if IsControlPlane(tr) {
+		return SmuxConfigLong(maxWirePayload)
+	}
+	return SmuxConfig(maxWirePayload)
+}
+
+// LivenessTimeout returns the control-stream pong timeout for a transport:
+// a relaxed window for ControlPlane transports (KCP batching + frame pacing
+// can delay control packets under load), and the conservative default for
+// conventional carriers so dead links are detected quickly.
+func LivenessTimeout(tr transport.Transport) time.Duration {
+	if IsControlPlane(tr) {
+		return 45 * time.Second
+	}
+	return control.DefaultTimeout
+}
+
+// ConnectAckTimeout returns the tunnel CONNECT ack read deadline for a
+// transport. ControlPlane transports (SFU renegotiation) may take ~30s to
+// start forwarding data frames, so they get a generous window; conventional
+// carriers use the conservative default.
+func ConnectAckTimeout(tr transport.Transport) time.Duration {
+	if IsControlPlane(tr) {
+		return 90 * time.Second
+	}
+	return 15 * time.Second
+}
+
 // 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

internal/runtime/runtime_test.go

@@ -41,11 +41,18 @@ func TestSmuxConfigDefault(t *testing.T) {
 		t.Fatalf("SmuxConfig(0) buffers = %+v", cfg)
 	}
 	if cfg.KeepAliveDisabled || cfg.KeepAliveInterval != 10*time.Second ||
-		cfg.KeepAliveTimeout != 120*time.Second {
+		cfg.KeepAliveTimeout != 30*time.Second {
 		t.Fatalf("SmuxConfig(0) keepalive = %+v", cfg)
 	}
 }
 
+func TestSmuxConfigLong(t *testing.T) {
+	cfg := runtime.SmuxConfigLong(0)
+	if cfg.KeepAliveInterval != 10*time.Second || cfg.KeepAliveTimeout != 120*time.Second {
+		t.Fatalf("SmuxConfigLong(0) keepalive = %+v", cfg)
+	}
+}
+
 func TestSmuxConfigShrinks(t *testing.T) {
 	// 100-byte wire payload minus smux+crypto overhead is far below default
 	// 32768, so MaxFrameSize must shrink.

internal/server/server.go

@@ -949,6 +949,13 @@ func (s *Server) startControlLoop(ctx context.Context, sess *smux.Session, strea
 	s.sessMu.Unlock()
 
 	liveness := s.liveness
+	// Relax the pong timeout only for transports with an isolated control
+	// plane (vp8channel); conventional carriers keep the conservative default
+	// so dead links are detected and reconnected promptly. A user-set timeout
+	// larger than the default is left untouched.
+	if runtime.IsControlPlane(s.ln) && liveness.Timeout <= control.DefaultTimeout {
+		liveness.Timeout = runtime.LivenessTimeout(s.ln)
+	}
 	onPong := liveness.OnPong
 	onMissedPong := liveness.OnMissedPong
 	onUnhealthy := liveness.OnUnhealthy

internal/transport/vp8channel/transport.go

@@ -531,8 +531,11 @@ func (p *streamTransport) drainControlOutbound() {
 func (p *streamTransport) ResetPeer() {
 	p.peerConfirmed.Store(false)
 	p.peerEpoch.Store(0)
+	// Preserve control epoch across reset to avoid breaking ping/pong routing.
+	// Control frames use a separate epoch path and must stay stable.
+	controlHdr := p.controlEpochHeader()
 	p.restartKCP(p.rotateEpochHeader())
-	p.restartControlKCP()
+	p.restartControlKCPWithHeader(controlHdr)
 }
 
 // Reconnect forwards to the underlying engine session.
@@ -553,8 +556,9 @@ func (p *streamTransport) SetReconnectCallback(cb func()) {
 		// need a new handshake and liveness does not time out.
 		p.peerConfirmed.Store(false)
 		p.peerEpoch.Store(0)
+		controlHdr := p.controlEpochHeader() // snapshot BEFORE data epoch rotation
 		p.restartKCP(p.rotateEpochHeader())
-		p.restartControlKCP()
+		p.restartControlKCPWithHeader(controlHdr)
 		if cb != nil {
 			cb()
 		}
@@ -793,7 +797,9 @@ func (p *streamTransport) restartKCP(epochHdr [epochHdrLen]byte) {
 	p.kcpMu.Unlock()
 }
 
-func (p *streamTransport) restartControlKCP() {
+// restartControlKCPWithHeader restarts the control KCP with a specific epoch header,
+// used to preserve the control epoch across carrier reconnects.
+func (p *streamTransport) restartControlKCPWithHeader(hdr [epochHdrLen]byte) {
 	p.drainControlOutbound()
 	p.controlKCPMu.Lock()
 	old := p.controlKCP
@@ -810,8 +816,7 @@ func (p *streamTransport) restartControlKCP() {
 			cb(data)
 		}
 	}
-	chdr := p.controlEpochHeader()
-	rt, err := startKCP(p.controlOutbound, controlCb, chdr)
+	rt, err := startKCP(p.controlOutbound, controlCb, hdr)
 	if err != nil {
 		return
 	}