Add BufferTarget.onFinalize, OutputOptions.onFinalize, ConcurrentRunner (#349)

* Add BufferTarget.onFinalize option for awaitable async action on finalize

The `finalized` event fires synchronously and its return value is ignored,
making it unsuitable for use cases where the muxer should wait (e.g.
uploading the buffer to S3, R2, or other object stores that require a
known Content-Length and therefore can't stream via StreamTarget).

Adds a new `BufferTargetOptions` type with an `onFinalize` callback that
the muxer awaits before resolving. Matches the existing callback pattern
used by `HlsOutputFormatOptions.onSegment`, `onMaster`, etc.

When used with PathedTarget, this provides proper backpressure: the next
segment won't start being produced until the previous one has finished
uploading, keeping memory bounded regardless of video length.

- Adds `BufferTargetOptions` type, exported from the package root
- `BufferTarget` constructor now accepts optional options (backward compatible)
- `_finalize()` awaits `onFinalize` before emitting the `finalized` event
- Adds runtime validation for non-function callbacks
- Updates "Upload to a server" docs in writing-hls.md with S3 pattern
- Documents `onFinalize` in writing-media-files.md BufferTarget section
- Adds tests for callback invocation, async awaiting, backward compat, and validation

* Add ConcurrentRunner, add OutputOptions.onFinalize, adjust docs accordingly, clean up tests

* give me more control baby

* forgot to tell about it

* polish guide and js docs

* vanilagy says remove this, docs will reveal it

---------

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

docs/guide/writing-hls.md

@@ -115,7 +115,9 @@ await output.finalize();
 
 ### Upload to a server
 
-This models a stream upload, where files are being uploaded *while* they are being created.
+#### Stream upload
+
+This code models a stream upload, where files are being uploaded *while* they are being created:
 
 ```ts
 const promises: Promise<Response>[] = [];
@@ -146,17 +148,52 @@ const output = new Output({
 			return new StreamTarget(writable);
 		},
 	),
+	onFinalize: () => Promise.all(promises),
 	// ...
 });
 
 // ...
 await output.finalize();
-
-await Promise.all(promises);
 // All files have been uploaded to the server
 ```
 
-If this is too fancy, you can always use `BufferTarget` instead and upload its contents to a server in a non-streaming way after the `finalized` event.
+#### Monolithic upload
+
+If streaming is not possible (e.g. when uploading to S3 via signed `PutObject`, which requires a known `Content-Length`), you can use [`BufferTarget`](../api/BufferTarget) with the [`onFinalize`](../api/BufferTargetOptions#onfinalize) option instead.
+
+You could call `fetch` directly but this would halt Mediabunny's internals until the upload has completed. Instead, using a [`ConcurrentRunner`](../api/ConcurrentRunner) allows Mediabunny to keep producing data internally while the upload is in flight, while also allowing multiple concurrent uploads:
+
+```ts
+import { ConcurrentRunner, ... } from 'mediabunny';
+
+// This Mediabunny utility class is used to allow up to two requests
+// to run concurrently. When this number is exceeded, backpressure is
+// automatically applied internally.
+const runner = new ConcurrentRunner(2);
+
+const output = new Output({
+	target: new PathedTarget(
+		'master.m3u8',
+		({ path, mimeType }) =>
+			new BufferTarget({
+				onFinalize: buffer => runner.run(() =>
+					fetch(`/upload?file=${encodeURIComponent(path)}`, {
+						method: 'PUT',
+						body: buffer,
+						headers: {
+							'Content-Type': mimeType,
+						},
+					})
+				),
+			}),
+	),
+	onFinalize: () => runner.flush(),
+	// ...
+});
+
+await output.finalize();
+// All files have been uploaded to the server
+```
 
 ## Adding tracks & media
 

docs/guide/writing-media-files.md

@@ -216,6 +216,10 @@ await output.finalize();
 const file = output.target.buffer; // => Uint8Array
 ```
 
+---
+
+An optional [`onFinalize`](../api/OutputOptions#onfinalize) callback can be provided in the output options. This function will be called at the end of `.finalize()` and can be used to do work once the output has been completed. If it returns a promise, it will be awaited by the output.
+
 ## Canceling an output
 
 Sometimes, you may want to cancel the ongoing creation of an output file. For this, use the `cancel` method:
@@ -288,6 +292,24 @@ output.target.buffer; // => ArrayBuffer
 
 This target is a great choice for small-ish files (< 100 MB), but since all data will be kept in memory, using it for large files is suboptimal. If the output gets very large, the page might crash due to memory exhaustion. For these cases, using `StreamTarget` is recommended.
 
+#### `onFinalize` callback
+
+`BufferTarget` accepts an `onFinalize` option, which is called with the complete buffer once the target has been finalized. Useful for uploading the final buffer to a server or object store (e.g. S3 `PutObject`, which requires a known `Content-Length`):
+
+```ts
+const output = new Output({
+	target: new BufferTarget({
+		onFinalize: async (buffer) => {
+			await fetch('/upload', { method: 'PUT', body: buffer });
+		},
+	}),
+	// ...
+});
+
+await output.finalize();
+// The upload has completed by the time finalize resolves.
+```
+
 ### `StreamTarget`
 
 This target passes you the data written by the `Output` in small chunks, requiring you to pipe that data elsewhere to manually assemble the final file. Example use cases include writing the file directly to disk, or uploading it to a server over the network.

scripts/check-docblocks.ts

@@ -16,7 +16,6 @@ const checkDocblocks = (filePath: string) => {
 		if (
 			ts.isInterfaceDeclaration(node)
 			|| ts.isClassDeclaration(node)
-			|| ts.isConstructorDeclaration(node)
 			|| ts.isMethodDeclaration(node)
 			|| ts.isGetAccessorDeclaration(node)
 			|| ts.isSetAccessorDeclaration(node)
@@ -61,13 +60,7 @@ const checkDocblocks = (filePath: string) => {
 			let name = 'anonymous';
 			const kind = ts.SyntaxKind[node.kind].replace(/Declaration|Statement/g, '').toLowerCase();
 
-			if (ts.isConstructorDeclaration(node)) {
+			if ('name' in node && node.name) {
-				// For constructors, use the parent class name
-				const parent = node.parent;
-				if (ts.isClassDeclaration(parent) && parent.name) {
-					name = parent.name.text;
-				}
-			} else if ('name' in node && node.name) {
 				if (ts.isIdentifier(node.name)) {
 					name = node.name.text;
 				} else if ('getText' in node.name) {

src/index.ts

@@ -130,6 +130,7 @@ export {
 	TargetEvents,
 	TargetRequest,
 	BufferTarget,
+	BufferTargetOptions,
 	FilePathTarget,
 	FilePathTargetOptions,
 	NullTarget,
@@ -141,6 +142,7 @@ export {
 } from './target';
 export {
 	AnyIterable,
+	ConcurrentRunner,
 	EventEmitter,
 	EventListenerOptions,
 	FilePath,

src/misc.ts

@@ -1235,3 +1235,66 @@ export class EventEmitter<TEvents extends Record<string, unknown>> {
 }
 
 export const ceilToMultipleOfTwo = (value: number) => Math.ceil(value / 2) * 2;
+
+/**
+ * Utility class for running async functions in parallel up to a certain level of parallelism. Can be used to apply
+ * backpressure only if the concurrency level would be exceeded.
+ *
+ * @group Miscellaneous
+ * @public
+*/
+export class ConcurrentRunner {
+	/** @internal */
+	_queue: Promise<unknown>[] = [];
+	/** @internal */
+	_errored = false;
+
+	/**
+	 * The maximum number of in-flight promises. You can also think of it as the "high water mark".
+	 * You can set this value to dynamically change the level of parallelism.
+	 */
+	parallelism: number;
+
+	constructor(parallelism: number) {
+		this.parallelism = parallelism;
+	}
+
+	/** Whether any function has errored. The runner is effectively bricked if this is `true`, by design. */
+	get errored() {
+		return this._errored;
+	}
+
+	/** The number of tasks currently running. */
+	get inFlightCount() {
+		return this._queue.length;
+	}
+
+	/**
+	 * Schedules an async function to be run. If the maximum allowed level of parallelism has not yet been reached,
+	 * the function will be executed immediately and `run()` will resolve immediately. Otherwise, the function will be
+	 * called as soon as any currently-running function finishes, and `run()` will only resolve then.
+	 *
+	 * Throws if the runner is errored.
+	 */
+	async run(fn: () => Promise<unknown>) {
+		if (this._errored) {
+			await Promise.race(this._queue); // Will surface the error
+		}
+
+		while (this._queue.length >= this.parallelism) {
+			await Promise.race(this._queue);
+		}
+
+		const promise = fn();
+		this._queue.push(promise);
+
+		void promise
+			.then(() => removeItem(this._queue, promise))
+			.catch(() => this._errored = true);
+	}
+
+	/** Waits for all currently running functions to finish. Throws if the runner is errored. */
+	async flush() {
+		await Promise.all(this._queue);
+	}
+}

src/output.ts

@@ -324,6 +324,11 @@ export type OutputOptions<
 	 * When this is a function, it will only be called if an init target is needed.
 	 */
 	initTarget?: T | (() => MaybePromise<T>);
+	/**
+	 * Optional; a callback to be called at the end of {@link Output.finalize}. Can be used to run logic once the
+	 * output has completed. If a promise is returned, it will be awaited internally by {@link Output.finalize}.
+	 */
+	onFinalize?: () => MaybePromise<unknown>;
 };
 
 /**
@@ -369,6 +374,8 @@ export class Output<
 	/** @internal */
 	private _initTarget: T | (() => MaybePromise<T>) | null;
 	/** @internal */
+	_onFinalize: (() => MaybePromise<unknown>) | null = null;
+	/** @internal */
 	_muxer: Muxer;
 	/** @internal */
 	_targets = new Set<Target>();
@@ -449,9 +456,13 @@ export class Output<
 				+ ' a Target.',
 			);
 		}
+		if (options.onFinalize !== undefined && typeof options.onFinalize !== 'function') {
+			throw new TypeError('options.onFinalize, when provided, must be a function.');
+		}
 
 		this.format = options.format;
 		this._target = options.target;
+		this._onFinalize = options.onFinalize ?? null;
 
 		this._initTarget = options.initTarget ?? null;
 		if (this._initTarget instanceof Target) {
@@ -881,6 +892,10 @@ export class Output<
 					}
 				}
 
+				if (this._onFinalize) {
+					await this._onFinalize();
+				}
+
 				this.state = 'finalized';
 			} finally {
 				release();

src/target.ts

@@ -89,6 +89,22 @@ export abstract class Target extends EventEmitter<TargetEvents> {
 const ARRAY_BUFFER_INITIAL_SIZE = 2 ** 16;
 const ARRAY_BUFFER_MAX_SIZE = 2 ** 32;
 
+/**
+ * Options for {@link BufferTarget}.
+ * @group Output targets
+ * @public
+ */
+export type BufferTargetOptions = {
+	/**
+	 * Called once the target has been finalized, with the complete output buffer. If you return a promise, it will be
+	 * used to apply backpressure internally.
+	 *
+	 * One use for this callback is for uploading to a server where the full buffer must be known before
+	 * sending (e.g. S3 PutObject) and stream-uploading is not an option.
+	 */
+	onFinalize?: (buffer: ArrayBuffer) => MaybePromise<unknown>;
+};
+
 /**
  * A target that writes data directly into an ArrayBuffer in memory. Great for performance, but not suitable for very
  * large files. The buffer will be available once the output has been finalized.
@@ -107,11 +123,22 @@ export class BufferTarget extends Target {
 	_maxPos = 0;
 	/** @internal */
 	_supportsResize: boolean;
+	/** @internal */
+	_options: BufferTargetOptions;
 
 	/** Creates a new {@link BufferTarget}. The buffer holding the data will be created and managed internally. */
-	constructor() {
+	constructor(options: BufferTargetOptions = {}) {
 		super();
 
+		if (!options || typeof options !== 'object') {
+			throw new TypeError('BufferTarget options, when provided, must be an object.');
+		}
+		if (options.onFinalize !== undefined && typeof options.onFinalize !== 'function') {
+			throw new TypeError('options.onFinalize, when provided, must be a function.');
+		}
+
+		this._options = options;
+
 		this._supportsResize = 'resize' in new ArrayBuffer(0);
 		if (this._supportsResize) {
 			try {
@@ -178,6 +205,11 @@ export class BufferTarget extends Target {
 	/** @internal */
 	async _finalize() {
 		this.buffer = this._buffer.slice(0, this._maxPos);
+
+		if (this._options.onFinalize) {
+			await this._options.onFinalize(this.buffer);
+		}
+
 		this._emit('finalized');
 	}
 

test/node/concurrent-runner.test.ts

@@ -0,0 +1,289 @@
+import { expect, test } from 'vitest';
+import { ConcurrentRunner, promiseWithResolvers } from '../../src/misc.js';
+
+test('parallelism of 1 runs tasks strictly sequentially', async () => {
+	const runner = new ConcurrentRunner(1);
+	let entered = false;
+	let overlapped = false;
+
+	const makeTask = () => async () => {
+		if (entered) {
+			overlapped = true;
+		}
+		entered = true;
+		await Promise.resolve();
+		await Promise.resolve();
+		entered = false;
+	};
+
+	const runPromises: Promise<void>[] = [];
+	for (let i = 0; i < 5; i++) {
+		runPromises.push(runner.run(makeTask()));
+	}
+
+	await Promise.all(runPromises);
+	await runner.flush();
+
+	expect(overlapped).toBe(false);
+	expect(entered).toBe(false);
+});
+
+test('run schedules tasks up to parallelism without waiting', async () => {
+	const runner = new ConcurrentRunner(3);
+	const d1 = promiseWithResolvers();
+	const d2 = promiseWithResolvers();
+	const d3 = promiseWithResolvers();
+
+	let started = 0;
+
+	await runner.run(async () => {
+		started++;
+		await d1.promise;
+	});
+	await runner.run(async () => {
+		started++;
+		await d2.promise;
+	});
+	await runner.run(async () => {
+		started++;
+		await d3.promise;
+	});
+
+	// All three should have started synchronously since we're under parallelism
+	expect(started).toBe(3);
+
+	d1.resolve();
+	d2.resolve();
+	d3.resolve();
+	await runner.flush();
+});
+
+test('run blocks when the queue is full and resumes as slots free up', async () => {
+	const runner = new ConcurrentRunner(2);
+	const d1 = promiseWithResolvers();
+	const d2 = promiseWithResolvers();
+	const d3 = promiseWithResolvers();
+
+	let startedThird = false;
+
+	await runner.run(() => d1.promise);
+	await runner.run(() => d2.promise);
+
+	// Kick off a third run. It must not resolve until one of the first two finishes.
+	const thirdRun = runner.run(async () => {
+		startedThird = true;
+		await d3.promise;
+	});
+
+	await flushMicrotasks();
+	expect(startedThird).toBe(false);
+
+	// Free a slot by resolving the first task.
+	d1.resolve();
+	await thirdRun;
+
+	expect(startedThird).toBe(true);
+
+	d2.resolve();
+	d3.resolve();
+	await runner.flush();
+});
+
+test('simultaneous run calls respect parallelism', async () => {
+	const runner = new ConcurrentRunner(2);
+	const deferreds = [
+		promiseWithResolvers(),
+		promiseWithResolvers(),
+		promiseWithResolvers(),
+		promiseWithResolvers(),
+	];
+	const started: number[] = [];
+
+	const runPromises = deferreds.map((d, i) => runner.run(async () => {
+		started.push(i);
+		await d.promise;
+	}));
+
+	// Give the runner a chance to start the first batch, then verify only the first two ran.
+	await flushMicrotasks();
+	expect(started).toEqual([0, 1]);
+
+	// The first two run() calls should have resolved already (they found open slots),
+	// the last two should still be waiting.
+	let settled = 0;
+	void Promise.all(runPromises).then(() => settled++);
+	await flushMicrotasks();
+	expect(settled).toBe(0);
+
+	// Unblock the first task. That should let task 2 in.
+	deferreds[0]!.resolve();
+	await flushMicrotasks();
+	expect(started).toEqual([0, 1, 2]);
+
+	// Unblock the second task. That should let task 3 in.
+	deferreds[1]!.resolve();
+	await flushMicrotasks();
+	expect(started).toEqual([0, 1, 2, 3]);
+
+	// All four run() calls should resolve once their slot has been acquired.
+	await Promise.all(runPromises);
+
+	deferreds[2]!.resolve();
+	deferreds[3]!.resolve();
+	await runner.flush();
+});
+
+test('errored task surfaces on the next run call', async () => {
+	const runner = new ConcurrentRunner(2);
+	const error = new Error('boom');
+
+	expect(runner.errored).toBe(false);
+
+	await runner.run(async () => {
+		throw error;
+	});
+
+	await flushMicrotasks();
+	expect(runner.errored).toBe(true);
+
+	await expect(runner.run(async () => {})).rejects.toBe(error);
+});
+
+test('errored task surfaces on flush', async () => {
+	const runner = new ConcurrentRunner(2);
+	const error = new Error('kaboom');
+
+	await runner.run(async () => {
+		throw error;
+	});
+
+	await expect(runner.flush()).rejects.toBe(error);
+	expect(runner.errored).toBe(true);
+});
+
+test('error from a slow task surfaces on subsequent run even after other tasks completed', async () => {
+	const runner = new ConcurrentRunner(2);
+	const slow = promiseWithResolvers();
+	const error = new Error('late');
+
+	await runner.run(async () => {
+		await slow.promise;
+		throw error;
+	});
+	await runner.run(async () => {});
+
+	// Let the fast task finish cleanly.
+	await flushMicrotasks();
+	expect(runner.errored).toBe(false);
+
+	// Now let the slow task reject.
+	slow.resolve();
+	await flushMicrotasks();
+	expect(runner.errored).toBe(true);
+
+	await expect(runner.run(async () => {})).rejects.toBe(error);
+});
+
+test('parallelism can be mutated at runtime to grow or shrink the in-flight limit', async () => {
+	const runner = new ConcurrentRunner(1);
+	const d1 = promiseWithResolvers();
+	const d2 = promiseWithResolvers();
+	const d3 = promiseWithResolvers();
+
+	let startedSecond = false;
+	let startedThird = false;
+
+	await runner.run(() => d1.promise);
+	expect(runner.inFlightCount).toBe(1);
+
+	// Grow to 2 before scheduling the next task — second run sees the new value and starts immediately.
+	runner.parallelism = 2;
+	await runner.run(async () => {
+		startedSecond = true;
+		await d2.promise;
+	});
+	expect(startedSecond).toBe(true);
+	expect(runner.inFlightCount).toBe(2);
+
+	// Shrink to 1 while 2 are in flight. No task is cancelled; a new run() must wait until queue < 1.
+	runner.parallelism = 1;
+	const third = runner.run(async () => {
+		startedThird = true;
+		await d3.promise;
+	});
+	await flushMicrotasks();
+	expect(startedThird).toBe(false);
+
+	// Draining one frees a slot but queue is still at 1 (>= new parallelism), third stays blocked.
+	d1.resolve();
+	await flushMicrotasks();
+	expect(startedThird).toBe(false);
+
+	// Draining the second lets third in.
+	d2.resolve();
+	await third;
+	expect(startedThird).toBe(true);
+
+	d3.resolve();
+	await runner.flush();
+});
+
+test('inFlightCount tracks currently running tasks', async () => {
+	const runner = new ConcurrentRunner(3);
+	const d1 = promiseWithResolvers();
+	const d2 = promiseWithResolvers();
+
+	expect(runner.inFlightCount).toBe(0);
+
+	await runner.run(() => d1.promise);
+	expect(runner.inFlightCount).toBe(1);
+
+	await runner.run(() => d2.promise);
+	expect(runner.inFlightCount).toBe(2);
+
+	d1.resolve();
+	await flushMicrotasks();
+	expect(runner.inFlightCount).toBe(1);
+
+	d2.resolve();
+	await runner.flush();
+	expect(runner.inFlightCount).toBe(0);
+});
+
+test('flush waits for all in-flight tasks', async () => {
+	const runner = new ConcurrentRunner(3);
+	const deferreds = [promiseWithResolvers(), promiseWithResolvers(), promiseWithResolvers()];
+	const completed: number[] = [];
+
+	for (let i = 0; i < deferreds.length; i++) {
+		await runner.run(async () => {
+			await deferreds[i]!.promise;
+			completed.push(i);
+		});
+	}
+
+	let flushResolved = false;
+	const flushPromise = runner.flush().then(() => {
+		flushResolved = true;
+	});
+
+	await flushMicrotasks();
+	expect(flushResolved).toBe(false);
+
+	deferreds[0]!.resolve();
+	await flushMicrotasks();
+	expect(flushResolved).toBe(false);
+
+	deferreds[1]!.resolve();
+	deferreds[2]!.resolve();
+	await flushPromise;
+
+	expect(flushResolved).toBe(true);
+	expect(completed.sort()).toEqual([0, 1, 2]);
+});
+
+const flushMicrotasks = async (iterations = 10) => {
+	for (let i = 0; i < iterations; i++) {
+		await Promise.resolve();
+	}
+};

test/node/output.test.ts

@@ -0,0 +1,36 @@
+import { expect, test } from 'vitest';
+import { Output } from '../../src/output.js';
+import { MkvOutputFormat } from '../../src/output-format.js';
+import { BufferTarget } from '../../src/target.js';
+import { EncodedVideoPacketSource } from '../../src/media-source.js';
+import { EncodedPacket } from '../../src/packet.js';
+
+test('Output, onFinalize', async () => {
+	let callCount = 0;
+	const output = new Output({
+		format: new MkvOutputFormat(),
+		target: new BufferTarget(),
+		onFinalize: async () => {
+			await new Promise(resolve => setTimeout(resolve, 200));
+
+			callCount++;
+		},
+	});
+
+	const source = new EncodedVideoPacketSource('avc');
+	output.addVideoTrack(source);
+
+	await output.start();
+
+	await source.add(new EncodedPacket(new Uint8Array(1024), 'key', 0, 0.5), {
+		decoderConfig: {
+			codec: 'avc1.640028',
+			codedWidth: 1920,
+			codedHeight: 1080,
+		},
+	});
+
+	await output.finalize();
+
+	expect(callCount).toBe(1);
+});

test/node/target.test.ts

@@ -0,0 +1,42 @@
+import path from 'node:path';
+import { expect, test } from 'vitest';
+import { Input } from '../../src/input.js';
+import { FilePathSource } from '../../src/source.js';
+import { ALL_FORMATS } from '../../src/input-format.js';
+import { Output } from '../../src/output.js';
+import { BufferTarget } from '../../src/target.js';
+import { Mp4OutputFormat } from '../../src/output-format.js';
+import { Conversion } from '../../src/conversion.js';
+
+const __dirname = new URL('.', import.meta.url).pathname;
+
+const samplePath = path.join(__dirname, '../public/video.mp4');
+
+test('BufferTarget onFinalize callback', async () => {
+	let received: ArrayBuffer | null = null;
+	let asyncCallbackDone = false;
+
+	using input = new Input({
+		source: new FilePathSource(samplePath),
+		formats: ALL_FORMATS,
+	});
+
+	const output = new Output({
+		format: new Mp4OutputFormat(),
+		target: new BufferTarget({
+			onFinalize: async (buffer) => {
+				received = buffer;
+				await new Promise(resolve => setTimeout(resolve, 20));
+				asyncCallbackDone = true;
+			},
+		}),
+	});
+
+	const conversion = await Conversion.init({ input, output, showWarnings: false });
+	await conversion.execute();
+
+	expect(received).not.toBeNull();
+	expect(received).toBe(output.target.buffer);
+	expect(asyncCallbackDone).toBe(true);
+	expect(output.target.buffer!.byteLength).toBeGreaterThan(0);
+});