Advanced Connection Management Features: In-Depth Analysis

[yashksaini-coder]

Issue: #553
PR: #1018
Status: Implementation Complete
Date: 2025-11-24
Author: @yashksaini-coder
Review: @sumanjeet0012 @acul71 @seetadev

---

Executive Summary

This document provides a comprehensive analysis of the advanced connection management features implemented in Python libp2p, bringing feature parity with JavaScript libp2p's connection manager. The implementation introduces sophisticated connection lifecycle management, resource controls, security features, and monitoring capabilities that significantly enhance the robustness and production-readiness of the Python libp2p implementation.

Key Achievements:

• ✅ 95-100% feature parity with JavaScript libp2p connection manager
• ✅ Backward compatible - existing code continues to work
• ✅ Production-ready - comprehensive resource management and security
• ✅ Well-documented - extensive examples and documentation

---

Overview

What Changed?

The connection management system has been completely overhauled, transforming from a simple connection dictionary to a sophisticated, multi-layered connection management infrastructure.

Before (Simple Model)

# Old: Simple 1:1 peer-to-connection mapping
connections: dict[ID, INetConn]

# Limited functionality:
# - No connection limits
# - No rate limiting
# - No automatic reconnection
# - No connection pruning
# - No security controls

After (Advanced Model)

# New: Multiple connections per peer with full management
connections: dict[ID, list[INetConn]]

# Comprehensive features:
# - Connection limits and pruning
# - Rate limiting per host
# - Automatic reconnection for KEEP_ALIVE peers
# - IP allow/deny lists
# - Priority-based dial queue
# - DNS resolution
# - Address sorting and management
# - Connection metrics and monitoring
# - Load balancing across connections

Why This Matters

1. Resource Protection: Prevents connection exhaustion attacks
2. Security: IP-based access control and rate limiting
3. Reliability: Automatic reconnection for critical peers
4. Performance: Efficient connection reuse and load balancing
5. Observability: Comprehensive metrics and monitoring
6. Production Readiness: Matches industry-standard implementations

---

Architecture & Design

System Architecture

The connection management system is built on a modular architecture with clear separation of concerns:

┌────────────────────────────────────────────────────┐
│                         Swarm (Orchestrator)       │
│  - Connection lifecycle management                 │
│  - Event coordination                              │
│  - Resource manager integration                    │
└────────────────────────────────────────────────────┘
                            │
        ┌───────────────────┼───────────────────┐
        │                   │                   │
        ▼                   ▼                   ▼
┌──────────────┐   ┌──────────────┐   ┌──────────────┐
│  Dial Queue  │   │ Reconnect    │   │  Connection  │
│              │   │ Queue        │   │  Pruner      │
│ - Priority   │   │ - KEEP_ALIVE │   │ - Multi-     │
│ - Concurrency│   │ - Backoff    │   │   criteria   │
│ - Timeouts   │   │ - Retries    │   │   sorting    │
└──────────────┘   └──────────────┘   └──────────────┘
        │                   │                   │
        └───────────────────┼───────────────────┘
                            │
        ┌───────────────────┼───────────────────┐
        │                   │                   │
        ▼                   ▼                   ▼
┌──────────────┐   ┌──────────────┐   ┌──────────────┐
│  Address     │   │  Connection  │   │  Rate        │
│  Manager     │   │  Gate        │   │  Limiter     │
│ - Sorting    │   │ - Allow/Deny │   │ - Token      │
│ - Filtering  │   │ - CIDR       │   │   bucket     │
│ - DNS        │   │ - Precedence │   │ - Per-host   │
└──────────────┘   └──────────────┘   └──────────────┘

Component Responsibilities

1. Swarm (Main Orchestrator)

• Location: libp2p/network/swarm.py
• Responsibilities:
  • Coordinates all connection management components
  • Manages connection lifecycle (creation, maintenance, cleanup)
  • Handles event notifications
  • Integrates with resource manager
  • Provides public API for connection operations

2. Dial Queue

• Location: libp2p/network/dial_queue.py
• Responsibilities:
  • Priority-based dial scheduling
  • Concurrency control (max parallel dials)
  • Queue length management
  • Timeout handling
  • Deduplication

3. Reconnection Queue

• Location: libp2p/network/reconnect_queue.py
• Responsibilities:
  • Automatic reconnection for KEEP_ALIVE peers
  • Exponential backoff with jitter
  • Retry limit enforcement
  • Parallel reconnection control

