Fix #599 - Record checkpoint on empty _changes result (#600)

* Set check point when the changes are empty (e.g. in filtered pull replications)
* Update checkpoint only when modified from previous recorded checkpoint
* Added checkpoint tests

Author: jjrodrig

CHANGES.md

@@ -1,3 +1,6 @@
+# 2.x.y (Unreleased)
+- [IMPROVED] Record checkpoint on empty `_changes` result in pull replications. This change optimizes 
+   filtered replications when changes in remote database doesn't match the replication filter. 
 # 2.4.0 (2019-01-15)
 - [NEW] `Database` methods `read`, `contains`, `create`, and `delete` now accept local
   (non-replicating documents). These documents must have their document ID prefixed with `_local/`

cloudant-sync-datastore-core/src/main/java/com/cloudant/sync/internal/replication/PullStrategy.java

@@ -236,9 +236,10 @@ private void replicate()
         }
 
         this.state.documentCounter = 0;
+
         while (!this.state.cancel) {
             this.state.batchCounter++;
-
+            final Object lastKnownCheckpoint = this.targetDb.getCheckpoint(this.getReplicationId());
             String msg = String.format(
                     "Batch %s started (completed %s changes so far)",
                     this.state.batchCounter,
@@ -247,7 +248,7 @@ private void replicate()
             logger.info(msg);
             long batchStartTime = System.currentTimeMillis();
 
-            ChangesResultWrapper changeFeeds = this.nextBatch();
+            ChangesResultWrapper changeFeeds = this.nextBatch(lastKnownCheckpoint);
             int batchChangesProcessed = 0;
 
             // So we can check whether all changes were processed during
@@ -264,6 +265,16 @@ private void replicate()
                 state.documentCounter += batchChangesProcessed;
             }
 
+            if (!this.state.cancel && (lastKnownCheckpoint == null || !lastKnownCheckpoint.equals(changeFeeds.getLastSeq()))) {
+                try {
+                    this.targetDb.putCheckpoint(this.getReplicationId(), changeFeeds.getLastSeq());
+                } catch (DocumentStoreException e) {
+                    logger.log(Level.WARNING, "Failed to put checkpoint doc, next replication " +
+                            "will " +
+                            "start from previous checkpoint", e);
+                }
+            }
+
             long batchEndTime = System.currentTimeMillis();
             msg = String.format(
                     "Batch %s completed in %sms (batch was %s changes)",
@@ -435,15 +446,6 @@ private int processOneChangesBatch(ChangesResultWrapper changeFeeds)
             }
         }
 
-        if (!this.state.cancel) {
-            try {
-                this.targetDb.putCheckpoint(this.getReplicationId(), changeFeeds.getLastSeq());
-            } catch (DocumentStoreException e) {
-                logger.log(Level.WARNING, "Failed to put checkpoint doc, next replication will " +
-                        "start from previous checkpoint", e);
-            }
-        }
-
         return changesProcessed;
     }
 
@@ -470,8 +472,7 @@ public String getReplicationId() throws DocumentStoreException {
         }
     }
 
