From e1b6e6419cb7d16f03eb9a7b079c97cbbd25a71e Mon Sep 17 00:00:00 2001 From: LFC <990479+MichaelScofield@users.noreply.github.com> Date: Wed, 14 Oct 2026 14:07:49 +0800 Subject: [PATCH] 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 f41e86f43a7c3f78d12782df0d65a0a23e04ce4c) Co-authored-by: Ning Sun --- src/common/function/src/admin/gc.rs | 3 +- src/common/function/src/admin/purge_table.rs | 6 ++- src/common/function/src/handlers.rs | 6 ++- src/common/function/src/state.rs | 1 + src/meta-client/src/client.rs | 44 +++++++++++----- src/meta-client/src/client/procedure.rs | 4 +- src/operator/src/procedure.rs | 53 ++++++++++++++++++-- 7 files changed, 97 insertions(+), 20 deletions(-) diff --git a/src/common/function/src/admin/gc.rs b/src/common/function/src/admin/gc.rs index 0090140a460..63b249cb54d 100644 --- a/src/common/function/src/admin/gc.rs +++ b/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!() diff --git a/src/common/function/src/admin/purge_table.rs b/src/common/function/src/admin/purge_table.rs index 7e5fcd84371..88ed26ae9f6 100644 --- a/src/common/function/src/admin/purge_table.rs +++ b/src/common/function/src/admin/purge_table.rs @@ -157,7 +157,11 @@ mod tests { async fn query_procedure_state(&self, _: &str) -> Result { 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 { diff --git a/src/common/function/src/handlers.rs b/src/common/function/src/handlers.rs index 6ee8064d6c7..9dafd853a84 100644 --- a/src/common/function/src/handlers.rs +++ b/src/common/function/src/handlers.rs @@ -105,7 +105,11 @@ pub trait ProcedureServiceHandler: Send + Sync { async fn query_procedure_state(&self, pid: &str) -> Result; /// 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; diff --git a/src/common/function/src/state.rs b/src/common/function/src/state.rs index f60b37dd36d..bf2405930c3 100644 --- a/src/common/function/src/state.rs +++ b/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(()) diff --git a/src/meta-client/src/client.rs b/src/meta-client/src/client.rs index e0932af5a69..44b6b21dd23 100644 --- a/src/meta-client/src/client.rs +++ b/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; /// 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 } } diff --git a/src/meta-client/src/client/procedure.rs b/src/meta-client/src/client/procedure.rs index 9c32b4fc880..222a525dcbb 100644 --- a/src/meta-client/src/client/procedure.rs +++ b/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 { +pub fn procedure_event_context(context: &ExecutorContext) -> Option { 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 Option { +pub fn procedure_actor(context: &ExecutorContext) -> Option { context.actor.as_ref().map(|username| ProcedureActor { username: username.clone(), }) diff --git a/src/operator/src/procedure.rs b/src/operator/src/procedure.rs index 03fb4f909c9..bf9ae106080 100644 --- a/src/operator/src/procedure.rs +++ b/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 { 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(