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 {