# TODO Analysis: Test Notify Issues

[acul71]

TODO Analysis: Test Notify Issues

Overview

This document analyzes three TODO issues in the test_notify.py module:

1. Missing tests for closed_stream and listen_close - tests/core/network/test_notify.py:8
2. closed_stream method not implemented - tests/core/network/test_notify.py:47
3. listen_close method not implemented - tests/core/network/test_notify.py:60

---

Issue 1: Missing Tests for Closed Stream and Listen Close

Issue Location

File: tests/core/network/test_notify.py:8
TODO Comment: TODO: Add tests for closed_stream, listen_close when those features are implemented in swarm

Current Status Analysis

Current Implementation

The test file has commented-out assertions for the unimplemented features:

"""
Test Notify and Notifee by ensuring that the proper events get called, and that
the stream passed into opened_stream is correct.

Note: Listen event does not get hit because MyNotifee is passed
into network after network has already started listening

TODO: Add tests for closed_stream, listen_close when those
features are implemented in swarm
"""

# ... test implementation ...

        # Wait for events
        assert await wait_for_event(events_0_0, Event.Connected, 1.0)
        assert await wait_for_event(events_0_0, Event.OpenedStream, 1.0)
        # assert await wait_for_event(
        #     events_0_0, Event.ClosedStream, 1.0
        # )  # Not implemented
        assert await wait_for_event(events_0_0, Event.Disconnected, 1.0)

The Problem

• Incomplete Test Coverage: Tests are missing for two important network events
• Feature Gap: The test suite doesn't validate closed_stream and listen_close functionality
• Documentation Gap: No test examples for how these events should work
• Implementation Dependency: Tests are waiting for swarm implementation

What Must Be Done

Required Changes

1. Implement Swarm Features: Add closed_stream and listen_close notification in swarm
2. Add Test Cases: Create comprehensive tests for both events
3. Update Test Infrastructure: Ensure test helpers support new events
4. Documentation: Add test examples and expected behavior

Implementation Options

Option A: Basic Test Implementation

@pytest.mark.trio
async def test_notify_with_closed_stream_and_listen_close(security_protocol):
    """Test that closed_stream and listen_close events are properly triggered."""
    
    # Helper to wait for specific event
    async def wait_for_event(events_list, event, timeout=1.0):
        with trio.move_on_after(timeout):
            while event not in events_list:
                await trio.sleep(0.01)
            return True
        return False

    # Event lists for notifees
    events_0 = []
    events_1 = []

    # Create two swarms
    async with SwarmFactory.create_batch_and_listen(2) as swarms:
        # Register notifees
        notifee_0 = MyNotifee(events_0)
        notifee_1 = MyNotifee(events_1)

        swarms[0].register_notifee(notifee_0)
        swarms[1].register_notifee(notifee_1)

        # Connect swarms
        await connect_swarm(swarms[0], swarms[1])

        # Create and close a stream to trigger closed_stream event
        stream = await swarms[0].new_stream(swarms[1].get_peer_id())
        await stream.close()

        # Wait for closed_stream event
        assert await wait_for_event(events_0, Event.ClosedStream, 1.0)
        assert await wait_for_event(events_1, Event.ClosedStream, 1.0)

        # Close listeners to trigger listen_close events
        await swarms[0].close()
        await swarms[1].close()

        # Wait for listen_close events
        assert await wait_for_event(events_0, Event.ListenClose, 1.0)
        assert await wait_for_event(events_1, Event.ListenClose, 1.0)

Option B: Comprehensive Test Suite

@pytest.mark.trio
async def test_closed_stream_notification(security_protocol):
    """Test closed_stream notification for various stream closure scenarios."""
    
    events = []
    
    async with SwarmFactory.create_batch_and_listen(2) as swarms:
        notifee = MyNotifee(events)
        swarms[0].register_notifee(notifee)
        
        # Connect swarms
        await connect_swarm(swarms[0], swarms[1])
        
        # Test 1: Normal stream close
        stream = await swarms[0].new_stream(swarms[1].get_peer_id())
        await stream.close()
        
        assert Event.ClosedStream in events
        events.clear()
        
        # Test 2: Stream reset
        stream = await swarms[0].new_stream(swarms[1].get_peer_id())
        await stream.reset()
        
        assert Event.ClosedStream in events
        events.clear()
        
        # Test 3: Connection close (should close all streams)
        await swarms[0].close_peer(swarms[1].get_peer_id())
        
        # Should get disconnected event but no additional closed_stream
        # since streams were already closed
        assert Event.Disconnected in events

