Bug 726054 - Desktop parity: track last sync time from final upload, not final fetch. r=nalexander

mobile/android/base/sync/middleware/Crypto5MiddlewareRepositorySession.java

@@ -146,7 +146,7 @@ public class Crypto5MiddlewareRepositorySession extends RepositorySession {
     }
 
     @Override
-    public void onFetchSucceeded(Record[] records, long end) {
+    public void onFetchSucceeded(Record[] records, long fetchEnd) {
       for (Record record : records) {
         try {
           this.onFetchedRecord(record);
@@ -154,12 +154,12 @@ public class Crypto5MiddlewareRepositorySession extends RepositorySession {
           this.onFetchFailed(e, record);
         }
       }
-      this.onFetchCompleted(end);
+      this.onFetchCompleted(fetchEnd);
     }
 
     @Override
-    public void onFetchCompleted(long end) {
+    public void onFetchCompleted(final long fetchEnd) {
-      next.onFetchCompleted(end);
+      next.onFetchCompleted(fetchEnd);
     }
 
     @Override
@@ -237,4 +237,9 @@ public class Crypto5MiddlewareRepositorySession extends RepositorySession {
   public void storeDone() {
     inner.storeDone();
   }
+
+  @Override
+  public void storeDone(long end) {
+    inner.storeDone(end);
+  }
 }

mobile/android/base/sync/repositories/RepositorySession.java

@@ -147,11 +147,18 @@ public abstract class RepositorySession {
   public abstract void store(Record record) throws NoStoreDelegateException;
 
   public void storeDone() {
+    // Our default behavior will be to assume that the Runnable is
+    // executed as soon as all the stores synchronously finish, so
+    // our end timestamp can just be… now.
+    storeDone(now());
+  }
+
+  public void storeDone(final long end) {
     Log.d(LOG_TAG, "Scheduling onStoreCompleted for after storing is done.");
     Runnable command = new Runnable() {
       @Override
       public void run() {
-        delegate.onStoreCompleted();
+        delegate.onStoreCompleted(end);
       }
     };
     storeWorkQueue.execute(command);

mobile/android/base/sync/repositories/RepositorySessionBundle.java

@@ -41,6 +41,7 @@ import java.io.IOException;
 
 import org.json.simple.parser.ParseException;
 import org.mozilla.gecko.sync.ExtendedJSONObject;
+import org.mozilla.gecko.sync.Logger;
 import org.mozilla.gecko.sync.NonObjectJSONException;
 
 import android.util.Log;
@@ -75,8 +76,11 @@ public class RepositorySessionBundle extends ExtendedJSONObject {
   }
 
   public void bumpTimestamp(long timestamp) {
-    if (timestamp > this.getTimestamp()) {
+    long existing = this.getTimestamp();
+    if (timestamp > existing) {
       this.setTimestamp(timestamp);
+    } else {
+      Logger.debug(LOG_TAG, "Timestamp " + timestamp + " not greater than " + existing + "; not bumping.");
     }
   }
 }

mobile/android/base/sync/repositories/Server11RepositorySession.java

@@ -44,6 +44,7 @@ import java.net.URI;
 import java.net.URISyntaxException;
 import java.util.ArrayList;
 import java.util.Date;
+import java.util.concurrent.atomic.AtomicLong;
 
 import org.json.simple.JSONArray;
 import org.mozilla.gecko.sync.CryptoRecord;
@@ -186,6 +187,20 @@ public class Server11RepositorySession extends RepositorySession {
 
 
   Server11Repository serverRepository;
+  AtomicLong uploadTimestamp = new AtomicLong(0);
+
+  private void bumpUploadTimestamp(long ts) {
+    while (true) {
+      long existing = uploadTimestamp.get();
+      if (existing > ts) {
+        return;
+      }
+      if (uploadTimestamp.compareAndSet(existing, ts)) {
+        return;
+      }
+    }
+  }
+
   public Server11RepositorySession(Repository repository) {
     super(repository);
     serverRepository = (Server11Repository) repository;
@@ -320,7 +335,7 @@ public class Server11RepositorySession extends RepositorySession {
   public void storeDone() {
     synchronized (recordsBufferMonitor) {
       flush();
-      super.storeDone();
+      storeDone(uploadTimestamp.get());
     }
   }
 
@@ -379,6 +394,10 @@ public class Server11RepositorySession extends RepositorySession {
             (success.size() > 0)) {
           Log.d(LOG_TAG, "Successful records: " + success.toString());
           // TODO: how do we notify without the whole record?
+
+          long ts = response.normalizedWeaveTimestamp();
+          Log.d(LOG_TAG, "Passing back upload X-Weave-Timestamp: " + ts);
+          bumpUploadTimestamp(ts);
         }
         if ((failed != null) &&
             (failed.object.size() > 0)) {

mobile/android/base/sync/repositories/delegates/DeferredRepositorySessionFetchRecordsDelegate.java

@@ -61,11 +61,11 @@ public class DeferredRepositorySessionFetchRecordsDelegate implements Repository
   }
 
   @Override
-  public void onFetchSucceeded(final Record[] records, final long end) {
+  public void onFetchSucceeded(final Record[] records, final long fetchEnd) {
     executor.execute(new Runnable() {
       @Override
       public void run() {
-        inner.onFetchSucceeded(records, end);  
+        inner.onFetchSucceeded(records, fetchEnd);
       }
     });       
   }
@@ -81,11 +81,11 @@ public class DeferredRepositorySessionFetchRecordsDelegate implements Repository
   }
 
   @Override
-  public void onFetchCompleted(final long end) {
+  public void onFetchCompleted(final long fetchEnd) {
     executor.execute(new Runnable() {
       @Override
       public void run() {
-        inner.onFetchCompleted(end);
+        inner.onFetchCompleted(fetchEnd);
       }
     });        
   }
@@ -97,4 +97,4 @@ public class DeferredRepositorySessionFetchRecordsDelegate implements Repository
     }
     throw new IllegalArgumentException("Can't re-defer this delegate.");
   }
-}
+}

mobile/android/base/sync/repositories/delegates/DeferredRepositorySessionStoreDelegate.java

@@ -81,12 +81,12 @@ public class DeferredRepositorySessionStoreDelegate implements
   }
 
   @Override
-  public void onStoreCompleted() {
+  public void onStoreCompleted(final long storeEnd) {
     executor.execute(new Runnable() {
       @Override
       public void run() {
-        inner.onStoreCompleted();
+        inner.onStoreCompleted(storeEnd);
       }
     });
   }
-}
+}

mobile/android/base/sync/repositories/delegates/RepositorySessionFetchRecordsDelegate.java

@@ -55,11 +55,11 @@ public interface RepositorySessionFetchRecordsDelegate {
    *        which the request was received.
    *        E.g., the (normalized) value of the X-Weave-Timestamp header.
    */
-  public void onFetchCompleted(long end);
+  public void onFetchCompleted(final long fetchEnd);
 
   // Shorthand for calling onFetchedRecord for each record in turn, then
   // calling onFetchCompleted.
-  public void onFetchSucceeded(Record[] records, long end);
+  public void onFetchSucceeded(Record[] records, final long fetchEnd);
 
   public RepositorySessionFetchRecordsDelegate deferredFetchDelegate(ExecutorService executor);
 }

mobile/android/base/sync/repositories/delegates/RepositorySessionStoreDelegate.java

@@ -51,6 +51,6 @@ import org.mozilla.gecko.sync.repositories.domain.Record;
 public interface RepositorySessionStoreDelegate {
   public void onRecordStoreFailed(Exception ex);
   public void onRecordStoreSucceeded(Record record);
-  public void onStoreCompleted();
+  public void onStoreCompleted(long storeEnd);
   public RepositorySessionStoreDelegate deferredStoreDelegate(ExecutorService executor);
 }

mobile/android/base/sync/synchronizer/RecordsChannel.java

@@ -51,8 +51,6 @@ import org.mozilla.gecko.sync.repositories.delegates.RepositorySessionFetchRecor
 import org.mozilla.gecko.sync.repositories.delegates.RepositorySessionStoreDelegate;
 import org.mozilla.gecko.sync.repositories.domain.Record;
 
-import android.util.Log;
-
 /**
  * Pulls records from `source`, applying them to `sink`.
  * Notifies its delegate of errors and completion.
@@ -104,7 +102,7 @@ class RecordsChannel implements
   public RepositorySession sink;
   private RecordsChannelDelegate delegate;
   private long timestamp;
-  private long end = -1;                     // Oo er, missus.
+  private long fetchEnd = -1;
 
   public RecordsChannel(RepositorySession source, RepositorySession sink, RecordsChannelDelegate delegate) {
     this.source    = source;
@@ -198,19 +196,19 @@ class RecordsChannel implements
   }
 
   @Override
-  public void onFetchSucceeded(Record[] records, long end) {
+  public void onFetchSucceeded(Record[] records, final long fetchEnd) {
     for (Record record : records) {
       this.toProcess.add(record);
     }
     this.consumer.doNotify();
-    this.onFetchCompleted(end);
+    this.onFetchCompleted(fetchEnd);
   }
 
   @Override
-  public void onFetchCompleted(long end) {
+  public void onFetchCompleted(final long fetchEnd) {
     Logger.info(LOG_TAG, "onFetchCompleted. Stopping consumer once stores are done.");
-    Logger.info(LOG_TAG, "Fetch timestamp is " + end);
+    Logger.info(LOG_TAG, "Fetch timestamp is " + fetchEnd);
-    this.end = end;
+    this.fetchEnd = fetchEnd;
     this.consumer.queueFilled();
   }
 
@@ -237,10 +235,11 @@ class RecordsChannel implements
   }
 
   @Override
-  public void onStoreCompleted() {
+  public void onStoreCompleted(long storeEnd) {
-    Logger.info(LOG_TAG, "onStoreCompleted. Notifying delegate of onFlowCompleted. End is " + end);
+    Logger.info(LOG_TAG, "onStoreCompleted. Notifying delegate of onFlowCompleted. " +
+                         "Fetch end is " + fetchEnd + ", store end is " + storeEnd);
     // TODO: synchronize on consumer callback?
-    delegate.onFlowCompleted(this, end);
+    delegate.onFlowCompleted(this, fetchEnd, storeEnd);
   }
 
   @Override

mobile/android/base/sync/synchronizer/RecordsChannelDelegate.java

@@ -38,7 +38,7 @@
 package org.mozilla.gecko.sync.synchronizer;
 
 public interface RecordsChannelDelegate {
-  public void onFlowCompleted(RecordsChannel recordsChannel, long end);
+  public void onFlowCompleted(RecordsChannel recordsChannel, long fetchEnd, long storeEnd);
   public void onFlowBeginFailed(RecordsChannel recordsChannel, Exception ex);
   public void onFlowStoreFailed(RecordsChannel recordsChannel, Exception ex);
   public void onFlowFinishFailed(RecordsChannel recordsChannel, Exception ex);

mobile/android/base/sync/synchronizer/SynchronizerSession.java

@@ -66,8 +66,15 @@ implements RecordsChannelDelegate,
   private RepositorySession sessionB;
   private RepositorySessionBundle bundleA;
   private RepositorySessionBundle bundleB;
+
+  // Bug 726054: just like desktop, we track our last interaction with the server,
+  // not the last record timestamp that we fetched. This ensures that we don't re-
+  // download the records we just uploaded, at the cost of skipping any records
+  // that a concurrently syncing client has uploaded.
   private long pendingATimestamp = -1;
   private long pendingBTimestamp = -1;
+  private long storeEndATimestamp = -1;
+  private long storeEndBTimestamp = -1;
   private boolean flowAToBCompleted = false;
   private boolean flowBToACompleted = false;
 
@@ -133,9 +140,11 @@ implements RecordsChannelDelegate,
     // TODO: failed record handling.
     final RecordsChannel channelBToA = new RecordsChannel(this.sessionB, this.sessionA, this);
     RecordsChannelDelegate channelDelegate = new RecordsChannelDelegate() {
-      public void onFlowCompleted(RecordsChannel recordsChannel, long end) {
+      public void onFlowCompleted(RecordsChannel recordsChannel, long fetchEnd, long storeEnd) {
-        info("First RecordsChannel flow completed. End is " + end + ". Starting next.");
+        info("First RecordsChannel flow completed. Fetch end is " + fetchEnd +
-        pendingATimestamp = end;
+             ". Store end is " + storeEnd + ". Starting next.");
+        pendingATimestamp = fetchEnd;
+        storeEndBTimestamp = storeEnd;
         flowAToBCompleted = true;
         channelBToA.flow();
       }
@@ -165,9 +174,12 @@ implements RecordsChannelDelegate,
   }
 
   @Override
-  public void onFlowCompleted(RecordsChannel channel, long end) {
+  public void onFlowCompleted(RecordsChannel channel, long fetchEnd, long storeEnd) {
-    info("Second RecordsChannel flow completed. End is " + end + ". Finishing.");
+    info("Second RecordsChannel flow completed. Fetch end is " + fetchEnd +
-    pendingBTimestamp = end;
+         ". Store end is " + storeEnd + ". Finishing.");
+
+    pendingBTimestamp = fetchEnd;
+    storeEndATimestamp = storeEnd;
     flowBToACompleted = true;
 
     // Finish the two sessions.
@@ -278,8 +290,8 @@ implements RecordsChannelDelegate,
 
     if (session == sessionA) {
       if (flowAToBCompleted) {
-        info("onFinishSucceeded: bumping session A's timestamp to " + pendingATimestamp);
+        info("onFinishSucceeded: bumping session A's timestamp to " + pendingATimestamp + " or " + storeEndATimestamp);
-        bundle.bumpTimestamp(pendingATimestamp);
+        bundle.bumpTimestamp(Math.max(pendingATimestamp, storeEndATimestamp));
         this.synchronizer.bundleA = bundle;
       }
       if (this.sessionB != null) {
@@ -289,8 +301,8 @@ implements RecordsChannelDelegate,
       }
     } else if (session == sessionB) {
       if (flowBToACompleted) {
-        info("onFinishSucceeded: bumping session B's timestamp to " + pendingBTimestamp);
+        info("onFinishSucceeded: bumping session B's timestamp to " + pendingBTimestamp + " or " + storeEndBTimestamp);
-        bundle.bumpTimestamp(pendingBTimestamp);
+        bundle.bumpTimestamp(Math.max(pendingBTimestamp, storeEndBTimestamp));
         this.synchronizer.bundleB = bundle;
         info("Notifying delegate.onSynchronized.");
         this.delegate.onSynchronized(this);