fix(sdk): inline iobuf into iggy_binary_protocol

.github/actions/rust/pre-merge/action.yml

@@ -91,7 +91,7 @@ runs:
             # Build -p flags for cargo build/test
             echo "$CRATES" | sed 's/^/-p /' | tr '\n' ' ' > /tmp/packages.txt
             # Build nextest filter expression for cargo nextest run
-            echo "$CRATES" | sed 's/^/package(/; s/$/)/' | paste -sd ' | ' > /tmp/nextest-filter.txt
+            echo "$CRATES" | sed 's/^/package(/; s/$/)/' | paste -sd '|' | sed 's/|/ | /g' > /tmp/nextest-filter.txt
             echo "::notice::DAG analysis: ${CRATE_COUNT} affected crates (of ${TOTAL_CRATES} total)"
           else
             echo "::notice::Full workspace affected (${TOTAL_CRATES} crates)"

.github/config/components.yml

@@ -51,17 +51,9 @@ components:
     paths:
       - "core/common/**"
 
-  # Leaf crate: zero-copy I/O buffer, depended on by binary_protocol and cluster
-  rust-iobuf:
-    depends_on:
-      - "rust-workspace"
-    paths:
-      - "core/iobuf/**"
-
   rust-binary-protocol:
     depends_on:
       - "rust-workspace" # Protocol is affected by workspace changes
-      - "rust-iobuf" # binary_protocol depends on iobuf
       - "ci-infrastructure" # CI changes trigger full regression
     paths:
       - "core/binary_protocol/**"
@@ -80,7 +72,6 @@ components:
   rust-cluster:
     depends_on:
       - "rust-workspace"
-      - "rust-iobuf" # cluster crates depend on iobuf
       - "rust-binary-protocol" # cluster crates depend on binary_protocol
     paths:
       - "core/clock/**"
@@ -104,7 +95,6 @@ components:
   rust:
     depends_on:
       - "rust-workspace"
-      - "rust-iobuf"
       - "rust-configs"
       - "rust-sdk"
       - "rust-common"

Cargo.toml

@@ -48,7 +48,6 @@ members = [
     "core/consensus",
     "core/harness_derive",
     "core/integration",
-    "core/iobuf",
     "core/journal",
     "core/message_bus",
     "core/metadata",
@@ -116,6 +115,7 @@ compio = { version = "0.18.0", features = [
     "ws",
     "fs",
 ] }
+compio-buf = "0.8.1"
 # Pin compio-driver >= 0.11.2 to fix musl compilation (compio-rs/compio#668)
 compio-driver = "0.11.4"
 configs = { path = "core/configs", version = "0.1.0" }
@@ -172,7 +172,6 @@ iggy_binary_protocol = { path = "core/binary_protocol", version = "0.10.0" }
 iggy_common = { path = "core/common", version = "0.10.0" }
 iggy_connector_sdk = { path = "core/connectors/sdk", version = "0.3.0" }
 integration = { path = "core/integration" }
-iobuf = { path = "core/iobuf" }
 journal = { path = "core/journal" }
 js-sys = "0.3"
 jsonwebtoken = { version = "10.3.0", features = ["rust_crypto"] }

DEPENDENCIES.md

@@ -515,7 +515,6 @@ interpolate_name: 0.2.4, "MIT",
 inventory: 0.3.24, "Apache-2.0 OR MIT",
 io-uring: 0.7.11, "Apache-2.0 OR MIT",
 io_uring_buf_ring: 0.2.3, "MIT",
-iobuf: 0.1.0, "Apache-2.0",
 ipconfig: 0.3.4, "Apache-2.0 OR MIT",
 ipnet: 2.12.0, "Apache-2.0 OR MIT",
 iri-string: 0.7.12, "Apache-2.0 OR MIT",

core/binary_protocol/Cargo.toml

@@ -29,10 +29,11 @@ repository = "https://github.com/apache/iggy"
 readme = "../../README.md"
 
 [dependencies]
+aligned-vec = { workspace = true }
 bytemuck = { workspace = true }
 bytes = { workspace = true }
+compio-buf = { workspace = true }
 enumset = { workspace = true }
-iobuf = { workspace = true }
 smallvec = { workspace = true }
 thiserror = { workspace = true }
 

core/binary_protocol/src/consensus/message.rs

@@ -15,12 +15,12 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use super::iobuf::{Frozen, Owned};
 use crate::consensus::{
     self, Command2, CommitHeader, ConsensusError, ConsensusHeader, DoViewChangeHeader,
     GenericHeader, PrepareHeader, PrepareOkHeader, RequestHeader, StartViewChangeHeader,
     StartViewHeader,
 };
-use iobuf::{Frozen, Owned};
 use smallvec::SmallVec;
 use std::{marker::PhantomData, mem::size_of};
 
@@ -46,13 +46,19 @@ where
     fn as_mut_slice(&mut self) -> &mut [u8];
 }
 
