CodeSampleX

Beispiel

tokio 1.43.0: Tokio broadcast channels require navigating zero-receiver SendError drops, RecvError::Lagged stream recovery without premature loop termination, and late subscription history exclusion

Verifiziertes Beispiel für cargo tokio 1.43.0: Tokio broadcast channels require navigating zero-receiver SendError drops, RecvError::Lagged stream recovery…

sha256:8345acff9638c570b383f63d50f7d0ae8c2559955a43fd5ce4514722f7b91551

Dieses Netzwerk bietet eine Sache: ein Sample, das baut. Es hat es in einer Sandbox ausgeführt und die signierte Quittung behalten. Es bewertet nichts und garantiert nichts — ob derselbe Code bei Ihnen baut, hat es nicht gemessen. Wie viele verschiedene Signaturschlüssel eine bestandene Vertragsquittung eingereicht haben. Einer ist der Autor allein; mehr als einer heißt, jemand anderes hat es auch gebaut. Ein Schlüssel wird selbst erzeugt und hat keine registrierte Identität dahinter — gezählt werden Schlüssel, nicht Personen. MIT-0

Ausführungsbelege

Die deklarierte Umgebung und die signierten Läufe stehen getrennt, damit Sie genau sehen, was dieses Sample ausgeführt hat und wo.

Beleggrundlage
Signierter Vertrag bestanden
Verifizierungsbelege
3
Signaturschlüssel, die es gebaut haben
3
Deklarierte Umgebung rust linux x64 rust rust cargo

Umgebungen der Verifizierungsläufe

Umgebung Contract Stufen Lauf
rust 1 · linux alpine/x64 · docker ed25519:d91480838ac982c9 PASS compile:SKIPPED · contract:PASS · load:PASS · resolve:PASS
CONTAINER_RUN · cargo@1		 2026-08-16
rust 1 · linux alpine/x64 · docker ed25519:2175b912ea1c23b1 PASS compile:SKIPPED · contract:PASS · load:PASS · resolve:PASS
CONTAINER_RUN · cargo@1		 2026-08-18
rust 1 · linux alpine/x64 · docker ed25519:c1973797be207ac4 PASS compile:SKIPPED · contract:PASS · load:PASS · resolve:PASS
CONTAINER_RUN · cargo@1rust:1-alpine@sha256:a10e64dd139b…		 2026-09-08

Fall

HOW
Ziel
Tokio broadcast channels require navigating zero-receiver SendError drops, RecvError::Lagged stream recovery without premature loop termination, and late subscription history exclusion
Pakete
Symbole
  • tokio::sync::broadcast::channel
  • tokio::sync::broadcast::Sender::send
  • tokio::sync::broadcast::Sender::subscribe
  • tokio::sync::broadcast::Receiver::recv
  • tokio::sync::broadcast::Receiver::resubscribe
Umgebung
rust
Erstellt
2026-08-16T06:46:26Z

Contract

  1. assert broadcast::channel(0) panics because broadcast capacity must be strictly positive
  2. assert tx.send returns Err(SendError) instead of buffering or returning Ok(0) when no active receivers exist
  3. assert tx.send returns Ok(active_receiver_count) rather than Ok(()) when subscribers are listening
  4. assert rx.recv returns Err(RecvError::Lagged) on overflow without closing the channel, allowing subsequent recv calls to recover
  5. assert tx.subscribe only receives messages sent strictly after subscription and excludes backlogged buffer history
  6. assert dropping tx permits draining all buffered messages before rx.recv yields Err(RecvError::Closed)
  7. assert rx.resubscribe creates an independent receiver that observes only future messages from the moment of call

Dateien

  • Cargo.lock
  • Cargo.toml
  • NOTES.md
  • csx.json
  • src/lib.rs

Quellartefakt herunterladen (tar.gz)

Quelltext

Cargo.lock
# This file is automatically @generated by Cargo.
# It is not intended for manual editing.
version = 4

[[package]]
name = "addr2line"
version = "0.25.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1b5d307320b3181d6d7954e663bd7c774a838b8220fe0593c86d9fb09f498b4b"
dependencies = [
 "gimli",
]

[[package]]
name = "adler2"
version = "2.0.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa"

[[package]]
name = "backtrace"
version = "0.3.76"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "bb531853791a215d7c62a30daf0dde835f381ab5de4589cfe7c649d2cbe92bd6"
dependencies = [
 "addr2line",
 "cfg-if",
 "libc",
 "miniz_oxide",
 "object",
 "rustc-demangle",
 "windows-link",
]

[[package]]
name = "cfg-if"
version = "1.0.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801"

[[package]]
name = "gimli"
version = "0.32.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e629b9b98ef3dd8afe6ca2bd0f89306cec16d43d907889945bc5d6687f2f13c7"

[[package]]
name = "libc"
version = "0.2.189"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2"

[[package]]
name = "memchr"
version = "2.8.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98"

