diff --git a/Overlord-Client/cmd/agent/webrtcpub/state.go b/Overlord-Client/cmd/agent/webrtcpub/state.go index 1ea9714..0b2fce5 100644 --- a/Overlord-Client/cmd/agent/webrtcpub/state.go +++ b/Overlord-Client/cmd/agent/webrtcpub/state.go @@ -24,10 +24,70 @@ type AudioWriter interface { } type writerEntry struct { - video VideoWriter + video *latestVideoWriter audio AudioWriter } +type latestVideoWriter struct { + writer VideoWriter + mu sync.Mutex + pending []byte + dur time.Duration + closed bool + wake chan struct{} +} + +func newLatestVideoWriter(writer VideoWriter) *latestVideoWriter { + queued := &latestVideoWriter{writer: writer, wake: make(chan struct{}, 1)} + go queued.run() + return queued +} + +func (w *latestVideoWriter) enqueue(frame []byte, dur time.Duration) { + w.mu.Lock() + if w.closed { + w.mu.Unlock() + return + } + w.pending = frame + w.dur = dur + w.mu.Unlock() + select { + case w.wake <- struct{}{}: + default: + } +} + +func (w *latestVideoWriter) close() { + w.mu.Lock() + w.closed = true + w.pending = nil + w.mu.Unlock() + select { + case w.wake <- struct{}{}: + default: + } +} + +func (w *latestVideoWriter) run() { + for range w.wake { + for { + w.mu.Lock() + if w.closed { + w.mu.Unlock() + return + } + frame, dur := w.pending, w.dur + w.pending = nil + w.mu.Unlock() + if len(frame) == 0 { + break + } + _ = w.writer.WriteH264(frame, dur) + } + } +} + var ( writersMu sync.RWMutex writers = map[string]map[string]writerEntry{} // kind → id → entry @@ -44,7 +104,10 @@ func registerVideoWriter(kind Kind, id string, w VideoWriter) { writers[string(kind)] = bucket } entry := bucket[id] - entry.video = w + if entry.video != nil { + entry.video.close() + } + entry.video = newLatestVideoWriter(w) bucket[id] = entry writersMu.Unlock() RequestKeyframe() @@ -72,6 +135,9 @@ func unregisterWriter(kind Kind, id string) { } writersMu.Lock() if bucket, ok := writers[string(kind)]; ok { + if entry, exists := bucket[id]; exists && entry.video != nil { + entry.video.close() + } delete(bucket, id) if len(bucket) == 0 { delete(writers, string(kind)) @@ -103,14 +169,19 @@ func WriteH264(kind Kind, nalu []byte, dur time.Duration) error { if len(nalu) == 0 { return nil } + frame := append([]byte(nil), nalu...) writersMu.RLock() - defer writersMu.RUnlock() bucket := writers[string(kind)] + targets := make([]*latestVideoWriter, 0, len(bucket)) for _, w := range bucket { if w.video != nil { - _ = w.video.WriteH264(nalu, dur) + targets = append(targets, w.video) } } + writersMu.RUnlock() + for _, target := range targets { + target.enqueue(frame, dur) + } return nil } @@ -119,13 +190,17 @@ func WriteAudio(kind Kind, pcm []int16) error { return nil } writersMu.RLock() - defer writersMu.RUnlock() bucket := writers[string(kind)] + targets := make([]AudioWriter, 0, len(bucket)) for _, w := range bucket { if w.audio != nil { - _ = w.audio.WriteAudio(pcm) + targets = append(targets, w.audio) } } + writersMu.RUnlock() + for _, target := range targets { + _ = target.WriteAudio(pcm) + } return nil } diff --git a/Overlord-Client/cmd/agent/webrtcpub/state_test.go b/Overlord-Client/cmd/agent/webrtcpub/state_test.go new file mode 100644 index 0000000..1f189a7 --- /dev/null +++ b/Overlord-Client/cmd/agent/webrtcpub/state_test.go @@ -0,0 +1,69 @@ +package webrtcpub + +import ( + "sync" + "testing" + "time" +) + +type blockingVideoWriter struct { + started chan struct{} + release chan struct{} + mu sync.Mutex + frames [][]byte +} + +func (w *blockingVideoWriter) WriteH264(frame []byte, _ time.Duration) error { + w.mu.Lock() + w.frames = append(w.frames, append([]byte(nil), frame...)) + count := len(w.frames) + w.mu.Unlock() + if count == 1 { + close(w.started) + <-w.release + } + return nil +} + +func TestWriteH264DoesNotBlockAndKeepsLatestPendingFrame(t *testing.T) { + kind := Kind("queue-test") + id := "slow-writer" + writer := &blockingVideoWriter{ + started: make(chan struct{}), + release: make(chan struct{}), + } + registerVideoWriter(kind, id, writer) + defer unregisterWriter(kind, id) + + if err := WriteH264(kind, []byte{1}, time.Millisecond); err != nil { + t.Fatal(err) + } + select { + case <-writer.started: + case <-time.After(time.Second): + t.Fatal("writer did not receive first frame") + } + + started := time.Now() + _ = WriteH264(kind, []byte{2}, time.Millisecond) + _ = WriteH264(kind, []byte{3}, time.Millisecond) + if elapsed := time.Since(started); elapsed > 100*time.Millisecond { + t.Fatalf("capture path blocked for %s", elapsed) + } + close(writer.release) + + deadline := time.Now().Add(time.Second) + for time.Now().Before(deadline) { + writer.mu.Lock() + frames := append([][]byte(nil), writer.frames...) + writer.mu.Unlock() + if len(frames) >= 2 { + if frames[0][0] != 1 || frames[1][0] != 3 { + t.Fatalf("expected first and latest frames, got %v", frames) + } + return + } + time.Sleep(5 * time.Millisecond) + } + t.Fatal("latest pending frame was not delivered") +} diff --git a/Overlord-Client/cmd/agent/webrtcpub/whip_pion.go b/Overlord-Client/cmd/agent/webrtcpub/whip_pion.go index 554d868..b83acbb 100644 --- a/Overlord-Client/cmd/agent/webrtcpub/whip_pion.go +++ b/Overlord-Client/cmd/agent/webrtcpub/whip_pion.go @@ -16,6 +16,7 @@ import ( "os" "strings" "sync" + "sync/atomic" "time" "github.com/pion/rtcp" @@ -46,23 +47,36 @@ func drainRTCP(sender *webrtc.RTPSender) { const h264SDPFmtpLine = "level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=640034" var rtcpKeyframeLogOnce sync.Once +var lastRTCPKeyframeRequest atomic.Int64 + +const rtcpKeyframeMinInterval = 750 * time.Millisecond func handleRTCPKeyframeFeedback() { if !honorRTCPKeyframes() { rtcpKeyframeLogOnce.Do(func() { - log.Printf("webrtcpub: ignoring RTCP video keyframe requests by default; set OVERLORD_WEBRTC_RTCP_KEYFRAMES=true to honor them") + log.Printf("webrtcpub: RTCP video keyframe recovery disabled by OVERLORD_WEBRTC_RTCP_KEYFRAMES") }) return } + now := time.Now().UnixNano() + for { + last := lastRTCPKeyframeRequest.Load() + if last > 0 && time.Duration(now-last) < rtcpKeyframeMinInterval { + return + } + if lastRTCPKeyframeRequest.CompareAndSwap(last, now) { + break + } + } RequestKeyframe() } func honorRTCPKeyframes() bool { switch strings.ToLower(strings.TrimSpace(os.Getenv("OVERLORD_WEBRTC_RTCP_KEYFRAMES"))) { - case "1", "true", "yes", "on": - return true - default: + case "0", "false", "no", "off": return false + default: + return true } } diff --git a/Overlord-Server/src/client-db-sync.ts b/Overlord-Server/src/client-db-sync.ts index f32bbdd..46efa4c 100644 --- a/Overlord-Server/src/client-db-sync.ts +++ b/Overlord-Server/src/client-db-sync.ts @@ -1,7 +1,16 @@ import { upsertClientRows, type ClientDbRow } from "./db"; import { metrics } from "./metrics"; -export const CLIENT_DB_SYNC_INTERVAL_MS = Number(process.env.OVERLORD_CLIENT_DB_SYNC_MS || 5000); +function positiveNumberEnv(name: string, fallback: number): number { + const value = Number(process.env[name] || fallback); + return Number.isFinite(value) && value > 0 ? value : fallback; +} + +export const CLIENT_DB_SYNC_INTERVAL_MS = positiveNumberEnv("OVERLORD_CLIENT_DB_SYNC_MS", 5000); +export const CLIENT_PRESENCE_SYNC_INTERVAL_MS = Math.max( + CLIENT_DB_SYNC_INTERVAL_MS, + positiveNumberEnv("OVERLORD_CLIENT_PRESENCE_SYNC_MS", 60_000), +); export const CLIENT_DB_SYNC_BATCH_SIZE = Math.max( 1, Number(process.env.OVERLORD_CLIENT_DB_SYNC_BATCH_SIZE || 500), @@ -65,7 +74,7 @@ setInterval(flushQueuedClientDbUpdates, CLIENT_DB_SYNC_INTERVAL_MS); export function shouldSyncClientToDb(clientId: string, now: number): boolean { const last = lastClientDbSync.get(clientId) || 0; - if (now - last < CLIENT_DB_SYNC_INTERVAL_MS) return false; + if (now - last < CLIENT_PRESENCE_SYNC_INTERVAL_MS) return false; lastClientDbSync.set(clientId, now); return true; } diff --git a/Overlord-Server/src/db/schema.ts b/Overlord-Server/src/db/schema.ts index 6c8131b..fcfa0bb 100644 --- a/Overlord-Server/src/db/schema.ts +++ b/Overlord-Server/src/db/schema.ts @@ -221,8 +221,24 @@ try { DELETE FROM client_search_fts WHERE rowid = old.rowid; END; `); + db.run(`DROP TRIGGER IF EXISTS clients_search_au`); db.run(` - CREATE TRIGGER IF NOT EXISTS clients_search_au AFTER UPDATE ON clients BEGIN + CREATE TRIGGER clients_search_au AFTER UPDATE ON clients + WHEN old.host IS NOT new.host + OR old.user IS NOT new.user + OR old.nickname IS NOT new.nickname + OR old.custom_tag IS NOT new.custom_tag + OR old.custom_tag_note IS NOT new.custom_tag_note + OR old.os IS NOT new.os + OR old.ip IS NOT new.ip + OR old.hwid IS NOT new.hwid + OR old.country IS NOT new.country + OR old.version IS NOT new.version + OR old.build_tag IS NOT new.build_tag + OR old.cpu IS NOT new.cpu + OR old.gpu IS NOT new.gpu + OR old.ram IS NOT new.ram + BEGIN DELETE FROM client_search_fts WHERE rowid = old.rowid; INSERT INTO client_search_fts( rowid, id, host, user, nickname, custom_tag, custom_tag_note, diff --git a/Overlord-Server/src/server/ws-console-rd-hvnc.ts b/Overlord-Server/src/server/ws-console-rd-hvnc.ts index bf92d00..48a7979 100644 --- a/Overlord-Server/src/server/ws-console-rd-hvnc.ts +++ b/Overlord-Server/src/server/ws-console-rd-hvnc.ts @@ -127,7 +127,10 @@ function getCaptureDllBytes(): Uint8Array | null { return null; } -const VIEWER_BACKPRESSURE_BYTES = 2 * 1024 * 1024; // 2 MB +const VIEWER_BACKPRESSURE_BYTES = Math.max( + 64 * 1024, + Number(process.env.OVERLORD_MEDIA_VIEWER_BACKPRESSURE_BYTES || 512 * 1024), +); type FrameBroadcastResult = { sent: boolean; diff --git a/Overlord-Server/src/sessions/sessionManager.ts b/Overlord-Server/src/sessions/sessionManager.ts index 8c4ff3e..f5f0173 100644 --- a/Overlord-Server/src/sessions/sessionManager.ts +++ b/Overlord-Server/src/sessions/sessionManager.ts @@ -41,6 +41,14 @@ const desktopAudioSessions = new Map(); const desktopAudioSessionsByClient = new Map>(); const dashboardSessions = new Map(); const chatSessions = new Map(); +const VIEWER_BACKPRESSURE_BYTES = Math.max( + 64 * 1024, + Number(process.env.OVERLORD_VIEWER_BACKPRESSURE_BYTES || 2 * 1024 * 1024), +); + +function viewerHasBackpressure(ws: ServerWebSocket): boolean { + return (ws.getBufferedAmount?.() ?? 0) > VIEWER_BACKPRESSURE_BYTES; +} function addSessionToClientIndex( index: Map>, @@ -346,6 +354,7 @@ export function safeSendViewer( payload: any, ): boolean { try { + if (viewerHasBackpressure(ws)) return false; ws.send(JSON.stringify(payload)); return true; } catch (err) { @@ -359,6 +368,7 @@ export function safeSendViewerFrame( header?: any, ): number { try { + if (viewerHasBackpressure(ws)) return 0; const meta = JSON.stringify(header || {}); const metaBytes = new TextEncoder().encode(meta); const metaLength = new Uint8Array(4); @@ -532,6 +542,7 @@ export function notifyDashboardClientEvent( }); for (const [id, session] of dashboardSessions) { try { + if (viewerHasBackpressure(session.viewer)) continue; session.viewer.send(msg); } catch { dashboardSessions.delete(id); @@ -553,6 +564,7 @@ export function notifyDashboardViewers(): void { const msg = JSON.stringify({ type: "clients_changed" }); for (const [id, session] of dashboardSessions) { try { + if (viewerHasBackpressure(session.viewer)) continue; session.viewer.send(msg); } catch { dashboardSessions.delete(id); @@ -580,6 +592,7 @@ export function getChatSessionCount(): number { export function broadcastChatMessage(msg: string): void { for (const [id, session] of chatSessions) { try { + if (viewerHasBackpressure(session.viewer)) continue; session.viewer.send(msg); } catch { chatSessions.delete(id); diff --git a/Overlord-Server/src/wsHandlers.ts b/Overlord-Server/src/wsHandlers.ts index a3018fa..bd27916 100644 --- a/Overlord-Server/src/wsHandlers.ts +++ b/Overlord-Server/src/wsHandlers.ts @@ -16,7 +16,7 @@ import { requestThumbnailRegen, } from "./thumbnails"; import { metrics } from "./metrics"; -import { flushQueuedClientDbUpdates, markClientDbSynced, queueClientDbUpdate, shouldSyncClientToDb } from "./client-db-sync"; +import { flushQueuedClientDbUpdates, queueClientDbUpdate, shouldSyncClientToDb } from "./client-db-sync"; import { setClientDisconnectInfo } from "./db"; export { clearClientSyncState } from "./client-db-sync"; @@ -225,14 +225,15 @@ export function handlePong(info: ClientInfo, payload: WireMessage) { if (rtt >= 0 && rtt < maxRttMs) { info.pingMs = rtt; - queueClientDbUpdate({ - id: info.id, - pingMs: info.pingMs, - lastSeen: info.lastSeen, - online: 1, - isAdmin: info.isAdmin, - }); - markClientDbSynced(info.id, nowTs); + if (shouldSyncClientToDb(info.id, nowTs)) { + queueClientDbUpdate({ + id: info.id, + pingMs: info.pingMs, + lastSeen: info.lastSeen, + online: 1, + isAdmin: info.isAdmin, + }); + } metrics.recordPing(rtt); } else { diff --git a/README.md b/README.md index 9a374ed..d6fb63a 100644 --- a/README.md +++ b/README.md @@ -373,18 +373,41 @@ If you only ever want P2P, you can comment out the `mediamtx:` service in your c ### LAN / public access -By default MediaMTX advertises `127.0.0.1` as an ICE candidate, which is enough for operators running their browser on the same machine as Docker. +MediaMTX automatically discovers addresses assigned directly to its network interfaces. This covers ordinary Linux host-network and many LAN deployments, but it cannot reliably infer a public address created by NAT, port forwarding, a cloud load balancer, or a reverse proxy. It also cannot discover which public DNS name you intend operators to use. -For viewers on a different machine: +Before using **WebRTC Relayed** from another machine, make sure MediaMTX advertises an address that both agents and operators can reach: -- **Linux** (host networking): nothing to configure — MediaMTX sees the host's eth0/wlan0 and gathers those candidates automatically. -- **Windows / macOS** (bridge networking): set `OVERLORD_WEBRTC_ADDITIONAL_HOSTS` to a comma-separated list including the server's reachable LAN or public IP. Either export it in your shell, or add it to a `.env` file next to the compose file: +- **Linux with host networking and no public NAT:** normally nothing is required; MediaMTX sees the host's interfaces automatically. +- **Public server behind NAT, port forwarding, or a cloud public-IP mapping:** set the public IP or a DNS name that resolves to it. +- **Windows / macOS Docker Desktop:** set the host's reachable LAN/public address because automatic interface discovery sees the container network, not necessarily the Docker host. +- **Same-machine testing on Docker Desktop:** retain `127.0.0.1` in the list. - ``` - OVERLORD_WEBRTC_ADDITIONAL_HOSTS=127.0.0.1,192.168.1.42 - ``` +Set `OVERLORD_WEBRTC_ADDITIONAL_HOSTS` as a comma-separated list in the shell or in a `.env` file next to the compose file: - Then `docker compose -f docker-compose.windows.yml up -d mediamtx` to apply. +```env +# LAN example +OVERLORD_WEBRTC_ADDITIONAL_HOSTS=192.168.1.42 + +# Public DNS plus LAN access +OVERLORD_WEBRTC_ADDITIONAL_HOSTS=stream.example.com,192.168.1.42 + +# Docker Desktop: same host plus other LAN machines +OVERLORD_WEBRTC_ADDITIONAL_HOSTS=127.0.0.1,192.168.1.42 +``` + +Ensure inbound UDP `8189` is forwarded to the MediaMTX host/container; TCP `8189` is the slower fallback. Recreate the sidecar after changing the setting: + +```bash +# Linux +docker compose up -d --force-recreate mediamtx + +# Windows / macOS Docker Desktop +docker compose -f docker-compose.windows.yml up -d --force-recreate mediamtx +``` + +To verify the selected route, open `chrome://webrtc-internals` during a relayed stream and confirm that the chosen remote candidate points to the expected server address and preferably uses UDP. + +Automatic public-IP discovery is possible with STUN, but it is not a complete replacement for this setting: it can require random UDP ports, does not discover your intended domain, and can fail behind restrictive or symmetric NAT. For predictable production deployments, explicitly setting the reachable IP/domain and forwarding fixed UDP port `8189` is recommended. ### Advanced MediaMTX customization diff --git a/docker-compose.yml b/docker-compose.yml index 9956a29..8b91204 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -95,12 +95,16 @@ services: # ICE on fixed ports so the compose / host firewall has a single target. MTX_WEBRTCLOCALUDPADDRESS: ":8189" MTX_WEBRTCLOCALTCPADDRESS: ":8189" - # Advertise 127.0.0.1 as an ICE candidate. Always required on Windows / - # macOS (bridge networking); harmless on Linux (host networking). To - # allow viewers on a different machine, override this with a comma- - # separated list including the host's LAN/public IP, e.g.: - # OVERLORD_WEBRTC_ADDITIONAL_HOSTS=127.0.0.1,192.168.1.42 - MTX_WEBRTCADDITIONALHOSTS: ${OVERLORD_WEBRTC_ADDITIONAL_HOSTS:-127.0.0.1} + # Keep the real-time packet queue bounded and use an MTU that remains + # safe across common VPN, IPv6, and tunneled network paths. + MTX_WRITEQUEUESIZE: ${OVERLORD_MEDIAMTX_WRITE_QUEUE_SIZE:-128} + MTX_UDPMAXPAYLOADSIZE: ${OVERLORD_MEDIAMTX_UDP_MAX_PAYLOAD_SIZE:-1200} + MTX_UDPREADBUFFERSIZE: ${OVERLORD_MEDIAMTX_UDP_READ_BUFFER_SIZE:-4194304} + # Leave blank to let MediaMTX advertise detected interfaces. Public-NAT + # deployments must set this to their public IP/DNS; LAN deployments can + # include both LAN and public addresses, for example: + # OVERLORD_WEBRTC_ADDITIONAL_HOSTS=192.168.1.42,stream.example.com + MTX_WEBRTCADDITIONALHOSTS: ${OVERLORD_WEBRTC_ADDITIONAL_HOSTS:-} volumes: overlord-data: diff --git a/scripts/start-prod-tuned.bat b/scripts/start-prod-tuned.bat index 49f7c0a..8a8d37a 100644 --- a/scripts/start-prod-tuned.bat +++ b/scripts/start-prod-tuned.bat @@ -22,6 +22,9 @@ REM OVERLORD_HEARTBEAT_INTERVAL_MS=30000 REM OVERLORD_HEARTBEAT_SWEEP_TICK_MS=1000 REM OVERLORD_CLIENT_DB_SYNC_MS=15000 REM OVERLORD_CLIENT_DB_SYNC_BATCH_SIZE=100 +REM OVERLORD_CLIENT_PRESENCE_SYNC_MS=60000 +REM OVERLORD_VIEWER_BACKPRESSURE_BYTES=2097152 +REM OVERLORD_MEDIA_VIEWER_BACKPRESSURE_BYTES=524288 REM OVERLORD_EVENT_LOOP_SAMPLE_MS=100 REM OVERLORD_EVENT_LOOP_HISTORY_SECONDS=60 REM OVERLORD_DASHBOARD_DEBOUNCE_MS=1000 @@ -46,6 +49,9 @@ if not defined OVERLORD_HEARTBEAT_INTERVAL_MS set OVERLORD_HEARTBEAT_INTERVAL_MS if not defined OVERLORD_HEARTBEAT_SWEEP_TICK_MS set OVERLORD_HEARTBEAT_SWEEP_TICK_MS=1000 if not defined OVERLORD_CLIENT_DB_SYNC_MS set OVERLORD_CLIENT_DB_SYNC_MS=15000 if not defined OVERLORD_CLIENT_DB_SYNC_BATCH_SIZE set OVERLORD_CLIENT_DB_SYNC_BATCH_SIZE=100 +if not defined OVERLORD_CLIENT_PRESENCE_SYNC_MS set OVERLORD_CLIENT_PRESENCE_SYNC_MS=60000 +if not defined OVERLORD_VIEWER_BACKPRESSURE_BYTES set OVERLORD_VIEWER_BACKPRESSURE_BYTES=2097152 +if not defined OVERLORD_MEDIA_VIEWER_BACKPRESSURE_BYTES set OVERLORD_MEDIA_VIEWER_BACKPRESSURE_BYTES=524288 if not defined OVERLORD_EVENT_LOOP_SAMPLE_MS set OVERLORD_EVENT_LOOP_SAMPLE_MS=100 if not defined OVERLORD_EVENT_LOOP_HISTORY_SECONDS set OVERLORD_EVENT_LOOP_HISTORY_SECONDS=60 if not defined OVERLORD_DASHBOARD_DEBOUNCE_MS set OVERLORD_DASHBOARD_DEBOUNCE_MS=1000 @@ -130,6 +136,9 @@ echo OVERLORD_HEARTBEAT_INTERVAL_MS=%OVERLORD_HEARTBEAT_INTERVAL_MS% echo OVERLORD_HEARTBEAT_SWEEP_TICK_MS=%OVERLORD_HEARTBEAT_SWEEP_TICK_MS% echo OVERLORD_CLIENT_DB_SYNC_MS=%OVERLORD_CLIENT_DB_SYNC_MS% echo OVERLORD_CLIENT_DB_SYNC_BATCH_SIZE=%OVERLORD_CLIENT_DB_SYNC_BATCH_SIZE% +echo OVERLORD_CLIENT_PRESENCE_SYNC_MS=%OVERLORD_CLIENT_PRESENCE_SYNC_MS% +echo OVERLORD_VIEWER_BACKPRESSURE_BYTES=%OVERLORD_VIEWER_BACKPRESSURE_BYTES% +echo OVERLORD_MEDIA_VIEWER_BACKPRESSURE_BYTES=%OVERLORD_MEDIA_VIEWER_BACKPRESSURE_BYTES% echo OVERLORD_EVENT_LOOP_SAMPLE_MS=%OVERLORD_EVENT_LOOP_SAMPLE_MS% echo OVERLORD_EVENT_LOOP_HISTORY_SECONDS=%OVERLORD_EVENT_LOOP_HISTORY_SECONDS% echo OVERLORD_DASHBOARD_DEBOUNCE_MS=%OVERLORD_DASHBOARD_DEBOUNCE_MS%