Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

raft: advance commit index safely #139

Draft
wants to merge 2 commits into
base: main
Choose a base branch
from
Draft

raft: advance commit index safely #139

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
87ac09e
raft: advance commit index safely
pav-kv Jan 26, 2024
4e08f52
raft: pass the leader term to append/commit methods
pav-kv Jan 29, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 43 additions & 0 deletions log.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,35 @@ type raftLog struct {
// they will be saved into storage.
unstable unstable

// leaderTerm is a term of the leader with whom our log is "consistent". The
// log is guaranteed to be a prefix of this term's leader log.
//
// The leaderTerm can be safely updated to `t` if:
// 1. the last entry in the log has term `t`, or, more generally,
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Are there cases where (2) is insufficient and (1) is needed? I'm curious why we can't make this more specific. You say "We use (1) to initialize leaderTerm" below. Is this important on startup?

Copy link
Contributor Author

@pav-kv pav-kv Jan 29, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Only (2) is necessary. (1) is a cheap "stateful" version of (2).

On a server restart, we forget who did the last append. (1) gives the best guess, and allows recovering to up-to-date state for idle raft groups that haven't seen recent appends except the current leader's empty entry. For non-idle groups, or followers whose log is significantly behind the leader's and doesn't end with entries at its term, recovering to raftLog.lastTerm() gives no value, and is equivalent to setting leaderTerm = 0. To recover, these will then have to "wait" for the next append message from this leader.

It would be more ideal if this field was stored in some HardState.LeaderTerm - then we would always recover to up-to-date state. Note that we can't reuse HardState.Term because HardState.Term can be > leaderTerm (for the same reason why r.Term can be > leaderTerm).

To bring analogy with Paxos, the local raftLog is an acceptor, and raftLog.leaderTerm is the id of the highest accepted proposal. The election term is sort of orthogonal to this - the election term (r.Term / HardState.Term) can briefly jump in front of the accepted proposal term until there is an accepted proposal at this new term.

If we ever want to bring this to the next level: MsgApp messages should not be rejected based on r.Term / HardState.Term. For correctness, it is only necessary to reject MsgApp if the message term is < raftLog.leaderTerm / HardState.LeaderTerm. I think this would reduce some unnecessary rejects during the leader election flux time. caveat: #139 (comment)

Copy link
Contributor Author

@pav-kv pav-kv Jan 30, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I should note the invariant here, and maybe check it in a few places or tests:

	RawNode.raft.Term >= raftLog.leaderTerm >= raftLog.lastTerm()

(1) initializes leaderTerm to lastTerm(), which is safe because raft.Term >= raftLog.lastTerm()
(2) maintains it

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

allows recovering to up-to-date state for idle raft groups that haven't seen recent appends except the current leader's empty entry. For non-idle groups, or followers whose log is significantly behind the leader's and doesn't end with entries at its term, recovering to raftLog.lastTerm() gives no value, and is equivalent to setting leaderTerm = 0. To recover, these will then have to "wait" for the next append message from this leader.

This is what I was hoping to clarify. Initializing to lastTerm is an opportunistic way to allow a restarted follower in idle raft groups to immediately advance its commit index on startup without needing to first accept a MsgApp.

Is it anything more than that? For an idle raft group, a follower with an up-to-date log but a lagging commit index may restart and never receive any new entries. If we didn't have (1) and we started discarding commit indexes in heartbeats with terms > raftLog.leaderTerm, would the commit index on the follower get stuck?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That's a good point. I think we indeed need (1). Also, (1) plays nicely with the invariant that I put above.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If we ever want to bring this to the next level: MsgApp messages should not be rejected based on r.Term / HardState.Term. For correctness, it is only necessary to reject MsgApp if the message term is < raftLog.leaderTerm / HardState.LeaderTerm. I think this would reduce some unnecessary rejects during the leader election flux time.

Does this mean MsgApp from old leader will be accepted by voters that voted to a new leader but yet received any log from the new leader? In that case, there may be committed entry in old leader but not in the new leader.

Copy link
Contributor Author

@pav-kv pav-kv Feb 1, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@joshuazh-x That's a good point. We don't want a quorum of nodes to accept entries unless we are sure these entries are consistent with the new leader's log.

So it's safest to accept only MsgApp.Term >= raft.Term. But we could sometimes accept MsgApp.Term < raft.Term if:

  • The MsgApp contains the (index, term) entry for which we voted when electing the raft.Term leader. If that election wins, we know the new leader will append right after this entry.
  • We would truncate MsgApp.Entries at the aforementioned index we voted for, and append it.
  • This guarantees that this append is consistent with the new leader's log (i.e. the new leader would send us the exact same entries).

A vote is a promise to the leader not to accept any entries that are not in the leader's log. If we can deduce that an entry is in the leader's log (before / other than by getting a MsgApp directly from this leader), we can always safely accept it.

It's unclear if such an optimization would give any value (like reduce replication latency in some cases; probably it does avoid a duplicate MsgApp from the new leader when the election races with the old leader appends), so I will leave it as an exercise for later :) Looks like a complication.

Filed #150 with a more general technique that will bring more benefits.

// 2. the last successful append was sent by the leader `t`.
//
// This is due to the following safety property (see raft paper §5.3):
//
// Log Matching: if two logs contain an entry with the same index and term,
// then the logs are identical in all entries up through the given index.
//
// We use (1) to initialize leaderTerm, and (2) to maintain it on updates.
//
// NB: (2) does not imply (1). If our log is behind the leader's log, the last
// entry term can be below leaderTerm.
//
// NB: leaderTerm does not necessarily match this raft node's term. It only
// does for the leader. For followers and candidates, when we first learn or
// bump to a new term, we don't have a proof that our log is consistent with
// the new term's leader (current or prospective). The new leader may override
// any suffix of the log after the committed index. Only when the first append
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

"the log", "the committed index"

Are you referring to the replicated log here, or this replica's local log? In other words, should these "the"s be replaced by "our"?

Copy link
Contributor Author

@pav-kv pav-kv Jan 29, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Both, but mostly the "replicated". The leader can override only a suffix of the "replicated" log after the "replicated" commit index.

"Our" log is lagging the leader's log, such as our committed index. For "our" log, this implies that the leader can override a suffix after our committed index (but we have no way of checking by how much it can go back).

// from the new leader succeeds, we can update leaderTerm.
//
// During normal operation, leaderTerm matches the node term though. During a
// leader change, it briefly lags behind, and matches again when the first
// append message succeeds.
leaderTerm uint64
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

My immediate feeling is that it's a little weird to add a leaderTerm into raftLog, which shouldn't care about the info (leader's Term). It should be part of the raft instead of raftLog.

Copy link
Contributor Author

@pav-kv pav-kv Jan 29, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

See comment https://github.com/etcd-io/raft/pull/139/files#r1469818174, specifically the Paxos analogy.

Semantically, raftLog is layered under raft.go. The raftLog is agnostic to election, and essentially only implements the "Paxos acceptor" role. So it makes quite a good sense to me to have a "last accepted term" notion in raftLog. Moving this logic up in raft.go would be a layering violation of sorts.

which shouldn't care about the info (leader's Term)

raftLog should care about the leader term with which it is consistent (i.e. what I called the leaderTerm here). It should not care about the RawNode.Term / HardState.Term, because the latter can be bumped arbitrarily when a new election happens, and this bump is not coordinated with raftLog.

There is an invariant at raft.go level that is relevant here: RawNode.Term >= raftLog.leaderTerm. We use RawNode.Term instead of raftLog.leaderTerm in a few places (notably, to reject append messages), and have only been safe because of this implicit invariant. I think we should make it more explicit.


// committed is the highest log position that is known to be in
// stable storage on a quorum of nodes.
committed uint64
Expand Down Expand Up @@ -88,6 +117,11 @@ func newLogWithSize(storage Storage, logger Logger, maxApplyingEntsSize entryEnc
if err != nil {
panic(err) // TODO(bdarnell)
}
lastTerm, err := storage.Term(lastIndex)
if err != nil {
panic(err) // TODO(pav-kv)
}
log.leaderTerm = lastTerm
log.unstable.offset = lastIndex + 1
log.unstable.offsetInProgress = lastIndex + 1
log.unstable.logger = logger
Expand All @@ -106,6 +140,15 @@ func (l *raftLog) String() string {

// maybeAppend returns (0, false) if the entries cannot be appended. Otherwise,
// it returns (last index of new entries, true).
//
// TODO(pav-kv): pass in the term of the leader who sent this update. It is only
// safe to handle this append if this term is >= l.leaderTerm. It is only safe
// to override an uncommitted suffix of entries if term > l.leaderTerm.
//
// TODO(pav-kv): introduce a struct that consolidates the append metadata. The
// (prevEntryIndex, prevEntryTerm, leaderTerm) tuple must always be carried
pav-kv marked this conversation as resolved.
Show resolved Hide resolved
// together, and safety of this append must be checked at the lowest layer here,
// rather than up in raft.go.
func (l *raftLog) maybeAppend(index, logTerm, committed uint64, ents ...pb.Entry) (lastnewi uint64, ok bool) {
if !l.matchTerm(index, logTerm) {
return 0, false
Expand Down
15 changes: 14 additions & 1 deletion raft.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -935,6 +935,8 @@ func (r *raft) becomeLeader() {
// so the preceding log append does not count against the uncommitted log
// quota of the new leader. In other words, after the call to appendEntry,
// r.uncommittedSize is still 0.

r.raftLog.leaderTerm = r.Term // the leader's log is consistent with itself
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: is is common to manipulate fields in raftLog directly? Should we tuck this behind a method? Same question below.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Unfortunately, the direct manipulation on raftLog and Progress fields is scattered across the raft.go too. This goes into the TODO above: all the correctness handling should be done at the level below.

In this particular case/PR though, we can move the leaderTerm update into the r.appendEntry call a few lines above, which in turn will delegate this to r.raftLog.append method. This will be more correct, I agree.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done this (see second commit).

Unfortunately, too many things have incorrect interfaces: the raftLog append and commit methods don't take the leader term into consideration. This lead to some tests testing scenarios which can never happen.

Maybe it would be good to reverse the order of commits here: first clean up the raftLog interfaces, and then this PR will be a small change.

r.logger.Infof("%x became leader at term %d", r.id, r.Term)
}

Expand Down Expand Up @@ -1735,6 +1737,7 @@ func (r *raft) handleAppendEntries(m pb.Message) {
return
}
if mlastIndex, ok := r.raftLog.maybeAppend(m.Index, m.LogTerm, m.Commit, m.Entries...); ok {
r.raftLog.leaderTerm = m.Term // the log is now consistent with the leader
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@nvanbenschoten This should move to r.raftLog.maybeAppend, relatedly to your other comment.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Also done in the second commit.

r.send(pb.Message{To: m.From, Type: pb.MsgAppResp, Index: mlastIndex})
return
}
Expand Down Expand Up @@ -1770,7 +1773,16 @@ func (r *raft) handleAppendEntries(m pb.Message) {
}

func (r *raft) handleHeartbeat(m pb.Message) {
r.raftLog.commitTo(m.Commit)
// It is only safe to advance the commit index if our log is a prefix of the
// leader's log. Otherwise, entries at this index may mismatch.
//
// TODO(pav-kv): move this logic to r.raftLog, which is more appropriate for
// handling safety. The raftLog can use leaderTerm for other safety checks.
// For example, unstable.truncateAndAppend currently may override a suffix of
// the log unconditionally, but it can only be done if m.Term > leaderTerm.
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm +1 on these additional safety checks / assertions, but they pose potential availability risk if we get them wrong and start rejecting valid state transitions. Should we make a habit of adding logging in the cases where we drop messages, so that any bugs are observable? For example, logging on an else branch here about how we're ignoring the commit index.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Otherwise, entries at this index may mismatch.

Did you ever see a real issue or can you create a test to reproduce the issue (mismatch)?

Copy link
Contributor Author

@pav-kv pav-kv Jan 29, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@ahrtr Currently, this will not occur because the leader cuts the commit index at the follower's Progress.Match. This is explained in #138. The solution to #138 is exactly to start sending commit indices that may be out-of-bound (but handling them correctly here on the follower).

I can confirm, however, that in a test environment in CRDB I've seen out-of-bound commit indices in this line, triggering panic in r.raftLog.commitTo. This happened in a particular kind of tests in which writes on the leader or other followers are sometimes not durable. It's bad to crash this follower only because writes on other participants are not durable - we should handle it more gracefully. Esp. in CRDB case when multiple raft groups are hosted by the same process.

This PR strengthens this code so that it handles out-of-bound indices correctly, and proceeds only if it's safe.

Re @nvanbenschoten comment about logging: I don't know if we should log dropped messages. I think we should only be reporting safety properties violation - that means there is a bug in raft or some state corruption. For cases (like this) when dropped messages can be a direct result of asynchrony / races / distributed nature of the system - it should be ok to silently disregard them (moreover, raft is built with the assumption that messages can be dropped arbitrarily).

I don't expect we would be dropping any messages here currently (except in the test environment case that I described above).

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

On a second thought, we might opt to distinguish legit message drops vs safety violations. Safety violations should crash offending nodes, or have other means of appearing on radars (e.g. a raft group can be "bricked", and this can surface in monitoring). See #18.
Legit messages drops, OTOH, don't have to be loud, but we can still log them for informational purposes (and, indeed, in cases of liveness bugs logs can help with debugging).

if m.Term == r.raftLog.leaderTerm {
r.raftLog.commitTo(min(m.Commit, r.raftLog.lastIndex()))
}
r.send(pb.Message{To: m.From, Type: pb.MsgHeartbeatResp, Context: m.Context})
}

Expand All @@ -1785,6 +1797,7 @@ func (r *raft) handleSnapshot(m pb.Message) {
if r.restore(s) {
r.logger.Infof("%x [commit: %d] restored snapshot [index: %d, term: %d]",
r.id, r.raftLog.committed, sindex, sterm)
r.raftLog.leaderTerm = m.Term // the log is now consistent with the leader
r.send(pb.Message{To: m.From, Type: pb.MsgAppResp, Index: r.raftLog.lastIndex()})
} else {
r.logger.Infof("%x [commit: %d] ignored snapshot [index: %d, term: %d]",
Expand Down
18 changes: 13 additions & 5 deletions raft_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1332,16 +1332,24 @@ func TestHandleMsgApp(t *testing.T) {
func TestHandleHeartbeat(t *testing.T) {
commit := uint64(2)
tests := []struct {
m pb.Message
wCommit uint64
m pb.Message
lastTerm uint64
wCommit uint64
}{
{pb.Message{From: 2, To: 1, Type: pb.MsgHeartbeat, Term: 2, Commit: commit + 1}, commit + 1},
{pb.Message{From: 2, To: 1, Type: pb.MsgHeartbeat, Term: 2, Commit: commit - 1}, commit}, // do not decrease commit
{pb.Message{From: 2, To: 1, Type: pb.MsgHeartbeat, Term: 2, Commit: commit + 1}, 2, commit + 1},
{pb.Message{From: 2, To: 1, Type: pb.MsgHeartbeat, Term: 2, Commit: commit - 1}, 2, commit}, // do not decrease commit
{pb.Message{From: 2, To: 1, Type: pb.MsgHeartbeat, Term: 2, Commit: commit - 1}, 1, commit},

// Do not increase the commit index if the log is not guaranteed to be a
// prefix of the leader's log.
{pb.Message{From: 2, To: 1, Type: pb.MsgHeartbeat, Term: 2, Commit: commit + 1}, 1, commit},
// Do not increase the commit index beyond our log size.
{pb.Message{From: 2, To: 1, Type: pb.MsgHeartbeat, Term: 2, Commit: commit + 10}, 2, commit + 1},
}

for i, tt := range tests {
storage := newTestMemoryStorage(withPeers(1, 2))
storage.Append([]pb.Entry{{Index: 1, Term: 1}, {Index: 2, Term: 2}, {Index: 3, Term: 3}})
Copy link
Contributor Author

@pav-kv pav-kv Jan 26, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This test was incorrect previously. The third entry term is 3, while the leader Term is 2. It was incorrect to update the commit index to 3 in this case, because the leader 2 would never commit this entry.

This basically demonstrates that we fixed the unsafety. Now it's impossible for the follower to advance the commit index to 3 in this case.

storage.Append([]pb.Entry{{Index: 1, Term: 1}, {Index: 2, Term: 1}, {Index: 3, Term: tt.lastTerm}})
sm := newTestRaft(1, 5, 1, storage)
sm.becomeFollower(2, 2)
sm.raftLog.commitTo(commit)
Expand Down
Loading