From f0de6b441ed0d979bd6becbbd233da146c86f815 Mon Sep 17 00:00:00 2001 From: neuronori Date: Sat, 20 Jun 2026 07:52:18 +0000 Subject: [PATCH 1/3] fix(vp8channel): preserve control epoch across reconnect/reset Control KCP epoch must stay stable across both carrier reconnects and liveness resets to avoid breaking ping/pong routing. Previously: - SetReconnectCallback rotated control epoch (fix: preserve) - ResetPeer also rotated control epoch (fix: preserve) Both now snapshot the control epoch header before data epoch rotation and use restartControlKCPWithHeader() with the preserved epoch. This prevents control frames from using a mismatched epoch after liveness timeout or carrier reconnect, which broke routing and caused reconnect loops. --- internal/transport/vp8channel/transport.go | 36 ++++++++++++++++++++-- 1 file changed, 34 insertions(+), 2 deletions(-) diff --git a/internal/transport/vp8channel/transport.go b/internal/transport/vp8channel/transport.go index efe90b8..09e57bc 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() } @@ -803,6 +807,34 @@ func (p *streamTransport) restartControlKCP() { p.controlKCPMu.Unlock() } +// 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 + 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) + } + } + rt, err := startKCP(p.controlOutbound, controlCb, hdr) + 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) From 384268b83e90b6e17f7bb087eb3e9144259eb61f Mon Sep 17 00:00:00 2001 From: nori Date: Wed, 24 Jun 2026 14:36:41 +0000 Subject: [PATCH 2/3] fix(runtime): scope relaxed liveness/keepalive to control-plane transports The #95 telemost fix widened three global windows (control pong 15s->45s, smux keepalive 30s->120s, tunnel ack 15s->90s). These live in shared packages, so they also slowed dead-link detection on jitsi/datachannel: a genuinely dead session stayed alive up to 120s before reconnecting, which presented as the link going down and not recovering. Restore the conservative defaults globally and apply the relaxed windows only when the transport implements transport.ControlPlane (vp8channel/ goolom), where KCP batching + SFU publisher-PC renegotiation legitimately go silent for ~25-30s. Conventional carriers reconnect promptly again. --- internal/client/client.go | 22 ++++++++---- internal/control/control.go | 16 ++++----- internal/runtime/runtime.go | 61 ++++++++++++++++++++++++++++---- internal/runtime/runtime_test.go | 9 ++++- internal/server/server.go | 7 ++++ 5 files changed, 92 insertions(+), 23 deletions(-) diff --git a/internal/client/client.go b/internal/client/client.go index 45139d2..6b97bd7 100644 --- a/internal/client/client.go +++ b/internal/client/client.go @@ -227,7 +227,7 @@ 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))) + sess, err := smux.Client(c.conn, runtime.SmuxConfigFor(ln)) if err != nil { return fmt.Errorf("smux client: %w", err) } @@ -517,7 +517,7 @@ func (c *Client) tryReopenSession( controlSmuxConn = controlConn } - sess, err := smux.Client(conn, smuxConfig(linkMaxPayload(c.ln))) + sess, err := smux.Client(conn, runtime.SmuxConfigFor(c.ln)) if err != nil { logger.Warnf("smux re-init failed (attempt %d): %v", attempt, err) return false @@ -570,6 +570,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 @@ -808,11 +816,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/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 0ff973f..8c37a51 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -921,6 +921,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 From 13f77b1da73f7c25f22d955237eeb4ac95d791b8 Mon Sep 17 00:00:00 2001 From: neuronori Date: Wed, 24 Jun 2026 14:54:19 +0000 Subject: [PATCH 3/3] test(e2e): prove dead-link detection window via blackhole carrier add a silent dead-link mode to the memory carrier (link stays up, frames vanish) and a flag-gated TestDeadLinkDetectionWindow that measures how long the datachannel client takes to notice a dead link and reconnect. on the conservative liveness window it reconnects in ~58s; with the buggy 120s smux keepalive it fails to detect within 75s. also drop dead restartControlKCP (both callers moved to restartControlKCPWithHeader) --- internal/e2e/deadlink_test.go | 120 +++++++++++++++++++++ internal/e2e/tunnel_test.go | 33 ++++++ internal/transport/vp8channel/transport.go | 27 ----- 3 files changed, 153 insertions(+), 27 deletions(-) create mode 100644 internal/e2e/deadlink_test.go 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/transport/vp8channel/transport.go b/internal/transport/vp8channel/transport.go index 09e57bc..a8097a3 100644 --- a/internal/transport/vp8channel/transport.go +++ b/internal/transport/vp8channel/transport.go @@ -780,33 +780,6 @@ 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() -} - // 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) {