TODO Analysis: Pubsub Module Issues

[acul71]

TODO Analysis: Pubsub Module Issues - Updated Status

Last Updated: August 2025
Go Comparison: Based on analysis of go-libp2p-pubsub
Python Repository: py-libp2p

Overview

This document analyzes five TODO/FIXME issues in the pubsub module with updated status based on Go implementation comparison:

1. GossipSub sendRPC piggybacking - libp2p/pubsub/gossipsub.py:265 - ⚠️ PARTIALLY IMPLEMENTED
2. Message ID type consistency - libp2p/pubsub/gossipsub.py:783 - ❌ CRITICAL ISSUE
3. Message ID parsing improvement - libp2p/pubsub/gossipsub.py:801-802 - ❌ SECURITY VULNERABILITY
4. Async validator throttling - libp2p/pubsub/pubsub.py:683 - ⚠️ PARTIALLY IMPLEMENTED

---

Issue 1: GossipSub sendRPC Piggybacking

Issue Location

File: libp2p/pubsub/gossipsub.py:265
Method: publish()
TODO Comment: # TODO: Go use sendRPC, which possibly piggybacks gossip/control messages.

Specification Requirements

GossipSub Version: v1.0+ (Core requirement)
Specification Reference: GossipSub v1.0 Specification

Specification Quote:

"Control Message Piggybacking"

Gossip and other control messages do not have to be transmitted in their own message. Instead, they can be coalesced and piggybacked on any other message in the regular flow, for any topic. This can lead to message rate reduction whenever there is some correlated flow between topics, which can be significant for densely connected peers.

For piggyback implementation details, consult the Go implementation.

Python Implementation Version: GossipSub v1.1 (supports /meshsub/1.1.0)
Specification Compliance: ⚠️ PARTIALLY COMPLIANT

Current Status: ⚠️ PARTIALLY IMPLEMENTED

Go Implementation Analysis

Location: gossipsub.go:1342-1387

Key Features:

func (gs *GossipSubRouter) sendRPC(p peer.ID, out *RPC, urgent bool) {
    // piggyback control message retries
    ctl, ok := gs.control[p]
    if ok {
        out.Control = ctl
        delete(gs.control, p)
    }
    
    // piggyback gossip
    ihave, ok := gs.gossip[p]
    if ok {
        out.Control.Ihave = append(out.Control.Ihave, ihave...)
        delete(gs.gossip, p)
    }
}

Go Strengths:

• Efficient batching of multiple message types
• Proper cleanup of pending messages
• Urgent message handling
• Queue-based message management
• Automatic RPC splitting for oversized messages

Python Implementation Status

Current Implementation:

# Pack(piggyback) GRAFT, PRUNE and IHAVE for the same peer into
# one control message and send it
await self._emit_control_msgs(
    peers_to_graft, peers_to_prune, peers_to_gossip
)

What's Implemented:

• Basic control message batching in _emit_control_msgs()
• Multiple message types in single RPC via pack_control_msgs()
• Heartbeat-based message aggregation

What's Missing (Specification Requirements):

• No proper sendRPC method like Go (v1.0+ REQUIRED - specification requirement)
• No urgent message handling (v1.0+ IMPLIED - not explicitly specified but needed for efficient implementation)
• No queue-based message management (v1.0+ IMPLIED - not explicitly specified but needed for proper piggybacking)
• No automatic RPC splitting (v1.0+ IMPLIED - not explicitly specified but needed for large messages)
• No retry mechanism for failed messages (v1.0+ IMPLIED - not explicitly specified but needed for reliability)
• Limited cross-topic piggybacking (v1.0+ REQUIRED - specification explicitly states "for any topic")
• No message size threshold handling (v1.2+ REQUIRED - specification requires message size thresholds for IDONTWANT)

Gap Analysis

Feature	Specification	Go Implementation	Python Implementation	Status
Message Batching	✅ Required	✅ Complete	⚠️ Partial	Medium Gap
Control Message Piggybacking	✅ Required	✅ Complete	⚠️ Partial	Medium Gap
Cross-topic Piggybacking	✅ Required ("for any topic")	✅ Complete	❌ Limited	High Gap
Urgent Message Handling	⚠️ Implied	✅ Complete	❌ Missing	High Gap
Queue Management	⚠️ Implied	✅ Complete	❌ Missing	High Gap
Retry Mechanism	⚠️ Implied	✅ Complete	❌ Missing	High Gap

What Must Be Done to Fix This Issue

Recommended Implementation

Create a unified sendRPC method (Specification Compliant):

