From 886cda08f953c1954dac600daf358b3f1c44883a Mon Sep 17 00:00:00 2001 From: pounceandmiss Date: Wed, 8 Jul 2026 02:50:16 -0700 Subject: [PATCH] calls: bound playout latency and add call telemetry --- calls/audio/latency_drift_test.go | 117 ++++++++++++++++++++++++++++ calls/audio/playout.go | 112 +++++++++++++++++++++++++- calls/audio/playout_test.go | 52 ++++++++++++- calls/audio/tg_to_xmpp.go | 94 +++++++++++++++++++++- calls/audio/xmpp_to_tg.go | 125 +++++++++++++++++++++++++++++- 5 files changed, 491 insertions(+), 9 deletions(-) create mode 100644 calls/audio/latency_drift_test.go diff --git a/calls/audio/latency_drift_test.go b/calls/audio/latency_drift_test.go new file mode 100644 index 0000000..042d300 --- /dev/null +++ b/calls/audio/latency_drift_test.go @@ -0,0 +1,117 @@ +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) + } +} diff --git a/calls/audio/playout.go b/calls/audio/playout.go index 2188b2e..a8a2ce1 100644 --- a/calls/audio/playout.go +++ b/calls/audio/playout.go @@ -8,8 +8,16 @@ import ( "github.com/pion/rtp" ) +// ceiling on buffered packets before Push force-trims back to the target. +// 20 packets is ~400ms at 20ms framing - past this we're just adding delay. +const DefaultMaxDepth = 20 + // pion's jitterbuffer + late-drop on Push + skip-ahead on persistent gap; -// PopOrSkip reports gap counts so the caller can run PLC over them +// PopOrSkip reports gap counts so the caller can run PLC over them. +// +// TargetDepth/MaxDepth bound playout latency: Push hard-trims to TargetDepth +// once depth exceeds MaxDepth; ShedOne walks depth back down one packet per +// pop. Dropped packets skip the audio forward briefly. type Playout struct { mu sync.Mutex jb *jitterbuffer.JitterBuffer @@ -17,6 +25,22 @@ type Playout struct { headSeq uint16 // expected next seq // caps the scan in PopOrSkip's slow path MaxSkip uint16 + // trim target; also the floor ShedOne won't drop below. Defaults to prime. + TargetDepth int + // depth ceiling; Push trims to TargetDepth once exceeded + MaxDepth int + // buffered packets (pushed minus popped); jitterbuffer exposes no length. + // Playout latency in packets: depth*OpusFrameMs ms of audio waiting. + depth int + // lifetime counters for telemetry (see Stats) + cPushed, cLateDrop, cPopped, cTrimDrop, cTrimEvents, cShedDrop int64 + maxDepth int // high-water depth +} + +// PlayoutStats snapshots a Playout's lifetime counters and current/peak depth. +type PlayoutStats struct { + Depth, MaxDepthSeen int + Pushed, LateDrop, Popped, TrimDrop, TrimEvents, ShedDrop int64 } var ErrEmpty = errors.New("playout: empty") @@ -27,8 +51,10 @@ func NewPlayout(primePackets uint16) *Playout { primePackets = 2 } return &Playout{ - jb: jitterbuffer.New(jitterbuffer.WithMinimumPacketCount(primePackets)), - MaxSkip: 64, + jb: jitterbuffer.New(jitterbuffer.WithMinimumPacketCount(primePackets)), + MaxSkip: 64, + TargetDepth: int(primePackets), + MaxDepth: DefaultMaxDepth, } } @@ -41,9 +67,85 @@ func (p *Playout) Push(pkt *rtp.Packet) { defer p.mu.Unlock() // int16 cast -> signed seq distance if p.started && int16(pkt.SequenceNumber-p.headSeq) < 0 { + p.cLateDrop++ return } p.jb.Push(pkt) + p.depth++ + p.cPushed++ + if p.depth > p.maxDepth { + p.maxDepth = p.depth + } + // hard cap: keep bounding latency even while the consumer is stalled + // (Push runs on the reader goroutine, independent of the ticker) + if p.MaxDepth > 0 && p.depth > p.MaxDepth { + p.cTrimEvents++ + p.trimLocked(p.TargetDepth) + } +} + +// Stats snapshots the lifetime counters and current/peak depth. +func (p *Playout) Stats() PlayoutStats { + p.mu.Lock() + defer p.mu.Unlock() + return PlayoutStats{ + Depth: p.depth, + MaxDepthSeen: p.maxDepth, + Pushed: p.cPushed, + LateDrop: p.cLateDrop, + Popped: p.cPopped, + TrimDrop: p.cTrimDrop, + TrimEvents: p.cTrimEvents, + ShedDrop: p.cShedDrop, + } +} + +// trimLocked drops the oldest buffered packets until at most target remain. +// Caller holds p.mu. +func (p *Playout) trimLocked(target int) { + // budget guards against spinning across a long run of missing seqs + budget := p.depth + int(p.MaxSkip) + for p.depth > target && budget > 0 { + budget-- + if pkt, err := p.jb.Pop(); err == nil { + p.started = true + p.headSeq = pkt.SequenceNumber + 1 + p.depth-- + p.cTrimDrop++ + } else { + // hole at the head; step over it toward the next present packet + p.jb.SetPlayoutHead(p.jb.PlayoutHead() + 1) + } + } +} + +// ShedOne drops the oldest buffered packet if depth exceeds target, returning +// whether it dropped. Called once per pop so post-stall latency drains back to +// target gradually rather than surfacing as call-long lag. +func (p *Playout) ShedOne(target int) bool { + p.mu.Lock() + defer p.mu.Unlock() + if p.depth <= target { + return false + } + if pkt, err := p.jb.Pop(); err == nil { + p.started = true + p.headSeq = pkt.SequenceNumber + 1 + p.depth-- + p.cShedDrop++ + return true + } + // hole at the head; step past it and let the next pop try again + p.jb.SetPlayoutHead(p.jb.PlayoutHead() + 1) + return false +} + +// Depth reports the number of buffered packets not yet popped, i.e. the +// current playout latency in packets (depth*OpusFrameMs ms of audio). +func (p *Playout) Depth() int { + p.mu.Lock() + defer p.mu.Unlock() + return p.depth } // returns (next playable packet, gap count to PLC over, err); @@ -55,6 +157,8 @@ func (p *Playout) PopOrSkip() (*rtp.Packet, int, error) { if pkt, err := p.jb.Pop(); err == nil { p.started = true p.headSeq = pkt.SequenceNumber + 1 + p.depth-- + p.cPopped++ return pkt, 0, nil } else if errors.Is(err, jitterbuffer.ErrPopWhileBuffering) { return nil, 0, ErrEmpty @@ -69,6 +173,8 @@ func (p *Playout) PopOrSkip() (*rtp.Packet, int, error) { p.jb.SetPlayoutHead(pkt.SequenceNumber + 1) p.started = true p.headSeq = pkt.SequenceNumber + 1 + p.depth-- + p.cPopped++ return pkt, int(i), nil } } diff --git a/calls/audio/playout_test.go b/calls/audio/playout_test.go index 24a509b..65d142c 100644 --- a/calls/audio/playout_test.go +++ b/calls/audio/playout_test.go @@ -70,6 +70,57 @@ func TestPlayoutSkipAhead(t *testing.T) { } } +func TestPlayoutHardCapTrim(t *testing.T) { + p := NewPlayout(2) + p.MaxDepth = 10 + p.TargetDepth = 3 + // Push one past the ceiling; the crossing push trims the oldest back to + // TargetDepth in one shot. + for seq := uint16(0); seq <= uint16(p.MaxDepth); seq++ { // seqs 0..10 + p.Push(pkt(seq)) + } + if got := p.Depth(); got != p.TargetDepth { + t.Fatalf("after overflow: depth=%d, want %d", got, p.TargetDepth) + } + // The three survivors are the newest packets (8, 9, 10); the next pop + // returns 8, proving the oldest were the ones dropped. + got, _, err := p.PopOrSkip() + if err != nil { + t.Fatalf("pop after trim: %v", err) + } + if got.SequenceNumber != 8 { + t.Errorf("oldest survivor seq=%d, want 8", got.SequenceNumber) + } +} + +func TestPlayoutShedOne(t *testing.T) { + p := NewPlayout(2) + p.TargetDepth = 2 + for seq := uint16(0); seq < 5; seq++ { + p.Push(pkt(seq)) + } + // Depth 5, target 2: ShedOne drops the oldest and reports true until depth + // reaches the target, then leaves the buffer alone. + if !p.ShedOne(p.TargetDepth) || p.Depth() != 4 { + t.Fatalf("first shed: dropped=%v depth=%d, want true/4", true, p.Depth()) + } + if !p.ShedOne(p.TargetDepth) || p.Depth() != 3 { + t.Fatalf("second shed: depth=%d, want 3", p.Depth()) + } + p.ShedOne(p.TargetDepth) // depth 3 -> 2 + if shed := p.ShedOne(p.TargetDepth); shed || p.Depth() != 2 { + t.Errorf("at target: shed=%v depth=%d, want false/2", shed, p.Depth()) + } + // Shedding drops from the front: the oldest remaining is seq 3. + got, _, err := p.PopOrSkip() + if err != nil { + t.Fatalf("pop after shed: %v", err) + } + if got.SequenceNumber != 3 { + t.Errorf("oldest survivor seq=%d, want 3", got.SequenceNumber) + } +} + func TestPlayoutSeqWraparound(t *testing.T) { p := NewPlayout(2) // Push packets straddling the uint16 boundary: 65534, 65535, 0, 1. @@ -91,4 +142,3 @@ func TestPlayoutSeqWraparound(t *testing.T) { } } } - diff --git a/calls/audio/tg_to_xmpp.go b/calls/audio/tg_to_xmpp.go index 2e9c714..e065e21 100644 --- a/calls/audio/tg_to_xmpp.go +++ b/calls/audio/tg_to_xmpp.go @@ -43,9 +43,37 @@ type TgToXmpp struct { unregister func() // returned by Ntg.OnFrame; called from Close + // telemetry counters (atomic; sampled by the periodic stats log) + cFramesIn atomic.Int64 + cFramesWrongChat atomic.Int64 + cFramesWrongSSRC atomic.Int64 + cFramesBadSize atomic.Int64 + cOpusOut atomic.Int64 + cWriteErrs atomic.Int64 + + statsMu sync.Mutex + lastLog time.Time + callStart time.Time // first frame seen; anchor for cumulative drift + prevSnap tgToXmppSnap + capture *captureWriter } +type tgToXmppSnap struct { + framesIn, wrongChat, wrongSSRC, badSize, opusOut, writeErrs int64 +} + +func (h *TgToXmpp) snap() tgToXmppSnap { + return tgToXmppSnap{ + framesIn: h.cFramesIn.Load(), + wrongChat: h.cFramesWrongChat.Load(), + wrongSSRC: h.cFramesWrongSSRC.Load(), + badSize: h.cFramesBadSize.Load(), + opusOut: h.cOpusOut.Load(), + writeErrs: h.cWriteErrs.Load(), + } +} + // builds the encoder, local opus track, registers ntgcalls frame handler; // Start() flips on the ntgcalls source func NewTgToXmpp(opts TgToXmppOptions) (*TgToXmpp, error) { @@ -135,18 +163,79 @@ func (h *TgToXmpp) Close() error { } func (h *TgToXmpp) onNtgFrames(chatID int64, frames []PCMFrame) { - if h.closed.Load() || chatID != h.opts.ChatID { + if h.closed.Load() { return } + if chatID != h.opts.ChatID { + h.cFramesWrongChat.Add(int64(len(frames))) + return + } + h.cFramesIn.Add(int64(len(frames))) for _, f := range frames { h.capture.writePCM(f.SSRC, f.Data) h.feedFrame(f) } + h.maybeLogStats() +} + +// maybeLogStats emits one aggregated line per statsInterval from the ntg +// callback goroutine (this direction has no ticker to hang it off). +func (h *TgToXmpp) maybeLogStats() { + now := time.Now() + h.statsMu.Lock() + if h.lastLog.IsZero() { + h.lastLog = now + h.callStart = now + h.prevSnap = h.snap() + h.statsMu.Unlock() + return + } + callStart := h.callStart + elapsed := now.Sub(h.lastLog) + if elapsed < statsInterval { + h.statsMu.Unlock() + return + } + prev := h.prevSnap + cur := h.snap() + h.lastLog = now + h.prevSnap = cur + h.statsMu.Unlock() + + secs := elapsed.Seconds() + if secs <= 0 { + secs = 1 + } + rate := func(d int64) float64 { return float64(d) / secs } + h.mu.Lock() + accBytes := len(h.acc) + ssrc := h.primarySSRC + h.mu.Unlock() + // deliver_drift_ms = frames ntgcalls delivered minus what real time allows. + // Grows if ntgcalls over-delivers: that excess is audio we forward to pion + // faster than real time, piling up in the XMPP client's jitter buffer. + sinceStart := now.Sub(callStart) + expected := sinceStart.Milliseconds() / NtgFrameMs + driftMs := (cur.framesIn - expected) * NtgFrameMs + h.log.WithFields(log.Fields{ + "interval_ms": elapsed.Milliseconds(), + "frames_in_per_s": rate(cur.framesIn - prev.framesIn), + "opus_out_per_s": rate(cur.opusOut - prev.opusOut), + "deliver_drift_ms": driftMs, + "call_secs": int64(sinceStart.Seconds()), + "wrong_chat": cur.wrongChat - prev.wrongChat, + "wrong_ssrc": cur.wrongSSRC - prev.wrongSSRC, + "bad_size": cur.badSize - prev.badSize, + "write_errs": cur.writeErrs - prev.writeErrs, + "acc_bytes": accBytes, + "primary_ssrc": ssrc, + }).Debug("tg->xmpp stats") } // returns number of opus packets emitted (0 or 1) func (h *TgToXmpp) feedFrame(f PCMFrame) int { if len(f.Data) != NtgFrameBytes { + h.cFramesBadSize.Add(1) h.log.WithField("len", len(f.Data)).Warn("unexpected ntg frame size") return 0 } @@ -156,6 +245,7 @@ func (h *TgToXmpp) feedFrame(f PCMFrame) int { if h.primarySSRC == 0 { h.primarySSRC = f.SSRC } else if f.SSRC != h.primarySSRC { + h.cFramesWrongSSRC.Add(1) return 0 } @@ -180,8 +270,10 @@ func (h *TgToXmpp) feedFrame(f PCMFrame) int { Data: payload, Duration: OpusFrameMs * time.Millisecond, }); err != nil { + h.cWriteErrs.Add(1) h.log.WithError(err).Warn("WriteSample failed") return 0 } + h.cOpusOut.Add(1) return 1 } diff --git a/calls/audio/xmpp_to_tg.go b/calls/audio/xmpp_to_tg.go index 842dbf3..c2f0865 100644 --- a/calls/audio/xmpp_to_tg.go +++ b/calls/audio/xmpp_to_tg.go @@ -40,12 +40,38 @@ type XmppToTg struct { // pion's per-track queue but NOT enqueued - otherwise every packet // from the setup window would back up and the ticker would replay // it at real time, manifesting as call-long lag. - livePush atomic.Bool - preLiveDiscards atomic.Int64 + livePush atomic.Bool + preLiveDiscards atomic.Int64 + + // telemetry counters (atomic; sampled by the periodic stats log) + cRtpIn atomic.Int64 // RTP packets read from pion + cRtpSkippedPT atomic.Int64 // dropped: payload type != negotiated opus PT + cShipTotal atomic.Int64 // 10ms frames handed to ntgcalls (== ntg feed rate) + cShipSilence atomic.Int64 // of those, silence frames (underrun/priming) + cPlcFrames atomic.Int64 // PLC 10ms frames synthesised over gaps + cGapEvents atomic.Int64 // ticks where a seq gap was skipped + cSendErrs atomic.Int64 // SendMicrophonePCM errors capture *captureWriter } +// counter snapshot for computing per-interval deltas in the stats log +type xmppToTgSnap struct { + rtpIn, rtpSkippedPT, shipTotal, shipSilence, plcFrames, gapEvents, sendErrs int64 +} + +func (h *XmppToTg) snap() xmppToTgSnap { + return xmppToTgSnap{ + rtpIn: h.cRtpIn.Load(), + rtpSkippedPT: h.cRtpSkippedPT.Load(), + shipTotal: h.cShipTotal.Load(), + shipSilence: h.cShipSilence.Load(), + plcFrames: h.cPlcFrames.Load(), + gapEvents: h.cGapEvents.Load(), + sendErrs: h.cSendErrs.Load(), + } +} + func NewXmppToTg(opts XmppToTgOptions) (*XmppToTg, error) { if opts.Ntg == nil { return nil, errors.New("audio: Ntg required") @@ -96,8 +122,13 @@ func (h *XmppToTg) Start() error { h.livePush.Store(true) h.wg.Add(1) go h.tickerLoop() - h.log.WithField("pre_live_discards", h.preLiveDiscards.Load()). - Info("xmpp->tg half started") + h.log.WithFields(log.Fields{ + "pre_live_discards": h.preLiveDiscards.Load(), + "prime_packets": h.playout.TargetDepth, + "target_depth": h.playout.TargetDepth, + "max_depth": h.playout.MaxDepth, + "max_depth_ms": h.playout.MaxDepth * OpusFrameMs, + }).Info("xmpp->tg half started") return nil } @@ -141,8 +172,10 @@ func (h *XmppToTg) reader(t *webrtc.TrackRemote) { if err != nil { return } + h.cRtpIn.Add(1) if pkt.PayloadType != expectedPT { // CN/DTMF/RED can share an SSRC; libopus rejects them + h.cRtpSkippedPT.Add(1) continue } if !h.livePush.Load() { @@ -167,42 +200,124 @@ func newTickState() *tickState { return &tickState{lastDecodedSamples: SamplesPerOpusFrame} } +// how often the aggregated stats line is emitted +const statsInterval = 5 * time.Second + +// a tick gap over this means the goroutine was descheduled (GC, blocking CGO, +// scheduler pressure) and missed ticks. 25ms = missed a 10ms tick with margin. +const tickStallThreshold = 25 * time.Millisecond + func (h *XmppToTg) tickerLoop() { defer h.wg.Done() ticker := time.NewTicker(NtgFrameMs * time.Millisecond) defer ticker.Stop() st := newTickState() + loopStart := time.Now() + last := loopStart + lastLog := loopStart + prev := h.snap() + var ticks, stalls, maxGapMs int64 for { select { case <-h.stop: return case <-ticker.C: } + now := time.Now() + gap := now.Sub(last) + last = now + ticks++ + if gap > tickStallThreshold { + stalls++ + if ms := gap.Milliseconds(); ms > maxGapMs { + maxGapMs = ms + } + } + h.processTick(st) + + if now.Sub(lastLog) >= statsInterval { + h.logStats(now.Sub(lastLog), now.Sub(loopStart), ticks, stalls, maxGapMs, &prev) + ticks, stalls, maxGapMs = 0, 0, 0 + lastLog = now + } } } +// logStats emits one aggregated telemetry line per statsInterval, with +// per-second rates over the interval and the current playout depth. +func (h *XmppToTg) logStats(elapsed, sinceStart time.Duration, ticks, stalls, maxGapMs int64, prev *xmppToTgSnap) { + cur := h.snap() + secs := elapsed.Seconds() + if secs <= 0 { + secs = 1 + } + rate := func(d int64) float64 { return float64(d) / secs } + ps := h.playout.Stats() + // feed_drift_ms = frames shipped to ntgcalls minus what real time allows. + // Ticker-paced, so ~0 normally; a growing value means the ticker is starved + // and we're under-feeding ntgcalls' capture. + expected := sinceStart.Milliseconds() / NtgFrameMs + feedDriftMs := (cur.shipTotal - expected) * NtgFrameMs + h.log.WithFields(log.Fields{ + "interval_ms": elapsed.Milliseconds(), + "call_secs": int64(sinceStart.Seconds()), + "ticks": ticks, + "tick_stalls": stalls, + "max_tick_gap_ms": maxGapMs, + "rtp_in_per_s": rate(cur.rtpIn - prev.rtpIn), + "rtp_skipped_pt": cur.rtpSkippedPT - prev.rtpSkippedPT, + "ship_per_s": rate(cur.shipTotal - prev.shipTotal), + "silence_per_s": rate(cur.shipSilence - prev.shipSilence), + "feed_drift_ms": feedDriftMs, + "plc_frames": cur.plcFrames - prev.plcFrames, + "gap_events": cur.gapEvents - prev.gapEvents, + "send_errs": cur.sendErrs - prev.sendErrs, + "depth": ps.Depth, + "depth_ms": ps.Depth * OpusFrameMs, + "peak_depth_ms": ps.MaxDepthSeen * OpusFrameMs, + "jb_pushed": ps.Pushed, + "jb_popped": ps.Popped, + "jb_late_drop": ps.LateDrop, + "jb_trim_drop": ps.TrimDrop, + "jb_trim_events": ps.TrimEvents, + "jb_shed_drop": ps.ShedDrop, + }).Debug("xmpp->tg stats") + *prev = cur +} + // one iteration of the playout loop func (h *XmppToTg) processTick(st *tickState) { if len(st.pending10ms) == 0 { + // walk any accumulated latency back toward the target depth, one + // packet per pop, so recovery after a stall is gradual rather than a + // single large skip + h.playout.ShedOne(h.playout.TargetDepth) pkt, plcCount, err := h.playout.PopOrSkip() if err != nil { // empty or still priming - feed silence + h.cShipSilence.Add(1) h.sendToNtg(h.silence) return } + if plcCount > 0 { + h.cGapEvents.Add(1) + } for i := 0; i < plcCount; i++ { plcOut := h.pcmBuf[:st.lastDecodedSamples*Channels] if err := h.dec.DecodePLC(plcOut); err != nil { h.log.WithError(err).Warn("DecodePLC failed") break } + before := len(st.pending10ms) st.pending10ms = appendChunks10ms(st.pending10ms, plcOut) + h.cPlcFrames.Add(int64(len(st.pending10ms) - before)) } n, err := h.dec.Decode(pkt.Payload, h.pcmBuf) if err != nil { h.log.WithError(err).Warn("opus decode failed") + h.cShipSilence.Add(1) h.sendToNtg(h.silence) return } @@ -227,7 +342,9 @@ func appendChunks10ms(dst [][]byte, pcm []int16) [][]byte { } func (h *XmppToTg) sendToNtg(pcm []byte) { + h.cShipTotal.Add(1) if err := h.opts.Ntg.SendMicrophonePCM(h.opts.ChatID, pcm); err != nil { + h.cSendErrs.Add(1) h.log.WithError(err).Debug("SendMicrophonePCM failed") } }