Method followerApply

kayak/runtime.go:413–469

(l *kt.Log, checkPrepare bool)

}

func (r *Runtime) followerApply(l *kt.Log, checkPrepare bool) (err error) {
	if l == nil {
		err = errors.Wrap(kt.ErrInvalidLog, "log is nil")
		return
	}
	if atomic.LoadUint32(&r.started) != 1 {
		err = kt.ErrStopped
		return
	}

	ctx, task := trace.NewTask(context.Background(), "Kayak.FollowerApply."+l.Type.String())
	defer task.End()

	tm := timer.NewTimer()

	var storageErr error

	defer func() {
		log.
			WithFields(log.Fields{
				"t":  l.Type.String(),
				"i":  l.Index,
				"se": storageErr,
			}).
			WithFields(tm.ToLogFields()).
			WithError(err).
			Debug("kayak follower apply")
	}()

	r.peersLock.RLock()
	defer r.peersLock.RUnlock()

	tm.Add("peers_lock")

	if r.role == proto.Leader {
		// not follower
		err = kt.ErrNotFollower
		return
	}

	// verify log structure
	switch l.Type {
	case kt.LogPrepare:
		err = r.followerPrepare(ctx, tm, l, checkPrepare)
	case kt.LogRollback:
		err = r.followerRollback(ctx, tm, l)
	case kt.LogCommit:
		storageErr, err = r.followerCommit(ctx, tm, l)
	}

	if err == nil {
		r.updateNextIndex(ctx, l)
		r.triggerLogAwaits(l)
	}

	return
}