Support clustered purge

- add 2 APIs(set_purge_seq, get_purge_seq).
- keep purge_seq in clouseau.
- adjust the commit logic to update purge seq.
- update the test code.

BugzID: 68280

Author: chguocloudant

src/main/scala/com/cloudant/clouseau/ClouseauTypeFactory.scala

@@ -41,6 +41,7 @@ case class UpdateDocMsg(id: String, doc: Document)
 case class DeleteDocMsg(id: String)
 case class CommitMsg(seq: Long)
 case class SetUpdateSeqMsg(seq: Long)
+case class SetPurgeSeqMsg(seq: Long)
 
 object ClouseauTypeFactory extends TypeFactory {
 
@@ -89,6 +90,8 @@ object ClouseauTypeFactory extends TypeFactory {
       Some(CommitMsg(toLong(reader.readTerm)))
     case ('set_update_seq, 2) =>
       Some(SetUpdateSeqMsg(toLong(reader.readTerm)))
+    case ('set_purge_seq, 2) =>
+      Some(SetPurgeSeqMsg(toLong(reader.readTerm)))
     case _ =>
       None
   }

src/main/scala/com/cloudant/clouseau/IndexService.scala

@@ -70,6 +70,8 @@ class IndexService(ctx: ServiceContext[IndexServiceArgs]) extends Service(ctx) w
   var reader = DirectoryReader.open(ctx.args.writer, true)
   var updateSeq = getCommittedSeq
   var pendingSeq = updateSeq
+  var purgeSeq = getCommittedPurgeSeq
+  var pendingPurgeSeq = purgeSeq
   var committing = false
   var forceRefresh = false
 
@@ -97,6 +99,8 @@ class IndexService(ctx: ServiceContext[IndexServiceArgs]) extends Service(ctx) w
         group2(request)
       case 'get_update_seq =>
         ('ok, updateSeq)
+      case 'get_purge_seq =>
+        ('ok, purgeSeq)
       case UpdateDocMsg(id: String, doc: Document) =>
         logger.debug("Updating %s".format(id))
         updateTimer.time {
@@ -117,6 +121,11 @@ class IndexService(ctx: ServiceContext[IndexServiceArgs]) extends Service(ctx) w
         pendingSeq = newSeq
         logger.debug("Pending sequence is now %d".format(newSeq))
         'ok
+      case SetPurgeSeqMsg(newSeq: Long) =>
+        pendingPurgeSeq = newSeq
+        logger.debug("Pending purge sequence is now %d".format(newSeq))
+        commit(pendingSeq, pendingPurgeSeq)
+        'ok
       case 'info =>
         ('ok, getInfo)
     }
@@ -148,12 +157,13 @@ class IndexService(ctx: ServiceContext[IndexServiceArgs]) extends Service(ctx) w
       }
       exit('deleted)
     case 'maybe_commit =>
-      commit(pendingSeq)
-    case ('committed, newSeq: Long) =>
-      updateSeq = newSeq
+      commit(pendingSeq, pendingPurgeSeq)
+    case ('committed, newUpdateSeq: Long, newPurgeSeq: Long) =>
+      updateSeq = newUpdateSeq
+      purgeSeq = newPurgeSeq
       forceRefresh = true
       committing = false
-      logger.info("Committed sequence %d".format(newSeq))
+      logger.debug("Committed sequence %d and %d".format(newUpdateSeq, newPurgeSeq))
     case 'commit_failed =>
       committing = false
   }
@@ -175,18 +185,19 @@ class IndexService(ctx: ServiceContext[IndexServiceArgs]) extends Service(ctx) w
     }
   }
 
-  private def commit(newSeq: Long) {
-    if (!committing && newSeq > updateSeq) {
+  private def commit(newUpdateSeq: Long, newPurgeSeq: Long) {
+    if (!committing && (newUpdateSeq > updateSeq || newPurgeSeq > purgeSeq)) {
       committing = true
       val index = self
       node.spawn((_) => {
         ctx.args.writer.setCommitData(ctx.args.writer.getCommitData +
-          ("update_seq" -> newSeq.toString))
+          ("update_seq" -> newUpdateSeq.toString) +
+          ("purge_seq" -> newPurgeSeq.toString))
         try {
           commitTimer.time {
             ctx.args.writer.commit()
           }
-          index ! ('committed, newSeq)
+          index ! ('committed, newUpdateSeq, newPurgeSeq)
         } catch {
           case e: AlreadyClosedException =>
             logger.error("Commit failed to closed writer", e)
@@ -569,7 +580,8 @@ class IndexService(ctx: ServiceContext[IndexServiceArgs]) extends Service(ctx) w
       ('doc_count, reader.numDocs),
       ('doc_del_count, reader.numDeletedDocs),
       ('pending_seq, pendingSeq),
-      ('committed_seq, getCommittedSeq)
+      ('committed_seq, getCommittedSeq),
+      ('purge_seq, pendingPurgeSeq)
     )
   }
 
@@ -590,6 +602,16 @@ class IndexService(ctx: ServiceContext[IndexServiceArgs]) extends Service(ctx) w
     }
   }
 
+  private def getCommittedPurgeSeq = {
+    val commitData = ctx.args.writer.getCommitData
+    commitData.get("purge_seq") match {
+      case null =>
+        0L
+      case seq =>
+        seq.toLong
+    }
+  }
+
   private def parseSort(v: Any): Sort = v match {
     case 'relevance =>
       Sort.RELEVANCE

src/test/scala/com/cloudant/clouseau/IndexServiceSpec.scala

@@ -399,15 +399,25 @@ class IndexServiceSpec extends SpecificationWithJUnit {
       node.call(service, UpdateDocMsg("bar", doc2)) must be equalTo 'ok
       node.call(service, UpdateDocMsg("zzz", doc3)) must be equalTo 'ok
 
-      node.call(service, Group1Msg("*:*", "_id", true,"<distance,lon,lat,0.2,57.15,km>", 0, 10)) must beLike {
+      node.call(service, Group1Msg("*:*", "_id", true, "<distance,lon,lat,0.2,57.15,km>", 0, 10)) must beLike {
         case ('ok, List((foo, _), (zzz, _), (bar, _))) => ok
       }
 
-      node.call(service, Group1Msg("*:*", "_id", true,"<distance,lon,lat,12,57.15,km>", 0, 10)) must beLike {
+      node.call(service, Group1Msg("*:*", "_id", true, "<distance,lon,lat,12,57.15,km>", 0, 10)) must beLike {
         case ('ok, List((bar, _), (zzz, _), (foo, _))) => ok
       }
     }
 
+    "support set/get purge seq" in new index_service {
+      node.call(service, 'get_purge_seq) must be equalTo ('ok, 0)
+      node.call(service, SetPurgeSeqMsg(1)) must be equalTo 'ok
+      Thread.sleep(100)
+      node.call(service, 'get_purge_seq) must be equalTo ('ok, 1)
+      node.call(service, SetPurgeSeqMsg(2)) must be equalTo 'ok
+      Thread.sleep(100)
+      node.call(service, 'get_purge_seq) must be equalTo ('ok, 2)
+    }
+
   }
 
   private def isSearchable(node: Node, service: Pid,