sync: support automatic periodic checkpoint syncs

abhizer · abhizer · commit c89ff35885ae · 2026-01-02T21:08:03.000Z
Adds a parameter `push_interval` to `SyncConfig` to specify the interval
at which to push the latest checkpoint to object storage. Default is
None, disabling the periodic push.

Adds a field `periodic` to response of `/checkpoint/sync_status`, that
maintains the latest periodic checkpoint that was synced. I chose adding
a field instead of setting it to `success` as I think the user should be
able to know if the sync was automatic or manual; it also affects allows
us to maintain the current the behavior of the python client downstream.

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
@@ -148,7 +148,9 @@ pub use feldera_types::config::{
     ConnectorConfig, FormatConfig, InputEndpointConfig, OutputEndpointConfig, PipelineConfig,
     RuntimeConfig, TransportConfig,
 };
-use feldera_types::config::{FileBackendConfig, FtConfig, FtModel, OutputBufferConfig, SyncConfig};
+use feldera_types::config::{
+    FileBackendConfig, FtConfig, FtModel, OutputBufferConfig, StorageBackendConfig, SyncConfig,
+};
 use feldera_types::constants::{STATE_FILE, STEPS_FILE};
 use feldera_types::format::json::{JsonFlavor, JsonParserConfig, JsonUpdateFormat};
 use feldera_types::program_schema::{SqlIdentifier, canonical_identifier};
@@ -522,6 +524,10 @@ impl Controller {
         self.inner.last_checkpoint()
     }
 
+    pub(crate) fn last_checkpoint_sync(&self) -> LastCheckpoint {
+        self.inner.last_checkpoint_sync()
+    }
+
     pub fn lir(&self) -> &LirCircuit {
         &self.inner.lir
     }
@@ -1693,14 +1699,21 @@ impl RunningCheckpointSync {
         Ok(Self::Waiting(uuid, join_handle, receiver))
     }
 
