Fix:Watcher waits for files to finish transferring before scanning #1362

 #2248

server/Watcher.js

@@ -6,7 +6,7 @@ const LibraryScanner = require('./scanner/LibraryScanner')
 const Task = require('./objects/Task')
 const TaskManager = require('./managers/TaskManager')
 
-const { filePathToPOSIX, isSameOrSubPath } = require('./utils/fileUtils')
+const { filePathToPOSIX, isSameOrSubPath, getFileMTimeMs } = require('./utils/fileUtils')
 
 /**
  * @typedef PendingFileUpdate
@@ -29,6 +29,8 @@ class FolderWatcher extends EventEmitter {
     /** @type {Task} */
     this.pendingTask = null
 
+    this.filesBeingAdded = new Set()
+
     /** @type {string[]} */
     this.ignoreDirs = []
     /** @type {string[]} */
@@ -64,14 +66,13 @@ class FolderWatcher extends EventEmitter {
     })
     watcher
       .on('add', (path) => {
-        this.onNewFile(library.id, path)
+        this.onFileAdded(library.id, filePathToPOSIX(path))
       }).on('change', (path) => {
         // This is triggered from metadata changes, not what we want
-        // this.onFileUpdated(path)
       }).on('unlink', path => {
-        this.onFileRemoved(library.id, path)
+        this.onFileRemoved(library.id, filePathToPOSIX(path))
       }).on('rename', (path, pathNext) => {
-        this.onRename(library.id, path, pathNext)
+        this.onFileRename(library.id, filePathToPOSIX(path), filePathToPOSIX(pathNext))
       }).on('error', (error) => {
         Logger.error(`[Watcher] ${error}`)
       }).on('ready', () => {
@@ -137,14 +138,31 @@ class FolderWatcher extends EventEmitter {
     return this.libraryWatchers.map(lib => lib.watcher.close())
   }
 
-  onNewFile(libraryId, path) {
+  /**
+   * Watcher detected file added
+   * 
+   * @param {string} libraryId 
+   * @param {string} path 
+   */
+  onFileAdded(libraryId, path) {
     if (this.checkShouldIgnorePath(path)) {
       return
     }
     Logger.debug('[Watcher] File Added', path)
     this.addFileUpdate(libraryId, path, 'added')
+
+    if (!this.filesBeingAdded.has(path)) {
+      this.filesBeingAdded.add(path)
+      this.waitForFileToAdd(path)
+    }
   }
 
+  /**
+   * Watcher detected file removed
+   * 
+   * @param {string} libraryId 
+   * @param {string} path 
+   */
   onFileRemoved(libraryId, path) {
     if (this.checkShouldIgnorePath(path)) {
       return
@@ -153,11 +171,13 @@ class FolderWatcher extends EventEmitter {
     this.addFileUpdate(libraryId, path, 'deleted')
   }
 
-  onFileUpdated(path) {
-    Logger.debug('[Watcher] Updated File', path)
-  }
-
-  onRename(libraryId, pathFrom, pathTo) {
+  /**
+   * Watcher detected file renamed
+   * 
+   * @param {string} libraryId 
+   * @param {string} path 
+   */
+  onFileRename(libraryId, pathFrom, pathTo) {
     if (this.checkShouldIgnorePath(pathTo)) {
       return
     }
@@ -166,13 +186,41 @@ class FolderWatcher extends EventEmitter {
   }
 
   /**
-   * File update detected from watcher
+   * Get mtimeMs from an added file every second until it is no longer changing
+   * Times out after 180s
+   * 
+   * @param {string} path 
+   * @param {number} [lastMTimeMs=0] 
+   * @param {number} [loop=0] 
+   */
+  async waitForFileToAdd(path, lastMTimeMs = 0, loop = 0) {
+    // Safety to catch infinite loop (180s)
+    if (loop >= 180) {
+      Logger.warn(`[Watcher] Waiting to add file at "${path}" timeout (loop ${loop}) - proceeding`)
+      return this.filesBeingAdded.delete(path)
+    }
+
+    const mtimeMs = await getFileMTimeMs(path)
+    if (mtimeMs === lastMTimeMs) {
+      if (lastMTimeMs) Logger.debug(`[Watcher] File finished adding at "${path}"`)
+      return this.filesBeingAdded.delete(path)
+    }
+    if (lastMTimeMs % 5 === 0) {
+      Logger.debug(`[Watcher] Waiting to add file at "${path}". mtimeMs=${mtimeMs} lastMTimeMs=${lastMTimeMs} (loop ${loop})`)
+    }
+    // Wait 1 second
+    await new Promise((resolve) => setTimeout(resolve, 1000))
+    this.waitForFileToAdd(path, mtimeMs, ++loop)
+  }
+
+  /**
+   * Queue file update
+   * 
    * @param {string} libraryId 
    * @param {string} path 
    * @param {string} type 
    */
   addFileUpdate(libraryId, path, type) {
-    path = filePathToPOSIX(path)
     if (this.pendingFilePaths.includes(path)) return
 
     // Get file library
@@ -222,12 +270,26 @@ class FolderWatcher extends EventEmitter {
       type
     })
 
-    // Notify server of update after "pendingDelay"
+    this.handlePendingFileUpdatesTimeout()
+  }
+
+  /**
+   * Wait X seconds before notifying scanner that files changed
+   * reset timer if files are still copying
+   */
+  handlePendingFileUpdatesTimeout() {
     clearTimeout(this.pendingTimeout)
     this.pendingTimeout = setTimeout(() => {
+      // Check that files are not still being added
+      if (this.pendingFileUpdates.some(pfu => this.filesBeingAdded.has(pfu.path))) {
+        Logger.debug(`[Watcher] Still waiting for pending files "${[...this.filesBeingAdded].join(', ')}"`)
+        return this.handlePendingFileUpdatesTimeout()
+      }
+
       LibraryScanner.scanFilesChanged(this.pendingFileUpdates, this.pendingTask)
       this.pendingTask = null
       this.pendingFileUpdates = []
+      this.filesBeingAdded.clear()
     }, this.pendingDelay)
   }
 

server/utils/fileUtils.js

@@ -38,22 +38,14 @@ function isSameOrSubPath(parentPath, childPath) {
 }
 module.exports.isSameOrSubPath = isSameOrSubPath
 
-async function getFileStat(path) {
+function getFileStat(path) {
   try {
-    var stat = await fs.stat(path)
-    return {
-      size: stat.size,
-      atime: stat.atime,
-      mtime: stat.mtime,
-      ctime: stat.ctime,
-      birthtime: stat.birthtime
-    }
+    return fs.stat(path)
   } catch (err) {
     Logger.error('[fileUtils] Failed to stat', err)
-    return false
+    return null
   }
 }
-module.exports.getFileStat = getFileStat
 
 async function getFileTimestampsWithIno(path) {
   try {
@@ -72,12 +64,25 @@ async function getFileTimestampsWithIno(path) {
 }
 module.exports.getFileTimestampsWithIno = getFileTimestampsWithIno
 
-async function getFileSize(path) {
-  var stat = await getFileStat(path)
-  if (!stat) return 0
-  return stat.size || 0
+/**
+ * Get file size
+ * 
+ * @param {string} path 
+ * @returns {Promise<number>}
+ */
+module.exports.getFileSize = async (path) => {
+  return (await getFileStat(path))?.size || 0
+}
+
+/**
+ * Get file mtimeMs
+ * 
+ * @param {string} path 
+ * @returns {Promise<number>} epoch timestamp
+ */
+module.exports.getFileMTimeMs = async (path) => {
+  return (await getFileStat(path))?.mtimeMs || 0
 }
-module.exports.getFileSize = getFileSize
 
 /**
  *