From 50c980e9ea7460f4a52ef795b5abc4682a15f8a3 Mon Sep 17 00:00:00 2001 From: Flowseal Date: Sun, 4 Oct 2026 21:53:26 +0300 Subject: [PATCH] WsPool refactoring, TCP fallback backoff, removed direct connections outside of a pool --- docs/EN/BuildFromSource.md | 2 +- docs/RU/BuildFromSource.md | 2 +- proxy/bridge.py | 64 +++++++++--- proxy/pool.py | 185 ++++++++++++++++++---------------- proxy/raw_websocket.py | 86 ++++++++-------- proxy/tg_ws_proxy.py | 144 +++----------------------- proxy/utils.py | 4 +- tests/test_h2_routing.py | 27 ++++- tests/test_pool.py | 196 ++++++++++++++++++++++++++++++++++-- tests/test_raw_websocket.py | 34 +++++++ tests/test_tcp_fallback.py | 109 ++++++++++++++++++++ ui/i18n/en.json | 2 +- ui/i18n/ru.json | 2 +- 13 files changed, 563 insertions(+), 294 deletions(-) create mode 100644 tests/test_tcp_fallback.py diff --git a/docs/EN/BuildFromSource.md b/docs/EN/BuildFromSource.md index f439ed8..0b8461d 100644 --- a/docs/EN/BuildFromSource.md +++ b/docs/EN/BuildFromSource.md @@ -56,7 +56,7 @@ tg-ws-proxy [--port PORT] [--host HOST] [--dc-ip DC:IP ...] [-v] | `--fake-tls-domain` | | Enable Fake TLS masquerading (ee-secret) with specified SNI domain | | `--proxy-protocol` | disabled | Accept HAProxy PROXY protocol v1 (for use behind nginx/haproxy with `proxy_protocol on`) | | `--buf-kb` | `256` | Buffer size in KB | -| `--pool-size` | `4` | Number of pre-allocated connections per DC | +| `--pool-size` | `4` | Number of ready WS connections per DC. `0` disables the direct DC→IP WS route | | `--log-file` | disabled | Path to file for saving logs | | `--log-max-mb` | `5` | Maximum log file size in MB (afterwards overwrites) | | `--log-backups` | `0` | Number of log backups after overwrite | diff --git a/docs/RU/BuildFromSource.md b/docs/RU/BuildFromSource.md index 959a21f..6cba6a4 100644 --- a/docs/RU/BuildFromSource.md +++ b/docs/RU/BuildFromSource.md @@ -56,7 +56,7 @@ tg-ws-proxy [--port PORT] [--host HOST] [--dc-ip DC:IP ...] [-v] | `--fake-tls-domain` | | Включить маскировку Fake TLS (ee-secret) с указанным SNI-доменом | | `--proxy-protocol` | выкл. | Принимать HAProxy PROXY protocol v1 (для работы за nginx/haproxy с `proxy_protocol on`) | | `--buf-kb` | `256` | Размер буфера в КБ | -| `--pool-size` | `4` | Количество заготовленных соединений на каждый DC | +| `--pool-size` | `4` | Количество заготовленных WS-соединений на каждый DC. `0` отключает прямой WS-маршрут DC→IP | | `--log-file` | выкл. | Путь к файлу, в который будут сохраняться логи | | `--log-max-mb` | `5` | Максимальный размер файла логов в МБ (после этого начинается перезапись) | | `--log-backups` | `0` | Количество сохранений логов после перезаписи | diff --git a/proxy/bridge.py b/proxy/bridge.py index 4ceef41..da68607 100644 --- a/proxy/bridge.py +++ b/proxy/bridge.py @@ -3,7 +3,7 @@ import logging import struct import time -from typing import List, Optional +from typing import Dict, List, Optional, Set, Tuple from urllib.parse import urlencode from .utils import * @@ -20,6 +20,18 @@ from .cf_h2 import bridge_h2 log = logging.getLogger('tg-mtproto-proxy') _st_I_le = struct.Struct(' None: + _tcp_failures.clear() + _tcp_retry_after.clear() + _tcp_connecting.clear() + ZERO_64 = b'\x00' * 64 @@ -171,8 +183,6 @@ async def do_fallback(reader, writer, relay_init, label, if ok: return True elif method == 'tcp' and fallback_dst: - log.info("[%s] DC%d%s -> TCP fallback to %s:443", - label, dc, media_tag, fallback_dst) ok = await _tcp_fallback( reader, writer, fallback_dst, 443, relay_init, label, ctx) @@ -271,19 +281,45 @@ async def _cfproxy_fallback(reader, writer, relay_init, label, async def _tcp_fallback(reader, writer, dst, port, relay_init, label, ctx: CryptoCtx): - try: - rr, rw = await asyncio.wait_for( - asyncio.open_connection(dst, port), timeout=10) - except Exception as exc: - log.warning("[%s] TCP fallback to %s:%d failed: %s", - label, dst, port, repr(exc)) + key = (dst, port) + if key in _tcp_connecting or time.monotonic() < _tcp_retry_after.get(key, 0): + log.debug("[%s] TCP fallback to %s:%d skipped (backoff or connecting)", + label, dst, port) return False - stats.connections_tcp_fallback += 1 - rw.write(relay_init) - await rw.drain() - await _bridge_tcp_reencrypt(reader, writer, rr, rw, label, ctx) - return True + _tcp_connecting.add(key) + rw = None + try: + try: + log.info("[%s] TCP fallback to %s:%d", label, dst, port) + rr, rw = await asyncio.wait_for( + asyncio.open_connection(dst, port), timeout=10) + rw.write(relay_init) + await rw.drain() + except Exception as exc: + failures = _tcp_failures.get(key, 0) + 1 + _tcp_failures[key] = failures + delay = min(TCP_BACKOFF_INITIAL * 2 ** min(failures - 1, 7), + TCP_BACKOFF_MAX) + _tcp_retry_after[key] = time.monotonic() + delay + log.warning("[%s] TCP fallback to %s:%d failed: %r; retry in %.0fs", + label, dst, port, exc, delay) + return False + finally: + _tcp_connecting.discard(key) + + _tcp_failures.pop(key, None) + _tcp_retry_after.pop(key, None) + stats.connections_tcp_fallback += 1 + await _bridge_tcp_reencrypt(reader, writer, rr, rw, label, ctx) + return True + finally: + if rw is not None: + rw.close() + try: + await rw.wait_closed() + except (OSError, ConnectionError): + pass async def bridge_ws_reencrypt(reader, writer, ws: RawWebSocket, label, diff --git a/proxy/pool.py b/proxy/pool.py index dd19f94..824671e 100644 --- a/proxy/pool.py +++ b/proxy/pool.py @@ -10,12 +10,11 @@ from typing import Dict, List, Optional, Tuple, Set from .raw_websocket import RawWebSocket, WsHandshakeError from .stats import stats from .config import proxy_config -from .utils import ws_domains, DC_DEFAULT_IPS +from .utils import ws_domains, DC_DEFAULT_IPS, WS_PATH, WS_PATH_TEST log = logging.getLogger('tg-mtproto-proxy') -# TODO: domains handling is broken: wrong is_media flag causes tcp_reset after handshake, -# but initial connection is still established no matter what is is_media flag is set to + class _WsPool: WS_POOL_MAX_AGE = 120.0 WS_POOL_CHECK_INTERVAL = 5.0 @@ -23,17 +22,20 @@ class _WsPool: REFILL_BACKOFF_MAX = 3600.0 def __init__(self): - self._idle: Dict[Tuple[int, bool], deque] = {} - self._refilling: Set[Tuple[int, bool]] = set() - self._rotating: Dict[Tuple[int, bool], asyncio.Task] = {} - self._refill_failures: Dict[Tuple[int, bool], int] = {} - self._refill_after: Dict[Tuple[int, bool], float] = {} - self.try_fronting_first = True + self._idle: Dict[Tuple[int, bool, bool], deque] = {} + self._refilling: Dict[Tuple[int, bool, bool], asyncio.Task] = {} + self._rotating: Dict[Tuple[int, bool, bool], asyncio.Task] = {} + self._refill_failures: Dict[Tuple[int, bool, bool], int] = {} + self._refill_after: Dict[Tuple[int, bool, bool], float] = {} + self.try_fronting_first = False - async def get(self, dc: int, is_media: bool, - target_ip: str, domains: List[str] + async def get(self, dc: int, is_media: bool, *, is_test_dc: bool = False ) -> Optional[RawWebSocket]: - key = (dc, is_media) + target_ip = proxy_config.dc_redirects.get(dc) + if not target_ip or proxy_config.pool_size <= 0: + return None + key = (dc, is_media, is_test_dc) + domains = ws_domains(dc, is_media) now = time.monotonic() bucket = self._idle.get(key) @@ -43,14 +45,12 @@ class _WsPool: while bucket: ws, created = bucket.popleft() age = now - created - if (age > self.WS_POOL_MAX_AGE or ws._closed - or ws.writer.transport.is_closing()): + if self._is_stale(ws, created, now): asyncio.create_task(self._quiet_close(ws)) continue stats.pool_hits += 1 - log.debug("WS pool hit DC%d%s (age=%.1fs, left=%d)", - dc, 'm' if is_media else '', age, len(bucket)) - self.report_success(dc, is_media) + log.debug("WS pool hit DC%d%s%s (age=%.1fs, left=%d)", + dc, 't' if is_test_dc else '', 'm' if is_media else '', age, len(bucket)) self._schedule_refill(key, target_ip, domains) return ws @@ -58,20 +58,25 @@ class _WsPool: self._schedule_refill(key, target_ip, domains) return None + def _is_stale(self, ws, created, now): + return (now - created >= self.WS_POOL_MAX_AGE or ws._closed + or ws.writer.transport.is_closing() or ws.reader.at_eof() + or ws.reader.exception() is not None) + def _schedule_refill(self, key, target_ip, domains): + if proxy_config.pool_size <= 0: + return + self._schedule_rotation(key, target_ip, domains) if (key in self._refilling or time.monotonic() < self._refill_after.get(key, 0)): return - self._refilling.add(key) - asyncio.create_task(self._refill(key, target_ip, domains)) - - def report_success(self, dc: int, is_media: bool) -> None: - key = (dc, is_media) - self._refill_failures.pop(key, None) - self._refill_after.pop(key, None) + self._refilling[key] = asyncio.create_task( + self._refill(key, target_ip, domains)) async def _refill(self, key, target_ip, domains): - dc, is_media = key + dc, is_media, is_test_dc = key + tasks = [] + adopted = set() try: bucket = self._idle.setdefault(key, deque()) needed = proxy_config.pool_size - len(bucket) @@ -79,19 +84,23 @@ class _WsPool: return connected = 0 tasks = [asyncio.create_task( - self._connect_one(target_ip, domains)) + self._connect_one(target_ip, domains, + WS_PATH_TEST if is_test_dc else WS_PATH)) for _ in range(needed)] - for t in tasks: + for t in asyncio.as_completed(tasks): try: ws = await t if ws: + if self._refilling.get(key) is not asyncio.current_task(): + return bucket.append((ws, time.monotonic())) + adopted.add(ws) connected += 1 - self._schedule_rotation(key, target_ip, domains) - except Exception: - pass + except Exception as exc: + log.debug("WS pool connect failed: %r", exc) if connected: - self.report_success(dc, is_media) + self._refill_failures.pop(key, None) + self._refill_after.pop(key, None) else: failures = self._refill_failures.get(key, 0) + 1 self._refill_failures[key] = failures @@ -102,12 +111,20 @@ class _WsPool: ) self._refill_after[key] = time.monotonic() + delay log.info( - "WS pool refill failed for DC%d%s, retry in %.0fs", - dc, 'm' if is_media else '', delay) - log.debug("WS pool refilled DC%d%s: %d ready", - dc, 'm' if is_media else '', len(bucket)) + "WS pool refill failed for DC%d%s%s, retry in %.0fs", + dc, 't' if is_test_dc else '', 'm' if is_media else '', delay) + log.debug("WS pool refilled DC%d%s%s: %d ready", + dc, 't' if is_test_dc else '', 'm' if is_media else '', len(bucket)) finally: - self._refilling.discard(key) + for task in tasks: + if not task.done(): + task.cancel() + results = await asyncio.gather(*tasks, return_exceptions=True) + for ws in results: + if ws is not None and not isinstance(ws, BaseException) and ws not in adopted: + await self._quiet_close(ws) + if self._refilling.get(key) is asyncio.current_task(): + self._refilling.pop(key, None) def _schedule_rotation(self, key, target_ip, domains): if key in self._rotating: @@ -116,28 +133,16 @@ class _WsPool: self._rotate(key, target_ip, domains)) async def _rotate(self, key, target_ip, domains): - dc, is_media = key + dc, is_media, is_test_dc = key try: - while True: - bucket = self._idle.get(key) - if not bucket: - return - - expires_at = min( - created + self.WS_POOL_MAX_AGE - for _, created in bucket) - await asyncio.sleep(min( - self.WS_POOL_CHECK_INTERVAL, - max(0, expires_at - time.monotonic()))) - + while proxy_config.pool_size > 0: + bucket = self._idle.setdefault(key, deque()) now = time.monotonic() expired = [] ready = deque() while bucket: ws, created = bucket.popleft() - if (now - created >= self.WS_POOL_MAX_AGE - or ws._closed - or ws.writer.transport.is_closing()): + if self._is_stale(ws, created, now): expired.append(ws) else: ready.append((ws, created)) @@ -147,49 +152,45 @@ class _WsPool: for ws in expired: asyncio.create_task(self._quiet_close(ws)) log.debug( - "WS pool rotated DC%d%s: %d stale, %d ready", - dc, 'm' if is_media else '', len(expired), len(bucket)) + "WS pool rotated DC%d%s%s: %d stale, %d ready", + dc, 't' if is_test_dc else '', 'm' if is_media else '', + len(expired), len(bucket)) if len(bucket) < proxy_config.pool_size: self._schedule_refill(key, target_ip, domains) + + wake_at = now + self.WS_POOL_CHECK_INTERVAL + if bucket: + wake_at = min(wake_at, min( + created + self.WS_POOL_MAX_AGE for _, created in bucket)) + if key not in self._refilling and self._refill_after.get(key, 0) > now: + wake_at = min(wake_at, self._refill_after[key]) + await asyncio.sleep(max(0, wake_at - time.monotonic())) finally: if self._rotating.get(key) is asyncio.current_task(): self._rotating.pop(key, None) - async def _connect_one(self, target_ip, domains) -> Optional[RawWebSocket]: + async def _connect_one(self, target_ip, domains, path=WS_PATH) -> Optional[RawWebSocket]: for domain in domains: - if self.try_fronting_first: - ws = await self._connect_fronted(target_ip, domain) - if ws: - return ws - try: - ws = await RawWebSocket.connect( - target_ip, domain, timeout=8) - self.try_fronting_first = False - return ws - except (asyncio.TimeoutError, ConnectionResetError): - if self.try_fronting_first: - return None - return await self._connect_fronted(target_ip, domain) - except WsHandshakeError as exc: - if exc.is_redirect: + modes = (True, False) if self.try_fronting_first else (False, True) + for fronted in modes: + try: + ws = await RawWebSocket.connect( + target_ip, domain, timeout=7 if fronted else 8, path=path, + sni="sprinthost.ru" if fronted else None) + except Exception as exc: + stats.ws_errors += 1 + log.debug("WS pool connect %s%s via %s (fronting=%s): %r", + domain, path, target_ip, fronted, exc) continue - return None - except Exception: - return None + self.try_fronting_first = fronted + if fronted: + stats.connections_fronting += 1 + log.debug("WS pool connected %s%s via %s (fronting=%s)", + domain, path, target_ip, fronted) + return ws return None - async def _connect_fronted(self, target_ip, domain) -> Optional[RawWebSocket]: - try: - ws = await RawWebSocket.connect( - target_ip, domain, timeout=7, sni="sprinthost.ru") - except Exception: - return None - - stats.connections_fronting += 1 - self.try_fronting_first = True - return ws - async def _quiet_close(self, ws): try: await ws.close() @@ -202,14 +203,21 @@ class _WsPool: continue for is_media in (False, True): domains = ws_domains(dc, is_media) - self._schedule_refill((dc, is_media), target_ip, domains) + key = (dc, is_media, proxy_config.force_test_dc) + self._schedule_refill(key, target_ip, domains) log.info("WS pool warmup started for %d DC(s)", len(proxy_config.dc_redirects)) def reset(self): loop = asyncio.get_running_loop() - for task in self._rotating.values(): + for task in list(self._rotating.values()) + list(self._refilling.values()): if not task.done() and task.get_loop() is loop: task.cancel() + for bucket in self._idle.values(): + for ws, _ in bucket: + try: + ws.writer.close() + except Exception as exc: + log.debug("WS pool close failed: %r", exc) self._idle.clear() self._refilling.clear() self._rotating.clear() @@ -217,6 +225,11 @@ class _WsPool: self._refill_after.clear() self.try_fronting_first = False + async def close(self): + tasks = list(self._rotating.values()) + list(self._refilling.values()) + self.reset() + await asyncio.gather(*tasks, return_exceptions=True) + class _CfWorkerPool: WS_POOL_MAX_AGE = 100.0 diff --git a/proxy/raw_websocket.py b/proxy/raw_websocket.py index 953fe86..fc0ad53 100644 --- a/proxy/raw_websocket.py +++ b/proxy/raw_websocket.py @@ -104,26 +104,26 @@ class RawWebSocket: timeout=min(timeout, 10), ) - set_sock_opts(writer.transport, proxy_config.buffer_size) - - ws_key = base64.b64encode(os.urandom(16)).decode() - - req = ( - f'GET {path} HTTP/1.1\r\n' - f'Host: {domain}\r\n' - f'Upgrade: websocket\r\n' - f'Connection: Upgrade\r\n' - f'Sec-WebSocket-Key: {ws_key}\r\n' - f'Sec-WebSocket-Version: 13\r\n' - f'Sec-WebSocket-Protocol: binary\r\n' - f'\r\n' - ) - - writer.write(req.encode()) - await writer.drain() - - response_lines: list[str] = [] try: + set_sock_opts(writer.transport, proxy_config.buffer_size) + + ws_key = base64.b64encode(os.urandom(16)).decode() + + req = ( + f'GET {path} HTTP/1.1\r\n' + f'Host: {domain}\r\n' + f'Upgrade: websocket\r\n' + f'Connection: Upgrade\r\n' + f'Sec-WebSocket-Key: {ws_key}\r\n' + f'Sec-WebSocket-Version: 13\r\n' + f'Sec-WebSocket-Protocol: binary\r\n' + f'\r\n' + ) + + writer.write(req.encode()) + await writer.drain() + + response_lines: list[str] = [] while True: line = await asyncio.wait_for(reader.readline(), timeout=timeout) @@ -131,34 +131,32 @@ class RawWebSocket: break response_lines.append( line.decode('utf-8', errors='replace').strip()) - except asyncio.TimeoutError: + + if not response_lines: + raise WsHandshakeError(0, 'empty response') + + first_line = response_lines[0] + parts = first_line.split(' ', 2) + try: + status_code = int(parts[1]) if len(parts) >= 2 else 0 + except ValueError: + status_code = 0 + + if status_code == 101: + return RawWebSocket(reader, writer) + + headers: dict[str, str] = {} + for hl in response_lines[1:]: + if ':' in hl: + k, v = hl.split(':', 1) + headers[k.strip().lower()] = v.strip() + + raise WsHandshakeError(status_code, first_line, headers, + location=headers.get('location')) + except BaseException: writer.close() raise - if not response_lines: - writer.close() - raise WsHandshakeError(0, 'empty response') - - first_line = response_lines[0] - parts = first_line.split(' ', 2) - try: - status_code = int(parts[1]) if len(parts) >= 2 else 0 - except ValueError: - status_code = 0 - - if status_code == 101: - return RawWebSocket(reader, writer) - - headers: dict[str, str] = {} - for hl in response_lines[1:]: - if ':' in hl: - k, v = hl.split(':', 1) - headers[k.strip().lower()] = v.strip() - - writer.close() - raise WsHandshakeError(status_code, first_line, headers, - location=headers.get('location')) - async def send(self, data: bytes): if self._closed: raise ConnectionError("WebSocket closed") diff --git a/proxy/tg_ws_proxy.py b/proxy/tg_ws_proxy.py index d526d11..e6d16af 100644 --- a/proxy/tg_ws_proxy.py +++ b/proxy/tg_ws_proxy.py @@ -10,7 +10,7 @@ import argparse import logging import socket as _socket -from typing import Dict, Optional, Set, Tuple +from typing import Optional, Set, Tuple if __name__ == '__main__' and (__package__ is None or __package__ == ''): @@ -22,8 +22,8 @@ if __name__ == '__main__' and (__package__ is None or __package__ == ''): from .utils import * from .stats import stats from .config import proxy_config, parse_dc_ip_list, start_cfproxy_domain_refresh, coerce_domain_list -from .bridge import MsgSplitter, CryptoCtx, do_fallback, bridge_ws_reencrypt -from .raw_websocket import RawWebSocket, WsHandshakeError, set_sock_opts +from .bridge import MsgSplitter, CryptoCtx, do_fallback, bridge_ws_reencrypt, reset_tcp_backoff +from .raw_websocket import set_sock_opts from .fake_tls import proxy_to_masking_domain, verify_client_hello, build_server_hello, FakeTlsStream, TLS_RECORD_HANDSHAKE from .balancer import balancer from .pool import ws_pool, cf_worker_pool @@ -34,14 +34,8 @@ from ._aes import Cipher, algorithms, modes log = logging.getLogger('tg-mtproto-proxy') -IP_FAIL_COOLDOWN = 3600.0 -DC_FAIL_COOLDOWN = 60.0 -WS_FAIL_TIMEOUT = 2.0 LISTENER_CHECK_INTERVAL = 5.0 LISTENER_RESTART_DELAY = 1.0 -ws_blacklist: Set[str] = set() -dc_fail_until: Dict[str, float] = {} -ip_fail_until: Dict[str, float] = {} cf_h2_pool: Optional[CfH2Pool] = None @@ -298,119 +292,11 @@ async def _handle_client(reader, writer, secret: bytes): relay_init = _generate_relay_init(proto_tag, dc_idx) ctx = _build_crypto_ctx(client_dec_prekey_iv, secret, relay_init) - dc_key = f'{dc}{"t" if is_test_dc else ""}{"m" if is_media else ""}' media_tag = " media" if is_media else "" - now = time.monotonic() - ws_path = WS_PATH_TEST if is_test_dc else WS_PATH - target = proxy_config.dc_redirects.get(dc) - is_any_cf_fallback = proxy_config.fallback_cfproxy or proxy_config.cfproxy_worker_domains - domains = ws_domains(dc, is_media) - ws = None - - # Fallback if DC not in config, if WS blacklisted for this DC/is_media or if connect to ip is timed out - if (dc not in proxy_config.dc_redirects - or dc_key in ws_blacklist - or now < ip_fail_until.get(target, 0) and is_any_cf_fallback): - - if dc not in proxy_config.dc_redirects: - log.info("[%s] DC%d not in config -> fallback", - label, dc) - elif dc_key in ws_blacklist: - log.info("[%s] DC%d%s WS blacklisted -> fallback", - label, dc, media_tag) - else: - # Try to get WS from pool first, might be accidental timeout - ws = await ws_pool.get( - dc, is_media, target, domains - ) if not is_test_dc else None - - if not ws: - log.info("[%s] DC%d%s WS connect to %s was timed out -> fallback", - label, dc, media_tag, target) - else: - log.info("[%s] DC%d%s WS connect to %s was timed out, but pool hit -> using WS", - label, dc, media_tag, target) - - if not ws: - splitter = None - try: - splitter = MsgSplitter(relay_init, proto_int) - except Exception: - pass - ok = await do_fallback( - clt_reader, clt_writer, relay_init, label, - dc, is_test_dc, is_media, media_tag, - ctx, splitter=splitter, h2_pool=cf_h2_pool, proto_tag=proto_tag) - if not ok: - log.warning("[%s] DC%d%s no fallback available", - label, dc, media_tag) - return - - ws_timeout = WS_FAIL_TIMEOUT if now < dc_fail_until.get(dc_key, 0) else 5.0 - ws_failed_redirect = False - ws_timed_out = False - all_redirects = True - - ws = ws or await ws_pool.get( - dc, is_media, target, domains - ) if not is_test_dc else None - if ws: - log.info("[%s] DC%d%s -> pool hit via %s", - label, dc, media_tag, target) - else: - for domain in domains: - url = f'wss://{domain}{ws_path}' - log.info("[%s] DC%d%s -> %s via %s", - label, dc, media_tag, url, target) - try: - ws = await RawWebSocket.connect(target, domain, - timeout=ws_timeout, - path=ws_path) - all_redirects = False - break - except WsHandshakeError as exc: - stats.ws_errors += 1 - if exc.is_redirect: - ws_failed_redirect = True - log.warning("[%s] DC%d%s got %d from %s -> %s", - label, dc, media_tag, - exc.status_code, domain, - exc.location or '?') - continue - else: - all_redirects = False - log.warning("[%s] DC%d%s WS handshake: %s", - label, dc, media_tag, exc.status_line) - except asyncio.TimeoutError: - stats.ws_errors += 1 - ws_timed_out = True - log.warning("[%s] DC%d%s WS connect timed out via %s", - label, dc, media_tag, domain) - break - except Exception as exc: - stats.ws_errors += 1 - all_redirects = False - log.warning("[%s] DC%d%s WS connect failed: %s", - label, dc, media_tag, repr(exc)) - - # WS failed -> fallback + ws = await ws_pool.get(dc, is_media, is_test_dc=is_test_dc) if ws is None: - if ws_timed_out: - ip_fail_until[target] = now + IP_FAIL_COOLDOWN - log.info("[%s] DC%d%s WS connect to %s timed out, cooldown for %ds", - label, dc, media_tag, target, int(IP_FAIL_COOLDOWN)) - - if ws_failed_redirect and all_redirects: - ws_blacklist.add(dc_key) - log.warning("[%s] DC%d%s blacklisted for WS (all 302)", - label, dc, media_tag) - elif ws_failed_redirect: - dc_fail_until[dc_key] = now + DC_FAIL_COOLDOWN - else: - dc_fail_until[dc_key] = now + DC_FAIL_COOLDOWN - log.info("[%s] DC%d%s WS failed for %ds", - label, dc, media_tag, int(DC_FAIL_COOLDOWN)) - + log.info("[%s] DC%d%s WS pool unavailable -> fallback", + label, dc, media_tag) splitter_fb = None try: splitter_fb = MsgSplitter(relay_init, proto_int) @@ -421,12 +307,12 @@ async def _handle_client(reader, writer, secret: bytes): dc, is_test_dc, is_media, media_tag, ctx, splitter=splitter_fb, h2_pool=cf_h2_pool, proto_tag=proto_tag) if ok: - log.info("[%s] DC%d%s fallback closed", - label, dc, media_tag) + log.info("[%s] DC%d%s fallback closed", label, dc, media_tag) + else: + log.warning("[%s] DC%d%s no fallback available", label, dc, media_tag) return - ip_fail_until.pop(target, None) - ws_pool.report_success(dc, is_media) + log.info("[%s] DC%d%s -> WS pool hit", label, dc, media_tag) stats.connections_ws += 1 splitter = None @@ -476,11 +362,9 @@ async def _run(stop_event: Optional[asyncio.Event] = None): global _server_instance, _server_stop_event, cf_h2_pool _server_stop_event = stop_event + reset_tcp_backoff() ws_pool.reset() cf_worker_pool.reset() - ws_blacklist.clear() - dc_fail_until.clear() - ip_fail_until.clear() _client_tasks.clear() cf_h2_pool = CfH2Pool() if proxy_config.h2_enabled else None @@ -567,8 +451,7 @@ async def _run(stop_event: Optional[asyncio.Event] = None): if now < next_stats: continue next_stats = now + 60 - bl = ', '.join(f'DC{k}' for k in sorted(ws_blacklist)) or 'none' - log.info("stats: %s | ws_bl: %s", stats.summary(), bl) + log.info("stats: %s", stats.summary()) if cf_h2_pool is not None: cf_h2_pool.log_stats() except asyncio.CancelledError: @@ -650,6 +533,7 @@ async def _run(stop_event: Optional[asyncio.Event] = None): if task is not None: await _quiet_cancel(task) await server.wait_closed() + await ws_pool.close() if cf_h2_pool is not None: await cf_h2_pool.close() cf_h2_pool = None @@ -695,7 +579,7 @@ def main(): ap.add_argument('--buf-kb', type=int, default=256, metavar='KB', help='Socket send/recv buffer size in KB (default 256)') ap.add_argument('--pool-size', type=int, default=4, metavar='N', - help='WS connection pool size per DC (default 4, min 0)') + help='WS connection pool size per DC (default 4, 0 disables direct WS)') ap.add_argument('--cfproxy-domain', action='append', default=None, metavar='DOMAIN', help='User defined Cloudflare-proxied domain for H2 media and WS fallback ' diff --git a/proxy/utils.py b/proxy/utils.py index a09fee2..65a507e 100644 --- a/proxy/utils.py +++ b/proxy/utils.py @@ -58,11 +58,11 @@ WS_PATH = '/apiws' WS_PATH_TEST = WS_PATH + '_test' -def ws_domains(dc: int, is_media) -> List[str]: +def ws_domains(dc: int, is_media: bool) -> List[str]: if dc == 203: dc = 2 if not is_media: - return [f'kws{dc}.web.telegram.org', f'kws{dc}-1.web.telegram.org'] + return [f'kws{dc}.web.telegram.org'] return [f'kws{dc}-1.web.telegram.org', f'kws{dc}.web.telegram.org'] diff --git a/tests/test_h2_routing.py b/tests/test_h2_routing.py index 3cc34c9..d850c95 100644 --- a/tests/test_h2_routing.py +++ b/tests/test_h2_routing.py @@ -19,8 +19,6 @@ class H2RoutingTest(unittest.IsolatedAsyncioTestCase): with patch.object(proxy_config, 'dc_redirects', {2: '149.154.167.51'}), \ patch.object(proxy_config, 'force_test_dc', False), \ patch.object(tg_ws_proxy, 'cf_h2_pool', pool), \ - patch.object(tg_ws_proxy, 'ws_blacklist', set()), \ - patch.object(tg_ws_proxy, 'ip_fail_until', {}), \ patch.object(tg_ws_proxy, 'set_sock_opts'), \ patch.object(tg_ws_proxy, '_read_client_init', AsyncMock( return_value=(b'', reader, writer, 'test'))), \ @@ -28,7 +26,6 @@ class H2RoutingTest(unittest.IsolatedAsyncioTestCase): 2, True, PROTO_TAG_INTERMEDIATE, b'')), \ patch.object(tg_ws_proxy, '_build_crypto_ctx'), \ patch.object(tg_ws_proxy.ws_pool, 'get', AsyncMock(return_value=ws)), \ - patch.object(tg_ws_proxy.ws_pool, 'report_success'), \ patch.object(tg_ws_proxy, 'bridge_ws_reencrypt', AsyncMock()) as direct, \ patch.object(tg_ws_proxy, 'do_fallback', AsyncMock()) as fallback: await tg_ws_proxy._handle_client(reader, writer, b'') @@ -36,6 +33,30 @@ class H2RoutingTest(unittest.IsolatedAsyncioTestCase): fallback.assert_not_awaited() pool.open.assert_not_awaited() + async def test_pool_miss_goes_straight_to_fallback_including_test_dcs(self): + for dc, forced, expected_test in [(2, False, False), + (10002, False, True), + (2, True, True)]: + with self.subTest(dc=dc, forced=forced): + reader = Mock() + writer = Mock(wait_closed=AsyncMock()) + writer.get_extra_info.return_value = None + with patch.object(proxy_config, 'force_test_dc', forced), \ + patch.object(tg_ws_proxy, 'set_sock_opts'), \ + patch.object(tg_ws_proxy, '_read_client_init', AsyncMock( + return_value=(b'', reader, writer, 'test'))), \ + patch.object(tg_ws_proxy, '_try_handshake', return_value=( + dc, False, PROTO_TAG_INTERMEDIATE, b'')), \ + patch.object(tg_ws_proxy, '_build_crypto_ctx'), \ + patch.object(tg_ws_proxy.ws_pool, 'get', AsyncMock(return_value=None)) as get, \ + patch('proxy.raw_websocket.RawWebSocket.connect', AsyncMock()) as connect, \ + patch.object(tg_ws_proxy, 'do_fallback', AsyncMock(return_value=True)) as fallback: + await tg_ws_proxy._handle_client(reader, writer, b'') + get.assert_awaited_once_with(2, False, is_test_dc=expected_test) + fallback.assert_awaited_once() + self.assertEqual(fallback.call_args.args[4:7], (2, expected_test, False)) + connect.assert_not_awaited() + async def test_cf_fallback_order_and_route_guards(self): cases = [ # media, test DC, CF enabled, Worker succeeds, H2 opens, expected diff --git a/tests/test_pool.py b/tests/test_pool.py index a62db1d..6973ee3 100644 --- a/tests/test_pool.py +++ b/tests/test_pool.py @@ -1,3 +1,4 @@ +import asyncio import time import unittest @@ -7,6 +8,8 @@ from unittest import mock from proxy.config import proxy_config from proxy.pool import _WsPool +from proxy.raw_websocket import RawWebSocket, WsHandshakeError +from proxy.utils import ws_domains class _StopRotation(Exception): @@ -15,8 +18,9 @@ class _StopRotation(Exception): def _open_ws(): transport = SimpleNamespace(is_closing=lambda: False) - writer = SimpleNamespace(transport=transport) - return SimpleNamespace(_closed=False, writer=writer) + writer = mock.Mock(transport=transport, drain=mock.AsyncMock(), + wait_closed=mock.AsyncMock()) + return RawWebSocket(asyncio.StreamReader(), writer) class WsPoolRotationTest(unittest.IsolatedAsyncioTestCase): @@ -35,23 +39,18 @@ class WsPoolRotationTest(unittest.IsolatedAsyncioTestCase): with mock.patch.object(pool, '_schedule_refill') as refill: await pool.warmup() keys = [call.args[0] for call in refill.call_args_list] - self.assertIn((2, False), keys) - self.assertEqual((2, True) in keys, media) + self.assertIn((2, False, False), keys) + self.assertEqual((2, True, False) in keys, media) async def test_refills_partially_populated_bucket(self): pool = _WsPool() - key = (2, False) + key = (2, False, False) pool._idle[key] = deque([ (_open_ws(), time.monotonic()), (_open_ws(), time.monotonic()), ]) - sleep_calls = 0 - async def stop_after_one_iteration(_delay): - nonlocal sleep_calls - sleep_calls += 1 - if sleep_calls > 1: - raise _StopRotation + raise _StopRotation with mock.patch.object(proxy_config, 'pool_size', 4): with mock.patch( @@ -67,5 +66,180 @@ class WsPoolRotationTest(unittest.IsolatedAsyncioTestCase): key, '149.154.167.220', ['example.com']) +class WsPoolTest(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self): + self.pool = _WsPool() + self.key = (2, False, False) + self.pool.WS_POOL_CHECK_INTERVAL = .01 + for name, value in [('dc_redirects', {2: '192.0.2.1'}), + ('pool_size', 1), ('force_test_dc', False)]: + patcher = mock.patch.object(proxy_config, name, value) + patcher.start() + self.addCleanup(patcher.stop) + self.addAsyncCleanup(self.pool.close) + + async def wait_for(self, predicate): + async def wait(): + while not predicate(): + await asyncio.sleep(.001) + await asyncio.wait_for(wait(), 1) + + async def test_regular_sessions_never_use_media_domain(self): + for dc in (2, 4): + with self.subTest(dc=dc), mock.patch( + 'proxy.pool.RawWebSocket.connect', + side_effect=WsHandshakeError(302, 'redirect')) as connect: + self.assertIsNone(await self.pool._connect_one('192.0.2.1', ws_domains(dc, False))) + self.assertEqual([call.args[1] for call in connect.call_args_list], + [f'kws{dc}.web.telegram.org'] * 2) + + async def test_media_tries_regular_domain_after_both_modes_fail(self): + for error in (asyncio.TimeoutError(), ConnectionResetError(), + WsHandshakeError(302, 'redirect'), WsHandshakeError(500, 'error')): + with self.subTest(error=type(error)), mock.patch( + 'proxy.pool.RawWebSocket.connect', + side_effect=[error, error, _open_ws()]) as connect: + self.assertIsNotNone(await self.pool._connect_one('192.0.2.1', ws_domains(2, True))) + self.assertEqual([call.args[1] for call in connect.call_args_list], + ['kws2-1.web.telegram.org'] * 2 + ['kws2.web.telegram.org']) + + async def test_concurrent_preference_change_does_not_skip_fronting(self): + ws = _open_ws() + + async def connect(*args, **kwargs): + if kwargs['sni'] is None: + self.pool.try_fronting_first = True + raise asyncio.TimeoutError() + return ws + + with mock.patch('proxy.pool.RawWebSocket.connect', side_effect=connect) as dial: + self.assertIs(await self.pool._connect_one('192.0.2.1', ws_domains(2, False)), ws) + self.assertEqual(dial.await_count, 2) + + async def test_miss_returns_without_waiting_and_starts_only_one_refill(self): + started = asyncio.Event() + + async def connect(*args): + started.set() + await asyncio.Future() + + with mock.patch.object(self.pool, '_connect_one', side_effect=connect) as dial: + for _ in range(10): + self.assertIsNone(await self.pool.get(2, False)) + await asyncio.wait_for(started.wait(), 1) + self.assertEqual(dial.await_count, 1) + await self.pool.close() + + async def test_empty_pool_recovers_in_background_after_backoff(self): + ws = _open_ws() + self.pool.REFILL_BACKOFF_INITIAL = .02 + with mock.patch.object(self.pool, '_connect_one', + side_effect=[None, ws]) as dial: + self.assertIsNone(await self.pool.get(2, False)) + await self.wait_for(lambda: self.key in self.pool._refill_after) + self.assertEqual(dial.await_count, 1) + # No new client requests are needed to resume refilling. + await self.wait_for(lambda: bool(self.pool._idle[self.key])) + self.assertEqual(dial.await_count, 2) + self.assertNotIn(self.key, self.pool._refill_failures) + await self.pool.close() + + async def test_ready_connection_is_available_before_slower_attempt(self): + ready = _open_ws() + slow_started = asyncio.Event() + calls = 0 + + async def connect(*args): + nonlocal calls + calls += 1 + if calls == 1: + slow_started.set() + await asyncio.Future() + return ready + + with mock.patch.object(proxy_config, 'pool_size', 2), \ + mock.patch.object(self.pool, '_connect_one', side_effect=connect): + await self.pool.get(2, False) + await asyncio.wait_for(slow_started.wait(), 1) + await self.wait_for(lambda: bool(self.pool._idle[self.key])) + self.assertIs(await self.pool.get(2, False), ready) + await self.pool.close() + + async def test_stale_connections_are_skipped(self): + for state in ('expired', 'eof', 'exception', 'closed', 'closing'): + with self.subTest(state=state): + stale, good = _open_ws(), _open_ws() + created = time.monotonic() + if state == 'expired': + created -= self.pool.WS_POOL_MAX_AGE + elif state == 'eof': + stale.reader.feed_eof() + elif state == 'exception': + stale.reader.set_exception(ConnectionResetError()) + elif state == 'closed': + stale._closed = True + else: + stale.writer.transport.is_closing = lambda: True + self.pool._idle[self.key] = deque([(stale, created), (good, time.monotonic())]) + with mock.patch.object(self.pool, '_schedule_refill'): + self.assertIs(await self.pool.get(2, False), good) + await asyncio.sleep(0) + + async def test_hit_does_not_reset_failed_refill_backoff(self): + self.pool._idle[self.key] = deque([(_open_ws(), time.monotonic())]) + self.pool._refill_failures[self.key] = 3 + self.pool._refill_after[self.key] = time.monotonic() + 60 + with mock.patch.object(self.pool, '_schedule_rotation'): + self.assertIsNotNone(await self.pool.get(2, False)) + self.assertEqual(self.pool._refill_failures[self.key], 3) + self.assertFalse(self.pool._refilling) + + async def test_zero_size_and_unconfigured_dc_do_not_connect(self): + with mock.patch.object(self.pool, '_connect_one') as dial: + self.assertIsNone(await self.pool.get(4, False)) + with mock.patch.object(proxy_config, 'pool_size', 0): + await self.pool.warmup() + self.assertIsNone(await self.pool.get(2, False)) + self.assertFalse(self.pool._rotating) + self.assertFalse(self.pool._refilling) + dial.assert_not_called() + + async def test_test_dc_has_separate_pool_and_path(self): + prod = _open_ws() + self.pool._idle[self.key] = deque([(prod, time.monotonic())]) + with mock.patch.object(self.pool, '_connect_one', return_value=_open_ws()) as dial: + self.assertIsNone(await self.pool.get(2, False, is_test_dc=True)) + test_key = (2, False, True) + await self.wait_for(lambda: bool(self.pool._idle.get(test_key))) + dial.assert_awaited_once_with('192.0.2.1', ['kws2.web.telegram.org'], '/apiws_test') + self.assertIs(self.pool._idle[self.key][0][0], prod) + await self.pool.close() + + async def test_close_cancels_refill_and_closes_unclaimed_connections(self): + ready = _open_ws() + idle = _open_ws() + started = asyncio.Event() + self.pool._idle[self.key] = deque([(idle, time.monotonic())]) + + async def connect(*args): + started.set() + try: + await asyncio.Future() + except asyncio.CancelledError: + # Completion raced with shutdown; it must not leak into the pool. + return ready + + with mock.patch.object(proxy_config, 'pool_size', 2), \ + mock.patch.object(self.pool, '_connect_one', side_effect=connect): + self.pool._schedule_refill(self.key, '192.0.2.1', ws_domains(2, False)) + await asyncio.wait_for(started.wait(), 1) + await self.pool.close() + self.assertFalse(self.pool._idle) + self.assertFalse(self.pool._rotating) + self.assertFalse(self.pool._refilling) + ready.writer.close.assert_called_once() + idle.writer.close.assert_called_once() + + if __name__ == '__main__': unittest.main() diff --git a/tests/test_raw_websocket.py b/tests/test_raw_websocket.py index 38f6be2..c1bcb1f 100644 --- a/tests/test_raw_websocket.py +++ b/tests/test_raw_websocket.py @@ -1,5 +1,6 @@ import asyncio import unittest +from unittest.mock import AsyncMock, Mock, patch from proxy.raw_websocket import RawWebSocket, WsHandshakeError, _xor_mask @@ -126,5 +127,38 @@ class HandshakeErrorTest(unittest.TestCase): self.assertFalse(WsHandshakeError(code, '').is_redirect) +class HandshakeCleanupTest(unittest.IsolatedAsyncioTestCase): + async def test_cancelling_pending_upgrade_closes_socket(self): + reader = asyncio.StreamReader() + writer = Mock(drain=AsyncMock()) + started = asyncio.Event() + + async def drain(): + started.set() + + writer.drain.side_effect = drain + with patch('proxy.raw_websocket.asyncio.open_connection', + AsyncMock(return_value=(reader, writer))), \ + patch('proxy.raw_websocket.set_sock_opts'): + task = asyncio.create_task(RawWebSocket.connect('192.0.2.1', 'example.org')) + await asyncio.wait_for(started.wait(), 1) + task.cancel() + with self.assertRaises(asyncio.CancelledError): + await task + writer.close.assert_called_once() + + async def test_failed_upgrade_closes_socket(self): + for error in (ConnectionResetError(), asyncio.TimeoutError()): + with self.subTest(error=type(error)): + reader = Mock(readline=AsyncMock(side_effect=error)) + writer = Mock(drain=AsyncMock()) + with patch('proxy.raw_websocket.asyncio.open_connection', + AsyncMock(return_value=(reader, writer))), \ + patch('proxy.raw_websocket.set_sock_opts'): + with self.assertRaises(type(error)): + await RawWebSocket.connect('192.0.2.1', 'example.org') + writer.close.assert_called_once() + + if __name__ == '__main__': unittest.main() diff --git a/tests/test_tcp_fallback.py b/tests/test_tcp_fallback.py new file mode 100644 index 0000000..cce4c7e --- /dev/null +++ b/tests/test_tcp_fallback.py @@ -0,0 +1,109 @@ +import asyncio +import time +import unittest +from unittest.mock import AsyncMock, Mock, patch + +from proxy import bridge + + +class TcpFallbackTest(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self): + bridge.reset_tcp_backoff() + self.addCleanup(bridge.reset_tcp_backoff) + self.key = ('192.0.2.1', 443) + self.remote = Mock(drain=AsyncMock(), wait_closed=AsyncMock()) + + async def fallback(self, dst='192.0.2.1', port=443): + return await bridge._tcp_fallback( + Mock(), Mock(), dst, port, b'init', 'test', Mock()) + + async def test_failure_backoff_doubles_caps_and_skips_new_connections(self): + with patch.object(bridge.asyncio, 'open_connection', + AsyncMock(side_effect=asyncio.TimeoutError())) as connect: + for delay in (30, 60, 120, 240, 480, 960, 1920, 3600, 3600): + bridge._tcp_retry_after[self.key] = 0 + start = time.monotonic() + self.assertFalse(await self.fallback()) + remaining = bridge._tcp_retry_after[self.key] - start + self.assertGreaterEqual(remaining, delay - .001) + self.assertLess(remaining, delay + 1) + count = connect.await_count + self.assertFalse(await self.fallback()) + self.assertEqual(connect.await_count, count) + self.assertFalse(bridge._tcp_connecting) + + async def test_other_ips_and_ports_are_not_blocked(self): + bridge._tcp_retry_after[self.key] = time.monotonic() + 60 + with patch.object(bridge.asyncio, 'open_connection', AsyncMock( + return_value=(Mock(), self.remote))) as connect, \ + patch.object(bridge, '_bridge_tcp_reencrypt', AsyncMock()): + self.assertTrue(await self.fallback('192.0.2.2')) + self.assertTrue(await self.fallback(port=80)) + self.assertEqual(connect.await_count, 2) + + async def test_success_clears_backoff_and_allows_following_sessions(self): + bridge._tcp_failures[self.key] = 5 + bridge._tcp_retry_after[self.key] = 0 + with patch.object(bridge.asyncio, 'open_connection', AsyncMock( + return_value=(Mock(), self.remote))) as connect, \ + patch.object(bridge, '_bridge_tcp_reencrypt', AsyncMock()): + self.assertTrue(await self.fallback()) + self.assertNotIn(self.key, bridge._tcp_failures) + self.assertNotIn(self.key, bridge._tcp_retry_after) + self.assertTrue(await self.fallback()) + self.assertEqual(connect.await_count, 2) + + async def test_concurrent_requests_only_start_one_connection_attempt(self): + started = asyncio.Event() + release = asyncio.Event() + + async def connect(*args): + started.set() + await release.wait() + raise ConnectionRefusedError() + + with patch.object(bridge.asyncio, 'open_connection', side_effect=connect) as dial: + first = asyncio.create_task(self.fallback()) + try: + await asyncio.wait_for(started.wait(), 1) + results = await asyncio.gather(*(self.fallback() for _ in range(8))) + self.assertEqual(results, [False] * 8) + self.assertEqual(dial.await_count, 1) + release.set() + self.assertFalse(await first) + finally: + first.cancel() + await asyncio.gather(first, return_exceptions=True) + + async def test_init_failure_closes_socket_and_enters_backoff(self): + self.remote.drain.side_effect = ConnectionResetError() + with patch.object(bridge.asyncio, 'open_connection', AsyncMock( + return_value=(Mock(), self.remote))), \ + patch.object(bridge, '_bridge_tcp_reencrypt', AsyncMock()) as forward: + self.assertFalse(await self.fallback()) + self.remote.close.assert_called_once() + forward.assert_not_awaited() + self.assertIn(self.key, bridge._tcp_retry_after) + + async def test_cancellation_releases_attempt_without_marking_ip_blocked(self): + started = asyncio.Event() + + async def drain(): + started.set() + await asyncio.Future() + + self.remote.drain.side_effect = drain + with patch.object(bridge.asyncio, 'open_connection', AsyncMock( + return_value=(Mock(), self.remote))): + task = asyncio.create_task(self.fallback()) + await asyncio.wait_for(started.wait(), 1) + task.cancel() + with self.assertRaises(asyncio.CancelledError): + await task + self.remote.close.assert_called_once() + self.assertFalse(bridge._tcp_connecting) + self.assertFalse(bridge._tcp_failures) + + +if __name__ == '__main__': + unittest.main() diff --git a/ui/i18n/en.json b/ui/i18n/en.json index 22a311f..7dbda52 100644 --- a/ui/i18n/en.json +++ b/ui/i18n/en.json @@ -49,7 +49,7 @@ "tip.verbose": "When enabled, more details are written to the log file — useful for troubleshooting", "tip.no_secure": "When enabled, the proxy will use port 80 for CF-proxy and CF-worker connections", "tip.buf_kb": "Receive/send buffer size in kilobytes.\nA larger value allocates more memory per socket", - "tip.pool": "How many parallel WebSocket sessions per data center can be kept open.\nIncreasing may help under high load", + "tip.pool": "How many ready WebSocket connections to keep per data center.\n0 disables the direct DC->IP WebSocket route", "tip.log_mb": "Maximum log file size; the file is overwritten when the limit is reached", "tip.autostart": "Launch TG WS Proxy on Windows login. If you move the app to another folder, autostart will reset", "tip.check_updates": "Check for updates on startup", diff --git a/ui/i18n/ru.json b/ui/i18n/ru.json index 07139ef..e58c0f5 100644 --- a/ui/i18n/ru.json +++ b/ui/i18n/ru.json @@ -49,7 +49,7 @@ "tip.verbose": "Если включено, в файл логов пишется больше подробностей — необходимо при поиске неполадок", "tip.no_secure": "Если включено, прокси будет использовать 80 порт для CF-прокси и CF-worker подключений", "tip.buf_kb": "Размер буфера приёма/передачи в килобайтах.\nБольше значение — больше выделение памяти на сокет", - "tip.pool": "Сколько параллельных WebSocket-сессий к одному датацентру можно держать.\nУвеличение может помочь при высокой нагрузке", + "tip.pool": "Сколько готовых WebSocket-соединений держать для каждого датацентра.\n0 отключает прямой DC->IP WebSocket-маршрут", "tip.log_mb": "Максимальный размер файла лога; при достижении лимита файл перезаписывается", "tip.autostart": "Запускать TG WS Proxy при входе в Windows. Если вы переместите программу в другую папку, автозапуск сбросится", "tip.check_updates": "При запуске проверять наличие обновлений",