fix: single client per transport, relative throughput assertion

- Use ONE Telethon client per transport mode (get_me + all downloads)
  to avoid AuthKeyDuplicatedError from multiple concurrent auth keys
- Add relative throughput check: fake-TLS must be >= 50% of obfs2
  (catches DRS regression independent of CI network speed)
- Keep absolute floor (0.5 MB/s default, configurable via env var)
- Increase timeouts for slow CI networks (20MB: 300s, 100MB: 900s)

Author: dvershinin

tests/test_direct_e2e.py

@@ -9,6 +9,9 @@
 to verify sustained data transfer works correctly — catches DRS delay
 regressions that throttle large downloads.
 
+Uses ONE client per transport mode to avoid AuthKeyDuplicatedError from
+Telegram seeing multiple connections with the same auth key.
+
 Requires the TG_STRING_SESSION environment variable (Telethon StringSession).
 Skips gracefully when the secret is absent (fork PRs, external contributors).
 
@@ -44,12 +47,12 @@
 }
 
 # Per-file download timeouts (seconds).
-DOWNLOAD_TIMEOUTS = {"1mb": 60, "20mb": 180, "100mb": 600}
+DOWNLOAD_TIMEOUTS = {"1mb": 60, "20mb": 300, "100mb": 900}
 
 # Minimum acceptable throughput (MB/s) for files >= 20 MB.
 # The DRS delay bug (v3.5.0) throttled downloads to ~2 MB/s.
-# This floor catches the regression while tolerating CI network variability.
-MIN_THROUGHPUT_MBPS = 3.0
+# Set conservatively — if even this floor is missed, something is broken.
+MIN_THROUGHPUT_MBPS = float(os.environ.get("MIN_THROUGHPUT_MBPS", "0.5"))
 
 
 def _patch_telethon_faketls():
@@ -95,9 +98,10 @@ def _writer_write_with_ccs(self, data, extra={}):
 async def _download_test_files(client, transport_label):
     """Download pre-saved test files and verify integrity.
 
-    Returns True if all files downloaded and verified successfully.
+    Returns dict of {label: throughput_mbps} for successful downloads,
+    or None for the label if that download failed.
     """
-    all_ok = True
+    throughputs = {}
     for label, info in TEST_FILES.items():
         timeout = DOWNLOAD_TIMEOUTS[label]
         print(f"  [{transport_label}] downloading {label} "
@@ -129,16 +133,17 @@ async def _download_test_files(client, transport_label):
         except asyncio.TimeoutError:
             elapsed = time.monotonic() - start
             got = len(buf.getvalue())
+            throughput = got / elapsed / 1048576 if elapsed > 0 else 0
             print(f"  [{transport_label}] {label}: FAIL — "
                   f"timed out after {elapsed:.0f}s "
                   f"({got}/{info['size']} bytes, "
-                  f"{got / elapsed / 1048576:.2f} MB/s)")
-            all_ok = False
+                  f"{throughput:.2f} MB/s)")
+            throughputs[label] = None
             continue
         except Exception as e:
             print(f"  [{transport_label}] {label}: FAIL — "
                   f"{type(e).__name__}: {e}")
-            all_ok = False
+            throughputs[label] = None
             continue
 
         elapsed = time.monotonic() - start
@@ -148,34 +153,34 @@ async def _download_test_files(client, transport_label):
         if len(data) != info["size"]:
             print(f"  [{transport_label}] {label}: FAIL — "
                   f"size mismatch: got {len(data)}, expected {info['size']}")
-            all_ok = False
+            throughputs[label] = None
             continue
 
         if actual_hash != info["sha256"]:
             print(f"  [{transport_label}] {label}: FAIL — "
                   f"SHA256 mismatch")
-            all_ok = False
+            throughputs[label] = None
             continue
 
         throughput_mb = len(data) / elapsed / 1048576
+        throughputs[label] = throughput_mb
 
-        # Assert minimum throughput for large files to catch DRS throttling.
+        # Assert minimum throughput for large files.
         if info["size"] >= 20 * 1048576 and throughput_mb < MIN_THROUGHPUT_MBPS:
             print(f"  [{transport_label}] {label}: FAIL — "
                   f"throughput {throughput_mb:.2f} MB/s < "
-                  f"{MIN_THROUGHPUT_MBPS} MB/s minimum "
-                  f"(DRS delay regression?)")
-            all_ok = False
+                  f"{MIN_THROUGHPUT_MBPS} MB/s minimum")
+            throughputs[label] = None
             continue
 
         print(f"  [{transport_label}] {label}: OK — "
-              f"{elapsed:.1f}s, {throughput_mb:.1f} MB/s")
+              f"{elapsed:.1f}s, {throughput_mb:.2f} MB/s")
 