async def sendRPC(
    self, 
    stream: INetStream, 
    publish_msgs: list[rpc_pb2.Message] | None = None,
    control_msg: rpc_pb2.ControlMessage | None = None,
    urgent: bool = False
) -> None:
    """
    Send RPC message with optional piggybacking of publish and control messages.
    Specification compliant implementation for GossipSub v1.0+.
    
    :param stream: Stream to send to
    :param publish_msgs: Optional list of publish messages
    :param control_msg: Optional control message to piggyback
    :param urgent: Whether this is an urgent message
    """
    rpc_msg = rpc_pb2.RPC()
    
    if publish_msgs:
        rpc_msg.publish.extend(publish_msgs)
    
    if control_msg:
        rpc_msg.control.CopyFrom(control_msg)
    
    # Check message size and split if necessary (like Go)
    if self._is_rpc_too_large(rpc_msg):
        rpc_msg = self._split_rpc_message(rpc_msg)
    
    await self.pubsub.write_msg(stream, rpc_msg)

Specification Compliance Notes:

• Cross-topic piggybacking: Must support piggybacking control messages from different topics
• Message rate reduction: Primary goal as specified
• Coalescing: Multiple control messages should be combined
• Regular flow: Should work with any message in the normal flow

---

Issue 2: Message ID Type Consistency

Issue Location

File: libp2p/pubsub/gossipsub.py:783
Method: handle_ihave()
FIXME Comment: # FIXME: Update type of message ID

Specification Requirements

GossipSub Version: v1.0+ (Core requirement)
Specification Reference: GossipSub v1.0 Specification

Specification Quote:

"The message cache (or mcache), is a data structure that stores message IDs and their corresponding messages, segmented into 'history windows.'"

Python Implementation Version: GossipSub v1.1 (supports /meshsub/1.1.0)
Specification Compliance: ❌ NON-COMPLIANT

Current Status: ❌ CRITICAL ISSUE

Go Implementation Analysis

Location: gossipsub.go:764-840

Key Features:

// Message IDs are consistently handled as strings
for _, mid := range ihave.GetMessageIDs() {
    if gs.p.seenMessage(mid) {
        continue
    }
    iwant[mid] = struct{}{}
}

// In handleIWant
for _, mid := range iwant.GetMessageIDs() {
    msg, count, ok := gs.mcache.GetForPeer(mid, p)
    if !ok {
        continue
    }
    ihave[mid] = msg.Message
}

Go Strengths:

• Consistent string-based message ID handling
• Type-safe operations
• No parsing required
• Direct string comparisons
• No security risks

Python Implementation Status

Current Issues:

# FIXME: Update type of message ID
msg_ids_wanted: list[Any] = [
    msg_id
    for msg_id in ihave_msg.messageIDs
    if literal_eval(msg_id) not in seen_seqnos_and_peers
]

Problems (Specification Violations):

• Uses literal_eval() for parsing (security risk) - v1.0+ VIOLATION
• Inconsistent type handling (Any vs specific types) - v1.0+ VIOLATION
• Performance overhead from parsing - v1.0+ VIOLATION
• Potential security vulnerabilities - v1.0+ CRITICAL VIOLATION
• Type confusion between string and tuple representations - v1.0+ VIOLATION

Gap Analysis

Feature	Go Implementation	Python Implementation	Status
Type Consistency	✅ String-based	❌ Mixed types	Critical Gap
Type Safety	✅ Compile-time	❌ Runtime errors	Critical Gap
Performance	✅ Direct operations	❌ Parsing overhead	High Gap
Security	✅ Safe	❌ Security risk	Critical Gap

What Must Be Done to Fix This Issue

Recommended Implementation

Implement consistent string-based handling:

from typing import NewType

# Define proper message ID type
MessageID = NewType("MessageID", str)

def parse_message_id_safe(msg_id_str: str) -> MessageID:
    """Safely handle message ID as string."""
    return MessageID(msg_id_str)

# Update methods to use proper types
async def handle_ihave(self, ihave_msg: rpc_pb2.ControlIHave, sender_peer_id: ID) -> None:
    msg_ids_wanted: list[MessageID] = [
        parse_message_id_safe(msg_id)
        for msg_id in ihave_msg.messageIDs
        if not self._is_msg_seen(msg_id)
    ]

---

Issue 3: Message ID Parsing Improvement

Issue Location

File: libp2p/pubsub/gossipsub.py:801-802
Method: handle_iwant()
FIXME Comments:

• # FIXME: Update type of message ID
• # FIXME: Find a better way to parse the msg ids

Specification Requirements

