feat(meta): decouple barrier collect and sync in global barrier manager

ci/workflows/main-cron.yml

@@ -143,7 +143,7 @@ steps:
           files: "*-junit.xml"
           format: "junit"
       - ./ci/plugins/upload-failure-logs
-    timeout_in_minutes: 13
+    timeout_in_minutes: 15
     retry: *auto-retry
 
   - label: "end-to-end test (parallel, in-memory) (release)"

ci/workflows/pull-request.yml

@@ -565,7 +565,7 @@ steps:
           config: ci/docker-compose.yml
           mount-buildkite-agent: true
       - ./ci/plugins/upload-failure-logs
-    timeout_in_minutes: 20
+    timeout_in_minutes: 25
     retry: *auto-retry
 
   - label: "end-to-end test (deterministic simulation)"

proto/stream_service.proto

@@ -22,7 +22,7 @@
   repeated stream_plan.SubscriptionUpstreamInfo subscriptions_to_remove = 11;
 }
 
-message BarrierCompleteResponse {
+message BarrierCollectResponse {
   message CreateMviewProgress {
     // Note: ideally we should use `executor_id`, but `actor_id` is ok-ish.
     // See <https://github.com/risingwavelabs/risingwave/issues/6236>.
@@ -34,9 +34,19 @@
     uint64 consumed_rows = 4;
     uint32 pending_barrier_num = 5;
   }
-  string request_id = 1;
-  common.Status status = 2;
+  uint64 partial_graph_id = 1;
+  // prev_epoch of barrier
+  uint64 epoch = 2;
   repeated CreateMviewProgress create_mview_progress = 3;
+}
+
+message BarrierCompleteRequest {
+  uint64 task_id = 1;
+  map<uint64, uint64> partial_graph_sync_epochs = 2;
+}
+
+message BarrierCompleteResponse {
+  uint64 task_id = 1;
   message LocalSstableInfo {
     reserved 1;
     reserved "compaction_group_id";
@@ -48,9 +58,6 @@
   uint32 worker_id = 5;
   map<uint32, hummock.TableWatermarks> table_watermarks = 6;
   repeated hummock.SstableInfo old_value_sstables = 7;
-  uint64 partial_graph_id = 8;
-  // prev_epoch of barrier
-  uint64 epoch = 9;
 }
 
 message WaitEpochCommitRequest {
@@ -85,6 +92,7 @@
     InjectBarrierRequest inject_barrier = 2;
     RemovePartialGraphRequest remove_partial_graph = 3;
     CreatePartialGraphRequest create_partial_graph = 4;
+    BarrierCompleteRequest complete_barrier = 5;
   }
 }
 
@@ -96,6 +104,7 @@
     InitResponse init = 1;
     BarrierCompleteResponse complete_barrier = 2;
     ShutdownResponse shutdown = 3;
+    BarrierCollectResponse collect_barrier = 4;
   }
 }
 

src/compute/src/rpc/service/monitor_service.rs

@@ -86,10 +86,16 @@ impl MonitorService for MonitorServiceImpl {
             Default::default()
         };
 
+        let mut barrier_traces_next_key = 0;
+        let barrier_traces_next_key = &mut barrier_traces_next_key;
         let barrier_traces = if let Some(reg) = self.stream_mgr.await_tree_reg() {
             reg.collect::<BarrierAwait>()
                 .into_iter()
-                .map(|(k, v)| (k.prev_epoch, v.to_string()))
+                .map(|(k, v)| {
+                    let key = *barrier_traces_next_key;
+                    *barrier_traces_next_key += 1;
+                    (key, format!("{:?}", (k.sync_graph_epochs, v.to_string())))
+                })
                 .collect()
         } else {
             Default::default()