feldera · ryzhyk · Jun 17, 2026 · Jun 11, 2026 · Jun 11, 2026 · Jun 11, 2026
diff --git a/crates/adapters/src/controller.rs b/crates/adapters/src/controller.rs
diff --git a/crates/adapters/src/controller/stats.rs b/crates/adapters/src/controller/stats.rs
@@ -67,7 +67,7 @@ use feldera_types::{
     memory_pressure::MemoryPressure,
     suspend::{PermanentSuspendError, SuspendError},
     time_series::SampleStatistics,
-    transaction::CommitProgressSummary,
+    transaction::{CommitProgressSummary, ConcurrentBootstrapProgress},
 };
 use parking_lot::{RwLock, RwLockReadGuard};
 use serde::{Deserialize, Serialize};
@@ -177,9 +177,22 @@ pub struct GlobalControllerMetrics {
     /// new and modified views.
     bootstrap_in_progress: AtomicBool,
 
+    /// A concurrent bootstrap is in its backfill phase: new/modified views
+    /// replay in the background while the pre-existing views stay live and
+    /// inputs keep flowing.
+    concurrent_backfill_in_progress: AtomicBool,
+
+    /// A concurrent bootstrap is in its synchronize (cutover) phase: inputs are
+    /// paused while recorded deltas drain into the new views before the state
+    /// transfer.
+    concurrent_synchronize_in_progress: AtomicBool,
+
     /// Transaction commit progress, if a transaction is committing.
     pub commit_progress: Mutex<Option<CommitProgressSummary>>,
 
+    /// Progress of a concurrent bootstrap, if one is in progress.
+    pub concurrent_bootstrap_progress: Mutex<Option<ConcurrentBootstrapProgress>>,
+
     /// Time at which the pipeline process started, in seconds since the epoch.
     pub start_time: DateTime<Utc>,
 
@@ -300,7 +313,10 @@ impl GlobalControllerMetrics {
         Self {
             state: Atomic::new(PipelineState::Paused),
             bootstrap_in_progress: AtomicBool::new(false),
+            concurrent_backfill_in_progress: AtomicBool::new(false),
+            concurrent_synchronize_in_progress: AtomicBool::new(false),
             commit_progress: Mutex::new(None),
+            concurrent_bootstrap_progress: Mutex::new(None),
             start_time,
             incarnation_uuid,
             initial_start_time,
@@ -409,6 +425,25 @@ impl GlobalControllerMetrics {
             .store(bootstrap_in_progress, Ordering::Release);
     }
 
+    fn concurrent_backfill_in_progress(&self) -> bool {
+        self.concurrent_backfill_in_progress.load(Ordering::Acquire)
+    }
+
+    fn set_concurrent_backfill_in_progress(&self, value: bool) {
+        self.concurrent_backfill_in_progress
+            .store(value, Ordering::Release);
+    }
+
+    fn concurrent_synchronize_in_progress(&self) -> bool {
+        self.concurrent_synchronize_in_progress
+            .load(Ordering::Acquire)
+    }
+
+    fn set_concurrent_synchronize_in_progress(&self, value: bool) {
+        self.concurrent_synchronize_in_progress
+            .store(value, Ordering::Release);
+    }
+
     fn set_step_requested(&self) -> bool {
         self.step_requested.swap(true, Ordering::AcqRel)
     }
@@ -417,6 +452,10 @@ impl GlobalControllerMetrics {
         *self.commit_progress.lock().unwrap() = commit_progress;
     }
 
+    pub fn set_concurrent_bootstrap_progress(&self, progress: Option<ConcurrentBootstrapProgress>) {
+        *self.concurrent_bootstrap_progress.lock().unwrap() = progress;
+    }
+
     pub fn update_output_stall_start(&self, stalled: bool) {
         let mut output_stall_start = self.output_stall_start.lock().unwrap();
         if stalled != output_stall_start.is_some() {
@@ -739,6 +778,24 @@ impl ControllerStatus {
             .set_bootstrap_in_progress(bootstrap_in_progress);
     }
 
+    pub fn concurrent_backfill_in_progress(&self) -> bool {
+        self.global_metrics.concurrent_backfill_in_progress()
+    }
+
+    pub fn set_concurrent_backfill_in_progress(&self, value: bool) {
+        self.global_metrics
+            .set_concurrent_backfill_in_progress(value);
+    }
+
+    pub fn concurrent_synchronize_in_progress(&self) -> bool {
+        self.global_metrics.concurrent_synchronize_in_progress()
+    }
+
+    pub fn set_concurrent_synchronize_in_progress(&self, value: bool) {
+        self.global_metrics
+            .set_concurrent_synchronize_in_progress(value);
+    }
+
     pub fn request_step(&self, circuit_thread_unparker: &Unparker) {
         let old = self.global_metrics.set_step_requested();
         if !old {
@@ -1355,6 +1412,12 @@ impl ControllerStatus {
                 0
             },
             commit_progress: self.global_metrics.commit_progress.lock().unwrap().clone(),
+            concurrent_bootstrap_progress: self
+                .global_metrics
+                .concurrent_bootstrap_progress
+                .lock()
+                .unwrap()
+                .clone(),
             transaction_initiators: ctx.transaction_info.initiators.to_api_type(),
             start_time: self.global_metrics.start_time,
             incarnation_uuid: self.global_metrics.incarnation_uuid,