ppn/olcrtc
3
0
Fork 0
mirror of https://github.com/openlibrecommunity/olcrtc.git synced 2026-07-03 14:05:39 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #106 from neuronori/fix/issue-105-peer-restart-reconnect

Browse Source 
fix(vp8channel): detect server restart via fresh peer epoch
This commit is contained in:
zarazaex
2026-06-29 03:58:49 +03:00
committed by GitHub
parent 6c8b6e6be7 f79439e61a
commit 64d3d29980
2 changed files with 189 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/transport/vp8channel/transport.go
+70 -6
View File
@@ -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.
internal/transport/vp8channel/transport_unit_test.go
+119 -1
View File
@@ -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 {