@pytest.mark.trio
async def test_listen_close_notification(security_protocol):
    """Test listen_close notification for various listener closure scenarios."""
    
    events = []
    
    async with SwarmFactory.create_batch_and_listen(1) as swarms:
        notifee = MyNotifee(events)
        swarms[0].register_notifee(notifee)
        
        # Test 1: Normal swarm close
        await swarms[0].close()
        
        # Should get listen_close events for all listening addresses
        listen_close_events = [e for e in events if e == Event.ListenClose]
        assert len(listen_close_events) > 0
        
        # Test 2: Individual listener close (if supported)
        # This would require additional swarm API for closing specific listeners

@pytest.mark.trio
async def test_notify_event_order(security_protocol):
    """Test that events are triggered in the correct order."""
    
    events = []
    
    async with SwarmFactory.create_batch_and_listen(2) as swarms:
        notifee = MyNotifee(events)
        swarms[0].register_notifee(notifee)
        
        # Connect swarms
        await connect_swarm(swarms[0], swarms[1])
        
        # Create and close stream
        stream = await swarms[0].new_stream(swarms[1].get_peer_id())
        await stream.close()
        
        # Close connection
        await swarms[0].close_peer(swarms[1].get_peer_id())
        
        # Verify event order
        expected_order = [
            Event.Connected,
            Event.OpenedStream,
            Event.ClosedStream,
            Event.Disconnected
        ]
        
        # Extract events in order they occurred
        ordered_events = []
        for expected_event in expected_order:
            if expected_event in events:
                ordered_events.append(expected_event)
        
        assert ordered_events == expected_order

Option C: Edge Case Testing

@pytest.mark.trio
async def test_notify_edge_cases(security_protocol):
    """Test edge cases for notify system."""
    
    events = []
    
    async with SwarmFactory.create_batch_and_listen(2) as swarms:
        notifee = MyNotifee(events)
        swarms[0].register_notifee(notifee)
        
        # Test 1: Multiple rapid stream operations
        streams = []
        for _ in range(5):
            stream = await swarms[0].new_stream(swarms[1].get_peer_id())
            streams.append(stream)
        
        # Close all streams rapidly
        for stream in streams:
            await stream.close()
        
        # Should get closed_stream events for all streams
        closed_stream_count = events.count(Event.ClosedStream)
        assert closed_stream_count == 5
        
        # Test 2: Stream operations after connection close
        await swarms[0].close_peer(swarms[1].get_peer_id())
        
        # Attempt to create new stream (should fail)
        with pytest.raises(Exception):
            await swarms[0].new_stream(swarms[1].get_peer_id())
        
        # Test 3: Multiple notifees
        events2 = []
        notifee2 = MyNotifee(events2)
        swarms[0].register_notifee(notifee2)
        
        # Reconnect and test
        await connect_swarm(swarms[0], swarms[1])
        stream = await swarms[0].new_stream(swarms[1].get_peer_id())
        await stream.close()
        
        # Both notifees should receive events
        assert Event.ClosedStream in events
        assert Event.ClosedStream in events2

Impact Analysis

• Test Coverage: High - Provides comprehensive validation of notify system
• Implementation Guidance: High - Tests define expected behavior
• Maintenance: Medium - Additional test complexity
• Documentation: High - Tests serve as usage examples

---

Issue 2: Closed Stream Method Not Implemented

Issue Location

File: tests/core/network/test_notify.py:47
Method: closed_stream() in MyNotifee class
TODO Comment: # TODO: It is not implemented yet.

Current Status Analysis

Current Implementation

The closed_stream method is a stub that does nothing:

class MyNotifee(INotifee):
    def __init__(self, events: list[Event]):
        self.events = events

    async def opened_stream(self, network: INetwork, stream: INetStream) -> None:
        self.events.append(Event.OpenedStream)

    async def closed_stream(self, network: INetwork, stream: INetStream) -> None:
        # TODO: It is not implemented yet.
        pass

    async def connected(self, network: INetwork, conn: INetConn) -> None:
        self.events.append(Event.Connected)

    async def disconnected(self, network: INetwork, conn: INetConn) -> None:
        self.events.append(Event.Disconnected)

    async def listen(self, network: INetwork, multiaddr: Multiaddr) -> None:
        self.events.append(Event.Listen)

    async def listen_close(self, network: INetwork, multiaddr: Multiaddr) -> None:
        # TODO: It is not implemented yet.
        pass

The Problem

• Missing Functionality: The method doesn't record closed stream events
• Test Incompleteness: Tests can't validate closed stream behavior
• Interface Mismatch: Method exists but doesn't fulfill its purpose
• Event Tracking Gap: No way to track when streams are closed

What Must Be Done

Required Changes

1. Implement Event Recording: Add closed stream events to the events list
2. Consistent Behavior: Match the pattern of other implemented methods
3. Test Integration: Enable tests to validate closed stream notifications
4. Error Handling: Add appropriate error handling if needed

Implementation Options

Option A: Simple Implementation

async def closed_stream(self, network: INetwork, stream: INetStream) -> None:
    """Record when a stream is closed."""
    self.events.append(Event.ClosedStream)

Option B: Enhanced Implementation with Validation

async def closed_stream(self, network: INetwork, stream: INetStream) -> None:
    """Record when a stream is closed with additional validation."""
    # Validate parameters
    if network is None:
        raise ValueError("network parameter cannot be None")
    if stream is None:
        raise ValueError("stream parameter cannot be None")
    
    # Record the event
    self.events.append(Event.ClosedStream)
    
    # Optional: Add logging for debugging
    # logger.debug(f"Stream closed: {stream}")

Option C: Implementation with Event Metadata

from dataclasses import dataclass
from typing import Any

@dataclass
class StreamEvent:
    event_type: Event
    network: INetwork
    stream: INetStream
    timestamp: float

class MyNotifee(INotifee):
    def __init__(self, events: list[Event]):
        self.events = events
        self.event_details: list[StreamEvent] = []

    async def closed_stream(self, network: INetwork, stream: INetStream) -> None:
        """Record when a stream is closed with detailed information."""
        import time
        
        self.events.append(Event.ClosedStream)
        self.event_details.append(StreamEvent(
            event_type=Event.ClosedStream,
            network=network,
            stream=stream,
            timestamp=time.time()
        ))

Impact Analysis

• Functionality: High - Enables closed stream event tracking
• Test Support: High - Allows tests to validate closed stream behavior
• Complexity: Low - Simple implementation
• Consistency: High - Matches existing method patterns

---

Issue 3: Listen Close Method Not Implemented

Issue Location

File: tests/core/network/test_notify.py:60
Method: listen_close() in MyNotifee class
TODO Comment: # TODO: It is not implemented yet.

Current Status Analysis

Current Implementation

The listen_close method is a stub that does nothing:

async def listen_close(self, network: INetwork, multiaddr: Multiaddr) -> None:
    # TODO: It is not implemented yet.
    pass

The Problem

• Missing Functionality: The method doesn't record listen close events
• Test Incompleteness: Tests can't validate listen close behavior
• Interface Mismatch: Method exists but doesn't fulfill its purpose
• Event Tracking Gap: No way to track when listeners stop listening

What Must Be Done

Required Changes

1. Implement Event Recording: Add listen close events to the events list
2. Consistent Behavior: Match the pattern of other implemented methods
3. Test Integration: Enable tests to validate listen close notifications
4. Error Handling: Add appropriate error handling if needed

Implementation Options

Option A: Simple Implementation

async def listen_close(self, network: INetwork, multiaddr: Multiaddr) -> None:
    """Record when a listener stops listening on a multiaddress."""
    self.events.append(Event.ListenClose)

Option B: Enhanced Implementation with Validation

