[PB-5727]: upload concurrently with worker pool to improve speed

src/apps/backups/BackupService.test.ts

@@ -1,10 +1,8 @@
-import { vi, Mock } from 'vitest';
+import { Mock } from 'vitest';
 import { mockDeep } from 'vitest-mock-extended';
 import { BackupService } from './BackupService';
 import LocalTreeBuilder from '../../context/local/localTree/application/LocalTreeBuilder';
 import { RemoteTreeBuilder } from '../../context/virtual-drive/remoteTree/application/RemoteTreeBuilder';
-import { FileBatchUploader } from '../../context/local/localFile/application/upload/FileBatchUploader';
-import { FileBatchUpdater } from '../../context/local/localFile/application/update/FileBatchUpdater';
 import { SimpleFolderCreator } from '../../context/virtual-drive/folders/application/create/SimpleFolderCreator';
 import { BackupInfo } from './BackupInfo';
 import { DriveDesktopError } from '../../context/shared/domain/errors/DriveDesktopError';
@@ -15,66 +13,59 @@
 import { UsageModule } from '../../backend/features/usage/usage.module';
 import { FolderMother } from '../../context/virtual-drive/folders/domain/__test-helpers__/FolderMother';
 import { BackupProgressTracker } from '../../backend/features/backup/backup-progress-tracker';
+import * as executeAsyncQueueModule from '../../backend/common/async-queue/execute-async-queue';
+import * as addFileToTrashModule from '../../infra/drive-server/services/files/services/add-file-to-trash';
+import { partialSpyOn } from '../../../tests/vitest/utils.helper';
 
