Fix NodeAv memory leak (fixes #392)

packages/server/src/audio-decoder.ts

@@ -12,10 +12,28 @@ import { CODEC_TO_CODEC_ID, getChannelLayout } from './misc';
 import { assert, toUint8Array } from '../../../src/misc';
 import { AvFrameAudioSampleResource } from './audio-sample';
 
+type NodeAvState = {
+	frame: NodeAv.Frame;
+	packet: NodeAv.Packet;
+	codecContext: NodeAv.CodecContext | null;
+};
+
+const freeState = (state: NodeAvState) => {
+	state.codecContext?.freeContext();
+	state.frame.free();
+	state.packet.free();
+};
+
+// Needed for proper freeing if close isn't called
+let finalizationRegistry: FinalizationRegistry<NodeAvState> | null = null;
+if (typeof FinalizationRegistry !== 'undefined') {
+	finalizationRegistry = new FinalizationRegistry<NodeAvState>((state) => {
+		freeState(state);
+	});
+}
+
 export class NodeAvAudioDecoder extends CustomAudioDecoder {
-	frame!: NodeAv.Frame;
-	packet!: NodeAv.Packet;
-	codecContext!: NodeAv.CodecContext;
+	state!: NodeAvState;
 
 	// eslint-disable-next-line @typescript-eslint/no-unused-vars
 	static override supports(codec: AudioCodec, config: AudioDecoderConfig): boolean {
@@ -29,10 +47,13 @@ export class NodeAvAudioDecoder extends CustomAudioDecoder {
 	}
 
 	async init(): Promise<void> {
-		this.frame = new NodeAv.Frame();
-		this.frame.alloc();
-		this.packet = new NodeAv.Packet();
-		this.packet.alloc();
+		const frame = new NodeAv.Frame();
+		frame.alloc();
+		const packet = new NodeAv.Packet();
+		packet.alloc();
+		this.state = { frame, packet, codecContext: null };
+
+		finalizationRegistry?.register(this, this.state, this);
 
 		const codecId = CODEC_TO_CODEC_ID[this.codec];
 		assert(codecId !== undefined);
@@ -57,22 +78,26 @@ export class NodeAvAudioDecoder extends CustomAudioDecoder {
 		const ret = await codecContext.open2();
 		NodeAv.FFmpegError.throwIfError(ret, 'Open codec context');
 
-		this.codecContext = codecContext;
+		this.state.codecContext = codecContext;
 	}
 
 	async decode(packet: EncodedPacket): Promise<void> {
-		this.packet.isKeyframe = packet.type === 'key';
-		this.packet.data = Buffer.from(packet.data);
-		this.packet.timeBase = { num: 1, den: this.config.sampleRate };
-		this.packet.pts = BigInt(Math.round(packet.timestamp * this.config.sampleRate));
-		this.packet.dts = NodeAv.AV_NOPTS_VALUE;
-		this.packet.duration = BigInt(Math.round(packet.duration * this.config.sampleRate));
+		assert(this.state.codecContext);
 
-		const ret = await this.codecContext.sendPacket(this.packet);
+		this.state.packet.isKeyframe = packet.type === 'key';
+		this.state.packet.data = Buffer.from(packet.data);
+		this.state.packet.timeBase = { num: 1, den: this.config.sampleRate };
+		this.state.packet.pts = BigInt(Math.round(packet.timestamp * this.config.sampleRate));
+		this.state.packet.dts = NodeAv.AV_NOPTS_VALUE;
+		this.state.packet.duration = BigInt(Math.round(packet.duration * this.config.sampleRate));
+
+		const ret = await this.state.codecContext.sendPacket(this.state.packet);
 		NodeAv.FFmpegError.throwIfError(ret, 'Send packet');
 
+		this.state.packet.unref(); // Don't need the data again, so just unref it
+
 		while (true) {
-			const receiveRet = await this.codecContext.receiveFrame(this.frame);
+			const receiveRet = await this.state.codecContext.receiveFrame(this.state.frame);
 			if (receiveRet === NodeAv.AVERROR_EAGAIN || receiveRet === NodeAv.AVERROR_EOF) {
 				break;
 			}
@@ -84,7 +109,7 @@ export class NodeAvAudioDecoder extends CustomAudioDecoder {
 	receiveFrame(ret: number) {
 		NodeAv.FFmpegError.throwIfError(ret, 'Receive frame');
 
-		const clone = this.frame.clone();
+		const clone = this.state.frame.clone();
 		if (!clone) {
 			throw new Error('Allocation failure during frame clone.');
 		}
@@ -94,13 +119,15 @@ export class NodeAvAudioDecoder extends CustomAudioDecoder {
 	}
 
 	async flush(): Promise<void> {
+		assert(this.state.codecContext);
+
 		// Send null packet to signal flush
-		const ret = await this.codecContext.sendPacket(null);
+		const ret = await this.state.codecContext.sendPacket(null);
 		NodeAv.FFmpegError.throwIfError(ret, 'Flush decoder');
 
 		// Keep receiving frames until no more are available
 		while (true) {
-			const receiveRet = await this.codecContext.receiveFrame(this.frame);
+			const receiveRet = await this.state.codecContext.receiveFrame(this.state.frame);
 			if (receiveRet === NodeAv.AVERROR_EAGAIN || receiveRet === NodeAv.AVERROR_EOF) {
 				// No more frames available
 				break;
@@ -109,12 +136,11 @@ export class NodeAvAudioDecoder extends CustomAudioDecoder {
 			this.receiveFrame(receiveRet);
 		}
 
-		this.codecContext.flushBuffers();
+		this.state.codecContext.flushBuffers();
 	}
 
 	close(): MaybePromise<void> {
-		this.codecContext.freeContext();
-		this.frame.free();
-		this.packet.free();
+		finalizationRegistry?.unregister(this);
+		freeState(this.state);
 	}
 }

packages/server/src/audio-encoder.ts

@@ -32,19 +32,39 @@ const AC3_SAMPLE_RATES = [32000, 44100, 48000];
 
 const FRAME_SIZE_FALLBACK = 1024; // Just 'cause
 
+type NodeAvState = {
+	frame: NodeAv.Frame;
+	packet: NodeAv.Packet;
+	codecContext: NodeAv.CodecContext | null;
+	resampler: NodeAv.SoftwareResampleContext | null;
+	dstFrame: NodeAv.Frame | null;
+};
+
+const freeState = (state: NodeAvState) => {
+	state.codecContext?.freeContext();
+	state.frame.free();
+	state.packet.free();
+	state.dstFrame?.free();
+	state.resampler?.free();
+};
+
+// Needed for proper freeing if close isn't called
+let finalizationRegistry: FinalizationRegistry<NodeAvState> | null = null;
+if (typeof FinalizationRegistry !== 'undefined') {
+	finalizationRegistry = new FinalizationRegistry<NodeAvState>((state) => {
+		freeState(state);
+	});
+}
+
 export class NodeAvAudioEncoder extends CustomAudioEncoder {
-	frame!: NodeAv.Frame;
-	packet!: NodeAv.Packet;
+	state!: NodeAvState;
 	avCodec!: NodeAv.Codec;
-	codecContext: NodeAv.CodecContext | null = null;
 	firstExpectedTimestamp: number | null = null;
 	outputTimestampOffset = 0;
 
-	resampler: NodeAv.SoftwareResampleContext | null = null;
 	inputParametersKey: string | null = null;
 	resamplerInputSampleRate: number | null = null;
 	nextResamplerPts: bigint | null = null;
-	dstFrame: NodeAv.Frame | null = null;
 	packetEmitted = false;
 	adtsHeaderTemplate: AdtsHeaderTemplate | null = null;
 
@@ -76,10 +96,13 @@ export class NodeAvAudioEncoder extends CustomAudioEncoder {
 	}
 
 	async init(): Promise<void> {
-		this.frame = new NodeAv.Frame();
-		this.frame.alloc();
-		this.packet = new NodeAv.Packet();
-		this.packet.alloc();
+		const frame = new NodeAv.Frame();
+		frame.alloc();
+		const packet = new NodeAv.Packet();
+		packet.alloc();
+		this.state = { frame, packet, codecContext: null, resampler: null, dstFrame: null };
+
+		finalizationRegistry?.register(this, this.state, this);
 
 		const codecId = CODEC_TO_CODEC_ID[this.codec];
 		assert(codecId !== undefined);
@@ -95,7 +118,7 @@ export class NodeAvAudioEncoder extends CustomAudioEncoder {
 	}
 
 	async createCodecContext() {
-		assert(this.codecContext === null);
+		assert(this.state.codecContext === null);
 
 		const codecContext = new NodeAv.CodecContext();
 		codecContext.allocContext3(this.avCodec);
@@ -122,28 +145,32 @@ export class NodeAvAudioEncoder extends CustomAudioEncoder {
 		const ret = await codecContext.open2();
 		NodeAv.FFmpegError.throwIfError(ret, 'Open codec context');
 
-		this.codecContext = codecContext;
+		this.state.codecContext = codecContext;
 	}
 
 	async encode(audioSample: AudioSample): Promise<void> {
-		if (this.codecContext === null) {
+		if (this.state.codecContext === null) {
 			await this.createCodecContext();
-			assert(this.codecContext);
+			assert(this.state.codecContext);
 		}
 
 		this.firstExpectedTimestamp ??= audioSample.timestamp;
 
 		if (audioSample._data instanceof AvFrameAudioSampleResource) {
-			this.frame.ref(audioSample._data.frame);
+			// Release any buffers still referenced from the previous encode before reffing the new frame, otherwise
+			// av_frame_ref leaks them
+			// https://github.com/Vanilagy/mediabunny/issues/392
+			this.state.frame.unref();
+			this.state.frame.ref(audioSample._data.frame);
 		} else {
-			copyAudioSampleToAvFrame(audioSample, this.frame);
+			copyAudioSampleToAvFrame(audioSample, this.state.frame);
 		}
 
-		this.frame.pts = BigInt(Math.round(audioSample.timestamp * this.config.sampleRate));
-		this.frame.duration = BigInt(Math.round(audioSample.duration * this.config.sampleRate));
-		this.frame.timeBase = new NodeAv.Rational(1, this.config.sampleRate);
+		this.state.frame.pts = BigInt(Math.round(audioSample.timestamp * this.config.sampleRate));
+		this.state.frame.duration = BigInt(Math.round(audioSample.duration * this.config.sampleRate));
+		this.state.frame.timeBase = new NodeAv.Rational(1, this.config.sampleRate);
 
-		const key = `${this.frame.sampleRate}:${this.frame.channels}:${this.frame.format}`;
+		const key = `${this.state.frame.sampleRate}:${this.state.frame.channels}:${this.state.frame.format}`;
 		if (this.inputParametersKey !== null && this.inputParametersKey !== key) {
 			throw new Error(
 				'Input audio parameters changed. For this audio encoder, you cannot change the input audio'
@@ -156,83 +183,83 @@ export class NodeAvAudioEncoder extends CustomAudioEncoder {
 		// 1. Format conversion is needed (sample format, sample rate, or channel count differs)
 		// 2. The codec requires fixed frame sizes
 		const requiresResampler
-			= this.codecContext.frameSize > 0
-				|| this.codecContext.sampleFormat !== this.frame.format
-				|| this.codecContext.sampleRate !== this.frame.sampleRate
-				|| this.codecContext.channels !== this.frame.channels;
+			= this.state.codecContext.frameSize > 0
+				|| this.state.codecContext.sampleFormat !== this.state.frame.format
+				|| this.state.codecContext.sampleRate !== this.state.frame.sampleRate
+				|| this.state.codecContext.channels !== this.state.frame.channels;
 
 		if (requiresResampler) {
-			if (!this.resampler) {
-				this.resampler = new NodeAv.SoftwareResampleContext();
-				this.resamplerInputSampleRate = this.frame.sampleRate;
+			if (!this.state.resampler) {
+				this.state.resampler = new NodeAv.SoftwareResampleContext();
+				this.resamplerInputSampleRate = this.state.frame.sampleRate;
 
-				const outLayout = getChannelLayout(this.codecContext.channels);
-				const inLayout = getChannelLayout(this.frame.channels);
+				const outLayout = getChannelLayout(this.state.codecContext.channels);
+				const inLayout = getChannelLayout(this.state.frame.channels);
 
-				const ret = this.resampler.allocSetOpts2(
-					outLayout, this.codecContext.sampleFormat, this.codecContext.sampleRate,
-					inLayout, this.frame.format as NodeAv.AVSampleFormat, this.frame.sampleRate,
+				const ret = this.state.resampler.allocSetOpts2(
+					outLayout, this.state.codecContext.sampleFormat, this.state.codecContext.sampleRate,
+					inLayout, this.state.frame.format as NodeAv.AVSampleFormat, this.state.frame.sampleRate,
 				);
 				NodeAv.FFmpegError.throwIfError(ret, 'allocSetOpts2');
 
-				const ret2 = this.resampler.init();
+				const ret2 = this.state.resampler.init();
 				NodeAv.FFmpegError.throwIfError(ret2, 'init');
 
-				this.dstFrame = new NodeAv.Frame();
-				this.dstFrame.alloc();
-				this.dstFrame.channelLayout = outLayout;
-				this.dstFrame.sampleRate = this.codecContext.sampleRate;
-				this.dstFrame.format = this.codecContext.sampleFormat;
-				this.dstFrame.nbSamples = this.codecContext.frameSize || FRAME_SIZE_FALLBACK;
-				this.dstFrame.duration = BigInt(this.dstFrame.nbSamples);
-				this.dstFrame.allocBuffer();
+				this.state.dstFrame = new NodeAv.Frame();
+				this.state.dstFrame.alloc();
+				this.state.dstFrame.channelLayout = outLayout;
+				this.state.dstFrame.sampleRate = this.state.codecContext.sampleRate;
+				this.state.dstFrame.format = this.state.codecContext.sampleFormat;
+				this.state.dstFrame.nbSamples = this.state.codecContext.frameSize || FRAME_SIZE_FALLBACK;
+				this.state.dstFrame.duration = BigInt(this.state.dstFrame.nbSamples);
+				this.state.dstFrame.allocBuffer();
 
-				this.nextResamplerPts = this.frame.pts;
+				this.nextResamplerPts = this.state.frame.pts;
 			}
 
-			const inputBuffers = this.frame.data;
+			const inputBuffers = this.state.frame.data;
 			if (!inputBuffers) {
 				throw new DOMException('Frame has no data', 'EncodingError');
 			}
-			await this.resampler.convert(null, 0, inputBuffers, this.frame.nbSamples);
+			await this.state.resampler.convert(null, 0, inputBuffers, this.state.frame.nbSamples);
 
 			await this.pullResampledFrames();
 		} else {
-			await this.sendFrameAndReceivePackets(this.frame);
+			await this.sendFrameAndReceivePackets(this.state.frame);
 		}
 	}
 
 	async pullResampledFrames() {
-		assert(this.codecContext);
-		assert(this.resampler);
-		assert(this.dstFrame);
+		assert(this.state.codecContext);
+		assert(this.state.resampler);
+		assert(this.state.dstFrame);
 		assert(this.nextResamplerPts !== null);
 
-		const frameSize = this.codecContext.frameSize || FRAME_SIZE_FALLBACK;
+		const frameSize = this.state.codecContext.frameSize || FRAME_SIZE_FALLBACK;
 
 		while (true) {
-			const available = this.resampler.getOutSamples(0);
+			const available = this.state.resampler.getOutSamples(0);
 			if (available < frameSize) {
 				break;
 			}
 
-			await this.resampler.convert(this.dstFrame.data, frameSize, null, 0);
+			await this.state.resampler.convert(this.state.dstFrame.data, frameSize, null, 0);
 
-			this.dstFrame.pts = this.nextResamplerPts;
+			this.state.dstFrame.pts = this.nextResamplerPts;
 
-			await this.sendFrameAndReceivePackets(this.dstFrame);
+			await this.sendFrameAndReceivePackets(this.state.dstFrame);
 			this.nextResamplerPts += BigInt(frameSize);
 		}
 	}
 
 	async sendFrameAndReceivePackets(frame: NodeAv.Frame | null) {
-		assert(this.codecContext);
+		assert(this.state.codecContext);
 
-		const ret = await this.codecContext.sendFrame(frame);
+		const ret = await this.state.codecContext.sendFrame(frame);
 		NodeAv.FFmpegError.throwIfError(ret, 'Send frame');
 
 		while (true) {
-			const receiveRet = await this.codecContext.receivePacket(this.packet);
+			const receiveRet = await this.state.codecContext.receivePacket(this.state.packet);
 			if (receiveRet === NodeAv.AVERROR_EAGAIN || receiveRet === NodeAv.AVERROR_EOF) {
 				break;
 			}
@@ -242,18 +269,18 @@ export class NodeAvAudioEncoder extends CustomAudioEncoder {
 	}
 
 	receivePacket(ret: number) {
-		assert(this.codecContext);
+		assert(this.state.codecContext);
 		assert(this.firstExpectedTimestamp !== null);
 		NodeAv.FFmpegError.throwIfError(ret, 'Receive packet');
 
-		if (!this.packet.data) {
+		if (!this.state.packet.data) {
 			return;
 		}
 
-		let timestamp = Number(this.packet.pts) / this.codecContext.sampleRate;
-		const duration = Number(this.packet.duration) / this.codecContext.sampleRate;
+		let timestamp = Number(this.state.packet.pts) / this.state.codecContext.sampleRate;
+		const duration = Number(this.state.packet.duration) / this.state.codecContext.sampleRate;
 
-		let data: Uint8Array = this.packet.data;
+		let data: Uint8Array = this.state.packet.data;
 
 		let metadata: EncodedAudioChunkMetadata | undefined;
 		if (this.packetEmitted) {
@@ -265,8 +292,8 @@ export class NodeAvAudioEncoder extends CustomAudioEncoder {
 			this.outputTimestampOffset = Math.max(this.firstExpectedTimestamp - timestamp, 0);
 
 			const codecString = this.config.codec;
-			let description = this.codecContext.extraData
-				? toUint8Array(this.codecContext.extraData)
+			let description = this.state.codecContext.extraData
+				? toUint8Array(this.state.codecContext.extraData)
 				: undefined;
 
 			if (this.codec === 'aac') {
@@ -311,8 +338,8 @@ export class NodeAvAudioEncoder extends CustomAudioEncoder {
 			metadata = {
 				decoderConfig: {
 					codec: codecString,
-					sampleRate: this.codecContext.sampleRate,
-					numberOfChannels: this.codecContext.channels,
+					sampleRate: this.state.codecContext.sampleRate,
+					numberOfChannels: this.state.codecContext.channels,
 					description,
 				},
 			};
@@ -344,53 +371,50 @@ export class NodeAvAudioEncoder extends CustomAudioEncoder {
 	}
 
 	async flush(): Promise<void> {
-		if (!this.codecContext) {
+		if (!this.state.codecContext) {
 			return;
 		}
 
 		outer:
-		if (this.resampler) {
+		if (this.state.resampler) {
 			assert(this.resamplerInputSampleRate !== null);
 
-			const currentOutSamples = this.resampler.getOutSamples(0);
+			const currentOutSamples = this.state.resampler.getOutSamples(0);
 			if (currentOutSamples === 0) {
 				break outer; // Clean cut-off point
 			}
 
-			const frameSize = this.codecContext.frameSize || FRAME_SIZE_FALLBACK;
+			const frameSize = this.state.codecContext.frameSize || FRAME_SIZE_FALLBACK;
 			assert(currentOutSamples < frameSize); // Because if it's more, it would've already been retrieved
 
 			const inputSamplesNeeded = Math.ceil(
-				((frameSize - currentOutSamples) / this.codecContext.sampleRate) * this.resamplerInputSampleRate,
+				((frameSize - currentOutSamples) / this.state.codecContext.sampleRate) * this.resamplerInputSampleRate,
 			);
-			this.resampler.injectSilence(inputSamplesNeeded);
+			this.state.resampler.injectSilence(inputSamplesNeeded);
 
 			await this.pullResampledFrames();
 		}
 
 		await this.sendFrameAndReceivePackets(null);
 
-		this.codecContext.freeContext();
-		this.codecContext = null;
+		this.state.codecContext.freeContext();
+		this.state.codecContext = null;
 		this.packetEmitted = false;
 		this.firstExpectedTimestamp = null;
 		this.outputTimestampOffset = 0;
 		this.adtsHeaderTemplate = null;
 
-		this.resampler?.free();
-		this.resampler = null;
+		this.state.resampler?.free();
+		this.state.resampler = null;
 		this.inputParametersKey = null;
 		this.resamplerInputSampleRate = null;
 		this.nextResamplerPts = null;
-		this.dstFrame?.free();
-		this.dstFrame = null;
+		this.state.dstFrame?.free();
+		this.state.dstFrame = null;
 	}
 
 	close(): MaybePromise<void> {
-		this.codecContext?.freeContext();
-		this.frame.free();
-		this.packet.free();
-		this.dstFrame?.free();
-		this.resampler?.free();
+		finalizationRegistry?.unregister(this);
+		freeState(this.state);
 	}
 }

packages/server/src/index.ts

@@ -78,6 +78,8 @@ export { AvFrameAudioSampleResource } from './audio-sample';
 export const toAvFrame = async (sample: VideoSample | AudioSample, frame: NodeAv.Frame) => {
 	if (sample instanceof VideoSample) {
 		if (sample._data instanceof AvFrameVideoSampleResource) {
+			// We're overriding the frame, so release whatever it referenced before, otherwise av_frame_ref leaks it
+			frame.unref();
 			frame.ref(sample._data.frame);
 		} else {
 			if (sample.format === null) {
@@ -92,6 +94,8 @@ export const toAvFrame = async (sample: VideoSample | AudioSample, frame: NodeAv
 		frame.timeBase = new NodeAv.Rational(1, 1e6);
 	} else {
 		if (sample._data instanceof AvFrameAudioSampleResource) {
+			// We're overriding the frame, so release whatever it referenced before, otherwise av_frame_ref leaks it
+			frame.unref();
 			frame.ref(sample._data.frame);
 		} else {
 			copyAudioSampleToAvFrame(sample, frame);

packages/server/src/video-decoder.ts

@@ -12,10 +12,28 @@ import { CODEC_TO_CODEC_ID, getHardwareDecoderCodec } from './misc';
 import { assert, binarySearchLessOrEqual, simplifyRational, toUint8Array } from '../../../src/misc';
 import { AvFrameVideoSampleResource } from './video-sample';
 
+type NodeAvState = {
+	frame: NodeAv.Frame;
+	packet: NodeAv.Packet;
+	codecContext: NodeAv.CodecContext | null;
+};
+
+const freeState = (state: NodeAvState) => {
+	state.codecContext?.freeContext();
+	state.frame.free();
+	state.packet.free();
+};
+
+// Needed for proper freeing if close isn't called
+let finalizationRegistry: FinalizationRegistry<NodeAvState> | null = null;
+if (typeof FinalizationRegistry !== 'undefined') {
+	finalizationRegistry = new FinalizationRegistry<NodeAvState>((state) => {
+		freeState(state);
+	});
+}
+
 export class NodeAvVideoDecoder extends CustomVideoDecoder {
-	frame!: NodeAv.Frame;
-	packet!: NodeAv.Packet;
-	codecContext: NodeAv.CodecContext | null = null;
+	state!: NodeAvState;
 	pixelAspectRatio!: Rational;
 
 	// Bookkeeping to restore the original timing information
@@ -33,14 +51,17 @@ export class NodeAvVideoDecoder extends CustomVideoDecoder {
 	}
 
 	async init() {
-		this.frame = new NodeAv.Frame();
-		this.frame.alloc();
-		this.packet = new NodeAv.Packet();
-		this.packet.alloc();
+		const frame = new NodeAv.Frame();
+		frame.alloc();
+		const packet = new NodeAv.Packet();
+		packet.alloc();
+		this.state = { frame, packet, codecContext: null };
+
+		finalizationRegistry?.register(this, this.state, this);
 	}
 
 	async initCodecContext(packet: EncodedPacket) {
-		assert(this.codecContext === null);
+		assert(this.state.codecContext === null);
 
 		const codecId = CODEC_TO_CODEC_ID[this.codec];
 		assert(codecId !== undefined);
@@ -85,28 +106,29 @@ export class NodeAvVideoDecoder extends CustomVideoDecoder {
 		const ret = await codecContext.open2();
 		NodeAv.FFmpegError.throwIfError(ret, 'Open codec context');
 
-		this.codecContext = codecContext;
+		this.state.codecContext = codecContext;
 	}
 
 	async decode(packet: EncodedPacket) {
-		if (this.codecContext === null) {
+		if (this.state.codecContext === null) {
 			await this.initCodecContext(packet);
-			assert(this.codecContext);
 		}
 
-		this.packet.isKeyframe = packet.type === 'key';
-		this.packet.data = Buffer.from(packet.data);
-		this.packet.timeBase = { num: 1, den: 1e6 };
-		this.packet.pts = BigInt(packet.microsecondTimestamp);
-		this.packet.dts = NodeAv.AV_NOPTS_VALUE;
-		this.packet.duration = BigInt(packet.microsecondDuration);
+		assert(this.state.codecContext);
+
+		this.state.packet.isKeyframe = packet.type === 'key';
+		this.state.packet.data = Buffer.from(packet.data);
+		this.state.packet.timeBase = { num: 1, den: 1e6 };
+		this.state.packet.pts = BigInt(packet.microsecondTimestamp);
+		this.state.packet.dts = NodeAv.AV_NOPTS_VALUE;
+		this.state.packet.duration = BigInt(packet.microsecondDuration);
 
 		if (packet.sideData.alpha) {
 			const matroskaBlockAdditional = Buffer.alloc(8 + packet.sideData.alpha.byteLength);
 			matroskaBlockAdditional[7] = 1; // BlockAddId
 			matroskaBlockAdditional.set(packet.sideData.alpha, 8);
 
-			this.packet.addSideData(NodeAv.AV_PKT_DATA_MATROSKA_BLOCKADDITIONAL, matroskaBlockAdditional);
+			this.state.packet.addSideData(NodeAv.AV_PKT_DATA_MATROSKA_BLOCKADDITIONAL, matroskaBlockAdditional);
 		}
 
 		const preciseTimingIndex = binarySearchLessOrEqual(
@@ -141,11 +163,13 @@ export class NodeAvVideoDecoder extends CustomVideoDecoder {
 			}
 		}
 
-		const ret = await this.codecContext.sendPacket(this.packet);
+		const ret = await this.state.codecContext.sendPacket(this.state.packet);
 		NodeAv.FFmpegError.throwIfError(ret, 'Send packet');
 
+		this.state.packet.unref(); // Don't need the data again, so just unref it
+
 		while (true) {
-			const receiveRet = await this.codecContext.receiveFrame(this.frame);
+			const receiveRet = await this.state.codecContext.receiveFrame(this.state.frame);
 			if (receiveRet === NodeAv.AVERROR_EAGAIN || receiveRet === NodeAv.AVERROR_EOF) {
 				break;
 			}
@@ -157,14 +181,14 @@ export class NodeAvVideoDecoder extends CustomVideoDecoder {
 	receiveFrame(ret: number) {
 		NodeAv.FFmpegError.throwIfError(ret, 'Receive frame');
 
-		this.frame.sampleAspectRatio = new NodeAv.Rational(this.pixelAspectRatio.num, this.pixelAspectRatio.den);
+		this.state.frame.sampleAspectRatio = new NodeAv.Rational(this.pixelAspectRatio.num, this.pixelAspectRatio.den);
 
-		let timestamp = Number(this.frame.pts) / 1e6;
-		let duration = Number(this.frame.duration) / 1e6;
+		let timestamp = Number(this.state.frame.pts) / 1e6;
+		let duration = Number(this.state.frame.duration) / 1e6;
 
 		const preciseTimingIndex = binarySearchLessOrEqual(
 			this.preciseTimings,
-			Number(this.frame.pts),
+			Number(this.state.frame.pts),
 			x => x.microsecondTimestamp,
 		);
 		const entry = preciseTimingIndex !== -1
@@ -173,7 +197,7 @@ export class NodeAvVideoDecoder extends CustomVideoDecoder {
 
 		// If there's a relevant timing entry, refine the frame's timing data to get better accuracy than
 		// microseconds
-		if (entry && entry.microsecondTimestamp === Number(this.frame.pts)) {
+		if (entry && entry.microsecondTimestamp === Number(this.state.frame.pts)) {
 			if (entry.timestampIsValid) {
 				timestamp = entry.timestamp;
 			}
@@ -182,7 +206,7 @@ export class NodeAvVideoDecoder extends CustomVideoDecoder {
 			}
 		}
 
-		const clone = this.frame.clone();
+		const clone = this.state.frame.clone();
 		if (!clone) {
 			throw new Error('Frame clone allocation failed.');
 		}
@@ -194,17 +218,17 @@ export class NodeAvVideoDecoder extends CustomVideoDecoder {
 	}
 
 	async flush() {
-		if (!this.codecContext) {
+		if (!this.state.codecContext) {
 			return;
 		}
 
 		// Send null packet to signal flush
-		const ret = await this.codecContext.sendPacket(null);
+		const ret = await this.state.codecContext.sendPacket(null);
 		NodeAv.FFmpegError.throwIfError(ret, 'Flush decoder');
 
 		// Keep receiving frames until no more are available
 		while (true) {
-			const receiveRet = await this.codecContext.receiveFrame(this.frame);
+			const receiveRet = await this.state.codecContext.receiveFrame(this.state.frame);
 			if (receiveRet === NodeAv.AVERROR_EAGAIN || receiveRet === NodeAv.AVERROR_EOF) {
 				// No more frames available
 				break;
@@ -213,12 +237,11 @@ export class NodeAvVideoDecoder extends CustomVideoDecoder {
 			this.receiveFrame(receiveRet);
 		}
 
-		this.codecContext.flushBuffers();
+		this.state.codecContext.flushBuffers();
 	}
 
 	close(): MaybePromise<void> {
-		this.codecContext?.freeContext();
-		this.frame.free();
-		this.packet.free();
+		finalizationRegistry?.unregister(this);
+		freeState(this.state);
 	}
 }

packages/server/src/video-encoder.ts

@@ -41,17 +41,36 @@ import {
 import { extractVideoCodecString } from '../../../src/codec';
 import { assert, binarySearchLessOrEqual, simplifyRational, toUint8Array } from '../../../src/misc';
 
+type NodeAvState = {
+	frame: NodeAv.Frame;
+	packet: NodeAv.Packet;
+	codecContext: NodeAv.CodecContext | null;
+	scaler: NodeAv.SoftwareScaleContext | null;
+	dstFrame: NodeAv.Frame | null;
+};
+
+const freeState = (state: NodeAvState) => {
+	state.codecContext?.freeContext();
+	state.frame.free();
+	state.packet.free();
+	state.scaler?.freeContext();
+	state.dstFrame?.free();
+};
+
+// Needed for proper freeing if close isn't called
+let finalizationRegistry: FinalizationRegistry<NodeAvState> | null = null;
+if (typeof FinalizationRegistry !== 'undefined') {
+	finalizationRegistry = new FinalizationRegistry<NodeAvState>((state) => {
+		freeState(state);
+	});
+}
+
 export class NodeAvVideoEncoder extends CustomVideoEncoder {
-	frame!: NodeAv.Frame;
-	packet!: NodeAv.Packet;
+	state!: NodeAvState;
 	avCodec!: NodeAv.Codec;
-	codecContext: NodeAv.CodecContext | null = null;
 	lastBuffer: Buffer | null = null;
 	packetEmitted = false;
-
-	scaler: NodeAv.SoftwareScaleContext | null = null;
 	lastScalerKey: string | null = null;
-	dstFrame: NodeAv.Frame | null = null;
 
 	// Bookkeeping to restore the original timing information
 	preciseTimings: {
@@ -68,12 +87,16 @@ export class NodeAvVideoEncoder extends CustomVideoEncoder {
 	}
 
 	async init(): Promise<void> {
-		this.frame = new NodeAv.Frame();
-		this.frame.alloc();
-		this.frame.timeBase = new NodeAv.Rational(1, 1e6);
+		const frame = new NodeAv.Frame();
+		frame.alloc();
+		frame.timeBase = new NodeAv.Rational(1, 1e6);
 
-		this.packet = new NodeAv.Packet();
-		this.packet.alloc();
+		const packet = new NodeAv.Packet();
+		packet.alloc();
+
+		this.state = { frame, packet, codecContext: null, scaler: null, dstFrame: null };
+
+		finalizationRegistry?.register(this, this.state, this);
 
 		const codecId = CODEC_TO_CODEC_ID[this.codec];
 		assert(codecId !== undefined);
@@ -97,7 +120,7 @@ export class NodeAvVideoEncoder extends CustomVideoEncoder {
 	}
 
 	async createCodecContext() {
-		assert(this.codecContext === null);
+		assert(this.state.codecContext === null);
 
 		const codecContext = new NodeAv.CodecContext();
 		codecContext.allocContext3(this.avCodec);
@@ -177,65 +200,69 @@ export class NodeAvVideoEncoder extends CustomVideoEncoder {
 		const ret = await codecContext.open2();
 		NodeAv.FFmpegError.throwIfError(ret, 'Open codec context');
 
-		this.codecContext = codecContext;
+		this.state.codecContext = codecContext;
 	}
 
 	async encode(videoSample: VideoSample, options: VideoEncoderEncodeOptions): Promise<void> {
-		if (this.codecContext === null) {
+		if (this.state.codecContext === null) {
 			await this.createCodecContext();
-			assert(this.codecContext);
 		}
+		assert(this.state.codecContext);
 
 		if (videoSample._data instanceof AvFrameVideoSampleResource) {
-			this.frame.ref(videoSample._data.frame);
+			// Release any buffers still referenced from the previous encode before reffing the new frame, otherwise
+			// av_frame_ref leaks them
+			// https://github.com/Vanilagy/mediabunny/issues/392
+			this.state.frame.unref();
+			this.state.frame.ref(videoSample._data.frame);
 		} else {
 			if (videoSample.format === null) {
 				throw new Error('Cannot encode foreign VideoSample with unknown (null) format.');
 			}
 
-			this.lastBuffer = await copyVideoSampleToAvFrame(videoSample, this.frame, this.lastBuffer);
+			this.lastBuffer = await copyVideoSampleToAvFrame(videoSample, this.state.frame, this.lastBuffer);
 		}
 
-		let frameToEncode = this.frame;
+		let frameToEncode = this.state.frame;
 
 		const requiresScaler
-			= this.codecContext.pixelFormat !== this.frame.format
-				|| this.codecContext.width !== this.frame.width
-				|| this.codecContext.height !== this.frame.height;
+			= this.state.codecContext.pixelFormat !== this.state.frame.format
+				|| this.state.codecContext.width !== this.state.frame.width
+				|| this.state.codecContext.height !== this.state.frame.height;
 
 		if (requiresScaler) {
-			if (!this.scaler) {
-				this.scaler = new NodeAv.SoftwareScaleContext();
+			if (!this.state.scaler) {
+				this.state.scaler = new NodeAv.SoftwareScaleContext();
 			}
 
-			const key = `${this.frame.width}x${this.frame.height}:${this.frame.format}`;
+			const key = `${this.state.frame.width}x${this.state.frame.height}:${this.state.frame.format}`;
 			const needsConfigure = key !== this.lastScalerKey;
 
 			if (needsConfigure) {
-				this.scaler.getContext(
-					this.frame.width, this.frame.height, this.frame.format as NodeAv.AVPixelFormat,
-					this.codecContext.width, this.codecContext.height, this.codecContext.pixelFormat,
+				this.state.scaler.getContext(
+					this.state.frame.width, this.state.frame.height, this.state.frame.format as NodeAv.AVPixelFormat,
+					this.state.codecContext.width, this.state.codecContext.height, this.state.codecContext.pixelFormat,
 					NodeAv.SWS_FAST_BILINEAR,
 				);
 
 				this.lastScalerKey = key;
 
-				const ret = this.scaler.initContext();
+				const ret = this.state.scaler.initContext();
 				NodeAv.FFmpegError.throwIfError(ret, 'initContext');
 			}
 
-			if (!this.dstFrame) {
-				this.dstFrame = new NodeAv.Frame();
-				this.dstFrame.alloc();
-				this.dstFrame.width = this.codecContext.width;
-				this.dstFrame.height = this.codecContext.height;
-				this.dstFrame.format = this.codecContext.pixelFormat;
-				this.dstFrame.allocBuffer();
+			if (!this.state.dstFrame) {
+				this.state.dstFrame = new NodeAv.Frame();
+				this.state.dstFrame.alloc();
+				this.state.dstFrame.width = this.state.codecContext.width;
+				this.state.dstFrame.height = this.state.codecContext.height;
+				this.state.dstFrame.format = this.state.codecContext.pixelFormat;
+				this.state.dstFrame.allocBuffer();
 			}
 
-			await this.scaler.scaleFrame(this.dstFrame, this.frame);
-			this.dstFrame.copyProps(this.frame);
-			frameToEncode = this.dstFrame;
+			await this.state.scaler.scaleFrame(this.state.dstFrame, this.state.frame);
+			this.state.dstFrame.copyProps(this.state.frame);
+			frameToEncode = this.state.dstFrame;
 		}
 
 		frameToEncode.pts = BigInt(videoSample.microsecondTimestamp);
@@ -282,12 +309,12 @@ export class NodeAvVideoEncoder extends CustomVideoEncoder {
 			}
 		}
 
-		const ret = await this.codecContext.sendFrame(frameToEncode);
+		const ret = await this.state.codecContext.sendFrame(frameToEncode);
 		NodeAv.FFmpegError.throwIfError(ret, 'Send frame');
 
 		// Keep receiving packets until no more are available for this frame
 		while (true) {
-			const receiveRet = await this.codecContext.receivePacket(this.packet);
+			const receiveRet = await this.state.codecContext.receivePacket(this.state.packet);
 			if (receiveRet === NodeAv.AVERROR_EAGAIN || receiveRet === NodeAv.AVERROR_EOF) {
 				break;
 			}
@@ -297,20 +324,20 @@ export class NodeAvVideoEncoder extends CustomVideoEncoder {
 	}
 
 	receivePacket(ret: number) {
-		assert(this.codecContext);
+		assert(this.state.codecContext);
 		NodeAv.FFmpegError.throwIfError(ret, 'Receive packet');
 
-		if (!this.packet.data) {
+		if (!this.state.packet.data) {
 			return;
 		}
-		let packetData = toUint8Array(this.packet.data);
+		let packetData = toUint8Array(this.state.packet.data);
 
-		let timestamp = Number(this.packet.pts) / 1e6;
-		let duration = Number(this.packet.duration) / 1e6;
+		let timestamp = Number(this.state.packet.pts) / 1e6;
+		let duration = Number(this.state.packet.duration) / 1e6;
 
 		const preciseTimingIndex = binarySearchLessOrEqual(
 			this.preciseTimings,
-			Number(this.packet.pts),
+			Number(this.state.packet.pts),
 			x => x.microsecondTimestamp,
 		);
 		const entry = preciseTimingIndex !== -1
@@ -319,7 +346,7 @@ export class NodeAvVideoEncoder extends CustomVideoEncoder {
 
 		// If there's a relevant timing entry, refine the packet's timing data to get better accuracy than
 		// microseconds
-		if (entry && entry.microsecondTimestamp === Number(this.packet.pts)) {
+		if (entry && entry.microsecondTimestamp === Number(this.state.packet.pts)) {
 			if (entry.timestampIsValid) {
 				timestamp = entry.timestamp;
 			}
@@ -346,14 +373,14 @@ export class NodeAvVideoEncoder extends CustomVideoEncoder {
 				let serializedRecord: Uint8Array;
 
 				if (this.codec === 'avc') {
-					const record = extractAvcDecoderConfigurationRecord(this.packet.data);
+					const record = extractAvcDecoderConfigurationRecord(this.state.packet.data);
 					if (!record) {
 						throw new Error('Invalid AVC data, could not extract decoder configuration record.');
 					}
 
 					serializedRecord = serializeAvcDecoderConfigurationRecord(record);
 				} else {
-					const record = extractHevcDecoderConfigurationRecord(this.packet.data);
+					const record = extractHevcDecoderConfigurationRecord(this.state.packet.data);
 					if (!record) {
 						throw new Error('Invalid HEVC data, could not extract decoder configuration record.');
 					}
@@ -488,14 +515,14 @@ export class NodeAvVideoEncoder extends CustomVideoEncoder {
 		}
 
 		const sideData: EncodedPacketSideData = {};
-		const matroskaBlockAdditional = this.packet.getSideData(NodeAv.AV_PKT_DATA_MATROSKA_BLOCKADDITIONAL);
+		const matroskaBlockAdditional = this.state.packet.getSideData(NodeAv.AV_PKT_DATA_MATROSKA_BLOCKADDITIONAL);
 		if (matroskaBlockAdditional) {
 			sideData.alpha = toUint8Array(matroskaBlockAdditional).subarray(8); // Skip the BlockAddId
 		}
 
 		const packet = new EncodedPacket(
 			packetData,
-			this.packet.isKeyframe ? 'key' : 'delta',
+			this.state.packet.isKeyframe ? 'key' : 'delta',
 			timestamp,
 			duration,
 			undefined,
@@ -507,21 +534,19 @@ export class NodeAvVideoEncoder extends CustomVideoEncoder {
 			// Create the decoder config
 			metadata.decoderConfig = {
 				codec: decoderConfigCodecString,
-				codedWidth: this.codecContext.width,
-				codedHeight: this.codecContext.height,
-				displayAspectWidth: this.config.displayWidth ?? this.codecContext.width,
-				displayAspectHeight: this.config.displayHeight ?? this.codecContext.height,
+				codedWidth: this.state.codecContext.width,
+				codedHeight: this.state.codecContext.height,
+				displayAspectWidth: this.config.displayWidth ?? this.state.codecContext.width,
+				displayAspectHeight: this.config.displayHeight ?? this.state.codecContext.height,
 				description: decoderConfigDescription ?? undefined,
 				colorSpace: {
-					primaries:
-							unmapColorPrimaries(this.codecContext.colorPrimaries) as VideoColorPrimaries,
-					matrix:
-							unmapMatrixCoefficients(this.codecContext.colorSpace) as VideoMatrixCoefficients,
+					primaries: unmapColorPrimaries(this.state.codecContext.colorPrimaries) as VideoColorPrimaries,
+					matrix: unmapMatrixCoefficients(this.state.codecContext.colorSpace) as VideoMatrixCoefficients,
 					transfer:
-							unmapTransferCharacteristics(this.codecContext.colorTrc) as VideoTransferCharacteristics,
-					fullRange: this.codecContext.colorRange === NodeAv.AVCOL_RANGE_JPEG
+						unmapTransferCharacteristics(this.state.codecContext.colorTrc) as VideoTransferCharacteristics,
+					fullRange: this.state.codecContext.colorRange === NodeAv.AVCOL_RANGE_JPEG
 						? true
-						: this.codecContext.colorRange === NodeAv.AVCOL_RANGE_MPEG
+						: this.state.codecContext.colorRange === NodeAv.AVCOL_RANGE_MPEG
 							? false
 							: undefined,
 				},
@@ -533,14 +558,14 @@ export class NodeAvVideoEncoder extends CustomVideoEncoder {
 	}
 
 	async flush(): Promise<void> {
-		if (this.codecContext) {
+		if (this.state.codecContext) {
 			// Send null frame to signal flush
-			const ret = await this.codecContext.sendFrame(null);
+			const ret = await this.state.codecContext.sendFrame(null);
 			NodeAv.FFmpegError.throwIfError(ret, 'Send frame');
 
 			// Keep receiving packets until no more are available
 			while (true) {
-				const receiveRet = await this.codecContext.receivePacket(this.packet);
+				const receiveRet = await this.state.codecContext.receivePacket(this.state.packet);
 				if (receiveRet === NodeAv.AVERROR_EAGAIN || receiveRet === NodeAv.AVERROR_EOF) {
 					break;
 				}
@@ -548,8 +573,8 @@ export class NodeAvVideoEncoder extends CustomVideoEncoder {
 				this.receivePacket(receiveRet);
 			}
 
-			this.codecContext.freeContext();
-			this.codecContext = null;
+			this.state.codecContext.freeContext();
+			this.state.codecContext = null;
 			// The codec is done now and can't be reused. Any subsequent encode call will first need to recreate a
 			// codec context.
 		}
@@ -558,10 +583,7 @@ export class NodeAvVideoEncoder extends CustomVideoEncoder {
 	}
 
 	close(): MaybePromise<void> {
-		this.codecContext?.freeContext();
-		this.frame.free();
-		this.packet.free();
-		this.scaler?.freeContext();
-		this.dstFrame?.free();
+		finalizationRegistry?.unregister(this);
+		freeState(this.state);
 	}
 }

test/node/server-extension.test.ts

@@ -871,6 +871,88 @@ describe('Video', async () => {
 		expect(plane[3]).toBe(0xff);
 	});
 
+	// https://github.com/Vanilagy/mediabunny/issues/392
+	test('Memory doesn\'t leak', async () => {
+		const encoder = new NodeAvVideoEncoder();
+		// @ts-expect-error Readonly
+		encoder.codec = 'avc';
+		// @ts-expect-error Readonly
+		encoder.config = {
+			codec: buildVideoCodecString('avc', 1920, 1080, 1e6),
+			width: 1920,
+			height: 1080,
+			bitrate: 1e6,
+		} satisfies VideoEncoderConfig;
+
+		const pendingPackets: EncodedPacket[] = [];
+		const decodedFrames: VideoSample[] = [];
+		let decoderConfig: VideoDecoderConfig | null = null;
+
+		// @ts-expect-error Readonly
+		encoder.onPacket = (packet: EncodedPacket, meta: EncodedVideoChunkMetadata) => {
+			if (meta.decoderConfig) {
+				decoderConfig = meta.decoderConfig;
+			}
+			pendingPackets.push(packet);
+		};
+
+		await encoder.init();
+
+		// Encode a single white frame to get one packet (and the decoder config) to drive the loop with
+		const data = new Uint8Array(1920 * 1080 * 4).fill(0xff); // White
+		using bootstrapSample = new VideoSample(data, {
+			format: 'RGBX',
+			codedWidth: 1920,
+			codedHeight: 1080,
+			timestamp: 0,
+			duration: 1 / 30,
+		});
+		await encoder.encode(bootstrapSample, {});
+		await encoder.flush();
+		assert(decoderConfig);
+		assert(pendingPackets.length > 0);
+
+		const sourcePacket = pendingPackets[0]!;
+		pendingPackets.length = 0;
+
+		const decoder = new NodeAvVideoDecoder();
+		// @ts-expect-error Readonly
+		decoder.codec = 'avc';
+		// @ts-expect-error Readonly
+		decoder.config = decoderConfig;
+		// @ts-expect-error Readonly
+		decoder.onSample = (sample: VideoSample) => {
+			decodedFrames.push(sample);
+		};
+		await decoder.init();
+
+		const iterations = 200;
+		let baselineRss = 0;
+
+		for (let i = 1; i <= iterations; i++) {
+			await decoder.decode(sourcePacket.clone({ timestamp: i / 30 }));
+
+			for (const frame of decodedFrames) {
+				await encoder.encode(frame, {});
+				frame.close();
+			}
+			decodedFrames.length = 0;
+			pendingPackets.length = 0;
+
+			if (i === 1) {
+				baselineRss = process.memoryUsage().rss;
+			}
+		}
+
+		const finalRss = process.memoryUsage().rss;
+		const growth = finalRss - baselineRss;
+
+		expect(growth).toBeLessThan(100 * 1024 * 1024);
+
+		await encoder.close();
+		await decoder.close();
+	});
+
 	describe('VideoSample transformation', () => {
 		// 400x400 image: red everywhere, with a 200x200 blue square filling the bottom-left quadrant.
 		const TEST_IMAGE = (() => {