diff --git a/crates/columns/src/cell_store/context.rs b/crates/columns/src/cell_store/context.rs index 1e4f1c7b7..a5734453b 100644 --- a/crates/columns/src/cell_store/context.rs +++ b/crates/columns/src/cell_store/context.rs @@ -4,7 +4,7 @@ use silver_common::{ ssz_view::{ BYTES_PER_CELL, BYTES_PER_KZG_COMMITMENT, BYTES_PER_KZG_PROOF, DATA_COLUMN_SIDECAR_GLOAS_MIN, DATA_COLUMN_SIDECAR_MIN, DataColumnSidecarFuluView, - DataColumnSidecarGloasView, + DataColumnSidecarGloasView, partial_column::PARTIAL_HEADER_FIXED, }, }; @@ -35,8 +35,9 @@ impl CommitmentContext { rows * BYTES_PER_KZG_COMMITMENT || DataColumnSidecarFuluView::kzg_proofs(bytes).len() != rows * BYTES_PER_KZG_PROOF || - bytes[20..356] != context[4..340] || - DataColumnSidecarFuluView::kzg_commitments(bytes) != &context[340..] + bytes[20..356] != context[4..PARTIAL_HEADER_FIXED] || + DataColumnSidecarFuluView::kzg_commitments(bytes) != + &context[PARTIAL_HEADER_FIXED..] { return None; } @@ -89,17 +90,20 @@ impl<'a> ContextData<'a> { } } + /// A Fulu context is stored as PartialDataColumnHeader SSZ, so it can + /// later be referenced verbatim as a partial frame's header ranges. pub(super) fn encoded_len(self) -> usize { - self.commitments().len() + if matches!(self, Self::Fulu { .. }) { 340 } else { 0 } + self.commitments().len() + + if matches!(self, Self::Fulu { .. }) { PARTIAL_HEADER_FIXED } else { 0 } } pub(super) fn write(self, out: &mut [u8]) { match self { Self::Fulu { signed_header, inclusion_proof, commitments } => { - out[..4].copy_from_slice(&340u32.to_le_bytes()); + out[..4].copy_from_slice(&(PARTIAL_HEADER_FIXED as u32).to_le_bytes()); out[4..212].copy_from_slice(signed_header); - out[212..340].copy_from_slice(inclusion_proof); - out[340..].copy_from_slice(commitments); + out[212..PARTIAL_HEADER_FIXED].copy_from_slice(inclusion_proof); + out[PARTIAL_HEADER_FIXED..].copy_from_slice(commitments); } Self::Gloas { commitments } => out.copy_from_slice(commitments), } @@ -111,10 +115,10 @@ impl<'a> ContextData<'a> { } match self { Self::Fulu { signed_header, inclusion_proof, commitments } => { - bytes[..4] == 340u32.to_le_bytes() && + bytes[..4] == (PARTIAL_HEADER_FIXED as u32).to_le_bytes() && bytes[4..212] == *signed_header && - bytes[212..340] == *inclusion_proof && - bytes[340..] == *commitments + bytes[212..PARTIAL_HEADER_FIXED] == *inclusion_proof && + bytes[PARTIAL_HEADER_FIXED..] == *commitments } Self::Gloas { commitments } => bytes == commitments, } diff --git a/crates/gossip/protobuf/gossipsub.proto b/crates/gossip/protobuf/gossipsub.proto index 4ce024fa3..27e6fa7a1 100644 --- a/crates/gossip/protobuf/gossipsub.proto +++ b/crates/gossip/protobuf/gossipsub.proto @@ -9,9 +9,14 @@ message RPC { message SubOpts { optional bool subscribe = 1; // subscribe or unsubscribe optional string topic_id = 2; + // Partial Messages extension: requesting partials implies + // supports_sending_partial, even when that flag is absent or false. + optional bool requests_partial = 3; + optional bool supports_sending_partial = 4; } optional ControlMessage control = 3; + optional PartialMessagesExtension partial = 10; } message Message { @@ -29,6 +34,20 @@ message ControlMessage { repeated ControlGraft graft = 3; repeated ControlPrune prune = 4; repeated ControlIDontWant idontwant = 5; + optional ControlExtensions extensions = 6; +} + +message ControlExtensions { + optional bool partial_messages = 10; +} + +// One (topic_id, group_id) pair per RPC frame; partial_message and +// parts_metadata are raw SSZ, not Snappy. +message PartialMessagesExtension { + optional bytes topic_id = 1; + optional bytes group_id = 2; + optional bytes partial_message = 3; + optional bytes parts_metadata = 4; } message ControlIHave { diff --git a/crates/gossip/src/generated/protobuf.gossipsub.rs b/crates/gossip/src/generated/protobuf.gossipsub.rs index 07839cb5f..073a8a179 100644 --- a/crates/gossip/src/generated/protobuf.gossipsub.rs +++ b/crates/gossip/src/generated/protobuf.gossipsub.rs @@ -9,6 +9,8 @@ pub struct RPC { pub publish: ::buffa::alloc::vec::Vec, ///Field 3: `control` pub control: ::buffa::MessageField, + ///Field 10: `partial` + pub partial: ::buffa::MessageField, #[doc(hidden)] pub __buffa_unknown_fields: ::buffa::UnknownFields, #[doc(hidden)] @@ -20,6 +22,7 @@ impl ::core::fmt::Debug for RPC { .field("subscriptions", &self.subscriptions) .field("publish", &self.publish) .field("control", &self.control) + .field("partial", &self.partial) .finish() } } @@ -52,6 +55,12 @@ impl ::buffa::Message for RPC { += 1u32 + ::buffa::encoding::varint_len(inner_size as u64) as u32 + inner_size; } + if self.partial.is_set() { + let inner_size = self.partial.compute_size(); + size + += 1u32 + ::buffa::encoding::varint_len(inner_size as u64) as u32 + + inner_size; + } for v in &self.subscriptions { let inner_size = v.compute_size(); size @@ -80,6 +89,15 @@ impl ::buffa::Message for RPC { ::buffa::encoding::encode_varint(self.control.cached_size() as u64, buf); self.control.write_to(buf); } + if self.partial.is_set() { + ::buffa::encoding::Tag::new( + 10u32, + ::buffa::encoding::WireType::LengthDelimited, + ) + .encode(buf); + ::buffa::encoding::encode_varint(self.partial.cached_size() as u64, buf); + self.partial.write_to(buf); + } for v in &self.subscriptions { ::buffa::encoding::Tag::new( 1u32, @@ -125,6 +143,20 @@ impl ::buffa::Message for RPC { depth, )?; } + 10u32 => { + if tag.wire_type() != ::buffa::encoding::WireType::LengthDelimited { + return ::core::result::Result::Err(::buffa::DecodeError::WireTypeMismatch { + field_number: 10u32, + expected: 2u8, + actual: tag.wire_type() as u8, + }); + } + ::buffa::Message::merge_length_delimited( + self.partial.get_or_insert_default(), + buf, + depth, + )?; + } 1u32 => { if tag.wire_type() != ::buffa::encoding::WireType::LengthDelimited { return ::core::result::Result::Err(::buffa::DecodeError::WireTypeMismatch { @@ -161,6 +193,7 @@ impl ::buffa::Message for RPC { } fn clear(&mut self) { self.control = ::buffa::MessageField::none(); + self.partial = ::buffa::MessageField::none(); self.subscriptions.clear(); self.publish.clear(); self.__buffa_unknown_fields.clear(); @@ -175,6 +208,8 @@ pub struct RPCView<'a> { pub publish: ::buffa::RepeatedView<'a, MessageView<'a>>, ///Field 3: `control` pub control: ::buffa::MessageFieldView>, + ///Field 10: `partial` + pub partial: ::buffa::MessageFieldView>, pub __buffa_unknown_fields: ::buffa::UnknownFieldsView<'a>, } impl<'a> RPCView<'a> { @@ -236,6 +271,27 @@ impl<'a> RPCView<'a> { } } } + 10u32 => { + if tag.wire_type() != ::buffa::encoding::WireType::LengthDelimited { + return ::core::result::Result::Err(::buffa::DecodeError::WireTypeMismatch { + field_number: 10u32, + expected: 2u8, + actual: tag.wire_type() as u8, + }); + } + if depth == 0 { + return Err(::buffa::DecodeError::RecursionLimitExceeded); + } + let sub = ::buffa::types::borrow_bytes(&mut cur)?; + match view.partial.as_mut() { + Some(existing) => existing._merge_into_view(sub, depth - 1)?, + None => { + view.partial = ::buffa::MessageFieldView::set( + PartialMessagesExtensionView::_decode_depth(sub, depth - 1)?, + ); + } + } + } 1u32 => { if tag.wire_type() != ::buffa::encoding::WireType::LengthDelimited { return ::core::result::Result::Err(::buffa::DecodeError::WireTypeMismatch { @@ -304,6 +360,14 @@ impl<'a> ::buffa::MessageView<'a> for RPCView<'a> { } None => ::buffa::MessageField::none(), }, + partial: match self.partial.as_option() { + Some(v) => { + ::buffa::MessageField::< + PartialMessagesExtension, + >::some(v.to_owned_message()) + } + None => ::buffa::MessageField::none(), + }, __buffa_unknown_fields: self .__buffa_unknown_fields .to_owned() @@ -330,6 +394,10 @@ pub mod rpc { pub subscribe: Option, ///Field 2: `topic_id` pub topic_id: Option<::buffa::alloc::string::String>, + ///Field 3: `requests_partial` + pub requests_partial: Option, + ///Field 4: `supports_sending_partial` + pub supports_sending_partial: Option, #[doc(hidden)] pub __buffa_unknown_fields: ::buffa::UnknownFields, #[doc(hidden)] @@ -340,6 +408,8 @@ pub mod rpc { f.debug_struct("SubOpts") .field("subscribe", &self.subscribe) .field("topic_id", &self.topic_id) + .field("requests_partial", &self.requests_partial) + .field("supports_sending_partial", &self.supports_sending_partial) .finish() } } @@ -372,6 +442,12 @@ pub mod rpc { if let Some(ref v) = self.topic_id { size += 1u32 + ::buffa::types::string_encoded_len(v) as u32; } + if self.requests_partial.is_some() { + size += 1u32 + ::buffa::types::BOOL_ENCODED_LEN as u32; + } + if self.supports_sending_partial.is_some() { + size += 1u32 + ::buffa::types::BOOL_ENCODED_LEN as u32; + } size += self.__buffa_unknown_fields.encoded_len() as u32; self.__buffa_cached_size.set(size); size @@ -392,6 +468,16 @@ pub mod rpc { .encode(buf); ::buffa::types::encode_string(v, buf); } + if let Some(v) = self.requests_partial { + ::buffa::encoding::Tag::new(3u32, ::buffa::encoding::WireType::Varint) + .encode(buf); + ::buffa::types::encode_bool(v, buf); + } + if let Some(v) = self.supports_sending_partial { + ::buffa::encoding::Tag::new(4u32, ::buffa::encoding::WireType::Varint) + .encode(buf); + ::buffa::types::encode_bool(v, buf); + } self.__buffa_unknown_fields.write_to(buf); } fn merge_field( @@ -432,6 +518,30 @@ pub mod rpc { buf, )?; } + 3u32 => { + if tag.wire_type() != ::buffa::encoding::WireType::Varint { + return ::core::result::Result::Err(::buffa::DecodeError::WireTypeMismatch { + field_number: 3u32, + expected: 0u8, + actual: tag.wire_type() as u8, + }); + } + self.requests_partial = ::core::option::Option::Some( + ::buffa::types::decode_bool(buf)?, + ); + } + 4u32 => { + if tag.wire_type() != ::buffa::encoding::WireType::Varint { + return ::core::result::Result::Err(::buffa::DecodeError::WireTypeMismatch { + field_number: 4u32, + expected: 0u8, + actual: tag.wire_type() as u8, + }); + } + self.supports_sending_partial = ::core::option::Option::Some( + ::buffa::types::decode_bool(buf)?, + ); + } _ => { self.__buffa_unknown_fields .push(::buffa::encoding::decode_unknown_field(tag, buf, depth)?); @@ -445,6 +555,8 @@ pub mod rpc { fn clear(&mut self) { self.subscribe = ::core::option::Option::None; self.topic_id = ::core::option::Option::None; + self.requests_partial = ::core::option::Option::None; + self.supports_sending_partial = ::core::option::Option::None; self.__buffa_unknown_fields.clear(); self.__buffa_cached_size.set(0); } @@ -455,6 +567,10 @@ pub mod rpc { pub subscribe: ::core::option::Option, ///Field 2: `topic_id` pub topic_id: ::core::option::Option<&'a str>, + ///Field 3: `requests_partial` + pub requests_partial: ::core::option::Option, + ///Field 4: `supports_sending_partial` + pub supports_sending_partial: ::core::option::Option, pub __buffa_unknown_fields: ::buffa::UnknownFieldsView<'a>, } impl<'a> SubOptsView<'a> { @@ -517,6 +633,30 @@ pub mod rpc { } view.topic_id = Some(::buffa::types::borrow_str(&mut cur)?); } + 3u32 => { + if tag.wire_type() != ::buffa::encoding::WireType::Varint { + return ::core::result::Result::Err(::buffa::DecodeError::WireTypeMismatch { + field_number: 3u32, + expected: 0u8, + actual: tag.wire_type() as u8, + }); + } + view.requests_partial = Some( + ::buffa::types::decode_bool(&mut cur)?, + ); + } + 4u32 => { + if tag.wire_type() != ::buffa::encoding::WireType::Varint { + return ::core::result::Result::Err(::buffa::DecodeError::WireTypeMismatch { + field_number: 4u32, + expected: 0u8, + actual: tag.wire_type() as u8, + }); + } + view.supports_sending_partial = Some( + ::buffa::types::decode_bool(&mut cur)?, + ); + } _ => { ::buffa::encoding::skip_field_depth(tag, &mut cur, depth)?; let span_len = before_tag.len() - cur.len(); @@ -548,6 +688,8 @@ pub mod rpc { SubOpts { subscribe: self.subscribe, topic_id: self.topic_id.map(|s| s.to_string()), + requests_partial: self.requests_partial, + supports_sending_partial: self.supports_sending_partial, __buffa_unknown_fields: self .__buffa_unknown_fields .to_owned() @@ -971,6 +1113,8 @@ pub struct ControlMessage { pub prune: ::buffa::alloc::vec::Vec, ///Field 5: `idontwant` pub idontwant: ::buffa::alloc::vec::Vec, + ///Field 6: `extensions` + pub extensions: ::buffa::MessageField, #[doc(hidden)] pub __buffa_unknown_fields: ::buffa::UnknownFields, #[doc(hidden)] @@ -984,6 +1128,7 @@ impl ::core::fmt::Debug for ControlMessage { .field("graft", &self.graft) .field("prune", &self.prune) .field("idontwant", &self.idontwant) + .field("extensions", &self.extensions) .finish() } } @@ -1010,6 +1155,12 @@ impl ::buffa::Message for ControlMessage { #[allow(unused_imports)] use ::buffa::Enumeration as _; let mut size = 0u32; + if self.extensions.is_set() { + let inner_size = self.extensions.compute_size(); + size + += 1u32 + ::buffa::encoding::varint_len(inner_size as u64) as u32 + + inner_size; + } for v in &self.ihave { let inner_size = v.compute_size(); size @@ -1047,6 +1198,15 @@ impl ::buffa::Message for ControlMessage { fn write_to(&self, buf: &mut impl ::buffa::bytes::BufMut) { #[allow(unused_imports)] use ::buffa::Enumeration as _; + if self.extensions.is_set() { + ::buffa::encoding::Tag::new( + 6u32, + ::buffa::encoding::WireType::LengthDelimited, + ) + .encode(buf); + ::buffa::encoding::encode_varint(self.extensions.cached_size() as u64, buf); + self.extensions.write_to(buf); + } for v in &self.ihave { ::buffa::encoding::Tag::new( 1u32, @@ -1105,6 +1265,20 @@ impl ::buffa::Message for ControlMessage { #[allow(unused_imports)] use ::buffa::Enumeration as _; match tag.field_number() { + 6u32 => { + if tag.wire_type() != ::buffa::encoding::WireType::LengthDelimited { + return ::core::result::Result::Err(::buffa::DecodeError::WireTypeMismatch { + field_number: 6u32, + expected: 2u8, + actual: tag.wire_type() as u8, + }); + } + ::buffa::Message::merge_length_delimited( + self.extensions.get_or_insert_default(), + buf, + depth, + )?; + } 1u32 => { if tag.wire_type() != ::buffa::encoding::WireType::LengthDelimited { return ::core::result::Result::Err(::buffa::DecodeError::WireTypeMismatch { @@ -1176,6 +1350,7 @@ impl ::buffa::Message for ControlMessage { self.__buffa_cached_size.get() } fn clear(&mut self) { + self.extensions = ::buffa::MessageField::none(); self.ihave.clear(); self.iwant.clear(); self.graft.clear(); @@ -1197,6 +1372,8 @@ pub struct ControlMessageView<'a> { pub prune: ::buffa::RepeatedView<'a, ControlPruneView<'a>>, ///Field 5: `idontwant` pub idontwant: ::buffa::RepeatedView<'a, ControlIDontWantView<'a>>, + ///Field 6: `extensions` + pub extensions: ::buffa::MessageFieldView>, pub __buffa_unknown_fields: ::buffa::UnknownFieldsView<'a>, } impl<'a> ControlMessageView<'a> { @@ -1237,6 +1414,27 @@ impl<'a> ControlMessageView<'a> { let before_tag = cur; let tag = ::buffa::encoding::Tag::decode(&mut cur)?; match tag.field_number() { + 6u32 => { + if tag.wire_type() != ::buffa::encoding::WireType::LengthDelimited { + return ::core::result::Result::Err(::buffa::DecodeError::WireTypeMismatch { + field_number: 6u32, + expected: 2u8, + actual: tag.wire_type() as u8, + }); + } + if depth == 0 { + return Err(::buffa::DecodeError::RecursionLimitExceeded); + } + let sub = ::buffa::types::borrow_bytes(&mut cur)?; + match view.extensions.as_mut() { + Some(existing) => existing._merge_into_view(sub, depth - 1)?, + None => { + view.extensions = ::buffa::MessageFieldView::set( + ControlExtensionsView::_decode_depth(sub, depth - 1)?, + ); + } + } + } 1u32 => { if tag.wire_type() != ::buffa::encoding::WireType::LengthDelimited { return ::core::result::Result::Err(::buffa::DecodeError::WireTypeMismatch { @@ -1340,6 +1538,14 @@ impl<'a> ::buffa::MessageView<'a> for ControlMessageView<'a> { graft: self.graft.iter().map(|v| v.to_owned_message()).collect(), prune: self.prune.iter().map(|v| v.to_owned_message()).collect(), idontwant: self.idontwant.iter().map(|v| v.to_owned_message()).collect(), + extensions: match self.extensions.as_option() { + Some(v) => { + ::buffa::MessageField::< + ControlExtensions, + >::some(v.to_owned_message()) + } + None => ::buffa::MessageField::none(), + }, __buffa_unknown_fields: self .__buffa_unknown_fields .to_owned() @@ -1361,6 +1567,539 @@ unsafe impl<'a> ::buffa::HasDefaultViewInstance for ControlMessageView<'a> { type Static = ControlMessageView<'static>; } #[derive(Clone, PartialEq, Default)] +pub struct ControlExtensions { + ///Field 10: `partial_messages` + pub partial_messages: Option, + #[doc(hidden)] + pub __buffa_unknown_fields: ::buffa::UnknownFields, + #[doc(hidden)] + pub __buffa_cached_size: ::buffa::__private::CachedSize, +} +impl ::core::fmt::Debug for ControlExtensions { + fn fmt(&self, f: &mut ::core::fmt::Formatter<'_>) -> ::core::fmt::Result { + f.debug_struct("ControlExtensions") + .field("partial_messages", &self.partial_messages) + .finish() + } +} +impl ControlExtensions { + /// Protobuf type URL for this message, for use with `Any::pack` and + /// `Any::unpack_if`. + /// + /// Format: `type.googleapis.com/` + pub const TYPE_URL: &'static str = "type.googleapis.com/gossipsub.ControlExtensions"; +} +unsafe impl ::buffa::DefaultInstance for ControlExtensions { + fn default_instance() -> &'static Self { + static VALUE: ::buffa::__private::OnceBox = ::buffa::__private::OnceBox::new(); + VALUE + .get_or_init(|| ::buffa::alloc::boxed::Box::new( + ControlExtensions::default(), + )) + } +} +impl ::buffa::Message for ControlExtensions { + /// Returns the total encoded size in bytes. + /// + /// The result is a `u32`; the protobuf specification requires all + /// messages to fit within 2 GiB (2,147,483,647 bytes), so a + /// compliant message will never overflow this type. + fn compute_size(&self) -> u32 { + #[allow(unused_imports)] + use ::buffa::Enumeration as _; + let mut size = 0u32; + if self.partial_messages.is_some() { + size += 1u32 + ::buffa::types::BOOL_ENCODED_LEN as u32; + } + size += self.__buffa_unknown_fields.encoded_len() as u32; + self.__buffa_cached_size.set(size); + size + } + fn write_to(&self, buf: &mut impl ::buffa::bytes::BufMut) { + #[allow(unused_imports)] + use ::buffa::Enumeration as _; + if let Some(v) = self.partial_messages { + ::buffa::encoding::Tag::new(10u32, ::buffa::encoding::WireType::Varint) + .encode(buf); + ::buffa::types::encode_bool(v, buf); + } + self.__buffa_unknown_fields.write_to(buf); + } + fn merge_field( + &mut self, + tag: ::buffa::encoding::Tag, + buf: &mut impl ::buffa::bytes::Buf, + depth: u32, + ) -> ::core::result::Result<(), ::buffa::DecodeError> { + #[allow(unused_imports)] + use ::buffa::bytes::Buf as _; + #[allow(unused_imports)] + use ::buffa::Enumeration as _; + match tag.field_number() { + 10u32 => { + if tag.wire_type() != ::buffa::encoding::WireType::Varint { + return ::core::result::Result::Err(::buffa::DecodeError::WireTypeMismatch { + field_number: 10u32, + expected: 0u8, + actual: tag.wire_type() as u8, + }); + } + self.partial_messages = ::core::option::Option::Some( + ::buffa::types::decode_bool(buf)?, + ); + } + _ => { + self.__buffa_unknown_fields + .push(::buffa::encoding::decode_unknown_field(tag, buf, depth)?); + } + } + ::core::result::Result::Ok(()) + } + fn cached_size(&self) -> u32 { + self.__buffa_cached_size.get() + } + fn clear(&mut self) { + self.partial_messages = ::core::option::Option::None; + self.__buffa_unknown_fields.clear(); + self.__buffa_cached_size.set(0); + } +} +#[derive(Clone, Debug, Default)] +pub struct ControlExtensionsView<'a> { + ///Field 10: `partial_messages` + pub partial_messages: ::core::option::Option, + pub __buffa_unknown_fields: ::buffa::UnknownFieldsView<'a>, +} +impl<'a> ControlExtensionsView<'a> { + /// Decode from `buf`, enforcing a recursion depth limit for nested messages. + /// + /// Called by [`::buffa::MessageView::decode_view`] with [`::buffa::RECURSION_LIMIT`] + /// and by generated sub-message decode arms with `depth - 1`. + /// + /// **Not part of the public API.** Named with a leading underscore to + /// signal that it is for generated-code use only. + #[doc(hidden)] + pub fn _decode_depth( + buf: &'a [u8], + depth: u32, + ) -> ::core::result::Result { + let mut view = Self::default(); + view._merge_into_view(buf, depth)?; + ::core::result::Result::Ok(view) + } + /// Merge fields from `buf` into this view (proto merge semantics). + /// + /// Repeated fields append; singular fields last-wins; singular + /// MESSAGE fields merge recursively. Used by sub-message decode + /// arms when the same field appears multiple times on the wire. + /// + /// **Not part of the public API.** + #[doc(hidden)] + pub fn _merge_into_view( + &mut self, + buf: &'a [u8], + depth: u32, + ) -> ::core::result::Result<(), ::buffa::DecodeError> { + let _ = depth; + #[allow(unused_variables)] + let view = self; + let mut cur: &'a [u8] = buf; + while !cur.is_empty() { + let before_tag = cur; + let tag = ::buffa::encoding::Tag::decode(&mut cur)?; + match tag.field_number() { + 10u32 => { + if tag.wire_type() != ::buffa::encoding::WireType::Varint { + return ::core::result::Result::Err(::buffa::DecodeError::WireTypeMismatch { + field_number: 10u32, + expected: 0u8, + actual: tag.wire_type() as u8, + }); + } + view.partial_messages = Some(::buffa::types::decode_bool(&mut cur)?); + } + _ => { + ::buffa::encoding::skip_field_depth(tag, &mut cur, depth)?; + let span_len = before_tag.len() - cur.len(); + view.__buffa_unknown_fields.push_raw(&before_tag[..span_len]); + } + } + } + ::core::result::Result::Ok(()) + } +} +impl<'a> ::buffa::MessageView<'a> for ControlExtensionsView<'a> { + type Owned = ControlExtensions; + fn decode_view(buf: &'a [u8]) -> ::core::result::Result { + Self::_decode_depth(buf, ::buffa::RECURSION_LIMIT) + } + fn decode_view_with_limit( + buf: &'a [u8], + depth: u32, + ) -> ::core::result::Result { + Self::_decode_depth(buf, depth) + } + /// Convert this view to the owned message type. + #[allow(clippy::redundant_closure)] + fn to_owned_message(&self) -> ControlExtensions { + #[allow(unused_imports)] + use ::buffa::alloc::string::ToString as _; + ControlExtensions { + partial_messages: self.partial_messages, + __buffa_unknown_fields: self + .__buffa_unknown_fields + .to_owned() + .unwrap_or_default(), + ..::core::default::Default::default() + } + } +} +unsafe impl ::buffa::DefaultViewInstance for ControlExtensionsView<'static> { + fn default_view_instance() -> &'static Self { + static VALUE: ::buffa::__private::OnceBox> = ::buffa::__private::OnceBox::new(); + VALUE + .get_or_init(|| ::buffa::alloc::boxed::Box::new( + ControlExtensionsView::default(), + )) + } +} +unsafe impl<'a> ::buffa::HasDefaultViewInstance for ControlExtensionsView<'a> { + type Static = ControlExtensionsView<'static>; +} +#[derive(Clone, PartialEq, Default)] +pub struct PartialMessagesExtension { + ///Field 1: `topic_id` + pub topic_id: Option<::buffa::alloc::vec::Vec>, + ///Field 2: `group_id` + pub group_id: Option<::buffa::alloc::vec::Vec>, + ///Field 3: `partial_message` + pub partial_message: Option<::buffa::alloc::vec::Vec>, + ///Field 4: `parts_metadata` + pub parts_metadata: Option<::buffa::alloc::vec::Vec>, + #[doc(hidden)] + pub __buffa_unknown_fields: ::buffa::UnknownFields, + #[doc(hidden)] + pub __buffa_cached_size: ::buffa::__private::CachedSize, +} +impl ::core::fmt::Debug for PartialMessagesExtension { + fn fmt(&self, f: &mut ::core::fmt::Formatter<'_>) -> ::core::fmt::Result { + f.debug_struct("PartialMessagesExtension") + .field("topic_id", &self.topic_id) + .field("group_id", &self.group_id) + .field("partial_message", &self.partial_message) + .field("parts_metadata", &self.parts_metadata) + .finish() + } +} +impl PartialMessagesExtension { + /// Protobuf type URL for this message, for use with `Any::pack` and + /// `Any::unpack_if`. + /// + /// Format: `type.googleapis.com/` + pub const TYPE_URL: &'static str = "type.googleapis.com/gossipsub.PartialMessagesExtension"; +} +unsafe impl ::buffa::DefaultInstance for PartialMessagesExtension { + fn default_instance() -> &'static Self { + static VALUE: ::buffa::__private::OnceBox = ::buffa::__private::OnceBox::new(); + VALUE + .get_or_init(|| ::buffa::alloc::boxed::Box::new( + PartialMessagesExtension::default(), + )) + } +} +impl ::buffa::Message for PartialMessagesExtension { + /// Returns the total encoded size in bytes. + /// + /// The result is a `u32`; the protobuf specification requires all + /// messages to fit within 2 GiB (2,147,483,647 bytes), so a + /// compliant message will never overflow this type. + fn compute_size(&self) -> u32 { + #[allow(unused_imports)] + use ::buffa::Enumeration as _; + let mut size = 0u32; + if let Some(ref v) = self.topic_id { + size += 1u32 + ::buffa::types::bytes_encoded_len(v) as u32; + } + if let Some(ref v) = self.group_id { + size += 1u32 + ::buffa::types::bytes_encoded_len(v) as u32; + } + if let Some(ref v) = self.partial_message { + size += 1u32 + ::buffa::types::bytes_encoded_len(v) as u32; + } + if let Some(ref v) = self.parts_metadata { + size += 1u32 + ::buffa::types::bytes_encoded_len(v) as u32; + } + size += self.__buffa_unknown_fields.encoded_len() as u32; + self.__buffa_cached_size.set(size); + size + } + fn write_to(&self, buf: &mut impl ::buffa::bytes::BufMut) { + #[allow(unused_imports)] + use ::buffa::Enumeration as _; + if let Some(ref v) = self.topic_id { + ::buffa::encoding::Tag::new( + 1u32, + ::buffa::encoding::WireType::LengthDelimited, + ) + .encode(buf); + ::buffa::types::encode_bytes(v, buf); + } + if let Some(ref v) = self.group_id { + ::buffa::encoding::Tag::new( + 2u32, + ::buffa::encoding::WireType::LengthDelimited, + ) + .encode(buf); + ::buffa::types::encode_bytes(v, buf); + } + if let Some(ref v) = self.partial_message { + ::buffa::encoding::Tag::new( + 3u32, + ::buffa::encoding::WireType::LengthDelimited, + ) + .encode(buf); + ::buffa::types::encode_bytes(v, buf); + } + if let Some(ref v) = self.parts_metadata { + ::buffa::encoding::Tag::new( + 4u32, + ::buffa::encoding::WireType::LengthDelimited, + ) + .encode(buf); + ::buffa::types::encode_bytes(v, buf); + } + self.__buffa_unknown_fields.write_to(buf); + } + fn merge_field( + &mut self, + tag: ::buffa::encoding::Tag, + buf: &mut impl ::buffa::bytes::Buf, + depth: u32, + ) -> ::core::result::Result<(), ::buffa::DecodeError> { + #[allow(unused_imports)] + use ::buffa::bytes::Buf as _; + #[allow(unused_imports)] + use ::buffa::Enumeration as _; + match tag.field_number() { + 1u32 => { + if tag.wire_type() != ::buffa::encoding::WireType::LengthDelimited { + return ::core::result::Result::Err(::buffa::DecodeError::WireTypeMismatch { + field_number: 1u32, + expected: 2u8, + actual: tag.wire_type() as u8, + }); + } + ::buffa::types::merge_bytes( + self.topic_id.get_or_insert_with(::buffa::alloc::vec::Vec::new), + buf, + )?; + } + 2u32 => { + if tag.wire_type() != ::buffa::encoding::WireType::LengthDelimited { + return ::core::result::Result::Err(::buffa::DecodeError::WireTypeMismatch { + field_number: 2u32, + expected: 2u8, + actual: tag.wire_type() as u8, + }); + } + ::buffa::types::merge_bytes( + self.group_id.get_or_insert_with(::buffa::alloc::vec::Vec::new), + buf, + )?; + } + 3u32 => { + if tag.wire_type() != ::buffa::encoding::WireType::LengthDelimited { + return ::core::result::Result::Err(::buffa::DecodeError::WireTypeMismatch { + field_number: 3u32, + expected: 2u8, + actual: tag.wire_type() as u8, + }); + } + ::buffa::types::merge_bytes( + self + .partial_message + .get_or_insert_with(::buffa::alloc::vec::Vec::new), + buf, + )?; + } + 4u32 => { + if tag.wire_type() != ::buffa::encoding::WireType::LengthDelimited { + return ::core::result::Result::Err(::buffa::DecodeError::WireTypeMismatch { + field_number: 4u32, + expected: 2u8, + actual: tag.wire_type() as u8, + }); + } + ::buffa::types::merge_bytes( + self + .parts_metadata + .get_or_insert_with(::buffa::alloc::vec::Vec::new), + buf, + )?; + } + _ => { + self.__buffa_unknown_fields + .push(::buffa::encoding::decode_unknown_field(tag, buf, depth)?); + } + } + ::core::result::Result::Ok(()) + } + fn cached_size(&self) -> u32 { + self.__buffa_cached_size.get() + } + fn clear(&mut self) { + self.topic_id = ::core::option::Option::None; + self.group_id = ::core::option::Option::None; + self.partial_message = ::core::option::Option::None; + self.parts_metadata = ::core::option::Option::None; + self.__buffa_unknown_fields.clear(); + self.__buffa_cached_size.set(0); + } +} +#[derive(Clone, Debug, Default)] +pub struct PartialMessagesExtensionView<'a> { + ///Field 1: `topic_id` + pub topic_id: ::core::option::Option<&'a [u8]>, + ///Field 2: `group_id` + pub group_id: ::core::option::Option<&'a [u8]>, + ///Field 3: `partial_message` + pub partial_message: ::core::option::Option<&'a [u8]>, + ///Field 4: `parts_metadata` + pub parts_metadata: ::core::option::Option<&'a [u8]>, + pub __buffa_unknown_fields: ::buffa::UnknownFieldsView<'a>, +} +impl<'a> PartialMessagesExtensionView<'a> { + /// Decode from `buf`, enforcing a recursion depth limit for nested messages. + /// + /// Called by [`::buffa::MessageView::decode_view`] with [`::buffa::RECURSION_LIMIT`] + /// and by generated sub-message decode arms with `depth - 1`. + /// + /// **Not part of the public API.** Named with a leading underscore to + /// signal that it is for generated-code use only. + #[doc(hidden)] + pub fn _decode_depth( + buf: &'a [u8], + depth: u32, + ) -> ::core::result::Result { + let mut view = Self::default(); + view._merge_into_view(buf, depth)?; + ::core::result::Result::Ok(view) + } + /// Merge fields from `buf` into this view (proto merge semantics). + /// + /// Repeated fields append; singular fields last-wins; singular + /// MESSAGE fields merge recursively. Used by sub-message decode + /// arms when the same field appears multiple times on the wire. + /// + /// **Not part of the public API.** + #[doc(hidden)] + pub fn _merge_into_view( + &mut self, + buf: &'a [u8], + depth: u32, + ) -> ::core::result::Result<(), ::buffa::DecodeError> { + let _ = depth; + #[allow(unused_variables)] + let view = self; + let mut cur: &'a [u8] = buf; + while !cur.is_empty() { + let before_tag = cur; + let tag = ::buffa::encoding::Tag::decode(&mut cur)?; + match tag.field_number() { + 1u32 => { + if tag.wire_type() != ::buffa::encoding::WireType::LengthDelimited { + return ::core::result::Result::Err(::buffa::DecodeError::WireTypeMismatch { + field_number: 1u32, + expected: 2u8, + actual: tag.wire_type() as u8, + }); + } + view.topic_id = Some(::buffa::types::borrow_bytes(&mut cur)?); + } + 2u32 => { + if tag.wire_type() != ::buffa::encoding::WireType::LengthDelimited { + return ::core::result::Result::Err(::buffa::DecodeError::WireTypeMismatch { + field_number: 2u32, + expected: 2u8, + actual: tag.wire_type() as u8, + }); + } + view.group_id = Some(::buffa::types::borrow_bytes(&mut cur)?); + } + 3u32 => { + if tag.wire_type() != ::buffa::encoding::WireType::LengthDelimited { + return ::core::result::Result::Err(::buffa::DecodeError::WireTypeMismatch { + field_number: 3u32, + expected: 2u8, + actual: tag.wire_type() as u8, + }); + } + view.partial_message = Some(::buffa::types::borrow_bytes(&mut cur)?); + } + 4u32 => { + if tag.wire_type() != ::buffa::encoding::WireType::LengthDelimited { + return ::core::result::Result::Err(::buffa::DecodeError::WireTypeMismatch { + field_number: 4u32, + expected: 2u8, + actual: tag.wire_type() as u8, + }); + } + view.parts_metadata = Some(::buffa::types::borrow_bytes(&mut cur)?); + } + _ => { + ::buffa::encoding::skip_field_depth(tag, &mut cur, depth)?; + let span_len = before_tag.len() - cur.len(); + view.__buffa_unknown_fields.push_raw(&before_tag[..span_len]); + } + } + } + ::core::result::Result::Ok(()) + } +} +impl<'a> ::buffa::MessageView<'a> for PartialMessagesExtensionView<'a> { + type Owned = PartialMessagesExtension; + fn decode_view(buf: &'a [u8]) -> ::core::result::Result { + Self::_decode_depth(buf, ::buffa::RECURSION_LIMIT) + } + fn decode_view_with_limit( + buf: &'a [u8], + depth: u32, + ) -> ::core::result::Result { + Self::_decode_depth(buf, depth) + } + /// Convert this view to the owned message type. + #[allow(clippy::redundant_closure)] + fn to_owned_message(&self) -> PartialMessagesExtension { + #[allow(unused_imports)] + use ::buffa::alloc::string::ToString as _; + PartialMessagesExtension { + topic_id: self.topic_id.map(|b| (b).to_vec()), + group_id: self.group_id.map(|b| (b).to_vec()), + partial_message: self.partial_message.map(|b| (b).to_vec()), + parts_metadata: self.parts_metadata.map(|b| (b).to_vec()), + __buffa_unknown_fields: self + .__buffa_unknown_fields + .to_owned() + .unwrap_or_default(), + ..::core::default::Default::default() + } + } +} +unsafe impl ::buffa::DefaultViewInstance for PartialMessagesExtensionView<'static> { + fn default_view_instance() -> &'static Self { + static VALUE: ::buffa::__private::OnceBox< + PartialMessagesExtensionView<'static>, + > = ::buffa::__private::OnceBox::new(); + VALUE + .get_or_init(|| ::buffa::alloc::boxed::Box::new( + PartialMessagesExtensionView::default(), + )) + } +} +unsafe impl<'a> ::buffa::HasDefaultViewInstance for PartialMessagesExtensionView<'a> { + type Static = PartialMessagesExtensionView<'static>; +} +#[derive(Clone, PartialEq, Default)] pub struct ControlIHave { ///Field 1: `topic_id` pub topic_id: Option<::buffa::alloc::string::String>, diff --git a/crates/gossip/src/lib.rs b/crates/gossip/src/lib.rs index ce2a6333a..374ab783c 100644 --- a/crates/gossip/src/lib.rs +++ b/crates/gossip/src/lib.rs @@ -8,12 +8,14 @@ mod generated; mod handler; mod mcache; mod message; +mod partial; pub use control::{ copy_grafts_to_protobuf_output, copy_prunes_to_protobuf_output, copy_subscribes_to_protobuf_output, copy_unsubscribes_to_protobuf_output, }; pub use handler::GossipHandler; +pub use partial::{PartialFrame, PartsMetadata}; use silver_common::{GossipMsgOut, NewGossipMsg, PeerEvent}; /// Events emitted by the GossipHandler. diff --git a/crates/gossip/src/partial.rs b/crates/gossip/src/partial.rs new file mode 100644 index 000000000..0df2e4765 --- /dev/null +++ b/crates/gossip/src/partial.rs @@ -0,0 +1,540 @@ +use std::{iter, time::Instant}; + +use buffa::{ + bytes::BufMut, + encoding::{Tag, WireType, encode_varint, varint_len}, + types::{encode_bytes, encode_string, string_encoded_len}, +}; +use silver_common::{ + CacheFrameError, CacheFrameRef, CacheSegment, TProducer, + ssz_view::{ + BYTES_PER_CELL, BYTES_PER_KZG_PROOF, + partial_column::{PartialSidecarPlan, parts_metadata_len, write_parts_metadata}, + }, +}; + +// Tags, varints, topic, group id, metadata field, and the SSZ +// offsets-plus-bitmap prefix; every piece is small and bounded. +const MAX_PARTIAL_FRAMING: usize = 256; + +/// Row masks for a `partsMetadata` field; both bitlists carry exactly +/// `n_rows` bits. +pub struct PartsMetadata { + pub available: u128, + pub requests: u128, + pub n_rows: usize, +} + +impl PartsMetadata { + fn encodable(&self) -> bool { + self.n_rows <= 128 && + (self.n_rows == 128 || + (self.available >> self.n_rows == 0 && self.requests >> self.n_rows == 0)) + } +} + +/// One outbound partial-column RPC frame: `RPC.partial` carrying a +/// `PartialMessagesExtension` for a single `(topic, group)` pair. The +/// fork identity is fixed by `plan` and `group_id` at construction and +/// never rewritten. Cell and proof payloads stay in their retained +/// records; only framing bytes are prepared here. +pub struct PartialFrame<'a> { + /// Full topic name (the protobuf `topic_id`). + pub topic: &'a str, + pub group_id: &'a [u8], + /// `None` omits `partialMessage`: a metadata-only frame. + pub plan: Option, + /// Fulu eager push only; the segment length must equal the plan's + /// `header_bytes`. + pub header: Option, + pub metadata: Option, +} + +impl PartialFrame<'_> { + /// Write the frame descriptor into the outgoing gossip cache. + /// `cells` and `proofs` are the selected rows' ranges in ascending + /// row order; counts and lengths must match the plan. + pub fn write( + &self, + producer: &mut TProducer, + cells: impl ExactSizeIterator + Clone, + proofs: impl ExactSizeIterator + Clone, + expires: Instant, + ) -> Result { + let k = self.plan.map_or(0, |plan| plan.cell_count()); + let header_bytes = self.plan.map_or(0, |plan| plan.header_bytes()); + // Range lengths back the declared varints; a mismatch would emit + // corrupt protobuf, so reject it here rather than on the wire. + // Metadata row counts must match the payload's and stay encodable. + if (self.plan.is_none() && self.metadata.is_none()) || + cells.len() != k || + proofs.len() != k || + cells.clone().any(|cell| segment_len(cell) != BYTES_PER_CELL) || + proofs.clone().any(|proof| segment_len(proof) != BYTES_PER_KZG_PROOF) || + self.header.map_or(0, segment_len) != header_bytes || + self.metadata.as_ref().is_some_and(|meta| { + !meta.encodable() || self.plan.is_some_and(|plan| plan.n_rows() != meta.n_rows) + }) + { + return Err(CacheFrameError::InvalidDescriptor); + } + + let ssz_len = self.plan.map_or(0, |plan| plan.ssz_len()); + let meta_len = self.metadata.as_ref().map(|meta| parts_metadata_len(meta.n_rows)); + let fields_len = 1 + + string_encoded_len(self.topic) + + 1 + + varint_len(self.group_id.len() as u64) + + self.group_id.len() + + self.plan.map_or(0, |_| 1 + varint_len(ssz_len as u64)); + let tail = meta_len.map_or(0, |m| 1 + varint_len(m as u64) + m); + let ext_len = fields_len + ssz_len + tail; + + // Framing layout: [lead | header list prefix | metadata field]. + let lead = 1 + + varint_len(ext_len as u64) + + fields_len + + self.plan.map_or(0, |plan| plan.prefix_len()); + let mid = if header_bytes > 0 { 4 } else { 0 }; + let framing_len = lead + mid + tail; + if framing_len > MAX_PARTIAL_FRAMING { + return Err(CacheFrameError::TooLarge); + } + + let mut buf = [0u8; MAX_PARTIAL_FRAMING]; + let mut cursor: &mut [u8] = &mut buf[..framing_len]; + Tag::new(10, WireType::LengthDelimited).encode(&mut cursor); + encode_varint(ext_len as u64, &mut cursor); + Tag::new(1, WireType::LengthDelimited).encode(&mut cursor); + encode_string(self.topic, &mut cursor); + Tag::new(2, WireType::LengthDelimited).encode(&mut cursor); + encode_bytes(self.group_id, &mut cursor); + if let Some(plan) = self.plan { + Tag::new(3, WireType::LengthDelimited).encode(&mut cursor); + encode_varint(ssz_len as u64, &mut cursor); + let mut prefix = [0u8; 33]; + let prefix = &mut prefix[..plan.prefix_len()]; + plan.write_prefix(prefix); + cursor.put_slice(prefix); + } + debug_assert_eq!(cursor.len(), mid + tail); + if header_bytes > 0 { + cursor.put_slice(&PartialSidecarPlan::HEADER_LIST_PREFIX); + } + if let Some(meta) = &self.metadata { + let mut buf = [0u8; parts_metadata_len(128)]; + let bytes = &mut buf[..meta_len.unwrap()]; + write_parts_metadata(meta.available, meta.requests, meta.n_rows, bytes); + Tag::new(4, WireType::LengthDelimited).encode(&mut cursor); + encode_bytes(bytes, &mut cursor); + } + debug_assert!(cursor.is_empty()); + + let header_prefix = + (header_bytes > 0).then_some(CacheSegment::Framing { offset: lead, length: 4 }); + let metadata_framing = + (tail > 0).then_some(CacheSegment::Framing { offset: lead + mid, length: tail }); + let count = 1 + + 2 * k + + usize::from(header_prefix.is_some()) * 2 + + usize::from(metadata_framing.is_some()); + CacheFrameRef::write(producer, expires, &buf[..framing_len], WithLen { + len: count, + inner: iter::once(CacheSegment::Framing { offset: 0, length: lead }) + .chain(cells) + .chain(proofs) + .chain(header_prefix) + .chain(self.header) + .chain(metadata_framing), + }) + } +} + +// `Chain` drops `ExactSizeIterator`; the segment count is known exactly. +struct WithLen { + inner: I, + len: usize, +} + +impl Iterator for WithLen { + type Item = I::Item; + + fn next(&mut self) -> Option { + self.len = self.len.saturating_sub(1); + self.inner.next() + } + + fn size_hint(&self) -> (usize, Option) { + (self.len, Some(self.len)) + } +} + +impl ExactSizeIterator for WithLen {} + +fn segment_len(segment: CacheSegment) -> usize { + match segment { + CacheSegment::Framing { length, .. } | + CacheSegment::Gossip { length, .. } | + CacheSegment::DataColumns { length, .. } | + CacheSegment::Shared { length, .. } => length, + } +} + +#[cfg(test)] +mod tests { + use buffa::{Message, MessageField, MessageView}; + + use crate::generated::{ + ControlExtensions, ControlMessage, PartialMessagesExtension, RPC, RPCView, rpc::SubOpts, + }; + + /// Registry wire numbers: SubOpts.requestsPartial=3, + /// supportsSendingPartial=4, ControlMessage.extensions=6, + /// ControlExtensions.partialMessages=10, RPC.partial=10. + #[test] + fn partial_fields_round_trip() { + let rpc = RPC { + subscriptions: vec![SubOpts { + subscribe: Some(true), + topic_id: Some("col_topic".into()), + requests_partial: Some(true), + supports_sending_partial: None, + ..Default::default() + }], + control: MessageField::some(ControlMessage { + extensions: MessageField::some(ControlExtensions { + partial_messages: Some(true), + ..Default::default() + }), + ..Default::default() + }), + partial: MessageField::some(PartialMessagesExtension { + topic_id: Some(b"col_topic".to_vec()), + group_id: Some(vec![0u8; 33]), + partial_message: Some(vec![0xab; 64]), + parts_metadata: None, + ..Default::default() + }), + ..Default::default() + }; + + let bytes = rpc.encode_to_vec(); + let view = RPCView::decode_view(&bytes).unwrap(); + + let sub = view.subscriptions.iter().next().unwrap(); + assert_eq!(sub.requests_partial, Some(true)); + // requestsPartial implies sending support; the flag itself stays + // absent on the wire and normalization happens at the consumer. + assert_eq!(sub.supports_sending_partial, None); + + assert_eq!( + view.control.as_option().unwrap().extensions.as_option().unwrap().partial_messages, + Some(true) + ); + + let partial = view.partial.as_option().unwrap(); + assert_eq!(partial.topic_id, Some(&b"col_topic"[..])); + assert_eq!(partial.group_id.unwrap().len(), 33); + assert_eq!(partial.partial_message, Some(&[0xab; 64][..])); + assert_eq!(partial.parts_metadata, None); + } + + #[test] + fn legacy_frames_decode_with_partial_fields_unset() { + let rpc = RPC { + subscriptions: vec![SubOpts { + subscribe: Some(true), + topic_id: Some("t".into()), + ..Default::default() + }], + control: MessageField::some(ControlMessage::default()), + ..Default::default() + }; + + let bytes = rpc.encode_to_vec(); + let view = RPCView::decode_view(&bytes).unwrap(); + + let sub = view.subscriptions.iter().next().unwrap(); + assert_eq!(sub.requests_partial, None); + assert_eq!(sub.supports_sending_partial, None); + assert!(!view.control.as_option().unwrap().extensions.is_set()); + assert!(!view.partial.is_set()); + } + + use std::time::{Duration, Instant}; + + use silver_common::{ + CacheFrameError, CacheFrameRef, CacheSegment, TCache, TCacheProducer, TProducer, + TRandomAccess, + ssz_view::{ + BYTES_PER_CELL, BYTES_PER_KZG_COMMITMENT, BYTES_PER_KZG_PROOF, + partial_column::{ + PARTIAL_HEADER_FIXED, PartialLayout, PartialSidecarPlan, fulu_group_id, + gloas_group_id, parts_metadata_len, write_parts_metadata, + }, + }, + }; + + use super::{PartialFrame, PartsMetadata}; + + const TOPIC: &str = "/eth2/aabbccdd/data_column_sidecar_7/ssz_snappy"; + + /// Reference bytes from buffa's own encoder over the same content. + fn reference_rpc(group_id: &[u8], ssz: Vec, metadata: Option>) -> Vec { + RPC { + partial: MessageField::some(PartialMessagesExtension { + topic_id: Some(TOPIC.as_bytes().to_vec()), + group_id: Some(group_id.to_vec()), + partial_message: Some(ssz), + parts_metadata: metadata, + ..Default::default() + }), + ..Default::default() + } + .encode_to_vec() + } + + fn reassemble(frame: CacheFrameRef, consumer: &mut TRandomAccess, now: Instant) -> Vec { + let view = frame.acquire(consumer, now).unwrap(); + let descriptor = view.descriptor_range(); + let mut wire = Vec::new(); + for segment in view.segments() { + if let Some(range) = segment.framing_range() { + wire.extend_from_slice(&descriptor.as_ref()[range]); + } else { + wire.extend_from_slice(segment.acquire(consumer, None).unwrap().as_ref()); + } + } + assert_eq!(wire.len(), view.wire_len()); + wire + } + + /// One source record holding k cells, k proofs, and `header_bytes` + /// of header SSZ; returns per-region segments. + fn source_record( + producer: &mut TProducer, + k: usize, + header_bytes: usize, + ) -> (Vec, Vec, Option, Vec) { + let len = k * (BYTES_PER_CELL + BYTES_PER_KZG_PROOF) + header_bytes; + let mut reservation = producer.reserve(len, true).unwrap(); + let read = reservation.read(); + { + let buf = reservation.buffer().unwrap(); + for (i, byte) in buf.iter_mut().enumerate() { + *byte = (i % 251) as u8; + } + if header_bytes > 0 { + let header = &mut buf[len - header_bytes..]; + header[0..4].copy_from_slice(&(PARTIAL_HEADER_FIXED as u32).to_le_bytes()); + } + } + let bytes = reservation.buffer().unwrap().to_vec(); + reservation.increment_offset(len); + + let cells = (0..k) + .map(|i| CacheSegment::Gossip { + read, + offset: i * BYTES_PER_CELL, + length: BYTES_PER_CELL, + }) + .collect(); + let proofs = (0..k) + .map(|i| CacheSegment::Gossip { + read, + offset: k * BYTES_PER_CELL + i * BYTES_PER_KZG_PROOF, + length: BYTES_PER_KZG_PROOF, + }) + .collect(); + let header = (header_bytes > 0).then_some(CacheSegment::Gossip { + read, + offset: len - header_bytes, + length: header_bytes, + }); + (cells, proofs, header, bytes) + } + + fn reference_ssz(plan: &PartialSidecarPlan, source: &[u8], header_bytes: usize) -> Vec { + let mut ssz = vec![0u8; plan.prefix_len()]; + plan.write_prefix(&mut ssz); + ssz.extend_from_slice(&source[..source.len() - header_bytes]); + if header_bytes > 0 { + ssz.extend_from_slice(&PartialSidecarPlan::HEADER_LIST_PREFIX); + ssz.extend_from_slice(&source[source.len() - header_bytes..]); + } + assert_eq!(ssz.len(), plan.ssz_len()); + ssz + } + + #[test] + fn fulu_frame_matches_buffa_reference_encoding() { + let mut producer = TCache::producer("", 1 << 18); + let mut consumer = producer.cache_ref().strict_random_access("", true).unwrap(); + let now = Instant::now(); + let header_bytes = PARTIAL_HEADER_FIXED + 3 * BYTES_PER_KZG_COMMITMENT; + let (cells, proofs, header, source) = source_record(&mut producer, 2, header_bytes); + + let group = fulu_group_id(&[0xab; 32]); + let plan = PartialSidecarPlan::new(PartialLayout::Fulu { header_bytes }, 0b101, 3).unwrap(); + let frame = PartialFrame { + topic: TOPIC, + group_id: &group, + plan: Some(plan), + header, + metadata: Some(PartsMetadata { available: 0b101, requests: 0b010, n_rows: 3 }), + } + .write( + &mut producer, + cells.iter().copied(), + proofs.iter().copied(), + now + Duration::from_secs(1), + ) + .unwrap(); + + let mut metadata = vec![0u8; parts_metadata_len(3)]; + write_parts_metadata(0b101, 0b010, 3, &mut metadata); + let reference = + reference_rpc(&group, reference_ssz(&plan, &source, header_bytes), Some(metadata)); + + assert_eq!(reassemble(frame, &mut consumer, now), reference); + } + + #[test] + fn gloas_frame_matches_buffa_reference_encoding() { + let mut producer = TCache::producer("", 1 << 18); + let mut consumer = producer.cache_ref().strict_random_access("", true).unwrap(); + let now = Instant::now(); + let (cells, proofs, header, source) = source_record(&mut producer, 1, 0); + assert!(header.is_none()); + + let group = gloas_group_id(&[0xcd; 32], 123_456); + let plan = PartialSidecarPlan::new(PartialLayout::Gloas, 0b10, 2).unwrap(); + let frame = PartialFrame { + topic: TOPIC, + group_id: &group, + plan: Some(plan), + header: None, + metadata: None, + } + .write( + &mut producer, + cells.iter().copied(), + proofs.iter().copied(), + now + Duration::from_secs(1), + ) + .unwrap(); + + let reference = reference_rpc(&group, reference_ssz(&plan, &source, 0), None); + assert_eq!(reassemble(frame, &mut consumer, now), reference); + } + + #[test] + fn frame_rejects_count_and_header_mismatches() { + let mut producer = TCache::producer("", 1 << 18); + let now = Instant::now(); + let (cells, proofs, _, _) = source_record(&mut producer, 2, 0); + + let group = fulu_group_id(&[0xab; 32]); + let plan = + PartialSidecarPlan::new(PartialLayout::Fulu { header_bytes: 0 }, 0b11, 2).unwrap(); + let missing_proof = PartialFrame { + topic: TOPIC, + group_id: &group, + plan: Some(plan), + header: None, + metadata: None, + } + .write( + &mut producer, + cells.iter().copied(), + proofs.iter().copied().take(1), + now + Duration::from_secs(1), + ); + assert!(matches!(missing_proof, Err(CacheFrameError::InvalidDescriptor))); + + let with_header = PartialSidecarPlan::new( + PartialLayout::Fulu { header_bytes: PARTIAL_HEADER_FIXED }, + 0b11, + 2, + ) + .unwrap(); + let missing_header = PartialFrame { + topic: TOPIC, + group_id: &group, + plan: Some(with_header), + header: None, + metadata: None, + } + .write( + &mut producer, + cells.iter().copied(), + proofs.iter().copied(), + now + Duration::from_secs(1), + ); + assert!(matches!(missing_header, Err(CacheFrameError::InvalidDescriptor))); + + let mismatched_rows = PartialFrame { + topic: TOPIC, + group_id: &group, + plan: Some(plan), + header: None, + metadata: Some(PartsMetadata { available: 0, requests: 0b1, n_rows: 3 }), + } + .write( + &mut producer, + cells.iter().copied(), + proofs.iter().copied(), + now + Duration::from_secs(1), + ); + assert!(matches!(mismatched_rows, Err(CacheFrameError::InvalidDescriptor))); + } + + #[test] + fn metadata_only_frame_matches_buffa_reference_encoding() { + let mut producer = TCache::producer("", 1 << 16); + let mut consumer = producer.cache_ref().strict_random_access("", true).unwrap(); + let now = Instant::now(); + + let group = fulu_group_id(&[0xee; 32]); + let frame = PartialFrame { + topic: TOPIC, + group_id: &group, + plan: None, + header: None, + metadata: Some(PartsMetadata { available: 0b11, requests: 0b100, n_rows: 4 }), + } + .write(&mut producer, std::iter::empty(), std::iter::empty(), now + Duration::from_secs(1)) + .unwrap(); + + let mut metadata = vec![0u8; parts_metadata_len(4)]; + write_parts_metadata(0b11, 0b100, 4, &mut metadata); + let reference = RPC { + partial: MessageField::some(PartialMessagesExtension { + topic_id: Some(TOPIC.as_bytes().to_vec()), + group_id: Some(group.to_vec()), + partial_message: None, + parts_metadata: Some(metadata), + ..Default::default() + }), + ..Default::default() + } + .encode_to_vec(); + assert_eq!(reassemble(frame, &mut consumer, now), reference); + + // Neither payload nor metadata: nothing to send. + let empty = PartialFrame { + topic: TOPIC, + group_id: &group, + plan: None, + header: None, + metadata: None, + } + .write( + &mut producer, + std::iter::empty(), + std::iter::empty(), + now + Duration::from_secs(1), + ); + assert!(matches!(empty, Err(CacheFrameError::InvalidDescriptor))); + } +} diff --git a/crates/network/src/p2p/streams/negotiate.rs b/crates/network/src/p2p/streams/negotiate.rs index 9970f8a46..099cbab82 100644 --- a/crates/network/src/p2p/streams/negotiate.rs +++ b/crates/network/src/p2p/streams/negotiate.rs @@ -238,10 +238,13 @@ impl NegotiateState { #[cfg(test)] mod tests { use quinn_proto::{Dir, Side}; - use silver_common::ALL_PROTOCOLS; + use silver_common::{ALL_PROTOCOLS, TRead}; use super::*; - use crate::p2p::streams::AcquiredRpcOutbound; + use crate::p2p::{ + quic::{Leased, OutboundGossip}, + streams::AcquiredRpcOutbound, + }; /// In-memory `StreamIo` for driving the negotiation state machine. struct MockIo { @@ -293,10 +296,10 @@ mod tests { None } - fn gossip_next(&mut self) -> Option { + fn gossip_next(&mut self) -> Option { None } - fn cluster_next(&mut self) -> Option> { + fn cluster_next(&mut self) -> Option> { None } fn remote_addr(&self) -> std::net::SocketAddr { @@ -306,7 +309,7 @@ mod tests { fn write_leased_to_stream( &mut self, _id: StreamId, - _data: crate::p2p::quic::Leased, + _data: Leased, ) -> Result { Ok(0) } diff --git a/crates/ssz/src/ssz_view.rs b/crates/ssz/src/ssz_view.rs index 956eb143e..89663d652 100644 --- a/crates/ssz/src/ssz_view.rs +++ b/crates/ssz/src/ssz_view.rs @@ -34,6 +34,8 @@ pub enum SszView { None, } +pub mod partial_column; + #[inline(always)] fn u64_le(buf: &[u8], off: usize) -> u64 { u64::from_le_bytes(buf[off..off + 8].try_into().unwrap()) diff --git a/crates/ssz/src/ssz_view/partial_column.rs b/crates/ssz/src/ssz_view/partial_column.rs new file mode 100644 index 000000000..549bb1830 --- /dev/null +++ b/crates/ssz/src/ssz_view/partial_column.rs @@ -0,0 +1,586 @@ +//! Partial data-column wire types (fulu/partial-columns and +//! gloas/partial-columns p2p specs). Row bitmaps index blobs, not +//! columns, and decode to `u128`: schedules beyond 128 rows are +//! unsupported. + +use super::{ + BYTES_PER_CELL, BYTES_PER_KZG_COMMITMENT, BYTES_PER_KZG_PROOF, MAX_BLOB_COMMITMENTS_PER_BLOCK, + u32_le, +}; + +pub const PARTIAL_COLUMNS_VERSION_BYTE: u8 = 0x00; +pub const FULU_GROUP_ID_SIZE: usize = 33; +pub const GLOAS_GROUP_ID_SIZE: usize = 41; + +// PartialDataColumnHeader: +// [0..4) offset to kzg_commitments (== 340) +// [4..212) signed_block_header (SignedBeaconBlockHeader, 208B) +// [212..340) kzg_commitments_inclusion_proof (Vector[Bytes32, 4]) +// [340..) kzg_commitments (n * 48B) +pub const PARTIAL_HEADER_FIXED: usize = 340; + +const PARTIAL_SIDECAR_FIXED_FULU: usize = 16; +const PARTIAL_SIDECAR_FIXED_GLOAS: usize = 12; + +/// Spec constant: bounds the Gloas SSZ payload, not the protobuf RPC. +pub const MAX_PARTIAL_DATA_COLUMN_SIDECAR_SIZE_GLOAS: usize = 8_585_741; +/// Derived from the Fulu SSZ schema bounds (4096-cell lists plus a +/// one-element header list). +pub const MAX_PARTIAL_DATA_COLUMN_SIDECAR_SIZE_FULU: usize = PARTIAL_SIDECAR_FIXED_FULU + + bitlist_bytes(MAX_BLOB_COMMITMENTS_PER_BLOCK) + + MAX_BLOB_COMMITMENTS_PER_BLOCK * (BYTES_PER_CELL + BYTES_PER_KZG_PROOF) + + 4 + + PARTIAL_HEADER_FIXED + + MAX_BLOB_COMMITMENTS_PER_BLOCK * BYTES_PER_KZG_COMMITMENT; + +pub fn fulu_group_id(block_root: &[u8; 32]) -> [u8; FULU_GROUP_ID_SIZE] { + let mut id = [PARTIAL_COLUMNS_VERSION_BYTE; FULU_GROUP_ID_SIZE]; + id[1..].copy_from_slice(block_root); + id +} + +pub fn gloas_group_id(block_root: &[u8; 32], slot: u64) -> [u8; GLOAS_GROUP_ID_SIZE] { + let mut id = [PARTIAL_COLUMNS_VERSION_BYTE; GLOAS_GROUP_ID_SIZE]; + id[1..33].copy_from_slice(block_root); + id[33..].copy_from_slice(&slot.to_le_bytes()); + id +} + +pub const fn bitlist_bytes(n_bits: usize) -> usize { + n_bits / 8 + 1 +} + +/// Decode a bitlist into a row mask. Canonical form is enforced: the +/// delimiter bit terminates the last byte, so a trailing zero byte or +/// empty input fails. `max_bits` is the trusted row count; anything +/// longer (or beyond the 128-row `u128` ceiling) fails. +pub fn bitlist_u128(buf: &[u8], max_bits: usize) -> Option<(u128, usize)> { + let last = *buf.last()?; + if last == 0 { + return None; + } + let delimiter = 7 - last.leading_zeros() as usize; + let n = (buf.len() - 1) * 8 + delimiter; + if n > max_bits || n > 128 { + return None; + } + let mut mask = 0u128; + for (i, &b) in buf.iter().enumerate() { + let b = if i == buf.len() - 1 { b ^ (1 << delimiter) } else { b }; + if b == 0 { + continue; + } + mask |= (b as u128) << (i * 8); + } + Some((mask, n)) +} + +/// `mask` must fit in `n_bits`; `out` must be exactly `bitlist_bytes(n_bits)`. +pub fn write_bitlist_u128(mask: u128, n_bits: usize, out: &mut [u8]) { + debug_assert!(n_bits <= 128 && (n_bits == 128 || mask >> n_bits == 0)); + debug_assert_eq!(out.len(), bitlist_bytes(n_bits)); + for (i, byte) in out.iter_mut().enumerate() { + *byte = if i * 8 < 128 { (mask >> (i * 8)) as u8 } else { 0 }; + } + out[n_bits / 8] |= 1 << (n_bits % 8); +} + +// PartialDataColumnSidecar (Fulu): +// [0..4) offset to cells_present_bitmap (== 16) +// [4..8) offset to partial_column +// [8..12) offset to kzg_proofs +// [12..16) offset to header (List[PartialDataColumnHeader, 1]) +// bitmap | cells (k * 2048) | proofs (k * 48) | header list +// A one-element header list is a 4-byte inner offset (== 4) plus the +// header SSZ; an omitted header is an empty list (zero bytes). +pub struct PartialDataColumnSidecarFuluView; + +impl PartialDataColumnSidecarFuluView { + /// Full structural check; returns the row mask. The bitmap must + /// have exactly the trusted `n_rows` bits, cell/proof region + /// lengths must match its popcount, and a cell-less sidecar is + /// only valid when it carries a header. + pub fn check_size(buf: &[u8], n_rows: usize) -> Option { + if buf.len() < PARTIAL_SIDECAR_FIXED_FULU + 1 || + buf.len() > MAX_PARTIAL_DATA_COLUMN_SIDECAR_SIZE_FULU + { + return None; + } + let o0 = u32_le(buf, 0) as usize; + let o1 = u32_le(buf, 4) as usize; + let o2 = u32_le(buf, 8) as usize; + let o3 = u32_le(buf, 12) as usize; + if o0 != PARTIAL_SIDECAR_FIXED_FULU || o1 < o0 || o2 < o1 || o3 < o2 || o3 > buf.len() { + return None; + } + let (rows, n) = bitlist_u128(&buf[o0..o1], n_rows)?; + let k = rows.count_ones() as usize; + if n != n_rows || o2 - o1 != k * BYTES_PER_CELL || o3 - o2 != k * BYTES_PER_KZG_PROOF { + return None; + } + let header = &buf[o3..]; + if header.is_empty() { + (k > 0).then_some(rows) + } else { + if header.len() < 4 || u32_le(header, 0) != 4 { + return None; + } + PartialDataColumnHeaderView::check_size(&header[4..]).then_some(rows) + } + } + + #[inline] + pub fn cells(buf: &[u8]) -> &[u8] { + &buf[u32_le(buf, 4) as usize..u32_le(buf, 8) as usize] + } + + #[inline] + pub fn proofs(buf: &[u8]) -> &[u8] { + &buf[u32_le(buf, 8) as usize..u32_le(buf, 12) as usize] + } + + /// The header SSZ, or empty when omitted. + #[inline] + pub fn header(buf: &[u8]) -> &[u8] { + let list = &buf[u32_le(buf, 12) as usize..]; + if list.is_empty() { list } else { &list[4..] } + } +} + +// PartialDataColumnSidecar (Gloas): three offsets, no header field. +// [0..4) offset to cells_present_bitmap (== 12) +// [4..8) offset to partial_column +// [8..12) offset to kzg_proofs +pub struct PartialDataColumnSidecarGloasView; + +impl PartialDataColumnSidecarGloasView { + /// As Fulu, but a partial payload must contain at least one cell. + pub fn check_size(buf: &[u8], n_rows: usize) -> Option { + if buf.len() < PARTIAL_SIDECAR_FIXED_GLOAS + 1 || + buf.len() > MAX_PARTIAL_DATA_COLUMN_SIDECAR_SIZE_GLOAS + { + return None; + } + let o0 = u32_le(buf, 0) as usize; + let o1 = u32_le(buf, 4) as usize; + let o2 = u32_le(buf, 8) as usize; + if o0 != PARTIAL_SIDECAR_FIXED_GLOAS || o1 < o0 || o2 < o1 || o2 > buf.len() { + return None; + } + let (rows, n) = bitlist_u128(&buf[o0..o1], n_rows)?; + let k = rows.count_ones() as usize; + if n != n_rows || + k == 0 || + o2 - o1 != k * BYTES_PER_CELL || + buf.len() - o2 != k * BYTES_PER_KZG_PROOF + { + return None; + } + Some(rows) + } + + #[inline] + pub fn cells(buf: &[u8]) -> &[u8] { + &buf[u32_le(buf, 4) as usize..u32_le(buf, 8) as usize] + } + + #[inline] + pub fn proofs(buf: &[u8]) -> &[u8] { + &buf[u32_le(buf, 8) as usize..] + } +} + +pub struct PartialDataColumnHeaderView; + +impl PartialDataColumnHeaderView { + pub fn check_size(buf: &[u8]) -> bool { + if buf.len() < PARTIAL_HEADER_FIXED || u32_le(buf, 0) as usize != PARTIAL_HEADER_FIXED { + return false; + } + let commitments = buf.len() - PARTIAL_HEADER_FIXED; + commitments.is_multiple_of(BYTES_PER_KZG_COMMITMENT) && + commitments / BYTES_PER_KZG_COMMITMENT <= MAX_BLOB_COMMITMENTS_PER_BLOCK + } + + #[inline] + pub fn signed_block_header(buf: &[u8]) -> &[u8; 208] { + super::fixed(buf, 4) + } + + #[inline] + pub fn inclusion_proof(buf: &[u8]) -> &[u8; 128] { + super::fixed(buf, 212) + } + + #[inline] + pub fn kzg_commitments(buf: &[u8]) -> &[u8] { + &buf[PARTIAL_HEADER_FIXED..] + } +} + +// PartialDataColumnPartsMetadata: { available, requests }, both +// bitlists of the block's row count. +// [0..4) offset to available (== 8) +// [4..8) offset to requests +pub struct PartialDataColumnPartsMetadataView; + +impl PartialDataColumnPartsMetadataView { + /// Returns (available, requests); both bitlists must have exactly + /// the trusted `n_rows` bits. + pub fn check_size(buf: &[u8], n_rows: usize) -> Option<(u128, u128)> { + if buf.len() < 10 { + return None; + } + let o0 = u32_le(buf, 0) as usize; + let o1 = u32_le(buf, 4) as usize; + if o0 != 8 || o1 < o0 || o1 > buf.len() { + return None; + } + let (available, n) = bitlist_u128(&buf[o0..o1], n_rows)?; + let (requests, n_req) = bitlist_u128(&buf[o1..], n_rows)?; + (n == n_rows && n_req == n_rows).then_some((available, requests)) + } +} + +pub const fn parts_metadata_len(n_rows: usize) -> usize { + 8 + 2 * bitlist_bytes(n_rows) +} + +pub fn write_parts_metadata(available: u128, requests: u128, n_rows: usize, out: &mut [u8]) { + debug_assert_eq!(out.len(), parts_metadata_len(n_rows)); + let bits = bitlist_bytes(n_rows); + out[0..4].copy_from_slice(&8u32.to_le_bytes()); + out[4..8].copy_from_slice(&((8 + bits) as u32).to_le_bytes()); + write_bitlist_u128(available, n_rows, &mut out[8..8 + bits]); + write_bitlist_u128(requests, n_rows, &mut out[8 + bits..]); +} + +/// Layout of one partial sidecar's SSZ: the offsets-plus-bitmap prefix +/// is written locally; cells, proofs, and any header bytes follow as +/// retained ranges and are never gathered. Construction validates the +/// invariants once, so encoding cannot fail. +#[derive(Clone, Copy)] +pub struct PartialSidecarPlan { + layout: PartialLayout, + rows: u128, + n_rows: usize, +} + +#[derive(Clone, Copy)] +pub enum PartialLayout { + /// `header_bytes` is the header SSZ length; 0 encodes the empty list. + Fulu { + header_bytes: usize, + }, + Gloas, +} + +impl PartialSidecarPlan { + /// `rows` must fit the declared count within the 128-row ceiling, a + /// Fulu header length must be structurally possible, and the fork's + /// payload rule holds: Gloas needs a cell, Fulu a cell or header. + pub fn new(layout: PartialLayout, rows: u128, n_rows: usize) -> Option { + let header_bytes = match layout { + PartialLayout::Fulu { header_bytes } => header_bytes, + PartialLayout::Gloas => 0, + }; + let header_ok = header_bytes == 0 || + (header_bytes >= PARTIAL_HEADER_FIXED && + (header_bytes - PARTIAL_HEADER_FIXED) + .is_multiple_of(BYTES_PER_KZG_COMMITMENT) && + (header_bytes - PARTIAL_HEADER_FIXED) / BYTES_PER_KZG_COMMITMENT <= + MAX_BLOB_COMMITMENTS_PER_BLOCK); + let rows_fit = n_rows <= 128 && (n_rows == 128 || rows >> n_rows == 0); + (header_ok && rows_fit && (rows != 0 || header_bytes > 0)).then_some(Self { + layout, + rows, + n_rows, + }) + } + + #[inline] + pub fn n_rows(&self) -> usize { + self.n_rows + } + + #[inline] + pub fn header_bytes(&self) -> usize { + match self.layout { + PartialLayout::Fulu { header_bytes } => header_bytes, + PartialLayout::Gloas => 0, + } + } + + #[inline] + pub fn cell_count(&self) -> usize { + self.rows.count_ones() as usize + } + + pub fn prefix_len(&self) -> usize { + let fixed = match self.layout { + PartialLayout::Fulu { .. } => PARTIAL_SIDECAR_FIXED_FULU, + PartialLayout::Gloas => PARTIAL_SIDECAR_FIXED_GLOAS, + }; + fixed + bitlist_bytes(self.n_rows) + } + + pub fn ssz_len(&self) -> usize { + let payload = self.cell_count() * (BYTES_PER_CELL + BYTES_PER_KZG_PROOF); + let header = match self.layout { + PartialLayout::Fulu { header_bytes: 0 } | PartialLayout::Gloas => 0, + PartialLayout::Fulu { header_bytes } => 4 + header_bytes, + }; + self.prefix_len() + payload + header + } + + /// Write offsets and bitmap into `out` (exactly `prefix_len` bytes). + pub fn write_prefix(&self, out: &mut [u8]) { + debug_assert_eq!(out.len(), self.prefix_len()); + let k = self.cell_count(); + let o0 = out.len() - bitlist_bytes(self.n_rows); + let o1 = self.prefix_len(); + let o2 = o1 + k * BYTES_PER_CELL; + let o3 = o2 + k * BYTES_PER_KZG_PROOF; + out[0..4].copy_from_slice(&(o0 as u32).to_le_bytes()); + out[4..8].copy_from_slice(&(o1 as u32).to_le_bytes()); + out[8..12].copy_from_slice(&(o2 as u32).to_le_bytes()); + if matches!(self.layout, PartialLayout::Fulu { .. }) { + out[12..16].copy_from_slice(&(o3 as u32).to_le_bytes()); + } + write_bitlist_u128(self.rows, self.n_rows, &mut out[o0..]); + } + + /// Fulu only: the header list's 4-byte inner offset, appended + /// between the proofs and the header bytes when a header is present. + pub const HEADER_LIST_PREFIX: [u8; 4] = 4u32.to_le_bytes(); +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn bitlist_round_trips() { + for (mask, n) in [(0u128, 0usize), (0b1011, 4), (0, 9), (u128::MAX, 128), (1 << 127, 128)] { + let mut buf = vec![0u8; bitlist_bytes(n)]; + write_bitlist_u128(mask, n, &mut buf); + assert_eq!(bitlist_u128(&buf, n), Some((mask, n)), "mask {mask:#x} n {n}"); + assert_eq!(bitlist_u128(&buf, 128), Some((mask, n))); + } + } + + #[test] + fn bitlist_rejects_non_canonical_and_oversized() { + assert!(bitlist_u128(&[], 128).is_none()); + assert!(bitlist_u128(&[0], 128).is_none(), "missing delimiter"); + assert!(bitlist_u128(&[0b11, 0], 128).is_none(), "trailing zero byte"); + let mut full = vec![0u8; bitlist_bytes(9)]; + write_bitlist_u128(0x1ff, 9, &mut full); + assert!(bitlist_u128(&full, 8).is_none(), "longer than trusted rows"); + let mut over = vec![0u8; bitlist_bytes(129)]; + over[16] = 2; + assert!(bitlist_u128(&over, 4096).is_none(), "beyond u128 rows"); + } + + #[test] + fn group_ids_match_spec_layout() { + let root = [0xab; 32]; + let fulu = fulu_group_id(&root); + assert_eq!(fulu[0], 0); + assert_eq!(&fulu[1..], &root); + let gloas = gloas_group_id(&root, 0x0102_0304); + assert_eq!(gloas[0], 0); + assert_eq!(&gloas[1..33], &root); + assert_eq!(gloas[33..], 0x0102_0304u64.to_le_bytes()); + } + + fn build(plan: &PartialSidecarPlan, header: &[u8]) -> Vec { + let k = plan.cell_count(); + let mut buf = vec![0u8; plan.prefix_len()]; + plan.write_prefix(&mut buf); + buf.extend_from_slice(&vec![0x11; k * BYTES_PER_CELL]); + buf.extend_from_slice(&vec![0x22; k * BYTES_PER_KZG_PROOF]); + if !header.is_empty() { + buf.extend_from_slice(&PartialSidecarPlan::HEADER_LIST_PREFIX); + buf.extend_from_slice(header); + } + assert_eq!(buf.len(), plan.ssz_len()); + buf + } + + fn test_header(commitments: usize) -> Vec { + let mut h = vec![0u8; PARTIAL_HEADER_FIXED + commitments * BYTES_PER_KZG_COMMITMENT]; + h[0..4].copy_from_slice(&(PARTIAL_HEADER_FIXED as u32).to_le_bytes()); + assert!(PartialDataColumnHeaderView::check_size(&h)); + h + } + + fn fulu_plan(header_bytes: usize, rows: u128, n_rows: usize) -> PartialSidecarPlan { + PartialSidecarPlan::new(PartialLayout::Fulu { header_bytes }, rows, n_rows).unwrap() + } + + #[test] + fn plan_rejects_unencodable_inputs() { + assert!(PartialSidecarPlan::new(PartialLayout::Gloas, 0b1, 129).is_none()); + assert!(PartialSidecarPlan::new(PartialLayout::Gloas, 0b100, 2).is_none()); + assert!(PartialSidecarPlan::new(PartialLayout::Gloas, 0, 2).is_none()); + assert!(PartialSidecarPlan::new(PartialLayout::Fulu { header_bytes: 0 }, 0, 2).is_none()); + assert!( + PartialSidecarPlan::new(PartialLayout::Fulu { header_bytes: 12 }, 0b1, 2).is_none() + ); + assert!( + PartialSidecarPlan::new( + PartialLayout::Fulu { header_bytes: PARTIAL_HEADER_FIXED + 1 }, + 0b1, + 2 + ) + .is_none() + ); + assert!(PartialSidecarPlan::new(PartialLayout::Gloas, 1 << 127, 128).is_some()); + assert!( + PartialSidecarPlan::new( + PartialLayout::Fulu { header_bytes: PARTIAL_HEADER_FIXED }, + 0, + 2 + ) + .is_some() + ); + } + + #[test] + fn fulu_sidecar_round_trips_with_and_without_header() { + let plan = fulu_plan(0, 0b10, 2); + let buf = build(&plan, &[]); + assert_eq!(PartialDataColumnSidecarFuluView::check_size(&buf, 2), Some(0b10)); + assert_eq!(PartialDataColumnSidecarFuluView::cells(&buf), &[0x11; BYTES_PER_CELL]); + assert_eq!(PartialDataColumnSidecarFuluView::proofs(&buf), &[0x22; BYTES_PER_KZG_PROOF]); + assert!(PartialDataColumnSidecarFuluView::header(&buf).is_empty()); + + let header = test_header(2); + let plan = fulu_plan(header.len(), 0, 2); + let buf = build(&plan, &header); + assert_eq!(PartialDataColumnSidecarFuluView::check_size(&buf, 2), Some(0)); + assert!(PartialDataColumnSidecarFuluView::check_size(&buf, 3).is_none()); + assert_eq!(PartialDataColumnSidecarFuluView::header(&buf), &header[..]); + } + + #[test] + fn fulu_rejects_cell_less_sidecar_without_header() { + // Hand-encoded: such a plan is unconstructible by design. + let mut buf = vec![0u8; 17]; + buf[0..4].copy_from_slice(&16u32.to_le_bytes()); + buf[4..8].copy_from_slice(&17u32.to_le_bytes()); + buf[8..12].copy_from_slice(&17u32.to_le_bytes()); + buf[12..16].copy_from_slice(&17u32.to_le_bytes()); + buf[16] = 0b100; // n = 2, no rows set + assert!(PartialDataColumnSidecarFuluView::check_size(&buf, 2).is_none()); + } + + #[test] + fn fulu_rejects_length_and_offset_corruption() { + let plan = fulu_plan(0, 0b1, 1); + let good = build(&plan, &[]); + assert!(PartialDataColumnSidecarFuluView::check_size(&good, 1).is_some()); + + let mut short = good.clone(); + short.pop(); + assert!(PartialDataColumnSidecarFuluView::check_size(&short, 1).is_none()); + + let mut bad_first = good.clone(); + bad_first[0] = 17; + assert!(PartialDataColumnSidecarFuluView::check_size(&bad_first, 1).is_none()); + + let mut crossed = good; + crossed[4..8].copy_from_slice(&u32::MAX.to_le_bytes()); + assert!(PartialDataColumnSidecarFuluView::check_size(&crossed, 1).is_none()); + } + + #[test] + fn gloas_sidecar_round_trips_and_requires_a_cell() { + let plan = PartialSidecarPlan::new(PartialLayout::Gloas, 0b101, 3).unwrap(); + let buf = build(&plan, &[]); + assert_eq!(PartialDataColumnSidecarGloasView::check_size(&buf, 3), Some(0b101)); + assert!(PartialDataColumnSidecarGloasView::check_size(&buf, 4).is_none()); + assert_eq!(PartialDataColumnSidecarGloasView::cells(&buf).len(), 2 * BYTES_PER_CELL); + assert_eq!(PartialDataColumnSidecarGloasView::proofs(&buf).len(), 2 * BYTES_PER_KZG_PROOF); + + // Hand-encoded cell-less payload: unconstructible as a plan. + let mut empty = vec![0u8; 13]; + empty[0..4].copy_from_slice(&12u32.to_le_bytes()); + empty[4..8].copy_from_slice(&13u32.to_le_bytes()); + empty[8..12].copy_from_slice(&13u32.to_le_bytes()); + empty[12] = 0b1000; // n = 3, no rows set + assert!(PartialDataColumnSidecarGloasView::check_size(&empty, 3).is_none()); + } + + #[test] + fn parts_metadata_round_trips_and_rejects_length_mismatch() { + let mut buf = vec![0u8; parts_metadata_len(6)]; + write_parts_metadata(0b110000, 0b1010, 6, &mut buf); + assert_eq!( + PartialDataColumnPartsMetadataView::check_size(&buf, 6), + Some((0b110000, 0b1010)) + ); + assert!(PartialDataColumnPartsMetadataView::check_size(&buf, 5).is_none()); + assert!(PartialDataColumnPartsMetadataView::check_size(&buf, 7).is_none()); + + // requests bitlist shorter than available: reject. + let mut mismatch = vec![0u8; 8 + bitlist_bytes(9) + bitlist_bytes(2)]; + mismatch[0..4].copy_from_slice(&8u32.to_le_bytes()); + mismatch[4..8].copy_from_slice(&((8 + bitlist_bytes(9)) as u32).to_le_bytes()); + write_bitlist_u128(0, 9, &mut mismatch[8..8 + bitlist_bytes(9)]); + write_bitlist_u128(0, 2, &mut mismatch[8 + bitlist_bytes(9)..]); + assert!(PartialDataColumnPartsMetadataView::check_size(&mismatch, 16).is_none()); + assert!(PartialDataColumnPartsMetadataView::check_size(&mismatch, 9).is_none()); + } + + /// Byte layouts written out by hand from the partial-columns specs, + /// independent of the encoders under test. + #[test] + fn encoders_match_spec_derived_fixtures() { + // Fulu: n = 2 rows, row 0 present. Offsets 16/17/2065/2113; + // bitmap 0b101 = row 0 plus the delimiter at bit 2. + let mut fulu = vec![ + 16, 0, 0, 0, // + 17, 0, 0, 0, // + 0x11, 0x08, 0, 0, // + 0x41, 0x08, 0, 0, // + 0b101, + ]; + let plan = fulu_plan(0, 0b01, 2); + let mut prefix = vec![0u8; plan.prefix_len()]; + plan.write_prefix(&mut prefix); + assert_eq!(prefix, fulu); + fulu.extend_from_slice(&[0xCC; BYTES_PER_CELL]); + fulu.extend_from_slice(&[0xDD; BYTES_PER_KZG_PROOF]); + assert_eq!(PartialDataColumnSidecarFuluView::check_size(&fulu, 2), Some(0b01)); + assert_eq!(PartialDataColumnSidecarFuluView::cells(&fulu), &[0xCC; BYTES_PER_CELL]); + assert_eq!(PartialDataColumnSidecarFuluView::proofs(&fulu), &[0xDD; BYTES_PER_KZG_PROOF]); + + // Gloas: n = 1 row, present. Offsets 12/13/2061; bitmap 0b11. + let mut gloas = vec![ + 12, 0, 0, 0, // + 13, 0, 0, 0, // + 0x0D, 0x08, 0, 0, // + 0b11, + ]; + let plan = PartialSidecarPlan::new(PartialLayout::Gloas, 0b1, 1).unwrap(); + let mut prefix = vec![0u8; plan.prefix_len()]; + plan.write_prefix(&mut prefix); + assert_eq!(prefix, gloas); + gloas.extend_from_slice(&[0xCC; BYTES_PER_CELL]); + gloas.extend_from_slice(&[0xDD; BYTES_PER_KZG_PROOF]); + assert_eq!(PartialDataColumnSidecarGloasView::check_size(&gloas, 1), Some(0b1)); + + // Metadata: n = 3, available 0b101, requests 0b010; each + // bitlist gains the delimiter at bit 3. + let metadata = [8, 0, 0, 0, 9, 0, 0, 0, 0b1101, 0b1010]; + let mut ours = vec![0u8; parts_metadata_len(3)]; + write_parts_metadata(0b101, 0b010, 3, &mut ours); + assert_eq!(ours, metadata); + assert_eq!( + PartialDataColumnPartsMetadataView::check_size(&metadata, 3), + Some((0b101, 0b010)) + ); + } +}