fix(recipe): prevent perpetual re-election if sequence node overflows

Once sequence node is incremented beyond 2147483647, the sequence
overflows into the negative space. When getting predecessors, cast
the node sequence from String to int before comparing.

Closes #730

Author: angxiang

kazoo/recipe/lock.py

@@ -276,15 +276,15 @@ def _get_predecessor(self, node):
         (e.g. rlock), this and also edge cases where the lock's ephemeral node
         is gone.
         """
-        node_sequence = node[len(self.prefix) :]
+        node_sequence = int(node[len(self.prefix) :])
         children = self.client.get_children(self.path)
         found_self = False
         # Filter out the contenders using the computed regex
         contender_matches = []
         for child in children:
             match = self._contenders_re.search(child)
             if match is not None:
-                contender_sequence = match.group(1)
+                contender_sequence = int(match.group(1))
                 # Only consider contenders with a smaller sequence number.
                 # A contender with a smaller sequence number has a higher
                 # priority.

kazoo/tests/test_lock.py

@@ -799,6 +799,46 @@ def _thread(sem, event, timeout):
 
 class TestSequence(unittest.TestCase):
     def test_get_predecessor(self):
+        """Validate selection of predecessors."""
+        pyLock_prefix = "514e5a831836450cb1a56c741e990fd8__lock__"
+        pyLock_predecessor = f"{pyLock_prefix}0000000030"
+        pyLock = f"{pyLock_prefix}0000000031"
+        pyLock_successor = f"{pyLock_prefix}0000000032"
+        children = [
+            "hello",
+            pyLock_predecessor,
+            "world",
+            pyLock,
+            pyLock_successor,
+        ]
+        client = MagicMock()
+        client.get_children.return_value = children
+        lock = Lock(client, "test")
+        assert lock._get_predecessor(pyLock) == pyLock_predecessor
+
+    def test_get_predecessor_with_overflowed_sequence(self):
+        """Validate selection of predecessors with negative sequence.
+
+        This can occur in case of an integer overflow, if the sequence
+        counter is incremented beyond 2147483647.
+        """
+        pyLock_prefix = "514e5a831836450cb1a56c741e990fd8__lock__"
+        pyLock_predecessor = f"{pyLock_prefix}-0000000032"
+        pyLock = f"{pyLock_prefix}-0000000031"
+        pyLock_successor = f"{pyLock_prefix}-0000000030"
+        children = [
+            "hello",
+            pyLock_predecessor,
+            "world",
+            pyLock,
+            pyLock_successor,
+        ]
+        client = MagicMock()
+        client.get_children.return_value = children
+        lock = Lock(client, "test")
+        assert lock._get_predecessor(pyLock) == pyLock_predecessor
+
+    def test_get_predecessor_no_predecessor(self):
         """Validate selection of predecessors."""
         goLock = "_c_8eb60557ba51e0da67eefc47467d3f34-lock-0000000031"
         pyLock = "514e5a831836450cb1a56c741e990fd8__lock__0000000032"