fix(metasrv): stop parking the heartbeat response on the region supervisor queue (#9474)

* fix(meta-srv): stop parking the heartbeat response on the region supervisor queue

Signed-off-by: lczllx <2181719471@qq.com>

* test(meta-srv): cover the heartbeat response path at the handler level

Signed-off-by: lczllx <2181719471@qq.com>

---------

Signed-off-by: lczllx <2181719471@qq.com>
(cherry picked from commit d1151b0ed6)
Signed-off-by: WenyXu <wenymedia@gmail.com>

src/meta-srv/src/handler/failure_handler.rs

@@ -52,9 +52,10 @@ impl HeartbeatHandler for RegionFailureHandler {
             return Ok(HandleControl::Continue);
         };
 
+        // Never blocks: the region lease has already been decided earlier in the
+        // chain, so a busy supervisor must not withhold the heartbeat response.
         self.heartbeat_acceptor
-            .accept(DatanodeHeartbeat::from(stat))
-            .await;
+            .accept(DatanodeHeartbeat::from(stat));
 
         Ok(HandleControl::Continue)
     }
@@ -62,6 +63,8 @@ impl HeartbeatHandler for RegionFailureHandler {
 
 #[cfg(test)]
 mod tests {
+    use std::time::Duration;
+
     use api::v1::meta::HeartbeatRequest;
     use common_catalog::consts::default_engine;
     use common_meta::datanode::{RegionManifestInfo, RegionStat, Stat};
@@ -73,7 +76,45 @@ mod tests {
     use crate::handler::{HeartbeatAccumulator, HeartbeatHandler};
     use crate::metasrv::builder::MetasrvBuilder;
     use crate::region::supervisor::tests::new_test_supervisor;
-    use crate::region::supervisor::{Event, HeartbeatAcceptor};
+    use crate::region::supervisor::{DatanodeHeartbeat, Event, HeartbeatAcceptor};
+
+    fn new_region_stat(region_id: u64) -> RegionStat {
+        RegionStat {
+            id: RegionId::from_u64(region_id),
+            rcus: 0,
+            wcus: 0,
+            approximate_bytes: 0,
+            engine: default_engine().to_string(),
+            role: RegionRole::Follower,
+            num_rows: 0,
+            memtable_size: 0,
+            manifest_size: 0,
+            sst_size: 0,
+            sst_num: 0,
+            index_size: 0,
+            region_manifest: RegionManifestInfo::Mito {
+                manifest_version: 0,
+                flushed_entry_id: 0,
+                file_removed_cnt: 0,
+            },
+            data_topic_latest_entry_id: 0,
+            metadata_topic_latest_entry_id: 0,
+            written_bytes: 0,
+            query_cpu_time: 0,
+            query_scanned_bytes: 0,
+            min_timestamp: None,
+            max_timestamp: None,
+        }
+    }
+
+    fn new_stat() -> Stat {
+        Stat {
+            id: 42,
+            region_stats: vec![new_region_stat(1), new_region_stat(2), new_region_stat(3)],
+            timestamp_millis: 1000,
+            ..Default::default()
+        }
+    }
 
     #[tokio::test]
     async fn test_handle_heartbeat() {
@@ -85,40 +126,7 @@ mod tests {
         let metasrv = builder.build().await.unwrap();
         let mut ctx = metasrv.new_ctx();
         let acc = &mut HeartbeatAccumulator::default();
-        fn new_region_stat(region_id: u64) -> RegionStat {
-            RegionStat {
-                id: RegionId::from_u64(region_id),
-                rcus: 0,
-                wcus: 0,
-                approximate_bytes: 0,
-                engine: default_engine().to_string(),
-                role: RegionRole::Follower,
-                num_rows: 0,
-                memtable_size: 0,
-                manifest_size: 0,
-                sst_size: 0,
-                sst_num: 0,
-                index_size: 0,
-                region_manifest: RegionManifestInfo::Mito {
-                    manifest_version: 0,
-                    flushed_entry_id: 0,
-                    file_removed_cnt: 0,
-                },
-                data_topic_latest_entry_id: 0,
-                metadata_topic_latest_entry_id: 0,
-                written_bytes: 0,
-                query_cpu_time: 0,
-                query_scanned_bytes: 0,
-                min_timestamp: None,
-                max_timestamp: None,
-            }
-        }
-        acc.stat = Some(Stat {
-            id: 42,
-            region_stats: vec![new_region_stat(1), new_region_stat(2), new_region_stat(3)],
-            timestamp_millis: 1000,
-            ..Default::default()
-        });
+        acc.stat = Some(new_stat());
 
         handler.handle(req, &mut ctx, acc).await.unwrap();
         let (tx, rx) = oneshot::channel();
@@ -126,4 +134,39 @@ mod tests {
         let detector = rx.await.unwrap();
         assert_eq!(detector.iter().collect::<Vec<_>>().len(), 3);
     }
+
+    #[tokio::test]
+    async fn test_handle_heartbeat_does_not_block_on_full_supervisor_queue() {
+        // Regression test for #9419: a full queue used to park the handler.
+        // Keep the supervisor alive, or the queue reports Closed.
+        let (_supervisor, sender) = new_test_supervisor();
+        // Built directly, so the supervisor stays undrained.
+        let handler = RegionFailureHandler {
+            heartbeat_acceptor: HeartbeatAcceptor::new(sender.clone()),
+        };
+
+        let stat = new_stat();
+        let mut queued = 0;
+        while sender
+            .try_send(Event::HeartbeatArrived(DatanodeHeartbeat::from(&stat)))
+            .is_ok()
+        {
+            queued += 1;
+        }
+        assert_eq!(queued, sender.max_capacity());
+
+        let builder = MetasrvBuilder::new();
+        let metasrv = builder.build().await.unwrap();
+        let mut ctx = metasrv.new_ctx();
+        let acc = &mut HeartbeatAccumulator::default();
+        // Without a stat the handler returns before the acceptor.
+        acc.stat = Some(stat);
+
+        let result = tokio::time::timeout(
+            Duration::from_secs(5),
+            handler.handle(&HeartbeatRequest::default(), &mut ctx, acc),
+        )
+        .await;
+        assert!(result.is_ok());
+    }
 }

src/meta-srv/src/metrics.rs

@@ -48,6 +48,9 @@ lazy_static! {
     ///  Heartbeat received by metasrv.
     pub static ref METRIC_META_HEARTBEAT_RECV: IntCounterVec =
         register_int_counter_vec!("greptime_meta_heartbeat_recv", "heartbeats received by metasrv", &["pusher_key"]).unwrap();
+    /// Heartbeats dropped by metasrv.
+    pub static ref METRIC_META_HEARTBEAT_DROPPED: IntCounterVec =
+        register_int_counter_vec!("greptime_meta_heartbeat_dropped", "heartbeats dropped by metasrv", &["reason"]).unwrap();
     /// The migration execute histogram.
     pub static ref METRIC_META_REGION_MIGRATION_EXECUTE: HistogramVec =
         register_histogram_vec!("greptime_meta_region_migration_execute", "meta region migration execute", &["state"]).unwrap();

src/meta-srv/src/region/supervisor.rs

@@ -35,6 +35,7 @@ use common_time::util::current_time_millis;
 use futures::{StreamExt, TryStreamExt};
 use snafu::{ResultExt, ensure};
 use store_api::storage::RegionId;
+use tokio::sync::mpsc::error::TrySendError;
 use tokio::sync::mpsc::{Receiver, Sender};
 use tokio::sync::oneshot;
 use tokio::time::{MissedTickBehavior, interval, interval_at};
@@ -43,6 +44,7 @@ use crate::discovery::utils::accept_ingest_workload;
 use crate::error::{self, Result};
 use crate::failure_detector::PhiAccrualFailureDetectorOptions;
 use crate::metasrv::{RegionStatAwareSelectorRef, SelectTarget, SelectorContext, SelectorRef};
+use crate::metrics::METRIC_META_HEARTBEAT_DROPPED;
 use crate::procedure::region_migration::manager::{
     RegionMigrationManagerRef, RegionMigrationTriggerReason, SubmitRegionMigrationTaskResult,
 };
@@ -356,10 +358,29 @@ impl HeartbeatAcceptor {
         Self { sender }
     }
 
-    /// Accepts heartbeats from datanodes.
-    pub(crate) async fn accept(&self, heartbeat: DatanodeHeartbeat) {
-        if let Err(err) = self.sender.send(Event::HeartbeatArrived(heartbeat)).await {
-            error!(err; "RegionSupervisor has stop receiving heartbeat.");
+    /// Accepts heartbeats from datanodes without waiting for queue capacity: a full queue
+    /// drops the heartbeat, counted in [`METRIC_META_HEARTBEAT_DROPPED`]. A dropped
+    /// heartbeat costs the failure detector one sample.
+    pub(crate) fn accept(&self, heartbeat: DatanodeHeartbeat) {
+        let datanode_id = heartbeat.datanode_id;
+        match self.sender.try_send(Event::HeartbeatArrived(heartbeat)) {
+            Ok(()) => {}
+            Err(TrySendError::Full(_)) => {
+                METRIC_META_HEARTBEAT_DROPPED
+                    .with_label_values(&["full"])
+                    .inc();
+                warn!(
+                    "Dropping heartbeat because the region supervisor event queue is full, datanode_id: {}, queue_capacity: {}",
+                    datanode_id,
+                    self.sender.max_capacity()
+                );
+            }
+            Err(TrySendError::Closed(err)) => {
+                METRIC_META_HEARTBEAT_DROPPED
+                    .with_label_values(&["closed"])
+                    .inc();
+                error!(err; "RegionSupervisor has stop receiving heartbeat.");
+            }
         }
     }
 }
@@ -899,14 +920,15 @@ pub(crate) mod tests {
     use tokio::time::sleep;
 
     use super::RegionSupervisorSelector;
+    use crate::metrics::METRIC_META_HEARTBEAT_DROPPED;
     use crate::procedure::region_migration::RegionMigrationTriggerReason;
     use crate::procedure::region_migration::manager::{
         RegionMigrationManager, SubmitRegionMigrationTaskResult,
     };
     use crate::procedure::region_migration::test_util::TestingEnv;
     use crate::region::supervisor::{
-        DatanodeHeartbeat, Event, RegionFailureDetectorControl, RegionSupervisor,
-        RegionSupervisorTicker,
+        DatanodeHeartbeat, Event, HeartbeatAcceptor, RegionFailureDetectorControl,
+        RegionSupervisor, RegionSupervisorTicker,
     };
     use crate::selector::test_utils::{RandomNodeSelector, new_test_selector_context};
 
@@ -940,6 +962,14 @@ pub(crate) mod tests {
         )
     }
 
+    fn filler_heartbeat() -> DatanodeHeartbeat {
+        DatanodeHeartbeat {
+            datanode_id: 1,
+            regions: vec![],
+            timestamp: 0,
+        }
+    }
+
     #[tokio::test]
     async fn test_heartbeat() {
         let (mut supervisor, sender) = new_test_supervisor();
@@ -1372,4 +1402,39 @@ pub(crate) mod tests {
         assert!(supervisor.failure_detector.contains(&detecting_region));
         assert!(supervisor.failover_counts.contains_key(&detecting_region));
     }
+
+    #[tokio::test]
+    async fn test_heartbeat_accept_does_not_block_on_full_queue() {
+        // Regression test for #9419: a full queue used to park the caller.
+        // The supervisor is never run, so the queue stays full.
+        let (mut supervisor, sender) = new_test_supervisor();
+
+        let mut queued = 0;
+        while sender
+            .try_send(Event::HeartbeatArrived(filler_heartbeat()))
+            .is_ok()
+        {
+            queued += 1;
+        }
+        assert_eq!(queued, sender.max_capacity());
+        assert_eq!(sender.capacity(), 0);
+
+        let acceptor = HeartbeatAcceptor::new(sender.clone());
+        let dropped = || {
+            METRIC_META_HEARTBEAT_DROPPED
+                .with_label_values(&["full"])
+                .get()
+        };
+        let dropped_before = dropped();
+        // The full queue drops the heartbeat and counts it.
+        acceptor.accept(filler_heartbeat());
+        assert_eq!(dropped(), dropped_before + 1);
+        assert_eq!(sender.capacity(), 0);
+
+        // Free one slot: the next heartbeat is accepted.
+        supervisor.receiver.try_recv().unwrap();
+        acceptor.accept(filler_heartbeat());
+        assert_eq!(dropped(), dropped_before + 1);
+        assert_eq!(sender.capacity(), 0);
+    }
 }