adds swarm and download improvements

ccbt.toml

@@ -1,4 +1,5 @@
 [network]
+swarm_acceleration_enabled = true
 max_global_peers = 200
 max_peers_per_torrent = 50
 pipeline_depth = 16
@@ -223,6 +224,8 @@ checkpoint_retention_days = 30
 
 [strategy]
 piece_selection = "rarest_first"
+straggler_duplication_enabled = true
+adaptive_request_timeout_enabled = true
 endgame_duplicates = 2
 endgame_threshold = 0.95
 pipeline_capacity = 4
@@ -237,7 +240,7 @@ adaptive_hybrid_phase_detection_window = 10
 
 [discovery]
 enable_dht = true
-min_peers_before_dht = 10
+min_peers_before_dht = 2
 enable_pex = true
 enable_udp_trackers = true
 enable_http_trackers = true

ccbt/config/config.py

@@ -101,10 +101,13 @@ def _optimization_profile_overlays() -> dict[
                 "piece_selection": "rarest_first",
                 "pipeline_capacity": 4,
                 "endgame_duplicates": 2,
+                "straggler_duplication_enabled": True,
+                "adaptive_request_timeout_enabled": True,
             },
             "network": {
                 "max_peers_per_torrent": 50,
                 "max_global_peers": 200,
+                "swarm_acceleration_enabled": True,
             },
             "discovery": {
                 "tracker_announce_interval": 60.0,
@@ -120,10 +123,17 @@ def _optimization_profile_overlays() -> dict[
                 "piece_selection": "bandwidth_weighted_rarest",
                 "pipeline_capacity": 8,
                 "endgame_duplicates": 3,
+                "straggler_duplication_enabled": True,
+                "straggler_max_copies": 3,
+                "duplicate_budget_fraction": 0.10,
+                "adaptive_request_timeout_enabled": True,
             },
             "network": {
                 "max_peers_per_torrent": 100,
                 "max_global_peers": 500,
+                "swarm_acceleration_enabled": True,
+                "warm_pipeline_depth": 32,
+                "ramp_productive_dial_cap": 24,
             },
             "discovery": {
                 "tracker_announce_interval": 30.0,
@@ -160,10 +170,12 @@ def _optimization_profile_overlays() -> dict[
                 "piece_selection": "rarest_first",
                 "pipeline_capacity": 2,
                 "endgame_duplicates": 1,
+                "straggler_duplication_enabled": False,
             },
             "network": {
                 "max_peers_per_torrent": 10,
                 "max_global_peers": 50,
+                "swarm_acceleration_enabled": False,
             },
             "discovery": {
                 "tracker_announce_interval": 120.0,

ccbt/discovery/tracker.py

@@ -27,6 +27,7 @@
 from ccbt.config.config import get_config
 from ccbt.core.bencode import BencodeDecoder
 from ccbt.models import PeerInfo
+from ccbt.performance.acceleration import accel_enabled
 from ccbt.security.ssl_context import CertificatePinner
 from ccbt.utils.tracker_utils import (
     tracker_url_implies_tls,
@@ -2241,9 +2242,20 @@ async def _announce_udp_sequential() -> list[Union[TrackerResponse, None]]:
             url_to_task[task] = url
 
         if udp_urls:
-            udp_task = asyncio.create_task(_announce_udp_sequential())
-            tasks.append(udp_task)
-            url_to_task[udp_task] = "udp://sequential-batch"
+            if accel_enabled(self.config):
+                # Acceleration: UDP announces are independent datagram
+                # exchanges; run them concurrently like HTTP. The UDP
+                # client already caps pending requests per host.
+                for url in udp_urls:
+                    torrent_copy = shared_torrent_data.copy()
+                    torrent_copy["announce"] = url
+                    udp_task = asyncio.create_task(_announce_single(torrent_copy))
+                    tasks.append(udp_task)
+                    url_to_task[udp_task] = url
+            else:
+                udp_task = asyncio.create_task(_announce_udp_sequential())
+                tasks.append(udp_task)
+                url_to_task[udp_task] = "udp://sequential-batch"
 
         # Wait for all announces to complete
         self.logger.debug(

ccbt/extensions/manager.py

@@ -376,6 +376,12 @@ async def handle_extension_protocol(
             self.extension_states["protocol"].error_count += 1
             self.extension_states["protocol"].last_error = str(e)
 
+    def register_pex_peer_callback(self, callback: Any) -> None:
+        """Register a callback dialed with inbound PEX peers (BEP 11)."""
+        pex_ext = self.extensions.get("pex")
+        if pex_ext is not None and hasattr(pex_ext, "add_peer_discovered_callback"):
+            pex_ext.add_peer_discovered_callback(callback)
+
     # PEX methods
     async def handle_pex_message(
         self,

ccbt/extensions/pex.py

@@ -56,6 +56,14 @@ def __init__(self):
         self.peer_flags: dict[tuple[str, int], int] = {}  # (ip, port) -> flags
         self.FLAG_PREFER_ENCRYPT = 0x01
         self.FLAG_SEED = 0x02
+        # Callbacks invoked with newly discovered peers so inbound PEX
+        # peers are dialed immediately, not only emitted as events.
+        self.peer_discovered_callbacks: list[Any] = []
+
+    def add_peer_discovered_callback(self, callback: Any) -> None:
+        """Register a sync or async callback for inbound PEX peers."""
+        if callback not in self.peer_discovered_callbacks:
+            self.peer_discovered_callbacks.append(callback)
 
     def encode_compact_peer(self, peer: PEXPeer) -> bytes:
         """Encode peer in compact format."""
@@ -389,6 +397,15 @@ def decode_pex_message(
 
     async def handle_added_peers(self, peer_id: str, peers: list[PEXPeer]) -> None:
         """Handle added peers from PEX."""
+        new_peers = [p for p in peers if p not in self.added_peers]
+        for callback in list(self.peer_discovered_callbacks):
+            if new_peers:
+                try:
+                    result = callback(list(new_peers))
+                    if result is not None and hasattr(result, "__await__"):
+                        await result
+                except Exception:
+                    pass  # A dialing callback must never break PEX parsing
         for pex_peer in peers:
             self.added_peers.add(pex_peer)
 

ccbt/models.py

@@ -808,6 +808,51 @@ class NetworkConfig(BaseModel):
         le=600.0,
         description="Base BitTorrent block request timeout in seconds",
     )
+    # Download acceleration (see ccbt/performance/acceleration.py)
+    swarm_acceleration_enabled: bool = Field(
+        default=True,
+        description="Enable acceleration: parallel raced dialing, straggler duplication, adaptive timeouts, snub detection, BEP6 allow-fast requests",
+    )
+    speculative_connect_timeout: float = Field(
+        default=5.0,
+        ge=1.0,
+        le=30.0,
+        description="First-pass connect timeout for speculative dials (retries use connection_timeout)",
+    )
+    ramp_connect_parallel_batches: int = Field(
+        default=4,
+        ge=1,
+        le=8,
+        description="Concurrent connect_to_peers batches while accelerating peer ramp-up",
+    )
+    ramp_productive_dial_cap: int = Field(
+        default=24,
+        ge=3,
+        le=50,
+        description="Parallel outbound dials while downloading but still below requestable target",
+    )
+    ramp_choked_probe_interval_s: float = Field(
+        default=2.0,
+        ge=0.5,
+        le=30.0,
+        description="Probe-dial interval when all active peers are remote-choked (replaces 15s hold)",
+    )
+    warm_pipeline_depth: int = Field(
+        default=32,
+        ge=1,
+        le=128,
+        description="Initial per-peer pipeline depth before BDP adaptation (acceleration)",
+    )
+    snub_threshold_s: float = Field(
+        default=10.0,
+        ge=1.0,
+        le=120.0,
+        description="Seconds an unchoked peer may deliver nothing (while others deliver) before being snubbed",
+    )
+    allow_fast_enabled: bool = Field(
+        default=True,
+        description="Request BEP6 allowed-fast pieces even while remote-choked",
+    )
     sparse_pipeline_stale_payload_cancel_s: float = Field(
         default=120.0,
         ge=0.0,
@@ -2547,6 +2592,33 @@ class StrategyConfig(BaseModel):
         le=32,
         description="Request pipeline capacity",
     )
+    # Download acceleration: speculative / straggler duplication
+    straggler_duplication_enabled: bool = Field(
+        default=True,
+        description="Duplicate straggler / last-blocks of a piece to other peers before endgame",
+    )
+    straggler_max_copies: int = Field(
+        default=3,
+        ge=1,
+        le=10,
+        description="Maximum simultaneous copies of one block (original + duplicates)",
+    )
+    duplicate_budget_fraction: float = Field(
+        default=0.10,
+        ge=0.0,
+        le=0.5,
+        description="Max redundant bytes as fraction of total downloaded bytes; duplication auto-disables beyond this",
+    )
+    adaptive_request_timeout_enabled: bool = Field(
+        default=True,
+        description="Use clamp(4 x peer p95 block latency, 5s, 30s) instead of fixed 60s request timeout",
+    )
+    dynamic_endgame_last_pieces: int = Field(
+        default=20,
+        ge=1,
+        le=1000,
+        description="Enter endgame when this many pieces remain, regardless of percentage threshold",
+    )
     streaming_mode: bool = Field(default=False, description="Enable streaming mode")
 
     # Advanced strategy settings
@@ -2653,11 +2725,23 @@ class DiscoveryConfig(BaseModel):
 
     enable_dht: bool = Field(default=True, description="Enable DHT")
     min_peers_before_dht: int = Field(
-        default=10,
+        default=2,
         ge=0,
         le=100,
         description="Minimum active peers before starting DHT discovery (0 = allow DHT immediately as fallback)",
     )
+    dht_max_peers_per_query_accelerated: int = Field(
+        default=100,
+        ge=10,
+        le=500,
+        description="DHT get_peers cap while requestable peers are scarce (acceleration)",
+    )
+    dht_query_interval_accelerated_s: float = Field(
+        default=5.0,
+        ge=1.0,
+        le=60.0,
+        description="DHT query interval while no requestable peers (acceleration)",
+    )
     enable_pex: bool = Field(default=True, description="Enable Peer Exchange")
     enable_udp_trackers: bool = Field(default=True, description="Enable UDP trackers")
     enable_http_trackers: bool = Field(default=True, description="Enable HTTP trackers")