async def listen_close(self, network: INetwork, multiaddr: Multiaddr) -> None:
    """Record when a listener stops listening on a multiaddress with validation."""
    # Validate parameters
    if network is None:
        raise ValueError("network parameter cannot be None")
    if multiaddr is None:
        raise ValueError("multiaddr parameter cannot be None")
    
    # Record the event
    self.events.append(Event.ListenClose)
    
    # Optional: Add logging for debugging
    # logger.debug(f"Listener closed on: {multiaddr}")

Option C: Implementation with Event Metadata

from dataclasses import dataclass
from typing import Any

@dataclass
class ListenEvent:
    event_type: Event
    network: INetwork
    multiaddr: Multiaddr
    timestamp: float

class MyNotifee(INotifee):
    def __init__(self, events: list[Event]):
        self.events = events
        self.listen_events: list[ListenEvent] = []

    async def listen_close(self, network: INetwork, multiaddr: Multiaddr) -> None:
        """Record when a listener stops listening with detailed information."""
        import time
        
        self.events.append(Event.ListenClose)
        self.listen_events.append(ListenEvent(
            event_type=Event.ListenClose,
            network=network,
            multiaddr=multiaddr,
            timestamp=time.time()
        ))

Impact Analysis

• Functionality: High - Enables listen close event tracking
• Test Support: High - Allows tests to validate listen close behavior
• Complexity: Low - Simple implementation
• Consistency: High - Matches existing method patterns

---

Swarm Implementation Requirements

Current Swarm Status

The swarm implementation has stub methods for the missing notifications:

async def notify_closed_stream(self, stream: INetStream) -> None:
    raise NotImplementedError

async def notify_listen_close(self, multiaddr: Multiaddr) -> None:
    raise NotImplementedError

Required Swarm Changes

1. Implement notify_closed_stream

async def notify_closed_stream(self, stream: INetStream) -> None:
    """Notify all notifees that a stream has been closed."""
    async with trio.open_nursery() as nursery:
        for notifee in self.notifees:
            nursery.start_soon(notifee.closed_stream, self, stream)

2. Implement notify_listen_close

async def notify_listen_close(self, multiaddr: Multiaddr) -> None:
    """Notify all notifees that a listener has stopped listening."""
    async with trio.open_nursery() as nursery:
        for notifee in self.notifees:
            nursery.start_soon(notifee.listen_close, self, multiaddr)

3. Integration Points

Stream Closure Integration:

# In NetStream._notify_closed method
async def _notify_closed(self) -> None:
    """Notify all listeners that the stream has been closed."""
    async with self._notify_lock:
        if hasattr(self.muxed_conn, "swarm"):
            swarm = getattr(self.muxed_conn, "swarm")
            
            # Call the new notify_closed_stream method
            if hasattr(swarm, "notify_closed_stream"):
                await swarm.notify_closed_stream(self)

Listener Closure Integration:

# In Swarm.close method
async def close(self) -> None:
    """Close the swarm instance and cleanup resources."""
    # Notify listen_close for all active listeners
    for maddr_str, listener in self.listeners.items():
        try:
            maddr = Multiaddr(maddr_str)
            await self.notify_listen_close(maddr)
        except Exception as e:
            logger.warning(f"Failed to notify listen_close for {maddr_str}: {e}")
    
    # ... rest of close implementation ...

---

Summary and Recommendations

Priority Order

1. High Priority: Implement closed_stream method in MyNotifee (Issue 2)

   • Simple implementation needed for test functionality
   • Enables immediate test improvements

2. High Priority: Implement listen_close method in MyNotifee (Issue 3)

   • Simple implementation needed for test functionality
   • Enables immediate test improvements

3. Medium Priority: Add comprehensive tests for closed_stream and listen_close (Issue 1)

   • Depends on swarm implementation
   • Provides validation and documentation

Implementation Strategy

1. Phase 1: Implement stub methods in MyNotifee class
2. Phase 2: Implement swarm notification methods
3. Phase 3: Add integration points in stream and listener code
4. Phase 4: Create comprehensive test suite
5. Phase 5: Add edge case and error handling tests

Testing Requirements

