fix: медиа и «прокси не настроен»: пакет на кадр, настоящие фронты, kwsN-1 только для медиа; beta.19 (#75)

* fix(tunnel): медиа и «прокси не настроен»: один пакет на кадр, настоящие фронты, kwsN-1 только для медиа

Три причины, каждая проверена на живом Telegram запросом req_DH_params:
сервер сверяет номер DC внутри него с собой и отвечает -444 на чужой.

- WebSocket-фронт разбирает один пакет на входящий кадр, остаток ждёт
  следующего. Клиент пишет запросы пачками, TGLock слал их одним кадром,
  и всё, что шло вторым, застревало: текст проходил, файлы вставали.
  Теперь поток режется на пакеты, как в Flowseal/tg-ws-proxy. Пачка через
  beta.18 висит на всех DC, через эту сборку принимается на всех.
- Кластер kwsN-1 отвергает обычное соединение с -444, а Telegram Desktop
  в ответ выключает прокси как настроенный неверно. При сбое kwsN основное
  соединение откатывалось именно туда. Теперь kwsN-1 только для медиа.
- Адреса из списка DC не говорят на TLS ни из какой сети (проверено с чистой
  сети), поэтому первый маршрут каждого DC не работал никогда. Закреплены
  настоящие фронты, и DC1, DC3, DC5 стали доступны по WebSocket, чего не было.
  У CDN203 своего фронта нет: остались TCP, Worker и CF-прокси.

Обрыв туннеля, через который уже шли данные, больше не ставит маршрут на
паузу. Таймаут соединения 8 с, как в tg-ws-proxy. probe_proxy.mjs получил
--identity и --batched: они и нашли эти баги.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* feat: журнал ошибок в диагностике, --log-file для роутеров; 2.0.0-beta.19

«Скопировать диагностику» отдавала только счётчики, и по ней нельзя было
понять, какой маршрут не ответил. Теперь в отчёте есть строки сбоев туннелей,
а домены Worker и CF-прокси в них заменены на <домен>.

--log-file и --log-max-mb (log_file, log_max_mb в настройках), как в
tg-ws-proxy: журнал идёт в файл с ротацией, а не в системный журнал роутера.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* feat(routes): запасной WebSocket на порту 80 родных адресов, включая CDN203

Родные адреса Telegram принимают WebSocket без TLS на порту 80. Измерено
req_DH_params: DC определяется адресом, имя хоста не важно, а 91.105.192.100
отдаёт ключи CDN. Для CDN203, откуда часто идёт видео, это первый прямой
WebSocket: на 443 фронта у него нет, и там, где провайдер закрыл только 443,
до него раньше было не достать без Worker.

Защита та же, что у родного TCP: внутри тот же обфусцированный MTProto. В Host
идёт адрес, имени Telegram в открытом виде нет. Маршрут разрешён только до
закреплённых адресов и пробуется последним.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix(routes): паузы маршрутов по результатам ревью, пределы и журнал

- Обрыв не наказывает маршрут, только если туннель работал и получал ответы
  15 с. Одного ответа мало: маршрут, который ответил раз и умер, оставался
  предпочтительным навсегда.
- Клиент спросил, не дождался ответа за 5 с и ушёл: маршрут считается
  неработающим. Иначе CDN, принявший TCP и молчащий, держал первое место, и
  очередь до порта 80 не доходила.
- Предел пакета 2 МиБ вместо 16: чужое устройство в LAN-режиме не заставит
  роутер копить десятки мегабайт на соединение.
- Журнал: разовая ошибка записи его больше не выключает, неудачная ротация
  не теряет бэкап и держит предел, права 0600 и на уже существующем файле.
- В отчёт диагностики не попадает Worker, заданный адресом.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: babin <Thartewerner536e@engineer.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

src/bin/cli.rs

@@ -6,12 +6,14 @@
 //! (by-sonic/tglock#10, by-sonic/tglock#17).
 
 use clap::Parser;
+use std::fs::File;
+use std::io::Write;
 use std::net::IpAddr;
-use std::path::PathBuf;
+use std::path::{Path, PathBuf};
 use std::process::ExitCode;
 use std::sync::atomic::Ordering;
-use std::sync::Arc;
-use std::time::Duration;
+use std::sync::{Arc, Mutex, OnceLock};
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
 use tglock_lib::cli_settings as config;
 use tglock_lib::{proxy, transport};
 
@@ -23,11 +25,144 @@ const STATUS_POLL: Duration = Duration::from_secs(1);
 /// пускают в `head`, закрывают терминал, перезапускают сборщик логов. Падать из
 /// за этого туннель не должен — он продолжает работать молча.
 fn say(text: &str) -> bool {
-    use std::io::Write;
+    if let Some(log) = LOG_FILE.get() {
+        // `false` здесь значило бы «писать больше некому», а это не так: диск
+        // могло временно заполнить. Журнал продолжается со следующей строки.
+        log.lock().unwrap().record(text);
+        return true;
+    }
     let mut out = std::io::stdout().lock();
     writeln!(out, "{text}").and_then(|()| out.flush()).is_ok()
 }
 
+/// Журнал вместо stdout, если задан `--log-file`.
+static LOG_FILE: OnceLock<Mutex<LogFile>> = OnceLock::new();
+
+/// Файл журнала с ротацией.
+///
+/// На роутере stdout сервиса уходит в системный журнал, и TGLock забивал его
+/// строками состояния (by-sonic/tglock#53). Заполненный файл переименовывается
+/// в `<имя>.1`, поэтому на диске лежит не больше двух файлов: flash и память
+/// роутера не резиновые.
+struct LogFile {
+    path: PathBuf,
+    /// `None` только внутри ротации: Windows не переименует открытый файл.
+    file: Option<File>,
+    written: u64,
+    limit: u64,
+    /// Об ошибке записи сообщается один раз, а не на каждой строке.
+    failing: bool,
+}
+
+impl LogFile {
+    fn open(path: PathBuf, limit: u64) -> std::io::Result<Self> {
+        let file = open_log(&path)?;
+        let written = file.metadata()?.len();
+        Ok(Self {
+            path,
+            file: Some(file),
+            written,
+            limit,
+            failing: false,
+        })
+    }
+
+    fn record(&mut self, text: &str) {
+        match self.write_line(text) {
+            Ok(()) => self.failing = false,
+            Err(error) if !self.failing => {
+                self.failing = true;
+                eprintln!(
+                    "tglock-cli: журнал {} не пишется: {error}",
+                    self.path.display()
+                );
+            }
+            Err(_) => {}
+        }
+    }
+
+    fn write_line(&mut self, text: &str) -> std::io::Result<()> {
+        let line = format!("{} {text}\n", timestamp(now_seconds()));
+        if self.written > 0 && self.written + line.len() as u64 > self.limit {
+            self.rotate()?;
+        }
+        if self.file.is_none() {
+            self.file = Some(open_log(&self.path)?);
+        }
+        let file = self.file.as_mut().ok_or(std::io::ErrorKind::NotFound)?;
+        file.write_all(line.as_bytes())?;
+        file.flush()?;
+        self.written += line.len() as u64;
+        Ok(())
+    }
+
+    fn rotate(&mut self) -> std::io::Result<()> {
+        let mut backup = self.path.clone().into_os_string();
+        backup.push(".1");
+        self.file = None;
+        // rename заменяет старый бэкап сам, и если он не удался, бэкап цел.
+        let renamed = std::fs::rename(&self.path, &backup);
+        let file = open_log(&self.path)?;
+        // Файл могли удалить или держать другой программой. Тогда предел
+        // важнее старых строк: журнал не должен съесть память роутера.
+        if renamed.is_err() {
+            // Отдельным открытием: файл для дозаписи на Windows нельзя обрезать.
+            File::options()
+                .write(true)
+                .truncate(true)
+                .open(&self.path)?;
+        }
+        self.file = Some(file);
+        self.written = 0;
+        Ok(())
+    }
+}
+
+/// В журнал попадает ссылка `tg://proxy` с секретом: читать его может только
+/// владелец, даже если файл появился раньше и с другими правами.
+fn open_log(path: &Path) -> std::io::Result<File> {
+    let mut options = std::fs::OpenOptions::new();
+    options.create(true).append(true);
+    #[cfg(unix)]
+    std::os::unix::fs::OpenOptionsExt::mode(&mut options, 0o600);
+    let file = options.open(path)?;
+    #[cfg(unix)]
+    file.set_permissions(std::os::unix::fs::PermissionsExt::from_mode(0o600))?;
+    Ok(file)
+}
+
+fn now_seconds() -> u64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .unwrap_or_default()
+        .as_secs()
+}
+
+/// `2026-10-31 09:30:00` в UTC. Перевод дней в дату — civil_from_days из
+/// https://howardhinnant.github.io/date_algorithms.html
+fn timestamp(seconds: u64) -> String {
+    let days = (seconds / 86_400) as i64 + 719_468;
+    let era = days.div_euclid(146_097);
+    let day_of_era = days.rem_euclid(146_097);
+    let year_of_era =
+        (day_of_era - day_of_era / 1_460 + day_of_era / 36_524 - day_of_era / 146_096) / 365;
+    let day_of_year = day_of_era - (365 * year_of_era + year_of_era / 4 - year_of_era / 100);
+    let shifted_month = (5 * day_of_year + 2) / 153;
+    let day = day_of_year - (153 * shifted_month + 2) / 5 + 1;
+    let month = if shifted_month < 10 {
+        shifted_month + 3
+    } else {
+        shifted_month - 9
+    };
+    let year = year_of_era + era * 400 + i64::from(month <= 2);
+    format!(
+        "{year:04}-{month:02}-{day:02} {:02}:{:02}:{:02}",
+        (seconds / 3600) % 24,
+        (seconds / 60) % 60,
+        seconds % 60
+    )
+}
+
 #[derive(Debug, Parser)]
 #[command(
     name = "tglock-cli",
