sync: add tests with periodic checkpointing

abhizer · abhizer · commit 17a9346c5fb7 · 2025-12-09T09:34:34.000Z
Signed-off-by: Abhinav Gyawali &lt;22275402+abhizer@users.noreply.github.com&gt;
diff --git a/crates/adapters/src/controller.rs b/crates/adapters/src/controller.rs
@@ -4007,7 +4007,7 @@ impl TransactionInfo {
 /// controller threads.
 pub struct ControllerInner {
     pub status: Arc<ControllerStatus>,
-    last_checkpoint: Arc<Mutex<LastCheckpoint>>,
+    last_checkpoint: Mutex<LastCheckpoint>,
     secrets_dir: PathBuf,
     num_api_connections: AtomicU64,
     command_sender: Sender<Command>,
diff --git a/python/tests/platform/test_checkpoint_sync.py b/python/tests/platform/test_checkpoint_sync.py
@@ -58,6 +58,8 @@ def test_checkpoint_sync(
         strict: bool = False,
         expect_empty: bool = False,
         standby: bool = False,
+        ft_interval: int = 60,
+        automated_checkpoint: bool = False,
     ):
         """
         CREATE TABLE t0 (c0 INT, c1 VARCHAR);
@@ -69,7 +71,8 @@ def test_checkpoint_sync(
 
         self.pipeline.set_runtime_config(
             RuntimeConfig(
-                fault_tolerance_model=ft, storage=Storage(config=storage_config)
+                fault_tolerance_model=ft, storage=Storage(config=storage_config),
+                checkpoint_interval_secs=ft_interval,
             )
         )
         self.pipeline.start()
@@ -103,7 +106,10 @@ def test_checkpoint_sync(
                 f"adhoc query returned {len(got_before)} but {processed} records were processed: {got_before}"
             )
 
-        self.pipeline.checkpoint(wait=True)
+        if not automated_checkpoint:
+            self.pipeline.checkpoint(wait=True)
+        else:
+            time.sleep(ft_interval)
         uuid = self.pipeline.sync_checkpoint(wait=True)
 
         self.pipeline.stop(force=True)
@@ -124,7 +130,8 @@ def test_checkpoint_sync(
         )
         self.pipeline.set_runtime_config(
             RuntimeConfig(
-                fault_tolerance_model=ft, storage=Storage(config=storage_config)
+                fault_tolerance_model=ft, storage=Storage(config=storage_config),
+                checkpoint_interval_secs=ft_interval,
             )
         )
 
@@ -177,9 +184,13 @@ def test_from_uuid(self):
     def test_without_clearing_storage(self):
         self.test_checkpoint_sync(clear_storage=False)
 
+    @enterprise_only
+    def test_automated_checkpoint(self):
+        self.test_checkpoint_sync(ft_interval=5, automated_checkpoint=True)
+
     @enterprise_only
     def test_autherr_fail(self):
-        with self.assertRaisesRegex(RuntimeError, "SignatureDoesNotMatch"):
+        with self.assertRaisesRegex(RuntimeError, "SignatureDoesNotMatch|Forbidden"):
             self.test_checkpoint_sync(auth_err=True, strict=True)
 
     @enterprise_only
