fix: two long-running HLS transcode issues (#355)

* fix: release targets from Output._targets on finalize

Long-running HLS transcodes leak memory. Every finalized BufferTarget
stays in _targets until the outer Output closes, pinning its buffer.
Writer.finalize() already does this cleanup for writer-based flows;
extend it to buffer-finalize paths via the public 'finalized' event.

* fix: ReadOrchestrator LRU eviction picks only drained workers

assert(pendingSlices.length === 0) fires under heavy concurrent reads
(e.g. multi-rendition HLS decode from BlobSource). LRU filter only
checked !running; workers with queued slices could be evicted.
Add pendingSlices.length === 0 to the filter.

* fix: export AppendOnlyStreamTarget from index

Missing from the re-export; public docs import it by name.

* fix: join HLS init segment path with root + playlist path

Init path went through _getTarget bare; segments got joined with rootPath + playlist.path. Playlist-relative URI then can't resolve when the playlist lives in a subdirectory.

* Fix targets not being cleaned up, fix paused workers with remaining pending slices, modify doc block, fixed isRoot not being changed on proxied requests

---------

Co-authored-by: Vanilagy <[email protected]>

dev/convert.html

@@ -23,7 +23,9 @@
 			chunked: true,
 			chunkSize: 2**20
 		});
-		const outputFormat = new Mediabunny.Mp4OutputFormat();
+		const outputFormat = new Mediabunny.HlsOutputFormat({
+			segmentFormat: new Mediabunny.MpegTsOutputFormat(),
+		});
 
 		const p = document.createElement('p');
 		p.textContent = 'Capturing...';
@@ -50,7 +52,7 @@
 
 		const output = new Mediabunny.Output({
 			format: outputFormat,
-			target
+			target: new Mediabunny.PathedTarget('master.m3u8', ({ path }) => new Mediabunny.BufferTarget()),
 		});
 
 		let input;
@@ -119,76 +121,13 @@
 				bitrate: 320000
 			},
 			*/
