From 44cfc02c00816972eac5de4ba2dd874d5c53ac1d Mon Sep 17 00:00:00 2001 From: mack42 Date: Fri, 10 Jul 2026 07:36:35 -0400 Subject: [PATCH 1/2] deps: replace unmaintained bincode and rustls-pemfile (#20, #21) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Drop two crates flagged by cargo-audit / cargo-deny as unmaintained. bincode (RUSTSEC-2025-0141) — the broker's cluster layer serialized Raft state and fabric RPCs with bincode. It now uses a new in-house `serde_bin` module: a full serde Serializer/Deserializer implementing bincode 1.x's wire model (little-endian fixint, u64 length prefixes, u32 enum-variant indices, structs-as-seq, non-self-describing), which round-trips openraft's generic types (Entry, Vote, SnapshotMeta, LogId, StoredMembership) unchanged. Verified by serde_bin unit tests, the fabric round-trip test, the full broker suite (121/121), and multi-process chaos (leader-kill failover with zero accepted loss + durable crash recovery). On-disk Raft log/snapshot encoding changes; the broker is unreleased so no migration is needed. (#20) rustls-pemfile (RUSTSEC-2025-0134) — the client's TLS PEM loading now uses the PEM support built into rustls-pki-types (already a dependency, and the crate that owns the CertificateDer/PrivateKeyDer types). No public API change. (#21) ramqp 0.8.0 -> 0.8.1, ramqp-broker 0.8.26 -> 0.8.27. --- CHANGELOG.md | 9 + Cargo.lock | 24 +- ramqp-broker/Cargo.toml | 3 +- ramqp-broker/src/cluster/fabric.rs | 37 +- ramqp-broker/src/cluster/node.rs | 45 +- ramqp-broker/src/cluster/queue_group.rs | 5 +- ramqp-broker/src/cluster/store.rs | 42 +- ramqp-broker/src/lib.rs | 1 + ramqp-broker/src/proxy.rs | 6 +- ramqp-broker/src/serde_bin.rs | 723 ++++++++++++++++++++++++ ramqp/Cargo.toml | 8 +- ramqp/src/transport/tls.rs | 19 +- 12 files changed, 821 insertions(+), 101 deletions(-) create mode 100644 ramqp-broker/src/serde_bin.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 0e5e548..c16ea64 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,15 @@ All notable changes to ramqp will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [0.8.1] - unreleased + +### Security +- Replaced the unmaintained `rustls-pemfile` crate (RUSTSEC-2025-0134) with the + PEM parsing built into `rustls-pki-types` (already present via `rustls`, and + the crate that owns the `CertificateDer`/`PrivateKeyDer` types we return). No + public API or behavior change; the `rustls` feature no longer pulls + `rustls-pemfile`. (#21) + ## [0.8.0] - unreleased **No `Cargo.toml` or code changes needed to upgrade**: `ramqp = "0.8"` is a diff --git a/Cargo.lock b/Cargo.lock index 742286b..fe3fe3b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -123,15 +123,6 @@ version = "0.22.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" -[[package]] -name = "bincode" -version = "1.3.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b1f45e9417d87227c7a56d22e471c6206462cba514c7590c09aff4cf6d1ddcad" -dependencies = [ - "serde", -] - [[package]] name = "bitflags" version = "2.13.0" @@ -1502,7 +1493,7 @@ checksum = "dc33ff2d4973d518d823d61aa239014831e521c75da58e3df4840d3f47749d09" [[package]] name = "ramqp" -version = "0.8.0" +version = "0.8.1" dependencies = [ "base64", "bytes", @@ -1515,7 +1506,6 @@ dependencies = [ "pin-project-lite", "ramqp-core", "rustls", - "rustls-pemfile", "slab", "thiserror 2.0.18", "tokio", @@ -1544,9 +1534,8 @@ dependencies = [ [[package]] name = "ramqp-broker" -version = "0.8.26" +version = "0.8.27" dependencies = [ - "bincode", "bytes", "futures-util", "openraft", @@ -1823,15 +1812,6 @@ dependencies = [ "zeroize", ] -[[package]] -name = "rustls-pemfile" -version = "2.2.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dce314e5fee3f39953d46bb63bb8a46d40c2f8fb7cc5a3b6cab2bde9721d6e50" -dependencies = [ - "rustls-pki-types", -] - [[package]] name = "rustls-pki-types" version = "1.14.1" diff --git a/ramqp-broker/Cargo.toml b/ramqp-broker/Cargo.toml index 0e06817..631accb 100644 --- a/ramqp-broker/Cargo.toml +++ b/ramqp-broker/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "ramqp-broker" -version = "0.8.26" +version = "0.8.27" edition = "2024" description = "A performance-first, highly-available AMQP 1.0 broker in Rust (in development)." license = "MIT" @@ -18,7 +18,6 @@ tracing-subscriber = { version = "0.3", features = ["env-filter"] } openraft = { version = "0.9.24", features = ["serde"] } serde = { version = "1.0.228", features = ["derive"] } serde_json = "1.0.150" -bincode = "1" redb = { version = "4", optional = true } [dev-dependencies] diff --git a/ramqp-broker/src/cluster/fabric.rs b/ramqp-broker/src/cluster/fabric.rs index 57ff568..84743b1 100644 --- a/ramqp-broker/src/cluster/fabric.rs +++ b/ramqp-broker/src/cluster/fabric.rs @@ -10,7 +10,7 @@ //! Hot-path shape (broker.md §3.2): //! - **Zero-copy bodies** — message payloads ride as the raw tail of a frame, //! sliced out of the read buffer as refcounted `Bytes`; they are never run -//! through serde. Only the small fixed header is bincode-encoded. +//! through serde. Only the small fixed header is serde_bin-encoded. //! - **Batched writes** — the writer task drains its queue and flushes once //! per wakeup, so a burst of deliveries/acks is one syscall, not N. //! - **Correlation ids, not lock-step RPC** — requests are pipelined and @@ -71,9 +71,9 @@ pub enum PublishStatus { /// A correlated request. The body's meaning depends on the kind. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub enum RequestKind { - /// Body: the bincode Raft RPC for `kind`. Reply body: its bincode result. + /// Body: the serde_bin Raft RPC for `kind`. Reply body: its serde_bin result. Raft(GroupRef, RaftKind), - /// Body: a bincode [`super::meta::MetaCommand`]. Reply body: bincode + /// Body: a serde_bin [`super::meta::MetaCommand`]. Reply body: serde_bin /// `Result`. MetaWrite, /// Start (or heal) this node's member of a queue group. Empty body/reply. @@ -83,20 +83,20 @@ pub enum RequestKind { /// The full replica set: `(node id, fabric address)`. members: Vec<(NodeId, String)>, }, - /// Ask which node leads a queue group. Reply body: bincode `Option`. + /// Ask which node leads a queue group. Reply body: serde_bin `Option`. WhoLeads { /// The queue name. queue: String, }, /// Publish one message to a queue this node leads. Body: the raw message. - /// Reply body: bincode [`PublishStatus`]. + /// Reply body: serde_bin [`PublishStatus`]. Publish { /// The queue name. queue: String, }, /// Open a subscription channel to a queue this node leads. Deliveries /// flow back as [`FabricHeader::Deliver`] frames carrying `sub_chan`. - /// Reply body: bincode `Result<(), Option>` (`Err` = not leader). + /// Reply body: serde_bin `Result<(), Option>` (`Err` = not leader). OpenSub { /// The queue name. queue: String, @@ -105,14 +105,14 @@ pub enum RequestKind { sub_chan: u64, }, /// Publish into a slot previously reserved via [`RequestKind::Reserve`] - /// (transaction commit). Body: the raw message. Reply body: bincode + /// (transaction commit). Body: the raw message. Reply body: serde_bin /// [`PublishStatus`]. PublishReserved { /// The queue name. queue: String, }, /// Reserve `count` capacity slots on a queue this node leads - /// (transaction commit phase 1). Reply body: bincode `bool`. + /// (transaction commit phase 1). Reply body: serde_bin `bool`. Reserve { /// The queue name. queue: String, @@ -243,9 +243,9 @@ impl OutFrame { } /// Encode one frame into `out`: -/// `[u32 total][u16 header_len][bincode header][raw body]`. +/// `[u32 total][u16 header_len][serde_bin header][raw body]`. fn encode_frame(frame: &OutFrame, out: &mut BytesMut) -> std::io::Result<()> { - let header = bincode::serialize(&frame.header).map_err(std::io::Error::other)?; + let header = crate::serde_bin::to_vec(&frame.header).map_err(std::io::Error::other)?; let header_len = u16::try_from(header.len()).map_err(std::io::Error::other)?; let total = 2 + header.len() + frame.body.len(); if total > MAX_FABRIC_FRAME { @@ -280,7 +280,7 @@ pub async fn read_frame( } let header_bytes = frame.split_to(header_len); let header: FabricHeader = - bincode::deserialize(&header_bytes).map_err(std::io::Error::other)?; + crate::serde_bin::from_slice(&header_bytes).map_err(std::io::Error::other)?; return Ok((header, frame.freeze())); } } @@ -408,7 +408,7 @@ impl ConnState { ) .await; let outcome: Result<(), Option> = match reply { - Ok(body) => match bincode::deserialize(&body) { + Ok(body) => match crate::serde_bin::from_slice(&body) { Ok(outcome) => outcome, Err(e) => { // A bad reply body still leaves the sub registered locally @@ -674,7 +674,8 @@ mod tests { buf.advance(4); assert_eq!(buf.len(), total); let header_len = buf.get_u16() as usize; - let header: FabricHeader = bincode::deserialize(&buf.split_to(header_len)).expect("header"); + let header: FabricHeader = + crate::serde_bin::from_slice(&buf.split_to(header_len)).expect("header"); assert_eq!( header, FabricHeader::Deliver { @@ -686,7 +687,7 @@ mod tests { } #[test] - fn raft_rpc_payloads_survive_bincode() { + fn raft_rpc_payloads_survive_serde_bin() { use openraft::raft::AppendEntriesRequest; use openraft::{BasicNode, Entry, EntryPayload, LogId, Vote}; @@ -707,9 +708,9 @@ mod tests { entries: vec![entry], leader_commit: Some(LogId::new(openraft::CommittedLeaderId::new(3, 1), 8)), }; - let bytes = bincode::serialize(&req).expect("serialize"); + let bytes = crate::serde_bin::to_vec(&req).expect("serialize"); let back: AppendEntriesRequest = - bincode::deserialize(&bytes).expect("deserialize"); + crate::serde_bin::from_slice(&bytes).expect("deserialize"); assert_eq!(back.entries.len(), 1); match &back.entries[0].payload { EntryPayload::Normal(QueueCommand::Enqueue { body, .. }) => { @@ -726,9 +727,9 @@ mod tests { (3, BasicNode::new("c:3")), ]), ); - let bytes = bincode::serialize(&m).expect("membership serialize"); + let bytes = crate::serde_bin::to_vec(&m).expect("membership serialize"); let back: openraft::Membership = - bincode::deserialize(&bytes).expect("membership deserialize"); + crate::serde_bin::from_slice(&bytes).expect("membership deserialize"); assert_eq!(back.get_node(&2).map(|n| n.addr.as_str()), Some("b:2")); } diff --git a/ramqp-broker/src/cluster/node.rs b/ramqp-broker/src/cluster/node.rs index a042645..e83c306 100644 --- a/ramqp-broker/src/cluster/node.rs +++ b/ramqp-broker/src/cluster/node.rs @@ -603,7 +603,7 @@ impl ClusterNode { Bytes::new(), ) .await - && let Ok(Some(leader)) = bincode::deserialize::>(&body) + && let Ok(Some(leader)) = crate::serde_bin::from_slice::>(&body) { return Some(leader); } @@ -742,12 +742,13 @@ impl ClusterNode { .conn() .await .map_err(|e| MetaWriteError::Other(e.to_string()))?; - let body = bincode::serialize(cmd).map_err(|e| MetaWriteError::Other(e.to_string()))?; + let body = + crate::serde_bin::to_vec(cmd).map_err(|e| MetaWriteError::Other(e.to_string()))?; let reply = conn .call(RequestKind::MetaWrite, Bytes::from(body)) .await .map_err(MetaWriteError::Other)?; - bincode::deserialize::>(&reply) + crate::serde_bin::from_slice::>(&reply) .map_err(|e| MetaWriteError::Other(e.to_string()))? } @@ -830,7 +831,7 @@ pub(crate) fn rendezvous_placement(name: &str, nodes: &[NodeId], want: usize) -> placement } -/// The generic fabric-backed Raft network: serializes RPCs with bincode and +/// The generic fabric-backed Raft network: serializes RPCs with serde_bin and /// rides the shared per-peer connection, tagged with the group id. #[derive(Debug, Clone)] pub(crate) struct FabricNetworkFactory { @@ -880,7 +881,7 @@ impl FabricRaftConn { &std::io::Error::other(msg), )) }; - let body = bincode::serialize(rpc).map_err(|e| unreachable(e.to_string()))?; + let body = crate::serde_bin::to_vec(rpc).map_err(|e| unreachable(e.to_string()))?; let conn = self .peer .conn() @@ -894,7 +895,7 @@ impl FabricRaftConn { .await .map_err(unreachable)?; let result: Result = - bincode::deserialize(&reply).map_err(|e| unreachable(e.to_string()))?; + crate::serde_bin::from_slice(&reply).map_err(|e| unreachable(e.to_string()))?; result.map_err(|e| { openraft::error::RPCError::RemoteError(openraft::error::RemoteError::new( self.target, @@ -1107,15 +1108,15 @@ async fn handle_request( let node = node.clone(); let writer = writer.clone(); tokio::spawn(async move { - let result: Result = match bincode::deserialize(&body) - { - Ok(cmd) => node.local_meta_write(&cmd).await, - Err(e) => Err(MetaWriteError::Other(e.to_string())), - }; + let result: Result = + match crate::serde_bin::from_slice(&body) { + Ok(cmd) => node.local_meta_write(&cmd).await, + Err(e) => Err(MetaWriteError::Other(e.to_string())), + }; send_reply( &writer, corr, - bincode::serialize(&result).map_err(|e| e.to_string()), + crate::serde_bin::to_vec(&result).map_err(|e| e.to_string()), ); }); } @@ -1141,7 +1142,7 @@ async fn handle_request( send_reply( writer, corr, - bincode::serialize(&leader).map_err(|e| e.to_string()), + crate::serde_bin::to_vec(&leader).map_err(|e| e.to_string()), ); } RequestKind::Publish { queue } => { @@ -1183,7 +1184,7 @@ async fn handle_request( send_reply( &writer, corr, - bincode::serialize(&ok).map_err(|e| e.to_string()), + crate::serde_bin::to_vec(&ok).map_err(|e| e.to_string()), ); }); } @@ -1206,7 +1207,7 @@ async fn handle_request( send_reply( &writer, corr, - bincode::serialize(&outcome).map_err(|e| e.to_string()), + crate::serde_bin::to_vec(&outcome).map_err(|e| e.to_string()), ); }); } @@ -1259,21 +1260,21 @@ where match kind { RaftKind::AppendEntries => { let rpc: openraft::raft::AppendEntriesRequest = - bincode::deserialize(body).map_err(|e| e.to_string())?; + crate::serde_bin::from_slice(body).map_err(|e| e.to_string())?; let result = raft.append_entries(rpc).await; - bincode::serialize(&result).map_err(|e| e.to_string()) + crate::serde_bin::to_vec(&result).map_err(|e| e.to_string()) } RaftKind::Vote => { let rpc: openraft::raft::VoteRequest = - bincode::deserialize(body).map_err(|e| e.to_string())?; + crate::serde_bin::from_slice(body).map_err(|e| e.to_string())?; let result = raft.vote(rpc).await; - bincode::serialize(&result).map_err(|e| e.to_string()) + crate::serde_bin::to_vec(&result).map_err(|e| e.to_string()) } RaftKind::InstallSnapshot => { let rpc: openraft::raft::InstallSnapshotRequest = - bincode::deserialize(body).map_err(|e| e.to_string())?; + crate::serde_bin::from_slice(body).map_err(|e| e.to_string())?; let result = raft.install_snapshot(rpc).await; - bincode::serialize(&result).map_err(|e| e.to_string()) + crate::serde_bin::to_vec(&result).map_err(|e| e.to_string()) } } } @@ -1435,7 +1436,7 @@ fn send_publish_status(writer: &mpsc::UnboundedSender, corr: u64, stat send_reply( writer, corr, - bincode::serialize(&status).map_err(|e| e.to_string()), + crate::serde_bin::to_vec(&status).map_err(|e| e.to_string()), ); } diff --git a/ramqp-broker/src/cluster/queue_group.rs b/ramqp-broker/src/cluster/queue_group.rs index 2808c6f..3881e4b 100644 --- a/ramqp-broker/src/cluster/queue_group.rs +++ b/ramqp-broker/src/cluster/queue_group.rs @@ -346,7 +346,7 @@ impl ReplicatedState for QueueState { spill.sync_all()?; spill_id = spill.id(); } - bincode::serialize(&PortableState { + crate::serde_bin::to_vec(&PortableState { next_msg_id: self.next_msg_id, messages, spill_id, @@ -355,7 +355,8 @@ impl ReplicatedState for QueueState { } fn restore_snapshot(&mut self, bytes: &[u8]) -> Result<(), String> { - let portable: PortableState = bincode::deserialize(bytes).map_err(|e| e.to_string())?; + let portable: PortableState = + crate::serde_bin::from_slice(bytes).map_err(|e| e.to_string())?; // Keep this state's paging config; rebuild contents. for (_, m) in std::mem::take(&mut self.messages) { self.drop_body(&m.body); diff --git a/ramqp-broker/src/cluster/store.rs b/ramqp-broker/src/cluster/store.rs index d79a592..05af01c 100644 --- a/ramqp-broker/src/cluster/store.rs +++ b/ramqp-broker/src/cluster/store.rs @@ -83,11 +83,11 @@ impl ReplicatedState for MetaState { } fn snapshot_bytes(&self) -> Result, String> { - bincode::serialize(self).map_err(|e| e.to_string()) + crate::serde_bin::to_vec(self).map_err(|e| e.to_string()) } fn restore_snapshot(&mut self, bytes: &[u8]) -> Result<(), String> { - *self = bincode::deserialize(bytes).map_err(|e| e.to_string())?; + *self = crate::serde_bin::from_slice(bytes).map_err(|e| e.to_string())?; Ok(()) } } @@ -131,7 +131,7 @@ pub enum SnapshotPersist { /// through here **before** the storage call returns. Restart recovery loads /// the same data back via [`RaftLogRecovery`]. /// -/// Entries and votes are pre-encoded (bincode) by the caller so the sink is +/// Entries and votes are pre-encoded (serde_bin) by the caller so the sink is /// type-erased and one implementation serves every group. pub trait RaftLogSink: Send + Sync + std::fmt::Debug { /// Durably append `(index, encoded entry)` pairs; returns once fsynced. @@ -326,21 +326,25 @@ impl SharedStore { { let mut inner = store.lock(); if let Some(vote) = &recovery.vote { - inner.vote = - Some(bincode::deserialize(vote).map_err(|e| format!("vote decode: {e}"))?); + inner.vote = Some( + crate::serde_bin::from_slice(vote).map_err(|e| format!("vote decode: {e}"))?, + ); } if let Some(purged) = &recovery.purged { - inner.last_purged = - Some(bincode::deserialize(purged).map_err(|e| format!("purge decode: {e}"))?); + inner.last_purged = Some( + crate::serde_bin::from_slice(purged) + .map_err(|e| format!("purge decode: {e}"))?, + ); } for (index, bytes) in &recovery.entries { - let entry: C::Entry = bincode::deserialize(bytes) + let entry: C::Entry = crate::serde_bin::from_slice(bytes) .map_err(|e| format!("log entry {index} decode: {e}"))?; inner.log.insert(*index, entry); } if let Some((meta_bytes, recovered_blob)) = &recovery.snapshot { - let meta: SnapshotMeta = bincode::deserialize(meta_bytes) - .map_err(|e| format!("snapshot meta decode: {e}"))?; + let meta: SnapshotMeta = + crate::serde_bin::from_slice(meta_bytes) + .map_err(|e| format!("snapshot meta decode: {e}"))?; let (data, blob) = match recovered_blob { SnapshotPersist::Inline(bytes) => { (bytes.clone(), SnapshotBlob::Memory(bytes.clone())) @@ -351,7 +355,7 @@ impl SharedStore { (data, SnapshotBlob::File(path.clone())) } }; - let payload: SnapshotPayload = bincode::deserialize(&data) + let payload: SnapshotPayload = crate::serde_bin::from_slice(&data) .map_err(|e| format!("snapshot payload decode: {e}"))?; inner .state @@ -474,7 +478,7 @@ where last_membership, state_bytes, }; - let data = bincode::serialize(&payload).map_err(|e| e.to_string())?; + let data = crate::serde_bin::to_vec(&payload).map_err(|e| e.to_string())?; // Deep paged states: park the blob on disk so the snapshot does // not double the queue's RSS. Written durably — this blob may // become the ONLY copy of the state once the log purges. @@ -505,7 +509,7 @@ where // durable snapshot would be silent total state loss on restart), // file blobs by path (recovery must never point at a deleted file). if let Some(sink) = self.persist() { - let encoded = bincode::serialize(&meta) + let encoded = crate::serde_bin::to_vec(&meta) .map_err(|e| openraft::StorageIOError::write_snapshot(None, &e))?; let persist_blob = match &blob { SnapshotBlob::Memory(bytes) => SnapshotPersist::Inline(bytes.clone()), @@ -551,8 +555,8 @@ where async fn save_vote(&mut self, vote: &Vote) -> Result<(), StorageError> { if let Some(sink) = self.persist() { - let encoded = - bincode::serialize(vote).map_err(|e| openraft::StorageIOError::write_vote(&e))?; + let encoded = crate::serde_bin::to_vec(vote) + .map_err(|e| openraft::StorageIOError::write_vote(&e))?; sink.save_vote(encoded) .await .map_err(|e| openraft::StorageIOError::write_vote(&std::io::Error::other(e)))?; @@ -596,7 +600,7 @@ where for entry in &entries { encoded.push(( entry.get_log_id().index, - bincode::serialize(entry) + crate::serde_bin::to_vec(entry) .map_err(|e| openraft::StorageIOError::write_logs(&e))?, )); } @@ -630,7 +634,7 @@ where async fn purge_logs_upto(&mut self, log_id: LogId) -> Result<(), StorageError> { if let Some(sink) = self.persist() { - let marker = bincode::serialize(&log_id) + let marker = crate::serde_bin::to_vec(&log_id) .map_err(|e| openraft::StorageIOError::write_logs(&e))?; sink.purge_upto(log_id.index, marker) .await @@ -693,7 +697,7 @@ where snapshot: Box>>, ) -> Result<(), StorageError> { let data = snapshot.into_inner(); - let payload: SnapshotPayload = bincode::deserialize(&data) + let payload: SnapshotPayload = crate::serde_bin::from_slice(&data) .map_err(|e| openraft::StorageIOError::read_snapshot(Some(meta.signature()), &e))?; let (blob, persist, old_blob) = { let mut inner = self.lock(); @@ -747,7 +751,7 @@ where }; // Record the installed snapshot durably (recovery restarts from it). if let Some(sink) = persist { - let encoded = bincode::serialize(meta) + let encoded = crate::serde_bin::to_vec(meta) .map_err(|e| openraft::StorageIOError::write_snapshot(None, &e))?; let persist_blob = match &blob { SnapshotBlob::Memory(bytes) => SnapshotPersist::Inline(bytes.clone()), diff --git a/ramqp-broker/src/lib.rs b/ramqp-broker/src/lib.rs index 4293483..6914072 100644 --- a/ramqp-broker/src/lib.rs +++ b/ramqp-broker/src/lib.rs @@ -41,6 +41,7 @@ mod proxy; mod queue; mod quorum; mod registry; +mod serde_bin; #[cfg(feature = "store-redb")] mod store; mod txn; diff --git a/ramqp-broker/src/proxy.rs b/ramqp-broker/src/proxy.rs index a86821b..c72458d 100644 --- a/ramqp-broker/src/proxy.rs +++ b/ramqp-broker/src/proxy.rs @@ -376,7 +376,9 @@ impl Proxy { .call(RequestKind::Reserve { queue, count }, Bytes::new()) .await { - Ok(body) => bincode::deserialize::(&body).unwrap_or(false), + Ok(body) => { + crate::serde_bin::from_slice::(&body).unwrap_or(false) + } Err(_) => false, }; let _ = reply.send(ok); @@ -648,7 +650,7 @@ impl Proxy { }; let reply = conn.call(req, body.clone()).await; let outcome = match reply.and_then(|b| { - bincode::deserialize::(&b).map_err(|e| e.to_string()) + crate::serde_bin::from_slice::(&b).map_err(|e| e.to_string()) }) { Ok(PublishStatus::Accepted) => PubOutcome::Accepted, Ok(PublishStatus::Rejected) => PubOutcome::Rejected, diff --git a/ramqp-broker/src/serde_bin.rs b/ramqp-broker/src/serde_bin.rs new file mode 100644 index 0000000..f71db7e --- /dev/null +++ b/ramqp-broker/src/serde_bin.rs @@ -0,0 +1,723 @@ +//! `serde_bin` — the broker's own compact binary serialization format, an +//! in-house replacement for the unmaintained `bincode` crate (RUSTSEC-2025-0141, +//! tracked in issue #20). +//! +//! It implements the full serde data model using the same wire model bincode 1.x +//! produced by default, so it round-trips every type the cluster layer already +//! serialized through bincode — **including openraft's generic types** +//! (`Entry`, `Vote`, `SnapshotMeta`, `LogId`, `StoredMembership`): +//! +//! - integers: little-endian, fixed width (`i8..=i128`, `u8..=u128`); +//! - `bool`: one byte (`0`/`1`); `char`: a `u32`; floats: IEEE-754 bits, LE; +//! - `Option`: a one-byte tag (`0` = none, `1` = some) then the value; +//! - sequences, maps, strings, and byte strings: a `u64` length prefix then the +//! elements/bytes; +//! - enum variants: a `u32` variant index then the variant's data; +//! - structs and tuples: their fields in declaration order, no framing. +//! +//! The format is **not self-describing** (there are no type tags on the wire), +//! so it cannot implement `deserialize_any` — exactly like bincode. Every type +//! we use has a shape known at both ends, so that costs us nothing. Both +//! directions are covered by the tests below and, end to end, by the cluster and +//! chaos suites (real Raft replication and snapshots ride this format). + +use std::fmt; + +use serde::Serialize; +use serde::de::{self, DeserializeOwned, DeserializeSeed, IntoDeserializer, Visitor}; +use serde::ser; + +// --------------------------------------------------------------------------- +// Public API — drop-in for the `bincode::{serialize, deserialize}` we replaced. +// --------------------------------------------------------------------------- + +/// Serialize `value` into a fresh byte vector. +pub fn to_vec(value: &T) -> Result, Error> { + let mut serializer = Serializer { out: Vec::new() }; + value.serialize(&mut serializer)?; + Ok(serializer.out) +} + +/// Deserialize a `T` from exactly `bytes`. Trailing bytes are an error — the +/// broker's framing always hands us an exact-length slice, so extra bytes mean a +/// corrupt or mismatched payload rather than a stream boundary. +pub fn from_slice(bytes: &[u8]) -> Result { + let mut de = Deserializer { input: bytes }; + let value = T::deserialize(&mut de)?; + if de.input.is_empty() { + Ok(value) + } else { + Err(Error::TrailingBytes(de.input.len())) + } +} + +// --------------------------------------------------------------------------- +// Error +// --------------------------------------------------------------------------- + +/// A serialization or deserialization failure. +#[derive(Debug)] +pub enum Error { + /// A `serde`-reported error (custom message from a (De)serialize impl). + Message(String), + /// The input ended before a value was fully decoded. + Eof, + /// Bytes remained after the top-level value was decoded. + TrailingBytes(usize), + /// A boolean byte was neither 0 nor 1. + InvalidBool(u8), + /// A `char` value was not a valid Unicode scalar. + InvalidChar(u32), + /// A string field was not valid UTF-8. + InvalidUtf8, + /// An `Option` tag byte was neither 0 nor 1. + InvalidOptionTag(u8), + /// A sequence/map was serialized without a known length (unsupported). + SequenceLengthRequired, + /// A `u64` length prefix did not fit in `usize` on this platform. + LengthOverflow(u64), + /// `deserialize_any`/`deserialize_ignored_any` on a non-self-describing + /// format — never needed by the types we use. + NotSelfDescribing, +} + +impl fmt::Display for Error { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Error::Message(m) => write!(f, "{m}"), + Error::Eof => write!(f, "unexpected end of input"), + Error::TrailingBytes(n) => write!(f, "{n} trailing byte(s) after value"), + Error::InvalidBool(b) => write!(f, "invalid bool byte {b}"), + Error::InvalidChar(c) => write!(f, "invalid char scalar {c:#x}"), + Error::InvalidUtf8 => write!(f, "invalid UTF-8 in string"), + Error::InvalidOptionTag(t) => write!(f, "invalid Option tag {t}"), + Error::SequenceLengthRequired => write!(f, "sequence length must be known"), + Error::LengthOverflow(n) => write!(f, "length {n} exceeds usize"), + Error::NotSelfDescribing => { + write!(f, "self-describing deserialization is not supported") + } + } + } +} + +impl std::error::Error for Error {} + +impl ser::Error for Error { + fn custom(msg: T) -> Self { + Error::Message(msg.to_string()) + } +} + +impl de::Error for Error { + fn custom(msg: T) -> Self { + Error::Message(msg.to_string()) + } +} + +// --------------------------------------------------------------------------- +// Serializer +// --------------------------------------------------------------------------- + +/// Writes the compact binary form into an owned buffer. Internal — the public +/// surface is [`to_vec`]. +struct Serializer { + out: Vec, +} + +impl Serializer { + #[inline] + fn w(&mut self, bytes: &[u8]) { + self.out.extend_from_slice(bytes); + } + #[inline] + fn write_len(&mut self, n: usize) { + self.w(&(n as u64).to_le_bytes()); + } + #[inline] + fn write_u32(&mut self, n: u32) { + self.w(&n.to_le_bytes()); + } +} + +macro_rules! ser_num { + ($($method:ident : $ty:ty,)*) => { $( + fn $method(self, v: $ty) -> Result<(), Error> { + self.w(&v.to_le_bytes()); + Ok(()) + } + )* }; +} + +impl ser::Serializer for &mut Serializer { + type Ok = (); + type Error = Error; + type SerializeSeq = Self; + type SerializeTuple = Self; + type SerializeTupleStruct = Self; + type SerializeTupleVariant = Self; + type SerializeMap = Self; + type SerializeStruct = Self; + type SerializeStructVariant = Self; + + ser_num! { + serialize_i8: i8, serialize_i16: i16, serialize_i32: i32, + serialize_i64: i64, serialize_i128: i128, + serialize_u8: u8, serialize_u16: u16, serialize_u32: u32, + serialize_u64: u64, serialize_u128: u128, + } + + fn serialize_bool(self, v: bool) -> Result<(), Error> { + self.out.push(u8::from(v)); + Ok(()) + } + fn serialize_f32(self, v: f32) -> Result<(), Error> { + self.w(&v.to_bits().to_le_bytes()); + Ok(()) + } + fn serialize_f64(self, v: f64) -> Result<(), Error> { + self.w(&v.to_bits().to_le_bytes()); + Ok(()) + } + fn serialize_char(self, v: char) -> Result<(), Error> { + self.write_u32(v as u32); + Ok(()) + } + fn serialize_str(self, v: &str) -> Result<(), Error> { + self.write_len(v.len()); + self.w(v.as_bytes()); + Ok(()) + } + fn serialize_bytes(self, v: &[u8]) -> Result<(), Error> { + self.write_len(v.len()); + self.w(v); + Ok(()) + } + fn serialize_none(self) -> Result<(), Error> { + self.out.push(0); + Ok(()) + } + fn serialize_some(self, v: &T) -> Result<(), Error> { + self.out.push(1); + v.serialize(self) + } + fn serialize_unit(self) -> Result<(), Error> { + Ok(()) + } + fn serialize_unit_struct(self, _name: &'static str) -> Result<(), Error> { + Ok(()) + } + fn serialize_unit_variant( + self, + _name: &'static str, + index: u32, + _variant: &'static str, + ) -> Result<(), Error> { + self.write_u32(index); + Ok(()) + } + fn serialize_newtype_struct( + self, + _name: &'static str, + v: &T, + ) -> Result<(), Error> { + v.serialize(self) + } + fn serialize_newtype_variant( + self, + _name: &'static str, + index: u32, + _variant: &'static str, + v: &T, + ) -> Result<(), Error> { + self.write_u32(index); + v.serialize(self) + } + fn serialize_seq(self, len: Option) -> Result { + let n = len.ok_or(Error::SequenceLengthRequired)?; + self.write_len(n); + Ok(self) + } + fn serialize_tuple(self, _len: usize) -> Result { + Ok(self) + } + fn serialize_tuple_struct(self, _name: &'static str, _len: usize) -> Result { + Ok(self) + } + fn serialize_tuple_variant( + self, + _name: &'static str, + index: u32, + _variant: &'static str, + _len: usize, + ) -> Result { + self.write_u32(index); + Ok(self) + } + fn serialize_map(self, len: Option) -> Result { + let n = len.ok_or(Error::SequenceLengthRequired)?; + self.write_len(n); + Ok(self) + } + fn serialize_struct(self, _name: &'static str, _len: usize) -> Result { + Ok(self) + } + fn serialize_struct_variant( + self, + _name: &'static str, + index: u32, + _variant: &'static str, + _len: usize, + ) -> Result { + self.write_u32(index); + Ok(self) + } + fn is_human_readable(&self) -> bool { + false + } +} + +// All the "compound" serialize traits just write their elements in order into +// the same buffer, so one impl per trait over `&mut Serializer` suffices. +macro_rules! ser_compound_elem { + ($trait:ident, $method:ident) => { + impl ser::$trait for &mut Serializer { + type Ok = (); + type Error = Error; + fn $method(&mut self, value: &T) -> Result<(), Error> { + value.serialize(&mut **self) + } + fn end(self) -> Result<(), Error> { + Ok(()) + } + } + }; +} +ser_compound_elem!(SerializeSeq, serialize_element); +ser_compound_elem!(SerializeTuple, serialize_element); +ser_compound_elem!(SerializeTupleStruct, serialize_field); +ser_compound_elem!(SerializeTupleVariant, serialize_field); + +impl ser::SerializeStruct for &mut Serializer { + type Ok = (); + type Error = Error; + fn serialize_field( + &mut self, + _key: &'static str, + value: &T, + ) -> Result<(), Error> { + value.serialize(&mut **self) + } + fn end(self) -> Result<(), Error> { + Ok(()) + } +} + +impl ser::SerializeStructVariant for &mut Serializer { + type Ok = (); + type Error = Error; + fn serialize_field( + &mut self, + _key: &'static str, + value: &T, + ) -> Result<(), Error> { + value.serialize(&mut **self) + } + fn end(self) -> Result<(), Error> { + Ok(()) + } +} + +impl ser::SerializeMap for &mut Serializer { + type Ok = (); + type Error = Error; + fn serialize_key(&mut self, key: &T) -> Result<(), Error> { + key.serialize(&mut **self) + } + fn serialize_value(&mut self, value: &T) -> Result<(), Error> { + value.serialize(&mut **self) + } + fn end(self) -> Result<(), Error> { + Ok(()) + } +} + +// --------------------------------------------------------------------------- +// Deserializer +// --------------------------------------------------------------------------- + +/// Reads the compact binary form from a byte slice. Internal — the public +/// surface is [`from_slice`]. +struct Deserializer<'de> { + input: &'de [u8], +} + +impl<'de> Deserializer<'de> { + #[inline] + fn take(&mut self, n: usize) -> Result<&'de [u8], Error> { + if self.input.len() < n { + return Err(Error::Eof); + } + let (head, tail) = self.input.split_at(n); + self.input = tail; + Ok(head) + } + #[inline] + fn read_u8(&mut self) -> Result { + Ok(self.take(1)?[0]) + } + #[inline] + fn read_u32(&mut self) -> Result { + Ok(u32::from_le_bytes(self.take(4)?.try_into().unwrap())) + } + #[inline] + fn read_len(&mut self) -> Result { + let n = u64::from_le_bytes(self.take(8)?.try_into().unwrap()); + usize::try_from(n).map_err(|_| Error::LengthOverflow(n)) + } +} + +macro_rules! de_num { + ($($method:ident : $ty:ty => $visit:ident, $n:expr,)*) => { $( + fn $method>(self, visitor: V) -> Result { + let bytes = self.take($n)?; + visitor.$visit(<$ty>::from_le_bytes(bytes.try_into().unwrap())) + } + )* }; +} + +impl<'de> de::Deserializer<'de> for &mut Deserializer<'de> { + type Error = Error; + + de_num! { + deserialize_i8: i8 => visit_i8, 1, + deserialize_i16: i16 => visit_i16, 2, + deserialize_i32: i32 => visit_i32, 4, + deserialize_i64: i64 => visit_i64, 8, + deserialize_i128: i128 => visit_i128, 16, + deserialize_u8: u8 => visit_u8, 1, + deserialize_u16: u16 => visit_u16, 2, + deserialize_u32: u32 => visit_u32, 4, + deserialize_u64: u64 => visit_u64, 8, + deserialize_u128: u128 => visit_u128, 16, + } + + fn deserialize_any>(self, _visitor: V) -> Result { + Err(Error::NotSelfDescribing) + } + fn deserialize_bool>(self, visitor: V) -> Result { + match self.read_u8()? { + 0 => visitor.visit_bool(false), + 1 => visitor.visit_bool(true), + b => Err(Error::InvalidBool(b)), + } + } + fn deserialize_f32>(self, visitor: V) -> Result { + let bits = u32::from_le_bytes(self.take(4)?.try_into().unwrap()); + visitor.visit_f32(f32::from_bits(bits)) + } + fn deserialize_f64>(self, visitor: V) -> Result { + let bits = u64::from_le_bytes(self.take(8)?.try_into().unwrap()); + visitor.visit_f64(f64::from_bits(bits)) + } + fn deserialize_char>(self, visitor: V) -> Result { + let scalar = self.read_u32()?; + let c = char::from_u32(scalar).ok_or(Error::InvalidChar(scalar))?; + visitor.visit_char(c) + } + fn deserialize_str>(self, visitor: V) -> Result { + let n = self.read_len()?; + let bytes = self.take(n)?; + let s = std::str::from_utf8(bytes).map_err(|_| Error::InvalidUtf8)?; + visitor.visit_borrowed_str(s) + } + fn deserialize_string>(self, visitor: V) -> Result { + self.deserialize_str(visitor) + } + fn deserialize_bytes>(self, visitor: V) -> Result { + let n = self.read_len()?; + let bytes = self.take(n)?; + visitor.visit_borrowed_bytes(bytes) + } + fn deserialize_byte_buf>(self, visitor: V) -> Result { + self.deserialize_bytes(visitor) + } + fn deserialize_option>(self, visitor: V) -> Result { + match self.read_u8()? { + 0 => visitor.visit_none(), + 1 => visitor.visit_some(self), + t => Err(Error::InvalidOptionTag(t)), + } + } + fn deserialize_unit>(self, visitor: V) -> Result { + visitor.visit_unit() + } + fn deserialize_unit_struct>( + self, + _name: &'static str, + visitor: V, + ) -> Result { + visitor.visit_unit() + } + fn deserialize_newtype_struct>( + self, + _name: &'static str, + visitor: V, + ) -> Result { + visitor.visit_newtype_struct(self) + } + fn deserialize_seq>(self, visitor: V) -> Result { + let len = self.read_len()?; + visitor.visit_seq(Access { + de: self, + remaining: len, + }) + } + fn deserialize_tuple>(self, len: usize, visitor: V) -> Result { + visitor.visit_seq(Access { + de: self, + remaining: len, + }) + } + fn deserialize_tuple_struct>( + self, + _name: &'static str, + len: usize, + visitor: V, + ) -> Result { + visitor.visit_seq(Access { + de: self, + remaining: len, + }) + } + fn deserialize_map>(self, visitor: V) -> Result { + let len = self.read_len()?; + visitor.visit_map(Access { + de: self, + remaining: len, + }) + } + fn deserialize_struct>( + self, + _name: &'static str, + fields: &'static [&'static str], + visitor: V, + ) -> Result { + visitor.visit_seq(Access { + de: self, + remaining: fields.len(), + }) + } + fn deserialize_enum>( + self, + _name: &'static str, + _variants: &'static [&'static str], + visitor: V, + ) -> Result { + visitor.visit_enum(self) + } + fn deserialize_identifier>(self, _visitor: V) -> Result { + // Structs deserialize field-by-field via `visit_seq`, and enum variants + // are selected by index in `EnumAccess`, so an identifier is never + // requested against this non-self-describing format. + Err(Error::NotSelfDescribing) + } + fn deserialize_ignored_any>(self, _visitor: V) -> Result { + Err(Error::NotSelfDescribing) + } + fn is_human_readable(&self) -> bool { + false + } +} + +/// Yields a fixed number of elements (a sequence, tuple, struct fields, or map +/// entries) from the underlying deserializer. +struct Access<'a, 'de> { + de: &'a mut Deserializer<'de>, + remaining: usize, +} + +impl<'de> de::SeqAccess<'de> for Access<'_, 'de> { + type Error = Error; + fn next_element_seed>( + &mut self, + seed: T, + ) -> Result, Error> { + if self.remaining == 0 { + return Ok(None); + } + self.remaining -= 1; + seed.deserialize(&mut *self.de).map(Some) + } + fn size_hint(&self) -> Option { + Some(self.remaining) + } +} + +impl<'de> de::MapAccess<'de> for Access<'_, 'de> { + type Error = Error; + fn next_key_seed>( + &mut self, + seed: K, + ) -> Result, Error> { + if self.remaining == 0 { + return Ok(None); + } + self.remaining -= 1; + seed.deserialize(&mut *self.de).map(Some) + } + fn next_value_seed>(&mut self, seed: V) -> Result { + seed.deserialize(&mut *self.de) + } + fn size_hint(&self) -> Option { + Some(self.remaining) + } +} + +// Enum variants are identified by the `u32` index we wrote; feed it to the seed +// through serde's integer deserializer, then read the variant payload. +impl<'de> de::EnumAccess<'de> for &mut Deserializer<'de> { + type Error = Error; + type Variant = Self; + fn variant_seed>(self, seed: V) -> Result<(V::Value, Self), Error> { + let index = self.read_u32()?; + let value = seed.deserialize(index.into_deserializer())?; + Ok((value, self)) + } +} + +impl<'de> de::VariantAccess<'de> for &mut Deserializer<'de> { + type Error = Error; + fn unit_variant(self) -> Result<(), Error> { + Ok(()) + } + fn newtype_variant_seed>(self, seed: T) -> Result { + seed.deserialize(self) + } + fn tuple_variant>(self, len: usize, visitor: V) -> Result { + de::Deserializer::deserialize_tuple(self, len, visitor) + } + fn struct_variant>( + self, + fields: &'static [&'static str], + visitor: V, + ) -> Result { + de::Deserializer::deserialize_tuple(self, fields.len(), visitor) + } +} + +#[cfg(test)] +mod tests { + use std::collections::BTreeMap; + + use serde::{Deserialize, Serialize}; + + use super::{Error, from_slice, to_vec}; + + fn round(value: &T) -> T + where + T: Serialize + serde::de::DeserializeOwned + PartialEq + std::fmt::Debug, + { + let bytes = to_vec(value).expect("serialize"); + let back: T = from_slice(&bytes).expect("deserialize"); + assert_eq!(value, &back, "round trip mismatch"); + back + } + + #[derive(Serialize, Deserialize, PartialEq, Debug)] + enum Shape { + Point, + Radius(f64), + Rect(u32, u32), + Named { id: u64, tag: String }, + } + + #[derive(Serialize, Deserialize, PartialEq, Debug)] + struct Everything { + b: bool, + i: i64, + u: u128, + f: f64, + c: char, + s: String, + bytes: Vec, + opt_some: Option, + opt_none: Option, + list: Vec, + map: BTreeMap, + nested: Vec>, + unit: (), + pair: (u8, String), + } + + #[test] + fn primitives_round_trip() { + round(&true); + round(&false); + round(&(-12345i64)); + round(&u64::MAX); + round(&0.0f64); + round(&(-1.5f32)); + round(&'∆'); + round(&"héllo".to_string()); + } + + #[test] + fn all_enum_variant_kinds_round_trip() { + round(&Shape::Point); + round(&Shape::Radius(2.5)); + round(&Shape::Rect(3, 4)); + round(&Shape::Named { + id: 7, + tag: "q".into(), + }); + } + + #[test] + fn options_and_collections_round_trip() { + round(&Some(9u32)); + round(&Option::::None); + round(&vec![1u8, 2, 3]); + let mut m = BTreeMap::new(); + m.insert("a".to_string(), 1i32); + m.insert("b".to_string(), -2); + round(&m); + } + + #[test] + fn deep_struct_round_trips() { + let mut map = BTreeMap::new(); + map.insert("x".to_string(), 10); + map.insert("y".to_string(), -20); + round(&Everything { + b: true, + i: -99, + u: u128::MAX, + f: 3.25, + c: 'z', + s: "wire".into(), + bytes: vec![9, 8, 7], + opt_some: Some(42), + opt_none: None, + list: vec![Shape::Point, Shape::Rect(1, 2), Shape::Radius(0.5)], + map, + nested: vec![vec![1, 2], vec![], vec![3]], + unit: (), + pair: (255, "end".into()), + }); + } + + #[test] + fn trailing_bytes_are_rejected() { + let mut bytes = to_vec(&7u32).unwrap(); + bytes.push(0); // one extra byte + let err = from_slice::(&bytes).unwrap_err(); + assert!(matches!(err, Error::TrailingBytes(1)), "got {err:?}"); + } + + #[test] + fn truncated_input_is_eof_not_panic() { + let bytes = to_vec(&(1u64, 2u64)).unwrap(); + let err = from_slice::<(u64, u64)>(&bytes[..bytes.len() - 1]).unwrap_err(); + assert!(matches!(err, Error::Eof), "got {err:?}"); + } +} diff --git a/ramqp/Cargo.toml b/ramqp/Cargo.toml index 2f60925..6e49608 100644 --- a/ramqp/Cargo.toml +++ b/ramqp/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "ramqp" -version = "0.8.0" +version = "0.8.1" edition = "2024" description = "Async AMQP 1.0 client for Rust on tokio — connects to RabbitMQ 4.x, ActiveMQ Artemis, and other AMQP 1.0 brokers." license = "MIT" @@ -34,7 +34,7 @@ tracing = "0.1" [features] default = [] # TLS backends -rustls = ["dep:tokio-rustls", "dep:rustls", "dep:webpki-roots", "dep:rustls-pemfile"] +rustls = ["dep:tokio-rustls", "dep:rustls", "dep:webpki-roots"] native-tls = ["dep:tokio-native-tls", "dep:native-tls"] # WebSocket transport ws = ["dep:tokio-tungstenite", "dep:http"] @@ -59,10 +59,6 @@ features = ["ring", "std", "tls12"] version = "1" optional = true -[dependencies.rustls-pemfile] -version = "2" -optional = true - [dependencies.tokio-native-tls] version = "0.3" optional = true diff --git a/ramqp/src/transport/tls.rs b/ramqp/src/transport/tls.rs index 72ce89c..a2ebe2d 100644 --- a/ramqp/src/transport/tls.rs +++ b/ramqp/src/transport/tls.rs @@ -73,26 +73,29 @@ pub async fn connect_rustls( .map_err(|e| ConnectError::new(ErrorKind::Tls).with_source(e)) } -/// Parse a PEM blob into a chain of DER certificates. +/// Parse a PEM blob into a chain of DER certificates. Uses `rustls-pki-types`' +/// built-in PEM support (the same crate that owns `CertificateDer`), so there is +/// no separate PEM-parsing dependency. #[cfg(feature = "rustls")] fn load_certs( pem: &[u8], ) -> Result>, ConnectError> { - let mut reader = std::io::Cursor::new(pem); - rustls_pemfile::certs(&mut reader) + use tokio_rustls::rustls::pki_types::CertificateDer; + use tokio_rustls::rustls::pki_types::pem::PemObject; + CertificateDer::pem_slice_iter(pem) .collect::, _>>() .map_err(|e| ConnectError::new(ErrorKind::Tls).with_source(e)) } -/// Parse a PEM blob into a single private key (PKCS#8 / PKCS#1 / SEC1). +/// Parse a PEM blob into a single private key (PKCS#8 / PKCS#1 / SEC1). Returns +/// an error if the blob contains no private key. #[cfg(feature = "rustls")] fn load_key( pem: &[u8], ) -> Result, ConnectError> { - let mut reader = std::io::Cursor::new(pem); - rustls_pemfile::private_key(&mut reader) - .map_err(|e| ConnectError::new(ErrorKind::Tls).with_source(e))? - .ok_or_else(|| ConnectError::msg(ErrorKind::Tls, "no private key found in PEM")) + use tokio_rustls::rustls::pki_types::PrivateKeyDer; + use tokio_rustls::rustls::pki_types::pem::PemObject; + PrivateKeyDer::from_pem_slice(pem).map_err(|e| ConnectError::new(ErrorKind::Tls).with_source(e)) } /// A certificate verifier that accepts any chain. Test-only; gated behind From 8214388aa1e8b5a5a25b822d051ab7133babdc25 Mon Sep 17 00:00:00 2001 From: mack42 Date: Fri, 10 Jul 2026 09:12:15 -0400 Subject: [PATCH 2/2] test(broker): re-runnable test battery, driver bins, and wire fuzzing Add a consistent, scriptable battery under ramqp-broker/scripts/ so the broker gets the same checks build over build ahead of a crates.io release. Not wired into CI; run by hand or via run-all.sh. Each stage sources lib.sh (logging, pass/fail tally, brokerd spawn/teardown, port + RSS helpers) and writes artifacts to a gitignored out//. Stages: static gates; the full suite under a flake-repeat loop; a soak/leak detector (RSS-flat + no-throughput-decay under connection churn); chaos (rolling kill -9 of cluster nodes with a zero-accepted-loss verifier, plus durable crash recovery); an interop matrix (ramqp and fe2o3-amqp clients x ramqp-broker / RabbitMQ / Artemis); robustness floods; and cargo-fuzz on the wire decoders. bench.sh (perf-vs-baseline) and cov.sh (coverage) are manual and never gate. Load/chaos/robustness drivers that bash cannot express live as example bins (examples/{loadgen,chaos,recover,robust}.rs) so cargo check --all-targets keeps them compiling. Fuzz targets (decode_frame, Value) live in ramqp-core/fuzz as their own workspace, so the main build never pulls libfuzzer. ramqp-broker 0.8.27 -> 0.8.28. --- .gitignore | 8 + Cargo.lock | 2 +- ramqp-broker/Cargo.toml | 2 +- ramqp-broker/README.md | 20 + ramqp-broker/examples/chaos.rs | 204 ++++++ ramqp-broker/examples/loadgen.rs | 157 +++++ ramqp-broker/examples/recover.rs | 97 +++ ramqp-broker/examples/robust.rs | 202 ++++++ ramqp-broker/scripts/00-gates.sh | 51 ++ ramqp-broker/scripts/10-suite.sh | 58 ++ ramqp-broker/scripts/20-soak.sh | 75 +++ ramqp-broker/scripts/30-chaos.sh | 138 ++++ ramqp-broker/scripts/40-interop.sh | 78 +++ ramqp-broker/scripts/50-robust.sh | 37 ++ ramqp-broker/scripts/60-fuzz.sh | 39 ++ ramqp-broker/scripts/README.md | 58 ++ ramqp-broker/scripts/analyze_soak.py | 60 ++ ramqp-broker/scripts/bench-baseline.json | 9 + ramqp-broker/scripts/bench.sh | 51 ++ ramqp-broker/scripts/bench_stats.py | 103 +++ ramqp-broker/scripts/cov.sh | 33 + ramqp-broker/scripts/lib.sh | 217 +++++++ ramqp-broker/scripts/run-all.sh | 62 ++ ramqp-core/fuzz/Cargo.lock | 626 +++++++++++++++++++ ramqp-core/fuzz/Cargo.toml | 37 ++ ramqp-core/fuzz/fuzz_targets/decode_frame.rs | 26 + ramqp-core/fuzz/fuzz_targets/value.rs | 12 + 27 files changed, 2460 insertions(+), 2 deletions(-) create mode 100644 ramqp-broker/examples/chaos.rs create mode 100644 ramqp-broker/examples/loadgen.rs create mode 100644 ramqp-broker/examples/recover.rs create mode 100644 ramqp-broker/examples/robust.rs create mode 100755 ramqp-broker/scripts/00-gates.sh create mode 100755 ramqp-broker/scripts/10-suite.sh create mode 100755 ramqp-broker/scripts/20-soak.sh create mode 100755 ramqp-broker/scripts/30-chaos.sh create mode 100755 ramqp-broker/scripts/40-interop.sh create mode 100755 ramqp-broker/scripts/50-robust.sh create mode 100755 ramqp-broker/scripts/60-fuzz.sh create mode 100644 ramqp-broker/scripts/README.md create mode 100644 ramqp-broker/scripts/analyze_soak.py create mode 100644 ramqp-broker/scripts/bench-baseline.json create mode 100755 ramqp-broker/scripts/bench.sh create mode 100644 ramqp-broker/scripts/bench_stats.py create mode 100755 ramqp-broker/scripts/cov.sh create mode 100755 ramqp-broker/scripts/lib.sh create mode 100755 ramqp-broker/scripts/run-all.sh create mode 100644 ramqp-core/fuzz/Cargo.lock create mode 100644 ramqp-core/fuzz/Cargo.toml create mode 100644 ramqp-core/fuzz/fuzz_targets/decode_frame.rs create mode 100644 ramqp-core/fuzz/fuzz_targets/value.rs diff --git a/.gitignore b/.gitignore index ce33856..e21a85e 100644 --- a/.gitignore +++ b/.gitignore @@ -3,6 +3,14 @@ debug target +# Test-script run artifacts (logs, captured metrics, spawned broker data dirs) +ramqp-broker/scripts/out/ + +# cargo-fuzz corpora/artifacts and coverage output +ramqp-core/fuzz/corpus/ +ramqp-core/fuzz/artifacts/ +ramqp-core/fuzz/coverage/ + # Stray local check binary /check_parsing diff --git a/Cargo.lock b/Cargo.lock index fe3fe3b..a8ec41c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1534,7 +1534,7 @@ dependencies = [ [[package]] name = "ramqp-broker" -version = "0.8.27" +version = "0.8.28" dependencies = [ "bytes", "futures-util", diff --git a/ramqp-broker/Cargo.toml b/ramqp-broker/Cargo.toml index 631accb..2c97917 100644 --- a/ramqp-broker/Cargo.toml +++ b/ramqp-broker/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "ramqp-broker" -version = "0.8.27" +version = "0.8.28" edition = "2024" description = "A performance-first, highly-available AMQP 1.0 broker in Rust (in development)." license = "MIT" diff --git a/ramqp-broker/README.md b/ramqp-broker/README.md index 9f84fad..ecbcb28 100644 --- a/ramqp-broker/README.md +++ b/ramqp-broker/README.md @@ -72,6 +72,26 @@ bound.run().await # } ``` +## Testing + +Beyond the in-tree suite (`cargo test --all-features`), a re-runnable battery +lives in [`scripts/`](scripts) — the same checks build over build, so +regressions in correctness, memory, HA, interop, or robustness surface before a +release. It is **not** wired into CI; run it by hand: + +```sh +ramqp-broker/scripts/run-all.sh --quick # gates + full suite (fast) +ramqp-broker/scripts/run-all.sh # + soak, chaos, interop, robustness, fuzz +``` + +Stages: static gates, the suite under a flake-repeat loop, a soak/leak detector +(RSS-flat + no-throughput-decay under churn), chaos (rolling `kill -9` of cluster +nodes with a zero-accepted-loss verifier, plus durable crash recovery), an +interop matrix (`ramqp` and `fe2o3-amqp` clients × ramqp-broker / RabbitMQ / +Artemis), robustness floods, and `cargo-fuzz` on the wire decoders. Performance +(`scripts/bench.sh`) and coverage (`scripts/cov.sh`) are manual and never gate. +See [`scripts/README.md`](scripts/README.md). + ## Numbers Untuned first numbers vs RabbitMQ 4.3.1 and Artemis on the same machine — diff --git a/ramqp-broker/examples/chaos.rs b/ramqp-broker/examples/chaos.rs new file mode 100644 index 0000000..4439b60 --- /dev/null +++ b/ramqp-broker/examples/chaos.rs @@ -0,0 +1,204 @@ +//! Zero-loss verifier for the chaos / fault-injection stage +//! (`scripts/30-chaos.sh`). The script spawns a cluster and kills+restarts +//! nodes underneath this client; this binary proves the HA contract: +//! +//! **every message the broker ACCEPTED is eventually delivered** — no +//! accepted-message loss across leader/follower failovers (at-least-once, so +//! duplicates are allowed and reported). +//! +//! A producer publishes seq `0..N` to a quorum queue, retrying each until it is +//! `Accepted` (the publisher-confirm pattern — a rejection during an election +//! is the cue to retry). A consumer drains concurrently, recording the set of +//! received seqs. At the end, every seq must have been received. The client +//! connects to a node the script keeps ALIVE, so the connection itself +//! survives; reconnect logic covers transient fabric errors regardless. +//! +//! Env: `CHAOS_PRODUCER_URL` `CHAOS_CONSUMER_URL` `CHAOS_QUEUE` `CHAOS_N` +//! `CHAOS_PAYLOAD` `CHAOS_DEADLINE_SECS`. + +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::time::{Duration, Instant}; + +use ramqp::types::messaging::{Body, DeliveryState}; +use ramqp::{Connection, ConnectionBuilder, Consumer, Message, Producer, Session}; + +fn env_usize(k: &str, d: usize) -> usize { + std::env::var(k) + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(d) +} +fn env_string(k: &str, d: &str) -> String { + std::env::var(k).unwrap_or_else(|_| d.to_owned()) +} + +fn seq_of(d: &ramqp::Delivery) -> Option { + let msg = d.message().ok()?; + match &msg.body { + Body::Data(sections) => { + let first = sections.first()?; + Some(u64::from_be_bytes(first.get(..8)?.try_into().ok()?)) + } + _ => None, + } +} + +async fn connect_producer(url: &str, addr: &str) -> Option<(Connection, Session, Producer)> { + let conn = ConnectionBuilder::new(url).connect().await.ok()?; + let session = conn.begin_session().await.ok()?; + let producer = session.create_producer(addr).await.ok()?; + Some((conn, session, producer)) +} + +async fn connect_consumer(url: &str, addr: &str) -> Option<(Connection, Session, Consumer)> { + let conn = ConnectionBuilder::new(url).connect().await.ok()?; + let session = conn.begin_session().await.ok()?; + let consumer = session.create_consumer(addr).await.ok()?; + Some((conn, session, consumer)) +} + +#[tokio::main(flavor = "multi_thread")] +async fn main() -> Result<(), Box> { + let purl = env_string("CHAOS_PRODUCER_URL", "amqp://127.0.0.1:5672"); + let curl = env_string("CHAOS_CONSUMER_URL", "amqp://127.0.0.1:5672"); + let queue = env_string("CHAOS_QUEUE", "/quorum/chaos"); + let n = env_usize("CHAOS_N", 20_000) as u64; + let payload = env_usize("CHAOS_PAYLOAD", 64).max(8); + let deadline_secs = env_usize("CHAOS_DEADLINE_SECS", 180) as u64; + + println!( + "chaos: producer={purl} consumer={curl} queue={queue} n={n} deadline={deadline_secs}s" + ); + + let start = Instant::now(); + let deadline = start + Duration::from_secs(deadline_secs); + let received: Arc> = Arc::new((0..n).map(|_| AtomicBool::new(false)).collect()); + let recv_count = Arc::new(AtomicUsize::new(0)); + let dupes = Arc::new(AtomicUsize::new(0)); + let accepted = Arc::new(AtomicUsize::new(0)); + let stop = Arc::new(AtomicBool::new(false)); + let step = (n / 10).max(1); + + // Producer: publish every seq, retrying until Accepted. + let producer = { + let (accepted, purl, queue) = (accepted.clone(), purl.clone(), queue.clone()); + tokio::spawn(async move { + let mut bundle = None; + let mut seq = 0u64; + while seq < n { + if Instant::now() >= deadline { + eprintln!("producer: DEADLINE with {seq}/{n} accepted"); + break; + } + if bundle.is_none() { + bundle = connect_producer(&purl, &queue).await; + if bundle.is_none() { + tokio::time::sleep(Duration::from_millis(100)).await; + continue; + } + } + let mut body = vec![0u8; payload]; + body[..8].copy_from_slice(&seq.to_be_bytes()); + let p = &bundle.as_ref().unwrap().2; + match p.send(Message::data(body)).await { + Ok(DeliveryState::Accepted(_)) => { + let c = accepted.fetch_add(1, Ordering::Relaxed) + 1; + seq += 1; + if (c as u64).is_multiple_of(step) { + println!( + "producer: {c}/{n} accepted ({:.0}s)", + start.elapsed().as_secs_f64() + ); + } + } + Ok(_) => tokio::time::sleep(Duration::from_millis(50)).await, // retry same seq + Err(_) => bundle = None, // reconnect + } + } + }) + }; + + // Consumer: drain concurrently, dedupe by seq. + let consumer = { + let (received, recv_count, dupes, stop, curl, queue) = ( + received.clone(), + recv_count.clone(), + dupes.clone(), + stop.clone(), + curl.clone(), + queue.clone(), + ); + tokio::spawn(async move { + let mut bundle = None; + while !stop.load(Ordering::Relaxed) && Instant::now() < deadline { + if bundle.is_none() { + bundle = connect_consumer(&curl, &queue).await; + if bundle.is_none() { + tokio::time::sleep(Duration::from_millis(100)).await; + continue; + } + } + let b = bundle.as_mut().unwrap(); + match tokio::time::timeout(Duration::from_secs(5), b.2.recv()).await { + Ok(Ok(d)) => { + if let Some(s) = seq_of(&d) { + if s < n && !received[s as usize].swap(true, Ordering::Relaxed) { + let c = recv_count.fetch_add(1, Ordering::Relaxed) + 1; + if (c as u64).is_multiple_of(step) { + println!( + "consumer: {c}/{n} received ({:.0}s)", + start.elapsed().as_secs_f64() + ); + } + } else { + dupes.fetch_add(1, Ordering::Relaxed); + } + } + let _ = b.2.accept(&d).await; + } + Ok(Err(_)) => bundle = None, // link/conn error → reconnect + Err(_) => { /* recv timeout: loop and re-check counts */ } + } + } + }) + }; + + let _ = producer.await; + // Producer done (or deadline). Let the consumer catch up to N or the deadline. + while recv_count.load(Ordering::Relaxed) < n as usize && Instant::now() < deadline { + tokio::time::sleep(Duration::from_millis(100)).await; + } + stop.store(true, Ordering::Relaxed); + let _ = consumer.await; + + // Verdict. + let acc = accepted.load(Ordering::Relaxed); + let rec = recv_count.load(Ordering::Relaxed); + let dup = dupes.load(Ordering::Relaxed); + let missing: Vec = (0..n) + .filter(|&s| !received[s as usize].load(Ordering::Relaxed)) + .collect(); + println!( + "chaos result: accepted={acc}/{n} received={rec}/{n} duplicates={dup} missing={} elapsed={:.0}s", + missing.len(), + start.elapsed().as_secs_f64() + ); + if acc < n as usize { + eprintln!( + "FAIL: producer could not get {} message(s) accepted before the deadline (liveness)", + n as usize - acc + ); + std::process::exit(2); + } + if !missing.is_empty() { + let sample: Vec = missing.iter().take(20).copied().collect(); + eprintln!( + "FAIL: {} ACCEPTED message(s) never delivered (loss). sample seqs: {sample:?}", + missing.len() + ); + std::process::exit(1); + } + println!("PASS: zero accepted-message loss across the chaos run"); + Ok(()) +} diff --git a/ramqp-broker/examples/loadgen.rs b/ramqp-broker/examples/loadgen.rs new file mode 100644 index 0000000..1e69283 --- /dev/null +++ b/ramqp-broker/examples/loadgen.rs @@ -0,0 +1,157 @@ +//! Sustained load generator for the soak / leak stage (`scripts/20-soak.sh`). +//! +//! Drives `LOAD_PAIRS` independent producer/consumer pairs against a broker for +//! `LOAD_SECS`, each a **closed loop** (send a window, drain it, accept) so the +//! queue depth stays bounded — any growth in the *broker's* RSS is then a real +//! leak, not backlog. With `LOAD_CHURN > 0` each pair tears down and reopens its +//! connection every N messages, exercising the connection open/close path that +//! hid the close-time settlement-drain requeue bug. +//! +//! Prints a throughput sample every `LOAD_REPORT_SECS` (`t= total= +//! rate=`) and a final summary line the soak script parses. +//! +//! Env: `LOAD_URL` `LOAD_ADDRESS` `LOAD_SECS` `LOAD_PAIRS` `LOAD_PAYLOAD` +//! `LOAD_WINDOW` `LOAD_CHURN` `LOAD_REPORT_SECS`. + +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::time::{Duration, Instant}; + +use ramqp::{ConnectionBuilder, Message}; + +fn env_usize(k: &str, d: usize) -> usize { + std::env::var(k) + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(d) +} +fn env_string(k: &str, d: &str) -> String { + std::env::var(k).unwrap_or_else(|_| d.to_owned()) +} + +#[tokio::main] +async fn main() -> Result<(), Box> { + let url = env_string("LOAD_URL", "amqp://127.0.0.1:5672"); + let base_addr = env_string("LOAD_ADDRESS", "/queues/soak"); + let secs = env_usize("LOAD_SECS", 60) as u64; + let pairs = env_usize("LOAD_PAIRS", 8).max(1); + let payload = env_usize("LOAD_PAYLOAD", 256).max(1); + let window = env_usize("LOAD_WINDOW", 100).max(1); + let churn = env_usize("LOAD_CHURN", 0); // reconnect every N msgs; 0 = never + let report = env_usize("LOAD_REPORT_SECS", 5).max(1) as u64; + + println!( + "loadgen: url={url} addr={base_addr}- pairs={pairs} payload={payload}B \ + window={window} churn={churn} secs={secs}" + ); + + let total = Arc::new(AtomicU64::new(0)); + let stop = Arc::new(AtomicBool::new(false)); + let start = Instant::now(); + let deadline = start + Duration::from_secs(secs); + + let reporter = { + let (total, stop) = (total.clone(), stop.clone()); + tokio::spawn(async move { + let mut last = 0u64; + let mut last_t = Instant::now(); + while !stop.load(Ordering::Relaxed) { + tokio::time::sleep(Duration::from_secs(report)).await; + let now = Instant::now(); + let cur = total.load(Ordering::Relaxed); + let dt = now.duration_since(last_t).as_secs_f64().max(1e-9); + println!( + "t={:.0}s total={} rate={:.0} msg/s", + now.duration_since(start).as_secs_f64(), + cur, + (cur - last) as f64 / dt + ); + last = cur; + last_t = now; + } + }) + }; + + let mut tasks = Vec::new(); + for i in 0..pairs { + let url = url.clone(); + let addr = format!("{base_addr}-{i}"); + let total = total.clone(); + tasks.push(tokio::spawn(async move { + 'session: while Instant::now() < deadline { + // (Re)establish the whole client stack; any setup hiccup just + // retries after a short backoff. + let Ok(conn) = ConnectionBuilder::new(&url).connect().await else { + tokio::time::sleep(Duration::from_millis(50)).await; + continue; + }; + let setup = async { + let session = conn.begin_session().await.ok()?; + let producer = session.create_producer(&addr).await.ok()?; + let consumer = session.create_consumer(&addr).await.ok()?; + Some((producer, consumer)) + }; + let Some((producer, mut consumer)) = setup.await else { + let _ = conn.close().await; + continue; + }; + + let mut since_reconnect = 0usize; + loop { + if Instant::now() >= deadline { + let _ = conn.close().await; + break 'session; + } + // Send one window of settled messages. + for _ in 0..window { + if producer + .send_settled(Message::data(vec![0u8; payload])) + .await + .is_err() + { + let _ = conn.close().await; + continue 'session; + } + } + // Drain exactly that window and accept each. + for _ in 0..window { + match consumer.recv().await { + Ok(d) => { + if consumer.accept(&d).await.is_err() { + let _ = conn.close().await; + continue 'session; + } + } + Err(_) => { + let _ = conn.close().await; + continue 'session; + } + } + } + total.fetch_add(window as u64, Ordering::Relaxed); + since_reconnect += window; + if churn > 0 && since_reconnect >= churn { + // Graceful close then reconnect: this is the path that + // must not requeue already-acked messages. + let _ = conn.close().await; + continue 'session; + } + } + } + })); + } + + for t in tasks { + let _ = t.await; + } + stop.store(true, Ordering::Relaxed); + let _ = reporter.await; + + let elapsed = start.elapsed().as_secs_f64().max(1e-9); + let n = total.load(Ordering::Relaxed); + println!( + "loadgen done: total={n} in {elapsed:.1}s = {:.0} msg/s", + n as f64 / elapsed + ); + Ok(()) +} diff --git a/ramqp-broker/examples/recover.rs b/ramqp-broker/examples/recover.rs new file mode 100644 index 0000000..0d5fbf1 --- /dev/null +++ b/ramqp-broker/examples/recover.rs @@ -0,0 +1,97 @@ +//! Two-phase durability check for `scripts/30-chaos.sh` (part B). +//! +//! `RECOVER_PHASE=produce` publishes seq `0..N` to a durable/quorum queue, +//! confirming each is `Accepted` (the on-disk durability confirm), then exits. +//! The script then **SIGKILLs** the broker and starts a fresh process on the +//! same data dir. `RECOVER_PHASE=consume` then drains and asserts every seq +//! survived the crash. This exercises real crash recovery (kill -9 + cold +//! start from disk), which is stronger than the in-process suite's graceful +//! restart. +//! +//! Env: `RECOVER_URL` `RECOVER_ADDRESS` `RECOVER_N` `RECOVER_PHASE`. + +use std::time::{Duration, Instant}; + +use ramqp::types::messaging::{Body, DeliveryState}; +use ramqp::{ConnectionBuilder, Message}; + +fn env_usize(k: &str, d: usize) -> usize { + std::env::var(k) + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(d) +} +fn env_string(k: &str, d: &str) -> String { + std::env::var(k).unwrap_or_else(|_| d.to_owned()) +} + +fn seq_of(d: &ramqp::Delivery) -> Option { + match &d.message().ok()?.body { + Body::Data(s) => Some(u64::from_be_bytes(s.first()?.get(..8)?.try_into().ok()?)), + _ => None, + } +} + +#[tokio::main] +async fn main() -> Result<(), Box> { + let url = env_string("RECOVER_URL", "amqp://127.0.0.1:5672"); + let addr = env_string("RECOVER_ADDRESS", "/durable/recovery"); + let n = env_usize("RECOVER_N", 5000) as u64; + let phase = env_string("RECOVER_PHASE", "produce"); + + let conn = ConnectionBuilder::new(&url).connect().await?; + let session = conn.begin_session().await?; + + match phase.as_str() { + "produce" => { + let producer = session.create_producer(&addr).await?; + for seq in 0..n { + let mut body = vec![0u8; 16]; + body[..8].copy_from_slice(&seq.to_be_bytes()); + // Retry until the durability confirm lands. + loop { + match producer.send(Message::data(body.clone())).await { + Ok(DeliveryState::Accepted(_)) => break, + _ => tokio::time::sleep(Duration::from_millis(20)).await, + } + } + } + conn.close().await?; + println!("recover/produce: {n} messages confirmed durable to {addr}"); + } + "consume" => { + let mut consumer = session.create_consumer(&addr).await?; + let mut seen = vec![false; n as usize]; + let mut count = 0u64; + let deadline = Instant::now() + Duration::from_secs(60); + while count < n && Instant::now() < deadline { + match tokio::time::timeout(Duration::from_secs(10), consumer.recv()).await { + Ok(Ok(d)) => { + if let Some(s) = seq_of(&d) + && s < n + && !seen[s as usize] + { + seen[s as usize] = true; + count += 1; + } + let _ = consumer.accept(&d).await; + } + _ => break, + } + } + conn.close().await.ok(); + let missing = seen.iter().filter(|&&b| !b).count(); + println!("recover/consume: recovered {count}/{n} (missing {missing}) from {addr}"); + if missing > 0 { + eprintln!("FAIL: {missing} durable message(s) did NOT survive the crash"); + std::process::exit(1); + } + println!("PASS: all {n} durable messages survived the crash"); + } + other => { + eprintln!("RECOVER_PHASE must be produce|consume, got {other}"); + std::process::exit(2); + } + } + Ok(()) +} diff --git a/ramqp-broker/examples/robust.rs b/ramqp-broker/examples/robust.rs new file mode 100644 index 0000000..dd0115b --- /dev/null +++ b/ramqp-broker/examples/robust.rs @@ -0,0 +1,202 @@ +//! Robustness / DoS-resilience driver for `scripts/50-robust.sh`. +//! +//! Hammers a **live** broker daemon with hostile traffic and, after each wave, +//! proves the broker is still alive by round-tripping a real message with the +//! `ramqp` client. The contract under test: adversarial peers get closed/reaped, +//! never crash the accept loop, exhaust fds, or wedge the broker for legitimate +//! clients. +//! +//! Waves: (1) connection flood — open/drop as fast as possible; (2) slow-loris — +//! connect, dribble a partial header, hold; (3) garbage flood — random bytes; +//! (4) malformed frames — a valid header then illegal frames. +//! +//! Env: `ROBUST_URL` (amqp URL), `ROBUST_SECS` (total), `ROBUST_CONNS` (fan-out). +//! Exits non-zero if any post-wave liveness check fails. + +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::{Duration, Instant}; + +use ramqp::{ConnectionBuilder, Message}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpStream; + +fn env_usize(k: &str, d: usize) -> usize { + std::env::var(k) + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(d) +} +fn env_string(k: &str, d: &str) -> String { + std::env::var(k).unwrap_or_else(|_| d.to_owned()) +} + +static SEED: AtomicU64 = AtomicU64::new(0x9e3779b97f4a7c15); +fn xorshift() -> u64 { + let mut x = SEED.fetch_add(0x2545f4914f6cdd1d, Ordering::Relaxed) | 1; + x ^= x << 13; + x ^= x >> 7; + x ^= x << 17; + x +} + +/// One legitimate produce→consume round trip. `true` = the broker is healthy. +async fn liveness(url: &str) -> bool { + let attempt = async { + let conn = ConnectionBuilder::new(url).connect().await.ok()?; + let session = conn.begin_session().await.ok()?; + let producer = session + .create_producer("/queues/robust-canary") + .await + .ok()?; + let mut consumer = session + .create_consumer("/queues/robust-canary") + .await + .ok()?; + producer + .send_settled(Message::data(b"ping".to_vec())) + .await + .ok()?; + let d = consumer.recv().await.ok()?; + consumer.accept(&d).await.ok()?; + conn.close().await.ok()?; + Some(()) + }; + matches!( + tokio::time::timeout(Duration::from_secs(10), attempt).await, + Ok(Some(())) + ) +} + +async fn conn_flood(target: &str, conns: usize, until: Instant) { + while Instant::now() < until { + let mut js = Vec::with_capacity(conns); + for _ in 0..conns { + let t = target.to_string(); + js.push(tokio::spawn(async move { + if let Ok(s) = TcpStream::connect(&t).await { + drop(s); // slam it shut with no handshake + } + })); + } + for j in js { + let _ = j.await; + } + } +} + +async fn slow_loris(target: &str, conns: usize, until: Instant) { + // Open many sockets, send a fragment of the header, then just hold them — + // the broker's inbound-handshake timeout must reap them. + let mut held = Vec::new(); + for _ in 0..conns { + if let Ok(mut s) = TcpStream::connect(target).await { + let _ = s.write_all(b"AM").await; // partial protocol header, never completed + held.push(s); + } + } + while Instant::now() < until { + tokio::time::sleep(Duration::from_millis(100)).await; + } + drop(held); +} + +async fn garbage_flood(target: &str, conns: usize, until: Instant) { + while Instant::now() < until { + let mut js = Vec::with_capacity(conns); + for _ in 0..conns { + let t = target.to_string(); + js.push(tokio::spawn(async move { + if let Ok(mut s) = TcpStream::connect(&t).await { + let mut buf = [0u8; 256]; + for b in buf.iter_mut() { + *b = xorshift() as u8; + } + let _ = s.write_all(&buf).await; + let mut sink = [0u8; 256]; + let _ = + tokio::time::timeout(Duration::from_millis(200), s.read(&mut sink)).await; + } + })); + } + for j in js { + let _ = j.await; + } + } +} + +async fn malformed_frames(target: &str, until: Instant) { + // A valid bare-AMQP header (the default broker offers it) then a series of + // illegal frames: undersized, oversized, bad data-offset, random tail. + let bad_frames: [Vec; 4] = [ + vec![0, 0, 0, 4, 2, 0, 0, 0], // size 4 < header len 8 + vec![0xFF, 0xFF, 0xFF, 0xFF, 2, 0, 0, 0], // size ~4GiB > max-frame-size + vec![0, 0, 0, 8, 1, 0, 0, 0], // data-offset 1 → header_len 4 < 8 + vec![0, 0, 0, 12, 2, 0, 0, 0, 0xDE, 0xAD, 0xBE, 0xEF], // garbage body + ]; + while Instant::now() < until { + if let Ok(mut s) = TcpStream::connect(target).await { + let _ = s.write_all(b"AMQP\x00\x01\x00\x00").await; + let mut echo = [0u8; 8]; + let _ = tokio::time::timeout(Duration::from_millis(200), s.read_exact(&mut echo)).await; + let f = &bad_frames[(xorshift() as usize) % bad_frames.len()]; + let _ = s.write_all(f).await; + let mut sink = [0u8; 512]; + let _ = tokio::time::timeout(Duration::from_millis(200), s.read(&mut sink)).await; + } + } +} + +#[tokio::main(flavor = "multi_thread")] +async fn main() -> Result<(), Box> { + let url = env_string("ROBUST_URL", "amqp://127.0.0.1:5672"); + let secs = env_usize("ROBUST_SECS", 20) as u64; + let conns = env_usize("ROBUST_CONNS", 200).max(1); + let authority = url.split("://").nth(1).unwrap_or(&url); + let target = authority.split('/').next().unwrap_or(authority).to_string(); + let phase = Duration::from_secs((secs / 4).max(2)); + + println!( + "robust: target={target} conns={conns} phase={}s", + phase.as_secs() + ); + + let mut failures = 0usize; + if !liveness(&url).await { + eprintln!("FAIL: broker not healthy before attacks even began"); + std::process::exit(3); + } + println!("liveness: ok (baseline)"); + + let waves: [(&str, _); 4] = [ + ("connection-flood", 0u8), + ("slow-loris", 1u8), + ("garbage-flood", 2u8), + ("malformed-frames", 3u8), + ]; + for (name, kind) in waves { + println!("wave: {name} for {}s ...", phase.as_secs()); + let until = Instant::now() + phase; + match kind { + 0 => conn_flood(&target, conns, until).await, + 1 => slow_loris(&target, conns, until).await, + 2 => garbage_flood(&target, conns, until).await, + _ => malformed_frames(&target, until).await, + } + // Give the broker a beat to reap, then check it still serves clients. + tokio::time::sleep(Duration::from_millis(500)).await; + if liveness(&url).await { + println!("liveness: ok (after {name})"); + } else { + eprintln!("FAIL: broker unresponsive after {name}"); + failures += 1; + } + } + + if failures == 0 { + println!("PASS: broker stayed live through all {} waves", waves.len()); + Ok(()) + } else { + eprintln!("FAIL: {failures} liveness check(s) failed"); + std::process::exit(1); + } +} diff --git a/ramqp-broker/scripts/00-gates.sh b/ramqp-broker/scripts/00-gates.sh new file mode 100755 index 0000000..89b0c11 --- /dev/null +++ b/ramqp-broker/scripts/00-gates.sh @@ -0,0 +1,51 @@ +#!/usr/bin/env bash +# Stage 0 — static gates: everything that must be clean before a release, +# independent of any running broker. Fast and deterministic. +# +# fmt · clippy -D warnings · check --all-features · docs · audit · deny +# +# These mirror (and slightly exceed) what CI enforces on `main`; running them +# here means a working branch never surprises CI. + +source "$(dirname "${BASH_SOURCE[0]}")/lib.sh" + +section "stage 0: static gates" + +check "rustfmt --check" \ + cargo fmt --all -- --check + +check "clippy -D warnings (all targets, all features)" \ + cargo clippy --all-targets --all-features -- -D warnings + +check "cargo check (all targets, all features)" \ + cargo check --all-targets --all-features + +RUSTDOCFLAGS="-D warnings" check "cargo doc (all features, no deps)" \ + cargo doc --all-features --no-deps -q + +# --- supply chain ---------------------------------------------------------- +# `cargo audit` and `cargo deny` both read the RustSec DB; keeping both gives an +# independent second opinion. Advisories/bans/sources are hard gates. Licenses +# is soft: the tree pulls the permissive Unicode-3.0 (url→idna→icu) which needs +# a deny.toml allowance — a config gap, not a compliance problem — so it warns +# rather than fails until that file lands. +if require_cmd cargo-audit; then + check "cargo audit" cargo audit +else + warn "cargo-audit not installed; skipping (cargo install cargo-audit)" +fi + +if require_cmd cargo-deny; then + check "cargo deny: advisories" cargo deny check advisories + check "cargo deny: bans" cargo deny check bans + check "cargo deny: sources" cargo deny check sources + if cargo deny check licenses >/dev/null 2>&1; then + pass "cargo deny: licenses" + else + warn "cargo deny: licenses — needs a deny.toml (Unicode-3.0 allowance); not gating" + fi +else + warn "cargo-deny not installed; skipping (cargo install cargo-deny)" +fi + +finish diff --git a/ramqp-broker/scripts/10-suite.sh b/ramqp-broker/scripts/10-suite.sh new file mode 100755 index 0000000..0f3b139 --- /dev/null +++ b/ramqp-broker/scripts/10-suite.sh @@ -0,0 +1,58 @@ +#!/usr/bin/env bash +# Stage 1 — the full test suite, both feature sets, plus a flake-repeat loop. +# +# The broker's nastiest bug so far (the close-time settlement drain requeuing +# acked messages) only showed up on *repeated* runs against a long-lived +# process. Deterministic-looking suites still hide races — so we run the broker +# suite several times and require every pass green. nextest gives fast parallel +# runs and a clean per-test report; we fall back to `cargo test` if it's absent. +# +# Knobs: RAMQP_SUITE_REPEAT (default 3). + +source "$(dirname "${BASH_SOURCE[0]}")/lib.sh" + +section "stage 1: test suite + flake loop" + +REPEAT="${RAMQP_SUITE_REPEAT:-3}" +LOG="$RAMQP_OUT/suite" + +if command -v cargo-nextest >/dev/null 2>&1; then + RUN=(cargo nextest run) + info "using cargo-nextest" +else + RUN=(cargo test) + warn "cargo-nextest not installed; using 'cargo test' (slower, no per-test retry report)" +fi + +# Full workspace (default-members), default features then all features. The +# all-features pass is the one that exercises store-redb (durable queues). +check "workspace suite — default features" \ + "${RUN[@]}" + +check "workspace suite — all features" \ + "${RUN[@]}" --all-features + +# nextest does not run doctests; cargo test --doc does. The broker README and +# lib carry runnable examples, so cover them explicitly. +check "doctests — all features" \ + cargo test --doc --all-features -q + +# Flake loop: hammer the broker suite specifically (the timing-sensitive part). +section "flake loop: broker suite ×$REPEAT (all features)" +flakes=0 +for i in $(seq 1 "$REPEAT"); do + if "${RUN[@]}" -p ramqp-broker --all-features >"$LOG.repeat-$i.log" 2>&1; then + ok "broker suite pass $i/$REPEAT" + else + flakes=$((flakes + 1)) + err "broker suite FAILED on pass $i/$REPEAT — see $LOG.repeat-$i.log" + tail -20 "$LOG.repeat-$i.log" || true + fi +done +if ((flakes == 0)); then + pass "flake loop: $REPEAT/$REPEAT green" +else + fail "flake loop: $flakes/$REPEAT runs failed (nondeterminism or a real race)" +fi + +finish diff --git a/ramqp-broker/scripts/20-soak.sh b/ramqp-broker/scripts/20-soak.sh new file mode 100755 index 0000000..891bb39 --- /dev/null +++ b/ramqp-broker/scripts/20-soak.sh @@ -0,0 +1,75 @@ +#!/usr/bin/env bash +# Stage 2 — soak / leak detector. +# +# Runs a separate brokerd process under sustained, churny load for N minutes and +# samples the BROKER's RSS the whole time. Two failure modes it catches: +# * a memory leak — RSS climbs instead of holding flat under bounded depth; +# * throughput decay — the class of regression that exposed the close-time +# settlement-drain bug (repeated busy connection closes degrading the broker). +# Connection churn is on by default so the close path is exercised hard. +# +# Knobs: RAMQP_SOAK_SECS (120) RAMQP_SOAK_PAIRS (8) RAMQP_SOAK_CHURN (500). + +source "$(dirname "${BASH_SOURCE[0]}")/lib.sh" + +section "stage 2: soak / leak" +require_cmd python3 || { finish; exit 1; } + +SECS="${RAMQP_SOAK_SECS:-120}" +PAIRS="${RAMQP_SOAK_PAIRS:-8}" +CHURN="${RAMQP_SOAK_CHURN:-500}" + +build_brokerd >/dev/null +info "building loadgen" +cargo build --release -q -p ramqp-broker --example loadgen +LOADGEN="$ROOT/target/release/examples/loadgen" + +PORT="$(free_port)" +spawn_brokerd soak-broker -- --listen "127.0.0.1:$PORT" +if ! wait_port "$PORT" 30; then + fail "soak: broker never came up on :$PORT" + finish; exit 1 +fi +ok "broker up on :$PORT (pid $BROKERD_PID), soaking ${SECS}s with churn=$CHURN" + +# Background RSS sampler. +RSSF="$RAMQP_OUT/soak-rss.tsv" +: >"$RSSF" +( while kill -0 "$BROKERD_PID" 2>/dev/null; do + printf '%s\t%s\n' "$(date +%s)" "$(rss_kib "$BROKERD_PID")" >>"$RSSF" + sleep 2 + done ) & +sampler=$! + +LOADLOG="$RAMQP_OUT/soak-loadgen.log" +LOAD_URL="amqp://127.0.0.1:$PORT" LOAD_ADDRESS="/queues/soak" \ + LOAD_SECS="$SECS" LOAD_PAIRS="$PAIRS" LOAD_CHURN="$CHURN" LOAD_REPORT_SECS=5 \ + "$LOADGEN" 2>&1 | tee "$LOADLOG" + +kill "$sampler" 2>/dev/null || true + +# --- evaluate -------------------------------------------------------------- +eval "$(python3 "$SCRIPTS_DIR/analyze_soak.py" "$RSSF" "$LOADLOG")" +info "RSS: early=${RSS_EARLY_KIB}KiB late=${RSS_LATE_KIB}KiB leak=${RSS_LEAK_KIB}KiB peak=${RSS_PEAK_KIB}KiB (${SAMPLES_RSS} samples)" +info "throughput: early=${RATE_EARLY} late=${RATE_LATE} msg/s (${SAMPLES_RATE} samples)" + +assert "soak: RSS sampler collected data" "[ ${SAMPLES_RSS:-0} -ge 5 ]" + +# Leak: fail only if RSS grew BOTH >50 MiB absolute AND >25% relative (avoids +# flagging normal allocator/steady-state jitter). +if [ "${RSS_LEAK_KIB:-0}" -gt 51200 ] && [ $(( RSS_LATE_KIB * 100 )) -gt $(( RSS_EARLY_KIB * 125 )) ]; then + fail "soak: broker RSS grew ${RSS_LEAK_KIB}KiB (${RSS_EARLY_KIB}→${RSS_LATE_KIB}) — possible leak" +else + pass "soak: broker RSS stayed flat under sustained churn" +fi + +# Throughput decay: late window must hold >=60% of the early window. +if [ "${RATE_EARLY:-0}" -gt 0 ] && [ $(( RATE_LATE * 100 )) -lt $(( RATE_EARLY * 60 )) ]; then + fail "soak: throughput decayed ${RATE_EARLY}→${RATE_LATE} msg/s (>40% drop) — degradation under long run" +else + pass "soak: throughput held steady (no degradation)" +fi + +check "soak: no broker panics" no_panics soak-broker + +finish diff --git a/ramqp-broker/scripts/30-chaos.sh b/ramqp-broker/scripts/30-chaos.sh new file mode 100755 index 0000000..c316705 --- /dev/null +++ b/ramqp-broker/scripts/30-chaos.sh @@ -0,0 +1,138 @@ +#!/usr/bin/env bash +# Stage 3 — chaos / fault injection against real broker PROCESSES (kill -9, +# cold restart from disk — stronger than the in-process suite's graceful stops). +# +# Part A: a 3-node quorum cluster (on-disk Raft via store-redb + --data-dir). +# A verifying client is pinned to node 1 (kept alive) while nodes 2 and 3 are +# killed and restarted one at a time (quorum 2/3 always held). Contract: every +# ACCEPTED message is eventually delivered — zero loss across failovers. +# Part B: single-node durable crash recovery — produce N, SIGKILL, cold start on +# the same data dir, and every durable message must still be there. +# +# Knobs: RAMQP_CHAOS_ROUNDS (4) RAMQP_CHAOS_N (20000) RAMQP_RECOVER_N (5000). + +source "$(dirname "${BASH_SOURCE[0]}")/lib.sh" + +section "stage 3: chaos / fault injection" + +ROUNDS="${RAMQP_CHAOS_ROUNDS:-4}" +N="${RAMQP_CHAOS_N:-20000}" +RN="${RAMQP_RECOVER_N:-5000}" + +build_brokerd store-redb >/dev/null # on-disk Raft is required to survive restart +info "building chaos + recover drivers" +cargo build --release -q -p ramqp-broker --features store-redb --example chaos --example recover +CHAOS="$ROOT/target/release/examples/chaos" +RECOVER="$ROOT/target/release/examples/recover" + +# --------------------------------------------------------------------------- +# Part A: rolling kills against a 3-node quorum cluster. +# --------------------------------------------------------------------------- +section "part A: 3-node cluster, rolling leader/follower kills ($ROUNDS rounds)" + +declare -A AMQP FAB DDIR PID +for i in 1 2 3; do + AMQP[$i]="$(free_port)" + FAB[$i]="$(free_port)" + DDIR[$i]="$RAMQP_OUT/chaos-node$i" + mkdir -p "${DDIR[$i]}" +done +SEEDS=(--seed "1=127.0.0.1:${FAB[1]}" --seed "2=127.0.0.1:${FAB[2]}" --seed "3=127.0.0.1:${FAB[3]}") + +start_node() { + local i="$1" + spawn_brokerd "chaos-node$i" -- \ + --listen "127.0.0.1:${AMQP[$i]}" \ + --node-id "$i" --cluster-listen "127.0.0.1:${FAB[$i]}" \ + "${SEEDS[@]}" --data-dir "${DDIR[$i]}" + PID[$i]="$BROKERD_PID" +} + +for i in 1 2 3; do start_node "$i"; done +for i in 1 2 3; do + if ! wait_port "${AMQP[$i]}" 30; then + fail "chaos: node $i never came up"; finish; exit 1 + fi +done +ok "3-node cluster up (AMQP ${AMQP[1]},${AMQP[2]},${AMQP[3]})" +sleep 5 # allow cluster formation before the client declares the quorum queue + +# Verifying client pinned to node 1 (never killed), running in the background. +CLIENTLOG="$RAMQP_OUT/chaos-client.log" +CHAOS_PRODUCER_URL="amqp://127.0.0.1:${AMQP[1]}" \ +CHAOS_CONSUMER_URL="amqp://127.0.0.1:${AMQP[1]}" \ +CHAOS_QUEUE="/quorum/chaos" CHAOS_N="$N" \ +CHAOS_DEADLINE_SECS=$((ROUNDS * 30 + 150)) \ + "$CHAOS" >"$CLIENTLOG" 2>&1 & +chaos_pid=$! + +# Rolling kill/restart of nodes 2 and 3 only (node 1 stays for the client; +# never more than one down at a time, so quorum is always held). +victims=(2 3) +for ((r = 0; r < ROUNDS; r++)); do + v="${victims[$((r % 2))]}" + info "round $r: SIGKILL node $v (pid ${PID[$v]})" + kill -9 "${PID[$v]}" 2>/dev/null || true + untrack_pid "${PID[$v]}" + wait_port_down "${AMQP[$v]}" 15 || warn "node $v port still up after kill" + sleep 4 # let the survivors re-elect / heal while the node is down + info "round $r: cold-restart node $v" + start_node "$v" + wait_port "${AMQP[$v]}" 30 || warn "node $v did not rebind :${AMQP[$v]}" + sleep 6 # let it rejoin + catch up before the next round touches the other node + kill -0 "$chaos_pid" 2>/dev/null || { info "chaos client finished early"; break; } +done + +info "kill rounds done; waiting for the verifying client to finish" +if wait "$chaos_pid"; then + pass "chaos: zero accepted-message loss across $ROUNDS kill/restart rounds" +else + rc=$? + fail "chaos: verifying client reported failure (rc=$rc) — see chaos-client.log" + tail -8 "$CLIENTLOG" | sed 's/^/ /' >&2 || true +fi +grep -E 'result:|PASS|FAIL' "$CLIENTLOG" | sed 's/^/ /' || true +check "chaos: no broker panics (cluster nodes)" no_panics chaos-node1 chaos-node2 chaos-node3 + +# Tear down the cluster before part B. +for i in 1 2 3; do kill "${PID[$i]}" 2>/dev/null || true; untrack_pid "${PID[$i]}"; done +sleep 1 + +# --------------------------------------------------------------------------- +# Part B: single-node durable crash recovery (kill -9 + cold start). +# --------------------------------------------------------------------------- +section "part B: durable crash recovery (produce → kill -9 → cold start → verify)" + +BPORT="$(free_port)" +BDIR="$RAMQP_OUT/recover-node" +mkdir -p "$BDIR" + +spawn_brokerd recover-broker -- --listen "127.0.0.1:$BPORT" --data-dir "$BDIR" +if ! wait_port "$BPORT" 30; then fail "recover: broker never came up"; finish; exit 1; fi + +info "producing $RN durable messages" +if RECOVER_URL="amqp://127.0.0.1:$BPORT" RECOVER_ADDRESS="/durable/recovery" \ + RECOVER_N="$RN" RECOVER_PHASE=produce "$RECOVER"; then + ok "durable produce confirmed" +else + fail "recover: durable produce phase failed"; finish; exit 1 +fi + +info "SIGKILL the broker (simulated crash)" +kill -9 "$BROKERD_PID" 2>/dev/null || true +untrack_pid "$BROKERD_PID" +wait_port_down "$BPORT" 15 || warn "recover: port still up after kill" + +info "cold-starting a fresh process on the same data dir" +spawn_brokerd recover-broker -- --listen "127.0.0.1:$BPORT" --data-dir "$BDIR" +if ! wait_port "$BPORT" 30; then fail "recover: broker did not restart"; finish; exit 1; fi + +if RECOVER_URL="amqp://127.0.0.1:$BPORT" RECOVER_ADDRESS="/durable/recovery" \ + RECOVER_N="$RN" RECOVER_PHASE=consume "$RECOVER"; then + pass "durable: all $RN messages survived kill -9 + cold start" +else + fail "durable: crash recovery lost messages" +fi +check "durable: no broker panics on recovery" no_panics recover-broker + +finish diff --git a/ramqp-broker/scripts/40-interop.sh b/ramqp-broker/scripts/40-interop.sh new file mode 100755 index 0000000..07c6f0c --- /dev/null +++ b/ramqp-broker/scripts/40-interop.sh @@ -0,0 +1,78 @@ +#!/usr/bin/env bash +# Stage 4 — interop matrix. +# +# Proves ramqp-broker speaks real AMQP 1.0, two ways: +# * the `ramqp` CLIENT interop suite (the same one CI runs against RabbitMQ and +# Artemis) passes against ramqp-broker, RabbitMQ 4.x, and Artemis — so our +# broker behaves like the brokers people deploy; +# * an INDEPENDENT client, `fe2o3-amqp`, interops with ramqp-broker — so we are +# not just agreeing with our own client's quirks. +# +# Each external broker leg is skipped (with a warning) if its container is down. +# Fresh per-run queues are used so pre-existing backlogs never taint results. + +source "$(dirname "${BASH_SOURCE[0]}")/lib.sh" + +section "stage 4: interop matrix" + +RID="$$" +ART_BIN=/var/lib/artemis-instance/bin/artemis + +# Run the ramqp client interop suite against a URL+address; log per target. +client_suite() { # url address logtag + RAMQP_BROKER_URL="$1" RAMQP_BROKER_ADDRESS="$2" \ + cargo test -q -p ramqp --test broker -- --ignored --test-threads=1 \ + >"$RAMQP_OUT/interop-$3.log" 2>&1 +} + +# --- ramqp-broker ---------------------------------------------------------- +build_brokerd >/dev/null +PORT="$(free_port)" +spawn_brokerd interop-broker -- --listen "127.0.0.1:$PORT" +if wait_port "$PORT" 30; then + check "ramqp client → ramqp-broker" \ + client_suite "amqp://127.0.0.1:$PORT" "/queues/interop-$RID" ramqp-broker +else + fail "interop: ramqp-broker never came up" +fi + +# --- RabbitMQ 4.x ---------------------------------------------------------- +if container_up rabbit; then + Q="ramqp_interop_$RID" + if curl -fsS -u guest:guest -X PUT "http://localhost:15672/api/queues/%2F/$Q" \ + -H content-type:application/json -d '{"durable":true}' >/dev/null 2>&1; then + check "ramqp client → RabbitMQ 4.x" \ + client_suite "amqp://guest:guest@localhost:5672" "/queues/$Q" rabbitmq + curl -fsS -u guest:guest -X DELETE "http://localhost:15672/api/queues/%2F/$Q" >/dev/null 2>&1 || true + else + warn "RabbitMQ mgmt API not reachable; skipping" + fi +else + warn "rabbit container not up; skipping RabbitMQ leg" +fi + +# --- ActiveMQ Artemis ------------------------------------------------------ +if container_up artemis; then + Q="ramqp_interop_$RID" + # Artemis auto-creates MULTICAST (drops pre-subscribe sends); pre-create an + # ANYCAST/queue-semantics address so produce-then-consume works. + if docker exec artemis "$ART_BIN" queue create --name "$Q" --address "$Q" \ + --anycast --durable --preserve-on-no-consumers --auto-create-address \ + --url tcp://localhost:61616 --user guest --password guest --silent >/dev/null 2>&1; then + check "ramqp client → Artemis" \ + client_suite "amqp://guest:guest@localhost:5682" "$Q" artemis + docker exec artemis "$ART_BIN" queue delete --name "$Q" \ + --url tcp://localhost:61616 --user guest --password guest >/dev/null 2>&1 || true + else + warn "could not create Artemis queue; skipping" + fi +else + warn "artemis container not up; skipping Artemis leg" +fi + +# --- independent client (fe2o3-amqp) → ramqp-broker ------------------------ +# These tests bring up ramqp-broker in-process and drive it with fe2o3-amqp. +check "fe2o3-amqp client → ramqp-broker (independent impl)" \ + cargo test -q -p ramqp-bench-compare --test fe2o3_interop + +finish diff --git a/ramqp-broker/scripts/50-robust.sh b/ramqp-broker/scripts/50-robust.sh new file mode 100755 index 0000000..7d95455 --- /dev/null +++ b/ramqp-broker/scripts/50-robust.sh @@ -0,0 +1,37 @@ +#!/usr/bin/env bash +# Stage 5 — robustness / DoS resilience against a LIVE broker daemon. +# +# The `robust` driver hits the broker with connection floods, slow-loris, +# garbage bytes, and malformed frames, and after every wave round-trips a real +# message with the ramqp client. The broker must stay live and responsive +# throughout — adversarial peers get reaped, they never wedge the accept loop, +# exhaust fds, or crash the process. +# +# Knobs: RAMQP_ROBUST_SECS (20, total across 4 waves) RAMQP_ROBUST_CONNS (200). + +source "$(dirname "${BASH_SOURCE[0]}")/lib.sh" + +section "stage 5: robustness" + +SECS="${RAMQP_ROBUST_SECS:-20}" +CONNS="${RAMQP_ROBUST_CONNS:-200}" + +build_brokerd >/dev/null +info "building robust driver" +cargo build --release -q -p ramqp-broker --example robust +ROBUST="$ROOT/target/release/examples/robust" + +PORT="$(free_port)" +spawn_brokerd robust-broker -- --listen "127.0.0.1:$PORT" +if ! wait_port "$PORT" 30; then + fail "robust: broker never came up"; finish; exit 1 +fi +ok "broker up on :$PORT (pid $BROKERD_PID)" + +check "robust: broker stays live under floods / slow-loris / malformed frames" \ + env "ROBUST_URL=amqp://127.0.0.1:$PORT" "ROBUST_SECS=$SECS" "ROBUST_CONNS=$CONNS" "$ROBUST" + +assert "robust: broker process still alive after all attacks" "kill -0 $BROKERD_PID 2>/dev/null" +check "robust: no broker panics" no_panics robust-broker + +finish diff --git a/ramqp-broker/scripts/60-fuzz.sh b/ramqp-broker/scripts/60-fuzz.sh new file mode 100755 index 0000000..9292a00 --- /dev/null +++ b/ramqp-broker/scripts/60-fuzz.sh @@ -0,0 +1,39 @@ +#!/usr/bin/env bash +# Stage 6 — fuzz the untrusted-wire decoders (a broker parses bytes from anyone +# who can open a socket, so this is the highest-value robustness surface). +# Bounded time per target; a crash writes a reproducer under fuzz/artifacts. +# +# Needs a nightly toolchain and cargo-fuzz. Skips (warns) if either is absent. +# +# Knobs: RAMQP_FUZZ_SECS (60, per target). + +source "$(dirname "${BASH_SOURCE[0]}")/lib.sh" + +section "stage 6: fuzz (decode_frame, Value)" + +FUZZ_SECS="${RAMQP_FUZZ_SECS:-60}" + +if ! rustup toolchain list 2>/dev/null | grep -q nightly; then + warn "no nightly toolchain (rustup toolchain install nightly); skipping fuzz" + finish; exit 0 +fi +if ! command -v cargo-fuzz >/dev/null 2>&1; then + warn "cargo-fuzz not installed (cargo install cargo-fuzz); skipping fuzz" + finish; exit 0 +fi + +cd "$ROOT/ramqp-core" +for target in decode_frame value; do + info "fuzzing $target for ${FUZZ_SECS}s" + if cargo +nightly fuzz run "$target" -- \ + -max_total_time="$FUZZ_SECS" -rss_limit_mb=4096 \ + >"$RAMQP_OUT/fuzz-$target.log" 2>&1; then + pass "fuzz: $target — no crash in ${FUZZ_SECS}s" + else + fail "fuzz: $target — CRASH found (repro in ramqp-core/fuzz/artifacts/$target, log: fuzz-$target.log)" + tail -25 "$RAMQP_OUT/fuzz-$target.log" | sed 's/^/ /' || true + fi +done +cd "$ROOT" + +finish diff --git a/ramqp-broker/scripts/README.md b/ramqp-broker/scripts/README.md new file mode 100644 index 0000000..694b711 --- /dev/null +++ b/ramqp-broker/scripts/README.md @@ -0,0 +1,58 @@ +# ramqp-broker test scripts + +A consistent, re-runnable battery for the broker — the same checks build over +build, so regressions in correctness, memory, HA, interop, or robustness are +caught before a crates.io release. **None of these run in CI** (yet); they are +run by hand or via `run-all.sh`. + +Every script sources [`lib.sh`](lib.sh) (logging, PASS/FAIL accounting, brokerd +spawn/teardown, port + RSS helpers), prints a clear per-check pass/fail, exits +non-zero on any failure, and writes artifacts (logs, metrics, spawned data +dirs) to a timestamped `out//` directory (gitignored). + +## Quick start + +```sh +cd +ramqp-broker/scripts/run-all.sh --quick # gates + full test suite (fast) +ramqp-broker/scripts/run-all.sh # the default battery +ramqp-broker/scripts/run-all.sh chaos soak # a specific subset, in order +``` + +## The stages + +| # | Script | What it proves | Needs | +|---|--------|----------------|-------| +| 0 | `00-gates.sh` | fmt, clippy `-D warnings`, `cargo check --all-features`, docs, `cargo audit`, `cargo deny` all clean | — | +| 1 | `10-suite.sh` | the full ~150-test suite passes on **both** feature sets, and stays green across a **flake-repeat loop** (the coop-budget requeue bug only surfaced on repeated runs) | nextest | +| 2 | `20-soak.sh` | sustained + churny load for N minutes leaves broker RSS **flat** and throughput **non-degrading** — the leak/degradation detector | — | +| 3 | `30-chaos.sh` | a 3-node cluster with rolling leader/follower **kills + restarts** loses **zero accepted messages** and recovers; durable data survives restart | store-redb | +| 4 | `40-interop.sh` | the `ramqp` client interop suite passes against ramqp-broker, **RabbitMQ**, and **Artemis**; the independent **fe2o3-amqp** client interops with ramqp-broker | docker | +| 5 | `50-robust.sh` | the broker stays **live and responsive** under connection floods, slow-loris, and malformed/oversized-frame floods against a live daemon | — | +| 6 | `60-fuzz.sh` | the untrusted-wire decoders (`decode_frame`, `Value`) survive bounded **fuzzing** with no panic/hang | cargo-fuzz (nightly) | + +## Manual-only (never in `run-all`, never in CI) + +| Script | Purpose | +|--------|---------| +| `bench.sh` | latency / depth / throughput harness; captures timestamped JSON and diffs against a committed `baseline.json` to flag perf drift. **Bench is never automated.** | +| `cov.sh` | `cargo llvm-cov` coverage report for the broker + core, to find untested paths. | + +## Tuning knobs (env) + +| Var | Default | Used by | +|-----|---------|---------| +| `RAMQP_OUT` | `out/` | all (share one dir across a run) | +| `RAMQP_SUITE_REPEAT` | 3 | suite flake loop | +| `RAMQP_SOAK_SECS` | 120 | soak | +| `RAMQP_SOAK_PAIRS` | 8 | soak concurrency | +| `RAMQP_CHAOS_ROUNDS` | 4 | chaos kill/restart rounds | +| `RAMQP_CHAOS_N` | 20000 | chaos messages to verify | +| `RAMQP_ROBUST_SECS` | 20 | robustness flood duration | +| `RAMQP_FUZZ_SECS` | 60 | per-target fuzz time | + +## Driver binaries + +Load/chaos/robustness drivers that can't be expressed in bash live as **example +binaries** in [`../examples/`](../examples) (`loadgen`, `chaos`, `robust`), so +they are compiled by the normal `cargo check --all-targets` and never rot. diff --git a/ramqp-broker/scripts/analyze_soak.py b/ramqp-broker/scripts/analyze_soak.py new file mode 100644 index 0000000..d6976d4 --- /dev/null +++ b/ramqp-broker/scripts/analyze_soak.py @@ -0,0 +1,60 @@ +#!/usr/bin/env python3 +"""Summarize a soak run into KEY=VALUE lines the soak script evals. + +Reads the RSS sample file (`\\t` per line) and the loadgen log +(`... rate= msg/s ...`), then reports early-vs-late medians so the caller can +assert on memory growth (leak) and throughput decay. Emits nothing but the +KEY=VALUE lines on stdout. +""" +import re +import statistics +import sys + + +def median(xs): + return statistics.median(xs) if xs else 0 + + +def window(xs, lo, hi): + n = len(xs) + if n == 0: + return [] + a, b = int(n * lo), int(n * hi) + return xs[a:max(b, a + 1)] + + +def main(): + rss_file, load_log = sys.argv[1], sys.argv[2] + + rss = [] + with open(rss_file) as f: + for line in f: + parts = line.split() + if len(parts) == 2 and parts[1].isdigit(): + rss.append(int(parts[1])) + + rates = [] + with open(load_log) as f: + for line in f: + m = re.search(r"rate=(\d+)", line) + if m: + rates.append(int(m.group(1))) + + # Discard the first RSS decile (warm-up allocation ramp) before measuring. + rss_early = median(window(rss, 0.10, 0.35)) + rss_late = median(window(rss, 0.75, 1.00)) + rate_early = median(window(rates, 0.00, 0.34)) + rate_late = median(window(rates, 0.66, 1.00)) + + print(f"RSS_EARLY_KIB={int(rss_early)}") + print(f"RSS_LATE_KIB={int(rss_late)}") + print(f"RSS_LEAK_KIB={int(rss_late - rss_early)}") + print(f"RSS_PEAK_KIB={int(max(rss) if rss else 0)}") + print(f"RATE_EARLY={int(rate_early)}") + print(f"RATE_LATE={int(rate_late)}") + print(f"SAMPLES_RSS={len(rss)}") + print(f"SAMPLES_RATE={len(rates)}") + + +if __name__ == "__main__": + main() diff --git a/ramqp-broker/scripts/bench-baseline.json b/ramqp-broker/scripts/bench-baseline.json new file mode 100644 index 0000000..e664e88 --- /dev/null +++ b/ramqp-broker/scripts/bench-baseline.json @@ -0,0 +1,9 @@ +{ + "p50_us": 91.2, + "p90_us": 120.4, + "p99_us": 202.9, + "p999_us": 283.6, + "max_us": 485.2, + "throughput_msgs": 286179.0, + "rss_mib": 24.7 +} diff --git a/ramqp-broker/scripts/bench.sh b/ramqp-broker/scripts/bench.sh new file mode 100755 index 0000000..0ac3ddf --- /dev/null +++ b/ramqp-broker/scripts/bench.sh @@ -0,0 +1,51 @@ +#!/usr/bin/env bash +# MANUAL performance runner — NOT part of run-all, NEVER in CI. Benchmarks must +# not gate anything (numbers move with the machine); this is for a human to spot +# drift build-over-build. +# +# Runs the `latency` bin several trials against the in-process ramqp-broker (or +# an external one via LAT_URL), medians the metrics, and diffs a committed +# baseline. `--save-baseline` stores the current run as the new baseline (commit +# it so drift shows up in git). +# +# scripts/bench.sh # run + compare to baseline +# scripts/bench.sh --save-baseline # run + store baseline +# LAT_URL=amqp://host:5672 scripts/bench.sh # against an external broker +# +# Knobs: BENCH_TRIALS (5) + latency-bin knobs (LAT_N, LAT_LAT_N, LAT_PAYLOAD, ...). + +source "$(dirname "${BASH_SOURCE[0]}")/lib.sh" + +section "manual perf bench (not a gate)" + +TRIALS="${BENCH_TRIALS:-5}" +BASELINE="$SCRIPTS_DIR/bench-baseline.json" +SAVE=0 +[[ "${1:-}" == "--save-baseline" ]] && SAVE=1 + +info "building latency bin" +cargo build --release -q -p ramqp-bench-compare --bin latency +LAT="$ROOT/target/release/latency" + +LOG="$RAMQP_OUT/bench-latency.log" +: >"$LOG" +for t in $(seq 1 "$TRIALS"); do + info "trial $t/$TRIALS" + "$LAT" 2>&1 | tee -a "$LOG" +done + +METRICS="$RAMQP_OUT/bench-metrics.json" +python3 "$SCRIPTS_DIR/bench_stats.py" parse "$LOG" >"$METRICS" +section "median over $TRIALS trials" +cat "$METRICS" + +if [[ $SAVE -eq 1 ]]; then + cp "$METRICS" "$BASELINE" + ok "baseline saved → $BASELINE (commit it to track perf over builds)" +elif [[ -f "$BASELINE" ]]; then + section "vs baseline" + python3 "$SCRIPTS_DIR/bench_stats.py" compare "$METRICS" "$BASELINE" \ + || warn "perf drift beyond tolerance (see table above)" +else + warn "no baseline yet — run 'scripts/bench.sh --save-baseline' to create one" +fi diff --git a/ramqp-broker/scripts/bench_stats.py b/ramqp-broker/scripts/bench_stats.py new file mode 100644 index 0000000..8081cfc --- /dev/null +++ b/ramqp-broker/scripts/bench_stats.py @@ -0,0 +1,103 @@ +#!/usr/bin/env python3 +"""Parse `latency`-bin output into median metrics; optionally diff a baseline. + +Usage: + bench_stats.py parse [ ...] > metrics.json + bench_stats.py compare + +`compare` prints a table and exits non-zero if any metric regressed beyond its +tolerance (latency worse if higher, throughput worse if lower). This is a manual +aid — it never runs in CI — so a non-zero exit just paints the drift red for a +human, it gates nothing. +""" +import json +import re +import statistics +import sys + +# metric -> (regex group parser). Latency line then throughput then RSS. +LAT_RE = re.compile( + r"p50\s+([\d.]+)\s+p90\s+([\d.]+)\s+p99\s+([\d.]+)\s+p99\.9\s+([\d.]+)\s+max\s+([\d.]+)" +) +THR_RE = re.compile(r"=\s*(\d+)\s*msg/s") +RSS_RE = re.compile(r"RSS:\s*([\d.]+)\s*MiB") + +# metric -> ("lower" means lower-is-better) , tolerance fraction +DIRECTION = { + "p50_us": ("lower", 0.25), + "p90_us": ("lower", 0.25), + "p99_us": ("lower", 0.30), + "p999_us": ("lower", 0.40), + "max_us": ("lower", 0.60), + "throughput_msgs": ("higher", 0.20), + "rss_mib": ("lower", 0.30), +} + + +def parse(logs): + cols = {k: [] for k in DIRECTION} + for path in logs: + with open(path) as f: + text = f.read() + for m in LAT_RE.finditer(text): + p50, p90, p99, p999, mx = (float(x) for x in m.groups()) + cols["p50_us"].append(p50) + cols["p90_us"].append(p90) + cols["p99_us"].append(p99) + cols["p999_us"].append(p999) + cols["max_us"].append(mx) + for m in THR_RE.finditer(text): + cols["throughput_msgs"].append(float(m.group(1))) + for m in RSS_RE.finditer(text): + cols["rss_mib"].append(float(m.group(1))) + return {k: round(statistics.median(v), 1) for k, v in cols.items() if v} + + +def compare(cur, base): + regressed = 0 + hdr = f"{'metric':<18}{'baseline':>12}{'current':>12}{'delta%':>10} verdict" + print(hdr) + print("-" * len(hdr)) + for k in DIRECTION: + if k not in cur or k not in base: + continue + direction, tol = DIRECTION[k] + b, c = base[k], cur[k] + if b == 0: + continue + delta = (c - b) / b * 100.0 + if direction == "lower": + bad = c > b * (1 + tol) + else: + bad = c < b * (1 - tol) + verdict = "REGRESSED" if bad else ("improved" if ( + (direction == "lower" and c < b * 0.9) or + (direction == "higher" and c > b * 1.1)) else "ok") + if bad: + regressed += 1 + print(f"{k:<18}{b:>12.1f}{c:>12.1f}{delta:>+9.1f}% {verdict}") + return regressed + + +def main(): + if len(sys.argv) < 3: + print(__doc__) + sys.exit(2) + mode = sys.argv[1] + if mode == "parse": + print(json.dumps(parse(sys.argv[2:]), indent=2)) + elif mode == "compare": + cur = json.load(open(sys.argv[2])) + base = json.load(open(sys.argv[3])) + n = compare(cur, base) + if n: + print(f"\n{n} metric(s) regressed beyond tolerance.") + sys.exit(1) + print("\nno regressions beyond tolerance.") + else: + print(__doc__) + sys.exit(2) + + +if __name__ == "__main__": + main() diff --git a/ramqp-broker/scripts/cov.sh b/ramqp-broker/scripts/cov.sh new file mode 100755 index 0000000..8d12717 --- /dev/null +++ b/ramqp-broker/scripts/cov.sh @@ -0,0 +1,33 @@ +#!/usr/bin/env bash +# MANUAL coverage report — NOT in run-all, NOT in CI. Runs the broker + core +# suites under llvm-cov instrumentation once and writes a text summary and an +# HTML report to out/. Use it to find untested paths, not as a gate. +# +# scripts/cov.sh + +source "$(dirname "${BASH_SOURCE[0]}")/lib.sh" + +section "coverage (ramqp-broker + ramqp-core, all features)" + +if ! command -v cargo-llvm-cov >/dev/null 2>&1; then + warn "cargo-llvm-cov not installed (cargo install cargo-llvm-cov)" + exit 0 +fi + +HTML="$RAMQP_OUT/coverage" +SUMMARY="$RAMQP_OUT/coverage-summary.txt" + +info "clean previous coverage data" +cargo llvm-cov clean --workspace + +info "running instrumented suites (this compiles + runs the tests once)…" +cargo llvm-cov --no-report --all-features -p ramqp-broker -p ramqp-core + +info "text summary" +cargo llvm-cov report --summary-only -p ramqp-broker -p ramqp-core | tee "$SUMMARY" + +info "html report" +cargo llvm-cov report --html --output-dir "$HTML" -p ramqp-broker -p ramqp-core + +ok "summary: $SUMMARY" +ok "html: $HTML/html/index.html" diff --git a/ramqp-broker/scripts/lib.sh b/ramqp-broker/scripts/lib.sh new file mode 100755 index 0000000..a366bf5 --- /dev/null +++ b/ramqp-broker/scripts/lib.sh @@ -0,0 +1,217 @@ +#!/usr/bin/env bash +# Shared helpers for the ramqp-broker test scripts. +# +# Every script `source`s this. It provides: consistent logging, PASS/FAIL +# accounting, a per-run output directory, a free-port finder, brokerd +# build/spawn/teardown, port-readiness waiting, and RSS sampling. Sourcing it +# installs a cleanup trap that kills every process the script spawned. +# +# Nothing here is wired into CI — these run by hand (or via run-all.sh) so the +# broker gets the same battery of checks build over build. +# +# NOTE: deliberately NOT `set -e`. This is a test orchestrator — commands are +# *expected* to fail and are handled explicitly via check/assert/`||` and the +# pass/fail tally. `set -e` would abort mid-stage on the first failing check (or +# on a helper whose last statement returns non-zero, e.g. a `[[ ]] && cmd` that +# doesn't match), which is exactly wrong here. Keep `-u` and pipefail. + +set -uo pipefail + +# --- locations ------------------------------------------------------------- + +# Repo root, robust to being invoked from anywhere. +SCRIPTS_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +ROOT="$(git -C "$SCRIPTS_DIR" rev-parse --show-toplevel 2>/dev/null || echo "${SCRIPTS_DIR%/ramqp-broker/scripts}")" + +# One timestamped output directory per run (override with RAMQP_OUT to share +# one dir across a run-all invocation). Gitignored. +: "${RAMQP_OUT:=$SCRIPTS_DIR/out/$(date +%Y%m%d-%H%M%S)}" +mkdir -p "$RAMQP_OUT" + +# --- logging --------------------------------------------------------------- + +if [[ -t 1 ]]; then + _C_RESET=$'\033[0m'; _C_RED=$'\033[31m'; _C_GRN=$'\033[32m' + _C_YEL=$'\033[33m'; _C_BLU=$'\033[34m'; _C_BOLD=$'\033[1m' +else + _C_RESET=""; _C_RED=""; _C_GRN=""; _C_YEL=""; _C_BLU=""; _C_BOLD="" +fi + +log() { printf '%s\n' "$*"; } +info() { printf '%s %s%s\n' "$_C_BLU" "$*" "$_C_RESET"; } +ok() { printf '%sok %s%s\n' "$_C_GRN" "$*" "$_C_RESET"; } +warn() { printf '%sWARN %s%s\n' "$_C_YEL" "$*" "$_C_RESET"; } +err() { printf '%sERR %s%s\n' "$_C_RED" "$*" "$_C_RESET" >&2; } +section() { printf '\n%s== %s ==%s\n' "$_C_BOLD" "$*" "$_C_RESET"; } + +# --- pass/fail accounting -------------------------------------------------- +# +# Each check calls pass/fail with a short label. A summary + exit code are +# emitted by `finish` (call it at the end, or let the EXIT trap do it). + +_PASS=0 +_FAIL=0 +declare -a _FAILED_LABELS=() + +pass() { _PASS=$((_PASS + 1)); ok "$*"; } +fail() { _FAIL=$((_FAIL + 1)); _FAILED_LABELS+=("$*"); err "FAIL: $*"; } + +# Run a command as a named check; record pass/fail from its exit status. +# Usage: check "label" cmd args... +check() { + local label="$1"; shift + info "▶ $label" + if "$@"; then pass "$label"; else fail "$label"; fi +} + +# Assert a condition (already-evaluated boolean via exit status of a command). +# Usage: assert "label" '[ "$x" -gt 0 ]' — pass the test as a string to eval. +assert() { + local label="$1"; local expr="$2" + if eval "$expr"; then pass "$label"; else fail "$label ($expr)"; fi +} + +finish() { + section "summary" + log "passed: $_PASS failed: $_FAIL artifacts: $RAMQP_OUT" + if ((_FAIL > 0)); then + for l in "${_FAILED_LABELS[@]}"; do err " - $l"; done + return 1 + fi + return 0 +} + +# --- process lifecycle ----------------------------------------------------- + +declare -a _SPAWNED_PIDS=() + +_cleanup() { + local pid + for pid in "${_SPAWNED_PIDS[@]:-}"; do + [[ -n "$pid" ]] || continue + kill "$pid" 2>/dev/null || true + done + # Give them a moment, then hard-kill any stragglers. + sleep 0.3 2>/dev/null || true + for pid in "${_SPAWNED_PIDS[@]:-}"; do + [[ -n "$pid" ]] || continue + kill -9 "$pid" 2>/dev/null || true + done +} +trap _cleanup EXIT INT TERM + +# Track a pid for cleanup. +track_pid() { _SPAWNED_PIDS+=("$1"); } + +# Forget a pid we deliberately killed (so cleanup doesn't warn on a reused pid). +untrack_pid() { + local target="$1" i + for i in "${!_SPAWNED_PIDS[@]}"; do + [[ "${_SPAWNED_PIDS[$i]}" == "$target" ]] && unset '_SPAWNED_PIDS[$i]' + done + return 0 # never let a non-match's `&&` short-circuit escape as our status +} + +# --- ports ----------------------------------------------------------------- + +# Print a currently-free localhost TCP port. Uses python for an atomic bind. +free_port() { + python3 - <<'PY' +import socket +s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) +s.bind(("127.0.0.1", 0)) +print(s.getsockname()[1]) +s.close() +PY +} + +# Wait until a TCP port accepts connections (or timeout). Usage: wait_port PORT [SECS] +wait_port() { + local port="$1" secs="${2:-30}" i + for ((i = 0; i < secs * 10; i++)); do + if (exec 3<>"/dev/tcp/127.0.0.1/$port") 2>/dev/null; then exec 3>&- 3<&-; return 0; fi + sleep 0.1 + done + return 1 +} + +# Wait until a TCP port STOPS accepting (a process we killed is really gone). +wait_port_down() { + local port="$1" secs="${2:-15}" i + for ((i = 0; i < secs * 10; i++)); do + if ! (exec 3<>"/dev/tcp/127.0.0.1/$port") 2>/dev/null; then return 0; fi + exec 3>&- 3<&- 2>/dev/null || true + sleep 0.1 + done + return 1 +} + +# --- brokerd --------------------------------------------------------------- + +# Build ramqp-brokerd once. Args: optional feature list (e.g. "store-redb"). +# Echoes the binary path. Caches per feature set within a run. +BROKERD="" +build_brokerd() { + local features="${1:-}" + local flag=() tag="plain" + if [[ -n "$features" ]]; then flag=(--features "$features"); tag="$features"; fi + info "building ramqp-brokerd (release, features: ${features:-none})" >&2 + cargo build --release -q -p ramqp-broker --bin ramqp-brokerd "${flag[@]}" >&2 + BROKERD="$ROOT/target/release/ramqp-brokerd" + [[ -x "$BROKERD" ]] || { err "brokerd not built at $BROKERD"; return 1; } + echo "$BROKERD" +} + +# Spawn a brokerd, tracking its pid and logging to $RAMQP_OUT/.log. +# Usage: spawn_brokerd NAME -- (env is inherited) +# Sets the global BROKERD_PID (do NOT call in $(...) — a subshell would lose the +# tracked pid so the cleanup trap could never reap it). +BROKERD_PID="" +spawn_brokerd() { + local name="$1"; shift + [[ "$1" == "--" ]] && shift + local logf="$RAMQP_OUT/$name.log" + # Append (not truncate) so a node restarted mid-stage keeps its earlier log — + # panic evidence from before a restart must not be lost. Names are per-stage, + # and each run gets a fresh RAMQP_OUT, so there is no cross-run bleed. + "$BROKERD" "$@" >>"$logf" 2>&1 & + BROKERD_PID=$! + track_pid "$BROKERD_PID" +} + +# Grep a spawned broker's log for panics/aborts — a broker that logged one is +# unhealthy even if the stage's functional assertions passed. Usage: no_panics NAME... +no_panics() { + local name rc=0 + for name in "$@"; do + local logf="$RAMQP_OUT/$name.log" + [[ -f "$logf" ]] || continue + if grep -qiE 'panicked|thread .* panicked|RUST_BACKTRACE|assertion failed|fatal runtime' "$logf"; then + err "$name.log contains a panic/abort:" + grep -iE 'panicked|assertion failed|fatal runtime' "$logf" | head -5 >&2 + rc=1 + fi + done + return $rc +} + +# --- sampling -------------------------------------------------------------- + +# VmRSS of a pid in KiB (empty if the process is gone). +rss_kib() { + local pid="$1" + awk '/^VmRSS:/{print $2}' "/proc/$pid/status" 2>/dev/null || true +} + +# --- misc ------------------------------------------------------------------ + +require_cmd() { + command -v "$1" >/dev/null 2>&1 || { err "required command not found: $1"; return 1; } +} + +# Is a docker container running by name? +container_up() { + docker ps --format '{{.Names}}' 2>/dev/null | grep -qx "$1" +} + +cd "$ROOT" diff --git a/ramqp-broker/scripts/run-all.sh b/ramqp-broker/scripts/run-all.sh new file mode 100755 index 0000000..458f521 --- /dev/null +++ b/ramqp-broker/scripts/run-all.sh @@ -0,0 +1,62 @@ +#!/usr/bin/env bash +# Orchestrator: run the ramqp-broker test battery, one stage per script, +# sharing a single timestamped output directory. NOT wired into CI — this is +# the "run the whole thing before a release" button. +# +# scripts/run-all.sh # default battery (gates suite interop robust chaos soak fuzz) +# scripts/run-all.sh --quick # gates + suite only (fast gate) +# scripts/run-all.sh gates suite # an explicit subset, in order +# +# Perf (bench.sh) and coverage (cov.sh) are deliberately NOT part of this — +# they are manual, and bench must never gate anything. +# +# Duration knobs (env): RAMQP_SOAK_SECS, RAMQP_FUZZ_SECS, RAMQP_CHAOS_ROUNDS. + +set -euo pipefail +HERE="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" + +# Shared output dir for every stage in this run. +export RAMQP_OUT="${RAMQP_OUT:-$HERE/out/$(date +%Y%m%d-%H%M%S)}" +mkdir -p "$RAMQP_OUT" + +DEFAULT_STAGES=(gates suite interop robust chaos soak fuzz) + +case "${1:-}" in + --quick) STAGES=(gates suite) ;; + --full) STAGES=("${DEFAULT_STAGES[@]}") ;; + "") STAGES=("${DEFAULT_STAGES[@]}") ;; + *) STAGES=("$@") ;; +esac + +declare -A SCRIPT=( + [gates]=00-gates.sh + [suite]=10-suite.sh + [soak]=20-soak.sh + [chaos]=30-chaos.sh + [interop]=40-interop.sh + [robust]=50-robust.sh + [fuzz]=60-fuzz.sh +) + +RESULTS="$RAMQP_OUT/run-all.summary" +: >"$RESULTS" +rc=0 +for stage in "${STAGES[@]}"; do + script="${SCRIPT[$stage]:-}" + if [[ -z "$script" ]]; then + echo "unknown stage: $stage (known: ${!SCRIPT[*]})" >&2 + exit 2 + fi + printf '\n\033[1m########## stage: %s ##########\033[0m\n' "$stage" + if bash "$HERE/$script"; then + echo "PASS $stage" >>"$RESULTS" + else + echo "FAIL $stage" >>"$RESULTS" + rc=1 + fi +done + +printf '\n\033[1m########## run-all summary ##########\033[0m\n' +cat "$RESULTS" +echo "artifacts: $RAMQP_OUT" +exit $rc diff --git a/ramqp-core/fuzz/Cargo.lock b/ramqp-core/fuzz/Cargo.lock new file mode 100644 index 0000000..15e9df0 --- /dev/null +++ b/ramqp-core/fuzz/Cargo.lock @@ -0,0 +1,626 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 4 + +[[package]] +name = "arbitrary" +version = "1.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3d036a3c4ab069c7b410a2ce876bd74808d2d0888a82667669f8e783a898bf1" + +[[package]] +name = "autocfg" +version = "1.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53" + +[[package]] +name = "bumpalo" +version = "3.20.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649" + +[[package]] +name = "bytes" +version = "1.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" + +[[package]] +name = "cc" +version = "1.2.66" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f5d6cac793997bd970000024b2934968efe83b382de4fdcf4fcb46b6ee4ad996" +dependencies = [ + "find-msvc-tools", + "jobserver", + "libc", + "shlex", +] + +[[package]] +name = "cfg-if" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" + +[[package]] +name = "displaydoc" +version = "0.2.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ac70aa55017e108007fbaf5aa0f54b021c98f92ff8af59d42eda9da96e3dd4f" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "find-msvc-tools" +version = "0.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582" + +[[package]] +name = "form_urlencoded" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb4cb245038516f5f85277875cdaa4f7d2c9a0fa0468de06ed190163b1581fcf" +dependencies = [ + "percent-encoding", +] + +[[package]] +name = "futures-core" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7e3450815272ef58cec6d564423f6e755e25379b217b0bc688e295ba24df6b1d" + +[[package]] +name = "futures-task" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "037711b3d59c33004d3856fbdc83b99d4ff37a24768fa1be9ce3538a1cde4393" + +[[package]] +name = "futures-util" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6" +dependencies = [ + "futures-core", + "futures-task", + "pin-project-lite", + "slab", +] + +[[package]] +name = "getrandom" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099" +dependencies = [ + "cfg-if", + "libc", + "r-efi", +] + +[[package]] +name = "icu_collections" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2984d1cd16c883d7935b9e07e44071dca8d917fd52ecc02c04d5fa0b5a3f191c" +dependencies = [ + "displaydoc", + "potential_utf", + "utf8_iter", + "yoke", + "zerofrom", + "zerovec", +] + +[[package]] +name = "icu_locale_core" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92219b62b3e2b4d88ac5119f8904c10f8f61bf7e95b640d25ba3075e6cac2c29" +dependencies = [ + "displaydoc", + "litemap", + "tinystr", + "writeable", + "zerovec", +] + +[[package]] +name = "icu_normalizer" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c56e5ee99d6e3d33bd91c5d85458b6005a22140021cc324cea84dd0e72cff3b4" +dependencies = [ + "icu_collections", + "icu_normalizer_data", + "icu_properties", + "icu_provider", + "smallvec", + "zerovec", +] + +[[package]] +name = "icu_normalizer_data" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "da3be0ae77ea334f4da67c12f149704f19f81d1adf7c51cf482943e84a2bad38" + +[[package]] +name = "icu_properties" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bee3b67d0ea5c2cca5003417989af8996f8604e34fb9ddf96208a033901e70de" +dependencies = [ + "icu_collections", + "icu_locale_core", + "icu_properties_data", + "icu_provider", + "zerotrie", + "zerovec", +] + +[[package]] +name = "icu_properties_data" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8e2bbb201e0c04f7b4b3e14382af113e17ba4f63e2c9d2ee626b720cbce54a14" + +[[package]] +name = "icu_provider" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "139c4cf31c8b5f33d7e199446eff9c1e02decfc2f0eec2c8d71f65befa45b421" +dependencies = [ + "displaydoc", + "icu_locale_core", + "writeable", + "yoke", + "zerofrom", + "zerotrie", + "zerovec", +] + +[[package]] +name = "idna" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3b0875f23caa03898994f6ddc501886a45c7d3d62d04d2d90788d47be1b1e4de" +dependencies = [ + "idna_adapter", + "smallvec", + "utf8_iter", +] + +[[package]] +name = "idna_adapter" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb68373c0d6620ef8105e855e7745e18b0d00d3bdb07fb532e434244cdb9a714" +dependencies = [ + "icu_normalizer", + "icu_properties", +] + +[[package]] +name = "jobserver" +version = "0.1.35" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1c00acbd29eabad4a2392fa0e921c874934dbbf4194312ad20f04a0ed67a3cb3" +dependencies = [ + "getrandom", + "libc", +] + +[[package]] +name = "js-sys" +version = "0.3.103" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53b44bfcdb3f8d5837a46dae1ca9660a837176eee74a28b229bc626816589102" +dependencies = [ + "cfg-if", + "futures-util", + "wasm-bindgen", +] + +[[package]] +name = "libc" +version = "0.2.186" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "68ab91017fe16c622486840e4c83c9a37afeff978bd239b5293d61ece587de66" + +[[package]] +name = "libfuzzer-sys" +version = "0.4.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a9fd2f41a1cba099f79a0b6b6c35656cf7c03351a7bae8ff0f28f25270f929d2" +dependencies = [ + "arbitrary", + "cc", +] + +[[package]] +name = "litemap" +version = "0.8.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92daf443525c4cce67b150400bc2316076100ce0b3686209eb8cf3c31612e6f0" + +[[package]] +name = "num-traits" +version = "0.2.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "071dfc062690e90b734c0b2273ce72ad0ffa95f0c74596bc250dcfd960262841" +dependencies = [ + "autocfg", +] + +[[package]] +name = "once_cell" +version = "1.21.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" + +[[package]] +name = "ordered-float" +version = "5.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b7d950ca161dc355eaf28f82b11345ed76c6e1f6eb1f4f4479e0323b9e2fbd0e" +dependencies = [ + "num-traits", +] + +[[package]] +name = "percent-encoding" +version = "2.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220" + +[[package]] +name = "pin-project-lite" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" + +[[package]] +name = "potential_utf" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0103b1cef7ec0cf76490e969665504990193874ea05c85ff9bab8b911d0a0564" +dependencies = [ + "zerovec", +] + +[[package]] +name = "proc-macro2" +version = "1.0.106" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8fd00f0bb2e90d81d1044c2b32617f68fcb9fa3bb7640c23e9c748e53fb30934" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "quote" +version = "1.0.46" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dfbc457d0c7a0759a614551b11a6409e5951f6c7537be1f1b7682b9ae9230368" +dependencies = [ + "proc-macro2", +] + +[[package]] +name = "r-efi" +version = "6.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" + +[[package]] +name = "ramqp-core" +version = "0.2.4" +dependencies = [ + "bytes", + "ordered-float", + "thiserror", + "tokio", + "url", + "uuid", +] + +[[package]] +name = "ramqp-core-fuzz" +version = "0.0.0" +dependencies = [ + "bytes", + "libfuzzer-sys", + "ramqp-core", +] + +[[package]] +name = "rustversion" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf54715a573b99ac80df0bc206da022bcd442c974952c7b9720069370852e21f" + +[[package]] +name = "serde" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a8e94ea7f378bd32cbbd37198a4a91436180c5bb472411e48b5ec2e2124ae9e" +dependencies = [ + "serde_core", +] + +[[package]] +name = "serde_core" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "41d385c7d4ca58e59fc732af25c3983b67ac852c1a25000afe1175de458b67ad" +dependencies = [ + "serde_derive", +] + +[[package]] +name = "serde_derive" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d540f220d3187173da220f885ab66608367b6574e925011a9353e4badda91d79" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "shlex" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" + +[[package]] +name = "slab" +version = "0.4.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" + +[[package]] +name = "smallvec" +version = "1.15.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ed6a63f02c8539c91a8685a86f4099661ba3da017932f6ebbea6de3f0fa7c90" + +[[package]] +name = "stable_deref_trait" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596" + +[[package]] +name = "syn" +version = "2.0.118" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1b9ae57f904213ebb649ce6895b8a66c66f0203b9319718f69a5612a065b1422" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "synstructure" +version = "0.13.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "728a70f3dbaf5bab7f0c4b1ac8d7ae5ea60a4b5549c8a5914361c99147a709d2" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "thiserror" +version = "2.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4288b5bcbc7920c07a1149a35cf9590a2aa808e0bc1eafaade0b80947865fbc4" +dependencies = [ + "thiserror-impl", +] + +[[package]] +name = "thiserror-impl" +version = "2.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ebc4ee7f67670e9b64d05fa4253e753e016c6c95ff35b89b7941d6b856dec1d5" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "tinystr" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c8323304221c2a851516f22236c5722a72eaa19749016521d6dff0824447d96d" +dependencies = [ + "displaydoc", + "zerovec", +] + +[[package]] +name = "tokio" +version = "1.52.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8fc7f01b389ac15039e4dc9531aa973a135d7a4135281b12d7c1bc79fd57fffe" +dependencies = [ + "bytes", + "pin-project-lite", +] + +[[package]] +name = "unicode-ident" +version = "1.0.24" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" + +[[package]] +name = "url" +version = "2.5.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff67a8a4397373c3ef660812acab3268222035010ab8680ec4215f38ba3d0eed" +dependencies = [ + "form_urlencoded", + "idna", + "percent-encoding", + "serde", +] + +[[package]] +name = "utf8_iter" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6c140620e7ffbb22c2dee59cafe6084a59b5ffc27a8859a5f0d494b5d52b6be" + +[[package]] +name = "uuid" +version = "1.23.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bf80a72845275afea99e7f2b434723d3bc7e38470fcd1c7ed39a599c73319a53" +dependencies = [ + "getrandom", + "js-sys", + "wasm-bindgen", +] + +[[package]] +name = "wasm-bindgen" +version = "0.2.126" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4b067c0c11094aef6b7a801c1e34a26affafdf3d051dba08456b868789aaf9a4" +dependencies = [ + "cfg-if", + "once_cell", + "rustversion", + "wasm-bindgen-macro", + "wasm-bindgen-shared", +] + +[[package]] +name = "wasm-bindgen-macro" +version = "0.2.126" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "167ce5e579f6bcf889c4f7175a8a5a585de84e8ff93976ce393efa5f2837aab1" +dependencies = [ + "quote", + "wasm-bindgen-macro-support", +] + +[[package]] +name = "wasm-bindgen-macro-support" +version = "0.2.126" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f3997c7839262f4ef12cf90b818d6340c18e80f263f1a94bf157d0ec4420380e" +dependencies = [ + "bumpalo", + "proc-macro2", + "quote", + "syn", + "wasm-bindgen-shared", +] + +[[package]] +name = "wasm-bindgen-shared" +version = "0.2.126" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc1b4cb0cc549fcf58d7dfc081778139b3d283a081644e833e84682ad71cea24" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "writeable" +version = "0.6.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ffae5123b2d3fc086436f8834ae3ab053a283cfac8fe0a0b8eaae044768a4c4" + +[[package]] +name = "yoke" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "709fe23a0424b6a435d82152b1bd3fdfb0833487d5fa90d05d42762a9891fef5" +dependencies = [ + "stable_deref_trait", + "yoke-derive", + "zerofrom", +] + +[[package]] +name = "yoke-derive" +version = "0.8.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "de844c262c8848816172cef550288e7dc6c7b7814b4ee56b3e1553f275f1858e" +dependencies = [ + "proc-macro2", + "quote", + "syn", + "synstructure", +] + +[[package]] +name = "zerofrom" +version = "0.1.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ec05a11813ea801ff6d75110ad09cd0824ddba17dfe17128ea0d5f68e6c5272" +dependencies = [ + "zerofrom-derive", +] + +[[package]] +name = "zerofrom-derive" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "11532158c46691caf0f2593ea8358fed6bbf68a0315e80aae9bd41fbade684a1" +dependencies = [ + "proc-macro2", + "quote", + "syn", + "synstructure", +] + +[[package]] +name = "zerotrie" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0f9152d31db0792fa83f70fb2f83148effb5c1f5b8c7686c3459e361d9bc20bf" +dependencies = [ + "displaydoc", + "yoke", + "zerofrom", +] + +[[package]] +name = "zerovec" +version = "0.11.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "90f911cbc359ab6af17377d242225f4d75119aec87ea711a880987b18cd7b239" +dependencies = [ + "yoke", + "zerofrom", + "zerovec-derive", +] + +[[package]] +name = "zerovec-derive" +version = "0.11.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "625dc425cab0dca6dc3c3319506e6593dcb08a9f387ea3b284dbd52a92c40555" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] diff --git a/ramqp-core/fuzz/Cargo.toml b/ramqp-core/fuzz/Cargo.toml new file mode 100644 index 0000000..71439cd --- /dev/null +++ b/ramqp-core/fuzz/Cargo.toml @@ -0,0 +1,37 @@ +# Fuzz targets for ramqp-core's untrusted-wire decoders. Its own `[workspace]` +# table decouples it from the parent workspace, so a normal `cargo build` / +# `cargo check --all-targets` at the repo root never pulls in libfuzzer or the +# nightly sanitizer toolchain. Run via `scripts/60-fuzz.sh` (needs nightly + +# cargo-fuzz). +[package] +name = "ramqp-core-fuzz" +version = "0.0.0" +publish = false +edition = "2021" + +[package.metadata] +cargo-fuzz = true + +[workspace] + +[dependencies] +libfuzzer-sys = "0.4" +bytes = "1" +ramqp-core = { path = ".." } + +[profile.release] +debug = 1 + +[[bin]] +name = "decode_frame" +path = "fuzz_targets/decode_frame.rs" +test = false +doc = false +bench = false + +[[bin]] +name = "value" +path = "fuzz_targets/value.rs" +test = false +doc = false +bench = false diff --git a/ramqp-core/fuzz/fuzz_targets/decode_frame.rs b/ramqp-core/fuzz/fuzz_targets/decode_frame.rs new file mode 100644 index 0000000..16ab4d8 --- /dev/null +++ b/ramqp-core/fuzz/fuzz_targets/decode_frame.rs @@ -0,0 +1,26 @@ +#![no_main] +//! Fuzz the frame decoder — the very first thing the broker runs on every byte +//! an untrusted client sends. It must never panic, hang, or over-allocate on +//! arbitrary input: it either decodes a frame, asks for more bytes, or returns +//! a protocol error. We drain repeatedly so multi-frame and trailing-garbage +//! inputs are exercised too. + +use bytes::BytesMut; +use libfuzzer_sys::fuzz_target; +use ramqp_core::transport::frame::decode_frame; + +fuzz_target!(|data: &[u8]| { + let mut buf = BytesMut::from(data); + // A 1 MiB cap mirrors a generous negotiated max-frame-size. Loop until the + // decoder stops making progress (needs-more, error, or empty). + loop { + match decode_frame(&mut buf, 1 << 20) { + Ok(Some(_frame)) => { + if buf.is_empty() { + break; + } + } + Ok(None) | Err(_) => break, + } + } +}); diff --git a/ramqp-core/fuzz/fuzz_targets/value.rs b/ramqp-core/fuzz/fuzz_targets/value.rs new file mode 100644 index 0000000..d3bca67 --- /dev/null +++ b/ramqp-core/fuzz/fuzz_targets/value.rs @@ -0,0 +1,12 @@ +#![no_main] +//! Fuzz the AMQP type-system value decoder. Every performative, message +//! section, and delivery annotation the broker parses bottoms out in this +//! decoder, so an arbitrary-bytes-to-`Value` decode must never panic or hang — +//! only decode or error. + +use libfuzzer_sys::fuzz_target; +use ramqp_core::codec::{Value, from_slice}; + +fuzz_target!(|data: &[u8]| { + let _ = from_slice::(data); +});