fix(webrtc): decode data-channel framing as a byte stream; SCTP stream-id parity by DTLS role

[yashksaini-coder]

What

Two pre-existing bugs that made every py↔go WebRTC-Direct connection fail, found by running the v1 code live against go-libp2p v0.49.0:

• Framing: we required one complete uvarint + pb.Message per SCTP message, but go-msgio writes the varint prefix and the body as two separate writes → two SCTP messages. py failed at Noise msg#1 with malformed handshake frame length (and WebRTCStream.on_data would have dropped every split frame after that). Now both the Noise channel and streams decode frames from an accumulating byte buffer: split or batched frames are fine, the buffer is bounded by MAX_MESSAGE_SIZE, frames decoded before a malformed one are still delivered, a malformed frame resets the stream (RESET sent to the peer, further bytes ignored), and a whole SCTP message is applied in one trio hop.
• SCTP stream-id parity: aiortc seeds its DCEP id allocator from the ICE role, which is inverted for a WebRTC-Direct listener (ICE-controlled but DTLS server), so our listener's ids collided with go's even dialer ids (assert stream_id not in self._data_channels). We now re-seed the allocator's parity from the DTLS role (RFC 8832: client even, server odd) and let aiortc keep allocating and recycling ids.

Also widens one Windows-sensitive loopback test's budget (two dials + two bounded closes).

Verified

Live against go-libp2p v0.49.0, both directions: go→py and py→go connect and complete Noise. Unit tests for split/batched frames, partial-delivery-then-reset, id parity and id reuse. tests/core/transport/webrtc: 230 passed; mypy/ruff/pyrefly clean.

Refs #1437 — first of two stacked PRs; the v2 flow (#1459) builds on this.

[yashksaini-coder]

@seetadev @acul71 This one's ready for review (thanks for the main syncs!). The two CI reds are main-side flakes, not this diff: test_publish_before_identify_completes and the anyio_service lifecycle test both fail on main's own runs too, and the webrtc suite is 231/231 inside those same jobs. This PR also fixes the [sdp-http-harness] WinError-10013 flake currently visible on main's Windows CI. Verified live against go-libp2p v0.49 in both directions.

[acul71]

Good work overall — the root cause analysis is correct and the fix is sound. Two things I'd like addressed before merging, plus a few smaller notes.

---

✅ What's good

• Root cause correctly identified. go-msgio's pbio fallback writes the varint prefix and body as two separate SCTP messages; the byte-stream accumulation approach (_decode_frames) matches how all mature WebRTC stacks handle this.
• _decode_frames() is a well-designed shared primitive. Generator mutates the buffer in place, yields complete frames, leaves partial tail — clean semantics used consistently by both DataChannelReadWriter and WebRTCStream.on_data.
• Bounds checking is correct. MAX_MESSAGE_SIZE (16 384) enforced before allocating; prefix window capped at _MAX_PREFIX = 3 bytes.
• Atomic Trio-thread application. All frames from one SCTP message are applied in a single _run_on_trio_thread hop — no interleaving.
• SCTP parity fix is minimal. Re-seeding only _data_channel_id (which aiortc bumps by 2) lets aiortc keep allocating and recycling IDs normally.
• Tests are thorough: split frame across 3 SCTP messages, batched frames, partial-delivery-then-reset, DTLS role parity (client even, server odd), ID recycling after channel close, harness TCP bind retry.
• CI results (run locally on this branch): make lint ✅, make typecheck ✅, make linux-docs ✅. make test: 3449 passed / 3 failed — all 3 failures are kad_dht assertion flakes unrelated to this PR (none of the touched files are in kad_dht). WebRTC suite: 231/231 passed.
• Newsfragment newsfragments/1437.bugfix.rst present, correct format, ends with newline ✅.

---

🔴 M1 — Hard assert in production channel-open code (_aiortc_helpers.py ~line 358)

assert hasattr(dtls, "_role"), "RTCDtlsTransport has no attribute '_role'"

_role is a private aiortc attribute. If a future aiortc upgrade renames or removes it, every open_stream() call will crash with AssertionError (not a clean WebRTCConnectionError). The "auto" fallback path already exists right below — please use it:

if not hasattr(dtls, "_role"):
    logger.warning(
        "RTCDtlsTransport._role missing (aiortc API changed?); "
        "using default SCTP stream-id parity"
    )
else:
    role = dtls._role

This is the same defensive pattern as the rest of the function. The assert style is fine inside set_private_attr (a dev-time guard), but not in a live connection path.

---

