adds bugfix, hydration , pools , connections, tests, and more

Author: Josephrp

.environement.md

@@ -694,6 +694,9 @@ def _get_env_config(self) -> dict[str, Any]:
             "CCBT_XET_MULTICAST_ADDRESS": "network.xet_multicast_address",
             "CCBT_XET_MULTICAST_PORT": "network.xet_multicast_port",
             "CCBT_PIPELINE_DEPTH": "network.pipeline_depth",
+            "CCBT_SPARSE_PIPELINE_STALE_PAYLOAD_CANCEL_S": (
+                "network.sparse_pipeline_stale_payload_cancel_s"
+            ),
             "CCBT_BLOCK_SIZE_KIB": "network.block_size_kib",
             "CCBT_CONNECTION_TIMEOUT": "network.connection_timeout",
             "CCBT_HANDSHAKE_TIMEOUT": "network.handshake_timeout",
@@ -704,6 +707,10 @@ def _get_env_config(self) -> dict[str, Any]:
             "CCBT_HANDSHAKE_TIMEOUT_HEALTHY_MIN": "network.handshake_timeout_healthy_min",
             "CCBT_HANDSHAKE_TIMEOUT_HEALTHY_MAX": "network.handshake_timeout_healthy_max",
             "CCBT_HANDSHAKE_ADAPTIVE_TIMEOUT_ENABLED": "network.handshake_adaptive_timeout_enabled",
+            # false = deprecated legacy (always max timeout in desperation band)
+            "CCBT_HANDSHAKE_TIMEOUT_DESPERATION_INTERPOLATE": (
+                "network.handshake_timeout_desperation_interpolate"
+            ),
             "CCBT_ADAPTIVE_TIMEOUT_HEALTH_PEER_SOURCE": (
                 "network.adaptive_timeout_health_peer_source"
             ),
@@ -716,12 +723,19 @@ def _get_env_config(self) -> dict[str, Any]:
             "CCBT_METADATA_EXCHANGE_TIMEOUT": "network.metadata_exchange_timeout",
             "CCBT_PEER_QUALITY_PROBATION_TIMEOUT": "network.peer_quality_probation_timeout",
             "CCBT_METADATA_PIECE_TIMEOUT": "network.metadata_piece_timeout",
+            "CCBT_BITFIELD_HAVE_WAIT_TIMEOUT_S": "network.bitfield_have_wait_timeout_s",
+            "CCBT_BITFIELD_HAVE_WAIT_METADATA_INCOMPLETE_MULTIPLIER": (
+                "network.bitfield_have_wait_metadata_incomplete_multiplier"
+            ),
             "CCBT_CONNECTION_HEALTH_CHECK_INTERVAL": "network.connection_health_check_interval",
             "CCBT_CONNECTION_VALIDATION_ENABLED": "network.connection_validation_enabled",
             "CCBT_PEER_VALIDATION_ENABLED": "network.peer_validation_enabled",
             "CCBT_SEND_BITFIELD_AFTER_METADATA": "network.send_bitfield_after_metadata",
             "CCBT_SEND_INTERESTED_AFTER_METADATA": "network.send_interested_after_metadata",
             "CCBT_MAX_CONCURRENT_CONNECTION_ATTEMPTS": "network.max_concurrent_connection_attempts",
+            "CCBT_CONNECT_TO_PEERS_PARALLEL_BATCHES": (
+                "network.connect_to_peers_parallel_batches"
+            ),
             "CCBT_MSE_INITIATOR_TIMEOUT_SCALE_ZERO_ACTIVE": (
                 "network.mse_initiator_timeout_scale_zero_active"
             ),
@@ -981,6 +995,28 @@ def _get_env_config(self) -> dict[str, Any]:
             "CCBT_TRACKER_PEER_COUNT_WEIGHT": "discovery.tracker_peer_count_weight",
             "CCBT_TRACKER_PERFORMANCE_WEIGHT": "discovery.tracker_performance_weight",
             "CCBT_DEFAULT_TRACKERS": "discovery.default_trackers",
