Repository: bakwc/PySyncObj

Issue:
Raft log not appying after restart
When using journal file as a storage - all operations should be applied after node restarts, without waiting connection with other nodes.

Discussion:

Patch:
diff --git a/pysyncobj/journal.py b/pysyncobj/journal.py
index 2ae88a3..820a712 100644
--- a/pysyncobj/journal.py
+++ b/pysyncobj/journal.py
@@ -3,7 +3,7 @@ import mmap
 import struct
 
 from .version import VERSION
-from .pickle import to_bytes
+from .pickle import to_bytes, loads, dumps
 
 class Journal(object):
 
@@ -28,12 +28,22 @@ class Journal(object):
     def _destroy(self):
         raise NotImplementedError
 
+    def setRaftCommitIndex(self, raftCommitIndex):
+        raise NotImplementedError
+
+    def getRaftCommitIndex(self):
+        raise NotImplementedError
+
+    def onOneSecondTimer(self):
+        pass
+
 
 class MemoryJournal(Journal):
 
     def __init__(self):
         self.__journal = []
         self.__bytesSize = 0
+        self.__lastCommitIndex = 0
 
     def add(self, command, idx, term):
         self.__journal.append((command, idx, term))
@@ -56,6 +66,13 @@ class MemoryJournal(Journal):
     def _destroy(self):
         pass
 
+    def setRaftCommitIndex(self, raftCommitIndex):
+        pass
+
+    def getRaftCommitIndex(self):
+        return 1
+
+
 
 class ResizableFile(object):
 
@@ -105,6 +122,27 @@ class ResizableFile(object):
         self.__mm.flush()
 
 
+class MetaStorer(object):
+    def __init__(self, path):
+        self.__path = path
+
+    def getMeta(self):
+        meta = {}
+        try:
+            meta = loads(open(self.__path, 'rb').read())
+        except:
+            pass
+        return meta
+
+    def storeMeta(self, meta):
+        with open(self.__path + '.tmp', 'wb') as f:
+            f.write(dumps(meta))
+            f.flush()
+        os.rename(self.__path + '.tmp', self.__path)
+
+    def getPath(self):
+        return self.__path
+
 
 JOURNAL_FORMAT_VERSION = 1
 APP_NAME = b'PYSYNCOBJ'
@@ -128,6 +166,9 @@ class FileJournal(Journal):
     def __init__(self, journalFile):
         self.__journalFile = ResizableFile(journalFile, defaultContent=self.__getDefaultHeader())
         self.__journal = []
+        self.__metaStorer = MetaStorer(journalFile + '.meta')
+        self.__meta = self.__metaStorer.getMeta()
+        self.__metaSaved = True
         currentOffset = FIRST_RECORD_OFFSET
         lastRecordOffset = self.__getLastRecordOffset()
         while currentOffset < lastRecordOffset:
@@ -197,6 +238,19 @@ class FileJournal(Journal):
     def flush(self):
         self.__journalFile.flush()
 
+    def setRaftCommitIndex(self, raftCommitIndex):
+        self.__meta['raftCommitIndex'] = raftCommitIndex
+        self.__metaSaved = False
+
+    def getRaftCommitIndex(self):
+        return self.__meta.get('raftCommitIndex', 1)
+
+    def onOneSecondTimer(self):
+        if not self.__metaSaved:
+            self.__metaStorer.storeMeta(self.__meta)
+            self.__metaSaved = True
+
+
 def createJournal(journalFile = None):
     if journalFile is None:
         return MemoryJournal()
