test: reduce test_log_rollback flakiness

test/conftest.py

@@ -58,6 +58,8 @@ PICO_SERVICE_ID = 32
 class ErrorCode:
     Loading = 116
     Other = 10000
+    NotALeader = 10001
+    StorageCorrupted = 10002
     TermMismatch = 10003
     RaftLogUnavailable = 10004
     RaftLogCompacted = 10005
@@ -71,6 +73,23 @@ class ErrorCode:
     LeaderUnknown = 10018
     PluginError = 10019
 
+    # Make sure this matches this list in picoplugin::error_code::ErrorCode::is_retriable_for_cas
+    retriable_for_cas = set(
+        [
+            LeaderUnknown,
+            NotALeader,
+            TermMismatch,
+            RaftLogCompacted,
+            RaftLogUnavailable,
+            CasEntryTermMismatch,
+            CasConflictFound,
+        ]
+    )
+
+    @classmethod
+    def is_retriable_for_cas(cls, code):
+        return code in cls.retriable_for_cas
+
 
 def eprint(*args, **kwargs):
     print(*args, file=sys.stderr, **kwargs)

test/int/test_network_effects.py

 import pytest
 import time
 
-from conftest import Cluster, Instance, TarantoolError, Retriable
+from conftest import Cluster, Instance, TarantoolError, Retriable, ErrorCode
 
 
 @pytest.fixture
@@ -65,12 +65,32 @@ def test_log_rollback(cluster3: Cluster):
 
     key = 0
 
-    def propose_state_change(srv: Instance, value):
+    def propose_state_change(srv: Instance, value, timeout=10):
+        deadline = time.time() + timeout
         nonlocal key
-        index = cluster3.cas(
-            "insert", "_pico_property", (f"check{key}", value), instance=srv
-        )
-        srv.raft_wait_index(index)
+        while True:
+            try:
+                index = cluster3.cas(
+                    "insert",
+                    "_pico_property",
+                    (f"check{key}", value),
+                    instance=srv,
+                )
+                srv.raft_wait_index(index)
+                break
+            except TarantoolError as e:
+                print(f"\x1b[33m### CaS error: {e}", end="")
+                if time.time() > deadline:
+                    print(", timeout\x1b[0m")
+                    raise
+
+                if ErrorCode.is_retriable_for_cas(e.args[0]):
+                    print(", retrying...\x1b[0m")
+                    time.sleep(0.1)
+                    continue
+
+                print(", failure!\x1b[0m")
+                raise
         key += 1
 
     propose_state_change(i1, "i1 is a leader")