diff --git a/Cargo.lock b/Cargo.lock index 2b70208..af1646d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -706,8 +706,6 @@ dependencies = [ "num-traits", "once_cell", "oorandom", - "plotters", - "rayon", "regex", "serde", "serde_derive", @@ -741,16 +739,6 @@ dependencies = [ "crossbeam-utils", ] -[[package]] -name = "crossbeam-deque" -version = "0.8.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9dd111b7b7f7d55b72c0a6ae361660ee5853c9af73f70c3c2ef6858b950e2e51" -dependencies = [ - "crossbeam-epoch", - "crossbeam-utils", -] - [[package]] name = "crossbeam-epoch" version = "0.9.18" @@ -3088,34 +3076,6 @@ dependencies = [ "time", ] -[[package]] -name = "plotters" -version = "0.3.7" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5aeb6f403d7a4911efb1e33402027fc44f29b5bf6def3effcc22d7bb75f2b747" -dependencies = [ - "num-traits", - "plotters-backend", - "plotters-svg", - "wasm-bindgen", - "web-sys", -] - -[[package]] -name = "plotters-backend" -version = "0.3.7" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "df42e13c12958a16b3f7f4386b9ab1f3e7933914ecea48da7139435263a4172a" - -[[package]] -name = "plotters-svg" -version = "0.3.7" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "51bae2ac328883f7acdfea3d66a7c35751187f870bc81f94563733a154d7a670" -dependencies = [ - "plotters-backend", -] - [[package]] name = "polyval" version = "0.6.2" @@ -3536,26 +3496,6 @@ dependencies = [ "rand_core 0.9.5", ] -[[package]] -name = "rayon" -version = "1.12.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fb39b166781f92d482534ef4b4b1b2568f42613b53e5b6c160e24cfbfa30926d" -dependencies = [ - "either", - "rayon-core", -] - -[[package]] -name = "rayon-core" -version = "1.13.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "22e18b0f0062d30d4230b2e85ff77fdfe4326feb054b9783a3460d8435c8ab91" -dependencies = [ - "crossbeam-deque", - "crossbeam-utils", -] - [[package]] name = "rcgen" version = "0.14.8" @@ -4331,7 +4271,6 @@ name = "swactor" version = "0.1.0" dependencies = [ "atomic-waker", - "criterion", "crossbeam-queue", "crossbeam-utils", "getrandom 0.2.17", @@ -4346,6 +4285,17 @@ dependencies = [ "web-time 0.2.4", ] +[[package]] +name = "swactor-benchmarks" +version = "0.1.0" +dependencies = [ + "criterion", + "serde", + "swactor", + "swactor-transport", + "telemetry", +] + [[package]] name = "swactor-engine" version = "0.1.0" @@ -4396,6 +4346,7 @@ name = "swactor-transport" version = "0.1.0" dependencies = [ "ed25519-dalek 2.2.0", + "proptest", "rand_core 0.6.4", "serde", "serde_json", @@ -4497,6 +4448,7 @@ dependencies = [ "futures-channel", "iroh", "libc", + "proptest", "serde", "serde_json", "swactor", diff --git a/Cargo.toml b/Cargo.toml index 10cdcad..36387fe 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -14,6 +14,7 @@ members = [ "xtask", "tools/vastai", "tools/actor-control-flow-lint", + "tools/benchmarks", ] default-members = [ ".", @@ -71,7 +72,6 @@ crossbeam-utils = "0.8.21" parking_lot = "0.12" [dev-dependencies] -criterion = { version = "0.5", features = ["html_reports"] } proptest = "1" proptest-state-machine = "0.3" loom = "0.7" diff --git a/apps/myelin/src/node/actor.rs b/apps/myelin/src/node/actor.rs index 37df28e..0a107b5 100644 --- a/apps/myelin/src/node/actor.rs +++ b/apps/myelin/src/node/actor.rs @@ -809,6 +809,10 @@ impl From<&stage::StageLifecycleEvent> for StageLifecycleWire { } pub(crate) fn register_codecs(registry: &mut CodecRegistry) { - registry.register::(JsonCodec::::default()); - registry.register::(JsonCodec::::default()); + registry + .register::(JsonCodec::::default()) + .expect("unique codec registration"); + registry + .register::(JsonCodec::::default()) + .expect("unique codec registration"); } diff --git a/apps/myelin/src/orchestration/actor.rs b/apps/myelin/src/orchestration/actor.rs index a7452c3..7fbed0d 100644 --- a/apps/myelin/src/orchestration/actor.rs +++ b/apps/myelin/src/orchestration/actor.rs @@ -523,8 +523,12 @@ impl From<&core::LifecycleEvent> for LifecycleEventWire { } pub(crate) fn register_codecs(registry: &mut CodecRegistry) { - registry.register::(JsonCodec::::default()); - registry.register::(JsonCodec::::default()); + registry + .register::(JsonCodec::::default()) + .expect("unique codec registration"); + registry + .register::(JsonCodec::::default()) + .expect("unique codec registration"); } impl From<&core::TokenObjectPayload> for TokenObjectPayloadWire { diff --git a/apps/myelin/src/orchestration/manual_control.rs b/apps/myelin/src/orchestration/manual_control.rs index b882acd..315c8f9 100644 --- a/apps/myelin/src/orchestration/manual_control.rs +++ b/apps/myelin/src/orchestration/manual_control.rs @@ -1392,7 +1392,9 @@ impl NetworkMessage for ManualControlReply { } pub(crate) fn register_codecs(registry: &mut CodecRegistry) { - registry.register::(JsonCodec::::default()); + registry + .register::(JsonCodec::::default()) + .expect("unique codec registration"); } struct NodeLane { diff --git a/crates/data-plane/src/blob_transfer.rs b/crates/data-plane/src/blob_transfer.rs index 66634b1..406a42e 100644 --- a/crates/data-plane/src/blob_transfer.rs +++ b/crates/data-plane/src/blob_transfer.rs @@ -99,5 +99,7 @@ impl fmt::Display for BlobTransferFailure { impl std::error::Error for BlobTransferFailure {} pub fn register_blob_transfer_codecs(registry: &mut CodecRegistry) { - registry.register::(JsonCodec::default()); + registry + .register::(JsonCodec::default()) + .expect("unique codec registration"); } diff --git a/crates/data-plane/src/namespace.rs b/crates/data-plane/src/namespace.rs index 9523ca3..284a83b 100644 --- a/crates/data-plane/src/namespace.rs +++ b/crates/data-plane/src/namespace.rs @@ -1494,6 +1494,10 @@ impl DirectoryClient { } pub fn register_namespace_codecs(registry: &mut CodecRegistry) { - registry.register::(JsonCodec::default()); - registry.register::(JsonCodec::default()); + registry + .register::(JsonCodec::default()) + .expect("unique codec registration"); + registry + .register::(JsonCodec::default()) + .expect("unique codec registration"); } diff --git a/crates/data-plane/src/protocol.rs b/crates/data-plane/src/protocol.rs index 3ad0294..3982263 100644 --- a/crates/data-plane/src/protocol.rs +++ b/crates/data-plane/src/protocol.rs @@ -367,9 +367,15 @@ impl NetworkMessage for HostStreamIn { } pub fn register_data_plane_codecs(registry: &mut CodecRegistry) { - registry.register::(JsonCodec::default()); - registry.register::(JsonCodec::default()); - registry.register::(JsonCodec::default()); + registry + .register::(JsonCodec::default()) + .expect("unique codec registration"); + registry + .register::(JsonCodec::default()) + .expect("unique codec registration"); + registry + .register::(JsonCodec::default()) + .expect("unique codec registration"); crate::namespace::register_namespace_codecs(registry); crate::blob_transfer::register_blob_transfer_codecs(registry); crate::source::register_blob_source_codecs(registry); diff --git a/crates/data-plane/src/source.rs b/crates/data-plane/src/source.rs index 60824c0..e482072 100644 --- a/crates/data-plane/src/source.rs +++ b/crates/data-plane/src/source.rs @@ -295,5 +295,7 @@ impl ActorInterface for FileBlobSourceActor { } pub fn register_blob_source_codecs(registry: &mut CodecRegistry) { - registry.register::(JsonCodec::default()); + registry + .register::(JsonCodec::default()) + .expect("unique codec registration"); } diff --git a/crates/distribution/src/messages.rs b/crates/distribution/src/messages.rs index babfbc9..06b0145 100644 --- a/crates/distribution/src/messages.rs +++ b/crates/distribution/src/messages.rs @@ -167,18 +167,27 @@ impl NetworkMessage for DirectoryGossip { /// Build a `CodecRegistry` with all distribution protocol messages registered. pub fn distribution_codec_registry() -> CodecRegistry { let mut cr = CodecRegistry::new(); - cr.register::(JsonCodec::::default()); - cr.register::(JsonCodec::::default()); - cr.register::(JsonCodec::::default()); + cr.register::(JsonCodec::::default()) + .expect("unique codec registration"); + cr.register::(JsonCodec::::default()) + .expect("unique codec registration"); + cr.register::(JsonCodec::::default()) + .expect("unique codec registration"); // §6.1 / §14.5: `IndirectAck` is folded into the shared registry so all six // SWIM message types decode through one uniform path; concrete drivers no // longer need to hand-dispatch it by tag. - cr.register::(JsonCodec::::default()); - cr.register::(JsonCodec::::default()); - cr.register::(JsonCodec::::default()); - cr.register::(JsonCodec::::default()); - cr.register::(JsonCodec::::default()); - cr.register::(JsonCodec::::default()); + cr.register::(JsonCodec::::default()) + .expect("unique codec registration"); + cr.register::(JsonCodec::::default()) + .expect("unique codec registration"); + cr.register::(JsonCodec::::default()) + .expect("unique codec registration"); + cr.register::(JsonCodec::::default()) + .expect("unique codec registration"); + cr.register::(JsonCodec::::default()) + .expect("unique codec registration"); + cr.register::(JsonCodec::::default()) + .expect("unique codec registration"); cr } @@ -228,25 +237,32 @@ pub fn actor_codec_registry() -> CodecRegistry { } }; Ok((tag.to_string(), bytes)) - }); + }) + .expect("unique codec encoder registration"); // A generic free fn (not a closure — closures are monomorphic and we decode // into six different inner types). fn d Deserialize<'de>>(bytes: &[u8]) -> Result { serde_json::from_slice(bytes).map_err(|e| Error::from(format!("decode: {e}"))) } - cr.register_decoder::("swactor_dist::Ping", |b| Ok(SwimIn::Ping(d(b)?))); - cr.register_decoder::("swactor_dist::Ack", |b| Ok(SwimIn::Ack(d(b)?))); - cr.register_decoder::("swactor_dist::PingReq", |b| Ok(SwimIn::PingReq(d(b)?))); + cr.register_decoder::("swactor_dist::Ping", |b| Ok(SwimIn::Ping(d(b)?))) + .expect("unique codec decoder registration"); + cr.register_decoder::("swactor_dist::Ack", |b| Ok(SwimIn::Ack(d(b)?))) + .expect("unique codec decoder registration"); + cr.register_decoder::("swactor_dist::PingReq", |b| Ok(SwimIn::PingReq(d(b)?))) + .expect("unique codec decoder registration"); cr.register_decoder::("swactor_dist::IndirectAck", |b| { Ok(SwimIn::IndirectAck(d(b)?)) - }); + }) + .expect("unique codec decoder registration"); cr.register_decoder::("swactor_dist::JoinRequest", |b| { Ok(SwimIn::JoinRequest(d(b)?)) - }); + }) + .expect("unique codec decoder registration"); cr.register_decoder::("swactor_dist::JoinResponse", |b| { Ok(SwimIn::JoinResponse(d(b)?)) - }); + }) + .expect("unique codec decoder registration"); // ── Registry: RegistryIn::Gossip ⇄ RegistryGossip frame ── use crate::registry_actor::RegistryIn; @@ -256,10 +272,12 @@ pub fn actor_codec_registry() -> CodecRegistry { serde_json::to_vec(g).map_err(|e| Error::from(format!("encode: {e}")))?, )), _ => Err(Error::from("RegistryIn: only Gossip is network-encodable")), - }); + }) + .expect("unique codec encoder registration"); cr.register_decoder::(RegistryGossip::type_tag(), |b| { Ok(RegistryIn::Gossip(d(b)?)) - }); + }) + .expect("unique codec decoder registration"); // ── Metadata: MetadataIn::Gossip ⇄ MetadataGossip frame ── use crate::node_metadata_actor::MetadataIn; @@ -269,10 +287,12 @@ pub fn actor_codec_registry() -> CodecRegistry { serde_json::to_vec(g).map_err(|e| Error::from(format!("encode: {e}")))?, )), _ => Err(Error::from("MetadataIn: only Gossip is network-encodable")), - }); + }) + .expect("unique codec encoder registration"); cr.register_decoder::(MetadataGossip::type_tag(), |b| { Ok(MetadataIn::Gossip(d(b)?)) - }); + }) + .expect("unique codec decoder registration"); // ── Directory: DirectoryIn::Gossip ⇄ DirectoryGossip frame ── use crate::directory_actor::DirectoryIn; @@ -282,10 +302,12 @@ pub fn actor_codec_registry() -> CodecRegistry { serde_json::to_vec(g).map_err(|e| Error::from(format!("encode: {e}")))?, )), _ => Err(Error::from("DirectoryIn: only Gossip is network-encodable")), - }); + }) + .expect("unique codec encoder registration"); cr.register_decoder::(DirectoryGossip::type_tag(), |b| { Ok(DirectoryIn::Gossip(d(b)?)) - }); + }) + .expect("unique codec decoder registration"); cr } diff --git a/crates/distribution/tests/routing.rs b/crates/distribution/tests/routing.rs index 47f0702..4b69866 100644 --- a/crates/distribution/tests/routing.rs +++ b/crates/distribution/tests/routing.rs @@ -739,15 +739,20 @@ mod directory_route_path { // The shared codec carries the directory's gossip frame *and* the app // protocol — the app registers its own type, exactly as a real app would. let mut codec = actor_codec_registry(); - codec.register_encoder::(|h: &Hello| { - Ok(( - Hello::type_tag().to_string(), - serde_json::to_vec(h).map_err(|e| Error::from(format!("encode: {e}")))?, - )) - }); - codec.register_decoder::(Hello::type_tag(), |b: &[u8]| { - serde_json::from_slice::(b).map_err(|e| Error::from(format!("decode: {e}"))) - }); + codec + .register_encoder::(|h: &Hello| { + Ok(( + Hello::type_tag().to_string(), + serde_json::to_vec(h).map_err(|e| Error::from(format!("encode: {e}")))?, + )) + }) + .expect("unique codec encoder registration"); + codec + .register_decoder::(Hello::type_tag(), |b: &[u8]| { + serde_json::from_slice::(b) + .map_err(|e| Error::from(format!("decode: {e}"))) + }) + .expect("unique codec decoder registration"); let codec = Arc::new(codec); let keys: Vec = (0..n).map(|_| Keypair::generate()).collect(); diff --git a/crates/job-runner/src/wire.rs b/crates/job-runner/src/wire.rs index b737436..be5095d 100644 --- a/crates/job-runner/src/wire.rs +++ b/crates/job-runner/src/wire.rs @@ -155,8 +155,16 @@ impl NetworkMessage for OutputChunk { /// Register the job wire messages with a codec registry (JSON). pub fn register_job_codecs(registry: &mut CodecRegistry) { - registry.register::>(JsonCodec::default()); - registry.register::>(JsonCodec::default()); - registry.register::>(JsonCodec::default()); - registry.register::>(JsonCodec::default()); + registry + .register::>(JsonCodec::default()) + .expect("unique codec registration"); + registry + .register::>(JsonCodec::default()) + .expect("unique codec registration"); + registry + .register::>(JsonCodec::default()) + .expect("unique codec registration"); + registry + .register::>(JsonCodec::default()) + .expect("unique codec registration"); } diff --git a/crates/telemetry/Cargo.toml b/crates/telemetry/Cargo.toml index eea19a7..91fc7bb 100644 --- a/crates/telemetry/Cargo.toml +++ b/crates/telemetry/Cargo.toml @@ -17,5 +17,8 @@ crossbeam-channel = "0.5" [target.'cfg(target_os = "linux")'.dependencies] libc = "0.2" +[dev-dependencies] +proptest = "1" + [lints] workspace = true diff --git a/crates/telemetry/src/publisher_actor.rs b/crates/telemetry/src/publisher_actor.rs index 7f5b323..d4bb659 100644 --- a/crates/telemetry/src/publisher_actor.rs +++ b/crates/telemetry/src/publisher_actor.rs @@ -76,5 +76,7 @@ impl ActorInterface for TelemetryPublisherActor { /// Register JSON encoding for remote telemetry publisher messages. pub fn register_telemetry_publisher_codec(registry: &mut CodecRegistry) { - registry.register::(JsonCodec::::default()); + registry + .register::(JsonCodec::::default()) + .expect("unique codec registration"); } diff --git a/crates/telemetry/src/wire.rs b/crates/telemetry/src/wire.rs index 7d5ecc3..9ed277b 100644 --- a/crates/telemetry/src/wire.rs +++ b/crates/telemetry/src/wire.rs @@ -87,7 +87,8 @@ impl Codec for TelemetryFrameCodec { /// Register the [`TelemetryFrame`] codec. pub fn register_telemetry_codec(cr: &mut CodecRegistry) { - cr.register::(TelemetryFrameCodec); + cr.register::(TelemetryFrameCodec) + .expect("unique codec registration"); } /// JSON event envelope for actor/control paths that can tolerate metadata size. diff --git a/crates/telemetry/tests/t_telemetry.rs b/crates/telemetry/tests/t_telemetry.rs index 69bd86b..50b61dd 100644 --- a/crates/telemetry/tests/t_telemetry.rs +++ b/crates/telemetry/tests/t_telemetry.rs @@ -56,9 +56,9 @@ fn mux_assigns_positions_when_drained() { assert_eq!(mux.assigned(), 0); assert_eq!(mux.dropped(), 0); - let mut positions: Vec = mux.drain().iter().map(|frame| frame.position.0).collect(); - positions.sort_unstable(); - assert_eq!(positions, (0..64).collect::>()); + let positions: Vec = mux.drain().iter().map(|frame| frame.position.0).collect(); + assert_eq!(positions.len(), 64); + assert!(positions.windows(2).all(|pair| pair[1] == pair[0] + 1)); assert_eq!(mux.assigned(), 64); } @@ -88,9 +88,8 @@ fn mux_concurrent_producers_assign_unique_positions_on_drain() { assert_eq!(mux.assigned(), 0); let frames = mux.drain(); assert_eq!(frames.len(), 128); - let mut positions: Vec = frames.iter().map(|frame| frame.position.0).collect(); - positions.sort_unstable(); - assert_eq!(positions, (0..128).collect::>()); + let positions: Vec = frames.iter().map(|frame| frame.position.0).collect(); + assert!(positions.windows(2).all(|pair| pair[1] == pair[0] + 1)); assert_eq!(mux.assigned(), 128); } @@ -106,24 +105,10 @@ fn mux_full_queue_drops_without_consuming_position() { let frames = mux.drain(); assert_eq!(frames.len(), 1); assert_eq!(frames[0].channel, LOG_CHANNEL); - assert_eq!(frames[0].position, Position(0)); assert_eq!(frames[0].payload, b"first"); assert_eq!(mux.assigned(), 1); } -#[test] -fn mux_one_submit_one_drained_frame_without_timing_sidecar() { - let mux = Mux::unbounded(stream()); - - assert!(mux.submit(RESOURCE_CHANNEL, resource(0).encode())); - let frames = mux.drain(); - - assert_eq!(frames.len(), 1); - assert_eq!(frames[0].channel, RESOURCE_CHANNEL); - assert_eq!(frames[0].position, Position(0)); - assert_eq!(frames[0].payload, resource(0).encode()); -} - #[test] fn wire_envelope_round_trips_numeric_channel_payload_and_position() { let stream = StreamId::new(NodeId::new("node-ünïcode-Ω"), Lifetime(7)); diff --git a/crates/telemetry/tests/t_telemetry_endpoint.rs b/crates/telemetry/tests/t_telemetry_endpoint.rs index 4dfb11a..77453d9 100644 --- a/crates/telemetry/tests/t_telemetry_endpoint.rs +++ b/crates/telemetry/tests/t_telemetry_endpoint.rs @@ -5,8 +5,8 @@ use swactor::actor::ActorAddress; use swactor::stats::{ActorSnapshot, StatsSnapshotKind}; use telemetry::frame::{FrameDelivery, TelemetryEvent}; use telemetry::{ - ChannelContent, ChannelContentKind, ChannelFilter, ChannelId, Lifetime, NodeId, Position, - Record, SourceFilter, StreamId, SubscriptionRequest, TelemetryEndpoint, + ChannelContent, ChannelContentKind, ChannelFilter, ChannelId, Lifetime, NodeId, Record, + SourceFilter, StreamId, SubscriptionRequest, TelemetryEndpoint, }; #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] @@ -50,44 +50,6 @@ fn endpoint_without_subscribers_drains_to_bitbucket() { assert_eq!(endpoint.bitbucketed(), 1); } -#[test] -fn channel_registration_allocates_numeric_ids() { - let endpoint = endpoint(); - - let stdout = endpoint.register_channel("stdout", ChannelContent::TextStream); - let stderr = endpoint.register_channel("stderr", ChannelContent::TextStream); - let runtime = endpoint.register_record::(); - - assert_eq!(stdout, ChannelId(1)); - assert_eq!(stderr, ChannelId(2)); - assert_eq!(runtime, ChannelId(3)); - let catalog = endpoint.catalog_snapshot(); - let removed_timing_name = ["telemetry", "frame_time"].join("."); - assert!( - !catalog - .channels - .values() - .any(|descriptor| descriptor.name == removed_timing_name) - ); -} - -#[test] -fn duplicate_channel_registration_rejects_conflicting_content() { - let endpoint = endpoint(); - - let first = endpoint.register_channel("stdout", ChannelContent::TextStream); - let duplicate = endpoint.register_channel("stdout", ChannelContent::TextStream); - let conflict = endpoint.try_register_channel( - "stdout", - ChannelContent::JsonRecord { - schema: Some("stdout.json".to_owned()), - }, - ); - - assert_eq!(first, duplicate); - assert!(conflict.is_err()); -} - #[test] fn subscription_snapshot_contains_stream_and_channel_metadata() { let endpoint = endpoint(); @@ -131,7 +93,6 @@ fn subscription_receives_only_future_matching_frames() { let delivery = frame_event(&events[0]); assert_eq!(delivery.channel.stream, stream()); assert_eq!(delivery.channel.channel, log); - assert_eq!(delivery.position, Position(1)); assert_eq!(delivery.payload, b"visible"); } @@ -181,34 +142,11 @@ fn endpoint_fans_out_ordered_frames_to_multiple_subscribers() { assert_eq!(tick.drained, 3); assert_eq!(tick.subscribers, 2); assert_eq!(tick.delivered, 6); - assert_eq!(positions(&left.drain_available()), vec![0, 1, 2]); - assert_eq!(positions(&right.drain_available()), vec![0, 1, 2]); -} - -#[test] -fn slow_subscriber_drops_without_blocking_fast_subscriber() { - let endpoint = endpoint(); - let producer = endpoint.producer(); - let log = producer.register_channel("runtime.log", ChannelContent::TextStream); - let slow = endpoint.subscribe_all_with_capacity("slow", 1); - let fast = endpoint.subscribe_all_with_capacity("fast", 8); - - for n in 0..4 { - producer.submit_text(log, format!("line-{n}")); - } - let tick = endpoint.tick(); - - assert_eq!(tick.drained, 4); - assert_eq!(tick.delivered, 5); - assert_eq!(tick.dropped_for_subscribers, 3); - assert_eq!(positions(&slow.drain_available()), vec![0]); - assert_eq!(positions(&fast.drain_available()), vec![0, 1, 2, 3]); - let slow_snapshot = endpoint - .subscriber_snapshots() - .into_iter() - .find(|snapshot| snapshot.name == "slow") - .expect("slow subscriber snapshot"); - assert_eq!(slow_snapshot.dropped, 3); + let left_positions = positions(&left.drain_available()); + let right_positions = positions(&right.drain_available()); + assert_eq!(left_positions, right_positions); + assert_eq!(left_positions.len(), 3); + assert!(left_positions.windows(2).all(|pair| pair[1] == pair[0] + 1)); } #[test] diff --git a/crates/telemetry/tests/telemetry_invariants.rs b/crates/telemetry/tests/telemetry_invariants.rs new file mode 100644 index 0000000..3b088ae --- /dev/null +++ b/crates/telemetry/tests/telemetry_invariants.rs @@ -0,0 +1,903 @@ +use std::collections::{BTreeMap, HashMap, HashSet, VecDeque}; + +use proptest::prelude::*; +use telemetry::frame::{Frame, FrameDelivery, TelemetryEvent}; +use telemetry::ingest::Consumer; +use telemetry::transport::Delivery; +use telemetry::wire::{WireError, decode_delivery, encode_delivery}; +use telemetry::{ + ChannelContent, ChannelDescriptor, ChannelId, ChannelRegistrationError, Lifetime, Position, + StreamId, TelemetryEndpoint, TelemetrySubscription, +}; + +#[derive(Clone, Debug)] +enum Action { + RegisterFresh { + suffix: String, + kind: u8, + }, + RegisterSame { + slot: usize, + }, + Submit { + channel: usize, + tail: Vec, + }, + Tick, + Subscribe { + capacity: usize, + }, + DrainSubscriber { + slot: usize, + }, + DropSubscriber { + slot: usize, + }, + IngestNew { + stream: u8, + position: u16, + channel: u16, + tail: Vec, + }, + IngestDuplicate { + slot: usize, + }, + InspectCatalog, + InspectStore, +} + +fn action_strategy() -> impl Strategy { + prop_oneof![ + 2 => (any::(), any::()) + .prop_map(|(suffix, kind)| Action::RegisterFresh { suffix, kind }), + 1 => any::().prop_map(|slot| Action::RegisterSame { slot }), + 4 => (any::(), prop::collection::vec(any::(), 0..129)) + .prop_map(|(channel, tail)| Action::Submit { channel, tail }), + 3 => Just(Action::Tick), + 2 => any::().prop_map(|capacity| Action::Subscribe { capacity }), + 2 => any::().prop_map(|slot| Action::DrainSubscriber { slot }), + 1 => any::().prop_map(|slot| Action::DropSubscriber { slot }), + 3 => ( + any::(), + any::(), + any::(), + prop::collection::vec(any::(), 0..129), + ) + .prop_map(|(stream, position, channel, tail)| Action::IngestNew { + stream, + position, + channel, + tail, + }), + 2 => any::().prop_map(|slot| Action::IngestDuplicate { slot }), + 1 => Just(Action::InspectCatalog), + 1 => Just(Action::InspectStore), + ] +} + +#[derive(Clone, Debug)] +struct Submission { + channel: ChannelId, + payload: Vec, + accepted: bool, + assigned_ordinal: Option, +} + +#[derive(Clone, Debug, PartialEq, Eq, Hash)] +enum EventKey { + Channel(ChannelId), + Frame(u64), +} + +struct SubscriberLedger { + subscription: TelemetrySubscription, + capacity: usize, + created_at_publication: u64, + last_live_publication: Option, + seen: HashSet, + last_dropped: u64, + observations: Vec, +} + +#[derive(Clone, Copy, Default)] +struct CounterSnapshot { + assigned: u64, + drained: u64, + mux_dropped: u64, + bitbucketed: u64, +} + +struct ObservationLedger { + actions: Vec, + channels: Vec, + submissions: BTreeMap, + pending: VecDeque, + assigned: Vec, + rejected: HashSet, + observed_positions: HashMap, + position_offset: Option, + publications: HashMap, + next_publication: u64, + subscribers: Vec>, + deliveries: Vec<(StreamId, Frame)>, + next_payload_id: u64, + expected_bitbucketed: u64, + previous_counters: CounterSnapshot, +} + +impl ObservationLedger { + fn new() -> Self { + Self { + actions: Vec::new(), + channels: Vec::new(), + submissions: BTreeMap::new(), + pending: VecDeque::new(), + assigned: Vec::new(), + rejected: HashSet::new(), + observed_positions: HashMap::new(), + position_offset: None, + publications: HashMap::new(), + next_publication: 0, + subscribers: Vec::new(), + deliveries: Vec::new(), + next_payload_id: 0, + expected_bitbucketed: 0, + previous_counters: CounterSnapshot::default(), + } + } + + fn publish(&mut self, key: EventKey) { + assert!( + self.publications + .insert(key, self.next_publication) + .is_none(), + "an endpoint event was published twice" + ); + self.next_publication += 1; + } +} + +struct Harness { + endpoint: TelemetryEndpoint, + consumer: Consumer, + ledger: ObservationLedger, + next_channel_name: u64, +} + +impl Harness { + fn new(mux_capacity: usize) -> Self { + let endpoint = TelemetryEndpoint::with_capacity( + StreamId::new("fuzz-endpoint", Lifetime(1)), + mux_capacity, + 16, + ); + let mut harness = Self { + endpoint, + consumer: Consumer::new(), + ledger: ObservationLedger::new(), + next_channel_name: 0, + }; + harness.register_fresh("initial".to_owned(), 0); + harness.ledger.actions.clear(); + harness + } + + fn content(kind: u8, suffix: &str) -> ChannelContent { + match kind % 3 { + 0 => ChannelContent::Bytes, + 1 => ChannelContent::TextStream, + _ => ChannelContent::JsonRecord { + schema: Some(suffix.to_owned()), + }, + } + } + + fn payload(&mut self, tail: Vec) -> (u64, Vec) { + let id = self.ledger.next_payload_id; + self.ledger.next_payload_id += 1; + let mut payload = Vec::with_capacity(8 + tail.len()); + payload.extend_from_slice(&id.to_be_bytes()); + payload.extend_from_slice(&tail); + (id, payload) + } + + fn payload_id(payload: &[u8]) -> u64 { + let prefix: [u8; 8] = payload[..8] + .try_into() + .expect("generated payload always carries an identity prefix"); + u64::from_be_bytes(prefix) + } + + fn register_fresh(&mut self, suffix: String, kind: u8) { + let name = format!("channel-{}-{suffix}", self.next_channel_name); + self.next_channel_name += 1; + let content = Self::content(kind, &suffix); + let id = self + .endpoint + .try_register_channel(name.clone(), content.clone()) + .expect("fresh generated channel name"); + let descriptor = self + .endpoint + .catalog_snapshot() + .channels + .values() + .find(|descriptor| descriptor.id == id) + .cloned() + .expect("new channel appears in catalog"); + assert_eq!(descriptor.name, name); + assert_eq!(descriptor.content, content); + self.ledger.channels.push(descriptor); + self.ledger.publish(EventKey::Channel(id)); + self.ledger.actions.push("RegisterFreshChannel".into()); + } + + fn register_same(&mut self, slot: usize) { + let descriptor = self.ledger.channels[slot % self.ledger.channels.len()].clone(); + let before = self.endpoint.catalog_snapshot().channels; + let id = self + .endpoint + .try_register_channel(descriptor.name.clone(), descriptor.content.clone()) + .expect("identical registration is legal"); + assert_eq!(id, descriptor.id); + assert_eq!(self.endpoint.catalog_snapshot().channels, before); + self.ledger.actions.push("RegisterSameChannelAgain".into()); + } + + fn submit(&mut self, channel_slot: usize, tail: Vec) { + let channel = self.ledger.channels[channel_slot % self.ledger.channels.len()].id; + let (id, payload) = self.payload(tail); + let accepted = self + .endpoint + .producer() + .submit_bytes(channel, payload.clone()); + assert!( + self.ledger + .submissions + .insert( + id, + Submission { + channel, + payload, + accepted, + assigned_ordinal: None, + }, + ) + .is_none() + ); + if accepted { + self.ledger.pending.push_back(id); + } else { + self.ledger.rejected.insert(id); + } + self.ledger + .actions + .push(format!("SubmitRegisteredPayload({accepted})")); + } + + fn tick(&mut self) { + let subscriberless = self.endpoint.subscriber_count() == 0; + let assigned_now: Vec = self.ledger.pending.drain(..).collect(); + let stats = self.endpoint.tick(); + assert_eq!(stats.drained, assigned_now.len()); + if subscriberless { + self.ledger.expected_bitbucketed += stats.drained as u64; + } + for id in assigned_now { + let ordinal = self.ledger.assigned.len() as u64; + let submission = self.ledger.submissions.get_mut(&id).unwrap(); + assert!(submission.assigned_ordinal.replace(ordinal).is_none()); + self.ledger.assigned.push(id); + self.ledger.publish(EventKey::Frame(id)); + } + self.ledger.actions.push("Tick".into()); + } + + fn subscribe(&mut self, raw_capacity: usize) { + let capacity = raw_capacity % 16 + 1; + let name = format!("subscriber-{}", self.ledger.subscribers.len()); + let subscription = self.endpoint.subscribe_all_with_capacity(name, capacity); + let snapshot_channels: HashSet = subscription + .snapshot() + .channels + .iter() + .map(|descriptor| descriptor.id) + .collect(); + let catalog_channels: HashSet = self + .ledger + .channels + .iter() + .map(|descriptor| descriptor.id) + .collect(); + assert_eq!(snapshot_channels, catalog_channels); + self.ledger.subscribers.push(Some(SubscriberLedger { + subscription, + capacity, + created_at_publication: self.ledger.next_publication, + last_live_publication: None, + seen: HashSet::new(), + last_dropped: 0, + observations: Vec::new(), + })); + self.ledger.actions.push("Subscribe".into()); + } + + fn live_subscriber_slots(&self) -> Vec { + self.ledger + .subscribers + .iter() + .enumerate() + .filter_map(|(slot, subscriber)| subscriber.as_ref().map(|_| slot)) + .collect() + } + + fn drain_subscriber(&mut self, requested_slot: usize) { + let live = self.live_subscriber_slots(); + if live.is_empty() { + self.subscribe(requested_slot); + return; + } + let slot = live[requested_slot % live.len()]; + let events = self.ledger.subscribers[slot] + .as_ref() + .unwrap() + .subscription + .drain_available(); + let capacity = self.ledger.subscribers[slot].as_ref().unwrap().capacity; + assert!(events.len() <= capacity); + + for event in events { + let key = match event { + TelemetryEvent::ChannelDeclared(descriptor) => { + let original = self + .ledger + .channels + .iter() + .find(|known| known.id == descriptor.id) + .expect("received declaration exists in catalog history"); + assert_eq!(&descriptor, original); + EventKey::Channel(descriptor.id) + } + TelemetryEvent::Frame(delivery) => { + self.check_delivery(&delivery); + EventKey::Frame(Self::payload_id(&delivery.payload)) + } + other => panic!("endpoint generated unexpected event: {other:?}"), + }; + let publication = *self + .ledger + .publications + .get(&key) + .expect("received event was previously published"); + let subscriber = self.ledger.subscribers[slot].as_mut().unwrap(); + assert!( + subscriber.seen.insert(key.clone()), + "subscriber saw a duplicate" + ); + if publication >= subscriber.created_at_publication { + if let Some(previous) = subscriber.last_live_publication { + assert!( + publication > previous, + "future events preserve publication order" + ); + } + subscriber.last_live_publication = Some(publication); + } + subscriber.observations.push(key); + } + self.ledger.actions.push("DrainSubscriber".into()); + } + + fn check_delivery(&mut self, delivery: &FrameDelivery) { + assert_eq!(delivery.channel.stream, self.endpoint.stream_id().clone()); + let id = Self::payload_id(&delivery.payload); + let submission = self + .ledger + .submissions + .get(&id) + .expect("observed frame corresponds to a submission"); + assert!(submission.accepted, "rejected submission appeared"); + assert_eq!(delivery.channel.channel, submission.channel); + assert_eq!(delivery.payload, submission.payload); + let ordinal = submission + .assigned_ordinal + .expect("observed frame was assigned during an earlier tick"); + let offset = i128::from(delivery.position.0) - i128::from(ordinal); + match self.ledger.position_offset { + Some(expected) => assert_eq!(offset, expected, "positions are gap-free across drains"), + None => self.ledger.position_offset = Some(offset), + } + if let Some(previous) = self.ledger.observed_positions.insert(id, delivery.position) { + assert_eq!(previous, delivery.position); + } + } + + fn drop_subscriber(&mut self, requested_slot: usize) { + let live = self.live_subscriber_slots(); + if live.is_empty() { + self.subscribe(requested_slot); + return; + } + let slot = live[requested_slot % live.len()]; + self.ledger.subscribers[slot].take(); + self.ledger.actions.push("DropSubscriber".into()); + } + + fn stream(raw: u8) -> StreamId { + match raw % 4 { + 0 => StreamId::new("shared-node", Lifetime(1)), + 1 => StreamId::new("shared-node", Lifetime(2)), + 2 => StreamId::new("other-node", Lifetime(1)), + _ => StreamId::new("other-node", Lifetime(9)), + } + } + + fn fresh_store_position(&self, stream: &StreamId, raw: u16) -> Position { + let occupied: HashSet = self + .ledger + .deliveries + .iter() + .filter(|(known_stream, _)| known_stream == stream) + .map(|(_, frame)| frame.position) + .collect(); + let mut position = u64::from(raw); + while occupied.contains(&Position(position)) { + position += 1; + } + Position(position) + } + + fn ingest_new(&mut self, raw_stream: u8, raw_position: u16, raw_channel: u16, tail: Vec) { + let stream = Self::stream(raw_stream); + let position = self.fresh_store_position(&stream, raw_position); + let (_, payload) = self.payload(tail); + let frame = Frame { + channel: ChannelId(10_000 + u32::from(raw_channel)), + position, + payload, + }; + assert!( + self.consumer + .accept(Delivery::new(stream.clone(), frame.clone())) + ); + self.ledger.deliveries.push((stream, frame)); + self.ledger.actions.push("IngestFrame".into()); + } + + fn ingest_duplicate(&mut self, requested_slot: usize) { + if self.ledger.deliveries.is_empty() { + self.ingest_new(0, 0, 0, Vec::new()); + return; + } + let (stream, frame) = + self.ledger.deliveries[requested_slot % self.ledger.deliveries.len()].clone(); + assert!( + !self + .consumer + .accept(Delivery::new(stream.clone(), frame.clone())) + ); + self.ledger.deliveries.push((stream, frame)); + self.ledger.actions.push("IngestDuplicate".into()); + } + + fn execute(&mut self, action: Action) { + let action_debug = format!("{action:?}"); + self.ledger.actions.push(format!("Attempt:{action_debug}")); + match action { + Action::RegisterFresh { suffix, kind } => self.register_fresh(suffix, kind), + Action::RegisterSame { slot } => self.register_same(slot), + Action::Submit { channel, tail } => self.submit(channel, tail), + Action::Tick => self.tick(), + Action::Subscribe { capacity } => self.subscribe(capacity), + Action::DrainSubscriber { slot } => self.drain_subscriber(slot), + Action::DropSubscriber { slot } => self.drop_subscriber(slot), + Action::IngestNew { + stream, + position, + channel, + tail, + } => self.ingest_new(stream, position, channel, tail), + Action::IngestDuplicate { slot } => self.ingest_duplicate(slot), + Action::InspectCatalog => self.ledger.actions.push("InspectCatalog".into()), + Action::InspectStore => self.ledger.actions.push("InspectStore".into()), + } + self.check_invariants(&action_debug); + } + + fn check_catalog(&self) { + let snapshot = self.endpoint.catalog_snapshot(); + assert_eq!(snapshot.channels.len(), self.ledger.channels.len()); + let by_id: HashMap = snapshot + .channels + .values() + .map(|descriptor| (descriptor.id, descriptor)) + .collect(); + let by_name: HashMap<&str, &ChannelDescriptor> = snapshot + .channels + .values() + .map(|descriptor| (descriptor.name.as_str(), descriptor)) + .collect(); + assert_eq!( + by_id.len(), + snapshot.channels.len(), + "channel ids are unique" + ); + assert_eq!( + by_name.len(), + snapshot.channels.len(), + "channel names are unique" + ); + for expected in &self.ledger.channels { + assert_eq!(by_id.get(&expected.id).copied(), Some(expected)); + assert_eq!(by_name.get(expected.name.as_str()).copied(), Some(expected)); + } + } + + fn expected_store(&self) -> BTreeMap> { + let mut expected = BTreeMap::>::new(); + for (stream, frame) in &self.ledger.deliveries { + expected + .entry(stream.clone()) + .or_default() + .entry(frame.position) + .or_insert_with(|| frame.clone()); + } + expected + } + + fn check_store(&self) { + let expected = self.expected_store(); + let actual_streams: HashSet = + self.consumer.store().stream_ids().cloned().collect(); + let expected_streams: HashSet = expected.keys().cloned().collect(); + assert_eq!(actual_streams, expected_streams); + + for (stream, expected_frames) in expected { + let stored = self + .consumer + .store() + .stream(&stream) + .expect("expected stream exists"); + let actual = stored.to_vec(); + let actual_positions: Vec = + actual.iter().map(|frame| frame.position).collect(); + assert!(actual_positions.windows(2).all(|pair| pair[0] < pair[1])); + assert_eq!( + actual_positions.len(), + actual_positions.iter().collect::>().len() + ); + assert_eq!( + actual, + expected_frames.values().cloned().collect::>(), + "first delivery wins and iteration follows position order" + ); + + let expected_gaps: Vec<(u64, u64)> = actual_positions + .windows(2) + .filter_map(|pair| { + (pair[1].0 > pair[0].0 + 1).then_some((pair[0].0 + 1, pair[1].0 - 1)) + }) + .collect(); + let actual_gaps: Vec<(u64, u64)> = stored + .gap_spans() + .into_iter() + .map(|gap| (gap.start, gap.end)) + .collect(); + assert_eq!(actual_gaps, expected_gaps); + } + } + + fn counter_snapshot(&self) -> CounterSnapshot { + CounterSnapshot { + assigned: self.endpoint.assigned(), + drained: self.endpoint.drained(), + mux_dropped: self.endpoint.mux_dropped(), + bitbucketed: self.endpoint.bitbucketed(), + } + } + + fn check_invariants(&mut self, action: &str) { + self.check_catalog(); + self.check_store(); + + assert!( + self.ledger + .assigned + .iter() + .all(|id| self.ledger.submissions[id].accepted) + ); + assert_eq!( + self.ledger.assigned.len(), + self.ledger.assigned.iter().collect::>().len(), + "accepted submission assigned at most once after {action}" + ); + assert!( + self.ledger + .rejected + .is_disjoint(&self.ledger.assigned.iter().copied().collect()) + ); + + let counters = self.counter_snapshot(); + assert!(counters.assigned >= self.ledger.previous_counters.assigned); + assert!(counters.drained >= self.ledger.previous_counters.drained); + assert!(counters.mux_dropped >= self.ledger.previous_counters.mux_dropped); + assert!(counters.bitbucketed >= self.ledger.previous_counters.bitbucketed); + assert_eq!(counters.assigned, self.ledger.assigned.len() as u64); + assert_eq!(counters.drained, self.ledger.assigned.len() as u64); + assert_eq!(counters.mux_dropped, self.ledger.rejected.len() as u64); + assert_eq!(counters.bitbucketed, self.ledger.expected_bitbucketed); + self.ledger.previous_counters = counters; + + let dropped_by_id: HashMap<_, _> = self + .endpoint + .subscriber_snapshots() + .into_iter() + .map(|snapshot| (snapshot.id, snapshot.dropped)) + .collect(); + for subscriber in self.ledger.subscribers.iter_mut().flatten() { + let dropped = dropped_by_id[&subscriber.subscription.id()]; + assert!(dropped >= subscriber.last_dropped); + subscriber.last_dropped = dropped; + } + } + + fn finish(&mut self) { + self.tick(); + for slot in self.live_subscriber_slots() { + self.drain_subscriber(slot); + self.check_invariants("final drain"); + } + self.check_invariants("trace completion"); + } +} + +proptest! { + #![proptest_config(ProptestConfig { + cases: 16, + max_shrink_iters: 20_000, + .. ProptestConfig::default() + })] + + #[test] + fn long_legal_action_sequences_never_violate_telemetry_invariants( + mux_capacity in 1usize..17, + actions in prop::collection::vec(action_strategy(), 256..1025), + ) { + let mut harness = Harness::new(mux_capacity); + for action in actions { + harness.execute(action); + } + harness.finish(); + } +} + +#[test] +fn conflicting_registration_fails_without_catalog_or_fanout_mutation() { + let endpoint = TelemetryEndpoint::with_capacity(StreamId::new("node", Lifetime(1)), 4, 8); + let id = endpoint + .try_register_channel("same", ChannelContent::Bytes) + .unwrap(); + let subscriber = endpoint.subscribe_all_with_capacity("observer", 8); + let before = endpoint.catalog_snapshot().channels; + + let error = endpoint + .try_register_channel("same", ChannelContent::TextStream) + .unwrap_err(); + assert!(matches!( + error, + ChannelRegistrationError::ConflictingName { .. } + )); + assert_eq!(endpoint.catalog_snapshot().channels, before); + assert!( + endpoint + .catalog_snapshot() + .channels + .values() + .any(|descriptor| descriptor.id == id) + ); + assert!(subscriber.try_recv().is_err()); +} + +#[test] +fn mux_rejection_never_appears_or_consumes_a_position() { + let endpoint = TelemetryEndpoint::with_capacity(StreamId::new("node", Lifetime(1)), 1, 8); + let channel = endpoint.register_channel("bytes", ChannelContent::Bytes); + let subscriber = endpoint.subscribe_all_with_capacity("observer", 8); + + assert!( + endpoint + .producer() + .submit_bytes(channel, b"accepted-0".to_vec()) + ); + assert!( + !endpoint + .producer() + .submit_bytes(channel, b"rejected".to_vec()) + ); + endpoint.tick(); + assert!( + endpoint + .producer() + .submit_bytes(channel, b"accepted-1".to_vec()) + ); + endpoint.tick(); + + let frames: Vec = subscriber + .drain_available() + .into_iter() + .filter_map(|event| match event { + TelemetryEvent::Frame(frame) => Some(frame), + _ => None, + }) + .collect(); + assert_eq!(frames.len(), 2); + assert_eq!(frames[0].payload, b"accepted-0"); + assert_eq!(frames[1].payload, b"accepted-1"); + assert_eq!(frames[1].position.0, frames[0].position.0 + 1); + assert_eq!(endpoint.mux_dropped(), 1); +} + +#[test] +fn full_and_disconnected_subscribers_do_not_affect_others() { + let endpoint = TelemetryEndpoint::with_capacity(StreamId::new("node", Lifetime(1)), 8, 8); + let channel = endpoint.register_channel("bytes", ChannelContent::Bytes); + let slow = endpoint.subscribe_all_with_capacity("slow", 1); + let fast = endpoint.subscribe_all_with_capacity("fast", 8); + let slow_id = slow.id(); + let fast_id = fast.id(); + + assert!(endpoint.producer().submit_bytes(channel, b"one".to_vec())); + assert!(endpoint.producer().submit_bytes(channel, b"two".to_vec())); + let tick = endpoint.tick(); + assert_eq!(tick.dropped_for_subscribers, 1); + + assert_eq!(slow.drain_available().len(), 1); + assert_eq!(fast.drain_available().len(), 2); + let dropped: HashMap<_, _> = endpoint + .subscriber_snapshots() + .into_iter() + .map(|snapshot| (snapshot.id, snapshot.dropped)) + .collect(); + assert_eq!(dropped[&slow_id], 1); + assert_eq!(dropped[&fast_id], 0); + + drop(slow); + assert!( + endpoint + .producer() + .submit_bytes(channel, b"after-drop".to_vec()) + ); + endpoint.tick(); + let event = fast + .try_recv() + .expect("live subscriber continues after peer drop"); + assert!(matches!(event, TelemetryEvent::Frame(frame) if frame.payload == b"after-drop")); + assert_eq!(endpoint.subscriber_count(), 1); +} + +#[test] +fn conflicting_duplicate_ingest_preserves_first_frame_and_other_streams() { + let mut consumer = Consumer::new(); + let first_stream = StreamId::new("same-node", Lifetime(1)); + let second_stream = StreamId::new("same-node", Lifetime(2)); + let first = Frame { + channel: ChannelId(999), + position: Position(7), + payload: vec![0, 1, 2, 255], + }; + let conflicting = Frame { + payload: b"replacement".to_vec(), + ..first.clone() + }; + + assert!(consumer.accept(Delivery::new(first_stream.clone(), first.clone()))); + assert!(!consumer.accept(Delivery::new(first_stream.clone(), conflicting))); + assert!(consumer.accept(Delivery::new(second_stream.clone(), first.clone()))); + assert_eq!( + consumer.store().stream(&first_stream).unwrap().to_vec(), + vec![first.clone()] + ); + assert_eq!( + consumer.store().stream(&second_stream).unwrap().to_vec(), + vec![first] + ); +} + +#[test] +fn invalid_actions_leave_all_ledger_invariants_intact() { + let mut harness = Harness::new(1); + harness.subscribe(1); + + harness.submit(0, b"accepted".to_vec()); + harness.submit(0, b"rejected".to_vec()); + assert_eq!(harness.endpoint.mux_dropped(), 1); + harness.check_invariants("full mux rejection"); + + let descriptor = harness.ledger.channels[0].clone(); + let before_catalog = harness.endpoint.catalog_snapshot().channels; + assert!(matches!( + harness + .endpoint + .try_register_channel(descriptor.name, ChannelContent::TextStream), + Err(ChannelRegistrationError::ConflictingName { .. }) + )); + assert_eq!(harness.endpoint.catalog_snapshot().channels, before_catalog); + harness.check_invariants("conflicting registration"); + + harness.register_fresh("fills-subscriber".into(), 0); + harness.register_fresh("drops-for-subscriber".into(), 0); + harness.check_invariants("full subscriber publication"); + + harness.ingest_new(0, 9, 7, b"first".to_vec()); + let (stream, first) = harness.ledger.deliveries.last().unwrap().clone(); + let conflicting = Frame { + payload: b"conflicting".to_vec(), + ..first + }; + assert!(!harness.consumer.accept(Delivery::new(stream, conflicting))); + harness.check_invariants("conflicting duplicate ingest"); +} + +fn sample_delivery_bytes() -> Vec { + encode_delivery( + &StreamId::new("node", Lifetime(9)), + &Frame { + channel: ChannelId(4), + position: Position(12), + payload: vec![0, 1, 2, 255], + }, + ) +} + +#[test] +fn telemetry_wire_rejects_every_truncation_and_trailing_bytes() { + let encoded = sample_delivery_bytes(); + for length in 0..encoded.len() { + assert!( + matches!( + decode_delivery(&encoded[..length]), + Err(WireError::Truncated | WireError::BadLength) + ), + "accepted truncation at {length}" + ); + } + let mut trailing = encoded; + trailing.push(0); + assert_eq!(decode_delivery(&trailing), Err(WireError::TrailingBytes)); +} + +#[test] +fn telemetry_wire_rejects_invalid_lengths_and_utf8() { + let mut invalid_length = sample_delivery_bytes(); + invalid_length[..4].copy_from_slice(&u32::MAX.to_le_bytes()); + assert!(matches!( + decode_delivery(&invalid_length), + Err(WireError::Truncated | WireError::BadLength) + )); + + let mut invalid_utf8 = sample_delivery_bytes(); + assert_eq!(u32::from_le_bytes(invalid_utf8[..4].try_into().unwrap()), 4); + invalid_utf8[4] = 0xff; + assert_eq!(decode_delivery(&invalid_utf8), Err(WireError::NotUtf8)); + + let (stream, frame) = decode_delivery(&sample_delivery_bytes()).unwrap(); + assert_eq!(stream, StreamId::new("node", Lifetime(9))); + assert_eq!(frame.payload, vec![0, 1, 2, 255]); +} + +proptest! { + #![proptest_config(ProptestConfig { + cases: 512, + max_shrink_iters: 10_000, + .. ProptestConfig::default() + })] + + #[test] + fn arbitrary_wire_bytes_decode_or_fail_without_panicking( + bytes in prop::collection::vec(any::(), 0..2048), + ) { + if let Ok((stream, frame)) = decode_delivery(&bytes) { + let canonical = encode_delivery(&stream, &frame); + prop_assert_eq!(decode_delivery(&canonical).unwrap(), (stream, frame)); + } + } +} diff --git a/crates/transport/Cargo.toml b/crates/transport/Cargo.toml index dfb6b2b..56972bb 100644 --- a/crates/transport/Cargo.toml +++ b/crates/transport/Cargo.toml @@ -15,5 +15,8 @@ rand_core = { version = "0.6", features = ["getrandom"] } serde = { version = "1", features = ["derive"] } serde_json = "1" +[dev-dependencies] +proptest = "1" + [lints] workspace = true diff --git a/crates/transport/src/codec.rs b/crates/transport/src/codec.rs index 21320a9..25c63cb 100644 --- a/crates/transport/src/codec.rs +++ b/crates/transport/src/codec.rs @@ -9,6 +9,7 @@ use std::any::{Any, TypeId}; use std::collections::HashMap; use std::sync::Arc; +use std::fmt; use swactor::actor::{ActorAddress, Message}; use swactor::Error; @@ -111,6 +112,31 @@ pub struct WireEnvelope { pub payload: Vec, } +/// Setup-time conflicts returned by [`CodecRegistry`] registration methods. +/// +/// Registrations are never replaced implicitly. A failed registration leaves +/// both registry maps unchanged. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum CodecRegistrationError { + EncoderAlreadyRegistered { message_type: &'static str }, + DecoderAlreadyRegistered { type_tag: String }, +} + +impl fmt::Display for CodecRegistrationError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::EncoderAlreadyRegistered { message_type } => { + write!(f, "encoder already registered for {message_type}") + } + Self::DecoderAlreadyRegistered { type_tag } => { + write!(f, "decoder already registered for '{type_tag}'") + } + } + } +} + +impl std::error::Error for CodecRegistrationError {} + // ─── CodecRegistry ────────────────────────────────────────────────────────── type EncodeFn = Box) -> Result<(String, Vec), Error> + Send + Sync>; @@ -145,11 +171,25 @@ impl CodecRegistry { /// Register a message type with its codec. /// /// Both encoding and decoding are handled by the same codec instance, keyed - /// symmetrically by `M::type_tag()`. - pub fn register>(&mut self, codec: C) { - let codec = Arc::new(codec); + /// symmetrically by `M::type_tag()`. If either the Rust type or wire tag is + /// already registered, this returns an error without changing either map. + pub fn register>( + &mut self, + codec: C, + ) -> Result<(), CodecRegistrationError> { + let message_type = std::any::type_name::(); + let type_id = TypeId::of::(); + let type_tag = M::type_tag(); + if self.encoders.contains_key(&type_id) { + return Err(CodecRegistrationError::EncoderAlreadyRegistered { message_type }); + } + if self.decoders.contains_key(type_tag) { + return Err(CodecRegistrationError::DecoderAlreadyRegistered { + type_tag: type_tag.to_string(), + }); + } - // Encoder side — closure downcasts Any → M, encodes, returns (tag, bytes) + let codec = Arc::new(codec); let encode_codec = codec.clone(); let encode_fn: EncodeFn = Box::new(move |msg: Box| { let typed = msg @@ -158,14 +198,14 @@ impl CodecRegistry { let bytes = encode_codec.encode(&*typed)?; Ok((M::type_tag().to_string(), bytes)) }); - self.encoders.insert(TypeId::of::(), encode_fn); - - // Decoder side — closure captures Arc let decode_fn: DecodeFn = Box::new(move |bytes: &[u8]| { let msg: M = codec.decode(bytes)?; Ok(Box::new(msg) as Box) }); - self.decoders.insert(M::type_tag().to_string(), decode_fn); + + self.encoders.insert(type_id, encode_fn); + self.decoders.insert(type_tag.to_string(), decode_fn); + Ok(()) } /// Register a **variant-multiplexing** encoder for one Rust type `M`. @@ -175,17 +215,26 @@ impl CodecRegistry { /// may refuse to encode local-only variants by returning `Err`. `M` is a /// plain [`Message`] (e.g. an actor `Incoming` enum), not a /// [`NetworkMessage`] — it has no single canonical `type_tag`. + /// + /// Returns an error without mutation if `M` already has an encoder. pub fn register_encoder( &mut self, e: impl Fn(&M) -> Result<(String, Vec), Error> + Send + Sync + 'static, - ) { + ) -> Result<(), CodecRegistrationError> { + let type_id = TypeId::of::(); + if self.encoders.contains_key(&type_id) { + return Err(CodecRegistrationError::EncoderAlreadyRegistered { + message_type: std::any::type_name::(), + }); + } let encode_fn: EncodeFn = Box::new(move |msg: Box| { let typed = msg .downcast::() .map_err(|_| Error::from("Transport: type downcast failed during encode"))?; e(&*typed) }); - self.encoders.insert(TypeId::of::(), encode_fn); + self.encoders.insert(type_id, encode_fn); + Ok(()) } /// Register a **fan-in** decoder mapping an arbitrary wire `type_tag` to one @@ -194,14 +243,22 @@ impl CodecRegistry { /// Unlike [`register`](Self::register), the tag is caller-chosen, so several /// tags can decode into the same actor enum `M`. `M` is a plain [`Message`] /// (e.g. an actor `Incoming` enum), not a [`NetworkMessage`]. + /// + /// Returns an error without mutation if `type_tag` already has a decoder. pub fn register_decoder( &mut self, type_tag: &str, d: impl Fn(&[u8]) -> Result + Send + Sync + 'static, - ) { + ) -> Result<(), CodecRegistrationError> { + if self.decoders.contains_key(type_tag) { + return Err(CodecRegistrationError::DecoderAlreadyRegistered { + type_tag: type_tag.to_string(), + }); + } let decode_fn: DecodeFn = Box::new(move |bytes: &[u8]| Ok(Box::new(d(bytes)?) as Box)); self.decoders.insert(type_tag.to_string(), decode_fn); + Ok(()) } /// Encode a type-erased message. Returns `(type_tag, payload_bytes)`. diff --git a/crates/transport/src/lib.rs b/crates/transport/src/lib.rs index 20c37e0..43000c5 100644 --- a/crates/transport/src/lib.rs +++ b/crates/transport/src/lib.rs @@ -12,7 +12,8 @@ pub mod json_codec; pub mod transport; pub use codec::{ - hex_decode, hex_encode, Codec, CodecRegistry, NetworkMessage, NodeId, WireEnvelope, + hex_decode, hex_encode, Codec, CodecRegistrationError, CodecRegistry, NetworkMessage, NodeId, + WireEnvelope, }; pub use json_codec::JsonCodec; pub use transport::{CodecRemoteSink, InMemoryTransport, Transport, TransportRouter}; diff --git a/crates/transport/tests/codec_invariants.rs b/crates/transport/tests/codec_invariants.rs new file mode 100644 index 0000000..d1975e9 --- /dev/null +++ b/crates/transport/tests/codec_invariants.rs @@ -0,0 +1,418 @@ +use std::any::TypeId; +use std::collections::HashSet; + +use proptest::prelude::*; +use swactor::actor::ActorAddress; +use swactor::Error; +use swactor_transport::{ + Codec, CodecRegistrationError, CodecRegistry, JsonCodec, NetworkMessage, WireEnvelope, +}; + +#[derive(Clone, Debug, PartialEq, Eq)] +struct Number(u64); + +impl NetworkMessage for Number { + fn type_tag() -> &'static str { + "contract::Number" + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +struct Text(String); + +impl NetworkMessage for Text { + fn type_tag() -> &'static str { + "contract::Text" + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +struct Blob(Vec); + +impl NetworkMessage for Blob { + fn type_tag() -> &'static str { + "contract::Blob" + } +} + +#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +struct Structured { + id: u64, + labels: Vec, +} + +impl NetworkMessage for Structured { + fn type_tag() -> &'static str { + "contract::Structured" + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +struct Fallible(u8); + +impl NetworkMessage for Fallible { + fn type_tag() -> &'static str { + "contract::Fallible" + } +} + +#[derive(Clone, Copy)] +struct NumberCodec; + +impl Codec for NumberCodec { + fn encode(&self, msg: &Number) -> Result, Error> { + Ok(msg.0.to_be_bytes().to_vec()) + } + + fn decode(&self, bytes: &[u8]) -> Result { + let bytes: [u8; 8] = bytes + .try_into() + .map_err(|_| Error::from("number payload must contain eight bytes"))?; + Ok(Number(u64::from_be_bytes(bytes))) + } +} + +#[derive(Clone, Copy)] +struct TextCodec; + +impl Codec for TextCodec { + fn encode(&self, msg: &Text) -> Result, Error> { + Ok(msg.0.as_bytes().to_vec()) + } + + fn decode(&self, bytes: &[u8]) -> Result { + let text = std::str::from_utf8(bytes) + .map_err(|error| Error::from(format!("invalid text payload: {error}")))?; + Ok(Text(text.to_owned())) + } +} + +#[derive(Clone, Copy)] +struct BlobCodec; + +impl Codec for BlobCodec { + fn encode(&self, msg: &Blob) -> Result, Error> { + Ok(msg.0.clone()) + } + + fn decode(&self, bytes: &[u8]) -> Result { + Ok(Blob(bytes.to_vec())) + } +} + +#[derive(Clone, Copy)] +struct FallibleCodec; + +impl Codec for FallibleCodec { + fn encode(&self, msg: &Fallible) -> Result, Error> { + if msg.0 == u8::MAX { + return Err(Error::from("refused value")); + } + Ok(vec![msg.0]) + } + + fn decode(&self, bytes: &[u8]) -> Result { + match bytes { + [value] if *value != u8::MAX => Ok(Fallible(*value)), + _ => Err(Error::from("malformed fallible payload")), + } + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +enum Sample { + Number(Number), + Text(Text), + Blob(Blob), + Structured(Structured), + Fallible(Fallible), +} + +impl Sample { + fn register(&self, registry: &mut CodecRegistry) -> Result<(), CodecRegistrationError> { + match self { + Self::Number(_) => registry.register::(NumberCodec), + Self::Text(_) => registry.register::(TextCodec), + Self::Blob(_) => registry.register::(BlobCodec), + Self::Structured(_) => registry.register::(JsonCodec::default()), + Self::Fallible(_) => registry.register::(FallibleCodec), + } + } + + fn type_id(&self) -> TypeId { + match self { + Self::Number(_) => TypeId::of::(), + Self::Text(_) => TypeId::of::(), + Self::Blob(_) => TypeId::of::(), + Self::Structured(_) => TypeId::of::(), + Self::Fallible(_) => TypeId::of::(), + } + } + + fn tag(&self) -> &'static str { + match self { + Self::Number(_) => Number::type_tag(), + Self::Text(_) => Text::type_tag(), + Self::Blob(_) => Blob::type_tag(), + Self::Structured(_) => Structured::type_tag(), + Self::Fallible(_) => Fallible::type_tag(), + } + } + + fn boxed(&self) -> Box { + match self { + Self::Number(value) => Box::new(value.clone()), + Self::Text(value) => Box::new(value.clone()), + Self::Blob(value) => Box::new(value.clone()), + Self::Structured(value) => Box::new(value.clone()), + Self::Fallible(value) => Box::new(value.clone()), + } + } + + fn assert_decoded(&self, decoded: Box) { + match self { + Self::Number(expected) => assert_eq!(*decoded.downcast::().unwrap(), *expected), + Self::Text(expected) => assert_eq!(*decoded.downcast::().unwrap(), *expected), + Self::Blob(expected) => assert_eq!(*decoded.downcast::().unwrap(), *expected), + Self::Structured(expected) => { + assert_eq!(*decoded.downcast::().unwrap(), *expected) + } + Self::Fallible(expected) => { + assert_eq!(*decoded.downcast::().unwrap(), *expected) + } + } + } +} + +fn address(seed: u8) -> ActorAddress { + ActorAddress([seed; 32]) +} + +fn assert_sample(registry: &CodecRegistry, sample: &Sample, destination: ActorAddress) -> Vec { + let (tag, payload) = registry + .encode(sample.type_id(), sample.boxed()) + .expect("registered sample encodes"); + assert_eq!(tag, sample.tag()); + sample.assert_decoded( + registry + .decode(&tag, &payload) + .expect("registered sample decodes"), + ); + + let (received_destination, decoded) = registry + .receive(WireEnvelope { + dest: destination, + type_tag: tag, + payload: payload.clone(), + }) + .expect("valid envelope receives"); + assert_eq!(received_destination, destination); + sample.assert_decoded(decoded); + payload +} + +fn fingerprint(registry: &CodecRegistry, samples: &[Sample]) -> Vec> { + let type_ids: HashSet = samples.iter().map(Sample::type_id).collect(); + let tags: HashSet<&str> = samples.iter().map(Sample::tag).collect(); + assert_eq!( + type_ids.len(), + samples.len(), + "registered TypeIds are unique" + ); + assert_eq!(tags.len(), samples.len(), "registered wire tags are unique"); + samples + .iter() + .enumerate() + .map(|(index, sample)| assert_sample(registry, sample, address(index as u8))) + .collect() +} + +#[derive(Clone, Debug)] +enum LegalOperation { + Encode(usize), + Decode(usize), + Receive(usize, u8), +} + +fn legal_operation() -> impl Strategy { + prop_oneof![ + any::().prop_map(LegalOperation::Encode), + any::().prop_map(LegalOperation::Decode), + (any::(), any::()).prop_map(|(slot, dest)| LegalOperation::Receive(slot, dest)), + ] +} + +proptest! { + #![proptest_config(ProptestConfig { + cases: 32, + max_shrink_iters: 10_000, + .. ProptestConfig::default() + })] + + #[test] + fn long_legal_action_sequences_preserve_every_registration( + number in any::(), + text in any::(), + blob in prop::collection::vec(any::(), 0..512), + structured_id in any::(), + labels in prop::collection::vec(any::(), 0..16), + fallible in 0u8..u8::MAX, + operations in prop::collection::vec(legal_operation(), 128..1025), + ) { + let samples = vec![ + Sample::Number(Number(number)), + Sample::Text(Text(text)), + Sample::Blob(Blob(blob)), + Sample::Structured(Structured { id: structured_id, labels }), + Sample::Fallible(Fallible(fallible)), + ]; + let mut registry = CodecRegistry::new(); + let mut registered = Vec::new(); + + for sample in &samples { + sample.register(&mut registry).expect("fresh type and tag register"); + registered.push(sample.clone()); + fingerprint(®istry, ®istered); + } + + for operation in operations { + let slot = match operation { + LegalOperation::Encode(slot) + | LegalOperation::Decode(slot) + | LegalOperation::Receive(slot, _) => slot % samples.len(), + }; + let sample = &samples[slot]; + match operation { + LegalOperation::Encode(_) => { + let (tag, _) = registry.encode(sample.type_id(), sample.boxed()).unwrap(); + prop_assert_eq!(tag, sample.tag()); + } + LegalOperation::Decode(_) => { + let (_, payload) = registry.encode(sample.type_id(), sample.boxed()).unwrap(); + sample.assert_decoded(registry.decode(sample.tag(), &payload).unwrap()); + } + LegalOperation::Receive(_, dest) => { + assert_sample(®istry, sample, address(dest)); + } + } + fingerprint(®istry, ®istered); + } + } +} + +#[test] +fn every_registration_duplicate_is_rejected_without_replacement() { + let sample = Sample::Number(Number(7)); + let mut registry = CodecRegistry::new(); + sample.register(&mut registry).unwrap(); + let before = fingerprint(®istry, std::slice::from_ref(&sample)); + + assert!(matches!( + registry.register::(NumberCodec), + Err(CodecRegistrationError::EncoderAlreadyRegistered { .. }) + )); + assert_eq!( + fingerprint(®istry, std::slice::from_ref(&sample)), + before + ); + + assert!(matches!( + registry.register_encoder::(|number| { + Ok(("replacement".to_owned(), number.0.to_le_bytes().to_vec())) + }), + Err(CodecRegistrationError::EncoderAlreadyRegistered { .. }) + )); + assert_eq!( + fingerprint(®istry, std::slice::from_ref(&sample)), + before + ); + + assert!(matches!( + registry.register_decoder::(Number::type_tag(), |_| Ok(Text("replacement".into()))), + Err(CodecRegistrationError::DecoderAlreadyRegistered { .. }) + )); + assert_eq!( + fingerprint(®istry, std::slice::from_ref(&sample)), + before + ); +} + +#[test] +fn symmetric_registration_is_atomic_when_only_encoder_conflicts() { + let mut registry = CodecRegistry::new(); + registry + .register_encoder::(|number| { + Ok(( + Number::type_tag().to_owned(), + number.0.to_be_bytes().to_vec(), + )) + }) + .unwrap(); + + assert!(matches!( + registry.register::(NumberCodec), + Err(CodecRegistrationError::EncoderAlreadyRegistered { .. }) + )); + assert!(registry + .decode(Number::type_tag(), &0u64.to_be_bytes()) + .is_err()); + let (tag, bytes) = registry + .encode(TypeId::of::(), Box::new(Number(9))) + .unwrap(); + assert_eq!(tag, Number::type_tag()); + assert_eq!(bytes, 9u64.to_be_bytes()); +} + +#[test] +fn symmetric_registration_is_atomic_when_only_decoder_conflicts() { + let mut registry = CodecRegistry::new(); + registry + .register_decoder::(Number::type_tag(), |bytes| { + Ok(Text(String::from_utf8_lossy(bytes).into_owned())) + }) + .unwrap(); + + assert!(matches!( + registry.register::(NumberCodec), + Err(CodecRegistrationError::DecoderAlreadyRegistered { .. }) + )); + assert!(registry + .encode(TypeId::of::(), Box::new(Number(9))) + .is_err()); + let decoded = registry.decode(Number::type_tag(), b"first").unwrap(); + assert_eq!(*decoded.downcast::().unwrap(), Text("first".into())); +} + +#[test] +fn operation_failures_do_not_change_registered_behavior() { + let samples = vec![Sample::Number(Number(11)), Sample::Fallible(Fallible(3))]; + let mut registry = CodecRegistry::new(); + for sample in &samples { + sample.register(&mut registry).unwrap(); + } + let before = fingerprint(®istry, &samples); + + assert!(registry + .encode(TypeId::of::(), Box::new(Text("unknown".into()))) + .is_err()); + assert_eq!(fingerprint(®istry, &samples), before); + + assert!(registry + .encode(TypeId::of::(), Box::new(Text("wrong".into()))) + .is_err()); + assert_eq!(fingerprint(®istry, &samples), before); + + assert!(registry.decode("contract::Unknown", b"anything").is_err()); + assert_eq!(fingerprint(®istry, &samples), before); + + assert!(registry.decode(Number::type_tag(), &[1, 2, 3]).is_err()); + assert_eq!(fingerprint(®istry, &samples), before); + + assert!(registry + .encode(TypeId::of::(), Box::new(Fallible(u8::MAX))) + .is_err()); + assert_eq!(fingerprint(®istry, &samples), before); + + assert!(registry.decode(Fallible::type_tag(), &[u8::MAX]).is_err()); + assert_eq!(fingerprint(®istry, &samples), before); +} diff --git a/crates/transport/tests/transport_api.rs b/crates/transport/tests/transport_api.rs index 161b2e0..c258de9 100644 --- a/crates/transport/tests/transport_api.rs +++ b/crates/transport/tests/transport_api.rs @@ -106,8 +106,10 @@ impl ActorInterface for PongActor { fn build_codec_registry() -> CodecRegistry { let mut cr = CodecRegistry::new(); - cr.register::(TestCodec); - cr.register::(TestCodec); + cr.register::(TestCodec) + .expect("unique codec registration"); + cr.register::(TestCodec) + .expect("unique codec registration"); cr } @@ -231,27 +233,6 @@ fn unregistered_type_produces_clear_error() { ); } -/// Given a WireEnvelope arrives with a type_tag not in the codec registry, -/// when CodecRegistry receives it, -/// then the error mentions "unknown type_tag". -#[test] -fn unknown_type_tag_on_receive_produces_clear_error() { - let codecs = CodecRegistry::new(); // empty registry - - let envelope = WireEnvelope { - dest: ActorAddress::default(), - type_tag: "nonexistent::Type".to_string(), - payload: vec![1, 2, 3], - }; - - let result = codecs.receive(envelope); - let err_msg = format!("{:?}", result.unwrap_err()); - assert!( - err_msg.contains("unknown type_tag"), - "Expected 'unknown type_tag' in error, got: {err_msg}" - ); -} - /// Given both local actors and transport routes exist, /// when an actor sends to another local actor, /// then the message is delivered locally (no serialization, transport never called). diff --git a/tools/actor-control-flow-lint/driver.rs b/tools/actor-control-flow-lint/driver.rs index 5b73224..1f5d9db 100755 --- a/tools/actor-control-flow-lint/driver.rs +++ b/tools/actor-control-flow-lint/driver.rs @@ -31,9 +31,9 @@ const EXECUTION_OWNERS: &[&str] = &[ "telemetry", ]; -/// This package owns only the compile-contract subprocess harness. It cannot be -/// used as a workspace dependency. -const TEST_SUPPORT_OWNERS: &[&str] = &["actor-control-flow-lint-tests"]; +/// Test and benchmark harness packages that must not be workspace dependencies. +const TEST_SUPPORT_OWNERS: &[&str] = + &["actor-control-flow-lint-tests", "swactor-benchmarks"]; /// Policy-bearing crates that must never enter an execution owner's dependency /// closure. diff --git a/tools/benchmarks/Cargo.toml b/tools/benchmarks/Cargo.toml new file mode 100644 index 0000000..898114c --- /dev/null +++ b/tools/benchmarks/Cargo.toml @@ -0,0 +1,23 @@ +[package] +name = "swactor-benchmarks" +version = "0.1.0" +edition = "2024" +license = "AGPL-3.0-only" +publish = false +autobenches = false + +[dependencies] +serde = { version = "1", features = ["derive"] } +swactor = { path = "../.." } +swactor-transport = { path = "../../crates/transport" } +telemetry = { path = "../../crates/telemetry" } + +[dev-dependencies] +criterion = { version = "0.5", default-features = false } + +[[bench]] +name = "criterion" +harness = false + +[lints] +workspace = true diff --git a/tools/benchmarks/benches/criterion.rs b/tools/benchmarks/benches/criterion.rs new file mode 100644 index 0000000..867841e --- /dev/null +++ b/tools/benchmarks/benches/criterion.rs @@ -0,0 +1,60 @@ +use std::hint::black_box; +use std::time::Duration; + +use criterion::measurement::WallTime; +use criterion::{ + BatchSize, BenchmarkGroup, BenchmarkId, Criterion, Throughput, criterion_group, criterion_main, +}; +use swactor_benchmarks::{SetupPolicy, WorkUnits, Workload, codec, telemetry, validate}; + +fn register(group: &mut BenchmarkGroup<'_, WallTime>, workload: &W) +where + W: Workload, +{ + validate(workload); + match workload.units() { + WorkUnits::Operations(operations) | WorkUnits::Frames(operations) => { + group.throughput(Throughput::Elements(operations)); + } + WorkUnits::Bytes(bytes) => { + group.throughput(Throughput::Bytes(bytes)); + } + } + + let batch_size = match workload.setup_policy() { + SetupPolicy::Batched => BatchSize::LargeInput, + SetupPolicy::PerExecution => BatchSize::PerIteration, + }; + group.bench_with_input( + BenchmarkId::from_parameter(workload.name()), + workload, + |bencher, workload| { + bencher.iter_batched( + || workload.setup(), + |mut state| black_box(workload.execute(&mut state)), + batch_size, + ); + }, + ); +} + +fn benchmark_workloads(criterion: &mut Criterion) { + let mut group = criterion.benchmark_group("swactor"); + for workload in codec::workloads() { + register(&mut group, &workload); + } + for workload in telemetry::workloads() { + register(&mut group, &workload); + } + group.finish(); +} + +criterion_group! { + name = benches; + config = Criterion::default() + .sample_size(20) + .warm_up_time(Duration::from_millis(500)) + .measurement_time(Duration::from_secs(1)); + targets = benchmark_workloads +} +criterion_main!(benches); diff --git a/tools/benchmarks/src/codec.rs b/tools/benchmarks/src/codec.rs new file mode 100644 index 0000000..b625f3b --- /dev/null +++ b/tools/benchmarks/src/codec.rs @@ -0,0 +1,641 @@ +use std::any::{Any, TypeId}; + +use crate::{WorkUnits, Workload}; +use serde::{Deserialize, Serialize}; +use swactor::Error; +use swactor::actor::ActorAddress; +use swactor_transport::{ + Codec, CodecRegistrationError, CodecRegistry, JsonCodec, NetworkMessage, WireEnvelope, +}; + +const PAYLOAD_SIZES: [usize; 3] = [32, 1024, 65_536]; + +#[derive(Clone, Debug, PartialEq, Eq)] +struct FixedMessage(u64); + +impl NetworkMessage for FixedMessage { + fn type_tag() -> &'static str { + "bench::FixedMessage" + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +struct BytesMessage(Vec); + +impl NetworkMessage for BytesMessage { + fn type_tag() -> &'static str { + "bench::BytesMessage" + } +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +struct StructuredMessage { + id: u64, + name: String, + fields: Vec, + nested: Vec>, +} + +impl NetworkMessage for StructuredMessage { + fn type_tag() -> &'static str { + "bench::StructuredMessage" + } +} + +#[derive(Clone, Copy)] +struct FixedCodec; + +impl Codec for FixedCodec { + fn encode(&self, message: &FixedMessage) -> Result, Error> { + Ok(message.0.to_le_bytes().to_vec()) + } + + fn decode(&self, bytes: &[u8]) -> Result { + let bytes: [u8; 8] = bytes + .try_into() + .map_err(|_| Error::from("fixed benchmark payload must contain eight bytes"))?; + Ok(FixedMessage(u64::from_le_bytes(bytes))) + } +} + +#[derive(Clone, Copy)] +struct BytesCodec; + +impl Codec for BytesCodec { + fn encode(&self, message: &BytesMessage) -> Result, Error> { + Ok(message.0.clone()) + } + + fn decode(&self, bytes: &[u8]) -> Result { + Ok(BytesMessage(bytes.to_vec())) + } +} + +#[derive(Clone, Copy, Debug)] +pub enum Format { + Fixed, + Bytes, + JsonFlat, + JsonNested, +} + +impl Format { + fn label(self) -> &'static str { + match self { + Self::Fixed => "fixed", + Self::Bytes => "bytes", + Self::JsonFlat => "json-flat", + Self::JsonNested => "json-nested", + } + } +} + +#[derive(Clone, Copy, Debug)] +pub enum DirectOperation { + Encode, + Decode, +} + +impl DirectOperation { + fn label(self) -> &'static str { + match self { + Self::Encode => "encode", + Self::Decode => "decode", + } + } +} + +#[derive(Clone, Copy, Debug)] +pub enum RegistryOperation { + Encode, + Decode, + Receive, +} + +impl RegistryOperation { + fn label(self) -> &'static str { + match self { + Self::Encode => "encode", + Self::Decode => "decode", + Self::Receive => "receive", + } + } +} + +#[derive(Clone, Copy, Debug)] +pub enum RegistrationKind { + Fresh, + DuplicateType, + DuplicateTag, +} + +impl RegistrationKind { + fn label(self) -> &'static str { + match self { + Self::Fresh => "fresh", + Self::DuplicateType => "duplicate-type", + Self::DuplicateTag => "duplicate-tag", + } + } +} + +#[derive(Clone, Copy, Debug)] +pub enum CodecWorkload { + Direct { + format: Format, + operation: DirectOperation, + size: usize, + }, + Registry { + format: Format, + operation: RegistryOperation, + size: usize, + }, + Registration(RegistrationKind), +} + +#[derive(Clone, Debug, PartialEq, Eq)] +enum CodecValue { + Fixed(FixedMessage), + Bytes(BytesMessage), + Structured(StructuredMessage), +} + +pub struct MessageState { + value: CodecValue, + encoded: Vec, + registry: Option, + destination: ActorAddress, +} + +pub struct RegistrationState { + registry: CodecRegistry, + kind: RegistrationKind, +} + +pub enum CodecState { + Message(MessageState), + Registration(RegistrationState), +} + +pub enum CodecOutput { + Bytes(Vec), + Tagged { + tag: String, + bytes: Vec, + }, + Decoded(Box), + Received { + destination: ActorAddress, + decoded: Box, + }, + Registration(Result<(), CodecRegistrationError>), +} + +fn deterministic_bytes(size: usize) -> Vec { + (0..size) + .map(|index| ((index.wrapping_mul(31) + 17) % 251) as u8) + .collect() +} + +fn structured_value(size: usize, nested: bool) -> StructuredMessage { + if !nested { + return StructuredMessage { + id: 0x5a5a_a5a5, + name: "f".repeat(size), + fields: Vec::new(), + nested: Vec::new(), + }; + } + + let width = 16; + let field_len = (size / (width * 2)).max(1); + let row_len = (size / (width * 16)).max(1); + StructuredMessage { + id: 0x5a5a_a5a5, + name: "nested".to_owned(), + fields: (0..width) + .map(|index| format!("{index:02}-{}", "w".repeat(field_len))) + .collect(), + nested: (0..width) + .map(|row| { + (0..row_len) + .map(|column| (row * row_len + column) as u64) + .collect() + }) + .collect(), + } +} + +fn value(format: Format, size: usize) -> CodecValue { + match format { + Format::Fixed => CodecValue::Fixed(FixedMessage(0x0123_4567_89ab_cdef)), + Format::Bytes => CodecValue::Bytes(BytesMessage(deterministic_bytes(size))), + Format::JsonFlat => CodecValue::Structured(structured_value(size, false)), + Format::JsonNested => CodecValue::Structured(structured_value(size, true)), + } +} + +fn direct_encode(value: &CodecValue) -> Vec { + match value { + CodecValue::Fixed(message) => FixedCodec.encode(message).unwrap(), + CodecValue::Bytes(message) => BytesCodec.encode(message).unwrap(), + CodecValue::Structured(message) => JsonCodec::::default() + .encode(message) + .unwrap(), + } +} + +fn direct_decode(template: &CodecValue, bytes: &[u8]) -> CodecValue { + match template { + CodecValue::Fixed(_) => CodecValue::Fixed(FixedCodec.decode(bytes).unwrap()), + CodecValue::Bytes(_) => CodecValue::Bytes(BytesCodec.decode(bytes).unwrap()), + CodecValue::Structured(_) => CodecValue::Structured( + JsonCodec::::default() + .decode(bytes) + .unwrap(), + ), + } +} + +fn register_value(registry: &mut CodecRegistry, value: &CodecValue) { + match value { + CodecValue::Fixed(_) => registry + .register::(FixedCodec) + .expect("fresh fixed benchmark registration"), + CodecValue::Bytes(_) => registry + .register::(BytesCodec) + .expect("fresh bytes benchmark registration"), + CodecValue::Structured(_) => registry + .register::(JsonCodec::::default()) + .expect("fresh JSON benchmark registration"), + } +} + +fn value_type_id(value: &CodecValue) -> TypeId { + match value { + CodecValue::Fixed(_) => TypeId::of::(), + CodecValue::Bytes(_) => TypeId::of::(), + CodecValue::Structured(_) => TypeId::of::(), + } +} + +fn value_tag(value: &CodecValue) -> &'static str { + match value { + CodecValue::Fixed(_) => FixedMessage::type_tag(), + CodecValue::Bytes(_) => BytesMessage::type_tag(), + CodecValue::Structured(_) => StructuredMessage::type_tag(), + } +} + +fn boxed_value(value: &CodecValue) -> Box { + match value { + CodecValue::Fixed(message) => Box::new(message.clone()), + CodecValue::Bytes(message) => Box::new(message.clone()), + CodecValue::Structured(message) => Box::new(message.clone()), + } +} + +fn decoded_matches(value: &CodecValue, decoded: &dyn Any) -> bool { + match value { + CodecValue::Fixed(expected) => decoded.downcast_ref::() == Some(expected), + CodecValue::Bytes(expected) => decoded.downcast_ref::() == Some(expected), + CodecValue::Structured(expected) => { + decoded.downcast_ref::() == Some(expected) + } + } +} + +fn verify_registration_state(state: &RegistrationState, kind: RegistrationKind) { + let probe = BytesMessage(deterministic_bytes(32)); + match kind { + RegistrationKind::Fresh => { + let (tag, bytes) = state + .registry + .encode(TypeId::of::(), Box::new(probe.clone())) + .unwrap(); + assert_eq!(tag, BytesMessage::type_tag()); + assert_eq!(bytes, probe.0); + } + RegistrationKind::DuplicateType => { + let (tag, bytes) = state + .registry + .encode(TypeId::of::(), Box::new(probe.clone())) + .unwrap(); + assert_eq!(tag, BytesMessage::type_tag()); + assert_eq!(bytes, probe.0); + assert!( + state + .registry + .decode(BytesMessage::type_tag(), &probe.0) + .is_err() + ); + } + RegistrationKind::DuplicateTag => { + assert!( + state + .registry + .encode(TypeId::of::(), Box::new(probe)) + .is_err() + ); + let decoded = state + .registry + .decode(BytesMessage::type_tag(), &0u64.to_le_bytes()) + .unwrap(); + assert_eq!( + *decoded.downcast::().unwrap(), + FixedMessage(0) + ); + } + } +} + +impl Workload for CodecWorkload { + type State = CodecState; + type Output = CodecOutput; + + fn name(&self) -> String { + match self { + Self::Direct { + format, + operation, + size, + } => format!( + "codec/direct/{}/{}/{}b", + format.label(), + operation.label(), + size + ), + Self::Registry { + format, + operation, + size, + } => format!( + "codec/registry/{}/{}/{}b", + format.label(), + operation.label(), + size + ), + Self::Registration(kind) => { + format!("codec/registration/{}", kind.label()) + } + } + } + + fn setup(&self) -> Self::State { + match self { + Self::Direct { format, size, .. } => { + let value = value(*format, *size); + let encoded = direct_encode(&value); + CodecState::Message(MessageState { + value, + encoded, + registry: None, + destination: ActorAddress([0x5a; 32]), + }) + } + Self::Registry { format, size, .. } => { + let value = value(*format, *size); + let encoded = direct_encode(&value); + let mut registry = CodecRegistry::new(); + register_value(&mut registry, &value); + CodecState::Message(MessageState { + value, + encoded, + registry: Some(registry), + destination: ActorAddress([0x5a; 32]), + }) + } + Self::Registration(kind) => { + let mut registry = CodecRegistry::new(); + match kind { + RegistrationKind::Fresh => {} + RegistrationKind::DuplicateType => registry + .register_encoder::(|message| { + Ok((BytesMessage::type_tag().to_owned(), message.0.clone())) + }) + .expect("initial benchmark encoder"), + RegistrationKind::DuplicateTag => registry + .register_decoder::(BytesMessage::type_tag(), |bytes| { + FixedCodec.decode(bytes) + }) + .expect("initial benchmark decoder"), + } + CodecState::Registration(RegistrationState { + registry, + kind: *kind, + }) + } + } + } + + fn execute(&self, state: &mut Self::State) -> Self::Output { + match (self, state) { + ( + Self::Direct { + operation: DirectOperation::Encode, + .. + }, + CodecState::Message(state), + ) => CodecOutput::Bytes(direct_encode(&state.value)), + ( + Self::Direct { + operation: DirectOperation::Decode, + .. + }, + CodecState::Message(state), + ) => CodecOutput::Decoded(boxed_value(&direct_decode(&state.value, &state.encoded))), + ( + Self::Registry { + operation: RegistryOperation::Encode, + .. + }, + CodecState::Message(state), + ) => { + let (tag, bytes) = state + .registry + .as_ref() + .unwrap() + .encode(value_type_id(&state.value), boxed_value(&state.value)) + .unwrap(); + CodecOutput::Tagged { tag, bytes } + } + ( + Self::Registry { + operation: RegistryOperation::Decode, + .. + }, + CodecState::Message(state), + ) => CodecOutput::Decoded( + state + .registry + .as_ref() + .unwrap() + .decode(value_tag(&state.value), &state.encoded) + .unwrap(), + ), + ( + Self::Registry { + operation: RegistryOperation::Receive, + .. + }, + CodecState::Message(state), + ) => { + let (destination, decoded) = state + .registry + .as_ref() + .unwrap() + .receive(WireEnvelope { + dest: state.destination, + type_tag: value_tag(&state.value).to_owned(), + payload: state.encoded.clone(), + }) + .unwrap(); + CodecOutput::Received { + destination, + decoded, + } + } + (Self::Registration(_), CodecState::Registration(state)) => { + CodecOutput::Registration(state.registry.register::(BytesCodec)) + } + _ => panic!("codec workload and state mismatch"), + } + } + + fn verify(&self, state: &Self::State, output: &Self::Output) { + match (self, state, output) { + ( + Self::Direct { + operation: DirectOperation::Encode, + .. + }, + CodecState::Message(state), + CodecOutput::Bytes(bytes), + ) => assert_eq!(bytes, &state.encoded), + ( + Self::Direct { + operation: DirectOperation::Decode, + .. + } + | Self::Registry { + operation: RegistryOperation::Decode, + .. + }, + CodecState::Message(state), + CodecOutput::Decoded(decoded), + ) => assert!(decoded_matches(&state.value, decoded.as_ref())), + ( + Self::Registry { + operation: RegistryOperation::Encode, + .. + }, + CodecState::Message(state), + CodecOutput::Tagged { tag, bytes }, + ) => { + assert_eq!(tag, value_tag(&state.value)); + assert_eq!(bytes, &state.encoded); + } + ( + Self::Registry { + operation: RegistryOperation::Receive, + .. + }, + CodecState::Message(state), + CodecOutput::Received { + destination, + decoded, + }, + ) => { + assert_eq!(*destination, state.destination); + assert!(decoded_matches(&state.value, decoded.as_ref())); + } + ( + Self::Registration(kind), + CodecState::Registration(state), + CodecOutput::Registration(result), + ) => { + assert_eq!(state.kind.label(), kind.label()); + match kind { + RegistrationKind::Fresh => assert!(result.is_ok()), + RegistrationKind::DuplicateType => assert!(matches!( + result, + Err(CodecRegistrationError::EncoderAlreadyRegistered { .. }) + )), + RegistrationKind::DuplicateTag => assert!(matches!( + result, + Err(CodecRegistrationError::DecoderAlreadyRegistered { .. }) + )), + } + verify_registration_state(state, *kind); + } + _ => panic!("codec workload, state, and output mismatch"), + } + } + + fn units(&self) -> WorkUnits { + match self { + Self::Registration(_) => WorkUnits::Operations(1), + Self::Direct { format, size, .. } | Self::Registry { format, size, .. } => { + WorkUnits::Bytes(direct_encode(&value(*format, *size)).len() as u64) + } + } + } +} + +pub fn workloads() -> Vec { + let mut workloads = Vec::new(); + + for operation in [DirectOperation::Encode, DirectOperation::Decode] { + workloads.push(CodecWorkload::Direct { + format: Format::Fixed, + operation, + size: 8, + }); + } + for format in [Format::Bytes, Format::JsonFlat, Format::JsonNested] { + for size in PAYLOAD_SIZES { + for operation in [DirectOperation::Encode, DirectOperation::Decode] { + workloads.push(CodecWorkload::Direct { + format, + operation, + size, + }); + } + } + } + + for operation in [ + RegistryOperation::Encode, + RegistryOperation::Decode, + RegistryOperation::Receive, + ] { + workloads.push(CodecWorkload::Registry { + format: Format::Fixed, + operation, + size: 8, + }); + } + for format in [Format::Bytes, Format::JsonFlat, Format::JsonNested] { + for size in PAYLOAD_SIZES { + for operation in [ + RegistryOperation::Encode, + RegistryOperation::Decode, + RegistryOperation::Receive, + ] { + workloads.push(CodecWorkload::Registry { + format, + operation, + size, + }); + } + } + } + + workloads.extend([ + CodecWorkload::Registration(RegistrationKind::Fresh), + CodecWorkload::Registration(RegistrationKind::DuplicateType), + CodecWorkload::Registration(RegistrationKind::DuplicateTag), + ]); + workloads +} diff --git a/tools/benchmarks/src/lib.rs b/tools/benchmarks/src/lib.rs new file mode 100644 index 0000000..d5b43c5 --- /dev/null +++ b/tools/benchmarks/src/lib.rs @@ -0,0 +1,60 @@ +pub mod codec; +pub mod telemetry; +pub mod workload; + +pub use workload::{SetupPolicy, WorkUnits, Workload, validate}; + +pub const BENCHMARK_COMMAND: &str = "cargo bench -p swactor-benchmarks --bench criterion"; +pub const INITIAL_BASELINE_COMMAND: &str = + "cargo bench -p swactor-benchmarks --bench criterion -- --save-baseline initial"; + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn every_workload_passes_its_correctness_check() { + for workload in codec::workloads() { + validate(&workload); + } + for workload in telemetry::workloads() { + validate(&workload); + } + } + + #[test] + fn benchmark_names_are_unique_and_stable() { + let codec_names: Vec = codec::workloads() + .into_iter() + .map(|workload| workload.name()) + .collect(); + let telemetry_names: Vec = telemetry::workloads() + .into_iter() + .map(|workload| workload.name()) + .collect(); + assert_eq!(codec_names.len(), 53); + assert_eq!(telemetry_names.len(), 31); + assert!(codec_names.contains(&"codec/direct/fixed/encode/8b".to_owned())); + assert!(codec_names.contains(&"codec/registry/json-nested/receive/65536b".to_owned())); + assert!( + telemetry_names.contains(&"telemetry/mux/submit-drain/65536b/batch-1024".to_owned()) + ); + assert!( + telemetry_names.contains(&"telemetry/mux/concurrent/producers-8/total-4096".to_owned()) + ); + + let mut names = codec_names; + names.extend(telemetry_names); + let total = names.len(); + names.sort(); + names.dedup(); + assert_eq!(names.len(), total); + assert!(names.iter().all(|name| !name.contains(char::is_whitespace))); + } + + #[test] + fn benchmark_commands_name_the_package() { + assert!(BENCHMARK_COMMAND.contains("-p swactor-benchmarks")); + assert!(INITIAL_BASELINE_COMMAND.contains("--save-baseline initial")); + } +} diff --git a/tools/benchmarks/src/telemetry.rs b/tools/benchmarks/src/telemetry.rs new file mode 100644 index 0000000..db7e229 --- /dev/null +++ b/tools/benchmarks/src/telemetry.rs @@ -0,0 +1,755 @@ +use std::collections::HashSet; +use std::sync::{Arc, Barrier}; +use std::thread::{self, JoinHandle}; + +use telemetry::frame::{Frame, TelemetryEvent}; +use telemetry::ingest::Consumer; +use telemetry::transport::Delivery; +use telemetry::wire::{decode_delivery, encode_delivery}; +use telemetry::{ + ChannelContent, ChannelId, Lifetime, Mux, Position, StreamId, TelemetryEndpoint, + TelemetrySubscription, +}; + +use crate::{SetupPolicy, WorkUnits, Workload}; + +const PAYLOAD_SIZES: [usize; 3] = [32, 1024, 65_536]; +const BATCH_SIZES: [usize; 3] = [1, 64, 1024]; +const STORE_BATCH: usize = 1024; +const CONCURRENT_TOTAL: usize = 4096; +const CONCURRENT_PAYLOAD_SIZE: usize = 32; + +#[derive(Clone, Copy, Debug)] +pub enum StoreKind { + Ordered, + Reordered, + Duplicate, + Gaps, +} + +impl StoreKind { + fn label(self) -> &'static str { + match self { + Self::Ordered => "ordered", + Self::Reordered => "reordered", + Self::Duplicate => "duplicate", + Self::Gaps => "gaps", + } + } +} + +#[derive(Clone, Copy, Debug)] +pub enum WireOperation { + Encode, + Decode, +} + +impl WireOperation { + fn label(self) -> &'static str { + match self { + Self::Encode => "encode", + Self::Decode => "decode", + } + } +} + +#[derive(Clone, Copy, Debug)] +pub enum TelemetryWorkload { + Mux { + payload_size: usize, + batch: usize, + reject_full: bool, + }, + Fanout { + payload_size: usize, + batch: usize, + subscribers: usize, + slow_subscriber: bool, + }, + Store(StoreKind), + Wire { + operation: WireOperation, + payload_size: usize, + }, + Concurrent { + producers: usize, + }, +} + +pub struct MuxState { + mux: Mux, + payload: Vec, + batch: usize, + reject_full: bool, +} + +pub struct FanoutState { + endpoint: TelemetryEndpoint, + channel: ChannelId, + subscriptions: Vec, + payload: Vec, + batch: usize, + slow_subscriber: bool, +} + +pub struct StoreState { + consumer: Consumer, + stream: StreamId, + deliveries: Vec, + kind: StoreKind, +} + +pub struct WireState { + stream: StreamId, + frame: Frame, + encoded: Vec, +} + +pub struct ConcurrentState { + endpoint: TelemetryEndpoint, + start: Arc, + handles: Vec>, + total: usize, +} + +pub enum TelemetryState { + Mux(MuxState), + Fanout(FanoutState), + Store(StoreState), + Wire(WireState), + Concurrent(ConcurrentState), +} + +pub enum TelemetryOutput { + Mux { + accepted: usize, + rejected: usize, + dropped: u64, + frames: Vec, + }, + Fanout { + accepted: usize, + drained: usize, + dropped_for_subscribers: usize, + events: Vec>, + drop_counts: Vec, + bitbucketed: u64, + }, + Store { + attempted: usize, + accepted: usize, + frames: Vec, + gaps: Vec<(u64, u64)>, + }, + WireEncoded(Vec), + WireDecoded(StreamId, Frame), + Concurrent { + accepted: usize, + frames: Vec, + dropped: u64, + }, +} + +fn deterministic_payload(size: usize) -> Vec { + (0..size) + .map(|index| ((index.wrapping_mul(29) + 11) % 251) as u8) + .collect() +} + +fn marked_payload(size: usize, marker: u64) -> Vec { + let mut payload = deterministic_payload(size.max(8)); + payload[..8].copy_from_slice(&marker.to_le_bytes()); + payload.truncate(size.max(8)); + payload +} + +fn marker(payload: &[u8]) -> u64 { + u64::from_le_bytes(payload[..8].try_into().unwrap()) +} + +fn contiguous(positions: &[u64]) -> bool { + positions.windows(2).all(|pair| pair[1] == pair[0] + 1) +} + +fn setup_mux(payload_size: usize, batch: usize, reject_full: bool) -> MuxState { + let mux = Mux::new(StreamId::new("bench-mux", Lifetime(1)), batch.max(1)); + let payload = deterministic_payload(payload_size); + if reject_full { + for _ in 0..batch { + assert!(mux.submit(ChannelId(1), payload.clone())); + } + } + MuxState { + mux, + payload, + batch, + reject_full, + } +} + +fn setup_fanout( + payload_size: usize, + batch: usize, + subscribers: usize, + slow_subscriber: bool, +) -> FanoutState { + let endpoint = TelemetryEndpoint::with_capacity( + StreamId::new("bench-endpoint", Lifetime(1)), + batch, + batch, + ); + let channel = endpoint.register_channel("payload", ChannelContent::Bytes); + let subscriptions = if slow_subscriber { + vec![ + endpoint.subscribe_all_with_capacity("slow", 1), + endpoint.subscribe_all_with_capacity("fast", batch), + ] + } else { + (0..subscribers) + .map(|index| endpoint.subscribe_all_with_capacity(format!("subscriber-{index}"), batch)) + .collect() + }; + FanoutState { + endpoint, + channel, + subscriptions, + payload: deterministic_payload(payload_size), + batch, + slow_subscriber, + } +} + +fn store_deliveries(kind: StoreKind) -> (StreamId, Vec) { + let stream = StreamId::new("bench-store", Lifetime(1)); + let positions: Vec = match kind { + StoreKind::Ordered | StoreKind::Duplicate => (0..STORE_BATCH as u64).collect(), + StoreKind::Reordered => (0..STORE_BATCH as u64).rev().collect(), + StoreKind::Gaps => (0..STORE_BATCH as u64) + .map(|position| position * 3) + .collect(), + }; + let deliveries = positions + .into_iter() + .map(|position| { + Delivery::new( + stream.clone(), + Frame { + channel: ChannelId(1), + position: Position(position), + payload: marked_payload(1024, position), + }, + ) + }) + .collect(); + (stream, deliveries) +} + +fn setup_store(kind: StoreKind) -> StoreState { + let (stream, deliveries) = store_deliveries(kind); + let mut consumer = Consumer::new(); + if matches!(kind, StoreKind::Gaps) { + for delivery in &deliveries { + assert!(consumer.accept(delivery.clone())); + } + } + StoreState { + consumer, + stream, + deliveries, + kind, + } +} + +fn setup_wire(payload_size: usize) -> WireState { + let stream = StreamId::new("bench-wire", Lifetime(7)); + let frame = Frame { + channel: ChannelId(3), + position: Position(11), + payload: deterministic_payload(payload_size), + }; + let encoded = encode_delivery(&stream, &frame); + WireState { + stream, + frame, + encoded, + } +} + +fn setup_concurrent(producers: usize) -> ConcurrentState { + let endpoint = TelemetryEndpoint::with_capacity( + StreamId::new("bench-concurrent", Lifetime(1)), + CONCURRENT_TOTAL, + 1, + ); + let start = Arc::new(Barrier::new(producers + 1)); + let per_producer = CONCURRENT_TOTAL / producers; + let mut handles = Vec::with_capacity(producers); + for producer_index in 0..producers { + let producer = endpoint.producer(); + let start = Arc::clone(&start); + let payloads: Vec> = (0..per_producer) + .map(|index| { + marked_payload( + CONCURRENT_PAYLOAD_SIZE, + (producer_index * per_producer + index) as u64, + ) + }) + .collect(); + handles.push(thread::spawn(move || { + start.wait(); + payloads + .into_iter() + .map(|payload| usize::from(producer.submit_bytes(ChannelId(1), payload))) + .sum() + })); + } + ConcurrentState { + endpoint, + start, + handles, + total: per_producer * producers, + } +} + +fn execute_mux(state: &mut MuxState) -> TelemetryOutput { + if state.reject_full { + let rejected = (0..state.batch) + .filter(|_| !state.mux.submit(ChannelId(1), state.payload.clone())) + .count(); + return TelemetryOutput::Mux { + accepted: 0, + rejected, + dropped: state.mux.dropped(), + frames: Vec::new(), + }; + } + + let accepted = (0..state.batch) + .filter(|_| state.mux.submit(ChannelId(1), state.payload.clone())) + .count(); + TelemetryOutput::Mux { + accepted, + rejected: state.batch - accepted, + dropped: state.mux.dropped(), + frames: state.mux.drain(), + } +} + +fn execute_fanout(state: &mut FanoutState) -> TelemetryOutput { + let producer = state.endpoint.producer(); + let accepted = (0..state.batch) + .filter(|_| producer.submit_bytes(state.channel, state.payload.clone())) + .count(); + let tick = state.endpoint.tick(); + let events = state + .subscriptions + .iter() + .map(TelemetrySubscription::drain_available) + .collect(); + let drop_counts = state + .endpoint + .subscriber_snapshots() + .into_iter() + .map(|snapshot| snapshot.dropped) + .collect(); + TelemetryOutput::Fanout { + accepted, + drained: tick.drained, + dropped_for_subscribers: tick.dropped_for_subscribers, + events, + drop_counts, + bitbucketed: state.endpoint.bitbucketed(), + } +} + +fn execute_store(state: &mut StoreState) -> TelemetryOutput { + if matches!(state.kind, StoreKind::Gaps) { + let stored = state.consumer.store().stream(&state.stream).unwrap(); + return TelemetryOutput::Store { + attempted: 0, + accepted: 0, + frames: stored.to_vec(), + gaps: stored + .gap_spans() + .into_iter() + .map(|gap| (gap.start, gap.end)) + .collect(), + }; + } + + let mut attempted = 0; + let mut accepted = 0; + for delivery in &state.deliveries { + attempted += 1; + accepted += usize::from(state.consumer.accept(delivery.clone())); + if matches!(state.kind, StoreKind::Duplicate) { + attempted += 1; + accepted += usize::from(state.consumer.accept(delivery.clone())); + } + } + let stored = state.consumer.store().stream(&state.stream).unwrap(); + TelemetryOutput::Store { + attempted, + accepted, + frames: stored.to_vec(), + gaps: Vec::new(), + } +} + +fn execute_wire(state: &WireState, operation: WireOperation) -> TelemetryOutput { + match operation { + WireOperation::Encode => { + TelemetryOutput::WireEncoded(encode_delivery(&state.stream, &state.frame)) + } + WireOperation::Decode => { + let (stream, frame) = decode_delivery(&state.encoded).unwrap(); + TelemetryOutput::WireDecoded(stream, frame) + } + } +} + +fn execute_concurrent(state: &mut ConcurrentState) -> TelemetryOutput { + state.start.wait(); + let accepted = state + .handles + .drain(..) + .map(|handle| handle.join().unwrap()) + .sum(); + TelemetryOutput::Concurrent { + accepted, + frames: state.endpoint.mux().drain(), + dropped: state.endpoint.mux_dropped(), + } +} + +impl Workload for TelemetryWorkload { + type State = TelemetryState; + type Output = TelemetryOutput; + + fn name(&self) -> String { + match self { + Self::Mux { + payload_size, + batch, + reject_full, + } => format!( + "telemetry/mux/{}/{payload_size}b/batch-{batch}", + if *reject_full { + "reject-full" + } else { + "submit-drain" + } + ), + Self::Fanout { + payload_size, + batch, + subscribers, + slow_subscriber, + } => format!( + "telemetry/fanout/{}/{payload_size}b/batch-{batch}", + if *slow_subscriber { + "slow-plus-fast".to_owned() + } else { + format!("subscribers-{subscribers}") + } + ), + Self::Store(kind) => format!("telemetry/store/{}", kind.label()), + Self::Wire { + operation, + payload_size, + } => format!("telemetry/wire/{}/{}b", operation.label(), payload_size), + Self::Concurrent { producers } => { + format!("telemetry/mux/concurrent/producers-{producers}/total-{CONCURRENT_TOTAL}") + } + } + } + + fn setup(&self) -> Self::State { + match self { + Self::Mux { + payload_size, + batch, + reject_full, + } => TelemetryState::Mux(setup_mux(*payload_size, *batch, *reject_full)), + Self::Fanout { + payload_size, + batch, + subscribers, + slow_subscriber, + } => TelemetryState::Fanout(setup_fanout( + *payload_size, + *batch, + *subscribers, + *slow_subscriber, + )), + Self::Store(kind) => TelemetryState::Store(setup_store(*kind)), + Self::Wire { payload_size, .. } => TelemetryState::Wire(setup_wire(*payload_size)), + Self::Concurrent { producers } => { + TelemetryState::Concurrent(setup_concurrent(*producers)) + } + } + } + + fn execute(&self, state: &mut Self::State) -> Self::Output { + match (self, state) { + (Self::Mux { .. }, TelemetryState::Mux(state)) => execute_mux(state), + (Self::Fanout { .. }, TelemetryState::Fanout(state)) => execute_fanout(state), + (Self::Store(_), TelemetryState::Store(state)) => execute_store(state), + (Self::Wire { operation, .. }, TelemetryState::Wire(state)) => { + execute_wire(state, *operation) + } + (Self::Concurrent { .. }, TelemetryState::Concurrent(state)) => { + execute_concurrent(state) + } + _ => panic!("telemetry workload and state mismatch"), + } + } + + fn verify(&self, state: &Self::State, output: &Self::Output) { + match (state, output) { + ( + TelemetryState::Mux(state), + TelemetryOutput::Mux { + accepted, + rejected, + dropped, + frames, + }, + ) => { + if state.reject_full { + assert_eq!(*accepted, 0); + assert_eq!(*rejected, state.batch); + assert_eq!(*dropped, state.batch as u64); + assert!(frames.is_empty()); + } else { + assert_eq!(*accepted, state.batch); + assert_eq!(*rejected, 0); + assert_eq!(*dropped, 0); + assert_eq!(frames.len(), state.batch); + let positions: Vec = frames.iter().map(|frame| frame.position.0).collect(); + assert!(contiguous(&positions)); + assert!(frames.iter().all(|frame| frame.payload == state.payload)); + } + } + ( + TelemetryState::Fanout(state), + TelemetryOutput::Fanout { + accepted, + drained, + dropped_for_subscribers, + events, + drop_counts, + bitbucketed, + }, + ) => { + assert_eq!(*accepted, state.batch); + assert_eq!(*drained, state.batch); + assert_eq!(events.len(), state.subscriptions.len()); + let expected_dropped = if state.slow_subscriber { + state.batch.saturating_sub(1) + } else { + 0 + }; + assert_eq!(*dropped_for_subscribers, expected_dropped); + assert_eq!( + drop_counts.iter().copied().sum::(), + expected_dropped as u64 + ); + for (index, events) in events.iter().enumerate() { + let frames: Vec<_> = events + .iter() + .map(|event| match event { + TelemetryEvent::Frame(frame) => frame, + other => panic!("unexpected fanout event: {other:?}"), + }) + .collect(); + let expected = if state.slow_subscriber && index == 0 { + usize::from(state.batch > 0) + } else { + state.batch + }; + assert_eq!(frames.len(), expected); + let positions: Vec = frames.iter().map(|frame| frame.position.0).collect(); + assert!(contiguous(&positions)); + assert!(frames.iter().all(|frame| frame.payload == state.payload)); + } + assert_eq!( + *bitbucketed, + if state.subscriptions.is_empty() { + state.batch as u64 + } else { + 0 + } + ); + } + ( + TelemetryState::Store(state), + TelemetryOutput::Store { + attempted, + accepted, + frames, + gaps, + }, + ) => { + assert_eq!(frames.len(), STORE_BATCH); + let positions: Vec = frames.iter().map(|frame| frame.position.0).collect(); + assert!(positions.windows(2).all(|pair| pair[0] < pair[1])); + assert!( + frames + .iter() + .all(|frame| marker(&frame.payload) == frame.position.0) + ); + if matches!(state.kind, StoreKind::Gaps) { + assert_eq!(*attempted, 0); + assert_eq!(*accepted, 0); + assert_eq!(gaps.len(), STORE_BATCH - 1); + assert!(gaps.iter().all(|(start, end)| *end == *start + 1)); + } else { + assert_eq!( + *attempted, + if matches!(state.kind, StoreKind::Duplicate) { + STORE_BATCH * 2 + } else { + STORE_BATCH + } + ); + assert_eq!(*accepted, STORE_BATCH); + assert!(gaps.is_empty()); + assert!(contiguous(&positions)); + } + } + (TelemetryState::Wire(state), TelemetryOutput::WireEncoded(encoded)) => { + assert_eq!(encoded, &state.encoded) + } + (TelemetryState::Wire(state), TelemetryOutput::WireDecoded(stream, frame)) => { + assert_eq!(stream, &state.stream); + assert_eq!(frame, &state.frame); + } + ( + TelemetryState::Concurrent(state), + TelemetryOutput::Concurrent { + accepted, + frames, + dropped, + }, + ) => { + assert_eq!(*accepted, state.total); + assert_eq!(*dropped, 0); + assert_eq!(frames.len(), state.total); + let positions: Vec = frames.iter().map(|frame| frame.position.0).collect(); + let payload_ids: HashSet = + frames.iter().map(|frame| marker(&frame.payload)).collect(); + assert!(contiguous(&positions)); + assert_eq!(payload_ids.len(), state.total); + assert_eq!( + frames + .iter() + .map(|frame| frame.payload.len()) + .sum::(), + state.total * CONCURRENT_PAYLOAD_SIZE + ); + } + _ => panic!("telemetry workload, state, and output mismatch"), + } + } + + fn units(&self) -> WorkUnits { + match self { + Self::Mux { + payload_size, + batch, + reject_full, + } => { + if *reject_full { + WorkUnits::Operations(*batch as u64) + } else { + WorkUnits::Bytes((*payload_size * *batch) as u64) + } + } + Self::Fanout { + payload_size, + batch, + subscribers, + slow_subscriber, + } => { + let copies = if *slow_subscriber { + 2 + } else { + (*subscribers).max(1) + }; + WorkUnits::Bytes((*payload_size * *batch * copies) as u64) + } + Self::Store(kind) => WorkUnits::Frames(if matches!(kind, StoreKind::Duplicate) { + (STORE_BATCH * 2) as u64 + } else { + STORE_BATCH as u64 + }), + Self::Wire { payload_size, .. } => WorkUnits::Bytes(*payload_size as u64), + Self::Concurrent { .. } => WorkUnits::Operations(CONCURRENT_TOTAL as u64), + } + } + + fn setup_policy(&self) -> SetupPolicy { + if matches!(self, Self::Concurrent { .. }) { + SetupPolicy::PerExecution + } else { + SetupPolicy::Batched + } + } +} + +pub fn workloads() -> Vec { + let mut workloads = Vec::new(); + for payload_size in PAYLOAD_SIZES { + for batch in BATCH_SIZES { + workloads.push(TelemetryWorkload::Mux { + payload_size, + batch, + reject_full: false, + }); + } + workloads.push(TelemetryWorkload::Mux { + payload_size, + batch: 64, + reject_full: true, + }); + } + + for subscribers in [0, 1, 8, 64] { + workloads.push(TelemetryWorkload::Fanout { + payload_size: 1024, + batch: 64, + subscribers, + slow_subscriber: false, + }); + } + workloads.push(TelemetryWorkload::Fanout { + payload_size: 1024, + batch: 64, + subscribers: 2, + slow_subscriber: true, + }); + + workloads.extend([ + TelemetryWorkload::Store(StoreKind::Ordered), + TelemetryWorkload::Store(StoreKind::Reordered), + TelemetryWorkload::Store(StoreKind::Duplicate), + TelemetryWorkload::Store(StoreKind::Gaps), + ]); + + for payload_size in PAYLOAD_SIZES { + workloads.push(TelemetryWorkload::Wire { + operation: WireOperation::Encode, + payload_size, + }); + workloads.push(TelemetryWorkload::Wire { + operation: WireOperation::Decode, + payload_size, + }); + } + + for producers in [1, 2, 4, 8] { + workloads.push(TelemetryWorkload::Concurrent { producers }); + } + workloads +} diff --git a/tools/benchmarks/src/workload.rs b/tools/benchmarks/src/workload.rs new file mode 100644 index 0000000..03d8610 --- /dev/null +++ b/tools/benchmarks/src/workload.rs @@ -0,0 +1,50 @@ +/// Engine-neutral throughput associated with one workload execution. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum WorkUnits { + Operations(u64), + Frames(u64), + Bytes(u64), +} + +impl WorkUnits { + pub fn amount(self) -> u64 { + match self { + Self::Operations(amount) | Self::Frames(amount) | Self::Bytes(amount) => amount, + } + } +} + +/// How independently each measured execution must be prepared. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub enum SetupPolicy { + #[default] + Batched, + PerExecution, +} + +/// A benchmark workload independent of its timing and reporting engine. +pub trait Workload { + type State; + type Output; + + fn name(&self) -> String; + fn setup(&self) -> Self::State; + fn execute(&self, state: &mut Self::State) -> Self::Output; + fn verify(&self, state: &Self::State, output: &Self::Output); + fn units(&self) -> WorkUnits; + + fn setup_policy(&self) -> SetupPolicy { + SetupPolicy::Batched + } +} + +/// Exercise correctness before an engine begins timing a workload. +pub fn validate(workload: &W) { + assert!( + workload.units().amount() > 0, + "workload units must be nonzero" + ); + let mut state = workload.setup(); + let output = workload.execute(&mut state); + workload.verify(&state, &output); +} diff --git a/xtask/src/demo/mod.rs b/xtask/src/demo/mod.rs index b021d9e..3bb5bda 100644 --- a/xtask/src/demo/mod.rs +++ b/xtask/src/demo/mod.rs @@ -334,14 +334,18 @@ fn run_supervisor(args: &[String]) -> Result<(), String> { )) .map_err(|e| format!("spawn edge ack relay: {e}"))?; let mut codec = CodecRegistry::new(); - codec.register_decoder::(node::ANNOUNCE_TAG, |bytes| { - serde_json::from_slice(bytes) - .map_err(|e| swactor::Error::from(format!("announce decode: {e}"))) - }); - codec.register_decoder::(edge::EDGE_ACK_TAG, |bytes| { - serde_json::from_slice(bytes) - .map_err(|e| swactor::Error::from(format!("edge ack decode: {e}"))) - }); + codec + .register_decoder::(node::ANNOUNCE_TAG, |bytes| { + serde_json::from_slice(bytes) + .map_err(|e| swactor::Error::from(format!("announce decode: {e}"))) + }) + .expect("unique codec decoder registration"); + codec + .register_decoder::(edge::EDGE_ACK_TAG, |bytes| { + serde_json::from_slice(bytes) + .map_err(|e| swactor::Error::from(format!("edge ack decode: {e}"))) + }) + .expect("unique codec decoder registration"); let mut routes = std::collections::HashMap::new(); routes.insert(node::ANNOUNCE_TAG.to_owned(), announce); routes.insert(edge::EDGE_ACK_TAG.to_owned(), ack_relay); diff --git a/xtask/src/demo/node.rs b/xtask/src/demo/node.rs index 3ac7cb8..b7e0a2f 100644 --- a/xtask/src/demo/node.rs +++ b/xtask/src/demo/node.rs @@ -167,11 +167,13 @@ pub fn run_node_role(supervisor_addr_json: &str, attempt: u64) -> Result<(), Str use distribution::transport_bridge::{Outbox, RelayMirror, RouteView}; use swactor_transport::CodecRegistry; let mut codec = CodecRegistry::new(); - codec.register_decoder::(edge::EDGE_PROVISION_TAG, |bytes| { - let provision: edge::EdgeProvision = serde_json::from_slice(bytes) - .map_err(|e| swactor::Error::from(format!("edge provision decode: {e}")))?; - Ok(edge::NodeEdgeMsg::Provision(provision)) - }); + codec + .register_decoder::(edge::EDGE_PROVISION_TAG, |bytes| { + let provision: edge::EdgeProvision = serde_json::from_slice(bytes) + .map_err(|e| swactor::Error::from(format!("edge provision decode: {e}")))?; + Ok(edge::NodeEdgeMsg::Provision(provision)) + }) + .expect("unique codec decoder registration"); let mut routes = std::collections::HashMap::new(); routes.insert(edge::EDGE_PROVISION_TAG.to_owned(), edge_agent); let relay_mirror: RelayMirror =