Method followerDoCommit

kayak/commit.go:180–228

(req *commitReq)

}

func (r *Runtime) followerDoCommit(req *commitReq) {
	if req.log == nil {
		log.Fatal("NO LOG FOR FOLLOWER COMMIT")
		return
	}

	waitCommitTask := trace.StartRegion(req.ctx, "waitForLastCommit")

	// check for last commit availability
	myLastCommit := atomic.LoadUint64(&r.lastCommit)
	if req.lastCommit != myLastCommit {
		// TODO(): need counter for retries, infinite commit re-order would cause troubles
		go func(req *commitReq) {
			_, _ = r.waitForLog(req.ctx, req.lastCommit)
			r.commitCh <- req
		}(req)
		waitCommitTask.End()
		return
	}

	waitCommitTask.End()
	req.tm.Add("queue")

	defer trace.StartRegion(req.ctx, "commitCycle").End()

	var err error

	// write log first
	if err = r.writeWAL(req.ctx, req.log); err != nil {
		return
	}

	req.tm.Add("write_wal")

	// do commit, not wrapping underlying handler commit error
	_, storageErr := r.doCommit(req.ctx, req.data, false)

	req.tm.Add("db_write")

	// mark last commit
	atomic.StoreUint64(&r.lastCommit, req.log.Index)

	req.result.Set(&commitResult{
		err:        err,
		storageErr: storageErr,
	})

	return
}

func (r *Runtime) getPrepareLog(ctx context.Context, l *kt.Log) (lastCommitIndex uint64, pl *kt.Log, err error) {
	defer trace.StartRegion(ctx, "getPrepareLog").End()