-pub trait FragmentedBacking<H>: MessageBacking<H> + ResponseBackingKind
+mod sealed {
+    pub trait Sealed {}
+}
+
+pub trait FragmentedBacking<H>: MessageBacking<H> + ResponseBackingKind + sealed::Sealed
 where
     H: ConsensusHeader,
 {
     fn fragments(&self) -> &[Frozen<MESSAGE_ALIGN>];
 }
 
+impl sealed::Sealed for ResponseBacking {}
+
 #[derive(Debug, Clone)]
 pub struct RequestBacking {
     owned: Owned<MESSAGE_ALIGN>,

core/binary_protocol/src/consensus/mod.rs

@@ -42,6 +42,20 @@
 mod command;
 mod error;
 mod header;
+// iobuf was moved here verbatim from the former `core/iobuf` crate. Its
+// implementation is intentionally untouched for rc2; the existing lints
+// predate the move and are tracked as tech debt for a 0.9.x cleanup pass.
+#[allow(
+    private_interfaces,
+    clippy::missing_errors_doc,
+    clippy::missing_panics_doc,
+    clippy::must_use_candidate,
+    clippy::return_self_not_must_use,
+    clippy::missing_const_for_fn,
+    clippy::non_send_fields_in_send_ty,
+    clippy::use_self
+)]
+pub mod iobuf;
 pub mod message;
 mod operation;
 

core/common/Cargo.toml

@@ -48,7 +48,6 @@ err_trail = { workspace = true }
 human-repr = { workspace = true }
 humantime = { workspace = true }
 iggy_binary_protocol = { workspace = true }
-iobuf = { workspace = true }
 lending-iterator = { workspace = true }
 moka = { workspace = true }
 once_cell = { workspace = true }

core/common/src/lib.rs

@@ -112,7 +112,6 @@ pub use types::personal_access_tokens::*;
 pub use types::segment::Segment;
 pub use types::segment_storage::*;
 pub use types::send_messages2;
-pub use types::send_messages2::*;
 pub use types::snapshot::*;
 pub use types::stats::*;
 pub use types::stream::*;

core/common/src/types/send_messages2.rs

@@ -17,8 +17,8 @@
 
 use crate::{INDEX_SIZE, IggyError, random_id, sharding::IggyNamespace};
 use bytes::{Bytes, BytesMut};
+use iggy_binary_protocol::consensus::iobuf::Owned;
 use iggy_binary_protocol::{Message, PrepareHeader, RequestHeader};
-use iobuf::Owned;
 use std::hash::Hasher;
 use twox_hash::XxHash3_64;
 
@@ -544,7 +544,7 @@ impl<'a> Iterator for SendMessages2IteratorWithOffsets<'a> {
     }
 }
 
