Merge pull request #355 from brave/compaction-callback

client/constants/messages.js

@@ -104,6 +104,11 @@ const messages = {
    * browser sends this to compact records in a category
    */
   COMPACT_SYNC_CATEGORY: _, /* @param {string} categoryName */
+  /**
+   * webview -> browser
+   * sent after compaction is competed.
+   */
+  COMPACTED_SYNC_CATEGORY: _, /* @param {string} categoryName */
   /**
    * webview -> browser
    * webview sends this to delete all site settings.

client/requestUtil.js

@@ -184,6 +184,8 @@ RequestUtil.prototype.parseAWSResponse = function (bytes) {
  * @param {number=} maxRecords Limit response to a given number of recods. By default the Sync lib will fetch all matching records, which might take a long time. If falsey, fetch all records.
  * @param {{
  *    compaction {boolean} // compact records while list object from S3
+ *    compactionDoneCb {function} // callback when compaction is done
+ *    compactionUpdateCb {function} // callback for client to update missing local records
  *  }} opts
  * @returns {Promise(Array.<Object>)}
  */
@@ -206,13 +208,20 @@ RequestUtil.prototype.list = function (category, startAt, maxRecords, nextContin
       }
       return new Promise((resolve, reject) => {
         s3ObjectsPromise.then((s3Objects) => {
+          const compactedRecords = this.compactObjects(s3Objects.contents)
+          if (opts.compactionUpdateCb) {
+            opts.compactionUpdateCb(compactedRecords)
+          }
+          // wait for 15 seconds between batches
           setTimeout(() => {
-            this.compactObjects(s3Objects.contents)
             if (s3Objects.isTruncated) {
               return this.list(category, startAt, maxRecords, s3Objects.nextContinuationToken, opts)
             }
             return new Promise((resolve, reject) => {
               // compaction is done
+              if (opts.compactionDoneCb) {
+                opts.compactionDoneCb()
+              }
               resolve()
             })
           }, 15000)
@@ -463,9 +472,11 @@ RequestUtil.prototype.put = function (category, record) {
 /**
  * Compact all records in a category
  * @param {string=} category - the category ID
+ * @returns {Array.<Object>} - records after compaction
  */
 RequestUtil.prototype.compactObjects = function (s3Objects) {
   let s3ObjectsToDelete = []
+  const compactedRecords = {}
   const recordObjects = this.s3ObjectsToRecords(s3Objects)
   recordObjects.forEach((recordObject) => {
     const record = recordObject.record
@@ -477,15 +488,18 @@ RequestUtil.prototype.compactObjects = function (s3Objects) {
         s3ObjectsToDelete = s3ObjectsToDelete.concat(cacheRecordObject.objects)
         console.log(cacheRecordObject.record)
         this.latestRecordsCache.set(id, recordObject)
+        compactedRecords[id] = record
       } else {
         s3ObjectsToDelete = s3ObjectsToDelete.concat(recordObject.objects)
         console.log(record)
       }
     } else {
       this.latestRecordsCache.set(id, recordObject)
+      compactedRecords[id] = record
     }
   })
   s3Helper.deleteObjects(this.s3, this.bucket, s3ObjectsToDelete)
+  return Object.values(compactedRecords)
 }
 
 RequestUtil.prototype.s3PostFormData = function (objectKey) {

client/sync.js

@@ -23,6 +23,7 @@ var clientKeys = {}
 var config = {}
 var seed
 var nextContinuationTokens = {}
+var isCompactionInProgress = false
 
 /**
  * Logs stuff on the visible HTML page.
@@ -190,9 +191,27 @@ const startSync = (requester) => {
     if (!proto.categories[category]) {
       throw new Error(`Unsupported sync category: ${category}`)
     }
-    requester.list(proto.categories[category], 0, 1000, '', {compaction: true}).then(() => {
-      logSync(`Compacting category: ${category}`)
-    })
+    const compactionDone = () => {
+      ipc.send(messages.COMPACTED_SYNC_CATEGORY, category)
+      isCompactionInProgress = false
+    }
+    const compactionUpdate = (records) => {
+      let jsRecords = []
+      for (let record of records) {
+        const jsRecord = recordUtil.syncRecordAsJS(record)
+        jsRecord.syncTimestamp = record.syncTimestamp
+        jsRecords.push(jsRecord)
+      }
+      logSync(`Compaction records update category: ${category}`)
+      ipc.send(messages.GET_EXISTING_OBJECTS, category, jsRecords, 0, false)
+    }
+    if (!isCompactionInProgress) {
+      requester.list(proto.categories[category], 0, 1000, '',
+        {compaction: true, compactionDoneCb: compactionDone, compactionUpdateCb: compactionUpdate}).then(() => {
+        logSync(`Compacting category: ${category}`)
+        isCompactionInProgress = true
+      })
+    }
   })
   ipc.on(messages.DELETE_SYNC_SITE_SETTINGS, (e) => {
     logSync(`Deleting siteSettings`)

test/client/requestUtil.js

@@ -404,7 +404,7 @@ test('client RequestUtil', (t) => {
 
     const testCanDoCompaction = (t) => {
       t.test('#compact bookmarks', (t) => {
-        t.plan(3)
+        t.plan(7)
         const recordObjectId = testHelper.newUuid()
         const record2ObjectId = testHelper.newUuid()
         const record = {
@@ -452,31 +452,47 @@ test('client RequestUtil', (t) => {
           requestUtil.put(proto.categories.BOOKMARKS, record2_update)
         }
         const consoleLogBak = console.log
+        let counter = 0
+        let compactedRecord = []
+        const compactionUpdate = (records) => {
+          compactedRecord = compactedRecord.concat(records)
+          counter = counter + 1
+        }
         // limit batch size to 10 to test cross batch compaction for around 40
         // objects
-        requestUtil.list(proto.categories.BOOKMARKS, 0, 10, '', {compaction: true}).then(() => {
+        requestUtil.list(proto.categories.BOOKMARKS, 0, 10, '', {compaction: true,
+          compactionUpdateCb: compactionUpdate,
+          compactionDoneCb: () => {
+            console.log = consoleLogBak
+            console.log('compaction is done')
+            t.equals(counter, 5)
+            t.equals(compactedRecord.length, 9)
+            t.equals(
+              compactedRecord.filter(record => record.objectId.toString() === record_update.objectId.toString()).length, 4)
+            t.equals(
+              compactedRecord.filter(record => record.objectId.toString() === record2_update.objectId.toString()).length, 5)
+            //  we already have 15 second timeout for each batch so no need to
+            //  do another wait after compaction is done
+            requestUtil.list(proto.categories.BOOKMARKS, 0, 0)
+              .then(s3Objects => requestUtil.s3ObjectsToRecords(s3Objects.contents))
+              .then((response) => {
+                t.equals(response.length, 2, `${t.name} check records number`)
+                const s3Record = response[0].record
+                const s3Record2 = response[1].record
+                t.deepEquals(s3Record.objectId, record.objectId, `${t.name}: objectId`)
+                t.deepEquals(s3Record.bookmark.site.title, record_update.bookmark.site.title, `${t.name}: bookmark.title`)
+                requestUtil.deleteCategory(proto.categories.BOOKMARKS)
+                  .then((_response) => {
+                    testCanLimitResponse(t)
+                  })
+                  .catch((error) => { t.fail(error) })
+              })
+              .catch((error) => { t.fail(error) })
+          }
+        }).then(() => {
+          // compaction is still in progress
           console.log = function() {}
         })
-        // takes about 1.5 minute for delete to be completely done and we also
-        // have 15 second timeout for each batch
-        setTimeout(() => {
-          console.log = consoleLogBak
-          requestUtil.list(proto.categories.BOOKMARKS, 0, 0)
-            .then(s3Objects => requestUtil.s3ObjectsToRecords(s3Objects.contents))
-            .then((response) => {
-              t.equals(response.length, 2, `${t.name} check records number`)
-              const s3Record = response[0].record
-              const s3Record2 = response[1].record
-              t.deepEquals(s3Record.objectId, record.objectId, `${t.name}: objectId`)
-              t.deepEquals(s3Record.bookmark.site.title, record_update.bookmark.site.title, `${t.name}: bookmark.title`)
-              requestUtil.deleteCategory(proto.categories.BOOKMARKS)
-                .then((_response) => {
-                  testCanLimitResponse(t)
-                })
-                .catch((error) => { t.fail(error) })
-            })
-            .catch((error) => { t.fail(error) })
-        }, 90000)
       })
     }