diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d48f63f..50b40fc 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -38,12 +38,19 @@ jobs: dashboard: runs-on: ubuntu-latest + defaults: + run: + working-directory: dashboard steps: - uses: actions/checkout@v4 - uses: actions/setup-node@v4 with: node-version: "20" - name: Install deps - run: cd dashboard && npm install + run: npm install + - name: Lint + run: npm run lint + - name: Test + run: npm test - name: Build - run: cd dashboard && npm run build + run: npm run build diff --git a/README.md b/README.md index a46e378..dbbb9a4 100644 --- a/README.md +++ b/README.md @@ -152,10 +152,10 @@ The full demo stack — engine, control plane, dashboard, Redis, Prometheus, and docker compose up -d --build ``` -Then, to stream a video into the running stack — the file must live under `Doc/dev/`, since that's the directory mounted into the engine container as `/media/`: +Then, to stream a video into the running stack — the file must live under `Resource/media/`, since that's the directory mounted into the engine container as `/media/`: ```bash -bash scripts/start_stream.sh Doc/dev/your-video.mp4 +bash scripts/start_stream.sh Resource/media/your-video.mp4 ``` Open `http://localhost:5173`, enter `127.0.0.1` / `5000` in Session Control, and click **Connect** to watch. Starting a stream and viewing it are deliberately separate steps — Connect only opens the viewer, it does not start a stream. @@ -172,7 +172,7 @@ All scripts are in `scripts/` and assume you're in the repo root. ```bash docker compose up -d --build # start the full stack first -bash scripts/start_stream.sh [/path/to/video.mp4] # defaults to a file under Doc/dev/ +bash scripts/start_stream.sh Resource/media/your-video.mp4 ``` Open `http://localhost:5173`, connect to `127.0.0.1:5000`, then run the script — it waits for services, posts the stream request to the engine's admin API, and confirms packets are actually flowing before exiting (the stream keeps running in the engine independently of the script). diff --git a/dashboard/package.json b/dashboard/package.json index c96e670..1e05c17 100644 --- a/dashboard/package.json +++ b/dashboard/package.json @@ -7,14 +7,18 @@ "dev": "vite", "build": "tsc -b && vite build", "preview": "vite preview", - "lint": "eslint src --ext .ts,.tsx", - "lint:fix": "eslint src --ext .ts,.tsx --fix" + "lint": "eslint src", + "lint:fix": "eslint src --fix", + "test": "vitest run", + "test:watch": "vitest" }, "dependencies": { "react": "^18.3.0", "react-dom": "^18.3.0" }, "devDependencies": { + "@testing-library/jest-dom": "^6.9.1", + "@testing-library/react": "^16.3.3", "@types/react": "^18.3.0", "@types/react-dom": "^18.3.0", "@vitejs/plugin-react": "^4.3.0", @@ -22,10 +26,12 @@ "eslint": "^9.39.5", "eslint-plugin-react": "^7.37.5", "eslint-plugin-react-hooks": "^5.2.0", + "jsdom": "^29.1.1", "postcss": "^8.4.0", "tailwindcss": "^3.4.0", "typescript": "^5.5.0", "typescript-eslint": "^8.68.0", - "vite": "^5.4.0" + "vite": "^5.4.0", + "vitest": "^2.1.9" } } diff --git a/dashboard/src/components/MetricsPanel.test.tsx b/dashboard/src/components/MetricsPanel.test.tsx new file mode 100644 index 0000000..a82c80b --- /dev/null +++ b/dashboard/src/components/MetricsPanel.test.tsx @@ -0,0 +1,70 @@ +import { describe, it, expect } from "vitest"; +import { render, screen } from "@testing-library/react"; +import { MetricsPanel } from "./MetricsPanel"; +import type { StreamMetrics } from "../types"; + +const baseMetrics: StreamMetrics = { + framesReceived: 100, + packetsReceived: 200, + bytesReceived: 1024 * 1024, + currentBitrateMbps: 2.5, + jitterMs: 3, + lossPercent: 0, + rttMs: 20, + streamActive: true, +}; + +describe("MetricsPanel", () => { + it("shows Connected when connected, Disconnected otherwise", () => { + const { rerender } = render(); + expect(screen.getByText("Connected")).toBeInTheDocument(); + + rerender(); + expect(screen.getByText("Disconnected")).toBeInTheDocument(); + }); + + it("shows the no-stream banner only when connected but no stream is active", () => { + render(); + expect(screen.getByText(/No stream running/)).toBeInTheDocument(); + }); + + it("hides the no-stream banner when a stream is active", () => { + render(); + expect(screen.queryByText(/No stream running/)).not.toBeInTheDocument(); + }); + + it("hides the no-stream banner when not connected at all", () => { + // connected=false means "idle", not "stream should be running but isn't" + render(); + expect(screen.queryByText(/No stream running/)).not.toBeInTheDocument(); + }); + + it("shows a network-degradation warning when loss or RTT is high", () => { + render( + , + ); + expect(screen.getByText(/Network degradation detected/)).toBeInTheDocument(); + }); + + it("suppresses the degradation warning while no stream is active", () => { + // Avoids flagging "degraded network" for numbers that are just placeholder zeros/stale. + render( + , + ); + expect(screen.queryByText(/Network degradation detected/)).not.toBeInTheDocument(); + }); + + it("renders counters formatted with locale separators and MB conversion", () => { + render( + , + ); + expect(screen.getByText("12,345")).toBeInTheDocument(); + expect(screen.getByText("2.00 MB")).toBeInTheDocument(); + }); +}); diff --git a/dashboard/src/components/ParticleBackground.tsx b/dashboard/src/components/ParticleBackground.tsx index f69c4c1..fe676d4 100644 --- a/dashboard/src/components/ParticleBackground.tsx +++ b/dashboard/src/components/ParticleBackground.tsx @@ -1,4 +1,4 @@ -import { useEffect, useRef, useState } from "react"; +import { useEffect, useRef } from "react"; import { useTheme } from "../contexts/ThemeContext"; interface Particle { @@ -21,8 +21,6 @@ export function ParticleBackground() { const canvasRef = useRef(null); const mouseRef = useRef({ x: 0, y: 0 }); const particlesRef = useRef([]); - const [fps, setFps] = useState(0); - const frameCountRef = useRef(0); const lastTimeRef = useRef(Date.now()); const { theme } = useTheme(); @@ -112,14 +110,7 @@ export function ParticleBackground() { const delta = now - lastTimeRef.current; if (delta >= 16) { - frameCountRef.current++; - if (delta >= 1000) { - setFps(frameCountRef.current); - frameCountRef.current = 0; - lastTimeRef.current = now; - } else { - lastTimeRef.current += 16; - } + lastTimeRef.current = now; // Background gradient based on theme const isDark = theme === "dark"; @@ -186,12 +177,6 @@ export function ParticleBackground() { } }); }); - - // FPS counter - ctx.globalAlpha = 1; - ctx.fillStyle = isDark ? "rgba(107, 114, 128, 0.6)" : "rgba(107, 114, 128, 0.4)"; - ctx.font = "12px monospace"; - ctx.fillText(`FPS: ${fps}`, 12, 24); } animationFrameId = requestAnimationFrame(animate); @@ -204,7 +189,6 @@ export function ParticleBackground() { window.removeEventListener("resize", resizeCanvas); window.removeEventListener("mousemove", handleMouseMove); }; - // eslint-disable-next-line react-hooks/exhaustive-deps }, [theme]); return ( diff --git a/dashboard/src/hooks/useStream.test.ts b/dashboard/src/hooks/useStream.test.ts new file mode 100644 index 0000000..147567a --- /dev/null +++ b/dashboard/src/hooks/useStream.test.ts @@ -0,0 +1,160 @@ +import { describe, it, expect } from "vitest"; +import { + nalType, + stripStartCode, + toLengthPrefixed, + buildAvcC, + codecFromSps, + readPtsUs, + kTypeVideo, + kTypeAudio, + kFrameHeaderLen, +} from "./useStream"; + +describe("nalType", () => { + it("reads the NAL type after a 4-byte start code", () => { + // type 7 = SPS + const data = new Uint8Array([0x00, 0x00, 0x00, 0x01, 0x67, 0xaa, 0xbb]); + expect(nalType(data)).toBe(7); + }); + + it("reads the NAL type after a 3-byte start code", () => { + // type 8 = PPS + const data = new Uint8Array([0x00, 0x00, 0x01, 0x68, 0xaa]); + expect(nalType(data)).toBe(8); + }); + + it("masks off nal_ref_idc bits, keeping only the 5-bit type", () => { + // 0x65 = 0110 0101 -> nal_ref_idc=011, type=00101=5 (IDR slice) + const data = new Uint8Array([0x00, 0x00, 0x00, 0x01, 0x65]); + expect(nalType(data)).toBe(5); + }); + + it("returns null with no start code", () => { + expect(nalType(new Uint8Array([0x01, 0x02, 0x03, 0x04]))).toBeNull(); + }); + + it("returns null when the start code is the entire buffer", () => { + expect(nalType(new Uint8Array([0x00, 0x00, 0x00, 0x01]))).toBeNull(); + }); +}); + +describe("stripStartCode", () => { + it("strips a 4-byte start code", () => { + const data = new Uint8Array([0x00, 0x00, 0x00, 0x01, 0x67, 0x42]); + expect(stripStartCode(data)).toEqual(new Uint8Array([0x67, 0x42])); + }); + + it("strips a 3-byte start code", () => { + const data = new Uint8Array([0x00, 0x00, 0x01, 0x68, 0x11]); + expect(stripStartCode(data)).toEqual(new Uint8Array([0x68, 0x11])); + }); + + it("returns the input unchanged when there is no start code", () => { + const data = new Uint8Array([0x67, 0x42, 0x00]); + expect(stripStartCode(data)).toEqual(data); + }); +}); + +describe("toLengthPrefixed", () => { + it("prefixes the NAL with its 4-byte big-endian length", () => { + const nal = new Uint8Array([0x65, 0x01, 0x02, 0x03]); + const out = toLengthPrefixed(nal); + expect(out).toEqual(new Uint8Array([0x00, 0x00, 0x00, 0x04, 0x65, 0x01, 0x02, 0x03])); + }); + + it("handles an empty NAL", () => { + expect(toLengthPrefixed(new Uint8Array([]))).toEqual(new Uint8Array([0, 0, 0, 0])); + }); + + it("handles lengths spanning multiple length bytes", () => { + const nal = new Uint8Array(300).fill(0xab); + const out = toLengthPrefixed(nal); + // 300 = 0x012C + expect(out.slice(0, 4)).toEqual(new Uint8Array([0x00, 0x00, 0x01, 0x2c])); + expect(out.length).toBe(304); + }); +}); + +describe("buildAvcC", () => { + // Regression test: buildAvcC once stripped the SPS/PPS NAL header byte, + // corrupting decoded width/height — avcC stores the full NAL (ISO 14496-15). + it("embeds the full SPS/PPS NAL unit, header byte included", () => { + const sps = new Uint8Array([0x67, 0x42, 0x00, 0x1e, 0xaa, 0xbb]); + const pps = new Uint8Array([0x68, 0xce, 0x3c, 0x80]); + const avcC = buildAvcC(sps, pps); + + // 11 fixed bytes + full sps + full pps (not sps.length - 1 / pps.length - 1) + expect(avcC.length).toBe(11 + sps.length + pps.length); + + expect(avcC[0]).toBe(1); // configurationVersion + expect(avcC[1]).toBe(sps[1]); // AVCProfileIndication + expect(avcC[2]).toBe(sps[2]); // profile_compatibility + expect(avcC[3]).toBe(sps[3]); // AVCLevelIndication + expect(avcC[4]).toBe(0xff); // lengthSizeMinusOne=3 (4-byte lengths) + expect(avcC[5]).toBe(0xe1); // numOfSequenceParameterSets=1 + + const spsLen = (avcC[6] << 8) | avcC[7]; + expect(spsLen).toBe(sps.length); + const embeddedSps = avcC.slice(8, 8 + spsLen); + expect(embeddedSps).toEqual(sps); // full NAL, header byte (0x67) included + + let o = 8 + spsLen; + expect(avcC[o]).toBe(1); // numOfPictureParameterSets + o += 1; + const ppsLen = (avcC[o] << 8) | avcC[o + 1]; + expect(ppsLen).toBe(pps.length); + o += 2; + const embeddedPps = avcC.slice(o, o + ppsLen); + expect(embeddedPps).toEqual(pps); // full NAL, header byte (0x68) included + }); +}); + +describe("codecFromSps", () => { + it("builds an avc1.PPCCLL codec string from profile/compat/level bytes", () => { + const sps = new Uint8Array([0x67, 0x42, 0x00, 0x1e]); + expect(codecFromSps(sps)).toBe("avc1.42001e"); + }); + + it("pads single hex digits with a leading zero", () => { + const sps = new Uint8Array([0x67, 0x0a, 0x00, 0x09]); + expect(codecFromSps(sps)).toBe("avc1.0a0009"); + }); +}); + +describe("readPtsUs", () => { + // Wire format [type][pts_us][payload]: pts_us is 8-byte big-endian at offset 1. + it("reads an 8-byte big-endian pts_us starting at offset 1", () => { + const data = new Uint8Array(kFrameHeaderLen + 2); + data[0] = kTypeVideo; + // pts_us = 0x0000000000033333 (209715 decimal, arbitrary test value) + const view = new DataView(data.buffer); + view.setBigUint64(1, 0x33333n, false); + expect(readPtsUs(data)).toBe(0x33333); + }); + + it("round-trips values the control plane actually sends (pts_us up to ~a few minutes)", () => { + const ptsUs = 176_664_900; // observed real value, ~176s into a clip + const data = new Uint8Array(kFrameHeaderLen); + data[0] = kTypeAudio; + new DataView(data.buffer).setBigUint64(1, BigInt(ptsUs), false); + expect(readPtsUs(data)).toBe(ptsUs); + }); + + it("reads 0 correctly", () => { + const data = new Uint8Array(kFrameHeaderLen); + expect(readPtsUs(data)).toBe(0); + }); +}); + +describe("wire format type markers", () => { + it("uses distinct, expected byte values for video and audio", () => { + expect(kTypeVideo).toBe(0x00); + expect(kTypeAudio).toBe(0x01); + expect(kTypeVideo).not.toBe(kTypeAudio); + }); + + it("header length accounts for 1 type byte + 8 pts bytes", () => { + expect(kFrameHeaderLen).toBe(9); + }); +}); diff --git a/dashboard/src/hooks/useStream.ts b/dashboard/src/hooks/useStream.ts index cc95dec..8572039 100644 --- a/dashboard/src/hooks/useStream.ts +++ b/dashboard/src/hooks/useStream.ts @@ -2,7 +2,7 @@ import { useEffect, useRef } from "react"; const WS_BASE = import.meta.env.VITE_WS_BASE || "ws://localhost:8000/ws"; -function nalType(data: Uint8Array): number | null { +export function nalType(data: Uint8Array): number | null { let i = 0; if (data.length >= 4 && data[0] === 0 && data[1] === 0 && data[2] === 0 && data[3] === 1) { i = 4; @@ -15,7 +15,7 @@ function nalType(data: Uint8Array): number | null { return data[i] & 0x1f; } -function stripStartCode(data: Uint8Array): Uint8Array { +export function stripStartCode(data: Uint8Array): Uint8Array { if (data.length >= 4 && data[0] === 0 && data[1] === 0 && data[2] === 0 && data[3] === 1) { return data.subarray(4); } @@ -25,7 +25,7 @@ function stripStartCode(data: Uint8Array): Uint8Array { return data; } -function toLengthPrefixed(nal: Uint8Array): Uint8Array { +export function toLengthPrefixed(nal: Uint8Array): Uint8Array { const out = new Uint8Array(4 + nal.length); const len = nal.length; out[0] = (len >>> 24) & 0xff; @@ -36,7 +36,7 @@ function toLengthPrefixed(nal: Uint8Array): Uint8Array { return out; } -function buildAvcC(sps: Uint8Array, pps: Uint8Array): Uint8Array { +export function buildAvcC(sps: Uint8Array, pps: Uint8Array): Uint8Array { // avcC box (ISO/IEC 14496-15): SPS/PPS entries store the complete NAL unit, // header byte included — not just the RBSP. const out = new Uint8Array(11 + sps.length + pps.length); @@ -58,7 +58,7 @@ function buildAvcC(sps: Uint8Array, pps: Uint8Array): Uint8Array { return out; } -function codecFromSps(sps: Uint8Array): string { +export function codecFromSps(sps: Uint8Array): string { const profile = sps[1].toString(16).padStart(2, "0"); const compat = sps[2].toString(16).padStart(2, "0"); const level = sps[3].toString(16).padStart(2, "0"); @@ -66,11 +66,11 @@ function codecFromSps(sps: Uint8Array): string { } // Wire format: [1B type][8B pts_us big-endian][payload]. -const kTypeVideo = 0x00; -const kTypeAudio = 0x01; -const kFrameHeaderLen = 9; +export const kTypeVideo = 0x00; +export const kTypeAudio = 0x01; +export const kFrameHeaderLen = 9; -function readPtsUs(data: Uint8Array): number { +export function readPtsUs(data: Uint8Array): number { const view = new DataView(data.buffer, data.byteOffset + 1, 8); return Number(view.getBigUint64(0, false)); } @@ -120,6 +120,11 @@ export function useStream( let sps: Uint8Array | null = null; let pps: Uint8Array | null = null; let needKey = true; + // pts is strictly increasing from the source; anything at or before what + // we've already decoded is a stale resend (see the engine's keyframe + // resend below), not new content — splicing it back in would break the + // H.264 reference chain and cause visible macroblocking. + let highestDecodedPtsUs = -1; let cancelled = false; let pingTimer: number | undefined; let statusTimer: number | undefined; @@ -138,6 +143,13 @@ export function useStream( let rafHandle: number | null = null; const kMaxQueuedFrames = 60; + // Engine/bridge bulk-drains the whole AudioBuffer on first poll — keep + // encoded packets and only decode/schedule near the video playhead so we + // can PTS-align instead of chaining ASAP on a suspended AudioContext. + const audioPacketQueue: { ptsUs: number; data: Uint8Array }[] = []; + const kAudioScheduleAheadSec = 0.75; + const kAudioLateSlackSec = 0.05; + const drawFrame = (frame: VideoFrame) => { if (canvasRef.current) { const ctx = canvasRef.current.getContext("2d"); @@ -149,6 +161,81 @@ export function useStream( frame.close(); }; + const videoMediaSec = (): number | null => { + if (renderAnchorPerfMs == null || renderAnchorPtsMs == null) return null; + return (renderAnchorPtsMs + (performance.now() - renderAnchorPerfMs)) / 1000; + }; + + const scheduleAudio = (audioData: AudioData) => { + if (!audioCtx || !gainNode) { + audioData.close(); + return; + } + const vMedia = videoMediaSec(); + if (vMedia == null || audioCtx.state !== "running") { + audioData.close(); + return; + } + try { + const { numberOfChannels, numberOfFrames, sampleRate } = audioData; + const ptsSec = audioData.timestamp / 1_000_000; + if (ptsSec < vMedia - kAudioLateSlackSec) { + audioData.close(); + return; + } + const buffer = audioCtx.createBuffer(numberOfChannels, numberOfFrames, sampleRate); + const channelData = new Float32Array(numberOfFrames); + for (let ch = 0; ch < numberOfChannels; ch++) { + audioData.copyTo(channelData, { planeIndex: ch, format: "f32-planar" }); + buffer.copyToChannel(channelData, ch); + } + const source = audioCtx.createBufferSource(); + source.buffer = buffer; + source.connect(gainNode); + // Align this chunk's media pts to the same clock the video render loop uses. + const startAt = Math.max( + audioCtx.currentTime + 0.01, + audioCtx.currentTime + (ptsSec - vMedia), + nextAudioTime, + ); + source.start(startAt); + nextAudioTime = startAt + buffer.duration; + } catch (err) { + console.error("audio schedule failed:", err); + } finally { + audioData.close(); + } + }; + + const pumpAudio = () => { + if (!audioDecoder || audioDecoder.state !== "configured" || !audioCtx) return; + if (audioCtx.state === "suspended") { + audioCtx.resume().catch(() => {}); + return; + } + if (audioCtx.state !== "running") return; + const vMedia = videoMediaSec(); + if (vMedia == null) return; + + while (audioPacketQueue.length > 0) { + const pkt = audioPacketQueue[0]; + const ptsSec = pkt.ptsUs / 1_000_000; + if (ptsSec < vMedia - kAudioLateSlackSec) { + audioPacketQueue.shift(); + continue; + } + if (ptsSec > vMedia + kAudioScheduleAheadSec) break; + audioPacketQueue.shift(); + try { + audioDecoder.decode( + new EncodedAudioChunk({ type: "key", timestamp: pkt.ptsUs, data: pkt.data }), + ); + } catch (err) { + console.error("audio decode failed:", err); + } + } + }; + const renderLoop = () => { if (cancelled) return; const nowMs = performance.now(); @@ -164,7 +251,18 @@ export function useStream( while (renderQueue.length > 0) { const framePtsMs = renderQueue[0].timestamp / 1000; - const dueAt = renderAnchorPerfMs! + (framePtsMs - renderAnchorPtsMs!); + let dueAt = renderAnchorPerfMs! + (framePtsMs - renderAnchorPtsMs!); + + // A pts far ahead of schedule (e.g. jumping from a resent stale + // keyframe to the live buffer's position) would idle out the gap — + // resync to "now" instead, the gap was never real playback time. + const kMaxScheduleAheadMs = 1000; + if (dueAt > nowMs + kMaxScheduleAheadMs) { + renderAnchorPerfMs = nowMs; + renderAnchorPtsMs = framePtsMs; + dueAt = nowMs; + } + if (nowMs < dueAt) break; const frame = renderQueue.shift()!; @@ -180,9 +278,9 @@ export function useStream( drawFrame(frame); } + pumpAudio(); rafHandle = requestAnimationFrame(renderLoop); }; - rafHandle = requestAnimationFrame(renderLoop); const closeDecoder = () => { try { @@ -193,6 +291,7 @@ export function useStream( decoder = null; configured = false; needKey = true; + highestDecodedPtsUs = -1; while (renderQueue.length > 0) { renderQueue.shift()!.close(); } @@ -210,32 +309,6 @@ export function useStream( audioConfigured = false; }; - const scheduleAudio = (audioData: AudioData) => { - if (!audioCtx || !gainNode) { - audioData.close(); - return; - } - try { - const { numberOfChannels, numberOfFrames, sampleRate } = audioData; - const buffer = audioCtx.createBuffer(numberOfChannels, numberOfFrames, sampleRate); - const channelData = new Float32Array(numberOfFrames); - for (let ch = 0; ch < numberOfChannels; ch++) { - audioData.copyTo(channelData, { planeIndex: ch, format: "f32-planar" }); - buffer.copyToChannel(channelData, ch); - } - const source = audioCtx.createBufferSource(); - source.buffer = buffer; - source.connect(gainNode); - const startAt = Math.max(audioCtx.currentTime, nextAudioTime); - source.start(startAt); - nextAudioTime = startAt + buffer.duration; - } catch (err) { - console.error("audio schedule failed:", err); - } finally { - audioData.close(); - } - }; - const ensureAudioDecoder = () => { if (typeof AudioDecoder === "undefined") return; if (audioConfigured && audioDecoder && audioDecoder.state === "configured") return; @@ -327,6 +400,7 @@ export function useStream( } }; + rafHandle = requestAnimationFrame(renderLoop); drawStatus(canvas, "Connecting to stream…"); try { @@ -374,15 +448,10 @@ export function useStream( // Confirm the payload is really an ADTS frame (sync word 0xFFF) before decoding. if (data.length < 2 || data[0] !== 0xff || (data[1] & 0xf0) !== 0xf0) return; ensureAudioDecoder(); - if (audioDecoder && audioDecoder.state === "configured") { - try { - audioDecoder.decode( - new EncodedAudioChunk({ type: "key", timestamp: ptsUs, data }), - ); - } catch (err) { - console.error("audio decode failed:", err); - } - } + // Copy — the WS buffer is reused. Queue until the video clock exists, + // then pump near the playhead (see pumpAudio). + audioPacketQueue.push({ ptsUs, data: data.slice() }); + pumpAudio(); return; } if (msgType !== kTypeVideo) return; @@ -413,7 +482,24 @@ export function useStream( if (needKey && type !== 5) { return; } - if (type === 5) needKey = false; + if (ptsUs <= highestDecodedPtsUs) return; + + // Detect a *timeline discontinuity* (missing NALs / stale cached keyframe + // jumped to live). Even one skipped reference frame causes macroblocking. + // ~100ms ≈ 3 frames at 30fps — tolerates jitter, rejects real gaps. + // (Do NOT measure against last keyframe: real GOPs are multi-second.) + const kMaxDecodeGapUs = 100_000; + if ( + type !== 5 && + highestDecodedPtsUs >= 0 && + ptsUs - highestDecodedPtsUs > kMaxDecodeGapUs + ) { + needKey = true; + return; + } + if (type === 5) { + needKey = false; + } try { // avcC declares 4-byte length-prefixed NALs, not Annex-B start codes. @@ -423,6 +509,7 @@ export function useStream( data: toLengthPrefixed(nal), }); decoder.decode(chunk); + highestDecodedPtsUs = ptsUs; } catch (err) { console.error("decode failed:", err); } @@ -444,6 +531,7 @@ export function useStream( websocket?.close(); closeDecoder(); closeAudioDecoder(); + audioPacketQueue.length = 0; try { audioCtx?.close(); } catch { diff --git a/dashboard/src/test/setup.ts b/dashboard/src/test/setup.ts new file mode 100644 index 0000000..f149f27 --- /dev/null +++ b/dashboard/src/test/setup.ts @@ -0,0 +1 @@ +import "@testing-library/jest-dom/vitest"; diff --git a/dashboard/vite.config.ts b/dashboard/vite.config.ts index d02d590..0dde8f8 100644 --- a/dashboard/vite.config.ts +++ b/dashboard/vite.config.ts @@ -1,3 +1,4 @@ +/// import { defineConfig } from "vite"; import react from "@vitejs/plugin-react"; @@ -10,4 +11,11 @@ export default defineConfig({ "/ws": { target: "ws://localhost:8000", ws: true }, }, }, + test: { + environment: "jsdom", + // Needed for @testing-library/react's auto-cleanup, which registers via + // a global afterEach and silently no-ops without it. + globals: true, + setupFiles: ["./src/test/setup.ts"], + }, }); diff --git a/docker-compose.yml b/docker-compose.yml index e5a2be7..f941bee 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -33,7 +33,7 @@ services: - "5001:5001/tcp" - "9100:9100" volumes: - - ./Doc/dev:/media:ro + - ./Resource/media:/media:ro healthcheck: test: ["CMD", "curl", "-f", "http://localhost:5001/admin/health"] interval: 10s diff --git a/engine/include/socketcast/transport.hpp b/engine/include/socketcast/transport.hpp index 530c16c..46bb2db 100644 --- a/engine/include/socketcast/transport.hpp +++ b/engine/include/socketcast/transport.hpp @@ -7,6 +7,7 @@ #include "socketcast/session.hpp" #include +#include #include #include #include @@ -88,6 +89,11 @@ class Transport { std::mutex param_sets_mutex_; std::vector cached_sps_; std::vector cached_pps_; + // Last keyframe, resent on a timer (not every poll — it's tens of KB) + // so a late joiner doesn't wait a full GOP for anything decodable. + std::vector cached_keyframe_; + uint64_t cached_keyframe_ts_us_{0}; + std::chrono::steady_clock::time_point last_keyframe_resend_{}; // Frame accumulation: flush when a new Annex B start code begins a NAL std::vector frame_accumulator_; diff --git a/engine/src/media_source.cpp b/engine/src/media_source.cpp index 119efcc..cbda8a6 100644 --- a/engine/src/media_source.cpp +++ b/engine/src/media_source.cpp @@ -5,6 +5,7 @@ #include #include #include +#include #include namespace socketcast { @@ -75,9 +76,20 @@ bool MediaSource::load_via_ffmpeg() { // No -tune zerolatency: it enables x264 sliced-threads, splitting each // picture into multiple VCL NALs — but this class assumes one NAL is one // picture (pts, per-NAL flushing), so slicing corrupted pts and images. + // + // -r forces CFR at fps_, matching the pts math below (pts = frame_index_ + // * 1e6/fps_). Without it, ffmpeg keeps the source's native rate, which + // silently desyncs pts from real time whenever they differ (e.g. a 25fps + // source assumed to be 30fps plays ~20% too fast). + // + // -g/-keyint_min: shorter GOPs so a lost P-frame (or late-joiner wait) + // recovers in ~2s instead of x264's default ~8s. const std::string cmd = "ffmpeg -loglevel error -y -i \"" + path_ + - "\" -an -c:v libx264 -preset ultrafast -f h264 pipe:1"; + "\" -an -r " + std::to_string(fps_) + + " -c:v libx264 -preset ultrafast -g " + std::to_string(fps_ * 2) + + " -keyint_min " + std::to_string(fps_) + + " -f h264 pipe:1"; FILE* pipe = popen(cmd.c_str(), "r"); if (pipe == nullptr) { return false; diff --git a/engine/src/transport.cpp b/engine/src/transport.cpp index 599615a..b325c19 100644 --- a/engine/src/transport.cpp +++ b/engine/src/transport.cpp @@ -460,6 +460,9 @@ std::string Transport::on_start_stream(const StreamRequest& req) { std::lock_guard lock(param_sets_mutex_); cached_sps_.clear(); cached_pps_.clear(); + cached_keyframe_.clear(); + cached_keyframe_ts_us_ = 0; + last_keyframe_resend_ = {}; } frame_accumulator_.clear(); if (frame_buffer_) { @@ -505,10 +508,13 @@ std::string Transport::on_get_frames() { bool first = true; // Always prepend cached SPS/PPS so a late-joining browser can configure - // WebCodecs even if it connects mid-GOP. + // WebCodecs mid-GOP. Also periodically re-prepend the last keyframe — + // without it a fresh viewer sees only undecodable P-frames until the next + // natural keyframe (~8.3s away by default), so video stays black that + // whole time. Resent on a timer, not every poll, since it's tens of KB. { std::lock_guard lock(param_sets_mutex_); - auto emit = [&](const std::vector& nal, bool key) { + auto emit = [&](const std::vector& nal, bool key, uint64_t ts_us) { if (nal.empty()) { return; } @@ -516,12 +522,20 @@ std::string Transport::on_get_frames() { oss << ","; } first = false; - oss << R"({"timestamp_us":0,"is_keyframe":)" << (key ? "true" : "false") + oss << R"({"timestamp_us":)" << ts_us << R"(,"is_keyframe":)" << (key ? "true" : "false") << R"(,"data_base64":")" << base64_encode(nal) << R"("})"; }; if (!frames.empty()) { - emit(cached_sps_, false); - emit(cached_pps_, false); + emit(cached_sps_, false, 0); + emit(cached_pps_, false, 0); + + constexpr auto kKeyframeResendInterval = std::chrono::milliseconds(500); + const auto now = std::chrono::steady_clock::now(); + if (!cached_keyframe_.empty() && + now - last_keyframe_resend_ >= kKeyframeResendInterval) { + emit(cached_keyframe_, true, cached_keyframe_ts_us_); + last_keyframe_resend_ = now; + } } } @@ -580,6 +594,11 @@ void Transport::cache_parameter_set(const std::vector& annex_b) { void Transport::push_captured_frame(const std::vector& frame_data, uint64_t ts_us, bool is_keyframe) { cache_parameter_set(frame_data); + if (is_keyframe) { + std::lock_guard lock(param_sets_mutex_); + cached_keyframe_ = frame_data; + cached_keyframe_ts_us_ = ts_us; + } frame_buffer_->push_frame(frame_data, ts_us, is_keyframe); } diff --git a/scripts/start_stream.sh b/scripts/start_stream.sh index a71b77a..e5b1499 100755 --- a/scripts/start_stream.sh +++ b/scripts/start_stream.sh @@ -1,9 +1,20 @@ #!/bin/bash -# Start a video stream into the SocketCast engine (Docker: file must be under Doc/dev → /media). +# Start a video stream into the SocketCast engine (Docker: file must be under Resource/media → /media). set -euo pipefail ROOT="$(cd "$(dirname "$0")/.." && pwd)" -VIDEO="${1:-$ROOT/Doc/dev/Rick-Astley-Never-Gonna-Give-You-Up-Official-Music-Video.mp4}" + +if [[ -n "${1:-}" ]]; then + VIDEO="$1" +else + read -rp "Path to video file (under Resource/media/): " VIDEO +fi + +if [[ -z "$VIDEO" ]]; then + echo "✗ No path given." + exit 1 +fi + VIDEO_PATH="/media/$(basename "$VIDEO")" echo "SocketCast Stream Starter" @@ -14,7 +25,7 @@ echo "" if [[ ! -f "$VIDEO" ]]; then echo "✗ Video not found on host: $VIDEO" - echo " Place the file under Doc/dev/ (mounted into the engine as /media/)." + echo " Place the file under Resource/media/ (mounted into the engine as /media/)." exit 1 fi