[[package]]
name = "miniz_oxide"
version = "0.8.9"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1fa76a2c86f704bdb222d66965fb3d63269ce38518b83cb0575fca855ebb6316"
dependencies = [
 "adler2",
]

[[package]]
name = "object"
version = "0.37.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ff76201f031d8863c38aa7f905eca4f53abbfa15f609db4277d44cd8938f33fe"
dependencies = [
 "memchr",
]

[[package]]
name = "pin-project-lite"
version = "0.2.17"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd"

[[package]]
name = "proc-macro2"
version = "1.0.107"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "985e7ec9bb745e6ce6535b544d84d6cd6f7ad8bd711c398938ae983b91a766d9"
dependencies = [
 "unicode-ident",
]

[[package]]
name = "quote"
version = "1.0.47"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1fbf4db142a473a8d80c26bbf18454ed458bf8d26c8219c331daecfdbd079001"
dependencies = [
 "proc-macro2",
]

[[package]]
name = "rustc-demangle"
version = "0.1.28"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b74b56ffa8bb2830709a538c2cbcae9aa062db0d2a42563bfb09bdaae44020eb"

[[package]]
name = "syn"
version = "2.0.119"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "872831b642d1a07999a962a351ed35b955ea2cfc8f3862091e2a240a84f17297"
dependencies = [
 "proc-macro2",
 "quote",
 "unicode-ident",
]

[[package]]
name = "tokio"
version = "1.43.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3d61fa4ffa3de412bfea335c6ecff681de2b609ba3c77ef3e00e521813a9ed9e"
dependencies = [
 "backtrace",
 "pin-project-lite",
 "tokio-macros",
]

[[package]]
name = "tokio-broadcast-semantics"
version = "0.1.0"
dependencies = [
 "tokio",
]

[[package]]
name = "tokio-macros"
version = "2.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6e06d43f1345a3bcd39f6a56dbb7dcab2ba47e68e8ac134855e7e2bdbaf8cab8"
dependencies = [
 "proc-macro2",
 "quote",
 "syn",
]

[[package]]
name = "unicode-ident"
version = "1.0.24"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75"

[[package]]
name = "windows-link"
version = "0.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5"
Cargo.toml
[package]
name = "tokio-broadcast-semantics"
version = "0.1.0"
edition = "2021"

[dependencies]
tokio = { version = "=1.43.0", features = ["sync", "macros", "rt"] }
NOTES.md
### Search Result
`search_known_solution` answered with a HIT (`MATCH: COMPATIBLE`, sample `sha256:9b9720fef342edf7ef51c61e4781caeafe87bc3863b3fdfe9efd67fdb0a2caf8` for `LocalSet` non-Send futures and `sha256:df2946c399f0c7e7df2da45bd620bd315d8fd33f401883465c71605cfba3bc53` for `block_in_place` runtime traps). To avoid duplicate coverage, this sample targets a distinct core API trap in Tokio: the return shapes, error types, lag handling, and subscription semantics of `tokio::sync::broadcast`.

### What a model would have written instead
Models typically assume `broadcast::channel` buffers messages prior to subscriber attachment, expect `tx.send` to return `Ok(())` or `Ok(0)`, and write naive `while let Ok(msg) = rx.recv().await` loops that treat any non-`Ok` result as channel closure.

### How the wrong version fails
Calling `tx.send` without active receivers fails loudly with a runtime `SendError`, while the common `while let Ok(msg) = rx.recv().await` consumer loop fails silently with a green build that permanently drops the listener task upon receiving the first `RecvError::Lagged` message under heavy load.
csx.json
{"case":{"caseId":"case:sha256:8bfc6371f1f2691dba4784bcc436262b5c6c63145ef2dd0b0e7bd449b75e0f04","contract":["assert broadcast::channel(0) panics because broadcast capacity must be strictly positive","assert tx.send returns Err(SendError) instead of buffering or returning Ok(0) when no active receivers exist","assert tx.send returns Ok(active_receiver_count) rather than Ok(()) when subscribers are listening","assert rx.recv returns Err(RecvError::Lagged) on overflow without closing the channel, allowing subsequent recv calls to recover","assert tx.subscribe only receives messages sent strictly after subscription and excludes backlogged buffer history","assert dropping tx permits draining all buffered messages before rx.recv yields Err(RecvError::Closed)","assert rx.resubscribe creates an independent receiver that observes only future messages from the moment of call"],"goal":"Tokio broadcast channels require navigating zero-receiver SendError drops, RecvError::Lagged stream recovery without premature loop termination, and late subscription history exclusion","kind":"HOW","packages":["pkg:cargo/tokio@1.43.0"],"schemaVersion":1,"symbols":["tokio::sync::broadcast::channel","tokio::sync::broadcast::Sender::send","tokio::sync::broadcast::Sender::subscribe","tokio::sync::broadcast::Receiver::recv","tokio::sync::broadcast::Receiver::resubscribe"]},"contractCommand":["cargo","test","--offline"],"environment":{"arch":"x64","ecosystem":"cargo","executionContext":"rust","language":"rust","os":"linux","packageManager":"cargo","runtime":"rust","schemaVersion":1},"license":"MIT-0","packages":["pkg:cargo/tokio@1.43.0"],"schemaVersion":1,"symbols":["tokio::sync::broadcast::channel","tokio::sync::broadcast::Sender::send","tokio::sync::broadcast::Sender::subscribe","tokio::sync::broadcast::Receiver::recv","tokio::sync::broadcast::Receiver::resubscribe"],"verifierAdapter":"cargo@1"}
src/lib.rs
#[cfg(test)]
mod tests {
    use std::panic::{catch_unwind, AssertUnwindSafe};
    use tokio::sync::broadcast::{self, error::RecvError, error::SendError};

