From a8a32604f7601cfa7cad7f989629529307d2fd9f Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Fri, 14 Aug 2026 10:12:45 -0400 Subject: [PATCH 1/2] fix: await upload session registration --- service/src/service/router.ts | 110 +++++++++++++-------- service/src/service/upload-session.test.ts | 85 ++++++++++++++++ service/src/service/upload-session.ts | 17 ++++ 3 files changed, 171 insertions(+), 41 deletions(-) create mode 100644 service/src/service/upload-session.test.ts create mode 100644 service/src/service/upload-session.ts diff --git a/service/src/service/router.ts b/service/src/service/router.ts index e0f8f44..72c7b64 100644 --- a/service/src/service/router.ts +++ b/service/src/service/router.ts @@ -23,6 +23,7 @@ import { jobsSubmitted } from '../metrics'; import { captureTraceCarrier, withSpan } from '../telemetry'; import { Jobs, Languages } from '../enum'; import { FileRefAuthorizationError, authorizeRequestedFiles } from './file-authorization'; +import { createUploadSessionRegistrar } from './upload-session'; import { prepareSandboxJobSecurity } from '../sandbox-egress'; import logger from '../logger'; @@ -431,9 +432,6 @@ router.post('/upload', uploadLimiter, async (req: t.AuthenticatedRequest, res: R reject(err instanceof Error ? err : new Error(String(err))); return; } - connection.set(`session:${session_id}`, sessionKey, 'EX', env.SESSION_CACHE_TTL); - logger.info(`[${INSTANCE_ID}] Upload: Session ID: ${session_id} | User ID: ${userId} | Session key: ${sessionKey}`); - const putHeaders: Record = { 'Content-Type': mimeType, /* file-server URL-decodes this header before storing metadata. @@ -444,26 +442,35 @@ router.post('/upload', uploadLimiter, async (req: t.AuthenticatedRequest, res: R if (readOnly) { putHeaders['X-Read-Only'] = 'true'; } - axios.put( - `${env.FILE_SERVER_URL}/sessions/${session_id}/objects/${fileId}`, - file, - { - headers: internalServiceHeaders(putHeaders), - maxBodyLength: planFileSize, - maxContentLength: planFileSize, - signal: abortController.signal, - } - ) + connection.set(`session:${session_id}`, sessionKey, 'EX', env.SESSION_CACHE_TTL) + .then(() => { + logger.info(`[${INSTANCE_ID}] Upload: Session ID: ${session_id} | User ID: ${userId} | Session key: ${sessionKey}`); + return axios.put( + `${env.FILE_SERVER_URL}/sessions/${session_id}/objects/${fileId}`, + file, + { + headers: internalServiceHeaders(putHeaders), + maxBodyLength: planFileSize, + maxContentLength: planFileSize, + signal: abortController.signal, + }, + ); + }) .then(response => { clearTimeout(uploadTimeout); resolve(response.data); }) .catch(error => { clearTimeout(uploadTimeout); + file.resume(); reject(error); }); }); + /* Busboy may take additional event-loop turns to drain the multipart + * body before `finish` aggregates this promise. Mark early failures as + * observed while preserving the original promise for Promise.all. */ + void uploadPromise.catch(() => undefined); uploadPromises.push(uploadPromise); }); @@ -541,7 +548,6 @@ router.post('/upload/batch', uploadLimiter, async (req: t.AuthenticatedRequest, let uploadId: string | undefined; let uploadVersionRaw: string | undefined; let readOnly = false; - let sessionKeySet = false; let hasResponded = false; let filesLimitReached = false; /* `SessionKeyResolutionError.status` spans 400 | 500 — 400 is a @@ -552,6 +558,14 @@ router.post('/upload/batch', uploadLimiter, async (req: t.AuthenticatedRequest, * 500 we see and convert it into a single 500 batch response on * `bb.on('finish')`. */ let serverError: SessionKeyResolutionError | undefined; + /* Redis registration is also a batch-level dependency fault. Keep file + * promises fulfilled while Busboy drains, then surface one 500. */ + let sessionRegistrationError: Error | undefined; + + const ensureSessionRegistered = createUploadSessionRegistrar((sessionKey) => { + logger.info(`[${INSTANCE_ID}] Batch upload: Session ID: ${session_id} | User ID: ${userId} | Session key: ${sessionKey}`); + return connection.set(`session:${session_id}`, sessionKey, 'EX', env.SESSION_CACHE_TTL); + }); const planFileSize = planLimits[req.planId ?? '']?.max_file_size ?? planLimits.default.max_file_size; /* See note on the single-upload busboy above for why preservePath is set. */ @@ -635,12 +649,6 @@ router.post('/upload/batch', uploadLimiter, async (req: t.AuthenticatedRequest, resolve({ status: 'error', filename, error: message }); return; } - if (!sessionKeySet) { - connection.set(`session:${session_id}`, sessionKey, 'EX', env.SESSION_CACHE_TTL); - sessionKeySet = true; - logger.info(`[${INSTANCE_ID}] Batch upload: Session ID: ${session_id} | User ID: ${userId} | Session key: ${sessionKey}`); - } - const putHeaders: Record = { 'Content-Type': mimeType, /* file-server URL-decodes this header before storing metadata. @@ -651,7 +659,26 @@ router.post('/upload/batch', uploadLimiter, async (req: t.AuthenticatedRequest, if (readOnly) { putHeaders['X-Read-Only'] = 'true'; } - axios.put( + const failSessionRegistration = (error: unknown): void => { + clearTimeout(uploadTimeout); + file.resume(); + const normalizedError = error instanceof Error ? error : new Error(String(error)); + sessionRegistrationError ??= normalizedError; + resolve({ status: 'error', filename, error: 'Failed to register upload session' }); + }; + const resolveUploadFailure = (error: unknown): void => { + clearTimeout(uploadTimeout); + if (abortController.signal.aborted) { + const reason = abortController.signal.reason === 'timeout' ? 'Upload timeout' : 'File size limit exceeded'; + resolve({ status: 'error', filename, error: reason }); + return; + } + file.resume(); + const message = error instanceof Error ? error.message : 'Unknown upload error'; + logger.error(`[${INSTANCE_ID}] Batch upload file failed: ${filename} | Session: ${session_id}`, { error: message }); + resolve({ status: 'error', filename, error: message }); + }; + const forwardFile = (): Promise => axios.put( `${env.FILE_SERVER_URL}/sessions/${session_id}/objects/${fileId}`, file, { @@ -659,23 +686,15 @@ router.post('/upload/batch', uploadLimiter, async (req: t.AuthenticatedRequest, maxBodyLength: planFileSize, maxContentLength: planFileSize, signal: abortController.signal, - } - ) - .then(response => { - clearTimeout(uploadTimeout); - resolve({ status: 'success', filename: response.data.filename, fileId: response.data.fileId }); - }) - .catch(error => { - clearTimeout(uploadTimeout); - if (abortController.signal.aborted) { - const reason = abortController.signal.reason === 'timeout' ? 'Upload timeout' : 'File size limit exceeded'; - resolve({ status: 'error', filename, error: reason }); - return; - } - const message = error instanceof Error ? error.message : 'Unknown upload error'; - logger.error(`[${INSTANCE_ID}] Batch upload file failed: ${filename} | Session: ${session_id}`, { error: message }); - resolve({ status: 'error', filename, error: message }); - }); + }, + ).then(response => { + clearTimeout(uploadTimeout); + resolve({ status: 'success', filename: response.data.filename, fileId: response.data.fileId }); + }, resolveUploadFailure); + + void ensureSessionRegistered(sessionKey) + .then(forwardFile, failSessionRegistration) + .catch(failSessionRegistration); }); uploadPromises.push(uploadPromise); @@ -701,6 +720,15 @@ router.post('/upload/batch', uploadLimiter, async (req: t.AuthenticatedRequest, try { const results = await Promise.all(uploadPromises); + if (sessionRegistrationError) { + logger.error( + `[${INSTANCE_ID}] Batch upload session registration failed for session ${session_id}:`, + sessionRegistrationError, + ); + res.status(500).json({ error: 'Error registering upload session' }); + return; + } + /* If sessionKey resolution faulted with a 500 status (server * misconfiguration — see `serverError` declaration above), * surface the fault as a single batch-level 500 instead of @@ -721,9 +749,9 @@ router.post('/upload/batch', uploadLimiter, async (req: t.AuthenticatedRequest, return; } - /* SessionKey was set inline in the per-file handler under - * `sessionKeySet`. No batch-level fallback needed: if zero files - * succeeded, no session was created. */ + /* SessionKey was set inline in the per-file handler through + * `ensureSessionRegistered`. No batch-level fallback is needed when + * no valid file reaches the forwarding step. */ let succeeded = 0; let failed = 0; diff --git a/service/src/service/upload-session.test.ts b/service/src/service/upload-session.test.ts new file mode 100644 index 0000000..47a2412 --- /dev/null +++ b/service/src/service/upload-session.test.ts @@ -0,0 +1,85 @@ +import { describe, expect, test } from 'bun:test'; +import { createUploadSessionRegistrar } from './upload-session'; + +function deferred(): { + promise: Promise; + resolve: () => void; + reject: (error: Error) => void; +} { + let resolve!: () => void; + let reject!: (error: Error) => void; + const promise = new Promise((resolvePromise, rejectPromise) => { + resolve = resolvePromise; + reject = rejectPromise; + }); + return { promise, resolve, reject }; +} + +describe('createUploadSessionRegistrar', () => { + test('waits for session registration before forwarding a file', async () => { + const registration = deferred(); + const events: string[] = []; + const ensureSessionRegistered = createUploadSessionRegistrar((sessionKey) => { + events.push(`session:set:start:${sessionKey}`); + return registration.promise.then(() => { + events.push('session:set:done'); + }); + }); + + const result = ensureSessionRegistered('user:user-1').then(async () => { + events.push('file:put'); + return 'uploaded'; + }); + + await Promise.resolve(); + expect(events).toEqual(['session:set:start:user:user-1']); + + registration.resolve(); + expect(await result).toBe('uploaded'); + expect(events).toEqual(['session:set:start:user:user-1', 'session:set:done', 'file:put']); + }); + + test('shares one pending registration across every file in a batch', async () => { + const registration = deferred(); + let registrations = 0; + const forwarded: string[] = []; + const ensureSessionRegistered = createUploadSessionRegistrar(() => { + registrations += 1; + return registration.promise; + }); + + const uploads = [ + ensureSessionRegistered('user:user-1').then(async () => { + forwarded.push('first'); + return 'first'; + }), + ensureSessionRegistered('user:user-1').then(async () => { + forwarded.push('second'); + return 'second'; + }), + ]; + + await Promise.resolve(); + expect(registrations).toBe(1); + expect(forwarded).toEqual([]); + + registration.resolve(); + expect(await Promise.all(uploads)).toEqual(['first', 'second']); + expect(forwarded).toEqual(['first', 'second']); + }); + + test('does not forward files when session registration fails', async () => { + const registration = deferred(); + let forwarded = false; + const ensureSessionRegistered = createUploadSessionRegistrar(() => registration.promise); + const result = ensureSessionRegistered('user:user-1').then(async () => { + forwarded = true; + return 'uploaded'; + }); + + registration.reject(new Error('Redis unavailable')); + + await expect(result).rejects.toThrow('Redis unavailable'); + expect(forwarded).toBe(false); + }); +}); diff --git a/service/src/service/upload-session.ts b/service/src/service/upload-session.ts new file mode 100644 index 0000000..77f6065 --- /dev/null +++ b/service/src/service/upload-session.ts @@ -0,0 +1,17 @@ +type RegisterUploadSession = (sessionKey: string) => Promise; + +/** + * Creates a request-scoped session registrar. Every file shares the first + * registration promise, allowing callers to wait for the Redis write without + * issuing duplicate SETs for a batch. + */ +export function createUploadSessionRegistrar( + registerSession: RegisterUploadSession, +): (sessionKey: string) => Promise { + let sessionRegistered: Promise | undefined; + + return (sessionKey: string): Promise => { + sessionRegistered ??= registerSession(sessionKey); + return sessionRegistered; + }; +} From ed3e94e41f9eb359f71dc457f552da3cbf1d6ab5 Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Fri, 14 Aug 2026 10:51:03 -0400 Subject: [PATCH 2/2] fix: bound batch session registration --- service/src/service/router.ts | 7 +++ service/src/service/upload-session.test.ts | 22 +++++++++ service/src/service/upload-session.ts | 56 ++++++++++++++++++++-- 3 files changed, 82 insertions(+), 3 deletions(-) diff --git a/service/src/service/router.ts b/service/src/service/router.ts index 72c7b64..d88dfdc 100644 --- a/service/src/service/router.ts +++ b/service/src/service/router.ts @@ -613,6 +613,13 @@ router.post('/upload/batch', uploadLimiter, async (req: t.AuthenticatedRequest, const uploadTimeout = setTimeout(() => { abortController.abort('timeout'); file.resume(); + /* If Redis is still pending, make that shared registration barrier + * fail before any file promise settles. This prevents Busboy from + * finishing with a 400 while a later Redis rejection arrives too + * late to be surfaced as the batch-level dependency failure. */ + if (ensureSessionRegistered.rejectPending( + new Error('Upload session registration timed out'), + )) return; resolve({ status: 'error', filename, error: 'Upload timeout' }); }, UPLOAD_TIMEOUT_MS); diff --git a/service/src/service/upload-session.test.ts b/service/src/service/upload-session.test.ts index 47a2412..13ac63c 100644 --- a/service/src/service/upload-session.test.ts +++ b/service/src/service/upload-session.test.ts @@ -82,4 +82,26 @@ describe('createUploadSessionRegistrar', () => { await expect(result).rejects.toThrow('Redis unavailable'); expect(forwarded).toBe(false); }); + + test('keeps a pending registration timeout terminal when Redis rejects later', async () => { + const registration = deferred(); + let forwarded = false; + const ensureSessionRegistered = createUploadSessionRegistrar(() => registration.promise); + const result = ensureSessionRegistered('user:user-1').then(async () => { + forwarded = true; + return 'uploaded'; + }); + + expect(ensureSessionRegistered.rejectPending( + new Error('Upload session registration timed out'), + )).toBe(true); + await expect(result).rejects.toThrow('Upload session registration timed out'); + expect(forwarded).toBe(false); + + registration.reject(new Error('Redis unavailable after timeout')); + await Promise.resolve(); + + expect(forwarded).toBe(false); + expect(ensureSessionRegistered.rejectPending(new Error('too late'))).toBe(false); + }); }); diff --git a/service/src/service/upload-session.ts b/service/src/service/upload-session.ts index 77f6065..2ce13b3 100644 --- a/service/src/service/upload-session.ts +++ b/service/src/service/upload-session.ts @@ -1,5 +1,15 @@ type RegisterUploadSession = (sessionKey: string) => Promise; +export interface UploadSessionRegistrar { + (sessionKey: string): Promise; + /** + * Rejects the shared registration barrier only while Redis is still + * pending. The underlying Redis promise remains observed, but its eventual + * outcome cannot reopen the barrier or forward a file after timeout. + */ + rejectPending(error: Error): boolean; +} + /** * Creates a request-scoped session registrar. Every file shares the first * registration promise, allowing callers to wait for the Redis write without @@ -7,11 +17,51 @@ type RegisterUploadSession = (sessionKey: string) => Promise; */ export function createUploadSessionRegistrar( registerSession: RegisterUploadSession, -): (sessionKey: string) => Promise { +): UploadSessionRegistrar { let sessionRegistered: Promise | undefined; + let registrationPending = false; + let resolveRegistration!: (value: unknown) => void; + let rejectRegistration!: (error: unknown) => void; + + const ensureSessionRegistered = (sessionKey: string): Promise => { + if (!sessionRegistered) { + registrationPending = true; + sessionRegistered = new Promise((resolve, reject) => { + resolveRegistration = resolve; + rejectRegistration = reject; + }); + + let registration: Promise; + try { + registration = registerSession(sessionKey); + } catch (error) { + registrationPending = false; + rejectRegistration(error); + return sessionRegistered; + } - return (sessionKey: string): Promise => { - sessionRegistered ??= registerSession(sessionKey); + void registration.then( + (value) => { + if (!registrationPending) return; + registrationPending = false; + resolveRegistration(value); + }, + (error: unknown) => { + if (!registrationPending) return; + registrationPending = false; + rejectRegistration(error); + }, + ); + } return sessionRegistered; }; + + ensureSessionRegistered.rejectPending = (error: Error): boolean => { + if (!registrationPending) return false; + registrationPending = false; + rejectRegistration(error); + return true; + }; + + return ensureSessionRegistered; }