From 660263881db80c6126c2dc73c12ed7e7c922f880 Mon Sep 17 00:00:00 2001 From: neuronori Date: Sun, 28 Jun 2026 22:24:27 +0000 Subject: [PATCH 1/3] fix(vp8channel): detect server restart via fresh peer epoch A restarted server rejoins the SFU with a new vp8channel epoch. The client stayed latched to the old epoch and dropped every frame from the new one, so recovery only happened once the relaxed control-liveness window expired (~70s). Watch for a frame from a different epoch after the latched peer has been silent past a short grace window and fire the carrier reconnect callback, driving a re-handshake against the new epoch in seconds. A live peer keeps the latch fresh via its ~2s keepalives, so the watchdog never trips under load or during SFU renegotiation. --- internal/transport/vp8channel/transport.go | 98 +++++++++++--- .../vp8channel/transport_unit_test.go | 124 ++++++++++++++++++ 2 files changed, 204 insertions(+), 18 deletions(-) diff --git a/internal/transport/vp8channel/transport.go b/internal/transport/vp8channel/transport.go index 808dc3c..7389c49 100644 --- a/internal/transport/vp8channel/transport.go +++ b/internal/transport/vp8channel/transport.go @@ -46,6 +46,12 @@ const ( inboundQueueSize = 4096 canSendHighWatermark = 90 // percent keepaliveIdlePeriod = 100 * time.Millisecond + // defaultPeerRestartGrace is how long the latched peer must be silent + // before a frame from a different epoch is read as a peer restart. The + // server emits a decodable keepalive every ~2s, so a few missed beats is + // a confident "peer is gone and a fresh one took its place" signal while + // staying clear of normal SFU jitter. See issue #105. + defaultPeerRestartGrace = 6 * time.Second ) var ( @@ -159,6 +165,19 @@ type streamTransport struct { localEpoch uint32 peerEpoch atomic.Uint32 + // lastPeerFrameNano stamps the wall-clock time of the most recent frame + // from the latched peer epoch. peerRestarting guards against firing the + // reconnect callback more than once per restart. peerRestartGrace is how + // long the latched peer must be silent before a frame from a different + // epoch is treated as a restarted peer rather than unrelated room noise. + // A restarted server rejoins the SFU with a fresh epoch and broadcasts + // decodable keepalives on it; spotting that lets the client re-handshake + // in seconds instead of waiting out the relaxed control-liveness window + // (~70s). See issue #105. + lastPeerFrameNano atomic.Int64 + peerRestarting atomic.Bool + peerRestartGrace time.Duration + kcp *kcpRuntime kcpMu sync.RWMutex // controlKCP is the isolated KCP session for the control plane. @@ -278,22 +297,23 @@ func newStreamTransport( } tr := &streamTransport{ - 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), - ctrlPeers: make(map[uint32]*peerControlKCP), + 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), + ctrlPeers: make(map[uint32]*peerControlKCP), + peerRestartGrace: defaultPeerRestartGrace, } // In single-peer mode, confirm the peer epoch on first successful KCP @@ -1110,6 +1130,10 @@ func (p *streamTransport) readVP8Track(track *webrtc.TrackRemote) { func (p *streamTransport) handleFirstPeer(peerEpoch uint32) { p.peerEpoch.Store(peerEpoch) p.peerConfirmed.Store(true) + // A fresh peer is latched: arm the restart watchdog and clear any pending + // restart flag so a later silence can re-trigger detection (issue #105). + p.lastPeerFrameNano.Store(time.Now().UnixNano()) + p.peerRestarting.Store(false) // Re-point our data KCP at the server so subsequent uplink frames are // addressed (dst=serverEpoch) instead of broadcast. The SFU forwards // every frame to every participant, so without a dst the server cannot @@ -1175,11 +1199,19 @@ func (p *streamTransport) handleIncomingFrame(frame []byte) { // handleSinglePeerData delivers a data frame in single-peer (client) mode. It // latches the first peer epoch seen and ignores frames from any other epoch. +// When the latched peer has gone silent past peerRestartGrace and a frame from +// a different epoch arrives, that is read as a server restart (the server +// rejoins the SFU with a fresh epoch) and triggers a fast carrier reconnect +// instead of waiting out the relaxed control-liveness window (issue #105). func (p *streamTransport) handleSinglePeerData(src uint32, kcpPayload []byte) { - if !p.peerConfirmed.Load() { + switch { + case !p.peerConfirmed.Load(): p.handleFirstPeer(src) - } else if src != p.peerEpoch.Load() { + case src != p.peerEpoch.Load(): + p.maybePeerRestart(src) return + default: + p.lastPeerFrameNano.Store(time.Now().UnixNano()) } if len(kcpPayload) == 0 { @@ -1193,6 +1225,36 @@ func (p *streamTransport) handleSinglePeerData(src uint32, kcpPayload []byte) { } } +// maybePeerRestart reads a frame from a non-latched epoch as a peer restart +// once the latched peer has been silent longer than peerRestartGrace. A live +// peer keeps the latch fresh by emitting a keepalive every ~2s, so a different +// epoch arriving after a silence gap means the old peer is gone and a fresh one +// (a restarted server) has taken its place. We fire the carrier reconnect +// callback exactly once per restart; the client side then resets its peer latch +// and re-handshakes against the new epoch, recovering in seconds instead of +// waiting out the relaxed control-liveness window (issue #105). +func (p *streamTransport) maybePeerRestart(src uint32) { + if p.peerRestartGrace <= 0 { + return + } + last := p.lastPeerFrameNano.Load() + if last == 0 || time.Since(time.Unix(0, last)) < p.peerRestartGrace { + return + } + if !p.peerRestarting.CompareAndSwap(false, true) { + return // a restart is already being handled + } + logger.Infof("vp8channel: peer restart detected old=0x%08x new=0x%08x - reconnecting", + p.peerEpoch.Load(), src) + + p.reconnectMu.Lock() + cb := p.reconnectFn + p.reconnectMu.Unlock() + if cb != nil { + go cb() + } +} + // handleControlFrame routes a control-plane VP8 frame. In multi-peer mode // (server) each data epoch gets its own per-peer control KCP created on demand. // In single-peer mode (client) the shared singleton control KCP is used. diff --git a/internal/transport/vp8channel/transport_unit_test.go b/internal/transport/vp8channel/transport_unit_test.go index 8032b8e..026de01 100644 --- a/internal/transport/vp8channel/transport_unit_test.go +++ b/internal/transport/vp8channel/transport_unit_test.go @@ -6,6 +6,7 @@ import ( "encoding/binary" "errors" "reflect" + "sync/atomic" "testing" "time" @@ -496,6 +497,129 @@ func TestReorderBufferRestoresOrderAndSurvivesLoss(t *testing.T) { } } +// mkPeerFrame builds a data-plane frame from epoch on the given transport's +// binding token, broadcast (dst=0), carrying payload. +func mkPeerFrame(token, epoch uint32, payload []byte) []byte { + frame := make([]byte, epochHdrLen+len(payload)) + copy(frame, vp8Keepalive) + binary.BigEndian.PutUint32(frame[tokenOff:srcOff], token) + binary.BigEndian.PutUint32(frame[srcOff:dstOff], epoch) + binary.BigEndian.PutUint32(frame[dstOff:crcOff], 0) + binary.BigEndian.PutUint32(frame[crcOff:epochHdrLen], epochCRC(token, epoch, 0)) + copy(frame[epochHdrLen:], payload) + return frame +} + +// TestPeerRestartTriggersReconnectAfterGrace guards issue #105: when the +// latched peer goes silent past peerRestartGrace and a frame from a fresh epoch +// arrives, the transport fires the carrier reconnect callback so the client +// re-handshakes against the restarted server instead of stalling for the full +// control-liveness window. +func TestPeerRestartTriggersReconnectAfterGrace(t *testing.T) { + reconnected := make(chan struct{}, 1) + tr := &streamTransport{ + stream: &fakeVideoStream{canSend: true}, + outbound: make(chan []byte, 16), + closeCh: make(chan struct{}), + writerDone: make(chan struct{}), + bindingToken: bindingToken("client"), + localEpoch: 0x100, + peerRestartGrace: 20 * time.Millisecond, + reconnectFn: func() { + select { + case reconnected <- struct{}{}: + default: + } + }, + } + defer func() { _ = tr.Close() }() + + // Latch the original server epoch. + tr.handleIncomingFrame(mkPeerFrame(tr.bindingToken, 0x200, []byte("hello"))) + if tr.peerEpoch.Load() != 0x200 { + t.Fatalf("peer epoch = 0x%08x, want 0x200", tr.peerEpoch.Load()) + } + + // A different epoch arriving inside the grace window must NOT reconnect. + tr.handleIncomingFrame(mkPeerFrame(tr.bindingToken, 0x300, []byte("early"))) + select { + case <-reconnected: + t.Fatal("reconnect fired inside grace window") + case <-time.After(10 * time.Millisecond): + } + + // After the latched peer has been silent past the grace window, a frame + // from the new epoch is read as a restart and fires the callback. + time.Sleep(25 * time.Millisecond) + tr.handleIncomingFrame(mkPeerFrame(tr.bindingToken, 0x300, []byte("restart"))) + select { + case <-reconnected: + case <-time.After(time.Second): + t.Fatal("reconnect did not fire after grace window") + } + if !tr.peerRestarting.Load() { + t.Fatal("peerRestarting flag not set after restart detection") + } +} + +// TestPeerRestartFiresOnlyOnce ensures repeated frames from the new epoch do +// not fire a reconnect storm before the latch is reset. +func TestPeerRestartFiresOnlyOnce(t *testing.T) { + var fires atomic.Uint32 + tr := &streamTransport{ + stream: &fakeVideoStream{canSend: true}, + outbound: make(chan []byte, 16), + closeCh: make(chan struct{}), + writerDone: make(chan struct{}), + bindingToken: bindingToken("client"), + localEpoch: 0x100, + peerRestartGrace: 10 * time.Millisecond, + reconnectFn: func() { fires.Add(1) }, + } + defer func() { _ = tr.Close() }() + + tr.handleIncomingFrame(mkPeerFrame(tr.bindingToken, 0x200, []byte("hello"))) + time.Sleep(15 * time.Millisecond) + for range 5 { + tr.handleIncomingFrame(mkPeerFrame(tr.bindingToken, 0x300, []byte("restart"))) + } + // Give any goroutines time to run. + time.Sleep(50 * time.Millisecond) + if got := fires.Load(); got != 1 { + t.Fatalf("reconnect fired %d times, want exactly 1", got) + } +} + +// TestLivePeerKeepsLatchFresh confirms a peer that keeps sending frames within +// the grace window never trips the restart watchdog, even if a stray frame from +// another epoch shows up (unrelated room participant). +func TestLivePeerKeepsLatchFresh(t *testing.T) { + var fires atomic.Uint32 + tr := &streamTransport{ + stream: &fakeVideoStream{canSend: true}, + outbound: make(chan []byte, 16), + closeCh: make(chan struct{}), + writerDone: make(chan struct{}), + bindingToken: bindingToken("client"), + localEpoch: 0x100, + peerRestartGrace: 40 * time.Millisecond, + reconnectFn: func() { fires.Add(1) }, + } + defer func() { _ = tr.Close() }() + + tr.handleIncomingFrame(mkPeerFrame(tr.bindingToken, 0x200, nil)) + // Keep the latched peer alive with frequent keepalives while a foreign + // epoch repeatedly shows up. The latch stays fresh, so no restart fires. + for range 8 { + tr.handleIncomingFrame(mkPeerFrame(tr.bindingToken, 0x200, nil)) + tr.handleIncomingFrame(mkPeerFrame(tr.bindingToken, 0x999, []byte("noise"))) + time.Sleep(10 * time.Millisecond) + } + if got := fires.Load(); got != 0 { + t.Fatalf("reconnect fired %d times for a live peer, want 0", got) + } +} + func TestSeqLessWrapAround(t *testing.T) { cases := []struct { a, b uint16 From e511d3204b5bc42154c5ce48ee02a329a19164d3 Mon Sep 17 00:00:00 2001 From: neuronori Date: Sun, 28 Jun 2026 23:41:56 +0000 Subject: [PATCH 2/3] Revert "fix(vp8channel): detect server restart via fresh peer epoch" This reverts commit 660263881db80c6126c2dc73c12ed7e7c922f880. --- internal/transport/vp8channel/transport.go | 98 +++----------- .../vp8channel/transport_unit_test.go | 124 ------------------ 2 files changed, 18 insertions(+), 204 deletions(-) diff --git a/internal/transport/vp8channel/transport.go b/internal/transport/vp8channel/transport.go index 7389c49..808dc3c 100644 --- a/internal/transport/vp8channel/transport.go +++ b/internal/transport/vp8channel/transport.go @@ -46,12 +46,6 @@ const ( inboundQueueSize = 4096 canSendHighWatermark = 90 // percent keepaliveIdlePeriod = 100 * time.Millisecond - // defaultPeerRestartGrace is how long the latched peer must be silent - // before a frame from a different epoch is read as a peer restart. The - // server emits a decodable keepalive every ~2s, so a few missed beats is - // a confident "peer is gone and a fresh one took its place" signal while - // staying clear of normal SFU jitter. See issue #105. - defaultPeerRestartGrace = 6 * time.Second ) var ( @@ -165,19 +159,6 @@ type streamTransport struct { localEpoch uint32 peerEpoch atomic.Uint32 - // lastPeerFrameNano stamps the wall-clock time of the most recent frame - // from the latched peer epoch. peerRestarting guards against firing the - // reconnect callback more than once per restart. peerRestartGrace is how - // long the latched peer must be silent before a frame from a different - // epoch is treated as a restarted peer rather than unrelated room noise. - // A restarted server rejoins the SFU with a fresh epoch and broadcasts - // decodable keepalives on it; spotting that lets the client re-handshake - // in seconds instead of waiting out the relaxed control-liveness window - // (~70s). See issue #105. - lastPeerFrameNano atomic.Int64 - peerRestarting atomic.Bool - peerRestartGrace time.Duration - kcp *kcpRuntime kcpMu sync.RWMutex // controlKCP is the isolated KCP session for the control plane. @@ -297,23 +278,22 @@ func newStreamTransport( } tr := &streamTransport{ - 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), - ctrlPeers: make(map[uint32]*peerControlKCP), - peerRestartGrace: defaultPeerRestartGrace, + 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), + ctrlPeers: make(map[uint32]*peerControlKCP), } // In single-peer mode, confirm the peer epoch on first successful KCP @@ -1130,10 +1110,6 @@ func (p *streamTransport) readVP8Track(track *webrtc.TrackRemote) { func (p *streamTransport) handleFirstPeer(peerEpoch uint32) { p.peerEpoch.Store(peerEpoch) p.peerConfirmed.Store(true) - // A fresh peer is latched: arm the restart watchdog and clear any pending - // restart flag so a later silence can re-trigger detection (issue #105). - p.lastPeerFrameNano.Store(time.Now().UnixNano()) - p.peerRestarting.Store(false) // Re-point our data KCP at the server so subsequent uplink frames are // addressed (dst=serverEpoch) instead of broadcast. The SFU forwards // every frame to every participant, so without a dst the server cannot @@ -1199,19 +1175,11 @@ func (p *streamTransport) handleIncomingFrame(frame []byte) { // handleSinglePeerData delivers a data frame in single-peer (client) mode. It // latches the first peer epoch seen and ignores frames from any other epoch. -// When the latched peer has gone silent past peerRestartGrace and a frame from -// a different epoch arrives, that is read as a server restart (the server -// rejoins the SFU with a fresh epoch) and triggers a fast carrier reconnect -// instead of waiting out the relaxed control-liveness window (issue #105). func (p *streamTransport) handleSinglePeerData(src uint32, kcpPayload []byte) { - switch { - case !p.peerConfirmed.Load(): + if !p.peerConfirmed.Load() { p.handleFirstPeer(src) - case src != p.peerEpoch.Load(): - p.maybePeerRestart(src) + } else if src != p.peerEpoch.Load() { return - default: - p.lastPeerFrameNano.Store(time.Now().UnixNano()) } if len(kcpPayload) == 0 { @@ -1225,36 +1193,6 @@ func (p *streamTransport) handleSinglePeerData(src uint32, kcpPayload []byte) { } } -// maybePeerRestart reads a frame from a non-latched epoch as a peer restart -// once the latched peer has been silent longer than peerRestartGrace. A live -// peer keeps the latch fresh by emitting a keepalive every ~2s, so a different -// epoch arriving after a silence gap means the old peer is gone and a fresh one -// (a restarted server) has taken its place. We fire the carrier reconnect -// callback exactly once per restart; the client side then resets its peer latch -// and re-handshakes against the new epoch, recovering in seconds instead of -// waiting out the relaxed control-liveness window (issue #105). -func (p *streamTransport) maybePeerRestart(src uint32) { - if p.peerRestartGrace <= 0 { - return - } - last := p.lastPeerFrameNano.Load() - if last == 0 || time.Since(time.Unix(0, last)) < p.peerRestartGrace { - return - } - if !p.peerRestarting.CompareAndSwap(false, true) { - return // a restart is already being handled - } - logger.Infof("vp8channel: peer restart detected old=0x%08x new=0x%08x - reconnecting", - p.peerEpoch.Load(), src) - - p.reconnectMu.Lock() - cb := p.reconnectFn - p.reconnectMu.Unlock() - if cb != nil { - go cb() - } -} - // handleControlFrame routes a control-plane VP8 frame. In multi-peer mode // (server) each data epoch gets its own per-peer control KCP created on demand. // In single-peer mode (client) the shared singleton control KCP is used. diff --git a/internal/transport/vp8channel/transport_unit_test.go b/internal/transport/vp8channel/transport_unit_test.go index 026de01..8032b8e 100644 --- a/internal/transport/vp8channel/transport_unit_test.go +++ b/internal/transport/vp8channel/transport_unit_test.go @@ -6,7 +6,6 @@ import ( "encoding/binary" "errors" "reflect" - "sync/atomic" "testing" "time" @@ -497,129 +496,6 @@ func TestReorderBufferRestoresOrderAndSurvivesLoss(t *testing.T) { } } -// mkPeerFrame builds a data-plane frame from epoch on the given transport's -// binding token, broadcast (dst=0), carrying payload. -func mkPeerFrame(token, epoch uint32, payload []byte) []byte { - frame := make([]byte, epochHdrLen+len(payload)) - copy(frame, vp8Keepalive) - binary.BigEndian.PutUint32(frame[tokenOff:srcOff], token) - binary.BigEndian.PutUint32(frame[srcOff:dstOff], epoch) - binary.BigEndian.PutUint32(frame[dstOff:crcOff], 0) - binary.BigEndian.PutUint32(frame[crcOff:epochHdrLen], epochCRC(token, epoch, 0)) - copy(frame[epochHdrLen:], payload) - return frame -} - -// TestPeerRestartTriggersReconnectAfterGrace guards issue #105: when the -// latched peer goes silent past peerRestartGrace and a frame from a fresh epoch -// arrives, the transport fires the carrier reconnect callback so the client -// re-handshakes against the restarted server instead of stalling for the full -// control-liveness window. -func TestPeerRestartTriggersReconnectAfterGrace(t *testing.T) { - reconnected := make(chan struct{}, 1) - tr := &streamTransport{ - stream: &fakeVideoStream{canSend: true}, - outbound: make(chan []byte, 16), - closeCh: make(chan struct{}), - writerDone: make(chan struct{}), - bindingToken: bindingToken("client"), - localEpoch: 0x100, - peerRestartGrace: 20 * time.Millisecond, - reconnectFn: func() { - select { - case reconnected <- struct{}{}: - default: - } - }, - } - defer func() { _ = tr.Close() }() - - // Latch the original server epoch. - tr.handleIncomingFrame(mkPeerFrame(tr.bindingToken, 0x200, []byte("hello"))) - if tr.peerEpoch.Load() != 0x200 { - t.Fatalf("peer epoch = 0x%08x, want 0x200", tr.peerEpoch.Load()) - } - - // A different epoch arriving inside the grace window must NOT reconnect. - tr.handleIncomingFrame(mkPeerFrame(tr.bindingToken, 0x300, []byte("early"))) - select { - case <-reconnected: - t.Fatal("reconnect fired inside grace window") - case <-time.After(10 * time.Millisecond): - } - - // After the latched peer has been silent past the grace window, a frame - // from the new epoch is read as a restart and fires the callback. - time.Sleep(25 * time.Millisecond) - tr.handleIncomingFrame(mkPeerFrame(tr.bindingToken, 0x300, []byte("restart"))) - select { - case <-reconnected: - case <-time.After(time.Second): - t.Fatal("reconnect did not fire after grace window") - } - if !tr.peerRestarting.Load() { - t.Fatal("peerRestarting flag not set after restart detection") - } -} - -// TestPeerRestartFiresOnlyOnce ensures repeated frames from the new epoch do -// not fire a reconnect storm before the latch is reset. -func TestPeerRestartFiresOnlyOnce(t *testing.T) { - var fires atomic.Uint32 - tr := &streamTransport{ - stream: &fakeVideoStream{canSend: true}, - outbound: make(chan []byte, 16), - closeCh: make(chan struct{}), - writerDone: make(chan struct{}), - bindingToken: bindingToken("client"), - localEpoch: 0x100, - peerRestartGrace: 10 * time.Millisecond, - reconnectFn: func() { fires.Add(1) }, - } - defer func() { _ = tr.Close() }() - - tr.handleIncomingFrame(mkPeerFrame(tr.bindingToken, 0x200, []byte("hello"))) - time.Sleep(15 * time.Millisecond) - for range 5 { - tr.handleIncomingFrame(mkPeerFrame(tr.bindingToken, 0x300, []byte("restart"))) - } - // Give any goroutines time to run. - time.Sleep(50 * time.Millisecond) - if got := fires.Load(); got != 1 { - t.Fatalf("reconnect fired %d times, want exactly 1", got) - } -} - -// TestLivePeerKeepsLatchFresh confirms a peer that keeps sending frames within -// the grace window never trips the restart watchdog, even if a stray frame from -// another epoch shows up (unrelated room participant). -func TestLivePeerKeepsLatchFresh(t *testing.T) { - var fires atomic.Uint32 - tr := &streamTransport{ - stream: &fakeVideoStream{canSend: true}, - outbound: make(chan []byte, 16), - closeCh: make(chan struct{}), - writerDone: make(chan struct{}), - bindingToken: bindingToken("client"), - localEpoch: 0x100, - peerRestartGrace: 40 * time.Millisecond, - reconnectFn: func() { fires.Add(1) }, - } - defer func() { _ = tr.Close() }() - - tr.handleIncomingFrame(mkPeerFrame(tr.bindingToken, 0x200, nil)) - // Keep the latched peer alive with frequent keepalives while a foreign - // epoch repeatedly shows up. The latch stays fresh, so no restart fires. - for range 8 { - tr.handleIncomingFrame(mkPeerFrame(tr.bindingToken, 0x200, nil)) - tr.handleIncomingFrame(mkPeerFrame(tr.bindingToken, 0x999, []byte("noise"))) - time.Sleep(10 * time.Millisecond) - } - if got := fires.Load(); got != 0 { - t.Fatalf("reconnect fired %d times for a live peer, want 0", got) - } -} - func TestSeqLessWrapAround(t *testing.T) { cases := []struct { a, b uint16 From f79439e61a3e3dab10e326224e0d0138d73adf55 Mon Sep 17 00:00:00 2001 From: neuronori Date: Sun, 28 Jun 2026 23:58:01 +0000 Subject: [PATCH 3/3] fix(vp8channel): rebuild carrier on peer restart via fresh epoch detect a server restart when a frame from a new epoch arrives after the latched peer has gone silent past a grace window, then drive the full carrier rebuild (stream.Reconnect) instead of a bare re-handshake over the stale carrier. the restarted server rejoins the sfu as a fresh participant, so re-handshaking on the old media path only times out; rebuilding the carrier re-establishes a path the new server answers on and recovers in seconds instead of waiting out the ~70s liveness window --- internal/transport/vp8channel/transport.go | 76 ++++++++++- .../vp8channel/transport_unit_test.go | 120 +++++++++++++++++- 2 files changed, 189 insertions(+), 7 deletions(-) diff --git a/internal/transport/vp8channel/transport.go b/internal/transport/vp8channel/transport.go index 808dc3c..170ef57 100644 --- a/internal/transport/vp8channel/transport.go +++ b/internal/transport/vp8channel/transport.go @@ -46,6 +46,12 @@ const ( inboundQueueSize = 4096 canSendHighWatermark = 90 // percent keepaliveIdlePeriod = 100 * time.Millisecond + // defaultPeerRestartGrace is how long the latched peer must be silent + // before a frame from a different epoch is read as a server restart. The + // server emits a decodable keepalive every ~2s, so a few missed beats is + // a confident "the latched peer is gone and a fresh one took its place" + // signal while staying clear of normal SFU jitter. See issue #105. + defaultPeerRestartGrace = 6 * time.Second ) var ( @@ -159,6 +165,19 @@ type streamTransport struct { localEpoch uint32 peerEpoch atomic.Uint32 + // lastPeerFrameNano stamps the wall-clock time of the most recent frame + // from the latched peer epoch. peerRestarting guards the carrier rebuild + // from firing more than once per restart. peerRestartGrace is how long the + // latched peer must be silent before a frame from a different epoch is read + // as a restarted server rather than unrelated room noise. A restarted + // server rejoins the SFU with a fresh epoch and broadcasts decodable + // keepalives on it; spotting that lets us rebuild the carrier in seconds + // instead of waiting out the relaxed control-liveness window (~70s). + // See issue #105. + lastPeerFrameNano atomic.Int64 + peerRestarting atomic.Bool + peerRestartGrace time.Duration + kcp *kcpRuntime kcpMu sync.RWMutex // controlKCP is the isolated KCP session for the control plane. @@ -291,9 +310,10 @@ func newStreamTransport( perTickBytes: perTickBytes, bindingToken: bindingToken(cfg.RoomURL), localEpoch: randomEpoch(), - peers: make(map[uint32]*kcpRuntime), - peerOut: make(map[uint32]chan []byte), - ctrlPeers: make(map[uint32]*peerControlKCP), + peers: make(map[uint32]*kcpRuntime), + peerOut: make(map[uint32]chan []byte), + ctrlPeers: make(map[uint32]*peerControlKCP), + peerRestartGrace: defaultPeerRestartGrace, } // In single-peer mode, confirm the peer epoch on first successful KCP @@ -1110,6 +1130,10 @@ func (p *streamTransport) readVP8Track(track *webrtc.TrackRemote) { func (p *streamTransport) handleFirstPeer(peerEpoch uint32) { p.peerEpoch.Store(peerEpoch) p.peerConfirmed.Store(true) + // Arm the restart watchdog against this fresh latch and clear any pending + // restart flag so a later silence can re-trigger detection (issue #105). + p.lastPeerFrameNano.Store(time.Now().UnixNano()) + p.peerRestarting.Store(false) // Re-point our data KCP at the server so subsequent uplink frames are // addressed (dst=serverEpoch) instead of broadcast. The SFU forwards // every frame to every participant, so without a dst the server cannot @@ -1174,12 +1198,20 @@ func (p *streamTransport) handleIncomingFrame(frame []byte) { } // handleSinglePeerData delivers a data frame in single-peer (client) mode. It -// latches the first peer epoch seen and ignores frames from any other epoch. +// latches the first peer epoch seen. When the latched peer has gone silent +// past peerRestartGrace and a frame from a different epoch arrives, that is +// read as a server restart (the server rejoins the SFU with a fresh epoch) and +// triggers a full carrier rebuild instead of waiting out the relaxed +// control-liveness window (issue #105). func (p *streamTransport) handleSinglePeerData(src uint32, kcpPayload []byte) { - if !p.peerConfirmed.Load() { + switch { + case !p.peerConfirmed.Load(): p.handleFirstPeer(src) - } else if src != p.peerEpoch.Load() { + case src != p.peerEpoch.Load(): + p.maybePeerRestart(src) return + default: + p.lastPeerFrameNano.Store(time.Now().UnixNano()) } if len(kcpPayload) == 0 { @@ -1193,6 +1225,38 @@ func (p *streamTransport) handleSinglePeerData(src uint32, kcpPayload []byte) { } } +// maybePeerRestart reads a frame from a non-latched epoch as a server restart +// once the latched peer has been silent longer than peerRestartGrace. A live +// peer keeps the latch fresh by emitting a keepalive every ~2s, so a different +// epoch arriving after a silence gap means the old peer is gone and a fresh one +// (a restarted server) has taken its place. +// +// Recovery drives the full carrier rebuild via stream.Reconnect - the same +// path control-liveness loss uses - rather than a bare re-handshake over the +// stale carrier. The restarted server rejoined the SFU as a fresh participant, +// so re-handshaking on the old media path just times out; only a carrier +// rebuild re-establishes a path the new server answers on. The carrier's +// reconnect callback then rotates our epoch, resets the peer latch and drives +// a fresh handshake. Firing this on the epoch change recovers in seconds +// instead of waiting out the relaxed control-liveness window (~70s, issue +// #105). We rebuild exactly once per restart; the flag clears when the next +// peer latches in handleFirstPeer. +func (p *streamTransport) maybePeerRestart(src uint32) { + if p.peerRestartGrace <= 0 { + return + } + last := p.lastPeerFrameNano.Load() + if last == 0 || time.Since(time.Unix(0, last)) < p.peerRestartGrace { + return + } + if !p.peerRestarting.CompareAndSwap(false, true) { + return // a rebuild is already in flight + } + logger.Infof("vp8channel: peer restart detected old=0x%08x new=0x%08x - rebuilding carrier", + p.peerEpoch.Load(), src) + go p.stream.Reconnect("peer restart") +} + // handleControlFrame routes a control-plane VP8 frame. In multi-peer mode // (server) each data epoch gets its own per-peer control KCP created on demand. // In single-peer mode (client) the shared singleton control KCP is used. diff --git a/internal/transport/vp8channel/transport_unit_test.go b/internal/transport/vp8channel/transport_unit_test.go index 8032b8e..740fb26 100644 --- a/internal/transport/vp8channel/transport_unit_test.go +++ b/internal/transport/vp8channel/transport_unit_test.go @@ -6,6 +6,7 @@ import ( "encoding/binary" "errors" "reflect" + "sync/atomic" "testing" "time" @@ -87,6 +88,8 @@ type fakeVideoStream struct { ended func(string) watched bool closed bool + + reconnects atomic.Int32 } func (s *fakeVideoStream) Connect(context.Context) error { return s.connectErr } @@ -101,7 +104,7 @@ func (s *fakeVideoStream) WatchConnection(context.Context) { s.watched = true 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) Reconnect(string) { s.reconnects.Add(1) } func (s *fakeVideoStream) SetTrackHandler(cb func(*webrtc.TrackRemote, *webrtc.RTPReceiver)) { s.trackCB = cb } @@ -441,6 +444,121 @@ func TestHandleIncomingFrameEpochFilteringAndReconnect(t *testing.T) { } } +// mkPeerFrame builds a broadcast data-plane frame (dst=0) from epoch on token, +// carrying payload. +func mkPeerFrame(token, epoch uint32, payload []byte) []byte { + frame := make([]byte, epochHdrLen+len(payload)) + copy(frame, vp8Keepalive) + binary.BigEndian.PutUint32(frame[tokenOff:srcOff], token) + binary.BigEndian.PutUint32(frame[srcOff:dstOff], epoch) + binary.BigEndian.PutUint32(frame[dstOff:crcOff], 0) + binary.BigEndian.PutUint32(frame[crcOff:epochHdrLen], epochCRC(token, epoch, 0)) + copy(frame[epochHdrLen:], payload) + return frame +} + +// TestPeerRestartRebuildsCarrierAfterGrace guards issue #105: when the latched +// peer goes silent past peerRestartGrace and a frame from a fresh epoch +// arrives, the transport rebuilds the carrier (stream.Reconnect) so the client +// re-handshakes against the restarted server instead of stalling for the full +// control-liveness window. +func TestPeerRestartRebuildsCarrierAfterGrace(t *testing.T) { + stream := &fakeVideoStream{canSend: true} + tr := &streamTransport{ + stream: stream, + outbound: make(chan []byte, 16), + closeCh: make(chan struct{}), + writerDone: make(chan struct{}), + bindingToken: bindingToken("client"), + localEpoch: 0x100, + peerRestartGrace: 20 * time.Millisecond, + } + defer func() { _ = tr.Close() }() + + // Latch the original server epoch. + tr.handleIncomingFrame(mkPeerFrame(tr.bindingToken, 0x200, []byte("hello"))) + if tr.peerEpoch.Load() != 0x200 { + t.Fatalf("peer epoch = 0x%08x, want 0x200", tr.peerEpoch.Load()) + } + + // A different epoch inside the grace window must NOT rebuild the carrier. + tr.handleIncomingFrame(mkPeerFrame(tr.bindingToken, 0x300, []byte("early"))) + time.Sleep(10 * time.Millisecond) + if got := stream.reconnects.Load(); got != 0 { + t.Fatalf("carrier rebuilt inside grace window: got %d, want 0", got) + } + + // After the latched peer has been silent past the grace window, a frame + // from the new epoch is read as a restart and rebuilds the carrier. + time.Sleep(15 * time.Millisecond) + tr.handleIncomingFrame(mkPeerFrame(tr.bindingToken, 0x300, []byte("restart"))) + deadline := time.Now().Add(time.Second) + for stream.reconnects.Load() == 0 && time.Now().Before(deadline) { + time.Sleep(5 * time.Millisecond) + } + if got := stream.reconnects.Load(); got != 1 { + t.Fatalf("carrier rebuilds after grace = %d, want 1", got) + } + if !tr.peerRestarting.Load() { + t.Fatal("peerRestarting flag not set after restart detection") + } +} + +// TestPeerRestartRebuildsOnlyOnce ensures repeated frames from the new epoch do +// not trigger a rebuild storm before the latch is reset. +func TestPeerRestartRebuildsOnlyOnce(t *testing.T) { + stream := &fakeVideoStream{canSend: true} + tr := &streamTransport{ + stream: stream, + outbound: make(chan []byte, 16), + closeCh: make(chan struct{}), + writerDone: make(chan struct{}), + bindingToken: bindingToken("client"), + localEpoch: 0x100, + peerRestartGrace: 10 * time.Millisecond, + } + defer func() { _ = tr.Close() }() + + tr.handleIncomingFrame(mkPeerFrame(tr.bindingToken, 0x200, []byte("hello"))) + time.Sleep(15 * time.Millisecond) + for range 5 { + tr.handleIncomingFrame(mkPeerFrame(tr.bindingToken, 0x300, []byte("restart"))) + } + time.Sleep(50 * time.Millisecond) + if got := stream.reconnects.Load(); got != 1 { + t.Fatalf("carrier rebuilt %d times, want exactly 1", got) + } +} + +// TestLivePeerKeepsLatchFresh confirms a peer that keeps sending frames within +// the grace window never trips the restart watchdog, even if a stray frame from +// another epoch shows up (unrelated room participant). +func TestLivePeerKeepsLatchFresh(t *testing.T) { + stream := &fakeVideoStream{canSend: true} + tr := &streamTransport{ + stream: stream, + outbound: make(chan []byte, 16), + closeCh: make(chan struct{}), + writerDone: make(chan struct{}), + bindingToken: bindingToken("client"), + localEpoch: 0x100, + peerRestartGrace: 40 * time.Millisecond, + } + defer func() { _ = tr.Close() }() + + tr.handleIncomingFrame(mkPeerFrame(tr.bindingToken, 0x200, nil)) + // Keep the latched peer alive with frequent keepalives while a foreign + // epoch repeatedly shows up. The latch stays fresh, so no rebuild fires. + for range 8 { + tr.handleIncomingFrame(mkPeerFrame(tr.bindingToken, 0x200, nil)) + tr.handleIncomingFrame(mkPeerFrame(tr.bindingToken, 0x999, []byte("noise"))) + time.Sleep(10 * time.Millisecond) + } + if got := stream.reconnects.Load(); got != 0 { + t.Fatalf("carrier rebuilt %d times for a live peer, want 0", got) + } +} + func seqList(pkts []*rtp.Packet) []uint16 { out := make([]uint16, len(pkts)) for i, p := range pkts {