[adapters]: last_sync should only be updated with successful syncs

swanandx · swanandx · commit 97d6bca65963 · 2026-03-26T18:28:15.000Z
Currently, even if sync fails, we update last_sync, due to which `periodic`
in checkpoint status reports a UUID whose sync actually failed.

This isn't correct, we should only set it if sync was successful.
periodic checkpoints gets retried anyways, so it if fine to ignore failed ones.

Signed-off-by: Swanand Mulay &lt;73115739+swanandx@users.noreply.github.com&gt;
diff --git a/crates/adapters/src/controller.rs b/crates/adapters/src/controller.rs
@@ -2209,11 +2209,14 @@ impl RunningCheckpointSync {
                 Ok(result) => {
                     join_handle.join().unwrap();
 
-                    let mut last_sync = circuit.controller.last_checkpoint_sync.lock().unwrap();
-                    *last_sync = LastCheckpoint {
-                        timestamp: Instant::now(),
-                        id: Some(uuid),
-                    };
+                    // Only update last_sync if the checkpoint sync succeeded.
+                    if result.is_ok() {
+                        let mut last_sync = circuit.controller.last_checkpoint_sync.lock().unwrap();
+                        *last_sync = LastCheckpoint {
+                            timestamp: Instant::now(),
+                            id: Some(uuid),
+                        };
+                    }
 
                     Some(result)
                 }
diff --git a/python/tests/platform/test_checkpoint_sync.py b/python/tests/platform/test_checkpoint_sync.py
@@ -284,6 +284,63 @@ def test_automated_checkpoint_sync(self):
     def test_automated_checkpoint_sync1(self):
         self.test_checkpoint_sync(ft_interval=5, automated_sync_interval=10)
 
+    @enterprise_only
+    @single_host_only
+    def test_automated_sync_auth_error(self):
+        """
+        CREATE TABLE t0 (c0 INT, c1 VARCHAR);
+        CREATE MATERIALIZED VIEW v0 AS SELECT * FROM t0;
+        """
+        # Configure pipeline with automatic periodic sync and bad credentials.
+        # The periodic sync should fail, and `periodic` in the sync status
+        # should remain None (not report a UUID for a failed sync).
+        storage_config = storage_cfg(
+            self.pipeline.name,
+            push_interval=5,
+            auth_err=True,
+        )
+        self.pipeline.set_runtime_config(
+            RuntimeConfig(
+                workers=FELDERA_TEST_NUM_WORKERS,
+                hosts=FELDERA_TEST_NUM_HOSTS,
+                fault_tolerance_model=FaultToleranceModel.AtLeastOnce,
+                storage=Storage(config=storage_config),
+                checkpoint_interval_secs=5,
+            )
+        )
+        self.pipeline.start()
+
+        # Insert data so the pipeline has something to checkpoint.
+        data = [{"c0": i, "c1": str(i)} for i in range(1, 10)]
+        self.pipeline.input_json("t0", data)
+        self.pipeline.wait_for_completion()
+
+        # Wait for a checkpoint to be created.
+        def checkpoint_exists() -> bool:
+            return len(self.pipeline.checkpoints()) > 0
+
+        wait_for_condition(
+            "checkpoint is created",
+            checkpoint_exists,
+            timeout_s=30.0,
+            poll_interval_s=0.5,
+        )
+
+        # Wait long enough for the periodic sync to have attempted (and failed).
+        time.sleep(15)
+
+        # The periodic sync should have failed due to bad credentials.
+        # `periodic` must be None — a failed sync should not report a UUID.
+        status = self.pipeline.client.sync_checkpoint_status(self.pipeline.name)
+        print(f"sync_status after failed periodic sync: {status}", file=sys.stderr)
+        self.assertIsNone(
+            status.get("periodic"),
+            f"periodic should be None after failed sync, got: {status}",
+        )
+
+        self.pipeline.stop(force=True)
+        self.pipeline.clear_storage()
+
     @enterprise_only
     @single_host_only
     def test_autherr_fail(self):