GossipSub Version: v1.0+ (Core requirement)
Specification Reference: GossipSub v1.0 Specification

Specification Quote:

"On receiving an IWANT(ids) message, the router will check its mcache and will forward any requested messages that are present in the mcache to the peer who sent the IWANT message."

Python Implementation Version: GossipSub v1.1 (supports /meshsub/1.1.0)
Specification Compliance: ❌ SECURITY NON-COMPLIANT

Current Status: ❌ SECURITY VULNERABILITY

Go Implementation Analysis

Key Features:

// Direct string operations - no parsing needed
for _, mid := range iwant.GetMessageIDs() {
    // Check if that peer has sent IDONTWANT before
    if _, ok := gs.unwanted[p][computeChecksum(mid)]; ok {
        continue
    }
    
    msg, count, ok := gs.mcache.GetForPeer(mid, p)
    if !ok {
        continue
    }
}

Go Strengths:

• No eval operations
• Direct string operations
• Secure by design
• High performance
• Type-safe operations

Python Implementation Status

Current Issues:

# SECURITY RISK: Using literal_eval
msg_ids: list[Any] = [literal_eval(msg) for msg in iwant_msg.messageIDs]

# Performance overhead
if literal_eval(msg_id) not in seen_seqnos_and_peers:

Problems (Specification Violations):

• Security Risk: literal_eval() can execute arbitrary code - v1.0+ CRITICAL VIOLATION
• Performance: Parsing overhead on every operation - v1.0+ VIOLATION
• Type Safety: No compile-time type checking - v1.0+ VIOLATION
• Maintainability: Complex parsing logic - v1.0+ VIOLATION
• Error Handling: No validation of parsed data - v1.0+ VIOLATION

Gap Analysis

Feature	Go Implementation	Python Implementation	Status
Security	✅ Safe	❌ Security risk	Critical Gap
Performance	✅ Direct operations	❌ Parsing overhead	High Gap
Type Safety	✅ Compile-time	❌ Runtime errors	Critical Gap
Error Handling	✅ Robust	❌ Limited	Medium Gap

What Must Be Done to Fix This Issue

Recommended Implementation

Replace unsafe parsing with direct string operations:

def safe_parse_message_id(msg_id_str: str) -> Tuple[bytes, bytes]:
    """
    Safely parse message ID using ast.literal_eval with validation.
    
    :param msg_id_str: String representation of message ID
    :return: Tuple of (seqno, from_id) as bytes
    :raises ValueError: If parsing fails
    """
    try:
        parsed = ast.literal_eval(msg_id_str)
        if not isinstance(parsed, tuple) or len(parsed) != 2:
            raise ValueError("Invalid message ID format")
        
        seqno, from_id = parsed
        if not isinstance(seqno, bytes) or not isinstance(from_id, bytes):
            raise ValueError("Message ID components must be bytes")
        
        return (seqno, from_id)
    except (ValueError, SyntaxError) as e:
        raise ValueError(f"Invalid message ID format: {e}")

# Update the method
async def handle_iwant(
    self, iwant_msg: rpc_pb2.ControlIWant, sender_peer_id: ID
) -> None:
    msg_ids: list[Tuple[bytes, bytes]] = [
        safe_parse_message_id(msg) for msg in iwant_msg.messageIDs
    ]

---

Issue 4: Async Validator Throttling

Issue Location

File: libp2p/pubsub/pubsub.py:683
Method: validate_msg()
TODO Comment: # TODO: Implement throttle on async validators

Specification Requirements

GossipSub Version: v1.1+ (Extended validators requirement)
Specification Reference: GossipSub v1.1 Specification

Specification Quote:

"In order to address this situation, all gossipsub v1.1 implementations must support extended validators with an enumerated decision interface."

Python Implementation Version: GossipSub v1.1 (supports /meshsub/1.1.0)
Specification Compliance: ⚠️ PARTIALLY COMPLIANT

Current Status: ⚠️ PARTIALLY IMPLEMENTED

Go Implementation Analysis

Key Features:

// Multiple throttling mechanisms
if score < gs.gossipThreshold {
    return nil
}

if gs.peerhave[p] > gs.params.MaxIHaveMessages {
    return nil
}

if gs.iasked[p] >= gs.params.MaxIHaveLength {
    return nil
}

Go Strengths:

• Multiple throttling levels
• Configurable thresholds
• Per-peer rate limiting
• Flood protection
• Score-based throttling

Python Implementation Status

Current Implementation:

# Basic throttling in async validator
if self._throttle_validator(peer_id):
    return

What's Implemented:

