telegabber/calls/audio/latency_drift_test.go

package audio

import (
	"testing"

	"github.com/pion/rtp"
)

// Deterministic playout-latency tests: drive the xmpp->tg path in simulated
// real time (1 step = 10ms), no ticker or goroutines. Producer pushes a 20ms
// packet every 2 steps; consumer runs processTick each step unless "stalled"
// (models a starved ticker). Each test checks latency drains back to baseline.

// callSim drives producer + consumer in lockstep over discrete 10ms steps.
type callSim struct {
	h       *XmppToTg
	st      *tickState
	payload []byte // one reusable 20ms opus packet; decodes fine repeatedly
	seq     uint16
	step    int // wall-clock step counter (10ms each)
}

func newCallSim(t *testing.T) *callSim {
	t.Helper()
	h, _ := newTestXmppToTg(t)
	enc, err := NewEncoder()
	if err != nil {
		t.Fatalf("NewEncoder: %v", err)
	}
	return &callSim{
		h:       h,
		st:      newTickState(),
		payload: encodeOpusFrame(t, enc),
		seq:     1000,
	}
}

// advance runs `steps` 10ms steps. The producer always pushes on schedule
// (every other step). The consumer runs processTick only when consume is true;
// when false the ticker goroutine is considered stalled for that step.
func (s *callSim) advance(steps int, consume bool) {
	for i := 0; i < steps; i++ {
		if s.step%2 == 0 { // one 20ms packet per 20ms of wall clock
			s.h.playout.Push(&rtp.Packet{
				Header:  rtp.Header{SequenceNumber: s.seq, PayloadType: opusPayloadType},
				Payload: s.payload,
			})
			s.seq++
		}
		if consume {
			s.h.processTick(s.st)
		}
		s.step++
	}
}

// depthMs is the current playout latency: buffered packets * 20ms.
func (s *callSim) depthMs() int { return s.h.playout.Depth() * OpusFrameMs }

// A single stall must not permanently raise latency; the backlog has to drain
// back to baseline.
func TestPlayoutLatencyRecoversAfterStall(t *testing.T) {
	s := newCallSim(t)

	// Warm up to steady state and record the baseline latency.
	s.advance(600, true) // 6s
	baseline := s.depthMs()
	t.Logf("baseline latency after warmup: %dms", baseline)

	// One 600ms stall: the ticker goroutine misses ticks while RTP keeps
	// arriving. ~30 packets (600ms of audio) pile up in the jitter buffer.
	s.advance(60, false)
	afterStall := s.depthMs()
	t.Logf("latency right after 600ms stall: %dms", afterStall)

	// A full minute of healthy steady-state playout to recover.
	s.advance(6000, true) // 60s
	settled := s.depthMs()
	t.Logf("latency after 60s of recovery: %dms", settled)

	// Allow one packet of slop around the baseline; ShedOne should have walked
	// the stall's backlog back down. Pre-fix, settled stayed up near afterStall.
	if settled > baseline+OpusFrameMs {
		t.Errorf("playout latency did not recover: baseline=%dms, settled=%dms "+
			"(stall added ~%dms that never drained)",
			baseline, settled, afterStall-baseline)
	}
}

// Many small stalls must not ratchet latency upward. Pre-fix, delay climbed
// monotonically (the reported 1s -> 8-10s); assert it stays bounded.
func TestPlayoutLatencyRatchetsUnderRepeatedStalls(t *testing.T) {
	s := newCallSim(t)
	s.advance(600, true) // warm up
	baseline := s.depthMs()

	const bound = 500 // ms; a sane jitter buffer should never exceed this

	// 30 cycles of {200ms stall, 20s healthy playout}. In wall-clock terms
	// that's ~10 minutes with a stall every 20s - a light, realistic hiccup
	// rate for a loaded host.
	worst := baseline
	for cycle := 0; cycle < 30; cycle++ {
		s.advance(20, false)  // 200ms stall
		s.advance(2000, true) // 20s recovery
		if d := s.depthMs(); d > worst {
			worst = d
		}
	}
	t.Logf("baseline=%dms, worst latency over run=%dms", baseline, worst)

	if worst > bound {
		t.Errorf("playout latency ratcheted past %dms (reached %dms); "+
			"the trim/shed mechanism is not bounding accumulated stalls",
			bound, worst)
	}
}