Handling Length-Prefixed Message Decoding in js-libp2p WebRTC Streams (interop testing)

[seetadev]

@r-zig : Hi Ron. Thanks so much for reaching out and for sharing the details of your work — it's awesome to hear you're making progress on stream handling over WebRTC in js-libp2p. That’s a tricky but rewarding area, and it’s great to see someone digging into the internals of stream multiplexing and encoding!

You're absolutely on the right track in expecting the decoding logic (like in it-protobuf-stream) to handle stream boundaries by first reading the length-prefix and then pulling the correct number of bytes. That's typically how message framing works over libp2p streams, especially when protobuf or other length-delimited encodings are in play.

The issue you're seeing (Reader.create_typed_array [as create]) usually occurs when the internal buffer doesn't yet contain enough bytes to fulfill a full read, especially after the length prefix has been read but not the payload. In cases like this, it-protobuf-stream or any async iterable parser will throw if the stream ends prematurely or pauses indefinitely without delivering the expected payload size.

A few thoughts and suggestions to consider:

---

✅ Expectations from it-protobuf-stream

You're correct that it's meant to handle length-prefixed streams — it reads the length prefix and then buffers until that many bytes are available before decoding. However, this assumes a "clean" stream — meaning each message is fully encoded and sent as a single, contiguous chunk.

If your sender is chunking or flushing too early (before the full message is passed through), or if the transport is dropping or delaying data, it could cause it-protobuf-stream to throw due to incomplete reads.

---

🔍 Things to check

1. Ensure full message is written
   If you're using something like stream.sink(pushableSource) or await writer.write(message), make sure:

   • The message is fully encoded before writing.
   • You're not doing partial writes or flushing the stream prematurely.

2. Use framing explicitly
   If you're not using it-protobuf-stream.encode() on the sender side, you might want to ensure that you wrap your outgoing messages with a proper length-prefix so that decode() on the receiver side knows what to expect.

   Example:

   const { encode } = require('it-protobuf-stream')
   const encoder = encode(MyProtobufMessage)
   pipe([message], encoder, stream.sink)

3. Fallback to it-length-prefixed
   Some libp2p streams just use it-length-prefixed to wrap byte payloads — especially if the actual encoding (e.g. protobuf, CBOR) is applied separately. You might want to try decoding the raw stream with it-length-prefixed first to see if the message boundaries are what you expect:

   const lp = require('it-length-prefixed')
   pipe(stream.source, lp.decode(), async source => {
     for await (const msg of source) {
       console.log('Message:', msg)
     }
   })

---

🧭 What do libp2p people use?

You're right to ask — libp2p generally doesn’t enforce a single encoding format. Developers often use:

• it-length-prefixed for framing
• it-protobuf-stream or custom stream transformers (like it-json, it-msgpack) for encoding/decoding structured data
• Their own codec stacks (multicodec, multiformats) especially in cross-language cases (e.g. Rust ↔ JS)

So your usage of it-protobuf-stream is valid, but it's worth checking if:

• The messages are encoded correctly and sent in one go.
• The stream is alive long enough to receive full messages.

---

🙌 Happy to help debug

If you want, feel free to share more of the sender and receiver logic — especially how you're encoding and writing to the stream — and I’d be happy to help debug further.

Thanks again for working on this and reaching out — the js-libp2p ecosystem needs contributors like you who aren't afraid to get into the gritty details of stream protocols!

Best,
Manu

[seetadev]

Pointers for Code fix

@r-zig: Thanks for sharing the code — it’s very helpful and well structured. You're clearly encoding messages using FileTransferMessage.encodeDelimited(...) on the sender side, which is the correct way to include a length-prefix that protobufjs (and by extension it-protobuf-stream) expects to parse properly.

From your description and the error:

Error: illegal buffer at Reader.create_typed_array [...]

this strongly indicates that pb.read(...) is trying to decode a message but isn’t receiving a complete length-prefixed buffer. In other words, it's reading the length prefix, then expecting a certain number of bytes to follow, but not getting that full payload — possibly due to:

1. incomplete stream chunking
2. a mismatch in how the message is encoded or framed
3. stream prematurely closed or drained between writes

---

🔍 Let’s analyze both ends step-by-step:

---

✅ Sender Side Highlights:

const buffer = FileTransferMessage.encodeDelimited(message).finish();
await this.peerDialer.send(buffer, peerId, this.fileTransferProtocol);

This is correct — encodeDelimited includes the length prefix and results in a complete, ready-to-send buffer.

The issue is likely not here unless peerDialer.send is chunking that buffer internally in a way that splits the message across multiple internal writes without proper framing.

---

⚠️ Receiver Side Concern:

You’re using:

const pb = pbStream(stream);
const message = await pb.read(FileTransferMessage);

