//! Attribute sets, stored once.
//!
//! A full table has millions of routes and far fewer distinct lists of path
//! attributes: a peer announces prefixes that share their attributes
//! together, about six to an UPDATE on the Internet today. A route in the
//! table is therefore a pointer to an [`AttrSet`], and each distinct set
//! exists once, in the [`AttrStore`].
//!
//! A set is kept as its bytes on the wire, which is a fifth the size of the
//! parsed form. What choosing between routes needs from it is worked out
//! once, when the set is stored, and kept beside the bytes.

use std::hash::BuildHasher;
use std::sync::{Arc, Mutex, MutexGuard, PoisonError};

use bgp_packet::message::{NlriNextHop, PathAttribute, RawAttribute, attr_type, split_attribute};
use bgp_primitives::parser::{EncodeError, ParserContext};
use bytes::{BufMut, BytesMut};
use foldhash::fast::RandomState;
use hashbrown::HashTable;

const OPTIONAL: u8 = 0x80;

/// The type of an AS_PATH segment whose AS numbers are in order; the other
/// type, AS_SET, is an unordered handful (RFC 4271 §4.3).
const AS_SEQUENCE: u8 = 2;

/// What route selection reads from a set of attributes (RFC 4271 §9.1.2.2).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct SelectionKey {
    /// LOCAL_PREF, if the set has one.
    pub local_pref: Option<u32>,
    /// MULTI_EXIT_DISC, if the set has one.
    pub med: Option<u32>,
    /// The AS that the route was learned from: the first AS of the path, or
    /// zero if the path is empty or begins with a set. MEDs are comparable
    /// only between routes with the same one.
    pub neighbor_as: u32,
    /// The length of the AS path as selection counts it: an AS_SET is one
    /// hop however many AS numbers it holds.
    pub as_path_len: u16,
    /// ORIGIN: 0 for IGP, 1 for EGP, 2 for INCOMPLETE.
    pub origin: u8,
    /// The AS the route originates from: the last AS of the path. Zero,
    /// which is no AS's number, if the path is empty or ends in a set and
    /// so has no one origin. Not part of selection; kept here because it
    /// is what a route's validity is asked about (RFC 6811), for every
    /// route, when the validation data changes.
    pub origin_as: u32,
}

/// Reads the selection key straight off attributes in the form an
/// [`AttrSet`] keeps, without building anything. `None` if the bytes are
/// not a well-formed list of attributes.
///
/// A list with the same attribute twice is not well formed (RFC 4271
/// §5). The decoder keeps only the first of such, so none comes from a
/// session; were one to get in some other way, the readers of a set, of
/// which this is one, could each take a different one of the two for the
/// route's path or next hop.
fn scan(mut wire: &[u8]) -> Option<SelectionKey> {
    // One bit for each type of attribute there can be.
    let mut seen = [0u64; 4];
    let mut key = SelectionKey {
        local_pref: None,
        med: None,
        neighbor_as: 0,
        as_path_len: 0,
        // A route without ORIGIN does not get this far; if one did, it
        // would be the least preferred.
        origin: 2,
        origin_as: 0,
    };
    let number = |value: &[u8]| Some(u32::from_be_bytes(<[u8; 4]>::try_from(value).ok()?));
    while !wire.is_empty() {
        let (attribute, rest) = split_attribute(wire).ok()?;
        wire = rest;
        let value = attribute.value;
        let (word, bit) = (
            usize::from(attribute.type_code / 64),
            1u64 << (attribute.type_code % 64),
        );
        if seen[word] & bit != 0 {
            return None;
        }
        seen[word] |= bit;
        match attribute.type_code {
            attr_type::ORIGIN => key.origin = *value.first()?,
            attr_type::AS_PATH => {
                let path = scan_as_path(value)?;
                key.as_path_len = path.hops;
                key.neighbor_as = path.first;
                key.origin_as = path.last;
            }
            attr_type::MULTI_EXIT_DISC => key.med = Some(number(value)?),
            attr_type::LOCAL_PREF => key.local_pref = Some(number(value)?),
            _ => {}
        }
    }
    Some(key)
}

/// What the selection key takes from an AS path.
struct PathScan {
    /// The length as selection counts it.
    hops: u16,
    /// The first AS, if the path begins with a sequence; else zero.
    first: u32,
    /// The last AS, if the path ends in a sequence; else zero.
    last: u32,
}

impl SelectionKey {
    /// The AS the route originates from: the last of its path. `None` if
    /// the path is empty or ends in a set, which has no one last AS.
    pub fn origin(&self) -> Option<u32> {
        Some(self.origin_as).filter(|origin| *origin != 0)
    }
}

/// Reads an AS_PATH value with four-octet AS numbers.
fn scan_as_path(mut segments: &[u8]) -> Option<PathScan> {
    let mut hops = 0usize;
    let (mut first, mut last) = (0, 0);
    let mut at_start = true;
    while let [kind, count, rest @ ..] = segments {
        let (asns, rest) = rest.split_at_checked(usize::from(*count) * 4)?;
        let ordered = *kind == AS_SEQUENCE;
        hops += if ordered { usize::from(*count) } else { 1 };
        let asn = |bytes: Option<&[u8; 4]>| bytes.map_or(0, |asn| u32::from_be_bytes(*asn));
        if at_start && ordered {
            first = asn(asns.first_chunk::<4>());
        }
        last = if ordered {
            asn(asns.last_chunk::<4>())
        } else {
            0
        };
        at_start = false;
        segments = rest;
    }
    if !segments.is_empty() {
        return None;
    }
    Some(PathScan {
        hops: u16::try_from(hops).unwrap_or(u16::MAX),
        first,
        last,
    })
}

/// Appends the AS numbers of an AS_PATH or AS4_PATH value with four-octet
/// AS numbers to `out`, and returns whether any segment is not a sequence.
/// Stops at what is not a segment.
fn read_as_path(mut segments: &[u8], out: &mut Vec<u32>) -> bool {
    let mut has_set = false;
    while let [kind, count, rest @ ..] = segments {
        let Some((asns, rest)) = rest.split_at_checked(usize::from(*count) * 4) else {
            break;
        };
        has_set |= *kind != AS_SEQUENCE;
        let (asns, _) = asns.as_chunks::<4>();
        out.extend(asns.iter().map(|asn| u32::from_be_bytes(*asn)));
        segments = rest;
    }
    has_set
}

/// One distinct list of path attributes, with the next hop that goes with
/// it. Made by an [`AttrStore`]; two routes have the same attributes
/// exactly when they point to the same `AttrSet`.
#[derive(Debug)]
pub struct AttrSet {
    hash: u64,
    key: SelectionKey,
    /// The attributes as on the wire, with four-octet AS numbers. A next
    /// hop that arrived in MP_REACH_NLRI is kept as an MP_REACH_NLRI that
    /// holds nothing else, the form RFC 6396 §4.3.4 uses in table dumps.
    wire: Box<[u8]>,
}

impl AttrSet {
    pub fn key(&self) -> &SelectionKey {
        &self.key
    }

    /// The attributes as they are on the wire.
    pub fn wire(&self) -> &[u8] {
        &self.wire
    }

    /// The attributes one by one, as they lie in the stored bytes. The
    /// bytes were checked when the set was made, so this is all of them;
    /// it stops at the first that does not split as an attribute all the
    /// same, rather than go on from the middle of one.
    fn raw_attributes(&self) -> impl Iterator<Item = RawAttribute<'_>> {
        let mut wire = &*self.wire;
        std::iter::from_fn(move || {
            let (attribute, rest) = split_attribute(wire).ok()?;
            wire = rest;
            Some(attribute)
        })
    }

    /// The value of the attribute of type `type_code`, if the set has one.
    fn attribute(&self, type_code: u8) -> Option<&[u8]> {
        self.raw_attributes()
            .find(|attribute| attribute.type_code == type_code)
            .map(|attribute| attribute.value)
    }

    /// The route's next hop: the one that came in MP_REACH_NLRI, or else
    /// the NEXT_HOP attribute. Read off the stored bytes without building
    /// anything else, for those who ask it of every route.
    pub fn next_hop(&self) -> Option<NlriNextHop> {
        let mut plain = None;
        for attribute in self.raw_attributes() {
            match attribute.type_code {
                attr_type::MP_REACH_NLRI => return decode_next_hop(attribute.value),
                attr_type::NEXT_HOP => plain = NlriNextHop::from_bytes(attribute.value),
                _ => {}
            }
        }
        plain
    }

    /// Whether the route carries the OTC attribute (RFC 9234): it came
    /// down from a provider or across from a peer, and is to go on down
    /// only. Read off the stored bytes without building anything else,
    /// for those who ask it of every route: a session to a peer or a
    /// provider asks it of most of the table, to pass over them.
    pub fn is_only_to_customer(&self) -> bool {
        self.attribute(attr_type::ONLY_TO_CUSTOMER).is_some()
    }

    /// Whether the route carries the BLACKHOLE community (RFC 7999), which
    /// asks for the packets to its prefix to be dropped. Read off the
    /// stored bytes without building anything else, for those who ask it
    /// of every route.
    pub fn is_blackhole(&self) -> bool {
        const BLACKHOLE: [u8; 4] = [0xFF, 0xFF, 0x02, 0x9A];
        self.attribute(attr_type::COMMUNITIES)
            .is_some_and(|communities| communities.as_chunks::<4>().0.contains(&BLACKHOLE))
    }

    /// Puts the AS numbers of the route's path in `out`, in place of what
    /// was there: in the order of the path, the neighbor's first and the
    /// origin's last, sets and all. Returns whether the path has a set in
    /// it.
    ///
    /// This is the whole path: what a session with two-octet AS numbers
    /// brought in AS4_PATH was put in its place when the UPDATE was read
    /// (RFC 6793 §4.2.3). It reads the stored bytes and builds nothing
    /// else, for those who ask it of every route.
    pub fn as_path(&self, out: &mut Vec<u32>) -> bool {
        out.clear();
        match self.attribute(attr_type::AS_PATH) {
            Some(segments) => read_as_path(segments, out),
            None => false,
        }
    }

    /// The attributes, parsed, and the next hop if it came in
    /// MP_REACH_NLRI.
    ///
    /// `None` if they do not parse. Sets made from parsed attributes or
    /// from an UPDATE the decoder accepted always do; one made from bytes
    /// of another source ([`AttrStore::intern_wire`]) was checked only as
    /// far as selection reads it.
    pub fn attributes(&self) -> Option<(Vec<PathAttribute>, Option<NlriNextHop>)> {
        parse(&self.wire)
    }
}

/// How the attributes of a set are written: with four-octet AS numbers.
fn four_octet() -> ParserContext {
    ParserContext {
        four_octet_asn: true,
        ibgp: true,
        add_path: Vec::new(),
    }
}

/// Writes `attributes` and `next_hop` in the form an [`AttrSet`] keeps.
fn encode(
    attributes: &[PathAttribute],
    next_hop: Option<&NlriNextHop>,
    out: &mut BytesMut,
) -> Result<(), EncodeError> {
    let ctx = four_octet();
    for attribute in attributes {
        // These name prefixes, which are not a property of the route.
        if matches!(
            attribute,
            PathAttribute::MpReachNlri(_) | PathAttribute::MpUnreachNlri(_)
        ) {
            continue;
        }
        attribute.to_wire(&ctx, out)?;
    }
    if let Some(next_hop) = next_hop {
        encode_next_hop(next_hop, out);
    }
    Ok(())
}

/// Writes a next hop as an MP_REACH_NLRI that holds nothing else: the
/// next hop's length, and the next hop.
fn encode_next_hop(next_hop: &NlriNextHop, out: &mut BytesMut) {
    out.put_u8(OPTIONAL);
    out.put_u8(attr_type::MP_REACH_NLRI);
    out.put_u8(1 + next_hop.wire_len());
    out.put_u8(next_hop.wire_len());
    next_hop.to_wire(out);
}

/// Reads the next hop out of the value of an MP_REACH_NLRI as
/// [`encode_next_hop`] writes it. `None` if the value is not one.
fn decode_next_hop(value: &[u8]) -> Option<NlriNextHop> {
    let (length, address) = value.split_first()?;
    NlriNextHop::from_bytes(address).filter(|_| usize::from(*length) == address.len())
}

/// Parses path attributes in the form [`AttrSet`] keeps and table dumps
/// use (RFC 6396 §4.3.4): four-octet AS numbers, and an MP_REACH_NLRI that
/// holds only a next hop. `None` if they are malformed.
fn parse(mut wire: &[u8]) -> Option<(Vec<PathAttribute>, Option<NlriNextHop>)> {
    let ctx = four_octet();
    let mut attributes = Vec::new();
    let mut next_hop = None;
    while !wire.is_empty() {
        let (raw, rest) = split_attribute(wire).ok()?;
        if raw.type_code == attr_type::MP_REACH_NLRI {
            next_hop = Some(decode_next_hop(raw.value)?);
        } else {
            let (_, attribute) = PathAttribute::from_wire(&ctx, raw.wire).ok()?;
            attributes.push(attribute);
        }
        wire = rest;
    }
    Some((attributes, next_hop))
}

/// How many locks the store is split over. Sessions intern concurrently,
/// once per UPDATE; this many locks keep them out of each other's way.
const STRIPES: usize = 64;

/// The one place attribute sets are kept. Safe to share between threads.
pub struct AttrStore {
    stripes: Box<[Mutex<HashTable<Arc<AttrSet>>>]>,
    hasher: RandomState,
}

impl Default for AttrStore {
    fn default() -> Self {
        Self::new()
    }
}

impl AttrStore {
    pub fn new() -> Self {
        Self {
            stripes: (0..STRIPES).map(|_| Mutex::default()).collect(),
            hasher: RandomState::default(),
        }
    }

    /// The stored set equal to `attributes` and `next_hop`, added if it was
    /// not there. `next_hop` is the one from MP_REACH_NLRI; routes whose
    /// next hop is the NEXT_HOP attribute pass `None`.
    ///
    /// Fails only if the attributes cannot be encoded, which attributes that
    /// were decoded from a message always can.
    pub fn intern(
        &self,
        attributes: &[PathAttribute],
        next_hop: Option<&NlriNextHop>,
    ) -> Result<Arc<AttrSet>, EncodeError> {
        let mut wire = BytesMut::with_capacity(256);
        encode(attributes, next_hop, &mut wire)?;
        Ok(self
            .intern_wire(&wire)
            .expect("attributes written here are well formed"))
    }

    /// As [`AttrStore::intern`], for attributes that are already in the
    /// form a set keeps: on the wire, with four-octet AS numbers, no
    /// MP_UNREACH_NLRI, and an MP_REACH_NLRI (if any) that holds only the
    /// next hop. That is how a table dump has them (RFC 6396 §4.3.4).
    ///
    /// Nothing is parsed: finding a set that is already stored costs a hash
    /// of the bytes and a comparison. `None` if the bytes are not a
    /// well-formed list of attributes.
    pub fn intern_wire(&self, wire: &[u8]) -> Option<Arc<AttrSet>> {
        let hash = self.hasher.hash_one(wire);
        let mut stripe = self.stripe(hash);
        if let Some(existing) = stripe.find(hash, |set| *set.wire == *wire) {
            return Some(Arc::clone(existing));
        }
        let set = Arc::new(AttrSet {
            hash,
            key: scan(wire)?,
            wire: wire.into(),
        });
        stripe.insert_unique(hash, Arc::clone(&set), |set| set.hash);
        Some(set)
    }

    /// The set for attributes as a session received them: `wire` is what
    /// the decoder kept of the UPDATE's attributes
    /// (`DecodedUpdate::attributes_wire`), and `next_hop` the next hop of
    /// the routes if they came in MP_REACH_NLRI.
    pub fn intern_received(
        &self,
        wire: &[u8],
        next_hop: Option<&NlriNextHop>,
    ) -> Option<Arc<AttrSet>> {
        self.intern_received_with(wire, &[], next_hop)
    }

    /// As [`AttrStore::intern_received`], with attributes of our own after
    /// the peer's: `added` is one or more whole attributes in the same
    /// form, of types that `wire` does not have.
    pub fn intern_received_with(
        &self,
        wire: &[u8],
        added: &[u8],
        next_hop: Option<&NlriNextHop>,
    ) -> Option<Arc<AttrSet>> {
        if added.is_empty() && next_hop.is_none() {
            return self.intern_wire(wire);
        }
        let mut whole = BytesMut::with_capacity(wire.len() + added.len() + 36);
        whole.put_slice(wire);
        whole.put_slice(added);
        if let Some(next_hop) = next_hop {
            encode_next_hop(next_hop, &mut whole);
        }
        self.intern_wire(&whole)
    }

