diff --git a/internal/client/client.go b/internal/client/client.go index c845383..6730d7c 100644 --- a/internal/client/client.go +++ b/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) } diff --git a/internal/control/control.go b/internal/control/control.go index 4991bea..0ca207f 100644 --- a/internal/control/control.go +++ b/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 diff --git a/internal/e2e/deadlink_test.go b/internal/e2e/deadlink_test.go new file mode 100644 index 0000000..900da6a --- /dev/null +++ b/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) + } +} diff --git a/internal/e2e/tunnel_test.go b/internal/e2e/tunnel_test.go index 5959676..f47ddaf 100644 --- a/internal/e2e/tunnel_test.go +++ b/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...) diff --git a/internal/runtime/runtime.go b/internal/runtime/runtime.go index cc934d4..4410a40 100644 --- a/internal/runtime/runtime.go +++ b/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 diff --git a/internal/runtime/runtime_test.go b/internal/runtime/runtime_test.go index d9a690e..1367a9f 100644 --- a/internal/runtime/runtime_test.go +++ b/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. diff --git a/internal/server/server.go b/internal/server/server.go index d654fbf..c5d390b 100644 --- a/internal/server/server.go +++ b/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 diff --git a/internal/transport/vp8channel/transport.go b/internal/transport/vp8channel/transport.go index a9b315e..6adf8fd 100644 --- a/internal/transport/vp8channel/transport.go +++ b/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 }