@@ -72,6 +207,15 @@ struct Args {
     /// Печатать только ошибки
     #[arg(short, long)]
     quiet: bool,
+
+    /// Писать журнал в этот файл, а не в stdout. Ошибки запуска всё равно идут
+    /// в stderr. Заполненный файл переименовывается в <имя>.1
+    #[arg(long, value_name = "PATH")]
+    log_file: Option<PathBuf>,
+
+    /// Размер файла журнала в МБ, после которого он сменяется. По умолчанию 5
+    #[arg(long, value_name = "MB")]
+    log_max_mb: Option<u32>,
 }
 
 impl Args {
@@ -85,6 +229,8 @@ impl Args {
             cf_proxy: self.cf_proxy.clone(),
             secret_file: self.secret_file.clone(),
             quiet: self.quiet,
+            log_file: self.log_file.clone(),
+            log_max_mb: self.log_max_mb,
         }
     }
 
@@ -129,6 +275,11 @@ async fn serve(args: Args) -> Result<(), String> {
     let listen = settings.listen;
     let stats = settings.stats();
     let quiet = settings.quiet;
+    if let Some(path) = &settings.log_file {
+        let log = LogFile::open(path.clone(), u64::from(settings.log_max_mb) * 1024 * 1024)
+            .map_err(|error| format!("не удалось открыть журнал {}: {error}", path.display()))?;
+        let _ = LOG_FILE.set(Mutex::new(log));
+    }
 
     // Bind before printing anything: a busy port must be an error, not a
     // daemon that reports success and silently does nothing.
@@ -291,6 +442,114 @@ mod tests {
         config::resolve(config::FileConfig::default(), parse(args).overrides()).unwrap()
     }
 
+    /// Пустой каталог для одного теста, удаляется вместе с ним.
+    struct TempDir(PathBuf);
+
+    impl TempDir {
+        fn new(name: &str) -> Self {
+            let path = std::env::temp_dir().join(format!(
+                "tglock-cli-{name}-{}-{:?}",
+                std::process::id(),
+                std::thread::current().id()
+            ));
+            let _ = std::fs::remove_dir_all(&path);
+            std::fs::create_dir_all(&path).unwrap();
+            Self(path)
+        }
+    }
+
+    impl Drop for TempDir {
+        fn drop(&mut self) {
+            let _ = std::fs::remove_dir_all(&self.0);
+        }
+    }
+
+    #[test]
+    fn log_rotation_keeps_the_newest_lines_in_two_bounded_files() {
+        let dir = TempDir::new("rotation");
+        let path = dir.0.join("tglock.log");
+        let mut log = LogFile::open(path.clone(), 200).unwrap();
+        for line in 0..40 {
+            log.write_line(&format!("line {line}")).unwrap();
+        }
+        let current = std::fs::read_to_string(&path).unwrap();
+        let backup = std::fs::read_to_string(dir.0.join("tglock.log.1")).unwrap();
+        assert!(current.contains("line 39"), "{current}");
+        assert!(current.len() <= 200 && backup.len() <= 200);
+        assert_eq!(std::fs::read_dir(&dir.0).unwrap().count(), 2);
+    }
+
+    #[test]
+    fn a_log_deleted_by_hand_is_recreated_and_stays_bounded() {
+        let dir = TempDir::new("deleted");
+        let path = dir.0.join("tglock.log");
+        let mut log = LogFile::open(path.clone(), 200).unwrap();
+        log.write_line("before").unwrap();
+        // На Windows открытый файл не удалить, поэтому он закрывается как при
+        // ротации. На unix запись шла бы в удалённый файл до предела.
+        log.file = None;
+        std::fs::remove_file(&path).unwrap();
+        for line in 0..40 {
+            log.write_line(&format!("line {line}")).unwrap();
+        }
+        let current = std::fs::read_to_string(&path).unwrap();
+        assert!(current.contains("line 39"), "{current}");
+        assert!(current.len() <= 200);
+    }
+
+    #[test]
+    fn a_failed_rotation_keeps_the_old_backup_and_the_size_limit() {
+        let dir = TempDir::new("failed-rotation");
+        let path = dir.0.join("tglock.log");
+        let mut log = LogFile::open(path.clone(), 200).unwrap();
+        std::fs::write(dir.0.join("tglock.log.1"), "old backup").unwrap();
+        log.write_line("x").unwrap();
+        // Файл исчез до ротации: переименовывать нечего.
+        log.file = None;
+        std::fs::remove_file(&path).unwrap();
+        log.rotate().unwrap();
+        assert_eq!(
+            std::fs::read_to_string(dir.0.join("tglock.log.1")).unwrap(),
+            "old backup"
+        );
+        assert_eq!(log.written, 0);
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn an_existing_log_with_wide_permissions_is_tightened() {
+        use std::os::unix::fs::PermissionsExt;
+        let dir = TempDir::new("permissions");
+        let path = dir.0.join("tglock.log");
+        std::fs::write(&path, "").unwrap();
+        std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o644)).unwrap();
+        LogFile::open(path.clone(), 200).unwrap();
+        let mode = std::fs::metadata(&path).unwrap().permissions().mode();
+        assert_eq!(mode & 0o777, 0o600);
+    }
+
+    #[test]
+    fn a_restarted_daemon_continues_the_log_and_counts_what_is_there() {
+        let dir = TempDir::new("restart");
+        let path = dir.0.join("tglock.log");
+        LogFile::open(path.clone(), 10_000)
+            .unwrap()
+            .write_line("first run")
+            .unwrap();
+        let mut log = LogFile::open(path.clone(), 10_000).unwrap();
+        assert!(log.written > 0);
+        log.write_line("second run").unwrap();
+        let text = std::fs::read_to_string(&path).unwrap();
+        assert!(text.contains("first run") && text.contains("second run"));
+    }
+
+    #[test]
+    fn timestamps_are_calendar_dates_in_utc() {
+        assert_eq!(timestamp(0), "1970-01-01 00:00:00");
+        assert_eq!(timestamp(951_782_400), "2000-02-29 00:00:00");
+        assert_eq!(timestamp(1_793_442_645), "2026-10-31 10:30:45");
+    }
+
     #[test]
     fn command_definition_is_valid() {
         Args::command().debug_assert();

src/cli_settings.rs

@@ -20,6 +20,9 @@ use std::path::{Path, PathBuf};
 /// Имя файла, который ищется рядом с бинарём, если `--config` не задан.
 pub const DEFAULT_FILE_NAME: &str = "tglock.toml";
 
+/// Размер журнала, после которого он сменяется. Как в Flowseal/tg-ws-proxy.
+pub const DEFAULT_LOG_MAX_MB: u32 = 5;
+
 /// Содержимое файла конфигурации.
 ///
 /// `deny_unknown_fields` намеренно: опечатка вроде `porrt = 1080` должна быть
@@ -38,6 +41,9 @@ pub struct FileConfig {
     pub secret: Option<String>,
     pub secret_file: Option<PathBuf>,
     pub quiet: Option<bool>,
+    /// Писать журнал в файл, а не в stdout (by-sonic/tglock#53).
+    pub log_file: Option<PathBuf>,
+    pub log_max_mb: Option<u32>,
 }
 
 impl FileConfig {
@@ -63,6 +69,8 @@ pub struct Overrides {
     pub cf_proxy: Vec<String>,
     pub secret_file: Option<PathBuf>,
     pub quiet: bool,
+    pub log_file: Option<PathBuf>,
+    pub log_max_mb: Option<u32>,
 }
 
 /// Откуда взять секрет прокси.
@@ -86,6 +94,8 @@ pub struct Resolved {
     pub cf_proxies: String,
     pub secret: SecretSource,
     pub quiet: bool,
+    pub log_file: Option<PathBuf>,
+    pub log_max_mb: u32,
 }
 
 impl Resolved {
@@ -139,6 +149,14 @@ pub fn resolve(file: FileConfig, cli: Overrides) -> Result<Resolved, String> {
         cli.cf_proxy
     };
 
+    let log_max_mb = cli
+        .log_max_mb
+        .or(file.log_max_mb)
+        .unwrap_or(DEFAULT_LOG_MAX_MB);
+    if log_max_mb == 0 {
+        return Err("log_max_mb должен быть не меньше 1".to_owned());
+    }
+
     // Секрет из файла конфигурации важнее отдельного файла: если человек вписал
     // его сюда, значит хотел держать всё в одном месте.
     let secret = match (&file.secret, cli.secret_file.or(file.secret_file)) {
@@ -156,6 +174,8 @@ pub fn resolve(file: FileConfig, cli: Overrides) -> Result<Resolved, String> {
         cf_proxies: cf_proxies.join(","),
         secret,
         quiet: cli.quiet || file.quiet.unwrap_or(false),
+        log_file: cli.log_file.or(file.log_file),
+        log_max_mb,
     })
 }
 
@@ -189,6 +209,31 @@ mod tests {
         assert!(!resolved.quiet);
     }
 
+    #[test]
+    fn log_file_comes_from_the_file_and_a_flag_overrides_it() {
+        let from_file =
+            || FileConfig::parse("log_file = \"/tmp/tglock.log\"\nlog_max_mb = 2").unwrap();
+        let resolved = resolve(from_file(), cli()).unwrap();
+        assert_eq!(resolved.log_file, Some(PathBuf::from("/tmp/tglock.log")));
+        assert_eq!(resolved.log_max_mb, 2);
+
+        let flags = Overrides {
+            log_file: Some(PathBuf::from("/var/log/t.log")),
+            log_max_mb: Some(1),
+            ..cli()
+        };
+        let resolved = resolve(from_file(), flags).unwrap();
+        assert_eq!(resolved.log_file, Some(PathBuf::from("/var/log/t.log")));
+        assert_eq!(resolved.log_max_mb, 1);
+    }
+
+    #[test]
+    fn a_zero_log_limit_is_refused() {
+        // Иначе файл менялся бы на каждой строке.
+        let file = FileConfig::parse("log_max_mb = 0").unwrap();
+        assert!(resolve(file, cli()).unwrap_err().contains("log_max_mb"));
+    }
+
     #[test]
     fn file_values_are_used_when_no_flags_are_given() {
         let file = FileConfig::parse(

src/mtproto.rs

@@ -354,6 +354,138 @@ fn build_crypto_context(
     })
 }
 
+/// Длина, больше которой пакет считается не настоящим. Файлы грузятся частями
+/// по 512 КиБ. Предел держит и память: в LAN-режиме чужое устройство могло бы
+/// заставить копить «пакет» на роутере. Если настоящий пакет окажется длиннее,
+/// поток просто пойдёт дальше без нарезки.
+const MAX_PACKET: usize = 2 * 1024 * 1024;
+
+#[derive(Clone, Copy)]
+enum Framing {
+    Abridged,
+    Intermediate,
+}
+
+/// Режет поток клиента на транспортные пакеты MTProto.
+///
+/// WebSocket-фронт Telegram разбирает один пакет на входящий кадр, а остаток
+/// кадра держит до следующего. Клиент пишет запросы пачками, и всё, что шло
+/// вторым в той же записи TCP, застревало на фронте: текст проходил, а файлы
+/// вставали. Проверено на живом фронте: `req_DH_params` вторым пакетом в кадре
+/// остаётся без ответа, пока не придёт следующий кадр, а отдельным кадром
+/// принимается сразу. Так же режет поток Flowseal/tg-ws-proxy.
+///
+/// Заголовок читается по копии: ключи `init` открыты, поэтому режим с секретом
+/// и SOCKS5 разбираются одинаково, а наружу уходят исходные байты.
+pub struct PacketSplitter {
+    framing: Option<Framing>,
+    decrypt: AesCtr,
+    wire: Vec<u8>,
+    plain: Vec<u8>,
+}
+
+impl PacketSplitter {
+    /// Разбирать поток, начатый этим `init`. Неизвестный протокол не режется.
+    pub fn from_init(init: &[u8; INIT_LEN]) -> Self {
+        let key: [u8; 32] = init[KEY_START..KEY_END].try_into().unwrap();
+        let iv: [u8; 16] = init[KEY_END..IV_END].try_into().unwrap();
+        let mut decrypt = AesCtr::new((&key).into(), (&iv).into());
+        let mut plain_init = *init;
+        decrypt.apply_keystream(&mut plain_init);
+        let framing = match [
+            plain_init[TAG_START],
+            plain_init[TAG_START + 1],
+            plain_init[TAG_START + 2],
+            plain_init[TAG_START + 3],
+        ] {
+            ABRIDGED => Some(Framing::Abridged),
+            INTERMEDIATE | PADDED_INTERMEDIATE => Some(Framing::Intermediate),
+            _ => None,
+        };
+        Self {
+            framing,
+            decrypt,
+            wire: Vec::new(),
+            plain: Vec::new(),
+        }
+    }
+
+    /// Принять зашифрованные байты, вернуть готовые к отправке пакеты.
+    pub fn push(&mut self, wire: &[u8]) -> Vec<Vec<u8>> {
+        let Some(framing) = self.framing else {
+            return vec![wire.to_vec()];
+        };
+        let start = self.plain.len();
+        self.wire.extend_from_slice(wire);
+        self.plain.extend_from_slice(wire);
+        self.decrypt.apply_keystream(&mut self.plain[start..]);
+
+        let mut packets = Vec::new();
+        let mut offset = 0;
+        while offset < self.plain.len() {
+            match packet_len(framing, &self.plain[offset..]) {
+                PacketLen::Incomplete => break,
+                PacketLen::Complete(length) => {
+                    if self.plain.len() - offset < length {
+                        break;
+                    }
+                    packets.push(self.wire[offset..offset + length].to_vec());
+                    offset += length;
+                }
+                // Длина не похожа на настоящую: поток не наш. Дальше он идёт
+                // как есть, иначе данные встанут в ожидании несуществующего конца.
+                PacketLen::Invalid => {
+                    packets.push(self.wire[offset..].to_vec());
+                    offset = self.plain.len();
+                    self.framing = None;
+                }
+            }
+        }
+        self.wire.drain(..offset);
+        self.plain.drain(..offset);
+        packets
+    }
+
+    /// Хвост незаконченного пакета: клиент закрыл соединение на середине.
+    pub fn finish(&mut self) -> Option<Vec<u8>> {
+        self.plain.clear();
+        (!self.wire.is_empty()).then(|| std::mem::take(&mut self.wire))
+    }
+}
+
+enum PacketLen {
+    Incomplete,
+    Complete(usize),
+    Invalid,
+}
+
+fn packet_len(framing: Framing, plain: &[u8]) -> PacketLen {
+    let (header, payload) = match framing {
+        Framing::Abridged => match plain {
+            [] => return PacketLen::Incomplete,
+            [0x7f | 0xff, length @ ..] => match length {
+                [a, b, c, ..] => (4, u32::from_le_bytes([*a, *b, *c, 0]) as usize * 4),
+                _ => return PacketLen::Incomplete,
+            },
+            [first, ..] => (1, (*first & 0x7f) as usize * 4),
+        },
+        Framing::Intermediate => match plain {
+            [a, b, c, d, ..] => {
+                // Старший бит просит quick ack и к длине не относится.
+                (
+                    4,
+                    (u32::from_le_bytes([*a, *b, *c, *d]) & 0x7fff_ffff) as usize,
+                )
+            }
+            _ => return PacketLen::Incomplete,
+        },
+    };
+    if payload == 0 || payload > MAX_PACKET {
+        return PacketLen::Invalid;
+    }
+    PacketLen::Complete(header + payload)
+}
+
 #[cfg(test)]
 pub(crate) fn test_client_init(secret: &[u8; 16], dc_index: i16) -> [u8; INIT_LEN] {
     tests::generate_client_init(secret, PADDED_INTERMEDIATE, dc_index)
@@ -423,6 +555,118 @@ pub(crate) fn test_relay_peer(relay_init: &[u8; INIT_LEN]) -> TestPeer {
 mod tests {
     use super::*;
 
+    /// Поток, каким его увидел бы фронт Telegram, и разделитель для него.
+    fn splitter_for(tag: [u8; 4], plain: &[u8]) -> (PacketSplitter, Vec<u8>) {
+        let init = generate_relay_init(tag, 2);
+        let mut wire = plain.to_vec();
+        test_relay_peer(&init).decrypt(&mut wire);
+        (PacketSplitter::from_init(&init), wire)
+    }
+
+    fn intermediate(payload: &[u8]) -> Vec<u8> {
+        let mut packet = (payload.len() as u32).to_le_bytes().to_vec();
+        packet.extend_from_slice(payload);
+        packet
+    }
+
+    #[test]
+    fn glued_packets_leave_one_per_frame() {
+        let first = intermediate(&[1; 40]);
+        let second = intermediate(&[2; 340]);
+        let plain = [first.clone(), second.clone()].concat();
+        for tag in [INTERMEDIATE, PADDED_INTERMEDIATE] {
+            let (mut splitter, wire) = splitter_for(tag, &plain);
+            let packets = splitter.push(&wire);
+            assert_eq!(packets.len(), 2);
+            assert_eq!(packets[0], wire[..first.len()]);
+            assert_eq!(packets[1], wire[first.len()..]);
+            assert_eq!(splitter.finish(), None);
+        }
+    }
+
+    #[test]
+    fn a_packet_is_held_until_it_is_complete() {
+        let packet = intermediate(&[7; 300]);
+        let (mut splitter, wire) = splitter_for(INTERMEDIATE, &packet);
+        // Границы TCP не совпадают с пакетами: режем в том числе внутри длины.
+        assert!(splitter.push(&wire[..2]).is_empty());
+        assert!(splitter.push(&wire[2..150]).is_empty());
+        assert_eq!(splitter.push(&wire[150..]), vec![wire.clone()]);
+    }
+
+    #[test]
+    fn a_tail_of_one_packet_and_the_start_of_the_next_stay_separate() {
+        let first = intermediate(&[1; 40]);
+        let second = intermediate(&[2; 80]);
+        let plain = [first.clone(), second.clone()].concat();
+        let (mut splitter, wire) = splitter_for(INTERMEDIATE, &plain);
+        let packets = splitter.push(&wire[..first.len() + 10]);
+        assert_eq!(packets, vec![wire[..first.len()].to_vec()]);
+        let packets = splitter.push(&wire[first.len() + 10..]);
+        assert_eq!(packets, vec![wire[first.len()..].to_vec()]);
+    }
+
+    #[test]
+    fn quick_ack_bit_is_not_part_of_the_length() {
+        let mut packet = intermediate(&[3; 16]);
+        packet[3] |= 0x80;
+        let (mut splitter, wire) = splitter_for(INTERMEDIATE, &packet);
+        assert_eq!(splitter.push(&wire), vec![wire.clone()]);
+    }
+
+    #[test]
+    fn abridged_lengths_count_in_words_with_either_header() {
+        let short = [vec![2], vec![9; 8]].concat();
+        let long = [vec![0x7f, 0x80, 0, 0], vec![9; 512]].concat();
+        let quick_ack = [vec![0x81], vec![9; 4]].concat();
+        let plain = [short.clone(), long.clone(), quick_ack.clone()].concat();
+        let (mut splitter, wire) = splitter_for(ABRIDGED, &plain);
+        let lengths: Vec<_> = splitter.push(&wire).iter().map(Vec::len).collect();
+        assert_eq!(lengths, [short.len(), long.len(), quick_ack.len()]);
+    }
+
+    #[test]
+    fn a_length_that_cannot_be_real_stops_the_splitting_without_losing_bytes() {
+        // Если длина мусорная, ждать конца пакета значит заморозить соединение.
+        let mut garbage = 0x7fff_fff0_u32.to_le_bytes().to_vec();
+        garbage.extend_from_slice(&[5; 32]);
+        let (mut splitter, wire) = splitter_for(INTERMEDIATE, &garbage);
+        assert_eq!(splitter.push(&wire), vec![wire.clone()]);
+        assert_eq!(splitter.push(b"more"), vec![b"more".to_vec()]);
+    }
+
+    #[test]
+    fn a_length_over_the_limit_is_never_buffered() {
+        let mut oversized = ((MAX_PACKET + 1) as u32).to_le_bytes().to_vec();
+        oversized.extend_from_slice(&[5; 32]);
+        let (mut splitter, wire) = splitter_for(INTERMEDIATE, &oversized);
+        assert_eq!(splitter.push(&wire), vec![wire.clone()]);
+        assert_eq!(splitter.finish(), None);
+    }
+
+    #[test]
+    fn the_largest_real_upload_part_still_arrives_whole() {
+        let packet = intermediate(&vec![7; 512 * 1024 + 256]);
+        let (mut splitter, wire) = splitter_for(INTERMEDIATE, &packet);
+        assert!(splitter.push(&wire[..65536]).is_empty());
+        assert_eq!(splitter.push(&wire[65536..]), vec![wire.clone()]);
+    }
+
+    #[test]
+    fn an_unknown_protocol_passes_through_untouched() {
+        let (mut splitter, wire) = splitter_for([0x11; 4], &[0; 16]);
+        assert_eq!(splitter.push(&wire), vec![wire.clone()]);
+    }
+
+    #[test]
+    fn the_unfinished_tail_is_returned_once() {
+        let packet = intermediate(&[1; 40]);
+        let (mut splitter, wire) = splitter_for(INTERMEDIATE, &packet);
+        assert!(splitter.push(&wire[..10]).is_empty());
+        assert_eq!(splitter.finish(), Some(wire[..10].to_vec()));
+        assert_eq!(splitter.finish(), None);
+    }
+
     pub(super) fn generate_client_init(
         secret: &[u8; 16],
         protocol_tag: [u8; 4],

src/proxy.rs

@@ -772,6 +772,7 @@ async fn ws_tunnel(
     ws_w.send(tungstenite::Message::Binary(init.to_vec()))
         .await
         .map_err(|error| tunnel_route_error(stats, &connected, dc, &error.to_string()))?;
+    let health = TunnelHealth::new();
 
     // Ping приходит в половину, которая читает, а отвечать на него должна та,
     // которая пишет: владелец у отправляющей половины строго один.
@@ -785,11 +786,12 @@ 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 !data.is_empty() {
+                        health.note_reply();
+                    }
                     if let Some(crypto) = &mut crypto {
                         crypto.apply(data.as_mut());
                     }
@@ -813,7 +815,7 @@ async fn ws_tunnel(
                             .is_ok_and(|code| code != 0)
                             || frame.code != tungstenite::protocol::frame::coding::CloseCode::Normal
                     });
-                    if !received_payload || transport_error {
+                    if !health.replied() || transport_error {
                         let reason = match frame {
                             Some(frame) => describe_ws_close(frame.code.into(), &frame.reason),
                             None => "WebSocket закрыт без ответа Telegram".to_owned(),
@@ -823,17 +825,18 @@ async fn ws_tunnel(
                     return Ok(());
                 }
                 Err(error) => {
-                    return Err(tunnel_route_error(
+                    return Err(tunnel_broke(
                         stats,
                         &connected,
                         dc,
                         &error.to_string(),
+                        &health,
                     ));
                 }
                 Ok(_) => {}
             }
         }
-        if !received_payload {
+        if !health.replied() {
             return Err(tunnel_route_error(
                 stats,
                 &connected,
@@ -846,32 +849,54 @@ async fn ws_tunnel(
 
     let upstream = async {
         let mut crypto = upstream_crypto;
+        // Worker пишет кадры в обычный TCP до Telegram, там они ничего не значат.
+        // Отдельный кадр на пакет нужен только WebSocket Telegram, на 443 и на 80.
+        let mut splitter = (connected.route.kind != crate::transport::RouteKind::CloudflareWorker)
+            .then(|| crate::mtproto::PacketSplitter::from_init(init));
         let mut buf = vec![0u8; 65536];
         loop {
             tokio::select! {
                 read = tcp_r.read(&mut buf) => match read {
-                    Ok(0) => break,
+                    Ok(0) => {
+                        if let Some(tail) = splitter.as_mut().and_then(|splitter| splitter.finish()) {
+                            let _ = ws_w.send(tungstenite::Message::Binary(tail)).await;
+                        }
+                        break;
+                    }
                     Err(error) => return Err(error.into()),
                     Ok(read) => {
+                        health.note_request();
                         if let Some(crypto) = &mut crypto {
                             crypto.apply(&mut buf[..read]);
                         }
-                        ws_w
-                            .send(tungstenite::Message::Binary(buf[..read].to_vec()))
-                            .await
-                            .map_err(|error| tunnel_route_error(stats, &connected, dc, &error.to_string()))?;
+                        let packets = match &mut splitter {
+                            Some(splitter) => splitter.push(&buf[..read]),
+                            None => vec![buf[..read].to_vec()],
+                        };
+                        for packet in packets {
+                            ws_w.feed(tungstenite::Message::Binary(packet)).await.map_err(|error| {
+                                tunnel_broke(stats, &connected, dc, &error.to_string(), &health)
+                            })?;
+                        }
+                        ws_w.flush().await.map_err(|error| {
+                            tunnel_broke(stats, &connected, dc, &error.to_string(), &health)
+                        })?;
                     }
                 },
                 payload = pong_rx.recv() => match payload {
                     Some(payload) => {
-                        ws_w.send(tungstenite::Message::Pong(payload)).await
-                            .map_err(|error| tunnel_route_error(stats, &connected, dc, &error.to_string()))?;
+                        ws_w.send(tungstenite::Message::Pong(payload)).await.map_err(|error| {
+                            tunnel_broke(stats, &connected, dc, &error.to_string(), &health)
+                        })?;
                     }
                     None => break,
                 },
             }
         }
         let _ = ws_w.close().await;
+        if health.went_silent() {
+            return Err(tunnel_route_error(stats, &connected, dc, SILENT_ROUTE));
+        }
         Ok::<(), Box<dyn std::error::Error + Send + Sync>>(())
     };
 
@@ -905,34 +930,38 @@ async fn cdn_tcp_tunnel(
         Some((upstream, downstream)) => (Some(upstream), Some(downstream)),
         None => (None, None),
     };
+    let health = TunnelHealth::new();
     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 {
+                if health.went_silent() {
+                    return Err(tunnel_route_error(stats, connected, dc, SILENT_ROUTE));
+                }
                 return Ok::<_, Box<dyn std::error::Error + Send + Sync>>(());
             }
+            health.note_request();
             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()))?;
+                .map_err(|error| tunnel_broke(stats, connected, dc, &error.to_string(), &health))?;
         }
     };
     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()))?;
+                .map_err(|error| tunnel_broke(stats, connected, dc, &error.to_string(), &health))?;
             if read == 0 {
-                return if received_payload {
+                return if health.replied() {
                     Ok(())
                 } else {
                     Err(tunnel_route_error(
@@ -943,7 +972,7 @@ async fn cdn_tcp_tunnel(
                     ))
                 };
             }
-            received_payload = true;
+            health.note_reply();
             if let Some(crypto) = &mut crypto {
                 crypto.apply(&mut buffer[..read]);
             }
@@ -989,6 +1018,92 @@ fn tunnel_route_error(
     message.into()
 }
 
+const SILENT_ROUTE: &str = "Telegram не ответил на запрос, клиент ушёл сам";
+
+/// Сколько туннель должен проработать, чтобы его обрыв не считался поломкой
+/// маршрута. Один ответ ничего не доказывает: DPI режет и после первых байт.
+const PROVEN_TUNNEL: Duration = if cfg!(test) {
+    Duration::from_millis(500)
+} else {
+    Duration::from_secs(15)
+};
+
+/// Сколько Telegram может молчать в ответ на запрос. Он отвечает за доли секунды.
+const SILENT_TUNNEL: Duration = if cfg!(test) {
+    Duration::from_millis(500)
+} else {
+    Duration::from_secs(5)
+};
+
+/// Что туннель показал о своём маршруте.
+struct TunnelHealth {
+    opened: std::time::Instant,
+    requested: AtomicBool,
+    replied: AtomicBool,
+}
+
+impl TunnelHealth {
+    fn new() -> Self {
+        Self {
+            opened: std::time::Instant::now(),
+            requested: AtomicBool::new(false),
+            replied: AtomicBool::new(false),
+        }
+    }
+
+    fn note_request(&self) {
+        self.requested.store(true, Ordering::Relaxed);
+    }
+
+    fn note_reply(&self) {
+        self.replied.store(true, Ordering::Relaxed);
+    }
+
+    fn replied(&self) -> bool {
+        self.replied.load(Ordering::Relaxed)
+    }
+
+    /// Маршрут работал по-настоящему: Telegram отвечал, и не один миг.
+    fn proven(&self) -> bool {
+        self.replied() && self.opened.elapsed() >= PROVEN_TUNNEL
+    }
+
+    /// Клиент спросил, Telegram промолчал, и клиент сдался.
+    ///
+    /// Раньше такой маршрут оставался предпочтительным: TCP открылся, значит
+    /// успех, а отказа никто не зафиксировал. Запасные маршруты до очереди так
+    /// и не доходили. Быстрый уход не считается: клиент сам закрывает лишние
+    /// соединения из своей гонки.
+    fn went_silent(&self) -> bool {
+        self.requested.load(Ordering::Relaxed)
+            && !self.replied()
+            && self.opened.elapsed() >= SILENT_TUNNEL
+    }
+}
+
+/// Соединение оборвалось на полпути.
+///
+/// Обрыв долгоживущего туннеля обычен, и пауза на рабочем маршруте оставила бы
+/// весь дата-центр без связи. Клиент сам откроет новое соединение. Маршрут
+/// наказывается, если он не успел доказать, что работает.
+fn tunnel_broke(
+    stats: &Stats,
+    connected: &crate::transport::ConnectedRoute,
+    dc: u16,
+    reason: &str,
+    health: &TunnelHealth,
+) -> Box<dyn std::error::Error + Send + Sync> {
+    if !health.proven() {
+        return tunnel_route_error(stats, connected, dc, reason);
+    }
+    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);
@@ -1726,6 +1841,225 @@ mod tests {
         }
     }
 
+    /// Точка фронта Telegram: принять WebSocket и вернуть его вместе с `init`.
+    #[allow(clippy::result_large_err)]
+    async fn accept_relay(
+        listener: TcpListener,
+    ) -> (
+        tokio_tungstenite::WebSocketStream<TcpStream>,
+        [u8; INIT_LEN],
+    ) {
+        use futures_util::StreamExt;
+        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();
+        let init = match ws.next().await.unwrap().unwrap() {
+            Message::Binary(init) => init,
+            other => panic!("expected init: {other:?}"),
+        };
+        (ws, init.try_into().unwrap())
+    }
+
+    async fn check_packets_per_frame(kind: crate::transport::RouteKind, expected_frames: usize) {
+        use futures_util::StreamExt;
+        let first = [&44_u32.to_le_bytes()[..], &[1; 44]].concat();
+        let second = [&300_u32.to_le_bytes()[..], &[2; 300]].concat();
+        let request = [first.clone(), second.clone()].concat();
+        let total = request.len();
+
+        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 ws, init) = accept_relay(listener).await;
+            let mut peer = crate::mtproto::test_relay_peer(&init);
+            let mut frames = Vec::new();
+            let mut received = 0;
+            while received < total {
+                if let Message::Binary(mut data) = ws.next().await.unwrap().unwrap() {
+                    peer.decrypt(&mut data);
+                    received += data.len();
+                    frames.push(data);
+                }
+            }
+            frames
+        });
+
+        let stats = Stats::new();
+        stats
+            .transport
+            .force_local_route_with(relay_port, kind, "/apiws".to_owned());
+        let (port, server) = start_proxy(stats.clone(), false).await;
+        let init = unambiguous_client_init(&stats.secret, 2);
+        let mut peer = crate::mtproto::test_client_peer(&init, &stats.secret);
+        let mut client = TcpStream::connect(("127.0.0.1", port)).await.unwrap();
+        client.write_all(&init).await.unwrap();
+        // Одна запись TCP: так клиент отправляет msgs_ack вместе с запросом.
+        let mut wire = request.clone();
+        peer.encrypt(&mut wire);
+        client.write_all(&wire).await.unwrap();
+
+        let frames = tokio::time::timeout(Duration::from_secs(5), relay)
+            .await
+            .expect("every packet must reach the relay")
+            .unwrap();
+        assert_eq!(frames.len(), expected_frames);
+        assert_eq!(frames.concat(), request);
+        if expected_frames == 2 {
+            assert_eq!(frames, vec![first, second]);
+        }
+        stats.stop();
+        drop(client);
+        server.await.unwrap().unwrap();
+    }
+
+    #[tokio::test]
+    async fn websocket_frontend_gets_one_packet_per_frame() {
+        // Фронт разбирает один пакет на кадр, остальное застревает до следующего.
+        check_packets_per_frame(crate::transport::RouteKind::TelegramIp, 2).await;
+        check_packets_per_frame(crate::transport::RouteKind::TelegramPlainWs, 2).await;
+    }
+
+    #[tokio::test]
+    async fn cloudflare_worker_needs_no_splitting() {
+        check_packets_per_frame(crate::transport::RouteKind::CloudflareWorker, 1).await;
+    }
+
+    /// Туннель, который ответил и через `lifetime` оборвался без закрывающего
+    /// кадра: так выглядит «Connection reset without closing handshake».
+    async fn break_tunnel_after(lifetime: Duration) -> Arc<Stats> {
+        use futures_util::SinkExt;
+        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 ws, _) = accept_relay(listener).await;
+            ws.send(Message::Binary(vec![1, 2, 3, 4])).await.unwrap();
+            tokio::time::sleep(lifetime).await;
+            drop(ws);
+        });
+        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 reply = Vec::new();
+        tokio::time::timeout(Duration::from_secs(5), client.read_to_end(&mut reply))
+            .await
+            .unwrap()
+            .unwrap();
+        relay.await.unwrap();
+        wait_until("connection task finished", || {
+            stats.active.load(Ordering::Relaxed) == 0
+        })
+        .await;
+        stats.stop();
+        server.await.unwrap().unwrap();
+        stats
+    }
+
+    #[tokio::test]
+    async fn a_tunnel_that_proved_itself_does_not_penalize_its_route_when_it_breaks() {
+        // Иначе один обрыв рабочего туннеля ставил маршрут на паузу, и DC
+        // оставался без связи.
+        let stats = break_tunnel_after(PROVEN_TUNNEL + Duration::from_millis(100)).await;
+        assert_eq!(stats.transport.route_failures(), 0);
+        assert_eq!(stats.ws_failures.load(Ordering::Relaxed), 1);
+        assert!(stats
+            .drain_events()
+            .iter()
+            .any(|event| event.contains("Обрыв туннеля DC2")));
+    }
+
+    #[tokio::test]
+    async fn a_route_that_answers_once_and_dies_is_still_penalized() {
+        // Иначе он навсегда оставался бы предпочтительным: DPI режет и после
+        // первых байт, и запасные маршруты до очереди не доходили.
+        let stats = break_tunnel_after(Duration::ZERO).await;
+        assert_eq!(stats.transport.route_failures(), 1);
+        assert!(stats
+            .drain_events()
+            .iter()
+            .any(|event| event.contains("Сбой туннеля DC2")));
+    }
+
+    /// Telegram молчит, клиент спрашивает и уходит через `patience`.
+    async fn leave_silent_route_after(patience: Duration, raw_tcp: bool) -> Arc<Stats> {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let relay_port = listener.local_addr().unwrap().port();
+        let (done_tx, done_rx) = tokio::sync::oneshot::channel::<()>();
+        let relay = tokio::spawn(async move {
+            // Соединение держится открытым и без ответа, пока клиент не уйдёт.
+            if raw_tcp {
+                let (stream, _) = listener.accept().await.unwrap();
+                let _ = done_rx.await;
+                drop(stream);
+            } else {
+                let (ws, _) = accept_relay(listener).await;
+                let _ = done_rx.await;
+                drop(ws);
+            }
+        });
+        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 mut client = TcpStream::connect(("127.0.0.1", port)).await.unwrap();
+        let init = unambiguous_client_init(&stats.secret, if raw_tcp { 203 } else { 2 });
+        client.write_all(&init).await.unwrap();
+        client.write_all(&[0; 16]).await.unwrap();
+        tokio::time::sleep(patience).await;
+        drop(client);
+        wait_until("connection task finished", || {
+            stats.active.load(Ordering::Relaxed) == 0
+        })
+        .await;
+        let _ = done_tx.send(());
+        relay.await.unwrap();
+        stats.stop();
+        server.await.unwrap().unwrap();
+        stats
+    }
+
+    #[tokio::test]
+    async fn a_silent_route_is_penalized_when_the_client_gives_up() {
+        // Иначе CDN, который принял TCP и молчит, оставался первым в списке, и
+        // запасной маршрут через порт 80 не получал ни одного шанса.
+        for raw_tcp in [true, false] {
+            let wait = SILENT_TUNNEL + Duration::from_millis(100);
+            let stats = leave_silent_route_after(wait, raw_tcp).await;
+            assert_eq!(stats.transport.route_failures(), 1, "raw_tcp {raw_tcp}");
+            assert!(stats
+                .drain_events()
+                .iter()
+                .any(|event| event.contains(SILENT_ROUTE)));
+        }
+    }
+
+    #[tokio::test]
+    async fn a_client_that_leaves_at_once_does_not_penalize_the_route() {
+        // Клиент сам закрывает лишние соединения из своей гонки.
+        for raw_tcp in [true, false] {
+            let stats = leave_silent_route_after(Duration::ZERO, raw_tcp).await;
+            assert_eq!(stats.transport.route_failures(), 0, "raw_tcp {raw_tcp}");
+        }
+    }
+
     #[tokio::test]
     async fn fragmented_encrypted_upload_and_download_preserve_both_streams() {
         check_fragmented_encrypted_duplex(false).await;

src/transport.rs

@@ -9,12 +9,15 @@ use tokio::time::Instant;
 use tokio_tungstenite::tungstenite::client::IntoClientRequest;
 use tokio_tungstenite::{MaybeTlsStream, WebSocketStream};
 
-const CONNECT_TIMEOUT: Duration = Duration::from_secs(4);
+// Живой адрес под нагрузкой не всегда укладывался в 4 с, а упавший маршрут уходит
+// на паузу. Столько же ждёт Flowseal/tg-ws-proxy.
+const CONNECT_TIMEOUT: Duration = Duration::from_secs(8);
 const FALLBACK_DELAY: Duration = Duration::from_millis(250);
 const MAX_CONNECTING: usize = 3;
 const FAILURE_BACKOFF_INITIAL: Duration = Duration::from_secs(30);
 const FAILURE_BACKOFF_MAX: Duration = Duration::from_secs(30 * 60);
 const HTTPS_PORT: u16 = 443;
+const HTTP_PORT: u16 = 80;
 const CDN203_IP: &str = "91.105.192.100";
 
 pub type TelegramWebSocket = WebSocketStream<MaybeTlsStream<TcpStream>>;
@@ -37,6 +40,9 @@ pub enum RouteKind {
     /// и Cloudflare сам доводит WebSocket до дата-центра. Worker и его лимиты
     /// не нужны — так работает CF-прокси в Flowseal/tg-ws-proxy.
     CloudflareProxy,
+    /// WebSocket без TLS на порту 80 родного адреса DC. Единственный прямой
+    /// WebSocket к CDN203, у которого нет фронта на 443.
+    TelegramPlainWs,
 }
 
 impl RouteKind {
@@ -48,6 +54,7 @@ impl RouteKind {
             Self::CloudflareWorker => 4,
             Self::TelegramTcp => 5,
             Self::CloudflareProxy => 6,
+            Self::TelegramPlainWs => 7,
         }
     }
 
@@ -59,6 +66,7 @@ impl RouteKind {
             4 => Some(Self::CloudflareWorker),
             5 => Some(Self::TelegramTcp),
             6 => Some(Self::CloudflareProxy),
+            7 => Some(Self::TelegramPlainWs),
             _ => None,
         }
     }
@@ -72,6 +80,7 @@ impl RouteKind {
             Self::CloudflareWorker => "Cloudflare Worker",
             Self::TelegramTcp => "Telegram TCP (MTProto)",
             Self::CloudflareProxy => "Cloudflare-прокси",
+            Self::TelegramPlainWs => "Telegram WebSocket (порт 80)",
         }
     }
 }
@@ -113,6 +122,23 @@ impl Route {
         }
     }
 