4. Connection Pruner

• Location: libp2p/network/connection_pruner.py
• Responsibilities:
  • Connection selection for pruning
  • Multi-criteria sorting (tag value, streams, direction, age)
  • Allow list protection
  • Automatic pruning when limits exceeded

5. Connection Gate

• Location: libp2p/network/connection_gate.py
• Responsibilities:
  • IP allow/deny list enforcement
  • CIDR block support
  • Precedence rules (deny > allow)
  • Private IP filtering

6. Rate Limiter

• Location: libp2p/network/rate_limiter.py
• Responsibilities:
  • Token bucket algorithm
  • Per-host/IP rate limiting
  • Configurable thresholds
  • Allow list bypass

7. Address Manager

• Location: libp2p/network/address_manager.py
• Responsibilities:
  • Address sorting (transport, certified, public/private, loopback)
  • Address filtering
  • Custom sorter support
  • Integration with connection gate

8. DNS Resolver

• Location: libp2p/network/dns_resolver.py
• Responsibilities:
  • DNS4/DNS6 resolution
  • DNSADDR support (partial)
  • Result caching
  • Async resolution

---

Core Features

1. Multiple Connections Per Peer

Overview

The system now supports multiple concurrent connections to the same peer, enabling:

• Load Balancing: Distribute streams across connections
• Fault Tolerance: If one connection fails, others remain available
• Performance: Parallel stream handling

Implementation

# Connection storage changed from 1:1 to 1:many
connections: dict[ID, list[INetConn]]  # Was: dict[ID, INetConn]

# Access methods
def get_connections(peer_id: ID | None = None) -> list[INetConn]:
    """Get all connections for a peer, or all connections if peer_id is None."""
    
def get_connection(peer_id: ID) -> INetConn | None:
    """Backward-compatible: returns first connection."""
    
@property
def connections_legacy(self) -> dict[ID, INetConn]:
    """Legacy 1:1 mapping for backward compatibility."""

Load Balancing

Two strategies are supported:

1. Round-Robin (Default):

   config = ConnectionConfig(load_balancing_strategy="round_robin")
   # Distributes streams evenly across connections

2. Least-Loaded:

   config = ConnectionConfig(load_balancing_strategy="least_loaded")
   # Selects connection with fewest active streams

Per-Peer Limits

config = ConnectionConfig(
    max_connections_per_peer=3  # Default: 3 connections per peer
)
# When limit exceeded, oldest connections are trimmed

Use Cases:

• High-throughput applications needing parallel streams
• Fault-tolerant systems requiring redundant connections
• Load distribution across network paths

---

2. Connection Limits & Resource Management

Global Connection Limits

config = ConnectionConfig(
    max_connections=300,  # Total connections (inbound + outbound)
    max_parallel_dials=100,  # Concurrent dial attempts
    max_dial_queue_length=500,  # Maximum queued dials
    max_peer_addrs_to_dial=25,  # Addresses to try per peer
    max_incoming_pending_connections=10  # Pending inbound limit
)

Connection Pruning

When max_connections is exceeded, connections are pruned based on:

1. Peer Tag Value (lowest first) - Peers with higher tags are more valuable
2. Stream Count (lowest first) - Connections with fewer streams
3. Direction (inbound first) - Prefer keeping outbound connections
4. Connection Age (oldest first) - Remove oldest connections

Pruning Algorithm:

# libp2p/network/connection_pruner.py:234-382
def sort_connections(
    self, connections: list[INetConn], peer_values: dict[ID, int]
) -> list[INetConn]:
    """
    Sort connections for pruning priority.
    Returns connections in order: first = should be pruned first.
    """

Protection Mechanisms:

• Connections in allow list are never pruned
• Connections with active streams are less likely to be pruned
• Pruning is automatic and transparent

Resource Manager Integration

The connection management system integrates with the resource manager:

# Pre-upgrade admission checks
pre_scope = resource_manager.open_connection(None, endpoint_ip=ip)

# Post-upgrade connection scoping
conn_scope = resource_manager.open_connection(peer_id, endpoint_ip=ip)

# Stream resource tracking
resource_manager.acquire_stream(peer_id, Direction.OUTBOUND)

---

3. Connection Queue Management

Dial Queue

Features:

• Priority-based scheduling: Higher priority dials processed first
• Concurrency control: Maximum parallel dials enforced
• Queue length limits: Prevents unbounded queue growth
• Deduplication: Prevents duplicate dial requests
• Timeout handling: Per-dial timeout enforcement
• DNS integration: Automatic DNS resolution
• Address sorting: Uses AddressManager for optimal address selection

Usage:

# Dial requests automatically go through the queue
await swarm.dial_peer(peer_id)  # Uses dial queue internally

# Queue configuration
config = ConnectionConfig(
    max_parallel_dials=100,  # Concurrent dials
    max_dial_queue_length=500,  # Queue size
    dial_timeout=10.0  # Per-dial timeout
)

Priority System:

# Default priority: 50
# Higher priority = processed first
# Lower priority = processed later

Reconnection Queue

Features:

• KEEP_ALIVE support: Automatic reconnection for tagged peers
• Exponential backoff: Configurable backoff with jitter
• Retry limits: Maximum retry attempts
• Parallel control: Maximum concurrent reconnects
• Cleanup: Automatic cleanup on failure

Configuration:

config = ConnectionConfig(
    reconnect_retries=5,  # Max retry attempts
    reconnect_retry_interval=1.0,  # Initial delay (seconds)
    reconnect_backoff_factor=2.0,  # Exponential multiplier
    max_parallel_reconnects=5  # Concurrent reconnects
)

Reconnection Flow:

1. Connection disconnects
2. Check if peer has KEEP_ALIVE tag
3. If yes, add to reconnection queue
4. Wait for retry interval (with backoff)
5. Attempt reconnection
6. On success: remove from queue
7. On failure: retry up to max attempts
8. On max failures: remove KEEP_ALIVE tag

---

4. Security & Access Control

IP Allow/Deny Lists

Features:

• CIDR support: Network blocks (e.g., 192.168.0.0/16)
• Single IP support: Individual IPs (e.g., 192.168.1.1)
• Precedence rules: Deny list takes precedence over allow list
• Allow list bypass: Connections in allow list skip rate limiting and connection limits

Configuration:

config = ConnectionConfig(
    allow_list=["192.168.1.0/24", "10.0.0.1"],  # Always allowed
    deny_list=["192.168.1.100", "10.0.0.0/8"]  # Always denied
)

Precedence Rules:

1. Deny list checked first - If IP in deny list → REJECT
2. Allow list checked second - If allow list exists and IP not in it → REJECT
3. Allow list exists and IP in it → ALLOW (bypasses other checks)
4. No allow list → ALLOW (subject to other limits)

Rate Limiting

Token Bucket Algorithm:

• Points: Number of tokens (connections) allowed
• Duration: Time window (default: 1 second)
• Per-host: Rate limiting is per IP address
• Allow list bypass: IPs in allow list skip rate limiting

Configuration:

config = ConnectionConfig(
    inbound_connection_threshold=5  # 5 connections per second per host
)

Behavior:

• Each host gets a token bucket
• Tokens refill at rate of threshold / duration
• Connection attempt consumes 1 token
• If no tokens available → connection rejected
• Allow list IPs bypass rate limiting

---

5. Address Management

Default Address Sorting

The default address sorter prioritizes addresses in this order:

1. Transport Priority (Highest):

   • TCP: 100
   • WebSocketSecure: 80
   • WebSocket: 60
   • WebRTC: 40
   • WebRTC-Direct: 20
   • WebTransport: 10

2. Certified Addresses: Certified addresses first (when implemented)

3. Circuit Relay: Circuit relay addresses last

4. Public vs Private: Public addresses before private

5. Loopback: Loopback addresses last

Implementation:

# libp2p/network/address_manager.py:183-243
def default_address_sorter(addresses: list[Multiaddr]) -> list[Multiaddr]:
    """Default address sorter matching JS libp2p behavior."""
    # Stable sort applied in reverse order
    # Last sort is most important

Custom Address Sorting

def custom_sorter(addresses: list[Multiaddr], peer_id: ID | None = None) -> list[Multiaddr]:
    # Custom sorting logic
    return sorted_addresses

config = ConnectionConfig(address_sorter=custom_sorter)

DNS Resolution

Supported Protocols:

• /dns4/ - IPv4 DNS resolution
• /dns6/ - IPv6 DNS resolution
• /dnsaddr/ - DNSADDR resolution (partial implementation)

Features:

• Async resolution: Non-blocking DNS lookups
• Caching: Results cached for performance
• Multiple IPs: Returns all resolved addresses
• Fallback: Returns original address on resolution failure

Usage:

# Automatic DNS resolution in dial queue
await swarm.dial_peer(peer_id)  # DNS addresses automatically resolved

# Manual resolution
from libp2p.network.dns_resolver import resolve_multiaddr
resolved = await resolve_multiaddr(Multiaddr("/dns4/example.com/tcp/4001"))

---

6. Connection State Management

Connection States

class ConnectionStatus(enum.Enum):
    PENDING = "pending"   # Connection being established
    OPEN = "open"         # Connection active and ready
    CLOSING = "closing"   # Connection is closing
    CLOSED = "closed"     # Connection fully closed

State Tracking

@dataclass
class ConnectionState:
    status: ConnectionStatus
    timeline: ConnectionTimeline  # open/close timestamps
    direction: str  # "inbound" or "outbound"

Direction Tracking

All connections are now tagged with direction:

• Inbound: Connections accepted from remote peers
• Outbound: Connections initiated to remote peers

Usage:

# Direction is set at connection creation
swarm_conn = SwarmConn(muxed_conn, swarm, direction="outbound")

# Access direction
direction = connection.direction  # "inbound" or "outbound"

---

7. Event System

Available Events

The system provides comprehensive event notifications via the INotifee interface:

class INotifee:
    async def connected(self, network: INetworkService, conn: INetConn) -> None:
        """Called when a connection is established."""
    
    async def disconnected(self, network: INetworkService, conn: INetConn) -> None:
        """Called when a connection is closed."""
    
    async def opened_stream(self, network: INetworkService, stream: INetStream) -> None:
        """Called when a stream is opened."""
    
    async def closed_stream(self, network: INetworkService, stream: INetStream) -> None:
        """Called when a stream is closed."""
    
    async def listen(self, network: INetworkService, multiaddr: Multiaddr) -> None:
        """Called when listening starts."""
    
    async def listen_close(self, network: INetworkService, multiaddr: Multiaddr) -> None:
        """Called when listening stops."""

Event Registration

class MyNotifee(INotifee):
    async def connected(self, network, conn):
        print(f"Connected to {conn.muxed_conn.peer_id}")
    
    async def disconnected(self, network, conn):
        print(f"Disconnected from {conn.muxed_conn.peer_id}")

notifee = MyNotifee()
swarm.register_notifee(notifee)

Automatic Reconnection Triggering

Disconnection events automatically trigger reconnection for KEEP_ALIVE peers:

async def notify_disconnected(self, conn: INetConn) -> None:
    peer_id = conn.muxed_conn.peer_id
    # Automatically triggers reconnection if peer has KEEP_ALIVE tag
    await self.reconnect_queue.maybe_reconnect(peer_id)

---

Configuration System

ConnectionConfig

The ConnectionConfig class centralizes all connection management settings:

@dataclass
class ConnectionConfig:
    # Connection Limits
    max_connections: int = 300
    max_connections_per_peer: int = 3
    max_parallel_dials: int = 100
    max_dial_queue_length: int = 500
    max_peer_addrs_to_dial: int = 25
    max_incoming_pending_connections: int = 10
    
    # Timeouts
    dial_timeout: float = 10.0
    connection_close_timeout: float = 1.0
    inbound_upgrade_timeout: float = 10.0
    outbound_upgrade_timeout: float = 10.0
    outbound_stream_protocol_negotiation_timeout: float = 10.0
    inbound_stream_protocol_negotiation_timeout: float = 10.0
    
    # Load Balancing
    load_balancing_strategy: str = "round_robin"  # or "least_loaded"
    
    # Rate Limiting
    inbound_connection_threshold: int = 5  # connections/second
    
    # Reconnection
    reconnect_retries: int = 5
    reconnect_retry_interval: float = 1.0
    reconnect_backoff_factor: float = 2.0
    max_parallel_reconnects: int = 5
    
    # Security
    allow_list: list[str] = field(default_factory=list)
    deny_list: list[str] = field(default_factory=list)
    
    # Address Management
    address_sorter: Callable[..., Any] | None = None

Configuration Validation

Current validations:

• Load balancing strategy must be "round_robin" or "least_loaded"
• All limits must be >= 1
• All timeouts must be >= 0
• Reconnect backoff factor must be > 0

Recommended Additional Validations:

# Relationship validations
if max_connections < max_parallel_dials:
    raise ValueError("max_connections must be >= max_parallel_dials")

if max_dial_queue_length < max_parallel_dials:
    raise ValueError("max_dial_queue_length must be >= max_parallel_dials")

if max_connections_per_peer > max_connections:
    raise ValueError("max_connections_per_peer must be <= max_connections")

Default Values Rationale

All defaults match JavaScript libp2p defaults:

Setting	Default	Source
max_connections	300	JS libp2p constants.ts
max_parallel_dials	100	JS libp2p constants.ts
max_dial_queue_length	500	JS libp2p constants.ts
dial_timeout	10.0s	JS libp2p (10000ms)
inbound_connection_threshold	5/sec	JS libp2p constants.ts

Reference: libp2p/network/config.py:1-5

---

Performance Analysis

Expected Performance Characteristics

Positive Impacts

1. Connection Pruning:

   • Prevents resource exhaustion
   • Improves system stability
   • Reduces memory usage under load

2. Rate Limiting:

   • Reduces CPU usage from connection storms
   • Prevents DoS attacks
   • Token bucket algorithm is O(1) per check

3. Dial Queue:

   • Better resource utilization
   • Prevents overwhelming the system
   • Priority queue: O(log n) for inserts

4. Connection Reuse:

   • Load balancing improves throughput
   • Reduces connection establishment overhead
   • Better stream distribution

Potential Overhead

1. Connection Tracking:

   • Dictionary lookups: O(1)
   • Minimal memory overhead per connection
   • Direction tracking: 1 string per connection

2. Pruning Logic:

   • Only runs when limits exceeded
   • Sorting: O(n log n) where n = connections to prune
   • Amortized cost is low

3. Rate Limiting:

   • Token bucket: O(1) operations
   • Per-host tracking: O(1) dictionary access
   • Minimal memory: ~24 bytes per tracked host

4. Queue Management:

   • Priority queue: O(log n) for inserts
   • Queue operations are efficient
   • Deduplication: O(1) dictionary checks

Memory Impact

Per Connection:

• Connection object: ~200-500 bytes (existing)
• Direction tracking: ~10 bytes (new)
• State tracking: ~50 bytes (new)
• Total overhead: ~60 bytes per connection

Per Host (Rate Limiting):

• Token bucket: ~24 bytes
• Timestamp: 8 bytes
• Total: ~32 bytes per tracked host

System-Wide:

• Dial queue: ~100 bytes per queued dial
• Reconnection queue: ~50 bytes per reconnection job
• Address manager: Minimal (stateless)
• DNS cache: ~200 bytes per cached resolution

Benchmarking Status

Current Status: ⚠️ Not Yet Benchmarked

Recommended Benchmarks:

1. Connection Establishment:

   • Time to establish 100/1000/10000 connections
   • Overhead of connection management components
   • Comparison: with vs without new features

2. Memory Usage:

   • Memory footprint with multiple connections per peer
   • Impact of connection state tracking
   • Rate limiter memory usage

3. Throughput:

   • Stream throughput with load balancing
   • Impact of connection pruning on active connections
   • Connection reuse efficiency

4. Latency:

   • Dial queue overhead
   • Connection selection latency
   • DNS resolution impact

Performance Recommendations:

• Default values are production-tested in JS libp2p
• Python-specific benchmarking would validate assumptions
• Consider profiling in production workloads

---

Security Features

1. IP-Based Access Control

Allow Lists

Use Cases:

• Trusted networks (e.g., internal infrastructure)
• Bypass connection limits for critical peers
• Whitelist specific IPs

Example:

config = ConnectionConfig(
    allow_list=[
        "192.168.1.0/24",  # Internal network
        "10.0.0.1",        # Specific trusted IP
    ]
)

Deny Lists

Use Cases:

• Block known malicious IPs
• Prevent access from specific networks
• Security policy enforcement

Example:

config = ConnectionConfig(
    deny_list=[
        "192.168.100.1",      # Known malicious IP
        "10.0.0.0/8",         # Block entire network
    ]
)

CIDR Support

Both allow and deny lists support CIDR notation:

config = ConnectionConfig(
    allow_list=["192.168.0.0/16"],  # All 192.168.x.x addresses
    deny_list=["10.0.0.0/8"]        # All 10.x.x.x addresses
)

2. Rate Limiting

Protection Against:

• Connection flooding attacks
• DoS attempts
• Resource exhaustion

Configuration:

config = ConnectionConfig(
    inbound_connection_threshold=5  # 5 connections/second per host
)

Behavior:

• Per-host/IP rate limiting
• Token bucket algorithm
• Allow list bypass
• Automatic token refill

3. Connection Limits

Protection Against:

• Resource exhaustion
• Memory exhaustion
• CPU overload

Mechanisms:

• Global connection limit
• Per-peer connection limit
• Automatic pruning
• Allow list protection

---

Monitoring & Metrics

Connection Metrics

The system provides comprehensive metrics via Swarm.get_metrics():

metrics = swarm.get_metrics()

# Available metrics:
metrics.inbound_connections      # Number of inbound connections
metrics.outbound_connections     # Number of outbound connections
metrics.inbound_pending          # Pending inbound connections
metrics.outbound_pending         # Pending outbound connections
metrics.protocol_streams         # Streams per protocol
metrics.get_protocol_streams_per_connection_90th_percentile()  # 90th percentile

Metrics Dictionary

metrics_dict = metrics.to_dict()
# Returns:
# {
#     "connections": {
#         "inbound": int,
#         "inbound_pending": int,
#         "outbound": int,
#         "outbound_pending": int
#     },
#     "protocol_streams_total": dict[str, int],
#     "protocol_streams_per_connection_90th_percentile": dict[str, float]
# }

Connection State Queries

# Get all connections
all_conns = swarm.get_connections()

# Get connections for specific peer
peer_conns = swarm.get_connections(peer_id)

# Get connection map
conn_map = swarm.get_connections_map()  # dict[ID, list[INetConn]]

# Get total connection count
total = swarm.get_total_connections()

Event-Based Monitoring

Register notifees for real-time monitoring:

class MonitoringNotifee(INotifee):
    def __init__(self):
        self.connection_count = 0
        self.disconnection_count = 0
    
    async def connected(self, network, conn):
        self.connection_count += 1
        print(f"Connection #{self.connection_count} established")
    
    async def disconnected(self, network, conn):
        self.disconnection_count += 1
        print(f"Connection closed (total: {self.disconnection_count})")

monitor = MonitoringNotifee()
swarm.register_notifee(monitor)

---

Migration Guide

Backward Compatibility

The implementation maintains full backward compatibility through:

1. Legacy Property:

   # Old code continues to work
   conn = swarm.connections_legacy[peer_id]  # Returns first connection

2. Backward-Compatible Method:

   # Old code continues to work
   conn = swarm.get_connection(peer_id)  # Returns first connection

3. Default Configuration:

   • All features use sensible defaults
   • No configuration required for basic usage
   • Existing code works without changes

Breaking Changes

Minimal breaking changes:

1. Direct Dictionary Access (Low Impact):

   # OLD (may break)
   conn = swarm.connections[peer_id]  # Now returns list, not single connection
   
   # NEW (recommended)
   conn = swarm.get_connection(peer_id)  # Returns first connection
   conns = swarm.get_connections(peer_id)  # Returns all connections

2. Connection Limits (Medium Impact):

   # OLD: No connection limits
   # NEW: Default limit of 300 connections
   
   # If you need more connections:
   config = ConnectionConfig(max_connections=1000)
   swarm = new_swarm(connection_config=config)

Migration Steps

Step 1: Update Connection Access

# Before
conn = swarm.connections[peer_id]

# After (Option 1: Get first connection)
conn = swarm.get_connection(peer_id)

# After (Option 2: Get all connections)
conns = swarm.get_connections(peer_id)
conn = conns[0] if conns else None

Step 2: Configure Connection Limits (if needed)

# If you need more than 300 connections
config = ConnectionConfig(max_connections=1000)
swarm = new_swarm(connection_config=config)

Step 3: Enable Advanced Features (optional)

# Enable rate limiting
config = ConnectionConfig(
    inbound_connection_threshold=10  # 10 connections/second
)

# Enable IP filtering
config = ConnectionConfig(
    allow_list=["192.168.1.0/24"],
    deny_list=["10.0.0.0/8"]
)

# Enable automatic reconnection
# (Requires KEEP_ALIVE tag on peers)

Testing Migration

1. Test Connection Access:

   # Verify backward-compatible methods work
   conn = swarm.get_connection(peer_id)
   assert conn is not None

2. Test Connection Limits:

   # Verify connection limits don't break existing behavior
   # (if you don't exceed 300 connections)

3. Test Multiple Connections:

   # Test new multiple-connection feature
   conns = swarm.get_connections(peer_id)
   assert len(conns) >= 1

---

Best Practices

1. Configuration Guidelines

Production Configuration

config = ConnectionConfig(
    # Connection limits (adjust based on capacity)
    max_connections=500,
    max_connections_per_peer=5,
    max_parallel_dials=100,
    
    # Security
    allow_list=["192.168.0.0/16"],  # Internal network
    deny_list=["10.0.0.0/8"],       # Blocked network
    inbound_connection_threshold=10,  # Rate limiting
    
    # Timeouts (adjust based on network conditions)
    dial_timeout=15.0,
    outbound_upgrade_timeout=15.0,
    
    # Reconnection (for critical peers)
    reconnect_retries=10,
    reconnect_backoff_factor=2.0,
)

High-Performance Configuration

config = ConnectionConfig(
    max_connections=1000,
    max_parallel_dials=200,
    max_connections_per_peer=10,
    load_balancing_strategy="least_loaded",  # Better for high throughput
)

Secure/Restrictive Configuration

config = ConnectionConfig(
    max_connections=100,
    max_connections_per_peer=1,
    inbound_connection_threshold=2,  # Very restrictive
    allow_list=["192.168.1.0/24"],   # Only allow specific network
    deny_list=["0.0.0.0/0"],         # Deny all others (if allow_list exists)
)

2. Connection Management

Use Multiple Connections for High Throughput

# Enable multiple connections per peer
config = ConnectionConfig(max_connections_per_peer=5)

# Load balancing automatically distributes streams
stream = await host.new_stream(peer_id, [protocol])

Use KEEP_ALIVE for Critical Peers

# Tag critical peers for automatic reconnection
# (Implementation depends on peer tag system)
peerstore.tag_peer(peer_id, "keep-alive", value=1)

# Reconnection happens automatically on disconnect

Monitor Connection Health

# Regular health checks
async def monitor_connections(swarm):
    while True:
        metrics = swarm.get_metrics()
        print(f"Connections: {metrics.inbound_connections + metrics.outbound_connections}")
        print(f"90th percentile streams: {metrics.get_protocol_streams_per_connection_90th_percentile()}")
        await trio.sleep(60)  # Check every minute

3. Security Best Practices

Use Allow Lists for Trusted Networks

config = ConnectionConfig(
    allow_list=[
        "192.168.1.0/24",  # Internal network
        "10.0.0.0/8",      # VPN network
    ]
)

Use Deny Lists for Known Threats

config = ConnectionConfig(
    deny_list=[
        "192.168.100.1",  # Known malicious IP
        "10.0.0.100",     # Blocked peer
    ]
)

Configure Rate Limiting

# Adjust based on expected load
config = ConnectionConfig(
    inbound_connection_threshold=5,  # Conservative
    # or
    inbound_connection_threshold=20,  # High-traffic
)

4. Performance Optimization

Custom Address Sorting

def prioritize_local_network(addresses: list[Multiaddr]) -> list[Multiaddr]:
    """Prioritize local network addresses."""
    local = [a for a in addresses if "192.168" in str(a)]
    remote = [a for a in addresses if "192.168" not in str(a)]
    return local + remote

config = ConnectionConfig(address_sorter=prioritize_local_network)

Tune Connection Limits

# For high-connection-count scenarios
config = ConnectionConfig(
    max_connections=1000,
    max_parallel_dials=200,
    max_dial_queue_length=1000,
)

Use Least-Loaded Load Balancing

# For better stream distribution
config = ConnectionConfig(
    load_balancing_strategy="least_loaded"
)

---

Comparison with JavaScript libp2p

Feature Parity

Feature	JS libp2p	Python libp2p	Status
Multiple connections per peer	✅	✅	Complete
Connection limits	✅	✅	Complete
Connection pruning	✅	✅	Complete
Rate limiting	✅	✅	Complete
Dial queue	✅	✅	Complete
Reconnection queue	✅	✅	Complete
IP allow/deny lists	✅	✅	Complete
Address sorting	✅	✅	Complete
DNS resolution	✅	✅	Complete (DNSADDR partial)
Connection metrics	✅	✅	Complete
90th percentile metrics	✅	✅	Complete
Direction tracking	✅	✅	Complete
Connection state	✅	✅	Complete
Event system	✅	✅	Complete
Timeout configuration	✅	✅	Complete

Overall Parity: 95-100%

API Compatibility

The Python implementation closely matches JS libp2p APIs:

JavaScript:

// Connection limits
connectionManager.maxConnections = 300;

// Rate limiting
connectionManager.inboundConnectionThreshold = 5;

// Allow/deny lists
connectionManager.allow = ["192.168.1.0/24"];
connectionManager.deny = ["10.0.0.0/8"];

Python:

# Connection limits
config = ConnectionConfig(max_connections=300)

# Rate limiting
config = ConnectionConfig(inbound_connection_threshold=5)

# Allow/deny lists
config = ConnectionConfig(
    allow_list=["192.168.1.0/24"],
    deny_list=["10.0.0.0/8"]
)

Default Values

All default values match JavaScript libp2p:

• maxConnections: 300 → max_connections: 300
• maxParallelDials: 100 → max_parallel_dials: 100
• dialTimeout: 10000ms → dial_timeout: 10.0s
• inboundConnectionThreshold: 5 → inbound_connection_threshold: 5

---

Code Examples

Basic Usage

from libp2p import new_host
from libp2p.network.config import ConnectionConfig

# Create host with default connection management
host = new_host()

# Connections are automatically managed
await host.connect(peer_info)

Advanced Configuration

from libp2p import new_host
from libp2p.network.config import ConnectionConfig

# Production configuration
config = ConnectionConfig(
    max_connections=500,
    max_connections_per_peer=5,
    inbound_connection_threshold=10,
    allow_list=["192.168.0.0/16"],
    load_balancing_strategy="least_loaded",
)

host = new_host(connection_config=config)

Monitoring Setup

from libp2p.network.metrics import calculate_connection_metrics

# Get metrics
metrics = swarm.get_metrics()

# Access metrics
print(f"Inbound: {metrics.inbound_connections}")
print(f"Outbound: {metrics.outbound_connections}")
print(f"90th percentile: {metrics.get_protocol_streams_per_connection_90th_percentile()}")

# Convert to dictionary
metrics_dict = metrics.to_dict()

Event Monitoring

from libp2p.abc import INotifee

class ConnectionMonitor(INotifee):
    async def connected(self, network, conn):
        print(f"Connected: {conn.muxed_conn.peer_id}")
    
    async def disconnected(self, network, conn):
        print(f"Disconnected: {conn.muxed_conn.peer_id}")

monitor = ConnectionMonitor()
swarm.register_notifee(monitor)

Custom Address Sorting

def custom_sorter(addresses: list[Multiaddr], peer_id: ID | None = None) -> list[Multiaddr]:
    # Prioritize IPv4 over IPv6
    ipv4 = [a for a in addresses if "/ip4/" in str(a)]
    ipv6 = [a for a in addresses if "/ip6/" in str(a)]
    return ipv4 + ipv6

config = ConnectionConfig(address_sorter=custom_sorter)

---

Testing & Validation

Test Coverage

The implementation includes comprehensive examples and test patterns:

• ✅ Connection limits examples
• ✅ Rate limiting examples
• ✅ Allow/deny list examples
• ✅ Connection state examples
• ✅ Comprehensive configuration examples

Recommended Tests

1. Connection Limit Enforcement:

   • Verify limits are enforced
   • Test pruning behavior
   • Test allow list protection

2. Rate Limiting:

   • Verify rate limits are enforced
   • Test allow list bypass
   • Test token bucket behavior

3. Reconnection:

   • Test KEEP_ALIVE reconnection
   • Test retry limits
   • Test backoff behavior

4. Load Balancing:

   • Test round-robin distribution
   • Test least-loaded selection
   • Test stream distribution

---

Conclusion

The advanced connection management features represent a significant enhancement to Python libp2p, bringing it to feature parity with JavaScript libp2p. The implementation is:

• ✅ Production-ready: Comprehensive resource management and security
• ✅ Backward compatible: Existing code continues to work
• ✅ Well-documented: Extensive examples and documentation
• ✅ Performant: Efficient algorithms and data structures
• ✅ Secure: IP-based access control and rate limiting
• ✅ Observable: Comprehensive metrics and monitoring

Key Takeaways

1. Default values are based on JS libp2p's proven defaults
2. Backward compatibility is maintained through legacy APIs
3. Configuration is flexible and comprehensive
4. Performance characteristics are expected to be positive
5. Security features provide robust protection
6. Monitoring capabilities enable observability

---

References

• JavaScript libp2p Connection Manager: https://github.com/libp2p/js-libp2p/blob/main/packages/libp2p/src/connection-manager/index.ts
• Implementation PR: Feat/advance connection management #1018