From 9277dc9e89e0643aed4bb4bd05858981aa136746 Mon Sep 17 00:00:00 2001 From: Alex Lutay <1928266+taurus-forever@users.noreply.github.com> Date: Fri, 3 Jul 2026 00:48:32 +0000 Subject: [PATCH] Persist raft currentTerm and votedFor to fix data divergence on simultaneous restart MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The raft paper (§5, "persistent state on all servers") requires currentTerm and votedFor to survive restarts. PySyncObj persisted only the log entries and raftCommitIndex, resetting currentTerm to 0 and votedFor to None on every start. When ALL cluster members restart simultaneously (e.g. a deployment or host reboot of a Patroni cluster using the raft DCS), the term space collapses: every node comes back with currentTerm=0 while its journal still holds entries with high terms from the previous run. This breaks the term monotonicity assumption behind the vote "up-to-dateness" check and allows committed data to be silently overwritten: 1. Pre-restart: log through idx 42, last term 5, all committed and applied. 2. All 3 members restart simultaneously. Member A wins the first election at term 1 and immediately-issued writes (Patroni writes its DCS keys on startup) become entries idx 43-45 with term 1. Member C acks them: majority -> committed and applied on A and C. 3. Member B, slightly slower to start, received nothing yet. Its election deadline fires and it campaigns at term 2 with last_log_term=5 - its stale pre-restart log looks MORE up-to-date than A/C's freshly committed term-1 entries, so A and C grant their votes. 4. B becomes leader and its append_entries (prevLogIdx=42, prevLogTerm=5) make A and C truncate the committed-and-applied entries 43-45. In addition, raftLastApplied is never rewound, so B's replacement entries at the reused indexes 44-45 are silently skipped on A and C. Result: the raft log converges and all future writes replicate fine, but the state machines differ permanently between members - A and C keep applied writes that B never applies, and miss B's entries at the reused indexes. For Patroni this manifests as different DCS data on different members while raft reports being in sync. A genuinely fresh first start is not affected (no high terms on disk yet), and a single member that did not restart re-synchronizes the terms from memory - only a simultaneous (re)start of members with existing raft data is vulnerable. Fix: - Journal gains setCurrentTerm(term, votedForNodeId) / getCurrentTerm() / getVotedForNodeId(). FileJournal stores both in the existing .meta file synchronously: raft requires term/vote to be durable before any message reflecting them is sent. Term changes only happen during elections, so the extra write is negligible. Existing .meta files without the new keys default to 0/None (backward compatible; the file stays readable by older versions since they ignore unknown keys). - SyncObj restores the state on startup. currentTerm is restored as max(persisted term, last log term) - currentTerm is always >= the term of the last log entry, so journals written by older versions (without a persisted term) are also protected after an upgrade. - All four term/vote mutation points (candidacy, vote grant, higher-term adoption in the request_vote and append_entries handlers) go through a helper that persists before the corresponding message is sent. - Defense in depth (should be unreachable with the fix): log CRITICAL when a truncation cuts below raftLastApplied (previously silent state-machine divergence), and clamp raftCommitIndex to the log end before persisting it, so a later restart cannot replay entries from a different leader epoch as already-committed. Tests: - test_simultaneousRestartDataDivergence deterministically reproduces the scenario above (fails with "assert None == 'post'" on the slow member without the fix; with the fix the stale member loses the election and receives the data instead). - test_journalKeepsCurrentTermAndVote covers persistence and legacy-meta defaults. Close: https://github.com/bakwc/PySyncObj/issues/201 Assisted-by: Claude:claude-5-fable --- pysyncobj/journal.py | 35 +++++++++++++++ pysyncobj/syncobj.py | 33 +++++++++++--- test_syncobj.py | 105 +++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 166 insertions(+), 7 deletions(-) diff --git a/pysyncobj/journal.py b/pysyncobj/journal.py index 44d5bd7..b11f9ae 100644 --- a/pysyncobj/journal.py +++ b/pysyncobj/journal.py @@ -35,6 +35,15 @@ def setRaftCommitIndex(self, raftCommitIndex): def getRaftCommitIndex(self): raise NotImplementedError + def setCurrentTerm(self, term, votedForNodeId): + raise NotImplementedError + + def getCurrentTerm(self): + raise NotImplementedError + + def getVotedForNodeId(self): + raise NotImplementedError + def onOneSecondTimer(self): pass @@ -45,6 +54,8 @@ def __init__(self): self.__journal = [] self.__bytesSize = 0 self.__lastCommitIndex = 0 + self.__currentTerm = 0 + self.__votedForNodeId = None def add(self, command, idx, term): self.__journal.append((command, idx, term)) @@ -73,6 +84,16 @@ def setRaftCommitIndex(self, raftCommitIndex): def getRaftCommitIndex(self): return 1 + def setCurrentTerm(self, term, votedForNodeId): + self.__currentTerm = term + self.__votedForNodeId = votedForNodeId + + def getCurrentTerm(self): + return self.__currentTerm + + def getVotedForNodeId(self): + return self.__votedForNodeId + class ResizableFile(object): @@ -246,6 +267,20 @@ def setRaftCommitIndex(self, raftCommitIndex): def getRaftCommitIndex(self): return self.__meta.get('raftCommitIndex', 1) + def setCurrentTerm(self, term, votedForNodeId): + self.__meta['currentTerm'] = term + self.__meta['votedForNodeId'] = votedForNodeId + # Raft requires currentTerm and votedFor to be durable before any message + # reflecting them (vote request/grant) is sent, so store synchronously. + self.__metaStorer.storeMeta(self.__meta) + self.__metaSaved = True + + def getCurrentTerm(self): + return self.__meta.get('currentTerm', 0) + + def getVotedForNodeId(self): + return self.__meta.get('votedForNodeId', None) + def onOneSecondTimer(self): if not self.__metaSaved: self.__metaStorer.storeMeta(self.__meta) diff --git a/pysyncobj/syncobj.py b/pysyncobj/syncobj.py index dddefb0..d25fdeb 100644 --- a/pysyncobj/syncobj.py +++ b/pysyncobj/syncobj.py @@ -163,6 +163,13 @@ def __init__(self, selfNode, otherNodes, conf=None, consumers=None, nodeClass = self.__raftLog = createJournal(self.__conf.journalFile) if len(self.__raftLog) == 0: self.__raftLog.add(_bchr(_COMMAND_TYPE.NO_OP), 1, self.__raftCurrentTerm) + # Restore raft persistent state (see raft paper §5 - currentTerm and votedFor + # must survive restarts, otherwise already-committed entries may be overwritten + # by a stale member after a simultaneous restart of the cluster). The max() with + # the last log term covers journals written by versions that did not store the + # term: currentTerm is always >= the term of the last log entry. + self.__raftCurrentTerm = max(self.__raftLog.getCurrentTerm(), self.__getCurrentLogTerm()) + self.__votedForNodeId = self.__raftLog.getVotedForNodeId() self.__raftCommitIndex = self.__raftLog.getRaftCommitIndex() self.__raftLastApplied = 1 self.__raftNextIndex = {} @@ -581,8 +588,7 @@ def _onTick(self, timeToWait=0.0): self.__raftElectionDeadline = monotonicTime() + self.__generateRaftTimeout() self.__raftLeader = None self.__setState(_RAFT_STATE.CANDIDATE) - self.__raftCurrentTerm += 1 - self.__votedForNodeId = self.__selfNode.id + self.__setCurrentTerm(self.__raftCurrentTerm + 1, self.__selfNode.id) self.__votesCount = 1 for node in self.__otherNodes: self.__transport.send(node, { @@ -856,8 +862,7 @@ def __onMessageReceived(self, node, message): if message['type'] == 'request_vote' and self.__selfNode is not None: if message['term'] > self.__raftCurrentTerm: - self.__raftCurrentTerm = message['term'] - self.__votedForNodeId = None + self.__setCurrentTerm(message['term'], None) self.__setState(_RAFT_STATE.FOLLOWER) self.__raftLeader = None @@ -873,7 +878,7 @@ def __onMessageReceived(self, node, message): if self.__votedForNodeId is not None: return - self.__votedForNodeId = node.id + self.__setCurrentTerm(self.__raftCurrentTerm, node.id) self.__raftElectionDeadline = monotonicTime() + self.__generateRaftTimeout() self.__transport.send(node, { @@ -887,8 +892,7 @@ def __onMessageReceived(self, node, message): self.__onLeaderChanged() self.__raftLeader = node if message['term'] > self.__raftCurrentTerm: - self.__raftCurrentTerm = message['term'] - self.__votedForNodeId = None + self.__setCurrentTerm(message['term'], None) self.__setState(_RAFT_STATE.FOLLOWER) newEntries = message.get('entries', []) serialized = message.get('serialized', None) @@ -930,6 +934,11 @@ def __onMessageReceived(self, node, message): if clusterChangeRequest is not None: self.__doChangeCluster(clusterChangeRequest, reverse=True) + if prevLogIdx + 1 <= self.__raftLastApplied: + logger.critical( + 'truncating already-applied log entries from index %d ' + '(last applied: %d) - state machine may have diverged from ' + 'other cluster members', prevLogIdx + 1, self.__raftLastApplied) self.__deleteEntriesFrom(prevLogIdx + 1) for entry in newEntries: self.__raftLog.add(*entry) @@ -953,6 +962,11 @@ def __onMessageReceived(self, node, message): self.__loadDumpFile(clearJournal=True) self.__sendNextNodeIdx(node, success=True) + # If the log was truncated below the commit index, don't persist a commit + # index covering entries that are no longer the committed ones. + if self.__raftCommitIndex > self.__getCurrentLogIndex(): + self.__raftCommitIndex = self.__getCurrentLogIndex() + if leaderCommitIndex > self.__raftCommitIndex: self.__raftCommitIndex = min(leaderCommitIndex, self.__getCurrentLogIndex()) @@ -1155,6 +1169,11 @@ def __onBecomeLeader(self): self.__sendAppendEntries() + def __setCurrentTerm(self, term, votedForNodeId): + self.__raftCurrentTerm = term + self.__votedForNodeId = votedForNodeId + self.__raftLog.setCurrentTerm(term, votedForNodeId) + def __setState(self, newState): oldState = self.__raftState self.__raftState = newState diff --git a/test_syncobj.py b/test_syncobj.py index 5d1dbf2..24fa932 100755 --- a/test_syncobj.py +++ b/test_syncobj.py @@ -849,6 +849,84 @@ def test_randomTest1(): removeFiles([e + '.meta' for e in journalFiles]) +def test_simultaneousRestartDataDivergence(): + # If raft persistent state (currentTerm, votedFor) is lost on restart, then after a + # simultaneous restart of all members the terms restart from 1 while journals still + # contain entries with higher terms. A member that starts slightly later can then + # out-vote the new leader with its stale (but high-term) log and truncate entries + # that were already committed and applied by the other members, permanently + # diverging their state machines while the raft log stays in sync. + random.seed(42) + + a = [getNextAddr(), getNextAddr(), getNextAddr()] + journalFiles = [getNextJournalFile(), getNextJournalFile(), getNextJournalFile()] + removeFiles(journalFiles) + removeFiles([e + '.meta' for e in journalFiles]) + + def createObj(i): + return TestObj(a[i], [a[j] for j in range(3) if j != i], + TEST_TYPE.JOURNAL_1, journalFile=journalFiles[i]) + + # Phase 1: build some history with lastLogTerm >= 2, so that the terms of a fresh + # post-restart election are lower than the terms already stored in the journals. + objs = [createObj(0), createObj(1), createObj(2)] + doTicks(objs, 10.0, stopFunc=lambda: all(o._isReady() for o in objs)) + assert all(o._isReady() for o in objs) + + # force a re-election to bump the term above 1 + oldTerm = max(o._getTerm() for o in objs) + follower = [o for o in objs if not o._isLeader()][0] + follower._SyncObj__raftElectionDeadline = 0 + doTicks(objs, 10.0, stopFunc=lambda: max(o._getTerm() for o in objs) > oldTerm and + any(o._isLeader() for o in objs)) + assert max(o._getTerm() for o in objs) >= 2 + + objs[0].addKeyValue('k1', 'pre') + doTicks(objs, 10.0, stopFunc=lambda: all(o.getValue('k1') == 'pre' for o in objs)) + assert all(o.getValue('k1') == 'pre' for o in objs) + + # let the journal meta be stored (it is saved on a one-second timer), then stop all + doTicks(objs, 2.5) + for o in objs: + o._destroy() + time.sleep(0.1) + + # Phase 2: members 0 and 2 restart quickly, member 1 is slow to start. + o0 = createObj(0) + o2 = createObj(2) + doTicks([o0, o2], 10.0, stopFunc=lambda: o0._isLeader() or o2._isLeader()) + assert o0._isLeader() or o2._isLeader() + + # a write is committed and applied by the two started members + (o0 if o0._isLeader() else o2).addKeyValue('w1', 'post') + doTicks([o0, o2], 10.0, stopFunc=lambda: o0.getValue('w1') == 'post' and + o2.getValue('w1') == 'post') + assert o0.getValue('w1') == 'post' + assert o2.getValue('w1') == 'post' + + # Phase 3: member 1 finally starts and tries to become leader. Member 0 (the + # current leader) is frozen (not ticked) so member 1 only needs member 2's vote, + # which makes the race deterministic. With a stale log whose last term is higher + # than the new entries' term it must NOT win; member 2 (with the up-to-date log) + # must win instead and replicate the data to member 1. + o1 = createObj(1) + doTicks([o1, o2], 15.0, stopFunc=lambda: o1._isLeader() or o1.getValue('w1') == 'post') + + # Everybody ticks again; the cluster must converge with no data loss. + objs = [o0, o1, o2] + doTicks(objs, 20.0, stopFunc=lambda: all(o.getValue('w1') == 'post' for o in objs)) + + for o in objs: + assert o.getValue('k1') == 'pre' + assert o.getValue('w1') == 'post' + + for o in objs: + o._destroy() + time.sleep(0.1) + removeFiles(journalFiles) + removeFiles([e + '.meta' for e in journalFiles]) + + # Ensure that raftLog after serialization is the same as in serialized data def test_logCompactionRegressionTest1(): random.seed(42) @@ -1171,6 +1249,33 @@ def test_journalTest2(): removeFiles([e + '.meta' for e in journalFiles]) +def test_journalKeepsCurrentTermAndVote(): + journalFiles = [getNextJournalFile()] + removeFiles(journalFiles) + removeFiles([e + '.meta' for e in journalFiles]) + + journal = createJournal(journalFiles[0]) + # defaults for a fresh (or legacy, pre-term-persistence) journal + assert journal.getCurrentTerm() == 0 + assert journal.getVotedForNodeId() is None + journal.setCurrentTerm(5, 'localhost:1234') + journal._destroy() + + journal = createJournal(journalFiles[0]) + assert journal.getCurrentTerm() == 5 + assert journal.getVotedForNodeId() == 'localhost:1234' + journal.setCurrentTerm(6, None) + journal._destroy() + + journal = createJournal(journalFiles[0]) + assert journal.getCurrentTerm() == 6 + assert journal.getVotedForNodeId() is None + journal._destroy() + + removeFiles(journalFiles) + removeFiles([e + '.meta' for e in journalFiles]) + + def test_applyJournalAfterRestart(): dumpFiles = [getNextDumpFile(), getNextDumpFile()] journalFiles = [getNextJournalFile(), getNextJournalFile()]