[ColbySerpa]

Summary

peeroxide's streams are now faster than Node.js's. Before this, a transfer ran at under 1 MB/s, whatever the network.

	Before	Now	Node.js (udx-native)
Windows, same machine, 50 MB	0.1 MB/s	290 MB/s	240 MB/s
Linux, same machine, 50 MB	0.9 MB/s	900 MB/s	600 MB/s
Simulated 100 Mbit/s link, 40 ms round trip, 10 MB	0.4 MB/s	9.7 MB/s	8.1 MB/s

• It also fixes Windows, where a node stopped receiving after one packet it sent went undelivered.
• No unsafe code, and no existing function signature changes.
• This replaces fix(libudx): keep receiving after an undelivered datagram on Windows #34 and perf(libudx): make streams as fast as udx-native by porting C libudx's send path #35.

Motivation

A stream sent one packet each time its timer fired: about every millisecond on Linux, and every 15.6 ms on Windows. That held a transfer to about 1 MB/s on Linux and 75 KB/s on Windows.

Changes

Crate	Change
libudx	The sending side now follows C libudx: its write queue, congestion control (BBR), pacing, loss detection and timers.
libudx	Selective acknowledgements from Node.js peers are read. They were ignored.
libudx	A packet's size counts the IP and UDP headers, so full packets are no longer split in two on the way.
libudx	On Windows, a socket keeps receiving after a packet it sent goes undelivered (was #34).
libudx	Two additions that C libudx doesn't have keep a burst of lost packets from slowing a stream. PROTOCOL.md describes both.
libudx	On Linux, packets are marked "don't fragment", as C libudx marks them, through rustix: a new dependency on Linux only, already in Cargo.lock.
libudx	A socket asks the system for the send and receive buffers C libudx asks for, through socket2: a new dependency, already in Cargo.lock.
libudx	A stream carries on where its sequence numbers wrap, after 2^32 packets. PROTOCOL.md listed that as a known limitation.
libudx	An option for servers at very high scale: a stream can be given a limit on what it holds in memory for its reader (set_read_buffer_limit). It is off by default. With it comes the zero-window probe, which PROTOCOL.md listed as missing.
peeroxide-dht	An encrypted frame's length and body go out in one write.
peeroxide-dht	SecretStream::get_ref reaches the stream under an encrypted connection, to set that limit.
libudx, peeroxide-dht	An example each that measures a transfer, to this crate or to a Node.js peer.
peeroxide-dht, peeroxide-cli	Two tests that failed from timing alone are fixed: one failed on macOS, one on WSL2.
CI	Tests and clippy also run on macOS.

Public API Changes

All changes below are: [x] Additive only (semver-compatible). No existing signature changes. Three behaviours change.

Crate	Symbol	Change
libudx	UdxAsyncStream::effective_mtu	New.
libudx	UdxStream::set_read_buffer_limit and read_buffer_limit, and both on UdxAsyncStream	New. A limit on what a stream holds for its reader. Off by default.
peeroxide-dht	SecretStream::get_ref	New.
libudx	UdxAsyncStream	A write returns once its bytes are queued, and flush returns at once. shutdown still waits until everything is acknowledged.
libudx	effective_mtu	Now the size of the whole IP packet, as in udx-native. Still 1200 to 1500.
libudx	UdxSocket::send_to	On Linux, a datagram too large for the path is dropped, where it was fragmented.

Testing

• cargo test --workspace
• cargo test -p peeroxide-cli --test live_commands -- --ignored
• cargo clippy --workspace --all-targets -- -D warnings

Also cargo doc, a check with Rust 1.85.0, and cargo package.

The CI passes on the fork, on GitHub's Linux and macOS runners.

The speeds are medians of 5 to 12 runs on one machine: Windows 11, and Linux under WSL2. The link is a local relay program that sets rate, delay, queue and loss. Mixed pairs, with this crate at one end and udx-native at the other, were measured too, with 0 to 5% of packets lost: each is as fast as two udx-native ends, or faster. The commit messages have every table.

Notes

• Only Linux has a safe call for "don't fragment". On Windows and macOS, a path that carries less than 1500 bytes still splits full packets in two, as before.
• Node.js's Linux figure is from its runs with no retransmission timeout. One struck about half its runs there, and each costs it a second.

[eshork]

Thanks for this, [@ColbySerpa](https://github.com/ColbySerpa). It's a big step up for libudx, and the C references made the port easy to follow. CI and our local suites, including the live network tests, pass on the head, and the speed numbers are impressive.

We found three issues a remote peer can trigger, and we'd like them fixed before merging:

1. A SACK range that wraps past u32::MAX panics the stream's task, and the writer then hangs forever.

2. DATA with seq = u32::MAX overflows SACK generation in the socket's shared receive task, which stops receiving for every stream on that socket.

3. UdxAsyncStream::shutdown() before the firewall hook fires returns Ok before anything is acknowledged.

Each inline comment has a suggested fix. I reproduced all three, and I checked the fixes locally: the reproduction tests pass, the workspace suite passes, clippy is clean, and the code builds on Rust 1.85. Reproduction tests are in the details block below; please add them as regression tests.

Please also mention the new socket2 dependency in the PR description.

[eshork]

Blocker: a remote peer can panic the stream's processor task here.

seq_after(range.end, range.start) is wrap-aware, so a range like 0xFFFF_FFF0..0x10 passes. BTreeMap::range then panics ("range start is greater than range end"), because start > end numerically and outgoing is non-empty. That is the normal state during a transfer.

What follows:

• The panic kills process_incoming, so terminate() never runs.
• The oneshots in pending_writes stay alive, so UdxStream::write/shutdown and poll_shutdown wait forever.
• poll_write stays Pending.
• Under panic = "abort", the whole process dies.

Main's for seq in range.start..range.end was merely empty for such a range.

Fix: validate each range the way C does (udx_sack_is_valid), then split a valid range that crosses the wrap into two lookups:

Suggested change

	if !seq_after(range.end, range.start) {
	continue;
	}
	if pkt.is_mtu_probe && self.mtu_state == MtuState::Search {
	self.on_mtu_probe_acked();
	// Only the packets still held, however wide the range claims to be.
	let sacked: Vec<u32> = self
	.outgoing
	.range(range.start..range.end)
	.map(|(&seq, _)| seq)
	.collect();
	if !self.sack_is_valid(range) {
	continue;
	}
	// Only the packets still held, however wide the range claims to be.
	// A valid range may still cross the wrap of the sequence space.
	let sacked: Vec<u32> = if range.start <= range.end {
	self.outgoing.range(range.start..range.end).map(|(&seq, _)| seq).collect()
	} else {
	self.outgoing
	.range(range.start..)
	.chain(self.outgoing.range(..range.end))
	.map(|(&seq, _)| seq)
	.collect()
	};

with:

/// Whether a SACK range lies within what was sent and is not yet
/// acknowledged. C: `udx_sack_is_valid`.
fn sack_is_valid(&self, range: &SackRange) -> bool {
    !seq_after(range.end, self.next_seq)
        && seq_after(range.end, range.start)
        && seq_after(self.next_seq, range.start)
        && !seq_after(self.remote_acked, range.start)
}

Only "skip if start > end" was not enough. It drops ranges that are legitimately wrapped, and still accepts ranges outside the window. Validating keeps us at parity with C's semantics.

Please add a regression test: the first repro in the review body, plus a unit test of sack_is_valid that includes a window crossing the wrap.

[eshork]

Blocker: one DATA packet with seq = u32::MAX takes down the socket's receive task.

receive_data buffers any out-of-order seq. Then first + 1 (and seq + 1 below) overflows. Main had the same arithmetic. But this PR builds the ack inside the socket's shared receive task (take_in → send_ack → take_ack → build_ack → here), so in debug builds the panic now stops receiving for every stream on the socket and for DHT datagrams. In release builds it wraps and sends {u32::MAX, 0}, which triggers the process_ack panic on a remote running this code.

Fix: use wrapping_add for all three + 1s in this function, i.e. first.wrapping_add(1) and seq.wrapping_add(1). Also make acceptance in receive_data wrap-aware (next comment).

Moving ack building back to the stream's own task would also avoid this, but it gives up one of this PR's wins, so we'd rather fix the arithmetic.

Please add a regression test: the second repro in the review body, plus a unit test with next_remote_seq = u32::MAX - 2, receiving u32::MAX out of order, that expects the range (u32::MAX, 0).

[eshork]

Part of the seq = u32::MAX fix: make this comparison wrap-aware, as C's seq_compare(stream->ack, seq) <= 0 is. Then a seq far "ahead" (≥ 2³¹) is treated as old and dropped, instead of being buffered:

Suggested change

	// Duplicate or already-received packet — ignore
	if seq < self.next_remote_seq {
	// Duplicate or already-received packet — ignore. C: `seq_compare`.
	if seq != self.next_remote_seq && !seq_after(seq, self.next_remote_seq) {

The same applies to:

• self.next_remote_seq = seq + 1 and self.next_remote_seq += 1 just below. Both should use wrapping_add(1).
• had_already in take_in, which uses the same numeric header.seq < guard.next_remote_seq.

[eshork]

Blocker (with the next two comments): UdxAsyncStream::shutdown() returns Ok before anything is acknowledged when the firewall hook hasn't fired yet.

The description says "shutdown still waits until everything is acknowledged". That holds once the stream is connected, but not before the hook fires:

• check_writable accepts writes while the hook is armed.
• This early return then reports success without queuing the end or waiting for anything.

I checked it: on a hook-armed stream with no peer, write_all(b"hello") followed by shutdown() returns Ok(()) immediately on this branch. On main, the same shutdown() is still pending after 500 ms, because poll_shutdown waited for the pending write's ack first. That false Ok is the regression.

(Separately, and not new: if the owner drops the stream at that point, the queued bytes are never sent, on main as well. The linger comment below fixes this too, so a shutdown() that now waits correctly isn't undone by a drop.)

For reference: C libudx never reaches this state. udx_stream_write and udx_stream_write_end both return UV_ENOTCONN until udx_stream_connect, and udx-native's _writev throws "Writing while not connected not currently supported". We can't adopt that here without changing the existing behaviour of writes before the hook, so we'd like the end queued after those writes instead.

Fix: keep the early Ok only when there is nothing to deliver. Otherwise queue the end after the writes, through an adapter-only variant of queue_end (next comment):

Suggested change

	if !inner.connected {
	return Poll::Ready(Ok(()));
	}
	Poll::Pending => return Poll::Pending,
	inner.queue_end()
	// Nothing written, and nowhere to send the end: done. A
	// stream waiting on its firewall hook with writes queued
	// sends the end after them once the hook lets a remote in.
	if !inner.connected && (!inner.awaiting_firewall() || inner.drained()) {
	return Poll::Ready(Ok(()));
	}
	inner.queue_end_awaiting_firewall()

This needs drained() to be pub(crate), plus a small accessor in stream.rs:

pub(crate) fn awaiting_firewall(&self) -> bool {
    self.firewall_hook.is_some()
}

If the hook rejects the remote, terminate() fails the end's oneshot, so shutdown() returns an error instead of a false Ok.

Please add regression tests: the third and fourth repros in the review body.

[eshork]

For the shutdown()-before-hook fix: please leave queue_end, and with it UdxStream::shutdown(), exactly as it is. It should still fail at once with "stream not connected" before the hook fires. We don't want this PR to change a public method's behaviour, and a graceful public variant can come in a follow-up.

Instead, add a crate-internal variant that only the async adapter uses, which also accepts a stream waiting on its hook:

Suggested change

	/// Queue the end of the stream, after everything written so far, and
	/// return a receiver that is told once the remote has acknowledged it.
	/// C: `udx_stream_write_end`.
	pub(crate) fn queue_end(&mut self) -> Result<oneshot::Receiver<Result<()>>> {
	if let Some(error) = self.failure() {
	return Err(UdxError::Io(error));
	}
	if self.terminated {
	return Err(UdxError::StreamClosed);
	}
	if !self.connected {
	return Err(UdxError::Io(std::io::Error::other("stream not connected")));
	}
	if self.ended {
	return Err(UdxError::Io(std::io::Error::other("stream already shut down")));
	}
	if self.udp.is_none() || self.remote_addr.is_none() {
	return Err(UdxError::StreamClosed);
	}
	self.ended = true;
	let udp = self.udp.clone().ok_or(UdxError::StreamClosed)?;
	let remote_addr = self.remote_addr.ok_or(UdxError::StreamClosed)?;
	let remote_id = self.remote_id;
	let (tx, rx) = oneshot::channel();
	self.push_write(QueuedWrite { data: Vec::new(), taken: 0, end: true, tx: Some(tx) });
	Ok(rx)
	}
	/// Queue the end of the stream, after everything written so far, and
	/// return a receiver that is told once the remote has acknowledged it.
	/// C: `udx_stream_write_end`.
	pub(crate) fn queue_end(&mut self) -> Result<oneshot::Receiver<Result<()>>> {
	self.queue_end_inner(false)
	}
	/// As [`queue_end`](Self::queue_end), and also on a stream still waiting
	/// on its firewall hook: the end then goes out after what was written,
	/// once the hook lets a remote in. For the async adapter's shutdown,
	/// whose writes are already queued that way.
	pub(crate) fn queue_end_awaiting_firewall(&mut self) -> Result<oneshot::Receiver<Result<()>>> {
	self.queue_end_inner(true)
	}
	fn queue_end_inner(&mut self, awaiting_firewall_ok: bool) -> Result<oneshot::Receiver<Result<()>>> {
	if let Some(error) = self.failure() {
	return Err(UdxError::Io(error));
	}
	if self.terminated {
	return Err(UdxError::StreamClosed);
	}
	let awaiting_firewall = awaiting_firewall_ok && self.firewall_hook.is_some();
	if !self.connected && !awaiting_firewall {
	return Err(UdxError::Io(std::io::Error::other("stream not connected")));
	}
	if self.ended {
	return Err(UdxError::Io(std::io::Error::other("stream already shut down")));
	}
	// A packet is addressed when it is sent, so the remote's address is
	// needed only once the stream is connected.
	if self.udp.is_none() || (self.connected && self.remote_addr.is_none()) {
	return Err(UdxError::StreamClosed);
	}
	self.ended = true;
	let (tx, rx) = oneshot::channel();
	self.push_write(QueuedWrite { data: Vec::new(), taken: 0, end: true, tx: Some(tx) });
	Ok(rx)
	}

[eshork]

For the shutdown()-before-hook fix: a stream that is dropped while its hook is still armed and writes are queued should linger, as a connected one does. Otherwise the processor exits before the hook can fire:

Suggested change

	let undelivered = (guard.connected && !guard.drained()).then_some(guard.remote_acked);
	let undelivered = ((guard.connected || guard.firewall_hook.is_some()) && !guard.drained())
	.then_some(guard.remote_acked);

The wait stays bounded by LINGER_TIMEOUT if no remote ever arrives.

[ColbySerpa]

Thanks for the careful review, and for writing the fixes out. All three are fixed, in three commits on top of the branch:

• a7bf755: SACK ranges are checked with sack_is_valid, and the lookup handles a valid range that crosses the wrap.
• 14a918b: wrapping_add in all three places of sack_ranges, the wrap-aware comparison in receive_data and for had_already in take_in, and wrapping_add where next_remote_seq advances.
• a8e41f6: UdxAsyncStream::shutdown() waits before the hook fires, through queue_end_awaiting_firewall. queue_end and UdxStream::shutdown() are as they were, and a stream dropped with its hook still armed lingers.

Each suggestion is applied as you wrote it.

The details block with your reproduction tests didn't come through in the review, so I wrote the tests from your descriptions: 13 of them, including the unit test of sack_is_valid with a window crossing the wrap, and the one that expects the range (u32::MAX, 0). I checked them by taking each fix out again: every time, one of the new tests fails. If your repros cover something these don't, send them and I'll add them.

There are three more commits, which you didn't ask for.

82d3518 makes a stream safe across the wrap of its own sequence numbers too, which PROTOCOL.md listed under Known Limitations: the cumulative ACK (split_off(&ack)), ack_threshold <= ack, an ACK number of zero, and the order in which packets are sent again and ranges are named. It seemed wrong to fix the wrap for what a remote sends and leave it for a stream's own numbers. Its main test sends 1,200 packets over a simulated lossy path with the wrap in thirteen different places, and requires the same packets every time. A transfer is as fast as before.

ea6b524 adds an option, and it is off by default: set_read_buffer_limit puts a limit on what a stream holds in memory for its reader. Most programs won't need it. It's for applications at very high scale: a server with a great many connections at once, taking in large amounts of data faster than it can write it away, can be overwhelmed, because a stream keeps everything a peer sends until its reader takes it, as the Node.js version (udx-native) does. On one stream, 100,000 unread packets were 115 MB. With a limit, a stream advertises the room it has left, the way C libudx does for a program that sets get_read_buffer_size, and turns away what doesn't fit, so what each stream holds is bounded and a sender is held to the reader's pace. A stream without a limit takes everything and advertises 4 MB, as before, and is as fast.

The Node.js version has no such option on its own side, but a sender on the Node.js version does keep to the limit of a stream on this crate. For that the commit numbers acknowledgements the way C libudx does: numbered zero, the Node.js version ignored the window they carry, and sent a third of its packets again in the new interop test. One difference remains: a sender on the Node.js version gives up on a stream whose reader stops altogether for about seventeen seconds, where a sender on this crate goes on asking. The commit also adds the zero-window probe that PROTOCOL.md listed as missing, so that a closed window opens again. PROTOCOL.md has a new section, Flow Control, with the details.

32c6781 adds SecretStream::get_ref, so that the limit can be set on an encrypted connection.

On the new head, cargo test --workspace, the live tests, clippy, cargo doc, the check on Rust 1.85 and cargo package all pass here, under Linux. The description now mentions socket2, the wrap and the limit, and the changelogs cover all of it.