fix: handle replica lag in WebDAV server

src/services/drive/drive-item.service.ts

@@ -1,14 +1,86 @@
 import { DriveItemRepository } from '../database/drive-item/drive-item.repository';
 import { DriveFileItem, DriveFolderItem } from '../../types/drive.types';
-import { ErrorUtils } from '../../utils/errors.utils';
+import { ErrorUtils, NotFoundError } from '../../utils/errors.utils';
 import { webdavLogger } from '../../utils/logger.utils';
 import { DriveFileService } from './drive-file.service';
 import { DriveFolderService } from './drive-folder.service';
 import { DriveItemBD } from '../database/drive-item/drive-item.domain';
 
+const RECENT_WRITE_TRUST_MS = 10_000;
+const RECENT_CHANGE_TRUST_MS = 60_000;
+
+const normalizePath = (path: string): string => (path.length > 1 && path.endsWith('/') ? path.slice(0, -1) : path);
+
+/** A not found that comes from our own recent change, not from an API failure. */
+class StaleItemError extends NotFoundError {
+  constructor(message: string) {
+    super(message);
+    this.name = 'StaleItemError';
+    Object.setPrototypeOf(this, StaleItemError.prototype);
+  }
+}
+
 export class DriveItemService {
   static readonly instance = new DriveItemService();
 
+  private readonly recentFileWrites = new Map<string, { item: DriveFileItem; expiresAt: number }>();
+  /** Where this server last put an item, keyed by uuid; a null path means it was deleted. */
+  private readonly recentChanges = new Map<string, { path: string | null; expiresAt: number }>();
+
+  private readonly rememberChange = (uuid: string, path: string | null): void => {
+    const now = Date.now();
+    for (const [changedUuid, change] of this.recentChanges) {
+      if (change.expiresAt <= now) this.recentChanges.delete(changedUuid);
+    }
+    this.recentChanges.set(uuid, { path, expiresAt: now + RECENT_CHANGE_TRUST_MS });
+  };
+
+  public rememberDeletion = (uuid: string): void => this.rememberChange(uuid, null);
+
+  public rememberMove = (uuid: string, path: string): void => this.rememberChange(uuid, normalizePath(path));
+
+  /** The replicas can keep returning an item from where it was before this server moved or
+   * deleted it, and acting on that is destructive, eg replacing a file that is now in the trash. */
+  public isStale = (uuid: string, path: string): boolean => {
+    const change = this.recentChanges.get(uuid);
+    if (!change || change.expiresAt <= Date.now()) return false;
+    return change.path !== normalizePath(path);
+  };
+
+  private readonly throwIfStale = (uuid: string, path: string, itemType: 'file' | 'folder'): void => {
+    if (!this.isStale(uuid, path)) return;
+    webdavLogger.info(`[LOOKUP] Ignoring '${path}', it was moved or deleted moments ago`, { uuid });
+    throw new StaleItemError(`${itemType === 'file' ? 'File' : 'Folder'} not found at path: ${path}`);
+  };
+
+  /** The API reads from replicas that can lag a moment behind a write, so a file this server
+   * just created can briefly look missing; remember it so that 404 is not trusted. */
+  public rememberFileWrite = (path: string, item: DriveFileItem): void => {
+    const now = Date.now();
+    for (const [writtenPath, write] of this.recentFileWrites) {
+      if (write.expiresAt <= now) this.recentFileWrites.delete(writtenPath);
+    }
+    this.recentFileWrites.set(path, { item, expiresAt: now + RECENT_WRITE_TRUST_MS });
+  };
+
+  /** Only while the path cache still maps to the written uuid: our own DELETE and MOVE clear it. */
+  private readonly getRecentFileWrite = (
+    cached: DriveItemBD | undefined,
+    path: string,
+    error: unknown,
+  ): DriveFileItem | undefined => {
+    if (!cached || !ErrorUtils.isNotFoundError(error)) return;
+
+    const write = this.recentFileWrites.get(path);
+    if (!write || write.expiresAt <= Date.now() || write.item.uuid !== cached.uuid) return;
+    if (this.isStale(write.item.uuid, path)) return;
+
+    webdavLogger.info(`[LOOKUP] '${path}' was written moments ago, ignoring the API's not found`, {
+      uuid: cached.uuid,
+    });
+    return write.item;
+  };
+
   /** Only drop the cached uuid when the API confirms the item is gone; evicting on a timeout
    * pushes every later request onto the slow path lookup. */
   private readonly dropCacheIfGone = async (uuid: string, error: unknown): Promise<void> => {
@@ -19,6 +91,7 @@ export class DriveItemService {
   private readonly tryGetFileByUuid = async (cached: DriveItemBD, path: string): Promise<DriveFileItem | undefined> => {
     try {
       const item = await DriveFileService.instance.getFileMetadata(cached.uuid);
+      this.throwIfStale(item.uuid, path, 'file');
       await DriveItemRepository.instance.createOrUpdate([
         {
           uuid: cached.uuid,
@@ -30,10 +103,15 @@ export class DriveItemService {
       ]);
       return item;
     } catch (error) {
-      webdavLogger.warn(ErrorUtils.withRequestId('File metadata by uuid failed, falling back to path lookup', error), {
-        path,
-        uuid: cached.uuid,
-      });
+      const recentWrite = this.getRecentFileWrite(cached, path, error);
+      if (recentWrite) return recentWrite;
+
+      if (!(error instanceof StaleItemError)) {
+        webdavLogger.warn(
+          ErrorUtils.withRequestId('File metadata by uuid failed, falling back to path lookup', error),
+          { path, uuid: cached.uuid },
+        );
+      }
       await this.dropCacheIfGone(cached.uuid, error);
     }
   };
@@ -44,6 +122,7 @@ export class DriveItemService {
   ): Promise<DriveFolderItem | undefined> => {
     try {
       const item = await DriveFolderService.instance.getFolderMetaByUuid(cached.uuid);
+      this.throwIfStale(item.uuid, path, 'folder');
       await DriveItemRepository.instance.createOrUpdate([
         {
           uuid: cached.uuid,
@@ -55,10 +134,12 @@ export class DriveItemService {
       ]);
       return item;
     } catch (error) {
-      webdavLogger.warn(
-        ErrorUtils.withRequestId('Folder metadata by uuid failed, falling back to path lookup', error),
-        { path, uuid: cached.uuid },
-      );
+      if (!(error instanceof StaleItemError)) {
+        webdavLogger.warn(
+          ErrorUtils.withRequestId('Folder metadata by uuid failed, falling back to path lookup', error),
+          { path, uuid: cached.uuid },
+        );
+      }
       await this.dropCacheIfGone(cached.uuid, error);
     }
   };
@@ -73,6 +154,7 @@ export class DriveItemService {
 
     try {
       const item = await DriveFileService.instance.getFileMetadataByPath(path);
+      this.throwIfStale(item.uuid, path, 'file');
       await DriveItemRepository.instance.createOrUpdate([
         {
           uuid: item.uuid,
@@ -84,6 +166,9 @@ export class DriveItemService {
       ]);
       return item;
     } catch (error) {
+      const recentWrite = this.getRecentFileWrite(cached, path, error);
+      if (recentWrite) return recentWrite;
+
       ErrorUtils.logIfUnexpected(webdavLogger, 'File lookup by path failed', error, { path });
       throw ErrorUtils.toLookupError('file', path, error);
     }
@@ -99,6 +184,7 @@ export class DriveItemService {
 
     try {
       const item = await DriveFolderService.instance.getFolderMetaByPath(path);
+      this.throwIfStale(item.uuid, path, 'folder');
       await DriveItemRepository.instance.createOrUpdate([
         {
           uuid: item.uuid,

src/utils/webdav.utils.ts

@@ -118,6 +118,7 @@ export class WebDavUtils {
       });
       webdavLogger.info(`[DELETE] [${driveItem.uuid}] ${type} trashed successfully`);
     }
+    DriveItemService.instance.rememberDeletion(driveItem.uuid);
     await DriveItemRepository.instance.delete([driveItem.uuid]);
   }
 

src/webdav/handlers/MOVE.handler.ts

@@ -2,6 +2,7 @@ import { Request, Response } from 'express';
 import { DriveFileService } from '../../services/drive/drive-file.service';
 import { DriveFolderService } from '../../services/drive/drive-folder.service';
 import { DriveItemRepository } from '../../services/database/drive-item/drive-item.repository';
+import { DriveItemService } from '../../services/drive/drive-item.service';
 import { WebDavMethodHandler } from '../../types/webdav.types';
 import { NotFoundError, PreconditionFailedError } from '../../utils/errors.utils';
 import { webdavLogger } from '../../utils/logger.utils';
@@ -96,6 +97,7 @@ export class MOVERequestHandler implements WebDavMethodHandler {
       }
     }
 
+    DriveItemService.instance.rememberMove(originalDriveItem.uuid, destinationResource.url);
     await DriveItemRepository.instance.delete([originalDriveItem.uuid]);
     await DriveItemRepository.instance.createOrUpdate([
       {

src/webdav/handlers/PROPFIND.handler.ts

@@ -4,6 +4,7 @@ import { DriveFileItem, DriveFolderItem } from '../../types/drive.types';
 import { DriveItemBD } from '../../services/database/drive-item/drive-item.domain';
 import { DriveItemRepository } from '../../services/database/drive-item/drive-item.repository';
 import { DriveFolderService } from '../../services/drive/drive-folder.service';
+import { DriveItemService } from '../../services/drive/drive-item.service';
 import { DriveUtils } from '../../utils/drive.utils';
 import { FormatUtils } from '../../utils/format.utils';
 import { Request, Response } from 'express';
@@ -99,6 +100,7 @@ export class PROPFINDRequestHandler implements WebDavMethodHandler {
 
     for (const folder of folderContent.folders) {
       const folderRelativePath = WebDavUtils.joinURL(relativePath, folder.plainName, '/');
+      if (DriveItemService.instance.isStale(folder.uuid, folderRelativePath)) continue;
 
       xmlNodes.push(
         this.driveFolderItemToXMLNode(
@@ -134,6 +136,7 @@ export class PROPFINDRequestHandler implements WebDavMethodHandler {
         relativePath,
         file.type ? `${file.plainName}.${file.type}` : file.plainName,
       );
+      if (DriveItemService.instance.isStale(file.uuid, fileRelativePath)) continue;
 
       xmlNodes.push(
         this.driveFileItemToXMLNode(

src/webdav/handlers/PUT.handler.ts

@@ -1,6 +1,7 @@
 import { Request, Response } from 'express';
 import { DriveFileService } from '../../services/drive/drive-file.service';
 import { DriveItemRepository } from '../../services/database/drive-item/drive-item.repository';
+import { DriveItemService } from '../../services/drive/drive-item.service';
 import { AuthService } from '../../services/auth.service';
 import { WebDavMethodHandler } from '../../types/webdav.types';
 import { ConflictError, ErrorUtils } from '../../utils/errors.utils';
@@ -12,9 +13,14 @@
 import { ThumbnailService } from '../../services/thumbnail.service';
 import { FormatUtils } from '../../utils/format.utils';
 import { UploadUtils } from '../../utils/upload.utils';
+import { AsyncUtils } from '../../utils/async.utils';
+import { DriveFileItem } from '../../types/drive.types';
+import { StorageTypes } from '@internxt/sdk/dist/drive';
+
+const STALE_READ_RETRY_DELAYS_MS = [0, 500, 1000];
 
 export class PUTRequestHandler implements WebDavMethodHandler {
   handle = async (req: Request, res: Response) => {
     let contentLength = Number(req.headers['content-length']);
     if (!contentLength || Number.isNaN(contentLength) || contentLength <= 0) {
       contentLength = 0;
@@ -110,22 +116,26 @@
 
     let file;
     if (driveFileItem?.itemType === 'file') {
+      file = await this.replaceExistingFile(driveFileItem, filePayload, resource.url);
+    } else {
       try {
-        file = await DriveFileService.instance.replaceFile(driveFileItem.uuid, filePayload);
+        file = await DriveFileService.instance.createFile(filePayload);
       } catch (error) {
-        webdavLogger.warn(
-          ErrorUtils.withRequestId(
-            `[PUT] File replace failed for '${resource.url}', falling back to delete and create: ${
-              error instanceof Error ? error.message : String(error)
-            }`,
-            error,
-          ),
+        if (!ErrorUtils.isAlreadyExistsError(error)) throw error;
+
+        const existingFile = await this.findUnseenExistingFile(
+          filePayload.folderUuid,
+          { plainName: filePayload.plainName, type: fileType },
+          resource.url,
         );
-        await WebDavUtils.deleteOrTrashItem(driveFileItem);
-        file = await DriveFileService.instance.createFile(filePayload);
+        if (!existingFile) throw error;
+
+        webdavLogger.info(
+          `[PUT] File '${resource.url}' already exists on Drive but the lookup missed it, it will be replaced...`,
+        );
+        isReplacement = true;
+        file = await this.replaceExistingFile(existingFile, filePayload, resource.url);
       }
-    } else {
-      file = await DriveFileService.instance.createFile(filePayload);
     }
     timings.driveUpload = driveTimer.stop();
 
@@ -138,6 +148,7 @@
         updatedAt: new Date(),
       },
     ]);
+    DriveItemService.instance.rememberFileWrite(resource.url, file);
 
     const thumbnailTimer = CLIUtils.timer();
     await ThumbnailService.instance.tryUploadThumbnail({
@@ -167,4 +178,56 @@
     res.header('ETag', WebDavUtils.getItemETag(file));
     res.status(statusCode).send();
   };
+
+  private readonly replaceExistingFile = async (
+    existingFile: DriveFileItem,
+    filePayload: StorageTypes.FileEntryByUuid,
+    url: string,
+  ): Promise<DriveFileItem> => {
+    try {
+      return await DriveFileService.instance.replaceFile(existingFile.uuid, filePayload);
+    } catch (error) {
+      webdavLogger.warn(
+        ErrorUtils.withRequestId(
+          `[PUT] File replace failed for '${url}', falling back to delete and create: ${
+            error instanceof Error ? error.message : String(error)
+          }`,
+          error,
+        ),
+      );
+      await WebDavUtils.deleteOrTrashItem(existingFile);
+      return await this.createFileAfterTrashing(filePayload);
+    }
+  };
+
+  /** The create's 409 check reads a replica that can still see the file we just trashed. */
+  private readonly createFileAfterTrashing = async (
+    filePayload: StorageTypes.FileEntryByUuid,
+  ): Promise<DriveFileItem> => {
+    let lastError: unknown;
+    for (const delayMs of STALE_READ_RETRY_DELAYS_MS) {
+      await AsyncUtils.sleep(delayMs);
+      try {
+        return await DriveFileService.instance.createFile(filePayload);
+      } catch (error) {
+        if (!ErrorUtils.isAlreadyExistsError(error)) throw error;
+        lastError = error;
+      }
+    }
+    throw lastError;
+  };
+
+  /** A 409 on create means the file exists even though the path lookup missed it, usually
+   * because the read replica has not caught up with a file created moments ago. */
+  private readonly findUnseenExistingFile = async (
+    folderUuid: string,
+    file: StorageTypes.FileStructure,
+    url: string,
+  ): Promise<DriveFileItem | undefined> => {
+    for (const delayMs of STALE_READ_RETRY_DELAYS_MS) {
+      await AsyncUtils.sleep(delayMs);
+      const existingFile = await DriveFileService.instance.findExistentFile(folderUuid, file);
+      if (existingFile && !DriveItemService.instance.isStale(existingFile.uuid, url)) return existingFile;
+    }
+  };
 }