From 54460226dac7d23a63588b4d47bce9f06e9e3916 Mon Sep 17 00:00:00 2001 From: MilindChatrabhuji Date: Mon, 26 Oct 2026 04:30:47 +0530 Subject: [PATCH] fix(export): stream chunks incrementally for database dumps to prevent OOM (#59) - Replace monolithic string concatenation and Blob response with ReadableStream - Read table rows in chunked batches (CHUNK_SIZE = 1000) using LIMIT and OFFSET pagination - Correctly format SQL NULL, booleans, and escaped strings in INSERT statements - Expand test suite in src/export/dump.test.ts to verify chunked pagination, type serialization, streaming reader, and stream error handling --- src/export/dump.test.ts | 127 +++++++++++++++++++++++++++++++++++++++- src/export/dump.ts | 122 ++++++++++++++++++++++++++------------ 2 files changed, 210 insertions(+), 39 deletions(-) diff --git a/src/export/dump.test.ts b/src/export/dump.test.ts index ca65b43..19912b1 100644 --- a/src/export/dump.test.ts +++ b/src/export/dump.test.ts @@ -128,7 +128,7 @@ describe('Database Dump Module', () => { ) }) - it('should return a 500 response when an error occurs', async () => { + it('should return a 500 response when an error occurs before streaming starts', async () => { const consoleErrorMock = vi .spyOn(console, 'error') .mockImplementation(() => {}) @@ -142,4 +142,129 @@ describe('Database Dump Module', () => { const jsonResponse: { error: string } = await response.json() expect(jsonResponse.error).toBe('Failed to create database dump') }) + + it('should paginate large tables in chunked batches across multiple queries', async () => { + const batch1 = Array.from({ length: 1000 }, (_, i) => ({ + id: i + 1, + val: `item_${i + 1}`, + })) + const batch2 = Array.from({ length: 1000 }, (_, i) => ({ + id: i + 1001, + val: `item_${i + 1001}`, + })) + const batch3 = Array.from({ length: 250 }, (_, i) => ({ + id: i + 2001, + val: `item_${i + 2001}`, + })) + + vi.mocked(executeOperation) + .mockResolvedValueOnce([{ name: 'large_table' }]) + .mockResolvedValueOnce([ + { sql: 'CREATE TABLE large_table (id INTEGER, val TEXT);' }, + ]) + .mockResolvedValueOnce(batch1) + .mockResolvedValueOnce(batch2) + .mockResolvedValueOnce(batch3) + + const response = await dumpDatabaseRoute(mockDataSource, mockConfig) + expect(response).toBeInstanceOf(Response) + + const dumpText = await response.text() + expect(dumpText).toContain( + 'CREATE TABLE large_table (id INTEGER, val TEXT);' + ) + expect(dumpText).toContain( + "INSERT INTO large_table VALUES (1, 'item_1');" + ) + expect(dumpText).toContain( + "INSERT INTO large_table VALUES (1001, 'item_1001');" + ) + expect(dumpText).toContain( + "INSERT INTO large_table VALUES (2250, 'item_2250');" + ) + + // Verify SQL queries were called with pagination + const calls = vi.mocked(executeOperation).mock.calls + const querySqls = calls.map( + (c) => (c[0] as Array<{ sql: string }>)[0].sql + ) + expect(querySqls).toContain( + 'SELECT * FROM large_table LIMIT 1000 OFFSET 0;' + ) + expect(querySqls).toContain( + 'SELECT * FROM large_table LIMIT 1000 OFFSET 1000;' + ) + expect(querySqls).toContain( + 'SELECT * FROM large_table LIMIT 1000 OFFSET 2000;' + ) + }) + + it('should format NULL, boolean, and numeric values correctly in INSERT statements', async () => { + vi.mocked(executeOperation) + .mockResolvedValueOnce([{ name: 'types_table' }]) + .mockResolvedValueOnce([ + { + sql: 'CREATE TABLE types_table (id INT, null_col TEXT, is_active BOOLEAN, score REAL);', + }, + ]) + .mockResolvedValueOnce([ + { id: 1, null_col: null, is_active: true, score: 98.6 }, + { id: 2, null_col: undefined, is_active: false, score: 0 }, + ]) + + const response = await dumpDatabaseRoute(mockDataSource, mockConfig) + const dumpText = await response.text() + + expect(dumpText).toContain( + 'INSERT INTO types_table VALUES (1, NULL, 1, 98.6);' + ) + expect(dumpText).toContain( + 'INSERT INTO types_table VALUES (2, NULL, 0, 0);' + ) + }) + + it('should stream response as a ReadableStream without buffering entire output in memory', async () => { + vi.mocked(executeOperation) + .mockResolvedValueOnce([{ name: 'stream_test' }]) + .mockResolvedValueOnce([ + { sql: 'CREATE TABLE stream_test (id INT);' }, + ]) + .mockResolvedValueOnce([{ id: 1 }, { id: 2 }]) + + const response = await dumpDatabaseRoute(mockDataSource, mockConfig) + + expect(response.body).toBeInstanceOf(ReadableStream) + const reader = response.body!.getReader() + const decoder = new TextDecoder() + let chunksCount = 0 + let fullOutput = '' + + while (true) { + const { done, value } = await reader.read() + if (done) break + chunksCount++ + fullOutput += decoder.decode(value, { stream: true }) + } + + expect(chunksCount).toBeGreaterThan(1) + expect(fullOutput).toContain('SQLite format 3\0') + expect(fullOutput).toContain('CREATE TABLE stream_test (id INT);') + expect(fullOutput).toContain('INSERT INTO stream_test VALUES (1);') + expect(fullOutput).toContain('INSERT INTO stream_test VALUES (2);') + }) + + it('should emit an error on the stream when an error occurs during streaming', async () => { + vi.mocked(executeOperation) + .mockResolvedValueOnce([{ name: 'broken_table' }]) + .mockRejectedValueOnce(new Error('Stream failure during fetch')) + + const response = await dumpDatabaseRoute(mockDataSource, mockConfig) + + expect(response.status).toBe(200) + expect(response.body).toBeInstanceOf(ReadableStream) + + await expect(response.text()).rejects.toThrow( + 'Stream failure during fetch' + ) + }) }) diff --git a/src/export/dump.ts b/src/export/dump.ts index 91a2e89..6d9e43b 100644 --- a/src/export/dump.ts +++ b/src/export/dump.ts @@ -3,67 +3,113 @@ import { StarbaseDBConfiguration } from '../handler' import { DataSource } from '../types' import { createResponse } from '../utils' +const CHUNK_SIZE = 1000 + +function formatSqlValue(value: unknown): string { + if (value === null || value === undefined) { + return 'NULL' + } + if (typeof value === 'string') { + return `'${value.replace(/'/g, "''")}'` + } + if (typeof value === 'boolean') { + return value ? '1' : '0' + } + return String(value) +} + export async function dumpDatabaseRoute( dataSource: DataSource, config: StarbaseDBConfiguration ): Promise { try { - // Get all table names + // Get all table names up front const tablesResult = await executeOperation( [{ sql: "SELECT name FROM sqlite_master WHERE type='table';" }], dataSource, config ) - const tables = tablesResult.map((row: any) => row.name) - let dumpContent = 'SQLite format 3\0' // SQLite file header + const tables = (tablesResult as Array<{ name: string }>).map( + (row) => row.name + ) + + const encoder = new TextEncoder() + const stream = new ReadableStream({ + async start(controller) { + try { + // SQLite file header + controller.enqueue(encoder.encode('SQLite format 3\0')) + + // Iterate through all tables + for (const table of tables) { + // Get table schema + const schemaResult = await executeOperation( + [ + { + sql: `SELECT sql FROM sqlite_master WHERE type='table' AND name='${table}';`, + }, + ], + dataSource, + config + ) - // Iterate through all tables - for (const table of tables) { - // Get table schema - const schemaResult = await executeOperation( - [ - { - sql: `SELECT sql FROM sqlite_master WHERE type='table' AND name='${table}';`, - }, - ], - dataSource, - config - ) + if (schemaResult && schemaResult.length) { + const schema = schemaResult[0].sql + controller.enqueue( + encoder.encode( + `\n-- Table: ${table}\n${schema};\n\n` + ) + ) + } - if (schemaResult.length) { - const schema = schemaResult[0].sql - dumpContent += `\n-- Table: ${table}\n${schema};\n\n` - } + // Stream table data in paginated batches + let offset = 0 + while (true) { + const dataResult = await executeOperation( + [ + { + sql: `SELECT * FROM ${table} LIMIT ${CHUNK_SIZE} OFFSET ${offset};`, + }, + ], + dataSource, + config + ) - // Get table data - const dataResult = await executeOperation( - [{ sql: `SELECT * FROM ${table};` }], - dataSource, - config - ) + if (!dataResult || dataResult.length === 0) { + break + } - for (const row of dataResult) { - const values = Object.values(row).map((value) => - typeof value === 'string' - ? `'${value.replace(/'/g, "''")}'` - : value - ) - dumpContent += `INSERT INTO ${table} VALUES (${values.join(', ')});\n` - } + let chunkSql = '' + for (const row of dataResult) { + const values = + Object.values(row).map(formatSqlValue) + chunkSql += `INSERT INTO ${table} VALUES (${values.join(', ')});\n` + } + controller.enqueue(encoder.encode(chunkSql)) - dumpContent += '\n' - } + offset += dataResult.length + if (dataResult.length < CHUNK_SIZE) { + break + } + } - // Create a Blob from the dump content - const blob = new Blob([dumpContent], { type: 'application/x-sqlite3' }) + controller.enqueue(encoder.encode('\n')) + } + + controller.close() + } catch (error) { + controller.error(error) + } + }, + }) const headers = new Headers({ 'Content-Type': 'application/x-sqlite3', 'Content-Disposition': 'attachment; filename="database_dump.sql"', }) - return new Response(blob, { headers }) + return new Response(stream, { headers }) } catch (error: any) { console.error('Database Dump Error:', error) return createResponse(undefined, 'Failed to create database dump', 500)