previousFetchTime parameter for FETCH_SYNC_RECORDS

client/constants/messages.js

@@ -57,7 +57,7 @@ const messages = {
    * with new records, do
    * GET_EXISTING_OBJECTS -> RESOLVE_SYNC_RECORDS -> RESOLVED_SYNC_RECORDS
    */
-  FETCH_SYNC_RECORDS: _, /* @param Array.<string> categoryNames, @param {number} startAt (in seconds or milliseconds), @param {number=} maxRecords limit response to a given max number of records. set to 0 or falsey to not limit the response */
+  FETCH_SYNC_RECORDS: _, /* @param Array.<string> categoryNames, @param {number} startAt (in seconds or milliseconds), @param {number=} maxRecords limit response to a given max number of records. set to 0 or falsey to not limit the response, @param {number} previousFetchTime (in milliseconds) */
   /**
    * browser -> webview
    * sent to fetch all sync devices. webview responds with RESOLVED_SYNC_RECORDS.

client/requestUtil.js

@@ -211,7 +211,7 @@ RequestUtil.prototype.parseAWSResponse = function (bytes) {
  *  }} opts
  * @returns {Promise(Array.<Object>)}
  */
-RequestUtil.prototype.list = function (category, startAt, maxRecords, nextContinuationToken, opts = {}) {
+RequestUtil.prototype.list = function (category, startAt, maxRecords, nextContinuationToken, previousFetchTime, opts = {}) {
   const prefix = `${this.apiVersion}/${this.userId}/${category}`
   let options = {
     MaxKeys: maxRecords || 1000,
@@ -223,7 +223,7 @@ RequestUtil.prototype.list = function (category, startAt, maxRecords, nextContin
   }
   if (startAt) { options.StartAfter = `${prefix}/${startAt}` }
   return this.withRetry(() => {
-    if (this.shouldListObject(startAt, category) || opts.compaction) {
+    if (this.shouldListObject(previousFetchTime, category) || opts.compaction) {
       const s3ObjectsPromise = s3Helper.listObjects(this.s3, options, !!maxRecords)
       if (!opts.compaction) {
         return s3ObjectsPromise
@@ -237,7 +237,7 @@ RequestUtil.prototype.list = function (category, startAt, maxRecords, nextContin
           // wait for 15 seconds between batches
           setTimeout(() => {
             if (s3Objects.isTruncated) {
-              return this.list(category, startAt, maxRecords, s3Objects.nextContinuationToken, opts)
+              return this.list(category, startAt, maxRecords, s3Objects.nextContinuationToken, previousFetchTime, opts)
             }
             return new Promise((resolve, reject) => {
               // compaction is done
@@ -305,16 +305,14 @@ const CATEGORIES_FOR_SQS = [proto.categories.BOOKMARKS, proto.categories.PREFERE
 
 /**
  * Checks do we need to use s3 list Object or SQS notifications
- * @param {number=} startAt return objects with timestamp >= startAt (e.g. 1482435340). Could be seconds or milliseconds
+ * @param {number=} previousFetchTime - the previous fetch time. Could be seconds or milliseconds
  * @param {string} category - the category ID
  * @returns {boolean}
 */
-RequestUtil.prototype.shouldListObject = function (startAt, category) {
+RequestUtil.prototype.shouldListObject = function (previousFetchTime, category) {
   let currentTime = new Date().getTime()
-  let startAtToCheck = this.normalizeTimestampToMs(startAt, currentTime)
-
-  return !startAtToCheck ||
-      (currentTime - startAtToCheck) > parseInt(s3Helper.SQS_RETENTION, 10) * 1000 ||
+  return !previousFetchTime ||
+      (currentTime - previousFetchTime) > parseInt(s3Helper.SQS_RETENTION, 10) * 1000 ||
       !CATEGORIES_FOR_SQS.includes(category) ||
       this.listInProgress === true
 }
@@ -336,17 +334,20 @@ RequestUtil.prototype.shouldRetireOldSQSQueue = function (createdTimestamp) {
 
 /**
  * Checks do we need to use s3 list Object or SQS notifications
- * @param {number=} startAt return objects with timestamp >= startAt (e.g. 1482435340). Could be seconds or milliseconds
+ * @param {number=} timeToNormalize could be seconds or milliseconds
  * @param {number=} currentTime currentTime in milliseconds
- * @returns {number=}
+ * @returns {number=} the time for sure in milliseconds
 */
-RequestUtil.prototype.normalizeTimestampToMs = function (startAt, currentTime) {
-  let startAtToCheck = startAt
-  if (startAtToCheck && currentTime.toString().length - startAtToCheck.toString().length >= 3) {
-    startAtToCheck *= 1000
+RequestUtil.prototype.normalizeTimestampToMs = function (timeToNormalize, currentTime) {
+  if (!timeToNormalize) {
+    return 0
+  }
+
+  if (timeToNormalize && currentTime.toString().length - timeToNormalize.toString().length >= 3) {
+    timeToNormalize *= 1000
   }
 
-  return startAtToCheck
+  return timeToNormalize
 }
 
 /**

client/sync.js

@@ -109,8 +109,8 @@ const startSync = (requester) => {
     return jsRecords
   }
 
-  ipc.on(messages.FETCH_SYNC_RECORDS, (e, categoryNames, startAt, limitResponse) => {
-    logSync(`fetching ${categoryNames} records after ${startAt}`)
+  ipc.on(messages.FETCH_SYNC_RECORDS, (e, categoryNames, startAt, limitResponse, previousFetchTime) => {
+    logSync(`fetching ${categoryNames} records after ${startAt} previous fetch is ${previousFetchTime}`)
     categoryNames.forEach((category) => {
       if (!proto.categories[category]) {
         throw new Error(`Unsupported sync category: ${category}`)
@@ -119,7 +119,7 @@ const startSync = (requester) => {
       if (nextContinuationTokens[category]) {
         continuationToken = nextContinuationTokens[category]
       }
-      requester.list(proto.categories[category], startAt, limitResponse, continuationToken).then((s3Objects) => {
+      requester.list(proto.categories[category], startAt, limitResponse, continuationToken, previousFetchTime).then((s3Objects) => {
         const jsRecords = getJSRecords(s3Objects.contents)
         logSync(`got ${jsRecords.length} decrypted records in ${category} after ${startAt}`)
         let lastRecordTimestamp
@@ -220,7 +220,7 @@ const startSync = (requester) => {
       ipc.send(messages.GET_EXISTING_OBJECTS, category, jsRecords, 0, false)
     }
     if (!isCompactionInProgress) {
-      requester.list(proto.categories[category], 0, 1000, '',
+      requester.list(proto.categories[category], 0, 1000, '', 0,
         {compaction: true, compactionDoneCb: compactionDone, compactionUpdateCb: compactionUpdate}).then(() => {
         logSync(`Compacting category: ${category}`)
         isCompactionInProgress = true