diff --git a/concurrency/src/lib.rs b/concurrency/src/lib.rs index 55c0355658..575e59d0ec 100644 --- a/concurrency/src/lib.rs +++ b/concurrency/src/lib.rs @@ -38,13 +38,6 @@ //! schedule that `parking_lot` permits. Tests that hinge on that //! interleaving need `RwLock` with explicit `read()` then //! `write()`, or a richer state machine in the facade. -//! * **`static FOO: Mutex = Mutex::new(...)` does not compile -//! under loom.** `loom::sync::Mutex::new` is plain `fn`, not -//! `const fn`, so a static initialiser fails to typecheck. Use -//! `OnceLock` for the static (the facade re-exports -//! `std::sync::OnceLock` under all backends) or move the -//! construction into a runtime initialiser gated by -//! `#[concurrency_mode(std)]`. //! * **`OnceLock` under loom/shuttle** is the real `std::sync::OnceLock`, //! not a model-aware shim. Loom and shuttle do not see the //! atomics inside `OnceLock::get_or_init`, so tests whose diff --git a/concurrency/src/sync/mod.rs b/concurrency/src/sync/mod.rs index 711b95d991..8fc333f723 100644 --- a/concurrency/src/sync/mod.rs +++ b/concurrency/src/sync/mod.rs @@ -36,8 +36,6 @@ //! `const fn`, but the facade exposes the lowest common //! denominator. So `static M: Mutex = Mutex::new(...)` compiles //! under the default and `parking_lot` backends and fails to -//! typecheck under the model-checker backends. Workaround for -//! tests that need a static: wrap the static in `OnceLock`. //! //! * **`OnceLock` under `loom`/`shuttle*` is re-exported from //! `std::sync` unchanged.** It is sound for laziness, but it uses diff --git a/config/src/external/overlay/algebra.rs b/config/src/external/overlay/algebra.rs index 9516fe5e7d..5fcf22bc4a 100644 --- a/config/src/external/overlay/algebra.rs +++ b/config/src/external/overlay/algebra.rs @@ -291,6 +291,16 @@ impl Footprint { pub fn is_empty(&self) -> bool { self.vpcs.is_empty() && self.peerings.is_empty() } + + #[must_use] + pub fn touches_vpc_named(&self, name: &str) -> bool { + self.vpcs.iter().any(|vpc| vpc.name() == name) + } + + #[must_use] + pub fn touches_peering_named(&self, name: &str) -> bool { + self.peerings.iter().any(|peering| peering.name() == name) + } } #[derive(Clone, Copy, PartialEq, Eq, Debug)] @@ -970,6 +980,36 @@ mod tests { use bolero::check; use concurrency::sync::atomic::{AtomicUsize, Ordering::Relaxed}; + #[test] + fn writing_a_vpc_does_not_imply_writing_its_peerings() { + static SEEN: AtomicUsize = AtomicUsize::new(0); + + check!() + .with_generator(Sequence::default()) + .for_each(|ops: &Vec| { + let mut draft = Draft::new(); + for op in ops { + let footprint = op.writes(&draft); + for vpc in &footprint.vpcs { + for (handle, spec) in draft.peerings() { + if spec.touches(*vpc) && !footprint.peerings.contains(&handle) { + SEEN.fetch_add(1, Relaxed); + } + } + } + op.apply(&mut draft).expect("a drawn operation applies"); + } + }); + + assert!( + SEEN.load(Relaxed) > 0, + "no drawn operation ever wrote a vpc while leaving one of its peerings unwritten. \ + Either the vocabulary changed and a peering-only frame filter is now sound -- in \ + which case say so where the filter is written -- or the generator stopped drawing \ + `AddPeering` against a vpc that already had one" + ); + } + static DRAWN: [AtomicUsize; 7] = [ AtomicUsize::new(0), AtomicUsize::new(0), diff --git a/config/src/external/overlay/completeness.rs b/config/src/external/overlay/completeness.rs new file mode 100644 index 0000000000..7b78fb5d4e --- /dev/null +++ b/config/src/external/overlay/completeness.rs @@ -0,0 +1,410 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright Open Network Fabric Authors + +use std::cell::RefCell; +use std::collections::{BTreeMap, BTreeSet}; + +use lpm::prefix::with_ports::{L4Protocol, PrefixPortsSet}; + +use super::Overlay; +use super::algebra::Sequence; +use super::vpc::Vpc; +use super::vpcpeering::{ + VpcExpose, VpcExposeMasquerade, VpcExposeNat, VpcExposeNatConfig, VpcExposePortForwarding, + VpcExposeStaticNat, VpcManifest, VpcPeering, +}; + +#[derive(Debug)] +enum Reach { + Spans(&'static [&'static str]), + Determined(&'static str), + Derived(&'static str), + Fixed(&'static str), +} + +const REACH: &[(&str, Reach)] = &[ + ( + "Overlay.vpc_table", + Reach::Determined("one vpc per `AddVpc`, in handle order"), + ), + ( + "Overlay.peering_table", + Reach::Determined("one peering per `AddPeering`, in handle order"), + ), + ("Vpc.name", Reach::Determined("`VpcHandle::name`")), + ("Vpc.id", Reach::Determined("`VpcHandle::id`")), + ("Vpc.vni", Reach::Determined("`VpcHandle::vni`")), + ( + "Vpc.interfaces", + Reach::Fixed( + "empty. No operation attaches an interface to a vpc, so no generated configuration \ + has one. Reaching the interface-bearing paths at all needs a new operation.", + ), + ), + ( + "Vpc.peerings", + Reach::Derived( + "collected from the peering table by `Overlay::validate`, not by the algebra", + ), + ), + ( + "VpcPeering.name", + Reach::Determined("`PeeringHandle::name`"), + ), + ( + "VpcPeering.left", + Reach::Determined("the peering's left handle"), + ), + ( + "VpcPeering.right", + Reach::Determined("the peering's right handle"), + ), + ( + "VpcPeering.gwgroup", + Reach::Fixed( + "the default group. `with_default_group` is the only constructor the algebra calls, \ + so nothing generated ever splits vpcs across gateway groups.", + ), + ), + ( + "VpcPeering.acl", + Reach::Fixed( + "absent. Peering-scoped ACLs are not in the vocabulary, so no generated configuration \ + carries one -- and an ACL is precisely a thing that changes a verdict, which is what \ + every property here asserts over.", + ), + ), + ( + "VpcManifest.name", + Reach::Determined("the side's vpc handle"), + ), + ( + "VpcManifest.exposes", + Reach::Determined("one per `AddExpose`, in slot order"), + ), + ( + "VpcExpose.default", + Reach::Fixed("false. `VpcExpose::empty` never sets it and no operation does either."), + ), + ( + "VpcExpose.ips", + Reach::Determined("one prefix, from the expose's peering, side and slot"), + ), + ( + "VpcExpose.ips.ports", + Reach::Fixed( + "unset. The algebra exposes whole prefixes, so a port-restricted expose is \ + unreachable, and with it every question about how ports partition an address.", + ), + ), + ( + "VpcExpose.nots", + Reach::Fixed( + "empty -- the survey renders it as no prefixes at all. An expose that carves holes out of its own range is unreachable, which is a \ + real hole rather than a canonicalisation: an exclusion is what makes a prefix set \ + non-contiguous, and non-contiguous is where a matcher goes wrong.", + ), + ), + ("VpcExpose.nat", Reach::Spans(&["absent", "present"])), + ( + "VpcExposeNat.as_range", + Reach::Determined("one prefix in the masquerade pool, from peering, side and slot"), + ), + ( + "VpcExposeNat.as_range.ports", + Reach::Fixed("unset, for the same reason as `VpcExpose.ips.ports`."), + ), + ( + "VpcExposeNat.not_as", + Reach::Fixed("empty, for the same reason as `VpcExpose.nots`."), + ), + ( + "VpcExposeNat.config", + Reach::Fixed( + "masquerade. `Flavour` has two members and only one of them makes a nat, so static \ + nat and port forwarding are both unreachable -- which the design note already names \ + as missing vocabulary.", + ), + ), + ( + "VpcExposeNat.proto", + Reach::Fixed("`Any`. No operation narrows an expose to tcp or udp."), + ), + ( + "VpcExposeMasquerade.idle_timeout", + Reach::Fixed( + "absent. `make_masquerade(None)` is the only call, so the timeout paths -- and every \ + question about a flow ageing out under a configuration that set one -- are never \ + entered.", + ), + ), + ( + "VpcExposeStaticNat", + Reach::Fixed("never constructed; see `VpcExposeNat.config`."), + ), + ( + "VpcExposePortForwarding.idle_timeout", + Reach::Fixed("never constructed; see `VpcExposeNat.config`."), + ), +]; + +#[derive(Default)] +struct Observed(BTreeMap<&'static str, BTreeSet>); + +impl Observed { + fn note(&mut self, field: &'static str, value: impl Into) { + self.0.entry(field).or_default().insert(value.into()); + } + + fn count(&mut self, field: &'static str, n: usize) { + self.note(field, n.to_string()); + } + + fn prefixes(&mut self, field: &'static str, set: &PrefixPortsSet) { + let listed: Vec = set + .into_iter() + .map(|entry| entry.prefix().to_string()) + .collect(); + self.note(field, listed.join(", ")); + } + + fn ports(&mut self, field: &'static str, set: &PrefixPortsSet) { + self.note( + field, + if set.into_iter().any(|entry| entry.ports().is_some()) { + "set" + } else { + "unset" + }, + ); + } +} + +fn survey(overlay: &Overlay, seen: &mut Observed) { + let Overlay { + vpc_table, + peering_table, + } = overlay; + seen.count("Overlay.vpc_table", vpc_table.len()); + seen.count("Overlay.peering_table", peering_table.len()); + + for vpc in vpc_table.values() { + let Vpc { + name, + id, + vni, + interfaces, + peerings, + } = vpc; + seen.note("Vpc.name", name.clone()); + seen.note("Vpc.id", id.to_string()); + seen.note("Vpc.vni", vni.as_u32().to_string()); + seen.count("Vpc.interfaces", interfaces.values().count()); + seen.count("Vpc.peerings", peerings.len()); + } + + for peering in peering_table.values() { + let VpcPeering { + name, + left, + right, + gwgroup, + acl, + } = peering; + seen.note("VpcPeering.name", name.clone()); + seen.note("VpcPeering.gwgroup", gwgroup.clone()); + seen.note( + "VpcPeering.acl", + if acl.is_some() { "present" } else { "absent" }, + ); + for (side, manifest) in [("VpcPeering.left", left), ("VpcPeering.right", right)] { + seen.note(side, manifest.name.clone()); + survey_manifest(manifest, seen); + } + } +} + +fn survey_manifest(manifest: &VpcManifest, seen: &mut Observed) { + let VpcManifest { name, exposes } = manifest; + seen.note("VpcManifest.name", name.clone()); + seen.count("VpcManifest.exposes", exposes.len()); + for expose in exposes { + let VpcExpose { + default, + ips, + nots, + nat, + } = expose; + seen.note("VpcExpose.default", default.to_string()); + seen.prefixes("VpcExpose.ips", ips); + seen.ports("VpcExpose.ips.ports", ips); + seen.prefixes("VpcExpose.nots", nots); + seen.note( + "VpcExpose.nat", + if nat.is_some() { "present" } else { "absent" }, + ); + if let Some(nat) = nat { + survey_nat(nat, seen); + } + } +} + +fn survey_nat(nat: &VpcExposeNat, seen: &mut Observed) { + let VpcExposeNat { + as_range, + not_as, + config, + proto, + } = nat; + seen.prefixes("VpcExposeNat.as_range", as_range); + seen.ports("VpcExposeNat.as_range.ports", as_range); + seen.prefixes("VpcExposeNat.not_as", not_as); + seen.note( + "VpcExposeNat.proto", + match proto { + L4Protocol::Tcp => "tcp", + L4Protocol::Udp => "udp", + L4Protocol::Any => "any", + }, + ); + match config { + VpcExposeNatConfig::Masquerade(VpcExposeMasquerade { idle_timeout }) => { + seen.note("VpcExposeNat.config", "masquerade"); + seen.note( + "VpcExposeMasquerade.idle_timeout", + if idle_timeout.is_some() { + "present" + } else { + "absent" + }, + ); + } + VpcExposeNatConfig::Static(VpcExposeStaticNat {}) => { + seen.note("VpcExposeNat.config", "static"); + seen.note("VpcExposeStaticNat", "constructed"); + } + VpcExposeNatConfig::PortForwarding(VpcExposePortForwarding { idle_timeout }) => { + seen.note("VpcExposeNat.config", "port-forwarding"); + seen.note( + "VpcExposePortForwarding.idle_timeout", + if idle_timeout.is_some() { + "present" + } else { + "absent" + }, + ); + } + } +} + +const CASES: usize = 512; + +fn survey_drawn(seen: &RefCell) { + let seen = std::panic::AssertUnwindSafe(seen); + bolero::check!() + .with_generator(Sequence::default()) + .with_iterations(CASES) + .for_each(|ops| { + let overlay = Sequence::fold(ops) + .overlay() + .unwrap_or_else(|e| panic!("{ops:?} does not assemble: {e}")); + survey(&overlay, &mut seen.borrow_mut()); + }); +} + +fn census() -> Observed { + let seen = RefCell::new(Observed::default()); + survey_drawn(&seen); + seen.into_inner() +} + +#[test] +fn every_surveyed_field_is_classified() { + let seen = census(); + let surveyed: BTreeSet<&str> = seen.0.keys().copied().collect(); + let classified: BTreeSet<&str> = REACH.iter().map(|(field, _)| *field).collect(); + + let unclassified: Vec<&&str> = surveyed.difference(&classified).collect(); + assert!( + unclassified.is_empty(), + "the survey records fields with no verdict in `REACH`: {unclassified:?}. A field added to \ + the overlay schema is not reachable by the algebra until an operation produces it, so say \ + which it is -- `Fixed` is a perfectly good answer and is what most of the table already \ + says." + ); + + let unreachable_by_construction: BTreeSet<&str> = REACH + .iter() + .filter( + |(_, reach)| matches!(reach, Reach::Fixed(why) if why.contains("never constructed")), + ) + .map(|(field, _)| *field) + .collect(); + let missing: Vec<&&str> = classified + .difference(&surveyed) + .filter(|field| !unreachable_by_construction.contains(**field)) + .collect(); + assert!( + missing.is_empty(), + "`REACH` gives a verdict for fields the survey never records: {missing:?}. Either the \ + field was removed from the schema, or the survey stopped visiting it -- and a survey that \ + has stopped visiting a field is a ratchet that has come loose." + ); +} + +#[test] +fn the_algebra_reaches_what_it_is_recorded_to_reach() { + let seen = census(); + + for (field, reach) in REACH { + let Some(values) = seen.0.get(field) else { + continue; + }; + let rendered: Vec<&str> = values.iter().map(String::as_str).collect(); + match reach { + Reach::Spans(expected) => assert_eq!( + rendered, *expected, + "`{field}` is recorded as spanning {expected:?} and {CASES} drawn configurations \ + show {rendered:?}" + ), + Reach::Determined(by) => assert!( + values.len() > 1, + "`{field}` is recorded as determined by {by}, which varies, and yet {CASES} drawn \ + configurations all show {rendered:?}. Either it is fixed after all, or the \ + generator has stopped varying what determines it" + ), + Reach::Derived(_) => {} + Reach::Fixed(why) => assert_eq!( + values.len(), + 1, + "`{field}` is recorded as fixed -- {why} -- and yet {CASES} drawn configurations \ + show {rendered:?}. If the vocabulary grew, say so here: this table is what tells \ + a reader of a green run which configurations it did not cover" + ), + } + } +} + +#[test] +fn report_what_the_algebra_reaches() { + let counted = |wanted: fn(&Reach) -> bool| -> usize { + REACH.iter().filter(|(_, reach)| wanted(reach)).count() + }; + eprintln!( + "of {} degrees of freedom in the overlay schema, the algebra spans {}, determines {}, \ + leaves {} to be derived, and fixes {}:", + REACH.len(), + counted(|reach| matches!(reach, Reach::Spans(_))), + counted(|reach| matches!(reach, Reach::Determined(_))), + counted(|reach| matches!(reach, Reach::Derived(_))), + counted(|reach| matches!(reach, Reach::Fixed(_))), + ); + for (field, reach) in REACH { + match reach { + Reach::Spans(values) => eprintln!(" spans {field}: {values:?}"), + Reach::Determined(by) => eprintln!(" determined {field}: by {by}"), + Reach::Derived(how) => eprintln!(" derived {field}: {how}"), + Reach::Fixed(why) => eprintln!(" FIXED {field}: {why}"), + } + } +} diff --git a/config/src/external/overlay/mod.rs b/config/src/external/overlay/mod.rs index fdf8d85186..85e0039a6e 100644 --- a/config/src/external/overlay/mod.rs +++ b/config/src/external/overlay/mod.rs @@ -6,6 +6,8 @@ pub mod acl; #[cfg(any(test, feature = "bolero"))] pub mod algebra; +#[cfg(test)] +mod completeness; pub mod tests; pub mod validation_tests; pub mod vpc; diff --git a/dataplane/Cargo.toml b/dataplane/Cargo.toml index dc3b553da5..d5cae60996 100644 --- a/dataplane/Cargo.toml +++ b/dataplane/Cargo.toml @@ -10,6 +10,8 @@ default = [] # Bin is gated out under loom (see `src/main.rs`); the feature exists so # `--features loom` resolves at workspace level and propagates to libs. loom = ["concurrency/loom"] +shuttle = ["concurrency/shuttle", "nat/shuttle", "flow-entry/shuttle"] +shuttle_dfs = ["concurrency/shuttle_dfs", "shuttle"] [dependencies] acl-filter = { workspace = true } @@ -67,6 +69,7 @@ routing = { workspace = true, features = ["testing"] } bolero = { workspace = true, default-features = false, features = ["alloc"] } n-vm = { workspace = true } tokio = { workspace = true, features = ["macros", "rt", "test-util", "time"] } +tracectl = { workspace = true, features = ["evidence"] } tracing = { workspace = true } tracing-subscriber = { workspace = true } tracing-test = { workspace = true, features = [] } diff --git a/dataplane/src/packet_processor/fuzz.rs b/dataplane/src/packet_processor/fuzz.rs index ccfc05691f..7b9bac5c7f 100644 --- a/dataplane/src/packet_processor/fuzz.rs +++ b/dataplane/src/packet_processor/fuzz.rs @@ -4,8 +4,11 @@ #![cfg(test)] #![cfg(not(miri))] -use acl_filter::{AclFilter, AclFilterContext, AclFilterContextWriter}; +use acl_filter::{ + AclFilter, AclFilterContext, AclFilterContextReaderFactory, AclFilterContextWriter, +}; use concurrency::sync::{Arc, Mutex}; +use config::external::GenId; use config::external::overlay::acl::Acl; use config::external::overlay::vpcpeering::VpcExpose; use config::external::overlay::vpcpeering::contract::{ @@ -13,135 +16,241 @@ use config::external::overlay::vpcpeering::contract::{ }; use config::external::overlay::{Overlay, ValidatedOverlay}; use flow_entry::flow_table::{FlowLookup, FlowTable}; -use flow_filter::{FlowFilter, FlowFilterContext, FlowFilterContextWriter}; +use flow_filter::{ + FlowFilter, FlowFilterContext, FlowFilterContextReaderFactory, FlowFilterContextWriter, +}; use lpm::prefix::Prefix; -use nat::masquerade::{MasqueradeConfig, NatAllocatorWriter}; -use nat::portfw::{PortForwarder, PortFwTableWriter}; -use nat::static_nat::NatTablesWriter; +use nat::masquerade::{MasqueradeConfig, NatAllocatorReaderFactory, NatAllocatorWriter}; +use nat::portfw::{PortForwarder, PortFwTableReaderFactory, PortFwTableWriter}; use nat::static_nat::setup::build_nat_configuration; +use nat::static_nat::{NatTablesReaderFactory, NatTablesWriter}; use nat::{IcmpErrorHandler, Masquerade, StaticNat}; use net::buffer::{PacketBufferMut, TestBuffer}; use net::eth::mac::{Mac, SourceMac}; use net::interface::InterfaceIndex; use net::packet::{DoneReason, Packet, VpcDiscriminant}; use net::vxlan::Vni; -use pipeline::{DynPipeline, NetworkFunction}; +use pipeline::{DynPipeline, NetworkFunction, PipelineData}; use routing::testing::RouterTables; use routing::testing::{FibGroup, FwAction, NhopKey, RouteOrigin}; +use routing::{AtableReaderFactory, FibTableReaderFactory, IfTableReaderFactory}; use routing::{EgressObject, FibEntry, PktInstruction, ResolvedEncapsulation, ResolvedVxlan, Vtep}; +use std::cell::{Cell, RefCell}; use std::net::IpAddr; +use std::time::Duration; + +const FIRST_GENID: GenId = 1; + +#[derive(Clone, Copy, PartialEq, Eq, Debug)] +pub(crate) enum Enact { + FlowFilter, + Acl, + StaticNat, + Masquerade, + PortForward, + Generation, + Everything, +} use super::egress::Egress; use super::ingress::Ingress; use super::ipforward::IpForwarder; -pub(crate) struct Fabric { - pipeline: DynPipeline, +pub(crate) struct Fleet { + flow_filter: FlowFilterContextWriter, + acl: AclFilterContextWriter, + static_nat: RefCell, + portfw: RefCell, + masquerade: RefCell, + genid: Cell, + randomize: Cell, + blueprint: Blueprint, +} + +pub(crate) struct Blueprint { + flow_filter: FlowFilterContextReaderFactory, + acl: AclFilterContextReaderFactory, + static_nat: NatTablesReaderFactory, + portfw: PortFwTableReaderFactory, + masquerade: NatAllocatorReaderFactory, + pipeline: Arc, + underlay: Option, flow_table: Arc, - _flow_filter: FlowFilterContextWriter, - _acl: AclFilterContextWriter, - _static_nat: NatTablesWriter, - _portfw: PortFwTableWriter, - _masquerade: NatAllocatorWriter, - _tables: Option, + declared: Arc<[Prefix]>, +} + +struct Underlay { + interfaces: IfTableReaderFactory, + fibs: FibTableReaderFactory, + adjacencies: AtableReaderFactory, +} + +pub(crate) struct Worker { + pipeline: DynPipeline, translations: Arc>, next_id: u64, } -impl Fabric { - pub(crate) fn build(exposes: &[VpcExpose]) -> Option { - Self::build_with_acl(exposes, None) +pub(crate) struct Fabric { + _tables: Option, + fleet: Fleet, + worker: Worker, +} + +impl Fleet { + pub(crate) fn lowering( + overlay: &ValidatedOverlay, + tables: Option<&RouterTables>, + flow_table: Arc, + ) -> Self { + let flow_filter = FlowFilterContextWriter::new(); + flow_filter.store( + FlowFilterContext::try_from(overlay).expect("a validated overlay lowers to tables"), + ); + + let acl = AclFilterContextWriter::new(); + acl.store(AclFilterContext::try_from(overlay).expect("a validated overlay lowers to acls")); + + let mut static_nat = NatTablesWriter::new(); + static_nat.update_nat_tables( + build_nat_configuration(overlay.vpc_table()) + .expect("a validated overlay lowers to nat"), + ); + + let mut portfw = PortFwTableWriter::new(); + portfw + .update_from_vpc_table(overlay.vpc_table()) + .expect("a validated overlay lowers to port forwarding"); + + let mut masquerade = NatAllocatorWriter::new(); + masquerade.update_nat_allocator( + MasqueradeConfig::new(overlay.vpc_table()).set_randomize(false), + FIRST_GENID, + &flow_table, + ); + + let blueprint = Blueprint { + flow_filter: flow_filter.get_reader_factory(), + acl: acl.get_reader_factory(), + static_nat: static_nat.get_reader_factory(), + portfw: portfw.reader().factory(), + masquerade: masquerade.get_reader_factory(), + pipeline: Arc::new(PipelineData::new(FIRST_GENID)), + underlay: tables.map(|tables| Underlay { + interfaces: tables.interface_factory(), + fibs: tables.fib_factory(), + adjacencies: tables.adjacency_factory(), + }), + flow_table, + declared: declared_public_ranges(overlay), + }; + + Self { + flow_filter, + acl, + static_nat: RefCell::new(static_nat), + portfw: RefCell::new(portfw), + masquerade: RefCell::new(masquerade), + genid: Cell::new(FIRST_GENID), + randomize: Cell::new(false), + blueprint, + } } - pub(crate) fn build_with_acl(exposes: &[VpcExpose], acl: Option<&Acl>) -> Option { - Self::assemble(exposes, acl, None) + pub(crate) fn reconfigure(&self, overlay: &ValidatedOverlay) { + self.enact(overlay, Enact::Everything); } - pub(crate) fn routed(exposes: &[VpcExpose], acl: Option<&Acl>) -> Option { - Self::assemble( - exposes, - acl, - Some(topology(&[vni(LOCAL_VNI), vni(REMOTE_VNI)])), - ) + pub(crate) fn enact_with_router_gap(&self, overlay: &ValidatedOverlay, gap: Duration) { + for step in [ + Enact::FlowFilter, + Enact::Acl, + Enact::StaticNat, + Enact::Masquerade, + Enact::PortForward, + ] { + self.enact(overlay, step); + } + std::thread::sleep(gap); + self.enact(overlay, Enact::Generation); } - pub(crate) fn routed_over(overlay: &Overlay, tables: RouterTables) -> Option { - Some(Self::with_overlay( - &overlay.clone().validate().ok()?, - Some(tables), - )) + pub(crate) fn randomizing(&self, on: bool) -> &Self { + self.randomize.set(on); + self } - pub(crate) fn routed_over_validated(overlay: &ValidatedOverlay, tables: RouterTables) -> Self { - Self::with_overlay(overlay, Some(tables)) + pub(crate) fn enact(&self, overlay: &ValidatedOverlay, which: Enact) { + let doing = |step: Enact| which == Enact::Everything || which == step; + if doing(Enact::FlowFilter) { + self.flow_filter.store( + FlowFilterContext::try_from(overlay).expect("a validated overlay lowers to tables"), + ); + } + if doing(Enact::Acl) { + self.acl.store( + AclFilterContext::try_from(overlay).expect("a validated overlay lowers to acls"), + ); + } + if doing(Enact::StaticNat) { + self.static_nat.borrow_mut().update_nat_tables( + build_nat_configuration(overlay.vpc_table()) + .expect("a validated overlay lowers to nat"), + ); + } + if doing(Enact::Masquerade) { + self.genid.set(self.genid.get() + 1); + self.masquerade.borrow_mut().update_nat_allocator( + MasqueradeConfig::new(overlay.vpc_table()).set_randomize(self.randomize.get()), + self.genid.get(), + &self.blueprint.flow_table, + ); + } + if doing(Enact::PortForward) { + self.portfw + .borrow_mut() + .update_from_vpc_table(overlay.vpc_table()) + .expect("a validated overlay lowers to port forwarding"); + } + if doing(Enact::Generation) { + self.blueprint.pipeline.set_genid(self.genid.get()); + } } - fn assemble( - exposes: &[VpcExpose], - acl: Option<&Acl>, - tables: Option, - ) -> Option { - let overlay = overlay_with_exposes_and_acl(exposes.to_vec(), acl) - .ok()? - .validate() - .ok()?; - Some(Self::with_overlay(&overlay, tables)) + pub(crate) fn blueprint(&self) -> &Blueprint { + &self.blueprint } +} - fn with_overlay(overlay: &ValidatedOverlay, tables: Option) -> Self { - let translations = Arc::new(Mutex::new(Translations::declaring(overlay))); - let flow_table = Arc::new(FlowTable::default()); - let mut pipeline = DynPipeline::new(); +impl Blueprint { + pub(crate) fn worker(&self) -> Worker { + let translations = Arc::new(Mutex::new(Translations::declaring(&self.declared))); + let mut pipeline = DynPipeline::new().set_data(self.pipeline.clone()); - if let Some(tables) = &tables { - pipeline = pipeline.add_stage(Ingress::new("ingress", tables.interfaces())); - pipeline = pipeline.add_stage(IpForwarder::new("ip-forward-1", tables.fibs())); + if let Some(underlay) = &self.underlay { + pipeline = pipeline.add_stage(Ingress::new("ingress", underlay.interfaces.handle())); + pipeline = pipeline.add_stage(IpForwarder::new("ip-forward-1", underlay.fibs.handle())); pipeline = pipeline.add_stage(Checkpoint::new( "after ip-forward-1", contract::decapsulated, )); } - pipeline = pipeline.add_stage(IcmpErrorHandler::new(flow_table.clone())); - pipeline = pipeline.add_stage(FlowLookup::new("flow-lookup", flow_table.clone())); - - let flow_filter = FlowFilterContextWriter::new(); - flow_filter.store( - FlowFilterContext::try_from(overlay).expect("a validated overlay lowers to tables"), - ); - pipeline = pipeline.add_stage(FlowFilter::new("flow-filter", flow_filter.get_reader())); + pipeline = pipeline.add_stage(IcmpErrorHandler::new(self.flow_table.clone())); + pipeline = pipeline.add_stage(FlowLookup::new("flow-lookup", self.flow_table.clone())); + pipeline = pipeline.add_stage(FlowFilter::new("flow-filter", self.flow_filter.handle())); pipeline = pipeline.add_stage(Checkpoint::new("after flow-filter", contract::placed)); - - let acl = AclFilterContextWriter::new(); - acl.store(AclFilterContext::try_from(overlay).expect("a validated overlay lowers to acls")); - pipeline = pipeline.add_stage(AclFilter::new("acl-filter", acl.get_reader())); - - let mut static_nat = NatTablesWriter::new(); - static_nat.update_nat_tables( - build_nat_configuration(overlay.vpc_table()) - .expect("a validated overlay lowers to nat"), - ); + pipeline = pipeline.add_stage(AclFilter::new("acl-filter", self.acl.handle())); pipeline = pipeline.add_stage(StaticNat::with_reader( "static-nat", - static_nat.get_reader(), + self.static_nat.handle(), )); - - let mut portfw = PortFwTableWriter::new(); - portfw - .update_from_vpc_table(overlay.vpc_table()) - .expect("a validated overlay lowers to port forwarding"); pipeline = pipeline.add_stage(PortForwarder::new( "port-forwarder", - portfw.reader(), - flow_table.clone(), + self.portfw.handle(), + self.flow_table.clone(), )); - let mut masquerade = NatAllocatorWriter::new(); - masquerade.update_nat_allocator( - MasqueradeConfig::new(overlay.vpc_table()).set_randomize(false), - 1, - &flow_table, - ); pipeline = pipeline.add_stage(Checkpoint::new( "before masquerade", contract::ready_to_translate, @@ -155,8 +264,8 @@ impl Fabric { )); pipeline = pipeline.add_stage(Masquerade::new( "masquerade", - flow_table.clone(), - masquerade.get_reader(), + self.flow_table.clone(), + self.masquerade.handle(), )); let checking = translations.clone(); @@ -167,40 +276,31 @@ impl Fabric { }, )); - if let Some(tables) = &tables { - pipeline = pipeline.add_stage(IpForwarder::new("ip-forward-2", tables.fibs())); + if let Some(underlay) = &self.underlay { + pipeline = pipeline.add_stage(IpForwarder::new("ip-forward-2", underlay.fibs.handle())); pipeline = pipeline.add_stage(Egress::new( "egress", - tables.interfaces(), - tables.adjacencies(), + underlay.interfaces.handle(), + underlay.adjacencies.handle(), )); pipeline = pipeline.add_stage(Checkpoint::new("after egress", contract::finished)); } - Self { + Worker { pipeline, - flow_table, - _flow_filter: flow_filter, - _acl: acl, - _static_nat: static_nat, - _portfw: portfw, - _masquerade: masquerade, - _tables: tables, translations, next_id: 0, } } +} +impl Worker { pub(crate) fn send(&mut self, packet: Packet) -> Packet { let mut out = self.send_batch(vec![packet]); assert_eq!(out.len(), 1, "the pipeline did not return the packet"); out.pop().unwrap_or_else(|| unreachable!()) } - pub(crate) fn flows(&self) -> Option { - self.flow_table.len() - } - pub(crate) fn send_batch( &mut self, mut packets: Vec>, @@ -218,6 +318,93 @@ impl Fabric { } } +impl Fabric { + pub(crate) fn build(exposes: &[VpcExpose]) -> Option { + Self::build_with_acl(exposes, None) + } + + pub(crate) fn build_with_acl(exposes: &[VpcExpose], acl: Option<&Acl>) -> Option { + Self::assemble(exposes, acl, None) + } + + pub(crate) fn routed(exposes: &[VpcExpose], acl: Option<&Acl>) -> Option { + Self::routed_sharing(exposes, acl, Arc::new(FlowTable::default())) + } + + pub(crate) fn routed_sharing( + exposes: &[VpcExpose], + acl: Option<&Acl>, + flow_table: Arc, + ) -> Option { + let overlay = overlay_with_exposes_and_acl(exposes.to_vec(), acl) + .ok()? + .validate() + .ok()?; + Some(Self::over( + &overlay, + Some(topology(&[vni(LOCAL_VNI), vni(REMOTE_VNI)])), + flow_table, + )) + } + + pub(crate) fn routed_over(overlay: &Overlay, tables: RouterTables) -> Option { + Some(Self::over( + &overlay.clone().validate().ok()?, + Some(tables), + Arc::new(FlowTable::default()), + )) + } + + pub(crate) fn routed_over_validated(overlay: &ValidatedOverlay, tables: RouterTables) -> Self { + Self::over(overlay, Some(tables), Arc::new(FlowTable::default())) + } + + fn assemble( + exposes: &[VpcExpose], + acl: Option<&Acl>, + tables: Option, + ) -> Option { + let overlay = overlay_with_exposes_and_acl(exposes.to_vec(), acl) + .ok()? + .validate() + .ok()?; + Some(Self::over(&overlay, tables, Arc::new(FlowTable::default()))) + } + + fn over( + overlay: &ValidatedOverlay, + tables: Option, + flow_table: Arc, + ) -> Self { + let fleet = Fleet::lowering(overlay, tables.as_ref(), flow_table); + let worker = fleet.blueprint().worker(); + Self { + _tables: tables, + fleet, + worker, + } + } + + pub(crate) fn send(&mut self, packet: Packet) -> Packet { + self.worker.send(packet) + } + + pub(crate) fn send_batch( + &mut self, + packets: Vec>, + ) -> Vec> { + self.worker.send_batch(packets) + } + + pub(crate) fn worker(&mut self) -> &mut Worker { + &mut self.worker + } + + pub(crate) fn flows(&self) -> Option { + self.fleet.blueprint().flow_table.len() + } +} + pub(crate) fn local() -> VpcDiscriminant { VpcDiscriminant::VNI(Vni::new_checked(LOCAL_VNI).unwrap_or_else(|_| unreachable!())) } @@ -324,7 +511,7 @@ pub(crate) struct Translations { was: std::collections::HashMap, from: std::collections::HashMap, given: std::collections::HashMap, - declared: Vec, + declared: Arc<[Prefix]>, } #[cfg(test)] @@ -384,29 +571,34 @@ impl Translations { self.given.clear(); } - fn declaring(overlay: &ValidatedOverlay) -> Self { - let mut declared = Vec::new(); - for vpc in overlay.vpc_table().values() { - for peering in vpc.peerings() { - for manifest in [peering.local(), peering.remote()] { - for expose in manifest.valexp() { - declared.extend( - expose - .public_ips() - .into_iter() - .map(lpm::prefix::PrefixWithOptionalPorts::prefix), - ); - } - } - } - } + fn declaring(declared: &Arc<[Prefix]>) -> Self { Self { - declared, + declared: declared.clone(), ..Self::default() } } } +#[cfg(test)] +fn declared_public_ranges(overlay: &ValidatedOverlay) -> Arc<[Prefix]> { + let mut declared = Vec::new(); + for vpc in overlay.vpc_table().values() { + for peering in vpc.peerings() { + for manifest in [peering.local(), peering.remote()] { + for expose in manifest.valexp() { + declared.extend( + expose + .public_ips() + .into_iter() + .map(lpm::prefix::PrefixWithOptionalPorts::prefix), + ); + } + } + } + } + declared.into() +} + #[cfg(test)] pub(crate) trait Load { fn next(&mut self) -> Option>; @@ -418,10 +610,14 @@ pub(crate) trait Load { fn checked(&self) -> bool; fn describe(&self) -> String; + + fn public(&self) -> Option<(IpAddr, u16)> { + None + } } #[cfg(test)] -pub(crate) fn drive(fabric: &mut Fabric, load: &mut dyn Load) { +pub(crate) fn drive(worker: &mut Worker, load: &mut dyn Load) { for _ in 0..64 { if load.finished() { return; @@ -432,7 +628,7 @@ pub(crate) fn drive(fabric: &mut Fabric, load: &mut dyn Load) { load.describe() ); }; - let out = fabric.send(packet); + let out = worker.send(packet); load.observe(&out); } panic!("a load did not finish in 64 steps: {}", load.describe()); @@ -450,7 +646,7 @@ pub(crate) type Poll = Vec; #[cfg(test)] pub(crate) fn run_schedule( - fabric: &mut Fabric, + worker: &mut Worker, loads: &mut [Box], schedule: &[Poll], ) -> Vec> { @@ -474,14 +670,14 @@ pub(crate) fn run_schedule( if burst.is_empty() { continue; } - for (answer, which) in fabric.send_batch(burst).iter().zip(&origin) { + for (answer, which) in worker.send_batch(burst).iter().zip(&origin) { loads[*which].observe(answer); } bursts.push(origin); } for load in loads { - drive(fabric, load.as_mut()); + drive(worker, load.as_mut()); } bursts } @@ -551,10 +747,32 @@ pub(crate) mod derive { } pub(crate) fn loads_for(overlay: &ValidatedOverlay, vary: &[Vary]) -> Vec> { + loads_where(overlay, vary, &|_| true) + } + + #[derive(Debug, Clone, Copy)] + pub(crate) struct Named<'a> { + pub(crate) local: &'a str, + pub(crate) remote: &'a str, + pub(crate) peering: &'a str, + } + + pub(crate) fn loads_where( + overlay: &ValidatedOverlay, + vary: &[Vary], + keep: &dyn Fn(Named<'_>) -> bool, + ) -> Vec> { let mut loads: Vec> = Vec::new(); let mut nth = 0usize; for vpc in overlay.vpc_table().values() { for peering in vpc.peerings() { + if !keep(Named { + local: vpc.name(), + remote: peering.remote().name(), + peering: peering.name(), + }) { + continue; + } let path = super::routed::Path::new(vpc.vni(), peering.remote_vni()); for expose in peering.local().valexp() { let Some(v) = vary.get(nth % vary.len().max(1)).copied() else { @@ -1998,7 +2216,7 @@ mod interleaved { }); } - for burst in run_schedule(&mut fabric, &mut loads, schedule) { + for burst in run_schedule(fabric.worker(), &mut loads, schedule) { let mut loads_in: Vec = burst.clone(); loads_in.sort_unstable(); loads_in.dedup(); @@ -2170,7 +2388,7 @@ mod offers { } } - for burst in run_schedule(&mut fabric, &mut loads, schedule) { + for burst in run_schedule(fabric.worker(), &mut loads, schedule) { let mut seen = burst.clone(); seen.sort_unstable(); seen.dedup(); @@ -2233,7 +2451,7 @@ mod generated { const SENDERS: usize = 6; const POLLS: usize = 8; - struct Generated; + pub(super) struct Generated; impl ValueGenerator for Generated { type Output = (Vec, Vec, Vec); @@ -2318,7 +2536,7 @@ mod generated { let mut loads = loads_for(&validated, vary); DERIVED.fetch_add(loads.len() as u64, Ordering::Relaxed); - for burst in run_schedule(&mut fabric, &mut loads, schedule) { + for burst in run_schedule(fabric.worker(), &mut loads, schedule) { let mut seen = burst.clone(); seen.sort_unstable(); seen.dedup(); @@ -2813,7 +3031,7 @@ mod routed { matches!(copy.vxlan_decap(), Some(Ok(_))).then_some(copy) } - fn inner() -> Packet { + pub(super) fn inner() -> Packet { build_test_udp_ipv4_packet("1.1.0.1", "3.3.3.1", 1234, 80) } @@ -2979,7 +3197,7 @@ mod routed { let mut load = Conversation::new(Path::fixture(), src, dst, flow.sport, flow.dport); - drive(&mut fabric, &mut load); + drive(fabric.worker(), &mut load); if load.checked() { ROUND_TRIPPED.fetch_add(1, Ordering::Relaxed); @@ -3008,6 +3226,7 @@ mod routed { dport: u16, sent: Option>, state: State, + public: Option<(IpAddr, u16)>, log: Vec, } @@ -3031,6 +3250,7 @@ mod routed { dport, sent: None, state: State::Opening, + public: None, log: Vec::new(), } } @@ -3075,6 +3295,7 @@ mod routed { return; }; self.note(&format!("request left as {public_src}:{}", port.get())); + self.public = Some((public_src, port.get())); self.state = State::Replying { public: (public_src, port.get()), }; @@ -3119,6 +3340,10 @@ mod routed { } impl Load for Conversation { + fn public(&self) -> Option<(IpAddr, u16)> { + self.public + } + fn next(&mut self) -> Option> { match self.state { State::Opening => { @@ -3462,3 +3687,1691 @@ mod routed { out.pop().unwrap_or_else(|| unreachable!()) } } + +#[cfg(test)] +mod model { + use super::derive::loads_for; + use super::routed::{Conversation, exposes, inner, inside, tunnelled}; + use super::*; + use concurrency::sync::Mutex; + use concurrency::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; + use concurrency::sync::{LazyLock, OnceLock}; + use concurrency::thread; + #[cfg_attr(not(feature = "shuttle"), allow(unused_imports))] + use concurrency::thread::BuilderExt; + use config::external::overlay::algebra::{Footprint, Sequence}; + use net::packet::test_utils::build_test_udp_ipv4_packet; + + type Tuple = (Option, Option); + + type Reported = (Vec, tracectl::evidence::Evidence); + + fn on_two_threads(body: &(impl Fn() + Sync)) { + thread::scope(|scope| { + let handles: Vec<_> = (0..2) + .map(|_| { + thread::Builder::new() + .spawn_scoped(scope, body) + .expect("spawn") + }) + .collect(); + for handle in handles { + handle.join().expect("join"); + } + }); + } + + #[concurrency::model_test] + fn a_lock_that_outlives_its_execution_is_not_model_checkable() { + static LOCK: OnceLock> = OnceLock::new(); + static RUNS: AtomicUsize = AtomicUsize::new(0); + + concurrency::stress(|| { + let lock = concurrency::sync::Arc::new(Mutex::new(0u32)); + let inner = lock.clone(); + let bump = move || *inner.lock() += 1; + on_two_threads(&bump); + assert_eq!(*lock.lock(), 2); + }); + + concurrency::stress(|| { + let first = RUNS.fetch_add(1, Ordering::Relaxed) == 0; + let bump = move || { + if first { + *LOCK.get_or_init(|| Mutex::new(0)).lock() += 1; + } + }; + on_two_threads(&bump); + }); + } + + #[concurrency::model_test] + fn a_pipeline_can_be_driven_inside_a_stress_run() { + let _eal = dpdk::test_support::start_eal(); + + let rt = cfg_select! { + feature = "shuttle" => None::, + _ => Some( + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("build tokio runtime") + ) + }; + let handle = rt.as_ref().map(tokio::runtime::Runtime::handle).cloned(); + + concurrency::stress(move || { + let table = Arc::new(FlowTable::default()); + let sending = table.clone(); + let entering = handle.clone(); + + thread::scope(|scope| { + let sender = thread::Builder::new() + .name("sender".to_owned()) + .spawn_scoped(scope, move || { + let _guard = entering.as_ref().map(tokio::runtime::Handle::enter); + let mut fabric = Fabric::routed_sharing(&exposes(), None, sending) + .unwrap_or_else(|| unreachable!("the fixture exposes do not validate")); + let out = fabric.send(tunnelled(&inner())); + assert!( + matches!(verdict(&out), Verdict::Delivered { .. }), + "a packet that is delivered single-threaded was not: {:?}", + verdict(&out) + ); + }) + .expect("spawn sender"); + + let reader = thread::Builder::new() + .name("reader".to_owned()) + .spawn_scoped(scope, move || { + for _ in 0..3 { + let seen = table.len(); + assert!(seen.is_some(), "the flow table refused a concurrent read"); + table.for_each_flow(|_, flow| { + let _ = flow.status(); + }); + thread::yield_now(); + } + }) + .expect("spawn reader"); + + sender.join().expect("sender panicked"); + reader.join().expect("reader panicked"); + }); + }); + } + fn _a_blueprint_crosses_a_thread_boundary(blueprint: &Blueprint) { + fn shareable(_: &T) {} + shareable(blueprint); + } + + #[concurrency::model_test] + fn two_workers_are_not_given_the_same_public_tuple() { + const FLOWS: u16 = 3; + + let _eal = dpdk::test_support::start_eal(); + + let rt = cfg_select! { + feature = "shuttle" => None::, + _ => Some( + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("build tokio runtime") + ) + }; + let handle = rt.as_ref().map(tokio::runtime::Runtime::handle).cloned(); + + concurrency::stress(move || { + let overlay = overlay_with_exposes_and_acl(exposes(), None) + .expect("the fixture exposes form an overlay") + .validate() + .expect("the fixture overlay validates"); + let tables = topology(&[vni(LOCAL_VNI), vni(REMOTE_VNI)]); + let fleet = Fleet::lowering(&overlay, Some(&tables), Arc::new(FlowTable::default())); + let blueprint = fleet.blueprint(); + + let workers = [("1.1.0.1", 1000u16), ("1.1.0.2", 2000u16)]; + let given: Vec = thread::scope(|scope| { + let running: Vec<_> = workers + .iter() + .map(|(host, first)| { + let entering = handle.clone(); + thread::Builder::new() + .name(format!("worker-{host}")) + .spawn_scoped(scope, move || { + let _guard = entering.as_ref().map(tokio::runtime::Handle::enter); + let recording = + tracectl::evidence::capture(format!("worker-{host}")); + let mut worker = blueprint.worker(); + let burst = (0..FLOWS) + .map(|n| { + tunnelled(&build_test_udp_ipv4_packet( + host, + "3.3.3.1", + first + n, + 80, + )) + }) + .collect(); + let tuples = worker + .send_batch(burst) + .iter() + .map(|out| { + assert!( + matches!(verdict(out), Verdict::Delivered { .. }), + "a packet that is delivered single-threaded was not: \ + {:?}", + verdict(out) + ); + let tenant = inside(out).expect( + "a delivered frame leaves this gateway tunnelled", + ); + ( + tenant.ip_source(), + tenant.transport_src_port().map(std::num::NonZero::get), + ) + }) + .collect::>(); + (tuples, recording.evidence()) + }) + .expect("spawn worker") + }) + .collect(); + running + .into_iter() + .map(|worker| worker.join().expect("worker panicked")) + .collect() + }); + + let (given, evidence): (Vec>, Vec<_>) = given.into_iter().unzip(); + let given: Vec<_> = given.into_iter().flatten().collect(); + let _explain = tracectl::evidence::dump_on_panic(evidence); + + let mut seen = std::collections::BTreeMap::new(); + for tuple in &given { + let count: &mut usize = seen.entry(format!("{tuple:?}")).or_default(); + *count += 1; + assert_eq!( + *count, + 1, + "{} distinct flows were translated and {tuple:?} was handed out twice, so a \ + reply to it cannot be attributed to either of them: {given:?}", + given.len() + ); + } + }); + } + + #[concurrency::model_test] + fn generated_traffic_survives_being_split_across_two_workers() { + const CASES: usize = 64; + + static SPLIT: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + static THIN: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + + let _eal = dpdk::test_support::start_eal(); + + let rt = cfg_select! { + feature = "shuttle" => None::, + _ => Some( + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("build tokio runtime") + ) + }; + let handle = rt.as_ref().map(tokio::runtime::Runtime::handle).cloned(); + + bolero::check!() + .with_max_len(MAX_INPUT_LEN) + .with_generator(generated::Generated) + .with_iterations(CASES) + .for_each(|(ops, vary, schedule)| { + let validated = Sequence::fold(ops) + .overlay() + .unwrap_or_else(|e| panic!("{ops:?} does not assemble: {e}")) + .validate() + .unwrap_or_else(|e| panic!("{ops:?} does not validate: {e}")); + + let vnis: Vec = validated + .vpc_table() + .values() + .map(config::external::overlay::vpc::ValidatedVpc::vni) + .collect(); + if vnis.is_empty() || loads_for(&validated, vary).len() < 2 { + THIN.fetch_add(1, Ordering::Relaxed); + return; + } + SPLIT.fetch_add(1, Ordering::Relaxed); + + let drawn = + concurrency::sync::Arc::new((validated, vnis, vary.clone(), schedule.clone())); + let entering = handle.clone(); + + concurrency::stress(move || { + let (validated, vnis, vary, schedule) = &*drawn; + let tables = topology(vnis); + let fleet = + Fleet::lowering(validated, Some(&tables), Arc::new(FlowTable::default())); + let blueprint = fleet.blueprint(); + + thread::scope(|scope| { + let running: Vec<_> = (0..2) + .map(|which| { + let entering = entering.clone(); + thread::Builder::new() + .name(format!("worker-{which}")) + .spawn_scoped(scope, move || { + let _guard = + entering.as_ref().map(tokio::runtime::Handle::enter); + let _evidence = + tracectl::evidence::capture(format!("worker-{which}")); + let mut worker = blueprint.worker(); + let mut mine: Vec> = + loads_for(validated, vary) + .into_iter() + .enumerate() + .filter(|(nth, _)| nth % 2 == which) + .map(|(_, load)| load) + .collect(); + run_schedule(&mut worker, &mut mine, schedule); + }) + .expect("spawn worker") + }) + .collect(); + for worker in running { + worker.join().expect("worker panicked"); + } + }); + }); + }); + + let (split, thin) = (SPLIT.load(Ordering::Relaxed), THIN.load(Ordering::Relaxed)); + eprintln!("split={split} thin={thin}"); + super::assert_covered( + split > 0, + "no drawn configuration ever implied enough traffic to load two workers, so this ran \ + nothing concurrently", + ); + } + + fn without_unwinding(body: impl FnOnce() -> T) -> Result { + cfg_select! { + feature = "shuttle" => Ok(body()), + feature = "loom" => Ok(body()), + _ => std::panic::catch_unwind(std::panic::AssertUnwindSafe(body)).map_err(|payload| { + payload + .downcast_ref::<&str>() + .map(|message| (*message).to_owned()) + .or_else(|| payload.downcast_ref::().cloned()) + .unwrap_or_else(|| "a panic carrying no message".to_owned()) + }), + } + } + + fn step(worker: &mut Worker, load: &mut dyn Load) { + let Some(packet) = load.next() else { + panic!( + "a load was asked for a packet it would not give: {}", + load.describe() + ); + }; + let out = worker.send(packet); + load.observe(&out); + } + + #[concurrency::model_test] + fn a_reply_is_translated_by_a_worker_that_never_saw_the_request() { + const FLOWS: u8 = 2; + + static CLOSED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + static ABANDONED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + + let _eal = dpdk::test_support::start_eal(); + + let rt = cfg_select! { + feature = "shuttle" => None::, + _ => Some( + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("build tokio runtime") + ) + }; + let handle = rt.as_ref().map(tokio::runtime::Runtime::handle).cloned(); + + concurrency::stress(move || { + let overlay = overlay_with_exposes_and_acl(exposes(), None) + .expect("the fixture exposes form an overlay") + .validate() + .expect("the fixture overlay validates"); + let tables = topology(&[vni(LOCAL_VNI), vni(REMOTE_VNI)]); + let fleet = Fleet::lowering(&overlay, Some(&tables), Arc::new(FlowTable::default())); + let blueprint = fleet.blueprint(); + let entering = handle.clone(); + + let mut opened: Vec> = thread::scope(|scope| { + let running: Vec<_> = (0..2u8) + .map(|which| { + let entering = entering.clone(); + thread::Builder::new() + .name(format!("open-{which}")) + .spawn_scoped(scope, move || { + let _guard = entering.as_ref().map(tokio::runtime::Handle::enter); + let _evidence = + tracectl::evidence::capture(format!("open-{which}")); + let mut worker = blueprint.worker(); + (0..FLOWS) + .map(|nth| { + let src = format!("1.1.{which}.{}", nth + 1); + let mut convo = Conversation::new( + super::routed::Path::fixture(), + src.parse().unwrap_or_else(|e| { + unreachable!("{src} is an address: {e}") + }), + "3.3.3.1" + .parse() + .unwrap_or_else(|e| unreachable!("{e}")), + u16::from(nth) + 1000, + 80, + ); + step(&mut worker, &mut convo); + convo + }) + .collect::>() + }) + .expect("spawn opener") + }) + .collect(); + running + .into_iter() + .map(|opener| opener.join().expect("opener panicked")) + .collect() + }); + + let second = opened + .pop() + .unwrap_or_else(|| unreachable!("two openers ran")); + let first = opened + .pop() + .unwrap_or_else(|| unreachable!("two openers ran")); + + let answered: Vec> = thread::scope(|scope| { + let running: Vec<_> = [(0u8, second), (1u8, first)] + .into_iter() + .map(|(which, mut theirs)| { + let entering = entering.clone(); + thread::Builder::new() + .name(format!("answer-{which}")) + .spawn_scoped(scope, move || { + let _guard = entering.as_ref().map(tokio::runtime::Handle::enter); + let _evidence = + tracectl::evidence::capture(format!("answer-{which}")); + let mut worker = blueprint.worker(); + for convo in &mut theirs { + if !convo.finished() { + step(&mut worker, convo); + } + } + theirs + }) + .expect("spawn answerer") + }) + .collect(); + running + .into_iter() + .map(|answerer| answerer.join().expect("answerer panicked")) + .collect() + }); + + for convo in answered.into_iter().flatten() { + if convo.checked() { + CLOSED.fetch_add(1, Ordering::Relaxed); + } else { + ABANDONED.fetch_add(1, Ordering::Relaxed); + } + } + }); + + let (closed, abandoned) = ( + CLOSED.load(Ordering::Relaxed), + ABANDONED.load(Ordering::Relaxed), + ); + eprintln!("closed={closed} abandoned={abandoned}"); + super::assert_covered( + closed > 0, + "no conversation was ever answered by the other worker, so nothing crossed", + ); + } + + fn unreachable( + code: net::icmp4::Icmp4DestUnreachable, + public: (IpAddr, u16), + target: IpAddr, + ) -> Packet { + let (IpAddr::V4(public_v4), IpAddr::V4(target_v4)) = (public.0, target) else { + unreachable!("the fixture is v4 throughout") + }; + let inner = + net::packet::test_utils::build_test_icmp4_destination_unreachable_packet_with_code( + code, + net::packet::test_utils::IcmpErrorAddrs { + outer_src: target_v4, + outer_dst: public_v4, + inner_src: public_v4, + inner_dst: target_v4, + }, + net::ip::NextHeader::UDP, + public.1, + 80, + ) + .unwrap_or_else(|e| unreachable!("the icmp error builds: {e:?}")); + super::routed::tunnelled_from(vni(REMOTE_VNI), &inner) + } + + #[concurrency::model_test] + fn an_icmp_teardown_leaves_another_workers_flow_alone() { + static REPORTED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + static SURVIVED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + + let _eal = dpdk::test_support::start_eal(); + + let rt = cfg_select! { + feature = "shuttle" => None::, + _ => Some( + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("build tokio runtime") + ) + }; + let handle = rt.as_ref().map(tokio::runtime::Runtime::handle).cloned(); + + concurrency::stress(move || { + let target: IpAddr = "3.3.3.1".parse().unwrap_or_else(|e| unreachable!("{e}")); + + for (code, tears_down) in [ + (net::icmp4::Icmp4DestUnreachable::Network, true), + ( + net::icmp4::Icmp4DestUnreachable::FragmentationNeeded { + next_hop_mtu: Some(1400.try_into().unwrap_or_else(|_| unreachable!())), + }, + false, + ), + ] { + let overlay = overlay_with_exposes_and_acl(exposes(), None) + .expect("the fixture exposes form an overlay") + .validate() + .expect("the fixture overlay validates"); + let tables = topology(&[vni(LOCAL_VNI), vni(REMOTE_VNI)]); + let fleet = + Fleet::lowering(&overlay, Some(&tables), Arc::new(FlowTable::default())); + let blueprint = fleet.blueprint(); + let entering = handle.clone(); + + let opening = entering.clone(); + let answering = entering.clone(); + let (doomed, mut spared): ((IpAddr, u16), Conversation) = thread::scope(|scope| { + let doomed = thread::Builder::new() + .name("open-doomed".to_owned()) + .spawn_scoped(scope, move || { + let _guard = opening.as_ref().map(tokio::runtime::Handle::enter); + let _evidence = tracectl::evidence::capture("open-doomed"); + let mut worker = blueprint.worker(); + let request = super::round_trip::udp( + "1.1.0.1".parse().unwrap_or_else(|e| unreachable!("{e}")), + target, + 1000, + 80, + ) + .unwrap_or_else(|| unreachable!("the fixture request builds")); + let out = worker.send(tunnelled(&request)); + let carried = inside(&out).unwrap_or_else(|| { + unreachable!( + "the request was not delivered tunnelled: {:?}", + verdict(&out) + ) + }); + let (Some(public), Some(port)) = + (carried.ip_source(), carried.transport_src_port()) + else { + unreachable!("a delivered request had no public tuple") + }; + (public, port.get()) + }) + .expect("spawn opener"); + + let spared = thread::Builder::new() + .name("open-spared".to_owned()) + .spawn_scoped(scope, move || { + let _guard = answering.as_ref().map(tokio::runtime::Handle::enter); + let _evidence = tracectl::evidence::capture("open-spared"); + let mut worker = blueprint.worker(); + let mut convo = Conversation::new( + super::routed::Path::fixture(), + "1.1.0.2".parse().unwrap_or_else(|e| unreachable!("{e}")), + target, + 2000, + 80, + ); + step(&mut worker, &mut convo); + convo + }) + .expect("spawn opener"); + + ( + doomed.join().expect("opener panicked"), + spared.join().expect("opener panicked"), + ) + }); + + let tearing = entering.clone(); + let keeping = entering.clone(); + let named = format!("{code:?}"); + let spared = thread::scope(|scope| { + let teardown = thread::Builder::new() + .name("teardown".to_owned()) + .spawn_scoped(scope, move || { + let _guard = tearing.as_ref().map(tokio::runtime::Handle::enter); + let _evidence = tracectl::evidence::capture("teardown"); + let mut worker = blueprint.worker(); + let out = worker.send(unreachable(code, doomed, target)); + matches!(verdict(&out), Verdict::Delivered { .. }) + }) + .expect("spawn teardown"); + + let answer = thread::Builder::new() + .name("answer".to_owned()) + .spawn_scoped(scope, move || { + let _guard = keeping.as_ref().map(tokio::runtime::Handle::enter); + let _evidence = tracectl::evidence::capture("answer"); + let mut worker = blueprint.worker(); + step(&mut worker, &mut spared); + spared + }) + .expect("spawn answer"); + + assert!( + teardown.join().expect("teardown panicked"), + "the icmp error never reached the flow it named, so this raced against \ + nothing" + ); + REPORTED.fetch_add(1, Ordering::Relaxed); + answer.join().expect("answer panicked") + }); + + assert!( + spared.checked(), + "a flow was disturbed by an icmp teardown of a different flow on another \ + worker. {}", + spared.describe() + ); + SURVIVED.fetch_add(1, Ordering::Relaxed); + + let mut worker = blueprint.worker(); + let reply = super::round_trip::udp(target, doomed.0, 80, doomed.1) + .unwrap_or_else(|| unreachable!("the reply builds")); + let out = worker.send(super::routed::tunnelled_from(vni(REMOTE_VNI), &reply)); + let delivered = matches!(verdict(&out), Verdict::Delivered { .. }); + assert_eq!( + delivered, + !tears_down, + "an icmp error with code {named} left the flow it named {}: {:?}", + if delivered { "alive" } else { "torn down" }, + verdict(&out) + ); + } + }); + + let (reported, survived) = ( + REPORTED.load(Ordering::Relaxed), + SURVIVED.load(Ordering::Relaxed), + ); + eprintln!("reported={reported} survived={survived}"); + super::assert_covered(reported > 0, "no icmp error ever reached the flow it named"); + } + + #[concurrency::model_test] + fn forwarding_survives_a_route_being_republished_underneath_it() { + const FLOWS: u8 = 2; + const CHURN: u8 = 6; + + static COMPLETED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + static PUBLISHED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + + let _eal = dpdk::test_support::start_eal(); + + let rt = cfg_select! { + feature = "shuttle" => None::, + _ => Some( + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("build tokio runtime") + ) + }; + let handle = rt.as_ref().map(tokio::runtime::Runtime::handle).cloned(); + + concurrency::stress(move || { + let overlay = overlay_with_exposes_and_acl(exposes(), None) + .expect("the fixture exposes form an overlay") + .validate() + .expect("the fixture overlay validates"); + let mut tables = topology(&[vni(LOCAL_VNI), vni(REMOTE_VNI)]); + let fleet = Fleet::lowering(&overlay, Some(&tables), Arc::new(FlowTable::default())); + let blueprint = fleet.blueprint(); + let entering = handle.clone(); + let start = Arc::new(concurrency::sync::Barrier::new(3)); + + thread::scope(|scope| { + let running: Vec<_> = (0..2u8) + .map(|which| { + let entering = entering.clone(); + let start = start.clone(); + thread::Builder::new() + .name(format!("forward-{which}")) + .spawn_scoped(scope, move || { + let _guard = entering.as_ref().map(tokio::runtime::Handle::enter); + let _evidence = + tracectl::evidence::capture(format!("forward-{which}")); + let mut worker = blueprint.worker(); + start.wait(); + let mut done = 0u64; + for nth in 0..FLOWS { + let src = format!("1.1.{which}.{}", nth + 1); + let mut convo = Conversation::new( + super::routed::Path::fixture(), + src.parse().unwrap_or_else(|e| unreachable!("{src}: {e}")), + "3.3.3.1".parse().unwrap_or_else(|e| unreachable!("{e}")), + u16::from(nth) + 1000, + 80, + ); + drive(&mut worker, &mut convo); + assert!( + convo.checked(), + "a conversation did not complete while routes were being \ + republished. {}", + convo.describe() + ); + done += 1; + } + done + }) + .expect("spawn forwarder") + }) + .collect(); + + let peer: IpAddr = PEER_VTEP.parse().unwrap_or_else(|_| unreachable!()); + let landing = + FibGroup::with_entry(FibEntry::with_inst(PktInstruction::Local(uplink()))); + start.wait(); + for nth in 0..CHURN { + let prefix = format!("9.9.{nth}.0"); + tables.route_via( + UNDERLAY_VRF, + Prefix::expect_from(( + prefix + .parse::() + .unwrap_or_else(|e| unreachable!("{e}")), + 24, + )), + nhop(&peer), + &landing, + ); + PUBLISHED.fetch_add(1, Ordering::Relaxed); + } + + for worker in running { + COMPLETED.fetch_add( + worker.join().expect("forwarder panicked"), + Ordering::Relaxed, + ); + } + }); + }); + + let (completed, published) = ( + COMPLETED.load(Ordering::Relaxed), + PUBLISHED.load(Ordering::Relaxed), + ); + eprintln!("completed={completed} published={published}"); + super::assert_covered( + completed > 0 && published > 0, + "either no conversation completed or no route was published, so nothing was raced", + ); + } + + #[concurrency::model_test] + fn a_next_hop_that_moves_is_never_seen_half_moved() { + const CHURN: u8 = 3; + const PER_ROUND: u8 = 2; + + static FRESH: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + static STALE: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + + type Seen = Result; + + fn waypoint(nth: u8) -> (IpAddr, InterfaceIndex) { + if nth == 0 { + ( + PEER_VTEP.parse().unwrap_or_else(|_| unreachable!()), + uplink(), + ) + } else { + ( + IpAddr::from([10, 0, 0, nth]), + InterfaceIndex::try_new(UPLINK + u32::from(nth)) + .unwrap_or_else(|_| unreachable!()), + ) + } + } + + fn framing(nth: u8) -> Mac { + Mac([0x02, 0, 0, 0, 0x11, nth]) + } + + fn towards(nth: u8) -> FibGroup { + let (remote, oif) = waypoint(nth); + let mut out = FibEntry::with_inst(PktInstruction::Encap(ResolvedEncapsulation::Vxlan( + ResolvedVxlan { + vni: vni(REMOTE_VNI), + remote, + dmac: framing(nth), + }, + ))); + out.add(PktInstruction::Egress(EgressObject::new( + Some(oif), + Some(remote), + ))); + FibGroup::with_entry(out) + } + + let _eal = dpdk::test_support::start_eal(); + + let rt = cfg_select! { + feature = "shuttle" => None::, + _ => Some( + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("build tokio runtime") + ) + }; + let handle = rt.as_ref().map(tokio::runtime::Runtime::handle).cloned(); + + concurrency::stress(move || { + let overlay = overlay_with_exposes_and_acl(exposes(), None) + .expect("the fixture exposes form an overlay") + .validate() + .expect("the fixture overlay validates"); + let mut tables = topology(&[vni(LOCAL_VNI), vni(REMOTE_VNI)]); + for nth in 1..=CHURN { + let (_, oif) = waypoint(nth); + tables.interface( + oif, + &format!("uplink-{nth}"), + SourceMac::new(framing(nth)).unwrap_or_else(|_| unreachable!()), + ); + tables.attach(oif, UNDERLAY_VRF); + } + for to in 0..=CHURN { + for over in 0..=CHURN { + tables.adjacency(waypoint(to).0, waypoint(over).1, framing(to)); + } + } + let fleet = Fleet::lowering(&overlay, Some(&tables), Arc::new(FlowTable::default())); + let blueprint = fleet.blueprint(); + let entering = handle.clone(); + let gate = Arc::new(concurrency::sync::Barrier::new(3)); + let mut reports = Vec::new(); + + thread::scope(|scope| { + let running: Vec<_> = (0..2u8) + .map(|which| { + let entering = entering.clone(); + let gate = gate.clone(); + thread::Builder::new() + .name(format!("probe-{which}")) + .spawn_scoped(scope, move || { + let _guard = entering.as_ref().map(tokio::runtime::Handle::enter); + let _evidence = + tracectl::evidence::capture(format!("probe-{which}")); + let mut worker = blueprint.worker(); + + let mut port = u16::from(which) * 1000 + 1000; + let send = |worker: &mut Worker, port: &mut u16| -> Seen { + *port += 1; + let src = format!("1.1.{which}.1"); + let out = worker.send(tunnelled(&build_test_udp_ipv4_packet( + &src, "3.3.3.1", *port, 80, + ))); + let Verdict::Delivered { + oif: Some(oif), + dst: Some(dst), + .. + } = verdict(&out) + else { + return Err(format!( + "it did not leave the gateway: {:?}", + verdict(&out) + )); + }; + (0..=CHURN) + .find(|nth| waypoint(*nth) == (dst, oif)) + .ok_or_else(|| { + format!( + "it left over interface {oif} towards {dst}, which \ + is no published next hop: either the \ + encapsulation and the egress came from different \ + versions, or the group was read while it was \ + being written" + ) + }) + }; + + let probe = |worker: &mut Worker, port: &mut u16| -> Seen { + without_unwinding(|| send(worker, port)).unwrap_or_else(|why| { + Err(format!("sending it panicked: {why}")) + }) + }; + + gate.wait(); + let mut seen = Vec::with_capacity( + usize::from(CHURN) * usize::from(PER_ROUND) + 1, + ); + for _ in 1..=CHURN { + for _ in 0..PER_ROUND { + seen.push(probe(&mut worker, &mut port)); + } + gate.wait(); + } + seen.push(probe(&mut worker, &mut port)); + seen + }) + .expect("spawn prober") + }) + .collect(); + + let key = nhop(&PEER_VTEP.parse().unwrap_or_else(|_| unreachable!())); + gate.wait(); + for round in 1..=CHURN { + tables.nexthop(REMOTE_VNI, &key, &towards(round)); + gate.wait(); + } + + for prober in running { + reports.push(prober.join().expect("prober panicked")); + } + }); + + for seen in reports { + for (nth, observed) in seen.iter().enumerate() { + let last = nth == seen.len() - 1; + let round = if last { + CHURN + } else { + u8::try_from(nth).unwrap_or_else(|_| unreachable!()) / PER_ROUND + 1 + }; + let version = match observed { + Ok(version) => *version, + Err(why) => panic!( + "a probe sent in round {round}, while the next hop was moving, is not \ + attributable to any version: {why}" + ), + }; + if last { + assert_eq!( + version, CHURN, + "a probe sent after the churn had finished was forwarded by version \ + {version}, not by version {CHURN}, the last one published. Every \ + publish returned before the barrier that released this probe, so a \ + reader still serving an earlier version is serving a next hop that \ + no longer exists" + ); + continue; + } + assert!( + version == round || version + 1 == round, + "a probe sent in round {round} was forwarded by version {version}. \ + Version {} was published before this round opened, so no reader may \ + still be serving anything older, and version {round} is the newest that \ + exists", + round - 1 + ); + if version == round { + FRESH.fetch_add(1, Ordering::Relaxed); + } else { + STALE.fetch_add(1, Ordering::Relaxed); + } + } + } + }); + + let (fresh, stale) = (FRESH.load(Ordering::Relaxed), STALE.load(Ordering::Relaxed)); + eprintln!("fresh={fresh} stale={stale}"); + super::assert_covered( + fresh > 0, + "no probe was ever forwarded by the version published in its own round, so the publish \ + never once landed inside the window it was racing", + ); + } + + #[concurrency::model_test] + fn re_enacting_a_configuration_under_load_disturbs_nothing() { + const FLOWS: u8 = 2; + const APPLIES: u8 = 4; + + static COMPLETED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + static ENACTED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + + let _eal = dpdk::test_support::start_eal(); + + let rt = cfg_select! { + feature = "shuttle" => None::, + _ => Some( + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("build tokio runtime") + ) + }; + let handle = rt.as_ref().map(tokio::runtime::Runtime::handle).cloned(); + + concurrency::stress(move || { + let overlay = overlay_with_exposes_and_acl(exposes(), None) + .expect("the fixture exposes form an overlay") + .validate() + .expect("the fixture overlay validates"); + let tables = topology(&[vni(LOCAL_VNI), vni(REMOTE_VNI)]); + let fleet = Fleet::lowering(&overlay, Some(&tables), Arc::new(FlowTable::default())); + let blueprint = fleet.blueprint(); + let entering = handle.clone(); + let gate = Arc::new(concurrency::sync::Barrier::new(3)); + let mut reports = Vec::new(); + + thread::scope(|scope| { + let running: Vec<_> = (0..2u8) + .map(|which| { + let entering = entering.clone(); + let gate = gate.clone(); + thread::Builder::new() + .name(format!("tenant-{which}")) + .spawn_scoped(scope, move || { + let _guard = entering.as_ref().map(tokio::runtime::Handle::enter); + let _evidence = + tracectl::evidence::capture(format!("tenant-{which}")); + let mut worker = blueprint.worker(); + gate.wait(); + let mut seen = Vec::new(); + for round in 1..=APPLIES { + for nth in 0..FLOWS { + seen.push(( + round, + without_unwinding(|| { + let src = format!("1.1.{which}.{}", nth + 1); + let mut convo = Conversation::new( + super::routed::Path::fixture(), + src.parse().unwrap_or_else(|e| { + unreachable!("{src}: {e}") + }), + "3.3.3.1" + .parse() + .unwrap_or_else(|e| unreachable!("{e}")), + u16::from(round) * 100 + u16::from(nth) + 1000, + 80, + ); + drive(&mut worker, &mut convo); + (convo.checked(), convo.describe()) + }), + )); + } + gate.wait(); + } + seen + }) + .expect("spawn tenant") + }) + .collect(); + + gate.wait(); + for _ in 0..APPLIES { + fleet.reconfigure(&overlay); + ENACTED.fetch_add(1, Ordering::Relaxed); + gate.wait(); + } + + for worker in running { + reports.push(worker.join().expect("tenant panicked")); + } + }); + + for seen in reports { + for (round, ran) in seen { + let (checked, described) = ran.unwrap_or_else(|why| { + panic!("a conversation in round {round} panicked mid-enactment: {why}") + }); + assert!( + checked, + "a conversation in round {round} did not survive the configuration it was \ + already running being enacted again. Every enactment before this round \ + had returned, and the one racing it changes nothing. {described}" + ); + COMPLETED.fetch_add(1, Ordering::Relaxed); + } + } + }); + + let (completed, enacted) = ( + COMPLETED.load(Ordering::Relaxed), + ENACTED.load(Ordering::Relaxed), + ); + eprintln!("completed={completed} enacted={enacted}"); + super::assert_covered( + completed > 0 && enacted > 0, + "either no conversation completed or no configuration was enacted, so nothing was raced", + ); + } + + #[concurrency::model_test] + fn a_configuration_change_leaves_traffic_outside_its_footprint_alone() { + const CASES: usize = 64; + const ROUNDS: u8 = 3; + + static FRAMED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + static BARREN: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + static ENGULFED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + static FRAMED_OUT: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + static RACED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + static DISTURBED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + + fn separate(drawn: u16, round: u8, which: usize) -> u16 { + let within = (drawn / 2).wrapping_add(u16::from(round).wrapping_mul(997)) % 32_000 + 1; + within + u16::try_from(which).unwrap_or_else(|_| unreachable!()) * 32_768 + } + + fn outside(footprint: &Footprint) -> impl Fn(derive::Named<'_>) -> bool + '_ { + move |named| { + !footprint.touches_peering_named(named.peering) + && !footprint.touches_vpc_named(named.local) + && !footprint.touches_vpc_named(named.remote) + } + } + + let _eal = dpdk::test_support::start_eal(); + + let rt = cfg_select! { + feature = "shuttle" => None::, + _ => Some( + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("build tokio runtime") + ) + }; + let handle = rt.as_ref().map(tokio::runtime::Runtime::handle).cloned(); + + bolero::check!() + .with_max_len(MAX_INPUT_LEN) + .with_generator(generated::Generated) + .with_iterations(CASES) + .for_each(|(ops, vary, _schedule)| { + let Some((change, built)) = ops.split_last() else { + BARREN.fetch_add(1, Ordering::Relaxed); + return; + }; + let before = Sequence::fold(built); + let footprint = change.writes(&before); + + let assemble = |draft: &config::external::overlay::algebra::Draft| { + draft + .overlay() + .unwrap_or_else(|e| panic!("{ops:?} does not assemble: {e}")) + .validate() + .unwrap_or_else(|e| panic!("{ops:?} does not validate: {e}")) + }; + let running = assemble(&before); + let enacted = assemble(&Sequence::fold(ops)); + + let framed = derive::loads_where(&running, vary, &outside(&footprint)).len(); + let total = derive::loads_for(&running, vary).len(); + FRAMED_OUT.fetch_add( + u64::try_from(total - framed).unwrap_or_else(|_| unreachable!()), + Ordering::Relaxed, + ); + if framed == 0 { + if total == 0 { &BARREN } else { &ENGULFED }.fetch_add(1, Ordering::Relaxed); + return; + } + FRAMED.fetch_add(1, Ordering::Relaxed); + + let vnis: Vec = enacted + .vpc_table() + .values() + .map(config::external::overlay::vpc::ValidatedVpc::vni) + .collect(); + if vnis.is_empty() { + BARREN.fetch_add(1, Ordering::Relaxed); + return; + } + + let drawn = concurrency::sync::Arc::new(( + running, + enacted, + vnis, + vary.clone(), + footprint, + *change, + )); + let entering = handle.clone(); + + concurrency::stress(move || { + let (running, enacted, vnis, vary, footprint, change) = &*drawn; + let tables = topology(vnis); + let fleet = + Fleet::lowering(running, Some(&tables), Arc::new(FlowTable::default())); + let blueprint = fleet.blueprint(); + let gate = Arc::new(concurrency::sync::Barrier::new(3)); + let mut reports = Vec::new(); + + thread::scope(|scope| { + let handles: Vec<_> = (0..2usize) + .map(|which| { + let entering = entering.clone(); + let gate = gate.clone(); + thread::Builder::new() + .name(format!("framed-{which}")) + .spawn_scoped(scope, move || { + let _guard = + entering.as_ref().map(tokio::runtime::Handle::enter); + let _evidence = + tracectl::evidence::capture(format!("framed-{which}")); + let mut worker = blueprint.worker(); + let outside = outside(footprint); + gate.wait(); + let mut seen = Vec::new(); + for round in 1..=ROUNDS { + let varied: Vec = vary + .iter() + .map(|v| derive::Vary { + sport: separate(v.sport, round, which), + ..*v + }) + .collect(); + let mine = + derive::loads_where(running, &varied, &outside); + seen.push(( + round, + without_unwinding(|| { + let mut ran = Vec::new(); + for mut load in mine { + for _ in 0..8 { + let Some(packet) = load.next() else { + break; + }; + let out = worker.send(packet); + load.observe(&out); + } + ran.push((load.checked(), load.describe())); + } + ran + }), + )); + gate.wait(); + } + seen + }) + .expect("spawn framed worker") + }) + .collect(); + + gate.wait(); + for _ in 0..ROUNDS { + fleet.reconfigure(enacted); + gate.wait(); + } + + for worker in handles { + reports.push(worker.join().expect("framed worker panicked")); + } + }); + + for seen in reports { + for (round, ran) in seen { + let ran = ran.unwrap_or_else(|why| { + panic!( + "carrying traffic in round {round} panicked while {change:?} \ + was being enacted: {why}" + ) + }); + for (checked, described) in ran { + if round == 1 { + if checked { &RACED } else { &DISTURBED } + .fetch_add(1, Ordering::Relaxed); + continue; + } + assert!( + checked, + "traffic outside the footprint of a configuration change did \ + not survive it, in round {round}. The change was {change:?}, \ + whose write set this load is outside of, so the configuration \ + it ran against and the one enacted agree about it entirely. \ + {}. {described}", + if round == 1 { + "This round races the enactment that carries the difference" + } else { + "That enactment returned before this round opened, and \ + this round races only a re-enactment of what is already \ + running" + } + ); + } + } + } + }); + }); + + let (framed, barren, engulfed, framed_out, raced) = ( + FRAMED.load(Ordering::Relaxed), + BARREN.load(Ordering::Relaxed), + ENGULFED.load(Ordering::Relaxed), + FRAMED_OUT.load(Ordering::Relaxed), + RACED.load(Ordering::Relaxed), + ); + eprintln!("framed={framed} barren={barren} engulfed={engulfed} framed_out={framed_out}"); + eprintln!( + "carried while the change was being enacted: {raced}, of which disturbed: {}", + DISTURBED.load(Ordering::Relaxed) + ); + super::assert_covered( + framed > 0, + "no draw left two loads outside the footprint of its last operation, so no \ + configuration change was ever carried under traffic", + ); + super::assert_covered( + raced > 0, + "no load was ever carried by the round that races the change itself, so every \ + assertion here was about a re-enactment of a configuration already running -- which \ + `re_enacting_a_configuration_under_load_disturbs_nothing` already covers", + ); + super::assert_covered( + framed_out > 0, + "no load was ever filtered out for being inside a footprint, so the frame was always \ + the whole configuration and this property is the re-enactment one with extra steps", + ); + } + + #[tokio::test] + #[dpdk::with_eal] + #[ignore = "an instrument, not a property: reports a rate and asserts nothing"] + #[allow(clippy::too_many_lines, reason = "one instrument, read top to bottom")] + async fn report_which_enactment_step_disturbs_traffic() { + use config::external::overlay::algebra::{ + Draft, Flavour, Op, PeeringHandle, Side, VpcHandle, + }; + + fn separate(drawn: u16, rep: u8, which: u16) -> u16 { + let within = (drawn / 2).wrapping_add(u16::from(rep).wrapping_mul(997)) % 32_000 + 1; + within + which * 32_768 + } + + const ROUNDS: usize = 60; + const REPS: u8 = 24; + + let vpc = VpcHandle; + let peering = PeeringHandle; + let built = vec![ + Op::AddVpc(vpc(0)), + Op::AddVpc(vpc(1)), + Op::AddPeering { + handle: peering(0), + left: vpc(0), + right: vpc(1), + }, + Op::SetFlavour { + peering: peering(0), + side: Side::Left, + slot: 0, + flavour: Flavour::Masquerade, + }, + Op::AddVpc(vpc(2)), + Op::AddVpc(vpc(3)), + Op::AddPeering { + handle: peering(1), + left: vpc(2), + right: vpc(3), + }, + ]; + let change = Op::SetFlavour { + peering: peering(1), + side: Side::Left, + slot: 0, + flavour: Flavour::Masquerade, + }; + let assemble = |draft: &Draft| { + draft + .overlay() + .expect("assembles") + .validate() + .expect("validates") + }; + let before = Sequence::fold(&built); + let running = assemble(&before); + let mut after_draft = before.clone(); + change.apply(&mut after_draft).expect("the change applies"); + let enacted = assemble(&after_draft); + let footprint = change.writes(&before); + + let vnis: Vec = enacted + .vpc_table() + .values() + .map(config::external::overlay::vpc::ValidatedVpc::vni) + .collect(); + let vary: Vec = (0..2) + .map(|n| derive::Vary { + host: n, + port: 0, + sport: 30000, + dport: 80, + burst: 2, + blast: false, + }) + .collect(); + let outside = |named: derive::Named<'_>| { + !footprint.touches_peering_named(named.peering) + && !footprint.touches_vpc_named(named.local) + && !footprint.touches_vpc_named(named.remote) + }; + eprintln!( + "loads: total={} framed={}", + derive::loads_for(&running, &vary).len(), + derive::loads_where(&running, &vary, &outside).len() + ); + + let handle = tokio::runtime::Handle::current(); + let steps = [ + Enact::Everything, + Enact::FlowFilter, + Enact::Acl, + Enact::StaticNat, + Enact::Masquerade, + Enact::PortForward, + Enact::Generation, + Enact::Everything, + ]; + for (nth, part) in steps.into_iter().enumerate() { + let gapped = nth == steps.len() - 1; + let disturbed = concurrency::sync::atomic::AtomicU64::new(0); + let carried = concurrency::sync::atomic::AtomicU64::new(0); + + for round in 0..ROUNDS { + let tables = topology(&vnis); + let fleet = + Fleet::lowering(&running, Some(&tables), Arc::new(FlowTable::default())); + let blueprint = fleet.blueprint(); + let gate = concurrency::sync::Barrier::new(3); + std::thread::scope(|scope| { + for which in 0..2u16 { + let handle = handle.clone(); + let (gate, disturbed, carried) = (&gate, &disturbed, &carried); + let (running, vary, outside) = (&running, &vary, &outside); + scope.spawn(move || { + let _guard = handle.enter(); + let mut worker = blueprint.worker(); + gate.wait(); + for rep in 0..REPS { + let mine = derive::loads_where( + running, + &vary + .iter() + .map(|v| derive::Vary { + sport: separate(v.sport, rep, which), + ..*v + }) + .collect::>(), + outside, + ); + for mut load in mine { + carried.fetch_add(1, Ordering::Relaxed); + let ran = without_unwinding(|| { + for _ in 0..8 { + let Some(packet) = load.next() else { break }; + let out = worker.send(packet); + load.observe(&out); + } + (load.checked(), load.describe()) + }); + let (ok, how) = ran.unwrap_or_else(|why| (false, why)); + if !ok && disturbed.fetch_add(1, Ordering::Relaxed) == 0 { + eprintln!(" {part:?} first, round {round}: {how}"); + } + } + } + }); + } + gate.wait(); + if gapped { + fleet.enact_with_router_gap(&enacted, Duration::from_micros(500)); + } else { + fleet.enact(&enacted, part); + } + }); + } + eprintln!( + "{part:?}{}: carried={} disturbed={}", + if gapped { " (with the router gap)" } else { "" }, + carried.load(Ordering::Relaxed), + disturbed.load(Ordering::Relaxed) + ); + } + } + + #[tokio::test] + #[dpdk::with_eal] + #[ignore = "an instrument, not a property: prints one trace and asserts nothing"] + #[allow(clippy::too_many_lines, reason = "one instrument, read top to bottom")] + async fn report_why_the_masquerade_swap_disturbs_traffic() { + use concurrency::sync::atomic::AtomicBool; + use config::external::overlay::algebra::{ + Draft, Flavour, Op, PeeringHandle, Side, VpcHandle, + }; + + fn separate(drawn: u16, rep: u8, which: u16) -> u16 { + let within = (drawn / 2).wrapping_add(u16::from(rep).wrapping_mul(997)) % 32_000 + 1; + within + which * 32_768 + } + + const ROUNDS: [usize; 2] = [200, 6000]; + const REPS: u8 = 24; + + fn reason(how: &str) -> String { + how.lines() + .next() + .unwrap_or(how) + .rsplit('|') + .next() + .unwrap_or(how) + .trim() + .to_owned() + } + + let vpc = VpcHandle; + let peering = PeeringHandle; + let built = vec![ + Op::AddVpc(vpc(0)), + Op::AddVpc(vpc(1)), + Op::AddPeering { + handle: peering(0), + left: vpc(0), + right: vpc(1), + }, + Op::SetFlavour { + peering: peering(0), + side: Side::Left, + slot: 0, + flavour: Flavour::Masquerade, + }, + Op::AddVpc(vpc(2)), + Op::AddVpc(vpc(3)), + Op::AddPeering { + handle: peering(1), + left: vpc(2), + right: vpc(3), + }, + ]; + let change = Op::SetFlavour { + peering: peering(1), + side: Side::Left, + slot: 0, + flavour: Flavour::Masquerade, + }; + let assemble = |draft: &Draft| { + draft + .overlay() + .expect("assembles") + .validate() + .expect("validates") + }; + let before = Sequence::fold(&built); + let running = assemble(&before); + let mut after_draft = before.clone(); + change.apply(&mut after_draft).expect("the change applies"); + let enacted = assemble(&after_draft); + let footprint = change.writes(&before); + + let vnis: Vec = enacted + .vpc_table() + .values() + .map(config::external::overlay::vpc::ValidatedVpc::vni) + .collect(); + let vary: Vec = (0..2) + .map(|n| derive::Vary { + host: n, + port: 0, + sport: 30000, + dport: 80, + burst: 2, + blast: false, + }) + .collect(); + let outside = |named: derive::Named<'_>| { + !footprint.touches_peering_named(named.peering) + && !footprint.touches_vpc_named(named.local) + && !footprint.touches_vpc_named(named.remote) + }; + + let handle = tokio::runtime::Handle::current(); + for (nth, randomize) in [false, true].into_iter().enumerate() { + eprintln!("\n== randomize={randomize} =="); + let explained = AtomicBool::new(false); + let carried = AtomicU64::new(0); + let disturbed = AtomicU64::new(0); + let why: Mutex> = + Mutex::new(std::collections::BTreeMap::new()); + let mut duplicated = 0u64; + + for round in 0..ROUNDS[nth] { + let tables = topology(&vnis); + let fleet = + Fleet::lowering(&running, Some(&tables), Arc::new(FlowTable::default())); + fleet.randomizing(randomize); + if randomize { + fleet.enact(&running, Enact::Masquerade); + } + let blueprint = fleet.blueprint(); + let gate = concurrency::sync::Barrier::new(3); + let handed: Mutex> = Mutex::new(Vec::new()); + let writing = tracectl::evidence::Capture::new("config-apply") + .depth(8192) + .keeping(|target| target.starts_with("dataplane_nat")) + .start(); + let written = writing.evidence(); + let traces: Mutex> = Mutex::new(vec![written]); + std::thread::scope(|scope| { + for which in 0..2u16 { + let handle = handle.clone(); + let (gate, explained, carried, disturbed) = + (&gate, &explained, &carried, &disturbed); + let (running, vary, outside) = (&running, &vary, &outside); + let (traces, why, handed) = (&traces, &why, &handed); + scope.spawn(move || { + let _guard = handle.enter(); + let recording = + tracectl::evidence::Capture::new(format!("masq-{which}")) + .depth(8192) + .keeping(|target| target.starts_with("dataplane_nat")) + .start(); + traces.lock().push(recording.evidence()); + let mut worker = blueprint.worker(); + gate.wait(); + let mut given = Vec::new(); + for rep in 0..REPS { + let mine = derive::loads_where( + running, + &vary + .iter() + .map(|v| derive::Vary { + sport: separate(v.sport, rep, which), + ..*v + }) + .collect::>(), + outside, + ); + for mut load in mine { + carried.fetch_add(1, Ordering::Relaxed); + let ran = without_unwinding(|| { + for _ in 0..8 { + let Some(packet) = load.next() else { break }; + let out = worker.send(packet); + load.observe(&out); + } + load.checked() + }); + given.extend(load.public()); + let (ok, how) = match ran { + Ok(ok) => (ok, load.describe()), + Err(why) => (false, format!("{why}. {}", load.describe())), + }; + if ok { + continue; + } + disturbed.fetch_add(1, Ordering::Relaxed); + *why.lock().entry(reason(&how)).or_default() += 1; + if !explained.swap(true, Ordering::Relaxed) { + eprintln!( + "\ndisturbed in round {round}, rep {rep}, worker \ + {which}: {how}\n" + ); + for trace in traces.lock().iter() { + trace.dump(); + } + } + } + } + handed.lock().extend(given); + }); + } + gate.wait(); + fleet.enact(&enacted, Enact::Masquerade); + }); + + let mut seen = std::collections::BTreeSet::new(); + for tuple in handed.lock().iter() { + if !seen.insert(*tuple) { + duplicated += 1; + eprintln!(" round {round}: {tuple:?} was handed to two live flows"); + } + } + } + + eprintln!( + "carried={} disturbed={} duplicated={duplicated}", + carried.load(Ordering::Relaxed), + disturbed.load(Ordering::Relaxed), + ); + for (reason, count) in why.lock().iter() { + eprintln!(" {count:>4} {reason}"); + } + } + } +} diff --git a/default.nix b/default.nix index c2a06e9055..c153120f62 100644 --- a/default.nix +++ b/default.nix @@ -87,10 +87,15 @@ let }; in if platform != "wasm32-wasip1" then over.pkgsCross.${platform'.info.nixarch} else over; + sysroot-stamp = '' + printf '%s' '${sanitize}' > "$out/.sanitize" + printf '%s' '${instrumentation}' > "$out/.instrumentation" + ''; sysroot = if platform != "wasm32-wasip1" then pkgs.symlinkJoin { name = "sysroot"; + postBuild = sysroot-stamp; paths = with pkgs.pkgsHostHost; [ pkgs.pkgsHostHost.libc.dev # fully qualified: bare `libc` resolves to the "gnu" function argument, not pkgs.pkgsHostHost.libc pkgs.pkgsHostHost.libc.out # (same as above) @@ -116,6 +121,7 @@ let else pkgs.symlinkJoin { name = "sysroot"; + postBuild = sysroot-stamp; paths = with pkgs.pkgsHostHost; [ fancy.hwloc.dev fancy.hwloc.static diff --git a/dpdk/src/acl/context.rs b/dpdk/src/acl/context.rs index 5e81675734..aaac8373e6 100644 --- a/dpdk/src/acl/context.rs +++ b/dpdk/src/acl/context.rs @@ -41,7 +41,7 @@ use core::fmt; use core::mem::ManuallyDrop; use core::ptr::NonNull; -use concurrency::sync::{Mutex, OnceLock}; +use concurrency::sync::OnceLock; use errno::Errno; use tracing::{debug, error, trace}; @@ -71,13 +71,7 @@ use super::rule::Rule; /// `loom`/`shuttle` model-checker backends, `concurrency::sync::Mutex::new` /// is not `const fn` (each instance registers with the scheduler), so a /// `static M: Mutex<()> = Mutex::new(())` would fail to typecheck on those -/// configurations. `OnceLock` + lazy init is the portable idiom across -/// all backends. See the module docs on `concurrency::sync`. /// -/// Why the concurrency facade rather than [`std::sync::Mutex`] directly: -/// the workspace policy is poison-as-panic ("poison is a fatal invariant -/// violation"); the facade applies that policy uniformly so call sites -/// never see `LockResult`. /// /// # Tracing reentrancy /// @@ -91,11 +85,38 @@ use super::rule::Rule; /// configuration never touches ACL, but custom layers (e.g. one that /// resolves the context name from a registry lookup for log enrichment) /// could trip this if added later. -static ACL_CREATE_LOCK: OnceLock> = OnceLock::new(); +static ACL_CREATE_LOCK: OnceLock = OnceLock::new(); -/// Lazy accessor for [`ACL_CREATE_LOCK`]. -fn acl_create_lock() -> &'static Mutex<()> { - ACL_CREATE_LOCK.get_or_init(|| Mutex::new(())) +concurrency::with_std! { + type RegistryMutex = concurrency::sync::Mutex<()>; + + fn hold(lock: &'static RegistryMutex) -> impl Sized { + lock.lock() + } +} + +concurrency::with_loom! { + // nosemgrep: rust-no-direct-std-sync-import + type RegistryMutex = std::sync::Mutex<()>; + + fn hold(lock: &'static RegistryMutex) -> impl Sized { + lock.lock() + .unwrap_or_else(|_| unreachable!("the acl registry lock is poisoned")) + } +} + +concurrency::with_shuttle! { + // nosemgrep: rust-no-direct-std-sync-import + type RegistryMutex = std::sync::Mutex<()>; + + fn hold(lock: &'static RegistryMutex) -> impl Sized { + lock.lock() + .unwrap_or_else(|_| unreachable!("the acl registry lock is poisoned")) + } +} + +fn hold_acl_registry() -> impl Sized { + hold(ACL_CREATE_LOCK.get_or_init(|| RegistryMutex::new(()))) } // --------------------------------------------------------------------------- @@ -475,7 +496,7 @@ impl AclContext> { // leaves DPDK's TAILQ in an unknown state, and continuing // silently could lead to use-after-free. Aborting via the // panic is the only safe answer. - let _create_guard = acl_create_lock().lock(); + let _create_guard = hold_acl_registry(); // Pre-flight: DPDK's `rte_acl_create` silently returns the existing // context for a duplicate name. Refuse if one is already registered. @@ -1181,7 +1202,7 @@ impl Drop for AclContext { // The facade panics on poison. Dropping while another holder // panicked mid-operation means the DPDK registry may be in an // unknown state; aborting via the panic is the only safe answer. - let _guard = acl_create_lock().lock(); + let _guard = hold_acl_registry(); // SAFETY: rte_acl_free is safe to call on any valid context pointer; `Drop` runs at // most once per `AclContext`, and the create-lock acquired above serialises against // `rte_acl_create` / `dump_all_contexts`. @@ -1211,7 +1232,7 @@ pub fn dump_all_contexts() { // expose a list in an inconsistent state to the walk. Facade panics // on poison (workspace policy -- a prior holder panic implies the // registry may be inconsistent). - let _guard = acl_create_lock().lock(); + let _guard = hold_acl_registry(); // SAFETY: rte_acl_list_dump takes no arguments and simply iterates an internal list. unsafe { dpdk_sys::rte_acl_list_dump() } } diff --git a/flow-filter/src/context/tables.rs b/flow-filter/src/context/tables.rs index dd2a051ba0..1c4a72fc64 100644 --- a/flow-filter/src/context/tables.rs +++ b/flow-filter/src/context/tables.rs @@ -347,18 +347,37 @@ impl fmt::Debug for AnyTable { } } -// Lazily initialized so this compiles under the loom backend, whose AtomicU64::new is not const -// (each instance registers with the loom executor). The atomic itself is still the backend atomic, -// so fetch_add() stays instrumented; only construction is deferred. On every other backend LazyLock -// is a thin wrapper over an otherwise-const atomic. -static TABLE_SEQ: LazyLock = LazyLock::new(|| AtomicU64::new(0)); +concurrency::with_std! { + static TABLE_SEQ: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + + fn next_in_sequence() -> u64 { + TABLE_SEQ.fetch_add(1, Ordering::Relaxed) + } +} + +concurrency::with_loom! { + // nosemgrep: rust-no-direct-std-sync-import + static TABLE_SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); + + fn next_in_sequence() -> u64 { + // nosemgrep: rust-no-direct-std-sync-import + TABLE_SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed) + } +} + +concurrency::with_shuttle! { + // nosemgrep: rust-no-direct-std-sync-import + static TABLE_SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); + + fn next_in_sequence() -> u64 { + // nosemgrep: rust-no-direct-std-sync-import + TABLE_SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed) + } +} /// A process-unique rte_acl context name (rte_acl rejects duplicate names). fn table_name(base: &str) -> String { - format!( - "flow_filter_{base}_{}", - TABLE_SEQ.fetch_add(1, Ordering::Relaxed) - ) + format!("flow_filter_{base}_{}", next_in_sequence()) } /// Build one table from backend-neutral rules using the selected backend. diff --git a/justfile b/justfile index f6257d8777..2f481f291f 100644 --- a/justfile +++ b/justfile @@ -205,6 +205,18 @@ fuzz target time="60s" *args="": # asan does not need that, and skipping the std rebuild keeps it far quicker. # `sanitize=NONE` drops instrumentation altogether, which buys roughly four times # the executions per second in exchange for only catching what the test asserts. + sysroot="${DATAPLANE_SYSROOT:-}" + if [ -n "${sysroot}" ] && [ -r "${sysroot}/.sanitize" ]; then + built_with="$(cat "${sysroot}/.sanitize")" + if [ "${built_with}" != "{{ sanitize }}" ]; then + printf 'refusing to fuzz: sanitize=%s was asked for, but this sysroot was built with sanitize=%s.\n' \ + "{{ sanitize }}" "${built_with:-}" >&2 + printf 'the C dependencies would not be instrumented. Re-enter the shell with:\n' >&2 + printf ' just sanitize=%s setup-roots && nix-shell --argstr sanitize %s\n' \ + "{{ sanitize }}" "{{ sanitize }}" >&2 + exit 1 + fi + fi corpus_dir="{{ fuzz_corpus_root }}/$(printf '%s' '{{ target }}' | tr -c 'A-Za-z0-9_.-' '_')" mkdir -p "${corpus_dir}" cargo bolero test '{{ target }}' --rustc-bootstrap -T '{{ time }}' \ diff --git a/nat/src/icmp_handler/nf.rs b/nat/src/icmp_handler/nf.rs index 7b7576bfe0..a919d0b796 100644 --- a/nat/src/icmp_handler/nf.rs +++ b/nat/src/icmp_handler/nf.rs @@ -142,8 +142,16 @@ impl IcmpErrorHandler { return; } - let flow_info_locked = flow.locked.read(); - let Some(dst_vpcd) = flow_info_locked.dst_vpcd else { + let (dst_vpcd, masquerading, port_forwarding) = { + let flow_info_locked = flow.locked.read(); + ( + flow_info_locked.dst_vpcd, + flow_info_locked.nat_state.is_some(), + flow_info_locked.port_fw_state.is_some(), + ) + }; + + let Some(dst_vpcd) = dst_vpcd else { warn!("Flow for {rev_flow_key} has no dst VPC discriminant set. This is a bug"); packet.done(DoneReason::InternalFailure); return; @@ -153,10 +161,10 @@ impl IcmpErrorHandler { packet.meta_mut().dst_vpcd = Some(dst_vpcd); // process the packet depending on the flow info - let result = if flow_info_locked.nat_state.is_some() { + let result = if masquerading { debug!("Icmp error is for vpc {dst_vpcd}. Will process with masquerade state"); handle_icmp_error_masquerading(packet, flow.as_ref()) - } else if flow_info_locked.port_fw_state.is_some() { + } else if port_forwarding { debug!("Icmp error is for vpc {dst_vpcd}. Will process with port-forwarding state"); handle_icmp_error_port_forwarding(packet, flow.as_ref()) } else { diff --git a/nat/src/masquerade/mod.rs b/nat/src/masquerade/mod.rs index 43e4488e3b..c8790bf223 100644 --- a/nat/src/masquerade/mod.rs +++ b/nat/src/masquerade/mod.rs @@ -22,7 +22,7 @@ mod test; //# REQ-9: A NAT MUST support "Hairpinning". // re exports pub use allocator_writer::MasqueradeConfig; -pub use allocator_writer::NatAllocatorWriter; +pub use allocator_writer::{NatAllocatorReaderFactory, NatAllocatorWriter}; pub use nf::Masquerade; use tracectl::trace_target; diff --git a/nat/src/static_nat/mod.rs b/nat/src/static_nat/mod.rs index a2bf321741..e520b7be92 100644 --- a/nat/src/static_nat/mod.rs +++ b/nat/src/static_nat/mod.rs @@ -11,6 +11,7 @@ pub mod setup; pub(crate) mod test; // re-exports +pub use natrw::NatTablesReaderFactory; pub use nf::{NatTablesWriter, StaticNat}; use tracectl::trace_target; diff --git a/nix/profiles.nix b/nix/profiles.nix index 3e0671a531..c05f346764 100644 --- a/nix/profiles.nix +++ b/nix/profiles.nix @@ -165,8 +165,6 @@ let sanitize.thread.RUSTFLAGS = [ "-Zsanitizer=thread" "-Zexternal-clangrt" - # gimli doesn't like thread sanitizer, but it shouldn't be an issue since that is all build time logic - "-Cunsafe-allow-abi-mismatch=sanitizer" ] ++ (map (flag: "-Clink-arg=${flag}") sanitize.thread.NIX_CFLAGS_LINK); # note: cfi _requires_ LTO and is fundamentally ill suited to debug builds diff --git a/routing/src/lib.rs b/routing/src/lib.rs index 3791867c8a..beb8234abb 100644 --- a/routing/src/lib.rs +++ b/routing/src/lib.rs @@ -27,17 +27,17 @@ mod router; mod routingdb; // re-exports -pub use atable::atablerw::AtableReader; +pub use atable::atablerw::{AtableReader, AtableReaderFactory}; pub use config::RouterConfig; pub use errors::RouterError; pub use evpn::Vtep; pub use fib::fibobjects::{EgressObject, FibEntry, PktInstruction}; -pub use fib::fibtable::FibTableReader; +pub use fib::fibtable::{FibTableReader, FibTableReaderFactory}; pub use fib::fibtype::FibKey; pub use frr::frrmi::FrrAppliedConfig; pub use frr::renderer::builder::Render; pub use interfaces::iftable::IfTable; -pub use interfaces::iftablerw::IfTableReader; +pub use interfaces::iftablerw::{IfTableReader, IfTableReaderFactory}; pub use interfaces::interface::{AttachConfig, Attachment, RouterInterfaceConfig}; pub use interfaces::interface::{IfDataEthernet, IfState, IfType, Interface}; pub use rib::encapsulation::{ diff --git a/routing/src/testing.rs b/routing/src/testing.rs index 9752a0e339..ea59df4bb2 100644 --- a/routing/src/testing.rs +++ b/routing/src/testing.rs @@ -15,11 +15,11 @@ use net::interface::InterfaceIndex; use net::vxlan::Vni; use crate::atable::adjacency::Adjacency; -use crate::atable::atablerw::{AtableReader, AtableWriter}; +use crate::atable::atablerw::{AtableReader, AtableReaderFactory, AtableWriter}; use crate::evpn::Vtep; -use crate::fib::fibtable::{FibTableReader, FibTableWriter}; +use crate::fib::fibtable::{FibTableReader, FibTableReaderFactory, FibTableWriter}; use crate::fib::fibtype::FibKey; -use crate::interfaces::iftablerw::{IfTableReader, IfTableWriter}; +use crate::interfaces::iftablerw::{IfTableReader, IfTableReaderFactory, IfTableWriter}; use crate::interfaces::interface::{IfDataEthernet, IfState, IfType, RouterInterfaceConfig}; use crate::rib::vrf::VrfId; @@ -143,6 +143,21 @@ impl RouterTables { self.adj_reader.clone() } + #[must_use] + pub fn interface_factory(&self) -> IfTableReaderFactory { + self.if_reader.factory() + } + + #[must_use] + pub fn fib_factory(&self) -> FibTableReaderFactory { + self.fib_reader.factory() + } + + #[must_use] + pub fn adjacency_factory(&self) -> AtableReaderFactory { + self.adj_reader.factory() + } + fn fib_mut(&mut self, vrfid: VrfId) -> &mut FibWriter { self.fibs .get_mut(&vrfid) diff --git a/tracectl/Cargo.toml b/tracectl/Cargo.toml index 08e33946b6..48849da4f5 100644 --- a/tracectl/Cargo.toml +++ b/tracectl/Cargo.toml @@ -5,6 +5,9 @@ license.workspace = true publish.workspace = true version.workspace = true +[features] +evidence = [] + [dependencies] clock = { workspace = true } color-eyre = { workspace = true , features = [ "capture-spantrace", "color-spantrace", "tracing-error", "track-caller" ] } diff --git a/tracectl/src/evidence.rs b/tracectl/src/evidence.rs new file mode 100644 index 0000000000..37f67d4dc9 --- /dev/null +++ b/tracectl/src/evidence.rs @@ -0,0 +1,532 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright Open Network Fabric Authors + +#![allow(clippy::disallowed_types)] + +use std::collections::VecDeque; +use std::fmt; +use std::fmt::Write as _; +// nosemgrep: rust-no-direct-std-sync-import +use std::sync::Arc; + +use tracing::field::{Field, Visit}; +use tracing::span::Attributes; +use tracing::subscriber::DefaultGuard; +use tracing::{Event, Id, Level, Metadata, Subscriber}; +use tracing_subscriber::layer::{Context, Layer, SubscriberExt}; +use tracing_subscriber::registry::LookupSpan; + +pub const DEFAULT_DEPTH: usize = 512; + +struct Line { + level: Level, + target: &'static str, + kind: Kind, + scope: String, + message: String, + fields: String, + repeats: usize, +} + +impl Line { + fn same_as(&self, other: &Self) -> bool { + self.level == other.level + && self.target == other.target + && self.kind == other.kind + && self.scope == other.scope + && self.message == other.message + && self.fields == other.fields + } +} + +#[derive(Clone, Copy, PartialEq, Eq)] +enum Kind { + Span, + Event, +} + +impl fmt::Display for Line { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let marker = match self.kind { + Kind::Span => "span", + Kind::Event => " ", + }; + write!(f, "{marker} {:>5} {}", self.level, self.target)?; + if !self.scope.is_empty() { + write!(f, " [{}]", self.scope)?; + } + if !self.message.is_empty() { + write!(f, ": {}", self.message)?; + } + if !self.fields.is_empty() { + write!(f, " {{{}}}", self.fields)?; + } + if self.repeats > 1 { + write!(f, " (x{})", self.repeats)?; + } + Ok(()) + } +} + +struct Log { + lines: VecDeque, + depth: usize, + dropped: usize, +} + +impl Log { + fn push(&mut self, line: Line) { + if let Some(last) = self.lines.back_mut() + && last.same_as(&line) + { + last.repeats += 1; + return; + } + if self.lines.len() == self.depth { + self.lines.pop_front(); + self.dropped += 1; + } + self.lines.push_back(line); + } +} + +#[derive(Default)] +struct Fields { + message: String, + rest: String, +} + +const FIELD_LIMIT: usize = 160; + +fn push_clipped(out: &mut String, value: &dyn fmt::Display) { + let rendered = value.to_string(); + if rendered.len() <= FIELD_LIMIT { + out.push_str(&rendered); + return; + } + let cut = rendered + .char_indices() + .map(|(at, _)| at) + .take_while(|at| *at <= FIELD_LIMIT) + .last() + .unwrap_or(0); + out.push_str(&rendered[..cut]); + let _ = write!(out, "...<{} more bytes>", rendered.len() - cut); +} + +impl Fields { + fn put(&mut self, field: &Field, value: &dyn fmt::Display) { + if field.name() == "message" { + push_clipped(&mut self.message, value); + return; + } + if !self.rest.is_empty() { + self.rest.push(' '); + } + let _ = write!(self.rest, "{}=", field.name()); + push_clipped(&mut self.rest, value); + } +} + +impl Visit for Fields { + fn record_f64(&mut self, field: &Field, value: f64) { + self.put(field, &value); + } + fn record_i64(&mut self, field: &Field, value: i64) { + self.put(field, &value); + } + fn record_u64(&mut self, field: &Field, value: u64) { + self.put(field, &value); + } + fn record_i128(&mut self, field: &Field, value: i128) { + self.put(field, &value); + } + fn record_u128(&mut self, field: &Field, value: u128) { + self.put(field, &value); + } + fn record_bool(&mut self, field: &Field, value: bool) { + self.put(field, &value); + } + fn record_str(&mut self, field: &Field, value: &str) { + self.put(field, &value); + } + fn record_error(&mut self, field: &Field, value: &(dyn std::error::Error + 'static)) { + self.put(field, &value); + } + fn record_debug(&mut self, field: &Field, value: &dyn fmt::Debug) { + self.put(field, &format_args!("{value:?}")); + } +} + +pub const MACHINERY: &[&str] = &["shuttle", "tokio", "runtime", "mio", "hyper"]; + +type Keep = Arc bool + Send + Sync>; + +concurrency::with_std! { + const RECORDS: bool = true; + + #[allow(clippy::unnecessary_wraps)] + fn install(subscriber: S) -> Option { + Some(tracing::subscriber::set_default(subscriber)) + } +} + +concurrency::with_loom! { + const RECORDS: bool = false; + + fn install(_subscriber: S) -> Option { + None + } +} + +concurrency::with_shuttle! { + const RECORDS: bool = false; + + fn install(_subscriber: S) -> Option { + None + } +} + +struct EvidenceLayer { + // nosemgrep: rust-no-direct-std-sync-import + log: Arc>, + keep: Keep, +} + +impl EvidenceLayer { + fn push(&self, line: Line) { + if let Ok(mut log) = self.log.lock() { + log.push(line); + } + } +} + +fn scope_of(ctx: &Context<'_, S>, event: Option<&Event<'_>>, id: Option<&Id>) -> String +where + S: Subscriber + for<'a> LookupSpan<'a>, +{ + let scope = match (event, id) { + (Some(event), _) => ctx.event_scope(event).map(from_root), + (None, Some(id)) => ctx.span_scope(id).map(from_root), + (None, None) => None, + }; + scope.unwrap_or_default() +} + +fn from_root<'a, S>(scope: tracing_subscriber::registry::Scope<'a, S>) -> String +where + S: Subscriber + for<'lookup> LookupSpan<'lookup>, +{ + let mut names: Vec<&str> = scope.map(|span| span.name()).collect(); + names.reverse(); + names.join(" > ") +} + +impl Layer for EvidenceLayer +where + S: Subscriber + for<'a> LookupSpan<'a>, +{ + fn enabled(&self, metadata: &Metadata<'_>, _ctx: Context<'_, S>) -> bool { + (self.keep)(metadata.target()) + } + + fn on_new_span(&self, attrs: &Attributes<'_>, id: &Id, ctx: Context<'_, S>) { + let mut fields = Fields::default(); + attrs.record(&mut fields); + let metadata = attrs.metadata(); + self.push(Line { + level: *metadata.level(), + target: metadata.target(), + kind: Kind::Span, + scope: scope_of(&ctx, None, Some(id)), + message: fields.message, + fields: fields.rest, + repeats: 1, + }); + } + + fn on_event(&self, event: &Event<'_>, ctx: Context<'_, S>) { + let mut fields = Fields::default(); + event.record(&mut fields); + let metadata = event.metadata(); + self.push(Line { + level: *metadata.level(), + target: metadata.target(), + kind: Kind::Event, + scope: scope_of(&ctx, Some(event), None), + message: fields.message, + fields: fields.rest, + repeats: 1, + }); + } +} + +#[must_use = "a recording that is dropped immediately captures nothing"] +pub struct Recording { + // nosemgrep: rust-no-direct-std-sync-import + log: Arc>, + label: String, + _installed: Option, +} + +#[must_use = "a recording that is dropped immediately captures nothing"] +pub fn capture(label: impl Into) -> Recording { + Capture::new(label).start() +} + +pub struct Capture { + label: String, + depth: usize, + keep: Keep, +} + +impl Capture { + #[must_use] + pub fn new(label: impl Into) -> Self { + Self { + label: label.into(), + depth: DEFAULT_DEPTH, + keep: Arc::new(|target: &str| !MACHINERY.iter().any(|noisy| target.starts_with(noisy))), + } + } + + #[must_use] + pub fn depth(mut self, depth: usize) -> Self { + self.depth = depth.max(1); + self + } + + #[must_use] + pub fn keeping(mut self, keep: impl Fn(&str) -> bool + Send + Sync + 'static) -> Self { + self.keep = Arc::new(keep); + self + } + + #[must_use = "a recording that is dropped immediately captures nothing"] + pub fn start(self) -> Recording { + let empty = Log { + lines: VecDeque::new(), + depth: self.depth, + dropped: 0, + }; + // nosemgrep: rust-no-direct-std-sync-import + let log = Arc::new(std::sync::Mutex::new(empty)); + let subscriber = tracing_subscriber::registry().with(EvidenceLayer { + log: log.clone(), + keep: self.keep, + }); + Recording { + log, + label: self.label, + _installed: install(subscriber), + } + } +} + +#[derive(Clone)] +pub struct Evidence { + // nosemgrep: rust-no-direct-std-sync-import + log: Arc>, + label: String, +} + +impl Evidence { + pub fn dump(&self) { + let Ok(log) = self.log.lock() else { + eprintln!("==== trace evidence ({}): buffer poisoned ====", self.label); + return; + }; + if log.lines.is_empty() { + let why = if RECORDS { + "Either the code under test emits no spans or events, or the level filter is above \ + them, or the targets were excluded -- see MACHINERY. Note that \ + `release_max_level_debug` compiles `trace!` out of release builds altogether, and \ + that a recording only sees the thread it was created on." + } else { + "This is a model-checker build, where a recording deliberately installs nothing: \ + loom and shuttle multiplex their tasks onto one OS thread and would corrupt the \ + thread-local the scoped dispatcher lives in. Use the replayable schedule the \ + backend printed instead. See this module's docs." + }; + eprintln!( + "==== trace evidence ({}): nothing captured ====\n{why}", + self.label + ); + return; + } + eprintln!("==== trace evidence ({}) ====", self.label); + if log.dropped > 0 { + eprintln!("[{} earlier lines dropped by the ring]", log.dropped); + } + for line in &log.lines { + eprintln!("{line}"); + } + eprintln!("==== end trace evidence ({}) ====", self.label); + } +} + +#[must_use = "the guard must outlive the assertions it is meant to explain"] +pub fn dump_on_panic(evidence: Vec) -> impl Sized { + struct OnPanic(Vec); + impl Drop for OnPanic { + fn drop(&mut self) { + if std::thread::panicking() { + for evidence in &self.0 { + evidence.dump(); + } + } + } + } + OnPanic(evidence) +} + +impl Recording { + #[must_use] + pub fn evidence(&self) -> Evidence { + Evidence { + log: self.log.clone(), + label: self.label.clone(), + } + } + + pub fn dump(&self) { + self.evidence().dump(); + } + + #[cfg(test)] + fn len(&self) -> usize { + self.log.lock().map_or(0, |log| log.lines.len()) + } + + #[cfg(test)] + fn rendered(&self) -> Vec { + self.log.lock().map_or_else( + |_| Vec::new(), + |log| log.lines.iter().map(ToString::to_string).collect(), + ) + } +} + +impl Drop for Recording { + fn drop(&mut self) { + if std::thread::panicking() { + self.dump(); + } + } +} + +#[cfg(test)] +#[cfg(not(any(feature = "loom", feature = "shuttle")))] +mod tests { + use super::*; + + #[test] + fn a_recording_captures_events_on_its_own_thread() { + let recording = capture("own thread"); + tracing::error!(port = 8080_u64, "a thing happened"); + assert_eq!(recording.len(), 1, "the event was not captured"); + } + + #[test] + fn a_recording_captures_spans_and_the_events_inside_them() { + let recording = capture("spans"); + let span = tracing::error_span!("outer", vni = 100_u64); + let _entered = span.enter(); + tracing::error!("inside"); + assert_eq!(recording.len(), 2, "expected the span and the event"); + } + + #[test] + fn the_ring_drops_the_oldest() { + let recording = Capture::new("bounded").depth(4).start(); + for n in 0..10_u64 { + tracing::error!(n, "line"); + } + assert_eq!(recording.len(), 4, "the ring did not bound itself"); + } + + #[test] + fn a_recording_does_not_capture_another_thread() { + let recording = capture("this thread"); + std::thread::scope(|scope| { + scope.spawn(|| tracing::error!("from elsewhere")); + }); + assert_eq!(recording.len(), 0, "a recording reached across a thread"); + } + + #[test] + fn a_scalar_field_is_read_as_a_scalar() { + let recording = capture("typed"); + tracing::error!(port = 8080_u64, name = "eth0", up = true, "hello"); + let rendered = recording.rendered().join("\n"); + + assert!(rendered.contains("port=8080"), "u64 field: {rendered}"); + assert!( + rendered.contains("name=eth0"), + "a str field went through the `Debug` fallback: {rendered}" + ); + assert!(rendered.contains("up=true"), "bool field: {rendered}"); + assert!( + rendered.contains(": hello"), + "the message was not lifted out of the fields: {rendered}" + ); + } + + #[test] + fn machinery_is_not_evidence() { + let recording = capture("filtered"); + tracing::error!(target: "shuttle::runtime::execution", "scheduling decision"); + tracing::error!(target: "dataplane_nat::masquerade", "allocated"); + assert_eq!(recording.len(), 1, "the machinery target was captured"); + } + + #[test] + fn a_capture_can_be_narrowed_to_one_crate() { + let recording = Capture::new("narrow") + .keeping(|target| target.starts_with("dataplane_nat")) + .start(); + tracing::error!(target: "dataplane_net::packet", "dropping"); + tracing::error!(target: "dataplane_nat::masquerade", "allocated"); + assert_eq!(recording.len(), 1, "the filter did not narrow"); + } + + #[test] + fn a_run_of_identical_lines_collapses() { + let recording = Capture::new("repeats").depth(8).start(); + for _ in 0..4 { + tracing::error!(target: "dataplane_net::buffer", "Dropping TestBuffer"); + } + assert_eq!(recording.len(), 1, "identical lines were not collapsed"); + assert!( + recording.rendered().join("\n").contains("(x4)"), + "the collapsed line does not say how many: {:?}", + recording.rendered() + ); + } + + #[test] + fn only_consecutive_lines_collapse() { + let recording = Capture::new("interleaved").depth(8).start(); + tracing::error!(target: "dataplane_net::buffer", "a"); + tracing::error!(target: "dataplane_net::buffer", "b"); + tracing::error!(target: "dataplane_net::buffer", "a"); + assert_eq!(recording.len(), 3, "non-consecutive lines were merged"); + } + + #[test] + fn an_enormous_field_is_clipped() { + let recording = capture("clipped"); + let huge = "x".repeat(8192); + tracing::error!(target: "dataplane_net::packet", packet = %huge, "in"); + let rendered = recording.rendered().join("\n"); + assert!( + rendered.len() < 512, + "an 8 KiB field was kept whole: {} bytes", + rendered.len() + ); + assert!( + rendered.contains("more bytes>"), + "the clip is not marked, so a reader cannot tell: {rendered}" + ); + } +} diff --git a/tracectl/src/lib.rs b/tracectl/src/lib.rs index 3e6d5bb735..4e18c10b4e 100644 --- a/tracectl/src/lib.rs +++ b/tracectl/src/lib.rs @@ -6,6 +6,9 @@ #![deny(clippy::all, clippy::pedantic)] #![allow(clippy::missing_errors_doc)] +#[cfg(any(test, feature = "evidence"))] +pub mod evidence; + pub mod control; pub mod display; pub mod targets;