+            "CCBT_TRACKER_UDP_PENDING_SOFT_CAP_PER_HOST": (
+                "discovery.tracker_udp_pending_soft_cap_per_host"
+            ),
+            "CCBT_TRACKER_UDP_MAX_PENDING_REQUESTS": (
+                "discovery.tracker_udp_max_pending_requests"
+            ),
+            "CCBT_TRACKER_UDP_WAIT_PACING_LOAD_RATIO": (
+                "discovery.tracker_udp_wait_pacing_load_ratio"
+            ),
+            "CCBT_TRACKER_INGRESS_HOLD_PENDING_QUEUE_THRESHOLD": (
+                "discovery.tracker_ingress_hold_pending_queue_threshold"
+            ),
+            # Deprecated to set false: legacy peer ordering; default true is supported path.
+            "CCBT_STRICT_TRACKER_SOURCE_CONNECT_PRIORITY": (
+                "discovery.strict_tracker_source_connect_priority"
+            ),
+            "CCBT_STRICT_TRACKER_PENDING_DHT_PEX_BOOST": (
+                "discovery.strict_tracker_pending_dht_pex_boost"
+            ),
+            "CCBT_STRICT_TRACKER_PENDING_TRACKER_PREFIX": (
+                "discovery.strict_tracker_pending_tracker_prefix"
+            ),
             "CCBT_PEX_INTERVAL": "discovery.pex_interval",
             "CCBT_STRICT_PRIVATE_MODE": "discovery.strict_private_mode",
             # BEP 32: IPv6 Extension for DHT
@@ -1074,6 +1110,13 @@ def _get_env_config(self) -> dict[str, Any]:
             "CCBT_TRACKER_IMMEDIATE_PER_SOURCE_CAP_MODE": (
                 "discovery.tracker_immediate_per_source_cap_mode"
             ),
+            "CCBT_TRACKER_IMMEDIATE_PER_TRACKER_COOLDOWN_ENABLED": (
+                "discovery.tracker_immediate_per_tracker_cooldown_enabled"
+            ),
+            "CCBT_MAX_TRACKER_URLS_PER_TORRENT": "discovery.max_tracker_urls_per_torrent",
+            "CCBT_ANNOUNCE_MAX_TRACKERS_PER_ROUND": (
+                "discovery.announce_max_trackers_per_round"
+            ),
             # XET chunk discovery
             "CCBT_XET_CHUNK_QUERY_BATCH_SIZE": "discovery.xet_chunk_query_batch_size",
             "CCBT_XET_CHUNK_QUERY_MAX_CONCURRENT": "discovery.xet_chunk_query_max_concurrent",
@@ -1378,6 +1421,42 @@ def export(self, fmt: str = "toml", encrypt_passwords: bool = True) -> str:
         msg = f"Unsupported export format: {fmt}"  # pragma: no cover
         raise ConfigurationError(msg)  # pragma: no cover
 
