From de7ed62d96f89c64bafe350af14ad3105e6c5fee Mon Sep 17 00:00:00 2001 From: nam Date: Wed, 4 Nov 2026 11:21:50 -0600 Subject: [PATCH] Jitter buffer v2, per-minute rx logging, websocket keepalive The cohost sounded jittery on the 2026-11-03 show. v1 skipped ahead on an underrun and kept playing from an empty buffer, so after the first hiccup every later wobble was another glitch. It also never corrected a slow guest clock: ~12 glitches/min in simulation at 30 ppm slow. - Underrun: hold the sequence position and refill to a larger target - Drift correction both ways (drop/insert a frame in quiet moments), quiet judged relative to the guest's own noise floor - 80 ms start, 300 ms max, relaxes after 2 calm minutes - Log 'rx 1m:' stats, the selected route, and call totals - Ping the guest websocket every 25 s (the proxy closed it every 90 s) - Simulation kept in test/jb_sim_test.go.txt Co-Authored-By: Claude Opus 5.5 --- README.md | 79 ++++++++++++++ call.go | 222 ++++++++++++++++++++++++++++------------ main.go | 21 ++++ test/jb_sim_test.go.txt | 86 ++++++++++++++++ web/static/host.js | 2 +- 5 files changed, 346 insertions(+), 64 deletions(-) create mode 100644 README.md create mode 100644 test/jb_sim_test.go.txt diff --git a/README.md b/README.md new file mode 100644 index 0000000..3d97666 --- /dev/null +++ b/README.md @@ -0,0 +1,79 @@ +# guestlink - MMO remote guest line + +Self-hosted replacement for Cleanfeed: one remote guest, live audio both ways, plus an +anonymous text chat between the guest and the host. Built 2026-11-02. + +| What | Where | +|---|---| +| Guest page | `https://guest.mmo.show/j/` (NPM on .158 -> 192.168.50.204:8790) | +| Host console | `http://192.168.50.204:8791` (LAN/VPN only, never proxied) | +| Media | UDP **50090** - one port for all WebRTC traffic; forwarded on the router to .204 | +| Service | `systemctl --user {status,restart} guestlink` · logs: `journalctl --user -u guestlink -f` | +| Invite token | `~/.local/state/guestlink/token` (0600) - survives restarts | + +## Audio path (same as `motu-cleanfeed`) +- guest voice -> `pacat` -> sink **`motu_guest`** -> From Computer 1-2 -> guest strip +- **`motu_mixminus.monitor`** (To Computer 3-4, show minus guest) -> `parec` -> guest + +Both are ordinary PipeWire client streams that exist only during a call. They never reopen +the MOTU. They carry `node.dont-fallback/dont-reconnect/dont-move`, so if `motu_guest` +disappears playback **fails** instead of landing on the default output (tested). +Restarting guestlink is safe for the rest of the rig. It only drops the guest's call. + +Opus 48 kHz stereo, 128 kb/s, in-band FEC. The guest page disables browser echo cancellation, +noise suppression and AGC. The "I'm wearing headphones" box is ticked by default; unticking it +turns echo cancellation back on for guests on speakers. + +The receive side has an adaptive jitter buffer (v2, 2026-11-04): 80 ms start, max 300 ms. +On an underrun it holds its place and refills to a larger target, so delayed packets are +played rather than thrown away. It corrects guest/MOTU clock drift in both directions by +dropping (guest fast) or adding (guest slow) a frame during a quiet moment. "Quiet" is +relative to the guest's own noise floor. The target rises on underruns and late packets +and relaxes after two calm minutes. Host console: `buffer x/y ms`, lost/FEC/concealed, +late, drift-drops, inserted, underruns. + +**Logs for diagnosing a bad night:** one `rx 1m:` line a minute per call (buffer, losses, +underruns, drift corrections), the network `route:` when it is chosen, and `call totals:` +when it ends. `journalctl --user -u guestlink --since ...` + +Why v2: v1 skipped ahead on an underrun and kept playing from an empty buffer, so after the +first hiccup every later wobble was another glitch. It also never corrected for a slow guest +clock. In simulation that was ~12 glitches/min with a guest clock 30 ppm slow, against ~0 +for v2. The 3 Nov 2026 cohost "jitter" was probably this, but that night was not measured. +The simulation is kept in `test/jb_sim_test.go.txt` (rename to `_test.go` in the package dir to run). + +The guest websocket is pinged every 25 s. Without that, the proxy closed it every 90 s +(signalling only, audio unaffected, but the guest saw "reconnecting..."). + +## Host console +New link (ends the session, clears chat), Revoke, Copy, End audio, Kick, Clear chat. It shows +live meters for both directions and the selected network route. The page never plays sound +(it is on the rig): a new guest message flashes the tab title. + +## Chat +Anonymous: no accounts, and the guest's name is optional (default "Guest"). The guest sees +the host as `-host-label` (default "Host"). Messages are kept **in memory only** (last 300), +are never written to disk or logs, and are cleared by New link / Clear / restart. Guest +messages are rate limited and rendered as plain text. + +## Public IP +Learned per call via STUN (google, then cloudflare), falling back to the A record of +guest.mmo.show. Override with `-public-ip`. + +## Build / test + cd ~/apps/guestlink && go build -tags nolibopusfile -o guestlink . && go build -tags nolibopusfile -o glprobe ./cmd/glprobe + systemctl --user restart guestlink # drops a live guest - not during a show + +`glprobe` is a headless test guest. **It creates a new link (ending the current session)** +unless given `-token`. Safe offline test on spare ports, with no PipeWire playback: + + ./guestlink -public 127.0.0.1:18790 -host 127.0.0.1:18791 -udp 50091 \ + -sink none -test-playout-file /tmp/rx.raw -state /tmp/glstate & + ./glprobe -hostws ws://127.0.0.1:18791/ws -guest ws://127.0.0.1:18790 -tone -20 + +## Fallback +`motu-cleanfeed` is untouched and still works. Don't run both at once: both would feed `motu_guest`. + +## Not done (v1) +One guest only. No TURN relay: a guest on a network that blocks all outbound UDP (some +corporate/hotel Wi-Fi) cannot connect; they can use a phone hotspot. No recording. diff --git a/call.go b/call.go index bc09aa9..7bb6226 100644 --- a/call.go +++ b/call.go @@ -246,7 +246,10 @@ func (c *Call) updateRoute() { continue } if pair, err := ice.GetSelectedCandidatePair(); err == nil && pair != nil { - c.route.Store(fmt.Sprintf("%s:%d <-> %s:%d (%s)", pair.Local.Address, pair.Local.Port, pair.Remote.Address, pair.Remote.Port, pair.Remote.Typ)) + r := fmt.Sprintf("%s:%d <-> %s:%d (%s)", pair.Local.Address, pair.Local.Port, pair.Remote.Address, pair.Remote.Port, pair.Remote.Typ) + if old, _ := c.route.Swap(r).(string); old != r { + log.Printf("route: %s", r) + } return } } @@ -258,9 +261,36 @@ func (c *Call) Close() { c.cancel() c.pc.Close() log.Printf("call closed after %s", time.Since(c.started).Round(time.Second)) + c.jbMu.Lock() + jb := c.jb + c.jbMu.Unlock() + if jb != nil { + k, _, t := jb.snapshot() + log.Printf("call totals: lost %d (fec %d) concealed %d late %d underruns %d drift-drops %d inserted %d, final target %d ms", + k.Lost, k.FEC, k.PLC, k.Late, k.Underruns, k.Dropped, k.Inserted, t) + } }) } +// logRx writes one line a minute about the guest's audio, so a bad night can be +// diagnosed afterwards (v1 only showed these numbers live on the host console). +func (c *Call) logRx(jb *jitterBuffer) { + t := time.NewTicker(time.Minute) + defer t.Stop() + var prev jbCounters + for { + select { + case <-c.ctx.Done(): + return + case <-t.C: + } + k, d, tg := jb.snapshot() + log.Printf("rx 1m: buffer %d/%d ms, lost %d (fec %d), concealed %d, late %d, underruns %d, drift-drops %d, inserted %d", + d, tg, k.Lost-prev.Lost, k.FEC-prev.FEC, k.PLC-prev.PLC, k.Late-prev.Late, k.Underruns-prev.Underruns, k.Dropped-prev.Dropped, k.Inserted-prev.Inserted) + prev = k + } +} + func (c *Call) status() map[string]any { st := map[string]any{ "pc": c.pcState.Load(), @@ -386,6 +416,7 @@ func (c *Call) startPlay(t *webrtc.TrackRemote) { c.jb = jb c.jbMu.Unlock() + go c.logRx(jb) go func() { // network -> jitter buffer for { p, _, err := t.ReadRTP() @@ -484,51 +515,101 @@ func (c *Call) playToFile(jb *jitterBuffer) error { // ---------------------------------------------------------------- jitter buffer // jitterBuffer reorders Opus packets, conceals losses (FEC, then PLC), adapts its depth to -// network jitter, and corrects clock drift between the guest's sound card and the MOTU by -// discarding quiet frames when it runs too deep. +// network jitter, and corrects clock drift between the guest's sound card and the MOTU. +// +// It is pulled once per 20 ms by the MOTU clock. Rules (v2, 2026-11-04): +// - Underrun (nothing buffered): HOLD the sequence position and refill to the target +// before playing again. The delayed packets are then played, not discarded as late. +// (v1 skipped ahead and kept playing from an empty buffer, so every later wobble +// became another glitch - ~12/min with a guest clock 30 ppm slow.) +// - Too deep (fast guest clock / after a stall): discard quiet frames. +// - Too shallow (slow guest clock): insert a concealment frame during a quiet moment. +// - Underruns and late packets raise the target; it relaxes after 2 calm minutes. type jitterBuffer struct { - mu sync.Mutex - dec *opus.Decoder - pkts map[uint16][]byte - nextSeq uint16 - started bool - target int // frames to hold before starting / after an underrun - misses int - lastUnd time.Time - pcm []int16 - lost int - fec int - plc int - dropped int - underrun int - late int + mu sync.Mutex + dec *opus.Decoder + pkts map[uint16][]byte + nextSeq uint16 + haveSeq bool // nextSeq is meaningful + playing bool // false = filling up to target + target int // frames + gap int // frames produced while refilling + lastBad time.Time + lastBump time.Time + lateRun int + insertNow bool + floor float64 // running estimate of the guest's background-noise RMS + pcm []int16 + c jbCounters +} + +type jbCounters struct { + Lost, FEC, PLC, Dropped, Inserted, Underruns, Late int +} + +const ( + jbFloor = 4 // 80 ms minimum target + jbMax = 15 // 300 ms maximum target + quietRMS = 0.003 // always counts as quiet (-50 dBFS) + quietCap = 0.02 // never counts as quiet (-34 dBFS) +) + +// isQuiet: below -50 dBFS, or within ~6 dB of the guest's own noise floor (so drift +// correction still works for a guest in a noisy room), but never above -34 dBFS. +func (j *jitterBuffer) isQuiet(pcm []int16) bool { + r := rms(pcm) + if r < j.floor { + j.floor = r + } else { + j.floor *= 1.0005 // creeps up ~2.5%/s so it follows a rising floor + } + lim := math.Min(quietCap, math.Max(quietRMS, 2*j.floor)) + return r < lim } func newJitterBuffer(dec *opus.Decoder) *jitterBuffer { - return &jitterBuffer{dec: dec, pkts: map[uint16][]byte{}, target: 3, pcm: make([]int16, 5760*channels), lastUnd: time.Now()} + return &jitterBuffer{dec: dec, pkts: map[uint16][]byte{}, target: jbFloor, pcm: make([]int16, 5760*channels), lastBad: time.Now(), floor: quietRMS} } func seqLess(a, b uint16) bool { return int16(a-b) < 0 } +func (j *jitterBuffer) reset() { + j.pkts = map[uint16][]byte{} + j.haveSeq, j.playing, j.lateRun, j.insertNow = false, false, 0, false +} + +// bump raises the target after an underrun or a late packet (at most once a second, so a +// burst of late packets counts once). +func (j *jitterBuffer) bump() { + j.lastBad = time.Now() + if j.target < jbMax && time.Since(j.lastBump) > time.Second { + j.target++ + j.lastBump = time.Now() + } +} + func (j *jitterBuffer) push(p *rtp.Packet) { j.mu.Lock() defer j.mu.Unlock() if len(p.Payload) == 0 { return } - if j.started { + if j.haveSeq { if seqLess(p.SequenceNumber, j.nextSeq) { - j.late++ + j.c.Late++ + j.bump() + if j.lateRun++; j.lateRun > 50 { // sender restarted its sequence numbers + j.reset() + } return } if d := uint16(p.SequenceNumber - j.nextSeq); d > 500 { // stream restarted - j.pkts = map[uint16][]byte{} - j.started = false + j.reset() } } + j.lateRun = 0 if len(j.pkts) > 100 { // hard cap - should never happen - j.pkts = map[uint16][]byte{} - j.started = false + j.reset() } j.pkts[p.SequenceNumber] = append([]byte(nil), p.Payload...) } @@ -554,15 +635,16 @@ func (j *jitterBuffer) decode(p []byte) []int16 { return append([]int16(nil), j.pcm[:n*channels]...) } -func (j *jitterBuffer) conceal() []int16 { +func (j *jitterBuffer) plcFrame() []int16 { out := make([]int16, frameSamps*channels) if err := j.dec.DecodePLC(out); err != nil { return j.silence() } - j.plc++ return out } +func (j *jitterBuffer) conceal() []int16 { j.c.PLC++; return j.plcFrame() } + func rms(pcm []int16) float64 { if len(pcm) == 0 { return 0 @@ -575,65 +657,73 @@ func rms(pcm []int16) float64 { return math.Sqrt(s / float64(len(pcm))) } +// fill is played while the buffer refills: a few concealment frames (Opus PLC fades out +// naturally), then silence. +func (j *jitterBuffer) fill() []int16 { + j.gap++ + if j.haveSeq && j.gap <= 3 { + return j.conceal() + } + return j.silence() +} + // next returns the next block of interleaved PCM to play (normally 20 ms). func (j *jitterBuffer) next() []int16 { j.mu.Lock() defer j.mu.Unlock() - // Relax the target after a quiet minute. - if j.target > 3 && time.Since(j.lastUnd) > 60*time.Second { + // Relax the target after two calm minutes. + if j.target > jbFloor && time.Since(j.lastBad) > 2*time.Minute { j.target-- - j.lastUnd = time.Now() + j.lastBad = time.Now() } - if !j.started { + if !j.playing { if len(j.pkts) < j.target { - return j.silence() + return j.fill() } - j.nextSeq = j.oldest() - j.started = true - j.misses = 0 + if !j.haveSeq { + j.nextSeq, j.haveSeq = j.oldest(), true + } + j.playing, j.gap = true, 0 + } + + if j.insertNow { // slow guest clock: stretch a quiet moment by one frame + j.insertNow = false + j.c.Inserted++ + return j.plcFrame() } for { - p, ok := j.pkts[j.nextSeq] - if ok { + if p, ok := j.pkts[j.nextSeq]; ok { delete(j.pkts, j.nextSeq) j.nextSeq++ - j.misses = 0 pcm := j.decode(p) depth := len(j.pkts) - // Drift / jitter-spike correction: too deep -> throw away quiet frames, - // or any frame if far too deep. - if (depth > j.target+3 && rms(pcm) < 0.003) || depth > j.target+12 { - j.dropped++ + quiet := j.isQuiet(pcm) + if (depth > j.target+2 && quiet) || depth > j.target+10 { + j.c.Dropped++ continue } + if depth < j.target-1 && quiet { + j.insertNow = true + } return pcm } if len(j.pkts) == 0 { - // Underrun: nothing to play. - j.misses++ - if j.misses == 1 { - j.underrun++ - if j.target < 10 { - j.target++ - } - j.lastUnd = time.Now() - } - if j.misses > 5 { - j.started = false // rebuffer to target, play silence meanwhile - return j.silence() - } - j.nextSeq++ - return j.conceal() + // Underrun: keep our place and refill to a larger target. + j.c.Underruns++ + j.lastBump = time.Time{} // an underrun always counts + j.bump() + j.playing = false + return j.fill() } - // Gap with later packets present: the packet is lost. - j.lost++ + // Gap with later packets present: this packet is lost (or too late to wait for). + j.c.Lost++ if fp, ok := j.pkts[j.nextSeq+1]; ok { out := make([]int16, frameSamps*channels) if err := j.dec.DecodeFEC(fp, out); err == nil { - j.fec++ + j.c.FEC++ j.nextSeq++ return out } @@ -643,11 +733,17 @@ func (j *jitterBuffer) next() []int16 { } } -func (j *jitterBuffer) stats() map[string]any { +func (j *jitterBuffer) snapshot() (c jbCounters, depthMs, targetMs int) { j.mu.Lock() defer j.mu.Unlock() + return j.c, len(j.pkts) * 20, j.target * 20 +} + +func (j *jitterBuffer) stats() map[string]any { + c, d, t := j.snapshot() return map[string]any{ - "depthMs": len(j.pkts) * 20, "targetMs": j.target * 20, - "lost": j.lost, "fec": j.fec, "plc": j.plc, "dropped": j.dropped, "underruns": j.underrun, "late": j.late, + "depthMs": d, "targetMs": t, + "lost": c.Lost, "fec": c.FEC, "plc": c.PLC, "dropped": c.Dropped, "inserted": c.Inserted, + "underruns": c.Underruns, "late": c.Late, } } diff --git a/main.go b/main.go index 8d4bdce..2f6ca69 100644 --- a/main.go +++ b/main.go @@ -297,6 +297,26 @@ func guestWS(w http.ResponseWriter, r *http.Request) { } c.SetReadLimit(64 << 10) me := &wsConn{c: c} + // Keepalive: ping every 25 s. Without it the proxy closed the idle socket every 90 s + // and the guest saw "reconnecting..." all show. Signalling only - audio is unaffected. + c.SetReadDeadline(time.Now().Add(70 * time.Second)) + c.SetPongHandler(func(string) error { c.SetReadDeadline(time.Now().Add(70 * time.Second)); return nil }) + pingDone := make(chan struct{}) + defer close(pingDone) + go func() { + t := time.NewTicker(25 * time.Second) + defer t.Stop() + for { + select { + case <-pingDone: + return + case <-t.C: + if err := c.WriteControl(websocket.PingMessage, nil, time.Now().Add(5*time.Second)); err != nil { + return + } + } + } + }() hub.mu.Lock() old := hub.guest @@ -339,6 +359,7 @@ func guestWS(w http.ResponseWriter, r *http.Request) { if err := c.ReadJSON(&m); err != nil { return } + c.SetReadDeadline(time.Now().Add(70 * time.Second)) hub.mu.Lock() current := hub.guest == me && hub.token == tok hub.mu.Unlock() diff --git a/test/jb_sim_test.go.txt b/test/jb_sim_test.go.txt new file mode 100644 index 0000000..7a060d7 --- /dev/null +++ b/test/jb_sim_test.go.txt @@ -0,0 +1,86 @@ +package main + +import ( + "fmt" + "math" + "math/rand" + "sort" + "testing" + + "github.com/pion/rtp" + "gopkg.in/hraban/opus.v2" +) + +// Simulate: sender emits a packet every 20ms*(1+ppm) of its clock; network adds jitter; +// playout pulls one frame every 20ms of MOTU clock. Count audible damage over 3 hours. +func sim(t *testing.T, name string, noiseFloor float64, ppm float64, jitterMs func(r *rand.Rand) float64) { + enc, _ := opus.NewEncoder(rate, channels, opus.AppAudio) + nr := rand.New(rand.NewSource(7)) + mk := func(i int, talk bool) []byte { + pcm := make([]int16, frameSamps*channels) + for k := range pcm { + n := nr.NormFloat64() * noiseFloor * 32768 + v := n + if talk { v += 8000 * math.Sin(float64(i*frameSamps+k/2)*2*math.Pi*220/rate) * (0.5 + 0.5*math.Sin(float64(i)*0.9)) } + pcm[k] = int16(v) + } + buf := make([]byte, 4000); n, _ := enc.Encode(pcm, buf); return append([]byte(nil), buf[:n]...) + } + var talkP, quietP [][]byte + for i := 0; i < 50; i++ { talkP = append(talkP, mk(i, true)); quietP = append(quietP, mk(i, false)) } + payloadFor := func(seq int) []byte { if seq%175 < 125 { return talkP[seq%50] }; return quietP[seq%50] } + dec, _ := opus.NewDecoder(rate, channels) + jb := newJitterBuffer(dec) + r := rand.New(rand.NewSource(1)) + const dur = 3 * 3600 * 1000.0 + type ev struct{ at float64; seq uint16 } + var arr []ev + period := 20.0 * (1 + ppm*1e-6) + for i := 0; float64(i)*period < dur; i++ { + arr = append(arr, ev{float64(i)*period + 50 + jitterMs(r), uint16(i)}) + } + sort.Slice(arr, func(a, b int) bool { return arr[a].at < arr[b].at }) + k := 0 + depthSum, frames, nearEmpty := 0, 0, 0 + for tms := 0.0; tms < dur; tms += 20 { + for k < len(arr) && arr[k].at <= tms { jb.push(&rtp.Packet{Header: rtp.Header{SequenceNumber: arr[k].seq}, Payload: payloadFor(int(arr[k].seq) + 65536*0)}); k++ } + jb.next() + d := len(jb.pkts); depthSum += d; frames++ + if d == 0 { nearEmpty++ } + } + s := jb.stats() + bad := s["plc"].(int) + s["fec"].(int) + s["dropped"].(int) + fmt.Printf("%-34s glitches(PLC+FEC)=%5d (~%.1f/min) underruns=%4d late-discarded=%5d drops=%4d inserted=%4d avg depth=%.0fms empty %.0f%% of time final target=%v\n", + name, bad, float64(bad)/180, s["underruns"], s["late"], s["dropped"], s["inserted"], float64(depthSum)/float64(frames)*20, 100*float64(nearEmpty)/float64(frames), s["targetMs"]) +} + +func TestSim(t *testing.T) { + for _, nf := range []float64{0.0005, 0.006} { + fmt.Printf("--- pauses at %.0f dBFS noise floor\n", 20*math.Log10(nf)) + gauss := func(sd float64) func(*rand.Rand) float64 { return func(r *rand.Rand) float64 { return math.Abs(r.NormFloat64() * sd) } } + spiky := func(r *rand.Rand) float64 { // typical home Wi-Fi: small jitter + occasional 80-200ms stalls + j := math.Abs(r.NormFloat64() * 4) + if r.Float64() < 0.002 { j += 80 + r.Float64()*120 } + return j + } + sim(t, "clean net, no drift", nf, 0, gauss(0.5)) + sim(t, "5ms jitter, no drift", nf, 0, gauss(5)) + sim(t, "5ms jitter, guest clock 30ppm FAST", nf, -30, gauss(5)) + sim(t, "5ms jitter, guest clock 30ppm SLOW", nf, 30, gauss(5)) + sim(t, "Wi-Fi-ish spikes, no drift", nf, 0, spiky) + sim(t, "Wi-Fi-ish spikes, guest 30ppm SLOW", nf, 30, spiky) + // correlated stall: every packet in a 150 ms window is held until the window ends + burst := func() func(*rand.Rand) float64 { + i, until := 0, -1.0 + return func(r *rand.Rand) float64 { + now := float64(i) * 20; i++ + if now > until && r.Float64() < 0.0005 { until = now + 150 } + j := math.Abs(r.NormFloat64() * 3) + if now <= until { j += until - now } + return j + } + } + sim(t, "burst stalls 150ms, no drift", nf, 0, burst()) + sim(t, "burst stalls 150ms, guest SLOW", nf, 30, burst()) +} +} diff --git a/web/static/host.js b/web/static/host.js index dc421bd..d160b68 100644 --- a/web/static/host.js +++ b/web/static/host.js @@ -39,7 +39,7 @@ function render(s){ $('#dIn').textContent = Math.max(-60, setMeter($('#mIn'), c.inPeak)).toFixed(0) + ' dB'; $('#dOut').textContent = Math.max(-60, setMeter($('#mOut'), c.outPeak)).toFixed(0) + ' dB'; const j = c.jb; - $('#jb').textContent = j ? `buffer ${j.depthMs}/${j.targetMs} ms · lost ${j.lost} (fec ${j.fec}) · concealed ${j.plc} · drift-drops ${j.dropped} · underruns ${j.underruns}` : ''; + $('#jb').textContent = j ? `buffer ${j.depthMs}/${j.targetMs} ms · lost ${j.lost} (fec ${j.fec}) · concealed ${j.plc} · late ${j.late} · drift-drops ${j.dropped} · inserted ${j.inserted} · underruns ${j.underruns}` : ''; } else { setMeter($('#mIn'), 0); setMeter($('#mOut'), 0); $('#dIn').textContent = $('#dOut').textContent = ''; $('#jb').textContent = ''; }