feat(meta-client): forward actor and event context for region follower RPCs [Backport release/v1.3] (#9571)

feat(meta-client): forward actor and event context for region follower RPCs (#9568)

The manage-region-follower chain submitted its procedures with an empty
executor context: `ProcedureServiceHandler::manage_region_follower`
accepted no query context, the operator built an
`ExecutorContext::default()`, and `MetaClient` dropped the context before
delegating to `RegionFollowerClient`. The submitting user (actor) and the
procedure event metadata therefore never reached the submitted
procedures.

- `ProcedureServiceHandler::manage_region_follower` now takes a
  `QueryContextRef`, consistent with `migrate_region`, `gc_regions` and
  `gc_table`.
- `ProcedureServiceOperator` derives the `ExecutorContext` (actor and
  `TriggerReason::Manual` event input) from the query context.
- The `RegionFollowerClient` trait methods now accept `&ExecutorContext`
  and `MetaClient` forwards it, so client implementations can transport
  the actor and event context on the wire.
- `procedure_event_context`/`procedure_actor` are now public so client
  implementations can reuse the trusted protocol derivation.

(cherry picked from commit f41e86f43a)

Co-authored-by: Ning Sun <sunng@protonmail.com>

src/common/function/src/admin/gc.rs

@@ -191,7 +191,7 @@ mod tests {
     use common_meta::rpc::procedure::{
         GcResponse, ManageRegionFollowerRequest, MigrateRegionRequest, ProcedureStateResponse,
     };
-    use session::context::QueryContext;
+    use session::context::{QueryContext, QueryContextRef};
 
     use super::*;
     use crate::handlers::ProcedureServiceHandler;
@@ -298,6 +298,7 @@ mod tests {
 
         async fn manage_region_follower(
             &self,
+            _query_ctx: QueryContextRef,
             _request: ManageRegionFollowerRequest,
         ) -> Result<()> {
             unreachable!()

src/common/function/src/admin/purge_table.rs

@@ -157,7 +157,11 @@ mod tests {
         async fn query_procedure_state(&self, _: &str) -> Result<ProcedureStateResponse> {
             unreachable!()
         }
-        async fn manage_region_follower(&self, _: ManageRegionFollowerRequest) -> Result<()> {
+        async fn manage_region_follower(
+            &self,
+            _query_ctx: QueryContextRef,
+            _request: ManageRegionFollowerRequest,
+        ) -> Result<()> {
             unreachable!()
         }
         fn catalog_manager(&self) -> &CatalogManagerRef {

src/common/function/src/handlers.rs

@@ -105,7 +105,11 @@ pub trait ProcedureServiceHandler: Send + Sync {
     async fn query_procedure_state(&self, pid: &str) -> Result<ProcedureStateResponse>;
 
     /// Manage a region follower to a region.
-    async fn manage_region_follower(&self, request: ManageRegionFollowerRequest) -> Result<()>;
+    async fn manage_region_follower(
+        &self,
+        query_ctx: QueryContextRef,
+        request: ManageRegionFollowerRequest,
+    ) -> Result<()>;
 
     /// Get the catalog manager
     fn catalog_manager(&self) -> &CatalogManagerRef;

src/common/function/src/state.rs

@@ -92,6 +92,7 @@ impl FunctionState {
 
             async fn manage_region_follower(
                 &self,
+                _query_ctx: QueryContextRef,
                 _request: ManageRegionFollowerRequest,
             ) -> Result<()> {
                 Ok(())

src/meta-client/src/client.rs

@@ -13,12 +13,11 @@
 // limitations under the License.
 
 mod ask_leader;
+mod cluster;
 mod config;
 pub mod heartbeat;
 mod load_balance;
 mod procedure;
-
-mod cluster;
 mod store;
 mod util;
 
@@ -69,7 +68,8 @@ use common_time::util::DefaultSystemTimer;
 use config::Client as ConfigClient;
 use futures::TryStreamExt;
 use heartbeat::{Client as HeartbeatClient, HeartbeatConfig};
-use procedure::{Client as ProcedureClient, procedure_actor, procedure_event_context};
+use procedure::Client as ProcedureClient;
+pub use procedure::{procedure_actor, procedure_event_context};
 use serde::de::DeserializeOwned;
 use snafu::{OptionExt, ResultExt};
 use store::Client as StoreClient;
@@ -318,15 +318,35 @@ impl MetaClient {
 pub type RegionFollowerClientRef = Arc<dyn RegionFollowerClient>;
 
 /// A trait for clients that can manage region followers.
+///
+/// The executor context carries the actor and the event input of the
+/// submission, so implementations can forward them to the metasrv and the
+/// region follower procedures record who submitted the operation.
 #[async_trait::async_trait]
 pub trait RegionFollowerClient: Sync + Send + Debug {
-    async fn add_region_follower(&self, request: AddRegionFollowerRequest) -> Result<()>;
+    async fn add_region_follower(
+        &self,
+        ctx: &ExecutorContext,
+        request: AddRegionFollowerRequest,
+    ) -> Result<()>;
 
-    async fn remove_region_follower(&self, request: RemoveRegionFollowerRequest) -> Result<()>;
+    async fn remove_region_follower(
+        &self,
+        ctx: &ExecutorContext,
+        request: RemoveRegionFollowerRequest,
+    ) -> Result<()>;
 
-    async fn add_table_follower(&self, request: AddTableFollowerRequest) -> Result<()>;
+    async fn add_table_follower(
+        &self,
+        ctx: &ExecutorContext,
+        request: AddTableFollowerRequest,
+    ) -> Result<()>;
 
-    async fn remove_table_follower(&self, request: RemoveTableFollowerRequest) -> Result<()>;
+    async fn remove_table_follower(
+        &self,
+        ctx: &ExecutorContext,
+        request: RemoveTableFollowerRequest,
+    ) -> Result<()>;
 
     async fn start(&self, urls: &[&str]) -> Result<()>;
 
@@ -370,31 +390,31 @@ impl ProcedureExecutor for MetaClient {
 
     async fn manage_region_follower(
         &self,
-        _ctx: &ExecutorContext,
+        ctx: &ExecutorContext,
         request: ManageRegionFollowerRequest,
     ) -> MetaResult<()> {
         if let Some(region_follower) = &self.region_follower {
             match request {
                 ManageRegionFollowerRequest::AddRegionFollower(add_region_follower_request) => {
                     region_follower
-                        .add_region_follower(add_region_follower_request)
+                        .add_region_follower(ctx, add_region_follower_request)
                         .await
                 }
                 ManageRegionFollowerRequest::RemoveRegionFollower(
                     remove_region_follower_request,
                 ) => {
                     region_follower
-                        .remove_region_follower(remove_region_follower_request)
+                        .remove_region_follower(ctx, remove_region_follower_request)
                         .await
                 }
                 ManageRegionFollowerRequest::AddTableFollower(add_table_follower_request) => {
                     region_follower
-                        .add_table_follower(add_table_follower_request)
+                        .add_table_follower(ctx, add_table_follower_request)
                         .await
                 }
                 ManageRegionFollowerRequest::RemoveTableFollower(remove_table_follower_request) => {
                     region_follower
-                        .remove_table_follower(remove_table_follower_request)
+                        .remove_table_follower(ctx, remove_table_follower_request)
                         .await
                 }
             }

src/meta-client/src/client/procedure.rs

@@ -49,7 +49,7 @@ use crate::error::Result;
 ///
 /// The caller can only supply reason/extensions. Protocol is derived here from
 /// the trusted, typed query channel held locally in the executor context.
-pub(crate) fn procedure_event_context(context: &ExecutorContext) -> Option<ProcedureEventContext> {
+pub fn procedure_event_context(context: &ExecutorContext) -> Option<ProcedureEventContext> {
     context.event_input.as_ref().map(|input| {
         let mut event_context = ProcedureEventContext::from(input);
         event_context.protocol = context
@@ -62,7 +62,7 @@ pub(crate) fn procedure_event_context(context: &ExecutorContext) -> Option<Proce
 }
 
 /// Builds the optional procedure actor transported by a procedure RPC.
-pub(crate) fn procedure_actor(context: &ExecutorContext) -> Option<ProcedureActor> {
+pub fn procedure_actor(context: &ExecutorContext) -> Option<ProcedureActor> {
     context.actor.as_ref().map(|username| ProcedureActor {
         username: username.clone(),
     })

src/operator/src/procedure.rs

@@ -124,10 +124,12 @@ impl ProcedureServiceHandler for ProcedureServiceOperator {
 
     async fn manage_region_follower(
         &self,
+        query_ctx: QueryContextRef,
         request: ManageRegionFollowerRequest,
     ) -> QueryResult<()> {
+        let executor_context = to_executor_context(query_ctx, TriggerReason::Manual);
         self.procedure_executor
-            .manage_region_follower(&ExecutorContext::default(), request)
+            .manage_region_follower(&executor_context, request)
             .await
             .map_err(BoxedError::new)
             .context(query_error::ProcedureServiceSnafu)
@@ -178,8 +180,11 @@ mod tests {
         ExecutorContext, ProcedureExecutor, ProcedureExecutorRef,
     };
     use common_meta::rpc::ddl::{DdlTask, SubmitDdlTaskRequest, SubmitDdlTaskResponse};
-    use common_meta::rpc::procedure::{MigrateRegionResponse, ProcedureStateResponse};
-    use session::context::QueryContextBuilder;
+    use common_meta::rpc::procedure::{
+        AddRegionFollowerRequest, ManageRegionFollowerRequest, MigrateRegionResponse,
+        ProcedureStateResponse,
+    };
+    use session::context::{QueryContextBuilder, QueryContextRef};
     use table::table_name::TableName;
 
     use super::*;
@@ -229,6 +234,14 @@ mod tests {
         ) -> common_meta::error::Result<ProcedureStateResponse> {
             unimplemented!()
         }
+        async fn manage_region_follower(
+            &self,
+            context: &ExecutorContext,
+            _: ManageRegionFollowerRequest,
+        ) -> common_meta::error::Result<()> {
+            self.contexts.lock().unwrap().push(context.clone());
+            Ok(())
+        }
         async fn list_procedures(
             &self,
             _: &ExecutorContext,
@@ -303,6 +316,40 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn test_manage_region_follower_forwards_query_context() {
+        let manager = Arc::new(TableMetadataManager::new(Arc::new(
+            MemoryKvBackend::default(),
+        )));
+        let executor = Arc::new(RecordingProcedureExecutor::default());
+        let operator = ProcedureServiceOperator::new(
+            executor.clone(),
+            catalog::memory::MemoryCatalogManager::new(),
+            manager,
+        );
+        let query_ctx: QueryContextRef = QueryContextBuilder::default().build().into();
+
+        operator
+            .manage_region_follower(
+                query_ctx,
+                ManageRegionFollowerRequest::AddRegionFollower(AddRegionFollowerRequest {
+                    region_id: 1,
+                    peer_id: 2,
+                }),
+            )
+            .await
+            .unwrap();
+
+        let contexts = executor.contexts.lock().unwrap();
+        assert_eq!(contexts.len(), 1);
+        // The default query context runs as the default user.
+        assert_eq!(contexts[0].actor.as_deref(), Some("greptime"));
+        assert_eq!(
+            contexts[0].event_input.as_ref().map(|input| input.reason),
+            Some(TriggerReason::Manual)
+        );
+    }
+
     #[tokio::test]
     async fn test_purge_table_missing_tombstone_is_table_not_found() {
         let manager = Arc::new(TableMetadataManager::new(Arc::new(