Which is also correct, assuming that stream is an iterable of complete binary chunks (as per libp2p's duplex protocol). But the failure points toward either:

• a corrupted or incomplete message
• stream-level buffering not being sufficient
• or FileTransferMessage.decodeDelimited() expecting more bytes than it gets.

---

✅ Recommendations / Fixes

1. Wrap incoming stream with buffer merging

Ensure that the stream you're passing to pbStream() isn't emitting partial chunks. Use it-concat or it-buffer-stream style wrappers to avoid message fragmentation.

But better: instead of trying to manage that manually, use lp.decode() to handle proper framing.

Try changing readMessages to:

import lp from 'it-length-prefixed';
import all from 'it-all';

async *readMessages(stream) {
    try {
        const source = lp.decode()(stream.source);
        for await (const buf of source) {
            const message = FileTransferMessage.decode(buf);
            yield message;
        }
    } catch (error) {
        console.error('❌ Error decoding message:', error);
    }
}

• This reads the stream one full message at a time, based on the length prefix, just like encodeDelimited() writes it.
• It avoids relying on protobufjs’ internal buffer mechanics (which can be sensitive to partial reads).

---

2. Test Raw Buffer Roundtrip Locally

Just to sanity check:

const original = FileTransferMessage.create({ ... });
const encoded = FileTransferMessage.encodeDelimited(original).finish();
const decoded = FileTransferMessage.decodeDelimited(encoded);
console.log(decoded);

If this fails, there’s a schema mismatch or encoding bug.

---

3. Inspect Stream Source Chunks

Log or debug what’s being emitted from the stream:

for await (const chunk of stream.source) {
    console.log('📦 Received raw chunk:', chunk.length);
}

If you see chunks with length < expected, or multiple tiny ones, then pb.read(...) won't be able to parse messages correctly.

---

🧩 tl;dr Fix

Replace pb.read(...) with lp.decode() + FileTransferMessage.decode(...), and make sure the stream source is not prematurely closed or draining between chunks.

Let me know if you'd like help patching the code directly — I can give you a PR-style diff.

You're really close to getting this solid — it's a common stumbling block with protobuf over streams, and you're handling it with great attention. Once it's stable, you’ve got a great baseline for robust file transfers over libp2p!

[seetadev]

@r-zig: Hi Ron. Thanks for the detailed follow-up — and great job digging deeper into the message behavior! Your observations are sharp and right on target for the kind of edge cases that show up with stream multiplexing and transport abstractions like WebRTC over WebSocket with a libp2p relay in the middle.

Let me break it down and offer some guidance based on your findings:

---

🔍 What You’re Observing

1. The first metadata message sent (145 bytes) sometimes arrives as 140 bytes
2. But with an artificial delay before sending the next message, the metadata message arrives correctly (145 bytes)
3. When messages are sent back-to-back (metadata → first chunk), the received metadata appears truncated or altered
4. Bytes at the beginning of the message differ — suggesting corruption or misalignment
5. You’re wondering if messages could be arriving out-of-order over WebRTC with a WebSocket-based relay

---

✅ Let’s Address It Step-by-Step

1. WebRTC over WebSocket via libp2p-relay: Is out-of-order delivery possible?

In general:

• WebRTC (data channels) is reliable and ordered by default, similar to TCP.
• WebSocket, which underlies many WebRTC relay implementations in libp2p, also provides in-order, reliable delivery.
• libp2p’s stream muxers (like mplex or yamux) preserve message order within each stream.

So under normal conditions, you should not see out-of-order delivery.

---

2. But: Message Fragmentation and Stream Reassembly Can Appear as Corruption

Your description sounds less like out-of-order arrival and more like stream framing confusion, particularly when:

• two messages are sent quickly back-to-back (metadata followed by chunk)
• the receiver starts reading before the full metadata message is available
• there's a race condition or buffering issue on how the stream is read

You mentioned using FileTransferMessage.encodeDelimited(...) — which is good — but the receiver must match it with a decoder that honors length-prefix framing. If you're reading raw bytes or using .read() from a protobuf stream that isn’t properly buffered, you might start decoding the chunk message as if it were still part of the metadata.

This would explain:

• why bytes differ at the start (you're reading part of the next message)
• why the length is 140 instead of 145 (only partial content before stream yields again)

---

🧪 Your Delay Test Confirms It

“When I delayed sending the next chunk, the message came through fine.”

This supports the theory: the stream reader is consuming data too early, not respecting the message boundaries, and mixing the second message's start into the first.

---

✅ Recommendations

1. Ensure proper framing on the receiver

Double-check that your receiver is not manually buffering or decoding messages without using length-prefixed framing.

The safest approach is:

import lp from 'it-length-prefixed'

async *readMessages(stream) {
  const source = lp.decode()(stream.source)
  for await (const buf of source) {
    yield FileTransferMessage.decode(buf)
  }
}

This will ensure that each message starts and ends cleanly, even if multiple messages arrive back-to-back in a single packet.

---

2. Avoid writing to stream too fast without flushing

If you’re using custom pushables or iterables, ensure that:

• you push one message at a time
• optionally, flush after writing (depending on the stream API)
• you give libp2p time to propagate and process the stream data

Sometimes, adding await new Promise(r => setTimeout(r, 10)) between sends (just during testing) can help you see if timing/race issues are the cause.

---

3. Sanity check with a hex dump

Try logging:

console.log('Sent buffer:', buffer.toString('hex').slice(0, 32))

and similarly log what you receive.

Compare the first 32–64 bytes to ensure they match. If not, the stream is likely being misread (not a transmission error, but a framing issue).

---

✅ Summary

So in short:

• No, WebRTC over WebSocket + relay should not deliver messages out of order
• Yes, it’s very possible to misread messages if your stream decoder isn't properly handling length-prefixed framing
• Your test with artificial delay confirms it’s a stream reading issue, not a network one

You're almost there — the good news is that this kind of issue, while subtle, has a well-defined fix once framing is handled cleanly.

Let me know if you’d like a full patched readMessages function or any help inspecting the buffers!