-// Mock the UsageModule
-vi.mock('../../backend/features/usage/usage.module', () => ({
-  UsageModule: {
-    validateSpace: vi.fn(),
-  },
-}));
-
-// Mock the Environment module
-vi.mock('@internxt/inxt-js', () => ({
-  Environment: {
-    get: vi.fn(),
-  },
-}));
+vi.mock(import('../../backend/features/usage/usage.module'));
+vi.mock(import('@internxt/inxt-js'));
 
 describe('BackupService', () => {
+  const executeAsyncQueueMock = partialSpyOn(executeAsyncQueueModule, 'executeAsyncQueue');
+  const addFileToTrashMock = partialSpyOn(addFileToTrashModule, 'addFileToTrash');
+
   let backupService: BackupService;
   let localTreeBuilder: LocalTreeBuilder;
   let remoteTreeBuilder: RemoteTreeBuilder;
-  let fileBatchUploader: FileBatchUploader;
-  let fileBatchUpdater: FileBatchUpdater;
   let simpleFolderCreator: SimpleFolderCreator;
   let mockValidateSpace: Mock;
   let abortController: AbortController;
   let tracker: BackupProgressTracker;
 
+  const info: BackupInfo = {
+    pathname: '/path/to/backup',
+    folderId: 123,
+    folderUuid: 'uuid',
+    tmpPath: '/tmp/path',
+    backupsBucket: 'backups-bucket',
+    name: 'backup-name',
+  };
+
   beforeEach(() => {
     localTreeBuilder = mockDeep<LocalTreeBuilder>();
     remoteTreeBuilder = mockDeep<RemoteTreeBuilder>();
-    fileBatchUploader = mockDeep<FileBatchUploader>();
-    fileBatchUpdater = mockDeep<FileBatchUpdater>();
     simpleFolderCreator = mockDeep<SimpleFolderCreator>();
     tracker = mockDeep<BackupProgressTracker>();
 
-    mockValidateSpace = UsageModule.validateSpace as Mock;
+    mockValidateSpace = vi.mocked(UsageModule.validateSpace);
     abortController = new AbortController();
 
-    // Setup default mock implementations
     vi.mocked(simpleFolderCreator.run).mockResolvedValue(FolderMother.any());
+    executeAsyncQueueMock.mockResolvedValue({ data: undefined });
+    addFileToTrashMock.mockResolvedValue({ data: true });
 
     backupService = new BackupService(
       localTreeBuilder,
       remoteTreeBuilder,
-      fileBatchUploader,
-      fileBatchUpdater,
       simpleFolderCreator,
+      {} as any,
+      'backups-bucket',
     );
 
     mockValidateSpace.mockClear();
   });
 
   it('should successfully run the backup process', async () => {
-    const info: BackupInfo = {
-      pathname: '/path/to/backup',
-      folderId: 123,
-      folderUuid: 'uuid',
-      tmpPath: '/tmp/path',
-      backupsBucket: 'backups-bucket',
-      name: 'backup-name',
-    };
     const localTree = LocalTreeMother.oneLevel(10);
     const remoteTree = RemoteTreeMother.oneLevel(10);
 
@@ -92,14 +83,6 @@
   });
 
   it('should return an error if local tree generation fails', async () => {
-    const info: BackupInfo = {
-      pathname: '/path/to/backup',
-      folderId: 123,
-      folderUuid: 'uuid',
-      tmpPath: '/tmp/path',
-      backupsBucket: 'backups-bucket',
-      name: 'backup-name',
-    };
     const error = new DriveDesktopError('NOT_EXISTS', 'Failed to generate local tree');
 
     vi.mocked(localTreeBuilder.run).mockResolvedValueOnce(left(error));
@@ -111,17 +94,8 @@
   });
 
   it('should return an error if remote tree generation fails', async () => {
-    const info: BackupInfo = {
-      pathname: '/path/to/backup',
-      folderId: 123,
-      folderUuid: 'uuid',
-      tmpPath: '/tmp/path',
-      backupsBucket: 'backups-bucket',
-      name: 'backup-name',
-    };
     const error = new DriveDesktopError('NOT_EXISTS', 'Failed to generate remote tree');
 
-    // Mock the behavior of dependencies
     vi.mocked(localTreeBuilder.run).mockResolvedValueOnce(right(LocalTreeMother.oneLevel(10)));
     vi.mocked(remoteTreeBuilder.run).mockResolvedValueOnce(left(error) as unknown as RemoteTree);
 
@@ -132,17 +106,8 @@
   });
 
   it('should return an error if there is not enough space', async () => {
-    const info: BackupInfo = {
-      pathname: '/path/to/backup',
-      folderId: 123,
-      folderUuid: 'uuid',
-      tmpPath: '/tmp/path',
-      backupsBucket: 'backups-bucket',
-      name: 'backup-name',
-    };
-
-    (localTreeBuilder.run as Mock).mockResolvedValueOnce(right(LocalTreeMother.oneLevel(10)));
-    (remoteTreeBuilder.run as Mock).mockResolvedValueOnce(RemoteTreeMother.oneLevel(10));
+    vi.mocked(localTreeBuilder.run).mockResolvedValueOnce(right(LocalTreeMother.oneLevel(10)));
+    vi.mocked(remoteTreeBuilder.run).mockResolvedValueOnce(RemoteTreeMother.oneLevel(10));
     mockValidateSpace.mockResolvedValueOnce({ data: { hasSpace: false } });
 
     const result = await backupService.run(info, abortController.signal, tracker);
@@ -151,15 +116,6 @@
   });
 
   it('should return an unknown error for unexpected issues', async () => {
-    const info: BackupInfo = {
-      pathname: '/path/to/backup',
-      folderId: 123,
-      folderUuid: 'uuid',
-      tmpPath: '/tmp/path',
-      backupsBucket: 'backups-bucket',
-      name: 'backup-name',
-    };
-
     vi.mocked(localTreeBuilder.run).mockImplementationOnce(() => {
       throw new Error('Unexpected error');
     });
@@ -169,4 +125,30 @@
     expect(result).toBeInstanceOf(DriveDesktopError);
     expect(result?.message).toBe('An unknown error occurred');
   });
+
+  it('should propagate fatal error from uploadAndCreate', async () => {
+    const fatalError = new DriveDesktopError('NOT_ENOUGH_SPACE', 'No space left');
+
+    vi.mocked(localTreeBuilder.run).mockResolvedValueOnce(right(LocalTreeMother.oneLevel(10)));
+    vi.mocked(remoteTreeBuilder.run).mockResolvedValueOnce(RemoteTreeMother.oneLevel(10));
+    mockValidateSpace.mockResolvedValueOnce({ data: { hasSpace: true } });
+    executeAsyncQueueMock.mockResolvedValueOnce({ error: fatalError });
+
+    const result = await backupService.run(info, abortController.signal, tracker);
+
+    expect(result).toBe(fatalError);
+  });
+
+  it('should propagate fatal error from uploadAndUpdate', async () => {
+    const fatalError = new DriveDesktopError('NOT_ENOUGH_SPACE', 'No space left');
+
+    vi.mocked(localTreeBuilder.run).mockResolvedValueOnce(right(LocalTreeMother.oneLevel(10)));
+    vi.mocked(remoteTreeBuilder.run).mockResolvedValueOnce(RemoteTreeMother.oneLevel(10));
+    mockValidateSpace.mockResolvedValueOnce({ data: { hasSpace: true } });
+    executeAsyncQueueMock.mockResolvedValueOnce({ data: undefined }).mockResolvedValueOnce({ error: fatalError });
+
+    const result = await backupService.run(info, abortController.signal, tracker);
+
+    expect(result).toBe(fatalError);
+  });
 });