    /// How many distinct sets are stored.
    pub fn len(&self) -> usize {
        (0..STRIPES).map(|i| self.stripe_at(i).len()).sum()
    }

    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }

    /// Drops the sets that no route refers to any more, and returns how
    /// many that was. Sets are not dropped as their last route goes, since
    /// that would mean a lock here for every route removed; this is called
    /// now and then instead.
    pub fn sweep(&self) -> usize {
        let mut dropped = 0;
        for i in 0..STRIPES {
            let mut stripe = self.stripe_at(i);
            let before = stripe.len();
            // The store's own reference is the only one left. Nobody can
            // be taking another: that is done under this lock.
            stripe.retain(|set| Arc::strong_count(set) > 1);
            dropped += before - stripe.len();
        }
        dropped
    }

    fn stripe(&self, hash: u64) -> MutexGuard<'_, HashTable<Arc<AttrSet>>> {
        self.stripe_at(super::spread(hash, STRIPES))
    }

    fn stripe_at(&self, index: usize) -> MutexGuard<'_, HashTable<Arc<AttrSet>>> {
        // A panic while holding the lock cannot leave the table half
        // changed in a way that matters: see `IgnorePoison` in the server.
        self.stripes[index]
            .lock()
            .unwrap_or_else(PoisonError::into_inner)
    }
}

#[cfg(test)]
mod tests {
    use std::net::Ipv4Addr;

    use bgp_packet::message::{
        AsPathAttribute, AsPathSegment, CommunitiesPathAttribute, LocalPrefPathAttribute,
        MultiExitDiscPathAttribute, NextHopPathAttribute, OriginPathAttribute,
    };

    use super::*;

    fn attributes(path: Vec<u32>) -> Vec<PathAttribute> {
        vec![
            PathAttribute::Origin(OriginPathAttribute::EGP),
            AsPathAttribute::from_asns(path),
            PathAttribute::NextHop(NextHopPathAttribute(Ipv4Addr::new(192, 0, 2, 1))),
            PathAttribute::MultiExitDisc(MultiExitDiscPathAttribute(50)),
            PathAttribute::Communities(CommunitiesPathAttribute(vec![(65000, 1)])),
        ]
    }

    #[test]
    fn equal_attributes_are_one_set() {
        let store = AttrStore::new();
        let a = store.intern(&attributes(vec![65001, 65002]), None).unwrap();
        let b = store.intern(&attributes(vec![65001, 65002]), None).unwrap();
        let c = store.intern(&attributes(vec![65001, 65003]), None).unwrap();
        assert!(Arc::ptr_eq(&a, &b));
        assert!(!Arc::ptr_eq(&a, &c));
        assert_eq!(store.len(), 2);

        // The next hop is part of what a set is.
        let next_hop = NlriNextHop::Ipv6("2001:db8::1".parse().unwrap());
        let d = store
            .intern(&attributes(vec![65001, 65002]), Some(&next_hop))
            .unwrap();
        assert!(!Arc::ptr_eq(&a, &d));
        assert_eq!(store.len(), 3);
    }

    #[test]
    fn a_set_gives_back_what_went_in() {
        let store = AttrStore::new();
        let next_hop = NlriNextHop::Ipv6WithLl {
            global: "2001:db8::1".parse().unwrap(),
            link_local: "fe80::1".parse().unwrap(),
        };
        let original = attributes(vec![65001, 4_200_000_001]);
        let set = store.intern(&original, Some(&next_hop)).unwrap();
        assert_eq!(set.attributes(), Some((original, Some(next_hop))));
    }

    #[test]
    fn the_selection_key_is_read_off_the_attributes() {
        let store = AttrStore::new();
        let mut attrs = vec![
            PathAttribute::Origin(OriginPathAttribute::INCOMPLETE),
            PathAttribute::AsPath(AsPathAttribute {
                segments: vec![
                    AsPathSegment {
                        ordered: true,
                        path: vec![65001, 65002],
                    },
                    // A set counts as one hop.
                    AsPathSegment {
                        ordered: false,
                        path: vec![65003, 65004, 65005],
                    },
                ],
            }),
            PathAttribute::LocalPref(LocalPrefPathAttribute(200)),
        ];
        let key = *store.intern(&attrs, None).unwrap().key();
        assert_eq!(
            key,
            SelectionKey {
                local_pref: Some(200),
                med: None,
                neighbor_as: 65001,
                as_path_len: 3,
                origin: 2,
                // The path ends in a set: it has no one origin.
                origin_as: 0,
            }
        );

        attrs.push(PathAttribute::MultiExitDisc(MultiExitDiscPathAttribute(7)));
        assert_eq!(store.intern(&attrs, None).unwrap().key().med, Some(7));
    }

    #[test]
    fn the_next_hop_is_read_off_a_set() {
        let store = AttrStore::new();
        let plain = store.intern(&attributes(vec![3333]), None).unwrap();
        assert_eq!(
            plain.next_hop(),
            Some(NlriNextHop::Ipv4(Ipv4Addr::new(192, 0, 2, 1)))
        );
        // The one that came with the routes counts, where there is one.
        let with_link_local = NlriNextHop::Ipv6WithLl {
            global: "2001:db8::1".parse().unwrap(),
            link_local: "fe80::1".parse().unwrap(),
        };
        let v6 = store
            .intern(&attributes(vec![3333]), Some(&with_link_local))
            .unwrap();
        assert_eq!(v6.next_hop(), Some(with_link_local));
        // A route of our own making may have none.
        let none = store
            .intern(&[PathAttribute::Origin(OriginPathAttribute::IGP)], None)
            .unwrap();
        assert_eq!(none.next_hop(), None);
    }

    #[test]
    fn the_path_is_read_off_a_set_as_it_really_is() {
        use bgp_packet::message::{AsPathAttribute, AsPathSegment};
        let store = AttrStore::new();
        let mut path = Vec::new();

        let set = store
            .intern(&attributes(vec![3333, 3333, 1299]), None)
            .unwrap();
        assert!(!set.as_path(&mut path));
        assert_eq!(path, [3333, 3333, 1299]);

        // A set is said to be there.
        let sequence = |path: Vec<u32>| AsPathSegment {
            ordered: true,
            path,
        };
        let mut with_set = attributes(vec![]);
        with_set[1] = PathAttribute::AsPath(AsPathAttribute {
            segments: vec![
                sequence(vec![3333]),
                AsPathSegment {
                    ordered: false,
                    path: vec![64500, 64501],
                },
            ],
        });
        let set = store.intern(&with_set, None).unwrap();
        assert!(set.as_path(&mut path));
        assert_eq!(path, [3333, 64500, 64501]);

        // No path at all.
        let set = store.intern(&attributes(vec![]), None).unwrap();
        assert!(!set.as_path(&mut path));
        assert!(path.is_empty());
    }

    /// The BLACKHOLE community is read off the stored bytes.
    #[test]
    fn a_blackhole_request_is_told_from_the_bytes() {
        let store = AttrStore::new();
        let with = |communities: Vec<(u16, u16)>| {
            let mut attributes = attributes(vec![3333]);
            attributes.retain(|attribute| !matches!(attribute, PathAttribute::Communities(_)));
            attributes.push(PathAttribute::Communities(CommunitiesPathAttribute(
                communities,
            )));
            store.intern(&attributes, None).unwrap()
        };
        assert!(with(vec![(3333, 100), (0xFFFF, 0x029A)]).is_blackhole());
        assert!(!with(vec![(3333, 666), (0xFFFF, 0xFF01)]).is_blackhole());
        assert!(
            !store
                .intern(&attributes(vec![3333]), None)
                .unwrap()
                .is_blackhole()
        );
    }

    /// The OTC attribute is read off the stored bytes, wherever in them it
    /// lies.
    #[test]
    fn otc_is_told_from_the_bytes() {
        let store = AttrStore::new();
        let without = store.intern(&attributes(vec![3333]), None).unwrap();
        assert!(!without.is_only_to_customer());

        let mut with = attributes(vec![3333]);
        with.insert(1, PathAttribute::OnlyToCustomer(3333));
        assert!(store.intern(&with, None).unwrap().is_only_to_customer());

        // After COMMUNITIES, which is the last of the usual attributes.
        let mut after = attributes(vec![3333]);
        after.push(PathAttribute::OnlyToCustomer(3333));
        assert!(store.intern(&after, None).unwrap().is_only_to_customer());
    }