-    private ChangesResultWrapper nextBatch() throws DocumentStoreException {
-        final Object lastCheckpoint = this.targetDb.getCheckpoint(this.getReplicationId());
+    private ChangesResultWrapper nextBatch(final Object lastCheckpoint) {
         logger.fine("last checkpoint " + lastCheckpoint);
 
         ChangesResult changeFeeds = null;

cloudant-sync-datastore-core/src/test/java/com/cloudant/sync/internal/replication/PullStrategyMockTest.java

@@ -16,6 +16,8 @@
 
 package com.cloudant.sync.internal.replication;
 
+import static org.junit.Assume.assumeNoException;
+import static org.mockito.Matchers.eq;
 import static org.mockito.Mockito.any;
 import static org.mockito.Mockito.anyInt;
 import static org.mockito.Mockito.anyString;
@@ -26,12 +28,12 @@
 import static org.mockito.Mockito.when;
 
 import com.cloudant.common.RequireRunningCouchDB;
+import com.cloudant.sync.event.Subscribe;
+import com.cloudant.sync.internal.documentstore.DocumentRevsList;
 import com.cloudant.sync.internal.mazha.ChangesResult;
 import com.cloudant.sync.internal.mazha.DocumentRevs;
 import com.cloudant.sync.internal.mazha.OkOpenRevision;
 import com.cloudant.sync.internal.mazha.OpenRevision;
-import com.cloudant.sync.internal.documentstore.DocumentRevsList;
-import com.cloudant.sync.event.Subscribe;
 import com.cloudant.sync.internal.util.JSONUtils;
 import com.cloudant.sync.replication.PullFilter;
 import com.cloudant.sync.util.TestUtils;
@@ -231,6 +233,107 @@ public Object answer(InvocationOnMock invocation) throws Throwable {
         verify(mockListener,never()).error(any(ReplicationStrategyErrored.class));
     }
 
+    @Test
+    public void testSetCheckpointWhenEmpty() throws Exception {
+        CouchDB mockRemoteDb = mock(CouchDB.class);
+        when(mockRemoteDb.changes((PullFilter) null, null, 1000)).then(new Answer<Object>() {
+            @Override
+            public Object answer(InvocationOnMock invocation) throws Throwable {
+                FileReader fr = new FileReader(TestUtils.loadFixture
+                        ("fixture/empty_changes.json"));
+                return JSONUtils.fromJson(fr, ChangesResult.class);
+            }
+        });
+        when(mockRemoteDb.exists()).thenReturn(true);
+
+        StrategyListener mockListener = mock(StrategyListener.class);
+        PullStrategy pullStrategy = super.getPullStrategy();
+        pullStrategy.sourceDb = mockRemoteDb;
+        pullStrategy.getEventBus().register(mockListener);
+        pullStrategy.run();
+
+        //should have 0 document
+        Assert.assertEquals(this.datastore.getDocumentCount(), 0);
+        //Checkpoint should be created in targetDb
+        String checkpoint =
+                (String) pullStrategy.targetDb.getCheckpoint(pullStrategy.getReplicationId());
+        Assert.assertEquals(checkpoint, "10-d9e5b0147af143e5b6d1979378ad957b");
+        //make sure the correct events were fired
+        verify(mockListener).complete(any(ReplicationStrategyCompleted.class));
+        verify(mockListener, never()).error(any(ReplicationStrategyErrored.class));
+    }
+
+    @Test
+    public void testDoNotSetCheckpointWhenNotModified() throws Exception {
+        try {
+            CouchDB mockRemoteDb = mock(CouchDB.class);
+            when(mockRemoteDb.changes((PullFilter) null, "10-d9e5b0147af143e5b6d1979378ad957b", 1000)).then(new Answer<Object>() {
+                @Override
+                public Object answer(InvocationOnMock invocation) throws Throwable {
+                    FileReader fr = new FileReader(TestUtils.loadFixture
+                            ("fixture/empty_changes.json"));
+                    return JSONUtils.fromJson(fr, ChangesResult.class);
+                }
+            });
+            when(mockRemoteDb.exists()).thenReturn(true);
+
+            DatastoreWrapper mockLocalDb = mock(DatastoreWrapper.class);
+            when(mockLocalDb.getCheckpoint(any(String.class))).thenReturn("10" +
+                    "-d9e5b0147af143e5b6d1979378ad957b");
+
+            StrategyListener mockListener = mock(StrategyListener.class);
+            PullStrategy pullStrategy = super.getPullStrategy();
+            pullStrategy.sourceDb = mockRemoteDb;
+            pullStrategy.targetDb = mockLocalDb;
+            pullStrategy.getEventBus().register(mockListener);
+            pullStrategy.run();
+
+            //make sure the correct events were fired
+            verify(mockListener).complete(any(ReplicationStrategyCompleted.class));
+            verify(mockListener, never()).error(any(ReplicationStrategyErrored.class));
+            verify(mockLocalDb, never()).putCheckpoint(anyString(), any());
+        } catch(UnsupportedOperationException uoe) {
+            assumeNoException("Cannot proxy DatastoreWrapper on Dalvik", uoe);
+        }
+    }
+
+    @Test
+    public void testSetCheckpointWhenModified() throws Exception {
+        try {
+            CouchDB mockRemoteDb = mock(CouchDB.class);
+            when(mockRemoteDb.changes((PullFilter) null, "9-d9e5b0147af143e5b6d1979378ad957b",
+                    1000)).then(new Answer<Object>() {
+                @Override
+                public Object answer(InvocationOnMock invocation) throws Throwable {
+                    FileReader fr = new FileReader(TestUtils.loadFixture
+                            ("fixture/empty_changes.json"));
+                    return JSONUtils.fromJson(fr, ChangesResult.class);
+                }
+            });
+            when(mockRemoteDb.exists()).thenReturn(true);
+
+            DatastoreWrapper mockLocalDb = mock(DatastoreWrapper.class);
+            when(mockLocalDb.getCheckpoint(any(String.class))).thenReturn("9" +
+                    "-d9e5b0147af143e5b6d1979378ad957b");
+
+            StrategyListener mockListener = mock(StrategyListener.class);
+            PullStrategy pullStrategy = super.getPullStrategy();
+            pullStrategy.sourceDb = mockRemoteDb;
+            pullStrategy.targetDb = mockLocalDb;
+            pullStrategy.getEventBus().register(mockListener);
+            pullStrategy.run();
+
+            //make sure the correct events were fired
+            verify(mockListener).complete(any(ReplicationStrategyCompleted.class));
+            verify(mockListener, never()).error(any(ReplicationStrategyErrored.class));
+            verify(mockLocalDb).putCheckpoint(anyString(), eq("10" +
+                    "-d9e5b0147af143e5b6d1979378ad957b"));
+        } catch (UnsupportedOperationException uoe) {
+            assumeNoException("Cannot proxy DatastoreWrapper on Dalvik", uoe);
+        }
+    }
+
+
     public class StrategyListener {
 
         @Subscribe

fixture/empty_changes.json

@@ -0,0 +1,5 @@
+{
+    "last_seq": "10-d9e5b0147af143e5b6d1979378ad957b",
+    "pending": 0,
+    "results": []
+}