-pub type FrozenBatchHeader = iobuf::Frozen<MESSAGE_ALIGN>;
+pub(crate) type FrozenBatchHeader = iggy_binary_protocol::consensus::iobuf::Frozen<MESSAGE_ALIGN>;
 
 pub fn convert_request_message(
     namespace: IggyNamespace,

core/consensus/Cargo.toml

@@ -33,7 +33,6 @@ bytemuck = { workspace = true }
 bytes = { workspace = true }
 iggy_binary_protocol = { workspace = true }
 iggy_common = { workspace = true }
-iobuf = { workspace = true }
 message_bus = { workspace = true }
 rand = { workspace = true }
 rand_xoshiro = { workspace = true }

core/consensus/src/plane_helpers.rs

@@ -20,10 +20,10 @@ use crate::{
     Consensus, IgnoreReason, Pipeline, PipelineEntry, PlaneKind, PrepareOkOutcome, Sequencer,
     Status, VsrConsensus,
 };
+use iggy_binary_protocol::consensus::iobuf::Owned;
 use iggy_binary_protocol::{
     Command2, GenericHeader, Message, PrepareHeader, PrepareOkHeader, ReplyHeader,
 };
-use iobuf::Owned;
 use message_bus::MessageBus;
 use std::ops::AsyncFnOnce;
 

core/journal/Cargo.toml

@@ -31,7 +31,6 @@ readme = "../../../README.md"
 bytemuck = { workspace = true }
 compio = { workspace = true }
 iggy_binary_protocol = { workspace = true }
-iobuf = { workspace = true }
 
 [dev-dependencies]
 tempfile = { workspace = true }

core/journal/src/prepare_journal.rs

@@ -18,9 +18,9 @@
 use crate::file_storage::FileStorage;
 use crate::{Journal, JournalHandle};
 use compio::io::AsyncWriteAtExt;
+use iggy_binary_protocol::consensus::iobuf::Owned;
 use iggy_binary_protocol::consensus::message::Message;
 use iggy_binary_protocol::consensus::{Command2, PrepareHeader};
-use iobuf::Owned;
 use std::cell::{Cell, Ref, RefCell};
 use std::fmt;
 use std::io;

core/metadata/Cargo.toml

@@ -46,7 +46,6 @@ tracing = { workspace = true }
 bytemuck = { workspace = true }
 bytes = { workspace = true }
 compio = { workspace = true }
-iobuf = { workspace = true }
 tempfile = { workspace = true }
 
 [lints.clippy]

core/metadata/src/impls/recovery.rs

@@ -158,8 +158,8 @@ where
 #[allow(clippy::cast_possible_truncation)]
 mod tests {
     use super::*;
+    use iggy_binary_protocol::consensus::iobuf::Owned;
     use iggy_binary_protocol::consensus::{Command2, Operation};
-    use iobuf::Owned;
     use journal::Journal;
     use tempfile::tempdir;
 

core/partitions/Cargo.toml

@@ -34,7 +34,6 @@ compio = { workspace = true }
 consensus = { workspace = true }
 iggy_binary_protocol = { workspace = true }
 iggy_common = { workspace = true }
-iobuf = { workspace = true }
 journal = { workspace = true }
 message_bus = { workspace = true }
 ringbuffer = { workspace = true }

core/partitions/src/iggy_partitions.rs

@@ -34,6 +34,7 @@ use consensus::{
     pipeline_prepare_common, replicate_preflight, replicate_to_next_in_chain, request_preflight,
     send_prepare_ok as send_prepare_ok_common,
 };
+use iggy_binary_protocol::consensus::iobuf::Frozen;
 use iggy_binary_protocol::{
     Command2, ConsensusHeader, GenericHeader, Message, Operation, PrepareHeader, PrepareOkHeader,
     RequestHeader,
@@ -43,7 +44,6 @@ use iggy_common::{
     send_messages2::{convert_request_message, decode_prepare_slice},
     sharding::{IggyNamespace, LocalIdx, ShardId},
 };
-use iobuf::Frozen;
 use journal::{Journal, JournalHandle};
 use message_bus::MessageBus;
 use std::cell::UnsafeCell;

core/partitions/src/journal.rs

@@ -15,9 +15,9 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use iggy_binary_protocol::consensus::iobuf::{Frozen, Owned};
 use iggy_binary_protocol::{Operation, PrepareHeader};
 use iggy_common::send_messages2::{COMMAND_HEADER_SIZE, SendMessages2Ref, decode_prepare_slice};
-use iobuf::{Frozen, Owned};
 use journal::{Journal, Storage};
 use std::io;
 use std::{

core/partitions/src/lib.rs

@@ -29,7 +29,7 @@ mod types;
 
 use iggy_binary_protocol::{Message, PrepareHeader};
 use iggy_common::IggyError;
-pub use iggy_common::{IggyMessage2, IggyMessage2Header, IggyMessages2};
+pub use iggy_common::send_messages2::{IggyMessage2, IggyMessage2Header, IggyMessages2};
 pub use iggy_partition::IggyPartition;
 pub use iggy_partitions::IggyPartitions;
 pub use types::{

core/partitions/src/messages_writer.rs

@@ -19,8 +19,8 @@ use compio::{
     fs::{File, OpenOptions},
     io::AsyncWriteAtExt,
 };
+use iggy_binary_protocol::consensus::iobuf::Frozen;
 use iggy_common::{IggyByteSize, IggyError};
-use iobuf::Frozen;
 use std::{
     rc::Rc,
     sync::atomic::{AtomicU64, Ordering},