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