From 16f2274c8971c7f34aee10f6c2e905149bcfb430 Mon Sep 17 00:00:00 2001 From: erichanwang Date: Tue, 28 Jul 2026 16:16:10 -0500 Subject: [PATCH] fix: honor sendFiles maxConcurrent --- .../client-typescript/src/cli/rocketride.ts | 36 ++++++++++-------- .../client-typescript/src/client/client.ts | 29 +++++++++----- .../tests/RocketRideClient.test.ts | 38 +++++++++++++++++++ packages/shell-api/versions/v0.d.ts | 2 +- 4 files changed, 78 insertions(+), 27 deletions(-) diff --git a/packages/client-typescript/src/cli/rocketride.ts b/packages/client-typescript/src/cli/rocketride.ts index a11befbcf..cee4916c4 100644 --- a/packages/client-typescript/src/cli/rocketride.ts +++ b/packages/client-typescript/src/cli/rocketride.ts @@ -922,21 +922,26 @@ export class RocketRideCLI { .option('--threads ', 'Number of threads to use for pipeline execution', '4') .option('--max-concurrent ', 'Maximum number of concurrent file uploads', '5') .option('--args ', 'Additional arguments to pass to pipeline execution') - .action(async (files, options) => { - // Validate required arguments - validation will happen in createAndConnectClient - if (!options.pipeline && !options.token) { - console.error('Error: Either --pipeline or --token must be specified for upload command. Use --pipeline/--token or set ROCKETRIDE_PIPELINE/ROCKETRIDE_TOKEN in .env file'); - process.exit(1); - } + .action(async (files, options) => { + // Validate required arguments - validation will happen in createAndConnectClient + if (!options.pipeline && !options.token) { + console.error('Error: Either --pipeline or --token must be specified for upload command. Use --pipeline/--token or set ROCKETRIDE_PIPELINE/ROCKETRIDE_TOKEN in .env file'); + process.exit(1); + } + const maxConcurrent = Number(options.maxConcurrent || '5'); + if (!Number.isFinite(maxConcurrent)) { + console.error('Error: --max-concurrent must be a finite number'); + process.exit(1); + } - this.args = { - command: 'upload', - ...options, - files, - threads: parseInt(options.threads), - max_concurrent: parseInt(options.maxConcurrent || '5'), - pipeline_args: options.args, - }; + this.args = { + command: 'upload', + ...options, + files, + threads: parseInt(options.threads), + max_concurrent: maxConcurrent, + pipeline_args: options.args, + }; this.uri = options.uri; try { @@ -1452,8 +1457,7 @@ export class RocketRideCLI { }); // Upload files - progress events come through event subscription - // Server handles concurrency automatically - const results = await this.client!.sendFiles(fileObjects, taskToken!); + const results = await this.client!.sendFiles(fileObjects, taskToken!, this.args.max_concurrent); const endTime = Date.now(); diff --git a/packages/client-typescript/src/client/client.ts b/packages/client-typescript/src/client/client.ts index fc1fafd09..0b88ac4db 100644 --- a/packages/client-typescript/src/client/client.ts +++ b/packages/client-typescript/src/client/client.ts @@ -1371,9 +1371,14 @@ export class RocketRideClient extends DAPClient { objinfo?: Record; mimetype?: string; }>, - token: string + token: string, + maxConcurrent = 5 ): Promise { const results: UPLOAD_RESULT[] = new Array(files.length); + if (!Number.isFinite(maxConcurrent)) { + throw new RangeError('maxConcurrent must be a finite number'); + } + const concurrency = Math.max(1, Math.floor(maxConcurrent)); /** * Helper function to send upload events through the event system. @@ -1483,16 +1488,20 @@ export class RocketRideClient extends DAPClient { results[index] = finalResult; }; - // Create a promise for every file - let server handle queuing - const uploadPromises = files.map((fileData, index) => - uploadFile(fileData, index).catch((err) => { - // Ensure errors don't kill the whole batch - console.error(`Upload failed for ${fileData.file.name}:`, err); - }) - ); + let nextIndex = 0; + const workers = Array.from({ length: Math.min(concurrency, files.length) }, async () => { + while (nextIndex < files.length) { + const index = nextIndex; + nextIndex += 1; + const fileData = files[index]!; + await uploadFile(fileData, index).catch((err) => { + // Ensure errors don't kill the whole batch + console.error(`Upload failed for ${fileData.file.name}:`, err); + }); + } + }); - // Wait for all uploads to complete - await Promise.all(uploadPromises); + await Promise.all(workers); return results; } diff --git a/packages/client-typescript/tests/RocketRideClient.test.ts b/packages/client-typescript/tests/RocketRideClient.test.ts index c613111b7..f4d03ea7f 100644 --- a/packages/client-typescript/tests/RocketRideClient.test.ts +++ b/packages/client-typescript/tests/RocketRideClient.test.ts @@ -2213,6 +2213,44 @@ describe('RocketRideClient URI normalization', () => { }); }); +describe('RocketRideClient sendFiles concurrency', () => { + it('honors maxConcurrent', async () => { + const client = new RocketRideClient({ auth: 'test-key', uri: 'http://localhost:5565' }); + let active = 0; + let maxActive = 0; + + (client as any).pipe = jest.fn(async () => { + active += 1; + maxActive = Math.max(maxActive, active); + return { + open: jest.fn(async () => undefined), + write: jest.fn(async () => undefined), + close: jest.fn(async () => { + await new Promise((resolve) => setTimeout(resolve, 10)); + active -= 1; + return undefined; + }), + }; + }); + + const files = Array.from({ length: 5 }, (_, index) => ({ + file: new File([`file-${index}`], `file-${index}.txt`, { type: 'text/plain' }), + })); + + await client.sendFiles(files, 'task-token', 2); + + expect((client as any).pipe).toHaveBeenCalledTimes(5); + expect(maxActive).toBeLessThanOrEqual(2); + }); + + it('rejects non-finite maxConcurrent', async () => { + const client = new RocketRideClient({ auth: 'test-key', uri: 'http://localhost:5565' }); + const files = [{ file: new File(['file'], 'file.txt', { type: 'text/plain' }) }]; + + await expect(client.sendFiles(files, 'task-token', Number.NaN)).rejects.toThrow(RangeError); + }); +}); + export async function isServerAvailable(): Promise { try { const client = new RocketRideClient({ diff --git a/packages/shell-api/versions/v0.d.ts b/packages/shell-api/versions/v0.d.ts index 6ceb1d746..298bf9ca0 100644 --- a/packages/shell-api/versions/v0.d.ts +++ b/packages/shell-api/versions/v0.d.ts @@ -2862,7 +2862,7 @@ export declare class RocketRideClient extends DAPClient { file: File; objinfo?: Record; mimetype?: string; - }>, token: string): Promise; + }>, token: string, maxConcurrent?: number): Promise; /** * Ask a question to RocketRide's AI and get an intelligent response. */