+    /// WebSocket to a native Telegram address on port 80.
+    ///
+    /// There is no TLS here, but no less protection than on `telegram_tcp`: the
+    /// payload is the same obfuscated MTProto stream, and MTProto authenticates
+    /// the server itself. Here the address picks the DC and the Host header does
+    /// not matter, so the address is all that goes in it.
+    fn plain_ws(ip: &str) -> Self {
+        Self {
+            connect_host: ip.to_owned(),
+            websocket_host: ip.to_owned(),
+            path: "/apiws".to_owned(),
+            kind: RouteKind::TelegramPlainWs,
+            port: HTTP_PORT,
+            secure: false,
+        }
+    }
+
     /// A production route: TLS on 443.
     fn https(connect_host: String, websocket_host: String, path: String, kind: RouteKind) -> Self {
         Self {
@@ -327,10 +353,17 @@ impl TransportEngine {
                         }
                         Err(error) => {
                             self.record_failure(&route);
-                            let endpoint = if route.kind == RouteKind::TelegramTcp {
-                                format!("{}:{} (MTProto TCP)", route.connect_host, route.port)
-                            } else {
-                                format!("{} (TLS {})", route.connect_host, route.websocket_host)
+                            let endpoint = match route.kind {
+                                RouteKind::TelegramTcp => {
+                                    format!("{}:{} (MTProto TCP)", route.connect_host, route.port)
+                                }
+                                RouteKind::TelegramPlainWs => {
+                                    format!("{}:{} (WebSocket)", route.connect_host, route.port)
+                                }
+                                _ => format!(
+                                    "{} (TLS {})",
+                                    route.connect_host, route.websocket_host
+                                ),
                             };
                             errors.push(format!("{endpoint} — {error}"));
                         }
@@ -379,8 +412,11 @@ impl TransportEngine {
                 RouteKind::CloudflareWorker | RouteKind::CloudflareProxy => 1,
                 RouteKind::TelegramIp | RouteKind::AlternateTelegramIp => 2,
                 RouteKind::SystemDns => 3,
-                // Last resort, raced separately; see race_connections().
-                RouteKind::TelegramTcp => 4,
+                // Last resort, raced separately; see race_connections(). The
+                // WebSocket goes first: its upgrade is an answer from Telegram,
+                // while TCP only shows that the address accepts connections.
+                RouteKind::TelegramPlainWs => 4,
+                RouteKind::TelegramTcp => 5,
             };
             (preferred_rank, kind_rank)
         });
@@ -479,10 +515,11 @@ pub(crate) fn cf_proxy_host(dc: u16, domain: &str) -> String {
     format!("kws{}.{}", dc, domain)
 }
 
-/// Native TCP to an ordinary data centre — every pinned address except the
-/// CDN, whose native transport is its primary route.
+/// Routes tried only once everything else failed: port 80, and native TCP to
+/// an ordinary data centre. The CDN's native TCP is its primary route.
 fn is_last_resort(route: &Route) -> bool {
-    route.kind == RouteKind::TelegramTcp && route.connect_host != CDN203_IP
+    route.kind == RouteKind::TelegramPlainWs
+        || (route.kind == RouteKind::TelegramTcp && route.connect_host != CDN203_IP)
 }
 
 /// Every address a Worker may be asked to reach, so a deployment can refuse
@@ -497,14 +534,118 @@ pub fn worker_allowed_destinations() -> Vec<&'static str> {
     all
 }
 
