diff --git a/crates/emulebb-ed2k/src/ed2k_tcp/download/session.rs b/crates/emulebb-ed2k/src/ed2k_tcp/download/session.rs index 27305966..765bffa9 100644 --- a/crates/emulebb-ed2k/src/ed2k_tcp/download/session.rs +++ b/crates/emulebb-ed2k/src/ed2k_tcp/download/session.rs @@ -879,6 +879,12 @@ pub(in crate::ed2k_tcp) async fn drive_download_session( } (OP_EMULEPROT, OP_HASHSETANSWER2) => { let hashset_answer = decode_hashset_answer2(&packet.payload)?; + let aich_generation = if session_state.aich_hashset_answer_pending { + session_state.aich_hashset_answer_pending = false; + session_state.aich_hashset_requested_generation + } else { + None + }; if !request_file_identifier .matches_relaxed(&hashset_answer.file_identifier) { @@ -904,9 +910,13 @@ pub(in crate::ed2k_tcp) async fn drive_download_session( .store_md4_hashset(file_hash_hex, hashset) .await?; } - if let Some(hashset) = hashset_answer.aich_hashset { + // Unrequested AICH sections cannot establish identity. + // Preserve usable MD4 data while ignoring an untrusted tree. + if let Some(hashset) = hashset_answer.aich_hashset + && let Some(generation) = aich_generation + { manifest = transfer_runtime - .store_aich_hashset(file_hash_hex, hashset) + .store_aich_hashset_for_generation(file_hash_hex, hashset, generation) .await?; request_file_identifier = Ed2kFileIdentifier::from_manifest(&manifest)?; } diff --git a/crates/emulebb-ed2k/src/ed2k_tcp/download/session/startup.rs b/crates/emulebb-ed2k/src/ed2k_tcp/download/session/startup.rs index c2c546b3..c40189b0 100644 --- a/crates/emulebb-ed2k/src/ed2k_tcp/download/session/startup.rs +++ b/crates/emulebb-ed2k/src/ed2k_tcp/download/session/startup.rs @@ -178,11 +178,17 @@ pub(super) async fn advance_download_startup(step: DownloadStartupStep<'_>) -> R session_state.aich_file_hash_requested = true; } + let request_md4 = !manifest.md4_hashset_acquired && !session_state.hashset_requested; + let request_aich = session_state.remote_supports_file_identifiers + && manifest.file_size > ED2K_PART_SIZE + && manifest.aich_root.is_some() + && !manifest.aich_hashset_acquired + && session_state.peer_advertised_aich_root + && session_state.aich_hashset_requested_generation.is_none(); if send_initial_requests && session_state.hello_complete && manifest.file_size != 0 - && !manifest.md4_hashset_acquired - && !session_state.hashset_requested + && (request_md4 || request_aich) && session_state.startup_file_response_received { if manifest.file_size <= ED2K_PART_SIZE { @@ -194,16 +200,10 @@ pub(super) async fn advance_download_startup(step: DownloadStartupStep<'_>) -> R encode_hashset_request2( request_file_identifier, Ed2kHashsetRequestOptions { - request_md4: true, - // Request the AICH hashset only when a peer has advertised - // an AICH root for this file. We can no longer key this off - // a trusted `manifest.aich_root` (network-learned roots are - // no longer promoted on first sight -- they need IP - // corroboration), so track the peer-advertised signal - // separately. This preserves the prior behaviour without - // making the request depend on a trust decision. - request_aich: manifest.file_size > ED2K_PART_SIZE - && session_state.peer_advertised_aich_root, + request_md4, + // Stock requests part hashes for an already accepted + // master. A peer advertisement is evidence, not trust. + request_aich, }, )? } else { @@ -225,6 +225,11 @@ pub(super) async fn advance_download_startup(step: DownloadStartupStep<'_>) -> R } })?; session_state.hashset_requested = true; + if request_aich { + session_state.aich_hashset_requested_generation = + Some(manifest.aich_trust_generation); + session_state.aich_hashset_answer_pending = true; + } session_state.hashset_requested_at = Some(tokio::time::Instant::now()); } } diff --git a/crates/emulebb-ed2k/src/ed2k_tcp/download/session/state.rs b/crates/emulebb-ed2k/src/ed2k_tcp/download/session/state.rs index a7e58461..4c6ad0c0 100644 --- a/crates/emulebb-ed2k/src/ed2k_tcp/download/session/state.rs +++ b/crates/emulebb-ed2k/src/ed2k_tcp/download/session/state.rs @@ -89,6 +89,11 @@ pub(super) struct DownloadSessionState { pub(super) peer_advertised_aich_root: bool, pub(super) aich_file_hash_requested: bool, pub(super) hashset_requested: bool, + /// One AICH-set attempt per connection, bound to the accepted generation. + /// Keep the attempted generation after consuming the reply: the wire has + /// no request ID with which to distinguish a delayed ABA response. + pub(super) aich_hashset_requested_generation: Option, + pub(super) aich_hashset_answer_pending: bool, pub(super) hashset_requested_at: Option, pub(super) upload_requested: bool, /// The next queue-rank packet is solicited by OP_STARTUPLOADREQ (or by an @@ -186,6 +191,8 @@ impl DownloadSessionState { peer_advertised_aich_root: false, aich_file_hash_requested: false, hashset_requested: false, + aich_hashset_requested_generation: None, + aich_hashset_answer_pending: false, hashset_requested_at: None, upload_requested: false, queue_rank_pending: false, diff --git a/crates/emulebb-ed2k/src/ed2k_tcp/tests/download/hashset_startup.rs b/crates/emulebb-ed2k/src/ed2k_tcp/tests/download/hashset_startup.rs index a3e1e6ea..e4799d65 100644 --- a/crates/emulebb-ed2k/src/ed2k_tcp/tests/download/hashset_startup.rs +++ b/crates/emulebb-ed2k/src/ed2k_tcp/tests/download/hashset_startup.rs @@ -2,8 +2,58 @@ use super::*; #[tokio::test] async fn large_file_download_starts_hashset_flow_before_peer_secure_ident_key_arrives() { + run_hashset_startup_case(StartupAichCase::Advertised).await; +} + +#[tokio::test] +async fn consistent_unrelated_peer_aich_tree_cannot_become_download_identity() { + run_hashset_startup_case(StartupAichCase::Unrelated).await; +} + +#[tokio::test] +async fn trusted_operator_master_allows_a_matching_modern_part_set() { + run_hashset_startup_case(StartupAichCase::Operator).await; +} + +#[tokio::test] +async fn trusted_master_allows_aich_only_acquisition_after_md4_is_known() { + run_hashset_startup_case(StartupAichCase::OperatorWithMd4).await; +} + +#[tokio::test] +async fn current_consensus_master_allows_a_matching_modern_part_set() { + run_hashset_startup_case(StartupAichCase::NetworkConsensus).await; +} + +#[tokio::test] +async fn late_modern_aich_answer_cannot_cross_a_consensus_aba_generation() { + run_hashset_startup_case(StartupAichCase::ConsensusAba).await; +} + +enum StartupAichCase { + Advertised, + Unrelated, + Operator, + OperatorWithMd4, + NetworkConsensus, + ConsensusAba, +} + +async fn run_hashset_startup_case(case: StartupAichCase) { + let unrelated_aich_tree = matches!(case, StartupAichCase::Unrelated); + let operator_root = matches!( + case, + StartupAichCase::Operator | StartupAichCase::OperatorWithMd4 + ); + let consensus_root = matches!( + case, + StartupAichCase::NetworkConsensus | StartupAichCase::ConsensusAba + ); + let late_aich_answer = matches!(case, StartupAichCase::ConsensusAba); + let trusted_root = operator_root || consensus_root; + let md4_preseeded = matches!(case, StartupAichCase::OperatorWithMd4); let root = unique_test_dir("ed2k-large-file-secure-ident-order"); - let transfer_runtime = Ed2kTransferRuntime::load_or_create(&root).unwrap(); + let transfer_runtime = Arc::new(Ed2kTransferRuntime::load_or_create(&root).unwrap()); let payload = vec![0x5A; (ED2K_PART_SIZE as usize) + 32_768]; let md4_hashset = payload .chunks(ED2K_PART_SIZE as usize) @@ -43,11 +93,51 @@ async fn large_file_download_starts_hashset_flow_before_peer_secure_ident_key_ar .store_piece_data(&file_hash_hex, 1, &payload[ED2K_PART_SIZE as usize..]) .await .unwrap(); - let source_aich = source_runtime + let mut source_aich = source_runtime .aich_hashset(&file_hash) .await .unwrap() .expect("missing source AICH hashset"); + if unrelated_aich_tree { + // A mathematically valid two-part tree unrelated to the MD4-identified + // payload. It must not authorize salvage or survive local completion. + source_aich.part_hashes = vec![[0x21; 20], [0x43; 20]]; + source_aich.master_hash = sha1::Sha1::digest( + source_aich + .part_hashes + .iter() + .flatten() + .copied() + .collect::>(), + ) + .into(); + } + let advertised_root = source_aich.master_hash; + if operator_root { + transfer_runtime + .reconcile_aich_root(&file_hash_hex, Some(advertised_root)) + .await + .unwrap(); + } else if consensus_root { + let observations = (0..10) + .map(|signer| { + ( + advertised_root, + IpAddr::V4(std::net::Ipv4Addr::new(203, signer, 0, 1)), + ) + }) + .collect::>(); + transfer_runtime + .record_network_aich_observations(&file_hash_hex, &observations) + .await + .unwrap(); + } + if md4_preseeded { + transfer_runtime + .store_md4_hashset(&file_hash_hex, md4_hashset.clone()) + .await + .unwrap(); + } let source_identifier = super::Ed2kFileIdentifier { file_hash, file_size: Some(payload.len() as u64), @@ -57,6 +147,7 @@ async fn large_file_download_starts_hashset_flow_before_peer_secure_ident_key_ar let listener = TcpListener::bind((test_bind_ip(), 0)).await.unwrap(); let peer_addr = listener.local_addr().unwrap(); let payload_for_server = payload.clone(); + let observed_runtime = Arc::clone(&transfer_runtime); let server = tokio::spawn(async move { let (mut stream, _) = listener.accept().await.unwrap(); @@ -115,21 +206,107 @@ async fn large_file_download_starts_hashset_flow_before_peer_secure_ident_key_ar requested_identifier.file_size, Some(payload_for_server.len() as u64) ); - assert!(request_options.request_md4); - assert!(request_options.request_aich); + assert_eq!(request_options.request_md4, !md4_preseeded); + assert_eq!(request_options.request_aich, trusted_root); + + let mut expected_generation = None; + if consensus_root { + let requested = observed_runtime + .manifest(&file_hash.to_string()) + .await + .unwrap(); + assert_eq!(requested.aich_root, Some(hex::encode(advertised_root))); + assert_eq!( + requested.aich_root_origin, + Some(crate::ed2k_transfer::Ed2kAichRootOrigin::NetworkConsensus) + ); + expected_generation = Some(requested.aich_trust_generation); + if late_aich_answer { + // WHY: receiving the real request is the causal barrier. Retire + // its generation before sending its otherwise matching answer; + // returning to the same master prevents a root mismatch from + // accidentally testing the wrong response-admission boundary. + let mut dissenting_root = advertised_root; + dissenting_root[0] ^= 0x80; + let demoted = observed_runtime + .record_network_aich_root( + &file_hash.to_string(), + Some(dissenting_root), + IpAddr::V4(std::net::Ipv4Addr::new(198, 0, 0, 1)), + ) + .await + .unwrap(); + assert!(demoted.aich_root.is_none()); + assert!(demoted.aich_trust_generation > requested.aich_trust_generation); + let renewed = observed_runtime + .record_network_aich_observations( + &file_hash.to_string(), + &[10, 11].map(|signer| { + ( + advertised_root, + IpAddr::V4(std::net::Ipv4Addr::new(203, signer, 0, 1)), + ) + }), + ) + .await + .unwrap(); + assert_eq!(renewed.aich_root, requested.aich_root); + assert_eq!(renewed.aich_root_origin, requested.aich_root_origin); + assert!(renewed.aich_trust_generation > demoted.aich_trust_generation); + assert!(!renewed.aich_hashset_acquired); + expected_generation = Some(renewed.aich_trust_generation); + } + } let hashset_answer = super::encode_hashset_answer2( &source_identifier, - Some(&md4_hashset), + (!md4_preseeded).then_some(md4_hashset.as_slice()), Some(&source_aich), ) .unwrap(); stream.write_all(&hashset_answer).await.unwrap(); + // With known MD4, STARTUPLOADREQ can precede processing this AICH-only + // answer. A reply to the following frame joins the actual receive path. + stream + .write_all(&encode_packet(OP_EMULEPROT, OP_PUBLICIP_REQ, &[])) + .await + .unwrap(); let start_upload = read_packet(&mut stream).await; assert_eq!(start_upload[0], OP_EDONKEYPROT); assert_eq!(start_upload[5], super::OP_STARTUPLOADREQ); assert_eq!(&start_upload[6..22], &file_hash.0); + let public_ip_answer = read_packet(&mut stream).await; + assert_eq!(public_ip_answer[0], OP_EMULEPROT); + assert_eq!(public_ip_answer[5], OP_PUBLICIP_ANSWER); + assert_eq!( + decode_public_ip_answer_payload(&public_ip_answer[6..]).unwrap(), + test_bind_ip() + ); + + // Observe before supplying any payload: a later local completion pass + // must not hide the single-peer acquisition bypass. + let before_payload = observed_runtime + .manifest(&file_hash.to_string()) + .await + .unwrap(); + assert!(before_payload.md4_hashset_acquired); + if trusted_root { + assert_eq!(before_payload.aich_root, Some(hex::encode(advertised_root))); + assert_eq!(before_payload.aich_hashset_acquired, !late_aich_answer); + assert_eq!(before_payload.aich_hashset.is_empty(), late_aich_answer); + if let Some(generation) = expected_generation { + assert_eq!(before_payload.aich_trust_generation, generation); + assert_eq!( + before_payload.aich_root_origin, + Some(crate::ed2k_transfer::Ed2kAichRootOrigin::NetworkConsensus) + ); + } + } else { + assert!(before_payload.aich_root.is_none()); + assert!(!before_payload.aich_hashset_acquired); + assert!(before_payload.aich_hashset.is_empty()); + } let accept = encode_accept_upload_req(); stream.write_all(&accept).await.unwrap(); @@ -243,6 +420,13 @@ async fn large_file_download_starts_hashset_flow_before_peer_secure_ident_key_ar let manifest = transfer_runtime.manifest(&file_hash_hex).await.unwrap(); assert!(manifest.completed); assert!(manifest.aich_hashset_acquired); + assert_eq!( + manifest.aich_root_origin, + Some(crate::ed2k_transfer::Ed2kAichRootOrigin::LocalComputed) + ); assert_eq!(manifest.aich_hashset.len(), 2); + if unrelated_aich_tree { + assert_ne!(manifest.aich_root, Some(hex::encode(advertised_root))); + } server.await.unwrap(); } diff --git a/crates/emulebb-ed2k/src/ed2k_tcp/tests/listener/startup.rs b/crates/emulebb-ed2k/src/ed2k_tcp/tests/listener/startup.rs index 5a49a9a2..67efdc02 100644 --- a/crates/emulebb-ed2k/src/ed2k_tcp/tests/listener/startup.rs +++ b/crates/emulebb-ed2k/src/ed2k_tcp/tests/listener/startup.rs @@ -298,10 +298,17 @@ async fn listener_established_session_tolerates_unknown_opcodes_and_continues() .await .unwrap(); let aich_root = [0x7B; 20]; - transfer_runtime - .reconcile_aich_root(&file_hash_hex, Some(aich_root)) + let proposed = transfer_runtime + .record_network_aich_root( + &file_hash_hex, + Some(aich_root), + std::net::IpAddr::from([203, 0, 113, 7]), + ) .await .unwrap(); + assert!(proposed.aich_root.is_none()); + assert!(proposed.aich_root_origin.is_none()); + assert!(!proposed.aich_hashset_acquired); transfer_runtime .store_md4_hashset(&no_sources_hash_hex, Vec::new()) .await diff --git a/crates/emulebb-ed2k/src/ed2k_transfer.rs b/crates/emulebb-ed2k/src/ed2k_transfer.rs index 4b44f023..5f1a6c6b 100644 --- a/crates/emulebb-ed2k/src/ed2k_transfer.rs +++ b/crates/emulebb-ed2k/src/ed2k_transfer.rs @@ -108,11 +108,12 @@ use manifest::Ed2kManifestCheckpointState; pub(crate) use manifest::expected_piece_length; pub use manifest::new_transfer_job; pub use model::{ - ED2K_CALLBACK_TICKET_MAX_LIFETIME, Ed2kCallbackDispatchOutcome, Ed2kCallbackDispatchPermit, - Ed2kCallbackDownload, Ed2kCallbackReservation, Ed2kCallbackRoute, Ed2kCallbackTicketError, - Ed2kCallbackTicketId, Ed2kCallbackTicketRequest, Ed2kDeliveryOwnership, Ed2kDeliveryState, - Ed2kFileNameObservation, Ed2kFileServingView, Ed2kLocalIngestSummary, Ed2kPieceState, - Ed2kReloadIndexEntry, Ed2kResumeManifest, Ed2kSourceHint, Ed2kTransferJob, Ed2kTransferState, + ED2K_CALLBACK_TICKET_MAX_LIFETIME, Ed2kAichRootOrigin, Ed2kCallbackDispatchOutcome, + Ed2kCallbackDispatchPermit, Ed2kCallbackDownload, Ed2kCallbackReservation, Ed2kCallbackRoute, + Ed2kCallbackTicketError, Ed2kCallbackTicketId, Ed2kCallbackTicketRequest, + Ed2kDeliveryOwnership, Ed2kDeliveryState, Ed2kFileNameObservation, Ed2kFileServingView, + Ed2kLocalIngestSummary, Ed2kPieceState, Ed2kReloadIndexEntry, Ed2kResumeManifest, + Ed2kSourceHint, Ed2kTransferJob, Ed2kTransferState, }; pub(crate) use model::{ Ed2kAichHashset, Ed2kCallbackClaim, Ed2kCallbackClaimError, Ed2kCallbackClaimToken, @@ -275,7 +276,7 @@ pub struct Ed2kTransferRuntime { metadata: MetadataStore, file_name_policy: Ed2kFileNamePolicy, shared_catalog: Ed2kSharedCatalog, - shared_catalog_generation: AtomicU64, + shared_catalog_generation: Arc, /// Keyset cursor for a persisted catalog intentionally loaded only through /// the first publishable startup cohort. `pending` is claimed atomically so /// at most one background hydration pass can append the remaining pages. @@ -323,6 +324,8 @@ pub struct Ed2kTransferRuntime { /// signer set lives here (in-memory, live session state, never persisted); /// the durable trust decision is the promoted `manifest.aich_root`. aich_root_corroboration: Arc>>, + #[cfg(test)] + aich_persist_barrier: Arc>>>, download_activity: Arc>>, /// Live per-source download state keyed by file hash -> peer endpoint, used /// to surface sourcesTransferring/partsAvailable and live transfer-source @@ -738,7 +741,7 @@ impl Ed2kTransferRuntime { metadata, file_name_policy, shared_catalog, - shared_catalog_generation: AtomicU64::new(0), + shared_catalog_generation: Arc::new(AtomicU64::new(0)), shared_catalog_hydration_cursor: StdMutex::new(None), shared_catalog_hydration_pending: AtomicBool::new(hydration_pending), servable_shared_hash_cache: Arc::new(StdMutex::new(ServableSharedHashCache::default())), @@ -753,6 +756,8 @@ impl Ed2kTransferRuntime { payload_handles: Arc::new(StdMutex::new(HashMap::new())), source_exchange: SourceExchangeState::default(), aich_root_corroboration: Arc::new(StdMutex::new(HashMap::new())), + #[cfg(test)] + aich_persist_barrier: Arc::new(StdMutex::new(None)), download_activity: Arc::new(StdMutex::new(HashMap::new())), download_sources: Arc::new(StdMutex::new(HashMap::new())), download_abuse: Arc::new(parking_lot::Mutex::new( diff --git a/crates/emulebb-ed2k/src/ed2k_transfer/aich_trust.rs b/crates/emulebb-ed2k/src/ed2k_transfer/aich_trust.rs index 7a597eb6..8786567b 100644 --- a/crates/emulebb-ed2k/src/ed2k_transfer/aich_trust.rs +++ b/crates/emulebb-ed2k/src/ed2k_transfer/aich_trust.rs @@ -11,11 +11,72 @@ //! percent of all proposing IPs. //! //! This module mirrors that policy for network-learned roots. Authoritative -//! roots (computed locally from a completed file, or read back from persisted -//! `.met`/manifest metadata) bypass corroboration entirely and are promoted -//! directly via `reconcile_aich_root`. +//! roots have explicit local/operator/import provenance. A persisted accepted +//! decision resumes VERIFIED as stock without changing that recorded origin; +//! a part set that was merely acquired never creates such a decision. -use std::collections::HashMap; +use std::collections::{BTreeMap, HashMap}; + +use super::{Ed2kAichRootOrigin, Ed2kResumeManifest}; + +/// Deterministically stop a test after SQL commit and before cache publication. +#[cfg(test)] +#[derive(Debug, Default)] +pub(super) struct AichPersistBarrier { + pub(super) committed: tokio::sync::Notify, + pub(super) release: tokio::sync::Notify, +} + +/// Merge only canonical AICH authority, preserving independent cached payload +/// and MD4 state. Existing part hashes remain valid only for the same root; +/// the durable generation still fences every outstanding response. +pub(super) fn merge_persisted_identity( + manifest: &mut Ed2kResumeManifest, + identity: &emulebb_metadata::MetadataTransferAichIdentity, +) -> anyhow::Result<()> { + let origin = identity + .origin + .as_deref() + .map(Ed2kAichRootOrigin::parse) + .transpose()?; + if manifest.aich_root != identity.root { + manifest.aich_hashset.clear(); + manifest.aich_hashset_acquired = false; + } + manifest.aich_root = identity.root.clone(); + manifest.aich_root_origin = origin; + manifest.aich_trust_generation = identity.generation; + Ok(()) +} + +/// Change accepted identity under the manifest lock. The generation and +/// dependent part set change atomically with the persisted decision. +pub(super) fn set_identity( + manifest: &mut Ed2kResumeManifest, + root: Option, + origin: Option, +) -> anyhow::Result { + anyhow::ensure!( + root.is_some() == origin.is_some(), + "incomplete AICH decision" + ); + if manifest.aich_root == root && manifest.aich_root_origin == origin { + return Ok(false); + } + let generation = manifest + .aich_trust_generation + .checked_add(1) + .ok_or_else(|| anyhow::anyhow!("AICH trust generation exhausted"))?; + i64::try_from(generation)?; + if manifest.aich_root != root { + manifest.aich_hashset.clear(); + manifest.aich_hashset_acquired = false; + } + manifest.aich_root = root; + manifest.aich_root_origin = origin; + manifest.aich_trust_generation = generation; + Ok(true) +} /// How many distinct IPs must propose the same AICH root before it is trusted. /// Mirrors `MINUNIQUEIPS_TOTRUST` (`SHAHashSet.cpp:40`). @@ -25,6 +86,12 @@ pub(super) const MINUNIQUEIPS_TOTRUST: usize = 10; /// before it is trusted. Mirrors `MINPERCENTAGE_TOTRUST` (`SHAHashSet.cpp:41`). pub(super) const MINPERCENTAGE_TOTRUST: u64 = 92; +/// Resource limits do not turn a truncated sample into trust. Once either +/// limit is exceeded, only independent authoritative metadata can establish +/// the root for this transfer; ordinary MD4 downloading remains available. +const MAX_SIGNERS: usize = 4_096; +const MAX_CANDIDATES: usize = 64; + /// Mask applied to a proposing IPv4 address (in network/big-endian byte order) /// before counting it as a unique signer. Mirrors the master's /// `dwIP &= 0x00F0FFFF` in `CAICHUntrustedHash::AddSigningIP` @@ -42,12 +109,15 @@ fn mask_signer_ip(octets: [u8; 4]) -> [u8; 4] { /// Per-file accumulator of network-proposed AICH roots keyed by signer. Keeping /// one root per masked signer prevents one responder from inflating multiple /// conflicting candidates. In-memory only (live session trust state, never -/// persisted): the durable trust decision lives in `manifest.aich_root`, which -/// is only written once a root is promoted. -#[derive(Debug, Default)] +/// persisted): the durable master, origin and generation describe an accepted +/// decision. Live vote history is bounded and released with the transfer. +#[derive(Debug, Clone, Default)] pub(super) struct AichRootCorroboration { /// Masked proposing IP -> first root proposed by that signer. proposals: HashMap<[u8; 4], [u8; 20]>, + /// Ordered roots make equal-weight evidence deterministic. + counts: BTreeMap<[u8; 20], usize>, + saturated: bool, } impl AichRootCorroboration { @@ -55,30 +125,29 @@ impl AichRootCorroboration { /// Returns `true` only when that signer had not voted before. pub(super) fn record(&mut self, root: [u8; 20], from_ip: [u8; 4]) -> bool { let masked = mask_signer_ip(from_ip); - if let std::collections::hash_map::Entry::Vacant(entry) = self.proposals.entry(masked) { - entry.insert(root); - true - } else { - false + if self.saturated || self.proposals.contains_key(&masked) { + return false; } + if self.proposals.len() == MAX_SIGNERS + || (!self.counts.contains_key(&root) && self.counts.len() == MAX_CANDIDATES) + { + self.saturated = true; + return false; + } + self.proposals.insert(masked, root); + *self.counts.entry(root).or_default() += 1; + true } /// Return the leading root if it now satisfies the master's trust gate: /// `>= MINUNIQUEIPS_TOTRUST` distinct IPs AND `>= MINPERCENTAGE_TOTRUST` /// percent of all proposing IPs. Mirrors `SHAHashSet.cpp:998`. pub(super) fn trusted_root(&self) -> Option<[u8; 20]> { - let total_signers = self.proposals.len(); - let mut counts = HashMap::<[u8; 20], usize>::new(); - for root in self.proposals.values() { - *counts.entry(*root).or_default() += 1; - } - let mut leader: Option<([u8; 20], usize)> = None; - for (root, count) in counts { - if leader.is_none_or(|(_, best)| count > best) { - leader = Some((root, count)); - } + if self.saturated { + return None; } - let (root, count) = leader?; + let total_signers = self.proposals.len(); + let (&root, &count) = self.counts.iter().max_by_key(|(_, count)| **count)?; if total_signers == 0 { return None; } @@ -152,4 +221,56 @@ mod tests { } assert_eq!(acc.trusted_root(), Some([1u8; 20])); } + + #[test] + fn signer_mask_matches_stock_little_endian_integer_mask() { + for octets in [[1, 2, 15, 255], [1, 2, 16, 1], [203, 0, 113, 7]] { + let stock = u32::from_le_bytes(octets) & 0x00F0_FFFF; + assert_eq!(mask_signer_ip(octets), stock.to_le_bytes()); + } + assert_eq!(mask_signer_ip([1, 2, 15, 255]), [1, 2, 0, 0]); + assert_eq!(mask_signer_ip([1, 2, 16, 1]), [1, 2, 16, 0]); + } + + #[test] + fn later_disagreement_reconsiders_consensus_at_stock_percentage_boundary() { + let mut acc = AichRootCorroboration::default(); + for signer in 0..92 { + acc.record([1; 20], [203, signer, 0, 1]); + } + for signer in 0..8 { + acc.record([2; 20], [198, signer, 0, 1]); + } + assert_eq!(acc.trusted_root(), Some([1; 20])); + acc.record([2; 20], [198, 8, 0, 1]); + assert_eq!(acc.trusted_root(), None); + } + + #[test] + fn candidate_limit_does_not_freeze_previously_trusted_sample() { + let mut acc = AichRootCorroboration::default(); + for signer in 0..10 { + acc.record([1; 20], [203, signer, 0, 1]); + } + assert_eq!(acc.trusted_root(), Some([1; 20])); + for candidate in 2..=u8::try_from(MAX_CANDIDATES).unwrap() { + acc.record([candidate; 20], [198, candidate, 0, 1]); + } + assert!(!acc.record([0xff; 20], [192, 0, 0, 1])); + assert_eq!(acc.counts.len(), MAX_CANDIDATES); + assert_eq!(acc.trusted_root(), None); + } + + #[test] + fn signer_limit_is_bounded_and_fails_closed() { + let mut acc = AichRootCorroboration::default(); + for signer in 0..MAX_SIGNERS { + let bytes = u16::try_from(signer).unwrap().to_be_bytes(); + acc.record([1; 20], [bytes[0], bytes[1], 0, 1]); + } + assert_eq!(acc.trusted_root(), Some([1; 20])); + assert!(!acc.record([1; 20], [200, 0, 0, 1])); + assert_eq!(acc.proposals.len(), MAX_SIGNERS); + assert_eq!(acc.trusted_root(), None); + } } diff --git a/crates/emulebb-ed2k/src/ed2k_transfer/completion_rehash.rs b/crates/emulebb-ed2k/src/ed2k_transfer/completion_rehash.rs index 59f32f1c..bebfd075 100644 --- a/crates/emulebb-ed2k/src/ed2k_transfer/completion_rehash.rs +++ b/crates/emulebb-ed2k/src/ed2k_transfer/completion_rehash.rs @@ -225,9 +225,14 @@ impl Ed2kTransferRuntime { manifest.payload_fingerprint = computation.payload_fingerprint; let recovery = computation.recovery; if let Some(set) = recovery.as_ref() - && (manifest.aich_root.is_none() || !manifest.aich_hashset_acquired) + && (manifest.aich_root.is_none() + || manifest.aich_root.as_deref() == Some(hex::encode(set.root).as_str())) { - manifest.aich_root = Some(hex::encode(set.root)); + super::aich_trust::set_identity( + &mut manifest, + Some(hex::encode(set.root)), + Some(super::Ed2kAichRootOrigin::LocalComputed), + )?; manifest.aich_hashset = if set.part_roots.len() == 1 { Vec::new() } else { diff --git a/crates/emulebb-ed2k/src/ed2k_transfer/duplicate.rs b/crates/emulebb-ed2k/src/ed2k_transfer/duplicate.rs index aa797eec..f62ed687 100644 --- a/crates/emulebb-ed2k/src/ed2k_transfer/duplicate.rs +++ b/crates/emulebb-ed2k/src/ed2k_transfer/duplicate.rs @@ -216,6 +216,8 @@ impl Ed2kTransferRuntime { replacement.md4_hashset = existing.md4_hashset; replacement.aich_hashset_acquired = existing.aich_hashset_acquired; replacement.aich_root = existing.aich_root; + replacement.aich_root_origin = existing.aich_root_origin; + replacement.aich_trust_generation = existing.aich_trust_generation; replacement.aich_hashset = existing.aich_hashset; replacement.upload_priority = existing.upload_priority; replacement.auto_upload_priority = existing.auto_upload_priority; diff --git a/crates/emulebb-ed2k/src/ed2k_transfer/metadata.rs b/crates/emulebb-ed2k/src/ed2k_transfer/metadata.rs index a54c10b0..9399e4ad 100644 --- a/crates/emulebb-ed2k/src/ed2k_transfer/metadata.rs +++ b/crates/emulebb-ed2k/src/ed2k_transfer/metadata.rs @@ -24,9 +24,9 @@ use super::manifest::{manifest_has_structural_progress, piece_count}; use super::transfer_sql::manifest_from_metadata; use super::upload_queue::upload_priority_score; use super::{ - Ed2kAichHashset, Ed2kFileNameObservation, Ed2kPieceState, Ed2kResumeManifest, Ed2kSharedEntry, - Ed2kSourceHint, Ed2kTransferDeletionRecoveryReport, Ed2kTransferJob, - Ed2kTransferRetirementRecoveryReport, Ed2kTransferRuntime, Ed2kTransferState, + Ed2kAichHashset, Ed2kAichRootOrigin, Ed2kFileNameObservation, Ed2kPieceState, + Ed2kResumeManifest, Ed2kSharedEntry, Ed2kSourceHint, Ed2kTransferDeletionRecoveryReport, + Ed2kTransferJob, Ed2kTransferRetirementRecoveryReport, Ed2kTransferRuntime, Ed2kTransferState, }; #[derive(Clone, Copy, PartialEq, Eq)] @@ -477,36 +477,59 @@ impl Ed2kTransferRuntime { Ok(manifest) } - /// Persist the canonical ED2K AICH root and part hashset after validating - /// the payload against the expected file size. + /// Activate a peer's part hashset only against an independently trusted + /// master. Internal tree consistency never establishes content identity. + #[cfg(test)] pub(crate) async fn store_aich_hashset( &self, file_hash: &str, aich_hashset: Ed2kAichHashset, ) -> Result { - let _guard = self.lock_manifest(file_hash).await; - let mut manifest = self.load_manifest_unlocked(file_hash).await?; - if let Some(existing_root) = manifest.aich_root.as_deref() { - let existing_root = decode_aich_hash_hex(existing_root)?; - if existing_root != aich_hashset.master_hash { - anyhow::bail!( - "refusing to replace AICH root for {} with conflicting data", - file_hash - ); - } + let generation = self.manifest(file_hash).await?.aich_trust_generation; + self.store_aich_hashset_for_generation(file_hash, aich_hashset, generation) + .await + } + + /// A stale response cannot reactivate a retired decision, even if a later + /// consensus selects the same master again. + pub(crate) async fn store_aich_hashset_for_generation( + &self, + file_hash: &str, + aich_hashset: Ed2kAichHashset, + expected_generation: u64, + ) -> Result { + let guard = self.lock_manifest(file_hash).await; + let mut manifest = self.load_aich_manifest_unlocked(file_hash).await?; + if manifest.aich_trust_generation != expected_generation { + return Ok(manifest); } validate_aich_hashset(manifest.file_size, &aich_hashset)?; - manifest.aich_root = Some(hex::encode(aich_hashset.master_hash)); + let existing_root = manifest + .aich_root + .as_deref() + .context("AICH hashset requires an independently trusted master")?; + ensure!( + decode_aich_hash_hex(existing_root)? == aich_hashset.master_hash, + "refusing to replace AICH root for {} with conflicting data", + file_hash + ); manifest.aich_hashset = aich_hashset.part_hashes.iter().map(hex::encode).collect(); manifest.aich_hashset_acquired = true; - self.store_manifest_unlocked(&manifest).await?; - self.upsert_verified_catalog_entry(&manifest).await; + let votes = self + .aich_root_corroboration + .lock() + .expect("aich corroboration mutex poisoned") + .get(file_hash) + .cloned(); + let (manifest, _guard) = self + .store_aich_transition_owned(guard, manifest, votes) + .await?; Ok(manifest) } - /// Persist an AUTHORITATIVE AICH root (computed locally from a completed - /// file, or read back from persisted `.met`/manifest metadata). Such a root - /// is trusted on sight and promoted directly. For roots learned over the + /// Accept an explicit operator-link AICH master. Local computation and + /// stock import have separate authority and do not enter this network + /// observation interface. For roots learned over the /// wire from a single peer, use [`record_network_aich_root`] instead, which /// requires corroboration before promotion. /// @@ -516,26 +539,35 @@ impl Ed2kTransferRuntime { file_hash: &str, aich_root: Option<[u8; 20]>, ) -> Result { - let _guard = self.lock_manifest(file_hash).await; - let mut manifest = self.load_manifest_unlocked(file_hash).await?; - let mut changed = false; + let guard = self.lock_manifest(file_hash).await; + let mut manifest = self.load_aich_manifest_unlocked(file_hash).await?; if let Some(aich_root) = aich_root { let encoded = hex::encode(aich_root); - if let Some(existing_root) = manifest.aich_root.as_deref() { - if existing_root != encoded { - anyhow::bail!( - "refusing to replace AICH root for {} with conflicting metadata", - file_hash - ); - } - } else { - manifest.aich_root = Some(encoded); - changed = true; + if let Some(existing_root) = manifest.aich_root.as_deref() + && existing_root != encoded + && manifest.aich_root_origin != Some(Ed2kAichRootOrigin::NetworkConsensus) + { + anyhow::bail!( + "refusing to replace AICH root for {} with conflicting metadata", + file_hash + ); } - } - if changed { - self.store_manifest_unlocked(&manifest).await?; - self.upsert_verified_catalog_entry(&manifest).await; + let origin = match manifest.aich_root_origin { + Some( + origin @ (Ed2kAichRootOrigin::LocalComputed | Ed2kAichRootOrigin::StockImport), + ) => origin, + _ => Ed2kAichRootOrigin::OperatorLink, + }; + if super::aich_trust::set_identity(&mut manifest, Some(encoded), Some(origin))? { + let (manifest, _guard) = self + .store_aich_transition_owned(guard, manifest, None) + .await?; + return Ok(manifest); + } + self.aich_root_corroboration + .lock() + .expect("aich corroboration mutex poisoned") + .remove(file_hash); } Ok(manifest) } @@ -552,9 +584,9 @@ impl Ed2kTransferRuntime { /// manifest's root is left unset, so salvage falls back to whole-part /// re-download. /// - /// If `manifest.aich_root` is already populated (authoritative, or already - /// promoted) this is a no-op. The MD4 hashset backstop independently - /// prevents corruption, so this is defense-in-depth + parity. + /// A live consensus keeps collecting votes and may lose trust. Explicit + /// authority is stable. Stock resumes a persisted accepted root as VERIFIED + /// (PartFile.cpp FT_AICH_HASH); retain its origin without recreating votes. pub async fn record_network_aich_root( &self, file_hash: &str, @@ -577,39 +609,50 @@ impl Ed2kTransferRuntime { file_hash: &str, observations: &[([u8; 20], std::net::IpAddr)], ) -> Result { - let _guard = self.lock_manifest(file_hash).await; - let mut manifest = self.load_manifest_unlocked(file_hash).await?; + let guard = self.lock_manifest(file_hash).await; + let mut manifest = self.load_aich_manifest_unlocked(file_hash).await?; if observations.is_empty() { return Ok(manifest); } - // Already have a trusted root: nothing to promote. If a peer proposes a - // root that conflicts with the trusted one, ignore it (the trusted root - // wins); matching is harmless. - if manifest.aich_root.is_some() { + if manifest.aich_root.is_some() + && manifest.aich_root_origin != Some(Ed2kAichRootOrigin::NetworkConsensus) + { return Ok(manifest); } - let promoted = { - let mut map = self + let mut accumulator = { + let map = self .aich_root_corroboration .lock() .expect("aich corroboration mutex poisoned"); - let accumulator = map.entry(file_hash.to_string()).or_default(); - for (aich_root, from_ip) in observations { - let std::net::IpAddr::V4(from_ip) = from_ip else { - continue; - }; - accumulator.record(*aich_root, from_ip.octets()); + // An accepted persisted consensus resumes VERIFIED exactly as + // stock. Only this runtime's live accumulator is reconsiderable. + if manifest.aich_root.is_some() && !map.contains_key(file_hash) { + return Ok(manifest); } - accumulator.trusted_root() + map.get(file_hash).cloned().unwrap_or_default() }; - if let Some(trusted) = promoted { - manifest.aich_root = Some(hex::encode(trusted)); - self.store_manifest_unlocked(&manifest).await?; - self.upsert_verified_catalog_entry(&manifest).await; + for (aich_root, from_ip) in observations { + let std::net::IpAddr::V4(from_ip) = from_ip else { + continue; + }; + accumulator.record(*aich_root, from_ip.octets()); } + let promoted = accumulator.trusted_root(); + + let origin = promoted.map(|_| Ed2kAichRootOrigin::NetworkConsensus); + if super::aich_trust::set_identity(&mut manifest, promoted.map(hex::encode), origin)? { + let (manifest, _guard) = self + .store_aich_transition_owned(guard, manifest, Some(accumulator)) + .await?; + return Ok(manifest); + } + self.aich_root_corroboration + .lock() + .expect("aich corroboration mutex poisoned") + .insert(file_hash.to_owned(), accumulator); Ok(manifest) } @@ -1051,6 +1094,10 @@ impl Ed2kTransferRuntime { "retired download disappeared before finalization" ); self.manifest_cache.lock().await.remove(file_hash); + self.aich_root_corroboration + .lock() + .expect("aich corroboration mutex poisoned") + .remove(file_hash); self.manifest_checkpoint_state .lock() .await @@ -1168,6 +1215,10 @@ impl Ed2kTransferRuntime { "ED2K transfer {file_hash} disappeared before deletion intent was persisted" ); self.manifest_cache.lock().await.remove(&file_hash); + self.aich_root_corroboration + .lock() + .expect("aich corroboration mutex poisoned") + .remove(&file_hash); self.manifest_checkpoint_state .lock() .await @@ -1406,7 +1457,7 @@ impl Ed2kTransferRuntime { /// never represented as an active download job. pub async fn manifest(&self, file_hash: &str) -> Result { let _guard = self.lock_manifest(file_hash).await; - self.load_manifest_unlocked(file_hash).await + self.load_aich_manifest_unlocked(file_hash).await } /// Resolve a read-only protocol view for a download or eligible complete diff --git a/crates/emulebb-ed2k/src/ed2k_transfer/model.rs b/crates/emulebb-ed2k/src/ed2k_transfer/model.rs index 1fefdb2e..71807cc5 100644 --- a/crates/emulebb-ed2k/src/ed2k_transfer/model.rs +++ b/crates/emulebb-ed2k/src/ed2k_transfer/model.rs @@ -468,6 +468,37 @@ impl From<&Ed2kResumeManifest> for Ed2kFileServingView { } } +/// Provenance of an accepted AICH master, separate from acquired part hashes. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum Ed2kAichRootOrigin { + LocalComputed, + OperatorLink, + StockImport, + NetworkConsensus, +} + +impl Ed2kAichRootOrigin { + pub(super) const fn as_str(self) -> &'static str { + match self { + Self::LocalComputed => "local_computed", + Self::OperatorLink => "operator_link", + Self::StockImport => "stock_import", + Self::NetworkConsensus => "network_consensus", + } + } + + pub(super) fn parse(value: &str) -> anyhow::Result { + match value { + "local_computed" => Ok(Self::LocalComputed), + "operator_link" => Ok(Self::OperatorLink), + "stock_import" => Ok(Self::StockImport), + "network_consensus" => Ok(Self::NetworkConsensus), + _ => anyhow::bail!("invalid accepted AICH origin {value:?}"), + } + } +} + /// Durable resume metadata for a download's managed piece payload. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] @@ -512,6 +543,10 @@ pub struct Ed2kResumeManifest { pub aich_hashset_acquired: bool, /// Canonical AICH root in lowercase hex when known. pub aich_root: Option, + /// Why this master may authorize verification; acquisition is not a reason. + pub aich_root_origin: Option, + /// Invalidates responses across trust/master changes, including A -> B -> A. + pub aich_trust_generation: u64, /// Canonical AICH per-part hashes in lowercase hex. pub aich_hashset: Vec, /// Upload-safe verified ranges. @@ -787,6 +822,8 @@ impl Ed2kResumeManifest { md4_hashset: Vec::new(), aich_hashset_acquired: false, aich_root: None, + aich_root_origin: None, + aich_trust_generation: 0, aich_hashset: Vec::new(), verified_ranges: Vec::new(), pieces: (0..piece_count) diff --git a/crates/emulebb-ed2k/src/ed2k_transfer/salvage.rs b/crates/emulebb-ed2k/src/ed2k_transfer/salvage.rs index aaa582cb..1db5ffd9 100644 --- a/crates/emulebb-ed2k/src/ed2k_transfer/salvage.rs +++ b/crates/emulebb-ed2k/src/ed2k_transfer/salvage.rs @@ -60,7 +60,7 @@ impl Ed2kTransferRuntime { recovery_body: &[u8], ) -> Result> { let _guard = self.lock_manifest(file_hash).await; - let mut manifest = self.load_manifest_unlocked(file_hash).await?; + let mut manifest = self.load_aich_manifest_unlocked(file_hash).await?; // Need a trusted AICH root that matches the answer's master hash. let Some(root_hex) = manifest.aich_root.as_deref() else { diff --git a/crates/emulebb-ed2k/src/ed2k_transfer/shared_catalog.rs b/crates/emulebb-ed2k/src/ed2k_transfer/shared_catalog.rs index 24e1526a..96750df1 100644 --- a/crates/emulebb-ed2k/src/ed2k_transfer/shared_catalog.rs +++ b/crates/emulebb-ed2k/src/ed2k_transfer/shared_catalog.rs @@ -475,9 +475,34 @@ impl Ed2kTransferRuntime { Ed2kMediaMetadata::default() }; } - let mut entries = self.shared_catalog.write().await; - entries.upsert_verified(file_hash, new_entry); - self.note_shared_catalog_entries_changed(); + // The caller's progress may be newer than SQL while its AICH authority + // is older. Preserve membership/media/MD4 state, but do not advertise a + // root retired by local/import acceptance during the work above. + // Lock order: catalog -> servable-hash cache -> metadata connection. + // The identity-only callback acquires no further locks and does no I/O. + let mut entries = Arc::clone(&self.shared_catalog).write_owned().await; + let metadata = self.metadata.clone(); + let hash = manifest.file_hash.clone(); + let generation = Arc::clone(&self.shared_catalog_generation); + let hashes = Arc::clone(&self.servable_shared_hash_cache); + let published = tokio::task::spawn_blocking(move || { + let mut hashes = hashes.lock().expect("servable hash cache mutex poisoned"); + metadata.publish_transfer_aich_identity(&hash, |identity| { + if let Some(entry) = new_entry.as_mut() { + entry.aich_root = identity.root; + } + entries.upsert_verified(file_hash, new_entry); + generation.fetch_add(1, Ordering::AcqRel); + hashes.hashes = None; + Ok(()) + }) + }) + .await; + match published { + Ok(Ok(())) => {} + Ok(Err(error)) => tracing::warn!(%error, "failed to fence shared AICH identity"), + Err(error) => tracing::warn!(%error, "shared AICH publication task panicked"), + } } /// Remove a locally verified file from the live serving/advertisement diff --git a/crates/emulebb-ed2k/src/ed2k_transfer/store.rs b/crates/emulebb-ed2k/src/ed2k_transfer/store.rs index 63262260..84ee7947 100644 --- a/crates/emulebb-ed2k/src/ed2k_transfer/store.rs +++ b/crates/emulebb-ed2k/src/ed2k_transfer/store.rs @@ -321,11 +321,12 @@ impl Ed2kTransferRuntime { if local_view.is_some() { Ok(local_view) } else { - Ok(self - .load_manifest_optional_unlocked(file_hash) - .await? - .as_ref() - .map(super::Ed2kFileServingView::from)) + match self.load_manifest_optional_unlocked(file_hash).await? { + Some(_) => Ok(Some(super::Ed2kFileServingView::from( + &self.load_aich_manifest_unlocked(file_hash).await?, + ))), + None => Ok(None), + } } } @@ -338,6 +339,72 @@ impl Ed2kTransferRuntime { .with_context(|| format!("missing ED2K transfer metadata for {file_hash}")) } + /// Merge only durable AICH identity into cached progress. Accepted local or + /// import writers share SQLite's transaction fence, not this runtime lock. + /// Never reload pieces/ranges or discard newer cache-only payload progress. + pub(super) async fn load_aich_manifest_unlocked( + &self, + file_hash: &str, + ) -> Result { + let mut manifest = self.load_manifest_unlocked(file_hash).await?; + let metadata = self.metadata.clone(); + let hash = file_hash.to_owned(); + let identity = + tokio::task::spawn_blocking(move || metadata.transfer_aich_identity_by_hash(&hash)) + .await + .context("AICH identity refresh task panicked")?? + .with_context(|| format!("missing AICH transfer identity for {file_hash}"))?; + let origin = identity + .origin + .as_deref() + .map(super::Ed2kAichRootOrigin::parse) + .transpose()?; + if manifest.aich_root != identity.root + || manifest.aich_root_origin != origin + || manifest.aich_trust_generation != identity.generation + { + let metadata = self.metadata.clone(); + let hash = file_hash.to_owned(); + let parsed_hash = file_hash.parse::()?; + let mut cached = manifest.clone(); + let mut cache = Arc::clone(&self.manifest_cache).lock_owned().await; + let mut catalog = Arc::clone(&self.shared_catalog).write_owned().await; + let generation = Arc::clone(&self.shared_catalog_generation); + let corroboration = Arc::clone(&self.aich_root_corroboration); + // Same publication order as owned decisions, omitting checkpoint + // bookkeeping because identity refresh must preserve dirty pieces. + return tokio::task::spawn_blocking(move || { + let mut map = corroboration + .lock() + .expect("aich corroboration mutex poisoned"); + metadata.publish_transfer_aich_identity(&hash, |identity| { + super::aich_trust::merge_persisted_identity(&mut manifest, &identity)?; + super::aich_trust::merge_persisted_identity(&mut cached, &identity)?; + if manifest.aich_root_origin + != Some(super::Ed2kAichRootOrigin::NetworkConsensus) + { + map.remove(&hash); + } + cache.insert(hash.clone(), cached); + let mut changed = false; + catalog.update_by_hash(&parsed_hash, |entry| { + if entry.aich_root != manifest.aich_root { + entry.aich_root = manifest.aich_root.clone(); + changed = true; + } + }); + if changed { + generation.fetch_add(1, std::sync::atomic::Ordering::AcqRel); + } + Ok(manifest) + }) + }) + .await + .context("AICH identity publication task panicked")?; + } + Ok(manifest) + } + /// Run a read-only probe against the manifest WITHOUT cloning it on the /// (hot) cache-hit path; only a cache miss pays the full load. For /// per-block predicates like the salvage-mode check. @@ -374,13 +441,113 @@ impl Ed2kTransferRuntime { // `synchronous = FULL`): run it on the blocking pool so the fsync // never parks an async worker thread. let metadata = self.metadata.clone(); - tokio::task::spawn_blocking(move || metadata.upsert_transfer_manifest(&row)) - .await - .context("ED2K manifest upsert task panicked")??; - self.mark_manifest_persisted_unlocked(manifest).await; + let stored = + tokio::task::spawn_blocking(move || metadata.upsert_transfer_manifest_reconciled(&row)) + .await + .context("ED2K manifest upsert task panicked")??; + let manifest = manifest_from_metadata(stored)?; + self.mark_manifest_persisted_unlocked(&manifest).await; Ok(()) } + /// Once a trust decision starts committing, its per-transfer guard and + /// publication belong to this task, not to the cancelable network caller. + /// Votes are staged until the durable decision and cache agree. Other + /// manifest writers cannot pass the guard during detached completion. + pub(super) async fn store_aich_transition_owned( + &self, + guard: tokio::sync::OwnedMutexGuard<()>, + manifest: Ed2kResumeManifest, + votes: Option, + ) -> Result<(Ed2kResumeManifest, tokio::sync::OwnedMutexGuard<()>)> { + let row = manifest_to_metadata(&manifest)?; + let transfer_dir = self.transfer_dir(&manifest.file_hash); + let metadata = self.metadata.clone(); + let cache = Arc::clone(&self.manifest_cache); + let checkpoints = Arc::clone(&self.manifest_checkpoint_state); + let corroboration = Arc::clone(&self.aich_root_corroboration); + let catalog = Arc::clone(&self.shared_catalog); + let catalog_generation = Arc::clone(&self.shared_catalog_generation); + let parsed_hash = manifest.file_hash.parse::()?; + #[cfg(test)] + let barrier = self.aich_persist_barrier.lock().unwrap().take(); + tokio::spawn(async move { + let guard = guard; + let result: Result = async { + tokio::fs::create_dir_all(&transfer_dir).await?; + let writer = metadata.clone(); + let stored = tokio::task::spawn_blocking(move || { + writer.upsert_transfer_manifest_reconciled(&row) + }) + .await + .context("AICH manifest upsert task panicked")??; + let mut manifest = manifest_from_metadata(stored)?; + #[cfg(test)] + if let Some(barrier) = barrier { + barrier.committed.notify_one(); + barrier.release.notified().await; + } + let mut cached = manifest.clone(); + let progress = manifest_progress_bytes(&manifest); + // Lock order: manifest -> cache -> checkpoints -> catalog -> + // corroboration -> metadata connection. No callback below may + // await, do I/O, re-enter metadata or acquire another lock. + let mut cache = cache.lock_owned().await; + let mut checkpoints = checkpoints.lock_owned().await; + let mut catalog = catalog.write_owned().await; + let hash = manifest.file_hash.clone(); + tokio::task::spawn_blocking(move || { + let mut map = corroboration + .lock() + .expect("aich corroboration mutex poisoned"); + metadata.publish_transfer_aich_identity(&hash, |identity| { + super::aich_trust::merge_persisted_identity(&mut manifest, &identity)?; + super::aich_trust::merge_persisted_identity(&mut cached, &identity)?; + cache.insert(hash.clone(), cached); + checkpoints.insert( + hash.clone(), + Ed2kManifestCheckpointState { + persisted_bytes_written: progress, + last_persisted_at: Instant::now(), + dirty_piece_indexes: std::collections::BTreeSet::new(), + }, + ); + if let Some(votes) = votes + && matches!( + manifest.aich_root_origin, + None | Some(super::Ed2kAichRootOrigin::NetworkConsensus) + ) + { + map.insert(hash.clone(), votes); + } else { + map.remove(&hash); + } + let mut changed = false; + catalog.update_by_hash(&parsed_hash, |entry| { + if entry.aich_root != manifest.aich_root { + entry.aich_root = manifest.aich_root.clone(); + changed = true; + } + }); + if changed { + catalog_generation.fetch_add(1, std::sync::atomic::Ordering::AcqRel); + } + Ok(manifest) + }) + }) + .await + .context("AICH publication task panicked")? + } + .await; + if let Err(error) = &result { + tracing::error!(%error, "AICH decision persistence failed"); + } + result.map(|manifest| (manifest, guard)) + }) + .await + .context("AICH decision completion task panicked")? + } + /// Persist mid-piece progress WITHOUT rewriting the manifest's child /// tables — the batched download checkpoint. Persists `current_piece` /// plus every piece dirtied by cache-only appends since the last durable diff --git a/crates/emulebb-ed2k/src/ed2k_transfer/tests.rs b/crates/emulebb-ed2k/src/ed2k_transfer/tests.rs index 9230016c..2e06a030 100644 --- a/crates/emulebb-ed2k/src/ed2k_transfer/tests.rs +++ b/crates/emulebb-ed2k/src/ed2k_transfer/tests.rs @@ -901,6 +901,10 @@ async fn completed_manifest_preserves_remote_aich_identity_over_local_rebuild() .unwrap(), ], }; + runtime + .reconcile_aich_root(&job.file_hash, Some(remote_aich.master_hash)) + .await + .unwrap(); runtime .store_aich_hashset(&job.file_hash, remote_aich.clone()) .await diff --git a/crates/emulebb-ed2k/src/ed2k_transfer/tests/aich_trust_corroboration.rs b/crates/emulebb-ed2k/src/ed2k_transfer/tests/aich_trust_corroboration.rs index 973fb9e9..7add0e66 100644 --- a/crates/emulebb-ed2k/src/ed2k_transfer/tests/aich_trust_corroboration.rs +++ b/crates/emulebb-ed2k/src/ed2k_transfer/tests/aich_trust_corroboration.rs @@ -4,11 +4,12 @@ //! A network-learned root must NOT become salvage-authorizing until at least //! `MINUNIQUEIPS_TOTRUST` (10) distinct IPs have proposed it at //! `MINPERCENTAGE_TOTRUST` (92%) agreement. Authoritative roots (locally -//! computed / persisted) are trusted on sight. +//! computed / explicitly supplied) have separate authority. An accepted +//! persisted decision resumes VERIFIED as stock while retaining its origin. use std::net::IpAddr; -use super::super::{Ed2kTransferRuntime, new_transfer_job}; +use super::super::{Ed2kAichHashset, Ed2kAichRootOrigin, Ed2kTransferRuntime, new_transfer_job}; use crate::paths::unique_test_dir; use emulebb_kad_proto::Ed2kHash; @@ -25,6 +26,27 @@ async fn runtime_with_job() -> (Ed2kTransferRuntime, String) { (runtime, job.file_hash) } +async fn runtime_with_servable_job() -> (Ed2kTransferRuntime, String) { + use md4::{Digest, Md4}; + + let root = unique_test_dir("ed2k-aich-servable-corroboration"); + let runtime = Ed2kTransferRuntime::load_or_create(&root).unwrap(); + let mut part_hashes = vec![[0x31; 16]; 4]; + part_hashes.push(Md4::digest(b"").into()); + let file_hash = Ed2kHash::from_bytes(Md4::digest(part_hashes.concat()).into()); + let job = new_transfer_job( + file_hash, + "servable-corroboration.bin".to_owned(), + 4 * super::super::ED2K_PART_SIZE, + ); + runtime.ensure_job(&job).await.unwrap(); + runtime + .store_md4_hashset(&job.file_hash, part_hashes) + .await + .unwrap(); + (runtime, job.file_hash) +} + #[tokio::test] async fn single_peer_root_is_not_trusted() { let (runtime, file_hash) = runtime_with_job().await; @@ -41,6 +63,709 @@ async fn single_peer_root_is_not_trusted() { assert!(reloaded.aich_root.is_none()); } +fn consistent_candidate() -> Ed2kAichHashset { + let part_hashes = vec![[0x21; 20], [0x43; 20], [0x65; 20], [0x87; 20]]; + let master_hash = super::super::hashset::reconstruct_aich_root_from_part_hashes( + 4 * super::super::ED2K_PART_SIZE, + &part_hashes, + ) + .unwrap(); + Ed2kAichHashset { + master_hash, + part_hashes, + } +} + +fn accept_independent_root( + runtime: &Ed2kTransferRuntime, + manifest: &super::super::Ed2kResumeManifest, + root: [u8; 20], + stock: bool, +) { + let fingerprint = emulebb_metadata::NativeFileFingerprint { + identity: emulebb_metadata::NativeFileIdentity::Unix { + device_id: 1, + inode_number: 2, + }, + size_bytes: manifest.file_size, + modified: emulebb_metadata::NativeFileTimestamp::new(0, 0).unwrap(), + changed: None, + }; + let path = runtime.root_dir.join("accepted-local.bin"); + if stock { + runtime + .metadata + .accept_imported_local_files(&[( + emulebb_metadata::MetadataFileImportRecord { + file_hash: manifest.file_hash.clone(), + display_name: "accepted-stock.bin".to_owned(), + file_size: manifest.file_size, + modified_s: 0, + md4_hashset: Vec::new(), + aich_root: Some(hex::encode(root)), + aich_hashset: Vec::new(), + upload_priority: "normal".to_owned(), + auto_upload_priority: false, + all_time_uploaded_bytes: 0, + all_time_upload_requests: 0, + all_time_upload_accepts: 0, + last_upload_request_ms: 0, + }, + path, + fingerprint, + )]) + .unwrap(); + } else { + runtime + .metadata + .upsert_verified_local_content(&emulebb_metadata::MetadataVerifiedLocalContent { + file_hash: manifest.file_hash.clone(), + display_name: "accepted-local.bin".to_owned(), + file_size: manifest.file_size, + part_size: manifest.piece_size, + md4_hashset: Vec::new(), + aich_root: Some(hex::encode(root)), + aich_hashset: Vec::new(), + path, + fingerprint, + upload_priority: "normal".to_owned(), + auto_upload_priority: false, + comment: String::new(), + rating: 0, + }) + .unwrap(); + } +} + +#[tokio::test] +async fn accepted_authority_retires_old_parts_without_another_vote_and_after_restart() { + for stock in [false, true] { + let (runtime, file_hash) = runtime_with_job().await; + let old = consistent_candidate(); + let observations = (0..10) + .map(|signer| (old.master_hash, ip(203, signer, 0, 1))) + .collect::>(); + runtime + .record_network_aich_observations(&file_hash, &observations) + .await + .unwrap(); + let mut stale = runtime + .store_aich_hashset(&file_hash, old.clone()) + .await + .unwrap(); + // Simulate a newer cache-only append: identity refresh must not reload + // old SQL piece rows over independently received payload progress. + stale.pieces[0].bytes_written = 123; + runtime.cache_manifest_unlocked(&stale).await; + let mut replacement = consistent_candidate(); + replacement.part_hashes[0] = [0x91; 20]; + replacement.master_hash = super::super::hashset::reconstruct_aich_root_from_part_hashes( + stale.file_size, + &replacement.part_hashes, + ) + .unwrap(); + accept_independent_root(&runtime, &stale, replacement.master_hash, stock); + + // Verify SQL before any runtime refresh or new peer observation. Both + // cold loaders must see one coherent accepted decision. + let persisted = runtime + .metadata + .transfer_manifest_by_hash(&file_hash) + .unwrap() + .unwrap(); + let listed = runtime.metadata.incomplete_transfer_manifests().unwrap(); + assert_eq!(listed[0].aich_root, persisted.aich_root); + assert_eq!( + persisted.aich_root, + Some(hex::encode(replacement.master_hash)) + ); + assert!(persisted.aich_trust_generation > stale.aich_trust_generation); + assert!(!persisted.aich_hashset_acquired); + assert!(persisted.aich_hashset.is_empty()); + let current = runtime.manifest(&file_hash).await.unwrap(); + assert_eq!(current.pieces[0].bytes_written, 123); + assert_eq!(current.aich_root, persisted.aich_root); + assert!(!current.aich_hashset_acquired); + let rejected = runtime + .store_aich_hashset_for_generation(&file_hash, old, stale.aich_trust_generation) + .await + .unwrap(); + assert!(!rejected.aich_hashset_acquired); + + // A delayed full checkpoint also cannot restore its old root, origin, + // generation or acquired set. Its independent piece progress survives. + let guard = runtime.lock_manifest(&file_hash).await; + runtime.store_manifest_unlocked(&stale).await.unwrap(); + drop(guard); + let directory = runtime.root_dir.clone(); + drop(runtime); + let resumed = Ed2kTransferRuntime::load_or_create(&directory).unwrap(); + let cold = resumed.manifest(&file_hash).await.unwrap(); + assert_eq!(cold.aich_root, persisted.aich_root); + assert_eq!(cold.aich_trust_generation, persisted.aich_trust_generation); + assert_eq!(cold.pieces[0].bytes_written, 123); + assert!(!cold.aich_hashset_acquired); + assert_eq!( + cold.aich_root_origin, + Some(if stock { + Ed2kAichRootOrigin::StockImport + } else { + Ed2kAichRootOrigin::LocalComputed + }) + ); + let acquired = resumed + .store_aich_hashset_for_generation(&file_hash, replacement, cold.aich_trust_generation) + .await + .unwrap(); + assert!(acquired.aich_hashset_acquired); + } +} + +#[tokio::test] +async fn internally_consistent_peer_hashset_does_not_establish_trust() { + let (runtime, file_hash) = runtime_with_job().await; + let candidate = consistent_candidate(); + runtime + .record_network_aich_root(&file_hash, Some(candidate.master_hash), ip(203, 0, 113, 5)) + .await + .unwrap(); + let error = runtime + .store_aich_hashset(&file_hash, candidate) + .await + .unwrap_err(); + assert!(error.to_string().contains("independently trusted master")); + let manifest = runtime.manifest(&file_hash).await.unwrap(); + assert!(manifest.aich_root.is_none()); + assert!(manifest.aich_hashset.is_empty()); + assert!(!manifest.aich_hashset_acquired); +} + +#[tokio::test] +async fn consistent_part_set_activates_only_after_independent_corroboration() { + let (runtime, file_hash) = runtime_with_job().await; + let candidate = consistent_candidate(); + for signer in 0..9 { + runtime + .record_network_aich_root( + &file_hash, + Some(candidate.master_hash), + ip(203, signer, 0, 1), + ) + .await + .unwrap(); + } + assert!( + runtime + .store_aich_hashset(&file_hash, candidate.clone()) + .await + .is_err() + ); + runtime + .record_network_aich_root(&file_hash, Some(candidate.master_hash), ip(203, 9, 0, 1)) + .await + .unwrap(); + let manifest = runtime + .store_aich_hashset(&file_hash, candidate.clone()) + .await + .unwrap(); + assert!(manifest.aich_hashset_acquired); + assert_eq!(manifest.aich_root, Some(hex::encode(candidate.master_hash))); + assert_eq!(manifest.aich_hashset.len(), 4); +} + +#[tokio::test] +async fn conflicting_consistent_hashset_cannot_replace_authoritative_identity() { + let (runtime, file_hash) = runtime_with_job().await; + let authoritative = [0x99; 20]; + runtime + .reconcile_aich_root(&file_hash, Some(authoritative)) + .await + .unwrap(); + assert!( + runtime + .store_aich_hashset(&file_hash, consistent_candidate()) + .await + .is_err() + ); + let manifest = runtime.manifest(&file_hash).await.unwrap(); + assert_eq!(manifest.aich_root, Some(hex::encode(authoritative))); + assert!(!manifest.aich_hashset_acquired); +} + +#[tokio::test] +async fn live_consensus_demotion_clears_part_set_and_rejects_aba_answer() { + let (runtime, file_hash) = runtime_with_job().await; + let candidate = consistent_candidate(); + let observations = (0..10) + .map(|signer| (candidate.master_hash, ip(203, signer, 0, 1))) + .collect::>(); + let accepted = runtime + .record_network_aich_observations(&file_hash, &observations) + .await + .unwrap(); + assert_eq!( + accepted.aich_root_origin, + Some(Ed2kAichRootOrigin::NetworkConsensus) + ); + runtime + .store_aich_hashset(&file_hash, candidate.clone()) + .await + .unwrap(); + let demoted = runtime + .record_network_aich_root(&file_hash, Some([0xff; 20]), ip(198, 0, 0, 1)) + .await + .unwrap(); + assert!(demoted.aich_root.is_none()); + assert!(demoted.aich_hashset.is_empty()); + assert!(!demoted.aich_hashset_acquired); + assert!(demoted.aich_trust_generation > accepted.aich_trust_generation); + + let observations = [10, 11].map(|signer| (candidate.master_hash, ip(203, signer, 0, 1))); + let renewed = runtime + .record_network_aich_observations(&file_hash, &observations) + .await + .unwrap(); + assert_eq!(renewed.aich_root, accepted.aich_root); + assert!(renewed.aich_trust_generation > demoted.aich_trust_generation); + let stale = runtime + .store_aich_hashset_for_generation( + &file_hash, + candidate.clone(), + accepted.aich_trust_generation, + ) + .await + .unwrap(); + assert!(!stale.aich_hashset_acquired); + let current = runtime + .store_aich_hashset_for_generation(&file_hash, candidate, renewed.aich_trust_generation) + .await + .unwrap(); + assert!(current.aich_hashset_acquired); +} + +#[tokio::test] +async fn accepted_consensus_resumes_verified_without_erasing_network_origin() { + let (runtime, file_hash) = runtime_with_job().await; + let root = [0x42; 20]; + let observations = (0..10) + .map(|signer| (root, ip(203, signer, 0, 1))) + .collect::>(); + let accepted = runtime + .record_network_aich_observations(&file_hash, &observations) + .await + .unwrap(); + let directory = runtime.root_dir.clone(); + drop(runtime); + let resumed = Ed2kTransferRuntime::load_or_create(&directory).unwrap(); + let observations = (0..20) + .map(|signer| ([0xff; 20], ip(198, signer, 0, 1))) + .collect::>(); + let manifest = resumed + .record_network_aich_observations(&file_hash, &observations) + .await + .unwrap(); + assert_eq!(manifest.aich_root, accepted.aich_root); + assert_eq!( + manifest.aich_trust_generation, + accepted.aich_trust_generation + ); + assert_eq!( + manifest.aich_root_origin, + Some(Ed2kAichRootOrigin::NetworkConsensus) + ); +} + +#[tokio::test] +async fn canceled_acceptance_finishes_publication_before_conflicting_vote() { + use std::{sync::Arc, time::Duration}; + + let (runtime, file_hash) = runtime_with_job().await; + let runtime = Arc::new(runtime); + let root = [0x42; 20]; + let observations = (0..9) + .map(|signer| (root, ip(203, signer, 0, 1))) + .collect::>(); + runtime + .record_network_aich_observations(&file_hash, &observations) + .await + .unwrap(); + let barrier = Arc::new(super::super::aich_trust::AichPersistBarrier::default()); + *runtime.aich_persist_barrier.lock().unwrap() = Some(Arc::clone(&barrier)); + let caller = { + let runtime = Arc::clone(&runtime); + let hash = file_hash.clone(); + tokio::spawn(async move { + runtime + .record_network_aich_root(&hash, Some(root), ip(203, 9, 0, 1)) + .await + }) + }; + tokio::time::timeout(Duration::from_secs(5), barrier.committed.notified()) + .await + .unwrap(); + // SQL has committed but publication is deliberately paused. Aborting the + // caller must not release the manifest gate or abandon the staged votes. + caller.abort(); + assert!(caller.await.unwrap_err().is_cancelled()); + assert!( + runtime.manifest_locks.lock().unwrap()[&file_hash] + .try_lock() + .is_err() + ); + assert!( + runtime.manifest_cache.lock().await[&file_hash] + .aich_root + .is_none() + ); + let disagreeing = { + let runtime = Arc::clone(&runtime); + let hash = file_hash.clone(); + tokio::spawn(async move { + runtime + .record_network_aich_root(&hash, Some([0xff; 20]), ip(198, 0, 0, 1)) + .await + }) + }; + barrier.release.notify_one(); + let demoted = tokio::time::timeout(Duration::from_secs(5), disagreeing) + .await + .unwrap() + .unwrap() + .unwrap(); + assert!(demoted.aich_root.is_none()); + assert_eq!(demoted.aich_trust_generation, 2); + let persisted = runtime + .metadata + .transfer_manifest_by_hash(&file_hash) + .unwrap() + .unwrap(); + assert_eq!(persisted.aich_root, demoted.aich_root); + assert_eq!( + persisted.aich_trust_generation, + demoted.aich_trust_generation + ); + let directory = runtime.root_dir.clone(); + drop(runtime); + let resumed = Ed2kTransferRuntime::load_or_create(&directory).unwrap(); + assert!( + resumed + .manifest(&file_hash) + .await + .unwrap() + .aich_root + .is_none() + ); +} + +async fn seed_servable_catalog( + runtime: &Ed2kTransferRuntime, + manifest: &super::super::Ed2kResumeManifest, +) { + assert!(manifest.md4_hashset_acquired); + let mut servable = manifest.clone(); + servable.pieces[0].state = super::super::Ed2kTransferState::Verified; + servable.pieces[0].bytes_written = super::super::ED2K_PART_SIZE; + servable.pieces[0].block_bitmap = None; + super::super::manifest::rebuild_verified_ranges(&mut servable); + let entry = super::super::Ed2kSharedEntry::from_manifest(&servable); + assert!(entry.is_servable()); + // The catalog must represent actual durable/cache MD4 authority, not a + // catalog-only entry that a valid later manifest refresh must remove. + let guard = runtime.lock_manifest(&manifest.file_hash).await; + runtime.store_manifest_unlocked(&servable).await.unwrap(); + drop(guard); + runtime + .upsert_verified_catalog_entry_cached(&servable) + .await; + assert!( + runtime + .shared_catalog + .read() + .await + .entry_by_hash(&manifest.file_hash.parse::().unwrap()) + .unwrap() + .is_servable() + ); +} + +#[tokio::test] +async fn canceled_demotion_retires_advertised_root_before_releasing_guard() { + use std::{sync::Arc, time::Duration}; + + let (runtime, file_hash) = runtime_with_servable_job().await; + let runtime = Arc::new(runtime); + let observations = (0..10) + .map(|signer| ([0x42; 20], ip(203, signer, 0, 1))) + .collect::>(); + let accepted = runtime + .record_network_aich_observations(&file_hash, &observations) + .await + .unwrap(); + seed_servable_catalog(&runtime, &accepted).await; + let barrier = Arc::new(super::super::aich_trust::AichPersistBarrier::default()); + *runtime.aich_persist_barrier.lock().unwrap() = Some(Arc::clone(&barrier)); + let caller = { + let runtime = Arc::clone(&runtime); + let hash = file_hash.clone(); + tokio::spawn(async move { + runtime + .record_network_aich_root(&hash, Some([0xff; 20]), ip(198, 0, 0, 1)) + .await + }) + }; + tokio::time::timeout(Duration::from_secs(10), barrier.committed.notified()) + .await + .unwrap(); + caller.abort(); + assert!(caller.await.unwrap_err().is_cancelled()); + barrier.release.notify_one(); + let guard = tokio::time::timeout(Duration::from_secs(10), runtime.lock_manifest(&file_hash)) + .await + .unwrap(); + assert!( + runtime.manifest_cache.lock().await[&file_hash] + .aich_root + .is_none() + ); + let catalog = runtime.shared_catalog.read().await; + let entry = catalog + .entry_by_hash(&file_hash.parse::().unwrap()) + .unwrap(); + assert!( + entry.is_servable(), + "MD4-verified availability must survive AICH demotion" + ); + assert!( + entry.aich_root.is_none(), + "a canceled caller must not leave retired authority advertised" + ); + drop(catalog); + drop(guard); +} + +#[tokio::test] +async fn authority_committed_before_publication_fences_older_task_and_catalog() { + use std::{sync::Arc, time::Duration}; + + for stock in [false, true] { + let (runtime, file_hash) = runtime_with_servable_job().await; + let runtime = Arc::new(runtime); + let observations = (0..9) + .map(|signer| ([0x42; 20], ip(203, signer, 0, 1))) + .collect::>(); + let previous = runtime + .record_network_aich_observations(&file_hash, &observations) + .await + .unwrap(); + seed_servable_catalog(&runtime, &previous).await; + let barrier = Arc::new(super::super::aich_trust::AichPersistBarrier::default()); + *runtime.aich_persist_barrier.lock().unwrap() = Some(Arc::clone(&barrier)); + let caller = { + let runtime = Arc::clone(&runtime); + let hash = file_hash.clone(); + tokio::spawn(async move { + runtime + .record_network_aich_root(&hash, Some([0x42; 20]), ip(203, 9, 0, 1)) + .await + }) + }; + tokio::time::timeout(Duration::from_secs(10), barrier.committed.notified()) + .await + .unwrap(); + let stale_projection = super::super::transfer_sql::manifest_from_metadata( + runtime + .metadata + .transfer_manifest_by_hash(&file_hash) + .unwrap() + .unwrap(), + ) + .unwrap(); + assert!(stale_projection.md4_hashset_acquired); + assert_eq!( + stale_projection.verified_ranges, + vec![super::super::Ed2kSharedRange { + start: 0, + end: super::super::ED2K_PART_SIZE, + }] + ); + // A different accepted authority commits after the older task's SQL + // write, while its cache/catalog publication is deliberately paused. + accept_independent_root(&runtime, &previous, [0x77; 20], stock); + let canonical = runtime + .metadata + .transfer_aich_identity_by_hash(&file_hash) + .unwrap() + .unwrap(); + barrier.release.notify_one(); + let published = tokio::time::timeout(Duration::from_secs(10), caller) + .await + .unwrap() + .unwrap() + .unwrap(); + assert_eq!(published.aich_root, canonical.root); + assert_eq!(published.aich_trust_generation, canonical.generation); + assert_eq!( + runtime.manifest_cache.lock().await[&file_hash].aich_root, + canonical.root + ); + let catalog = runtime.shared_catalog.read().await; + let entry = catalog + .entry_by_hash(&file_hash.parse::().unwrap()) + .unwrap(); + assert!(entry.is_servable()); + assert_eq!(entry.aich_root, canonical.root); + assert!( + !runtime + .aich_root_corroboration + .lock() + .unwrap() + .contains_key(&file_hash) + ); + drop(catalog); + // A later generic refresh retaining the old manifest must preserve its + // servable MD4 ranges but cannot republish the retired consensus root. + runtime + .upsert_verified_catalog_entry_cached(&stale_projection) + .await; + let catalog = runtime.shared_catalog.read().await; + let entry = catalog + .entry_by_hash(&file_hash.parse::().unwrap()) + .unwrap(); + assert!(entry.is_servable()); + assert_eq!(entry.aich_root, canonical.root); + } +} + +#[tokio::test] +async fn operator_identity_supersedes_live_consensus_without_peer_veto() { + let (runtime, file_hash) = runtime_with_job().await; + let candidate = consistent_candidate(); + let observations = (0..10) + .map(|signer| (candidate.master_hash, ip(203, signer, 0, 1))) + .collect::>(); + runtime + .record_network_aich_observations(&file_hash, &observations) + .await + .unwrap(); + runtime + .store_aich_hashset(&file_hash, candidate) + .await + .unwrap(); + let root = [0x77; 20]; + let manifest = runtime + .reconcile_aich_root(&file_hash, Some(root)) + .await + .unwrap(); + assert_eq!( + manifest.aich_root_origin, + Some(Ed2kAichRootOrigin::OperatorLink) + ); + assert!(manifest.aich_hashset.is_empty()); + assert!( + !runtime + .aich_root_corroboration + .lock() + .unwrap() + .contains_key(&file_hash) + ); + let observations = (0..20) + .map(|signer| ([0xff; 20], ip(198, signer, 0, 1))) + .collect::>(); + let manifest = runtime + .record_network_aich_observations(&file_hash, &observations) + .await + .unwrap(); + assert_eq!(manifest.aich_root, Some(hex::encode(root))); +} + +#[tokio::test] +async fn deleting_transfer_releases_its_bounded_votes() { + let (runtime, file_hash) = runtime_with_job().await; + runtime + .record_network_aich_root(&file_hash, Some([1; 20]), ip(203, 0, 0, 1)) + .await + .unwrap(); + assert!( + runtime + .aich_root_corroboration + .lock() + .unwrap() + .contains_key(&file_hash) + ); + assert!(runtime.delete_transfer_files(&file_hash).await.unwrap()); + assert!( + !runtime + .aich_root_corroboration + .lock() + .unwrap() + .contains_key(&file_hash) + ); +} + +#[tokio::test] +async fn new_local_authority_replaces_cached_consensus_before_more_network_votes() { + let (runtime, file_hash) = runtime_with_job().await; + let candidate = consistent_candidate(); + let observations = (0..10) + .map(|signer| (candidate.master_hash, ip(203, signer, 0, 1))) + .collect::>(); + let accepted = runtime + .record_network_aich_observations(&file_hash, &observations) + .await + .unwrap(); + runtime + .store_aich_hashset(&file_hash, candidate) + .await + .unwrap(); + let root = [0x77; 20]; + // Model independently accepted local evidence. No payload file is needed: + // metadata admission never rehashes bytes supplied by its trusted caller. + runtime + .metadata + .upsert_verified_local_content(&emulebb_metadata::MetadataVerifiedLocalContent { + file_hash: file_hash.clone(), + display_name: "accepted-local.bin".to_owned(), + file_size: accepted.file_size, + part_size: accepted.piece_size, + md4_hashset: Vec::new(), + aich_root: Some(hex::encode(root)), + aich_hashset: Vec::new(), + path: runtime.root_dir.join("accepted-local.bin"), + fingerprint: emulebb_metadata::NativeFileFingerprint { + identity: emulebb_metadata::NativeFileIdentity::Unix { + device_id: 1, + inode_number: 2, + }, + size_bytes: accepted.file_size, + modified: emulebb_metadata::NativeFileTimestamp::new(0, 0).unwrap(), + changed: None, + }, + upload_priority: "normal".to_owned(), + auto_upload_priority: false, + comment: String::new(), + rating: 0, + }) + .unwrap(); + let manifest = runtime + .record_network_aich_root(&file_hash, Some([0xff; 20]), ip(198, 0, 0, 1)) + .await + .unwrap(); + assert_eq!(manifest.aich_root, Some(hex::encode(root))); + assert_eq!( + manifest.aich_root_origin, + Some(Ed2kAichRootOrigin::LocalComputed) + ); + assert!(manifest.aich_trust_generation > accepted.aich_trust_generation); + assert!(!manifest.aich_hashset_acquired); + assert!( + !runtime + .aich_root_corroboration + .lock() + .unwrap() + .contains_key(&file_hash) + ); +} + #[tokio::test] async fn ten_unique_ip_majority_root_is_promoted() { let (runtime, file_hash) = runtime_with_job().await; diff --git a/crates/emulebb-ed2k/src/ed2k_transfer/tests/corruption_blackbox.rs b/crates/emulebb-ed2k/src/ed2k_transfer/tests/corruption_blackbox.rs index 09d65134..76e2eaa1 100644 --- a/crates/emulebb-ed2k/src/ed2k_transfer/tests/corruption_blackbox.rs +++ b/crates/emulebb-ed2k/src/ed2k_transfer/tests/corruption_blackbox.rs @@ -175,6 +175,10 @@ async fn salvage_verdicts_ban_only_the_corrupt_block_writer() { .store_md4_hashset(&job.file_hash, part_hashes) .await .unwrap(); + runtime + .reconcile_aich_root(&job.file_hash, Some(aich.master_hash)) + .await + .unwrap(); runtime .store_aich_hashset(&job.file_hash, aich.clone()) .await diff --git a/crates/emulebb-ed2k/src/ed2k_transfer/tests/file_status_parts.rs b/crates/emulebb-ed2k/src/ed2k_transfer/tests/file_status_parts.rs index 60cd078e..51e14858 100644 --- a/crates/emulebb-ed2k/src/ed2k_transfer/tests/file_status_parts.rs +++ b/crates/emulebb-ed2k/src/ed2k_transfer/tests/file_status_parts.rs @@ -59,6 +59,8 @@ fn manifest(file_size: u64, completed: bool, ranges: Vec) -> Ed md4_hashset: Vec::new(), aich_hashset_acquired: false, aich_root: None, + aich_root_origin: None, + aich_trust_generation: 0, aich_hashset: Vec::new(), verified_ranges: ranges, pieces, diff --git a/crates/emulebb-ed2k/src/ed2k_transfer/tests/ich_salvage.rs b/crates/emulebb-ed2k/src/ed2k_transfer/tests/ich_salvage.rs index 0531b553..00ab9a9c 100644 --- a/crates/emulebb-ed2k/src/ed2k_transfer/tests/ich_salvage.rs +++ b/crates/emulebb-ed2k/src/ed2k_transfer/tests/ich_salvage.rs @@ -279,6 +279,10 @@ async fn ich_rehash_miss_keeps_aich_salvage_bitmap() { fs::write(&path, &payload).unwrap(); super::super::build_aich_hashset_from_payload(&path, payload.len() as u64).unwrap() }; + runtime + .reconcile_aich_root(&job.file_hash, Some(aich.master_hash)) + .await + .unwrap(); runtime .store_aich_hashset(&job.file_hash, aich.clone()) .await diff --git a/crates/emulebb-ed2k/src/ed2k_transfer/tests/salvage.rs b/crates/emulebb-ed2k/src/ed2k_transfer/tests/salvage.rs index 2b02fd6a..101c9795 100644 --- a/crates/emulebb-ed2k/src/ed2k_transfer/tests/salvage.rs +++ b/crates/emulebb-ed2k/src/ed2k_transfer/tests/salvage.rs @@ -100,6 +100,10 @@ async fn salvage_keeps_good_blocks_and_reverifies_after_one_block() { .store_md4_hashset(&job.file_hash, md4_parts) .await .unwrap(); + runtime + .reconcile_aich_root(&job.file_hash, Some(aich.master_hash)) + .await + .unwrap(); runtime .store_aich_hashset(&job.file_hash, aich.clone()) .await @@ -211,6 +215,10 @@ async fn block_bitmap_persists_round_trip() { .store_md4_hashset(&job.file_hash, md4_parts) .await .unwrap(); + runtime + .reconcile_aich_root(&job.file_hash, Some(aich.master_hash)) + .await + .unwrap(); runtime .store_aich_hashset(&job.file_hash, aich.clone()) .await diff --git a/crates/emulebb-ed2k/src/ed2k_transfer/transfer_sql.rs b/crates/emulebb-ed2k/src/ed2k_transfer/transfer_sql.rs index 7d7b4c03..17d8c428 100644 --- a/crates/emulebb-ed2k/src/ed2k_transfer/transfer_sql.rs +++ b/crates/emulebb-ed2k/src/ed2k_transfer/transfer_sql.rs @@ -47,6 +47,10 @@ pub(super) fn manifest_to_metadata( md4_hashset: manifest.md4_hashset.clone(), aich_hashset_acquired: manifest.aich_hashset_acquired, aich_root: manifest.aich_root.clone(), + aich_root_origin: manifest + .aich_root_origin + .map(|origin| origin.as_str().to_owned()), + aich_trust_generation: manifest.aich_trust_generation, aich_hashset: manifest.aich_hashset.clone(), verified_ranges: manifest .verified_ranges @@ -148,6 +152,12 @@ pub(super) fn manifest_from_metadata( md4_hashset: manifest.md4_hashset, aich_hashset_acquired: manifest.aich_hashset_acquired, aich_root: manifest.aich_root, + aich_root_origin: manifest + .aich_root_origin + .as_deref() + .map(super::Ed2kAichRootOrigin::parse) + .transpose()?, + aich_trust_generation: manifest.aich_trust_generation, aich_hashset: manifest.aich_hashset, verified_ranges: manifest .verified_ranges @@ -390,6 +400,8 @@ mod tests { md4_hashset: Vec::new(), aich_hashset_acquired: false, aich_root: None, + aich_root_origin: None, + aich_trust_generation: 0, aich_hashset: Vec::new(), verified_ranges: Vec::new(), pieces: Vec::new(), diff --git a/crates/emulebb-metadata/src/content_identity.rs b/crates/emulebb-metadata/src/content_identity.rs index cfa384ce..27dd238e 100644 --- a/crates/emulebb-metadata/src/content_identity.rs +++ b/crates/emulebb-metadata/src/content_identity.rs @@ -21,7 +21,8 @@ pub(crate) fn check_accepted_identity( FROM known_files f LEFT JOIN verified_local_content c ON c.known_file_id = f.id WHERE f.ed2k_hash = ?1 - AND (c.known_file_id IS NOT NULL + AND (f.aich_root_origin IS NOT NULL + OR c.known_file_id IS NOT NULL OR EXISTS (SELECT 1 FROM transfers t WHERE t.known_file_id = f.id AND t.completed = 1 AND t.final_rehash_pending = 0))", )?; diff --git a/crates/emulebb-metadata/src/download_history_store.rs b/crates/emulebb-metadata/src/download_history_store.rs index 3a2f73a9..df000354 100644 --- a/crates/emulebb-metadata/src/download_history_store.rs +++ b/crates/emulebb-metadata/src/download_history_store.rs @@ -180,6 +180,8 @@ mod tests { md4_hashset: Vec::new(), aich_hashset_acquired: true, aich_root: Some("1111111111111111111111111111111111111111".to_string()), + aich_root_origin: Some("local_computed".to_string()), + aich_trust_generation: 1, aich_hashset: Vec::new(), verified_ranges: vec![MetadataTransferRange { start: 0, @@ -284,6 +286,8 @@ mod tests { manifest.display_name = "renamed.bin".to_string(); manifest.aich_root = None; + manifest.aich_root_origin = None; + manifest.aich_hashset_acquired = false; store.upsert_transfer_manifest(&manifest).unwrap(); let metadata_refresh = store .download_history_by_hash(&manifest.file_hash) diff --git a/crates/emulebb-metadata/src/lib.rs b/crates/emulebb-metadata/src/lib.rs index 12154ff8..05bd4e14 100644 --- a/crates/emulebb-metadata/src/lib.rs +++ b/crates/emulebb-metadata/src/lib.rs @@ -58,8 +58,8 @@ pub use search_store::normalized_search_query; pub use store::MetadataStore; pub use text::normalize_search_text; pub use transfer_model::{ - MetadataFileImportRecord, MetadataSharedSourceFailure, MetadataTransferCatalogEntry, - MetadataTransferCounts, MetadataTransferDeletionTombstone, + MetadataFileImportRecord, MetadataSharedSourceFailure, MetadataTransferAichIdentity, + MetadataTransferCatalogEntry, MetadataTransferCounts, MetadataTransferDeletionTombstone, MetadataTransferDeletionTombstonePhase, MetadataTransferDeliveryAttempt, MetadataTransferDeliveryAttemptPhase, MetadataTransferDeliveryOwnership, MetadataTransferDeliveryState, MetadataTransferDirectoryDeletionAuthority, diff --git a/crates/emulebb-metadata/src/local_file_store.rs b/crates/emulebb-metadata/src/local_file_store.rs index abc35877..8003bbb0 100644 --- a/crates/emulebb-metadata/src/local_file_store.rs +++ b/crates/emulebb-metadata/src/local_file_store.rs @@ -317,6 +317,21 @@ impl MetadataStore { content: &MetadataVerifiedLocalContent, now: i64, ) -> Result<()> { + Self::upsert_verified_local_content_with_origin_tx(tx, content, now, "local_computed") + } + + /// Stock cache acceptance shares geometry/fingerprint admission with native + /// hashing, but never pretends that Rust read and hashed the imported bytes. + pub(crate) fn upsert_verified_local_content_with_origin_tx( + tx: &rusqlite::Transaction<'_>, + content: &MetadataVerifiedLocalContent, + now: i64, + origin: &'static str, + ) -> Result<()> { + ensure!( + matches!(origin, "local_computed" | "stock_import"), + "invalid local AICH origin" + ); ensure!( content.path.is_absolute(), "local file path must be absolute" @@ -355,15 +370,19 @@ impl MetadataStore { .collect::>>()?; tx.execute( "INSERT INTO known_files( - ed2k_hash, size_bytes, display_name, aich_root, + ed2k_hash, size_bytes, display_name, aich_root, aich_root_origin, first_seen_at_ms, last_seen_at_ms, updated_at_ms - ) VALUES (?1, ?2, ?3, ?4, ?5, ?5, ?5) + ) VALUES (?1, ?2, ?3, ?4, ?6, ?5, ?5, ?5) ON CONFLICT(ed2k_hash) DO UPDATE SET size_bytes = excluded.size_bytes, aich_root = coalesce(excluded.aich_root, known_files.aich_root), + aich_root_origin = CASE + WHEN excluded.aich_root IS NULL THEN known_files.aich_root_origin + WHEN known_files.aich_root_origin = 'local_computed' THEN known_files.aich_root_origin + ELSE excluded.aich_root_origin END, last_seen_at_ms = excluded.last_seen_at_ms, updated_at_ms = excluded.updated_at_ms", - params![hash, size, content.display_name, aich_root, now], + params![hash, size, content.display_name, aich_root, now, aich_root.as_ref().map(|_| origin)], )?; let (known_file_id, stored_size): (i64, i64) = tx.query_row( "SELECT id, size_bytes FROM known_files WHERE ed2k_hash = ?1", @@ -374,6 +393,7 @@ impl MetadataStore { stored_size == size, MetadataContentConflict("ED2K identity has conflicting file size") ); + crate::transfer_store::reconcile_transfer_aich_authority_tx(tx, known_file_id)?; let inserted = tx.execute( "INSERT INTO verified_local_content( known_file_id, display_name, part_size_bytes, part_count, aich_root, diff --git a/crates/emulebb-metadata/src/profile_import.rs b/crates/emulebb-metadata/src/profile_import.rs index c26e1588..ce59e0b7 100644 --- a/crates/emulebb-metadata/src/profile_import.rs +++ b/crates/emulebb-metadata/src/profile_import.rs @@ -58,7 +58,7 @@ impl MetadataStore { let fingerprint = NativeFileFingerprint::from_storage_bytes(&binding.fingerprint) .ok_or_else(|| anyhow::anyhow!("invalid live fingerprint"))?; let tx = conn.transaction()?; - Self::upsert_verified_local_content_tx( + Self::upsert_verified_local_content_with_origin_tx( &tx, &MetadataVerifiedLocalContent { file_hash: entry.file_hash.clone(), @@ -76,6 +76,7 @@ impl MetadataStore { rating: 0, }, unix_ms(), + "stock_import", )?; tx.execute( "UPDATE known_files SET aich_root=coalesce(aich_root,?2) WHERE ed2k_hash=?1", @@ -134,7 +135,7 @@ impl MetadataStore { // WHY: stock has already hashed these bytes. Rebind its accepted // content identity to this platform's fingerprint instead of scheduling // a full payload read merely because Rust has not seen the file before. - Self::upsert_verified_local_content_tx( + Self::upsert_verified_local_content_with_origin_tx( &tx, &MetadataVerifiedLocalContent { file_hash: entry.file_hash.clone(), @@ -152,6 +153,7 @@ impl MetadataStore { rating: 0, }, now, + "stock_import", )?; if let Some(root) = &entry.aich_root { tx.execute( @@ -237,6 +239,14 @@ mod tests { store .accept_imported_local_files(&[(enriched.clone(), path.clone(), fingerprint.clone())]) .unwrap(); + let origin: String = store + .connection() + .unwrap() + .query_row("SELECT aich_root_origin FROM known_files", [], |row| { + row.get(0) + }) + .unwrap(); + assert_eq!(origin, "stock_import"); enriched.aich_root = Some("22".repeat(20)); assert!( store diff --git a/crates/emulebb-metadata/src/profile_store.rs b/crates/emulebb-metadata/src/profile_store.rs index e6dd3a1c..f4506a29 100644 --- a/crates/emulebb-metadata/src/profile_store.rs +++ b/crates/emulebb-metadata/src/profile_store.rs @@ -832,6 +832,8 @@ mod tests { md4_hashset: Vec::new(), aich_hashset_acquired: false, aich_root: None, + aich_root_origin: None, + aich_trust_generation: 0, aich_hashset: Vec::new(), verified_ranges: Vec::new(), upload_priority: "normal".to_string(), diff --git a/crates/emulebb-metadata/src/schema.rs b/crates/emulebb-metadata/src/schema.rs index a0fe3e28..ca607960 100644 --- a/crates/emulebb-metadata/src/schema.rs +++ b/crates/emulebb-metadata/src/schema.rs @@ -16,7 +16,7 @@ //! rejected: this module contains no migration or legacy interpretation. pub const SCHEMA_ID: &str = "emulebb.metadata"; -pub const SCHEMA_VERSION: i64 = 67; +pub const SCHEMA_VERSION: i64 = 68; pub const SCHEMA_SQL: &str = include_str!("schema.sql"); #[cfg(test)] diff --git a/crates/emulebb-metadata/src/schema.sql b/crates/emulebb-metadata/src/schema.sql index 65897d79..61dc83b9 100644 --- a/crates/emulebb-metadata/src/schema.sql +++ b/crates/emulebb-metadata/src/schema.sql @@ -67,6 +67,8 @@ CREATE TABLE known_files ( display_name TEXT NOT NULL, content_type TEXT NOT NULL DEFAULT '', aich_root BLOB CHECK(aich_root IS NULL OR length(aich_root) = 20), + aich_root_origin TEXT CHECK(aich_root_origin IN + ('local_computed', 'operator_link', 'stock_import')), media_artist TEXT NOT NULL DEFAULT '', media_album TEXT NOT NULL DEFAULT '', media_title TEXT NOT NULL DEFAULT '', @@ -84,7 +86,8 @@ CREATE TABLE known_files ( last_upload_requested_at_ms INTEGER NOT NULL DEFAULT 0 CHECK(last_upload_requested_at_ms >= 0), first_seen_at_ms INTEGER NOT NULL CHECK(first_seen_at_ms >= 0), last_seen_at_ms INTEGER NOT NULL CHECK(last_seen_at_ms >= 0), - updated_at_ms INTEGER NOT NULL CHECK(updated_at_ms >= 0) + updated_at_ms INTEGER NOT NULL CHECK(updated_at_ms >= 0), + CHECK((aich_root IS NULL) = (aich_root_origin IS NULL)) ) STRICT; CREATE TABLE known_file_origins ( @@ -323,6 +326,12 @@ CREATE TABLE transfers ( completed INTEGER NOT NULL DEFAULT 0 CHECK(completed IN (0, 1)), md4_hashset_acquired INTEGER NOT NULL DEFAULT 0 CHECK(md4_hashset_acquired IN (0, 1)), aich_hashset_acquired INTEGER NOT NULL DEFAULT 0 CHECK(aich_hashset_acquired IN (0, 1)), + -- A live consensus belongs to the transfer, never accepted known/local + -- content. Persist the decision and origin separately from mere acquisition. + aich_root BLOB CHECK(aich_root IS NULL OR length(aich_root) = 20), + aich_root_origin TEXT CHECK(aich_root_origin IN + ('local_computed', 'operator_link', 'stock_import', 'network_consensus')), + aich_trust_generation INTEGER NOT NULL DEFAULT 0 CHECK(aich_trust_generation >= 0), visible_state TEXT NOT NULL CHECK(visible_state IN ('completed', 'completing', 'downloading', 'queued')), -- Set after all parts verify and cleared only after the authoritative -- whole-file ED2K MD4 completion rehash succeeds or demotes bad parts. @@ -374,6 +383,8 @@ CREATE TABLE transfers ( OR (delivery_state = 'delivered' AND delivery_ownership = 'owned' AND delivered_path_id IS NOT NULL AND source_fingerprint IS NOT NULL) OR (delivery_state = 'delivered' AND delivery_ownership = 'unowned' AND delivered_path_id IS NOT NULL) ), + CHECK((aich_root IS NULL) = (aich_root_origin IS NULL)), + CHECK(aich_hashset_acquired = 0 OR aich_root IS NOT NULL), CHECK((completed = 1) = (visible_state = 'completed')), CHECK(payload_fingerprint IS NULL OR (completed = 1 AND final_rehash_pending = 0)), UNIQUE(known_file_id) diff --git a/crates/emulebb-metadata/src/schema/tests.rs b/crates/emulebb-metadata/src/schema/tests.rs index 33e17730..c034d509 100644 --- a/crates/emulebb-metadata/src/schema/tests.rs +++ b/crates/emulebb-metadata/src/schema/tests.rs @@ -100,6 +100,8 @@ fn schema_rejects_wrong_storage_types_null_keys_and_out_of_range_values() { let conn = store.connection().unwrap(); for sql in [ "INSERT INTO known_files(ed2k_hash,size_bytes,display_name,first_seen_at_ms,last_seen_at_ms,updated_at_ms) VALUES ('abcdefghijklmnop',1,'sample',0,0,0)", + "INSERT INTO known_files(ed2k_hash,size_bytes,display_name,aich_root,first_seen_at_ms,last_seen_at_ms,updated_at_ms) VALUES (zeroblob(16),1,'sample',zeroblob(20),0,0,0)", + "INSERT INTO known_files(ed2k_hash,size_bytes,display_name,aich_root,aich_root_origin,first_seen_at_ms,last_seen_at_ms,updated_at_ms) VALUES (zeroblob(16),1,'sample',zeroblob(20),'network_consensus',0,0,0)", "INSERT INTO known_files(ed2k_hash,size_bytes,display_name,first_seen_at_ms,last_seen_at_ms,updated_at_ms) VALUES (zeroblob(16),'invalid','sample',0,0,0)", "INSERT INTO download_history(ed2k_hash,size_bytes,display_name,first_completed_at_ms,last_completed_at_ms) VALUES (NULL,1,'sample',0,0)", "INSERT INTO known_files(ed2k_hash,size_bytes,display_name,media_length_seconds,first_seen_at_ms,last_seen_at_ms,updated_at_ms) VALUES (zeroblob(16),1,'sample',4294967296,0,0,0)", diff --git a/crates/emulebb-metadata/src/transfer_model.rs b/crates/emulebb-metadata/src/transfer_model.rs index c3b83a6d..7a123a84 100644 --- a/crates/emulebb-metadata/src/transfer_model.rs +++ b/crates/emulebb-metadata/src/transfer_model.rs @@ -61,6 +61,13 @@ pub struct MetadataTransferDirectoryDeletionAuthority { pub marker_id: Uuid, } +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct MetadataTransferAichIdentity { + pub root: Option, + pub origin: Option, + pub generation: u64, +} + #[derive(Debug, Clone, PartialEq, Eq)] pub struct MetadataTransferManifest { pub file_hash: String, @@ -76,6 +83,8 @@ pub struct MetadataTransferManifest { pub md4_hashset: Vec, pub aich_hashset_acquired: bool, pub aich_root: Option, + pub aich_root_origin: Option, + pub aich_trust_generation: u64, pub aich_hashset: Vec, pub verified_ranges: Vec, pub pieces: Vec, diff --git a/crates/emulebb-metadata/src/transfer_store.rs b/crates/emulebb-metadata/src/transfer_store.rs index a6cc7b6a..5d3e8965 100644 --- a/crates/emulebb-metadata/src/transfer_store.rs +++ b/crates/emulebb-metadata/src/transfer_store.rs @@ -10,21 +10,33 @@ use crate::{ native_path::{optional_path_from_row, path_from_row}, store::{bool_to_i64, decode_fixed_hex, unix_ms, upsert_local_path}, transfer_model::{ - MetadataSharedSourceFailure, MetadataTransferCatalogEntry, MetadataTransferCounts, - MetadataTransferDeletionTombstone, MetadataTransferDeletionTombstonePhase, - MetadataTransferDeliveryAttempt, MetadataTransferDeliveryAttemptPhase, - MetadataTransferDeliveryOwnership, MetadataTransferDeliveryState, - MetadataTransferDirectoryDeletionAuthority, MetadataTransferFileNameObservation, - MetadataTransferManifest, MetadataTransferMediaMetadata, MetadataTransferPiece, - MetadataTransferPublishEntry, MetadataTransferRange, MetadataTransferSource, + MetadataSharedSourceFailure, MetadataTransferAichIdentity, MetadataTransferCatalogEntry, + MetadataTransferCounts, MetadataTransferDeletionTombstone, + MetadataTransferDeletionTombstonePhase, MetadataTransferDeliveryAttempt, + MetadataTransferDeliveryAttemptPhase, MetadataTransferDeliveryOwnership, + MetadataTransferDeliveryState, MetadataTransferDirectoryDeletionAuthority, + MetadataTransferFileNameObservation, MetadataTransferManifest, + MetadataTransferMediaMetadata, MetadataTransferPiece, MetadataTransferPublishEntry, + MetadataTransferRange, MetadataTransferSource, }, }; impl super::MetadataStore { pub fn upsert_transfer_manifest(&self, manifest: &MetadataTransferManifest) -> Result<()> { + self.upsert_transfer_manifest_reconciled(manifest) + .map(|_| ()) + } + + /// Persist progress while retaining any newer accepted AICH authority. The + /// returned row is the exact manifest committed, including identity-only + /// reconciliation; callers must publish this row rather than their stale + /// input. Piece progress and MD4 verification are never reset here. + pub fn upsert_transfer_manifest_reconciled( + &self, + manifest: &MetadataTransferManifest, + ) -> Result { crate::transfer_validation::validate_manifest(manifest)?; let hash = decode_fixed_hex(&manifest.file_hash, 16, "ED2K hash")?; - let aich_root = optional_fixed_hex(manifest.aich_root.as_deref(), 20, "AICH root")?; let now = unix_ms(); let mut conn = self.connection()?; // WHY: this transaction reads deletion authority before writing paths and @@ -32,6 +44,16 @@ impl super::MetadataStore { // snapshot and fail its write upgrade with SQLITE_BUSY_SNAPSHOT after a // concurrent connection commits; acquire the writer slot before reading. let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + let mut reconciled = manifest.clone(); + reconcile_manifest_aich_identity(&tx, &hash, &mut reconciled)?; + let manifest = &reconciled; + let aich_root = optional_fixed_hex(manifest.aich_root.as_deref(), 20, "AICH root")?; + let accepted_aich_root = + if manifest.aich_root_origin.as_deref() == Some("network_consensus") { + None + } else { + aich_root.as_deref() + }; crate::content_identity::check_accepted_identity( &tx, &hash, @@ -112,17 +134,24 @@ impl super::MetadataStore { tx.prepare_cached( r#" - INSERT INTO known_files(ed2k_hash, size_bytes, display_name, aich_root, first_seen_at_ms, last_seen_at_ms, updated_at_ms) - VALUES (?1, ?2, ?3, ?4, ?5, ?5, ?5) + INSERT INTO known_files(ed2k_hash, size_bytes, display_name, aich_root, aich_root_origin, first_seen_at_ms, last_seen_at_ms, updated_at_ms) + VALUES (?1, ?2, ?3, ?4, ?6, ?5, ?5, ?5) ON CONFLICT(ed2k_hash) DO UPDATE SET size_bytes = excluded.size_bytes, display_name = CASE WHEN EXISTS(SELECT 1 FROM verified_local_content WHERE known_file_id = known_files.id) THEN known_files.display_name ELSE excluded.display_name END, aich_root = coalesce(excluded.aich_root, known_files.aich_root), + aich_root_origin = CASE + WHEN excluded.aich_root IS NULL THEN known_files.aich_root_origin + WHEN known_files.aich_root_origin = 'local_computed' THEN known_files.aich_root_origin + WHEN known_files.aich_root_origin = 'stock_import' + AND excluded.aich_root_origin = 'operator_link' THEN known_files.aich_root_origin + ELSE excluded.aich_root_origin END, last_seen_at_ms = excluded.last_seen_at_ms, updated_at_ms = excluded.updated_at_ms "#, - )?.execute(params![hash, i64::try_from(manifest.file_size)?, manifest.display_name, aich_root, now])?; + )?.execute(params![hash, i64::try_from(manifest.file_size)?, manifest.display_name, accepted_aich_root, now, + accepted_aich_root.and(manifest.aich_root_origin.as_deref())])?; let known_file_id: i64 = tx .prepare_cached("SELECT id FROM known_files WHERE ed2k_hash = ?1")? .query_row( @@ -154,13 +183,17 @@ impl super::MetadataStore { visible_state, final_rehash_pending, control_state, category_id, display_name, delivery_state, delivery_ownership, delivered_path_id, source_fingerprint, payload_fingerprint, created_at_ms, updated_at_ms, - completed_at_ms, removed_at_ms, part_size_bytes, part_count, upload_priority, auto_upload_priority, comment, rating + completed_at_ms, removed_at_ms, part_size_bytes, part_count, upload_priority, auto_upload_priority, comment, rating, + aich_root, aich_root_origin, aich_trust_generation ) - VALUES (?1, ?16, ?17, ?18, ?2, ?12, ?3, ?4, ?5, ?13, ?15, ?6, ?10, ?11, ?7, ?7, ?8, ?9, ?19, ?20, ?21, ?22, ?23, ?24) + VALUES (?1, ?16, ?17, ?18, ?2, ?12, ?3, ?4, ?5, ?13, ?15, ?6, ?10, ?11, ?7, ?7, ?8, ?9, ?19, ?20, ?21, ?22, ?23, ?24, ?25, ?26, ?27) ON CONFLICT(known_file_id) DO UPDATE SET completed = excluded.completed, md4_hashset_acquired = excluded.md4_hashset_acquired, aich_hashset_acquired = excluded.aich_hashset_acquired, + aich_root = excluded.aich_root, + aich_root_origin = excluded.aich_root_origin, + aich_trust_generation = excluded.aich_trust_generation, visible_state = excluded.visible_state, final_rehash_pending = excluded.final_rehash_pending, control_state = excluded.control_state, @@ -207,6 +240,7 @@ impl super::MetadataStore { bool_to_i64(manifest.aich_hashset_acquired), i64::try_from(manifest.piece_size)?, i64::try_from(manifest.pieces.len())?, manifest.upload_priority, manifest.auto_upload_priority, manifest.comment, i64::from(manifest.rating), + aich_root, manifest.aich_root_origin, i64::try_from(manifest.aich_trust_generation)?, ])?; let transfer_id: i64 = tx .prepare_cached("SELECT id FROM transfers WHERE known_file_id = ?1")? @@ -216,12 +250,40 @@ impl super::MetadataStore { &tx, manifest, &hash, - aich_root.as_deref(), + accepted_aich_root, completed_at_ms, completion_transition, )?; tx.commit()?; - Ok(()) + Ok(reconciled) + } + + /// Indexed identity-only read for cached transfer users. Do not load piece, + /// source, range or part-hash tables just to fence an AICH response. + pub fn transfer_aich_identity_by_hash( + &self, + file_hash: &str, + ) -> Result> { + let hash = decode_fixed_hex(file_hash, 16, "ED2K hash")?; + let conn = self.connection()?; + read_transfer_aich_identity(&conn, &hash) + } + + /// Publish a bounded in-memory AICH projection while holding the existing + /// connection mutex. Accepted identity writers cannot commit between this + /// read and publication. The callback must be synchronous, must not perform + /// I/O, and must not re-enter MetadataStore. Acquire runtime async guards + /// before invoking this helper on the blocking pool. + pub fn publish_transfer_aich_identity( + &self, + file_hash: &str, + publish: impl FnOnce(MetadataTransferAichIdentity) -> Result, + ) -> Result { + let hash = decode_fixed_hex(file_hash, 16, "ED2K hash")?; + let conn = self.connection()?; + let identity = read_transfer_aich_identity(&conn, &hash)? + .ok_or_else(|| anyhow::anyhow!("missing AICH transfer identity for {file_hash}"))?; + publish(identity) } pub fn transfer_manifest_by_hash( @@ -256,8 +318,8 @@ impl super::MetadataStore { transfers.completed, transfers.md4_hashset_acquired, transfers.aich_hashset_acquired, CASE - WHEN known_files.aich_root IS NULL THEN NULL - ELSE lower(hex(known_files.aich_root)) + WHEN transfers.aich_root IS NULL THEN NULL + ELSE lower(hex(transfers.aich_root)) END, transfers.upload_priority, transfers.auto_upload_priority, transfers.comment, transfers.rating, @@ -265,7 +327,9 @@ impl super::MetadataStore { delivered_paths.platform, delivered_paths.native_path, transfers.source_fingerprint, transfers.payload_fingerprint, transfers.final_rehash_pending, transfers.delivery_state, - transfers.delivery_ownership + transfers.delivery_ownership, + transfers.aich_root_origin, + transfers.aich_trust_generation FROM known_files JOIN transfers ON transfers.known_file_id = known_files.id LEFT JOIN local_paths delivered_paths ON delivered_paths.id = transfers.delivered_path_id @@ -287,6 +351,8 @@ impl super::MetadataStore { md4_hashset_acquired: row.get::<_, i64>(6)? != 0, aich_hashset_acquired: row.get::<_, i64>(7)? != 0, aich_root: row.get(8)?, + aich_root_origin: row.get(23)?, + aich_trust_generation: crate::sql_value::unsigned::(row, 24)?, upload_priority: row.get(9)?, auto_upload_priority: row.get::<_, i64>(10)? != 0, comment: row.get(11)?, @@ -495,8 +561,8 @@ impl super::MetadataStore { transfers.completed, transfers.md4_hashset_acquired, transfers.aich_hashset_acquired, CASE - WHEN known_files.aich_root IS NULL THEN NULL - ELSE lower(hex(known_files.aich_root)) + WHEN transfers.aich_root IS NULL THEN NULL + ELSE lower(hex(transfers.aich_root)) END, transfers.upload_priority, transfers.auto_upload_priority, transfers.comment, transfers.rating, @@ -504,7 +570,9 @@ impl super::MetadataStore { delivered_paths.platform, delivered_paths.native_path, transfers.source_fingerprint, transfers.payload_fingerprint, transfers.final_rehash_pending, transfers.delivery_state, - transfers.delivery_ownership + transfers.delivery_ownership, + transfers.aich_root_origin, + transfers.aich_trust_generation FROM known_files JOIN transfers ON transfers.known_file_id = known_files.id LEFT JOIN local_paths delivered_paths ON delivered_paths.id = transfers.delivered_path_id @@ -525,6 +593,8 @@ impl super::MetadataStore { md4_hashset_acquired: row.get::<_, i64>(6)? != 0, aich_hashset_acquired: row.get::<_, i64>(7)? != 0, aich_root: row.get(8)?, + aich_root_origin: row.get(23)?, + aich_trust_generation: crate::sql_value::unsigned::(row, 24)?, upload_priority: row.get(9)?, auto_upload_priority: row.get::<_, i64>(10)? != 0, comment: row.get(11)?, @@ -1833,6 +1903,168 @@ fn delete_transfer_and_unreferenced_identity( Ok(()) } +fn read_transfer_aich_identity( + conn: &rusqlite::Connection, + hash: &[u8], +) -> Result> { + Ok(conn + .query_row( + "SELECT CASE WHEN t.aich_root IS NULL THEN NULL ELSE lower(hex(t.aich_root)) END, + t.aich_root_origin, t.aich_trust_generation + FROM known_files f JOIN transfers t ON t.known_file_id = f.id + WHERE f.ed2k_hash = ?1 AND t.delete_requested_at_ms IS NULL", + params![hash], + |row| { + Ok(MetadataTransferAichIdentity { + root: row.get(0)?, + origin: row.get(1)?, + generation: crate::sql_value::unsigned::(row, 2)?, + }) + }, + ) + .optional()?) +} + +fn reconcile_manifest_aich_identity( + conn: &rusqlite::Connection, + hash: &[u8], + manifest: &mut MetadataTransferManifest, +) -> Result<()> { + let current = read_transfer_aich_identity(conn, hash)?; + if let Some(current) = ¤t + && manifest.aich_trust_generation < current.generation + { + // A checkpoint may contain newer payload progress but older authority. + // Keep the former, and never resurrect its old acquired AICH part set. + manifest.aich_root = current.root.clone(); + manifest.aich_root_origin = current.origin.clone(); + manifest.aich_trust_generation = current.generation; + manifest.aich_hashset.clear(); + manifest.aich_hashset_acquired = false; + } + let accepted = conn + .query_row( + "SELECT lower(hex(aich_root)), aich_root_origin FROM known_files + WHERE ed2k_hash = ?1 AND aich_root IS NOT NULL AND aich_root_origin IS NOT NULL", + params![hash], + |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)), + ) + .optional()?; + if let Some((root, stored_origin)) = accepted { + if manifest.aich_root.is_some() { + ensure!( + manifest.aich_root.as_ref() == Some(&root), + crate::MetadataContentConflict("ED2K identity has conflicting AICH root") + ); + } + // Preserve legitimate local/import enrichment of the same accepted + // root; lower-authority checkpoints cannot downgrade its provenance. + let origin = match manifest.aich_root_origin.as_deref() { + Some("local_computed") => "local_computed".to_owned(), + Some("stock_import") if stored_origin != "local_computed" => "stock_import".to_owned(), + _ => stored_origin, + }; + if manifest.aich_root.as_ref() != Some(&root) + || manifest.aich_root_origin.as_ref() != Some(&origin) + { + if manifest.aich_root.as_ref() != Some(&root) { + manifest.aich_hashset.clear(); + manifest.aich_hashset_acquired = false; + } + manifest.aich_trust_generation = match current.as_ref() { + Some(current) + if current.root.as_ref() == Some(&root) + && current.origin.as_ref() == Some(&origin) => + { + current.generation + } + current => manifest.aich_trust_generation.max( + current + .map_or(0, |current| current.generation) + .checked_add(1) + .ok_or_else(|| anyhow::anyhow!("AICH trust generation exhausted"))?, + ), + }; + i64::try_from(manifest.aich_trust_generation)?; + manifest.aich_root = Some(root); + manifest.aich_root_origin = Some(origin); + } + } + if let Some(current) = current { + let minimum_generation = + if manifest.aich_root != current.root || manifest.aich_root_origin != current.origin { + current + .generation + .checked_add(1) + .ok_or_else(|| anyhow::anyhow!("AICH trust generation exhausted"))? + } else { + current.generation + }; + manifest.aich_trust_generation = manifest.aich_trust_generation.max(minimum_generation); + i64::try_from(manifest.aich_trust_generation)?; + } + Ok(()) +} + +/// Local hashing and trusted stock acceptance update the transfer in the same +/// transaction as accepted identity. Cold readers therefore never combine a +/// new root with an old acquired part set or generation. +pub(crate) fn reconcile_transfer_aich_authority_tx( + tx: &rusqlite::Transaction<'_>, + known_file_id: i64, +) -> Result<()> { + let row = tx + .query_row( + "SELECT t.id, f.aich_root, f.aich_root_origin, t.aich_root, + t.aich_root_origin, t.aich_trust_generation + FROM known_files f JOIN transfers t ON t.known_file_id = f.id + WHERE f.id = ?1 AND f.aich_root IS NOT NULL + AND f.aich_root_origin IS NOT NULL", + params![known_file_id], + |row| { + Ok(( + row.get::<_, i64>(0)?, + row.get::<_, Vec>(1)?, + row.get::<_, String>(2)?, + row.get::<_, Option>>(3)?, + row.get::<_, Option>(4)?, + crate::sql_value::unsigned::(row, 5)?, + )) + }, + ) + .optional()?; + let Some((transfer_id, root, origin, previous_root, previous_origin, generation)) = row else { + return Ok(()); + }; + let root_changed = previous_root.as_ref() != Some(&root); + if !root_changed && previous_origin.as_ref() == Some(&origin) { + return Ok(()); + } + let generation = generation + .checked_add(1) + .ok_or_else(|| anyhow::anyhow!("AICH trust generation exhausted"))?; + if root_changed { + tx.execute( + "DELETE FROM transfer_aich_part_hashes WHERE transfer_id = ?1", + params![transfer_id], + )?; + } + tx.execute( + "UPDATE transfers SET aich_root = ?2, aich_root_origin = ?3, + aich_trust_generation = ?4, + aich_hashset_acquired = CASE WHEN ?5 THEN 0 ELSE aich_hashset_acquired END + WHERE id = ?1", + params![ + transfer_id, + root, + origin, + i64::try_from(generation)?, + root_changed + ], + )?; + Ok(()) +} + #[derive(Debug)] struct TransferRow { transfer_id: i64, @@ -1845,6 +2077,8 @@ struct TransferRow { md4_hashset_acquired: bool, aich_hashset_acquired: bool, aich_root: Option, + aich_root_origin: Option, + aich_trust_generation: u64, upload_priority: String, auto_upload_priority: bool, comment: String, @@ -2045,6 +2279,8 @@ fn manifest_from_row( )?, aich_hashset_acquired: row.aich_hashset_acquired, aich_root: row.aich_root, + aich_root_origin: row.aich_root_origin, + aich_trust_generation: row.aich_trust_generation, aich_hashset: read_hex_list( conn, "SELECT lower(hex(aich_hash)) FROM transfer_aich_part_hashes WHERE transfer_id = ?1 ORDER BY part_index", diff --git a/crates/emulebb-metadata/src/transfer_store/tests.rs b/crates/emulebb-metadata/src/transfer_store/tests.rs index c83b87a6..5e7a35b0 100644 --- a/crates/emulebb-metadata/src/transfer_store/tests.rs +++ b/crates/emulebb-metadata/src/transfer_store/tests.rs @@ -37,6 +37,8 @@ fn completed_download_manifest() -> MetadataTransferManifest { md4_hashset: Vec::new(), aich_hashset_acquired: false, aich_root: None, + aich_root_origin: None, + aich_trust_generation: 0, aich_hashset: Vec::new(), verified_ranges: vec![MetadataTransferRange { start: 0, end: 16 }], pieces: Vec::new(), @@ -65,6 +67,221 @@ fn completed_pending_download_manifest() -> MetadataTransferManifest { manifest } +#[test] +fn accepted_network_decision_roundtrips_without_becoming_known_content_identity() { + let store = MetadataStore::in_memory().unwrap(); + let mut manifest = completed_pending_download_manifest(); + manifest.completed = false; + manifest.payload_fingerprint = None; + manifest.aich_root = Some("11".repeat(20)); + manifest.aich_root_origin = Some("network_consensus".to_string()); + manifest.aich_trust_generation = 1; + store.upsert_transfer_manifest(&manifest).unwrap(); + assert_eq!( + store + .transfer_manifest_by_hash(&manifest.file_hash) + .unwrap(), + Some(manifest.clone()) + ); + let known_root: Option> = store + .connection() + .unwrap() + .query_row("SELECT aich_root FROM known_files", [], |row| row.get(0)) + .unwrap(); + assert!(known_root.is_none()); + + manifest.aich_root = None; + manifest.aich_root_origin = None; + manifest.aich_trust_generation = 2; + store.upsert_transfer_manifest(&manifest).unwrap(); + assert_eq!( + store + .transfer_manifest_by_hash(&manifest.file_hash) + .unwrap(), + Some(manifest) + ); +} + +#[test] +fn root_origin_and_part_set_cannot_manufacture_an_accepted_decision() { + let store = MetadataStore::in_memory().unwrap(); + let manifest = completed_pending_download_manifest(); + let mut invalid = manifest.clone(); + invalid.aich_root = Some("11".repeat(20)); + assert!(store.upsert_transfer_manifest(&invalid).is_err()); + invalid = manifest.clone(); + invalid.aich_root_origin = Some("network_consensus".to_string()); + assert!(store.upsert_transfer_manifest(&invalid).is_err()); + invalid = manifest.clone(); + invalid.aich_hashset_acquired = true; + assert!(store.upsert_transfer_manifest(&invalid).is_err()); + invalid = manifest; + invalid.aich_trust_generation = u64::MAX; + assert!(store.upsert_transfer_manifest(&invalid).is_err()); +} + +#[test] +fn accepted_local_identity_and_stale_checkpoint_preserve_md4_verified_progress() { + let store = MetadataStore::in_memory().unwrap(); + let mut stale = completed_download_manifest(); + stale.completed = false; + stale.aich_root = Some("11".repeat(20)); + stale.aich_root_origin = Some("network_consensus".to_owned()); + stale.aich_trust_generation = 1; + stale.aich_hashset_acquired = true; + stale.aich_hashset = vec!["33".repeat(20)]; + stale.pieces = vec![MetadataTransferPiece { + piece_index: 0, + state: "Verified".to_owned(), + bytes_written: 16, + block_bitmap: None, + ich_corrupted: false, + }]; + store.upsert_transfer_manifest(&stale).unwrap(); + let content = crate::MetadataVerifiedLocalContent { + file_hash: stale.file_hash.clone(), + display_name: "local-authority.bin".to_owned(), + file_size: stale.file_size, + part_size: stale.piece_size, + md4_hashset: Vec::new(), + aich_root: Some("22".repeat(20)), + aich_hashset: Vec::new(), + path: std::env::temp_dir().join("aich-new-authority.bin"), + fingerprint: fingerprint(stale.file_size, 21), + upload_priority: "normal".to_owned(), + auto_upload_priority: false, + comment: String::new(), + rating: 0, + }; + store.upsert_verified_local_content(&content).unwrap(); + let accepted = store + .transfer_manifest_by_hash(&stale.file_hash) + .unwrap() + .unwrap(); + assert_eq!(accepted.aich_root, content.aich_root); + assert_eq!(accepted.aich_root_origin.as_deref(), Some("local_computed")); + assert_eq!(accepted.aich_trust_generation, 2); + assert!(!accepted.aich_hashset_acquired); + assert!(accepted.aich_hashset.is_empty()); + assert_eq!(accepted.verified_ranges, stale.verified_ranges); + assert_eq!(accepted.pieces, stale.pieces); + assert_eq!(accepted.md4_hashset_acquired, stale.md4_hashset_acquired); + + stale.comment = "newer independent checkpoint metadata".to_owned(); + let canonical = store.upsert_transfer_manifest_reconciled(&stale).unwrap(); + assert_eq!(canonical.aich_root, accepted.aich_root); + assert_eq!(canonical.aich_root_origin, accepted.aich_root_origin); + assert_eq!( + canonical.aich_trust_generation, + accepted.aich_trust_generation + ); + assert!(canonical.aich_hashset.is_empty()); + assert!(!canonical.aich_hashset_acquired); + assert_eq!(canonical.comment, stale.comment); + assert_eq!(canonical.verified_ranges, stale.verified_ranges); + assert_eq!(canonical.pieces, stale.pieces); + assert_eq!( + store.transfer_manifest_by_hash(&stale.file_hash).unwrap(), + Some(canonical) + ); +} + +#[test] +fn local_acceptance_of_same_master_preserves_acquired_parts_and_progress() { + let store = MetadataStore::in_memory().unwrap(); + let mut manifest = completed_pending_download_manifest(); + manifest.md4_hashset = vec!["22".repeat(16)]; + manifest.aich_root = Some("11".repeat(20)); + manifest.aich_root_origin = Some("network_consensus".to_owned()); + manifest.aich_trust_generation = 7; + manifest.aich_hashset_acquired = true; + manifest.aich_hashset = vec!["33".repeat(20)]; + store.upsert_transfer_manifest(&manifest).unwrap(); + let content = crate::MetadataVerifiedLocalContent { + file_hash: manifest.file_hash.clone(), + display_name: "same-master.bin".to_owned(), + file_size: manifest.file_size, + part_size: manifest.piece_size, + md4_hashset: manifest.md4_hashset.clone(), + aich_root: manifest.aich_root.clone(), + aich_hashset: manifest.aich_hashset.clone(), + path: std::env::temp_dir().join("emulebb-same-master.bin"), + fingerprint: fingerprint(manifest.file_size, 201), + upload_priority: "normal".to_owned(), + auto_upload_priority: false, + comment: String::new(), + rating: 0, + }; + store.upsert_verified_local_content(&content).unwrap(); + manifest.aich_root_origin = Some("local_computed".to_owned()); + manifest.aich_trust_generation = 8; + // Provenance changed, but the acquired set is still bound to the same + // master. Whole-manifest equality also preserves independent MD4/progress. + assert_eq!( + store + .transfer_manifest_by_hash(&manifest.file_hash) + .unwrap(), + Some(manifest.clone()) + ); + store.upsert_verified_local_content(&content).unwrap(); + assert_eq!( + store + .transfer_manifest_by_hash(&manifest.file_hash) + .unwrap(), + Some(manifest) + ); +} + +#[test] +fn operator_master_is_authority_for_network_and_local_content_writers() { + let store = MetadataStore::in_memory().unwrap(); + let mut manifest = completed_download_manifest(); + manifest.completed = false; + manifest.aich_root = Some("11".repeat(20)); + manifest.aich_root_origin = Some("operator_link".to_string()); + manifest.aich_trust_generation = 1; + store.upsert_transfer_manifest(&manifest).unwrap(); + let mut proposal = manifest.clone(); + proposal.aich_root = Some("22".repeat(20)); + proposal.aich_root_origin = Some("network_consensus".to_string()); + proposal.aich_trust_generation = 2; + assert!( + store + .upsert_transfer_manifest(&proposal) + .unwrap_err() + .is::() + ); + + let content = crate::MetadataVerifiedLocalContent { + file_hash: manifest.file_hash.clone(), + display_name: "local.bin".to_string(), + file_size: manifest.file_size, + part_size: manifest.piece_size, + md4_hashset: Vec::new(), + aich_root: Some("22".repeat(20)), + aich_hashset: Vec::new(), + path: std::env::temp_dir().join("aich-operator-conflict.bin"), + fingerprint: fingerprint(manifest.file_size, 9), + upload_priority: "normal".to_string(), + auto_upload_priority: false, + comment: String::new(), + rating: 0, + }; + assert!( + store + .upsert_verified_local_content(&content) + .unwrap_err() + .is::() + ); + assert_eq!( + store + .transfer_manifest_by_hash(&manifest.file_hash) + .unwrap(), + Some(manifest) + ); + assert_eq!(store.table_count("verified_local_content").unwrap(), 0); +} + fn fingerprint(size_bytes: u64, sequence: u64) -> NativeFileFingerprint { NativeFileFingerprint { identity: crate::NativeFileIdentity::Unix { @@ -159,10 +376,12 @@ fn accepted_identity_survives_conflicting_transfer_and_index_observations() { let mut manifest = completed_download_manifest(); manifest.completed = false; manifest.aich_root = Some("11".repeat(20)); + manifest.aich_root_origin = Some("operator_link".to_string()); + manifest.aich_trust_generation = 1; store.upsert_transfer_manifest(&manifest).unwrap(); store.connection().unwrap().execute( "INSERT INTO verified_local_content(known_file_id, display_name, part_size_bytes, part_count, aich_root) - SELECT known_files.id, transfers.display_name, transfers.part_size_bytes, transfers.part_count, aich_root FROM known_files JOIN transfers ON transfers.known_file_id = known_files.id", + SELECT known_files.id, transfers.display_name, transfers.part_size_bytes, transfers.part_count, known_files.aich_root FROM known_files JOIN transfers ON transfers.known_file_id = known_files.id", [], ).unwrap(); @@ -200,6 +419,7 @@ fn accepted_identity_survives_conflicting_transfer_and_index_observations() { let mut sparse = manifest.clone(); sparse.aich_root = None; + sparse.aich_root_origin = None; store.upsert_transfer_manifest(&sparse).unwrap(); assert_eq!( store @@ -325,6 +545,8 @@ fn transfer_manifest_roundtrips_sql_tables() { let mut manifest = completed_pending_download_manifest(); manifest.display_name = "Sample.Transfer.bin".to_string(); manifest.aich_root = Some("1111111111111111111111111111111111111111".to_string()); + manifest.aich_root_origin = Some("operator_link".to_string()); + manifest.aich_trust_generation = 1; manifest.aich_hashset_acquired = true; manifest.aich_hashset = vec!["2222222222222222222222222222222222222222".to_string()]; manifest.pieces = vec![MetadataTransferPiece { @@ -425,6 +647,8 @@ fn upload_demand_counters_persist_on_known_files() { md4_hashset: Vec::new(), aich_hashset_acquired: false, aich_root: None, + aich_root_origin: None, + aich_trust_generation: 0, aich_hashset: Vec::new(), verified_ranges: vec![MetadataTransferRange { start: 0, @@ -554,6 +778,12 @@ fn deleting_download_preserves_same_hash_local_content() { manifest.md4_hashset = vec!["22222222222222222222222222222222".to_string()]; manifest.aich_hashset = vec!["3333333333333333333333333333333333333333".to_string()]; store.upsert_transfer_manifest(&manifest).unwrap(); + assert_eq!( + store + .transfer_manifest_by_hash(&manifest.file_hash) + .unwrap(), + Some(manifest.clone()) + ); let path = std::env::temp_dir().join("emulebb-same-hash-local-content.bin"); let content = crate::MetadataVerifiedLocalContent { file_hash: manifest.file_hash.clone(), @@ -571,8 +801,13 @@ fn deleting_download_preserves_same_hash_local_content() { rating: 0, }; store.upsert_verified_local_content(&content).unwrap(); - // Content identity enrichment is shared; transfer-owned hashsets stay independent. + // A newly accepted master retires parts stored without that authority. MD4 + // hashes and verified progress remain independent of this AICH transition. manifest.aich_root = content.aich_root.clone(); + manifest.aich_root_origin = Some("local_computed".to_string()); + manifest.aich_trust_generation += 1; + manifest.aich_hashset.clear(); + manifest.aich_hashset_acquired = false; assert_eq!( store .transfer_manifest_by_hash(&manifest.file_hash) @@ -653,6 +888,8 @@ fn delete_transfer_manifest_removes_transfer_rows() { md4_hashset: Vec::new(), aich_hashset_acquired: false, aich_root: None, + aich_root_origin: None, + aich_trust_generation: 0, aich_hashset: Vec::new(), verified_ranges: Vec::new(), pieces: vec![MetadataTransferPiece { @@ -930,6 +1167,8 @@ fn destructive_deletion_intent_is_idempotent_and_required_for_finalization() { md4_hashset: Vec::new(), aich_hashset_acquired: false, aich_root: None, + aich_root_origin: None, + aich_trust_generation: 0, aich_hashset: Vec::new(), verified_ranges: vec![MetadataTransferRange { start: 0, end: 1 }], pieces: vec![MetadataTransferPiece { @@ -1346,6 +1585,8 @@ fn delete_transfer_manifest_clears_soft_known_file_references() { md4_hashset: Vec::new(), aich_hashset_acquired: false, aich_root: None, + aich_root_origin: None, + aich_trust_generation: 0, aich_hashset: Vec::new(), verified_ranges: Vec::new(), pieces: vec![MetadataTransferPiece { @@ -1464,6 +1705,8 @@ fn completed_catalog_supports_hash_keyset_pages() { md4_hashset: Vec::new(), aich_hashset_acquired: false, aich_root: None, + aich_root_origin: None, + aich_trust_generation: 0, aich_hashset: Vec::new(), verified_ranges: vec![MetadataTransferRange { start: 0, diff --git a/crates/emulebb-metadata/src/transfer_validation.rs b/crates/emulebb-metadata/src/transfer_validation.rs index e20b2390..0bb5bfba 100644 --- a/crates/emulebb-metadata/src/transfer_validation.rs +++ b/crates/emulebb-metadata/src/transfer_validation.rs @@ -5,6 +5,24 @@ use crate::{ use anyhow::{Context, Result, ensure}; pub(crate) fn validate_manifest(manifest: &MetadataTransferManifest) -> Result<()> { + ensure!( + manifest.aich_root.is_some() == manifest.aich_root_origin.is_some(), + "AICH root and accepted origin must be present together" + ); + if let Some(origin) = manifest.aich_root_origin.as_deref() { + ensure!( + matches!( + origin, + "local_computed" | "operator_link" | "stock_import" | "network_consensus" + ), + "invalid AICH root origin" + ); + } + ensure!( + !manifest.aich_hashset_acquired || manifest.aich_root.is_some(), + "acquired AICH part hashes require an accepted master" + ); + i64::try_from(manifest.aich_trust_generation)?; ensure!( manifest.piece_size > 0, "transfer piece size must be positive"