-    fn poll(&mut self) -> Option<Result<(), Arc<ControllerError>>> {
+    fn poll(&mut self, circuit: &mut CircuitThread) -> Option<Result<(), Arc<ControllerError>>> {
         let uuid = self.uuid();
 
         match replace(self, Self::Done(uuid)) {
             Self::Error(_, error) => Some(Err(error)),
             Self::Waiting(uuid, join_handle, mut receiver) => match receiver.try_recv() {
                 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),
+                    };
+
                     Some(result)
                 }
                 Err(TryRecvError::Empty) => {
@@ -1716,9 +1729,21 @@ impl RunningCheckpointSync {
     }
 }
 
-struct SyncCheckpointRequest {
-    uuid: uuid::Uuid,
-    cb: SyncCheckpointCallbackFn,
+enum SyncCheckpointRequest {
+    Scheduled(uuid::Uuid),
+    Requested {
+        uuid: uuid::Uuid,
+        cb: SyncCheckpointCallbackFn,
+    },
+}
+
+impl SyncCheckpointRequest {
+    fn uuid(&self) -> uuid::Uuid {
+        match self {
+            Self::Scheduled(uuid) => *uuid,
+            Self::Requested { uuid, .. } => *uuid,
+        }
+    }
 }
 
 #[derive(Clone)]
@@ -2136,17 +2161,22 @@ impl CircuitThread {
             output_backpressure_warning = None;
 
             match trigger.trigger(
-                self.last_checkpoint().timestamp,
+                self.last_checkpoint(),
+                self.last_checkpoint_sync(),
                 self.replaying(),
                 self.circuit.bootstrap_in_progress(),
                 self.checkpoint_requested(),
+                self.sync_checkpoint_requested(),
             ) {
                 Action::Step => {
                     if !self.step()? {
                         break;
                     }
                 }
                 Action::Checkpoint => self.checkpoint_requests.push(CheckpointRequest::Scheduled),
+                Action::SyncCheckpoint(chk) => self
+                    .sync_checkpoint_requests
+                    .push(SyncCheckpointRequest::Scheduled(chk)),
                 Action::Park(Some(deadline)) => self.parker.park_deadline(deadline),
                 Action::Park(None) => self.parker.park(),
             }
@@ -2341,6 +2371,10 @@ impl CircuitThread {
         self.controller.last_checkpoint()
     }
 
+    fn last_checkpoint_sync(&self) -> LastCheckpoint {
+        self.controller.last_checkpoint_sync()
+    }
+
     fn update_last_checkpoint(&self, result: &Result<Checkpoint, ControllerError>) {
         *self.controller.last_checkpoint.lock().unwrap() = LastCheckpoint {
             timestamp: Instant::now(),
@@ -2437,10 +2471,11 @@ impl CircuitThread {
                         .push(CheckpointRequest::SuspendCommand(reply_callback));
                 }
                 Command::SyncCheckpoint((uuid, reply_callback)) => {
-                    self.sync_checkpoint_requests.push(SyncCheckpointRequest {
-                        uuid,
-                        cb: reply_callback,
-                    });
+                    self.sync_checkpoint_requests
+                        .push(SyncCheckpointRequest::Requested {
+                            uuid,
+                            cb: reply_callback,
+                        });
                 }
             }
         }
@@ -2803,8 +2838,10 @@ impl CircuitThread {
             uuid: uuid::Uuid,
             result: Result<(), Arc<ControllerError>>,
         ) {
-            for request in requests.extract_if(.., |x| x.uuid == uuid) {
-                (request.cb)(result.clone());
+            for request in requests.extract_if(.., |x| x.uuid() == uuid) {
+                if let SyncCheckpointRequest::Requested { cb, .. } = request {
+                    (cb)(result.clone());
+                }
             }
         }
 
@@ -2814,7 +2851,7 @@ impl CircuitThread {
 
         match running_sync {
             RunningCheckpointSync::Waiting(uuid, _, _) => {
-                let Some(result) = running_sync.poll() else {
+                let Some(result) = running_sync.poll(self) else {
                     self.running_checkpoint_sync = Some(running_sync);
                     return;
                 };
@@ -2836,7 +2873,7 @@ impl CircuitThread {
         };
 
         if self.running_checkpoint_sync.is_none() {
-            self.running_checkpoint_sync = Some(RunningCheckpointSync::new(self, req.uuid));
+            self.running_checkpoint_sync = Some(RunningCheckpointSync::new(self, req.uuid()));
         }
     }
 }
@@ -3168,6 +3205,9 @@ struct StepTrigger {
     /// Time between automatic checkpoints.
     checkpoint_interval: Option<Duration>,
 
+    /// Time between automatic checkpoint syncs.
+    sync_interval: Option<Duration>,
+
     /// The circuit is bootstrapping. Used to detect the transition from bootstrapping
     /// to normal mode.
     bootstrapping: bool,
@@ -3184,6 +3224,9 @@ enum Action {
 
     /// Step the circuit.
     Step,
+
+    /// Synchronize a checkpoint to object storage.
+    SyncCheckpoint(uuid::Uuid),
 }
 
 impl StepTrigger {
@@ -3193,19 +3236,30 @@ impl StepTrigger {
         let max_buffering_delay = Duration::from_micros(config.max_buffering_delay_usecs);
         let min_batch_size_records = config.min_batch_size_records;
         let checkpoint_interval = config.fault_tolerance.checkpoint_interval();
+        let sync_interval = config.storage.as_ref().and_then(|s| match &s.backend {
+            StorageBackendConfig::File(file) => file
+                .sync
+                .as_ref()
+                .and_then(|s| s.push_interval)
+                .map(Duration::from_secs),
+            _ => None,
+        });
+
         Self {
             controller,
             buffer_timeout: None,
             max_buffering_delay,
             min_batch_size_records,
             checkpoint_interval,
             bootstrapping,
+            sync_interval,
         }
     }
 
     /// Determines when to trigger the next step, given:
     ///
-    /// - The time of the last checkpoint.
+    /// - The metadata about the last checkpoint.
+    /// - The metadata about the last sync checkpoint.
     /// - Whether we're currently `replaying`.
     /// - Whether the pipeline is currently `bootstrapping`.
     /// - Whether the pipeline is currently `running`.
@@ -3214,10 +3268,12 @@ impl StepTrigger {
     /// Returns the action for the controller to take.
     fn trigger(
         &mut self,
-        last_checkpoint: Instant,
+        last_checkpoint: LastCheckpoint,
+        last_sync: LastCheckpoint,
         replaying: bool,
         bootstrapping: bool,
         checkpoint_requested: bool,
+        sync_checkpoint_requested: bool,
     ) -> Action {
         // If any input endpoints are blocking suspend, then those are the only
         // ones that we count; otherwise, count all of them.
@@ -3238,7 +3294,12 @@ impl StepTrigger {
         // Time of the next checkpoint.
         let checkpoint = self
             .checkpoint_interval
-            .map(|interval| last_checkpoint + interval);
+            .map(|interval| last_checkpoint.timestamp + interval);
+
+        // Time of the next checkpoint sync.
+        let sync_checkpoint = self
+            .sync_interval
+            .map(|interval| last_sync.timestamp + interval);
 
         // Used to force a step regardless of input
         let committing = self.controller.transaction_commit_requested();
@@ -3258,6 +3319,13 @@ impl StepTrigger {
             step(self)
         } else if checkpoint.is_some_and(|t| now >= t) && !checkpoint_requested {
             Action::Checkpoint
+        } else if sync_checkpoint.is_some_and(|t| now >= t)
+            && !sync_checkpoint_requested
+            && let Some(chk) = last_checkpoint.id
+            && !chk.is_nil()
+            && Some(chk) != last_sync.id
+        {
+            Action::SyncCheckpoint(chk)
         } else if self.controller.status.unset_step_requested()
             || buffered_records > self.min_batch_size_records
             || self.buffer_timeout.is_some_and(|t| now >= t)
@@ -4284,6 +4352,7 @@ impl TransactionInfo {
 pub struct ControllerInner {
     pub status: Arc<ControllerStatus>,
     last_checkpoint: Mutex<LastCheckpoint>,
+    last_checkpoint_sync: Mutex<LastCheckpoint>,
     secrets_dir: PathBuf,
     num_api_connections: AtomicU64,
     command_sender: Sender<Command>,
@@ -4354,6 +4423,7 @@ impl ControllerInner {
             command_sender,
             catalog: Arc::new(catalog),
             last_checkpoint: Default::default(),
+            last_checkpoint_sync: Default::default(),
             lir,
             trace_snapshot: Arc::new(TokioMutex::new(BTreeMap::new())),
             next_input_id: Atomic::new(0),
@@ -4451,6 +4521,10 @@ impl ControllerInner {
         self.last_checkpoint.lock().unwrap().clone()
     }
 
+    fn last_checkpoint_sync(&self) -> LastCheckpoint {
+        self.last_checkpoint_sync.lock().unwrap().clone()
+    }
+
     fn get_transaction_number(&self) -> u64 {
         self.transaction_number.load(Ordering::Acquire)
     }
diff --git a/crates/adapters/src/server.rs b/crates/adapters/src/server.rs
@@ -160,6 +160,10 @@ impl CheckpointSyncState {
             }
         }
     }
+
+    fn completed_periodic(&mut self, uuid: uuid::Uuid) {
+        self.status.periodic = Some(uuid);
+    }
 }
 
 #[derive(Clone, Debug, Default)]
@@ -1681,8 +1685,17 @@ async fn checkpoint_status(state: WebData<ServerState>) -> impl Responder {
 }
 
 #[get("/checkpoint/sync_status")]
-async fn sync_checkpoint_status(state: WebData<ServerState>) -> impl Responder {
-    HttpResponse::Ok().json(state.sync_checkpoint_state.lock().unwrap().status.clone())
+async fn sync_checkpoint_status(
+    state: WebData<ServerState>,
+) -> Result<HttpResponse, PipelineError> {
+    let mut sync_state = state.sync_checkpoint_state.lock().unwrap();
+
+    let controller = state.controller()?;
+    if let Some(chk) = controller.last_checkpoint_sync().id {
+        sync_state.completed_periodic(chk);
+    }
+
+    Ok(HttpResponse::Ok().json(sync_state.status.clone()))
 }
 
 /// Suspends the pipeline and terminate the circuit.
diff --git a/crates/feldera-types/src/checkpoint.rs b/crates/feldera-types/src/checkpoint.rs
@@ -58,6 +58,9 @@ pub struct CheckpointSyncStatus {
 
     /// Most recently failed checkpoint sync, and the associated error.
     pub failure: Option<CheckpointSyncFailure>,
+
+    /// Most recently successful automated periodic checkpoint sync.
+    pub periodic: Option<Uuid>,
 }
 
 /// Information about a failed checkpoint sync.
diff --git a/crates/feldera-types/src/config.rs b/crates/feldera-types/src/config.rs
@@ -527,6 +527,12 @@ pub struct SyncConfig {
     #[serde(default = "default_pull_interval")]
     pub pull_interval: u64,
 
+    /// The interval (in seconds) between each push of checkpoints to object store.
+    ///
+    /// Default: disabled (no periodic push).
+    #[serde(default)]
+    pub push_interval: Option<u64>,
+
     /// Extra flags to pass to `rclone`.
     ///
     /// WARNING: Supplying incorrect or conflicting flags can break `rclone`.
diff --git a/openapi.json b/openapi.json
@@ -11518,6 +11518,13 @@
             "default": 10,
             "minimum": 0
           },
+          "push_interval": {
+            "type": "integer",
+            "format": "int64",
+            "description": "The interval (in seconds) between each push of checkpoints to object store.\n\nDefault: disabled (no periodic push).",
+            "nullable": true,
+            "minimum": 0
+          },
           "region": {
             "type": "string",
             "description": "The region that this bucket is in.\n\nLeave empty for Minio or the default region (`us-east-1` for AWS).",
diff --git a/python/feldera/pipeline.py b/python/feldera/pipeline.py
@@ -825,10 +825,11 @@ def sync_checkpoint_status(self, uuid: str) -> CheckpointStatus:
 
         resp = self.client.sync_checkpoint_status(self.name)
         success = resp.get("success")
+        periodic = resp.get("periodic")
 
         fail = resp.get("failure") or {}
 
-        if uuid == success:
+        if uuid == success or uuid == periodic:
             return CheckpointStatus.Success
 
         fail = resp.get("failure") or {}
@@ -843,6 +844,26 @@ def sync_checkpoint_status(self, uuid: str) -> CheckpointStatus:
 
         return CheckpointStatus.Unknown
 
+    def last_successful_checkpoint_sync(self) -> UUID:
+        """
+        Returns the UUID of the last successfully synced checkpoint.
+
+        :return: The UUID of the last successfully synced checkpoint.
+        """
+
+        resp = self.client.sync_checkpoint_status(self.name)
+        success = resp.get("success")
+        periodic = resp.get("periodic")
+
+        if success is None and periodic is None:
+            raise RuntimeError("no checkpoints have been synced yet")
+        elif success is None:
+            return UUID(periodic)
+        elif periodic is None:
+            return UUID(success)
+        else:
+            return max(UUID(success), UUID(periodic))
+
     def query(self, query: str) -> Generator[Mapping[str, Any], None, None]:
         """
         Executes an ad-hoc SQL query on this pipeline and returns a generator
diff --git a/python/tests/platform/test_checkpoint_sync.py b/python/tests/platform/test_checkpoint_sync.py

Original file line number	Diff line number	Diff line change
`@@ -160,6 +160,10 @@ impl CheckpointSyncState {`
`160`	`160`	`}`
`161`	`161`	`}`
`162`	`162`	`}`
	`163`	`+`
	`164`	`+ fn completed_periodic(&mut self, uuid: uuid::Uuid) {`
	`165`	`+ self.status.periodic = Some(uuid);`
	`166`	`+ }`
`163`	`167`	`}`
`164`	`168`
`165`	`169`	`#[derive(Clone, Debug, Default)]`
`@@ -1681,8 +1685,17 @@ async fn checkpoint_status(state: WebData<ServerState>) -> impl Responder {`
`1681`	`1685`	`}`
`1682`	`1686`
`1683`	`1687`	`#[get("/checkpoint/sync_status")]`
`1684`		`-async fn sync_checkpoint_status(state: WebData<ServerState>) -> impl Responder {`
`1685`		`- HttpResponse::Ok().json(state.sync_checkpoint_state.lock().unwrap().status.clone())`
	`1688`	`+async fn sync_checkpoint_status(`
	`1689`	`+ state: WebData<ServerState>,`
	`1690`	`+) -> Result<HttpResponse, PipelineError> {`
	`1691`	`+ let mut sync_state = state.sync_checkpoint_state.lock().unwrap();`
	`1692`	`+`
	`1693`	`+ let controller = state.controller()?;`
	`1694`	`+ if let Some(chk) = controller.last_checkpoint_sync().id {`
	`1695`	`+ sync_state.completed_periodic(chk);`
	`1696`	`+ }`
	`1697`	`+`
	`1698`	`+ Ok(HttpResponse::Ok().json(sync_state.status.clone()))`
`1686`	`1699`	`}`
`1687`	`1700`
`1688`	`1701`	`/// Suspends the pipeline and terminate the circuit.`
Original file line number	Diff line number	Diff line change
`@@ -58,6 +58,9 @@ pub struct CheckpointSyncStatus {`
`58`	`58`
`59`	`59`	`/// Most recently failed checkpoint sync, and the associated error.`
`60`	`60`	`pub failure: Option<CheckpointSyncFailure>,`
	`61`	`+`
	`62`	`+ /// Most recently successful automated periodic checkpoint sync.`
	`63`	`+ pub periodic: Option<Uuid>,`
`61`	`64`	`}`
`62`	`65`
`63`	`66`	`/// Information about a failed checkpoint sync.`