• Unit tests for MyNotifee methods
• Integration tests for swarm notification system
• End-to-end tests for stream lifecycle
• Edge case tests for rapid operations
• Error handling tests for invalid parameters

Documentation Needs

• API documentation for new notification methods
• Test examples showing expected behavior
• Integration guide for custom notifee implementations
• Troubleshooting guide for notification issues

Backward Compatibility

• All changes maintain existing API compatibility
• New methods are additive and don't break existing code
• Optional features can be gradually adopted

[bomanaps]

I will start working on this this week and also open up issues on it.
I will be going in this order:

1. Implement closed_stream in MyNotifee
2. Implement listen_close in MyNotifee
3. Implement swarm notification methods (notify_closed_stream & notify_listen_close)
4. Add integration points in stream and listener lifecycle
5. Write the tests for closed_stream and listen_close

[acul71]

Great. Tag me when you'll open the Issue/PR

[acul71]

Libp2p Notification System: Use Cases & Core Module Integration

Overview

The notification system in py-libp2p is the nervous system of the peer-to-peer network, allowing different components to stay synchronized with network state changes without direct dependencies. It uses an event-driven architecture where components can register as "notifees" to receive notifications about network lifecycle events.

Core Notification Events

The system provides 6 key notification events:

1. opened_stream - When a new network stream is created
2. closed_stream - When a stream is closed or reset
3. connected - When a new peer connection is established
4. disconnected - When a peer connection is closed
5. listen - When a listener starts on a multiaddress
6. listen_close - When a listener is closed

Core Modules Using the Notification System

1. Swarm (Core Network Layer)

File: libp2p/network/swarm.py

Role: Central orchestrator of the notification system

What it does:

• Manages notifees: Maintains a list of registered INotifee objects
• Triggers notifications: Calls notification methods at key lifecycle events
• Coordinates events: Ensures all notifees are notified of network state changes

Key methods:

def register_notifee(self, notifee: INotifee) -> None
async def notify_opened_stream(self, stream: INetStream) -> None
async def notify_connected(self, conn: INetConn) -> None
async def notify_disconnected(self, conn: INetConn) -> None
async def notify_listen(self, multiaddr: Multiaddr) -> None
async def notify_listen_close(self, multiaddr: Multiaddr) -> None
async def notify_closed_stream(self, stream: INetStream) -> None

Integration points:

• Connection establishment: Calls notify_connected() when new connections are added
• Listener lifecycle: Calls notify_listen() and notify_listen_close() during listener management
• Stream management: Calls notify_opened_stream() when streams are created
• Cleanup: Calls notify_disconnected() and notify_closed_stream() during resource cleanup

2. PubSub (Publish/Subscribe System)

Files: libp2p/pubsub/pubsub.py and libp2p/pubsub/pubsub_notifee.py

Role: Primary consumer of the notification system

What it does:

• Registers a notifee: PubsubNotifee is automatically registered with the swarm
• Manages peer lifecycle: Uses connection events to track pubsub peers
• Handles message routing: Automatically creates/destroys pubsub streams based on peer connections

Key implementation:

# In Pubsub.__init__():
self.host.get_network().register_notifee(
    PubsubNotifee(peer_send, dead_peer_send)
)

Event handling:

• connected: Adds new peers to initiator_peers_queue for pubsub stream creation
• disconnected: Adds dead peers to dead_peers_queue for cleanup
• opened_stream/closed_stream: Minimal processing (just checkpoints)

Use case: Automatic peer discovery and management - When peers connect/disconnect, pubsub automatically creates or tears down the appropriate pubsub streams without manual intervention.

3. SwarmConnection (Connection Management)

File: libp2p/network/connection/swarm_connection.py

Role: Handles individual connection lifecycle

What it does:

• Stream creation: Calls notify_opened_stream() when new streams are added to the connection
• Connection cleanup: Calls notify_disconnected() when connections are closed
• Resource management: Ensures proper notification during connection lifecycle

Key methods:

async def _add_stream(self, muxed_stream: IMuxedStream) -> NetStream:
    net_stream = NetStream(muxed_stream)
    self.streams.add(net_stream)
    await self.swarm.notify_opened_stream(net_stream)  # ← Notification here
    return net_stream

