fix: limit DRS delays to slow-start phases, add media download E2E tests

DRS inter-record delays still throttled sustained downloads in v3.5.1 —
the burst threshold (16KB) fired at batch boundaries during streaming.
Now delays only apply during phases 1+2 (first ~60 records, ~140KB).
Phase 3 (max-size records) runs at full TCP speed. The DRS counter
resets after 1s inactivity so new responses still get realistic timing.

Add multi-size media download E2E tests (1/20/100 MB) through both
obfuscated2 and fake-TLS proxy modes with SHA256 integrity verification.
These catch DRS delay regressions that throttle large transfers.

Closes #70

Author: dvershinin

CHANGELOG.md

@@ -2,6 +2,15 @@
 
 All notable changes to this project will be documented in this file.
 
+## [3.5.2] - 2026-03-28
+
+### Fixed
+- **DRS delays still throttled sustained downloads** — delays are now limited to slow-start phases only (first ~60 records, ~140KB). Once the connection reaches phase 3 (max-size TLS records), all delays are skipped. The DRS counter resets after 1s of inactivity, so new responses still get realistic slow-start timing. This fully resolves media download degradation reported in [#70](https://github.com/GetPageSpeed/MTProxy/issues/70)
+
+### Added
+- **Media download E2E tests** — CI now downloads pre-saved files (1 MB, 20 MB, 100 MB) through both obfuscated2 and fake-TLS proxy modes, verifying data integrity and throughput. Catches DRS delay regressions that throttle large transfers
+- DRS bulk transfer delay bounds test in Docker test suite
+
 ## [3.5.1] - 2026-03-27
 
 ### Fixed

net/net-tcp-drs.c

@@ -147,9 +147,11 @@ int cpu_tcp_aes_crypto_ctr128_encrypt_output_drs (connection_job_t C) /* {{{ */
 
     assert (rwm_encrypt_decrypt_to (&c->out, &c->out_p, len, T->write_aeskey, 1) == len);
 
-    /* Inter-record delay: skip during bulk transfers for full throughput */
+    /* Inter-record delay: skip during bulk transfers and sustained transfers
+       (phase 3 = max-size records = no real server adds delays here).
+       Delays only apply during slow-start phases 1+2 (~140KB, ~60 records). */
     if (drs_delays_enabled && (c->flags & C_IS_TLS) && c->out.total_bytes > 0) {
-      if (c->out.total_bytes > DRS_BURST_THRESHOLD) {
+      if (c->out.total_bytes > DRS_BURST_THRESHOLD || drs->record_index >= DRS_PHASE2_END) {
         drs_delays_skipped++;
         continue;
       }

tests/seed_test_media.py

@@ -0,0 +1,101 @@
+#!/usr/bin/env python3
+"""One-time utility: upload test files to Saved Messages for E2E download tests.
+
+Connects using the asian bot's SQLite session and uploads 1 MB, 20 MB, and
+100 MB random files.  Prints message IDs and SHA256 hashes to hard-code in
+test_direct_e2e.py.
+
+Usage:
+    python3 tests/seed_test_media.py [--session PATH]
+
+Requires: telethon
+"""
+
+import argparse
+import asyncio
+import hashlib
+import io
+import os
+import shutil
+import sys
+import tempfile
+
+# Same credentials used in test_direct_e2e.py / CI.
+API_ID = 2834
+API_HASH = "68875f756c9b437a8b916ca3de215815"
+
+DEFAULT_SESSION = os.path.expanduser("~/Projects/tgp/sessions/asian.session")
+
+TEST_FILES = [
+    ("mtproxy_test_1mb.bin", 1 * 1024 * 1024),
+    ("mtproxy_test_20mb.bin", 20 * 1024 * 1024),
+    ("mtproxy_test_100mb.bin", 100 * 1024 * 1024),
+]
+
+
+async def main(session_path: str):
+    from telethon import TelegramClient
+
+    # Copy session to a temp file to avoid locking the original.
+    tmp_dir = tempfile.mkdtemp(prefix="seed_media_")
+    tmp_session = os.path.join(tmp_dir, "session")
+    shutil.copy2(session_path, tmp_session + ".session")
+
+    client = TelegramClient(tmp_session, api_id=API_ID, api_hash=API_HASH)
+
+    try:
+        await client.connect()
+        if not await client.is_user_authorized():
+            print("ERROR: session is not authorized", file=sys.stderr)
+            sys.exit(1)
+
+        me = await client.get_me()
+        print(f"Connected as user_id={me.id} ({me.first_name})\n")
+
+        print("Uploading test files to Saved Messages...\n")
+        results = []
+
+        for name, size in TEST_FILES:
+            print(f"  Generating {name} ({size / 1024 / 1024:.0f} MB)...")
+            data = os.urandom(size)
+            sha = hashlib.sha256(data).hexdigest()
+
+            print(f"  Uploading {name}...")
+            msg = await client.send_file(
+                "me",
+                io.BytesIO(data),
+                file_name=name,
+                caption=f"MTProxy E2E test file — {size} bytes — do not delete",
+                force_document=True,
+            )
+            print(f"  Done: msg_id={msg.id}\n")
+            results.append((name, size, msg.id, sha))
+
+        print("=" * 60)
+        print("Hard-code these in tests/test_direct_e2e.py:\n")
+        print("TEST_FILES = {")
+        for name, size, msg_id, sha in results:
+            label = name.replace("mtproxy_test_", "").replace(".bin", "")
+            print(f'    "{label}": {{"msg_id": {msg_id}, "size": {size}, "sha256": "{sha}"}},')
+        print("}")
+        print()
+
+    finally:
+        await client.disconnect()
+        shutil.rmtree(tmp_dir, ignore_errors=True)
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="Seed test media for MTProxy E2E tests")
+    parser.add_argument(
+        "--session",
+        default=DEFAULT_SESSION,
+        help=f"Path to Telethon SQLite session (default: {DEFAULT_SESSION})",
+    )
+    args = parser.parse_args()
+
+    if not os.path.exists(args.session):
+        print(f"ERROR: session file not found: {args.session}", file=sys.stderr)
+        sys.exit(1)
+
+    asyncio.run(main(args.session))

tests/test_direct_e2e.py

@@ -5,6 +5,10 @@
 get_me() to verify the full data path works.  Tests both obfuscated2
 (dd-prefix) and fake-TLS (ee-prefix) transport modes.
 
+Also downloads pre-saved test files (1 MB, 20 MB, 100 MB) through the proxy
+to verify sustained data transfer works correctly — catches DRS delay
+regressions that throttle large downloads.
+
 Requires the TG_STRING_SESSION environment variable (Telethon StringSession).
 Skips gracefully when the secret is absent (fork PRs, external contributors).
 
@@ -21,13 +25,27 @@
 """
 
 import asyncio
+import hashlib
+import io
 import os
 import sys
+import time
 
 # Official Telegram macOS client credentials (public, well-known).
 API_ID = 2834
 API_HASH = "68875f756c9b437a8b916ca3de215815"
 
+# Pre-saved test files in Saved Messages of the test account (user 8326553636).
+# Seeded by tests/seed_test_media.py — do not delete from Saved Messages.
+TEST_FILES = {
+    "1mb": {"msg_id": 116459, "size": 1048576, "sha256": "2c331300901dc5a4a58fcce075212f9506da9cdb281203067760b0c2715eab59"},
+    "20mb": {"msg_id": 116462, "size": 20971520, "sha256": "30ff0930d860a323c4e9e1f31a6758a7d208f0c53e57c4ea4361b8397e277816"},
+    "100mb": {"msg_id": 116467, "size": 104857600, "sha256": "ca81cb2e62c70e488b74a347caea415989bfdaf64d47aa80e898d01974fb2ed0"},
+}
+
+# Per-file download timeouts (seconds).
+DOWNLOAD_TIMEOUTS = {"1mb": 60, "20mb": 180, "100mb": 600}
+
 
 def _patch_telethon_faketls():
     """Patch TelethonFakeTLS bugs.
@@ -69,6 +87,78 @@ def _writer_write_with_ccs(self, data, extra={}):
     tls_io.FakeTLSStreamWriter.write = _writer_write_with_ccs
 
 
+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.
+    """
+    all_ok = True
+    for label, info in TEST_FILES.items():
+        timeout = DOWNLOAD_TIMEOUTS[label]
+        print(f"  [{transport_label}] downloading {label} "
+              f"(msg_id={info['msg_id']}, timeout={timeout}s)...",
+              flush=True)
+
+        try:
+            msg = await asyncio.wait_for(
+                client.get_messages("me", ids=info["msg_id"]),
+                timeout=15,
+            )
+        except Exception as e:
+            print(f"  [{transport_label}] {label}: SKIP — "
+                  f"could not fetch message: {type(e).__name__}: {e}")
+            continue
+
+        if msg is None or msg.media is None:
+            print(f"  [{transport_label}] {label}: SKIP — "
+                  f"message {info['msg_id']} not found (test data not seeded)")
+            continue
+
+        buf = io.BytesIO()
+        start = time.monotonic()
+        try:
+            await asyncio.wait_for(
+                client.download_media(msg, file=buf),
+                timeout=timeout,
+            )
+        except asyncio.TimeoutError:
+            elapsed = time.monotonic() - start
+            got = len(buf.getvalue())
+            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
+            continue
+        except Exception as e:
+            print(f"  [{transport_label}] {label}: FAIL — "
+                  f"{type(e).__name__}: {e}")
+            all_ok = False
+            continue
+
+        elapsed = time.monotonic() - start
+        data = buf.getvalue()
+        actual_hash = hashlib.sha256(data).hexdigest()
+
+        if len(data) != info["size"]:
+            print(f"  [{transport_label}] {label}: FAIL — "
+                  f"size mismatch: got {len(data)}, expected {info['size']}")
+            all_ok = False
+            continue
+
+        if actual_hash != info["sha256"]:
+            print(f"  [{transport_label}] {label}: FAIL — "
+                  f"SHA256 mismatch")
+            all_ok = False
+            continue
+
+        throughput_mb = len(data) / elapsed / 1048576
+        print(f"  [{transport_label}] {label}: OK — "
+              f"{elapsed:.1f}s, {throughput_mb:.1f} MB/s")
+
+    return all_ok
+
+
 async def test_obfs2_get_me(host, port, secret, session_str):
     """Connect via obfuscated2 (dd-prefix) and call get_me()."""
     from telethon import TelegramClient
@@ -110,6 +200,41 @@ async def test_obfs2_get_me(host, port, secret, session_str):
             pass
 
 
+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")
+    except Exception as e:
+        print(f"[obfs2-media] 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()."""
     try:
@@ -158,6 +283,48 @@ async def test_faketls_get_me(host, port, secret, domain, session_str):
             pass
 
 
+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
+
+    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")
+    except Exception as e:
+        print(f"[fake-tls-media] FAIL: {type(e).__name__}: {e}")
+        return False
+    finally:
+        try:
+            await client.disconnect()
+        except Exception:
+            pass
+
+
 def main():
     session_str = os.environ.get("TG_STRING_SESSION", "")
     if not session_str:
@@ -178,16 +345,30 @@ def main():
 
     print("=== Direct Mode E2E Tests ===\n")
 
-    # Test 1: obfuscated2
+    # Test 1: obfuscated2 get_me
     ok = asyncio.run(test_obfs2_get_me(host, obfs2_port, secret, session_str))
-    results.append(("obfs2", ok))
+    results.append(("obfs2-get_me", ok))
     print()
 
-    # Test 2: fake-TLS
+    # Test 2: fake-TLS get_me
     ok = asyncio.run(
         test_faketls_get_me(host, tls_port, secret, domain, session_str)
     )
-    results.append(("fake-tls", ok))
+    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)
+    )
+    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))
 
     print("\n=== Results ===")
     all_ok = True