src/apps/backups/BackupService.ts

@@ -1,6 +1,11 @@
+import { Environment } from '@internxt/inxt-js';
 import { Service } from 'diod';
-import { FileBatchUpdater } from '../../context/local/localFile/application/update/FileBatchUpdater';
-import { FileBatchUploader } from '../../context/local/localFile/application/upload/FileBatchUploader';
+import { executeAsyncQueue } from '../../backend/common/async-queue/execute-async-queue';
+import {
+  createBackupUpdateExecutor,
+  ModifiedFilePair,
+} from '../../backend/features/backup/upload/create-backup-update-executor';
+import { DEFAULT_CONCURRENCY } from '../../backend/features/backup/upload/constants';
 import { LocalFile } from '../../context/local/localFile/domain/LocalFile';
 import { AbsolutePath } from '../../context/local/localFile/infrastructure/AbsolutePath';
 import LocalTreeBuilder from '../../context/local/localTree/application/LocalTreeBuilder';
@@ -10,8 +15,6 @@ import { SimpleFolderCreator } from '../../context/virtual-drive/folders/applica
 import { RemoteTreeBuilder } from '../../context/virtual-drive/remoteTree/application/RemoteTreeBuilder';
 import { RemoteTree } from '../../context/virtual-drive/remoteTree/domain/RemoteTree';
 import { BackupInfo } from './BackupInfo';
-import { AddedFilesBatchCreator } from './batches/AddedFilesBatchCreator';
-import { ModifiedFilesBatchCreator } from './batches/ModifiedFilesBatchCreator';
 import { DiffFilesCalculatorService, FilesDiff } from './diff/DiffFilesCalculatorService';
 import { FoldersDiff, FoldersDiffCalculator } from './diff/FoldersDiffCalculator';
 import { relative } from './utils/relative';
@@ -24,15 +27,16 @@ import { BackupProgressTracker } from '../../backend/features/backup/backup-prog
 import { RetryError } from '../shared/retry/RetryError';
 import { Either, left, right } from '../../context/shared/domain/Either';
 import { addFileToTrash } from '../../infra/drive-server/services/files/services/add-file-to-trash';