+    def get_runtime_env_diagnostics(self) -> dict[str, Any]:
+        """Return runtime env + dotenv provenance diagnostics for support reports."""
+        import os
+
+        return {
+            "dotenv_loader_requested": str(os.getenv("CCBT_LOAD_DOTENV", "")).strip(),
+            "dotenv_loaded": str(os.getenv("CCBT_DOTENV_LOADED", "0")).strip(),
+            "dotenv_path_effective": str(
+                os.getenv("CCBT_DOTENV_PATH_EFFECTIVE", "")
+            ).strip(),
+            "dotenv_keys_loaded": str(
+                os.getenv("CCBT_DOTENV_KEYS_LOADED", "0")
+            ).strip(),
+            "max_peers_per_torrent_effective": int(
+                getattr(self.config.network, "max_peers_per_torrent", 0) or 0
+            ),
+            "tracker_immediate_connect_burst_total": int(
+                getattr(
+                    self.config.discovery,
+                    "tracker_immediate_connect_burst_total",
+                    0,
+                )
+                or 0
+            ),
+            "tracker_immediate_per_tracker_cooldown_enabled": bool(
+                getattr(
+                    self.config.discovery,
+                    "tracker_immediate_per_tracker_cooldown_enabled",
+                    True,
+                )
+            ),
+            "target_requestable_peers": int(
+                getattr(self.config.discovery, "target_requestable_peers", 0) or 0
+            ),
+        }
+
     def save_config(self) -> None:
         """Save current configuration to file.
 

ccbt/config/config.py

@@ -123,6 +123,7 @@ def maybe_load_dotenv_from_env() -> None:
     """If ``CCBT_LOAD_DOTENV`` is truthy, merge ``.env`` into the process environment."""
     raw_flag = os.getenv(_LOAD_DOTENV_FLAG)
     if not _truthy_env(raw_flag):
+        os.environ.setdefault("CCBT_DOTENV_LOADED", "0")
         logger.debug(
             "Skipping dotenv load: %s is not truthy (value=%r)",
             _LOAD_DOTENV_FLAG,
@@ -133,6 +134,9 @@ def maybe_load_dotenv_from_env() -> None:
     raw_path = os.getenv(_DOTENV_PATH_VAR)
     path = Path(raw_path).expanduser() if raw_path else Path.cwd() / ".env"
     n = load_dotenv_file(path)
+    os.environ["CCBT_DOTENV_LOADED"] = "1"
+    os.environ["CCBT_DOTENV_PATH_EFFECTIVE"] = str(path)
+    os.environ["CCBT_DOTENV_KEYS_LOADED"] = str(int(n))
     logger.debug(
         "Loaded %d variable(s) from dotenv (path=%s, CCBT_LOAD_DOTENV set)",
         n,

ccbt/config/env_bootstrap.py

@@ -0,0 +1,87 @@
+"""Deduplicate tracker announce URLs that target the same host:port endpoint.
+
+Multiple schemes (https/http/udp) to the same endpoint create redundant announces
+and multiply load on the shared UDP tracker client. We keep the highest-priority
+scheme per endpoint while preserving first-seen ordering of endpoints.
+"""
+
+from __future__ import annotations
+
+from typing import Optional, Tuple
+from urllib.parse import urlparse
+
+# Prefer TLS-capable HTTP when the same host:port is reachable as both.
+_SCHEME_PRIORITY: dict[str, int] = {
+    "https": 3,
+    "http": 2,
+    "udp": 1,
+}
+
+
+def _default_port_for_scheme(scheme: str) -> Optional[int]:
+    s = scheme.lower()
+    if s == "http":
+        return 80
+    if s == "https":
+        return 443
+    return None
+
+
+def tracker_endpoint_key(url: str) -> Optional[Tuple[str, int]]:
+    """Return (host_lower, port) for deduplication, or None if not dedupeable."""
+    try:
+        parsed = urlparse(url.strip())
+        host = (parsed.hostname or "").lower()
+        if not host:
+            return None
+        port = parsed.port
+        if port is None:
+            port = _default_port_for_scheme(parsed.scheme or "")
+        if port is None:
+            return None
+        return (host, int(port))
+    except (TypeError, ValueError):
+        return None
+
+
+def dedupe_tracker_urls_by_host_port(urls: list[str]) -> list[str]:
+    """Collapse URLs that share the same host:port, preferring https > http > udp.
+
+    Order: first occurrence of each endpoint in ``urls`` defines output position.
+    Unparseable URLs are appended in original order (string-deduped).
+    """
+    if not urls:
+        return []
+
+    best_by_endpoint: dict[tuple[str, int], tuple[int, str]] = {}
+    order: list[tuple[str, int]] = []
+    seen_ep: set[tuple[str, int]] = set()
+    unparsed: list[str] = []
+    unparsed_seen: set[str] = set()
+
+    for raw in urls:
+        if not isinstance(raw, str):
+            continue
+        u = raw.strip()
+        if not u:
+            continue
+        key = tracker_endpoint_key(u)
+        if key is None:
+            if u not in unparsed_seen:
+                unparsed_seen.add(u)
+                unparsed.append(u)
+            continue
+        scheme = (urlparse(u).scheme or "").lower()
+        pri = _SCHEME_PRIORITY.get(scheme, 0)
+        if key not in seen_ep:
+            seen_ep.add(key)
+            order.append(key)
+            best_by_endpoint[key] = (pri, u)
+        else:
+            old_pri, _old_url = best_by_endpoint[key]
+            if pri > old_pri:
+                best_by_endpoint[key] = (pri, u)
+
+    out = [best_by_endpoint[k][1] for k in order]
+    out.extend(unparsed)
+    return out

ccbt/discovery/tracker_dedupe.py

@@ -30,6 +30,7 @@
 from urllib.parse import urlparse
 
 from ccbt.config.config import get_config
+from ccbt.session.peer_discovery_telemetry import observe_udp_tracker_pending_window
 
 if TYPE_CHECKING:
     from ccbt.models import PeerInfo
@@ -166,6 +167,11 @@ def __init__(self, peer_id: Optional[bytes] = None, test_mode: bool = False):
         self._timeout_warning_host_state: dict[str, tuple[float, int]] = {}
         self._tracker_response_timeout_alpha: float = 0.25
         self._tracker_response_timeout_ema: dict[str, float] = {}
+        self._tracker_timeout_floor_scale: dict[str, float] = {}
+        self._pending_request_host_by_tid: dict[int, str] = {}
+        self._pending_request_soft_cap_per_host: int = 24
+        self._udp_wait_pacing_load_ratio: float = 0.5
+        self._last_udp_pending_gauge_monotonic: float = 0.0
 
         # Background tasks
         self._cleanup_task: Optional[asyncio.Task] = None
@@ -195,6 +201,7 @@ def __init__(self, peer_id: Optional[bytes] = None, test_mode: bool = False):
         self._xet_chunk_registry: dict[tuple[bytes, Optional[str]], list[PeerInfo]] = {}
 
         self.logger = logging.getLogger(__name__)
+        self._refresh_udp_pending_settings_from_config()
         if not test_mode and not _udp_singleton_construct_in_progress():
             msg = (
                 "AsyncUDPTrackerClient must be obtained via get_udp_tracker_client() "
@@ -203,6 +210,29 @@ def __init__(self, peer_id: Optional[bytes] = None, test_mode: bool = False):
             )
             raise RuntimeError(msg)
 
+    def _refresh_udp_pending_settings_from_config(self) -> None:
+        """Apply discovery.* limits for the process-wide UDP tracker singleton."""
+        disc = getattr(self.config, "discovery", None)
+        if disc is None:
+            return
+        with contextlib.suppress(Exception):
+            self._pending_request_soft_cap_per_host = int(
+                getattr(disc, "tracker_udp_pending_soft_cap_per_host", 24)
+            )
+            self._max_pending_requests = int(
+                getattr(disc, "tracker_udp_max_pending_requests", 128)
+            )
+            self._udp_wait_pacing_load_ratio = float(
+                getattr(disc, "tracker_udp_wait_pacing_load_ratio", 0.5)
+            )
+
+    def _maybe_emit_udp_pending_gauge(self) -> None:
+        now = time.monotonic()
+        if now - self._last_udp_pending_gauge_monotonic < 0.25:
+            return
+        self._last_udp_pending_gauge_monotonic = now
+        observe_udp_tracker_pending_window(len(self.pending_requests))
+
     @property
     def socket_ready(self) -> bool:
         """Check if socket is ready.
