ddidderr/lanspread
1
0
Fork 0
Code Issues 6 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(peer): preserve manifest entry shape consensus

Browse Source 
Treat directory/file shape as part of each peer manifest vote and reject
portable aliases before aggregating descriptors. This keeps majority selection
deterministic and avoids collapsing conflicting entries that share a path or
size.

This preserves the previously staged consensus hardening before the protocol-8
catalog-authority cutover layered in the working tree.

Test Plan:
- `git diff --cached --check` -- passed
This commit is contained in:
ddidderr
2026-08-10 13:58:37 +02:00
parent 290af433c7
commit 36c4785775
1 changed files with 258 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/lanspread-peer/src/peer_db.rs
+258 -91
View File
@@ -2,7 +2,7 @@
use std::{
cmp::Reverse,
collections::HashMap,
collections::{BTreeMap, HashMap},
net::SocketAddr,
time::{Duration, Instant},
};
@@ -10,7 +10,7 @@ use std::{
use lanspread_db::db::{Availability, Game, GameCatalog, GameFileDescription};
use lanspread_proto::{GameSummary, LibraryDelta, LibrarySnapshot};
use crate::library::compute_library_digest;
use crate::{game_paths::portable_name_key, library::compute_library_digest};
pub type PeerId = String;
/// Information about a discovered peer.
@@ -639,7 +639,7 @@ impl PeerGameDB {
.map(|(size, _)| size)
}
/// Validates file sizes across all peers and returns only the files with majority consensus.
/// Validates entry shapes and file sizes across all peers.
///
/// Returns a tuple of (`validated_files`, `peer_whitelist`, `file_peer_map`) where
/// `peer_whitelist` contains peers that have at least one majority-approved file and
@@ -653,7 +653,7 @@ impl PeerGameDB {
self.validate_file_sizes_majority_from(game_id, &game_files)
}
/// Validates file-size consensus over caller-sanitized complete peer manifests.
/// Validates entry consensus over caller-sanitized complete peer manifests.
pub(crate) fn validate_file_sizes_majority_from(
&self,
game_id: &str,
@@ -663,125 +663,112 @@ impl PeerGameDB {
return Ok((Vec::new(), Vec::new(), HashMap::new()));
}
let (file_size_map, _peer_files) = collect_file_sizes(game_files);
let entry_consensus_map = collect_manifest_entries(game_files)?;
let (validated_files, peer_scores, file_peer_map) =
self.validate_each_file_consensus(game_id, file_size_map)?;
self.validate_each_entry_consensus(game_id, entry_consensus_map)?;
let peer_whitelist = create_peer_whitelist(peer_scores);
Ok((validated_files, peer_whitelist, file_peer_map))
}
/// Validates consensus for each file and returns validated files with peer scores.
fn validate_each_file_consensus(
/// Validates consensus for each entry and returns validated descriptions with peer scores.
fn validate_each_entry_consensus(
&self,
game_id: &str,
file_size_map: FileSizeMap,
entry_consensus_map: EntryConsensusMap,
) -> eyre::Result<FileConsensusAggregation> {
let mut validated_files = Vec::new();
let mut peer_whitelist_scores: HashMap<SocketAddr, usize> = HashMap::new();
let mut file_peer_map: HashMap<String, Vec<SocketAddr>> = HashMap::new();
for (relative_path, size_map) in file_size_map {
let total_peers: usize = size_map.values().map(Vec::len).sum();
for (relative_path, descriptor_map) in entry_consensus_map {
let total_peers: usize = descriptor_map.values().map(Vec::len).sum();
if total_peers == 0 {
continue; // Skip files with no size information
continue;
}
let (consensus_size, consensus_peers) =
self.determine_size_consensus(&size_map, total_peers, &relative_path)?;
let (consensus_descriptor, consensus_peers) =
self.determine_entry_consensus(&descriptor_map, total_peers, &relative_path)?;
update_peer_scores(&consensus_peers, &mut peer_whitelist_scores);
if let Some((size, peers)) = consensus_size
&& let Some(file_desc) =
self.create_validated_file_description(game_id, &relative_path, size, &peers)
{
if let Some((descriptor, peers)) = consensus_descriptor {
file_peer_map.insert(relative_path.clone(), peers.clone());
validated_files.push(file_desc);
validated_files.push(GameFileDescription {
game_id: game_id.to_owned(),
relative_path,
is_dir: descriptor.is_dir,
size: descriptor.size,
});
}
}
Ok((validated_files, peer_whitelist_scores, file_peer_map))
}
/// Determines the consensus size for a file based on peer reports.
/// Determines the consensus shape and size for an entry based on peer reports.
///
/// # Panics
///
/// Panics if `size_map.iter().next()` returns None when `total_peers` == 1
/// Panics if `descriptor_map.iter().next()` returns None when `total_peers` == 1
#[allow(clippy::unused_self)]
fn determine_size_consensus(
fn determine_entry_consensus(
&self,
size_map: &HashMap<u64, Vec<SocketAddr>>,
descriptor_map: &BTreeMap<EntryDescriptor, Vec<SocketAddr>>,
total_peers: usize,
relative_path: &str,
) -> eyre::Result<(ConsensusResult, Vec<SocketAddr>)> {
if total_peers == 1 {
// Only one peer has this file - trust it
let (&size, peers) = size_map
// Only one peer has this entry - trust it.
let (&descriptor, peers) = descriptor_map
.iter()
.next()
.expect("size_map should have at least one entry when total_peers == 1");
return Ok((Some((size, peers.clone())), peers.clone()));
.expect("descriptor_map should have an entry when total_peers == 1");
return Ok((Some((descriptor, peers.clone())), peers.clone()));
}
let (majority_size, _majority_count) = find_majority_size(size_map);
let (majority_descriptor, _majority_count) = find_majority_descriptor(descriptor_map);
if let Some(size) = majority_size {
let majority_peers = &size_map[&size];
if let Some(descriptor) = majority_descriptor {
let majority_peers = &descriptor_map[&descriptor];
let is_majority = majority_peers.len() > total_peers / 2;
if is_majority {
// We have a clear majority
Ok((Some((size, majority_peers.clone())), majority_peers.clone()))
Ok((
Some((descriptor, majority_peers.clone())),
majority_peers.clone(),
))
} else if total_peers == 2 {
// Two peers with different sizes - ambiguous, fail
// Two peers with different descriptions - ambiguous, fail.
eyre::bail!(
"File size ambiguity for '{}': two peers report different sizes, cannot determine majority",
"Manifest entry ambiguity for '{}': two peers report different shapes or sizes, cannot determine majority",
relative_path
);
}
// If no majority and more than 2 peers, we fall back to plurality (largest group)
// If no majority and more than 2 peers, fall back to the unique plurality.
else {
Ok((Some((size, majority_peers.clone())), majority_peers.clone()))
Ok((
Some((descriptor, majority_peers.clone())),
majority_peers.clone(),
))
}
} else {
// No clear majority and it's a tie between different sizes
// No clear majority and the largest groups are tied.
if total_peers == 2 {
eyre::bail!(
"File size ambiguity for '{}': two peers report different sizes, cannot determine majority",
"Manifest entry ambiguity for '{}': two peers report different shapes or sizes, cannot determine majority",
relative_path
);
}
// For more than 2 peers, we could fall back to plurality, but for now let's be strict
eyre::bail!(
"File size ambiguity for '{}': no clear majority among {} peers",
"Manifest entry ambiguity for '{}': no unique plurality among {} peers",
relative_path,
total_peers
);
}
}
/// Creates a validated file description from consensus data.
fn create_validated_file_description(
&self,
game_id: &str,
relative_path: &str,
size: u64,
peers: &[SocketAddr],
) -> Option<GameFileDescription> {
if let Some(first_peer) = peers.first()
&& let Some(peer_id) = self.addr_index.get(first_peer)
&& let Some(files) = self.peers.get(peer_id).and_then(|p| p.files.get(game_id))
&& let Some(file_desc) = files
.iter()
.find(|f| f.relative_path == relative_path && f.size == size)
{
return Some(file_desc.clone());
}
None
}
/// Returns peers that haven't been seen within the timeout duration.
#[must_use]
pub fn get_stale_peers(&self, timeout: Duration) -> Vec<SocketAddr> {
@@ -807,14 +794,27 @@ impl PeerGameDB {
// Type aliases for consensus validation
// =============================================================================
/// Type alias for file size mapping: path -> size -> peers
type FileSizeMap = HashMap<String, HashMap<u64, Vec<SocketAddr>>>;
/// The complete description that peers vote on for one exact protocol path.
#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)]
struct EntryDescriptor {
is_dir: bool,
size: u64,
}
/// Type alias for peer file mapping: peer -> path -> size
type PeerFileMap = HashMap<SocketAddr, HashMap<String, u64>>;
impl From<&GameFileDescription> for EntryDescriptor {
fn from(description: &GameFileDescription) -> Self {
Self {
is_dir: description.is_dir,
size: description.size,
}
}
}
/// Type alias for consensus result: (size, peers) or None
type ConsensusResult = Option<(u64, Vec<SocketAddr>)>;
/// Type alias for entry consensus mapping: path -> descriptor -> peers.
type EntryConsensusMap = BTreeMap<String, BTreeMap<EntryDescriptor, Vec<SocketAddr>>>;
/// Type alias for consensus result: (descriptor, peers) or None.
type ConsensusResult = Option<(EntryDescriptor, Vec<SocketAddr>)>;
/// Type alias for the aggregated majority validation result.
pub type MajorityValidationResult = (
@@ -834,51 +834,89 @@ type FileConsensusAggregation = (
// Helper functions for consensus validation
// =============================================================================
/// Collects file sizes from all peers and organizes them by path and size.
fn collect_file_sizes(
/// Collects exact entry descriptions from all peers.
///
/// A portable alias cannot be merged here: the current transfer map carries one
/// protocol spelling per entry and therefore cannot translate that spelling for
/// a peer that advertised an alias-equivalent path. Reject the aggregate rather
/// than constructing a manifest that is unsafe on case-insensitive filesystems.
fn collect_manifest_entries(
game_files: &[(SocketAddr, Vec<GameFileDescription>)],
) -> (FileSizeMap, PeerFileMap) {
let mut file_size_map: FileSizeMap = HashMap::new();
let mut peer_files: PeerFileMap = HashMap::new();
) -> eyre::Result<EntryConsensusMap> {
let mut peer_entries: BTreeMap<SocketAddr, BTreeMap<String, EntryDescriptor>> = BTreeMap::new();
for (peer_addr, files) in game_files {
let mut peer_file_sizes = HashMap::new();
let entries = peer_entries.entry(*peer_addr).or_default();
for file in files {
if !file.is_dir && !peer_file_sizes.contains_key(&file.relative_path) {
let size = file.size;
file_size_map
.entry(file.relative_path.clone())
.or_default()
.entry(size)
.or_default()
.push(*peer_addr);
peer_file_sizes.insert(file.relative_path.clone(), size);
let descriptor = EntryDescriptor::from(file);
if let Some(previous) = entries.get(&file.relative_path) {
if *previous != descriptor {
eyre::bail!(
"peer {peer_addr} reported conflicting shapes or sizes for {}",
file.relative_path
);
}
// Repeated identical entries from one peer are one vote.
continue;
}
entries.insert(file.relative_path.clone(), descriptor);
}
peer_files.insert(*peer_addr, peer_file_sizes);
}
(file_size_map, peer_files)
let mut alias_paths = BTreeMap::<String, String>::new();
let mut entry_consensus_map: EntryConsensusMap = BTreeMap::new();
for (peer_addr, entries) in peer_entries {
for (relative_path, descriptor) in entries {
let alias = portable_manifest_path_key(&relative_path);
if let Some(previous_path) = alias_paths.get(&alias)
&& previous_path != &relative_path
{
let (first, second) = if previous_path < &relative_path {
(previous_path.as_str(), relative_path.as_str())
} else {
(relative_path.as_str(), previous_path.as_str())
};
eyre::bail!("peer manifests contain platform-alias paths: {first} and {second}");
}
alias_paths.insert(alias, relative_path.clone());
entry_consensus_map
.entry(relative_path)
.or_default()
.entry(descriptor)
.or_default()
.push(peer_addr);
}
}
Ok(entry_consensus_map)
}
/// Finds the majority size from a map of sizes to peer lists.
fn find_majority_size(size_map: &HashMap<u64, Vec<SocketAddr>>) -> (Option<u64>, usize) {
let mut majority_size = None;
fn portable_manifest_path_key(relative_path: &str) -> String {
relative_path
.split('/')
.map(portable_name_key)
.collect::<Vec<_>>()
.join("/")
}
/// Finds the unique most popular descriptor from a map of peer votes.
fn find_majority_descriptor(
descriptor_map: &BTreeMap<EntryDescriptor, Vec<SocketAddr>>,
) -> (Option<EntryDescriptor>, usize) {
let mut majority_descriptor = None;
let mut majority_count = 0;
for (&size, peers) in size_map {
for (&descriptor, peers) in descriptor_map {
let count = peers.len();
if count > majority_count {
majority_count = count;
majority_size = Some(size);
majority_descriptor = Some(descriptor);
} else if count == majority_count {
// Tie between different sizes - ambiguous, fail
majority_size = None;
break;
majority_descriptor = None;
}
}
(majority_size, majority_count)
(majority_descriptor, majority_count)
}
/// Updates peer scores based on consensus participation.
@@ -948,6 +986,7 @@ mod tests {
use std::net::SocketAddr;
use super::*;
use crate::{download::ValidatedDownloadManifest, test_support::TempDir};
fn addr(port: u16) -> SocketAddr {
SocketAddr::from(([127, 0, 0, 1], port))
@@ -979,6 +1018,15 @@ mod tests {
}
}
fn directory_desc(game_id: &str, relative_path: &str) -> GameFileDescription {
GameFileDescription {
game_id: game_id.to_string(),
relative_path: relative_path.to_string(),
is_dir: true,
size: 0,
}
}
#[test]
fn aggregation_counts_only_ready_peers_as_download_sources() {
let ready_addr = addr(12000);
@@ -1218,4 +1266,123 @@ mod tests {
Some(&vec![honest_a, honest_b])
);
}
#[test]
fn majority_validation_preserves_empty_directory_shape() {
let peer_a = addr(12008);
let peer_b = addr(12009);
let peer_c = addr(12010);
let version = file_desc("game", "game/version.ini", 8);
let manifests = vec![
(
peer_c,
vec![version.clone(), file_desc("game", "game/empty-child", 0)],
),
(
peer_b,
vec![version.clone(), directory_desc("game", "game/empty-child")],
),
(
peer_a,
vec![version, directory_desc("game", "game/empty-child")],
),
];
let (validated, peer_whitelist, file_peer_map) = PeerGameDB::new()
.validate_file_sizes_majority_from("game", &manifests)
.expect("directory shape should win the majority vote");
assert_eq!(peer_whitelist, vec![peer_a, peer_b, peer_c]);
assert_eq!(
validated
.iter()
.map(|entry| entry.relative_path.as_str())
.collect::<Vec<_>>(),
vec!["game/empty-child", "game/version.ini"]
);
let empty_child = validated
.iter()
.find(|entry| entry.relative_path == "game/empty-child")
.expect("empty directory should survive aggregation");
assert!(empty_child.is_dir);
assert_eq!(empty_child.size, 0);
assert_eq!(
file_peer_map.get("game/empty-child"),
Some(&vec![peer_a, peer_b])
);
let temp = TempDir::new("lanspread-peer-db-empty-directory");
let manifest = ValidatedDownloadManifest::from_protocol_v7(
temp.path(),
"game",
validated,
&GameCatalog::from_ids(["game".to_owned()]),
)
.expect("aggregated directory should enter the validated storage manifest");
let storage_entries = manifest.transfer_entries().collect::<Vec<_>>();
assert_eq!(storage_entries.len(), 1);
let storage_entry = storage_entries[0];
assert!(storage_entry.is_dir());
assert_eq!(storage_entry.size(), 0);
}
#[test]
fn two_peer_file_directory_shape_conflict_is_deterministic() {
let peer_a = addr(12011);
let peer_b = addr(12012);
let version = file_desc("game", "game/version.ini", 8);
let first_order = vec![
(
peer_a,
vec![version.clone(), directory_desc("game", "game/cache")],
),
(
peer_b,
vec![version.clone(), file_desc("game", "game/cache", 0)],
),
];
let mut reverse_order = first_order.clone();
reverse_order.reverse();
let first_error = PeerGameDB::new()
.validate_file_sizes_majority_from("game", &first_order)
.expect_err("a tied file/directory shape must be rejected")
.to_string();
let reverse_error = PeerGameDB::new()
.validate_file_sizes_majority_from("game", &reverse_order)
.expect_err("input ordering must not resolve a tied shape")
.to_string();
assert_eq!(first_error, reverse_error);
assert!(first_error.contains("different shapes or sizes"));
assert!(first_error.contains("game/cache"));
}
#[test]
fn cross_peer_platform_aliases_fail_deterministically() {
let peer_a = addr(12013);
let peer_b = addr(12014);
let version = file_desc("game", "game/version.ini", 8);
let first_order = vec![
(
peer_a,
vec![version.clone(), directory_desc("game", "game/Data")],
),
(peer_b, vec![version, directory_desc("game", "game/data")]),
];
let mut reverse_order = first_order.clone();
reverse_order.reverse();
let first_error = PeerGameDB::new()
.validate_file_sizes_majority_from("game", &first_order)
.expect_err("portable path aliases cannot share one transfer spelling")
.to_string();
let reverse_error = PeerGameDB::new()
.validate_file_sizes_majority_from("game", &reverse_order)
.expect_err("alias rejection must not depend on peer ordering")
.to_string();
assert_eq!(first_error, reverse_error);
assert!(first_error.contains("game/Data and game/data"));
}
}