-    return all_ok
+    return throughputs
 
 
-async def test_obfs2_get_me(host, port, secret, session_str):
-    """Connect via obfuscated2 (dd-prefix) and call get_me()."""
+async def test_obfs2_all(host, port, secret, session_str):
+    """Run all obfs2 tests with a single client connection."""
     from telethon import TelegramClient
     from telethon.network.connection import (
         ConnectionTcpMTProxyRandomizedIntermediate,
@@ -196,67 +201,35 @@ async def test_obfs2_get_me(host, port, secret, session_str):
         await asyncio.wait_for(client.connect(), timeout=30)
         if not client.is_connected():
             print("[obfs2] FAIL: client did not connect")
-            return False
+            return False, {}
 
         me = await asyncio.wait_for(client.get_me(), timeout=15)
         if me is None:
             print("[obfs2] FAIL: get_me() returned None")
-            return False
-
-        print(f"[obfs2] OK: get_me() returned user_id={me.id}")
-        return True
-    except Exception as e:
-        print(f"[obfs2] FAIL: {type(e).__name__}: {e}")
-        return False
-    finally:
-        try:
-            await client.disconnect()
-        except Exception:
-            pass
+            return False, {}
 
+        print(f"[obfs2] get_me OK: user_id={me.id}")
 
-async def test_obfs2_media_download(host, port, secret, session_str):
-    """Download pre-saved files through obfuscated2 proxy."""
-    from telethon import TelegramClient
-    from telethon.network.connection import (
-        ConnectionTcpMTProxyRandomizedIntermediate,
-    )
-    from telethon.sessions import StringSession
-
-    print(f"[obfs2-media] Connecting to {host}:{port} ...", flush=True)
-
-    client = TelegramClient(
-        StringSession(session_str),
-        api_id=API_ID,
-        api_hash=API_HASH,
-        connection=ConnectionTcpMTProxyRandomizedIntermediate,
-        proxy=(host, port, "dd" + secret),
-    )
-
-    try:
-        await asyncio.wait_for(client.connect(), timeout=30)
-        if not client.is_connected():
-            print("[obfs2-media] FAIL: client did not connect")
-            return False
-
-        return await _download_test_files(client, "obfs2-media")
+        throughputs = await _download_test_files(client, "obfs2")
+        has_failure = any(v is None for v in throughputs.values())
+        return not has_failure, throughputs
     except Exception as e:
-        print(f"[obfs2-media] FAIL: {type(e).__name__}: {e}")
-        return False
+        print(f"[obfs2] FAIL: {type(e).__name__}: {e}")
+        return False, {}
     finally:
         try:
             await client.disconnect()
         except Exception:
             pass
 
 
-async def test_faketls_get_me(host, port, secret, domain, session_str):
-    """Connect via fake-TLS (ee-prefix) and call get_me()."""
+async def test_faketls_all(host, port, secret, domain, session_str):
+    """Run all fake-TLS tests with a single client connection."""
     try:
         from TelethonFakeTLS import ConnectionTcpMTProxyFakeTLS
     except ImportError:
         print("[fake-tls] SKIP: TelethonFakeTLS not installed")
-        return True
+        return True, {}
 
     from telethon import TelegramClient
     from telethon.sessions import StringSession
@@ -279,60 +252,21 @@ async def test_faketls_get_me(host, port, secret, domain, session_str):
         await asyncio.wait_for(client.connect(), timeout=30)
         if not client.is_connected():
             print("[fake-tls] FAIL: client did not connect")
-            return False
+            return False, {}
 
         me = await asyncio.wait_for(client.get_me(), timeout=15)
         if me is None:
             print("[fake-tls] FAIL: get_me() returned None")
-            return False
-
-        print(f"[fake-tls] OK: get_me() returned user_id={me.id}")
-        return True
-    except Exception as e:
-        print(f"[fake-tls] FAIL: {type(e).__name__}: {e}")
-        return False
-    finally:
-        try:
-            await client.disconnect()
-        except Exception:
-            pass
-
+            return False, {}
 
-async def test_faketls_media_download(host, port, secret, domain, session_str):
-    """Download pre-saved files through fake-TLS proxy (exercises DRS)."""
-    try:
-        from TelethonFakeTLS import ConnectionTcpMTProxyFakeTLS
-    except ImportError:
-        print("[fake-tls-media] SKIP: TelethonFakeTLS not installed")
-        return True
+        print(f"[fake-tls] get_me OK: user_id={me.id}")
 
-    from telethon import TelegramClient
-    from telethon.sessions import StringSession
-
-    _patch_telethon_faketls()
-
-    proxy_secret = secret + domain.encode().hex()
-    print(f"[fake-tls-media] Connecting to {host}:{port} (domain={domain}) ...",
-          flush=True)
-
-    client = TelegramClient(
-        StringSession(session_str),
-        api_id=API_ID,
-        api_hash=API_HASH,
-        connection=ConnectionTcpMTProxyFakeTLS,
-        proxy=(host, port, proxy_secret),
-    )
-
-    try:
-        await asyncio.wait_for(client.connect(), timeout=30)
-        if not client.is_connected():
-            print("[fake-tls-media] FAIL: client did not connect")
-            return False
-
-        return await _download_test_files(client, "fake-tls-media")
+        throughputs = await _download_test_files(client, "fake-tls")
+        has_failure = any(v is None for v in throughputs.values())
+        return not has_failure, throughputs
     except Exception as e:
-        print(f"[fake-tls-media] FAIL: {type(e).__name__}: {e}")
-        return False
+        print(f"[fake-tls] FAIL: {type(e).__name__}: {e}")
+        return False, {}
     finally:
         try:
             await client.disconnect()
@@ -356,38 +290,39 @@ def main():
     tls_port = int(os.environ.get("DIRECT_TLS_PORT", "9443"))
     domain = os.environ.get("EE_DOMAIN", "ya.ru")
 
-    results = []
-
     print("=== Direct Mode E2E Tests ===\n")
 
-    # Test 1: obfuscated2 get_me
-    ok = asyncio.run(test_obfs2_get_me(host, obfs2_port, secret, session_str))
-    results.append(("obfs2-get_me", ok))
-    print()
-
-    # Test 2: fake-TLS get_me
-    ok = asyncio.run(
-        test_faketls_get_me(host, tls_port, secret, domain, session_str)
+    # Test 1: obfuscated2 (control — no DRS delays)
+    obfs2_ok, obfs2_tp = asyncio.run(
+        test_obfs2_all(host, obfs2_port, secret, session_str)
     )
-    results.append(("fake-tls-get_me", ok))
     print()
 
-    # Test 3: obfuscated2 media download (control — no DRS delays)
-    ok = asyncio.run(
-        test_obfs2_media_download(host, obfs2_port, secret, session_str)
+    # Test 2: fake-TLS (exercises DRS delays)
+    tls_ok, tls_tp = asyncio.run(
+        test_faketls_all(host, tls_port, secret, domain, session_str)
     )
-    results.append(("obfs2-media", ok))
-    print()
 
-    # Test 4: fake-TLS media download (exercises DRS delays)
-    ok = asyncio.run(
-        test_faketls_media_download(host, tls_port, secret, domain, session_str)
-    )
-    results.append(("fake-tls-media", ok))
+    # Compare throughputs: fake-TLS must not be dramatically slower than obfs2.
+    # If fake-TLS is < 50% of obfs2 for the same file, DRS delays are the cause.
+    print("\n=== Throughput Comparison ===")
+    comparison_ok = True
+    for label in TEST_FILES:
+        o = obfs2_tp.get(label)
+        t = tls_tp.get(label)
+        if o is None or t is None:
+            continue
+        ratio = t / o if o > 0 else 0
+        status = "OK" if ratio >= 0.5 else "FAIL"
+        if status == "FAIL":
+            comparison_ok = False
+        print(f"  {label}: obfs2={o:.2f} MB/s, fake-tls={t:.2f} MB/s, "
+              f"ratio={ratio:.0%} {status}")
 
     print("\n=== Results ===")
     all_ok = True
-    for name, ok in results:
+    for name, ok in [("obfs2", obfs2_ok), ("fake-tls", tls_ok),
+                      ("tls-vs-obfs2-ratio", comparison_ok)]:
         status = "PASS" if ok else "FAIL"
         print(f"  {name}: {status}")
         if not ok: