diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 864c8b3..9dd68c1 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -34,7 +34,7 @@ jobs: - name: Install system deps if: runner.os == 'Linux' run: sudo apt-get update && sudo apt-get install -y libudev-dev libdbus-1-dev pkg-config - - run: cargo test --workspace + - run: cargo test --workspace --locked working-directory: rsLXMF lint: @@ -60,10 +60,36 @@ jobs: run: sudo apt-get update && sudo apt-get install -y libudev-dev libdbus-1-dev pkg-config - run: cargo fmt --all -- --check working-directory: rsLXMF - - run: cargo clippy --workspace --all-targets -- -D warnings + - run: cargo clippy --workspace --all-targets --all-features --locked -- -D warnings working-directory: rsLXMF - name: Check public API documentation env: RUSTDOCFLAGS: "-D warnings" - run: cargo doc --workspace --no-deps + run: cargo doc --workspace --no-deps --locked + working-directory: rsLXMF + + msrv: + name: Rust 1.85 MSRV + runs-on: ubuntu-latest + timeout-minutes: 30 + steps: + - uses: actions/checkout@v5 + with: + path: rsLXMF + - uses: actions/checkout@v5 + with: + repository: ${{ github.repository_owner }}/rsReticulum + ref: main + path: rsReticulum + - uses: dtolnay/rust-toolchain@1.85.0 + with: + components: clippy + - uses: Swatinem/rust-cache@v2 + with: + workspaces: rsLXMF -> target + cache-bin: false + - name: Install system deps + run: sudo apt-get update && sudo apt-get install -y libudev-dev libdbus-1-dev pkg-config + - name: Clippy (all targets and features) + run: cargo clippy --workspace --all-targets --all-features --locked -- -D warnings working-directory: rsLXMF diff --git a/clippy.toml b/clippy.toml new file mode 100644 index 0000000..566161e --- /dev/null +++ b/clippy.toml @@ -0,0 +1,2 @@ +# Keep Clippy suggestions aligned with the public MSRV in Cargo.toml. +msrv = "1.85" diff --git a/crates/lxmf-core/src/delivery_ratchet.rs b/crates/lxmf-core/src/delivery_ratchet.rs index 3827496..ae08a23 100644 --- a/crates/lxmf-core/src/delivery_ratchet.rs +++ b/crates/lxmf-core/src/delivery_ratchet.rs @@ -159,15 +159,14 @@ impl DeliveryRatchetState { } } - if control_file_trusted - && (!control_existed || control_changed) - && let Err(error) = control.save_verified(&control_path, identity) - { - tracing::warn!( - path = %control_path.display(), - %error, - "could not persist initial ratchet control state; new announces remain deferred until persistence recovers" - ); + if control_file_trusted && (!control_existed || control_changed) { + if let Err(error) = control.save_verified(&control_path, identity) { + tracing::warn!( + path = %control_path.display(), + %error, + "could not persist initial ratchet control state; new announces remain deferred until persistence recovers" + ); + } } Ok(Self { @@ -211,12 +210,13 @@ impl DeliveryRatchetState { } let announce_time = AnnounceTime::new(wall_now, cache_now)?; - if let DeliveryAnnounceKind::PathResponse { tag: Some(tag) } = kind - && let Some(packet) = self + if let DeliveryAnnounceKind::PathResponse { tag: Some(tag) } = kind { + if let Some(packet) = self .destination .cached_path_response_packet(tag, cache_now)? - { - return Ok(packet); + { + return Ok(packet); + } } if !self.control_file_trusted { @@ -292,15 +292,15 @@ impl DeliveryRatchetState { /// Best-effort shutdown checkpoint. Invalid files detected during load are /// never overwritten by this operation. pub fn save(&self, identity: &Identity) { - if self.ring_file_trusted - && let Err(error) = self.ring.save_verified(&self.ring_path, identity) - { - tracing::warn!(%error, "failed to save ratchet ring"); + if self.ring_file_trusted { + if let Err(error) = self.ring.save_verified(&self.ring_path, identity) { + tracing::warn!(%error, "failed to save ratchet ring"); + } } - if self.control_file_trusted - && let Err(error) = self.control.save_verified(&self.control_path, identity) - { - tracing::warn!(%error, "failed to save ratchet control state"); + if self.control_file_trusted { + if let Err(error) = self.control.save_verified(&self.control_path, identity) { + tracing::warn!(%error, "failed to save ratchet control state"); + } } } diff --git a/crates/lxmf-core/src/link_delivery.rs b/crates/lxmf-core/src/link_delivery.rs index 778c30a..fb4fed4 100644 --- a/crates/lxmf-core/src/link_delivery.rs +++ b/crates/lxmf-core/src/link_delivery.rs @@ -755,8 +755,10 @@ impl LinkDeliveryManager { .try_send(TransportMessage::DeregisterDestination { hash: link_id }); } self.direct_links.remove(&dest_hash); - } else if let Some(delivery) = self.pending.get_mut(&link_id) - && delivery.reusable + } else if let Some(delivery) = self + .pending + .get_mut(&link_id) + .filter(|delivery| delivery.reusable) { let msg_hash = message.hash; let attempts = message.delivery_attempts; @@ -1327,32 +1329,37 @@ impl LinkDeliveryManager { match delivery.state { DeliveryState::Idle => {} DeliveryState::Identifying if delivery.link.is_active() => { - if !delivery.reusable - && let (Some(pub_key), Some(sign_key)) = + if !delivery.reusable { + if let (Some(pub_key), Some(sign_key)) = (&self.identity_pub, &self.identity_key) - && let Ok(identify_data) = delivery.link.identify(pub_key, sign_key) - { - let id_header = rns_wire::header::PacketHeader { - flags: rns_wire::flags::PacketFlags { - header_type: rns_wire::flags::HeaderType::Header1, - context_flag: false, - transport_type: rns_wire::flags::TransportType::Broadcast, - destination_type: rns_wire::flags::DestinationType::Link, - packet_type: rns_wire::flags::PacketType::Data, - }, - hops: 0, - transport_id: None, - destination_hash: *link_id, - context: rns_wire::context::PacketContext::LinkIdentify, - }; - let mut id_raw = id_header.pack(); - id_raw.extend_from_slice(&identify_data); - let _ = self.transport_tx.try_send(TransportMessage::Outbound( - OutboundRequest { - raw: Bytes::from(id_raw), - destination_hash: *link_id, - }, - )); + { + if let Ok(identify_data) = delivery.link.identify(pub_key, sign_key) + { + let id_header = rns_wire::header::PacketHeader { + flags: rns_wire::flags::PacketFlags { + header_type: rns_wire::flags::HeaderType::Header1, + context_flag: false, + transport_type: + rns_wire::flags::TransportType::Broadcast, + destination_type: + rns_wire::flags::DestinationType::Link, + packet_type: rns_wire::flags::PacketType::Data, + }, + hops: 0, + transport_id: None, + destination_hash: *link_id, + context: rns_wire::context::PacketContext::LinkIdentify, + }; + let mut id_raw = id_header.pack(); + id_raw.extend_from_slice(&identify_data); + let _ = self.transport_tx.try_send(TransportMessage::Outbound( + OutboundRequest { + raw: Bytes::from(id_raw), + destination_hash: *link_id, + }, + )); + } + } } // Reusable Direct links follow upstream LXMF and identify // after a successful delivery, not before the transfer. @@ -1809,25 +1816,27 @@ impl LinkDeliveryManager { } pub fn handle_hmu(&mut self, link_id: &[u8; 16], hmu_data: &[u8]) { - let event = if let Some(delivery) = self.pending.get_mut(link_id) - && let Some(ref mut transfer) = delivery.transfer - { - transfer.handle_hmu(hmu_data); - let progress = delivery_resource_progress(delivery); - if let Some(progress) = progress - && should_update_resource_progress(delivery.message.progress, progress) - { - delivery.message.progress = progress; + let event = if let Some(delivery) = self.pending.get_mut(link_id) { + if let Some(ref mut transfer) = delivery.transfer { + transfer.handle_hmu(hmu_data); + let progress = delivery_resource_progress(delivery); + if let Some(progress) = progress { + if should_update_resource_progress(delivery.message.progress, progress) { + delivery.message.progress = progress; + } + } + progress.map(|_| { + delivery_event( + LxmfDeliveryEventKind::TransferProgress, + *link_id, + delivery, + Some(delivery.message.progress), + None, + ) + }) + } else { + None } - progress.map(|_| { - delivery_event( - LxmfDeliveryEventKind::TransferProgress, - *link_id, - delivery, - Some(delivery.message.progress), - None, - ) - }) } else { None }; @@ -1864,10 +1873,10 @@ impl LinkDeliveryManager { } } let progress = delivery_resource_progress(delivery); - if let Some(progress) = progress - && should_update_resource_progress(delivery.message.progress, progress) - { - delivery.message.progress = progress; + if let Some(progress) = progress { + if should_update_resource_progress(delivery.message.progress, progress) { + delivery.message.progress = progress; + } } progress.map(|_| { delivery_event( @@ -1887,28 +1896,33 @@ impl LinkDeliveryManager { /// Apply an inbound resource proof; returns `true` when the proof was accepted. pub fn handle_resource_proof(&mut self, link_id: &[u8; 16], proof_data: &[u8]) -> bool { let mut event = None; - let accepted = if let Some(delivery) = self.pending.get_mut(link_id) - && let Some(ref mut transfer) = delivery.transfer - && transfer.handle_proof(proof_data) - { - let progress = delivery_resource_proof_progress(delivery).unwrap_or(1.0); - delivery.message.progress = progress; - event = Some(delivery_event( - LxmfDeliveryEventKind::TransferProgress, - *link_id, - delivery, - Some(progress), - None, - )); - if delivery.remaining_segments.is_empty() { - delivery.state = DeliveryState::Complete; + let accepted = if let Some(delivery) = self.pending.get_mut(link_id) { + if delivery + .transfer + .as_mut() + .is_some_and(|transfer| transfer.handle_proof(proof_data)) + { + let progress = delivery_resource_proof_progress(delivery).unwrap_or(1.0); + delivery.message.progress = progress; + event = Some(delivery_event( + LxmfDeliveryEventKind::TransferProgress, + *link_id, + delivery, + Some(progress), + None, + )); + if delivery.remaining_segments.is_empty() { + delivery.state = DeliveryState::Complete; + } else { + let rtt = delivery.link.rtt.unwrap_or(Duration::from_millis(500)); + let next_segment = delivery.remaining_segments.remove(0); + delivery.transfer = Some(OutboundTransfer::from_prebuilt(next_segment, rtt)); + delivery.state = DeliveryState::Transferring; + } + true } else { - let rtt = delivery.link.rtt.unwrap_or(Duration::from_millis(500)); - let next_segment = delivery.remaining_segments.remove(0); - delivery.transfer = Some(OutboundTransfer::from_prebuilt(next_segment, rtt)); - delivery.state = DeliveryState::Transferring; + false } - true } else { false }; @@ -1927,16 +1941,17 @@ impl LinkDeliveryManager { let mut rejected_hash = [0u8; 32]; rejected_hash.copy_from_slice(&reject_data[..32]); - if let Some(delivery) = self.pending.get_mut(link_id) - && let Some(ref mut transfer) = delivery.transfer - && transfer.resource.resource_hash == rejected_hash - { - transfer.handle_cancel(); - delivery.remaining_segments.clear(); - delivery.message.mark_rejected(); - delivery.state = DeliveryState::Rejected; - delivery.failure_reason = Some("resource rejected".to_string()); - return true; + if let Some(delivery) = self.pending.get_mut(link_id) { + if let Some(ref mut transfer) = delivery.transfer { + if transfer.resource.resource_hash == rejected_hash { + transfer.handle_cancel(); + delivery.remaining_segments.clear(); + delivery.message.mark_rejected(); + delivery.state = DeliveryState::Rejected; + delivery.failure_reason = Some("resource rejected".to_string()); + return true; + } + } } false @@ -1979,15 +1994,18 @@ impl LinkDeliveryManager { /// Apply an inbound link-packet proof; returns `true` when the packet delivery is complete. pub fn handle_link_packet_proof(&mut self, link_id: &[u8; 16], proof_data: &[u8]) -> bool { - if let Some(delivery) = self.pending.get_mut(link_id) - && delivery.state == DeliveryState::AwaitingProof - && let Some(packet_hash) = delivery.packet_proof_hash - && delivery - .link - .validate_packet_proof(&packet_hash, proof_data) - { - delivery.state = DeliveryState::Complete; - return true; + if let Some(delivery) = self.pending.get_mut(link_id) { + if delivery.state == DeliveryState::AwaitingProof { + if let Some(packet_hash) = delivery.packet_proof_hash { + if delivery + .link + .validate_packet_proof(&packet_hash, proof_data) + { + delivery.state = DeliveryState::Complete; + return true; + } + } + } } false } @@ -2122,30 +2140,30 @@ impl LinkDeliveryManager { .pending_backchannel_deliveries .iter() .find_map(|(key, delivery)| (delivery.message.hash == Some(msg_hash)).then_some(*key)); - if let Some(key) = pending_key - && let Some(delivery) = self.pending_backchannel_deliveries.remove(&key) - { - self.backchannel_links.remove(&delivery.dest_hash); - self.delivery_events.push_back(backchannel_delivery_event( - BackchannelDeliveryEventInput { - kind: LxmfDeliveryEventKind::Failed, - message: &delivery.message, - dest_hash: delivery.dest_hash, + if let Some(key) = pending_key { + if let Some(delivery) = self.pending_backchannel_deliveries.remove(&key) { + self.backchannel_links.remove(&delivery.dest_hash); + self.delivery_events.push_back(backchannel_delivery_event( + BackchannelDeliveryEventInput { + kind: LxmfDeliveryEventKind::Failed, + message: &delivery.message, + dest_hash: delivery.dest_hash, + link_id: delivery.link_id, + representation: delivery.representation, + progress: Some(delivery.message.progress), + reason: Some(reason.to_string()), + link_state: LinkState::Closed, + delivery_state: DeliveryState::Failed, + }, + )); + results.push(DeliveryResult::Failed { link_id: delivery.link_id, - representation: delivery.representation, - progress: Some(delivery.message.progress), - reason: Some(reason.to_string()), - link_state: LinkState::Closed, - delivery_state: DeliveryState::Failed, - }, - )); - results.push(DeliveryResult::Failed { - link_id: delivery.link_id, - msg_hash: delivery.message.hash, - dest_hash: delivery.dest_hash, - message: delivery.message, - reason: reason.to_string(), - }); + msg_hash: delivery.message.hash, + dest_hash: delivery.dest_hash, + message: delivery.message, + reason: reason.to_string(), + }); + } } results @@ -2535,11 +2553,11 @@ fn finish_reusable_delivery( link_id: &[u8; 16], delivery: &mut PendingDelivery, ) { - if !delivery.backchannel_identified - && let (Some(pub_key), Some(sign_key)) = (identity_pub, identity_key) - { - delivery.backchannel_identified = - send_link_identify(transport_tx, link_id, &delivery.link, pub_key, sign_key); + if !delivery.backchannel_identified { + if let (Some(pub_key), Some(sign_key)) = (identity_pub, identity_key) { + delivery.backchannel_identified = + send_link_identify(transport_tx, link_id, &delivery.link, pub_key, sign_key); + } } delivery.transfer = None; diff --git a/crates/lxmf-core/src/message.rs b/crates/lxmf-core/src/message.rs index c3dc001..9b9e8bf 100644 --- a/crates/lxmf-core/src/message.rs +++ b/crates/lxmf-core/src/message.rs @@ -391,19 +391,19 @@ impl LxMessage { self.transient_id = Some(tid); let mut stamp_value = 0; - if let Some(target_cost) = propagation_stamp_cost - && self.propagation_stamp.is_none() - { - let (stamp, value) = crate::stamper::generate_stamp( - &tid, - target_cost, - crate::constants::STAMP_WORKBLOCK_EXPAND_ROUNDS_PN, - ) - .ok_or_else(|| { - MessageError::PackFailed("failed to generate propagation stamp".to_string()) - })?; - self.propagation_stamp = Some(stamp); - stamp_value = value; + if let Some(target_cost) = propagation_stamp_cost { + if self.propagation_stamp.is_none() { + let (stamp, value) = crate::stamper::generate_stamp( + &tid, + target_cost, + crate::constants::STAMP_WORKBLOCK_EXPAND_ROUNDS_PN, + ) + .ok_or_else(|| { + MessageError::PackFailed("failed to generate propagation stamp".to_string()) + })?; + self.propagation_stamp = Some(stamp); + stamp_value = value; + } } if let Some(ref prop_stamp) = self.propagation_stamp { @@ -791,9 +791,7 @@ impl LxMessage { /// /// Python reference: LXMessage.py:301-332. pub fn get_stamp(&mut self) -> Option> { - if let Some(ticket) = self.outbound_ticket - && let Some(message_id) = self.message_id - { + if let (Some(ticket), Some(message_id)) = (self.outbound_ticket, self.message_id) { let mut material = Vec::with_capacity(TICKET_LENGTH + 32); material.extend_from_slice(&ticket); material.extend_from_slice(&message_id); @@ -816,13 +814,14 @@ impl LxMessage { // 4. Generate PoW stamp let cost = self.stamp_cost.unwrap(); - if let Some(message_id) = self.message_id - && let Some((stamp, value)) = + if let Some(message_id) = self.message_id { + if let Some((stamp, value)) = crate::stamper::generate_stamp(&message_id, cost, STAMP_WORKBLOCK_EXPAND_ROUNDS) - { - self.stamp_value = Some(value as u16); - self.stamp = Some(stamp.to_vec()); - return Some(stamp.to_vec()); + { + self.stamp_value = Some(value as u16); + self.stamp = Some(stamp.to_vec()); + return Some(stamp.to_vec()); + } } None @@ -1347,7 +1346,7 @@ fn deserialize_bin_or_str<'de, D: serde::Deserializer<'de>>( deserializer: D, ) -> Result { struct BinOrStrVisitor; - impl<'de> serde::de::Visitor<'de> for BinOrStrVisitor { + impl serde::de::Visitor<'_> for BinOrStrVisitor { type Value = String; fn expecting(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result { f.write_str("string or binary data") diff --git a/crates/lxmf-core/src/persist.rs b/crates/lxmf-core/src/persist.rs index 8d91bb7..8c8bf83 100644 --- a/crates/lxmf-core/src/persist.rs +++ b/crates/lxmf-core/src/persist.rs @@ -68,15 +68,14 @@ pub fn write_file_atomic(path: &Path, data: &[u8]) -> io::Result<()> { fs::rename(&tmp, path) })(); - if result.is_err() - && tmp.exists() - && let Err(e) = fs::remove_file(&tmp) - { - tracing::error!( - "Error while cleaning temporary file {} for {}: {e}", - tmp.display(), - path.display() - ); + if result.is_err() && tmp.exists() { + if let Err(e) = fs::remove_file(&tmp) { + tracing::error!( + "Error while cleaning temporary file {} for {}: {e}", + tmp.display(), + path.display() + ); + } } result diff --git a/crates/lxmf-core/src/propagation_client.rs b/crates/lxmf-core/src/propagation_client.rs index 7b1df90..2b06f0f 100644 --- a/crates/lxmf-core/src/propagation_client.rs +++ b/crates/lxmf-core/src/propagation_client.rs @@ -295,23 +295,25 @@ impl PropagationClient { continue; } let node_hex = self.outbound_propagation_node.map(|h| hex_encode(&h)); - if let Some(node_hex) = node_hex - && let Some(pub_key) = known_identities.get(&node_hex) - { - let ed25519_bytes: [u8; 32] = pub_key[32..64] - .try_into() - .expect("known_identities values are [u8; 64]; slice [32..64] is always 32 bytes"); - if let Ok(verify_key) = Ed25519PublicKey::from_bytes(&ed25519_bytes) - { - self.handle_link_proof(data, &verify_key, &ed25519_bytes); + if let Some(node_hex) = node_hex { + if let Some(pub_key) = known_identities.get(&node_hex) { + let ed25519_bytes: [u8; 32] = pub_key[32..64] + .try_into() + .expect("known_identities values are [u8; 64]; slice [32..64] is always 32 bytes"); + if let Ok(verify_key) = + Ed25519PublicKey::from_bytes(&ed25519_bytes) + { + self.handle_link_proof(data, &verify_key, &ed25519_bytes); + } } } } rns_wire::context::PacketContext::Response => { - if let Some(ref mut link) = self.link - && let Ok((_request_id, response_data)) = link.handle_response(data) - { - self.handle_response_data(&response_data); + if let Some(ref mut link) = self.link { + if let Ok((_request_id, response_data)) = link.handle_response(data) + { + self.handle_response_data(&response_data); + } } } rns_wire::context::PacketContext::ResourceAdv => { @@ -745,10 +747,10 @@ impl PropagationClient { context: rns_wire::context::PacketContext, plaintext: &[u8], ) { - if let Some(link) = self.link.as_ref() - && let Ok(encrypted) = link.encrypt(plaintext) - { - self.send_link_packet(context, rns_wire::flags::PacketType::Data, &encrypted); + if let Some(link) = self.link.as_ref() { + if let Ok(encrypted) = link.encrypt(plaintext) { + self.send_link_packet(context, rns_wire::flags::PacketType::Data, &encrypted); + } } } @@ -805,10 +807,10 @@ impl PropagationClient { if let Some(arr) = value.as_array() { self.available_messages.clear(); for item in arr { - if let Some(id_bytes) = item.as_slice() - && id_bytes.len() == 32 - { - self.available_messages.push(id_bytes.to_vec()); + if let Some(id_bytes) = item.as_slice() { + if id_bytes.len() == 32 { + self.available_messages.push(id_bytes.to_vec()); + } } } @@ -864,14 +866,15 @@ impl PropagationClient { } pub fn tick(&mut self) { - if let Some(started) = self.started_at - && started.elapsed() > self.timeout - && self.status.state != PropagationClientState::Idle - && self.status.state != PropagationClientState::Complete - { - self.cleanup(); - self.status.state = PropagationClientState::Failed; - return; + if let Some(started) = self.started_at { + if started.elapsed() > self.timeout + && self.status.state != PropagationClientState::Idle + && self.status.state != PropagationClientState::Complete + { + self.cleanup(); + self.status.state = PropagationClientState::Failed; + return; + } } match self.status.state { @@ -895,31 +898,32 @@ impl PropagationClient { } fn send_identify(&mut self) { - if let (Some(link), Some(link_id)) = (&mut self.link, self.link_id) - && let (Some(pub_key), Some(sign_key)) = (&self.identity_pub, &self.identity_key) - && let Ok(identify_data) = link.identify(pub_key, sign_key) - { - let id_header = rns_wire::header::PacketHeader { - flags: rns_wire::flags::PacketFlags { - header_type: rns_wire::flags::HeaderType::Header1, - context_flag: false, - transport_type: rns_wire::flags::TransportType::Broadcast, - destination_type: rns_wire::flags::DestinationType::Link, - packet_type: rns_wire::flags::PacketType::Data, - }, - hops: 0, - transport_id: None, - destination_hash: link_id, - context: rns_wire::context::PacketContext::LinkIdentify, - }; - let mut id_raw = id_header.pack(); - id_raw.extend_from_slice(&identify_data); - let _ = self - .transport_tx - .try_send(TransportMessage::Outbound(OutboundRequest { - raw: Bytes::from(id_raw), - destination_hash: link_id, - })); + if let (Some(link), Some(link_id)) = (&mut self.link, self.link_id) { + if let (Some(pub_key), Some(sign_key)) = (&self.identity_pub, &self.identity_key) { + if let Ok(identify_data) = link.identify(pub_key, sign_key) { + let id_header = rns_wire::header::PacketHeader { + flags: rns_wire::flags::PacketFlags { + header_type: rns_wire::flags::HeaderType::Header1, + context_flag: false, + transport_type: rns_wire::flags::TransportType::Broadcast, + destination_type: rns_wire::flags::DestinationType::Link, + packet_type: rns_wire::flags::PacketType::Data, + }, + hops: 0, + transport_id: None, + destination_hash: link_id, + context: rns_wire::context::PacketContext::LinkIdentify, + }; + let mut id_raw = id_header.pack(); + id_raw.extend_from_slice(&identify_data); + let _ = + self.transport_tx + .try_send(TransportMessage::Outbound(OutboundRequest { + raw: Bytes::from(id_raw), + destination_hash: link_id, + })); + } + } } } diff --git a/crates/lxmf-core/src/propagation_node.rs b/crates/lxmf-core/src/propagation_node.rs index d3d58da..fca1fab 100644 --- a/crates/lxmf-core/src/propagation_node.rs +++ b/crates/lxmf-core/src/propagation_node.rs @@ -203,11 +203,12 @@ pub(crate) fn prepare_sync_offer_snapshot( .propagation_sync_limit .map(kilobytes_to_bytes_fail_closed); let mut handled_messages = snapshot.policy.handled_messages.clone(); - if let Some(path) = snapshot.peer_path.as_ref() - && let Ok(data) = std::fs::read(path) - && let Some(peer) = LxmPeer::from_bytes_with_handled(&data) - { - handled_messages.extend(peer.handled_messages); + if let Some(path) = snapshot.peer_path.as_ref() { + if let Ok(data) = std::fs::read(path) { + if let Some(peer) = LxmPeer::from_bytes_with_handled(&data) { + handled_messages.extend(peer.handled_messages); + } + } } let mut candidates = snapshot.candidates; @@ -631,10 +632,10 @@ impl PropagationNode { self.advance_offer_generation(); } - if before > after - && let Some(ref dir) = self.storage_path - { - self.cleanup_orphaned_files(dir); + if before > after { + if let Some(ref dir) = self.storage_path { + self.cleanup_orphaned_files(dir); + } } } @@ -652,10 +653,10 @@ impl PropagationNode { if filename.ends_with(".peer") || filename.ends_with(".msgpack") { continue; } - if let Some((tid, _, _)) = PropagationEntry::parse_filename(&filename) - && !self.store.contains(&tid) - { - let _ = std::fs::remove_file(&path); + if let Some((tid, _, _)) = PropagationEntry::parse_filename(&filename) { + if !self.store.contains(&tid) { + let _ = std::fs::remove_file(&path); + } } } } @@ -939,17 +940,19 @@ impl PropagationNode { let mut reads = Vec::new(); if let Some(wants_arr) = arr[0].as_array() { for want_val in wants_arr { - if let Some(tid) = parse_store_id(want_val) - && let Some(ref dir) = self.storage_path - && let Some(entry) = self.store.get(&tid) - // Ownership gate (Python LXMRouter.py:1479): a client - // may only download messages addressed to itself. - && entry.destination_hash == *client_dest_hash + if let (Some(tid), Some(dir)) = + (parse_store_id(want_val), self.storage_path.as_ref()) { - reads.push(PlannedRead { - path: dir.join(entry.filename()), - stamped: entry.stamped, - }); + if let Some(entry) = self.store.get(&tid) { + // Ownership gate (Python LXMRouter.py:1479): a client + // may only download messages addressed to itself. + if entry.destination_hash == *client_dest_hash { + reads.push(PlannedRead { + path: dir.join(entry.filename()), + stamped: entry.stamped, + }); + } + } } } } @@ -1111,10 +1114,11 @@ impl PropagationNode { // Compatibility wrapper: production uses the staged task // path and exposes this delta to the daemon for off-loop // persistence. - if !terminal_handled_ids.is_empty() - && let Err(error) = self.persist_peer_handled(policy, &terminal_handled_ids) - { - tracing::warn!(%error, "failed to persist terminal offer dispositions"); + if !terminal_handled_ids.is_empty() { + if let Err(error) = self.persist_peer_handled(policy, &terminal_handled_ids) + { + tracing::warn!(%error, "failed to persist terminal offer dispositions"); + } } return offer; } @@ -1228,12 +1232,13 @@ impl PropagationNode { if let Ok(entries) = std::fs::read_dir(dir) { for entry in entries.flatten() { let path = entry.path(); - if path.extension().map(|e| e == "peer").unwrap_or(false) - && let Ok(data) = std::fs::read(&path) - && let Some(mut peer) = LxmPeer::from_bytes_with_handled(&data) - { - self.prune_handled_against_store(&mut peer); - peers.push(peer); + if path.extension().map(|e| e == "peer").unwrap_or(false) { + if let Ok(data) = std::fs::read(&path) { + if let Some(mut peer) = LxmPeer::from_bytes_with_handled(&data) { + self.prune_handled_against_store(&mut peer); + peers.push(peer); + } + } } } } diff --git a/crates/lxmf-core/src/propagation_sync.rs b/crates/lxmf-core/src/propagation_sync.rs index bae8686..d1bbb5b 100644 --- a/crates/lxmf-core/src/propagation_sync.rs +++ b/crates/lxmf-core/src/propagation_sync.rs @@ -400,22 +400,25 @@ impl PropagationSyncTask { continue; } let node_hex = self.node_dest_hash.map(|h| hex_encode(&h)); - if let Some(node_hex) = node_hex - && let Some(pub_key) = known_identities.get(&node_hex) - { - let ed25519_bytes: [u8; 32] = pub_key[32..64].try_into().unwrap(); - if let Ok(verify_key) = Ed25519PublicKey::from_bytes(&ed25519_bytes) - { - self.handle_link_proof(data, &verify_key, &ed25519_bytes); + if let Some(node_hex) = node_hex { + if let Some(pub_key) = known_identities.get(&node_hex) { + let ed25519_bytes: [u8; 32] = + pub_key[32..64].try_into().unwrap(); + if let Ok(verify_key) = + Ed25519PublicKey::from_bytes(&ed25519_bytes) + { + self.handle_link_proof(data, &verify_key, &ed25519_bytes); + } } } } rns_wire::context::PacketContext::ResourceHmu => { - if let Some(ref link) = self.link - && let Ok(plaintext) = link.decrypt(data) - && let Some(ref mut transfer) = self.active_transfer - { - transfer.handle_hmu(&plaintext); + if let Some(ref link) = self.link { + if let Ok(plaintext) = link.decrypt(data) { + if let Some(ref mut transfer) = self.active_transfer { + transfer.handle_hmu(&plaintext); + } + } } } rns_wire::context::PacketContext::ResourceReq => { @@ -434,12 +437,16 @@ impl PropagationSyncTask { } } rns_wire::context::PacketContext::Response => { - if self.state == SyncTaskState::AwaitingResponse - && let Some(ref mut link) = self.link - && let Ok((_request_id, response_data)) = link.handle_response(data) - { - let offer_response = OfferResponse::from_msgpack(&response_data); - self.handle_offer_response(offer_response); + if self.state == SyncTaskState::AwaitingResponse { + if let Some(ref mut link) = self.link { + if let Ok((_request_id, response_data)) = + link.handle_response(data) + { + let offer_response = + OfferResponse::from_msgpack(&response_data); + self.handle_offer_response(offer_response); + } + } } } rns_wire::context::PacketContext::ResourceRcl @@ -694,10 +701,10 @@ impl PropagationSyncTask { fn prepare_transfer_for_ids(&mut self, ids: &[PropagationTransientId]) { if ids.is_empty() { - if let Some(node_hash) = self.node_dest_hash - && let Ok(mut node) = self.propagation_node.lock() - { - node.complete_sync(&node_hash); + if let Some(node_hash) = self.node_dest_hash { + if let Ok(mut node) = self.propagation_node.lock() { + node.complete_sync(&node_hash); + } } self.state = SyncTaskState::Complete; return; @@ -746,13 +753,12 @@ impl PropagationSyncTask { } pub fn tick(&mut self) { - if let Some(started) = self.sync_started - && started.elapsed() > self.sync_timeout - && self.state != SyncTaskState::Idle - { - self.cleanup_sync(); - self.state = SyncTaskState::Failed; - return; + if let Some(started) = self.sync_started { + if started.elapsed() > self.sync_timeout && self.state != SyncTaskState::Idle { + self.cleanup_sync(); + self.state = SyncTaskState::Failed; + return; + } } match self.state { @@ -760,12 +766,13 @@ impl PropagationSyncTask { if self.terminal_result.is_none() && self.offer_policy.is_none() && self.last_sync.elapsed() >= self.sync_interval - && let Some(node_hash) = self.node_dest_hash { - if self.message_count() > 0 { - self.start_sync(node_hash); - } else { - self.last_sync = Instant::now(); + if let Some(node_hash) = self.node_dest_hash { + if self.message_count() > 0 { + self.start_sync(node_hash); + } else { + self.last_sync = Instant::now(); + } } } } @@ -777,22 +784,22 @@ impl PropagationSyncTask { self.drive_transfers(); } SyncTaskState::Complete | SyncTaskState::Failed => { - if self.terminal_result.is_none() - && let Some(peer_hash) = self.node_dest_hash - { - let complete = self.state == SyncTaskState::Complete; - self.terminal_result = Some(PeerSyncTerminalResult { - peer_hash, - state: if complete { - PeerSyncTerminalState::Complete - } else { - PeerSyncTerminalState::Failed - }, - offer_generation: complete - .then_some(self.active_offer_generation) - .flatten(), - generation_exhausted: complete && self.generation_exhausted, - }); + if self.terminal_result.is_none() { + if let Some(peer_hash) = self.node_dest_hash { + let complete = self.state == SyncTaskState::Complete; + self.terminal_result = Some(PeerSyncTerminalResult { + peer_hash, + state: if complete { + PeerSyncTerminalState::Complete + } else { + PeerSyncTerminalState::Failed + }, + offer_generation: complete + .then_some(self.active_offer_generation) + .flatten(), + generation_exhausted: complete && self.generation_exhausted, + }); + } } self.cleanup_sync(); self.last_sync = Instant::now(); @@ -895,10 +902,10 @@ impl PropagationSyncTask { self.active_offer_generation = Some(generation); self.generation_exhausted = generation_exhausted; self.record_handled_updates(&terminal_handled_ids); - if offer.transient_ids.is_empty() - && let Ok(mut node) = self.propagation_node.lock() - { - node.complete_sync(&node_hash); + if offer.transient_ids.is_empty() { + if let Ok(mut node) = self.propagation_node.lock() { + node.complete_sync(&node_hash); + } } offer } @@ -1127,10 +1134,10 @@ impl PropagationSyncTask { let completed_ids = std::mem::take(&mut self.active_transfer_ids); self.record_handled_updates(&completed_ids); - if let Some(node_hash) = self.node_dest_hash - && let Ok(mut node) = self.propagation_node.lock() - { - node.complete_sync(&node_hash); + if let Some(node_hash) = self.node_dest_hash { + if let Ok(mut node) = self.propagation_node.lock() { + node.complete_sync(&node_hash); + } } self.state = SyncTaskState::Complete; } diff --git a/crates/lxmf-core/src/router.rs b/crates/lxmf-core/src/router.rs index bccf34a..fde8132 100644 --- a/crates/lxmf-core/src/router.rs +++ b/crates/lxmf-core/src/router.rs @@ -480,10 +480,10 @@ impl LxmRouter { } let now = now_f64(); - if message.outbound_ticket.is_none() - && let Some(ticket) = self.ticket_store.find(&message.destination_hash, now) - { - message.outbound_ticket = Some(ticket.token); + if message.outbound_ticket.is_none() { + if let Some(ticket) = self.ticket_store.find(&message.destination_hash, now) { + message.outbound_ticket = Some(ticket.token); + } } if message.stamp.is_none() && message.stamp_cost.is_none() { @@ -514,16 +514,16 @@ impl LxmRouter { } } - if message.method == DeliveryMethod::Opportunistic - && let Ok(packed) = message.pack_payload() - { - let content_size = packed - .len() - .saturating_sub(TIMESTAMP_SIZE + STRUCT_OVERHEAD); - // Approximates ENCRYPTED_PACKET_MAX_CONTENT for default RNS parameters. - let max_content = 295; - if content_size > max_content { - message.method = DeliveryMethod::Direct; + if message.method == DeliveryMethod::Opportunistic { + if let Ok(packed) = message.pack_payload() { + let content_size = packed + .len() + .saturating_sub(TIMESTAMP_SIZE + STRUCT_OVERHEAD); + // Approximates ENCRYPTED_PACKET_MAX_CONTENT for default RNS parameters. + let max_content = 295; + if content_size > max_content { + message.method = DeliveryMethod::Direct; + } } } @@ -817,13 +817,14 @@ impl LxmRouter { if let Some(mut msg) = self.pending_deferred_stamps.remove(message_hash) { msg.cancel(); - if self + let active_matches = self .active_deferred_stamp .as_ref() - .is_some_and(|job| job.message_hash == *message_hash) - && let Some(job) = self.active_deferred_stamp.take() - { - job.handle.cancel(); + .is_some_and(|job| job.message_hash == *message_hash); + if active_matches { + if let Some(job) = self.active_deferred_stamp.take() { + job.handle.cancel(); + } } return true; } @@ -1170,8 +1171,7 @@ impl LxmRouter { return false; } if !configured_static - && let Some(h) = hops - && h as usize > self.config.ext.autopeer_maxdepth + && hops.is_some_and(|h| h as usize > self.config.ext.autopeer_maxdepth) { return false; } @@ -1546,8 +1546,11 @@ impl LxmRouter { let mut i = 0; while i < self.pending_outbound.len() { - if let Some(limit) = self.config.ext.processing_limit - && processed >= limit + if self + .config + .ext + .processing_limit + .is_some_and(|limit| processed >= limit) { break; } @@ -1665,8 +1668,11 @@ impl LxmRouter { let mut i = 0; while i < self.pending_outbound.len() { - if let Some(limit) = self.config.ext.processing_limit - && processed >= limit + if self + .config + .ext + .processing_limit + .is_some_and(|limit| processed >= limit) { break; } @@ -1955,20 +1961,16 @@ impl LxmRouter { fn run_periodic_jobs(&mut self) { // Job cadences match the Python LXMRouter jobloop. - if self.processing_count.is_multiple_of(JOB_TRANSIENT_INTERVAL) { + if self.processing_count % JOB_TRANSIENT_INTERVAL == 0 { self.propagation_store.clean_transient_caches(now_f64()); } - if self.processing_count.is_multiple_of(JOB_STORE_INTERVAL) - && self.config.propagation_enabled - { + if self.processing_count % JOB_STORE_INTERVAL == 0 && self.config.propagation_enabled { self.cull_propagation(); } - if self.processing_count.is_multiple_of(JOB_PEERSYNC_INTERVAL) { + if self.processing_count % JOB_PEERSYNC_INTERVAL == 0 { self.clean_throttled_peers(); } - if self.processing_count.is_multiple_of(JOB_ROTATE_INTERVAL) - && self.config.propagation_enabled - { + if self.processing_count % JOB_ROTATE_INTERVAL == 0 && self.config.propagation_enabled { self.rotate_peers(); } } diff --git a/crates/lxmf-tools/src/commands/lxmd.rs b/crates/lxmf-tools/src/commands/lxmd.rs index acaa8b9..2eb468e 100644 --- a/crates/lxmf-tools/src/commands/lxmd.rs +++ b/crates/lxmf-tools/src/commands/lxmd.rs @@ -841,10 +841,10 @@ impl LxmdRunner { if let Ok(entries) = std::fs::read_dir(&received_dir) { for entry in entries.flatten() { let path = entry.path(); - if let Some(name) = path.file_stem().and_then(|n| n.to_str()) - && let Ok(rr) = ReceivedRatchet::load(&path) - { - received_ratchets.insert(name.to_string(), rr); + if let Some(name) = path.file_stem().and_then(|n| n.to_str()) { + if let Ok(rr) = ReceivedRatchet::load(&path) { + received_ratchets.insert(name.to_string(), rr); + } } } } @@ -852,17 +852,17 @@ impl LxmdRunner { // known_identities format: concat of [dest_hash:16][pubkey:64] let ki_path = paths.known_identities_path.clone(); let mut known_identities: HashMap = HashMap::new(); - if ki_path.exists() - && let Ok(data) = std::fs::read(&ki_path) - { - let mut pos = 0; - while pos + 80 <= data.len() { - let mut dh = [0u8; 16]; - dh.copy_from_slice(&data[pos..pos + 16]); - let mut pk = [0u8; 64]; - pk.copy_from_slice(&data[pos + 16..pos + 80]); - known_identities.insert(hex::encode(dh), pk); - pos += 80; + if ki_path.exists() { + if let Ok(data) = std::fs::read(&ki_path) { + let mut pos = 0; + while pos + 80 <= data.len() { + let mut dh = [0u8; 16]; + dh.copy_from_slice(&data[pos..pos + 16]); + let mut pk = [0u8; 64]; + pk.copy_from_slice(&data[pos + 16..pos + 80]); + known_identities.insert(hex::encode(dh), pk); + pos += 80; + } } } @@ -1577,14 +1577,15 @@ impl LxmdRunner { // Upstream control unpeer breaks the live peering without // mutating the operator's configured static-peer set. self.router.remove_peer(&peer_hash); - if let Some(ref node) = self.propagation_node - && let Ok(mut node) = node.lock() - && let Err(error) = node.delete_peer(&peer_hash) - { - tracing::warn!( - peer = %hex::encode(peer_hash), - "failed to remove persisted propagation peer: {error}" - ); + if let Some(ref node) = self.propagation_node { + if let Ok(mut node) = node.lock() { + if let Err(error) = node.delete_peer(&peer_hash) { + tracing::warn!( + peer = %hex::encode(peer_hash), + "failed to remove persisted propagation peer: {error}" + ); + } + } } if let Err(e) = self.router.save_state(&self.data_dir) { tracing::warn!("Failed to save router state after control unpeer: {e}"); @@ -1607,11 +1608,11 @@ impl LxmdRunner { // announce timebase (or a lower current target) does not make // a completed key stale if its measured value is still high // enough for the current policy. - if let Some((key, value)) = result.peering_key - && value >= peer.peering_cost as u32 - { - peer.peering_key = Some((key, value)); - applied = true; + if let Some((key, value)) = result.peering_key { + if value >= peer.peering_cost as u32 { + peer.peering_key = Some((key, value)); + applied = true; + } } } @@ -1619,13 +1620,15 @@ impl LxmdRunner { if let (Some(node), Some(peer)) = ( self.propagation_node.as_ref(), self.router.peers.get(&result.peer_hash), - ) && let Ok(node) = node.lock() - && let Err(error) = node.save_peer(peer) - { - tracing::warn!( - peer = %hex::encode(result.peer_hash), - "failed to persist generated peering key: {error}" - ); + ) { + if let Ok(node) = node.lock() { + if let Err(error) = node.save_peer(peer) { + tracing::warn!( + peer = %hex::encode(result.peer_hash), + "failed to persist generated peering key: {error}" + ); + } + } } } else if current_cost { // A bounded key search can fail for an excessive advertised @@ -1690,13 +1693,13 @@ impl LxmdRunner { .then(|| OutboundOfferPolicy::from(peer)) }); if let Some(policy) = policy { - if let Some(sync) = self.propagation_sync.as_mut() - && sync.request_sync_now_with_policy(policy) - { - if let Some(peer) = self.router.peers.get_mut(&peer_hash) { - peer.begin_sync(); + if let Some(sync) = self.propagation_sync.as_mut() { + if sync.request_sync_now_with_policy(policy) { + if let Some(peer) = self.router.peers.get_mut(&peer_hash) { + peer.begin_sync(); + } + self.pending_peer_syncs.remove(&peer_hash); } - self.pending_peer_syncs.remove(&peer_hash); } return; } @@ -2058,51 +2061,53 @@ impl LxmdRunner { ps.drain_events(&self.known_identities); ps.tick(); let updates = ps.take_handled_updates(); - if !updates.is_empty() - && let Some(peer_hash) = ps.node_dest_hash() - { - peer_handled_updates = Some((peer_hash, updates)); + if !updates.is_empty() { + if let Some(peer_hash) = ps.node_dest_hash() { + peer_handled_updates = Some((peer_hash, updates)); + } } peer_terminal_result = ps.take_terminal_peer_result(); } let mut peers_to_persist = HashSet::new(); - if let Some((peer_hash, updates)) = peer_handled_updates - && let Some(peer) = self.router.peers.get_mut(&peer_hash) - { - for transient_id in updates { - peer.add_handled_message(&transient_id); + if let Some((peer_hash, updates)) = peer_handled_updates { + if let Some(peer) = self.router.peers.get_mut(&peer_hash) { + for transient_id in updates { + peer.add_handled_message(&transient_id); + } + peers_to_persist.insert(peer_hash); } - peers_to_persist.insert(peer_hash); } - if let Some(result) = peer_terminal_result - && let Some(peer) = self.router.peers.get_mut(&result.peer_hash) - { - match result.state { - lxmf_core::propagation_sync::PeerSyncTerminalState::Complete => { - peer.sync_complete(); - if result.generation_exhausted - && let Some(generation) = result.offer_generation - { - peer.mark_offer_generation_processed(generation); + if let Some(result) = peer_terminal_result { + if let Some(peer) = self.router.peers.get_mut(&result.peer_hash) { + match result.state { + lxmf_core::propagation_sync::PeerSyncTerminalState::Complete => { + peer.sync_complete(); + if result.generation_exhausted { + if let Some(generation) = result.offer_generation { + peer.mark_offer_generation_processed(generation); + } + } + } + lxmf_core::propagation_sync::PeerSyncTerminalState::Failed => { + peer.sync_failed(); } } - lxmf_core::propagation_sync::PeerSyncTerminalState::Failed => { - peer.sync_failed(); - } + peers_to_persist.insert(result.peer_hash); } - peers_to_persist.insert(result.peer_hash); } for peer_hash in peers_to_persist { if let (Some(node), Some(peer)) = ( self.propagation_node.as_ref(), self.router.peers.get(&peer_hash), - ) && let Ok(node) = node.lock() - && let Err(error) = node.save_peer(peer) - { - tracing::warn!( - peer = %hex::encode(peer_hash), - "failed to persist peer sync state: {error}" - ); + ) { + if let Ok(node) = node.lock() { + if let Err(error) = node.save_peer(peer) { + tracing::warn!( + peer = %hex::encode(peer_hash), + "failed to persist peer sync state: {error}" + ); + } + } } } @@ -2129,18 +2134,18 @@ impl LxmdRunner { client.start_download(); self.last_propagation_check = now; tracing::debug!("auto-triggered propagation download"); - } else if let Some(node) = self.router.outbound_propagation_node - && queue_unknown_propagation_node_path_request( + } else if let Some(node) = self.router.outbound_propagation_node { + if queue_unknown_propagation_node_path_request( &self.transport_tx, node, &mut self.last_propagation_check, now, - ) - { - tracing::debug!( - node = %hex::encode(node), - "propagation node identity unknown; requesting path before download" - ); + ) { + tracing::debug!( + node = %hex::encode(node), + "propagation node identity unknown; requesting path before download" + ); + } } } let status = client.transfer_status(); @@ -2158,52 +2163,58 @@ impl LxmdRunner { for msg_data in downloaded_messages { self.handle_propagation_downloaded_data(&msg_data); } - if acknowledge_propagation && let Some(client) = self.propagation_client.as_mut() { - client.acknowledge_transfer(); - } - - if let Some(interval) = self.config.announce_interval - && now - self.last_peer_announce > interval as f64 - { - let tx = self.transport_tx.clone(); - if let Ok(raw) = self.create_announce_packet() { - let dest = self.lxmf_dest_hash; - let _ = tx.try_send(TransportMessage::Outbound( - rns_transport::messages::OutboundRequest { - raw: Bytes::from(raw), - destination_hash: dest, - }, - )); - self.last_peer_announce = now; - tracing::debug!("periodic peer announce sent"); + if acknowledge_propagation { + if let Some(client) = self.propagation_client.as_mut() { + client.acknowledge_transfer(); } } - if self.config.propagation_enabled - && let Some(interval) = self.config.node_announce_interval - && now - self.last_node_announce > interval as f64 - && let Ok(raw) = self.create_propagation_announce_packet() - { - let dest = self.propagation_dest_hash; - let _ = self.transport_tx.try_send(TransportMessage::Outbound( - rns_transport::messages::OutboundRequest { - raw: Bytes::from(raw), - destination_hash: dest, - }, - )); - if self.should_announce_control() - && let Ok(raw) = - create_control_announce_packet(&self.identity, self.control_dest_hash) - { - let _ = self.transport_tx.try_send(TransportMessage::Outbound( - rns_transport::messages::OutboundRequest { - raw: Bytes::from(raw), - destination_hash: self.control_dest_hash, - }, - )); + if let Some(interval) = self.config.announce_interval { + if now - self.last_peer_announce > interval as f64 { + let tx = self.transport_tx.clone(); + if let Ok(raw) = self.create_announce_packet() { + let dest = self.lxmf_dest_hash; + let _ = tx.try_send(TransportMessage::Outbound( + rns_transport::messages::OutboundRequest { + raw: Bytes::from(raw), + destination_hash: dest, + }, + )); + self.last_peer_announce = now; + tracing::debug!("periodic peer announce sent"); + } + } + } + + if self.config.propagation_enabled { + if let Some(interval) = self.config.node_announce_interval { + if now - self.last_node_announce > interval as f64 { + if let Ok(raw) = self.create_propagation_announce_packet() { + let dest = self.propagation_dest_hash; + let _ = self.transport_tx.try_send(TransportMessage::Outbound( + rns_transport::messages::OutboundRequest { + raw: Bytes::from(raw), + destination_hash: dest, + }, + )); + if self.should_announce_control() { + if let Ok(raw) = create_control_announce_packet( + &self.identity, + self.control_dest_hash, + ) { + let _ = self.transport_tx.try_send(TransportMessage::Outbound( + rns_transport::messages::OutboundRequest { + raw: Bytes::from(raw), + destination_hash: self.control_dest_hash, + }, + )); + } + } + self.last_node_announce = now; + tracing::debug!("periodic propagation node announce sent"); + } + } } - self.last_node_announce = now; - tracing::debug!("periodic propagation node announce sent"); } if now - self.last_cull > 300.0 { @@ -2212,10 +2223,10 @@ impl LxmdRunner { // jobloop cadences. The propagation node's own store (separate // from the router's) ages out expired messages and enforces the // weight cap here — previously this never ran. - if let Some(ref pn) = self.propagation_node - && let Ok(mut node) = pn.lock() - { - node.tick(); + if let Some(ref pn) = self.propagation_node { + if let Ok(mut node) = pn.lock() { + node.tick(); + } } self.last_cull = now; } @@ -2264,20 +2275,21 @@ impl LxmdRunner { .insert(event.destination_hash, event.hops.max(1)); if event.name_hash == delivery_name_hash { - if let Some(ref data) = event.app_data - && let Some((display_name, stamp_cost)) = + if let Some(ref data) = event.app_data { + if let Some((display_name, stamp_cost)) = lxmf_core::handlers::parse_announce_app_data(data) - { - if let Some(name) = display_name { - tracing::info!(dest = %dest_hex, name = %name, "announce display name"); - } - if let Some(cost) = stamp_cost { - self.router.set_stamp_cost(event.destination_hash, cost); - tracing::debug!( - dest = %dest_hex, - stamp_cost = cost, - "learned delivery stamp cost from announce" - ); + { + if let Some(name) = display_name { + tracing::info!(dest = %dest_hex, name = %name, "announce display name"); + } + if let Some(cost) = stamp_cost { + self.router.set_stamp_cost(event.destination_hash, cost); + tracing::debug!( + dest = %dest_hex, + stamp_cost = cost, + "learned delivery stamp cost from announce" + ); + } } } let triggered = self @@ -2290,9 +2302,13 @@ impl LxmdRunner { "delivery announce made pending outbound messages eligible" ); } - } else if event.name_hash == propagation_name_hash - && let Some(ref data) = event.app_data - && let Some(pn) = lxmf_core::handlers::parse_pn_announce_data(data) + } else if let Some((data, pn)) = event + .app_data + .as_deref() + .filter(|_| event.name_hash == propagation_name_hash) + .and_then(|data| { + lxmf_core::handlers::parse_pn_announce_data(data).map(|pn| (data, pn)) + }) { self.router .set_stamp_cost(event.destination_hash, pn.stamp_cost); @@ -2347,14 +2363,15 @@ impl LxmdRunner { if let Some(sync) = self.propagation_sync.as_mut() { sync.cancel_peer_sync(&event.destination_hash); } - if let Some(node) = self.propagation_node.as_ref() - && let Ok(mut node) = node.lock() - && let Err(error) = node.delete_peer(&event.destination_hash) - { - tracing::warn!( - peer = %dest_hex, - "failed to remove retired propagation peer: {error}" - ); + if let Some(node) = self.propagation_node.as_ref() { + if let Ok(mut node) = node.lock() { + if let Err(error) = node.delete_peer(&event.destination_hash) { + tracing::warn!( + peer = %dest_hex, + "failed to remove retired propagation peer: {error}" + ); + } + } } } else if peer_changed { let offer_constraints_changed = @@ -2379,13 +2396,15 @@ impl LxmdRunner { if let (Some(node), Some(peer)) = ( self.propagation_node.as_ref(), self.router.peers.get(&event.destination_hash), - ) && let Ok(node) = node.lock() - && let Err(error) = node.save_peer(peer) - { - tracing::warn!( - peer = %dest_hex, - "failed to persist propagation peer policy: {error}" - ); + ) { + if let Ok(node) = node.lock() { + if let Err(error) = node.save_peer(peer) { + tracing::warn!( + peer = %dest_hex, + "failed to persist propagation peer policy: {error}" + ); + } + } } } tracing::debug!( @@ -2404,34 +2423,35 @@ impl LxmdRunner { ); } } - if let Some(pub_key) = event.public_key - && self.known_identities.get(&dest_hex) != Some(&pub_key) - { - self.known_identities.insert(dest_hex.clone(), pub_key); - tracing::debug!(dest = %dest_hex, "learned identity key from announce"); + if let Some(pub_key) = event.public_key { + if self.known_identities.get(&dest_hex) != Some(&pub_key) { + self.known_identities.insert(dest_hex.clone(), pub_key); + tracing::debug!(dest = %dest_hex, "learned identity key from announce"); + } } // Python Identity._remember_ratchet: persist only the single // changed ratchet, off the daemon loop. Identity keys and the // ring stay on the periodic/shutdown saves. - if let Some(ratchet_key) = event.ratchet - && self + if let Some(ratchet_key) = event.ratchet { + if self .received_ratchets .get(&dest_hex) .is_none_or(|rr| rr.ratchet_pub != ratchet_key) - { - let rr = ReceivedRatchet::new(ratchet_key); - self.received_ratchets.insert(dest_hex.clone(), rr); - tracing::debug!(dest = %dest_hex, "learned ratchet from announce"); - let path = self - .received_ratchets_dir - .join(format!("{dest_hex}.ratchet")); - let dir = self.received_ratchets_dir.clone(); - tokio::task::spawn_blocking(move || { - std::fs::create_dir_all(&dir).ok(); - if let Err(e) = rr.save(&path) { - tracing::warn!("Failed to persist received ratchet: {e}"); - } - }); + { + let rr = ReceivedRatchet::new(ratchet_key); + self.received_ratchets.insert(dest_hex.clone(), rr); + tracing::debug!(dest = %dest_hex, "learned ratchet from announce"); + let path = self + .received_ratchets_dir + .join(format!("{dest_hex}.ratchet")); + let dir = self.received_ratchets_dir.clone(); + tokio::task::spawn_blocking(move || { + std::fs::create_dir_all(&dir).ok(); + if let Err(e) = rr.save(&path) { + tracing::warn!("Failed to persist received ratchet: {e}"); + } + }); + } } } seen @@ -2451,9 +2471,10 @@ impl LxmdRunner { direction: LinkResourceDirection::Inbound, .. } - ) && let Some(event) = delivery_resource_event_from_runtime(event) - { - self.router.handle_inbound_resource_event(event); + ) { + if let Some(event) = delivery_resource_event_from_runtime(event) { + self.router.handle_inbound_resource_event(event); + } } } @@ -2617,17 +2638,17 @@ impl LxmdRunner { }; let mut accepted = 0usize; - if let Some(ref node) = self.propagation_node - && let Ok(mut node) = node.lock() - { - for entry in result.entries { - let stamp_value = u8::try_from(entry.stamp_value).unwrap_or(u8::MAX); - if node.accept_stamped_propagated_blob( - &entry.lxmf_data, - &entry.stamp_data, - stamp_value, - ) { - accepted += 1; + if let Some(ref node) = self.propagation_node { + if let Ok(mut node) = node.lock() { + for entry in result.entries { + let stamp_value = u8::try_from(entry.stamp_value).unwrap_or(u8::MAX); + if node.accept_stamped_propagated_blob( + &entry.lxmf_data, + &entry.stamp_data, + stamp_value, + ) { + accepted += 1; + } } } } @@ -2640,26 +2661,26 @@ impl LxmdRunner { "processed inbound propagation Resource" ); - if claim.should_close_link() - && let Some(command_tx) = self.prop_link_command_tx.clone() - { - let link_id = claim.link_id(); - tokio::spawn(async move { - if command_tx - .send(rns_runtime::link_manager::LinkManagerCommand::CloseLink { - link_id, - reason: rns_runtime::prelude::CloseReason::DestinationClosed, - send_teardown: true, - }) - .await - .is_err() - { - tracing::debug!( - link_id = %hex::encode(link_id), - "propagation Link already closed before validation teardown" - ); - } - }); + if claim.should_close_link() { + if let Some(command_tx) = self.prop_link_command_tx.clone() { + let link_id = claim.link_id(); + tokio::spawn(async move { + if command_tx + .send(rns_runtime::link_manager::LinkManagerCommand::CloseLink { + link_id, + reason: rns_runtime::prelude::CloseReason::DestinationClosed, + send_teardown: true, + }) + .await + .is_err() + { + tracing::debug!( + link_id = %hex::encode(link_id), + "propagation Link already closed before validation teardown" + ); + } + }); + } } } @@ -2910,14 +2931,15 @@ impl LxmdRunner { // Also deposit into the propagation store (if enabled) so peers can // download it via offer/get sync. - if let Some(ref pn) = self.propagation_node - && let Ok(mut node) = pn.lock() - && node.accept_message(&msg) - { - tracing::info!( - from = %hex::encode(msg.source_hash), - "propagation: message accepted into store" - ); + if let Some(ref pn) = self.propagation_node { + if let Ok(mut node) = pn.lock() { + if node.accept_message(&msg) { + tracing::info!( + from = %hex::encode(msg.source_hash), + "propagation: message accepted into store" + ); + } + } } let messages_dir = self.messages_dir.clone(); @@ -2951,10 +2973,10 @@ impl LxmdRunner { } // Execute on_inbound command if configured - if let Some(ref cmd) = self.config.on_inbound_command - && let Err(e) = execute_on_inbound(cmd, &msg_path.to_string_lossy()) - { - tracing::error!("on_inbound command failed: {e}"); + if let Some(ref cmd) = self.config.on_inbound_command { + if let Err(e) = execute_on_inbound(cmd, &msg_path.to_string_lossy()) { + tracing::error!("on_inbound command failed: {e}"); + } } // Update known identity from sender @@ -3019,24 +3041,25 @@ impl LxmdRunner { } let hops = route_hops_for(&self.route_hops, prop_hash); self.ensure_link_delivery(); - if let Some(ref mut ld) = self.link_delivery - && let Err(err) = ld + if let Some(ref mut ld) = self.link_delivery { + if let Err(err) = ld .start_packed_delivery(message, prop_hash, hops, packed, false) - { - let reason = err.error.to_string(); - tracing::warn!( - error = %reason, - prop = %hex::encode(prop_hash), - "failed to start propagated link delivery" - ); - requeue_after_path_request( - &mut self.router, - &self.transport_tx, - *err.message, - prop_hash, - &reason, - false, - ); + { + let reason = err.error.to_string(); + tracing::warn!( + error = %reason, + prop = %hex::encode(prop_hash), + "failed to start propagated link delivery" + ); + requeue_after_path_request( + &mut self.router, + &self.transport_tx, + *err.message, + prop_hash, + &reason, + false, + ); + } } } None => { @@ -3052,17 +3075,18 @@ impl LxmdRunner { OutboundAction::Failed(_) | OutboundAction::Expired(_) => continue, }; - if message.stamp.is_none() - && let Some(cost) = self.router.get_stamp_cost(&message.destination_hash) - && cost > 0 - { - tracing::info!( - dest = %hex::encode(message.destination_hash), - cost = cost, - "generating stamp" - ); - message.stamp_cost = Some(cost); - message.get_stamp(); + if message.stamp.is_none() { + if let Some(cost) = self.router.get_stamp_cost(&message.destination_hash) { + if cost > 0 { + tracing::info!( + dest = %hex::encode(message.destination_hash), + cost = cost, + "generating stamp" + ); + message.stamp_cost = Some(cost); + message.get_stamp(); + } + } } let dest_hex = hex::encode(dest_hash); @@ -3316,23 +3340,23 @@ impl LxmdRunner { } let hops = route_hops_for(&self.route_hops, dest_hash); self.ensure_link_delivery(); - if let Some(ref mut ld) = self.link_delivery - && let Err(err) = ld.start_delivery(message, dest_hash, hops) - { - let reason = err.error.to_string(); - tracing::warn!( - error = %reason, - dest = %dest_hex, - "failed to start oversized direct link delivery" - ); - requeue_after_path_request( - &mut self.router, - &self.transport_tx, - *err.message, - dest_hash, - &reason, - false, - ); + if let Some(ref mut ld) = self.link_delivery { + if let Err(err) = ld.start_delivery(message, dest_hash, hops) { + let reason = err.error.to_string(); + tracing::warn!( + error = %reason, + dest = %dest_hex, + "failed to start oversized direct link delivery" + ); + requeue_after_path_request( + &mut self.router, + &self.transport_tx, + *err.message, + dest_hash, + &reason, + false, + ); + } } continue; } @@ -3491,11 +3515,11 @@ impl LxmdRunner { let ki_path = ratchet_dir.join("known_identities"); let mut data = Vec::with_capacity(self.known_identities.len() * 80); for (hash_hex, pk) in &self.known_identities { - if let Ok(hash_bytes) = hex::decode(hash_hex) - && hash_bytes.len() == 16 - { - data.extend_from_slice(&hash_bytes); - data.extend_from_slice(pk); + if let Ok(hash_bytes) = hex::decode(hash_hex) { + if hash_bytes.len() == 16 { + data.extend_from_slice(&hash_bytes); + data.extend_from_slice(pk); + } } } if let Err(e) = rns_identity::persistence::atomic_write(&ki_path, &data) { @@ -3850,9 +3874,7 @@ pub(crate) async fn main() { tracing::info!("On-inbound command: {}", cmd); } - if !shutdown.is_triggered() - && let Some(ref send_args) = args.send - { + if let Some(send_args) = args.send.as_ref().filter(|_| !shutdown.is_triggered()) { let dest_hex = normalize_hash_hex(&send_args[0]); let content = match args.send_file.as_ref() { Some(path) => match std::fs::read_to_string(path) { @@ -4060,9 +4082,10 @@ pub(crate) async fn main() { direction: LinkResourceDirection::Inbound, .. } - ) && let Some(event) = delivery_resource_event_from_runtime(event) - { - runner.router.handle_inbound_resource_event(event); + ) { + if let Some(event) = delivery_resource_event_from_runtime(event) { + runner.router.handle_inbound_resource_event(event); + } } runner.drain_link_packets(); } diff --git a/crates/lxmf-tools/src/daemon.rs b/crates/lxmf-tools/src/daemon.rs index b3e04a5..78b9b6f 100644 --- a/crates/lxmf-tools/src/daemon.rs +++ b/crates/lxmf-tools/src/daemon.rs @@ -321,16 +321,17 @@ impl DaemonConfig { dc.outbound_propagation_node = Some(trimmed.to_string()); } } - if get_int(Some(sec), "propagation_stamp_cost_target").is_none() - && let Some(cost) = sec.get_uint("propagation_stamp_cost") - { - dc.propagation_stamp_cost = cost as u8; + if get_int(Some(sec), "propagation_stamp_cost_target").is_none() { + if let Some(cost) = sec.get_uint("propagation_stamp_cost") { + dc.propagation_stamp_cost = cost as u8; + } } if get_float(Some(sec), "propagation_message_max_accepted_size").is_none() && get_float(Some(sec), "propagation_transfer_max_accepted_size").is_none() - && let Some(limit) = sec.get_uint("propagation_limit") { - dc.propagation_limit_kb = limit as usize; + if let Some(limit) = sec.get_uint("propagation_limit") { + dc.propagation_limit_kb = limit as usize; + } } dc.enforce_stamps = sec.get_bool_or("enforce_stamps", false); } @@ -339,14 +340,14 @@ impl DaemonConfig { if !dc.auth_required { dc.auth_required = sec.get_bool_or("auth_required", false); } - if dc.control_allowed.is_empty() - && let Some(allowed) = sec.get("allowed") - { - dc.control_allowed = allowed - .split(',') - .map(|s| s.trim().to_string()) - .filter(|s| !s.is_empty()) - .collect(); + if dc.control_allowed.is_empty() { + if let Some(allowed) = sec.get("allowed") { + dc.control_allowed = allowed + .split(',') + .map(|s| s.trim().to_string()) + .filter(|s| !s.is_empty()) + .collect(); + } } } diff --git a/crates/lxmf-tools/src/lxmd_control.rs b/crates/lxmf-tools/src/lxmd_control.rs index c54dda3..1c2369b 100644 --- a/crates/lxmf-tools/src/lxmd_control.rs +++ b/crates/lxmf-tools/src/lxmd_control.rs @@ -92,10 +92,10 @@ pub async fn resolve_remote_identity_hash( let wait = async { while let Some(event) = ann_rx.recv().await { - if event.destination_hash == remote_destination_hash - && let Some(identity_hash) = event.identity_hash - { - return Ok(identity_hash); + if event.destination_hash == remote_destination_hash { + if let Some(identity_hash) = event.identity_hash { + return Ok(identity_hash); + } } } Err(LinkClientError::PubkeyNotDiscovered) @@ -123,10 +123,10 @@ pub fn decode_control_response(response: &[u8]) -> ControlResponse { return ControlResponse::Success; }; - if let Some(code) = value.as_u64() - && let Some(error) = peer_error_from_code(code as u8) - { - return ControlResponse::Error(error); + if let Some(code) = value.as_u64() { + if let Some(error) = peer_error_from_code(code as u8) { + return ControlResponse::Error(error); + } } if value.is_nil() {