    #[test]
    fn assert_channel_zero_capacity_panics() {
        let panic_result = catch_unwind(AssertUnwindSafe(|| {
            let _ = broadcast::channel::<i32>(0);
        }));
        assert!(panic_result.is_err(), "broadcast::channel(0) must panic");
    }

    #[tokio::test]
    async fn assert_send_without_active_receivers_returns_send_error() {
        let (tx, rx) = broadcast::channel::<&str>(16);
        drop(rx);

        // Models assume broadcast channels buffer without receivers or return Ok(0)
        let result = tx.send("hello");
        match result {
            Err(SendError(val)) => assert_eq!(val, "hello"),
            Ok(_) => panic!("expected SendError when no receivers exist"),
        }
    }

    #[tokio::test]
    async fn assert_send_with_active_receivers_returns_receiver_count() {
        let (tx, mut rx1) = broadcast::channel::<&str>(16);
        let mut rx2 = tx.subscribe();
        let mut rx3 = tx.subscribe();

        let recipient_count = tx.send("ping").expect("send should succeed");
        assert_eq!(recipient_count, 3, "send() returns the exact count of active subscribers");

        assert_eq!(rx1.recv().await, Ok("ping"));
        assert_eq!(rx2.recv().await, Ok("ping"));
        assert_eq!(rx3.recv().await, Ok("ping"));
    }

    #[tokio::test]
    async fn assert_receiver_lagged_error_reports_missed_count_and_recovers() {
        let capacity = 2;
        let (tx, mut rx) = broadcast::channel::<u32>(capacity);

        // Fill and overflow the buffer: send 3 items into capacity 2
        tx.send(100).unwrap();
        tx.send(200).unwrap();
        tx.send(300).unwrap();

        // First recv returns Err(Lagged(1)) rather than silently skipping or failing closed
        let first_recv = rx.recv().await;
        assert_eq!(first_recv, Err(RecvError::Lagged(1)));

        // Subsequent recv calls continue normally with unlagged messages
        assert_eq!(rx.recv().await, Ok(200));
        assert_eq!(rx.recv().await, Ok(300));
    }

    #[tokio::test]
    async fn assert_late_subscribe_does_not_replay_buffered_messages() {
        let (tx, mut rx1) = broadcast::channel::<u32>(16);

        tx.send(1).unwrap();
        tx.send(2).unwrap();

        // rx2 subscribes after messages 1 and 2 are sent
        let mut rx2 = tx.subscribe();

        tx.send(3).unwrap();

        assert_eq!(rx1.recv().await, Ok(1));
        assert_eq!(rx1.recv().await, Ok(2));
        assert_eq!(rx1.recv().await, Ok(3));

        // rx2 only observes messages sent AFTER subscription
        assert_eq!(rx2.recv().await, Ok(3));
    }

    #[tokio::test]
    async fn assert_channel_closure_drains_buffer_before_closed_error() {
        let (tx, mut rx) = broadcast::channel::<&str>(16);

        tx.send("final-msg-1").unwrap();
        tx.send("final-msg-2").unwrap();
        drop(tx); // All senders dropped

        // Buffered messages are drained successfully
        assert_eq!(rx.recv().await, Ok("final-msg-1"));
        assert_eq!(rx.recv().await, Ok("final-msg-2"));

        // Only after draining does it return Closed
        assert_eq!(rx.recv().await, Err(RecvError::Closed));
    }

    #[tokio::test]
    async fn assert_receiver_resubscribe_creates_independent_future_only_receiver() {
        let (tx, mut rx1) = broadcast::channel::<i32>(16);

        tx.send(10).unwrap();

        // Receiver::resubscribe starts from the current moment, not rx1's queue position
        let mut rx2 = rx1.resubscribe();

        tx.send(20).unwrap();

        assert_eq!(rx1.recv().await, Ok(10));
        assert_eq!(rx1.recv().await, Ok(20));

        assert_eq!(rx2.recv().await, Ok(20));
    }
}

Ursprungs-Seeder

csx-seed