0.7.1 bug fixes

.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

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
 

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 = [

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<ChannelMessage>`'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() {

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<ChannelMessage>`'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(())
+}