Proposal: Add a Stream State Machine to py-libp2p

[seetadev]

Description

We propose introducing a formal Stream State mechanism in py-libp2p to track the lifecycle of a Stream. This would allow both internal components and external consumers to determine whether a stream is open, closed, reset, or has errored.

This mirrors behavior in go-libp2p and js-libp2p, and is aligned with good practices from transport protocols like TCP and QUIC.

Motivation

Currently, py-libp2p has no explicit notion of a stream’s state. As a result:

• Errors from operations on a closed or reset stream are only caught at runtime.
• There’s no unified API to check whether a stream is in a valid operational state.
• Bugs like double-close, write-after-close, or inconsistent state transitions are harder to catch.
• This makes reasoning about stream lifecycle harder for transport and application layers.

Introducing a structured StreamState will:

• Improve debuggability and error prevention
• Allow future stream muxers (e.g., Yamux, Mplex) to better coordinate state
• Enable uniform stream lifecycle handling across libp2p implementations

Current Implementation

• There is no state tracking today in the Stream class.
• Streams are managed directly using asyncio reader/writer objects.
• No guards exist for write/read/reset based on state.

Proposed Change

Step 1: Define a StreamState Enum

from enum import Enum, auto

class StreamState(Enum):
    INIT = auto()       # Created, not yet started
    OPEN = auto()       # Ready and transmitting data
    RESET = auto()      # Reset by remote or local peer
    CLOSED = auto()     # Gracefully closed
    ERRORED = auto()    # Unexpected failure

Step 2: Add State to Stream Class

class Stream:
    def __init__(...):
        self._state = StreamState.INIT

Update transitions in:

• start() → OPEN
• reset() → RESET
• close() → CLOSED
• on Exception → ERRORED

Expose via:

@property
def state(self) -> StreamState:
    return self._state

Step 3: Guard Operations

Add guards in critical methods:

async def read(self, n):
    if self._state != StreamState.OPEN:
        raise StreamClosedError("Cannot read from stream; not open")

Apply similar logic to write(), reset(), and close().

---

Example Usage

stream = await libp2p_host.new_stream(peer_id)
print(stream.state)  # Should be OPEN
await stream.close()
print(stream.state)  # Should be CLOSED

---

Design Notes

• Aligns with go-libp2p-core/network.Stream semantics.
• State machine supports future enhancements: e.g., idle, half-close, draining.
• Can help stream muxer implementations (like Yamux) enforce protocol correctness.

Testing Plan

• Unit tests for valid and invalid state transitions
• Tests for read/write/reset after close or reset
• Add integration test with a mock stream peer

Next Steps

• Gather feedback from maintainers/community
• Finalize StreamState enum and transition logic
• Implement in Stream class
• Add tests and documentation

Related Topics

• Feature parity with go-libp2p
• Supports ongoing work in stream muxer abstraction
• Future integration with yamux-py or multiplexers for clean lifecycle management