fix: bound memory buffering for streamed file uploads

service/src/file-server.ts

@@ -11,11 +11,11 @@ import { sendFileDownload } from './file-download';
 import path from 'path';
 import IORedis from 'ioredis';
 import express from 'express';
-import { Client } from 'minio';
+import { createMinioClient } from './minio-client';
 import { nanoid } from 'nanoid';
 import { PassThrough } from 'stream';
 import { pipeline } from 'stream/promises';
-import type { BucketItem, BucketItemStat, ClientOptions } from 'minio';
+import type { BucketItem, BucketItemStat, Client } from 'minio';
 import type { Readable } from 'stream';
 import type * as tls from 'tls';
 import type * as t from './types';
@@ -40,69 +40,6 @@ app.use(httpMetricsMiddleware);
 
 const bucketName = process.env.MINIO_BUCKET ?? 'test-bucket';
 
-type IamProviderModule = { IamAwsProvider?: new (opts: object) => unknown; default?: new (opts: object) => unknown };
-
-async function createMinioClient(): Promise<Client> {
-  const irsaExplicit = process.env.MINIO_USE_IRSA?.toLowerCase() === 'true';
-  const irsaEnvVars = Boolean(process.env.AWS_WEB_IDENTITY_TOKEN_FILE) && Boolean(process.env.AWS_ROLE_ARN);
-  const useIrsa = irsaExplicit || irsaEnvVars;
-
-  const baseConfig: ClientOptions = {
-    endPoint: process.env.MINIO_ENDPOINT ?? 'localhost',
-    port: process.env.MINIO_NO_PORT?.toLowerCase() === 'true' ? undefined : parseInt(process.env.MINIO_PORT ?? '9000'),
-    useSSL: process.env.MINIO_USE_SSL?.toLowerCase() === 'true',
-    region: process.env.MINIO_REGION ?? process.env.AWS_REGION ?? 'us-east-1',
-  };
-
-  if (useIrsa) {
-    logger.info('Using IRSA (IamAwsProvider) for S3 authentication', {
-      tokenFile: process.env.AWS_WEB_IDENTITY_TOKEN_FILE,
-      roleArn: process.env.AWS_ROLE_ARN,
-      region: baseConfig.region,
-    });
-
-    /** IamAwsProvider exists in minio 8.0.6+ but isn't exported from main module
-     * Try multiple import paths for compatibility with different runtimes (bun, ts-node, node)
-     */
-    let IamAwsProviderClass: new (opts: object) => unknown;
-    try {
-      const mod = await import('minio/dist/main/IamAwsProvider.js') as IamProviderModule;
-      IamAwsProviderClass = (mod.IamAwsProvider ?? mod.default)!;
-    } catch (primaryError) {
-      try {
-        // Fallback for bun: resolve path using require if available (CJS context)
-        let resolvePath = 'node_modules/minio/';
-        try {
-          // eslint-disable-next-line @typescript-eslint/no-require-imports
-          resolvePath = require.resolve('minio').replace(/dist\/.*$/, '');
-        } catch {
-          // require.resolve not available (ESM context), use default path
-        }
-        const mod = await import(`${resolvePath}dist/main/IamAwsProvider.js`) as IamProviderModule;
-        IamAwsProviderClass = (mod.IamAwsProvider ?? mod.default)!;
-      } catch (fallbackError) {
-        logger.error('Failed to load IamAwsProvider', { primaryError, fallbackError });
-        throw new Error('Could not load IamAwsProvider for IRSA authentication. Ensure minio >= 8.0.6 is installed.');
-      }
-    }
-
-    const credentialsProvider = new IamAwsProviderClass({});
-
-    return new Client({
-      ...baseConfig,
-      credentialsProvider: credentialsProvider as ClientOptions['credentialsProvider'],
-    });
-  }
-
-  logger.info('Using explicit credentials for MinIO/S3 authentication');
-  return new Client({
-    ...baseConfig,
-    accessKey: process.env.MINIO_ACCESS_KEY ?? '',
-    secretKey: process.env.MINIO_SECRET_KEY ?? '',
-    sessionToken: process.env.MINIO_SESSION_TOKEN,
-  });
-}
-
 let minioClient: Client;
 let storageInitialized = false;
 

service/src/minio-client.test.ts

