Test: Better persistence retry mechanism, and cover more cases

test/broker/persist_module_helper.py

@@ -8,6 +8,26 @@ import time
 from typing import Any, Optional
 from types import ModuleType
 
+import time
+from functools import wraps
+
+def retry(retries=5, delay=1):
+    def decorator(func):
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            last_exception = None
+            for attempt in range(retries):
+                try:
+                    return func(*args, **kwargs)
+                except Exception as e:
+                    print(f"Retrying {func.__name__} {attempt}/{retries}")
+                    last_exception = e
+                    if attempt < retries - 1:
+                        time.sleep(delay)
+            raise last_exception
+        return wrapper
+    return decorator
+
 
 def connect_client(
     port: int,
@@ -77,7 +97,7 @@ def publish_messages(
         mosq_test.do_send_receive(sock, publish_packet, puback_packet, "puback")
 
 
-def check_db_once(
+def check_db(
     persist_help: ModuleType,
     port: int,
     username: str,
@@ -163,35 +183,3 @@ def check_db_once(
                 0,
                 persist_help.ms_queued,
             )
-
-def check_db(
-    persist_help: ModuleType,
-    port: int,
-    username: str,
-    subscription_topic: str,
-    client_msg_counts: dict[str, int],
-    publisher_id: str,
-    num_published_msgs: int,
-    retain_end: int = 0,
-    message_expiry: int = 0,
-    qos: int = 1,
-    check_session_expiry_time: bool = True,
-):
-    for i in range(5):
-        try:
-            check_db_once(
-                persist_help,
-                port,
-                username,
-                subscription_topic,
-                client_msg_counts,
-                publisher_id,
-                num_published_msgs,
-                retain_end,
-                message_expiry,
-                qos,
-                check_session_expiry_time,
-            )
-            break
-        except ValueError:
-            time.sleep(1)

test/broker/persist_sqlite.py

@@ -3,6 +3,7 @@ from pathlib import Path
 import sqlite3
 import mosq_paths
 import mosq_test
+from persist_module_helper import retry
 
 mosq_test.require_features(["WITH_PLUGINS", "WITH_PLUGIN_PERSIST_SQLITE"])
 
@@ -116,6 +117,7 @@ def cleanup(port):
     return rc
 
 
+@retry()
 def check_version_infos(port, database_schema_version):
     con = sqlite3.connect(Path(str(port), "mosquitto.sqlite3"))
     cur = con.cursor()
@@ -134,6 +136,7 @@ def check_version_infos(port, database_schema_version):
     con.close()
 
 
+@retry()
 def check_counts(
     port,
     clients=0,
@@ -192,6 +195,7 @@ def check_counts(
     con.close()
 
 
+@retry()
 def check_client(
     port,
     client_id,
@@ -287,6 +291,7 @@ def modify_client(port: int, client_id: str, sub_expiry_time: int):
     return num_modified_rows
 
 
+@retry()
 def check_subscription(
     port, client_id, topic, subscription_options, subscription_identifier
 ):
@@ -321,6 +326,7 @@ def check_subscription(
     con.close()
 
 
+@retry()
 def check_client_msg(
     port, client_id, cmsg_id, store_id, dup, direction, mid, qos, retain, state
 ):
@@ -388,6 +394,7 @@ def check_client_msg(
         con.close()
 
 
+@retry()
 def check_base_msg(
     port,
     expiry_time,
@@ -476,6 +483,7 @@ def modify_base_msgs(
     return num_modified_rows
 
 
+@retry()
 def check_retain(port, topic, store_id):
     con = sqlite3.connect(Path(str(port), "mosquitto.sqlite3"))
     cur = con.cursor()
@@ -487,6 +495,7 @@ def check_retain(port, topic, store_id):
     con.close()
 
 
+@retry()
 def check_will(
     port,
     client_id: str,