fix: маршруты, медиа и Worker; Android APK и статический ARM64 CLI (#60)

* fix: address routing, media, worker and platform issues

* ci: use available Android tools and verify Windows CLI artifact

* fix(gui): return a result from asynchronous stop command

* test(android): verify installed APK and proxy lifecycle on emulator

* fix(media): use native MTProto for CDN203 and retain recent diagnostics

* Fix owned WebSocket split after transport boxing

* Clarify CDN transport and connection status guarantees

* Recognize accessible Android power button labels

* chore(release): prepare 2.0.0-beta.15 with verified draft publication

---------

Co-authored-by: babin <Thartewerner536e@engineer.com>

src/proxy.rs

@@ -1,5 +1,5 @@
 use crate::config::ListenConfig;
-use std::collections::{HashSet, VecDeque};
+use std::collections::VecDeque;
 use std::net::{IpAddr, Ipv4Addr, SocketAddr};
 use std::sync::atomic::{AtomicBool, AtomicU16, AtomicU32, Ordering};
 use std::sync::{Arc, Mutex};
@@ -71,15 +71,17 @@ pub struct Stats {
     shutdown: Mutex<Option<tokio::sync::watch::Sender<bool>>>,
 }
 
-/// Однократные сообщения о том, что происходит с подключениями.
+/// Сообщения о подключениях со склейкой недавних повторов.
 ///
-/// Однократные намеренно: отклонённый адрес повторяется десятки раз в минуту,
+/// Склейка намеренна: отклонённый адрес повторяется десятки раз в минуту,
 /// и без склейки журнал превратился бы в одну строку, повторённую сто раз.
 /// Число повторов при этом не теряется — оно в счётчике `blocked`.
 #[derive(Default)]
 struct Events {
     pending: VecDeque<String>,
-    seen: HashSet<String>,
+    // FIFO of recent distinct messages, independent from the consumer queue.
+    // Once full, evict the oldest entry rather than silencing all future logs.
+    seen: VecDeque<String>,
 }
 
 /// Сколько разных событий помним, чтобы буфер не рос без границы.
@@ -132,13 +134,20 @@ impl Stats {
         })
     }
 