@@ -0,0 +1,101 @@
+import { expect, test } from 'bun:test';
+import { createHash } from 'node:crypto';
+import { Readable } from 'node:stream';
+import { createMinioClient } from './minio-client';
+
+const MiB = 1024 * 1024;
+
+// Exercise the real SDK against a local S3 HTTP fixture. No storage account,
+// Redis, or file-server listener is needed to test the production client.
+test.each([1024, 8 * MiB, 20 * MiB + 17])(
+  'unknown-length upload of %i bytes uses bounded parts without losing bytes',
+  async size => {
+    const parts = new Map<number, Buffer>();
+    const lengths: number[] = [];
+    const uploaded: { body?: Buffer; contentType?: string | null; originalFilename?: string | null } = {};
+    const xml = (body: string) => new Response(body, {
+      headers: { 'Content-Type': 'application/xml' },
+    });
+    const server = Bun.serve({
+      hostname: '127.0.0.1',
+      port: 0,
+      async fetch(req) {
+        const url = new URL(req.url);
+        if (req.method === 'GET' && url.searchParams.has('uploads')) {
+          return xml('<ListMultipartUploadsResult><IsTruncated>false</IsTruncated></ListMultipartUploadsResult>');
+        }
+        if (req.method === 'POST' && url.searchParams.has('uploads')) {
+          uploaded.contentType = req.headers.get('content-type');
+          uploaded.originalFilename = req.headers.get('x-amz-meta-original-filename');
+          return xml('<InitiateMultipartUploadResult><UploadId>test-upload</UploadId></InitiateMultipartUploadResult>');
+        }
+        if (req.method === 'PUT' && url.searchParams.has('partNumber')) {
+          const body = Buffer.from(await req.arrayBuffer());
+          lengths.push(body.length);
+          if (Number(req.headers.get('content-length')) !== body.length ||
+              req.headers.get('content-md5') !== createHash('md5').update(body).digest('base64')) {
+            return new Response('Invalid part length or checksum', { status: 400 });
+          }
+          parts.set(Number(url.searchParams.get('partNumber')), body);
+          return new Response(null, {
+            headers: { ETag: `"${createHash('md5').update(body).digest('hex')}"` },
+          });
+        }
+        if (req.method === 'POST' && url.searchParams.has('uploadId')) {
+          const manifest = await req.text();
+          const ordered = [...manifest.matchAll(/<PartNumber>(\d+)<\/PartNumber>/g)]
+            .map(match => parts.get(Number(match[1])));
+          if (ordered.length !== parts.size || ordered.some(part => !part)) {
+            return new Response('Invalid multipart completion', { status: 400 });
+          }
+          uploaded.body = Buffer.concat(ordered as Buffer[]);
+          return xml('<CompleteMultipartUploadResult><Location>http://localhost/test-bucket/input.bin</Location><Bucket>test-bucket</Bucket><Key>input.bin</Key><ETag>"complete"</ETag></CompleteMultipartUploadResult>');
+        }
+        return new Response('Unexpected S3 request', { status: 400 });
+      },
+    });
+    const settings: Record<string, string | undefined> = {
+      MINIO_ENDPOINT: '127.0.0.1',
+      MINIO_PORT: String(server.port),
+      MINIO_NO_PORT: 'false',
+      MINIO_USE_SSL: 'false',
+      MINIO_REGION: 'us-east-1',
+      MINIO_USE_IRSA: 'false',
+      AWS_WEB_IDENTITY_TOKEN_FILE: undefined,
+      AWS_ROLE_ARN: undefined,
+      MINIO_ACCESS_KEY: 'test-access',
+      MINIO_SECRET_KEY: 'test-secret',
+      MINIO_SESSION_TOKEN: undefined,
+    };
+    const saved = Object.fromEntries(Object.keys(settings).map(key => [key, process.env[key]]));
+    try {
+      for (const [key, value] of Object.entries(settings)) {
+        if (value === undefined) delete process.env[key];
+        else process.env[key] = value;
+      }
+      const client = await createMinioClient();
+      const expected = Buffer.alloc(size);
+      for (let i = 0; i < expected.length; i++) expected[i] = i % 251;
+      function* chunks() {
+        for (let offset = 0; offset < size; offset += 64 * 1024) {
+          yield expected.subarray(offset, Math.min(size, offset + 64 * 1024));
+        }
+      }
+      await client.putObject('test-bucket', 'input.bin', Readable.from(chunks()), undefined, {
+        'Content-Type': 'application/octet-stream',
+        'X-Amz-Meta-Original-Filename': 'input.bin',
+      });
+      expect(lengths.length).toBe(Math.ceil(size / (8 * MiB)));
+      expect(lengths.every(length => length <= 8 * MiB)).toBe(true);
+      expect(uploaded.body?.equals(expected)).toBe(true);
+      expect(uploaded.contentType).toBe('application/octet-stream');
+      expect(uploaded.originalFilename).toBe('input.bin');
+    } finally {
+      for (const [key, value] of Object.entries(saved)) {
+        if (value === undefined) delete process.env[key];
+        else process.env[key] = value;
+      }
+      await server.stop(true);
+    }
+  },
+);

