fronting refactoring: implemented into ws_pool

proxy/pool.py

@@ -19,7 +19,6 @@ class _WsPool:
     def __init__(self):
         self._idle: Dict[Tuple[int, bool], deque] = {}
         self._refilling: Set[Tuple[int, bool]] = set()
-        self.fronting_until: float = 0.0
 
     async def get(self, dc: int, is_media: bool,
                   target_ip: str, domains: List[str]
@@ -62,7 +61,7 @@ class _WsPool:
             if needed <= 0:
                 return
             tasks = [asyncio.create_task(
-                self._connect_one(target_ip, domains, time.monotonic() < self.fronting_until))
+                self._connect_one(target_ip, domains))
                 for _ in range(needed)]
             for t in tasks:
                 try:
@@ -77,11 +76,19 @@ class _WsPool:
             self._refilling.discard(key)
 
     @staticmethod
-    async def _connect_one(target_ip, domains, fronting_active) -> Optional[RawWebSocket]:
+    async def _connect_one(target_ip, domains) -> Optional[RawWebSocket]:
         for domain in domains:
             try:
                 return await RawWebSocket.connect(
-                    target_ip, domain, timeout=8, sni="sprinthost.ru" if fronting_active else None)
+                    target_ip, domain, timeout=8)
+            except asyncio.TimeoutError:
+                try:
+                    ws = await RawWebSocket.connect(
+                        target_ip, domain, timeout=7, sni="sprinthost.ru")
+                    stats.connections_fronting += 1
+                    return ws
+                except Exception:
+                    return None
             except WsHandshakeError as exc:
                 if exc.is_redirect:
                     continue
@@ -109,7 +116,6 @@ class _WsPool:
     def reset(self):
         self._idle.clear()
         self._refilling.clear()
-        self.fronting_until = 0.0
 
 
 class _CfWorkerPool:

proxy/tg_ws_proxy.py

@@ -36,13 +36,11 @@ log = logging.getLogger('tg-mtproto-proxy')
 IP_FAIL_COOLDOWN = 3600.0
 DC_FAIL_COOLDOWN = 60.0
 WS_FAIL_TIMEOUT = 2.0
-FRONTING_COOLDOWN = 1800.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] = {}
-fronting_until: float = 0.0
 
 
 def _try_handshake(handshake: bytes, secret: bytes) -> Optional[Tuple[int, bool, bytes, bytes]]:
@@ -250,8 +248,6 @@ def _build_crypto_ctx(client_dec_prekey_iv, secret, relay_init):
 
 
 async def _handle_client(reader, writer, secret: bytes):
-    global fronting_until
-    
     stats.connections_total += 1
     stats.connections_active += 1
     peer = writer.get_extra_info('peername')
@@ -336,7 +332,6 @@ async def _handle_client(reader, writer, secret: bytes):
             return
 
         ws_timeout = WS_FAIL_TIMEOUT if now < dc_fail_until.get(dc_key, 0) else 5.0
-        fronting_active = now < fronting_until
 
         domains = ws_domains(dc, is_media)
         ws = None
@@ -348,25 +343,6 @@ async def _handle_client(reader, writer, secret: bytes):
         if ws:
             log.info("[%s] DC%d%s -> pool hit via %s",
                      label, dc, media_tag, target)
-        elif fronting_active:
-            # TODO: Move fronting logic into bridge.py where other fallbacks are handled
-            log.info("[%s] DC%d%s -> fronting / Host %s",
-                     label, dc, media_tag, domains[0])
-            try:
-                ws = await RawWebSocket.connect(target, domains[0],
-                                                timeout=5.0, path=ws_path,
-                                                sni="sprinthost.ru")
-            except Exception as exc:
-                stats.ws_errors += 1
-                log.warning("[%s] DC%d%s fronting failed: %s",
-                            label, dc, media_tag, repr(exc))
-            if ws:
-                stats.connections_fronting += 1
-                fronting_until = now + FRONTING_COOLDOWN
-                ws_pool.fronting_until = fronting_until
-            else:
-                fronting_until = 0.0
-                ws_pool.fronting_until = 0.0
         else:
             for domain in domains:
                 url = f'wss://{domain}{ws_path}'
@@ -403,31 +379,12 @@ async def _handle_client(reader, writer, secret: bytes):
                     log.warning("[%s] DC%d%s WS connect failed: %s",
                                 label, dc, media_tag, repr(exc))
 
-        # Fronting fallback if WS timed out
-        # TODO: Move fronting logic into bridge.py where other fallbacks are handled
-        # and don't forget about WsPool fronting fallback
-        if ws is None and ws_timed_out and not fronting_active:
-            log.info("[%s] DC%d%s -> fronting fallback (Host %s)",
-                     label, dc, media_tag, domains[0])
-            try:
-                ws = await RawWebSocket.connect(target, domains[0],
-                                                timeout=5.0, path=ws_path,
-                                                sni="sprinthost.ru")
-            except Exception as exc:
-                stats.ws_errors += 1
-                log.warning("[%s] DC%d%s fronting failed: %s",
-                            label, dc, media_tag, repr(exc))
-            if ws:
-                fronting_until = now + FRONTING_COOLDOWN
-                ws_pool.fronting_until = now + FRONTING_COOLDOWN
-                stats.connections_fronting += 1
-                log.info("[%s] DC%d%s fronting OK for %ds",
-                         label, dc, media_tag, int(FRONTING_COOLDOWN))
-
         # WS failed -> fallback
         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)
@@ -502,7 +459,7 @@ _client_tasks: Set[asyncio.Task] = set()
 
 
 async def _run(stop_event: Optional[asyncio.Event] = None):
-    global _server_instance, _server_stop_event, fronting_until
+    global _server_instance, _server_stop_event
     _server_stop_event = stop_event
 
     ws_pool.reset()
@@ -511,7 +468,6 @@ async def _run(stop_event: Optional[asyncio.Event] = None):
     dc_fail_until.clear()
     ip_fail_until.clear()
     _client_tasks.clear()
-    fronting_until = 0.0
 
     if proxy_config.fallback_cfproxy:
         user = proxy_config.cfproxy_user_domains