-			video: (track) => ({
+			video: [
-				discard: true,
+				{ height: 1080 },
-				//discard: true,
+				{ height: 720 },
-				//discard: !tracks.includes(track),
+				{ height: 480 },
-				//forceTranscode: true,
+				{ height: 360 },
-				//forceTranscode: true,
+				{ height: 240 },
-				//width: 1280,
+			],
-				//forceTranscode: true,
-				//allowRotationMetadata: false,
-				//width: 720,
-				//frameRate: 30,
-				//bitrate: Mediabunny.QUALITY_VERY_LOW,
-				//discard: true,
-				/*
-				process: (sample) => {
-					if (!ctx) {
-						// Create a canvas for image compositing
-						const canvas = new OffscreenCanvas(
-							sample.displayWidth,
-							sample.displayHeight,
-						);
-						ctx = canvas.getContext('2d');
-					}
-
-					console.log(ctx.canvas.width, ctx.canvas.height);
-
-					ctx.clearRect(0, 0, ctx.canvas.width, ctx.canvas.height);
-					sample.drawWithFit(ctx, { fit: 'fill' });
-					//ctx.drawImage(watermark, 32, 32);
-
-					return ctx.canvas;
-				},
-				*/
-				//width: 300,
-				//alpha: 'keep',
-				//width: 320,
-				//discard: true,
-				//discard: true,
-				//crop: {
-				//	left: 0,
-				//	top: 0,
-				//	width: 500,
-				//	height: 500,
-				//},
-				//rotate: 90,
-				//width: 200,
-				//height: 500,
-				//fit: 'contain',
-				//forceTranscode: true,
-				//codec: 'avc',
-				//fit: 'contain',
-				//frameRate: 27.123,
-				//width: 320,
-				//forceTranscode: true,
-				//codec: 'av1',
-				//discard: true,
-				//width: 1280,
-				//discard: true,
-				//width: 640
-				//forceTranscode: true,
-				//rotate: 90
-				//width: 720 ?? 2160,
-				//height: 1280 ?? 3840,
-				//fit: 'contain',
-				//rotate: 90,
-				//width: 512,
-				//height: 512,
-				//width: 200,
-				//height: 100,
-			}),
 			tags: {} ?? {
 				title: 'Bigggy',
 				artist: 'Buck Bunny',
@@ -207,7 +146,7 @@
 				}
 			},
 			trim: {
-				end: 5,
+				//end: 5,
 				//start,
 				//end: start + 5,
 				////start: 0,

src/hls/hls-muxer.ts

@@ -898,6 +898,11 @@ export class HlsMuxer extends Muxer {
 				target: new PathedTarget(
 					fullSegmentPath,
 					async (request: TargetRequest) => {
+						const proxiedRequest: TargetRequest = {
+							...request,
+							isRoot: false,
+						};
+
 						if (request.isRoot) {
 							if (playlist.singleFile) {
 								const slice = playlist.singleFile.target.slice(playlist.singleFile.nextOffset);
@@ -905,7 +910,7 @@ export class HlsMuxer extends Muxer {
 
 								return slice;
 							} else {
-								const target = await this.output._getTarget(request);
+								const target = await this.output._getTarget(proxiedRequest);
 								outputTarget = target;
 								target.on('write', ({ end }) => segmentSize = Math.max(segmentSize, end));
 
@@ -913,7 +918,7 @@ export class HlsMuxer extends Muxer {
 							}
 						}
 
-						return this.output._getTarget(request);
+						return this.output._getTarget(proxiedRequest);
 					},
 				),
 				initTarget: async () => {
@@ -955,8 +960,12 @@ export class HlsMuxer extends Muxer {
 							info: null,
 						};
 
+						const fullInitPath = joinPaths(
+							joinPaths(pathedTarget.rootPath, playlist.path),
+							initPath,
+						);
 						const target = await this.output._getTarget({
-							path: initPath,
+							path: fullInitPath,
 							isRoot: false,
 							mimeType: playlist.segmentFormat.mimeType,
 						});

src/hls/hls-segmented-input.ts

@@ -11,7 +11,7 @@ import { ENCRYPTION_KEY_CACHE_GROUP, Input } from '../input';
 import { Segment, SegmentedInput, SegmentedInputTrackDeclaration, SegmentRetrievalOptions } from '../segmented-input';
 import { toDataView, joinPaths, last, assert, binarySearchLessOrEqual, arrayArgmin, wait } from '../misc';
 import { readAllLines, readBytes, Reader } from '../reader';
-import { CustomPathedSource, ReadableStreamSource, SourceRef } from '../source';
+import { CustomPathedSource, ReadableStreamSource, SourceRef, SourceRequest } from '../source';
 import { HlsDemuxer } from './hls-demuxer';
 import {
 	AttributeList,
@@ -568,6 +568,11 @@ export class HlsSegmentedInput extends SegmentedInput {
 				async (request) => {
 					assert(request.isRoot); // Shouldn't fail since we don't allow recursive HLS
 
+					const proxiedRequest: SourceRequest = {
+						...request,
+						isRoot: false,
+					};
+
 					let ref: SourceRef;
 					const needsSlice = hlsSegment.location.offset > 0 || hlsSegment.location.length !== null;
 
@@ -576,7 +581,7 @@ export class HlsSegmentedInput extends SegmentedInput {
 						|| hlsSegment.encryption.method === 'SAMPLE-AES'
 						|| hlsSegment.encryption.method === 'SAMPLE-AES-CTR'
 					) {
-						ref = await this.input._getSourceCached(request);
+						ref = await this.input._getSourceCached(proxiedRequest);
 
 						if (needsSlice) {
 							const slice = ref.source.slice(
@@ -591,7 +596,7 @@ export class HlsSegmentedInput extends SegmentedInput {
 						const encryption = hlsSegment.encryption;
 						assert(encryption.iv);
 
-						let ciphertextRef = await this.input._getSourceCached(request);
+						let ciphertextRef = await this.input._getSourceCached(proxiedRequest);
 						if (needsSlice) {
 							// Slice before decrypting
 							const slice = ciphertextRef.source.slice(

src/index.ts

@@ -129,16 +129,17 @@ export {
 	Target,
 	TargetEvents,
 	TargetRequest,
+	AppendOnlyStreamTarget,
 	BufferTarget,
 	BufferTargetOptions,
 	FilePathTarget,
 	FilePathTargetOptions,
 	NullTarget,
+	PathedTarget,
 	RangedTarget,
 	StreamTarget,
 	StreamTargetOptions,
 	StreamTargetChunk,
-	PathedTarget,
 } from './target';
 export {
 	AnyIterable,

src/output.ts

@@ -378,7 +378,7 @@ export class Output<
 	/** @internal */
 	_muxer: Muxer;
 	/** @internal */
-	_targets = new Set<Target>();
+	_unfinalizedTargets = new Set<Target>();
 	/** @internal */
 	_rootWriterPromise: Promise<Writer> | null = null;
 	/** @internal */
@@ -439,12 +439,7 @@ export class Output<
 			throw new TypeError('options.target must be a Target or a PathedTarget.');
 		}
 		if (options.target instanceof Target) {
-			if (options.target._output) {
+			this._rememberTarget(options.target);
-				throw new Error('Target is already used for another output.');
-			}
-
-			options.target._output = this;
-			this._targets.add(options.target);
 		}
 		if (
 			options.initTarget !== undefined
@@ -466,8 +461,7 @@ export class Output<
 
 		this._initTarget = options.initTarget ?? null;
 		if (this._initTarget instanceof Target) {
-			this._initTarget._output = this;
+			this._rememberTarget(this._initTarget);
-			this._targets.add(this._initTarget);
 		}
 
 		this._muxer = options.format._createMuxer(this);
@@ -498,18 +492,23 @@ export class Output<
 		assert(this._target instanceof PathedTarget);
 
 		const target = await this._getTargetValidated(request);
-		target._output = this;
 		this._emit('target', { target, request, isRoot: request.isRoot });
 
 		if (this.state === 'canceled') {
 			await target._close();
 		} else {
-			this._targets.add(target);
+			this._rememberTarget(target);
 		}
 
 		return target;
 	}
 
+	/** @internal */
+	_rememberTarget(target: Target) {
+		this._unfinalizedTargets.add(target);
+		target.on('finalized', () => this._unfinalizedTargets.delete(target), { once: true });
+	}
+
 	/** @internal */
 	async _getInitTarget(): Promise<T> {
 		assert(this._initTarget !== null);
@@ -519,12 +518,11 @@ export class Output<
 		}
 
 		const target = await this._initTarget();
-		target._output = this;
 
 		if (this.state === 'canceled') {
 			await target._close();
 		} else {
-			this._targets.add(target);
+			this._rememberTarget(target);
 		}
 
 		return target;
@@ -558,13 +556,11 @@ export class Output<
 		const result = this._getTargetValidated(request);
 
 		const handleResult = (target: T) => {
-			target._output = this;
-
 			if (this.state === 'canceled') {
 				// Promise thrown away here, but no way to surface it to the user really
 				void target._close();
 			} else {
-				this._targets.add(target);
+				this._rememberTarget(target);
 			}
 
 			this._emit('target', { target, request, isRoot: true });
@@ -849,8 +845,8 @@ export class Output<
 				const promises = this._tracks.map(x => x.source._flushOrWaitForOngoingClose(true)); // Force close
 				await Promise.all(promises);
 
-				await Promise.all([...this._targets].map(target => target._close()));
+				await Promise.all([...this._unfinalizedTargets].map(target => target._close()));
-				this._targets.clear();
+				this._unfinalizedTargets.clear();
 			} finally {
 				release();
 			}

src/source.ts

@@ -558,7 +558,7 @@ export class BlobSource extends Source {
 			}
 		}
 
-		worker.running = false;
+		this._orchestrator.signalWorkerStoppedRunning(worker);
 
 		if (worker.aborted) {
 			// MDN: "Calling this method signals a loss of interest in the stream by a consumer."
@@ -853,7 +853,7 @@ export class UrlSource extends PathedSource {
 			while (true) {
 				if (worker.currentPos >= worker.targetPos || worker.aborted) {
 					abortController.abort();
-					worker.running = false;
+					this._orchestrator.signalWorkerStoppedRunning(worker);
 
 					return;
 				}
@@ -1201,7 +1201,7 @@ export class StreamSource extends Source {
 			}
 		}
 
-		worker.running = false;
+		this._orchestrator.signalWorkerStoppedRunning(worker);
 	}
 
 	/** @internal */
@@ -1608,6 +1608,8 @@ class ReadOrchestrator {
 		minReadPosition: number,
 		maxReadPosition: number,
 	): MaybePromise<ReadResult | null> {
+		assert(!this.disposed);
+
 		const prefetchRange = this.options.prefetchProfile(innerStart, innerEnd, this.workers);
 		const outerStart = Math.max(prefetchRange.start, minReadPosition);
 		const outerEnd = Math.min(prefetchRange.end, this.fileSize ?? Infinity, maxReadPosition);
@@ -1868,7 +1870,11 @@ class ReadOrchestrator {
 			for (let i = 0; i < this.workers.length; i++) {
 				const worker = this.workers[i]!;
 
-				if (!worker.running && (!oldestWorker || worker.age < oldestWorker.age)) {
+				if (
+					!worker.running
+					&& worker.pendingSlices.length === 0
+					&& (!oldestWorker || worker.age < oldestWorker.age)
+				) {
 					oldestIndex = i;
 					oldestWorker = worker;
 				}
@@ -1952,13 +1958,18 @@ class ReadOrchestrator {
 		// Here we merge everything into one "megaworker" that spans the entire file. We assume the passed-in worker
 		// is already configured to be a megaworker.
 
+		const uniqueSlices = new Set(worker.pendingSlices);
+
 		for (let i = 0; i < this.workers.length; i++) {
 			const otherWorker = this.workers[i]!;
 			if (otherWorker === worker) {
 				continue;
 			}
 
-			worker.pendingSlices.push(...otherWorker.pendingSlices);
+			for (const slice of otherWorker.pendingSlices) {
+				uniqueSlices.add(slice);
+			}
+
 			otherWorker.aborted = true;
 			otherWorker.pendingSlices.length = 0;
 			this.workers.splice(i, 1);
@@ -1967,9 +1978,13 @@ class ReadOrchestrator {
 
 		for (let i = 0; i < this.queuedReads.length; i++) {
 			const queuedRead = this.queuedReads[i]!;
-			worker.pendingSlices.push(...queuedRead.pendingSlices);
+
+			for (const slice of queuedRead.pendingSlices) {
+				uniqueSlices.add(slice);
+			}
 		}
 
+		worker.pendingSlices = [...uniqueSlices];
 		this.queuedReads.length = 0;
 	}
 
@@ -2101,6 +2116,16 @@ class ReadOrchestrator {
 		}
 	}
 
+	signalWorkerStoppedRunning(worker: ReadWorker) {
+		worker.running = false;
+
+		// When a worker stops running, that means it has hit its targetPos. It might still have pendingSlices assigned,
+		// but this is because those pending slices cover data that other workers are assigned to fill. Since targetPos
+		// has been reached, we can confidently say that this worker has completed its share of work on the pending
+		// slices and must no longer care about them.
+		worker.pendingSlices.length = 0;
+	}
+
 	/** Called when a worker reaches the end of the underlying data and must be cleaned up. */
 	onWorkerFinished(worker: ReadWorker) {
 		const index = this.workers.indexOf(worker);

src/target.ts

@@ -7,7 +7,6 @@
  */
 
 import type { FileHandle } from 'node:fs/promises';
-import { Output } from './output';
 import * as nodeAlias from './node';
 import { assert, EventEmitter, FilePath, MaybePromise } from './misc';
 
@@ -39,7 +38,7 @@ export type TargetEvents = {
  */
 export abstract class Target extends EventEmitter<TargetEvents> {
 	/** @internal */
-	_output: Output | null = null;
+	_writerAcquired = false;
 
 	/** @internal */
 	_monotonicity: boolean | null = null; // null = unknown
@@ -582,6 +581,17 @@ export class StreamTarget extends Target {
 	}
 }
 
+/**
+ * This target writes to a `WritableStream<Uint8Array>`, meaning all writes are necessarily append-only and involve no
+ * seeking. Great for streaming data to a source that can only accept sequential data, like an HTTP server processing
+ * an incoming upload.
+ *
+ * Note that using this target *requires* that the underlying format write data sequentially. Not all formats do this,
+ * and this target will throw for the formats that don't. Check the guide for more.
+ *
+ * @group Output targets
+ * @public
+ */
 export class AppendOnlyStreamTarget extends Target {
 	/** @internal */
 	_writable: WritableStream<Uint8Array>;
@@ -787,7 +797,6 @@ export class RangedTarget extends Target {
 
 		this._baseTarget = baseTarget;
 		this._offset = offset;
-		this._output = baseTarget._output;
 	}
 
 	/** @internal */

src/writer.ts

@@ -17,8 +17,13 @@ export class Writer {
 	private pos = 0;
 
 	constructor(target: Target, isMonotonic: boolean) {
+		if (target._writerAcquired) {
+			throw new Error('Can\'t have multiple Writers for the same Target.');
+		}
+
 		this.target = target;
 		target._setMonotonicity(isMonotonic);
+		target._writerAcquired = true;
 	}
 
 	start() {
@@ -55,10 +60,8 @@ export class Writer {
 	/** Called after muxing has finished. */
 	async finalize() {
 		assert(this.started && !this.finalized);
-		assert(this.target._output);
 
 		await this.target._finalize();
-		this.target._output._targets.delete(this.target);
 
 		this.finalized = true;
 	}

test/node/hls-input.test.ts

@@ -948,3 +948,24 @@ test.concurrent('Widevine encryption (SAMPLE-AES-CTR) succeeds with buffer keys'
 	assert(lastPacket);
 	expect(lastPacket.timestamp + lastPacket.duration).toBe(60);
 });
+
+test.concurrent('SourceRequest.isRoot', async () => {
+	using input = new Input({
+		source: new CustomPathedSource(
+			'https://test-streams.mux.dev/x36xhzz/x36xhzz.m3u8',
+			({ path, isRoot }) => {
+				if (isRoot) {
+					expect(path).toBe('https://test-streams.mux.dev/x36xhzz/x36xhzz.m3u8');
+				}
+
+				return new UrlSource(path);
+			},
+		),
+		formats: ALL_FORMATS,
+	});
+
+	const videoTrack = await input.getPrimaryVideoTrack();
+	assert(videoTrack);
+
+	await videoTrack.computeDuration();
+});

test/node/hls-output.test.ts

@@ -2903,3 +2903,37 @@ test('Append-only stream with monotonicity violation', async () => {
 	await expect(source.add(new EncodedPacket(avcPacketData, 'key', 2, 0), avcMetadata))
 		.rejects.toThrow('AppendOnlyStreamTarget');
 });
+
+test('Relative paths & isRoot', async () => {
+	const output = new Output({
+		format: new HlsOutputFormat({
+			segmentFormat: new CmafOutputFormat(),
+			getPlaylistPath: () => `a/folder/playlist.m3u8`,
+		}),
+		target: new PathedTarget('path/to/master.m3u8', (request) => {
+			if (request.isRoot) {
+				expect(request.path).toBe('path/to/master.m3u8');
+			} else {
+				expect(request.path.startsWith('path/to/a/folder/')).toBe(true);
+			}
+
+			return new BufferTarget();
+		}),
+	});
+
+	const source = videoSource();
+	output.addVideoTrack(source);
+
+	await output.start();
+
+	await source.add(new EncodedPacket(avcPacketData, 'key', 0, 0), avcMetadata);
+	await source.add(new EncodedPacket(avcPacketData, 'delta', 0.5, 0), avcMetadata);
+	await source.add(new EncodedPacket(avcPacketData, 'delta', 1, 0), avcMetadata);
+	await source.add(new EncodedPacket(avcPacketData, 'delta', 1.5, 0), avcMetadata);
+	await source.add(new EncodedPacket(avcPacketData, 'key', 2, 0), avcMetadata);
+	await source.add(new EncodedPacket(avcPacketData, 'delta', 2.5, 0), avcMetadata);
+	await source.add(new EncodedPacket(avcPacketData, 'delta', 3, 0), avcMetadata);
+	await source.add(new EncodedPacket(avcPacketData, 'delta', 3.5, 0), avcMetadata);
+
+	await output.finalize();
+});