service/src/minio-client.ts

@@ -0,0 +1,69 @@
+import { Client, type ClientOptions } from 'minio';
+import logger from './fileServerLogger';
+
+type IamProviderModule = { IamAwsProvider?: new (opts: object) => unknown; default?: new (opts: object) => unknown };
+
+export async function createMinioClient(): Promise<Client> {
+  const irsaExplicit = process.env.MINIO_USE_IRSA?.toLowerCase() === 'true';
+  const irsaEnvVars = Boolean(process.env.AWS_WEB_IDENTITY_TOKEN_FILE) && Boolean(process.env.AWS_ROLE_ARN);
+  const useIrsa = irsaExplicit || irsaEnvVars;
+
+  const baseConfig: ClientOptions = {
+    // Unknown-length streams otherwise grow SDK parts to 528 MiB (the 5 TiB
+    // object limit / 10,000 parts). Bound each multipart buffer instead.
+    partSize: 8 * 1024 * 1024,
+    endPoint: process.env.MINIO_ENDPOINT ?? 'localhost',
+    port: process.env.MINIO_NO_PORT?.toLowerCase() === 'true' ? undefined : parseInt(process.env.MINIO_PORT ?? '9000'),
+    useSSL: process.env.MINIO_USE_SSL?.toLowerCase() === 'true',
+    region: process.env.MINIO_REGION ?? process.env.AWS_REGION ?? 'us-east-1',
+  };
+
+  if (useIrsa) {
+    logger.info('Using IRSA (IamAwsProvider) for S3 authentication', {
+      tokenFile: process.env.AWS_WEB_IDENTITY_TOKEN_FILE,
+      roleArn: process.env.AWS_ROLE_ARN,
+      region: baseConfig.region,
+    });
+
+    /** IamAwsProvider exists in minio 8.0.6+ but isn't exported from main module
+     * Try multiple import paths for compatibility with different runtimes (bun, ts-node, node)
+     */
+    let IamAwsProviderClass: new (opts: object) => unknown;
+    try {
+      const mod = await import('minio/dist/main/IamAwsProvider.js') as IamProviderModule;
+      IamAwsProviderClass = (mod.IamAwsProvider ?? mod.default)!;
+    } catch (primaryError) {
+      try {
+        // Fallback for bun: resolve path using require if available (CJS context)
+        let resolvePath = 'node_modules/minio/';
+        try {
+          // eslint-disable-next-line @typescript-eslint/no-require-imports
+          resolvePath = require.resolve('minio').replace(/dist\/.*$/, '');
+        } catch {
+          // require.resolve not available (ESM context), use default path
+        }
+        const mod = await import(`${resolvePath}dist/main/IamAwsProvider.js`) as IamProviderModule;
+        IamAwsProviderClass = (mod.IamAwsProvider ?? mod.default)!;
+      } catch (fallbackError) {
+        logger.error('Failed to load IamAwsProvider', { primaryError, fallbackError });
+        throw new Error('Could not load IamAwsProvider for IRSA authentication. Ensure minio >= 8.0.6 is installed.');
+      }
+    }
+
+    const credentialsProvider = new IamAwsProviderClass({});
+
+    return new Client({
+      ...baseConfig,
+      credentialsProvider: credentialsProvider as ClientOptions['credentialsProvider'],
+    });
+  }
+
+  logger.info('Using explicit credentials for MinIO/S3 authentication');
+  return new Client({
+    ...baseConfig,
+    accessKey: process.env.MINIO_ACCESS_KEY ?? '',
+    secretKey: process.env.MINIO_SECRET_KEY ?? '',
+    sessionToken: process.env.MINIO_SESSION_TOKEN,
+  });
+}
+