@@ -380,10 +410,16 @@ def _get_adaptive_wait_timeout(
         queue_pressure = pending_count / effective_cap if effective_cap > 0 else 0.0
         queue_pressure = max(0.0, min(1.0, queue_pressure))
 
-        # When there is queue pressure, reduce per-request timeout to avoid long stalls.
-        queue_scale = 1.0 - (0.45 * queue_pressure)
-
+        # Queue pressure scaling with congestion floor + hysteresis.
+        # Slightly steeper than legacy 0.45 to shorten waits under multiplex load.
+        queue_scale = 1.0 - (0.55 * queue_pressure)
         host_key = self._get_tracker_host(tracker_host)
+        previous_floor = float(self._tracker_timeout_floor_scale.get(host_key, 0.65))
+        target_floor = 0.65 if queue_pressure < 0.7 else 0.8
+        floor_scale = (0.85 * previous_floor) + (0.15 * target_floor)
+        self._tracker_timeout_floor_scale[host_key] = floor_scale
+        queue_scale = max(floor_scale, queue_scale)
+
         host_ema = self._tracker_response_timeout_ema.get(host_key)
         host_scale = 1.0
         if host_ema is not None and host_ema > 0:
@@ -636,6 +672,7 @@ async def start(self) -> None:
         CRITICAL: Socket must be initialized during daemon startup via start_udp_tracker_client().
         Socket recreation is not supported as it breaks session logic.
         """
+        self._refresh_udp_pending_settings_from_config()
         # Note: Assert socket should never be recreated during runtime
         # If socket is already initialized and healthy, return immediately
         # Socket recreation breaks session logic and causes WinError 10022 on Windows
@@ -2446,6 +2483,7 @@ def _prune_stale_pending_requests(
             future = self.pending_requests.pop(transaction_id, None)
             self._pending_request_timestamps.pop(transaction_id, None)
             self.pending_immediate_callbacks.pop(transaction_id, None)
+            self._pending_request_host_by_tid.pop(transaction_id, None)
             if future is not None and not future.done():
                 future.cancel()
             self._mark_stale_transaction(transaction_id, now=now)
@@ -2468,6 +2506,7 @@ def _prune_stale_pending_requests(
                 future = self.pending_requests.pop(transaction_id, None)
                 self._pending_request_timestamps.pop(transaction_id, None)
                 self.pending_immediate_callbacks.pop(transaction_id, None)
+                self._pending_request_host_by_tid.pop(transaction_id, None)
                 if future is not None and not future.done():
                     future.cancel()
                 self._mark_stale_transaction(transaction_id, now=now)
@@ -2493,6 +2532,29 @@ async def _wait_for_response(
         future = asyncio.Future()
         now = time.time()
         start_wait = now
+        host_pending = sum(
+            1 for h in self._pending_request_host_by_tid.values() if h == host
+        )
+        if host_pending >= self._pending_request_soft_cap_per_host:
+            self.logger.debug(
+                "Tracker host pending soft cap reached: host=%s pending=%d cap=%d",
+                host,
+                host_pending,
+                self._pending_request_soft_cap_per_host,
+            )
+            return None
+        effective_cap_pre = self._get_effective_pending_request_cap()
+        pending_pre = len(self.pending_requests)
+        pace_threshold = self._udp_wait_pacing_load_ratio * effective_cap_pre
+        if effective_cap_pre > 0 and pending_pre > int(pace_threshold):
+            # Pace new waits when the shared UDP client is heavily loaded so responses
+            # can drain before adding more in-flight transactions.
+            half_span = max(1.0, pace_threshold)
+            pressure = min(
+                1.0,
+                (pending_pre - pace_threshold) / half_span,
+            )
+            await asyncio.sleep(0.04 + 0.12 * pressure)
         pruned_count = self._prune_stale_pending_requests(
             now=now, timeout=timeout, additional_new=1
         )
@@ -2523,11 +2585,14 @@ async def _wait_for_response(
         )
         self.pending_requests[transaction_id] = future
         self._pending_request_timestamps[transaction_id] = now
+        self._pending_request_host_by_tid[transaction_id] = host
         if immediate_peers_callback is not None:
             self.pending_immediate_callbacks[transaction_id] = immediate_peers_callback
         else:
             self.pending_immediate_callbacks.pop(transaction_id, None)
 
+        self._maybe_emit_udp_pending_gauge()
+
         try:
             response = await asyncio.wait_for(future, timeout=adaptive_timeout)
             elapsed = time.time() - start_wait
@@ -2566,6 +2631,7 @@ async def _wait_for_response(
             self.pending_requests.pop(transaction_id, None)
             self._pending_request_timestamps.pop(transaction_id, None)
             self.pending_immediate_callbacks.pop(transaction_id, None)
+            self._pending_request_host_by_tid.pop(transaction_id, None)
             self._cleanup_stale_response_transaction_ids(now=time.time())
 
     @staticmethod