-fn canonical_dc(dc: u16) -> u16 {
-    if dc == 203 {
-        2
-    } else {
-        dc
+/// Настоящий WebSocket-фронт Telegram: адрес и имена, которые он обслуживает.
+struct WebFront {
+    ip: &'static str,
+    /// Основной кластер. Принимает и обычные соединения, и медиа.
+    host: &'static str,
+    /// Медиа-кластер (`-1`), только для медиа; см. `routes_for_dc`.
+    media_host: &'static str,
+}
+
+const fn front(ip: &'static str, host: &'static str, media_host: &'static str) -> WebFront {
+    WebFront {
+        ip,
+        host,
+        media_host,
     }
 }
 
+/// Где Telegram принимает WebSocket. Первый фронт каждого DC — тот, на который
+/// указывает `kwsN.web.telegram.org`.
+///
+/// Это не адреса из `telegram_ips`: те говорят только на родном MTProto и на
+/// TLS не отвечают из любой сети. Фронт выбирает DC по имени хоста, а не по
+/// заголовку клиента, так что имя и адрес здесь нельзя менять местами: `.220`
+/// на `kws1` отвечает редиректом. Каждая пара проверена запросом `req_DH_params`:
+/// сервер принимает только свой номер DC, поэтому за фронтом стоит именно он.
+fn web_fronts(dc: u16) -> &'static [WebFront] {
+    const DC1: &[WebFront] = &[
+        front(
+            "149.154.174.100",
+            "kws1.web.telegram.org",
+            "kws1-1.web.telegram.org",
+        ),
+        front(
+            "149.154.174.200",
+            "zws1.web.telegram.org",
+            "zws1-1.web.telegram.org",
+        ),
+        front(
+            "149.154.175.209",
+            "pluto.web.telegram.org",
+            "pluto-1.web.telegram.org",
+        ),
+    ];
+    const DC2: &[WebFront] = &[
+        front(
+            "149.154.167.220",
+            "kws2.web.telegram.org",
+            "kws2-1.web.telegram.org",
+        ),
+        front(
+            "149.154.167.99",
+            "kws2.web.telegram.org",
+            "kws2-1.web.telegram.org",
+        ),
+    ];
+    const DC3: &[WebFront] = &[
+        front(
+            "149.154.174.100",
+            "kws3.web.telegram.org",
+            "kws3-1.web.telegram.org",
+        ),
+        front(
+            "149.154.174.200",
+            "zws3.web.telegram.org",
+            "zws3-1.web.telegram.org",
+        ),
+        front(
+            "149.154.175.209",
+            "aurora.web.telegram.org",
+            "aurora-1.web.telegram.org",
+        ),
+    ];
+    const DC4: &[WebFront] = &[
+        front(
+            "149.154.167.220",
+            "kws4.web.telegram.org",
+            "kws4-1.web.telegram.org",
+        ),
+        front(
+            "149.154.167.99",
+            "kws4.web.telegram.org",
+            "kws4-1.web.telegram.org",
+        ),
+    ];
+    const DC5: &[WebFront] = &[
+        front(
+            "149.154.170.100",
+            "kws5.web.telegram.org",
+            "kws5-1.web.telegram.org",
+        ),
+        front(
+            "149.154.170.200",
+            "zws5.web.telegram.org",
+            "zws5-1.web.telegram.org",
+        ),
+        front(
+            "149.154.170.96",
+            "flora.web.telegram.org",
+            "flora-1.web.telegram.org",
+        ),
+    ];
+    match dc {
+        1 => DC1,
+        2 => DC2,
+        3 => DC3,
+        4 => DC4,
+        5 => DC5,
+        _ => &[],
+    }
+}
+
+/// Родные адреса MTProto: сюда ходит прямой TCP и Worker. TLS здесь нет.
 fn telegram_ips(dc: u16) -> &'static [&'static str] {
     match dc {
         1 => &["149.154.175.50"],
@@ -518,32 +659,33 @@ fn telegram_ips(dc: u16) -> &'static [&'static str] {
 }
 
 pub fn routes_for_dc(dc: u16, media: bool) -> Vec<Route> {
-    if telegram_ips(dc).is_empty() {
+    if dc == 203 {
+        // CDN203 has no WebSocket frontend of its own: its address speaks only
+        // native obfuscated2 on 443, and kws2 on any frontend is plain DC2, where
+        // a CDN session has neither its keys nor its data. Port 80 of the same
+        // address does lead to the CDN. Worker and CF proxy routes are added on
+        // top in routes_for_key.
+        return vec![Route::cdn_tcp(), Route::plain_ws(CDN203_IP)];
+    }
+    let fronts = web_fronts(dc);
+    if fronts.is_empty() {
         return Vec::new();
     }
-    let websocket_dc = canonical_dc(dc);
-    let primary = format!("kws{}.web.telegram.org", websocket_dc);
-    let secondary = format!("kws{}-1.web.telegram.org", websocket_dc);
-    let websocket_hosts = if media {
-        [secondary, primary]
-    } else {
-        [primary, secondary]
-    };
-    let ips = telegram_ips(dc);
-    let mut routes = Vec::new();
-    if dc == 203 {
-        // CDN203 speaks native obfuscated2 on this exact IP:443. A WebSocket
-        // handshake on the same address can time out even while MTProto is
-        // healthy. Keep the destination unchanged and use its native wire
-        // transport; never redirect CDN authorization to ordinary DC2.
-        routes.push(Route::cdn_tcp());
-    }
 
-    for websocket_host in &websocket_hosts {
-        for (index, ip) in ips.iter().enumerate() {
+    // Only media may use the `-1` hosts. That cluster rejects an ordinary
+    // connection's key exchange with -444 (measured on kws2-1 and kws4-1), and
+    // Telegram Desktop answers -444 by turning the proxy off as misconfigured.
+    let clusters: &[fn(&WebFront) -> &'static str] = if media {
+        &[|front| front.media_host, |front| front.host]
+    } else {
+        &[|front| front.host]
+    };
+    let mut routes = Vec::new();
+    for host in clusters {
+        for (index, front) in fronts.iter().enumerate() {
             routes.push(Route::https(
-                (*ip).to_owned(),
-                websocket_host.clone(),
+                front.ip.to_owned(),
+                host(front).to_owned(),
                 "/apiws".to_owned(),
                 if index == 0 {
                     RouteKind::TelegramIp
@@ -552,28 +694,18 @@ pub fn routes_for_dc(dc: u16, media: bool) -> Vec<Route> {
                 },
             ));
         }
-        // DC203 uses the DC2 hostname only for TLS/HTTP virtual hosting.
-        // Resolving it would connect to DC2 itself and send CDN sessions to
-        // the wrong data centre. Its fallback must retain the CDN address.
-        if dc != 203 {
-            routes.push(Route::https(
-                websocket_host.clone(),
-                websocket_host.clone(),
-                "/apiws".to_owned(),
-                RouteKind::SystemDns,
-            ));
-        }
-    }
-    // Where the WebSocket frontends are filtered but the address itself is
-    // reachable, plain MTProto to the DC's primary address still works —
-    // measured from a Russian ISP where TLS to .51/.91/.50/.100/.5 was cut
-    // while native TCP to each of them answered req_pq (main and media).
-    // 149.154.167.220 is a WebSocket frontend only and does not speak native
-    // MTProto, so it gets no TCP route. Raced only after everything above
-    // failed (race_connections).
-    if dc != 203 {
-        routes.push(Route::telegram_tcp(ips[0]));
+        let name = host(&fronts[0]).to_owned();
+        routes.push(Route::https(
+            name.clone(),
+            name,
+            "/apiws".to_owned(),
+            RouteKind::SystemDns,
+        ));
     }
+    // Where the frontends are out of reach, the DC's own address may still
+    // work. Raced only after everything above failed (race_connections).
+    routes.push(Route::plain_ws(telegram_ips(dc)[0]));
+    routes.push(Route::telegram_tcp(telegram_ips(dc)[0]));
     routes
 }
 
@@ -615,6 +747,11 @@ async fn connect_route_with_config(
             "native MTProto TCP разрешён только до закреплённых адресов Telegram".to_owned(),
         );
     }
+    if route.kind == RouteKind::TelegramPlainWs && !allowed_plain_ws_route(route) {
+        return Err(
+            "WebSocket без TLS разрешён только до закреплённых адресов Telegram".to_owned(),
+        );
+    }
     let tcp = tokio::time::timeout(
         CONNECT_TIMEOUT,
         TcpStream::connect((route.connect_host.as_str(), route.port)),
@@ -643,8 +780,8 @@ async fn connect_route_with_config(
     );
 
     if !route.secure {
-        // Raw TCP has already returned above. Plain WebSocket is only used by
-        // local fixtures; production WebSocket routes are always HTTPS.
+        // Raw TCP has already returned above. Besides local fixtures, plain
+        // WebSocket is only allowed to pinned Telegram addresses, checked above.
         return tokio::time::timeout(
             CONNECT_TIMEOUT,
             tokio_tungstenite::client_async(request, MaybeTlsStream::Plain(tcp)),
@@ -670,6 +807,17 @@ async fn connect_route_with_config(
     .map_err(|error| format!("TLS/WebSocket handshake: {}", error))
 }
 
+fn allowed_plain_ws_route(route: &Route) -> bool {
+    #[cfg(test)]
+    if route.connect_host == "127.0.0.1" {
+        return !route.secure;
+    }
+    worker_allowed_destinations().contains(&route.connect_host.as_str())
+        && route.port == HTTP_PORT
+        && route.websocket_host == route.connect_host
+        && !route.secure
+}
+
 fn allowed_tcp_route(route: &Route) -> bool {
     #[cfg(test)]
     if route.connect_host == "127.0.0.1" && !route.secure {
@@ -724,7 +872,73 @@ mod tests {
     fn media_prefers_dash_one_websocket_host() {
         let routes = routes_for_dc(2, true);
         assert_eq!(routes[0].websocket_host, "kws2-1.web.telegram.org");
-        assert_eq!(routes[0].connect_host, "149.154.167.51");
+        assert_eq!(routes[0].connect_host, "149.154.167.220");
+    }
+
+    #[test]
+    fn only_media_may_reach_the_dash_one_cluster() {
+        // That cluster answers an ordinary key exchange with -444, and Telegram
+        // Desktop reacts by disabling the proxy as misconfigured.
+        for dc in [1, 2, 3, 4, 5] {
+            let dash_one = |route: &Route| route.websocket_host.contains("-1.");
+            assert!(
+                !routes_for_dc(dc, false).iter().any(dash_one),
+                "DC{dc} ordinary connection on a media-only cluster"
+            );
+            let media = routes_for_dc(dc, true);
+            assert!(dash_one(&media[0]), "DC{dc}: {:?}", media[0]);
+            // The main cluster accepts media too, so it stays as a fallback.
+            assert!(media
+                .iter()
+                .any(|route| route.websocket_host == format!("kws{dc}.web.telegram.org")));
+        }
+    }
+
+    #[test]
+    fn websocket_routes_only_dial_real_frontends() {
+        // The native addresses speak MTProto and answer TLS with garbage from
+        // any network, so a WebSocket route there can never connect.
+        for dc in [1, 2, 3, 4, 5, 203] {
+            for media in [false, true] {
+                for route in routes_for_dc(dc, media) {
+                    if matches!(
+                        route.kind,
+                        RouteKind::TelegramTcp | RouteKind::TelegramPlainWs | RouteKind::SystemDns
+                    ) {
+                        continue;
+                    }
+                    assert!(
+                        web_fronts(dc)
+                            .iter()
+                            .any(|front| front.ip == route.connect_host),
+                        "DC{dc}: {route:?}"
+                    );
+                    assert!(route.secure, "DC{dc}: {route:?}");
+                    assert_ne!(route.connect_host, telegram_ips(dc)[0], "DC{dc}");
+                }
+            }
+        }
+    }
+
+    #[test]
+    fn frontends_are_only_dialled_under_a_name_they_serve() {
+        // A frontend picks the DC from the Host header, not from the client's
+        // init: kws2 is DC2 whatever the header says, and .220 redirects kws1.
+        for (dc, name) in [
+            (1, "kws1."),
+            (2, "kws2."),
+            (3, "kws3."),
+            (4, "kws4."),
+            (5, "kws5."),
+        ] {
+            let routes = routes_for_dc(dc, false);
+            assert!(routes[0].websocket_host.starts_with(name), "{routes:?}");
+            let front = &web_fronts(dc)[0];
+            assert_eq!(routes[0].connect_host, front.ip);
+        }
+        assert!(routes_for_dc(1, false)
+            .iter()
+            .all(|route| route.connect_host != "149.154.167.220"));
     }
 
     #[test]
@@ -734,7 +948,8 @@ mod tests {
         assert!(routes[0].websocket_host.is_empty());
         assert!(!routes[0].secure);
         assert_eq!(routes[0].connect_host, "91.105.192.100");
-        assert_eq!(routes[1].websocket_host, "kws2.web.telegram.org");
+        assert_eq!(routes[1], Route::plain_ws("91.105.192.100"));
+        assert_eq!(routes.len(), 2, "CDN203 has no WebSocket frontend on 443");
         assert!(routes
             .iter()
             .all(|route| route.connect_host == "91.105.192.100"));
@@ -816,15 +1031,27 @@ mod tests {
                 let routes = engine.routes_for_key(DcKey { dc, media });
                 assert!(!routes.is_empty(), "DC{dc} must have at least one route");
                 for route in routes {
-                    assert_eq!(route.port, 443, "{route:?}");
-                    if route.kind == RouteKind::TelegramTcp {
-                        assert!(
-                            telegram_ips(dc).contains(&route.connect_host.as_str()),
-                            "{route:?}"
-                        );
-                        assert!(allowed_tcp_route(&route));
-                    } else {
-                        assert!(route.secure, "{route:?}");
+                    match route.kind {
+                        RouteKind::TelegramTcp => {
+                            assert_eq!(route.port, 443, "{route:?}");
+                            assert!(
+                                telegram_ips(dc).contains(&route.connect_host.as_str()),
+                                "{route:?}"
+                            );
+                            assert!(allowed_tcp_route(&route));
+                        }
+                        // The only route without TLS: pinned address, and no
+                        // Telegram name on the wire.
+                        RouteKind::TelegramPlainWs => {
+                            assert_eq!(route.port, 80, "{route:?}");
+                            assert_eq!(route.connect_host, telegram_ips(dc)[0]);
+                            assert_eq!(route.websocket_host, route.connect_host);
+                            assert!(allowed_plain_ws_route(&route));
+                        }
+                        _ => {
+                            assert_eq!(route.port, 443, "{route:?}");
+                            assert!(route.secure, "{route:?}");
+                        }
                     }
                 }
             }
@@ -837,17 +1064,13 @@ mod tests {
         engine.set_cf_proxy_domains(&["example.com".to_owned()]);
         for dc in [1, 2, 3, 4, 5] {
             let ordered = engine.ordered_candidates(DcKey { dc, media: true });
-            let first_tcp = ordered
+            let first_fallback = ordered
                 .iter()
-                .position(|route| route.kind == RouteKind::TelegramTcp)
-                .expect("every DC has a native TCP fallback");
-            assert!(
-                ordered[first_tcp..].iter().all(is_last_resort),
-                "DC{dc}: {ordered:?}"
-            );
-            assert!(ordered[..first_tcp]
-                .iter()
-                .all(|route| !is_last_resort(route)));
+                .position(is_last_resort)
+                .expect("every DC has a last-resort route");
+            assert_eq!(ordered[first_fallback].kind, RouteKind::TelegramPlainWs);
+            assert_eq!(ordered[first_fallback + 1].kind, RouteKind::TelegramTcp);
+            assert_eq!(ordered.len(), first_fallback + 2, "DC{dc}: {ordered:?}");
         }
         let cdn = engine.ordered_candidates(DcKey {
             dc: 203,
@@ -855,6 +1078,25 @@ mod tests {
         });
         assert_eq!(cdn[0], Route::cdn_tcp());
         assert!(!is_last_resort(&cdn[0]));
+        assert_eq!(cdn.last(), Some(&Route::plain_ws(CDN203_IP)));
+    }
+
+    #[tokio::test]
+    async fn plain_websocket_refuses_anything_but_a_pinned_address() {
+        for route in [
+            Route::plain_ws("203.0.113.5"),
+            Route {
+                websocket_host: "kws2.web.telegram.org".to_owned(),
+                ..Route::plain_ws("149.154.167.51")
+            },
+            Route {
+                port: 8080,
+                ..Route::plain_ws("149.154.167.51")
+            },
+        ] {
+            let error = connect_route(&route).await.unwrap_err();
+            assert!(error.contains("без TLS"), "{route:?}: {error}");
+        }
     }
 
     #[tokio::test]
@@ -1163,6 +1405,8 @@ mod tests {
             RouteKind::SystemDns,
             RouteKind::CloudflareWorker,
             RouteKind::TelegramTcp,
+            RouteKind::CloudflareProxy,
+            RouteKind::TelegramPlainWs,
         ] {
             assert_eq!(RouteKind::from_ui_code(kind.ui_code()), Some(kind));
             assert_eq!(route_label(kind.ui_code()), kind.label());
@@ -1471,6 +1715,20 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    #[ignore = "requires live Telegram network access"]
+    async fn port_80_answers_on_every_native_address() {
+        for dc in [1, 2, 3, 4, 5, 203] {
+            let route = Route::plain_ws(telegram_ips(dc)[0]);
+            match connect_route(&route).await {
+                Ok(TelegramConnection::WebSocket(mut websocket)) => {
+                    websocket.close(None).await.unwrap()
+                }
+                other => panic!("DC{dc}: {other:?}"),
+            }
+        }
+    }
+
     #[tokio::test]
     #[ignore = "requires live Telegram network access"]
     async fn connects_to_all_production_data_centers() {