async def _notify_disconnected(self) -> None:
    await self.swarm.notify_disconnected(self)  # ← Notification here

4. NetStream (Stream Management)

File: libp2p/network/stream/net_stream.py

Role: Manages individual stream lifecycle

What it does:

• Stream closure: Uses notify_all() to notify about stream closure events
• Resource cleanup: Ensures proper notification when streams are closed or reset
• State management: Tracks stream state and triggers appropriate notifications

Key implementation:

async def _notify_closed(self) -> None:
    async with self._notify_lock:
        if hasattr(self.muxed_conn, "swarm"):
            swarm = getattr(self.muxed_conn, "swarm")
            if hasattr(swarm, "notify_all"):
                await swarm.notify_all(
                    lambda notifiee: notifiee.closed_stream(swarm, self)
                )

5. BasicHost (Host Layer)

File: libp2p/host/basic_host.py

Role: Provides access to the notification system

What it does:

• Exposes network: Provides get_network() method that allows access to swarm's notification methods
• Integration point: Higher-level components register notifees through the host
• Protocol management: Handles protocol negotiation and stream management

Practical Use Cases

1. Peer Discovery and Management

Scenario: New peer connects to the network
Flow:

1. Swarm establishes connection → notify_connected()
2. PubSub receives notification → adds peer to queue
3. PubSub creates pubsub stream → notify_opened_stream()
4. Peer is now available for message routing

2. Resource Management

Scenario: Stream is closed
Flow:

1. NetStream closes → _notify_closed()
2. Swarm receives notification → notify_closed_stream()
3. All notifees are notified → cleanup resources
4. Stream is removed from connection tracking

3. Listener Lifecycle Management

Scenario: Host starts/stops listening
Flow:

1. Host starts listening → notify_listen()
2. Host stops listening → notify_listen_close()
3. Components can track listener state changes
4. Resources are properly allocated/deallocated

4. Network Partition Handling

Scenario: Connection drops unexpectedly
Flow:

1. Connection fails → notify_disconnected()
2. PubSub receives notification → removes peer from routing
3. Resources are cleaned up
4. System can attempt reconnection

5. Debugging and Monitoring

Scenario: Network state tracking
Flow:

1. All network events trigger notifications
2. Monitoring components can track state changes
3. Debug information is available for troubleshooting
4. Performance metrics can be collected

Benefits of the Notification System

1. Decoupled Architecture

• Components don't need direct dependencies on each other
• Easy to add new components that react to network events
• Clean separation of concerns

2. Event-Driven Design

• Asynchronous processing of network events
• Non-blocking notification delivery
• Scalable to many notifees

3. Resource Management

• Automatic cleanup of resources
• Proper lifecycle management
• Memory leak prevention

4. Extensibility

• Easy to add new notification types
• Components can register/unregister dynamically
• Flexible event handling

Integration Patterns

1. Registration Pattern

# Component registers with swarm
swarm.register_notifee(MyNotifee())

2. Event Handling Pattern

class MyNotifee(INotifee):
    async def connected(self, network: INetwork, conn: INetConn) -> None:
        # Handle new connection
        pass
    
    async def disconnected(self, network: INetwork, conn: INetConn) -> None:
        # Handle connection loss
        pass

3. Queue-Based Processing

# PubSub uses queues for async processing
peer_send, peer_receive = trio.open_memory_channel[ID](0)
dead_peer_send, dead_peer_receive = trio.open_memory_channel[ID](0)

Future Extensions

The notification system is designed to be extensible. Potential future additions could include:

1. Performance metrics notifications
2. Security event notifications
3. Protocol-specific notifications
4. Network topology change notifications
5. Resource usage notifications

Conclusion

The notification system is fundamental to py-libp2p's architecture, providing a robust, scalable way for components to stay synchronized with network state changes. It enables the complex peer-to-peer networking functionality while maintaining clean, decoupled code architecture.

[seetadev]

@acul71 : Great, thank you so much Luca for the wonderful efforts and this great write-up. Appreciate your initiative.