Fix snapshot calculation in ludicrous mode (#5636)

In ludicrous mode snapshot calculation is not happening correctly. This is resulting in no
snapshot being taken ever and hence w directory size is increasing with time.
For snapshot creation we need last applied raft index(such that all raft index entries before this index are also applied) and max timestamp out of all committed transactions. Committed
transactions are sent by Zero to all alpha leaders via subscription. In ludicrous mode we don't
send transaction to Zero while committing. Hence Zero doesn't have anything to send as part of
Oracle updates. Because of this, possible snapshot timestamp(maxCommitTs) in
calculateSnapshot() is always 0. Hence no snapshot is ever taken.

This PR addresses two issues:

Correction while marking raft index done. Now any raft index will only be marked as done once
we have applied mutations for proposal(raft entry).
For ludicrous mode, use StartTs of any transaction for calculating maxCommitTs snapshot in
calculateSnapshot()

Author: ashish-goswami

worker/draft.go

@@ -215,7 +215,7 @@ func newNode(store *raftwal.DiskStorage, gid uint32, id uint64, myAddr string) *
 		ops:     make(map[op]*y.Closer),
 	}
 	if x.WorkerConfig.LudicrousMode {
-		n.ex = newExecutor()
+		n.ex = newExecutor(&m.Applied)
 	}
 	return n
 }
@@ -420,7 +420,7 @@ func (n *node) applyMutations(ctx context.Context, proposal *pb.Proposal) (rerr
 	})
 
 	if x.WorkerConfig.LudicrousMode {
-		n.ex.addEdges(ctx, m.StartTs, m.Edges)
+		n.ex.addEdges(ctx, proposal.Index, m.StartTs, m.Edges)
 		return nil
 	}
 
@@ -489,14 +489,7 @@ func (n *node) applyCommitted(proposal *pb.Proposal) error {
 			span.Annotatef(nil, "While applying mutations: %v", err)
 			return err
 		}
-		if x.WorkerConfig.LudicrousMode {
-			ts := proposal.Mutations.StartTs
-			return n.commitOrAbort(proposal.Key, &pb.OracleDelta{
-				Txns: []*pb.TxnStatus{
-					{StartTs: ts, CommitTs: ts},
-				},
-			})
-		}
+
 		span.Annotate(nil, "Done")
 		return nil
 	}
@@ -1430,13 +1423,18 @@ func (n *node) calculateSnapshot(startIdx uint64, discardN int) (*pb.Snapshot, e
 				span.Annotatef(nil, "Error: %v", err)
 				return nil, err
 			}
+
+			var start uint64
 			if proposal.Mutations != nil {
-				start := proposal.Mutations.StartTs
+				start = proposal.Mutations.StartTs
 				if start >= minPendingStart && snapshotIdx == 0 {
 					snapshotIdx = entry.Index - 1
 				}
 			}
-			if proposal.Delta != nil {
+			// In ludicrous mode commitTs for any transaction is same as startTs.
+			if x.WorkerConfig.LudicrousMode {
+				maxCommitTs = x.Max(maxCommitTs, start)
+			} else if proposal.Delta != nil {
 				for _, txn := range proposal.Delta.GetTxns() {
 					maxCommitTs = x.Max(maxCommitTs, txn.CommitTs)
 				}

worker/executor.go

@@ -33,6 +33,7 @@ type subMutation struct {
 	edges   []*pb.DirectedEdge
 	ctx     context.Context
 	startTs uint64
+	index   uint64
 }
 
 type executor struct {
@@ -41,12 +42,14 @@ type executor struct {
 	sync.RWMutex
 	predChan map[string]chan *subMutation
 	closer   *y.Closer
+	applied  *y.WaterMark
 }
 
-func newExecutor() *executor {
+func newExecutor(applied *y.WaterMark) *executor {
 	ex := &executor{
 		predChan: make(map[string]chan *subMutation),
 		closer:   y.NewCloser(0),
+		applied:  applied,
 	}
 	go ex.shutdown()
 	return ex
@@ -80,6 +83,7 @@ func (e *executor) processMutationCh(ch chan *subMutation) {
 			glog.Errorf("Error while waiting for writes: %v", err)
 		}
 
+		e.applied.Done(payload.index)
 		atomic.AddInt64(&e.pendingSize, -esize)
 	}
 }
@@ -111,7 +115,7 @@ const (
 	executorAddEdges          = "executor.addEdges"
 )
 
-func (e *executor) addEdges(ctx context.Context, startTs uint64, edges []*pb.DirectedEdge) {
+func (e *executor) addEdges(ctx context.Context, index, startTs uint64, edges []*pb.DirectedEdge) {
 	rampMeter(&e.pendingSize, maxPendingEdgesSize, executorAddEdges)
 
 	payloadMap := make(map[string]*subMutation)
@@ -122,6 +126,7 @@ func (e *executor) addEdges(ctx context.Context, startTs uint64, edges []*pb.Dir
 			payloadMap[edge.Attr] = &subMutation{
 				ctx:     ctx,
 				startTs: startTs,
+				index:   index,
 			}
 			payload = payloadMap[edge.Attr]
 		}
@@ -138,6 +143,7 @@ func (e *executor) addEdges(ctx context.Context, startTs uint64, edges []*pb.Dir
 	default:
 		// Closer is not closed. And we have the Lock, so sending on channel should be safe.
 		for attr, payload := range payloadMap {
+			e.applied.Begin(index)
 			e.getChannelUnderLock(attr) <- payload
 		}
 	}