-    /// Запомнить событие, если такого ещё не было.
+    /// Запомнить событие, если его нет среди последних EVENT_LIMIT разных сообщений.
     pub fn note(&self, message: impl Into<String>) {
         let message = message.into();
         let mut events = self.events.lock().unwrap();
-        if events.seen.len() >= EVENT_LIMIT || !events.seen.insert(message.clone()) {
+        if events.seen.contains(&message) {
             return;
         }
+        if events.seen.len() == EVENT_LIMIT {
+            events.seen.pop_front();
+        }
+        events.seen.push_back(message.clone());
+        if events.pending.len() == EVENT_LIMIT {
+            events.pending.pop_front();
+        }
         events.pending.push_back(message);
     }
 
@@ -683,13 +692,9 @@ fn dc_from_init(init: &[u8; 64]) -> Option<(u16, bool)> {
 
 // -- WebSocket tunnel -------------------------------------------------------
 
-/// Keeps `Stats::ws` equal to the number of *established* tunnels.
-///
-/// Counting attempts instead would let the interface announce «Telegram на
-/// связи» while the WebSocket handshake is still failing over between routes,
-/// which takes seconds per route. Reporting a working tunnel that does not
-/// exist yet is the whole reason users saw «прокси подключён, а Telegram не
-/// работает».
+/// Counts opened upstream transports, not pending connection attempts.
+/// WebSocket routes have completed their upgrade; native CDN routes have an
+/// open TCP socket. Neither alone proves a successful Telegram protocol reply.
 struct EstablishedTunnel<'a>(&'a Stats);
 
 impl<'a> EstablishedTunnel<'a> {
@@ -715,7 +720,7 @@ async fn ws_tunnel(
 ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     use futures_util::{SinkExt, StreamExt};
 
-    let (ws, connected) = match stats.transport.connect(dc, media).await {
+    let (connection, connected) = match stats.transport.connect(dc, media).await {
         Ok(connected) => connected,
         Err(error) => {
             // Единственное место, где известно, ПОЧЕМУ туннеля нет. Раньше
@@ -729,6 +734,13 @@ async fn ws_tunnel(
     let _tunnel = EstablishedTunnel::new(stats);
     stats.note_tunnel(dc, connected.route.kind.ui_code());
 
+    let ws = match connection {
+        crate::transport::TelegramConnection::WebSocket(ws) => *ws,
+        crate::transport::TelegramConnection::Tcp(remote) => {
+            return cdn_tcp_tunnel(tcp, remote, init, crypto, stats, &connected, dc).await;
+        }
+    };
+
     let (mut tcp_r, mut tcp_w) = tokio::io::split(tcp);
     let (mut ws_w, mut ws_r) = ws.split();
     let (upstream_crypto, downstream_crypto) = match crypto.map(|crypto| crypto.split()) {
@@ -738,7 +750,8 @@ async fn ws_tunnel(
 
     // Send buffered init as first frame
     ws_w.send(tungstenite::Message::Binary(init.to_vec()))
-        .await?;
+        .await
+        .map_err(|error| tunnel_route_error(stats, &connected, dc, &error.to_string()))?;
 
     // Ping приходит в половину, которая читает, а отвечать на него должна та,
     // которая пишет: владелец у отправляющей половины строго один.
@@ -752,9 +765,11 @@ async fn ws_tunnel(
     // сессия встаёт при живом туннеле (by-sonic/tglock#42, #32).
     let downstream = async {
         let mut crypto = downstream_crypto;
+        let mut received_payload = false;
         while let Some(message) = ws_r.next().await {
             match message {
                 Ok(tungstenite::Message::Binary(mut data)) => {
+                    received_payload |= !data.is_empty();
                     if let Some(crypto) = &mut crypto {
                         crypto.apply(data.as_mut());
                     }
@@ -766,10 +781,48 @@ async fn ws_tunnel(
                         break;
                     }
                 }
-                Ok(tungstenite::Message::Close(_)) | Err(_) => break,
+                Ok(tungstenite::Message::Close(frame)) => {
+                    // Telegram may send code 1000 even for transport errors;
+                    // the actual error is a signed decimal in the reason.
+                    // https://core.telegram.org/mtproto/transports#websocket
+                    let transport_error = frame.as_ref().is_some_and(|frame| {
+                        frame
+                            .reason
+                            .trim()
+                            .parse::<i32>()
+                            .is_ok_and(|code| code != 0)
+                            || frame.code != tungstenite::protocol::frame::coding::CloseCode::Normal
+                    });
+                    if !received_payload || transport_error {
+                        let reason = match frame {
+                            Some(frame) => {
+                                format!("WebSocket закрыт: {} {}", frame.code, frame.reason)
+                            }
+                            None => "WebSocket закрыт без ответа Telegram".to_owned(),
+                        };
+                        return Err(tunnel_route_error(stats, &connected, dc, &reason));
+                    }
+                    return Ok(());
+                }
+                Err(error) => {
+                    return Err(tunnel_route_error(
+                        stats,
+                        &connected,
+                        dc,
+                        &error.to_string(),
+                    ));
+                }
                 Ok(_) => {}
             }
         }
+        if !received_payload {
+            return Err(tunnel_route_error(
+                stats,
+                &connected,
+                dc,
+                "WebSocket завершён без ответа Telegram",
+            ));
+        }
         Ok::<(), Box<dyn std::error::Error + Send + Sync>>(())
     };
 
@@ -779,19 +832,22 @@ async fn ws_tunnel(
         loop {
             tokio::select! {
                 read = tcp_r.read(&mut buf) => match read {
-                    Ok(0) | Err(_) => break,
+                    Ok(0) => break,
+                    Err(error) => return Err(error.into()),
                     Ok(read) => {
                         if let Some(crypto) = &mut crypto {
                             crypto.apply(&mut buf[..read]);
                         }
                         ws_w
                             .send(tungstenite::Message::Binary(buf[..read].to_vec()))
-                            .await?;
+                            .await
+                            .map_err(|error| tunnel_route_error(stats, &connected, dc, &error.to_string()))?;
                     }
                 },
                 payload = pong_rx.recv() => match payload {
                     Some(payload) => {
-                        ws_w.send(tungstenite::Message::Pong(payload)).await?;
+                        ws_w.send(tungstenite::Message::Pong(payload)).await
+                            .map_err(|error| tunnel_route_error(stats, &connected, dc, &error.to_string()))?;
                     }
                     None => break,
                 },
@@ -809,6 +865,94 @@ async fn ws_tunnel(
     Ok(())
 }
 
+/// DC203's pinned CDN endpoint speaks obfuscated MTProto over TCP. Preserve the
+/// same generated init and independent CTR streams used for a WebSocket route.
+/// The transport engine restricts this connection to the exact CDN destination.
+async fn cdn_tcp_tunnel(
+    client: TcpStream,
+    mut remote: TcpStream,
+    init: &[u8; 64],
+    crypto: Option<crate::mtproto::CryptoContext>,
+    stats: &Stats,
+    connected: &crate::transport::ConnectedRoute,
+    dc: u16,
+) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    remote
+        .write_all(init)
+        .await
+        .map_err(|error| tunnel_route_error(stats, connected, dc, &error.to_string()))?;
+    let (mut client_r, mut client_w) = client.into_split();
+    let (mut remote_r, mut remote_w) = remote.into_split();
+    let (upstream_crypto, downstream_crypto) = match crypto.map(|crypto| crypto.split()) {
+        Some((upstream, downstream)) => (Some(upstream), Some(downstream)),
+        None => (None, None),
+    };
+    let upstream = async {
+        let mut crypto = upstream_crypto;
+        let mut buffer = vec![0; 65536];
+        loop {
+            let read = client_r.read(&mut buffer).await?;
+            if read == 0 {
+                return Ok::<_, Box<dyn std::error::Error + Send + Sync>>(());
+            }
+            if let Some(crypto) = &mut crypto {
+                crypto.apply(&mut buffer[..read]);
+            }
+            remote_w
+                .write_all(&buffer[..read])
+                .await
+                .map_err(|error| tunnel_route_error(stats, connected, dc, &error.to_string()))?;
+        }
+    };
+    let downstream = async {
+        let mut crypto = downstream_crypto;
+        let mut buffer = vec![0; 65536];
+        let mut received_payload = false;
+        loop {
+            let read = remote_r
+                .read(&mut buffer)
+                .await
+                .map_err(|error| tunnel_route_error(stats, connected, dc, &error.to_string()))?;
+            if read == 0 {
+                return if received_payload {
+                    Ok(())
+                } else {
+                    Err(tunnel_route_error(
+                        stats,
+                        connected,
+                        dc,
+                        "CDN TCP закрыт без ответа Telegram",
+                    ))
+                };
+            }
+            received_payload = true;
+            if let Some(crypto) = &mut crypto {
+                crypto.apply(&mut buffer[..read]);
+            }
+            client_w.write_all(&buffer[..read]).await?;
+        }
+    };
+    tokio::select! {
+        result = upstream => result,
+        result = downstream => result,
+    }
+}
+
+fn tunnel_route_error(
+    stats: &Stats,
+    connected: &crate::transport::ConnectedRoute,
+    dc: u16,
+    reason: &str,
+) -> Box<dyn std::error::Error + Send + Sync> {
+    stats.transport.report_route_failure(connected);
+    let message = format!(
+        "Сбой туннеля DC{dc} через {}: {reason}",
+        connected.route.connect_host
+    );
+    stats.note(message.clone());
+    message.into()
+}
+
 async fn tcp_relay(a: TcpStream, b: TcpStream) {
     let (mut ar, mut aw) = tokio::io::split(a);
     let (mut br, mut bw) = tokio::io::split(b);
@@ -953,6 +1097,7 @@ mod tests {
             .send(Message::Binary(wire))
             .await
             .map_err(|e| e.to_string())?;
+        websocket.close(None).await.map_err(|e| e.to_string())?;
         Ok((requested, init, request))
     }
 
@@ -1040,6 +1185,52 @@ mod tests {
         );
     }
 
+    #[test]
+    fn diagnostics_keep_new_events_after_the_first_sixty_four() {
+        let stats = Stats::with_secret([0; 16]);
+        for index in 0..=EVENT_LIMIT {
+            stats.note(format!("route failure {index}"));
+        }
+        {
+            let events = stats.events.lock().unwrap();
+            assert_eq!(events.seen.len(), EVENT_LIMIT);
+            assert_eq!(events.pending.len(), EVENT_LIMIT);
+        }
+        let events = stats.drain_events();
+        assert_eq!(events.first().unwrap(), "route failure 1");
+        assert_eq!(
+            events.last().unwrap(),
+            &format!("route failure {EVENT_LIMIT}")
+        );
+
+        // Draining must not discard deduplication history, but must not stop
+        // the next new error from reaching the operator either.
+        stats.note(format!("route failure {EVENT_LIMIT}"));
+        stats.note("a new upstream error");
+        assert_eq!(stats.drain_events(), vec!["a new upstream error"]);
+    }
+
+    #[test]
+    fn an_evicted_event_can_be_reported_again_without_repeating_recent_events() {
+        let stats = Stats::with_secret([0; 16]);
+        stats.note("original error");
+        assert_eq!(stats.drain_events(), vec!["original error"]);
+        stats.note("original error");
+        assert!(stats.drain_events().is_empty());
+
+        // More distinct events than the retained history used to silence
+        // diagnostics permanently, even when the frontend drained every one.
+        for index in 0..EVENT_LIMIT * 2 {
+            let message = format!("error {index}");
+            stats.note(message.clone());
+            assert_eq!(stats.drain_events(), vec![message]);
+        }
+        stats.note("original error");
+        stats.note("original error");
+        assert_eq!(stats.drain_events(), vec!["original error"]);
+        assert_eq!(stats.events.lock().unwrap().seen.len(), EVENT_LIMIT);
+    }
+
     /// Диагностика обязана показывать пару из одного соединения.
     ///
     /// Пока это были два независимых поля, при десятках одновременных
@@ -1411,6 +1602,250 @@ mod tests {
         let _ = server.await.unwrap();
     }
 
+    #[allow(clippy::result_large_err)]
+    #[tokio::test]
+    async fn post_handshake_failures_are_reported_and_penalize_the_route() {
+        use futures_util::{SinkExt, StreamExt};
+        use tungstenite::protocol::{frame::coding::CloseCode, CloseFrame};
+
+        // No reply, abrupt socket loss, and a Telegram transport error carried
+        // in a nominally normal WebSocket close all used to disappear silently.
+        for mode in 0..4 {
+            let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+            let relay_port = listener.local_addr().unwrap().port();
+            let relay = tokio::spawn(async move {
+                let (tcp, _) = listener.accept().await.unwrap();
+                let mut ws = tokio_tungstenite::accept_hdr_async(
+                    tcp,
+                    |_: &Request, mut response: Response| {
+                        response
+                            .headers_mut()
+                            .insert("Sec-WebSocket-Protocol", "binary".parse().unwrap());
+                        Ok(response)
+                    },
+                )
+                .await
+                .unwrap();
+                assert!(matches!(ws.next().await, Some(Ok(Message::Binary(_)))));
+                if mode >= 2 {
+                    ws.send(Message::Binary(vec![1, 2, 3, 4])).await.unwrap();
+                }
+                if mode != 1 {
+                    ws.close(Some(CloseFrame {
+                        code: CloseCode::Normal,
+                        reason: if mode == 2 { " -404 " } else { "" }.into(),
+                    }))
+                    .await
+                    .unwrap();
+                }
+            });
+            let stats = Stats::new();
+            stats.transport.force_local_route(relay_port);
+            let (port, server) = start_proxy(stats.clone(), false).await;
+            let mut client = TcpStream::connect(("127.0.0.1", port)).await.unwrap();
+            client
+                .write_all(&unambiguous_client_init(&stats.secret, 2))
+                .await
+                .unwrap();
+            let mut received = Vec::new();
+            tokio::time::timeout(Duration::from_secs(5), client.read_to_end(&mut received))
+                .await
+                .unwrap()
+                .unwrap();
+            relay.await.unwrap();
+            wait_until("connection task finished", || {
+                stats.active.load(Ordering::Relaxed) == 0
+            })
+            .await;
+            let failed = u32::from(mode != 3);
+            assert_eq!(
+                stats.ws_failures.load(Ordering::Relaxed),
+                failed,
+                "mode {mode}"
+            );
+            assert_eq!(stats.transport.route_failures(), failed, "mode {mode}");
+            let events = stats.drain_events();
+            assert_eq!(
+                events
+                    .iter()
+                    .any(|event| event.contains("Сбой туннеля DC2 через 127.0.0.1")),
+                mode != 3
+            );
+            if mode == 2 {
+                assert!(events.iter().any(|event| event.contains("-404")));
+            }
+            stats.stop();
+            server.await.unwrap().unwrap();
+        }
+    }
+
+    #[tokio::test]
+    async fn fragmented_encrypted_upload_and_download_preserve_both_streams() {
+        check_fragmented_encrypted_duplex(false).await;
+    }
+
+    #[tokio::test]
+    async fn cdn_tcp_preserves_fragmented_encrypted_duplex_and_clean_close() {
+        check_fragmented_encrypted_duplex(true).await;
+    }
+
+    #[tokio::test]
+    async fn cdn_tcp_closing_before_a_reply_is_reported_as_a_route_failure() {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let relay_port = listener.local_addr().unwrap().port();
+        let relay = tokio::spawn(async move {
+            let (mut stream, _) = listener.accept().await.unwrap();
+            let mut init = [0; INIT_LEN];
+            stream.read_exact(&mut init).await.unwrap();
+        });
+        let stats = Stats::new();
+        stats.transport.force_local_route_with(
+            relay_port,
+            crate::transport::RouteKind::TelegramTcp,
+            String::new(),
+        );
+        let (port, server) = start_proxy(stats.clone(), false).await;
+        let mut client = TcpStream::connect(("127.0.0.1", port)).await.unwrap();
+        client
+            .write_all(&unambiguous_client_init(&stats.secret, 203))
+            .await
+            .unwrap();
+        wait_until("CDN failure is counted", || {
+            stats.ws_failures.load(Ordering::Relaxed) == 1
+        })
+        .await;
+        assert_eq!(stats.transport.route_failures(), 1);
+        assert!(stats
+            .drain_events()
+            .iter()
+            .any(|event| event.contains("CDN TCP закрыт без ответа")));
+        relay.await.unwrap();
+        stats.stop();
+        server.await.unwrap().unwrap();
+    }
+
+    #[allow(clippy::result_large_err)]
+    async fn check_fragmented_encrypted_duplex(raw_tcp: bool) {
+        use futures_util::{SinkExt, StreamExt};
+        let request: Vec<u8> = (0..131_072).map(|i| (i % 251) as u8).collect();
+        let response: Vec<u8> = (0..262_144).map(|i| (i % 239) as u8).collect();
+        let expected_request = request.clone();
+        let relay_response = response.clone();
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let relay_port = listener.local_addr().unwrap().port();
+        let relay = tokio::spawn(async move {
+            let (mut tcp, _) = listener.accept().await.unwrap();
+            if raw_tcp {
+                let mut init = [0; INIT_LEN];
+                tcp.read_exact(&mut init).await.unwrap();
+                let mut peer = crate::mtproto::test_relay_peer(&init);
+                let mut download = relay_response;
+                peer.encrypt(&mut download);
+                let (mut reader, mut writer) = tcp.into_split();
+                let send = async {
+                    for chunk in download.chunks(8191) {
+                        writer.write_all(chunk).await.unwrap();
+                        tokio::task::yield_now().await;
+                    }
+                };
+                let receive = async {
+                    let mut received = vec![0; expected_request.len()];
+                    reader.read_exact(&mut received).await.unwrap();
+                    peer.decrypt(&mut received);
+                    assert_eq!(received, expected_request);
+                };
+                tokio::join!(send, receive);
+                writer.shutdown().await.unwrap();
+                return;
+            }
+            let mut ws =
+                tokio_tungstenite::accept_hdr_async(tcp, |_: &Request, mut response: Response| {
+                    response
+                        .headers_mut()
+                        .insert("Sec-WebSocket-Protocol", "binary".parse().unwrap());
+                    Ok(response)
+                })
+                .await
+                .unwrap();
+            let init = match ws.next().await.unwrap().unwrap() {
+                Message::Binary(init) => init,
+                other => panic!("expected init: {other:?}"),
+            };
+            let mut peer = crate::mtproto::test_relay_peer(&init.try_into().unwrap());
+            let mut download = relay_response;
+            peer.encrypt(&mut download);
+            let (mut sink, mut stream) = ws.split();
+            let send = async {
+                for chunk in download.chunks(8191) {
+                    sink.send(Message::Binary(chunk.to_vec())).await.unwrap();
+                }
+            };
+            let receive = async {
+                let mut received = Vec::new();
+                while received.len() < expected_request.len() {
+                    if let Message::Binary(mut chunk) = stream.next().await.unwrap().unwrap() {
+                        peer.decrypt(&mut chunk);
+                        received.extend_from_slice(&chunk);
+                    }
+                }
+                assert_eq!(received, expected_request);
+            };
+            tokio::join!(send, receive);
+            sink.close().await.unwrap();
+        });
+        let stats = Stats::new();
+        if raw_tcp {
+            stats.transport.force_local_route_with(
+                relay_port,
+                crate::transport::RouteKind::TelegramTcp,
+                String::new(),
+            );
+        } else {
+            stats.transport.force_local_route(relay_port);
+        }
+        let (port, server) = start_proxy(stats.clone(), false).await;
+        let init = unambiguous_client_init(&stats.secret, if raw_tcp { 203 } else { -4 });
+        let mut peer = crate::mtproto::test_client_peer(&init, &stats.secret);
+        let mut client = TcpStream::connect(("127.0.0.1", port)).await.unwrap();
+        // A mobile TCP stream can split the 64-byte init and subsequent data
+        // at arbitrary boundaries. WebSocket boundaries need not match these.
+        for chunk in init.chunks(7) {
+            client.write_all(chunk).await.unwrap();
+            tokio::task::yield_now().await;
+        }
+        let mut upload = request;
+        peer.encrypt(&mut upload);
+        let (mut reader, mut writer) = client.into_split();
+        let send = async {
+            for chunk in upload.chunks(137) {
+                writer.write_all(chunk).await.unwrap();
+                tokio::task::yield_now().await;
+            }
+        };
+        let receive = async {
+            let mut received = vec![0; response.len()];
+            reader.read_exact(&mut received).await.unwrap();
+            peer.decrypt(&mut received);
+            assert_eq!(received, response);
+        };
+        tokio::time::timeout(Duration::from_secs(10), async {
+            tokio::join!(send, receive);
+            relay.await.unwrap();
+        })
+        .await
+        .expect("both encrypted directions must make progress");
+        assert_eq!(stats.ws_failures.load(Ordering::Relaxed), 0);
+        if raw_tcp {
+            assert_eq!(
+                stats.last_route(),
+                crate::transport::RouteKind::TelegramTcp.ui_code()
+            );
+            assert_eq!(stats.last_dc(), 203);
+        }
+        stats.stop();
+        server.await.unwrap().unwrap();
+    }
+
     #[tokio::test]
     async fn connects_through_the_documented_worker_contract() {
         // Locks the contract in docs/CLOUDFLARE_WORKER.md: a server that