sync: prevent syncing checkpoints from being GCed

abhizer · abhizer · commit 7f19dc031d00 · 2025-09-03T10:59:54.000Z
Only clears the `sync_checkpoint_request` once the sync has
actually completed. Checkpoints will only get GCed once the
sync process has completed.

Signed-off-by: Abhinav Gyawali &lt;22275402+abhizer@users.noreply.github.com&gt;
diff --git a/crates/adapters/src/controller/mod.rs b/crates/adapters/src/controller/mod.rs
@@ -1034,6 +1034,17 @@ impl Controller {
     }
 }
 
+struct SyncCheckpointRequest {
+    uuid: Arc<TokioMutex<Option<uuid::Uuid>>>,
+    cb: Option<SyncCheckpointCallbackFn>,
+}
+
+impl SyncCheckpointRequest {
+    fn uuid(&self) -> Option<uuid::Uuid> {
+        *self.uuid.blocking_lock()
+    }
+}
+
 struct CircuitThread {
     controller: Arc<ControllerInner>,
     circuit: DBSPHandle,
@@ -1048,7 +1059,7 @@ struct CircuitThread {
     checkpoint_requests: Vec<CheckpointRequest>,
 
     /// Currently only allows one request at a time.
-    sync_checkpoint_request: Option<(uuid::Uuid, SyncCheckpointCallbackFn)>,
+    sync_checkpoint_request: Option<SyncCheckpointRequest>,
 
     /// Storage backend for writing checkpoints.
     storage: Option<Arc<dyn StorageBackend>>,
@@ -1467,12 +1478,18 @@ impl CircuitThread {
             CHECKPOINT_WRITTEN.record((written_after - written_before) / 1_000_000);
             CHECKPOINT_PROCESSED_RECORDS.store(processed_records, Ordering::Relaxed);
 
-            // TODO: check the requested checkpoint UUID
-            if this.sync_checkpoint_request.is_none() {
+            let sync_requested = this
+                .sync_checkpoint_request
+                .as_ref()
+                .map(|s| s.uuid())
+                .flatten();
+
+            if sync_requested.is_none() {
                 if let Err(error) = this.circuit.gc_checkpoint() {
                     warn!("error removing old checkpoints: {error}");
                 }
             }
+
             if let Some(ft) = &mut this.ft {
                 ft.checkpointed()?;
             }
@@ -1557,7 +1574,10 @@ impl CircuitThread {
                         .push(CheckpointRequest::SuspendCommand(reply_callback));
                 }
                 Command::SyncCheckpoint((uuid, reply_callback)) => {
-                    self.sync_checkpoint_request = Some((uuid, reply_callback));
+                    self.sync_checkpoint_request = Some(SyncCheckpointRequest {
+                        uuid: Arc::new(TokioMutex::new(Some(uuid))),
+                        cb: Some(reply_callback),
+                    });
                 }
             }
         }
@@ -1882,7 +1902,15 @@ impl CircuitThread {
     }
 
     fn sync_checkpoint(&mut self) {
-        let Some((uuid, cb)) = self.sync_checkpoint_request.take() else {
+        let Some((uuid_lock, Some(cb))) = self
+            .sync_checkpoint_request
+            .as_mut()
+            .map(|u| (u.uuid.clone(), u.cb.take()))
+        else {
+            return;
+        };
+
+        let Some(uuid) = uuid_lock.blocking_lock().clone() else {
             return;
         };
 
@@ -1940,6 +1968,8 @@ impl CircuitThread {
                 } else {
                     cb(Ok(()))
                 }
+
+                uuid_lock.blocking_lock().take()
             })
             .expect("failed to spawn s3-synchronizer thread");
     }
diff --git a/python/feldera/pipeline.py b/python/feldera/pipeline.py
@@ -685,6 +685,7 @@ def sync_checkpoint(self, wait: bool = False, timeout_s=300) -> str:
             checkpoint to complete syncing.
 
         :raises FelderaAPIError: If no checkpoints have been made.
+        :raises RuntimeError: If syncing the checkpoint fails.
         """
 
         uuid = self.client.sync_checkpoint(self.name)
@@ -702,6 +703,11 @@ def sync_checkpoint(self, wait: bool = False, timeout_s=300) -> str:
 pipeline '{self.name}' to sync checkpoint '{uuid}'"""
                 )
             status = self.sync_checkpoint_status(uuid)
+            if status == CheckpointStatus.Failure:
+                raise RuntimeError(
+                    f"failed to sync checkpoint '{uuid}': ", status.get_error()
+                )
+
             if status in [CheckpointStatus.InProgress, CheckpointStatus.Unknown]:
                 time.sleep(0.1)
                 continue
@@ -716,6 +722,9 @@ def sync_checkpoint_status(self, uuid: str) -> CheckpointStatus:
         If the checkpoint is currently being synchronized, returns
         `CheckpointStatus.Unknown`.
 
+        Failures are not raised as runtime errors and must be explicitly
+        checked.
+
         :param uuid: The checkpoint uuid.
         """
 
@@ -731,6 +740,7 @@ def sync_checkpoint_status(self, uuid: str) -> CheckpointStatus:
         if uuid == fail.get("uuid"):
             failure = CheckpointStatus.Failure
             failure.error = fail.get("error", "")
+            logging.error(f"failed to sync checkpoint '{uuid}': {failure.error}")
             return failure
 
         if (success is None) or UUID(uuid) > UUID(success):
diff --git a/python/feldera/rest/feldera_client.py b/python/feldera/rest/feldera_client.py
@@ -986,7 +986,7 @@ def query_as_json(
             stream=True,
         )
 
-        for chunk in resp.iter_lines(chunk_size=50000000):
+        for chunk in resp.iter_lines(chunk_size=1024):
             if chunk:
                 yield json.loads(chunk, parse_float=Decimal)
 

Original file line number	Diff line number	Diff line change
`@@ -986,7 +986,7 @@ def query_as_json(`
`986`	`986`	`stream=True,`
`987`	`987`	`)`
`988`	`988`
`989`		`- for chunk in resp.iter_lines(chunk_size=50000000):`
	`989`	`+ for chunk in resp.iter_lines(chunk_size=1024):`
`990`	`990`	`if chunk:`
`991`	`991`	`yield json.loads(chunk, parse_float=Decimal)`
`992`	`992`