• Basic semaphore-based throttling
• Configurable concurrent validator limits
• Async/await support
• Test coverage

What's Missing (Specification Requirements):

• Less comprehensive than Go (v1.1+ REQUIRED - extended validators must be supported)
• Fewer throttling mechanisms (v1.1+ IMPLIED - needed for proper validator handling)
• No score-based throttling (v1.1+ IMPLIED - needed for peer scoring integration)
• Limited flood protection (v1.1+ IMPLIED - needed for spam protection)

Gap Analysis

Feature	Go Implementation	Python Implementation	Status
Basic Throttling	✅ Complete	✅ Complete	No Gap
Multiple Levels	✅ Complete	⚠️ Partial	Medium Gap
Score-based	✅ Complete	❌ Missing	Medium Gap
Flood Protection	✅ Complete	⚠️ Limited	Medium Gap

What Must Be Done to Fix This Issue

Recommended Enhancement

Add more comprehensive throttling:

class ValidatorThrottler:
    def __init__(self, max_concurrent_validators: int = 10):
        self.semaphore = trio.Semaphore(max_concurrent_validators)
        self.validator_stats: dict[str, int] = {}
        self.score_threshold = 0.0
        self.flood_protection = True
    
    async def run_validator(
        self, 
        validator: AsyncValidatorFn, 
        msg_forwarder: ID, 
        msg: rpc_pb2.Message,
        peer_score: float = 0.0
    ) -> bool:
        # Score-based throttling
        if peer_score < self.score_threshold:
            return False
        
        # Flood protection
        if self.flood_protection and self._is_flooding(msg_forwarder):
            return False
        
        async with self.semaphore:
            return await validator(msg_forwarder, msg)

---

Summary of Current Status

Implementation Status

Issue	Specification	Go Status	Python Status	Gap	Priority
sendRPC Piggybacking	v1.0+ REQUIRED	✅ Complete	⚠️ Partial	High	Medium
Message ID Type Consistency	v1.0+ REQUIRED	✅ Complete	❌ Broken	Critical	Critical
Message ID Parsing	v1.0+ REQUIRED	✅ Secure	❌ Security Risk	Critical	Critical
Async Validator Throttling	v1.1+ REQUIRED	✅ Comprehensive	⚠️ Partial	Medium	Low

Critical Issues in Python Implementation

1. Security Vulnerability: Use of literal_eval() in message ID parsing
2. Type Inconsistency: Mixed Any types for message IDs
3. Performance Issues: Parsing overhead on every operation
4. Incomplete Piggybacking: Missing proper sendRPC implementation

Recommendations

1. Immediate (Critical): Replace literal_eval() with direct string operations
2. High Priority: Implement consistent string-based message ID handling
3. Medium Priority: Complete sendRPC piggybacking implementation
4. Low Priority: Enhance throttling mechanisms

Go Implementation Best Practices

The Go implementation demonstrates several best practices that should be adopted:

1. Type Safety: Consistent string-based message ID handling
2. Security: No eval operations, direct string operations
3. Performance: Efficient batching and piggybacking
4. Comprehensive Throttling: Multiple protection mechanisms
5. Clean Architecture: Proper separation of concerns

---

Implementation Strategy

Phase 1: Security Fixes (Critical)

1. Replace literal_eval() with safe parsing methods (v1.0+ REQUIRED)
2. Implement consistent string-based message ID handling (v1.0+ REQUIRED)
3. Add comprehensive validation for message ID parsing (v1.0+ REQUIRED)
4. Update all message ID usage throughout the codebase (v1.0+ REQUIRED)

Phase 2: Type Safety Improvements (High Priority)

1. Define proper types for message IDs
2. Update protocol buffer handling for consistency
3. Add type annotations throughout the codebase
4. Implement compile-time type checking

Phase 3: Performance Optimizations (Medium Priority)

1. Complete sendRPC implementation with piggybacking (v1.0+ REQUIRED)
2. Add message batching for control messages (v1.0+ REQUIRED)
3. Implement queue-based message management (v1.0+ REQUIRED)
4. Add retry mechanisms for failed messages (v1.0+ REQUIRED)

Phase 4: Advanced Features (Low Priority)

1. Enhance throttling mechanisms to match Go implementation (v1.1+ REQUIRED)
2. Add score-based throttling for better resource management (v1.1+ REQUIRED)
3. Implement flood protection mechanisms (v1.1+ REQUIRED)
4. Add monitoring and metrics for performance tracking (v1.1+ REQUIRED)

Files That Would Need Updates

Core Implementation

