Release video frames against the audio clock
With the codec fixed the picture stopped stuttering but ran ahead of the sound, and for a structural reason: the audio takes a longer road to the call — our PCM goes through revoice's own ffmpeg and its buffer — while frames went out the moment they were decoded. Each frame now carries its position in the stream and waits until the audio that belongs with it has actually played, using the player's own playback position as the shared clock. Frames that fall more than 250 ms behind are dropped rather than shown late, so the picture recovers by itself instead of accumulating drift. The queue is bounded by bytes and never by pausing the stream: one ffmpeg feeds both outputs, so blocking the video pipe would stop the audio whose clock we are waiting on — a deadlock. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Leonid Pershin
co-authored by
Claude Opus 5
Claude Opus 5
parent
ec50e4c4df
commit
bc567e953e
+9
-1
@@ -379,7 +379,15 @@ export class GuildPlayer extends EventEmitter<GuildPlayerEvents> {
|
||||
private async startScreenShare(video: NonNullable<PlaybackInput["video"]>): Promise<void> {
|
||||
const room = this.connection?.room;
|
||||
if (!room) return;
|
||||
const publisher = new VideoPublisher({ width: video.width, height: video.height });
|
||||
const publisher = new VideoPublisher({
|
||||
width: video.width,
|
||||
height: video.height,
|
||||
fps: video.fps,
|
||||
// Frames are released against the sound that has actually played, which is
|
||||
// the only clock both sides share: the audio path runs through revoice's
|
||||
// own ffmpeg and buffer, so it always trails the raw frames.
|
||||
audioClock: () => this.media?.seconds ?? 0,
|
||||
});
|
||||
try {
|
||||
await publisher.start(room, video.stream, (reason) => {
|
||||
this.notify(`📺 Видео отключено: ${reason}.`);
|
||||
|
||||
+218
-143
@@ -1,143 +1,218 @@
|
||||
import { createRequire } from "node:module";
|
||||
import type { Readable } from "node:stream";
|
||||
import type { LocalVideoTrack, Room, VideoSource } from "@livekit/rtc-node";
|
||||
import { config } from "../config.js";
|
||||
import { logger } from "../logger.js";
|
||||
|
||||
// revoice.js pulls in the CommonJS build of @livekit/rtc-node, and the native
|
||||
// FFI client is per-module-instance: importing the ESM build here would give us
|
||||
// a second instance whose tracks the room (created by revoice) cannot publish.
|
||||
// So we require the very same module object it uses.
|
||||
const require = createRequire(import.meta.url);
|
||||
const rtc = require("@livekit/rtc-node") as typeof import("@livekit/rtc-node");
|
||||
|
||||
const log = logger.child({ mod: "video" });
|
||||
/** Give up on the video track if the source never produced a picture. */
|
||||
const FIRST_FRAME_TIMEOUT_MS = 15_000;
|
||||
|
||||
export interface VideoPublisherOptions {
|
||||
width: number;
|
||||
height: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* Publishes a raw I420 stream into a LiveKit room as a screen share. ffmpeg
|
||||
* feeds frames at playback speed, so this class only has to slice the byte
|
||||
* stream into frames and hand them over.
|
||||
*/
|
||||
export class VideoPublisher {
|
||||
private readonly frameSize: number;
|
||||
private source: VideoSource | null = null;
|
||||
private track: LocalVideoTrack | null = null;
|
||||
private trackSid: string | null = null;
|
||||
private room: Room | null = null;
|
||||
private stream: Readable | null = null;
|
||||
private pending: Buffer[] = [];
|
||||
private pendingBytes = 0;
|
||||
private frames = 0;
|
||||
private firstFrameTimer: NodeJS.Timeout | null = null;
|
||||
|
||||
constructor(private readonly options: VideoPublisherOptions) {
|
||||
// I420: one luma plane plus two half-resolution chroma planes.
|
||||
this.frameSize = Math.floor((options.width * options.height * 3) / 2);
|
||||
}
|
||||
|
||||
get isPublishing(): boolean {
|
||||
return this.trackSid !== null;
|
||||
}
|
||||
|
||||
/** Publishes the track and starts pumping frames. Resolves once published. */
|
||||
async start(room: Room, stream: Readable, onGiveUp?: (reason: string) => void): Promise<void> {
|
||||
const { width, height } = this.options;
|
||||
const source = new rtc.VideoSource(width, height);
|
||||
// As a camera the picture appears inside the bot's own tile; a screen share
|
||||
// is rendered as a separate tile by every client.
|
||||
const asScreen = config.VIDEO_SOURCE === "screen";
|
||||
const track = rtc.LocalVideoTrack.createVideoTrack(asScreen ? "screen" : "video", source);
|
||||
|
||||
const options = new rtc.TrackPublishOptions();
|
||||
options.source = asScreen ? rtc.TrackSource.SOURCE_SCREENSHARE : rtc.TrackSource.SOURCE_CAMERA;
|
||||
|
||||
const participant = room.localParticipant;
|
||||
if (!participant) throw new Error("room has no local participant yet");
|
||||
const publication = await participant.publishTrack(track, options);
|
||||
|
||||
this.room = room;
|
||||
this.source = source;
|
||||
this.track = track;
|
||||
this.trackSid = publication.sid ?? null;
|
||||
this.stream = stream;
|
||||
|
||||
stream.on("data", (chunk: Buffer) => this.consume(chunk));
|
||||
stream.once("error", (err) => log.debug({ err }, "video stream error"));
|
||||
|
||||
this.firstFrameTimer = setTimeout(() => {
|
||||
if (this.frames > 0) return;
|
||||
log.warn("no video frames arrived, dropping the screen share");
|
||||
onGiveUp?.("источник не отдал видео");
|
||||
void this.stop();
|
||||
}, FIRST_FRAME_TIMEOUT_MS);
|
||||
this.firstFrameTimer.unref?.();
|
||||
|
||||
log.info(
|
||||
{ width, height, source: config.VIDEO_SOURCE, sid: publication.sid },
|
||||
"video track published",
|
||||
);
|
||||
}
|
||||
|
||||
private consume(chunk: Buffer): void {
|
||||
if (!this.source) return;
|
||||
this.pending.push(chunk);
|
||||
this.pendingBytes += chunk.length;
|
||||
if (this.pendingBytes < this.frameSize) return;
|
||||
|
||||
let buffer = this.pending.length === 1 ? (this.pending[0] as Buffer) : Buffer.concat(this.pending, this.pendingBytes);
|
||||
while (buffer.length >= this.frameSize) {
|
||||
const frame = buffer.subarray(0, this.frameSize);
|
||||
try {
|
||||
this.source.captureFrame(
|
||||
new rtc.VideoFrame(frame, this.options.width, this.options.height, rtc.VideoBufferType.I420),
|
||||
);
|
||||
this.frames += 1;
|
||||
} catch (err) {
|
||||
log.debug({ err }, "captureFrame failed");
|
||||
}
|
||||
buffer = buffer.subarray(this.frameSize);
|
||||
}
|
||||
// Keep only the tail of a partial frame for the next chunk.
|
||||
this.pending = buffer.length > 0 ? [Buffer.from(buffer)] : [];
|
||||
this.pendingBytes = buffer.length;
|
||||
}
|
||||
|
||||
async stop(): Promise<void> {
|
||||
if (this.firstFrameTimer) {
|
||||
clearTimeout(this.firstFrameTimer);
|
||||
this.firstFrameTimer = null;
|
||||
}
|
||||
this.stream?.removeAllListeners("data");
|
||||
this.stream = null;
|
||||
this.pending = [];
|
||||
this.pendingBytes = 0;
|
||||
|
||||
const { room, trackSid } = this;
|
||||
this.trackSid = null;
|
||||
if (room?.localParticipant && trackSid) {
|
||||
try {
|
||||
await room.localParticipant.unpublishTrack(trackSid, true);
|
||||
} catch (err) {
|
||||
log.debug({ err }, "unpublishTrack failed");
|
||||
}
|
||||
}
|
||||
|
||||
try {
|
||||
this.track?.close?.();
|
||||
await this.source?.close?.();
|
||||
} catch (err) {
|
||||
log.debug({ err }, "closing video track failed");
|
||||
}
|
||||
this.track = null;
|
||||
this.source = null;
|
||||
this.room = null;
|
||||
this.frames = 0;
|
||||
}
|
||||
}
|
||||
import { createRequire } from "node:module";
|
||||
import type { Readable } from "node:stream";
|
||||
import type { LocalVideoTrack, Room, VideoSource } from "@livekit/rtc-node";
|
||||
import { config } from "../config.js";
|
||||
import { logger } from "../logger.js";
|
||||
|
||||
// revoice.js pulls in the CommonJS build of @livekit/rtc-node, and the native
|
||||
// FFI client is per-module-instance: importing the ESM build here would give us
|
||||
// a second instance whose tracks the room (created by revoice) cannot publish.
|
||||
// So we require the very same module object it uses.
|
||||
const require = createRequire(import.meta.url);
|
||||
const rtc = require("@livekit/rtc-node") as typeof import("@livekit/rtc-node");
|
||||
|
||||
const log = logger.child({ mod: "video" });
|
||||
/** Give up on the video track if the source never produced a picture. */
|
||||
const FIRST_FRAME_TIMEOUT_MS = 15_000;
|
||||
/** How far behind the sound a frame may be before it is dropped instead of shown. */
|
||||
const MAX_LATE_SECONDS = 0.25;
|
||||
/**
|
||||
* Frames wait in memory until the sound catches up, so the queue is bounded by
|
||||
* bytes rather than count — 720p frames are four times the size of 360p ones.
|
||||
* Never by pausing the stream: one ffmpeg feeds both outputs, so blocking the
|
||||
* video pipe would also stop the audio whose clock we are waiting for.
|
||||
*/
|
||||
const MAX_QUEUE_BYTES = 48 * 1024 * 1024;
|
||||
|
||||
export interface VideoPublisherOptions {
|
||||
width: number;
|
||||
height: number;
|
||||
fps: number;
|
||||
/**
|
||||
* Seconds of audio actually played out so far. Frames are released against
|
||||
* this clock, which is what keeps picture and sound together.
|
||||
*/
|
||||
audioClock(): number;
|
||||
}
|
||||
|
||||
interface QueuedFrame {
|
||||
/** Position in the stream, so a dropped frame does not shift the rest. */
|
||||
index: number;
|
||||
data: Buffer;
|
||||
}
|
||||
|
||||
/**
|
||||
* Publishes a raw I420 stream into a LiveKit room.
|
||||
*
|
||||
* Frames are not handed over as they arrive: the audio takes a longer road to
|
||||
* the call (through revoice's own ffmpeg and its buffer), so a frame released on
|
||||
* arrival runs ahead of the sound. Instead every frame carries its position in
|
||||
* the stream and waits until the audio clock reaches it.
|
||||
*/
|
||||
export class VideoPublisher {
|
||||
private readonly frameSize: number;
|
||||
private source: VideoSource | null = null;
|
||||
private track: LocalVideoTrack | null = null;
|
||||
private trackSid: string | null = null;
|
||||
private room: Room | null = null;
|
||||
private stream: Readable | null = null;
|
||||
private partial: Buffer[] = [];
|
||||
private partialBytes = 0;
|
||||
private queue: QueuedFrame[] = [];
|
||||
private produced = 0;
|
||||
private published = 0;
|
||||
private dropped = 0;
|
||||
private pump: NodeJS.Timeout | null = null;
|
||||
private firstFrameTimer: NodeJS.Timeout | null = null;
|
||||
|
||||
private readonly maxQueuedFrames: number;
|
||||
|
||||
constructor(private readonly options: VideoPublisherOptions) {
|
||||
// I420: one luma plane plus two half-resolution chroma planes.
|
||||
this.frameSize = Math.floor((options.width * options.height * 3) / 2);
|
||||
this.maxQueuedFrames = Math.max(12, Math.floor(MAX_QUEUE_BYTES / this.frameSize));
|
||||
}
|
||||
|
||||
get isPublishing(): boolean {
|
||||
return this.trackSid !== null;
|
||||
}
|
||||
|
||||
/** Publishes the track and starts pumping frames. Resolves once published. */
|
||||
async start(room: Room, stream: Readable, onGiveUp?: (reason: string) => void): Promise<void> {
|
||||
const { width, height } = this.options;
|
||||
const source = new rtc.VideoSource(width, height);
|
||||
// As a camera the picture appears inside the bot's own tile; a screen share
|
||||
// is rendered as a separate tile by every client.
|
||||
const asScreen = config.VIDEO_SOURCE === "screen";
|
||||
const track = rtc.LocalVideoTrack.createVideoTrack(asScreen ? "screen" : "video", source);
|
||||
|
||||
const options = new rtc.TrackPublishOptions();
|
||||
options.source = asScreen ? rtc.TrackSource.SOURCE_SCREENSHARE : rtc.TrackSource.SOURCE_CAMERA;
|
||||
|
||||
const participant = room.localParticipant;
|
||||
if (!participant) throw new Error("room has no local participant yet");
|
||||
const publication = await participant.publishTrack(track, options);
|
||||
|
||||
this.room = room;
|
||||
this.source = source;
|
||||
this.track = track;
|
||||
this.trackSid = publication.sid ?? null;
|
||||
this.stream = stream;
|
||||
|
||||
stream.on("data", (chunk: Buffer) => this.consume(chunk));
|
||||
stream.once("error", (err) => log.debug({ err }, "video stream error"));
|
||||
|
||||
// Checked often enough for smooth release at any sane frame rate.
|
||||
this.pump = setInterval(() => this.release(), 5);
|
||||
this.pump.unref?.();
|
||||
|
||||
this.firstFrameTimer = setTimeout(() => {
|
||||
if (this.published > 0) return;
|
||||
log.warn("no video frames arrived, dropping the video track");
|
||||
onGiveUp?.("источник не отдал видео");
|
||||
void this.stop();
|
||||
}, FIRST_FRAME_TIMEOUT_MS);
|
||||
this.firstFrameTimer.unref?.();
|
||||
|
||||
log.info(
|
||||
{ width, height, source: config.VIDEO_SOURCE, sid: publication.sid },
|
||||
"video track published",
|
||||
);
|
||||
}
|
||||
|
||||
private consume(chunk: Buffer): void {
|
||||
if (!this.source) return;
|
||||
this.partial.push(chunk);
|
||||
this.partialBytes += chunk.length;
|
||||
if (this.partialBytes < this.frameSize) return;
|
||||
|
||||
let buffer =
|
||||
this.partial.length === 1 ? (this.partial[0] as Buffer) : Buffer.concat(this.partial, this.partialBytes);
|
||||
while (buffer.length >= this.frameSize) {
|
||||
this.queue.push({ index: this.produced, data: Buffer.from(buffer.subarray(0, this.frameSize)) });
|
||||
this.produced += 1;
|
||||
buffer = buffer.subarray(this.frameSize);
|
||||
}
|
||||
// Keep only the tail of a partial frame for the next chunk.
|
||||
this.partial = buffer.length > 0 ? [Buffer.from(buffer)] : [];
|
||||
this.partialBytes = buffer.length;
|
||||
|
||||
// The queue grows while the audio buffer fills; beyond that, something is
|
||||
// wrong and holding more frames would only cost memory.
|
||||
while (this.queue.length > this.maxQueuedFrames) {
|
||||
this.queue.shift();
|
||||
this.dropped += 1;
|
||||
}
|
||||
this.release();
|
||||
}
|
||||
|
||||
/** Releases every frame the sound has already caught up with. */
|
||||
private release(): void {
|
||||
if (!this.source) return;
|
||||
const { fps } = this.options;
|
||||
const now = this.options.audioClock();
|
||||
const tolerance = 1 / (2 * fps);
|
||||
|
||||
while (this.queue.length > 0) {
|
||||
const frame = this.queue[0] as QueuedFrame;
|
||||
const dueAt = frame.index / fps;
|
||||
if (now + tolerance < dueAt) break;
|
||||
|
||||
this.queue.shift();
|
||||
if (now > dueAt + MAX_LATE_SECONDS) {
|
||||
this.dropped += 1;
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
this.source.captureFrame(
|
||||
new rtc.VideoFrame(frame.data, this.options.width, this.options.height, rtc.VideoBufferType.I420),
|
||||
);
|
||||
this.published += 1;
|
||||
} catch (err) {
|
||||
log.debug({ err }, "captureFrame failed");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async stop(): Promise<void> {
|
||||
for (const timer of [this.firstFrameTimer, this.pump]) {
|
||||
if (timer) clearTimeout(timer as NodeJS.Timeout);
|
||||
}
|
||||
if (this.pump) clearInterval(this.pump);
|
||||
this.firstFrameTimer = null;
|
||||
this.pump = null;
|
||||
|
||||
this.stream?.removeAllListeners("data");
|
||||
this.stream = null;
|
||||
this.partial = [];
|
||||
this.partialBytes = 0;
|
||||
this.queue = [];
|
||||
|
||||
if (this.published > 0) {
|
||||
log.info({ published: this.published, dropped: this.dropped }, "video track finished");
|
||||
}
|
||||
|
||||
const { room, trackSid } = this;
|
||||
this.trackSid = null;
|
||||
if (room?.localParticipant && trackSid) {
|
||||
try {
|
||||
await room.localParticipant.unpublishTrack(trackSid, true);
|
||||
} catch (err) {
|
||||
log.debug({ err }, "unpublishTrack failed");
|
||||
}
|
||||
}
|
||||
|
||||
try {
|
||||
this.track?.close?.();
|
||||
await this.source?.close?.();
|
||||
} catch (err) {
|
||||
log.debug({ err }, "closing video track failed");
|
||||
}
|
||||
this.track = null;
|
||||
this.source = null;
|
||||
this.room = null;
|
||||
this.produced = 0;
|
||||
this.published = 0;
|
||||
this.dropped = 0;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -169,8 +169,8 @@ export interface PlaybackInput {
|
||||
cleanup(): void;
|
||||
/** Resolves with a reason if the downloader died on its own, for reporting. */
|
||||
failure?: Promise<string | null>;
|
||||
/** Raw I420 frames to publish as a screen share, when video is on. */
|
||||
video?: { stream: Readable; width: number; height: number };
|
||||
/** Raw I420 frames to publish alongside the sound, when video is on. */
|
||||
video?: { stream: Readable; width: number; height: number; fps: number };
|
||||
}
|
||||
|
||||
export interface PlaybackOptions {
|
||||
@@ -252,7 +252,12 @@ export async function openPlayback(
|
||||
input: pipeline.audio,
|
||||
inputOptions: pipeline.audioInputOptions,
|
||||
cleanup: () => pipeline.kill(),
|
||||
video: { stream: pipeline.video, width: pipeline.width, height: pipeline.height },
|
||||
video: {
|
||||
stream: pipeline.video,
|
||||
width: pipeline.width,
|
||||
height: pipeline.height,
|
||||
fps: config.VIDEO_FPS,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user