diff --git a/pysyncobj/node.py b/pysyncobj/node.py
index 0396bd6..bebe4f7 100644
--- a/pysyncobj/node.py
+++ b/pysyncobj/node.py
@@ -66,29 +66,13 @@ class TCPNode(Node):
         """
 
         super(TCPNode, self).__init__(address, **kwargs)
-        self.__address = address
-        self.__host, port = address.rsplit(':', 1)
-        self.__port = int(port)
-        self.__ip = globalDnsResolver().resolve(self.host)
-
-    @property
-    def address(self):
-        return self.__address
-
-    @property
-    def host(self):
-        return self.__host
-
-    @property
-    def port(self):
-        return self.__port
-
-    @property
-    def ip(self):
-        return self.__ip
+        self.address = address
+        self.host, port = address.rsplit(':', 1)
+        self.port = int(port)
+        self.ip = globalDnsResolver().resolve(self.host)
 
     def __repr__(self):
         v = vars(self)
-        filtered = ['_id', '_TCPNode__address', '_TCPNode__host', '_TCPNode__port', '_TCPNode__ip']
+        filtered = ['_id', 'address', 'host', 'port', 'ip']
         formatted = ['{} = {}'.format(key, repr(v[key])) for key in v if key not in filtered]
         return '{}({}{})'.format(type(self).__name__, repr(self.id), (', ' + ', '.join(formatted)) if len(formatted) else '')
diff --git a/pysyncobj/syncobj.py b/pysyncobj/syncobj.py
index a12661a..ddf44cc 100644
--- a/pysyncobj/syncobj.py
+++ b/pysyncobj/syncobj.py
@@ -161,7 +161,7 @@ class SyncObj(object):
         self.__raftLog = createJournal(self.__conf.journalFile)
         if len(self.__raftLog) == 0:
             self.__raftLog.add(_bchr(_COMMAND_TYPE.NO_OP), 1, self.__raftCurrentTerm)
-        self.__raftCommitIndex = 1
+        self.__raftCommitIndex = self.__raftLog.getRaftCommitIndex()
         self.__raftLastApplied = 1
         self.__raftNextIndex = {}
         self.__lastResponseTime = {}
@@ -180,6 +180,7 @@ class SyncObj(object):
         self.__onTickCallbacksLock = threading.Lock()
 
         self.__startTime = monotonicTime()
+        self.__numOneSecondDumps = 0
         globalDnsResolver().setTimeouts(self.__conf.dnsCacheTime, self.__conf.dnsFailCacheTime)
         globalDnsResolver().setPreferredAddrFamily(self.__conf.preferredAddrType)
         self.__serializer = Serializer(self.__conf.fullDumpFile,
@@ -540,6 +541,7 @@ class SyncObj(object):
 
         if not self.__transport.ready:
             time.sleep(timeToWait)
+            self.__applyLogEntries()
             return
 
         if self.__needLoadDumpFile:
@@ -547,6 +549,11 @@ class SyncObj(object):
                 self.__loadDumpFile(clearJournal=False)
             self.__needLoadDumpFile = False
 
+        workTime = monotonicTime() - self.__startTime
+        if workTime > self.__numOneSecondDumps:
+            self.__numOneSecondDumps += 1
+            self.__raftLog.onOneSecondTimer()
+
         if self.__raftState in (_RAFT_STATE.FOLLOWER, _RAFT_STATE.CANDIDATE) and self.__selfNode is not None:
             if self.__raftElectionDeadline < monotonicTime() and self.__connectedToAnyone():
                 self.__raftElectionDeadline = monotonicTime() + self.__generateRaftTimeout()
@@ -575,6 +582,7 @@ class SyncObj(object):
                         count += 1
                 if count > (len(self.__otherNodes) + 1) / 2:
                     self.__raftCommitIndex = nextCommitIndex
+                    self.__raftLog.setRaftCommitIndex(self.__raftCommitIndex)
                 else:
                     break
             self.__leaderCommitIndex = self.__raftCommitIndex
@@ -587,6 +595,27 @@ class SyncObj(object):
                 self.__setState(_RAFT_STATE.FOLLOWER)
                 self.__raftLeader = None
 
+        needSendAppendEntries = self.__applyLogEntries()
+
+        if self.__raftState == _RAFT_STATE.LEADER:
+            if monotonicTime() > self.__newAppendEntriesTime or needSendAppendEntries:
+                self.__sendAppendEntries()
+
+        if not self.__onReadyCalled and self.__raftLastApplied == self.__leaderCommitIndex:
+            if self.__conf.onReady:
+                self.__conf.onReady()
+            self.__onReadyCalled = True
+
+        self._checkCommandsToApply()
+        self.__tryLogCompaction()
+
+        with self.__onTickCallbacksLock:
+            for callback in self.__onTickCallbacks:
+                callback()
+
+        self._poller.poll(timeToWait)
+
+    def __applyLogEntries(self):
         needSendAppendEntries = False
 
         if self.__raftCommitIndex > self.__raftLastApplied:
@@ -605,30 +634,14 @@ class SyncObj(object):
 
                     self.__raftLastApplied += 1
                 except SyncObjExceptionWrongVer as e:
-                    logging.error('request to switch to unsupported code version (self version: %d, requested version: %d)' %
+                    logging.error(
+                        'request to switch to unsupported code version (self version: %d, requested version: %d)' %
                         (self.__selfCodeVersion, e.ver))
 
-
             if not self.__conf.appendEntriesUseBatch:
                 needSendAppendEntries = True
 
-        if self.__raftState == _RAFT_STATE.LEADER:
-            if monotonicTime() > self.__newAppendEntriesTime or needSendAppendEntries:
-                self.__sendAppendEntries()
-
-        if not self.__onReadyCalled and self.__raftLastApplied == self.__leaderCommitIndex:
-            if self.__conf.onReady:
-                self.__conf.onReady()
-            self.__onReadyCalled = True
-
-        self._checkCommandsToApply()
-        self.__tryLogCompaction()
-
-        with self.__onTickCallbacksLock:
-            for callback in self.__onTickCallbacks:
-                callback()
-
-        self._poller.poll(timeToWait)
+        return needSendAppendEntries
 
     def addOnTickCallback(self, callback):
         with self.__onTickCallbacksLock:
@@ -642,76 +655,25 @@ class SyncObj(object):
                 # callback not in list, ignore
                 pass
 
-    def isNodeConnected(self, node):
-        """
-        Checks if the given node is connected
-        :param node: node to check
-        :type node: Node
-        :rtype: bool
-        """
-        return node in self.__connectedNodes
-
-    @property
-    def selfNode(self):
-        """
-        :rtype: Node
-        """
-        return self.__selfNode
-
-    @property
-    def otherNodes(self):
-        """
-        :rtype: set of Node
-        """
-        return self.__otherNodes.copy()
-
-    @property
-    def readonlyNodes(self):
-        """
-        :rtype: set of Node
-        """
-        return self.__readonlyNodes.copy()
-
-    @property
-    def raftLastApplied(self):
-        """
-        :rtype: int
-        """
-        return self.__raftLastApplied
-
-    @property
-    def raftCommitIndex(self):
-        """
-        :rtype: int
-        """
-        return self.__raftCommitIndex
-
-    @property
-    def raftCurrentTerm(self):
-        """
-        :rtype: int
-        """
-        return self.__raftCurrentTerm
-
     def getStatus(self):
         """Dumps different debug info about cluster to dict and return it"""
 
         status = {}
         status['version'] = VERSION
         status['revision'] = 'deprecated'
-        status['self'] = self.selfNode
+        status['self'] = self.__selfNode
         status['state'] = self.__raftState
         status['leader'] = self.__raftLeader
         status['partner_nodes_count'] = len(self.__otherNodes)
         for node in self.__otherNodes:
-            status['partner_node_status_server_' + node.id] = 2 if self.isNodeConnected(node) else 0
+            status['partner_node_status_server_' + node.id] = 2 if node in self.__connectedNodes else 0
         status['readonly_nodes_count'] = len(self.__readonlyNodes)
         for node in self.__readonlyNodes:
-            status['readonly_node_status_server_' + node.id] = 2 if self.isNodeConnected(node) else 0
+            status['readonly_node_status_server_' + node.id] = 2 if node in self.__connectedNodes else 0
         status['log_len'] = len(self.__raftLog)
-        status['last_applied'] = self.raftLastApplied
-        status['commit_idx'] = self.raftCommitIndex
-        status['raft_term'] = self.raftCurrentTerm
+        status['last_applied'] = self.__raftLastApplied
+        status['commit_idx'] = self.__raftCommitIndex
+        status['raft_term'] = self.__raftCurrentTerm
         status['next_node_idx_count'] = len(self.__raftNextIndex)
         for node, idx in iteritems(self.__raftNextIndex):
             status['next_node_idx_server_' + node.id] = idx
@@ -877,6 +839,7 @@ class SyncObj(object):
                     self.__sendNextNodeIdx(node, success=True)
 
             self.__raftCommitIndex = min(leaderCommitIndex, self.__getCurrentLogIndex())
+            self.__raftLog.setRaftCommitIndex(self.__raftCommitIndex)
 
         if message['type'] == 'apply_command':
             if 'request_id' in message:
@@ -1221,7 +1184,7 @@ class SyncObj(object):
         if adding:
             newNode = requestNode
             # Node already exists in cluster
-            if newNode == self.__selfNode or newNode in self.__otherNodes:
+            if newNode is self.__selfNode or newNode in self.__otherNodes:
                 return False
             self.__otherNodes.add(newNode)
             self.__raftNextIndex[newNode] = self.__getCurrentLogIndex() + 1
@@ -1232,7 +1195,7 @@ class SyncObj(object):
             return True
         else:
             oldNode = requestNode
-            if oldNode == self.__selfNode:
+            if oldNode is self.__selfNode:
                 return False
             if oldNode not in self.__otherNodes:
                 return False
diff --git a/pysyncobj/utility.py b/pysyncobj/utility.py
index 7a00dd4..e82cfbe 100644
--- a/pysyncobj/utility.py
+++ b/pysyncobj/utility.py
@@ -54,6 +54,12 @@ class TcpUtility(Utility):
         self.__error = None
 
     def executeCommand(self, node, command):
+        """
+        Executes command on the given node.
+        :param node: where to execute the command
+        :type node: Node or str
+        """
+
         self.__result = None
         self.__error = None
 
@@ -64,7 +70,7 @@ class TcpUtility(Utility):
                 self.__error = 'invalid address to connect'
                 return
 
-        self.__isConnected = self.__connection.connect(node.ip, node.port)
+        self.__isConnected = self.__connection.connect(node.host, node.port)
         if not self.__isConnected:
             self.__error = "can't connected"
             return


Test patch:
diff --git a/test_syncobj.py b/test_syncobj.py
index 1fb6c6d..d1abf3a 100755
--- a/test_syncobj.py
+++ b/test_syncobj.py
@@ -120,6 +120,7 @@ class TestObj(SyncObj):
 
     @replicated
     def addValue(self, value):
+        print(" === CALL ADD VALUE ===", value)
         self.__counter += value
         return self.__counter
 
@@ -714,6 +715,7 @@ def _checkSameLeader2(objs):
 def test_randomTest1():
     journalFiles = [getNextJournalFile(), getNextJournalFile(), getNextJournalFile()]
     removeFiles(journalFiles)
+    removeFiles([e + '.meta' for e in journalFiles])
 
     random.seed(12)
 
@@ -784,6 +786,7 @@ def test_randomTest1():
         raise AssertionError('Values not equal')
 
     removeFiles(journalFiles)
+    removeFiles([e + '.meta' for e in journalFiles])
 
 
 # Ensure that raftLog after serialization is the same as in serialized data
@@ -987,6 +990,7 @@ def test_journalTest1():
     journalFiles = [getNextJournalFile(), getNextJournalFile()]
     removeFiles(dumpFiles)
     removeFiles(journalFiles)
+    removeFiles([e + '.meta' for e in journalFiles])
 
     random.seed(42)
 
@@ -1070,11 +1074,13 @@ def test_journalTest1():
 
     removeFiles(dumpFiles)
     removeFiles(journalFiles)
+    removeFiles([e + '.meta' for e in journalFiles])
 
 
 def test_journalTest2():
     journalFiles = [getNextJournalFile()]
     removeFiles(journalFiles)
+    removeFiles([e + '.meta' for e in journalFiles])
 
     removeFiles(journalFiles)
     journal = createJournal(journalFiles[0])
@@ -1102,6 +1108,54 @@ def test_journalTest2():
     assert journal[0] == (b'cmd2', 2, 0)
     journal._destroy()
     removeFiles(journalFiles)
+    removeFiles([e + '.meta' for e in journalFiles])
+
+
+def test_applyJournalAfterRestart():
+    dumpFiles = [getNextDumpFile(), getNextDumpFile()]
+    journalFiles = [getNextJournalFile(), getNextJournalFile()]
+    removeFiles(dumpFiles)
+    removeFiles(journalFiles)
+    removeFiles([e + '.meta' for e in journalFiles])
+
+    random.seed(42)
+
+    a = [getNextAddr(), getNextAddr()]
+
+    o1 = TestObj(a[0], [a[1]], TEST_TYPE.JOURNAL_1, dumpFile=dumpFiles[0], journalFile=journalFiles[0])
+    o2 = TestObj(a[1], [a[0]], TEST_TYPE.JOURNAL_1, dumpFile=dumpFiles[1], journalFile=journalFiles[1])
+    objs = [o1, o2]
+    doTicks(objs, 10, stopFunc=lambda: o1._isReady() and o2._isReady())
+
+    assert o1._getLeader().address in a
+    assert o1._getLeader() == o2._getLeader()
+
+    o1.addValue(150)
+    o2.addValue(200)
+
+    doTicks(objs, 10, stopFunc=lambda: o1.getCounter() == 350 and o2.getCounter() == 350)
+
+    assert o1.getCounter() == 350
+    assert o2.getCounter() == 350
+    doTicks(objs, 2)
+
+    o1._destroy()
+    o2._destroy()
+
+    del o1
+    del o2
+
+    removeFiles(dumpFiles)
+
+    o1 = TestObj(a[0], [a[1]], TEST_TYPE.JOURNAL_1, dumpFile=dumpFiles[0], journalFile=journalFiles[0])
+    objs = [o1]
+    doTicks(objs, 10, o1.getCounter() == 350)
+
+    assert o1.getCounter() == 350
+
+    removeFiles(dumpFiles)
+    removeFiles(journalFiles)
+    removeFiles([e + '.meta' for e in journalFiles])
 
 
 def test_autoTick1():