    /// What a session received, with attributes of our own added after it,
    /// is one set with the parsed form of the whole; the bytes are not
    /// parsed to find that out, so the bytes must say the same thing.
    #[test]
    fn received_attributes_with_ours_added_are_the_same_set_as_the_parsed_form() {
        let store = AttrStore::new();
        let ctx = four_octet();
        let received = attributes(vec![65001, 65002]);
        let mut wire = BytesMut::new();
        encode(&received, None, &mut wire).unwrap();
        let ours = [
            PathAttribute::OnlyToCustomer(65001),
            PathAttribute::LocalPref(LocalPrefPathAttribute(150)),
        ];
        let mut added = BytesMut::new();
        for attribute in &ours {
            attribute.to_wire(&ctx, &mut added).unwrap();
        }
        let next_hop = NlriNextHop::Ipv6("2001:db8::1".parse().unwrap());

        let mut whole = received.clone();
        whole.extend(ours.iter().cloned());
        let parsed = store.intern(&whole, Some(&next_hop)).unwrap();
        let from_bytes = store
            .intern_received_with(&wire, &added, Some(&next_hop))
            .unwrap();
        assert!(Arc::ptr_eq(&parsed, &from_bytes));
        assert!(from_bytes.is_only_to_customer());
        assert_eq!(from_bytes.next_hop(), Some(next_hop));

        // Without a next hop too.
        let parsed = store.intern(&whole, None).unwrap();
        let from_bytes = store.intern_received_with(&wire, &added, None).unwrap();
        assert!(Arc::ptr_eq(&parsed, &from_bytes));

        // An attribute of ours of a type the peer sent is not a set.
        let mut repeated = BytesMut::new();
        PathAttribute::MultiExitDisc(MultiExitDiscPathAttribute(1))
            .to_wire(&ctx, &mut repeated)
            .unwrap();
        assert!(store.intern_received_with(&wire, &repeated, None).is_none());
    }

    /// Two of the same attribute are not a set: its readers could each
    /// take a different one.
    #[test]
    fn the_same_attribute_twice_is_not_a_set() {
        let store = AttrStore::new();
        let origin = [0x40, 1, 1, 0];
        let path = |asn: u8| [0x40, 2, 6, 2, 1, 0, 0, 0, asn];
        let once = [&origin[..], &path(1)].concat();
        assert!(store.intern_wire(&once).is_some());
        let twice = [&origin[..], &path(1), &path(2)].concat();
        assert!(store.intern_wire(&twice).is_none());
        // Whatever the type, known here or not.
        let unknown = [0xc0, 200, 1, 7];
        assert!(store.intern_wire(&[&once[..], &unknown].concat()).is_some());
        assert!(
            store
                .intern_wire(&[&once[..], &unknown, &unknown].concat())
                .is_none()
        );
    }

    #[test]
    fn attributes_on_the_wire_are_interned_without_parsing_them() {
        let store = AttrStore::new();
        let next_hop = NlriNextHop::Ipv6("2001:db8::1".parse().unwrap());
        let mut attrs = attributes(vec![65001, 65002, 65003]);
        attrs.push(PathAttribute::LocalPref(LocalPrefPathAttribute(300)));
        let parsed = store.intern(&attrs, Some(&next_hop)).unwrap();

        // The same set by its bytes is the same set, key and all.
        let by_wire = store.intern_wire(parsed.wire()).unwrap();
        assert!(Arc::ptr_eq(&parsed, &by_wire));

        // And one that arrives as bytes first gets the same key as it would
        // have had from the parsed form.
        let other = AttrStore::new();
        let fresh = other.intern_wire(parsed.wire()).unwrap();
        assert_eq!(fresh.key(), parsed.key());
        assert_eq!(fresh.key().as_path_len, 3);
        assert_eq!(fresh.key().neighbor_as, 65001);
        assert_eq!(fresh.key().origin_as, 65003);
        assert_eq!(fresh.attributes(), Some((attrs, Some(next_hop))));

        // Bytes that are not attributes are refused.
        assert!(other.intern_wire(&[0x40, 0x01, 0x05, 0x00]).is_none());
        assert!(
            other
                .intern_wire(&[0x40, 0x02, 0x03, 0x02, 0x01, 0x00])
                .is_none()
        );
    }

    #[test]
    fn sweeping_drops_only_sets_nobody_holds() {
        let store = AttrStore::new();
        let kept = store.intern(&attributes(vec![65001]), None).unwrap();
        drop(store.intern(&attributes(vec![65002]), None).unwrap());
        assert_eq!(store.len(), 2);

        assert_eq!(store.sweep(), 1);
        assert_eq!(store.len(), 1);
        // The one that is left is still the one that is found.
        let again = store.intern(&attributes(vec![65001]), None).unwrap();
        assert!(Arc::ptr_eq(&kept, &again));
    }
}