Repository: risingwavelabs/risingwave
File: src/meta/src/barrier/context/recovery.rs (Rust)

Code before:
            })
        else {
            return Ok(None);
        };

        let (state_table_committed_epochs, state_table_log_epochs) = self
            .hummock_manager
            .on_current_version(|version| {
                Self::resolve_hummock_version_epochs(
                    background_jobs
                        .keys()
                        .map(|job_id| (*job_id, &info[job_id])),
                    version,
                )
            })
            .await?;

        let mv_depended_subscriptions = self
            .metadata_manager
            .get_mv_depended_subscriptions(Some(database_id))
            .await?;

        let fragment_relations = self
            .metadata_manager
            .catalog_controller
            .get_fragment_downstream_relations(
                info.values()
                    .flatten()
                    .map(|(fragment_id, _)| *fragment_id as _)
                    .collect(),
            )
            .await?;

        // get split assignments for all actors
        let mut source_splits = HashMap::new();
        for (_, fragment) in info.values().flatten() {
            for (actor_id, info) in &fragment.actors {
                source_splits.insert(*actor_id, info.splits.clone());
            }
        }

        let cdc_table_snapshot_splits =
            reload_cdc_table_snapshot_splits(&self.env.meta_store_ref().conn, Some(database_id))
                .await?;

        self.refresh_manager
            .remove_trackers_by_database(database_id);

        Ok(Some(DatabaseRuntimeInfoSnapshot {
            job_infos: info,
            state_table_committed_epochs,

Review comment (refactor):
Can we try further refactor the code of database recovery, to only reload all necessary metadata in async context of `reload_database_runtime_info`, and after `reload_database_runtime_info` finishes we then render the actors and split assignments? In current implementation, there are still async method calls after `render_actor_assignments`, which are `recovery_table_with_upstream_sinks` and `load_stream_actors`. 

Though `recovery_table_with_upstream_sinks` takes `InflightDatabaseInfo` as input, it doesn't actually depend on having actor rendered, and it only modifies the fragment stream node of fragments in `InflightDatabaseInfo`.  `load_stream_actors` can be non-async if we already load `job_extra_info` when we reload it along with other metadata under async contexts.

Code after:
        let background_jobs = self
            .list_background_job_progress(Some(database_id))
            .await
            .context("recover background job progress of database should not fail")?;
        tracing::info!(?database_id, "recovered background job progress");

        // This is a quick path to accelerate the process of dropping and canceling streaming jobs.
        let dropped_table_ids = self
            .scheduled_barriers
            .pre_apply_drop_cancel(Some(database_id));
        self.metadata_manager
            .catalog_controller
            .complete_dropped_tables(dropped_table_ids)
            .await;

        let adaptive_parallelism_strategy = {
            let system_params_reader = self
                .metadata_manager
                .catalog_controller
                .env
                .system_params_reader()
                .await;
            system_params_reader.adaptive_parallelism_strategy()
        };

        let active_streaming_nodes =
            ActiveStreamingWorkerNodes::new_snapshot(self.metadata_manager.clone()).await?;

        let loaded = self
            .load_fragment_context(Some(database_id))
            .await
            .inspect_err(|err| {
                warn!(error = %err.as_report(), "load fragment context failed");
            })?;

        let mut all_info = self
            .render_actor_assignments(
                Some(database_id),
                &loaded,
                &active_streaming_nodes,
                adaptive_parallelism_strategy,
            )
            .inspect_err(|err| {
                warn!(error = %err.as_report(), "render actor assignments failed");
            })?;

        let mut database_info = all_info
            .remove(&database_id)
            .map_or_else(HashMap::new, |table_map| {
                HashMap::from([(database_id, table_map)])
            });