diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f8589d1..8e7cfb4 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -34,7 +34,6 @@ jobs: cargo check --all-targets --no-default-features cargo test --features js_tests cargo test --no-default-features --features js_tests - cargo test --benches build-extra: runs-on: ubuntu-latest diff --git a/CHANGELOG.md b/CHANGELOG.md index d017ea5..a723de2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,12 +2,17 @@ All notable changes to this Rust implementation of hypercore-protocol will be documented here. -### unreleased +### 7.0.1 + +* Rewrite serveral async methods to return owned futures and fix a busy loop ([PR 148](https://github.com/datrs/hypercore-protocol-rs/pull/24)). + +### 7.0.0 BIG CHANGES: * Encryption and framing of streams has been moved out of this crate into `hypercore_handshake` and `uint24le_framing` respectively. This had big impacts on the public API. Now `Protocol::new` just takes a `impl CipherTrait` argument. -* Remove dependence on `hypercore` instead we use `hypercore_schema`. +* Remove dependence on `hypercore` instead we use `hypercore_schema` (so hypercore related features have been removed). * Bumped to edition 2024. +* Dropped support for async-std (and its feature flag) ### 0.6.1 diff --git a/Cargo.toml b/Cargo.toml index dbf4c86..5e7a3cc 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "hypercore-protocol" -version = "0.7.0" +version = "0.7.1" license = "MIT OR Apache-2.0" description = "Replication protocol for Hypercore feeds" authors = [ diff --git a/src/mqueue.rs b/src/mqueue.rs index 1bd912c..348f503 100644 --- a/src/mqueue.rs +++ b/src/mqueue.rs @@ -16,7 +16,7 @@ use futures::{Sink, Stream}; use hypercore_handshake::{CipherTrait, state_machine::PUBLIC_KEYLEN}; use tracing::{error, instrument, trace}; -use crate::message::ChannelMessage; +use crate::message::{ChannelMessage, Message}; /// Message IO layer that encodes/decodes `ChannelMessage` over a byte stream. /// @@ -73,10 +73,24 @@ impl MessageIo { break; } - // Batch all queued messages + // Batch queued messages, but never batch an `Open`/`Close` message together + // with anything else. `Vec`'s multi-message wire encoding + // groups messages by channel number to avoid repeating it per message, but + // `Open`/`Close` don't have an outer channel number at all (it's embedded in + // their own payload, framed via a dedicated 2-byte prefix) — only the + // single-message path encodes them correctly. So an `Open`/`Close` at the + // front of the queue is sent alone; a batch otherwise stops right before one. let mut messages = vec![]; - while let Some(msg) = self.write_queue.pop_front() { - messages.push(msg); + while let Some(front) = self.write_queue.front() { + let front_is_open_or_close = + matches!(front.message, Message::Open(_) | Message::Close(_)); + if front_is_open_or_close && !messages.is_empty() { + break; + } + messages.push(self.write_queue.pop_front().expect("front just checked")); + if front_is_open_or_close { + break; + } } let buf = match messages.to_encoded_bytes() { diff --git a/tests/basic.rs b/tests/basic.rs index 0618d74..22c4c35 100644 --- a/tests/basic.rs +++ b/tests/basic.rs @@ -176,3 +176,38 @@ async fn open_close_channels() -> anyhow::Result<()> { fn want(start: u64, length: u64) -> Message { Message::Want(Want { start, length }) } + +/// Regression test: two `Open` messages queued before the first flush must not get batched +/// into one multi-message write. `Vec`'s multi-message encoding groups +/// messages by channel number to save repeating it, but `Open`/`Close` don't have an outer +/// channel number at all (it's embedded in their own payload, framed via a dedicated 2-byte +/// prefix) — only the single-message path encodes them correctly. Unlike `open_close_channels` +/// above (which fully establishes key1 before ever opening key2, so each `Open` is always +/// flushed alone), this opens both keys back-to-back with no driving in between, so they're +/// still queued together when the first flush happens. +#[tokio::test] +async fn two_opens_queued_before_first_flush() -> anyhow::Result<()> { + let (proto_a, proto_b) = create_pair(); + + let key1 = [4u8; 32]; + let key2 = [5u8; 32]; + + proto_a.open(key1).await?; + proto_a.open(key2).await?; + proto_b.open(key1).await?; + proto_b.open(key2).await?; + + let next_a = drive_until_channel(proto_a); + let next_b = drive_until_channel(proto_b); + let (proto_a, _channel_a1) = next_a.await??; + let (proto_b, _channel_b1) = next_b.await??; + + let next_a = drive_until_channel(proto_a); + let next_b = drive_until_channel(proto_b); + let (proto_a, _channel_a2) = next_a.await??; + let (proto_b, _channel_b2) = next_b.await??; + + assert_eq!(proto_a.channels().count(), 2); + assert_eq!(proto_b.channels().count(), 2); + Ok(()) +}