+import { createBackupUploadExecutor } from '../../backend/features/backup/upload/create-backup-upload-executor';
 
 @Service()
 export class BackupService {
   constructor(
     private readonly localTreeBuilder: LocalTreeBuilder,
     private readonly remoteTreeBuilder: RemoteTreeBuilder,
-    private readonly fileBatchUploader: FileBatchUploader,
-    private readonly fileBatchUpdater: FileBatchUpdater,
     private readonly simpleFolderCreator: SimpleFolderCreator,
+    private readonly environment: Environment,
+    private readonly bucket: string,
   ) {}
 
   // TODO: PB-3897 - Change Signature of this method for a better error handling
@@ -188,52 +192,51 @@ export class BackupService {
 
     const { added, modified, deleted } = filesDiff;
 
-    logger.debug({ tag: 'BACKUPS', msg: 'Files added', count: added.length });
-    await this.uploadAndCreate(local.root.path, added, remote, signal, tracker);
+    if (added.length > 0) {
+      logger.debug({ tag: 'BACKUPS', msg: 'Files added', count: added.length });
+      await this.uploadAndCreate(local.root.path, added, remote, signal, tracker);
+    }
 
-    logger.debug({ tag: 'BACKUPS', msg: 'Files modified', count: modified.size });
-    await this.uploadAndUpdate(modified, local, remote, signal, tracker);
+    if (modified.size > 0) {
+      logger.debug({ tag: 'BACKUPS', msg: 'Files modified', count: modified.size });
+      await this.uploadAndUpdate(modified, signal, tracker);
+    }
 
-    logger.debug({ tag: 'BACKUPS', msg: 'Files deleted', count: deleted.length });
-    await this.deleteRemoteFiles(deleted, signal, tracker);
+    if (deleted.length > 0) {
+      logger.debug({ tag: 'BACKUPS', msg: 'Files deleted', count: deleted.length });
+      await this.deleteRemoteFiles(deleted, signal, tracker);
+    }
   }
 
   private async uploadAndCreate(
     localRootPath: string,
     added: Array<LocalFile>,
-    tree: RemoteTree,
+    remoteTree: RemoteTree,
     signal: AbortSignal,
     tracker: BackupProgressTracker,
   ): Promise<void> {
-    const batches = AddedFilesBatchCreator.run(added);
+    const executor = createBackupUploadExecutor(localRootPath, remoteTree, this.bucket, this.environment, tracker);
 
-    for (const batch of batches) {
-      if (signal.aborted) {
-        return;
-      }
-      // eslint-disable-next-line no-await-in-loop
-      await this.fileBatchUploader.run(localRootPath, tree, batch, signal);
-      tracker.incrementProcessed(batch.length);
+    const result = await executeAsyncQueue(added, executor, { concurrency: DEFAULT_CONCURRENCY, signal });
+
+    if (result.error) {
+      throw result.error;
     }
   }
 
   private async uploadAndUpdate(
     modified: Map<LocalFile, File>,
-    localTree: LocalTree,
-    remoteTree: RemoteTree,
     signal: AbortSignal,
     tracker: BackupProgressTracker,
   ): Promise<void> {
-    const batches = ModifiedFilesBatchCreator.run(modified);
+    const items: ModifiedFilePair[] = Array.from(modified.entries());
 
-    for (const batch of batches) {
-      logger.debug({ tag: 'BACKUPS', msg: 'Signal aborted', aborted: signal.aborted });
-      if (signal.aborted) {
-        return;
-      }
-      // eslint-disable-next-line no-await-in-loop
-      await this.fileBatchUpdater.run(localTree.root, remoteTree, Array.from(batch.keys()), signal);
-      tracker.incrementProcessed(batch.size);
+    const executor = createBackupUpdateExecutor(this.bucket, this.environment, tracker);
+
+    const result = await executeAsyncQueue(items, executor, { concurrency: DEFAULT_CONCURRENCY, signal });
+
+    if (result.error) {
+      throw result.error;
     }
   }