fix: publish stream close verdict before reader EOF

CHANGELOG.md

@@ -1,5 +1,12 @@
 # Changelog
 
+## 0.3.3
+
+### Fixed
+
+- Publish a stream's completion or truncation verdict before closing its reader
+  channel, so a reader that has drained the payload observes the correct end.
+
 ## 0.3.2
 
 ### Added

Cargo.toml

@@ -11,16 +11,16 @@ exclude = ["examples"]
 resolver = "3"
 
 [workspace.package]
-version = "0.3.2"
+version = "0.3.3"
 edition = "2024"
 rust-version = "1.85"
 license = "GPL-3.0-only"
 repository = "https://github.com/tinyhumansai/tinybus"
 
 [workspace.dependencies]
-tinybus = { path = "crates/tinybus", version = "=0.3.2", default-features = false }
-tinybus-macros = { path = "crates/tinybus-macros", version = "=0.3.2" }
-tinybus-module = { path = "crates/tinybus-module", version = "=0.3.2" }
+tinybus = { path = "crates/tinybus", version = "=0.3.3", default-features = false }
+tinybus-macros = { path = "crates/tinybus-macros", version = "=0.3.3" }
+tinybus-module = { path = "crates/tinybus-module", version = "=0.3.3" }
 tokio = { version = "1", features = ["rt-multi-thread", "sync", "time"] }
 
 async-trait = "0.1"

crates/tinybus/src/module/client_pinned_tests.rs

