[WIP] Split membership sync endpoints

conn/raft_server.go

@@ -78,6 +78,13 @@ func (p *proposals) Store(key string, pctx *ProposalCtx) bool {
 	return true
 }
 
+func (p *proposals) Ctx(key string) context.Context {
+	if pctx := p.Get(key); pctx != nil {
+		return pctx.Ctx
+	}
+	return context.Background()
+}
+
 func (p *proposals) Get(key string) *ProposalCtx {
 	p.RLock()
 	defer p.RUnlock()

contrib/blockade/main.go

@@ -105,7 +105,10 @@ func testCommon(remove, join string, minAlphasUp int) error {
 		if err := increment(minAlphasUp); err != nil {
 			return err
 		}
-		// Then join.
+		// Then join, if available.
+		if len(join) == 0 {
+			continue
+		}
 		if err := run(ctxb, join); err != nil {
 			return err
 		}
@@ -155,10 +158,16 @@ func runTests() error {
 	fmt.Println("===> Slow TEST: OK")
 
 	if err := testCommon("blockade stop", "blockade start --all", 2); err != nil {
-		fmt.Printf("Error testRestart: %v\n", err)
+		fmt.Printf("Error testRestart with stop: %v\n", err)
+		return err
+	}
+	fmt.Println("===> Restart TEST1: OK")
+
+	if err := testCommon("blockade restart", "", 3); err != nil {
+		fmt.Printf("Error testRestart with restart: %v\n", err)
 		return err
 	}
-	fmt.Println("===> Restart TEST: OK")
+	fmt.Println("===> Restart TEST2: OK")
 
 	if err := testCommon("blockade partition", "blockade join", 2); err != nil {
 		fmt.Printf("Error testPartitions: %v\n", err)

dgraph/cmd/zero/oracle.go

@@ -86,10 +86,6 @@ func (o *Oracle) purgeBelow(minTs uint64) {
 	o.Lock()
 	defer o.Unlock()
 
-	// TODO: HACK. Remove this later.
-	glog.Infof("Not purging below: %d", minTs)
-	return
-
 	// Dropping would be cheaper if abort/commits map is sharded
 	for ts := range o.commits {
 		if ts < minTs {
@@ -203,7 +199,9 @@ func (o *Oracle) sendDeltasToSubscribers() {
 			// Don't goto slurp_loop, because it would break from select immediately.
 		}
 
-		glog.V(2).Infof("DoneUntil: %d. Sending delta: %+v\n", o.doneUntil.DoneUntil(), delta)
+		if glog.V(3) {
+			glog.Infof("DoneUntil: %d. Sending delta: %+v\n", o.doneUntil.DoneUntil(), delta)
+		}
 		o.Lock()
 		for id, ch := range o.subscribers {
 			select {
@@ -446,40 +444,6 @@ func (s *Server) SyncedUntil() uint64 {
 	return syncUntil
 }
 
-func (s *Server) purgeOracle() {
-	ticker := time.NewTicker(time.Second * 30)
-	defer ticker.Stop()
-
-	var lastPurgeTs uint64
-OUTER:
-	for {
-		<-ticker.C
-		groups := s.KnownGroups()
-		var minTs uint64
-		for _, group := range groups {
-			pl := s.Leader(group)
-			if pl == nil {
-				glog.Errorf("No healthy connection found to leader of group %d\n", group)
-				goto OUTER
-			}
-			c := pb.NewWorkerClient(pl.Get())
-			num, err := c.PurgeTs(context.Background(), &api.Payload{})
-			if err != nil {
-				glog.Errorf("Error while fetching minTs from group %d, err: %v\n", group, err)
-				goto OUTER
-			}
-			if minTs == 0 || num.Val < minTs {
-				minTs = num.Val
-			}
-		}
-
-		if minTs > 0 && minTs != lastPurgeTs {
-			s.orc.purgeBelow(minTs)
-			lastPurgeTs = minTs
-		}
-	}
-}
-
 func (s *Server) TryAbort(ctx context.Context,
 	txns *pb.TxnTimestamps) (*pb.OracleDelta, error) {
 	delta := &pb.OracleDelta{}