feat(refresh): implement coordinated refresh progress tracking

- Introduced `RefreshProgressTracker` to manage progress across multiple actors during refresh operations, preventing race conditions.
- Updated data structures to track per-actor progress for list and load phases.
- Added new `RefreshProgress` protobuf message for communication.
- Enhanced `BarrierCompleteResult` to include refresh progress data.
- Integrated the tracker into `DatabaseCheckpointControl` and updated related components for compatibility.
- Added migration for new refresh job table and related functionality.

Next steps include integrating the tracker with barrier checkpoint control and updating RPC call sites to handle refresh progress.

Author: tabVersion

src/meta/model/migration/src/lib.rs

@@ -56,6 +56,7 @@ mod m20250916_120000_add_refresh_fields;
 mod m20251005_000000_fragment_splits;
 mod m20251016_220528_fragment_parallelism;
 mod m20251022_294610_source_refresh_mode;
+mod m20251030_120000_refresh_jobs;
 mod utils;
 
 pub struct Migrator;
@@ -150,6 +151,7 @@ impl MigratorTrait for Migrator {
             Box::new(m20251005_000000_fragment_splits::Migration),
             Box::new(m20251016_220528_fragment_parallelism::Migration),
             Box::new(m20251022_294610_source_refresh_mode::Migration),
+            Box::new(m20251030_120000_refresh_jobs::Migration),
         ]
     }
 }

src/meta/model/migration/src/m20251030_120000_refresh_jobs.rs

@@ -0,0 +1,72 @@
+use sea_orm_migration::prelude::*;
+
+#[derive(DeriveMigrationName)]
+pub struct Migration;
+
+#[async_trait::async_trait]
+impl MigrationTrait for Migration {
+    async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        manager
+            .create_table(
+                sea_orm_migration::prelude::Table::create()
+                    .table(RefreshJob::Table)
+                    .col(
+                        ColumnDef::new(RefreshJob::TableId)
+                            .integer()
+                            .not_null()
+                            .primary_key(),
+                    )
+                    .col(
+                        ColumnDef::new(RefreshJob::JobCreateTime)
+                            .date_time()
+                            .not_null()
+                            .default(Expr::current_timestamp()),
+                    )
+                    .col(ColumnDef::new(RefreshJob::LastTriggerTime).date_time())
+                    .col(ColumnDef::new(RefreshJob::TriggerIntervalSecs).big_integer())
+                    .col(
+                        ColumnDef::new(RefreshJob::CurrentStatus)
+                            .string_len(16)
+                            .not_null()
+                            .default("IDLE"),
+                    )
+                    .foreign_key(
+                        ForeignKey::create()
+                            .name("fk_refresh_job_table")
+                            .from(RefreshJob::Table, RefreshJob::TableId)
+                            .to(Table::Table, Table::TableId)
+                            .on_delete(ForeignKeyAction::Cascade),
+                    )
+                    .to_owned(),
+            )
+            .await
+    }
+
+    async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        manager
+            .drop_table(
+                sea_orm_migration::prelude::Table::drop()
+                    .table(RefreshJob::Table)
+                    .to_owned(),
+            )
+            .await
+    }
+}
+
+#[derive(DeriveIden)]
+enum RefreshJob {
+    #[sea_orm(iden = "refresh_job")]
+    Table,
+    TableId,
+    JobCreateTime,
+    LastTriggerTime,
+    TriggerIntervalSecs,
+    CurrentStatus,
+}
+
+#[derive(DeriveIden)]
+enum Table {
+    #[sea_orm(iden = "table")]
+    Table,
+    TableId,
+}

src/meta/model/src/lib.rs

@@ -53,6 +53,7 @@ pub mod iceberg_tables;
 pub mod index;
 pub mod object;
 pub mod object_dependency;
+pub mod refresh_job;
 pub mod schema;
 pub mod secret;
 pub mod serde_seaql_migration;

src/meta/model/src/prelude.rs