@@ -13,7 +13,7 @@ static RECORD: ReleaseRecord = ReleaseRecord {
     id: "ordinary-pinned-client-test",
     bus_name: "ai.tinyhumans.test.OrdinaryPinnedClient",
     object_path: "/ai/tinyhumans/test/OrdinaryPinnedClient",
-    version: "0.3.2",
+    version: "0.3.3",
     release_url: "https://example.invalid/tinybus",
     // Ordinary fixtures and linked host factories intentionally have no
     // release archive/library pair to attest.
@@ -284,7 +284,7 @@ fn linked_ordinary_factory() -> crate::Result<crate::module::LinkedModule> {
             schema: MANIFEST_SCHEMA,
             module: ModuleIdentity {
                 name: RECORD.id.to_owned(),
-                version: Version::new(0, 3, 2),
+                version: Version::new(0, 3, 3),
                 description: String::new(),
                 homepage: None,
                 license: String::new(),
@@ -330,7 +330,7 @@ static RESOLVER_LINKED_RECORD: ReleaseRecord = ReleaseRecord {
     id: "ordinary-binding-linked-readiness",
     bus_name: "ai.tinyhumans.test.OrdinaryBindingLinkedReadiness",
     object_path: "/ai/tinyhumans/test/OrdinaryBindingLinkedReadiness",
-    version: "0.3.2",
+    version: "0.3.3",
     release_url: "https://example.invalid/tinybus",
     assets: &[],
 };
@@ -411,7 +411,7 @@ fn resolver_linked_factory() -> crate::Result<crate::module::LinkedModule> {
             schema: MANIFEST_SCHEMA,
             module: ModuleIdentity {
                 name: record.id.to_owned(),
-                version: Version::new(0, 3, 2),
+                version: Version::new(0, 3, 3),
                 description: String::new(),
                 homepage: None,
                 license: String::new(),
@@ -1038,7 +1038,7 @@ async fn ordinary_binding_refuses_an_unsupported_resolver_without_proxy_or_busin
         id: "ordinary-binding-unsupported",
         bus_name: "ai.tinyhumans.test.OrdinaryBindingUnsupported",
         object_path: "/ai/tinyhumans/test/OrdinaryBindingUnsupported",
-        version: "0.3.2",
+        version: "0.3.3",
         release_url: "https://example.invalid/tinybus",
         assets: &[],
     };

crates/tinybus/src/stream/mod.rs

@@ -252,6 +252,8 @@ pub(crate) struct StreamRegistry {
     limits: std::sync::RwLock<StreamLimits>,
     inbound: std::sync::Mutex<HashMap<String, Arc<Inbound>>>,
     next_id: AtomicU64,
+    #[cfg(test)]
+    after_close_seal: std::sync::Mutex<Option<Box<dyn FnOnce() + Send>>>,
 }
 
 impl StreamRegistry {
@@ -260,6 +262,8 @@ impl StreamRegistry {
             limits: std::sync::RwLock::new(StreamLimits::default()),
             inbound: std::sync::Mutex::new(HashMap::new()),
             next_id: AtomicU64::new(1),
+            #[cfg(test)]
+            after_close_seal: std::sync::Mutex::new(None),
         }
     }
 
@@ -462,15 +466,29 @@ impl StreamRegistry {
         // the *fast* case — into "no such stream".
         let received = stream.received.load(Ordering::Relaxed);
         stream.touch();
+        let truncated = received != total_len;
+        if truncated {
+            stream.finish(Outcome::Aborted("the sender closed a truncated stream"));
+        } else {
+            stream.finish(Outcome::Complete);
+        }
         stream.seal();
 
-        if received != total_len {
-            stream.finish(Outcome::Aborted("the sender closed a truncated stream"));
+        #[cfg(test)]
+        if let Some(hook) = self
+            .after_close_seal
+            .lock()
+            .expect("stream close test hook lock")
+            .take()
+        {
+            hook();
+        }
+
+        if truncated {
             return Err(Error::protocol(format!(
                 "stream closed after {received} bytes, {total_len} declared"
             )));
         }
-        stream.finish(Outcome::Complete);
         Ok(Value::Null)
     }
 

crates/tinybus/src/stream/stream_tests.rs

@@ -18,7 +18,8 @@ use crate::message::Message;
 use crate::name::{BusName, InterfaceName, MemberName, ObjectPath};
 use crate::ports::Transport;
 use crate::stream::{
-    MAX_CHUNK_LEN, STREAM_INTERFACE, STREAM_PATH, StreamDescriptor, StreamLimits, StreamRef,
+    MAX_CHUNK_LEN, STREAM_INTERFACE, STREAM_PATH, StreamDescriptor, StreamLimits, StreamReader,
+    StreamRef, StreamRegistry,
 };
 use crate::transport::memory::{MemoryBus, MemoryTransport};
 
@@ -96,6 +97,92 @@ fn payload(len: usize) -> Vec<u8> {
     (0..len).map(|i| (i % 251) as u8).collect()
 }
 
+fn stream_header(member: &str, sender: &BusName) -> crate::message::Header {
+    let mut header = Message::method_call(
+        BusName::new("ai.tinyhumans.StreamReceiver").unwrap(),
+        ObjectPath::new(STREAM_PATH).unwrap(),
+        InterfaceName::new(STREAM_INTERFACE).unwrap(),
+        MemberName::new(member).unwrap(),
+        Value::Null,
+    )
+    .header;
+    header.sender = Some(sender.clone());
+    header
+}
+
+async fn close_verdict_is_published_before_reader_observes_eof(
+    declared_len: u64,
+    close_len: u64,
+    expected_terminal: &str,
+    expected_close_error: Option<&str>,
+) {
+    let registry = Arc::new(StreamRegistry::new());
+    let owner = BusName::new("ai.tinyhumans.StreamSender").unwrap();
+    let open = registry
+        .dispatch(
+            &stream_header("Open", &owner),
+            serde_json::json!([StreamDescriptor::with_len(declared_len)]),
+        )
+        .await
+        .unwrap();
+    let id = open.as_str().unwrap().to_string();
+    let mut reader: StreamReader = registry.take_reader(&id).unwrap();
+    registry
+        .dispatch(
+            &stream_header("Write", &owner),
+            serde_json::json!([id.clone(), 0, super::base64::encode(b"abc")]),
+        )
+        .await
+        .unwrap();
+    assert_eq!(
+        reader.next_chunk().await.unwrap().as_deref(),
+        Some(b"abc".as_slice())
+    );
+
+    let (close_sealed_tx, close_sealed_rx) = tokio::sync::oneshot::channel();
+    let release_close = Arc::new(std::sync::Barrier::new(2));
+    let close_release = Arc::clone(&release_close);
+    *registry
+        .after_close_seal
+        .lock()
+        .expect("stream close test hook lock") = Some(Box::new(move || {
+        let _ = close_sealed_tx.send(());
+        close_release.wait();
+    }));
+
+    let close_registry = Arc::clone(&registry);
+    let close_member = MemberName::new("Close").unwrap();
+    let close_header = stream_header("Close", &owner);
+    let close = tokio::task::spawn_blocking(move || {
+        close_registry.close(
+            &close_header,
+            &close_member,
+            serde_json::json!([id, close_len]),
+        )
+    });
+
+    close_sealed_rx
+        .await
+        .expect("Close should reach the per-registry scheduling hook");
+    let terminal = reader.next_chunk().await;
+    // Always release and join the paused synchronous Close before asserting, so
+    // a deliberately failing RED run cannot strand a blocking task.
+    release_close.wait();
+    let close_result = close.await.expect("Close task should complete");
+
+    let terminal = match terminal {
+        Ok(None) => "clean EOF".to_string(),
+        Ok(Some(_)) => "unexpected extra chunk".to_string(),
+        Err(Error::StreamAborted { reason }) => format!("aborted: {reason}"),
+        Err(error) => format!("error: {}", error.wire_name()),
+    };
+    assert_eq!(terminal, expected_terminal);
+    match expected_close_error {
+        Some(wire_name) => assert_eq!(close_result.unwrap_err().wire_name(), wire_name),
+        None => assert!(close_result.is_ok()),
+    }
+}
+
 /// A broker, a sink service that owns [`SINK`], and a client.
 async fn bus() -> (Connection, Connection) {
     let bus = MemoryBus::new();
@@ -370,6 +457,22 @@ async fn closing_short_of_the_declared_length_is_an_error_not_a_short_read() {
     assert_eq!(error.wire_name(), "ai.tinyhumans.tinybus.Error.Protocol");
 }
 
+#[tokio::test]
+async fn a_complete_close_publishes_its_verdict_before_reader_eof() {
+    close_verdict_is_published_before_reader_observes_eof(3, 3, "clean EOF", None).await;
+}
+
+#[tokio::test]
+async fn a_truncated_close_publishes_its_abort_before_reader_eof() {
+    close_verdict_is_published_before_reader_observes_eof(
+        4,
+        4,
+        "aborted: the sender closed a truncated stream",
+        Some("ai.tinyhumans.tinybus.Error.Protocol"),
+    )
+    .await;
+}
+
 #[tokio::test]
 async fn a_reader_sees_an_error_rather_than_an_eof_when_the_sender_aborts() {
     let (client, service) = bus().await;

scripts/check-coverage.sh

@@ -4,11 +4,11 @@ cd "$(dirname "${BASH_SOURCE[0]}")/.."
 cargo build --locked --example module_clock --example module_clock_two --example module_wrong_target --example module_context --example module_context_eager --example module_host_callback --example module_controlled_init --example module_terminal --no-default-features --features modules,macros,__native-fixture-support
 set -euo pipefail
 target_dir=$(cargo metadata --format-version 1 --no-deps | python3 -c 'import json,sys; print(json.load(sys.stdin)["target_directory"])')
-module=$(find "$target_dir/debug/examples" -maxdepth 1 -type f -name 'libmodule_clock.so' -print -quit)
-module_two=$(find "$target_dir/debug/examples" -maxdepth 1 -type f -name 'libmodule_clock_two.so' -print -quit)
-wrong_target=$(find "$target_dir/debug/examples" -maxdepth 1 -type f -name 'libmodule_wrong_target.so' -print -quit)
-context_module=$(find "$target_dir/debug/examples" -maxdepth 1 -type f -name 'libmodule_context.so' -print -quit)
-context_eager_module=$(find "$target_dir/debug/examples" -maxdepth 1 -type f -name 'libmodule_context_eager.so' -print -quit)
+module=$(find "$target_dir/debug/examples" -maxdepth 1 -type f \( -name 'libmodule_clock.so' -o -name 'libmodule_clock.dylib' \) -print -quit)
+module_two=$(find "$target_dir/debug/examples" -maxdepth 1 -type f \( -name 'libmodule_clock_two.so' -o -name 'libmodule_clock_two.dylib' \) -print -quit)
+wrong_target=$(find "$target_dir/debug/examples" -maxdepth 1 -type f \( -name 'libmodule_wrong_target.so' -o -name 'libmodule_wrong_target.dylib' \) -print -quit)
+context_module=$(find "$target_dir/debug/examples" -maxdepth 1 -type f \( -name 'libmodule_context.so' -o -name 'libmodule_context.dylib' \) -print -quit)
+context_eager_module=$(find "$target_dir/debug/examples" -maxdepth 1 -type f \( -name 'libmodule_context_eager.so' -o -name 'libmodule_context_eager.dylib' \) -print -quit)
 host_callback_module=$(find "$target_dir/debug/examples" -maxdepth 1 -type f \( -name 'libmodule_host_callback.so' -o -name 'libmodule_host_callback.dylib' \) -print -quit)
 controlled_init_module=$(find "$target_dir/debug/examples" -maxdepth 1 -type f \( -name 'libmodule_controlled_init.so' -o -name 'libmodule_controlled_init.dylib' \) -print -quit)
 terminal_module=$(find "$target_dir/debug/examples" -maxdepth 1 -type f \( -name 'libmodule_terminal.so' -o -name 'libmodule_terminal.dylib' \) -print -quit)

scripts/test-features.sh

@@ -1,6 +1,12 @@
 #!/usr/bin/env bash
 set -euo pipefail
 cd "$(dirname "${BASH_SOURCE[0]}")/.."
+case "${RUNNER_OS:-$(uname -s)}" in
+  Linux|Darwin|macOS)
+    python3 scripts/test_check_coverage.py
+    ;;
+  Windows|MINGW*|MSYS*|CYGWIN*) ;;
+esac
 cargo check --locked --workspace --no-default-features
 cargo check --locked --package tinybus --no-default-features --features modules
 cargo metadata --locked --format-version 1 --manifest-path examples/Cargo.toml >/dev/null