diff --git a/.env.example b/.env.example index 76a03a7fad..3e3f7879ef 100644 --- a/.env.example +++ b/.env.example @@ -12,3 +12,7 @@ NOTIFICATIONS_URL= # Desktop client identification INTERNXT_DESKTOP_HEADER_KEY= + +# Download and prefetch settings +INTERNXT_DRIVE_DOWNLOAD_BLOCK_SIZE_MB= +INTERNXT_DRIVE_READ_PREFETCH_BLOCKS_AHEAD= diff --git a/src/backend/features/fuse/on-read/download-cache/constants.ts b/src/backend/features/fuse/on-read/download-cache/constants.ts index 1fc635e970..be8ddd93b7 100644 --- a/src/backend/features/fuse/on-read/download-cache/constants.ts +++ b/src/backend/features/fuse/on-read/download-cache/constants.ts @@ -1,7 +1,46 @@ /** - * 4MB blocks — matches the chunk size used by the legacy downloader, proven to work well - * for this codebase. Each block is downloaded in full on first access regardless of how - * small the FUSE read is, so subsequent reads within the same block are served from disk. + * 4MB default blocks — lower latency on slow links while preserving cache locality. + * Each block is downloaded in full on first access regardless of how small the FUSE read is, + * so subsequent reads within the same block are served from disk. */ -export const BLOCK_SIZE = 4 * 1024 * 1024; +const DEFAULT_BLOCK_SIZE_MB = 4; +const ALLOWED_BLOCK_SIZE_MB = new Set([1, 2, 4, 8]); +const PREFETCH_DEFAULT_BLOCKS_AHEAD = 5; +const PREFETCH_MAX_BLOCKS_AHEAD = 8; + +function getConfiguredBlockSizeInMb() { + const configuredValue = process.env.INTERNXT_DRIVE_DOWNLOAD_BLOCK_SIZE_MB; + + if (!configuredValue) { + return DEFAULT_BLOCK_SIZE_MB; + } + + const parsedValue = Number.parseInt(configuredValue, 10); + if (!ALLOWED_BLOCK_SIZE_MB.has(parsedValue)) { + return DEFAULT_BLOCK_SIZE_MB; + } + + return parsedValue; +} + +function getPrefetchBlocksAhead({ configuredValue }: { configuredValue: string | undefined }) { + if (!configuredValue) { + return PREFETCH_DEFAULT_BLOCKS_AHEAD; + } + + const parsed = Number.parseInt(configuredValue, 10); + if (Number.isNaN(parsed)) { + return PREFETCH_DEFAULT_BLOCKS_AHEAD; + } + + return Math.max(0, Math.min(parsed, PREFETCH_MAX_BLOCKS_AHEAD)); +} + +export function getReadPrefetchBlocksAhead() { + return getPrefetchBlocksAhead({ + configuredValue: process.env.INTERNXT_DRIVE_READ_PREFETCH_BLOCKS_AHEAD, + }); +} + +export const BLOCK_SIZE = getConfiguredBlockSizeInMb() * 1024 * 1024; export const BITS_PER_BYTE = 8; diff --git a/src/backend/features/fuse/on-read/download-cache/download-and-save-block.ts b/src/backend/features/fuse/on-read/download-cache/download-and-save-block.ts index d60fffae70..f390afadf6 100644 --- a/src/backend/features/fuse/on-read/download-cache/download-and-save-block.ts +++ b/src/backend/features/fuse/on-read/download-cache/download-and-save-block.ts @@ -36,6 +36,7 @@ export async function downloadAndCacheBlock({ blockLength, }: Props): Promise> { if (isAborted(state)) return { data: undefined }; + if (blockLength <= 0 || blockStart >= virtualFile.size) return { data: undefined }; try { const download = await downloadBlockWithRetry({ diff --git a/src/backend/features/fuse/on-read/download-cache/expand-to-block-boundaries.test.ts b/src/backend/features/fuse/on-read/download-cache/expand-to-block-boundaries.test.ts index 8c7e5f9600..dab2c8d00c 100644 --- a/src/backend/features/fuse/on-read/download-cache/expand-to-block-boundaries.test.ts +++ b/src/backend/features/fuse/on-read/download-cache/expand-to-block-boundaries.test.ts @@ -40,4 +40,26 @@ describe('expandToBlockBoundaries', () => { expect(result).toStrictEqual({ blockStart: BLOCK_SIZE, blockLength: partialLastBlockLength }); }); + + it('returns zero block length when read starts exactly at EOF', () => { + const fileSize = BLOCK_SIZE * 2; + + const result = expandToBlockBoundaries({ + range: { position: fileSize, length: 1 }, + fileSize, + }); + + expect(result).toStrictEqual({ blockStart: fileSize, blockLength: 0 }); + }); + + it('returns zero block length when read starts after EOF', () => { + const fileSize = BLOCK_SIZE * 2; + + const result = expandToBlockBoundaries({ + range: { position: fileSize + 123, length: BLOCK_SIZE }, + fileSize, + }); + + expect(result).toStrictEqual({ blockStart: fileSize, blockLength: 0 }); + }); }); diff --git a/src/backend/features/fuse/on-read/download-cache/expand-to-block-boundaries.ts b/src/backend/features/fuse/on-read/download-cache/expand-to-block-boundaries.ts index ebfc009bd5..0fc05d76aa 100644 --- a/src/backend/features/fuse/on-read/download-cache/expand-to-block-boundaries.ts +++ b/src/backend/features/fuse/on-read/download-cache/expand-to-block-boundaries.ts @@ -10,8 +10,18 @@ export function expandToBlockBoundaries({ range, fileSize }: { range: ReadRange; blockStart: number; blockLength: number; } { - const blockStart = Math.floor(range.position / BLOCK_SIZE) * BLOCK_SIZE; - const end = range.position + range.length; - const blockEnd = Math.min(Math.ceil(end / BLOCK_SIZE) * BLOCK_SIZE, fileSize); + if (fileSize <= 0 || range.length <= 0 || range.position >= fileSize) { + return { blockStart: Math.max(0, Math.min(range.position, fileSize)), blockLength: 0 }; + } + + const clampedStart = Math.max(0, range.position); + const clampedEndExclusive = Math.min(clampedStart + range.length, fileSize); + + if (clampedEndExclusive <= clampedStart) { + return { blockStart: clampedStart, blockLength: 0 }; + } + + const blockStart = Math.floor(clampedStart / BLOCK_SIZE) * BLOCK_SIZE; + const blockEnd = Math.min(Math.ceil(clampedEndExclusive / BLOCK_SIZE) * BLOCK_SIZE, fileSize); return { blockStart, blockLength: blockEnd - blockStart }; } diff --git a/src/backend/features/fuse/on-read/download-cache/hydration-state.test.ts b/src/backend/features/fuse/on-read/download-cache/hydration-state.test.ts index 255461e2ba..ca6b188244 100644 --- a/src/backend/features/fuse/on-read/download-cache/hydration-state.test.ts +++ b/src/backend/features/fuse/on-read/download-cache/hydration-state.test.ts @@ -164,6 +164,15 @@ describe('hydration-state lifecycle', () => { expect(getHydratedBytes(state)).toBe(BLOCK_SIZE); }); + it('does not mark bytes when the range starts after EOF', () => { + const fileSize = BLOCK_SIZE * 2; + const state = getOrCreateHydrationState('out-of-bounds', fileSize); + + markBlocksInRangeDownloaded(state, { position: fileSize + BLOCK_SIZE, length: BLOCK_SIZE }); + + expect(getHydratedBytes(state)).toBe(0); + }); + it('treats an empty file as fully hydrated without marking any blocks', () => { const state = getOrCreateHydrationState('empty-contents-id', 0); diff --git a/src/backend/features/fuse/on-read/download-cache/hydration-state.ts b/src/backend/features/fuse/on-read/download-cache/hydration-state.ts index 282433bfbc..d556e5593f 100644 --- a/src/backend/features/fuse/on-read/download-cache/hydration-state.ts +++ b/src/backend/features/fuse/on-read/download-cache/hydration-state.ts @@ -133,10 +133,20 @@ export function getHydratedBytes(state: FileHydrationState): number { return state.hydratedBytes; } -function blocksWithinRange({ position, length }: ReadRange): Array { - if (length <= 0) return []; - const first = blockIndexForByte(position); - const last = blockIndexForByte(position + length - 1); +function blocksWithinRange(state: FileHydrationState, { position, length }: ReadRange): Array { + if (length <= 0 || state.totalBlocks === 0 || position >= state.fileSize) { + return []; + } + + const clampedStart = Math.max(0, position); + const clampedEndExclusive = Math.min(position + length, state.fileSize); + + if (clampedEndExclusive <= clampedStart) { + return []; + } + + const first = blockIndexForByte(clampedStart); + const last = blockIndexForByte(clampedEndExclusive - 1); const blocks: number[] = []; for (let block = first; block <= last; block++) { blocks.push(block); @@ -145,11 +155,11 @@ function blocksWithinRange({ position, length }: ReadRange): Array { } export function isRangeHydrated(state: FileHydrationState, { position, length }: ReadRange): boolean { - return blocksWithinRange({ position, length }).every((block) => getBit(state.bitmap, block)); + return blocksWithinRange(state, { position, length }).every((block) => getBit(state.bitmap, block)); } export function markBlocksInRangeDownloaded(state: FileHydrationState, { position, length }: ReadRange): void { - for (const block of blocksWithinRange({ position, length })) { + for (const block of blocksWithinRange(state, { position, length })) { if (!getBit(state.bitmap, block)) { setBit(state.bitmap, block); state.hydratedBytes += blockByteLength(state, block); @@ -167,7 +177,7 @@ function blockByteLength(state: FileHydrationState, block: number): number { * Call after waiting for existing in-flight blocks to identify the remaining work. */ export function getMissingBlocks(state: FileHydrationState, { position, length }: ReadRange): number[] { - return blocksWithinRange({ position, length }).filter( + return blocksWithinRange(state, { position, length }).filter( (block) => !getBit(state.bitmap, block) && !state.blocksBeingDownloaded.has(block), ); } @@ -177,7 +187,7 @@ export function getBlocksBeingDownloaded( { position, length }: ReadRange, ): Map>> { const blocksBeingDownloadedWithinRange = new Map>>(); - for (const block of blocksWithinRange({ position, length })) { + for (const block of blocksWithinRange(state, { position, length })) { const existing = state.blocksBeingDownloaded.get(block); if (existing) blocksBeingDownloadedWithinRange.set(block, existing); } diff --git a/src/backend/features/fuse/on-read/handle-read-callback.test.ts b/src/backend/features/fuse/on-read/handle-read-callback.test.ts index 28191196b2..11df1f6072 100644 --- a/src/backend/features/fuse/on-read/handle-read-callback.test.ts +++ b/src/backend/features/fuse/on-read/handle-read-callback.test.ts @@ -3,9 +3,9 @@ import * as readChunkModule from './read-chunk-from-disk'; import * as fileExistsModule from './download-cache/file-exists-on-disk'; import * as allocateFileModule from './download-cache/allocate-file'; import * as downloadAndSaveBlockModule from './download-cache/download-and-save-block'; -import * as downloadFileModule from '../../../../infra/environment/download-file/download-file'; import { clearHydrationState } from './download-cache/hydration-state'; -import { partialSpyOn, call } from '../../../../../tests/vitest/utils.helper'; +import { BLOCK_SIZE } from './download-cache/constants'; +import { partialSpyOn, call, testSleep } from '../../../../../tests/vitest/utils.helper'; import { type File } from '../../../../context/virtual-drive/files/domain/File'; import { FuseIOError, FuseNoSuchFileOrDirectoryError } from '../../../../apps/drive/fuse/callbacks/FuseErrors'; @@ -13,7 +13,6 @@ const readChunkFromDiskMock = partialSpyOn(readChunkModule, 'readChunkFromDisk') const fileExistsOnDiskMock = partialSpyOn(fileExistsModule, 'fileExistsOnDisk'); const allocateFileMock = partialSpyOn(allocateFileModule, 'allocateFile'); const downloadAndCacheBlockMock = partialSpyOn(downloadAndSaveBlockModule, 'downloadAndCacheBlock'); -const downloadFileRangeMock = partialSpyOn(downloadFileModule, 'downloadFileRange'); const virtualFile = { contentsId: 'contents-123', @@ -92,35 +91,27 @@ describe('handleReadCallback', () => { }); describe('when process is a thumbnail generator', () => { - it('should download the exact requested range without block expansion', async () => { - const chunk = Buffer.from('image-header'); - downloadFileRangeMock.mockResolvedValue({ data: chunk }); + it('should route thumbnail reads through cache hydration', async () => { const deps = createDeps({ processName: 'pool-org.gnome.', range: { position: 0, length: 32768 } }); const result = await handleReadCallback(deps); - expect(result.data).toBe(chunk); - call(downloadFileRangeMock).toMatchObject({ - fileId: virtualFile.contentsId, - bucketId: deps.bucketId, - mnemonic: deps.mnemonic, - range: { position: 0, length: 32768 }, - }); + expect(result.data).toStrictEqual(Buffer.from('data')); + expect(downloadAndCacheBlockMock).toHaveBeenCalledOnce(); }); - it('should not allocate a cache file or download blocks', async () => { - downloadFileRangeMock.mockResolvedValue({ data: Buffer.from('bytes') }); + it('should allocate and download cache blocks when needed', async () => { + fileExistsOnDiskMock.mockResolvedValue(false); const deps = createDeps({ processName: 'pool-org.gnome.' }); await handleReadCallback(deps); - expect(fileExistsOnDiskMock).not.toHaveBeenCalled(); - expect(allocateFileMock).not.toHaveBeenCalled(); - expect(downloadAndCacheBlockMock).not.toHaveBeenCalled(); + expect(fileExistsOnDiskMock).toHaveBeenCalled(); + expect(allocateFileMock).toHaveBeenCalledOnce(); + expect(downloadAndCacheBlockMock).toHaveBeenCalledOnce(); }); it('should not emit download progress or register the file', async () => { - downloadFileRangeMock.mockResolvedValue({ data: Buffer.from('bytes') }); const deps = createDeps({ processName: 'pool-org.gnome.' }); await handleReadCallback(deps); @@ -129,8 +120,8 @@ describe('handleReadCallback', () => { expect(deps.saveToRepository).not.toHaveBeenCalled(); }); - it('should return EIO when the ranged download fails', async () => { - downloadFileRangeMock.mockResolvedValue({ error: new Error('network error') }); + it('should return EIO when block hydration fails', async () => { + downloadAndCacheBlockMock.mockResolvedValue({ error: new Error('network error') }); const deps = createDeps({ processName: 'pool-org.gnome.' }); const result = await handleReadCallback(deps); @@ -139,6 +130,28 @@ describe('handleReadCallback', () => { }); }); + describe('when process is a normal reader', () => { + it('should prefetch blocks while reading the requested range', async () => { + const largeVirtualFile = { + ...virtualFile, + contentsId: 'large-normal-reader-file', + size: BLOCK_SIZE * 8 + 100, + } as unknown as File; + const deps = createDeps({ processName: 'vlc', range: { position: 0, length: 10 } }); + + fileExistsOnDiskMock.mockResolvedValue(false); + readChunkFromDiskMock.mockResolvedValue(Buffer.from('data')); + deps.findVirtualFile = vi.fn().mockResolvedValue(largeVirtualFile); + + await handleReadCallback(deps); + + await testSleep(0); + + expect(downloadAndCacheBlockMock).toHaveBeenCalledTimes(6); + expect(readChunkFromDiskMock).toHaveBeenCalled(); + }); + }); + describe('when allocating the cache file', () => { it('returns EIO and does not download when allocation fails', async () => { fileExistsOnDiskMock.mockResolvedValue(false); diff --git a/src/backend/features/fuse/on-read/handle-read-callback.ts b/src/backend/features/fuse/on-read/handle-read-callback.ts index fdc3dc6ff1..9d4a75667e 100644 --- a/src/backend/features/fuse/on-read/handle-read-callback.ts +++ b/src/backend/features/fuse/on-read/handle-read-callback.ts @@ -1,20 +1,17 @@ import { logger } from '@internxt/drive-desktop-core/build/backend'; import { type TemporalFile } from '../../../../context/storage/TemporalFiles/domain/TemporalFile'; import { type File } from '../../../../context/virtual-drive/files/domain/File'; -import { - type FuseError, - FuseIOError, - FuseNoSuchFileOrDirectoryError, -} from '../../../../apps/drive/fuse/callbacks/FuseErrors'; -import { downloadFileRange } from '../../../../infra/environment/download-file/download-file'; +import { type FuseError, FuseNoSuchFileOrDirectoryError } from '../../../../apps/drive/fuse/callbacks/FuseErrors'; import { type Result } from '../../../../context/shared/domain/Result'; import { readChunkFromDisk } from './read-chunk-from-disk'; import nodePath from 'node:path'; import { PATHS } from '../../../../core/electron/paths'; import { EMPTY } from './constants'; +import { getReadPrefetchBlocksAhead } from './download-cache/constants'; import { readOrHydrate } from './read-or-hydrate'; import { type HandleReadDeps, type ReadRange } from './types'; import { isThumbnailProcess } from './thumbnail-processes'; + export type HandleReadCallbackProps = HandleReadDeps & { findVirtualFile: (path: string) => Promise; findTemporalFile: (path: string) => Promise; @@ -50,11 +47,25 @@ export async function handleReadCallback({ if (isThumbnailProcess(processName)) { logger.debug({ - msg: '[ReadCallback] thumbnail process, downloading exact range', + msg: '[ReadCallback] thumbnail process, reading through cache hydration', process: processName, file: virtualFile.nameWithExtension, }); - return readExactRangeForThumbnail({ bucketId, mnemonic, network, virtualFile, range }); + + const filePath = nodePath.join(PATHS.DOWNLOADED, virtualFile.contentsId); + return readOrHydrate({ + bucketId, + mnemonic, + network, + // Thumbnail reads should not spam progress updates in UI. + onDownloadProgress: () => undefined, + // Thumbnail reads should not register files as offline available. + saveToRepository: async () => undefined, + virtualFile, + filePath, + range, + prefetchBlocksAhead: 0, + }); } const filePath = nodePath.join(PATHS.DOWNLOADED, virtualFile.contentsId); @@ -68,6 +79,7 @@ export async function handleReadCallback({ virtualFile, filePath, range, + prefetchBlocksAhead: getReadPrefetchBlocksAhead(), }); } @@ -87,27 +99,3 @@ async function readFromTemporalFile( const chunk = await readChunkFromDisk(temporalFile.contentFilePath, length, position); return { data: chunk ?? EMPTY }; } - -type ThumbnailRangeProps = Pick & { - virtualFile: File; -}; - -async function readExactRangeForThumbnail({ - bucketId, - mnemonic, - network, - virtualFile, - range, -}: ThumbnailRangeProps): Promise> { - const { signal } = new AbortController(); - const result = await downloadFileRange({ - fileId: virtualFile.contentsId, - bucketId, - mnemonic, - network, - range, - signal, - }); - if (result.error) return { error: new FuseIOError(result.error.message) }; - return { data: result.data }; -} diff --git a/src/backend/features/fuse/on-read/read-or-hydrate.test.ts b/src/backend/features/fuse/on-read/read-or-hydrate.test.ts index 4fbd395dc6..4b0e803c00 100644 --- a/src/backend/features/fuse/on-read/read-or-hydrate.test.ts +++ b/src/backend/features/fuse/on-read/read-or-hydrate.test.ts @@ -43,7 +43,6 @@ function createDeps(overrides: Partial = {}): ReadOrHydrateDe describe('readOrHydrate', () => { beforeEach(() => { clearHydrationState(); - vi.clearAllMocks(); fileExistsOnDiskMock.mockResolvedValue(true); allocateFileMock.mockResolvedValue(undefined); downloadAndCacheBlockMock.mockResolvedValue({ data: undefined }); @@ -493,4 +492,102 @@ describe('readOrHydrate', () => { expect(saveToRepository).toHaveBeenCalledOnce(); expect(downloadFinished).toHaveBeenCalledOnce(); }); + + describe('schedulePrefetchBlocksAhead (via readOrHydrate)', () => { + it('prefetches next blocks when enabled for sequential reads', async () => { + const multiBlockFile = { + ...virtualFile, + contentsId: 'multi-block-file', + size: BLOCK_SIZE * 4 + 100, + } as unknown as File; + const state = getOrCreateHydrationState(multiBlockFile.contentsId, multiBlockFile.size); + markBlocksInRangeDownloaded(state, { position: 0, length: BLOCK_SIZE }); + downloadAndCacheBlockMock.mockImplementation(async ({ state: innerState, blockStart, blockLength }) => { + markBlocksInRangeDownloaded(innerState, { position: blockStart, length: blockLength }); + return { data: undefined }; + }); + + const result = await readOrHydrate({ + ...createDeps(), + virtualFile: multiBlockFile, + filePath: '/tmp/cache-file', + range: { position: 0, length: 131072 }, + prefetchBlocksAhead: 2, + }); + + expect(result.data).toStrictEqual(Buffer.from('data')); + expect(downloadAndCacheBlockMock).toHaveBeenCalledTimes(2); + expect(downloadAndCacheBlockMock).toHaveBeenNthCalledWith( + 1, + expect.objectContaining({ + blockStart: BLOCK_SIZE, + }), + ); + expect(downloadAndCacheBlockMock).toHaveBeenNthCalledWith( + 2, + expect.objectContaining({ + blockStart: BLOCK_SIZE * 2, + }), + ); + }); + + it('does not prefetch when the requested block already reaches end of file', async () => { + const twoBlockFile = { + ...virtualFile, + contentsId: 'two-block-file', + size: BLOCK_SIZE * 2, + } as unknown as File; + const state = getOrCreateHydrationState(twoBlockFile.contentsId, twoBlockFile.size); + markBlocksInRangeDownloaded(state, { position: BLOCK_SIZE, length: BLOCK_SIZE }); + + const result = await readOrHydrate({ + ...createDeps(), + virtualFile: twoBlockFile, + filePath: '/tmp/cache-file', + range: { position: BLOCK_SIZE, length: 4096 }, + prefetchBlocksAhead: 3, + }); + + expect(result.data).toStrictEqual(Buffer.from('data')); + expect(downloadAndCacheBlockMock).not.toHaveBeenCalled(); + }); + + it('prefetches only missing blocks ahead and skips already hydrated ones', async () => { + const fiveBlockFile = { + ...virtualFile, + contentsId: 'five-block-file', + size: BLOCK_SIZE * 5, + } as unknown as File; + const state = getOrCreateHydrationState(fiveBlockFile.contentsId, fiveBlockFile.size); + markBlocksInRangeDownloaded(state, { position: 0, length: BLOCK_SIZE }); + markBlocksInRangeDownloaded(state, { position: BLOCK_SIZE, length: BLOCK_SIZE }); + downloadAndCacheBlockMock.mockImplementation(async ({ state: innerState, blockStart, blockLength }) => { + markBlocksInRangeDownloaded(innerState, { position: blockStart, length: blockLength }); + return { data: undefined }; + }); + + const result = await readOrHydrate({ + ...createDeps(), + virtualFile: fiveBlockFile, + filePath: '/tmp/cache-file', + range: { position: 0, length: 131072 }, + prefetchBlocksAhead: 3, + }); + + expect(result.data).toStrictEqual(Buffer.from('data')); + expect(downloadAndCacheBlockMock).toHaveBeenCalledTimes(2); + expect(downloadAndCacheBlockMock).toHaveBeenNthCalledWith( + 1, + expect.objectContaining({ + blockStart: BLOCK_SIZE * 2, + }), + ); + expect(downloadAndCacheBlockMock).toHaveBeenNthCalledWith( + 2, + expect.objectContaining({ + blockStart: BLOCK_SIZE * 3, + }), + ); + }); + }); }); diff --git a/src/backend/features/fuse/on-read/read-or-hydrate.ts b/src/backend/features/fuse/on-read/read-or-hydrate.ts index 0b02e20180..d36b923f7c 100644 --- a/src/backend/features/fuse/on-read/read-or-hydrate.ts +++ b/src/backend/features/fuse/on-read/read-or-hydrate.ts @@ -28,6 +28,7 @@ type Props = HandleReadDeps & { virtualFile: File; filePath: string; range: ReadRange; + prefetchBlocksAhead?: number; }; export async function readOrHydrate({ @@ -39,6 +40,7 @@ export async function readOrHydrate({ virtualFile, filePath, range, + prefetchBlocksAhead = 0, }: Props): Promise> { logger.debug({ msg: '[ReadCallback] read request:', @@ -70,6 +72,20 @@ export async function readOrHydrate({ if (downloadResult.error) return { error: fuseIOErrorFrom(downloadResult.error) }; } + if (prefetchBlocksAhead > 0) { + schedulePrefetchBlocksAhead({ + bucketId, + mnemonic, + network, + onDownloadProgress, + virtualFile, + filePath, + state: state.data, + range, + blocksAhead: prefetchBlocksAhead, + }); + } + await finalizeFullyHydratedFileIfNeeded(saveToRepository, virtualFile, state.data); if (wasAborted(state.data)) return { data: EMPTY }; @@ -80,6 +96,102 @@ export async function readOrHydrate({ } } +function schedulePrefetchBlocksAhead({ + bucketId, + mnemonic, + network, + onDownloadProgress, + virtualFile, + filePath, + state, + range, + blocksAhead, +}: { + onDownloadProgress: HandleReadDeps['onDownloadProgress']; + bucketId: HandleReadDeps['bucketId']; + mnemonic: HandleReadDeps['mnemonic']; + network: HandleReadDeps['network']; + virtualFile: File; + filePath: string; + range: ReadRange; + state: FileHydrationState; + blocksAhead: number; +}) { + if (wasAborted(state)) { + return; + } + + const { blockStart, blockLength } = expandToBlockBoundaries({ range, fileSize: virtualFile.size }); + const nextBlockStart = blockStart + blockLength; + + if (nextBlockStart >= virtualFile.size) { + return; + } + + const missingBlocksAhead = getMissingBlocks(state, { + position: nextBlockStart, + length: blocksAhead * BLOCK_SIZE, + }); + if (missingBlocksAhead.length === 0) { + return; + } + + const prefetchedBlocks = missingBlocksAhead.slice(0, blocksAhead); + + logger.debug({ + msg: '[ReadCallback] prefetching next blocks', + file: virtualFile.nameWithExtension, + blocks: prefetchedBlocks, + }); + + for (const block of prefetchedBlocks) { + const start = block * BLOCK_SIZE; + const end = Math.min(start + BLOCK_SIZE, virtualFile.size); + const blockLength = end - start; + + if (blockLength <= 0) { + logger.debug({ + msg: '[ReadCallback] skipping invalid prefetch block outside file bounds', + file: virtualFile.nameWithExtension, + block, + start, + end, + fileSize: virtualFile.size, + }); + continue; + } + + const download = downloadAndCacheBlock({ + bucketId, + mnemonic, + network, + onDownloadProgress, + virtualFile, + filePath, + state, + blockStart: start, + blockLength, + }); + + setBlockDownloadInFlight(state, block, download); + + void download + .then((result) => { + if (result.error) { + logger.warn({ + msg: '[ReadCallback] next-block prefetch failed', + file: virtualFile.nameWithExtension, + block, + error: result.error, + }); + } + }) + .finally(() => { + clearBlockDownloadInFlight(state, block, download); + }); + } +} + async function ensureFileAllocated( filePath: string, virtualFile: File, @@ -118,6 +230,10 @@ async function ensureRangeDownloaded({ }): Promise> { const { blockStart, blockLength } = expandToBlockBoundaries({ range, fileSize: virtualFile.size }); + if (blockLength <= 0 || blockStart >= virtualFile.size) { + return { data: undefined }; + } + const blocksBeingDownloaded = getBlocksBeingDownloaded(state, { position: blockStart, length: blockLength }); if (blocksBeingDownloaded.size > 0) { logger.debug({ @@ -140,6 +256,8 @@ async function ensureRangeDownloaded({ const downloads = missingBlocks.map((block) => { const start = block * BLOCK_SIZE; const end = Math.min(start + BLOCK_SIZE, virtualFile.size); + const boundedBlockLength = end - start; + const download = downloadAndCacheBlock({ bucketId, mnemonic, @@ -149,7 +267,7 @@ async function ensureRangeDownloaded({ filePath, state, blockStart: start, - blockLength: end - start, + blockLength: boundedBlockLength, }); setBlockDownloadInFlight(state, block, download); download.finally(() => clearBlockDownloadInFlight(state, block, download)); diff --git a/src/backend/features/virtual-drive/services/operations/read.service.ts b/src/backend/features/virtual-drive/services/operations/read.service.ts index e875de543a..c5dc245263 100644 --- a/src/backend/features/virtual-drive/services/operations/read.service.ts +++ b/src/backend/features/virtual-drive/services/operations/read.service.ts @@ -12,6 +12,7 @@ import { logger } from '@internxt/drive-desktop-core/build/backend'; import { getCredentials } from '../../../../../apps/main/auth/get-credentials'; import { DependencyInjectionUserProvider } from '../../../../../apps/shared/dependency-injection/DependencyInjectionUserProvider'; import { buildNetworkClient } from '../../../../../infra/environment/download-file/build-network-client'; +import { shouldEmitProgress, type ProgressReporterState } from './should-emit-progress'; export async function read( path: string, @@ -21,6 +22,7 @@ export async function read( container: Container, ): Promise> { try { + const progressReporterState: ProgressReporterState = { lastUpdateAt: 0 }; const { mnemonic } = getCredentials(); const user = DependencyInjectionUserProvider.get(); const network = buildNetworkClient({ bridgeUser: user.bridgeUser, userId: user.userId }); @@ -31,6 +33,10 @@ export async function read( findVirtualFile: (p) => container.get(FirstsFileSearcher).run({ path: p }), findTemporalFile: (p) => container.get(TemporalFileByPathFinder).run(p), onDownloadProgress: (name, extension, bytesDownloaded, fileSize, elapsedTime) => { + if (!shouldEmitProgress({ bytesDownloaded, fileSize, state: progressReporterState })) { + return; + } + tracker.downloadUpdate(name, extension, { percentage: Math.min(bytesDownloaded / fileSize, 1), elapsedTime, diff --git a/src/backend/features/virtual-drive/services/operations/should-emit-progress.test.ts b/src/backend/features/virtual-drive/services/operations/should-emit-progress.test.ts new file mode 100644 index 0000000000..cd24a3b4ab --- /dev/null +++ b/src/backend/features/virtual-drive/services/operations/should-emit-progress.test.ts @@ -0,0 +1,54 @@ +import { shouldEmitProgress, type ProgressReporterState } from './should-emit-progress'; + +describe('should-emit-progress', () => { + let state: ProgressReporterState; + + beforeEach(() => { + vi.useFakeTimers(); + vi.setSystemTime(new Date('2026-01-01T00:00:00.000Z')); + state = { lastUpdateAt: 0 }; + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it('should emit when enough time has elapsed since the last update', () => { + state.lastUpdateAt = Date.now() - 300; + + const result = shouldEmitProgress({ + bytesDownloaded: 10, + fileSize: 100, + state, + }); + + expect(result).toBe(true); + expect(state.lastUpdateAt).toBe(Date.now()); + }); + + it('should not emit when the interval has not elapsed and download is not finished', () => { + state.lastUpdateAt = Date.now() - 100; + + const result = shouldEmitProgress({ + bytesDownloaded: 10, + fileSize: 100, + state, + }); + + expect(result).toBe(false); + expect(state.lastUpdateAt).toBe(Date.now() - 100); + }); + + it('should emit the final update even when the interval has not elapsed', () => { + state.lastUpdateAt = Date.now() - 100; + + const result = shouldEmitProgress({ + bytesDownloaded: 100, + fileSize: 100, + state, + }); + + expect(result).toBe(true); + expect(state.lastUpdateAt).toBe(Date.now()); + }); +}); diff --git a/src/backend/features/virtual-drive/services/operations/should-emit-progress.ts b/src/backend/features/virtual-drive/services/operations/should-emit-progress.ts new file mode 100644 index 0000000000..214b261986 --- /dev/null +++ b/src/backend/features/virtual-drive/services/operations/should-emit-progress.ts @@ -0,0 +1,24 @@ +const PROGRESS_UPDATE_INTERVAL_MS = 250; + +export type ProgressReporterState = { + lastUpdateAt: number; +}; + +type Props = { + bytesDownloaded: number; + fileSize: number; + state: ProgressReporterState; +}; + +export function shouldEmitProgress({ bytesDownloaded, fileSize, state }: Props) { + const now = Date.now(); + const reachedEnd = bytesDownloaded >= fileSize; + const elapsedSinceLastUpdate = now - state.lastUpdateAt; + + if (!reachedEnd && elapsedSinceLastUpdate < PROGRESS_UPDATE_INTERVAL_MS) { + return false; + } + + state.lastUpdateAt = now; + return true; +} diff --git a/src/context/storage/StorageFiles/application/download/download-with-progress-tracking.test.ts b/src/context/storage/StorageFiles/application/download/download-with-progress-tracking.test.ts index f66e89f6c1..e772903e39 100644 --- a/src/context/storage/StorageFiles/application/download/download-with-progress-tracking.test.ts +++ b/src/context/storage/StorageFiles/application/download/download-with-progress-tracking.test.ts @@ -63,4 +63,38 @@ describe('downloadWithProgressTracking', () => { size: virtualFile.size, }); }); + + it('throttles progress updates and still emits completion progress', async () => { + const storeMock = partialSpyOn(repository, 'store'); + const nowMock = partialSpyOn(Date, 'now'); + + const virtualFile = FileMother.fromPartial({ + size: 100, + path: 'folder/test-file.txt', + }); + + const handler = { elapsedTime: vi.fn(() => elapsedTime) }; + const stream = Readable.from('hello'); + const metadata = { name: virtualFile.name, type: virtualFile.type, size: virtualFile.size }; + + downloader.run.mockResolvedValue({ stream, metadata, handler }); + nowMock.mockReturnValueOnce(1_000).mockReturnValueOnce(1_001).mockReturnValueOnce(1_002).mockReturnValueOnce(1_003); + + storeMock.mockImplementation(async (_file, _readable, onProgress) => { + [10, 20, 30, 100].forEach((bytes) => onProgress(bytes)); + }); + + await downloadWithProgressTracking({ + virtualFile, + tracker, + downloader: downloader as unknown as StorageFileDownloader, + repository, + }); + + calls(tracker.downloadUpdate).toHaveLength(2); + calls(tracker.downloadUpdate).toMatchObject([ + [metadata.name, metadata.type, { percentage: 0.1, elapsedTime }], + [metadata.name, metadata.type, { percentage: 1, elapsedTime }], + ]); + }); }); diff --git a/src/context/storage/StorageFiles/application/download/download-with-progress-tracking.ts b/src/context/storage/StorageFiles/application/download/download-with-progress-tracking.ts index b2a897d4d6..1e231d763f 100644 --- a/src/context/storage/StorageFiles/application/download/download-with-progress-tracking.ts +++ b/src/context/storage/StorageFiles/application/download/download-with-progress-tracking.ts @@ -4,6 +4,28 @@ import { StorageFilesRepository } from '../../domain/StorageFilesRepository'; import { File } from '../../../../../context/virtual-drive/files/domain/File'; import { StorageFile } from '../../domain/StorageFile'; +const PROGRESS_UPDATE_INTERVAL_MS = 250; + +type CreateThrottledProgressReporterProps = { + onUpdate: (bytesWritten: number) => void; +}; + +function createThrottledProgressReporter({ onUpdate }: CreateThrottledProgressReporterProps) { + let lastUpdateAt = 0; + + return function report({ bytesWritten, totalBytes }: { bytesWritten: number; totalBytes: number }) { + const now = Date.now(); + const reachedEnd = bytesWritten >= totalBytes; + + if (!reachedEnd && now - lastUpdateAt < PROGRESS_UPDATE_INTERVAL_MS) { + return; + } + + lastUpdateAt = now; + onUpdate(bytesWritten); + }; +} + type Props = { virtualFile: File; tracker: DownloadProgressTracker; @@ -20,10 +42,15 @@ export async function downloadWithProgressTracking({ virtualFile, tracker, downl tracker.downloadStarted(virtualFile.name, virtualFile.type); const { stream, metadata, handler } = await downloader.run(storage, virtualFile); + const reportProgress = createThrottledProgressReporter({ + onUpdate: (bytesWritten) => { + const percentage = Math.min(bytesWritten / virtualFile.size, 1); + tracker.downloadUpdate(metadata.name, metadata.type, { percentage, elapsedTime: handler.elapsedTime() }); + }, + }); await repository.store(storage, stream, (bytesWritten) => { - const percentage = Math.min(bytesWritten / virtualFile.size, 1); - tracker.downloadUpdate(metadata.name, metadata.type, { percentage, elapsedTime: handler.elapsedTime() }); + reportProgress({ bytesWritten, totalBytes: virtualFile.size }); }); tracker.downloadFinished(metadata.name, metadata.type); diff --git a/src/infra/drive-server/client/interceptors/rate-limiter/attach-rate-limiter-interceptors.test.ts b/src/infra/drive-server/client/interceptors/rate-limiter/attach-rate-limiter-interceptors.test.ts index 5b350ae7d7..fdb927f965 100644 --- a/src/infra/drive-server/client/interceptors/rate-limiter/attach-rate-limiter-interceptors.test.ts +++ b/src/infra/drive-server/client/interceptors/rate-limiter/attach-rate-limiter-interceptors.test.ts @@ -34,7 +34,7 @@ describe('attachRateLimiterInterceptors', () => { it('should create a request interceptor with a fresh delay state', () => { attachRateLimiterInterceptors(instance); - call(createRequestInterceptor).toMatchObject({ pending: null }); + call(createRequestInterceptor).toMatchObject({ pending: null, requestKey: null }); }); it('should register the request interceptor on the instance', () => { @@ -49,7 +49,7 @@ describe('attachRateLimiterInterceptors', () => { call(createResponseInterceptor).toMatchObject([ instance, { limit: null, remaining: null, reset: null }, - { pending: null }, + { pending: null, requestKey: null }, ]); }); @@ -60,7 +60,7 @@ describe('attachRateLimiterInterceptors', () => { }); it('should share the same delay state between request and response interceptors', () => { - const delayState = { pending: null }; + const delayState = { pending: null, requestKey: null }; attachRateLimiterInterceptors(instance); diff --git a/src/infra/drive-server/client/interceptors/rate-limiter/attach-rate-limiter-interceptors.ts b/src/infra/drive-server/client/interceptors/rate-limiter/attach-rate-limiter-interceptors.ts index bea17de34c..61e4269cf5 100644 --- a/src/infra/drive-server/client/interceptors/rate-limiter/attach-rate-limiter-interceptors.ts +++ b/src/infra/drive-server/client/interceptors/rate-limiter/attach-rate-limiter-interceptors.ts @@ -5,7 +5,7 @@ import { createResponseInterceptor } from './create-response-interceptor'; export function attachRateLimiterInterceptors(instance: AxiosInstance): void { const state: RateLimitState = { limit: null, remaining: null, reset: null }; - const delayState: DelayState = { pending: null }; + const delayState: DelayState = { pending: null, requestKey: null }; instance.interceptors.request.use(createRequestInterceptor(delayState)); diff --git a/src/infra/drive-server/client/interceptors/rate-limiter/create-request-interceptor.test.ts b/src/infra/drive-server/client/interceptors/rate-limiter/create-request-interceptor.test.ts index 9af3b5a0c1..a70edbb73f 100644 --- a/src/infra/drive-server/client/interceptors/rate-limiter/create-request-interceptor.test.ts +++ b/src/infra/drive-server/client/interceptors/rate-limiter/create-request-interceptor.test.ts @@ -6,7 +6,7 @@ describe('createRequestInterceptor', () => { const mockConfig = { url: '/test' } as InternalAxiosRequestConfig; it('should return the config immediately when there is no pending delay', async () => { - const state: DelayState = { pending: null }; + const state: DelayState = { pending: null, requestKey: null }; const interceptor = createRequestInterceptor(state); const result = await interceptor(mockConfig); @@ -20,6 +20,7 @@ describe('createRequestInterceptor', () => { pending: new Promise((resolve) => { resolveDelay = resolve; }), + requestKey: 'GET:/test', }; const interceptor = createRequestInterceptor(state); @@ -39,17 +40,18 @@ describe('createRequestInterceptor', () => { expect(result).toBe(mockConfig); }); - it('should make multiple concurrent requests wait for the same delay', async () => { + it('should make multiple concurrent requests with the same key wait for the same delay', async () => { let resolveDelay!: () => void; const state: DelayState = { pending: new Promise((resolve) => { resolveDelay = resolve; }), + requestKey: 'GET:/a', }; const interceptor = createRequestInterceptor(state); const configA = { url: '/a' } as InternalAxiosRequestConfig; - const configB = { url: '/b' } as InternalAxiosRequestConfig; + const configB = { url: '/a' } as InternalAxiosRequestConfig; const promiseA = interceptor(configA); const promiseB = interceptor(configB); @@ -61,4 +63,20 @@ describe('createRequestInterceptor', () => { expect(resultA).toBe(configA); expect(resultB).toBe(configB); }); + + it('should not wait when there is a pending delay for a different request key', async () => { + let resolveDelay!: () => void; + const state: DelayState = { + pending: new Promise((resolve) => { + resolveDelay = resolve; + }), + requestKey: 'GET:/different', + }; + const interceptor = createRequestInterceptor(state); + + const result = await interceptor(mockConfig); + + expect(result).toBe(mockConfig); + resolveDelay(); + }); }); diff --git a/src/infra/drive-server/client/interceptors/rate-limiter/create-request-interceptor.ts b/src/infra/drive-server/client/interceptors/rate-limiter/create-request-interceptor.ts index 696ff992bb..031c40389a 100644 --- a/src/infra/drive-server/client/interceptors/rate-limiter/create-request-interceptor.ts +++ b/src/infra/drive-server/client/interceptors/rate-limiter/create-request-interceptor.ts @@ -1,11 +1,19 @@ import type { InternalAxiosRequestConfig } from 'axios'; import { DelayState } from './rate-limiter.types'; +function getRequestKey(config: InternalAxiosRequestConfig) { + const method = config.method?.toUpperCase() ?? 'GET'; + const url = config.url ?? ''; + + return `${method}:${url}`; +} + export function createRequestInterceptor( delayState: DelayState, ): (config: InternalAxiosRequestConfig) => Promise { return async (config: InternalAxiosRequestConfig) => { - if (delayState.pending) { + const currentRequestKey = getRequestKey(config); + if (delayState.pending && delayState.requestKey === currentRequestKey) { await delayState.pending; } diff --git a/src/infra/drive-server/client/interceptors/rate-limiter/create-response-interceptor.test.ts b/src/infra/drive-server/client/interceptors/rate-limiter/create-response-interceptor.test.ts index 29dcd6c5c0..8d773a4167 100644 --- a/src/infra/drive-server/client/interceptors/rate-limiter/create-response-interceptor.test.ts +++ b/src/infra/drive-server/client/interceptors/rate-limiter/create-response-interceptor.test.ts @@ -74,7 +74,7 @@ describe('createResponseInterceptor', () => { beforeEach(() => { state = { limit: null, remaining: null, reset: null }; - delayState = { pending: null }; + delayState = { pending: null, requestKey: null }; retryResponse = makeResponse(); instance = { request: vi.fn().mockResolvedValue(retryResponse) } as unknown as AxiosInstance; }); @@ -134,6 +134,7 @@ describe('createResponseInterceptor', () => { call(waitForDelay).toMatchObject([delayState, 3050]); call(instance.request).toMatchObject(config); expect(result).toBe(retryResponse); + expect(delayState.requestKey).toBeNull(); }); it('should default to 5000ms when state.reset is null', async () => { diff --git a/src/infra/drive-server/client/interceptors/rate-limiter/create-response-interceptor.ts b/src/infra/drive-server/client/interceptors/rate-limiter/create-response-interceptor.ts index dd64b7598e..236f61664a 100644 --- a/src/infra/drive-server/client/interceptors/rate-limiter/create-response-interceptor.ts +++ b/src/infra/drive-server/client/interceptors/rate-limiter/create-response-interceptor.ts @@ -6,6 +6,10 @@ import { waitForDelay } from './wait-for-delay'; import { addJitter } from './add-jitter'; import { MAX_RETRIES, RETRY_CONFIG_KEY } from '../../drive-server.constants'; +function getRequestKey({ method, url }: { method?: string; url?: string }) { + return `${method?.toUpperCase() ?? 'GET'}:${url ?? ''}`; +} + export function createResponseInterceptor( instance: AxiosInstance, state: RateLimitState, @@ -49,7 +53,9 @@ export function createResponseInterceptor( maxRetries: MAX_RETRIES, }); + delayState.requestKey = getRequestKey(config); await waitForDelay(delayState, waitMs); + delayState.requestKey = null; config[RETRY_CONFIG_KEY] = retryCount + 1; return instance.request(config); diff --git a/src/infra/drive-server/client/interceptors/rate-limiter/rate-limiter.types.ts b/src/infra/drive-server/client/interceptors/rate-limiter/rate-limiter.types.ts index 07d6f9890e..1b97d98528 100644 --- a/src/infra/drive-server/client/interceptors/rate-limiter/rate-limiter.types.ts +++ b/src/infra/drive-server/client/interceptors/rate-limiter/rate-limiter.types.ts @@ -5,7 +5,10 @@ export type RateLimitState = { remaining: number | null; reset: number | null; }; -export type DelayState = { pending: Promise | null }; +export type DelayState = { + pending: Promise | null; + requestKey: string | null; +}; export type ResponseInterceptor = { onFulfilled: (response: AxiosResponse) => AxiosResponse; diff --git a/src/infra/drive-server/client/interceptors/rate-limiter/update-state-from-headers.test.ts b/src/infra/drive-server/client/interceptors/rate-limiter/update-state-from-headers.test.ts index ba0cbbc6e6..63fd98a306 100644 --- a/src/infra/drive-server/client/interceptors/rate-limiter/update-state-from-headers.test.ts +++ b/src/infra/drive-server/client/interceptors/rate-limiter/update-state-from-headers.test.ts @@ -51,4 +51,28 @@ describe('updateStateFromHeaders', () => { expect(state.remaining).toBe(199); expect(state.reset).toBe(5000); }); + + it('should normalize reset header in seconds to milliseconds', () => { + const state: RateLimitState = { limit: null, remaining: null, reset: null }; + + updateStateFromHeaders(state, { + 'x-internxt-ratelimit-reset': '10', + }); + + expect(state.reset).toBe(10000); + }); + + it('should ignore invalid numeric header values', () => { + const state: RateLimitState = { limit: 10, remaining: 5, reset: 1000 }; + + updateStateFromHeaders(state, { + 'x-internxt-ratelimit-limit': 'NaN', + 'x-internxt-ratelimit-remaining': 'bad', + 'x-internxt-ratelimit-reset': 'oops', + }); + + expect(state.limit).toBe(10); + expect(state.remaining).toBe(5); + expect(state.reset).toBe(1000); + }); }); diff --git a/src/infra/drive-server/client/interceptors/rate-limiter/update-state-from-headers.ts b/src/infra/drive-server/client/interceptors/rate-limiter/update-state-from-headers.ts index 6efc9cebea..4448432f04 100644 --- a/src/infra/drive-server/client/interceptors/rate-limiter/update-state-from-headers.ts +++ b/src/infra/drive-server/client/interceptors/rate-limiter/update-state-from-headers.ts @@ -1,18 +1,45 @@ import { AxiosResponse } from 'axios'; import { RateLimitState } from './rate-limiter.types'; +const MAX_REASONABLE_RESET_SECONDS = 120; + +function parseNumberHeader(value: string | null | undefined) { + if (!value) { + return null; + } + + const parsed = Number.parseInt(value, 10); + if (Number.isNaN(parsed)) { + return null; + } + + return parsed; +} + +function normalizeResetToMs(resetValue: number) { + if (resetValue <= MAX_REASONABLE_RESET_SECONDS) { + return resetValue * 1000; + } + + return resetValue; +} + export function updateStateFromHeaders(state: RateLimitState, headers: AxiosResponse['headers']): void { const limitHeader = headers['x-internxt-ratelimit-limit']; const remainingHeader = headers['x-internxt-ratelimit-remaining']; const resetHeader = headers['x-internxt-ratelimit-reset']; - if (limitHeader) { - state.limit = parseInt(limitHeader, 10); + const limit = parseNumberHeader(limitHeader); + const remaining = parseNumberHeader(remainingHeader); + const reset = parseNumberHeader(resetHeader); + + if (limit !== null) { + state.limit = limit; } - if (remainingHeader) { - state.remaining = parseInt(remainingHeader, 10); + if (remaining !== null) { + state.remaining = remaining; } - if (resetHeader) { - state.reset = parseInt(resetHeader, 10); + if (reset !== null) { + state.reset = normalizeResetToMs(reset); } } diff --git a/src/infra/environment/download-file/download-file.test.ts b/src/infra/environment/download-file/download-file.test.ts index e805d8c900..ed5979c5bb 100644 --- a/src/infra/environment/download-file/download-file.test.ts +++ b/src/infra/environment/download-file/download-file.test.ts @@ -67,4 +67,19 @@ describe('downloadFileRange', () => { }, }); }); + + it('returns empty data and skips network when range length is non-positive', async () => { + const result = await downloadFileRange({ + fileId: 'file-id', + bucketId: 'bucket-id', + mnemonic: 'mnemonic', + network: {} as never, + range: { position: 10, length: 0 }, + signal: new AbortController().signal, + }); + + expect(result).toStrictEqual({ data: Buffer.alloc(0) }); + expect(sdkDownloadFileMock).not.toHaveBeenCalled(); + expect(axiosGetMock).not.toHaveBeenCalled(); + }); }); diff --git a/src/infra/environment/download-file/download-file.ts b/src/infra/environment/download-file/download-file.ts index 9f250d1589..fe980800f6 100644 --- a/src/infra/environment/download-file/download-file.ts +++ b/src/infra/environment/download-file/download-file.ts @@ -5,6 +5,7 @@ import { buildCryptoLib } from './build-crypto-lib'; import { DownloadFileProps } from './types'; import { decryptAtOffset } from './decrypt-at-offset'; import { type Result } from '../../../context/shared/domain/Result'; +import { logger } from '@internxt/drive-desktop-core/build/backend'; export async function downloadFileRange({ signal, @@ -14,6 +15,8 @@ export async function downloadFileRange({ network, range, }: DownloadFileProps): Promise> { + if (range.length <= 0) return { data: Buffer.alloc(0) }; + let encryptedBytes: Buffer | undefined; let decryptedBuffer: Buffer | undefined; let operationError: Error | undefined; @@ -61,12 +64,23 @@ export async function downloadFileRange({ ); } catch (error) { if (signal.aborted) return abortedDownloadResult(); + + logger.warn({ + msg: '[DOWNLOAD RANGE] Download failed', + fileId, + bucketId, + rangeStart: range.position, + rangeLength: range.length, + error, + }); + return { error: error instanceof Error ? error : new Error('Unknown error occurred') }; } if (signal.aborted) return abortedDownloadResult(); if (operationError) return { error: operationError }; if (!decryptedBuffer) return { error: new Error('Decryption did not produce a buffer') }; + return { data: decryptedBuffer }; } @@ -80,6 +94,8 @@ async function fetchEncryptedRange( length: number, signal: AbortSignal, ): Promise { + if (length <= 0) return Buffer.alloc(0); + const response = await axios.get(url, { responseType: 'stream', signal, @@ -89,9 +105,23 @@ async function fetchEncryptedRange( }); return new Promise((resolve, reject) => { - const chunks: Uint8Array[] = []; - response.data.on('data', (chunk: Uint8Array) => chunks.push(chunk)); - response.data.on('end', () => resolve(Buffer.concat(chunks))); + let bytesRead = 0; + let buffer = Buffer.alloc(length); + + response.data.on('data', (chunk: Uint8Array) => { + const source = Buffer.from(chunk); + const requiredLength = bytesRead + source.length; + + if (requiredLength > buffer.length) { + const next = Buffer.alloc(Math.max(buffer.length * 2, requiredLength)); + buffer.copy(next, 0, 0, bytesRead); + buffer = next; + } + + source.copy(buffer, bytesRead); + bytesRead += source.length; + }); + response.data.on('end', () => resolve(buffer.subarray(0, bytesRead))); response.data.on('error', reject); }); }