feat: enhance file download and caching mechanisms

.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=

src/backend/features/fuse/on-read/download-cache/constants.ts

@@ -1,7 +1,25 @@
 /**
- * 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]);
+
+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;
+}
+
+export const BLOCK_SIZE = getConfiguredBlockSizeInMb() * 1024 * 1024;
 export const BITS_PER_BYTE = 8;

src/backend/features/fuse/on-read/download-cache/download-and-save-block.ts

@@ -36,6 +36,7 @@ export async function downloadAndCacheBlock({
   blockLength,
 }: Props): Promise<Result<void, Error>> {
   if (isAborted(state)) return { data: undefined };
+  if (blockLength <= 0 || blockStart >= virtualFile.size) return { data: undefined };
 
   try {
     const download = await downloadBlockWithRetry({

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 });
+  });
 });

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 };
 }

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);
 

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<number> {
-  if (length <= 0) return [];
-  const first = blockIndexForByte(position);
-  const last = blockIndexForByte(position + length - 1);
+function blocksWithinRange(state: FileHydrationState, { position, length }: ReadRange): Array<number> {
+  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<number> {
 }
 
 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<number, Promise<Result<void, Error>>> {
   const blocksBeingDownloadedWithinRange = new Map<number, Promise<Result<void, Error>>>();
-  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);
   }

src/backend/features/fuse/on-read/handle-read-callback.test.ts

@@ -3,17 +3,16 @@ 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';
 
 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);

src/backend/features/fuse/on-read/handle-read-callback.ts

@@ -1,12 +1,7 @@
 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';
@@ -15,6 +10,35 @@ import { EMPTY } from './constants';
 import { readOrHydrate } from './read-or-hydrate';
 import { type HandleReadDeps, type ReadRange } from './types';
 import { isThumbnailProcess } from './thumbnail-processes';
+
+const PREFETCH_DEFAULT_BLOCKS_AHEAD = 5;
+const PREFETCH_MAX_BLOCKS_AHEAD = 8;
+
+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));
+}
+
+function getThumbnailPrefetchBlocksAhead() {
+  return getPrefetchBlocksAhead({
+    configuredValue: process.env.INTERNXT_DRIVE_THUMBNAIL_PREFETCH_BLOCKS_AHEAD,
+  });
+}
+
+function getReadPrefetchBlocksAhead() {
+  return getPrefetchBlocksAhead({
+    configuredValue: process.env.INTERNXT_DRIVE_READ_PREFETCH_BLOCKS_AHEAD,
+  });
+}
+
 export type HandleReadCallbackProps = HandleReadDeps & {
   findVirtualFile: (path: string) => Promise<File | undefined>;
   findTemporalFile: (path: string) => Promise<TemporalFile | undefined>;
@@ -50,11 +74,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: getThumbnailPrefetchBlocksAhead(),
+    });
   }
 
   const filePath = nodePath.join(PATHS.DOWNLOADED, virtualFile.contentsId);
@@ -68,6 +106,7 @@ export async function handleReadCallback({
     virtualFile,
     filePath,
     range,
+    prefetchBlocksAhead: getReadPrefetchBlocksAhead(),
   });
 }
 
@@ -87,27 +126,3 @@ async function readFromTemporalFile(
   const chunk = await readChunkFromDisk(temporalFile.contentFilePath, length, position);
   return { data: chunk ?? EMPTY };
 }
-
-type ThumbnailRangeProps = Pick<HandleReadCallbackProps, 'bucketId' | 'mnemonic' | 'network' | 'range'> & {
-  virtualFile: File;
-};
-
-async function readExactRangeForThumbnail({
-  bucketId,
-  mnemonic,
-  network,
-  virtualFile,
-  range,
-}: ThumbnailRangeProps): Promise<Result<Buffer, FuseError>> {
-  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 };
-}