standby: pull once more after activation

abhizer · abhizer · commit bfa8dcadb128 · 2025-10-08T11:22:14.000Z
Once we receive an activation request, pull once more
to be up to date with the latest checkpoint in S3. If
there is no newer checkpoint, the download should not
take too long.

Signed-off-by: Abhinav Gyawali &lt;22275402+abhizer@users.noreply.github.com&gt;
diff --git a/crates/adapters/src/controller/sync.rs b/crates/adapters/src/controller/sync.rs
@@ -191,7 +191,13 @@ where
     }
 
     let mut prev = uuid::Uuid::nil();
-    while !is_activated() {
+    let mut pull_once_again_after_activation = false;
+
+    // This should run at least once before checking for activation, to ensure
+    // that we pull a checkpoint if one is available.
+    // Also, if we receive an activation signal, we run one more iteration to
+    // ensure that we have the latest checkpoint before activating.
+    loop {
         match pull_and_gc(storage.backend.clone(), sync, &mut prev) {
             Err(err) => {
                 if sync.fail_if_no_checkpoint {
@@ -207,6 +213,15 @@ where
             return Ok(());
         }
 
+        if is_activated() {
+            if pull_once_again_after_activation {
+                // We've already done one iteration after activation, now we're done
+                break;
+            }
+            pull_once_again_after_activation = true;
+            // Continue to pull one more time to get the latest checkpoint
+        }
+
         std::thread::sleep(std::time::Duration::from_secs(sync.pull_interval));
     }
 
diff --git a/python/feldera/rest/_httprequests.py b/python/feldera/rest/_httprequests.py
@@ -56,6 +56,8 @@ def send_request(
         """
         self.headers["Content-Type"] = content_type
 
+        prev_resp: Optional[requests.Response] = None
+
         try:
             conn_timeout = self.config.connection_timeout
             timeout = self.config.timeout
@@ -102,6 +104,8 @@ def send_request(
                         verify=self.requests_verify,
                     )
 
+                prev_resp = request
+
                 try:
                     resp = self.__validate(request, stream=stream)
                     logging.debug("got response: %s", str(resp))
@@ -134,6 +138,10 @@ def send_request(
         except requests.exceptions.ConnectionError as err:
             raise FelderaCommunicationError(str(err)) from err
 
+        raise FelderaAPIError(
+            "Max retries exceeded, couldn't successfully connect to Feldera", prev_resp
+        )
+
     def get(
         self,
         path: str,
diff --git a/python/feldera/rest/feldera_client.py b/python/feldera/rest/feldera_client.py
@@ -185,7 +185,7 @@ def __wait_for_pipeline_state(
                 elapsed = time.monotonic() - start_time
                 if elapsed > timeout_s:
                     raise TimeoutError(
-                        f"Timed out waiting for pipeline {pipeline_name} to"
+                        f"Timed out waiting for pipeline {pipeline_name} to "
                         f"transition to '{state}' state"
                     )
 
diff --git a/python/tests/platform/test_checkpoint_sync.py b/python/tests/platform/test_checkpoint_sync.py
@@ -150,8 +150,7 @@ def test_checkpoint_sync(
 
             if standby:
                 assert self.pipeline.status() == PipelineStatus.STANDBY
-
-                self.pipeline.activate(timeout_s=10)
+                self.pipeline.activate()
 
         got_after = list(self.pipeline.query("SELECT * FROM v0"))
 

Original file line number	Diff line number	Diff line change
`@@ -185,7 +185,7 @@ def __wait_for_pipeline_state(`
`185`	`185`	`elapsed = time.monotonic() - start_time`
`186`	`186`	`if elapsed > timeout_s:`
`187`	`187`	`raise TimeoutError(`
`188`		`- f"Timed out waiting for pipeline {pipeline_name} to"`
	`188`	`+ f"Timed out waiting for pipeline {pipeline_name} to "`
`189`	`189`	`f"transition to '{state}' state"`
`190`	`190`	`)`
`191`	`191`