feat(network/onion): detect liveness via TcpStream close (#5168)

<!--
Thank you for contributing to nervosnetwork/ckb!

If you haven't already, please read
[CONTRIBUTING](https://github.com/nervosnetwork/ckb/blob/develop/CONTRIBUTING.md)
document.

If you're unsure about anything, just ask; somebody should be along to
answer within a day or two.

**Important**: We use Squash Merge to merge PRs, so the PR title will
become the commit message.
Please ensure your PR title follows the [Conventional Commit
Messages](https://www.conventionalcommits.org/) format.

The most important prefixes you should use:

- `fix:`: represents bug fixes, and results in a SemVer patch bump.
- `feat:`: represents a new feature, and results in a SemVer minor bump.
- `<prefix>!:` (e.g. `feat!:`): represents a breaking change (indicated
by the !) and results in a SemVer major bump.

Other conventional prefixes are also acceptable (e.g., `docs:`,
`refactor:`, `test:`, `chore:`, etc.).
-->
### What problem does this PR solve?

Issue Number: close #5161 <!-- REMOVE this line if no issue to close -->
Follow-up to #5271.

Problem Summary:

The onion service currently polls the Tor daemon every 3 seconds via
`get_uptime()` to detect connection loss. This adds unnecessary latency
(up to 3s for failure detection) and constant I/O overhead. A passive,
event-driven approach would react instantly and reduce resource usage.

### What is changed and how it works?

What's Changed:

- An `oneshot` channel notifies `TorController` when either copy
direction terminates (EOF or error).
- Replaced the 3-second ticker loop in `OnionService` with a branch
awaiting `wait_for_disconnect()`.

### Related changes

### Check List <!--REMOVE the items that are not applicable-->

Tests <!-- At least one of them must be included. -->

- [ ] Unit test
- [ ] Integration test
- [x] Manual test (add detailed scripts or steps below)
- [ ] No code

Manual test steps:
1. `ckb run` with config listen_on_onion = true.
2. Observe the onion service starts successfully.
3. Kill/exit the Tor process.
4. Verify the node logs `"Tor control connection lost"` within
milliseconds, not after a multi-second delay.
5. Restart the Tor daemon and verify the onion service can be
re-established.

Side effects

- ~~Performance regression~~
- ~~Breaking backward compatibility~~

Author: jetjinser

util/onion/src/onion_service.rs

@@ -128,22 +128,15 @@ impl OnionService {
         );
 
         self.handle.spawn(async move {
-            let mut ticker = tokio::time::interval(tokio::time::Duration::from_secs(3));
-            loop {
-                tokio::select! {
-                    _ = ticker.tick() => {
-                        let uptime = tor_controller.get_uptime().await;
-                        if let Err(err) = uptime {
-                            error!("Failed to get tor server uptime: {:?}", err);
-                            drop(tor_server_alive_tx);
-                            return;
-                        }
-                    }
-                    _ = stop_rx.cancelled() => {
-                        info!("OnionService received stop signal, exiting...");
-                        drop(tor_server_alive_tx);
-                        return;
-                    }
+            tokio::select! {
+                _ = tor_controller.wait_for_disconnect() => {
+                    error!("OnionService disconnected, retrying...");
+                    drop(tor_server_alive_tx);
+
+                }
+                _ = stop_rx.cancelled() => {
+                    info!("OnionService received stop signal, exiting...");
+                    drop(tor_server_alive_tx);
                 }
             }
         });

util/onion/src/tor_connection.rs

@@ -11,7 +11,7 @@ use strum::FromRepr;
 use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
 use tokio::net::tcp::OwnedWriteHalf;
 use tokio::net::TcpStream;
-use tokio::sync::mpsc;
+use tokio::sync::{mpsc, oneshot};
 
 /// Tor control protocol status codes.
 ///
@@ -206,6 +206,7 @@ impl Response {
 pub struct TorConnection {
     write: OwnedWriteHalf,
     line_rx: mpsc::UnboundedReceiver<String>,
+    disconnect_rx: oneshot::Receiver<()>,
 }
 
 impl TorConnection {
@@ -215,6 +216,7 @@ impl TorConnection {
     pub(crate) fn connect(stream: TcpStream, handle: Handle) -> Self {
         let (read, write) = stream.into_split();
         let (line_tx, line_rx) = mpsc::unbounded_channel();
+        let (disconnect_tx, disconnect_rx) = oneshot::channel();
 
         handle.spawn(async move {
             let mut reader = BufReader::new(read);
@@ -237,9 +239,19 @@ impl TorConnection {
                 }
             }
             drop(line_tx);
+            _ = disconnect_tx.send(());
         });
 
-        TorConnection { write, line_rx }
+        TorConnection {
+            write,
+            line_rx,
+            disconnect_rx,
+        }
+    }
+
+    /// Wait for the underlying TCP connection to close.
+    pub async fn wait_for_disconnect(self) -> bool {
+        self.disconnect_rx.await.is_ok()
     }
 
     /// Load protocol information from the Tor controller.

util/onion/src/tor_controller.rs

@@ -101,6 +101,16 @@ impl TorController {
         Ok(())
     }
 
+    /// Waits asynchronously until the underlying TCP connection to the Tor
+    /// controller is severed (either cleanly closed or due to an I/O error).
+    ///
+    /// Returns `Some(())` when the connection is lost, or `None` if the internal
+    /// notification channel was closed unexpectedly (which should not occur during
+    /// normal operation).
+    pub async fn wait_for_disconnect(self) -> bool {
+        self.inner.wait_for_disconnect().await
+    }
+
     /// Add a new v3 onion service to the Tor server.
     pub async fn add_onion_v3(
         &mut self,