reticulum-ref/rsLXMF
1
0
Fork 0
mirror of https://github.com/ratspeak/rsLXMF synced 2026-08-12 18:08:14 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

ci: enforce strict Clippy across supported targets

Browse source
This commit is contained in:
DeFiDude 2026-08-06 18:49:06 -06:00
parent c0bd1ef82f
commit 2e124b1006
13 changed files with 738 additions and 652 deletions
Download patch file Download diff file Expand all files Collapse all files

32
.github/workflows/ci.yml vendored
View file

@ -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

2
clippy.toml Normal file
View file

@ -0,0 +1,2 @@
# Keep Clippy suggestions aligned with the public MSRV in Cargo.toml.
msrv = "1.85"

42
crates/lxmf-core/src/delivery_ratchet.rs
View file

@ -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");
}
}
}

250
crates/lxmf-core/src/link_delivery.rs
View file

@ -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;

45
crates/lxmf-core/src/message.rs
View file

@ -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<Vec<u8>> {
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<String, D::Error> {
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")

17
crates/lxmf-core/src/persist.rs
View file

@ -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

112
crates/lxmf-core/src/propagation_client.rs
View file

@ -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,
}));
}
}
}
}

71
crates/lxmf-core/src/propagation_node.rs
View file

@ -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);
}
}
}
}
}

123
crates/lxmf-core/src/propagation_sync.rs
View file

@ -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;
}

68
crates/lxmf-core/src/router.rs
View file

@ -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();
}
}

583
crates/lxmf-tools/src/commands/lxmd.rs
View file

@ -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<String, [u8; 64]> = 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();
}

29
crates/lxmf-tools/src/daemon.rs
View file

@ -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();
}
}
}

16
crates/lxmf-tools/src/lxmd_control.rs
View file

@ -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() {