🟡 M2 — Latent head[-1] access when consumed == 0 in _decode_frames (stream.py ~line 534)

length, consumed = decode_varint_with_size(bytes(head))
if head[consumed - 1] & 0x80:   # ← head[-1] if consumed == 0

In practice consumed >= 1 for any non-empty input with the current varint implementation, and while buf: guards the outer loop. But if decode_varint_with_size ever returns (0, 0) for a non-empty byte (implementation contract not documented), head[-1] silently checks the wrong byte. One-line fix:

if consumed == 0:
    return  # wait for more bytes

immediately after the decode_varint_with_size call.

---

🟡 M3 — _reset_locally() not idempotent (stream.py)

A batch containing a peer-sent RESET flag followed by a malformed frame causes _reset_locally() to be called twice (once from _apply_batch_on_trio_thread, once from _reset_on_trio_thread). The second call is benign today but _enqueue_eof_sentinel_locked is not guarded. Easy fix:

def _reset_locally(self) -> None:
    if self._state == StreamState.RESET:
        return
    self._state = StreamState.RESET
    self._enqueue_eof_sentinel_locked()

---

🔵 Minor / Nice-to-have

• m1 — newsfragments/1437.bugfix.rst is very long (500+ word single paragraph). Towncrier reflows it, but 2–3 concise user-facing bullets would read better in the changelog.
• m2 — test_ids_follow_dtls_role_and_are_recycled uses asyncio.run() in a plain def test. Inconsistent with the @pytest.mark.trio convention. Consider restructuring as a @pytest.mark.trio async def test.
• m3 — Redundant assert hasattr pattern (the manual assert on _role + the one inside set_private_attr mix two styles — just use one).

---

❓ Questions for the author

1. [M1] Was RTCDtlsTransport._role present since aiortc 1.15 (the floor in pyproject.toml)? Is there a CI check that would catch a private rename?
2. [M2] Is there a test that exercises _decode_frames with consumed == 0 from a non-empty buffer?
3. [DataChannelReadWriter._fill] When a FIN is received, _closed = True is set and the loop breaks, leaving any further frames in self._raw unprocessed. Is the invariant "Noise handshake FIN frames carry no payload" documented anywhere?
4. [Loopback timeout] The 30 s → 60 s widening: was this observed to be flaky on Linux CI too, or only Windows?
5. [Stacked PRs] Is #1460 safe to merge independently and #1459 rebased on top, or should both land together?

---

Summary

Quality	Good
Security impact	Low
Merge readiness	Needs fixes — please address M1 and M2 (both are one-liners), and optionally M3
Confidence	High

The framing and SCTP-parity fixes are correct and critical for go↔py interop. M1 is the main blocker (a hard assert in a live connection path). M2 and M3 are easy defensive fixes worth adding. Everything else is polish.

Requesting changes on M1 + M2; the rest is up to you.

[yashksaini-coder]

@acul71 All addressed in d87292c1:

• M1 — _role read no longer asserts: missing attribute degrades to aiortc's default parity with a logger.warning. (Kept the set_private_attr assert style only for the dev-time guard, per your m3.)
• M2 — consumed == 0 now raises ValueError("empty varint length prefix"). I chose raise over return: returning would keep growing the buffer forever on a decoder-contract violation, while raising routes through the existing malformed-frame → RESET path. New test test_decode_frames_rejects_zero_consumed patches the decoder to return (0, 0).
• M3 — _reset_locally is idempotent, and _apply_batch_on_trio_thread stops delivering a batch once the stream is RESET (also covers feat(webrtc): WebRTC-Direct v2 flow (libp2p/specs#715) — listener accepts v2, dialer opt-in #1459's M2), each with a test.
• m1 — newsfragment wrapped. m2 — kept asyncio.run there deliberately: test_aiortc_helpers.py/test_udp_mux.py document that convention ("sync wrappers so they run outside trio_mode"); happy to convert if you'd rather.

Your questions: (1) _role exists throughout aiortc ≥1.15 (our floor); a future rename now warns + falls back, and the parity/recycling loopback test fails loudly on an actual collision. (2) Yes — added, see M2. (3) Documented now in _fill's docstring: frames after FIN/RESET are dropped by design; go never closes channel 0, and py↔py tears the channel down at FIN. (4) Windows only — never seen on Linux. (5) Either works: my preference is merge #1460 first (standalone v1 bugfix) and I rebase #1459, but merging #1459 and closing this as superseded (like the #1449 stack) is equally fine.

[acul71]

Superseded by #1459, which already includes this stack (framing as a byte stream + SCTP stream-id parity by DTLS role).