fix(admin): bound request log export memory

apps/web/src/app/admin/api/api-request-log/download/route.test.ts

@@ -0,0 +1,92 @@
+import { NextRequest } from 'next/server';
+import { randomBytes } from 'node:crypto';
+import { eq } from 'drizzle-orm';
+import { strFromU8, unzipSync } from 'fflate';
+import { api_request_log } from '@kilocode/db/schema';
+import { db } from '@/lib/drizzle';
+import { getUserFromAuth } from '@/lib/user/server';
+import { defineTestUser } from '@/tests/helpers/user.helper';
+import { GET } from './route';
+
+jest.mock('next/server', () => {
+  const actual = jest.requireActual('next/server');
+  return { ...actual, connection: jest.fn() };
+});
+
+jest.mock('@/lib/user/server', () => ({
+  getUserFromAuth: jest.fn(),
+}));
+
+const mockedGetUserFromAuth = jest.mocked(getUserFromAuth);
+const TEST_USER_ID = 'api-request-log-download-test-user';
+const TEST_MODEL = 'poolside/laguna-s-2.1:free';
+const BATCH_SIZE = 25;
+
+function createRequest() {
+  const params = new URLSearchParams({
+    userId: TEST_USER_ID,
+    startDate: '2026-08-01',
+    endDate: '2026-08-01',
+    model: TEST_MODEL,
+  });
+  return new NextRequest(`http://localhost:3000/admin/api/api-request-log/download?${params}`);
+}
+
+function readEntry(entries: Record<string, Uint8Array>, suffix: string): string {
+  const name = Object.keys(entries).find(entryName => entryName.endsWith(suffix));
+  if (!name) throw new Error(`Missing archive entry ending in ${suffix}`);
+  return strFromU8(entries[name]);
+}
+
+describe('GET /admin/api/api-request-log/download', () => {
+  beforeEach(() => {
+    mockedGetUserFromAuth.mockResolvedValue({
+      user: defineTestUser({ is_admin: true }),
+      authFailedResponse: null,
+    });
+  });
+
+  afterEach(async () => {
+    await db.delete(api_request_log).where(eq(api_request_log.kilo_user_id, TEST_USER_ID));
+  });
+
+  it('streams a complete ZIP across backpressured DB batches', async () => {
+    // The first batch must exceed both the Node and web stream queues. This
+    // keeps page two blocked until the test starts consuming the response.
+    const payload = randomBytes(128 * 1024).toString('base64');
+    const rows = await db
+      .insert(api_request_log)
+      .values(
+        Array.from({ length: BATCH_SIZE + 1 }, (_, index) => ({
+          created_at: '2026-08-01T12:00:00.000Z',
+          kilo_user_id: TEST_USER_ID,
+          provider: 'test-provider',
+          model: TEST_MODEL,
+          request: { index },
+          response: JSON.stringify({ output: index, payload }),
+        }))
+      )
+      .returning({ id: api_request_log.id });
+
+    const response = await GET(createRequest());
+
+    expect(response.status).toBe(200);
+    expect(response.headers.get('Content-Type')).toBe('application/zip');
+    expect(response.headers.get('Content-Disposition')).toBe(
+      'attachment; filename="api-request-log_api-request-log-download-test-user_2026-08-01_2026-08-01_poolside-laguna-s-2.1-free.zip"'
+    );
+
+    const bytes = new Uint8Array(await response.arrayBuffer());
+    expect(Array.from(bytes.subarray(0, 4))).toEqual([0x50, 0x4b, 0x03, 0x04]);
+
+    const entries = unzipSync(bytes);
+    expect(Object.keys(entries)).toHaveLength((BATCH_SIZE + 1) * 2);
+    expect(readEntry(entries, `_${rows[0].id}_request.json`)).toBe(
+      JSON.stringify({ index: 0 }, null, 2)
+    );
+    expect(JSON.parse(readEntry(entries, `_${rows[BATCH_SIZE].id}_response.json`))).toEqual({
+      output: BATCH_SIZE,
+      payload,
+    });
+  });
+});

apps/web/src/app/admin/api/api-request-log/download/route.ts

@@ -12,7 +12,7 @@ import { Readable } from 'node:stream';
 // extract it ("Error 79 - Inappropriate file type or format").
 export const maxDuration = 800;
 
-const BATCH_SIZE = 100;
+const BATCH_SIZE = 25;
 
 function formatTimestamp(isoString: string): string {
   return isoString.replaceAll(':', '-').replaceAll(' ', '_');
@@ -128,6 +128,45 @@ export async function GET(request: NextRequest) {
   }
 
   const archive = archiver('zip', { zlib: { level: 6 } });
+  let totalAppendedEntries = 0;
+  let totalProcessedEntries = 0;
+
+  archive.on('entry', () => {
+    totalProcessedEntries += 1;
+  });
+
+  const waitForEntries = (target: number) => {
+    if (totalProcessedEntries >= target) return Promise.resolve();
+    if (archive.destroyed) {
+      return Promise.reject(new Error('Archive closed before all entries were processed'));
+    }
+
+    return new Promise<void>((resolve, reject) => {
+      const cleanup = () => {
+        archive.off('entry', onEntry);
+        archive.off('error', onError);
+        archive.off('close', onClose);
+      };
+      const onEntry = () => {
+        if (totalProcessedEntries >= target) {
+          cleanup();
+          resolve();
+        }
+      };
+      const onError = (error: Error) => {
+        cleanup();
+        reject(error);
+      };
+      const onClose = () => {
+        cleanup();
+        reject(new Error('Archive closed before all entries were processed'));
+      };
+
+      archive.on('entry', onEntry);
+      archive.once('error', onError);
+      archive.once('close', onClose);
+    });
+  };
 
   // Fetch and archive rows in batches using cursor-based pagination to
   // avoid loading the entire result set into memory at once.
@@ -150,36 +189,32 @@ export async function GET(request: NextRequest) {
         const requestExt = isJson(row.request) ? 'json' : 'txt';
         const requestContent = tryFormatJson(row.request);
         if (requestContent) {
+          totalAppendedEntries += 1;
           archive.append(requestContent, { name: `${ts}_${id}_request.${requestExt}` });
         }
 
         const responseExt = isJson(row.response) ? 'json' : 'txt';
         const responseContent = tryFormatJson(row.response);
         if (responseContent) {
+          totalAppendedEntries += 1;
           archive.append(responseContent, { name: `${ts}_${id}_response.${responseExt}` });
         }
 
         if (row.error !== null && row.error !== undefined) {
           const errorContent = tryFormatJson(row.error);
           if (errorContent) {
+            totalAppendedEntries += 1;
             archive.append(errorContent, { name: `${ts}_${id}_error.json` });
           }
         }
       }
 
       cursor = rows[rows.length - 1].id;
 
-      // Yield between batches while the archive's readable buffer is above
-      // its high-water mark, so we don't buffer unbounded data in memory.
-      // Polling via setImmediate rather than waiting on a single 'drain'
-      // event: archiver's internal queue pauses once it's out of entries, so
-      // after we stop appending its writable side may never go back above
-      // HWM and 'drain' would never fire again - listening for it would
-      // deadlock the stream.
-      const hwm = archive.readableHighWaterMark ?? 16 * 1024;
-      while (archive.readableLength > hwm) {
-        await new Promise<void>(resolve => setImmediate(resolve));
-      }
+      // Archiver maintains its own input queue, which is not reflected by the
+      // readable stream's high-water mark. Wait until this batch is emitted so
+      // large exports remain bounded even when compression is slower than DB reads.
+      await waitForEntries(totalAppendedEntries);
     }
 
     await archive.finalize();