• libp2p/pubsub/gossipsub.py - All four issues
• libp2p/pubsub/pubsub.py - Async validator throttling
• libp2p/pubsub/mcache.py - Message ID type updates

Type Definitions

• libp2p/custom_types.py - Add MessageID type
• libp2p/pubsub/pb/rpc_pb2.py - Protocol buffer definitions

Tests

• tests/core/pubsub/test_gossipsub.py - Message ID parsing tests
• tests/core/pubsub/test_pubsub.py - Validator throttling tests
• tests/core/pubsub/test_mcache.py - Message cache tests

Configuration

• libp2p/pubsub/pubsub.py - Add throttling configuration
• libp2p/pubsub/gossipsub.py - Add piggybacking configuration

Conclusion

The analysis reveals significant gaps between the Go and Python implementations, particularly in security and type consistency. The most critical issues are:

1. Security vulnerability from using literal_eval() for message ID parsing
2. Type inconsistency in message ID handling
3. Performance issues from unnecessary parsing overhead
4. Incomplete piggybacking implementation

The Go implementation provides a robust, secure, and efficient foundation that the Python implementation should strive to match while maintaining Python's strengths in async programming and rapid development.

Priority Actions:

1. Critical: Fix message ID parsing security vulnerability
2. High: Implement consistent string-based message ID handling
3. Medium: Complete sendRPC piggybacking implementation
4. Low: Enhance throttling mechanisms

Implementing these fixes will significantly improve the robustness, performance, and maintainability of the pubsub module.

[acul71]

@sumanjeet0012 @seetadev
The analysis reveals significant gaps between the Go and Python implementations, particularly in security and type consistency. The most critical issues are:

1. Security vulnerability from using literal_eval() for message ID parsing
2. Type inconsistency in message ID handling
3. Performance issues from unnecessary parsing overhead
4. Incomplete piggybacking implementation

The Go implementation provides a robust, secure, and efficient foundation that the Python implementation should strive to match while maintaining Python's strengths in async programming and rapid development.

Priority Actions:

1. Critical: Fix message ID parsing security vulnerability
2. High: Implement consistent string-based message ID handling
3. Medium: Complete sendRPC piggybacking implementation
4. Low: Enhance throttling mechanisms

Implementing these fixes will significantly improve the robustness, performance, and maintainability of the pubsub module.

[sukhman-sukh]

Hey @acul71 @seetadev,
I analysed the Issue 1 mentioned by you for sendRPC().
We are already batching and piggybacking in Python (the way it is implemented is different, but it surely works). For the things left, we need to first add a message size threshold and then use queues to manage them with an urgent flag.
Message backoff and retries can be added incrementally to it.
If I am on the right track, then should I start working on it?

[acul71]

Yes right track, go for it!

[sukhman-sukh]

Also, py-libp2p handles sendRPC differently. It does not explicitly mention it, but suitable for the required changes. Should we still shift to go-libp2p's way of handling Messages by passing it to the sendRPC pattern? In that case, it will have some architectural changes. I want your suggestion on this.

[acul71]

Regarding your questions about the sendRPC implementation:
Current Batching Implementation: You're correct that py-libp2p already implements basic batching and piggybacking through the _emit_control_msgs() method. The current approach is functional but different from Go's explicit sendRPC pattern.

Architectural Changes: The decision to adopt Go's sendRPC pattern should be based on:

• Specification Compliance: The GossipSub v1.0+ specification explicitly requires control message piggybacking
• Performance Benefits: Unified message handling could reduce overhead
• Maintainability: Consistency with Go implementation makes the codebase easier to maintain

Recommended Approach:
Phase 1: Fix the critical security issues (message ID parsing)
Phase 2: Implement message size thresholds and queue management
Phase 3: Consider adopting the sendRPC pattern if performance benefits are significant

[acul71]

@sukhman-sukh

Recommended Implementation Strategy

Phase 1: Critical Security Fixes (High Priority)

1. Fix Message ID Parsing: Replace literal_eval() with direct string handling
2. Implement Type Consistency: Ensure message IDs are handled as strings throughout
3. Add Input Validation: Validate message ID format before processing

Phase 2: Performance Enhancements (Medium Priority)

1. Add Message Size Thresholds: Implement proper message size handling
2. Implement Queue Management: Add queue-based message handling
3. Add Urgent Message Support: Implement priority-based message handling

Phase 3: Architectural Improvements (Low Priority)

1. Consider sendRPC Pattern: Evaluate if the benefits justify the architectural changes
2. Enhanced Throttling: Implement more sophisticated backoff mechanisms
3. Cross-topic Piggybacking: Improve support for multi-topic message batching