feat: Add suspend and resume actions to watchers

  When we started reacting to OS's suspend and resume events, we took
  the simple route of stopping and restarting both the remote and local
  watchers each time.

  However, this is time consuming, especially with the local watcher
  since it does a full scan of the local synchronization directory on
  start, and quite unnecessary since there little change we missed a
  local event while the computer was suspended.

  This is why we now introduce suspend and resume actions which, for the
  local watcher at least, only take care of stopping and restarting
  their event subscriptions.
  This should improve performance when resuming a suspended computer.

  For now the remote watcher still does a complete stop and restart as
  it does not trigger expensive computing.

core/local/channel_watcher/index.js

@@ -165,6 +165,18 @@ class ChannelWatcher {
     await scanDone
   }
 
+  resume() {
+    log.info('Resuming watcher...')
+
+    return this.producer.resume()
+  }
+
+  suspend() {
+    log.info('Suspending watcher...')
+
+    return this.producer.suspend()
+  }
+
   async stop() /*: Promise<*> */ {
     log.info('Stopping watcher...')
 

core/local/channel_watcher/parcel_producer.js

@@ -102,15 +102,7 @@ class Producer {
   async start() {
     log.info('Starting producer...')
 
-    this.watcher = await parcel.subscribe(
-      this.config.syncPath,
-      async (err, events) => {
-        // FIXME: use async queue to run processEvents calls in order
-        await this.processEvents(events)
-      },
-      { backend }
-    )
-    if (!this.watcher) throw new Error('Could not start @parcel/watcher')
+    await this.subscribe()
 
     this.events.emit('buffering-start')
 
@@ -123,6 +115,47 @@ class Producer {
     this.events.emit('buffering-end')
   }
 
+  async resume() {
+    log.info('Resuming producer...')
+
+    await this.subscribe()
+  }
+
+  async suspend() {
+    log.info('Suspending producer...')
+
+    await this.unsubscribe()
+  }
+
+  async stop() {
+    log.info('Stopping producer...')
+
+    await this.unsubscribe()
+  }
+
+  async subscribe() {
+    if (!this.watcher) {
+      this.watcher = await parcel.subscribe(
+        this.config.syncPath,
+        async (err, events) => {
+          // FIXME: use async queue to run processEvents calls in order
+          await this.processEvents(events)
+        },
+        { backend }
+      )
+    }
+    if (!this.watcher) throw new Error('Could not start @parcel/watcher')
+  }
+
+  async unsubscribe() {
+    if (this.watcher) {
+      await this.watcher.unsubscribe()
+      // XXX: unsubscribe() resolves before it was actually finished
+      await Promise.delay(1000)
+      this.watcher = null
+    }
+  }
+
   async scan(relPath /*: string */) {
     const stopParcelScanMeasure = measureTime('Parcel#scan')
     const scanEvents = await parcel.scan(
@@ -187,17 +220,6 @@ class Producer {
       this.channel.push(batch)
     }
   }
-
-  async stop() {
-    log.info('Stopping producer...')
-
-    if (this.watcher) {
-      await this.watcher.unsubscribe()
-      // XXX: unsubscribe() resolves before it was actually finished
-      await Promise.delay(1000)
-      this.watcher = null
-    }
-  }
 }
 
 module.exports = Producer

core/local/chokidar/event_buffer.js

@@ -106,7 +106,7 @@ class EventBuffer /*:: <EventType> */ {
   }
 
   switchMode(mode /*: EventBufferMode */) /*: void */ {
-    this.flush()
+    this.clearTimeout()
     this.mode = mode
   }
 

core/local/chokidar/watcher.js

@@ -190,6 +190,7 @@ class LocalWatcher {
         .on('ready', () => {
           stopChokidarScanMeasure()
           log.info('Folder scan done')
+          this.buffer.flush()
           this.buffer.switchMode('timeout')
         })
         .on('raw', async (event, path, details) => {
@@ -223,6 +224,68 @@ class LocalWatcher {
     return started
   }
 
+  async resume() {
+    log.info('Resuming watcher...')
+
+    if (this.watcher && this.watcher.getWatched().length === 0) {
+      this.watcher.add('.')
+    }
+
+    // Flush previously buffered events
+    this.buffer.flush()
+    // Restart flushes loop
+    this.buffer.switchMode('timeout')
+  }
+
+  async suspend() {
+    log.info('Suspending watcher...')
+
+    // Stop flushes loop but keep buffered events
+    this.buffer.switchMode('idle')
+
+    // Stop underlying Chokidar watcher
+    if (this.watcher) {
+      this.watcher.unwatch('.')
+    }
+  }
+
+  async stop(force /*: ?bool */ = false) {
+    log.info('Stopping watcher...')
+
+    if (!this.watcher) return
+
+    if (force || !this.initialScanParams.flushed) {
+      // Drop buffered events
+      this.buffer.clear()
+    } else {
+      // XXX manually fire events for added file, because chokidar will cancel
+      // them if they are still in the awaitWriteFinish period
+      for (let relpath in this.watcher._pendingWrites) {
+        try {
+          const fullpath = path.join(this.watcher.options.cwd, relpath)
+          const curStat = await stater.stat(fullpath)
+          this.watcher.emit('add', relpath, curStat)
+        } catch (err) {
+          log.warn('Could not fire remaining add events', { err })
+        }
+      }
+    }
+
+    // Stop underlying Chokidar watcher
+    await this.watcher.close()
+    this.watcher = null
+    // Flush buffer and stop flushes loop
+    this.buffer.flush()
+    this.buffer.switchMode('idle')
+
+    if (!force) {
+      // Give some time for awaitWriteFinish events to be managed
+      return new Promise(resolve => {
+        setTimeout(resolve, 1000)
+      })
+    }
+  }
+
   // TODO: Start checksuming as soon as an add/change event is buffered
   // TODO: Put flushed event batches in a queue
   async onFlush(rawEvents /*: ChokidarEvent[] */) {
@@ -283,42 +346,6 @@ class LocalWatcher {
     this.endInitialScan()
   }
 
-  async stop(force /*: ?bool */ = false) {
-    log.info('Stopping watcher...')
-
-    if (!this.watcher) return
-
-    if (force || !this.initialScanParams.flushed) {
-      // Drop buffered events
-      this.buffer.clear()
-    } else {
-      // XXX manually fire events for added file, because chokidar will cancel
-      // them if they are still in the awaitWriteFinish period
-      for (let relpath in this.watcher._pendingWrites) {
-        try {
-          const fullpath = path.join(this.watcher.options.cwd, relpath)
-          const curStat = await stater.stat(fullpath)
-          this.watcher.emit('add', relpath, curStat)
-        } catch (err) {
-          log.warn('Could not fire remaining add events', { err })
-        }
-      }
-    }
-
-    // Stop underlying Chokidar watcher
-    await this.watcher.close()
-    this.watcher = null
-    // Flush buffer and stop flushes loop
-    this.buffer.switchMode('idle')
-
-    if (!force) {
-      // Give some time for awaitWriteFinish events to be managed
-      return new Promise(resolve => {
-        setTimeout(resolve, 1000)
-      })
-    }
-  }
-
   resetInitialScanParams() {
     this.initialScanParams = {
       paths: [],

core/local/index.js

@@ -115,6 +115,17 @@ class Local /*:: implements Reader, Writer */ {
     return this.watcher.start()
   }
 
+  resume() {
+    syncDir.ensureExistsSync(this)
+    this.syncDirCheckInterval = syncDir.startIntervalCheck(this)
+    return this.watcher.resume()
+  }
+
+  suspend() {
+    clearInterval(this.syncDirCheckInterval)
+    return this.watcher.suspend()
+  }
+
   /** Stop watching the file system */
   stop() {
     clearInterval(this.syncDirCheckInterval)

core/local/watcher.js

@@ -26,6 +26,8 @@ import type { Checksumer } from './checksumer'
 export interface Watcher {
   checksumer: Checksumer,
   start (): Promise<*>,
+  resume (): Promise<*>,
+  suspend (): Promise<*>,
   stop (force: ?bool): Promise<*>,
   onFatal (listener: Error => any): void,
   fatal (err: Error): void,

core/remote/index.js

@@ -117,6 +117,15 @@ class Remote /*:: implements Reader, Writer */ {
     return this.warningsPoller.start()
   }
 
+  async resume() {
+    await this.watcher.resume()
+    return this.warningsPoller.start()
+  }
+
+  async suspend() {
+    await Promise.all([this.watcher.suspend(), this.warningsPoller.stop()])
+  }
+
   async stop() {
     await Promise.all([this.watcher.stop(), this.warningsPoller.stop()])
   }

core/remote/watcher/index.js

@@ -110,7 +110,8 @@ class RemoteWatcher {
 
   async start() {
     if (!this.running) {
-      log.info('Starting watcher')
+      log.info('Starting watcher...')
+
       this.running = true
       this.startClock()
 
@@ -127,9 +128,21 @@ class RemoteWatcher {
     }
   }
 
+  resume() {
+    log.info('Resuming watcher...')
+
+    return this.start()
+  }
+
+  suspend() {
+    log.info('Suspending watcher...')
+
+    return this.stop()
+  }
+
   async stop() {
     if (this.running) {
-      log.info('Stopping watcher')
+      log.info('Stopping watcher...')
 
       if (this.realtimeManager) {
         this.realtimeManager.stop()

core/sync/index.js

@@ -298,14 +298,21 @@ class Sync {
     }
   }
 
-  async started() {
-    await this.lifecycle.started()
-  }
+  async resume() {
+    log.info('resuming synchronization')
 
-  // Manually force a full synchronization
-  async forceSync() {
-    await this.stop()
-    await this.start()
+    try {
+      this.lifecycle.begin('start')
+    } catch (err) {
+      return
+    }
+
+    this.events.once('power-suspend', this.suspend)
+
+    this.lifecycle.unblockFor('all')
+    this.local.resume()
+    this.remote.resume()
+    this.lifecycle.end('start')
   }
 
   suspend() {
@@ -319,19 +326,14 @@ class Sync {
       return
     }
 
-    this.local.stop()
-    this.remote.stop()
+    this.local.suspend()
+    this.remote.suspend()
     clearInterval(this.retryInterval)
     this.retryInterval = null
     this.lifecycle.unblockFor('all')
     this.lifecycle.end('stop')
   }
 
-  async resume() {
-    log.info('resuming synchronization')
-    await this.start()
-  }
-
   // Stop the synchronization
   async stop() /*: Promise<void> */ {
     // In case an interval timer was started, we clear it to make sure it won't
@@ -361,10 +363,20 @@ class Sync {
     this.lifecycle.end('stop')
   }
 
+  async started() {
+    await this.lifecycle.started()
+  }
+
   async stopped() {
     await this.lifecycle.stopped()
   }
 
+  // Manually force a full synchronization
+  async forceSync() {
+    await this.stop()
+    await this.start()
+  }
+
   fatal(err /*: Error */) {
     log.fatal(`Sync fatal: ${err.message}`, { err, sentry: true })