lace_sans_io/core/

listing.rs

//! Complete advertisement listings and their changes between rounds.

use std::collections::BTreeMap;
use std::sync::Arc;

use super::{AdvertisementRecord, LogicError};
use crate::records::RecordHash;

/// Hash-ordered listing entries. Entries are shared between the listings and
/// requests that contain them.
pub type ListingEntries = BTreeMap<RecordHash, Arc<AdvertisementRecord>>;

/// One complete listing: the frozen local listing at its sender and the
/// complete peer listing at its receiver, with the canonical byte size `B` of
/// the listing rendered as an `::offers` item: its label line, every
/// entry's lines, and the terminating LF.
///
/// An endpoint keeps one local listing for its logical session and changes it
/// in place each round; a direct receiver shares its map until it freezes its
/// request, so the next change finds the map unshared.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AdvertisementListing {
    pub entries: Arc<ListingEntries>,
    pub(super) canonical_bytes: u64,
}

impl Default for AdvertisementListing {
    fn default() -> Self {
        Self {
            entries: Arc::default(),
            canonical_bytes: crate::iltp::EMPTY_LISTING_BYTES,
        }
    }
}

impl AdvertisementListing {
    /// Size a complete listing from its entries' measured bytes. Fails with
    /// `ResourceLimit` when the block would exceed the ILTP item cap.
    #[cfg(test)]
    pub(super) fn new(entries: ListingEntries) -> Result<Self, LogicError> {
        let listing = Self {
            canonical_bytes: crate::iltp::EMPTY_LISTING_BYTES
                + entries.values().map(|e| e.canonical_bytes()).sum::<u64>(),
            entries: Arc::new(entries),
        };
        listing.check_item_cap()?;
        Ok(listing)
    }

    pub fn canonical_bytes(&self) -> u64 {
        self.canonical_bytes
    }

    /// Add an entry the listing does not hold.
    pub(super) fn insert(&mut self, entry: Arc<AdvertisementRecord>) -> Result<(), LogicError> {
        let hash = entry.record_hash();
        let bytes = entry.canonical_bytes();
        if Arc::make_mut(&mut self.entries)
            .insert(hash, entry)
            .is_some()
        {
            return Err(LogicError::Integration(format!(
                "listing change adds {hash}, which the listing holds"
            )));
        }
        self.canonical_bytes += bytes;
        Ok(())
    }

    /// Remove the entry for `hash`, which the listing holds.
    pub(super) fn withdraw(
        &mut self,
        hash: &RecordHash,
    ) -> Result<Arc<AdvertisementRecord>, LogicError> {
        let entry = Arc::make_mut(&mut self.entries)
            .remove(hash)
            .ok_or_else(|| {
                LogicError::Integration(format!(
                    "listing change withdraws {hash}, which the listing lacks"
                ))
            })?;
        self.canonical_bytes -= entry.canonical_bytes();
        Ok(entry)
    }

    /// The sender checks its complete listing against the item cap every
    /// round, whatever encoding carries it.
    pub(super) fn check_item_cap(&self) -> Result<(), LogicError> {
        if self.canonical_bytes > crate::iltp::ITEM_MAX_BYTES as u64 {
            return Err(LogicError::ResourceLimit(
                "advertisement listing exceeds ILTP item limit".into(),
            ));
        }
        Ok(())
    }
}

#[cfg(test)]
impl FromIterator<AdvertisementRecord> for AdvertisementListing {
    fn from_iter<I: IntoIterator<Item = AdvertisementRecord>>(records: I) -> Self {
        Self::new(
            records
                .into_iter()
                .map(|record| (record.record_hash(), Arc::new(record)))
                .collect(),
        )
        .expect("test listing fits the item cap")
    }
}

/// What a complete listing added and withdrew since the previous complete
/// listing of its logical session.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ListingChange {
    /// Every entry is added: the previous listing was empty, as before a
    /// session's first listing.
    All,
    Delta {
        added: Vec<Arc<AdvertisementRecord>>,
        withdrawn: Vec<Arc<AdvertisementRecord>>,
    },
}

impl Default for ListingChange {
    /// No change.
    fn default() -> Self {
        Self::Delta {
            added: Vec::new(),
            withdrawn: Vec::new(),
        }
    }
}

impl ListingChange {
    /// The change of a listing whose previous listing was `previous`, before
    /// any entry is added or withdrawn.
    pub(super) fn starting_from(previous: &ListingEntries) -> Self {
        if previous.is_empty() {
            Self::All
        } else {
            Self::default()
        }
    }

    pub(super) fn note_added(&mut self, entry: &Arc<AdvertisementRecord>) {
        if let Self::Delta { added, .. } = self {
            added.push(entry.clone());
        }
    }

    pub(super) fn note_withdrawn(&mut self, entry: Arc<AdvertisementRecord>) {
        if let Self::Delta { withdrawn, .. } = self {
            withdrawn.push(entry);
        }
    }

    /// The added entries of `listing`, the complete listing after the change.
    pub(super) fn added<'a>(
        &'a self,
        listing: &'a ListingEntries,
    ) -> impl Iterator<Item = &'a Arc<AdvertisementRecord>> + 'a {
        let (all, delta) = match self {
            Self::All => (Some(listing.values()), None),
            Self::Delta { added, .. } => (None, Some(added.iter())),
        };
        all.into_iter().flatten().chain(delta.into_iter().flatten())
    }

    pub(super) fn withdrawn(&self) -> &[Arc<AdvertisementRecord>] {
        match self {
            Self::All => &[],
            Self::Delta { withdrawn, .. } => withdrawn,
        }
    }

    /// The change from `previous` to `current`, by comparing them.
    #[cfg(test)]
    pub(super) fn between(previous: &ListingEntries, current: &ListingEntries) -> Self {
        Self::Delta {
            added: current
                .iter()
                .filter(|(hash, _)| !previous.contains_key(hash))
                .map(|(_, entry)| entry.clone())
                .collect(),
            withdrawn: previous
                .iter()
                .filter(|(hash, _)| !current.contains_key(hash))
                .map(|(_, entry)| entry.clone())
                .collect(),
        }
    }
}