Repository: tikv/tikv (language: rust)

Issue:
Ensure `Deregister::Conn` Is Always Triggered in CDC Connection Handling
## Bug Report

<!-- Thanks for your bug report! Don't worry if you can't fill out all the sections. -->

Currently, in the CDC connection handling logic, `Deregister::Conn(conn_id)` may not be triggered if an error occurs early in `recv_req`. This can lead to connection leaks, as the connection might not be properly deregistered.

The issue occurs in the following code:

```rust
let recv_req = async move {
    let mut stream = stream.map_err(|e| format!("{:?}", e));
    if let Some(request) = stream.try_next().await? {
        Self::set_conn_version(&scheduler, conn_id, version, explicit_features)?;
        Self::handle_request(&scheduler, &peer, request, conn_id)?;
    }
    while let Some(request) = stream.try_next().await? {
        Self::handle_request(&scheduler, &peer, request, conn_id)?;
    }
    let deregister = Deregister::Conn(conn_id);
    if let Err(e) = scheduler.schedule(Task::Deregister(deregister)) {
        error!("cdc deregister failed"; "error" => ?e, "conn_id" => ?conn_id);
    }
    Ok::<(), String>(())
};
```

If an error occurs before reaching the deregistration step, `Deregister::Conn(conn_id)` will not be executed, leading to a dangling connection.

### What version of TiKV are you using?
<!-- You can run `tikv-server --version` -->

master

### What did you expect?

`Deregister::Conn(conn_id)` should always be executed, regardless of whether `recv_req` completes successfully or encounters an error.

### Impact

* Connections may not be properly deregistered.
* Potential resource leaks affecting CDC stability.



Discussion / hints:

Fix patch:
diff --git a/components/cdc/src/service.rs b/components/cdc/src/service.rs
index 0ecd2eb606c..6f1a7c08d08 100644
--- a/components/cdc/src/service.rs
+++ b/components/cdc/src/service.rs
@@ -455,13 +455,10 @@ impl Service {
             while let Some(request) = stream.try_next().await? {
                 Self::handle_request(&scheduler, &peer, request, conn_id)?;
             }
-            let deregister = Deregister::Conn(conn_id);
-            if let Err(e) = scheduler.schedule(Task::Deregister(deregister)) {
-                error!("cdc deregister failed"; "error" => ?e, "conn_id" => ?conn_id);
-            }
             Ok::<(), String>(())
         };
 
+        let scheduler_dereg = self.scheduler.clone();
         let peer = ctx.peer();
         ctx.spawn(async move {
             if let Err(e) = recv_req.await {
@@ -469,6 +466,11 @@ impl Service {
             } else {
                 info!("cdc receive closed"; "downstream" => peer, "conn_id" => ?conn_id);
             }
+
+            let deregister = Deregister::Conn(conn_id);
+            if let Err(e) = scheduler_dereg.schedule(Task::Deregister(deregister)) {
+                error!("cdc deregister failed"; "error" => ?e, "conn_id" => ?conn_id);
+            }
         });
 
         let peer = ctx.peer();


Test patch: