TODOs of WIP: Certified-Address-Book interface for Peer-Store #753

[acul71]

In refer to #753

Certified Address Book Implementation - TODO Analysis

Overview

This document provides a comprehensive analysis of all TODO items identified in the Certified Address Book implementation for py-libp2p. The implementation introduces cryptographically signed peer records with envelope-based verification, bringing py-libp2p in line with other libp2p implementations.

Core Certified Address Book TODOs

1. Periodic Peer Store Cleanup

Location: libp2p/peer/peerstore.py:41

# TODO: Set up an async task for periodic peer-store cleanup
# for expired addresses and records.

Description: The current implementation lacks automatic cleanup of expired peer records and addresses. This could lead to memory leaks over time.

Impact:

• Memory usage may grow unbounded
• Stale data could affect performance
• No automatic garbage collection

Suggested Implementation:

async def start_cleanup_task(self, cleanup_interval: int = 3600) -> None:
    """Start periodic cleanup of expired peer records and addresses."""
    while True:
        await trio.sleep(cleanup_interval)
        await self._cleanup_expired_records()

async def _cleanup_expired_records(self) -> None:
    """Remove expired peer records and addresses."""
    current_time = int(time.time())
    expired_peers = []
    
    for peer_id, peer_data in self.peer_data_map.items():
        if peer_data.is_expired():
            expired_peers.append(peer_id)
    
    for peer_id in expired_peers:
        self.maybe_delete_peer_record(peer_id)
        del self.peer_data_map[peer_id]

2. Enhanced maybe_delete_peer_record Usage

Location: libp2p/peer/peerstore.py:182

# TODO: Make proper use of this function
def maybe_delete_peer_record(self, peer_id: ID) -> None:

Description: The maybe_delete_peer_record function is currently only called in limited contexts and could be better integrated throughout the codebase.

Current Usage:

• Called in add_addrs() method
• Called in get_peer_record() method

Suggested Improvements:

• Call during periodic cleanup
• Call when addresses are manually cleared
• Call when peer data is cleared
• Add logging for debugging

3. Record Storage Limits

Location: libp2p/peer/peerstore.py:213

# TODO: Put up a limit on the number of records to be stored ?

Description: No limit is currently enforced on the number of peer records that can be stored, which could lead to memory issues in high-peer-count scenarios.

Suggested Implementation:

class PeerStore(IPeerStore):
    def __init__(self, max_records: int = 10000) -> None:
        self.peer_data_map = defaultdict(PeerData)
        self.addr_update_channels: dict[ID, MemorySendChannel[Multiaddr]] = {}
        self.peer_record_map: dict[ID, PeerRecordState] = {}
        self.max_records = max_records

    def _enforce_record_limit(self) -> None:
        """Enforce maximum number of stored records."""
        if len(self.peer_record_map) > self.max_records:
            # Remove oldest records based on sequence number
            sorted_records = sorted(
                self.peer_record_map.items(),
                key=lambda x: x[1].seq
            )
            records_to_remove = len(self.peer_record_map) - self.max_records
            for peer_id, _ in sorted_records[:records_to_remove]:
                self.maybe_delete_peer_record(peer_id)
                del self.peer_record_map[peer_id]

4. Address Overwriting Strategy

Location: libp2p/peer/peerstore.py:218-219

# TODO: In case of overwriting a record, what should be do with the
# old addresses, do we overwrite them with the new addresses too ?

Description: When a new peer record is consumed, the current implementation replaces all addresses. The strategy for handling old addresses needs to be defined.

Current Behavior:

• New addresses from the record completely replace old addresses
• No merging or preservation of existing addresses

Suggested Strategies:

Option A: Complete Replacement (Current)

# Keep current behavior - new record completely replaces old addresses
self.add_addrs(peer_id, record.addrs, ttl)

Option B: Merge Addresses

# Merge new addresses with existing ones, removing duplicates
existing_addrs = set(self.addrs(peer_id))
new_addrs = set(record.addrs)
merged_addrs = list(existing_addrs.union(new_addrs))
self.clear_addrs(peer_id)
self.add_addrs(peer_id, merged_addrs, ttl)

Option C: Configurable Strategy

class AddressMergeStrategy(Enum):
    REPLACE = "replace"
    MERGE = "merge"
    APPEND = "append"

def consume_peer_record(self, envelope: Envelope, ttl: int, 
                       merge_strategy: AddressMergeStrategy = AddressMergeStrategy.REPLACE) -> bool:

Crypto-Related TODOs

5. ECDSA Support

Location: libp2p/peer/envelope.py:159

# TODO: Add suport fot ECDSA parsing also

Description: The envelope parsing currently supports Ed25519, RSA, and Secp256k1 key types, but lacks ECDSA support.

Current Implementation:

def pub_key_from_protobuf(pb_key: cryto_pb.PublicKey) -> PublicKey:
    if pb_key.Type == cryto_pb.KeyType.Ed25519:
        return Ed25519PublicKey.from_bytes(pb_key.Data)
    elif pb_key.Type == cryto_pb.KeyType.RSA:
        return RSAPublicKey.from_bytes(pb_key.Data)
    elif pb_key.Type == cryto_pb.KeyType.Secp256k1:
        return Secp256k1PublicKey.from_bytes(pb_key.Data)
    # TODO: Add suport fot ECDSA parsing also
    else:
        raise ValueError(f"Unknown key type: {pb_key.Type}")

Suggested Implementation:

from libp2p.crypto.ecdsa import ECDSAPublicKey

def pub_key_from_protobuf(pb_key: cryto_pb.PublicKey) -> PublicKey:
    if pb_key.Type == cryto_pb.KeyType.Ed25519:
        return Ed25519PublicKey.from_bytes(pb_key.Data)
    elif pb_key.Type == cryto_pb.KeyType.RSA:
        return RSAPublicKey.from_bytes(pb_key.Data)
    elif pb_key.Type == cryto_pb.KeyType.Secp256k1:
        return Secp256k1PublicKey.from_bytes(pb_key.Data)
    elif pb_key.Type == cryto_pb.KeyType.ECDSA:
        return ECDSAPublicKey.from_bytes(pb_key.Data)
    else:
        raise ValueError(f"Unknown key type: {pb_key.Type}")

Integration and Interface TODOs

6. Peer Record Transfer Mechanism

Mentioned in GitHub Discussion: The implementation needs interfaces for sending and receiving peer records.

Current Status: The core functionality exists but lacks integration points.

Suggested Interfaces:

class IPeerRecordTransfer(ABC):
    @abstractmethod
    async def send_peer_record(self, peer_id: ID, envelope: Envelope) -> None:
        """Send a peer record to another peer."""
        pass
    
    @abstractmethod
    async def receive_peer_record(self, data: bytes) -> tuple[Envelope, PeerRecord]:
        """Receive and validate a peer record from another peer."""
        pass

class PeerRecordTransfer(IPeerRecordTransfer):
    def __init__(self, peerstore: IPeerStore):
        self.peerstore = peerstore
    
    async def send_peer_record(self, peer_id: ID, envelope: Envelope) -> None:
        """Send peer record via identify protocol or direct message."""
        # Implementation needed
        pass
    
    async def receive_peer_record(self, data: bytes) -> tuple[Envelope, PeerRecord]:
        """Receive and process incoming peer record."""
        envelope, record = consume_envelope(data, record.domain())
        self.peerstore.consume_peer_record(envelope, ttl=3600)
        return envelope, record

7. Type Checking Improvements

Mentioned in GitHub Discussion: Protobuf-generated code has type checking issues.

Current Issues:

• # type: ignore[attr-defined, name-defined] comments throughout
• Protobuf-generated classes not properly typed

Suggested Solutions:

Option A: Generate Type Stubs

# Generate .pyi files for protobuf modules
protoc --python_out=. --pyi_out=. libp2p/peer/pb/*.proto

Option B: Use mypy-protobuf

# pyproject.toml
[tool.mypy-protobuf]
mypy_plugin = true

Option C: Custom Type Definitions

# libp2p/peer/pb/__init__.py
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from .peer_record_pb2 import PeerRecord as PeerRecordPB
    from .envelope_pb2 import Envelope as EnvelopePB
    from .crypto_pb2 import PublicKey as PublicKeyPB
else:
    PeerRecordPB = None
    EnvelopePB = None
    PublicKeyPB = None

Performance and Optimization TODOs

8. Lazy Loading and Caching

Current Implementation: Basic caching exists in the Envelope class.

Potential Improvements:

class PeerStore(IPeerStore):
    def __init__(self) -> None:
        self.peer_data_map = defaultdict(PeerData)
        self.addr_update_channels: dict[ID, MemorySendChannel[Multiaddr]] = {}
        self.peer_record_map: dict[ID, PeerRecordState] = {}
        self._record_cache: dict[ID, PeerRecord] = {}  # Cache decoded records
        self._cache_ttl = 300  # 5 minutes

    def _get_cached_record(self, peer_id: ID) -> PeerRecord | None:
        """Get cached record if not expired."""
        if peer_id in self._record_cache:
            state = self.peer_record_map.get(peer_id)
            if state and time.time() - state.last_access < self._cache_ttl:
                return self._record_cache[peer_id]
        return None

9. Batch Operations

Suggested Enhancement:

def consume_peer_records(self, envelopes: list[Envelope], ttl: int) -> list[bool]:
    """Consume multiple peer records in a single operation."""
    results = []
    for envelope in envelopes:
        results.append(self.consume_peer_record(envelope, ttl))
    return results

Testing and Validation TODOs

10. Comprehensive Test Coverage

Current Status: Basic tests exist but could be expanded.

Missing Test Cases:

• Edge cases for sequence number generation
• Memory pressure scenarios
• Concurrent access patterns
• Network failure scenarios
• Malformed envelope handling

Suggested Test Additions:

def test_concurrent_sequence_generation():
    """Test thread safety of timestamp_seq()."""
    import threading
    import concurrent.futures
    
    def generate_seq():
        return timestamp_seq()
    
    with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
        futures = [executor.submit(generate_seq) for _ in range(100)]
        results = [future.result() for future in futures]
    
    # Ensure all sequence numbers are unique and increasing
    assert len(set(results)) == len(results)
    assert results == sorted(results)

def test_memory_pressure_scenario():
    """Test behavior under memory pressure."""
    store = PeerStore(max_records=5)
    
    # Add more records than the limit
    for i in range(10):
        keypair = create_new_key_pair()
        peer_id = ID.from_pubkey(keypair.public_key)
        record = PeerRecord(peer_id, [Multiaddr(f"/ip4/127.0.0.1/tcp/{4000+i}")], i)
        envelope = seal_record(record, keypair.private_key)
        store.consume_peer_record(envelope, ttl=3600)
    
    # Verify only the newest records are kept
    assert len(store.peer_record_map) <= 5

Documentation TODOs

11. API Documentation

Current Status: Basic docstrings exist but could be enhanced.

Suggested Improvements:

• Add usage examples
• Document error conditions
• Provide migration guide from old address book
• Add performance characteristics

12. Integration Guide

Missing: Guide for integrating certified address book with existing applications.

Suggested Content:

• Migration from basic address book
• Configuration options
• Best practices
• Troubleshooting guide

Priority Matrix

TODO Item	Priority	Effort	Impact	Dependencies
Periodic Cleanup	High	Medium	High	None
ECDSA Support	Medium	Low	Medium	Crypto module
Record Limits	High	Low	High	None
Address Strategy	Medium	Medium	Medium	None
Type Checking	Low	High	Low	Build system
Transfer Mechanism	High	High	High	Network layer
Enhanced Testing	Medium	Medium	Medium	None

Implementation Timeline

Phase 1 (Immediate - 1-2 weeks)

1. Implement periodic cleanup task
2. Add record storage limits
3. Define address overwriting strategy
4. Add ECDSA support

Phase 2 (Short-term - 2-4 weeks)

1. Implement peer record transfer mechanism
2. Enhance test coverage
3. Improve type checking
4. Add performance optimizations

Phase 3 (Medium-term - 1-2 months)

1. Comprehensive documentation
2. Integration examples
3. Performance benchmarking
4. Security audit

Conclusion

The Certified Address Book implementation is functionally complete but requires several enhancements for production readiness. The most critical TODOs are the periodic cleanup mechanism, record storage limits, and the peer record transfer interface. These improvements will ensure the implementation is robust, performant, and ready for integration with GossipSub 1.1 and other libp2p protocols.

References

• GitHub Pull Request #753
• Go-libp2p PeerRecord Implementation
• Libp2p Specification
• GossipSub 1.1 Requirements