refactor(stream): own and recreate shared context and local barrier manager in recovery

src/compute/src/rpc/service/config_service.rs

@@ -34,7 +34,7 @@ impl ConfigService for ConfigServiceImpl {
     ) -> Result<Response<ShowConfigResponse>, Status> {
         let batch_config = serde_json::to_string(self.batch_mgr.config())
             .map_err(|e| e.to_status(Code::Internal, "compute"))?;
-        let stream_config = serde_json::to_string(&self.stream_mgr.context().config())
+        let stream_config = serde_json::to_string(&self.stream_mgr.env.config())
             .map_err(|e| e.to_status(Code::Internal, "compute"))?;
 
         let show_config_response = ShowConfigResponse {

src/compute/src/rpc/service/exchange_service.rs

@@ -106,8 +106,8 @@ impl ExchangeService for ExchangeServiceImpl {
 
         let receiver = self
             .stream_mgr
-            .context()
-            .take_receiver((up_actor_id, down_actor_id))?;
+            .take_receiver((up_actor_id, down_actor_id))
+            .await?;
 
         // Map the remaining stream to add-permits.
         let add_permits_stream = request_stream.map_ok(|req| match req.value.unwrap() {

src/compute/src/rpc/service/stream_service.rs

@@ -84,7 +84,7 @@ impl StreamService for StreamServiceImpl {
     ) -> std::result::Result<Response<BroadcastActorInfoTableResponse>, Status> {
         let req = request.into_inner();
 
-        let res = self.mgr.update_actor_info(&req.info);
+        let res = self.mgr.update_actor_info(req.info).await;
         match res {
             Err(e) => {
                 error!(error = %e.as_report(), "failed to update actor info table actor");