perf(gc): pack file reference exchange (#9009)

* perf(gc): pack file reference exchange

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* fix(gc): address packed reference review feedback

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* fix(gc): stop without retry when maintenance is enabled

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* fix(meta): avoid logging malformed mailbox payloads

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

---------

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

src/common/meta/src/error.rs

@@ -269,6 +269,20 @@ pub enum Error {
         error: prost::DecodeError,
     },
 
+    #[snafu(display("Failed to decode packed file references"))]
+    DecodePackedFileRefs {
+        #[snafu(implicit)]
+        location: Location,
+        #[snafu(source)]
+        error: base64::DecodeError,
+    },
+
+    #[snafu(display("Invalid packed file references framing"))]
+    InvalidPackedFileRefs {
+        #[snafu(implicit)]
+        location: Location,
+    },
+
     #[snafu(display("Failed to encode object into json"))]
     EncodeJson {
         #[snafu(implicit)]
@@ -1212,6 +1226,8 @@ impl ErrorExt for Error {
             | PayloadNotExist { .. }
             | ConvertRawKey { .. }
             | DecodeProto { .. }
+            | DecodePackedFileRefs { .. }
+            | InvalidPackedFileRefs { .. }
             | BuildTableMeta { .. }
             | TableRouteNotFound { .. }
             | TableRepartNotFound { .. }

src/common/meta/src/heartbeat/handler.rs

@@ -94,7 +94,21 @@ impl HeartbeatResponseHandlerExecutor for HandlerGroupExecutor {
                 Ok(HandleControl::Done) => break,
                 Ok(HandleControl::Continue) => {}
                 Err(e) => {
-                    error!(e;"Error while handling: {:?}", ctx.response);
+                    let mailbox_message_id = ctx
+                        .response
+                        .mailbox_message
+                        .as_ref()
+                        .map(|message| message.id);
+                    let json_payload_len =
+                        ctx.response.mailbox_message.as_ref().and_then(|message| {
+                            message.payload.as_ref().map(|payload| match payload {
+                                api::v1::meta::mailbox_message::Payload::Json(json) => json.len(),
+                            })
+                        });
+                    error!(
+                        %e;
+                        "Error while handling heartbeat response: mailbox_message_id={mailbox_message_id:?}, json_payload_len={json_payload_len:?}"
+                    );
                     break;
                 }
             }
@@ -102,3 +116,86 @@ impl HeartbeatResponseHandlerExecutor for HandlerGroupExecutor {
         Ok(())
     }
 }
+
+#[cfg(test)]
+mod error_log_tests {
+    use std::fmt::Debug;
+    use std::sync::{Arc, Mutex};
+
+    use api::v1::meta::mailbox_message::Payload;
+    use api::v1::meta::{HeartbeatResponse, MailboxMessage};
+    use common_telemetry::tracing::field::{Field, Visit};
+    use common_telemetry::tracing::{Event, Subscriber};
+    use common_telemetry::tracing_subscriber::layer::{Context, SubscriberExt};
+    use common_telemetry::{tracing, tracing_subscriber};
+
+    use super::parse_mailbox_message::ParseMailboxMessageHandler;
+    use super::*;
+    use crate::heartbeat::mailbox::HeartbeatMailbox;
+
+    #[derive(Clone, Default)]
+    struct LogCapture(Arc<Mutex<Vec<String>>>);
+
+    impl<S> tracing_subscriber::Layer<S> for LogCapture
+    where
+        S: Subscriber,
+    {
+        fn on_event(&self, event: &Event<'_>, _ctx: Context<'_, S>) {
+            let mut visitor = FieldVisitor::default();
+            event.record(&mut visitor);
+            self.0.lock().unwrap().push(visitor.fields.join(", "));
+        }
+    }
+
+    #[derive(Default)]
+    struct FieldVisitor {
+        fields: Vec<String>,
+    }
+
+    impl Visit for FieldVisitor {
+        fn record_debug(&mut self, field: &Field, value: &dyn Debug) {
+            self.fields.push(format!("{}={value:?}", field.name()));
+        }
+
+        fn record_str(&mut self, field: &Field, value: &str) {
+            self.fields.push(format!("{}={value:?}", field.name()));
+        }
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn test_handler_error_log_excludes_mailbox_payload() {
+        let payload_sentinel = "MALFORMED_PACKED_PAYLOAD_MUST_NOT_BE_LOGGED".repeat(256);
+        let payload = format!(
+            r#"{{"PackedGcRegions":{{"regions":[],"packed_file_refs_manifest":"{}"}}"#,
+            payload_sentinel
+        );
+        let payload_len = payload.len();
+        let capture = LogCapture::default();
+        let subscriber = tracing_subscriber::registry().with(capture.clone());
+        let _guard = tracing::subscriber::set_default(subscriber);
+        let (mailbox_tx, _) = tokio::sync::mpsc::channel(1);
+        let ctx = HeartbeatResponseHandlerContext::new(
+            Arc::new(HeartbeatMailbox::new(mailbox_tx)),
+            HeartbeatResponse {
+                mailbox_message: Some(MailboxMessage {
+                    id: 42,
+                    subject: "unsafe subject".to_string(),
+                    to: "unsafe recipient".to_string(),
+                    from: "unsafe sender".to_string(),
+                    payload: Some(Payload::Json(payload)),
+                    ..Default::default()
+                }),
+                ..Default::default()
+            },
+        );
+        let executor = HandlerGroupExecutor::new(vec![Arc::new(ParseMailboxMessageHandler)]);
+
+        executor.handle(ctx).await.unwrap();
+
+        let logs = capture.0.lock().unwrap().join("\n");
+        assert!(logs.contains("Error while handling heartbeat response"));
+        assert!(logs.contains("mailbox_message_id=Some(42)"));
+        assert!(logs.contains(&format!("json_payload_len=Some({payload_len})")));
+        assert!(!logs.contains(&payload_sentinel));
+    }
+}

src/common/meta/src/instruction.rs

@@ -16,16 +16,19 @@ use std::collections::{HashMap, HashSet};
 use std::fmt::{Display, Formatter};
 use std::time::Duration;
 
+use base64::Engine;
 use common_error::ext::{ErrorExt, RetryHint};
 use common_error::status_code::StatusCode;
 use serde::{Deserialize, Deserializer, Serialize, Serializer};
+use snafu::ResultExt as _;
 use store_api::region_engine::SyncRegionFromRequest;
 use store_api::region_request::{RegionFlushReason, RegionRequirements};
-use store_api::storage::{FileRefsManifest, GcReport, RegionId, RegionNumber};
+use store_api::storage::{FileId, FileRef, FileRefsManifest, GcReport, RegionId, RegionNumber};
 use strum::Display;
 use table::metadata::TableId;
 use table::table_name::TableName;
 
+use crate::error::{DecodePackedFileRefsSnafu, InvalidPackedFileRefsSnafu};
 use crate::flow_name::FlowName;
 use crate::key::schema_name::SchemaName;
 use crate::key::{FlowId, FlowPartitionId};
@@ -579,6 +582,24 @@ impl Display for GetFileRefs {
     }
 }
 
+/// Instruction to get file references using the packed manifest format.
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct GetPackedFileRefs {
+    /// List of region IDs to get file references from active FileHandles (in-memory).
+    pub query_regions: Vec<RegionId>,
+    /// Mapping from the src region IDs (whose file references to look for) to
+    /// the dst region IDs (where to read the manifests).
+    /// Key: The source region IDs (where files originally came from).
+    /// Value: The set of destination region IDs (whose manifests need to be read).
+    pub related_regions: HashMap<RegionId, HashSet<RegionId>>,
+}
+
+impl Display for GetPackedFileRefs {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        write!(f, "GetPackedFileRefs(region_ids={:?})", self.query_regions)
+    }
+}
+
 /// Instruction to trigger garbage collection for a region.
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
 pub struct GcRegions {
@@ -590,6 +611,147 @@ pub struct GcRegions {
     pub full_file_listing: bool,
 }
 
+/// Instruction to trigger garbage collection with a packed file-reference manifest.
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct PackedGcRegions {
+    /// The region ID to perform GC on.
+    pub regions: Vec<RegionId>,
+    /// Packed manifest used by the packed-only GC protocol.
+    pub packed_file_refs_manifest: PackedFileRefsManifest,
+    /// Whether to perform a full file listing to find orphan files.
+    pub full_file_listing: bool,
+}
+
+impl Display for PackedGcRegions {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        let file_refs_count = self.packed_file_refs_manifest.file_refs.len();
+        write!(
+            f,
+            "PackedGcRegions(regions={:?}, file_refs_count={}, full_file_listing={})",
+            self.regions, file_refs_count, self.full_file_listing
+        )
+    }
+}
+
+/// A packed, JSON-compatible encoding of a file reference manifest.
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
+pub struct PackedFileRefsManifest {
+    /// Required packed file references for each region.
+    pub file_refs: HashMap<RegionId, PackedRegionFileRefs>,
+    #[serde(default)]
+    pub manifest_version: HashMap<RegionId, u64>,
+    #[serde(default)]
+    pub cross_region_refs: HashMap<RegionId, HashSet<RegionId>>,
+}
+
+/// Packed file references for one region, encoded with STANDARD Base64.
+///
+/// `files` is required and contains 16-byte UUID records for references whose
+/// `index_version` is `None`. `indexed` is required and contains 24-byte records
+/// for references whose `index_version` is `Some`: a 16-byte UUID followed by
+/// an 8-byte big-endian `u64` version. Decoding fails when either field is not
+/// valid STANDARD Base64 or its decoded bytes do not align to the corresponding
+/// record width.
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
+pub struct PackedRegionFileRefs {
+    /// Required STANDARD Base64-encoded 16-byte UUID records.
+    pub files: String,
+    /// Required STANDARD Base64-encoded 16-byte UUID and 8-byte version records.
+    pub indexed: String,
+}
+
+impl PackedFileRefsManifest {
+    /// Packs a file reference manifest using required [`PackedRegionFileRefs`]
+    /// STANDARD Base64 fields.
+    ///
+    /// References whose `index_version` is `None` are encoded as 16-byte UUID
+    /// `files` records; references whose `index_version` is `Some` are encoded
+    /// as 24-byte `indexed` records: a 16-byte UUID followed by an 8-byte
+    /// big-endian `u64` version.
+    pub fn from_manifest(manifest: &FileRefsManifest) -> Self {
+        Self {
+            file_refs: manifest
+                .file_refs
+                .iter()
+                .map(|(region, refs)| (*region, PackedRegionFileRefs::from_refs(refs)))
+                .collect(),
+            manifest_version: manifest.manifest_version.clone(),
+            cross_region_refs: manifest.cross_region_refs.clone(),
+        }
+    }
+
+    /// Decodes this packed representation into a file reference manifest.
+    ///
+    /// Returns a typed error for invalid STANDARD Base64 or malformed record
+    /// framing without retaining encoded payloads.
+    pub fn into_manifest(self) -> crate::error::Result<FileRefsManifest> {
+        let mut file_refs = HashMap::new();
+        for (region, encoded) in self.file_refs {
+            file_refs.insert(region, encoded.into_refs(region)?);
+        }
+        Ok(FileRefsManifest {
+            file_refs,
+            manifest_version: self.manifest_version,
+            cross_region_refs: self.cross_region_refs,
+        })
+    }
+}
+
+impl PackedRegionFileRefs {
+    /// Packs file references as the documented STANDARD Base64 record streams.
+    pub fn from_refs(refs: &HashSet<FileRef>) -> Self {
+        let engine = base64::engine::general_purpose::STANDARD;
+        let mut files = Vec::new();
+        let mut indexed = Vec::new();
+        for file_ref in refs {
+            match file_ref.index_version {
+                None => files.extend_from_slice(file_ref.file_id.as_bytes()),
+                Some(version) => {
+                    indexed.extend_from_slice(file_ref.file_id.as_bytes());
+                    indexed.extend_from_slice(&version.to_be_bytes());
+                }
+            }
+        }
+        Self {
+            files: engine.encode(files),
+            indexed: engine.encode(indexed),
+        }
+    }
+
+    /// Decodes the documented STANDARD Base64 record streams for `region`.
+    fn into_refs(self, region: RegionId) -> crate::error::Result<HashSet<FileRef>> {
+        let engine = base64::engine::general_purpose::STANDARD;
+        let files = engine
+            .decode(self.files)
+            .context(DecodePackedFileRefsSnafu)?;
+        let indexed = engine
+            .decode(self.indexed)
+            .context(DecodePackedFileRefsSnafu)?;
+        if files.len() % 16 != 0 || indexed.len() % 24 != 0 {
+            return InvalidPackedFileRefsSnafu.fail();
+        }
+
+        let mut refs = HashSet::new();
+        for bytes in files.chunks_exact(16) {
+            let mut id = [0; 16];
+            id.copy_from_slice(bytes);
+            refs.insert(FileRef::new(region, FileId::from_bytes(id), None));
+        }
+        for bytes in indexed.chunks_exact(24) {
+            let mut id = [0; 16];
+            id.copy_from_slice(&bytes[..16]);
+            let mut version = [0; 8];
+            version.copy_from_slice(&bytes[16..]);
+            refs.insert(FileRef::new(
+                region,
+                FileId::from_bytes(id),
+                Some(u64::from_be_bytes(version)),
+            ));
+        }
+        Ok(refs)
+    }
+}
+
 impl Display for GcRegions {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         write!(
@@ -625,6 +787,26 @@ impl Display for GetFileRefsReply {
     }
 }
 
+/// Reply for GetPackedFileRefs instruction.
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct GetPackedFileRefsReply {
+    pub packed_file_refs_manifest: PackedFileRefsManifest,
+    pub success: bool,
+    pub error: Option<InstructionError>,
+}
+
+impl Display for GetPackedFileRefsReply {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "GetPackedFileRefsReply(success={}, file_refs_count={}, error={:?})",
+            self.success,
+            self.packed_file_refs_manifest.file_refs.len(),
+            self.error
+        )
+    }
+}
+
 /// Reply for GC instruction.
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
 pub struct GcRegionsReply {
@@ -827,8 +1009,12 @@ pub enum Instruction {
     FlushRegions(FlushRegions),
     /// Gets file references for regions.
     GetFileRefs(GetFileRefs),
+    /// Gets file references using the packed manifest format.
+    GetPackedFileRefs(GetPackedFileRefs),
     /// Triggers garbage collection for a region.
     GcRegions(GcRegions),
+    /// Triggers garbage collection with a packed file-reference manifest.
+    PackedGcRegions(PackedGcRegions),
     /// Temporary suspend serving reads or writes
     Suspend,
     /// Makes regions enter staging state.
@@ -1111,6 +1297,7 @@ pub enum InstructionReply {
     DowngradeRegions(DowngradeRegionsReply),
     FlushRegions(FlushRegionReply),
     GetFileRefs(GetFileRefsReply),
+    GetPackedFileRefs(GetPackedFileRefsReply),
     GcRegions(GcRegionsReply),
     EnterStagingRegions(EnterStagingRegionsReply),
     SyncRegions(SyncRegionsReply),
@@ -1131,6 +1318,9 @@ impl Display for InstructionReply {
             }
             Self::FlushRegions(reply) => write!(f, "InstructionReply::FlushRegions({})", reply),
             Self::GetFileRefs(reply) => write!(f, "InstructionReply::GetFileRefs({})", reply),
+            Self::GetPackedFileRefs(reply) => {
+                write!(f, "InstructionReply::GetPackedFileRefs({})", reply)
+            }
             Self::GcRegions(reply) => write!(f, "InstructionReply::GcRegion({})", reply),
             Self::EnterStagingRegions(reply) => {
                 write!(
@@ -1221,6 +1411,7 @@ impl InstructionReply {
 #[cfg(test)]
 mod tests {
     use std::collections::HashSet;
+    use std::error::Error as _;
 
     use common_error::mock::MockError;
     use common_wal::options::WalOptions;
@@ -1771,6 +1962,161 @@ mod tests {
         }
     }
 
+    #[test]
+    fn test_legacy_gc_file_refs_compatibility() {
+        #[derive(Debug, Deserialize)]
+        struct LegacyGetFileRefs {
+            query_regions: Vec<RegionId>,
+            related_regions: HashMap<RegionId, HashSet<RegionId>>,
+        }
+
+        #[derive(Debug, Deserialize)]
+        struct LegacyGetFileRefsReply {
+            file_refs_manifest: FileRefsManifest,
+            success: bool,
+            error: Option<InstructionError>,
+        }
+
+        #[derive(Debug, Deserialize)]
+        enum LegacyInstruction {
+            GetFileRefs(LegacyGetFileRefs),
+        }
+
+        let get_file_refs = Instruction::GetFileRefs(GetFileRefs {
+            query_regions: vec![RegionId::new(7, 3)],
+            related_regions: HashMap::new(),
+        });
+        let get_file_refs_json = serde_json::to_string(&get_file_refs).unwrap();
+        let legacy_get_file_refs: LegacyInstruction =
+            serde_json::from_str(&get_file_refs_json).unwrap();
+        let LegacyInstruction::GetFileRefs(legacy_get_file_refs) = legacy_get_file_refs;
+        assert_eq!(legacy_get_file_refs.query_regions.len(), 1);
+        assert!(legacy_get_file_refs.related_regions.is_empty());
+
+        let reply = GetFileRefsReply {
+            file_refs_manifest: FileRefsManifest::default(),
+            success: true,
+            error: None,
+        };
+        let reply_json = serde_json::to_string(&reply).unwrap();
+        let legacy_reply: LegacyGetFileRefsReply = serde_json::from_str(&reply_json).unwrap();
+        let current_reply: GetFileRefsReply = serde_json::from_str(&reply_json).unwrap();
+        assert!(legacy_reply.success);
+        assert!(legacy_reply.error.is_none());
+        assert!(legacy_reply.file_refs_manifest.file_refs.is_empty());
+        assert_eq!(current_reply, reply);
+
+        let packed = Instruction::PackedGcRegions(PackedGcRegions {
+            regions: vec![],
+            packed_file_refs_manifest: PackedFileRefsManifest::default(),
+            full_file_listing: false,
+        });
+        let packed_json = serde_json::to_string(&packed).unwrap();
+        assert!(serde_json::from_str::<LegacyInstruction>(&packed_json).is_err());
+    }
+
+    #[test]
+    fn test_packed_gc_regions_round_trip_is_distinct() {
+        let instruction = Instruction::PackedGcRegions(PackedGcRegions {
+            regions: vec![RegionId::new(7, 3)],
+            packed_file_refs_manifest: PackedFileRefsManifest::default(),
+            full_file_listing: true,
+        });
+        let serialized = serde_json::to_string(&instruction).unwrap();
+        assert!(serialized.contains("PackedGcRegions"));
+        assert_eq!(
+            serde_json::from_str::<Instruction>(&serialized).unwrap(),
+            instruction
+        );
+    }
+
+    #[test]
+    fn test_packed_file_refs_manifest_round_trip_and_validation() {
+        let region = RegionId::new(7, 3);
+        let file_id = FileId::parse_str("00000000-0000-0000-0000-000000000001").unwrap();
+        let mut manifest = FileRefsManifest::default();
+        manifest.file_refs.insert(
+            region,
+            HashSet::from([
+                FileRef::new(region, file_id, None),
+                FileRef::new(region, file_id, Some(0)),
+                FileRef::new(region, file_id, Some(u64::MAX)),
+            ]),
+        );
+        manifest.manifest_version.insert(region, 42);
+        manifest
+            .cross_region_refs
+            .insert(region, HashSet::from([RegionId::new(7, 4)]));
+        let packed = PackedFileRefsManifest::from_manifest(&manifest);
+        assert_eq!(packed.clone().into_manifest().unwrap(), manifest);
+        let mut missing_files = serde_json::to_value(&packed).unwrap();
+        missing_files
+            .get_mut("file_refs")
+            .and_then(serde_json::Value::as_object_mut)
+            .and_then(|file_refs| file_refs.values_mut().next())
+            .and_then(serde_json::Value::as_object_mut)
+            .unwrap()
+            .remove("files");
+        assert!(serde_json::from_value::<PackedFileRefsManifest>(missing_files).is_err());
+
+        let mut missing_indexed = serde_json::to_value(&packed).unwrap();
+        missing_indexed
+            .get_mut("file_refs")
+            .and_then(serde_json::Value::as_object_mut)
+            .and_then(|file_refs| file_refs.values_mut().next())
+            .and_then(serde_json::Value::as_object_mut)
+            .unwrap()
+            .remove("indexed");
+        assert!(serde_json::from_value::<PackedFileRefsManifest>(missing_indexed).is_err());
+
+        let mut malformed = packed.clone();
+        malformed.file_refs.get_mut(&region).unwrap().indexed = "!".to_string();
+        let err = malformed.into_manifest().unwrap_err();
+        assert!(matches!(
+            &err,
+            crate::error::Error::DecodePackedFileRefs { .. }
+        ));
+        assert!(
+            err.source()
+                .is_some_and(|source| source.is::<base64::DecodeError>())
+        );
+        assert_eq!(err.status_code(), StatusCode::Unexpected);
+        assert_eq!(err.retry_hint(), RetryHint::NonRetryable);
+
+        let mut malformed = packed;
+        malformed.file_refs.get_mut(&region).unwrap().indexed =
+            base64::engine::general_purpose::STANDARD.encode([0; 1]);
+        let err = malformed.into_manifest().unwrap_err();
+        assert!(matches!(
+            &err,
+            crate::error::Error::InvalidPackedFileRefs { .. }
+        ));
+        assert!(err.source().is_none());
+        assert_eq!(err.status_code(), StatusCode::Unexpected);
+        assert_eq!(err.retry_hint(), RetryHint::NonRetryable);
+    }
+
+    #[test]
+    fn test_packed_file_refs_is_smaller_than_legacy() {
+        let region = RegionId::new(7, 3);
+        let mut manifest = FileRefsManifest::default();
+        manifest.file_refs.insert(
+            region,
+            (0..500)
+                .map(|i| {
+                    let id =
+                        FileId::parse_str(&format!("00000000-0000-0000-0000-{i:012}")).unwrap();
+                    FileRef::new(region, id, None)
+                })
+                .collect(),
+        );
+        let legacy = serde_json::to_vec(&manifest).unwrap().len();
+        let packed = serde_json::to_vec(&PackedFileRefsManifest::from_manifest(&manifest))
+            .unwrap()
+            .len();
+        assert!(packed * 2 < legacy, "packed={packed}, legacy={legacy}");
+    }
+
     #[test]
     fn test_serialize_get_file_refs_instruction_reply() {
         let mut manifest = FileRefsManifest::default();

src/datanode/src/heartbeat.rs

@@ -18,6 +18,7 @@ use std::sync::atomic::{AtomicBool, Ordering};
 use std::time::Duration;
 
 use api::v1::meta::heartbeat_request::NodeWorkloads;
+use api::v1::meta::mailbox_message::Payload;
 use api::v1::meta::{DatanodeWorkloads, HeartbeatRequest, NodeInfo, Peer, RegionRole, RegionStat};
 use common_base::Plugins;
 use common_meta::cache_invalidator::CacheInvalidatorRef;
@@ -146,7 +147,15 @@ impl HeartbeatTask {
                 None
             }) {
                 if let Some(msg) = res.mailbox_message.as_ref() {
-                    info!("Received mailbox message: {msg:?}, meta_client id: {client_id:?}");
+                    let payload_len = msg
+                        .payload
+                        .as_ref()
+                        .map(|Payload::Json(payload)| payload.len())
+                        .unwrap_or_default();
+                    info!(
+                        message_id = msg.id,
+                        payload_len, client_id, "Received mailbox message"
+                    );
                 }
                 if let Some(lease) = res.region_lease.as_ref() {
                     metrics::LAST_RECEIVED_HEARTBEAT_ELAPSED
@@ -193,7 +202,8 @@ impl HeartbeatTask {
         ctx: HeartbeatResponseHandlerContext,
         handler_executor: HeartbeatResponseHandlerExecutorRef,
     ) -> Result<()> {
-        trace!("Heartbeat response: {:?}", ctx.response);
+        let mailbox_message_id = ctx.response.mailbox_message.as_ref().map(|msg| msg.id);
+        trace!(?mailbox_message_id, "Handling heartbeat response");
         handler_executor
             .handle(ctx)
             .await
@@ -379,7 +389,8 @@ impl HeartbeatTask {
                         .set(last_sent.elapsed().as_millis() as i64);
                     // Resets the timer.
                     last_sent = Instant::now();
-                    debug!("Sending heartbeat request: {:?}", req);
+                    let mailbox_message_id = req.mailbox_message.as_ref().map(|msg| msg.id);
+                    debug!(?mailbox_message_id, "Sending heartbeat request");
                     if let Err(e) = tx.send(req).await {
                         error!(e; "Failed to send heartbeat to metasrv");
                         match Self::create_streams(

src/datanode/src/heartbeat/handler.rs

@@ -41,9 +41,9 @@ use crate::heartbeat::handler::apply_staging_manifest::ApplyStagingManifestsHand
 use crate::heartbeat::handler::close_region::CloseRegionsHandler;
 use crate::heartbeat::handler::downgrade_region::DowngradeRegionsHandler;
 use crate::heartbeat::handler::enter_staging::EnterStagingRegionsHandler;
-use crate::heartbeat::handler::file_ref::GetFileRefsHandler;
+use crate::heartbeat::handler::file_ref::{GetFileRefsHandler, GetPackedFileRefsHandler};
 use crate::heartbeat::handler::flush_region::FlushRegionsHandler;
-use crate::heartbeat::handler::gc_worker::GcRegionsHandler;
+use crate::heartbeat::handler::gc_worker::{GcRegionsHandler, PackedGcRegionsHandler};
 use crate::heartbeat::handler::open_region::OpenRegionsHandler;
 use crate::heartbeat::handler::remap_manifest::RemapManifestHandler;
 use crate::heartbeat::handler::sync_region::SyncRegionHandler;
@@ -135,7 +135,11 @@ impl RegionHeartbeatResponseHandler {
                 .into(),
             ))),
             Instruction::GetFileRefs(_) => Ok(Some(Box::new(GetFileRefsHandler.into()))),
+            Instruction::GetPackedFileRefs(_) => {
+                Ok(Some(Box::new(GetPackedFileRefsHandler.into())))
+            }
             Instruction::GcRegions(_) => Ok(Some(Box::new(GcRegionsHandler.into()))),
+            Instruction::PackedGcRegions(_) => Ok(Some(Box::new(PackedGcRegionsHandler.into()))),
             Instruction::InvalidateCaches(_) => InvalidHeartbeatResponseSnafu.fail(),
             Instruction::Suspend => Ok(None),
             Instruction::EnterStagingRegions(_) => {
@@ -159,7 +163,9 @@ pub enum InstructionHandlers {
     DowngradeRegions(DowngradeRegionsHandler),
     UpgradeRegions(UpgradeRegionsHandler),
     GetFileRefs(GetFileRefsHandler),
+    GetPackedFileRefs(GetPackedFileRefsHandler),
     GcRegions(GcRegionsHandler),
+    PackedGcRegions(PackedGcRegionsHandler),
     EnterStagingRegions(EnterStagingRegionsHandler),
     SyncRegions(SyncRegionHandler),
     RemapManifest(RemapManifestHandler),
@@ -192,7 +198,9 @@ impl_from_handler!(
     DowngradeRegionsHandler => DowngradeRegions,
     UpgradeRegionsHandler => UpgradeRegions,
     GetFileRefsHandler => GetFileRefs,
+    GetPackedFileRefsHandler => GetPackedFileRefs,
     GcRegionsHandler => GcRegions,
+    PackedGcRegionsHandler => PackedGcRegions,
     EnterStagingRegionsHandler => EnterStagingRegions,
     SyncRegionHandler => SyncRegions,
     RemapManifestHandler => RemapManifest,
@@ -240,7 +248,9 @@ dispatch_instr!(
     DowngradeRegions => DowngradeRegions,
     UpgradeRegions => UpgradeRegions,
     GetFileRefs => GetFileRefs,
+    GetPackedFileRefs => GetPackedFileRefs,
     GcRegions => GcRegions,
+    PackedGcRegions => PackedGcRegions,
     EnterStagingRegions => EnterStagingRegions,
     SyncRegions => SyncRegions,
     RemapManifest => RemapManifest,

src/datanode/src/heartbeat/handler/file_ref.rs

@@ -13,7 +13,10 @@
 // limitations under the License.
 
 use common_error::ext::ErrorExt;
-use common_meta::instruction::{GetFileRefs, GetFileRefsReply, InstructionError, InstructionReply};
+use common_meta::instruction::{
+    GetFileRefs, GetFileRefsReply, GetPackedFileRefs, GetPackedFileRefsReply, InstructionError,
+    InstructionReply, PackedFileRefsManifest,
+};
 use store_api::storage::FileRefsManifest;
 
 use crate::heartbeat::handler::{HandlerContext, InstructionHandler};
@@ -65,3 +68,54 @@ impl InstructionHandler for GetFileRefsHandler {
         }
     }
 }
+
+pub struct GetPackedFileRefsHandler;
+
+#[async_trait::async_trait]
+impl InstructionHandler for GetPackedFileRefsHandler {
+    type Instruction = GetPackedFileRefs;
+
+    async fn handle(
+        &self,
+        ctx: &HandlerContext,
+        get_file_refs: Self::Instruction,
+    ) -> Option<InstructionReply> {
+        let region_server = &ctx.region_server;
+        let Some(mito_engine) = region_server.mito_engine() else {
+            return Some(InstructionReply::GetPackedFileRefs(
+                GetPackedFileRefsReply {
+                    packed_file_refs_manifest: PackedFileRefsManifest::default(),
+                    success: false,
+                    error: Some(InstructionError::legacy_internal_retryable(
+                        "MitoEngine not found",
+                    )),
+                },
+            ));
+        };
+        match mito_engine
+            .get_snapshot_of_file_refs(get_file_refs.query_regions, get_file_refs.related_regions)
+            .await
+        {
+            Ok(all_file_refs) => Some(InstructionReply::GetPackedFileRefs(
+                GetPackedFileRefsReply {
+                    packed_file_refs_manifest: PackedFileRefsManifest::from_manifest(
+                        &all_file_refs,
+                    ),
+                    success: true,
+                    error: None,
+                },
+            )),
+            Err(e) => Some(InstructionReply::GetPackedFileRefs(
+                GetPackedFileRefsReply {
+                    packed_file_refs_manifest: PackedFileRefsManifest::default(),
+                    success: false,
+                    error: Some(InstructionError::new(
+                        e.status_code(),
+                        format!("Failed to get file refs: {}", e.output_msg()),
+                        e.retry_hint(),
+                    )),
+                },
+            )),
+        }
+    }
+}

src/datanode/src/heartbeat/handler/gc_worker.rs

@@ -15,7 +15,7 @@
 use std::collections::{BTreeMap, HashMap};
 use std::sync::Arc;
 
-use common_meta::instruction::{GcRegions, GcRegionsReply, InstructionReply};
+use common_meta::instruction::{GcRegions, GcRegionsReply, InstructionReply, PackedGcRegions};
 use common_meta::key::table_info::TableInfoManager;
 use common_meta::key::table_route::TableRouteManager;
 use common_telemetry::{debug, warn};
@@ -34,6 +34,7 @@ use crate::error::{GcMitoEngineSnafu, GetMetadataSnafu, Result, UnexpectedSnafu}
 use crate::heartbeat::handler::{HandlerContext, InstructionHandler};
 
 pub struct GcRegionsHandler;
+pub struct PackedGcRegionsHandler;
 
 #[async_trait::async_trait]
 impl InstructionHandler for GcRegionsHandler {
@@ -67,7 +68,7 @@ impl InstructionHandler for GcRegionsHandler {
                 .push(rid);
         }
 
-        let file_refs_manifest = gc_regions.file_refs_manifest.clone();
+        let file_refs_manifest = gc_regions.file_refs_manifest;
         let full_file_listing = gc_regions.full_file_listing;
 
         let ctx_clone = ctx.clone();
@@ -138,6 +139,36 @@ impl InstructionHandler for GcRegionsHandler {
     }
 }
 
+#[async_trait::async_trait]
+impl InstructionHandler for PackedGcRegionsHandler {
+    type Instruction = PackedGcRegions;
+
+    async fn handle(
+        &self,
+        ctx: &HandlerContext,
+        instruction: Self::Instruction,
+    ) -> Option<InstructionReply> {
+        let manifest = match instruction.packed_file_refs_manifest.into_manifest() {
+            Ok(manifest) => manifest,
+            Err(err) => {
+                return Some(InstructionReply::GcRegions(GcRegionsReply {
+                    result: Err(common_meta::instruction::InstructionError::from_error(&err)),
+                }));
+            }
+        };
+        GcRegionsHandler
+            .handle(
+                ctx,
+                GcRegions {
+                    regions: instruction.regions,
+                    file_refs_manifest: manifest,
+                    full_file_listing: instruction.full_file_listing,
+                },
+            )
+            .await
+    }
+}
+
 impl GcRegionsHandler {
     /// Create a GC worker for the given table and region IDs.
     async fn create_gc_worker(

src/meta-srv/src/error.rs

@@ -368,9 +368,9 @@ pub enum Error {
         location: Location,
     },
 
-    #[snafu(display("Failed to deserialize from json: {}", input))]
+    #[snafu(display("Failed to deserialize from json payload of length {}", input_len))]
     DeserializeFromJson {
-        input: String,
+        input_len: usize,
         #[snafu(source)]
         error: serde_json::error::Error,
         #[snafu(implicit)]

src/meta-srv/src/event/gc.rs

@@ -218,6 +218,7 @@ mod tests {
     use common_event_recorder::testing::assert_event_contract;
     use common_event_recorder::{EventTypeFilter, PersistentEventContext, TriggerReason};
     use common_meta::key::TableMetadataManager;
+    use common_meta::key::runtime_switch::RuntimeSwitchManager;
     use common_meta::kv_backend::memory::MemoryKvBackend;
     use common_meta::sequence::SequenceBuilder;
     use common_procedure::{
@@ -588,11 +589,13 @@ mod tests {
     fn batch_gc_procedure() -> BatchGcProcedure {
         let kv_backend = Arc::new(MemoryKvBackend::new());
         let table_metadata_manager = Arc::new(TableMetadataManager::new(kv_backend.clone()));
+        let runtime_switch_manager = Arc::new(RuntimeSwitchManager::new(kv_backend.clone()));
         let mailbox_sequence = SequenceBuilder::new("test_batch_gc_event", kv_backend).build();
         let mailbox = MailboxContext::new(mailbox_sequence);
         BatchGcProcedure::new(
             mailbox.mailbox().clone(),
             table_metadata_manager,
+            runtime_switch_manager,
             "localhost".to_string(),
             vec![RegionId::new(1024, 1)],
             true,

src/meta-srv/src/gc/ctx.rs

@@ -27,6 +27,7 @@ use common_meta::ddl_manager::DdlManagerRef;
 #[cfg(feature = "enterprise")]
 use common_meta::key::DroppedTableName;
 use common_meta::key::TableMetadataManagerRef;
+use common_meta::key::runtime_switch::RuntimeSwitchManagerRef;
 use common_meta::key::table_repart::TableRepartValue;
 use common_meta::key::table_route::PhysicalTableRouteValue;
 #[cfg(feature = "enterprise")]
@@ -155,6 +156,8 @@ pub(crate) struct DefaultGcSchedulerCtx {
     pub(crate) table_metadata_manager: TableMetadataManagerRef,
     /// Procedure manager.
     pub(crate) procedure_manager: ProcedureManagerRef,
+    /// Runtime switch manager used by recovered and newly submitted GC procedures.
+    pub(crate) runtime_switch_manager: RuntimeSwitchManagerRef,
     /// DDL manager used to submit the existing purge procedure.
     #[cfg(feature = "enterprise")]
     pub(crate) ddl_manager: DdlManagerRef,
@@ -176,6 +179,7 @@ impl DefaultGcSchedulerCtx {
     pub fn try_new(
         table_metadata_manager: TableMetadataManagerRef,
         procedure_manager: ProcedureManagerRef,
+        runtime_switch_manager: RuntimeSwitchManagerRef,
         #[cfg(feature = "enterprise")] ddl_manager: DdlManagerRef,
         meta_peer_client: MetaPeerClientRef,
         mailbox: MailboxRef,
@@ -184,6 +188,7 @@ impl DefaultGcSchedulerCtx {
         Ok(Self {
             table_metadata_manager,
             procedure_manager,
+            runtime_switch_manager,
             #[cfg(feature = "enterprise")]
             ddl_manager,
             #[cfg(feature = "enterprise")]
@@ -320,6 +325,7 @@ impl DefaultGcSchedulerCtx {
         let procedure = BatchGcProcedure::new(
             self.mailbox.clone(),
             self.table_metadata_manager.clone(),
+            self.runtime_switch_manager.clone(),
             self.server_addr.clone(),
             region_ids.to_vec(),
             full_file_listing,

src/meta-srv/src/gc/procedure.rs

@@ -18,8 +18,14 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use api::v1::meta::MailboxMessage;
-use common_meta::instruction::{self, GcRegions, GetFileRefs, GetFileRefsReply, InstructionReply};
+use common_error::ext::PlainError;
+use common_error::status_code::StatusCode;
+use common_meta::instruction::{
+    self, GetPackedFileRefs, GetPackedFileRefsReply, InstructionReply, PackedFileRefsManifest,
+    PackedGcRegions, PackedRegionFileRefs,
+};
 use common_meta::key::TableMetadataManagerRef;
+use common_meta::key::runtime_switch::RuntimeSwitchManagerRef;
 use common_meta::key::table_repart::TableRepartValue;
 use common_meta::key::table_route::PhysicalTableRouteValue;
 use common_meta::lock_key::{RegionLock, TableLock};
@@ -34,6 +40,7 @@ use common_telemetry::tracing::Instrument as _;
 use common_telemetry::tracing_context::TracingContext;
 use common_telemetry::{debug, error, info, warn};
 use futures::future::join_all;
+use futures::stream::{FuturesUnordered, StreamExt};
 use itertools::Itertools as _;
 use serde::{Deserialize, Serialize};
 use snafu::ResultExt as _;
@@ -49,14 +56,14 @@ use crate::metrics::{METRIC_META_GC_DATANODE_CALLS_TOTAL, METRIC_META_GC_FAILED_
 use crate::procedure::utils::{instruction_error_result, instruction_to_error};
 use crate::service::mailbox::{Channel, MailboxReceiver, MailboxRef};
 
-async fn send_get_file_refs_inner(
+async fn send_get_packed_file_refs_inner(
     mailbox: &MailboxRef,
     server_addr: &str,
     peer: &Peer,
-    instruction: GetFileRefs,
+    instruction: GetPackedFileRefs,
     timeout: Duration,
 ) -> Result<MailboxReceiver> {
-    let instruction = instruction::Instruction::GetFileRefs(instruction);
+    let instruction = instruction::Instruction::GetPackedFileRefs(instruction);
     let tracing_ctx = TracingContext::from_current_span();
     let msg = MailboxMessage::json_message(
         &format!("Get file references: {}", instruction),
@@ -75,25 +82,25 @@ async fn send_get_file_refs_inner(
         .await
 }
 
-async fn recv_get_file_refs_reply(
+async fn recv_get_packed_file_refs_reply(
     peer: &Peer,
     mailbox_rx: MailboxReceiver,
-) -> Result<GetFileRefsReply> {
+) -> Result<GetPackedFileRefsReply> {
     let reply = match mailbox_rx.await {
         Ok(reply_msg) => HeartbeatMailbox::json_reply(&reply_msg)?,
         Err(e) => {
             error!(
-                e; "Failed to receive reply from datanode {} for GetFileRefs instruction",
+                e; "Failed to receive reply from datanode {} for GetPackedFileRefs instruction",
                 peer,
             );
             return Err(e);
         }
     };
 
-    let InstructionReply::GetFileRefs(reply) = reply else {
+    let InstructionReply::GetPackedFileRefs(reply) = reply else {
         return error::UnexpectedInstructionReplySnafu {
-            mailbox_message: format!("{:?}", reply),
-            reason: "Unexpected reply of the GetFileRefs instruction",
+            mailbox_message: "unexpected instruction reply for GetPackedFileRefs".to_string(),
+            reason: "Unexpected reply of the GetPackedFileRefs instruction",
         }
         .fail();
     };
@@ -104,12 +111,11 @@ async fn recv_get_file_refs_reply(
 async fn send_gc_regions_inner(
     mailbox: &MailboxRef,
     peer: &Peer,
-    gc_regions: &GcRegions,
+    instruction: instruction::Instruction,
     server_addr: &str,
     timeout: Duration,
     description: &str,
 ) -> Result<MailboxReceiver> {
-    let instruction = instruction::Instruction::GcRegions(gc_regions.clone());
     let tracing_ctx = TracingContext::from_current_span();
     let msg = MailboxMessage::json_message(
         &format!("{}: {}", description, instruction),
@@ -128,9 +134,43 @@ async fn send_gc_regions_inner(
         .await
 }
 
+fn scoped_gc_instruction(
+    regions: Vec<RegionId>,
+    full_manifest: &FileRefsManifest,
+    full_file_listing: bool,
+) -> instruction::Instruction {
+    let packed_file_refs_manifest = PackedFileRefsManifest {
+        file_refs: regions
+            .iter()
+            .filter_map(|region| {
+                full_manifest
+                    .file_refs
+                    .get(region)
+                    .map(|refs| (*region, PackedRegionFileRefs::from_refs(refs)))
+            })
+            .collect(),
+        manifest_version: regions
+            .iter()
+            .filter_map(|region| {
+                full_manifest
+                    .manifest_version
+                    .get(region)
+                    .map(|version| (*region, *version))
+            })
+            .collect(),
+        cross_region_refs: HashMap::new(),
+    };
+
+    instruction::Instruction::PackedGcRegions(PackedGcRegions {
+        regions,
+        packed_file_refs_manifest,
+        full_file_listing,
+    })
+}
+
 async fn recv_gc_regions_reply(
     peer: &Peer,
-    gc_regions: &GcRegions,
+    regions: &[RegionId],
     description: &str,
     mailbox_rx: MailboxReceiver,
 ) -> Result<GcReport> {
@@ -147,7 +187,7 @@ async fn recv_gc_regions_reply(
 
     let InstructionReply::GcRegions(reply) = reply else {
         return error::UnexpectedInstructionReplySnafu {
-            mailbox_message: format!("{:?}", reply),
+            mailbox_message: "unexpected instruction reply for GcRegions".to_string(),
             reason: "Unexpected reply of the GcRegions instruction",
         }
         .fail();
@@ -158,14 +198,16 @@ async fn recv_gc_regions_reply(
         Ok(report) => Ok(report),
         Err(e) => {
             error!(
-                e; "Datanode {} reported error during GC for regions {:?}",
-                peer, gc_regions
+                e; "Datanode {} reported error during GC for {} regions",
+                peer, regions.len()
             );
             instruction_error_result(
                 &e,
                 format!(
-                    "Datanode {} reported error during GC for regions {:?}: {}",
-                    peer, gc_regions, e
+                    "Datanode {} reported error during GC for {} regions: {}",
+                    peer,
+                    regions.len(),
+                    e
                 ),
             )
         }
@@ -177,6 +219,7 @@ async fn recv_gc_regions_reply(
 pub struct BatchGcProcedure {
     mailbox: MailboxRef,
     table_metadata_manager: TableMetadataManagerRef,
+    runtime_switch_manager: RuntimeSwitchManagerRef,
     data: BatchGcData,
 }
 
@@ -217,9 +260,11 @@ pub enum State {
 impl BatchGcProcedure {
     pub const TYPE_NAME: &'static str = "metasrv-procedure::BatchGcProcedure";
 
+    #[allow(clippy::too_many_arguments)]
     pub fn new(
         mailbox: MailboxRef,
         table_metadata_manager: TableMetadataManagerRef,
+        runtime_switch_manager: RuntimeSwitchManagerRef,
         server_addr: String,
         regions: Vec<RegionId>,
         full_file_listing: bool,
@@ -229,6 +274,7 @@ impl BatchGcProcedure {
         Self {
             mailbox,
             table_metadata_manager,
+            runtime_switch_manager,
             data: BatchGcData {
                 state: State::Start,
                 server_addr,
@@ -251,6 +297,7 @@ impl BatchGcProcedure {
     pub fn new_update_repartition_for_test(
         mailbox: MailboxRef,
         table_metadata_manager: TableMetadataManagerRef,
+        runtime_switch_manager: RuntimeSwitchManagerRef,
         server_addr: String,
         regions: Vec<RegionId>,
         file_refs: FileRefsManifest,
@@ -259,6 +306,7 @@ impl BatchGcProcedure {
         Self {
             mailbox,
             table_metadata_manager,
+            runtime_switch_manager,
             data: BatchGcData {
                 state: State::UpdateRepartition,
                 server_addr,
@@ -286,6 +334,22 @@ impl BatchGcProcedure {
         })
     }
 
+    async fn check_maintenance_mode(&self) -> ProcedureResult<()> {
+        let enabled = self
+            .runtime_switch_manager
+            .maintenance_mode()
+            .await
+            .context(error::RuntimeSwitchManagerSnafu)
+            .map_err(ProcedureError::retry_later)?;
+        if enabled {
+            return Err(ProcedureError::external(PlainError::new(
+                "maintenance mode is enabled".to_string(),
+                StatusCode::IllegalState,
+            )));
+        }
+        Ok(())
+    }
+
     fn merge_gc_report(&mut self, report: GcReport) {
         let accumulated = self.data.gc_report.get_or_insert_default();
         // Deleted objects are cumulative, while these sets describe the latest outcome
@@ -644,10 +708,10 @@ impl BatchGcProcedure {
             }
         }
 
-        // Send GetFileRefs instructions to each datanode
+        // Send packed GetFileRefs instructions to each datanode
         let mut all_file_refs: HashMap<RegionId, HashSet<_>> = HashMap::new();
         let mut all_manifest_versions = HashMap::new();
-        let mut all_cross_region_refs = HashMap::new();
+        let mut all_cross_region_refs: HashMap<RegionId, HashSet<RegionId>> = HashMap::new();
 
         let mut peers = HashSet::new();
         peers.extend(datanode2query_regions.keys().cloned());
@@ -655,33 +719,34 @@ impl BatchGcProcedure {
 
         let mailbox = &self.mailbox;
         let server_addr = &self.data.server_addr;
-        let mut tasks = Vec::new();
-
+        // Each future owns both send and receive, so a completed reply is merged and
+        // dropped immediately rather than retained in a second join_all buffer.
+        let mut tasks = FuturesUnordered::new();
         for peer in peers {
             let regions = datanode2query_regions.remove(&peer).unwrap_or_default();
             let related_regions_for_peer =
                 datanode2related_regions.remove(&peer).unwrap_or_default();
-
             if regions.is_empty() && related_regions_for_peer.is_empty() {
                 continue;
             }
-
             tasks.push(async move {
-                let instruction = GetFileRefs {
-                    query_regions: regions.clone(),
-                    related_regions: related_regions_for_peer.clone(),
+                let instruction = GetPackedFileRefs {
+                    query_regions: regions,
+                    related_regions: related_regions_for_peer,
                 };
-
-                let reply =
-                    send_get_file_refs_inner(mailbox, server_addr, &peer, instruction, timeout)
-                        .await;
-
-                (peer, regions, related_regions_for_peer, reply)
+                let rx = send_get_packed_file_refs_inner(
+                    mailbox,
+                    server_addr,
+                    &peer,
+                    instruction,
+                    timeout,
+                )
+                .await?;
+                let reply = recv_get_packed_file_refs_reply(&peer, rx).await?;
+                Ok::<_, crate::error::Error>((peer, reply))
             });
         }
 
-        let mut recv_tasks = Vec::new();
-        // store error to make sure metrics doesn't ignore other peers
         let mut first_error = None;
         let mut record_get_file_refs_error = |e| {
             METRIC_META_GC_DATANODE_CALLS_TOTAL
@@ -691,80 +756,56 @@ impl BatchGcProcedure {
                 first_error = Some(e);
             }
         };
-        for (peer, regions, related_regions_for_peer, reply) in join_all(tasks).await {
-            match reply {
-                Ok(mailbox_rx) => {
-                    recv_tasks.push(async move {
-                        let reply = recv_get_file_refs_reply(&peer, mailbox_rx).await;
-                        (peer, regions, related_regions_for_peer, reply)
-                    });
-                }
-                Err(e) => record_get_file_refs_error(e),
-            }
-        }
-
-        let replies = join_all(recv_tasks).await;
-
-        for (peer, regions, related_regions_for_peer, reply) in replies {
-            let reply = match reply {
+        while let Some(result) = tasks.next().await {
+            let (peer, reply) = match result {
                 Ok(reply) => reply,
                 Err(e) => {
                     record_get_file_refs_error(e);
                     continue;
                 }
             };
-            debug!(
-                "Got file references from datanode: {:?}, query_regions: {:?}, related_regions: {:?}, reply: {:?}",
-                peer, regions, related_regions_for_peer, reply
-            );
-
             if !reply.success {
-                METRIC_META_GC_DATANODE_CALLS_TOTAL
-                    .with_label_values(&["get_file_refs", "error"])
-                    .inc();
                 let err = if let Some(error) = &reply.error {
                     instruction_to_error(
                         error,
-                        format!(
-                            "Failed to get file references from datanode {}: {:?}",
-                            peer, error
-                        ),
+                        format!("Failed to get file references from datanode {peer}"),
                     )
                 } else {
                     error::UnexpectedSnafu {
-                        violated: format!(
-                            "Failed to get file references from datanode {}: {:?}",
-                            peer, reply.error
-                        ),
+                        violated:
+                            "Datanode returned an unsuccessful GetPackedFileRefs reply without an error"
+                                .to_string(),
                     }
                     .build()
                 };
                 record_get_file_refs_error(err);
                 continue;
             }
+            let manifest = match reply.packed_file_refs_manifest.into_manifest() {
+                Ok(manifest) => manifest,
+                Err(err) => {
+                    record_get_file_refs_error(error::Error::Other {
+                        source: common_error::ext::BoxedError::new(err),
+                        location: snafu::Location::new(file!(), line!(), 0),
+                    });
+                    continue;
+                }
+            };
             METRIC_META_GC_DATANODE_CALLS_TOTAL
                 .with_label_values(&["get_file_refs", "success"])
                 .inc();
-
-            // Merge the file references from this datanode
-            for (region_id, file_refs) in reply.file_refs_manifest.file_refs {
-                all_file_refs
-                    .entry(region_id)
-                    .or_default()
-                    .extend(file_refs);
+            for (region_id, refs) in manifest.file_refs {
+                all_file_refs.entry(region_id).or_default().extend(refs);
             }
-
-            // region manifest version should be the smallest one among all peers, so outdated region can be detected
-            for (region_id, version) in reply.file_refs_manifest.manifest_version {
+            for (region_id, version) in manifest.manifest_version {
                 let entry = all_manifest_versions.entry(region_id).or_insert(version);
                 *entry = (*entry).min(version);
             }
-
-            for (region_id, related_region_ids) in reply.file_refs_manifest.cross_region_refs {
-                let entry = all_cross_region_refs
+            for (region_id, related) in manifest.cross_region_refs {
+                all_cross_region_refs
                     .entry(region_id)
-                    .or_insert_with(HashSet::new);
-                entry.extend(related_region_ids);
+                    .or_default()
+                    .extend(related);
             }
         }
 
@@ -813,28 +854,23 @@ impl BatchGcProcedure {
         let tasks = datanode2regions
             .into_iter()
             .map(|(peer, regions_for_peer)| {
-                let gc_regions = GcRegions {
-                    regions: regions_for_peer.clone(),
-                    // file_refs_manifest could be somewhere large. But still intentionally clone per datanode here:
-                    // this path is admin-triggered or scheduler-triggered, peer count is expected to be bounded, and
-                    // and abnormal manifest growth should be addressed at the source
-                    file_refs_manifest: file_refs.clone(),
-                    full_file_listing,
-                };
-                let region_count = gc_regions.regions.len() as u64;
+                let region_count = regions_for_peer.len() as u64;
+                let regions = regions_for_peer.clone();
+                let instruction =
+                    scoped_gc_instruction(regions_for_peer, file_refs, full_file_listing);
 
                 async move {
                     let report = send_gc_regions_inner(
                         mailbox,
                         &peer,
-                        &gc_regions,
+                        instruction,
                         server_addr,
                         timeout,
                         "Batch GC",
                     )
                     .await;
 
-                    (peer, gc_regions, region_count, report)
+                    (peer, regions, region_count, report)
                 }
             });
 
@@ -851,12 +887,12 @@ impl BatchGcProcedure {
                 first_error = Some(e);
             }
         };
-        for (peer, gc_regions, region_count, report) in join_all(tasks).await {
+        for (peer, regions, region_count, report) in join_all(tasks).await {
             match report {
                 Ok(mailbox_rx) => {
                     recv_tasks.push(async move {
                         let report =
-                            recv_gc_regions_reply(&peer, &gc_regions, "Batch GC", mailbox_rx).await;
+                            recv_gc_regions_reply(&peer, &regions, "Batch GC", mailbox_rx).await;
                         (peer, region_count, report)
                     });
                 }
@@ -887,12 +923,12 @@ impl BatchGcProcedure {
 
             if need_retry.is_empty() {
                 info!(
-                    "GC report from datanode {}: successfully deleted files for regions {:?}",
+                    "GC report from datanode {}: successfully deleted files for region IDs {:?}",
                     peer, success
                 );
             } else {
                 warn!(
-                    "GC report from datanode {}: successfully deleted files for regions {:?}, need retry for regions {:?}",
+                    "GC report from datanode {}: successfully deleted files for region IDs {:?}, need retry for region IDs {:?}",
                     peer, success, need_retry
                 );
             }
@@ -921,6 +957,8 @@ impl Procedure for BatchGcProcedure {
     }
 
     async fn execute(&mut self, ctx: &ProcedureContext) -> ProcedureResult<Status> {
+        self.check_maintenance_mode().await?;
+
         match self.data.state {
             State::Start => {
                 let _regions_span = common_telemetry::tracing::debug_span!(
@@ -1122,11 +1160,15 @@ mod tests {
 
     use api::v1::meta::MailboxMessage;
     use api::v1::meta::mailbox_message::Payload;
-    use common_meta::instruction::{GcRegionsReply, InstructionReply};
+    use common_meta::instruction::{GcRegionsReply, Instruction, InstructionReply};
     use common_meta::key::TableMetadataManager;
+    use common_meta::key::runtime_switch::RuntimeSwitchManager;
     use common_meta::kv_backend::memory::MemoryKvBackend;
+    use common_meta::kv_backend::test_util::MockKvBackendBuilder;
     use common_meta::peer::Peer;
     use common_meta::sequence::SequenceBuilder;
+    use common_procedure::Context as ProcedureContext;
+    use common_procedure_test::MockContextProvider;
     use common_time::util::current_time_millis;
     use store_api::storage::FileId;
     use tokio::sync::mpsc;
@@ -1135,6 +1177,38 @@ mod tests {
     use crate::procedure::test_util::{MailboxContext, send_mock_reply};
     use crate::service::mailbox::Channel;
 
+    #[test]
+    fn test_scoped_gc_instruction_selects_and_scopes_manifest() {
+        let region = RegionId::new(7, 3);
+        let other = RegionId::new(7, 4);
+        let mut manifest = FileRefsManifest::default();
+        manifest.file_refs.insert(region, HashSet::new());
+        manifest.file_refs.insert(other, HashSet::new());
+        manifest.manifest_version.insert(region, 42);
+
+        let packed = scoped_gc_instruction(vec![region], &manifest, true);
+        let Instruction::PackedGcRegions(packed) = packed else {
+            panic!("expected packed instruction");
+        };
+        assert_eq!(packed.regions, vec![region]);
+        assert_eq!(
+            packed.packed_file_refs_manifest.manifest_version[&region],
+            42
+        );
+        assert!(
+            !packed
+                .packed_file_refs_manifest
+                .file_refs
+                .contains_key(&other)
+        );
+        assert!(
+            packed
+                .packed_file_refs_manifest
+                .cross_region_refs
+                .is_empty()
+        );
+    }
+
     #[test]
     fn test_done_with_gc_report_keeps_report() {
         let region_id = RegionId::new(1024, 1);
@@ -1215,6 +1289,7 @@ mod tests {
 
         let kv_backend = Arc::new(MemoryKvBackend::new());
         let table_metadata_manager = Arc::new(TableMetadataManager::new(kv_backend.clone()));
+        let runtime_switch_manager = Arc::new(RuntimeSwitchManager::new(kv_backend.clone()));
         let mailbox_sequence =
             SequenceBuilder::new("test_batch_gc_partial_report", kv_backend).build();
         let mut mailbox = MailboxContext::new(mailbox_sequence);
@@ -1230,6 +1305,7 @@ mod tests {
         let mut procedure = BatchGcProcedure::new(
             mailbox.mailbox().clone(),
             table_metadata_manager,
+            runtime_switch_manager,
             "localhost".to_string(),
             vec![first_region, second_region],
             true,
@@ -1265,11 +1341,15 @@ mod tests {
     fn batch_gc_procedure() -> BatchGcProcedure {
         let kv_backend = Arc::new(MemoryKvBackend::new());
         let table_metadata_manager = Arc::new(TableMetadataManager::new(kv_backend.clone()));
-        let mailbox_sequence = SequenceBuilder::new("test_batch_gc_procedure", kv_backend).build();
+        let runtime_switch_manager = Arc::new(RuntimeSwitchManager::new(kv_backend));
+        let mailbox_sequence =
+            SequenceBuilder::new("test_batch_gc_procedure", Arc::new(MemoryKvBackend::new()))
+                .build();
         let mailbox = MailboxContext::new(mailbox_sequence);
         BatchGcProcedure::new(
             mailbox.mailbox().clone(),
             table_metadata_manager,
+            runtime_switch_manager,
             "localhost".to_string(),
             vec![RegionId::new(1024, 1)],
             true,
@@ -1277,4 +1357,113 @@ mod tests {
             HashMap::new(),
         )
     }
+
+    #[tokio::test]
+    async fn test_maintenance_mode_gates_every_gc_state() {
+        let states = [
+            State::Start,
+            State::Acquiring,
+            State::Gcing,
+            State::UpdateRepartition,
+        ];
+
+        for state in states {
+            let kv_backend = Arc::new(MemoryKvBackend::new());
+            let table_metadata_manager = Arc::new(TableMetadataManager::new(kv_backend.clone()));
+            let runtime_switch_manager = Arc::new(RuntimeSwitchManager::new(kv_backend.clone()));
+            runtime_switch_manager.set_maintenance_mode().await.unwrap();
+            let mailbox_sequence =
+                SequenceBuilder::new("test_batch_gc_maintenance_gate", kv_backend).build();
+            let mut mailbox = MailboxContext::new(mailbox_sequence);
+            let (tx, mut rx) = mpsc::channel(1);
+            mailbox
+                .insert_heartbeat_response_receiver(Channel::Datanode(1), tx)
+                .await;
+            let mut procedure = BatchGcProcedure::new(
+                mailbox.mailbox().clone(),
+                table_metadata_manager,
+                runtime_switch_manager,
+                "localhost".to_string(),
+                vec![RegionId::new(1024, 1)],
+                true,
+                Duration::from_secs(10),
+                HashMap::new(),
+            );
+            procedure.data.state = state.clone();
+            let dump_before = procedure.dump().unwrap();
+
+            let ctx = ProcedureContext {
+                procedure_id: common_procedure::ProcedureId::random(),
+                provider: Arc::new(MockContextProvider::default()),
+                event_context: None,
+            };
+            let err = procedure.execute(&ctx).await.unwrap_err();
+
+            assert!(!err.is_retry_later());
+            assert_eq!(
+                common_error::ext::ErrorExt::retry_hint(&err),
+                common_error::ext::RetryHint::NonRetryable
+            );
+            assert_eq!(procedure.data.state, state);
+            assert_eq!(procedure.dump().unwrap(), dump_before);
+            if matches!(state, State::Acquiring | State::Gcing) {
+                assert!(rx.try_recv().is_err());
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn test_maintenance_mode_read_error_retries_without_advancing_state() {
+        let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let failing_once_calls = calls.clone();
+        let kv_backend = Arc::new(
+            MockKvBackendBuilder::default()
+                .range_fn(Arc::new(move |_| {
+                    if failing_once_calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst) == 0 {
+                        common_meta::error::UnexpectedSnafu {
+                            err_msg: "maintenance read failed",
+                        }
+                        .fail()
+                    } else {
+                        Ok(common_meta::rpc::store::RangeResponse {
+                            kvs: vec![],
+                            more: false,
+                        })
+                    }
+                }))
+                .build()
+                .unwrap(),
+        );
+        let table_metadata_manager = Arc::new(TableMetadataManager::new(kv_backend.clone()));
+        let runtime_switch_manager = Arc::new(RuntimeSwitchManager::new(kv_backend.clone()));
+        let mailbox_sequence =
+            SequenceBuilder::new("test_batch_gc_maintenance_read_error", kv_backend).build();
+        let mailbox = MailboxContext::new(mailbox_sequence);
+        let mut procedure = BatchGcProcedure::new(
+            mailbox.mailbox().clone(),
+            table_metadata_manager,
+            runtime_switch_manager,
+            "localhost".to_string(),
+            vec![RegionId::new(1024, 1)],
+            true,
+            Duration::from_secs(10),
+            HashMap::new(),
+        );
+        let ctx = ProcedureContext {
+            procedure_id: common_procedure::ProcedureId::random(),
+            provider: Arc::new(MockContextProvider::default()),
+            event_context: None,
+        };
+        let dump_before = procedure.dump().unwrap();
+
+        assert!(procedure.execute(&ctx).await.unwrap_err().is_retry_later());
+        assert_eq!(procedure.data.state, State::Start);
+        assert_eq!(procedure.dump().unwrap(), dump_before);
+
+        assert!(matches!(
+            procedure.execute(&ctx).await.unwrap(),
+            Status::Executing { .. }
+        ));
+        assert_eq!(procedure.data.state, State::Acquiring);
+    }
 }

src/meta-srv/src/handler.rs

@@ -428,9 +428,11 @@ impl HeartbeatMailbox {
                 .as_ref()
                 .with_context(|| UnexpectedInstructionReplySnafu {
                     mailbox_message: msg.to_string(),
-                    reason: format!("empty payload, msg: {msg:?}"),
+                    reason: "empty JSON payload".to_string(),
                 })?;
-        serde_json::from_str(payload).context(DeserializeFromJsonSnafu { input: payload })
+        serde_json::from_str(payload).context(DeserializeFromJsonSnafu {
+            input_len: payload.len(),
+        })
     }
 
     /// Parses the [Instruction] from [MailboxMessage].
@@ -443,9 +445,11 @@ impl HeartbeatMailbox {
                 .as_ref()
                 .with_context(|| UnexpectedInstructionReplySnafu {
                     mailbox_message: msg.to_string(),
-                    reason: format!("empty payload, msg: {msg:?}"),
+                    reason: "empty JSON payload".to_string(),
                 })?;
-        serde_json::from_str(payload).context(DeserializeFromJsonSnafu { input: payload })
+        serde_json::from_str(payload).context(DeserializeFromJsonSnafu {
+            input_len: payload.len(),
+        })
     }
 
     pub fn create(pushers: Pushers, sequence: Sequence) -> MailboxRef {
@@ -526,7 +530,12 @@ impl Mailbox for HeartbeatMailbox {
         msg.id = message_id;
 
         let pusher_id = ch.pusher_id();
-        debug!("Sending mailbox message {msg:?} to {pusher_id}");
+        let payload_len = msg
+            .payload
+            .as_ref()
+            .map(|Payload::Json(payload)| payload.len())
+            .unwrap_or_default();
+        debug!(message_id, payload_len, %pusher_id, "Sending mailbox message");
 
         let (tx, rx) = oneshot::channel();
         let _ = self.senders.insert(message_id, tx);
@@ -548,7 +557,12 @@ impl Mailbox for HeartbeatMailbox {
         msg.id = message_id;
 
         let pusher_id = ch.pusher_id();
-        debug!("Sending mailbox message {msg:?} to {pusher_id}");
+        let payload_len = msg
+            .payload
+            .as_ref()
+            .map(|Payload::Json(payload)| payload.len())
+            .unwrap_or_default();
+        debug!(message_id, payload_len, %pusher_id, "Sending one-way mailbox message");
 
         self.pushers.push(pusher_id, msg).await?;
 
@@ -560,7 +574,18 @@ impl Mailbox for HeartbeatMailbox {
     }
 
     async fn on_recv(&self, id: MessageId, maybe_msg: Result<MailboxMessage>) -> Result<()> {
-        debug!("Received mailbox message {maybe_msg:?}");
+        let payload_len = maybe_msg
+            .as_ref()
+            .ok()
+            .and_then(|msg| msg.payload.as_ref())
+            .map(|Payload::Json(payload)| payload.len())
+            .unwrap_or_default();
+        debug!(
+            message_id = id,
+            payload_len,
+            success = maybe_msg.is_ok(),
+            "Received mailbox message"
+        );
 
         let _ = self.timeouts.remove(&id);
 
@@ -568,7 +593,15 @@ impl Mailbox for HeartbeatMailbox {
             tx.send(maybe_msg)
                 .map_err(|_| error::MailboxClosedSnafu { id }.build())?;
         } else if let Ok(finally_msg) = maybe_msg {
-            warn!("The response arrived too late: {finally_msg:?}");
+            let payload_len = finally_msg
+                .payload
+                .as_ref()
+                .map(|Payload::Json(payload)| payload.len())
+                .unwrap_or_default();
+            warn!(
+                message_id = id,
+                payload_len, "The mailbox response arrived too late"
+            );
         }
 
         Ok(())

src/meta-srv/src/key.rs

@@ -87,7 +87,9 @@ impl FromStr for LeaseValue {
     type Err = error::Error;
 
     fn from_str(value: &str) -> crate::Result<Self> {
-        serde_json::from_str(value).context(error::DeserializeFromJsonSnafu { input: value })
+        serde_json::from_str(value).context(error::DeserializeFromJsonSnafu {
+            input_len: value.len(),
+        })
     }
 }
 

src/meta-srv/src/metasrv/builder.rs

@@ -530,6 +530,7 @@ impl MetasrvBuilder {
             let gc_scheduler_ctx = DefaultGcSchedulerCtx::try_new(
                 table_metadata_manager.clone(),
                 procedure_manager.clone(),
+                runtime_switch_manager.clone(),
                 #[cfg(feature = "enterprise")]
                 ddl_manager.clone(),
                 meta_peer_client.clone(),

src/meta-srv/src/service/heartbeat.rs

@@ -206,7 +206,13 @@ where
 
     /// Handles the incoming heartbeat request, and returns whether to continue the session.
     async fn handle_request(&mut self, req: HeartbeatRequest, is_handshake: bool) -> bool {
-        debug!("Receiving heartbeat request: {:?}", req);
+        let mailbox_message_id = req.mailbox_message.as_ref().map(|msg| msg.id);
+        debug!(
+            sender_id = %self.sender_id,
+            ?mailbox_message_id,
+            is_handshake,
+            "Receiving heartbeat request"
+        );
 
         let sender_id = self.sender_id.to_string();
         METRIC_META_HEARTBEAT_RECV
@@ -224,7 +230,12 @@ where
 
         let is_not_leader = res.as_ref().is_ok_and(|r| r.is_not_leader());
 
-        debug!("Sending heartbeat response: {:?}", res);
+        debug!(
+            sender_id = %self.sender_id,
+            success = res.is_ok(),
+            is_not_leader,
+            "Sending heartbeat response"
+        );
 
         if self.tx.send(res).await.is_err() {
             info!(

src/store-api/src/storage/file.rs

@@ -51,6 +51,11 @@ impl FileId {
     pub fn as_bytes(&self) -> &[u8] {
         self.0.as_bytes()
     }
+
+    /// Constructs a file id from its packed 16-byte UUID representation.
+    pub fn from_bytes(bytes: [u8; 16]) -> FileId {
+        FileId(Uuid::from_bytes(bytes))
+    }
 }
 
 impl From<FileId> for Uuid {

tests-integration/src/tests/gc.rs

@@ -208,6 +208,7 @@ async fn test_gc_basic(store_type: &StorageType) {
     let procedure = BatchGcProcedure::new(
         metasrv.mailbox().clone(),
         metasrv.table_metadata_manager().clone(),
+        metasrv.runtime_switch_manager().clone(),
         metasrv.options().grpc.server_addr.clone(),
         regions.clone(),
         false,                   // full_file_listing

tests-integration/src/tests/gc/repart.rs

@@ -90,6 +90,7 @@ CREATE TABLE test_cleanup_repartition (
     let mut procedure = BatchGcProcedure::new_update_repartition_for_test(
         metasrv.mailbox().clone(),
         metasrv.table_metadata_manager().clone(),
+        metasrv.runtime_switch_manager().clone(),
         metasrv.options().grpc.server_addr.clone(),
         regions,
         manifest,
@@ -183,6 +184,7 @@ async fn test_cleanup_region_repartition_preserve_uninvolved_entries() {
     let mut procedure = BatchGcProcedure::new_update_repartition_for_test(
         metasrv.mailbox().clone(),
         metasrv.table_metadata_manager().clone(),
+        metasrv.runtime_switch_manager().clone(),
         metasrv.options().grpc.server_addr.clone(),
         regions,
         manifest,
@@ -267,6 +269,7 @@ async fn test_cleanup_region_repartition_remove_when_tmp_refs_empty() {
     let mut procedure = BatchGcProcedure::new_update_repartition_for_test(
         metasrv.mailbox().clone(),
         metasrv.table_metadata_manager().clone(),
+        metasrv.runtime_switch_manager().clone(),
         metasrv.options().grpc.server_addr.clone(),
         regions,
         manifest,

tests-integration/tests/repartition.rs

@@ -812,6 +812,7 @@ async fn trigger_table_gc(metasrv: &Arc<Metasrv>, table_name: &str) {
     let procedure = BatchGcProcedure::new(
         metasrv.mailbox().clone(),
         metasrv.table_metadata_manager().clone(),
+        metasrv.runtime_switch_manager().clone(),
         metasrv.options().grpc.server_addr.clone(),
         region_ids.clone(),
         false,                   // full_file_listing