@@ -33,6 +33,7 @@ pub use super::hummock_version_stats::Entity as HummockVersionStats;
 pub use super::index::Entity as Index;
 pub use super::object::Entity as Object;
 pub use super::object_dependency::Entity as ObjectDependency;
+pub use super::refresh_job::Entity as RefreshJob;
 pub use super::schema::Entity as Schema;
 pub use super::secret::Entity as Secret;
 pub use super::session_parameter::Entity as SessionParameter;

src/meta/model/src/refresh_job.rs

@@ -0,0 +1,68 @@
+// Copyright 2025 RisingWave Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use sea_orm::entity::prelude::*;
+use serde::{Deserialize, Serialize};
+
+use crate::TableId;
+
+#[derive(
+    Clone, Debug, PartialEq, Eq, EnumIter, DeriveActiveEnum, Serialize, Deserialize, Copy, Hash,
+)]
+#[sea_orm(rs_type = "String", db_type = "string(None)")]
+pub enum RefreshJobStatus {
+    #[sea_orm(string_value = "IDLE")]
+    Idle,
+    #[sea_orm(string_value = "RUNNING")]
+    Running,
+    #[sea_orm(string_value = "DISABLED")]
+    Disabled,
+}
+
+impl RefreshJobStatus {
+    pub fn is_active(self) -> bool {
+        matches!(self, RefreshJobStatus::Idle | RefreshJobStatus::Running)
+    }
+}
+
+#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq, Serialize, Deserialize)]
+#[sea_orm(table_name = "refresh_job")]
+pub struct Model {
+    #[sea_orm(primary_key, auto_increment = false)]
+    pub table_id: TableId,
+    pub job_create_time: DateTime,
+    pub last_trigger_time: Option<DateTime>,
+    pub trigger_interval_secs: Option<i64>,
+    pub current_status: RefreshJobStatus,
+}
+
+#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
+pub enum Relation {
+    #[sea_orm(
+        belongs_to = "super::table::Entity",
+        from = "Column::TableId",
+        to = "super::table::Column::TableId",
+        on_update = "Cascade",
+        on_delete = "Cascade"
+    )]
+    Table,
+}
+
+impl Related<super::table::Entity> for Entity {
+    fn to() -> RelationDef {
+        Relation::Table.def()
+    }
+}
+
+impl ActiveModelBehavior for ActiveModel {}

src/meta/node/src/server.rs

@@ -35,7 +35,7 @@ use risingwave_meta::manager::{META_NODE_ID, MetadataManager};
 use risingwave_meta::rpc::ElectionClientRef;
 use risingwave_meta::rpc::election::dummy::DummyElectionClient;
 use risingwave_meta::rpc::intercept::MetricsMiddlewareLayer;
-use risingwave_meta::stream::ScaleController;
+use risingwave_meta::stream::{GlobalRefreshManager, ScaleController};
 use risingwave_meta_service::AddressInfo;
 use risingwave_meta_service::backup_service::BackupServiceImpl;
 use risingwave_meta_service::cloud_service::CloudServiceImpl;
@@ -506,6 +506,14 @@ pub async fn start_service_as_election_leader(
         env.opts.iceberg_gc_interval_sec,
     ));
 
+    let (refresh_manager, refresh_handle, refresh_shutdown) = GlobalRefreshManager::start(
+        metadata_manager.clone(),
+        barrier_scheduler.clone(),
+        env.shared_actor_infos().clone(),
+    )
+    .await?;
+    sub_tasks.push((refresh_handle, refresh_shutdown));
+
     let scale_controller = Arc::new(ScaleController::new(
         &metadata_manager,
         source_manager.clone(),
@@ -521,6 +529,7 @@ pub async fn start_service_as_election_leader(
         sink_manager.clone(),
         scale_controller.clone(),
         barrier_scheduler.clone(),
+        refresh_manager.clone(),
     )
     .await;
     tracing::info!("GlobalBarrierManager started");
@@ -579,6 +588,7 @@ pub async fn start_service_as_election_leader(
         barrier_manager.clone(),
         stream_manager.clone(),
         metadata_manager.clone(),
+        refresh_manager.clone(),
     );
     let sink_coordination_srv = SinkCoordinationServiceImpl::new(sink_manager);
     let hummock_srv = HummockServiceImpl::new(

src/meta/service/src/stream_service.rs

@@ -25,7 +25,7 @@ use risingwave_meta::controller::fragment::StreamingJobInfo;
 use risingwave_meta::controller::utils::FragmentDesc;
 use risingwave_meta::manager::MetadataManager;
 use risingwave_meta::model::ActorId;
-use risingwave_meta::stream::{SourceManagerRunningInfo, ThrottleConfig};
+use risingwave_meta::stream::{GlobalRefreshManagerRef, SourceManagerRunningInfo, ThrottleConfig};
 use risingwave_meta::{MetaError, model};
 use risingwave_meta_model::{FragmentId, ObjectId, SinkId, SourceId, StreamingParallelism};
 use risingwave_pb::meta::alter_connector_props_request::AlterConnectorPropsObject;
@@ -55,6 +55,7 @@ pub struct StreamServiceImpl {
     barrier_manager: BarrierManagerRef,
     stream_manager: GlobalStreamManagerRef,
     metadata_manager: MetadataManager,
+    refresh_manager: GlobalRefreshManagerRef,
 }
 
 impl StreamServiceImpl {
@@ -64,13 +65,15 @@ impl StreamServiceImpl {
         barrier_manager: BarrierManagerRef,
         stream_manager: GlobalStreamManagerRef,
         metadata_manager: MetadataManager,
+        refresh_manager: GlobalRefreshManagerRef,
     ) -> Self {
         StreamServiceImpl {
             env,
             barrier_scheduler,
             barrier_manager,
             stream_manager,
             metadata_manager,
+            refresh_manager,
         }
     }
 }
@@ -486,7 +489,7 @@ impl StreamManagerService for StreamServiceImpl {
                 .chain(source_fragments.values().flatten().copied())
                 .collect();
 
-            let guard = self.env.shared_actor_info.read_guard();
+            let guard = self.env.shared_actor_infos().read_guard();
             guard
                 .iter_over_fragments()
                 .filter(|(frag_id, _)| all_fragment_ids.contains(frag_id))
@@ -588,14 +591,9 @@ impl StreamManagerService for StreamServiceImpl {
 
         tracing::info!("Refreshing table with id: {}", req.table_id);
 
-        // Create refresh manager and execute refresh
-        let refresh_manager = risingwave_meta::stream::RefreshManager::new(
-            self.metadata_manager.clone(),
-            self.barrier_scheduler.clone(),
-        );
-
-        let response = refresh_manager
-            .refresh_table(req, self.env.shared_actor_infos())
+        let response = self
+            .refresh_manager
+            .trigger_manual_refresh(req, self.env.shared_actor_infos())
             .await?;
 
         Ok(Response::new(response))

src/meta/src/barrier/context/context_impl.rs

@@ -15,7 +15,7 @@
 use std::collections::{HashMap, HashSet};
 use std::sync::Arc;
 
-use anyhow::{Context, anyhow};
+use anyhow::Context;
 use risingwave_common::catalog::{DatabaseId, FragmentTypeFlag, TableId};
 use risingwave_common::id::JobId;
 use risingwave_meta_model::ActorId;
@@ -28,6 +28,7 @@ use risingwave_pb::stream_service::barrier_complete_response::{
 use risingwave_pb::stream_service::streaming_control_stream_request::PbInitRequest;
 use risingwave_rpc_client::StreamingControlHandle;
 
+use crate::MetaResult;
 use crate::barrier::command::CommandContext;
 use crate::barrier::context::{GlobalBarrierWorkerContext, GlobalBarrierWorkerContextImpl};
 use crate::barrier::progress::TrackingJob;
@@ -39,8 +40,7 @@ use crate::barrier::{
 };
 use crate::hummock::CommitEpochInfo;
 use crate::model::FragmentDownstreamRelation;
-use crate::stream::{REFRESH_TABLE_PROGRESS_TRACKER, SourceChange, SplitState};
-use crate::{MetaError, MetaResult};
+use crate::stream::{SourceChange, SplitState};
 
 impl GlobalBarrierWorkerContext for GlobalBarrierWorkerContextImpl {
     #[await_tree::instrument]
@@ -134,17 +134,9 @@ impl GlobalBarrierWorkerContext for GlobalBarrierWorkerContextImpl {
         }
 
         for ((table_id, associated_source_id), actors) in list_finished_info {
-            let allow_yield = {
-                let mut lock_handle = REFRESH_TABLE_PROGRESS_TRACKER.lock();
-                let single_task_tracker =
-                    lock_handle.inner.get_mut(&table_id).ok_or_else(|| {
-                        MetaError::from(anyhow!("Table tracker not found for table {}", table_id))
-                    })?;
-                single_task_tracker.report_list_finished(actors.iter().copied());
-                let allow_yield = single_task_tracker.is_list_finished()?;
-
-                Ok::<_, MetaError>(allow_yield)
-            }?;
+            let allow_yield = self
+                .refresh_manager
+                .mark_list_stage_finished(table_id, &actors)?;
 
             if !allow_yield {
                 continue;
@@ -194,17 +186,9 @@ impl GlobalBarrierWorkerContext for GlobalBarrierWorkerContextImpl {
         }
 
         for ((table_id, associated_source_id), actors) in load_finished_info {
-            let allow_yield = {
-                let mut lock_handle = REFRESH_TABLE_PROGRESS_TRACKER.lock();
-                let single_task_tracker =
-                    lock_handle.inner.get_mut(&table_id).ok_or_else(|| {
-                        MetaError::from(anyhow!("Table tracker not found for table {}", table_id))
-                    })?;
-                single_task_tracker.report_load_finished(actors.iter().copied());
-                let allow_yield = single_task_tracker.is_load_finished()?;
-
-                Ok::<_, MetaError>(allow_yield)
-            }?;
+            let allow_yield = self
+                .refresh_manager
+                .mark_load_stage_finished(table_id, &actors)?;
 
             if !allow_yield {
                 continue;
@@ -244,29 +228,18 @@ impl GlobalBarrierWorkerContext for GlobalBarrierWorkerContextImpl {
         refresh_finished_table_job_ids: Vec<JobId>,
     ) -> MetaResult<()> {
         for job_id in refresh_finished_table_job_ids {
-            {
-                let table_id = &job_id.as_mv_table_id();
-                let mut lock_handle = REFRESH_TABLE_PROGRESS_TRACKER.lock();
-                let remove_res = lock_handle.inner.remove(table_id);
-                debug_assert!(remove_res.is_some());
-
-                // try remove the table_id from the table_id_by_database_id
-                lock_handle
-                    .table_id_by_database_id
-                    .values_mut()
-                    .for_each(|table_ids| {
-                        table_ids.remove(table_id);
-                    });
-            }
+            let table_id = job_id.as_mv_table_id();
+
+            self.refresh_manager.mark_refresh_complete(table_id).await?;
 
             // Update the table's refresh state back to Idle (refresh complete)
             self.metadata_manager
                 .catalog_controller
-                .set_table_refresh_state(job_id.as_mv_table_id(), RefreshState::Idle)
+                .set_table_refresh_state(table_id, RefreshState::Idle)
                 .await
                 .context("Failed to set table refresh state to Idle")?;
 
-            tracing::info!(%job_id, "Table refresh completed, state updated to Idle");
+            tracing::info!(%job_id, %table_id, "Table refresh completed, state updated to Idle");
         }
 
         Ok(())
@@ -309,9 +282,16 @@ impl CommandContext {
             }
 
             Command::DropStreamingJobs {
+                streaming_job_ids,
                 unregistered_state_table_ids,
                 ..
             } => {
+                for job_id in streaming_job_ids {
+                    barrier_manager_context
+                        .refresh_manager
+                        .remove_progress_tracker(job_id.as_mv_table_id());
+                }
+
                 barrier_manager_context
                     .hummock_manager
                     .unregister_table_ids(unregistered_state_table_ids.iter().cloned())