diff --git a/Cargo.lock b/Cargo.lock index d5de69c2e..bddbf4432 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1149,7 +1149,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -2087,6 +2087,7 @@ dependencies = [ "libc", "lnomad", "md-5", + "metrics", "rand_core 0.6.4", "regex", "rmp-serde", @@ -2362,6 +2363,16 @@ dependencies = [ "autocfg", ] +[[package]] +name = "metrics" +version = "0.24.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "89550ee9f79e88fef3119de263694973a8adb26c21d75322164fb8c493039fe2" +dependencies = [ + "portable-atomic", + "rapidhash", +] + [[package]] name = "mime" version = "0.3.17" @@ -2516,7 +2527,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -3102,6 +3113,15 @@ dependencies = [ "rand_core 0.9.5", ] +[[package]] +name = "rapidhash" +version = "4.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5da7e78a036ce858e8d55b7e7dc8ba3a88b78350fd2155d3591bbd966b58589e" +dependencies = [ + "rustversion", +] + [[package]] name = "ratatui" version = "0.30.2" @@ -3407,7 +3427,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.11.0", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -3891,7 +3911,7 @@ dependencies = [ "getrandom 0.3.4", "once_cell", "rustix 1.1.3", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -3938,7 +3958,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "230a1b821ccbd75b185820a1f1ff7b14d21da1e442e22c0863ea5f08771a8874" dependencies = [ "rustix 1.1.3", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -4634,7 +4654,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] diff --git a/THIRD-PARTY-NOTICES b/THIRD-PARTY-NOTICES index 06b2ba6e3..992104ddc 100644 --- a/THIRD-PARTY-NOTICES +++ b/THIRD-PARTY-NOTICES @@ -29,15 +29,15 @@ Linked into lnsd, lnstest, lncp, lnstatus, lnomad, lblogd and lnflash. Built for x86_64-unknown-linux-musl and aarch64-unknown-linux-musl. -362 third-party crates, by licence: +364 third-party crates, by licence: - Apache-2.0 249 + Apache-2.0 250 BSD-2-Clause 1 BSD-3-Clause 6 CC0-1.0 1 CDLA-Permissive-2.0 2 ISC 5 - MIT 96 + MIT 97 MPL-2.0 2 Unicode-3.0 1 Zlib 1 @@ -925,7 +925,7 @@ Applies to: rusticata-macros 4.1.0 rustix 0.38.44 rustix 1.1.3 - rustls 0.23.41 + rustls 0.23.45 rustls-acme 0.15.3 scopeguard 1.2.0 serde-pickle 1.2.0 @@ -6517,6 +6517,215 @@ APPENDIX: How to apply the Apache License to your work. Apache-2.0 — Apache License 2.0 ------------------------------------------------------------------------ +Applies to: + rapidhash 4.5.1 + + Apache License + Version 2.0, January 2004 + http://www.apache.org/licenses/ + +TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION + +1. Definitions. + + "License" shall mean the terms and conditions for use, reproduction, + and distribution as defined by Sections 1 through 9 of this document. + + "Licensor" shall mean the copyright owner or entity authorized by + the copyright owner that is granting the License. + + "Legal Entity" shall mean the union of the acting entity and all + other entities that control, are controlled by, or are under common + control with that entity. For the purposes of this definition, + "control" means (i) the power, direct or indirect, to cause the + direction or management of such entity, whether by contract or + otherwise, or (ii) ownership of fifty percent (50%) or more of the + outstanding shares, or (iii) beneficial ownership of such entity. + + "You" (or "Your") shall mean an individual or Legal Entity + exercising permissions granted by this License. + + "Source" form shall mean the preferred form for making modifications, + including but not limited to software source code, documentation + source, and configuration files. + + "Object" form shall mean any form resulting from mechanical + transformation or translation of a Source form, including but + not limited to compiled object code, generated documentation, + and conversions to other media types. + + "Work" shall mean the work of authorship, whether in Source or + Object form, made available under the License, as indicated by a + copyright notice that is included in or attached to the work + (an example is provided in the Appendix below). + + "Derivative Works" shall mean any work, whether in Source or Object + form, that is based on (or derived from) the Work and for which the + editorial revisions, annotations, elaborations, or other modifications + represent, as a whole, an original work of authorship. For the purposes + of this License, Derivative Works shall not include works that remain + separable from, or merely link (or bind by name) to the interfaces of, + the Work and Derivative Works thereof. + + "Contribution" shall mean any work of authorship, including + the original version of the Work and any modifications or additions + to that Work or Derivative Works thereof, that is intentionally + submitted to Licensor for inclusion in the Work by the copyright owner + or by an individual or Legal Entity authorized to submit on behalf of + the copyright owner. For the purposes of this definition, "submitted" + means any form of electronic, verbal, or written communication sent + to the Licensor or its representatives, including but not limited to + communication on electronic mailing lists, source code control systems, + and issue tracking systems that are managed by, or on behalf of, the + Licensor for the purpose of discussing and improving the Work, but + excluding communication that is conspicuously marked or otherwise + designated in writing by the copyright owner as "Not a Contribution." + + "Contributor" shall mean Licensor and any individual or Legal Entity + on behalf of whom a Contribution has been received by Licensor and + subsequently incorporated within the Work. + +2. Grant of Copyright License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + copyright license to reproduce, prepare Derivative Works of, + publicly display, publicly perform, sublicense, and distribute the + Work and such Derivative Works in Source or Object form. + +3. Grant of Patent License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + (except as stated in this section) patent license to make, have made, + use, offer to sell, sell, import, and otherwise transfer the Work, + where such license applies only to those patent claims licensable + by such Contributor that are necessarily infringed by their + Contribution(s) alone or by combination of their Contribution(s) + with the Work to which such Contribution(s) was submitted. If You + institute patent litigation against any entity (including a + cross-claim or counterclaim in a lawsuit) alleging that the Work + or a Contribution incorporated within the Work constitutes direct + or contributory patent infringement, then any patent licenses + granted to You under this License for that Work shall terminate + as of the date such litigation is filed. + +4. Redistribution. You may reproduce and distribute copies of the + Work or Derivative Works thereof in any medium, with or without + modifications, and in Source or Object form, provided that You + meet the following conditions: + + (a) You must give any other recipients of the Work or + Derivative Works a copy of this License; and + + (b) You must cause any modified files to carry prominent notices + stating that You changed the files; and + + (c) You must retain, in the Source form of any Derivative Works + that You distribute, all copyright, patent, trademark, and + attribution notices from the Source form of the Work, + excluding those notices that do not pertain to any part of + the Derivative Works; and + + (d) If the Work includes a "NOTICE" text file as part of its + distribution, then any Derivative Works that You distribute must + include a readable copy of the attribution notices contained + within such NOTICE file, excluding those notices that do not + pertain to any part of the Derivative Works, in at least one + of the following places: within a NOTICE text file distributed + as part of the Derivative Works; within the Source form or + documentation, if provided along with the Derivative Works; or, + within a display generated by the Derivative Works, if and + wherever such third-party notices normally appear. The contents + of the NOTICE file are for informational purposes only and + do not modify the License. You may add Your own attribution + notices within Derivative Works that You distribute, alongside + or as an addendum to the NOTICE text from the Work, provided + that such additional attribution notices cannot be construed + as modifying the License. + + You may add Your own copyright statement to Your modifications and + may provide additional or different license terms and conditions + for use, reproduction, or distribution of Your modifications, or + for any such Derivative Works as a whole, provided Your use, + reproduction, and distribution of the Work otherwise complies with + the conditions stated in this License. + +5. Submission of Contributions. Unless You explicitly state otherwise, + any Contribution intentionally submitted for inclusion in the Work + by You to the Licensor shall be under the terms and conditions of + this License, without any additional terms or conditions. + Notwithstanding the above, nothing herein shall supersede or modify + the terms of any separate license agreement you may have executed + with Licensor regarding such Contributions. + +6. Trademarks. This License does not grant permission to use the trade + names, trademarks, service marks, or product names of the Licensor, + except as required for reasonable and customary use in describing the + origin of the Work and reproducing the content of the NOTICE file. + +7. Disclaimer of Warranty. Unless required by applicable law or + agreed to in writing, Licensor provides the Work (and each + Contributor provides its Contributions) on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + implied, including, without limitation, any warranties or conditions + of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A + PARTICULAR PURPOSE. You are solely responsible for determining the + appropriateness of using or redistributing the Work and assume any + risks associated with Your exercise of permissions under this License. + +8. Limitation of Liability. In no event and under no legal theory, + whether in tort (including negligence), contract, or otherwise, + unless required by applicable law (such as deliberate and grossly + negligent acts) or agreed to in writing, shall any Contributor be + liable to You for damages, including any direct, indirect, special, + incidental, or consequential damages of any character arising as a + result of this License or out of the use or inability to use the + Work (including but not limited to damages for loss of goodwill, + work stoppage, computer failure or malfunction, or any and all + other commercial damages or losses), even if such Contributor + has been advised of the possibility of such damages. + +9. Accepting Warranty or Additional Liability. While redistributing + the Work or Derivative Works thereof, You may choose to offer, + and charge a fee for, acceptance of support, warranty, indemnity, + or other liability obligations and/or rights consistent with this + License. However, in accepting such obligations, You may act only + on Your own behalf and on Your sole responsibility, not on behalf + of any other Contributor, and only if You agree to indemnify, + defend, and hold each Contributor harmless for any liability + incurred by, or claims asserted against, such Contributor by reason + of your accepting any such warranty or additional liability. + +END OF TERMS AND CONDITIONS + +APPENDIX: How to apply the Apache License to your work. + + To apply the Apache License to your work, attach the following + boilerplate notice, with the fields enclosed by brackets "[]" + replaced with your own identifying information. (Don't include + the brackets!) The text should be enclosed in the appropriate + comment syntax for the file format. We also recommend that a + file or class name and description of purpose be included on the + same "printed page" as the copyright notice for easier + identification within third-party archives. + +Copyright 2024 Liam Gray + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. + +------------------------------------------------------------------------ +Apache-2.0 — Apache License 2.0 +------------------------------------------------------------------------ + Applies to: ring 0.17.14 @@ -9373,7 +9582,7 @@ ISC — ISC License ------------------------------------------------------------------------ Applies to: - rustls-webpki 0.103.13 + rustls-webpki 0.103.15 Except as otherwise noted, this project is licensed under the following (ISC-style) terms: @@ -10006,7 +10215,7 @@ MIT — MIT License ------------------------------------------------------------------------ Applies to: - h2 0.4.15 + h2 0.4.19 Copyright (c) 2017 h2 authors @@ -10347,6 +10556,33 @@ SOFTWARE. MIT — MIT License ------------------------------------------------------------------------ +Applies to: + metrics 0.24.6 + +Copyright (c) 2021 Metrics Contributors + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. + +------------------------------------------------------------------------ +MIT — MIT License +------------------------------------------------------------------------ + Applies to: mio 1.1.1 diff --git a/leviculum-core/src/node/link_management.rs b/leviculum-core/src/node/link_management.rs index cf80140fb..6ec431be1 100644 --- a/leviculum-core/src/node/link_management.rs +++ b/leviculum-core/src/node/link_management.rs @@ -494,6 +494,12 @@ impl NodeCore { /// `self.transport.register_destination(link_id)`. This helper /// ensures the reverse cleanup always happens together. fn remove_link(&mut self, link_id: &LinkId) { + // leviculum#77: the close reported next counts as a handshake failure + // if this link never established. + if let Some(link) = self.links.get(link_id) { + let established = link.established_at_secs().is_some(); + self.note_link_removing(established); + } // Fail any in-flight resource transfers before dropping the link, so // the app waiting on the transfer is notified instead of hanging // (mirrors RNS Link.link_closed() cancelling both incoming and @@ -572,8 +578,37 @@ impl NodeCore { /// `if is_initiator`), so an incoming link never acquires an alias and /// there is nothing to resolve. If responder-side re-keying is ever /// added, this becomes the next stale-id bug. + /// + /// Only a pending inbound request can be refused. Any other link (one + /// already established, or one this node dialled) is closed instead, as + /// [`Self::close_link`] would: close packet, `LinkClosed`, counted as a + /// close. Silently dropping a live link would leave the completion + /// mirror and the lifecycle counters believing it alive (leviculum#77, + /// Codex review on #76). Its events go out with the node's next output. pub fn reject_link(&mut self, link_id: &LinkId) { + let Some(link) = self.links.get_mut(link_id) else { + return; + }; + if !link.is_initiator() && link.established_at_secs().is_none() { + self.remove_link(link_id); + self.note_link_rejected(); + return; + } + let is_initiator = link.is_initiator(); + let destination_hash = *link.destination_hash(); + if let Ok(close_packet) = link.build_close_packet(&mut self.rng) { + link.close(); + self.route_link_packet(link_id, &close_packet); + } else { + link.close(); + } self.remove_link(link_id); + self.emit_link_closed( + *link_id, + LinkCloseReason::Normal, + is_initiator, + destination_hash, + ); } /// Close a link gracefully @@ -636,6 +671,55 @@ impl NodeCore { .count() } + /// Live links, counted by role and by destination (CIRISEdge#819, + /// leviculum#57). + /// + /// "Live" is established and not closed: `Active` or `Stale`. Pending + /// links (request out, no proof yet) are counted apart. Per destination, + /// the initiator count is links this node dialled to that destination, + /// and the responder count is links peers opened to this node's own + /// destination. A count that only grows names its owner: the initiator + /// side keeps a link alive with its keepalives, so a rising initiator + /// count toward one destination is this node not closing what it + /// dialled. + pub fn link_census(&self) -> crate::node::LinkCensus { + let mut census = crate::node::LinkCensus::default(); + let mut by_dest: alloc::collections::BTreeMap = + alloc::collections::BTreeMap::new(); + for link in self.links.values() { + match link.state() { + LinkState::Active | LinkState::Stale => {} + LinkState::Closed => continue, + _ => { + census.pending += 1; + continue; + } + } + let entry = by_dest.entry(*link.destination_hash()).or_default(); + if link.is_initiator() { + census.initiator += 1; + entry.0 += 1; + } else { + census.responder += 1; + entry.1 += 1; + } + } + census.by_destination = by_dest + .into_iter() + .map( + |(destination_hash, (initiator, responder))| crate::node::DestinationLinks { + destination_hash, + initiator, + responder, + }, + ) + .collect(); + census + .by_destination + .sort_by_key(|d| core::cmp::Reverse(d.initiator + d.responder)); + census + } + /// Return all local links as RPC-exportable rows. /// /// One entry per [`crate::link::Link`] currently in the `links` map, @@ -1235,6 +1319,7 @@ impl NodeCore { match link.build_rtt_packet(rtt_seconds_wire, &mut self.rng) { Ok(rtt_packet) => { link.record_rtt_sent(now_ms); + self.link_telemetry.lifecycle.established_initiator += 1; self.events.push(NodeEvent::LinkEstablished { link_id, is_initiator: true, @@ -1250,6 +1335,7 @@ impl NodeCore { ); // Emit the event anyway, the link IS established even if the // RTT packet could not be sent. + self.link_telemetry.lifecycle.established_initiator += 1; self.events.push(NodeEvent::LinkEstablished { link_id, is_initiator: true, @@ -1480,6 +1566,7 @@ impl NodeCore { link.stale_close_timeout_secs().saturating_mul(MS_PER_SECOND), link.hops(), ); + self.link_telemetry.lifecycle.established_responder += 1; self.events.push(NodeEvent::LinkEstablished { link_id, is_initiator: false, @@ -4222,6 +4309,7 @@ impl NodeCore { is_initiator: bool, destination_hash: DestinationHash, ) { + self.note_link_closed(reason); // Translate a re-keyed wire id back to the caller-visible // original and consume the mapping — the close is the last // event for this link (Codeberg #66). Registries may hold diff --git a/leviculum-core/src/node/mod.rs b/leviculum-core/src/node/mod.rs index 2e9918eda..18da5a373 100644 --- a/leviculum-core/src/node/mod.rs +++ b/leviculum-core/src/node/mod.rs @@ -569,6 +569,8 @@ pub struct NodeCore { /// it, so a cancel, a duplicate ADV, a timeout poll or a link close treat /// it as they always did. Taking a job is what moves the parts out. pending_assembly: AssemblyQueue, + /// Link lifecycle counters (leviculum#77). + link_telemetry: telemetry::LinkTelemetry, /// Resources whose assembly is running outside the node, by link. One per /// link at most: a sender waits for the proof before its next segment. /// Holds no more than the caller asked for at a time. @@ -674,6 +676,7 @@ impl NodeCore { link_origin_ids: BTreeMap::new(), defer_resource_assembly: false, pending_assembly: AssemblyQueue::default(), + link_telemetry: telemetry::LinkTelemetry::default(), assembling_resources: BTreeMap::new(), max_incoming_resource_size, resource_window_policy, @@ -12825,3 +12828,37 @@ mod tests { assert!(!has_timeout); } } + +/// Live links by role, overall and per destination: see +/// [`NodeCore::link_census`] (CIRISEdge#819, leviculum#57). Defined last so +/// the fork moves no line an upstream citation points at. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct LinkCensus { + /// Established links this node dialled. + pub initiator: usize, + /// Established links peers opened to this node. + pub responder: usize, + /// Links still establishing. + pub pending: usize, + /// Per destination, most links first. + pub by_destination: Vec, +} + +/// One destination's share of a [`LinkCensus`]. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct DestinationLinks { + /// For an initiator link the destination dialled; for a responder link + /// this node's own destination the peer linked to. + pub destination_hash: DestinationHash, + pub initiator: usize, + pub responder: usize, +} +#[cfg(test)] +mod mvr_link_census; + +// Link telemetry (+ciris, leviculum#77), declared last so the fork moves no +// line an upstream citation points at. +mod telemetry; +pub use telemetry::{ + link_close_reason_name, LinkInfo, LinkLifecycle, LinkRole, LINK_CLOSE_REASONS, +}; diff --git a/leviculum-core/src/node/mvr_link_census.rs b/leviculum-core/src/node/mvr_link_census.rs new file mode 100644 index 000000000..711692242 --- /dev/null +++ b/leviculum-core/src/node/mvr_link_census.rs @@ -0,0 +1,176 @@ +//! mvr: the link census counts live links by role and destination, and +//! links leave it when they close (CIRISEdge#819, leviculum#57). +//! +//! Sans-I/O: two endpoints over one mock mesh hop. + +extern crate std; + +use std::string::String; +use std::vec::Vec; + +use rand_core::OsRng; + +use crate::destination::{Destination, DestinationType, Direction}; +use crate::identity::Identity; +use crate::link::{LinkCloseReason, LinkState}; +use crate::node::{LinkCensus, LinkRole, NodeCore, NodeCoreBuilder}; +use crate::test_utils::{MockClock, MockInterface, TEST_TIME_MS}; +use crate::transport::{Action, InterfaceId, TickOutput}; + +type Node = NodeCore; + +fn add_iface(node: &mut Node, name: &'static str) -> usize { + let idx = node + .transport + .register_interface(std::boxed::Box::new(MockInterface::new(name, 0))); + node.set_interface_name(idx, String::from(name)); + idx +} + +fn packets(output: &TickOutput) -> Vec> { + output + .actions + .iter() + .map(|a| match a { + Action::SendPacket { data, .. } | Action::Broadcast { data, .. } => data.clone(), + }) + .collect() +} + +fn deliver(target: &mut Node, iface: usize, pkts: Vec>) -> Vec> { + let mut out = Vec::new(); + for pkt in pkts { + out.extend(packets(&target.handle_packet(InterfaceId(iface), &pkt))); + } + out +} + +#[test] +fn the_census_counts_live_links_by_role_and_destination() { + let identity = Identity::generate(&mut OsRng); + let signing_key = identity.ed25519_verifying().to_bytes(); + let mut b = NodeCoreBuilder::new().build( + OsRng, + MockClock::new(TEST_TIME_MS), + crate::traits::NoStorage, + ); + let mut dest = Destination::new( + Some(identity), + Direction::In, + DestinationType::Single, + "mvrapp", + &["census"], + ) + .unwrap(); + dest.set_accepts_links(true); + let dest_hash = *dest.hash(); + b.register_destination(dest); + let mut a = NodeCoreBuilder::new().build( + OsRng, + MockClock::new(TEST_TIME_MS), + crate::traits::NoStorage, + ); + let b_mesh = add_iface(&mut b, "B_mesh"); + let a_mesh = add_iface(&mut a, "A_mesh"); + assert_eq!(a.link_census(), LinkCensus::default()); + + // Three links from A to B's destination; the last stays pending. + let mut ids = Vec::new(); + for i in 0..3 { + let (link_id, _, out) = a.connect(dest_hash, &signing_key).expect("connect"); + ids.push(link_id); + if i == 2 { + break; + } + let mut to_b = packets(&out); + for _ in 0..8 { + if to_b.is_empty() { + break; + } + let back = deliver(&mut b, b_mesh, to_b); + to_b = deliver(&mut a, a_mesh, back); + } + } + + let census = a.link_census(); + assert_eq!( + (census.initiator, census.responder, census.pending), + (2, 0, 1) + ); + assert_eq!(census.by_destination.len(), 1); + assert_eq!(census.by_destination[0].destination_hash, dest_hash); + assert_eq!(census.by_destination[0].initiator, 2); + + let census = b.link_census(); + assert_eq!((census.initiator, census.responder), (0, 2)); + assert_eq!(census.by_destination[0].responder, 2); + + // The list names each link with its role and state; idle and age are + // known only once established. + let list = a.link_list(); + assert_eq!(list.len(), 3); + let live: Vec<_> = list.iter().filter(|l| l.age_secs.is_some()).collect(); + assert_eq!(live.len(), 2); + assert!(live + .iter() + .all(|l| l.role == LinkRole::Initiator && l.state == LinkState::Active)); + assert!(b.link_list().iter().all(|l| l.role == LinkRole::Responder)); + a.transport.clock().advance(7_000); + assert!(a + .link_list() + .iter() + .filter(|l| l.age_secs.is_some()) + .all(|l| l.idle_secs == Some(7) && l.age_secs == Some(7))); + + let lc = a.link_lifecycle(); + assert_eq!((lc.established_initiator, lc.established_responder), (2, 0)); + assert_eq!(b.link_lifecycle().established_responder, 2); + + // A closed link leaves the census and is counted by reason, on both ends. + let close = a.close_link(&ids[0]); + assert_eq!(a.link_census().initiator, 1); + assert_eq!(a.link_lifecycle().closed(LinkCloseReason::Normal), 1); + deliver(&mut b, b_mesh, packets(&close)); + assert_eq!(b.link_lifecycle().closed(LinkCloseReason::PeerClosed), 1); + + // A request B refuses is counted as rejected on B. + let (refused, _, out) = a.connect(dest_hash, &signing_key).expect("connect"); + let _ = deliver(&mut b, b_mesh, packets(&out)); + let pending_on_b = b + .link_list() + .into_iter() + .find(|l| l.age_secs.is_none()) + .expect("B holds the pending request"); + b.reject_link(&pending_on_b.link_id); + assert_eq!(b.link_lifecycle().rejected, 1); + assert_eq!(b.link_lifecycle().handshake_failed_total(), 0); + // Rejecting a link that is already live is a close, not a refusal: the + // counters keep agreeing with the census. + let live_on_b = b + .link_list() + .into_iter() + .find(|l| l.age_secs.is_some()) + .expect("an established link on B"); + b.reject_link(&live_on_b.link_id); + let lc = b.link_lifecycle(); + assert_eq!(lc.rejected, 1, "not counted as a refusal"); + assert_eq!(lc.closed(LinkCloseReason::Normal), 1); + let census = b.link_census(); + assert_eq!( + lc.established() - lc.closed_total(), + (census.initiator + census.responder) as u64 + ); + let _ = a.close_link(&refused); + + // A link that never established is a handshake failure, not a close. + let _ = a.close_link(&ids[2]); + let lc = a.link_lifecycle(); + assert_eq!(lc.handshake_failed(LinkCloseReason::Normal), 2); + assert_eq!(lc.closed_total(), 1); + // And the counters agree with the census. + let census = a.link_census(); + assert_eq!( + lc.established() - lc.closed_total(), + (census.initiator + census.responder) as u64 + ); +} diff --git a/leviculum-core/src/node/telemetry.rs b/leviculum-core/src/node/telemetry.rs new file mode 100644 index 000000000..cfb2f201d --- /dev/null +++ b/leviculum-core/src/node/telemetry.rs @@ -0,0 +1,278 @@ +//! Link telemetry (+ciris, leviculum#77): the live link list and the +//! cumulative lifecycle counters a host reads to see what its links are +//! doing in the field. +//! +//! Both are plain snapshots. The counters only grow (for a monotonic +//! metrics counter), and nothing here allocates on the hot path: a close +//! or an establishment bumps a fixed-size counter. + +use alloc::vec::Vec; + +use rand_core::CryptoRngCore; + +use crate::constants::MS_PER_SECOND; +use crate::destination::DestinationHash; +use crate::link::{LinkCloseReason, LinkId, LinkState}; +use crate::traits::{Clock, Storage}; + +use super::NodeCore; + +/// Which end of a link this node is. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum LinkRole { + /// This node dialled the link; its keepalives keep it alive. + Initiator, + /// A peer opened the link to one of this node's destinations. + Responder, +} + +/// One live link, as a host lists it. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct LinkInfo { + /// The id the application was given (a re-keyed link reports its + /// original id, the same one its events carry). + pub link_id: LinkId, + /// For an initiator link the destination dialled; for a responder link + /// this node's own destination. + pub destination_hash: DestinationHash, + pub role: LinkRole, + pub state: LinkState, + /// Seconds since establishment, or `None` while still establishing. + pub age_secs: Option, + /// Seconds since anything arrived on the link (one-second resolution), + /// or `None` if nothing has yet. + pub idle_secs: Option, + /// Measured round-trip time in milliseconds, or `None` before a + /// sample exists (the link's working fallback is not a measurement). + pub rtt_ms: Option, + /// The interface the link is sent on, if attached. + pub interface_index: Option, +} + +/// Why a link ended, as a counter index. One slot per [`LinkCloseReason`]. +const REASONS: usize = 7; + +fn reason_slot(reason: LinkCloseReason) -> usize { + match reason { + LinkCloseReason::Normal => 0, + LinkCloseReason::Timeout => 1, + LinkCloseReason::InvalidProof => 2, + LinkCloseReason::PeerClosed => 3, + LinkCloseReason::Stale => 4, + LinkCloseReason::ChannelExhausted => 5, + LinkCloseReason::Blackholed => 6, + } +} + +/// Every [`LinkCloseReason`], in counter order. +pub const LINK_CLOSE_REASONS: [LinkCloseReason; REASONS] = [ + LinkCloseReason::Normal, + LinkCloseReason::Timeout, + LinkCloseReason::InvalidProof, + LinkCloseReason::PeerClosed, + LinkCloseReason::Stale, + LinkCloseReason::ChannelExhausted, + LinkCloseReason::Blackholed, +]; + +/// A stable lowercase name for a close reason, for a metric label. +pub fn link_close_reason_name(reason: LinkCloseReason) -> &'static str { + match reason { + LinkCloseReason::Normal => "normal", + LinkCloseReason::Timeout => "timeout", + LinkCloseReason::InvalidProof => "invalid_proof", + LinkCloseReason::PeerClosed => "peer_closed", + LinkCloseReason::Stale => "stale", + LinkCloseReason::ChannelExhausted => "channel_exhausted", + LinkCloseReason::Blackholed => "blackholed", + } +} + +/// Cumulative link lifecycle counters since the node started. +/// +/// Every link that ends is counted once: in `closed` if it had been +/// established, in `handshake_failed` if it never was, or in `rejected` if +/// the application refused its request. So +/// `established - closed_total()` is the established links alive now, and +/// a census that disagrees with it is a bookkeeping fault. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct LinkLifecycle { + /// Links established with this node as initiator. + pub established_initiator: u64, + /// Links established with this node as responder. + pub established_responder: u64, + /// Incoming link requests the application refused (`reject_link`). + pub rejected: u64, + closed: [u64; REASONS], + handshake_failed: [u64; REASONS], +} + +impl LinkLifecycle { + /// Links established, either role. + pub fn established(&self) -> u64 { + self.established_initiator + self.established_responder + } + + /// Established links that ended with `reason`. + pub fn closed(&self, reason: LinkCloseReason) -> u64 { + self.closed[reason_slot(reason)] + } + + /// Links that ended with `reason` before they were established. + pub fn handshake_failed(&self, reason: LinkCloseReason) -> u64 { + self.handshake_failed[reason_slot(reason)] + } + + /// Established links that ended, any reason. + pub fn closed_total(&self) -> u64 { + self.closed.iter().sum() + } + + /// Links that never established, any reason. + pub fn handshake_failed_total(&self) -> u64 { + self.handshake_failed.iter().sum() + } +} + +/// The node's telemetry state: the counters, and whether the link being +/// torn down right now had been established (noted when it leaves the +/// table, read when its close is reported). +#[derive(Debug, Default)] +pub(crate) struct LinkTelemetry { + pub(super) lifecycle: LinkLifecycle, + pub(super) removing_was_established: Option, +} + +impl NodeCore { + /// Every link in the table, live or establishing, with its role, state, + /// age and idle time (leviculum#77). + pub fn link_list(&self) -> Vec { + let now_secs = self.transport.clock().now_ms() / MS_PER_SECOND; + self.links + .iter() + .filter(|(_, link)| link.state() != LinkState::Closed) + .map(|(wire_id, link)| { + let established = link.established_at_secs(); + LinkInfo { + link_id: self + .link_origin_ids + .get(wire_id) + .copied() + .unwrap_or(*wire_id), + destination_hash: *link.destination_hash(), + role: if link.is_initiator() { + LinkRole::Initiator + } else { + LinkRole::Responder + }, + state: link.state(), + age_secs: established.map(|at| now_secs.saturating_sub(at)), + idle_secs: established + .map(|_| now_secs.saturating_sub(link.last_inbound_secs())), + rtt_ms: link.rtt_us().map(|us| us / 1000), + interface_index: link.attached_interface(), + } + }) + .collect() + } + + /// Cumulative link lifecycle counters (leviculum#77). + pub fn link_lifecycle(&self) -> LinkLifecycle { + self.link_telemetry.lifecycle + } + + /// Called by `reject_link`: a request refused before it established, + /// which reports no close (Codex review on #76). + pub(super) fn note_link_rejected(&mut self) { + self.link_telemetry.removing_was_established = None; + self.link_telemetry.lifecycle.rejected += 1; + } + + /// Called by `remove_link` as a link leaves the table. + pub(super) fn note_link_removing(&mut self, was_established: bool) { + self.link_telemetry.removing_was_established = Some(was_established); + } + + /// Called by `emit_link_closed`, right after the matching removal. + pub(super) fn note_link_closed(&mut self, reason: LinkCloseReason) { + // Every close follows its removal; an unmatched one is counted as an + // established link closing, the conservative reading. + let established = self + .link_telemetry + .removing_was_established + .take() + .unwrap_or(true); + let lc = &mut self.link_telemetry.lifecycle; + let slot = reason_slot(reason); + if established { + lc.closed[slot] += 1; + } else { + lc.handshake_failed[slot] += 1; + } + } +} + +impl NodeCore { + /// Established links (Active or Stale), either role: the count the + /// driver's completion mirror must agree with (leviculum#77). One pass + /// over the table, no allocation. + pub fn established_link_count(&self) -> usize { + self.links + .values() + .filter(|l| matches!(l.state(), LinkState::Active | LinkState::Stale)) + .count() + } +} + +impl crate::transport::TransportStats { + /// The drop counter for one reason (leviculum#77), so a metrics bridge + /// can walk [`crate::transport::DropReason::ALL`] instead of naming + /// twenty-one accessors. Defined here so the fork moves no line of + /// `transport.rs`. + pub fn drops_for(&self, reason: crate::transport::DropReason) -> u64 { + use crate::transport::DropReason as D; + match reason { + D::OverheardTransportId => self.drops_overheard_transport_id, + D::InvalidAnnounce => self.drops_invalid_announce, + D::PlainGroupMultihop => self.drops_plain_group_multihop, + D::NoPath => self.drops_no_path, + D::Ifac => self.drops_ifac, + D::Duplicate => self.drops_duplicate, + D::AnnounceOverMaxHops => self.drops_announce_over_max_hops, + D::AnnounceReplay => self.drops_announce_replay, + D::AnnounceRateLimited => self.drops_announce_rate_limited, + D::IngressBurstAnnounce => self.drops_ingress_burst_announce, + D::LrproofInvalid => self.drops_lrproof_invalid, + D::LrproofNoLink => self.drops_lrproof_no_link, + D::LinkDataNoLink => self.drops_link_data_no_link, + D::LinkRepeatEcho => self.drops_link_repeat_echo, + D::ForwardMaxHops => self.drops_forward_max_hops, + D::BlackholedAnnounce => self.drops_blackholed_announce, + D::SingleDecryptFail => self.drops_single_decrypt_fail, + D::GroupDecryptFail => self.drops_group_decrypt_fail, + D::UnknownContext => self.drops_unknown_context, + D::NoSuchInterface => self.drops_no_such_interface, + D::NextHopIsRequestor => self.drops_next_hop_is_requestor, + } + } +} + +#[cfg(test)] +mod tests { + use crate::transport::{DropReason, TransportStats}; + + #[test] + fn drops_for_covers_every_reason_and_sums_to_the_total() { + let mut stats = TransportStats::default(); + for (i, reason) in DropReason::ALL.iter().enumerate() { + for _ in 0..=i { + stats.record_drop(*reason); + } + } + for (i, reason) in DropReason::ALL.iter().enumerate() { + assert_eq!(stats.drops_for(*reason), i as u64 + 1, "{reason:?}"); + } + let sum: u64 = DropReason::ALL.iter().map(|r| stats.drops_for(*r)).sum(); + assert_eq!(sum, stats.packets_dropped()); + } +} diff --git a/leviculum-nrf/Cargo.lock b/leviculum-nrf/Cargo.lock index ccc7b0c6e..e6ef6b3ae 100644 --- a/leviculum-nrf/Cargo.lock +++ b/leviculum-nrf/Cargo.lock @@ -944,7 +944,7 @@ dependencies = [ [[package]] name = "leviculum-core" -version = "0.10.0-dev" +version = "0.29.0+ciris.1" dependencies = [ "aes", "cbc", @@ -985,7 +985,7 @@ version = "0.1.0" [[package]] name = "leviculum-lxmf" -version = "0.10.0-dev" +version = "0.29.0+ciris.1" dependencies = [ "leviculum-core", "rand_core 0.6.4", diff --git a/leviculum-std/Cargo.toml b/leviculum-std/Cargo.toml index 0f7233915..30438ed93 100644 --- a/leviculum-std/Cargo.toml +++ b/leviculum-std/Cargo.toml @@ -37,6 +37,11 @@ socket2.workspace = true libc.workspace = true tokio-serial.workspace = true tracing = { workspace = true, features = ["std"] } +# The metrics facade (leviculum#77, CIRISServer FSD UNIFIED_TELEMETRY §5): +# libraries emit through it, the host installs the recorder and exporter. +# With no recorder installed every call is a no-op, so no exporter is a +# dependency here. +metrics = { version = "0.24", default-features = false } # Used by `test_support::tracing_setup` (and from commit 2 onward # `test_support::event_log`) for the test-harness subscriber. The # crate's runtime presence is a few KB; production builds that don't diff --git a/leviculum-std/src/driver/mod.rs b/leviculum-std/src/driver/mod.rs index 264ed3e24..cd3424192 100644 --- a/leviculum-std/src/driver/mod.rs +++ b/leviculum-std/src/driver/mod.rs @@ -266,6 +266,7 @@ pub(crate) struct PlaneCounters { /// `retry_dropped`, these never paid for an IFAC signature and never /// entered a retry queue — that saving is the point of the breaker. shed_packets: AtomicU64, + mirror_alarms: AtomicU64, // leviculum#77: crate::telemetry::MirrorWatch alarms } /// Sender half of the split control/data node-event channels (Codeberg #71). @@ -1511,6 +1512,7 @@ pub struct ReticulumNode { /// Per-transfer ceiling for multi-segment resource assembly /// (leviculum#62). pub(crate) max_assembled_resource_size: usize, + pub(crate) metrics_baseline: Mutex, // leviculum#77 /// Merged event receiver for consuming events. `None` either because the /// node was built with `without_events()`, or because /// `take_event_receiver()` already handed it out. @@ -1724,6 +1726,7 @@ impl ReticulumNode { control_dropped, plane_counters: Arc::new(PlaneCounters::default()), max_assembled_resource_size: segments::DEFAULT_MAX_ASSEMBLED_RESOURCE_SIZE, + metrics_baseline: Mutex::default(), event_rx, shutdown_tx: None, runner_handle: None, @@ -2624,6 +2627,85 @@ impl ReticulumNode { self.inner.lock_recover().storage().identity_evictions() } + /// Every link in the table with its role, state, age and idle time + /// (leviculum#77). Ids are the ones the application was given. + /// Takes the core lock briefly; walks the link table once. + pub fn link_list(&self) -> Vec { + self.inner.lock_recover().link_list() + } + + /// Cumulative link lifecycle counters: established by role, closed by + /// reason, handshake failures by reason (leviculum#77). + pub fn link_lifecycle(&self) -> leviculum_core::node::LinkLifecycle { + self.inner.lock_recover().link_lifecycle() + } + + /// The node core's heap census: accounted bytes by component, the link + /// table among them (leviculum#77). An estimate from container sizes, + /// not an allocator reading. Takes the core lock and walks every table. + pub fn heap_census(&self) -> leviculum_core::heap_census::NodeHeapCensus { + self.inner.lock_recover().heap_census() + } + + /// The completion mirror's established-link count beside the core's, + /// and how many sustained divergences the event loop has alarmed on + /// (leviculum#77). The two can differ for an instant while events are + /// in flight; the loop alarms only when they stay apart. + pub fn link_count_check(&self) -> crate::telemetry::LinkCountCheck { + crate::telemetry::LinkCountCheck { + mirror: self.completions.established_len(), + core: self.inner.lock_recover().established_link_count(), + alarms: self + .plane_counters + .mirror_alarms + .load(std::sync::atomic::Ordering::Relaxed), + } + } + + /// Record the node's metrics onto the installed `metrics` recorder + /// (leviculum#77). Call it on the host's cadence, typically right before + /// each scrape; with no recorder installed it costs the snapshots and + /// nothing else. The names are `crate::telemetry::METRIC_CATALOG`. + /// Counters are process totals that each node adds its growth to, so + /// several nodes in one process sum and a replacement node carries on. + /// Takes the core lock once for all core snapshots. + pub fn publish_metrics(&self) { + let (lifecycle, census, transport, known_evictions, heap) = { + let core = self.inner.lock_recover(); + ( + core.link_lifecycle(), + core.link_census(), + core.transport_stats(), + core.storage().identity_evictions(), + core.heap_census(), + ) + }; + crate::telemetry::record( + &crate::telemetry::Snapshot { + lifecycle, + census, + mirror: self.completions.established_len(), + mirror_alarms: self + .plane_counters + .mirror_alarms + .load(std::sync::atomic::Ordering::Relaxed), + transport, + known_evictions, + plane: self.plane_stats(), + heap, + }, + &mut self.metrics_baseline.lock_recover(), + ); + } + + /// Live links by role, overall and per destination (CIRISEdge#819, + /// leviculum#57). A field gauge for link accumulation: see + /// [`leviculum_core::node::NodeCore::link_census`] for how to read it. + /// Takes the core lock briefly; walks the link table once. + pub fn link_census(&self) -> leviculum_core::node::LinkCensus { + self.inner.lock_recover().link_census() + } + /// Current saturation of every bounded structure the driver owns — /// depths, capacities, cumulative drops (leviculum#60). See /// [`PlaneStats`] for what each limit does when it saturates. @@ -4672,6 +4754,13 @@ async fn run_event_loop( // (a detach racing a just-accepted add); held here, applied on arrival. let mut pending_removals: std::collections::HashSet = std::collections::HashSet::new(); + // leviculum#77: the completion mirror and the core must agree on how many + // links are established. They differ for an instant while events are in + // flight between the core and this loop, so only a disagreement that + // outlasts crate::telemetry::MIRROR_DIVERGENCE_GRACE is alarmed, once per episode. + let mut mirror_check_at = tokio::time::Instant::now() + crate::telemetry::MIRROR_CHECK_INTERVAL; + let mut mirror_watch = + crate::telemetry::MirrorWatch::new(crate::telemetry::MIRROR_DIVERGENCE_GRACE); let mut shutdown = channels.shutdown; let mut next_poll = tokio::time::Instant::now(); let mut next_flush = tokio::time::Instant::now() + Duration::from_secs(flush_interval_secs); @@ -5318,6 +5407,25 @@ async fn run_event_loop( } } + if tokio::time::Instant::now() >= mirror_check_at { + mirror_check_at = tokio::time::Instant::now() + crate::telemetry::MIRROR_CHECK_INTERVAL; + let mirror = completions.established_len(); + let core = inner.lock_recover().established_link_count(); + if let Some(alarm) = + mirror_watch.observe(mirror, core, std::time::Instant::now()) + { + plane_counters + .mirror_alarms + .fetch_add(1, std::sync::atomic::Ordering::Relaxed); + tracing::warn!( + event = "LINK_MIRROR_DIVERGED", + mirror = alarm.mirror, + core = alarm.core, + for_secs = alarm.for_secs, + ); + } + } + // Advance next_poll based on next_deadline_ms let interval = match next { Some(deadline_ms) => { diff --git a/leviculum-std/src/event_catalog_completeness.rs b/leviculum-std/src/event_catalog_completeness.rs index 21224c0c7..4019e2f37 100644 --- a/leviculum-std/src/event_catalog_completeness.rs +++ b/leviculum-std/src/event_catalog_completeness.rs @@ -506,7 +506,10 @@ fn scan(file: &str, text: &str) -> Vec { sites } -fn all_sites() -> Vec { +/// Every production source file of every workspace member, comments +/// stripped, `#[cfg(test)]` modules left out: `(path relative to the +/// workspace root, text)`. +fn production_sources() -> Vec<(String, String)> { let root = workspace_root(); let mut files = Vec::new(); for member in workspace_members(&root) { @@ -527,7 +530,7 @@ fn all_sites() -> Vec { .collect(); let test_only = cfg_test_module_names(&texts); - let mut sites = Vec::new(); + let mut out = Vec::new(); for (path, text) in &texts { let stem = path .file_stem() @@ -551,11 +554,76 @@ fn all_sites() -> Vec { .unwrap_or(path) .to_string_lossy() .to_string(); - sites.extend(scan(&rel, text)); + out.push((rel, text.clone())); + } + out +} + +fn all_sites() -> Vec { + production_sources() + .iter() + .flat_map(|(rel, text)| scan(rel, text)) + .collect() +} + +/// One `counter!`/`gauge!`/`histogram!` call site (leviculum#77): the +/// metric name and the label keys written into the call. +#[derive(Debug)] +struct MetricSite { + file: String, + line: usize, + name: String, + labels: Vec, +} + +/// Find the metric macro call sites in one file. A name has to be a string +/// literal for the catalogue to be checkable, so a call whose first +/// argument is not one is reported with an empty name, and fails. +fn scan_metrics(file: &str, text: &str) -> Vec { + let mut sites = Vec::new(); + for mac in ["counter!(", "gauge!(", "histogram!("] { + let mut from = 0; + while let Some(rel) = text[from..].find(mac) { + let at = from + rel; + from = at + mac.len(); + // `describe_counter!(` and friends name a variable, not a literal. + if text[..at].ends_with("describe_") { + continue; + } + let args_start = at + mac.len(); + let Some(close) = text[args_start..].find(')') else { + continue; + }; + let args = &text[args_start..args_start + close]; + let mut quoted = args.split('"').skip(1).step_by(2); + let name = if args.trim_start().starts_with('"') { + quoted.next().unwrap_or_default().to_string() + } else { + String::new() + }; + let labels = args + .split("=>") + .take(args.matches("=>").count()) + .filter_map(|left| left.rsplit('"').nth(1).map(str::to_string)) + .collect(); + sites.push(MetricSite { + file: file.to_string(), + line: text[..at].matches('\n').count() + 1, + name, + labels, + }); + } } sites } +fn all_metric_sites() -> Vec { + production_sources() + .iter() + .flat_map(|(rel, text)| scan_metrics(rel, text)) + .collect() +} + // --------------------------------------------------------------------- // The tests // --------------------------------------------------------------------- @@ -671,3 +739,74 @@ fn no_catalogue_entry_requires_a_message_key() { "EVENT_CATALOG entries requiring a `message` key: {offenders:?}" ); } + +/// The metric scanner reads names and label keys from a real call shape. +#[test] +fn metric_scanner_reads_names_and_label_keys() { + let text = r#" + counter!("a.b", "role" => "x").absolute(1); + gauge!("c.d").set(2.0); + describe_counter!(spec.name, spec.unit, spec.description); + histogram!(name); + "#; + let sites = scan_metrics("f.rs", text); + assert_eq!(sites.len(), 3, "{sites:?}"); + assert_eq!(sites[0].name, "a.b"); + assert_eq!(sites[0].labels, vec!["role".to_string()]); + assert_eq!(sites[1].name, "c.d"); + assert!(sites[1].labels.is_empty()); + assert_eq!( + sites[2].name, "", + "a non-literal name is reported, not skipped" + ); +} + +/// Every metric the code records is in `METRIC_CATALOG`, with only the label +/// keys the catalogue allows; and every catalogue entry is recorded +/// somewhere (leviculum#77). A metric outside the catalogue has no unit, no +/// description and no label contract, and an entry nothing records is a +/// promise a dashboard will wait on forever. +#[test] +fn every_recorded_metric_is_catalogued_and_every_entry_recorded() { + use crate::telemetry::METRIC_CATALOG; + let sites = all_metric_sites(); + assert!( + sites.len() >= METRIC_CATALOG.len(), + "the scan found only {} metric sites", + sites.len() + ); + let catalogue: BTreeMap<&str, &[&str]> = + METRIC_CATALOG.iter().map(|m| (m.name, m.labels)).collect(); + assert_eq!( + catalogue.len(), + METRIC_CATALOG.len(), + "a name is catalogued twice" + ); + let mut problems = Vec::new(); + let mut recorded = BTreeSet::new(); + for site in &sites { + match catalogue.get(site.name.as_str()) { + None => problems.push(format!( + "{}:{}: `{}` is recorded but not in METRIC_CATALOG", + site.file, site.line, site.name + )), + Some(allowed) => { + recorded.insert(site.name.as_str()); + for label in &site.labels { + if !allowed.contains(&label.as_str()) { + problems.push(format!( + "{}:{}: `{}` carries label `{label}`, which its entry does not allow", + site.file, site.line, site.name + )); + } + } + } + } + } + for name in catalogue.keys() { + if !recorded.contains(name) { + problems.push(format!("`{name}` is catalogued but nothing records it")); + } + } + assert!(problems.is_empty(), "{}", problems.join("\n")); +} diff --git a/leviculum-std/src/event_log.rs b/leviculum-std/src/event_log.rs index 5eba36f84..b4fc69c02 100644 --- a/leviculum-std/src/event_log.rs +++ b/leviculum-std/src/event_log.rs @@ -1022,6 +1022,12 @@ pub const EVENT_CATALOG: &[EventSchema] = &[ name: "HELPER_TICK", required_keys: &["i"], }, + // leviculum#77: the completion mirror and the core disagreed on the + // established-link count for longer than the event loop's grace. + EventSchema { + name: "LINK_MIRROR_DIVERGED", + required_keys: &["mirror", "core", "for_secs"], + }, ]; /// Where the buffer is dumped on a panicking drop. @@ -1317,7 +1323,7 @@ enum SinkMode { /// hundred lines apart in `handle_announce`, and nothing between them /// can take seconds. What can is the emission itself: the driver's /// event loop calls `tracing::debug!` while it holds the core mutex -/// (`apply_inbound`, `leviculum-std/src/driver/mod.rs:4540`), the layer +/// (`apply_inbound`, `leviculum-std/src/driver/mod.rs:5179`), the layer /// wrote the line with a blocking `write(2)` on that very thread, and a /// `write(2)` to a USB disk under writeback throttling blocks for /// seconds. Every other task then queued behind the core mutex, so the diff --git a/leviculum-std/src/lib.rs b/leviculum-std/src/lib.rs index 2d04c1885..1e8acdd60 100644 --- a/leviculum-std/src/lib.rs +++ b/leviculum-std/src/lib.rs @@ -95,3 +95,18 @@ pub use rpc::{ rpc_drop_path, rpc_query, rpc_query_hash_param, rpc_query_transport_tables, TRANSPORT_TABLE_NAMES, }; + +/// Metrics through the `metrics` facade (+ciris, leviculum#77). Declared +/// last so the fork moves no line an upstream citation points at. +pub mod telemetry; +pub use telemetry::LinkCountCheck; + +// The types the telemetry API returns (leviculum#77, CIRISEdge#819), so a +// consumer of this crate alone can name them. Kept here, last, so the fork +// moves no line an upstream citation points at. +pub use leviculum_core::heap_census::NodeHeapCensus; +pub use leviculum_core::link::LinkState; +pub use leviculum_core::node::{ + link_close_reason_name, DestinationLinks, LinkCensus, LinkInfo, LinkLifecycle, LinkRole, + LINK_CLOSE_REASONS, +}; diff --git a/leviculum-std/src/telemetry.rs b/leviculum-std/src/telemetry.rs new file mode 100644 index 000000000..11fff5c42 --- /dev/null +++ b/leviculum-std/src/telemetry.rs @@ -0,0 +1,589 @@ +//! Metrics through the `metrics` facade (+ciris, leviculum#77). +//! +//! The no_std core keeps its counter structs; this module bridges them onto +//! the facade once, on std hosts, as CIRISServer's unified-telemetry FSD +//! asks (§3.2: libraries emit, the host installs the recorder and the +//! exporter). With no recorder installed every macro here is a no-op. +//! +//! The bridge is pulled, not pushed: the host calls +//! [`crate::driver::ReticulumNode::publish_metrics`] on its own cadence, +//! typically right before each scrape, and every value is set from the +//! node's snapshots at that moment. +//! +//! Counters are process totals. Each node remembers what it last published +//! ([`CounterBaseline`]) and adds only its new growth to one process-wide +//! sum per series, which is what `absolute` records. Two nodes in one +//! process therefore sum, a node that replaces another carries on instead +//! of sitting under the old total (`absolute` keeps the largest value it +//! has seen; Codex review on #76), and a recorder installed late still +//! reads full totals. Gauges are levels and are simply set: with several +//! nodes in one process the last publish wins. +//! +//! [`METRIC_CATALOG`] is the one list of names. Every name the code records +//! must be in it and every entry must be recorded somewhere, with only the +//! label keys it lists; `event_catalog_completeness` checks both against the +//! source, the way it checks `EVENT_CATALOG`. Label values are always bounded +//! enumerations (role, reason, plane, component): no link id, destination or +//! peer key ever becomes a label. + +use leviculum_core::heap_census::NodeHeapCensus; +use leviculum_core::node::{link_close_reason_name, LinkCensus, LinkLifecycle, LINK_CLOSE_REASONS}; +use leviculum_core::transport::{DropReason, TransportStats}; +use metrics::{counter, describe_counter, describe_gauge, gauge, Unit}; + +use crate::driver::PlaneStats; + +/// Counter or gauge. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum MetricKind { + /// Monotonic, cumulative since the node started. + Counter, + /// A level read now. + Gauge, +} + +/// One metric this crate records. +#[derive(Debug, Clone, Copy)] +pub struct MetricSpec { + /// OpenTelemetry-style dotted name; a Prometheus exporter renders + /// `leviculum.link.closed` as `leviculum_link_closed_total`. + pub name: &'static str, + pub kind: MetricKind, + pub unit: Unit, + /// The only label keys the metric may carry. Their values are bounded + /// enumerations. + pub labels: &'static [&'static str], + pub description: &'static str, +} + +const fn counter_spec( + name: &'static str, + unit: Unit, + labels: &'static [&'static str], + description: &'static str, +) -> MetricSpec { + MetricSpec { + name, + kind: MetricKind::Counter, + unit, + labels, + description, + } +} + +const fn gauge_spec( + name: &'static str, + unit: Unit, + labels: &'static [&'static str], + description: &'static str, +) -> MetricSpec { + MetricSpec { + name, + kind: MetricKind::Gauge, + unit, + labels, + description, + } +} + +/// Every metric leviculum-std records. +pub const METRIC_CATALOG: &[MetricSpec] = &[ + // Link lifecycle (P0) + counter_spec( + "leviculum.link.established", + Unit::Count, + &["role"], + "Links established, by this node's role (initiator: it dialled; responder: a peer did).", + ), + counter_spec( + "leviculum.link.closed", + Unit::Count, + &["reason"], + "Established links that ended, by close reason.", + ), + counter_spec( + "leviculum.link.handshake_failed", + Unit::Count, + &["reason"], + "Links that ended before they were established, by close reason, plus \ + reason=rejected for requests the application refused.", + ), + gauge_spec( + "leviculum.link.live", + Unit::Count, + &["role"], + "Established links (active or stale) now, by role. The initiator keeps a link alive \ + with its keepalives, so a rising initiator count is this node not closing what it dialled.", + ), + gauge_spec( + "leviculum.link.pending", + Unit::Count, + &[], + "Links still establishing.", + ), + gauge_spec( + "leviculum.link.destinations", + Unit::Count, + &[], + "Destinations with at least one established link.", + ), + gauge_spec( + "leviculum.link.max_per_destination", + Unit::Count, + &[], + "Established links to the busiest destination. Per-destination detail is on \ + ReticulumNode::link_census, never a label.", + ), + // The mirror invariant (P0) + gauge_spec( + "leviculum.link.mirror", + Unit::Count, + &[], + "Links the driver's completion mirror holds as established.", + ), + gauge_spec( + "leviculum.link.mirror_skew", + Unit::Count, + &[], + "Completion mirror minus the core's established links. Zero in steady state; \ + a value that stays non-zero is a lost LinkEstablished or LinkClosed.", + ), + counter_spec( + "leviculum.link.mirror_divergence_alarms", + Unit::Count, + &[], + "Times the mirror and the core disagreed for longer than the alarm's grace \ + (each also logged as LINK_MIRROR_DIVERGED).", + ), + // Transport + counter_spec( + "leviculum.transport.packets", + Unit::Count, + &["direction"], + "Packets sent, received, forwarded, and forwarded on a link.", + ), + counter_spec( + "leviculum.transport.announces_processed", + Unit::Count, + &[], + "Announces processed.", + ), + counter_spec( + "leviculum.transport.dropped", + Unit::Count, + &["reason"], + "Packets dropped, by classified reason. overheard-transport-id is normal on a \ + shared medium; the others are loss.", + ), + counter_spec( + "leviculum.transport.suppressed", + Unit::Count, + &["kind"], + "Work withheld on purpose, not loss: link requests redirected to a local client, \ + path requests batched onto a pending discovery, discovery retries withheld.", + ), + counter_spec( + "leviculum.known_destination.evictions", + Unit::Count, + &[], + "Known destinations evicted at the identity cap.", + ), + // Driver planes and queues + gauge_spec( + "leviculum.events.queued", + Unit::Count, + &["plane"], + "Node events waiting for the application, per plane.", + ), + gauge_spec( + "leviculum.events.capacity", + Unit::Count, + &["plane"], + "Capacity of each node-event plane.", + ), + counter_spec( + "leviculum.events.dropped", + Unit::Count, + &["plane"], + "Node events dropped because the application did not drain its plane.", + ), + gauge_spec( + "leviculum.retry.queued", + Unit::Count, + &[], + "Packets queued for retry across all interfaces.", + ), + counter_spec( + "leviculum.retry.dropped", + Unit::Count, + &[], + "Packets dropped because an interface's retry queue was full.", + ), + counter_spec( + "leviculum.shed.packets", + Unit::Count, + &[], + "Packets shed by an open per-peer circuit breaker.", + ), + gauge_spec( + "leviculum.completion.recent", + Unit::Count, + &[], + "Terminal outcomes held in the completion registry's recent ring.", + ), + // Memory (P1: per-link memory on std) + gauge_spec( + "leviculum.memory.bytes", + Unit::Bytes, + &["component"], + "Heap the node core accounts for, by component (the core's own census, an \ + estimate from container sizes).", + ), + gauge_spec( + "leviculum.memory.link_bytes_mean", + Unit::Bytes, + &[], + "Accounted link-table bytes per link.", + ), +]; + +/// The node's state at one moment, everything a publish records. +pub(crate) struct Snapshot { + pub lifecycle: LinkLifecycle, + pub census: LinkCensus, + pub mirror: usize, + pub mirror_alarms: u64, + pub transport: TransportStats, + pub known_evictions: u64, + pub plane: PlaneStats, + pub heap: NodeHeapCensus, +} + +/// Register every catalogue entry's unit and description with whatever +/// recorder is installed now. Done on every publish, not once per process: +/// a recorder installed (or swapped for a local one) after the first +/// publish must still learn them (Codex review on #76). Twenty-four calls, +/// against a scrape. +fn describe_all() { + for spec in METRIC_CATALOG { + match spec.kind { + MetricKind::Counter => describe_counter!(spec.name, spec.unit, spec.description), + MetricKind::Gauge => describe_gauge!(spec.name, spec.unit, spec.description), + } + } +} + +/// The counter totals one node last published, per series, so the next +/// publish adds only the difference to the process-wide total. +#[derive(Debug, Default)] +pub(crate) struct CounterBaseline(std::collections::HashMap<(&'static str, &'static str), u64>); + +/// Every node's contributions to each counter series, summed over the life +/// of the process. +static PROCESS_TOTALS: std::sync::OnceLock< + std::sync::Mutex>, +> = std::sync::OnceLock::new(); + +impl CounterBaseline { + /// Add what this node's series `name{label}` gained since its last + /// publish to the process-wide total, and return that total. + /// + /// Recording the process total with `absolute` keeps every property at + /// once: two nodes in one process sum, a node that replaces another + /// carries on from where the series stood, and a recorder installed + /// after earlier publishes still reads full totals. A node's own totals + /// never fall, so a smaller value adds nothing. + fn accumulate(&mut self, name: &'static str, label: &'static str, total: u64) -> u64 { + let last = self.0.insert((name, label), total).unwrap_or(0); + let delta = total.saturating_sub(last); + let mut totals = PROCESS_TOTALS + .get_or_init(Default::default) + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let entry = totals.entry((name, label)).or_insert(0); + *entry += delta; + *entry + } +} + +/// Record a snapshot onto whatever recorder is installed. +pub(crate) fn record(s: &Snapshot, b: &mut CounterBaseline) { + describe_all(); + + // Link lifecycle + let lc = &s.lifecycle; + counter!("leviculum.link.established", "role" => "initiator").absolute(b.accumulate( + "leviculum.link.established", + "initiator", + lc.established_initiator, + )); + counter!("leviculum.link.established", "role" => "responder").absolute(b.accumulate( + "leviculum.link.established", + "responder", + lc.established_responder, + )); + for reason in LINK_CLOSE_REASONS { + let name = link_close_reason_name(reason); + counter!("leviculum.link.closed", "reason" => name).absolute(b.accumulate( + "leviculum.link.closed", + name, + lc.closed(reason), + )); + counter!("leviculum.link.handshake_failed", "reason" => name).absolute(b.accumulate( + "leviculum.link.handshake_failed", + name, + lc.handshake_failed(reason), + )); + } + // A request the application refused ended before it established too. + counter!("leviculum.link.handshake_failed", "reason" => "rejected").absolute(b.accumulate( + "leviculum.link.handshake_failed", + "rejected", + lc.rejected, + )); + gauge!("leviculum.link.live", "role" => "initiator").set(s.census.initiator as f64); + gauge!("leviculum.link.live", "role" => "responder").set(s.census.responder as f64); + gauge!("leviculum.link.pending").set(s.census.pending as f64); + gauge!("leviculum.link.destinations").set(s.census.by_destination.len() as f64); + let busiest = s + .census + .by_destination + .first() + .map(|d| d.initiator + d.responder) + .unwrap_or(0); + gauge!("leviculum.link.max_per_destination").set(busiest as f64); + + // The mirror invariant + let core = s.census.initiator + s.census.responder; + gauge!("leviculum.link.mirror").set(s.mirror as f64); + gauge!("leviculum.link.mirror_skew").set(s.mirror as f64 - core as f64); + counter!("leviculum.link.mirror_divergence_alarms").absolute(b.accumulate( + "leviculum.link.mirror_divergence_alarms", + "", + s.mirror_alarms, + )); + + // Transport + let t = &s.transport; + counter!("leviculum.transport.packets", "direction" => "sent").absolute(b.accumulate( + "leviculum.transport.packets", + "sent", + t.packets_sent(), + )); + counter!("leviculum.transport.packets", "direction" => "received").absolute(b.accumulate( + "leviculum.transport.packets", + "received", + t.packets_received(), + )); + counter!("leviculum.transport.packets", "direction" => "forwarded").absolute(b.accumulate( + "leviculum.transport.packets", + "forwarded", + t.packets_forwarded(), + )); + counter!("leviculum.transport.packets", "direction" => "forwarded_link").absolute( + b.accumulate( + "leviculum.transport.packets", + "forwarded_link", + t.packets_forwarded_link(), + ), + ); + counter!("leviculum.transport.announces_processed").absolute(b.accumulate( + "leviculum.transport.announces_processed", + "", + t.announces_processed(), + )); + for reason in DropReason::ALL { + counter!("leviculum.transport.dropped", "reason" => reason.kebab()).absolute(b.accumulate( + "leviculum.transport.dropped", + reason.kebab(), + t.drops_for(reason), + )); + } + counter!("leviculum.transport.suppressed", "kind" => "lr_local_client_redirect").absolute( + b.accumulate( + "leviculum.transport.suppressed", + "lr_local_client_redirect", + t.lr_local_client_redirects(), + ), + ); + counter!("leviculum.transport.suppressed", "kind" => "path_request_pending").absolute( + b.accumulate( + "leviculum.transport.suppressed", + "path_request_pending", + t.path_request_pending_suppressions(), + ), + ); + counter!("leviculum.transport.suppressed", "kind" => "path_request_retry").absolute( + b.accumulate( + "leviculum.transport.suppressed", + "path_request_retry", + t.path_request_retry_withholds(), + ), + ); + counter!("leviculum.known_destination.evictions").absolute(b.accumulate( + "leviculum.known_destination.evictions", + "", + s.known_evictions, + )); + + // Planes and queues + let p = &s.plane; + gauge!("leviculum.events.queued", "plane" => "control").set(p.control_queued as f64); + gauge!("leviculum.events.queued", "plane" => "data").set(p.data_queued as f64); + gauge!("leviculum.events.capacity", "plane" => "control").set(p.control_capacity as f64); + gauge!("leviculum.events.capacity", "plane" => "data").set(p.data_capacity as f64); + counter!("leviculum.events.dropped", "plane" => "control").absolute(b.accumulate( + "leviculum.events.dropped", + "control", + p.control_dropped_total, + )); + counter!("leviculum.events.dropped", "plane" => "data").absolute(b.accumulate( + "leviculum.events.dropped", + "data", + p.data_dropped_total, + )); + gauge!("leviculum.retry.queued").set(p.retry_queued as f64); + counter!("leviculum.retry.dropped").absolute(b.accumulate( + "leviculum.retry.dropped", + "", + p.retry_dropped_total, + )); + counter!("leviculum.shed.packets").absolute(b.accumulate( + "leviculum.shed.packets", + "", + p.shed_packets_total, + )); + gauge!("leviculum.completion.recent").set(p.recent_outcomes as f64); + + // Memory + let h = &s.heap; + for (component, bytes) in [ + ("node", h.node_struct), + ("links", h.links), + ("resources", h.resources), + ("events", h.events), + ("requests", h.requests), + ("destinations", h.destinations), + ("transport", h.transport), + ("storage", h.storage), + ] { + gauge!("leviculum.memory.bytes", "component" => component).set(bytes as f64); + } + let per_link = if h.link_count == 0 { + 0.0 + } else { + h.links as f64 / h.link_count as f64 + }; + gauge!("leviculum.memory.link_bytes_mean").set(per_link); +} + +/// How often the event loop compares the completion mirror with the core's +/// established-link count. +pub(crate) const MIRROR_CHECK_INTERVAL: std::time::Duration = std::time::Duration::from_secs(10); +/// How long the two may disagree before it is a divergence rather than +/// events in flight. +pub(crate) const MIRROR_DIVERGENCE_GRACE: std::time::Duration = std::time::Duration::from_secs(30); + +/// The completion mirror's view of the established links beside the core's +/// (leviculum#77). See [`crate::driver::ReticulumNode::link_count_check`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct LinkCountCheck { + /// Links the completion mirror holds as established. + pub mirror: usize, + /// Links the core holds as established (active or stale). + pub core: usize, + /// Sustained divergences alarmed on since the node started. + pub alarms: u64, +} + +/// The completion-mirror invariant's alarm (leviculum#77): the mirror and the +/// core must agree on the established-link count. They differ for an instant +/// while events are in flight, so a disagreement is a divergence only once it +/// has lasted `grace`, and it is alarmed once per episode. +#[derive(Debug)] +pub(crate) struct MirrorWatch { + grace: std::time::Duration, + diverged_since: Option, + alarmed: bool, +} + +/// A divergence that has outlasted the grace. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct MirrorAlarm { + pub mirror: usize, + pub core: usize, + pub for_secs: u64, +} + +impl MirrorWatch { + pub(crate) fn new(grace: std::time::Duration) -> Self { + MirrorWatch { + grace, + diverged_since: None, + alarmed: false, + } + } + + /// One comparison. Returns an alarm the first time a disagreement has + /// lasted `grace`; agreement ends the episode. + pub(crate) fn observe( + &mut self, + mirror: usize, + core: usize, + now: std::time::Instant, + ) -> Option { + if mirror == core { + self.diverged_since = None; + self.alarmed = false; + return None; + } + let since = *self.diverged_since.get_or_insert(now); + let lasted = now.saturating_duration_since(since); + if self.alarmed || lasted < self.grace { + return None; + } + self.alarmed = true; + Some(MirrorAlarm { + mirror, + core, + for_secs: lasted.as_secs(), + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::time::{Duration, Instant}; + + #[test] + fn a_brief_disagreement_is_not_a_divergence() { + let mut w = MirrorWatch::new(Duration::from_secs(30)); + let t = Instant::now(); + assert_eq!(w.observe(5, 4, t), None); + assert_eq!(w.observe(5, 4, t + Duration::from_secs(10)), None); + assert_eq!(w.observe(5, 5, t + Duration::from_secs(20)), None); + // The clock starts again after agreement. + assert_eq!(w.observe(6, 5, t + Duration::from_secs(40)), None); + } + + #[test] + fn a_lasting_divergence_alarms_once_per_episode() { + let mut w = MirrorWatch::new(Duration::from_secs(30)); + let t = Instant::now(); + assert_eq!(w.observe(7, 4, t), None); + assert_eq!( + w.observe(7, 4, t + Duration::from_secs(30)), + Some(MirrorAlarm { + mirror: 7, + core: 4, + for_secs: 30 + }) + ); + assert_eq!(w.observe(8, 4, t + Duration::from_secs(60)), None, "once"); + assert_eq!(w.observe(4, 4, t + Duration::from_secs(70)), None); + // A new episode alarms again. + assert_eq!(w.observe(3, 4, t + Duration::from_secs(80)), None); + assert!(w.observe(3, 4, t + Duration::from_secs(110)).is_some()); + } +} diff --git a/leviculum-std/tests/telemetry_publish.rs b/leviculum-std/tests/telemetry_publish.rs new file mode 100644 index 000000000..f44441573 --- /dev/null +++ b/leviculum-std/tests/telemetry_publish.rs @@ -0,0 +1,265 @@ +//! leviculum#77: the link telemetry a host reads, end to end on real nodes. +//! The link list, the lifecycle counters, the mirror check, and the metrics +//! `publish_metrics` records onto an installed recorder. + +use std::collections::BTreeMap; +use std::net::{SocketAddr, TcpListener as StdTcpListener}; +use std::sync::atomic::{AtomicU16, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use leviculum_core::{Destination, DestinationType, Direction, Identity}; +use leviculum_std::driver::{ReticulumNode, ReticulumNodeBuilder}; +// Named through leviculum_std alone, as a consumer without a direct +// leviculum-core dependency would (Codex review on PR #76). +use leviculum_std::{ + EventReceiver, LinkCensus, LinkCloseReason, LinkCountCheck, LinkInfo, LinkLifecycle, LinkRole, + NodeEvent, NodeHeapCensus, +}; +use metrics::{ + Counter, CounterFn, Gauge, GaugeFn, Histogram, Key, KeyName, Metadata, Recorder, SharedString, + Unit, +}; + +static PORT_COUNTER: AtomicU16 = AtomicU16::new(60910); + +fn next_port() -> u16 { + loop { + let candidate = PORT_COUNTER.fetch_add(1, Ordering::Relaxed); + if candidate >= 60990 { + PORT_COUNTER.store(60910, Ordering::Relaxed); + continue; + } + if StdTcpListener::bind(("127.0.0.1", candidate)).is_ok() { + return candidate; + } + } +} + +/// A recorder that keeps the last value written to every series, keyed +/// `name{label=value,...}`. +#[derive(Default, Clone)] +struct Capture(Arc>>, Arc>>); + +struct Series(String, Arc>>); + +impl CounterFn for Series { + fn increment(&self, value: u64) { + *self.1.lock().unwrap().entry(self.0.clone()).or_default() += value as f64; + } + fn absolute(&self, value: u64) { + self.1.lock().unwrap().insert(self.0.clone(), value as f64); + } +} + +impl GaugeFn for Series { + fn increment(&self, value: f64) { + *self.1.lock().unwrap().entry(self.0.clone()).or_default() += value; + } + fn decrement(&self, value: f64) { + *self.1.lock().unwrap().entry(self.0.clone()).or_default() -= value; + } + fn set(&self, value: f64) { + self.1.lock().unwrap().insert(self.0.clone(), value); + } +} + +fn series_name(key: &Key) -> String { + let labels: Vec = key + .labels() + .map(|l| format!("{}={}", l.key(), l.value())) + .collect(); + if labels.is_empty() { + key.name().to_string() + } else { + format!("{}{{{}}}", key.name(), labels.join(",")) + } +} + +impl Recorder for Capture { + fn describe_counter(&self, name: KeyName, _: Option, _: SharedString) { + self.1.lock().unwrap().push(name.as_str().to_string()); + } + fn describe_gauge(&self, name: KeyName, _: Option, _: SharedString) { + self.1.lock().unwrap().push(name.as_str().to_string()); + } + fn describe_histogram(&self, _: KeyName, _: Option, _: SharedString) {} + fn register_counter(&self, key: &Key, _: &Metadata<'_>) -> Counter { + Counter::from_arc(Arc::new(Series(series_name(key), self.0.clone()))) + } + fn register_gauge(&self, key: &Key, _: &Metadata<'_>) -> Gauge { + Gauge::from_arc(Arc::new(Series(series_name(key), self.0.clone()))) + } + fn register_histogram(&self, _: &Key, _: &Metadata<'_>) -> Histogram { + Histogram::noop() + } +} + +impl Capture { + fn publish(&self, node: &ReticulumNode) -> BTreeMap { + metrics::with_local_recorder(self, || node.publish_metrics()); + self.0.lock().unwrap().clone() + } +} + +struct TestNode { + node: ReticulumNode, + rx: EventReceiver, +} + +async fn start(builder: ReticulumNodeBuilder) -> TestNode { + let storage = tempfile::tempdir().expect("tempdir"); + let mut node = builder + .storage_path(storage.path().to_path_buf()) + .build() + .await + .expect("build node"); + std::mem::forget(storage); + node.start().await.expect("start node"); + let rx = node.take_event_receiver().expect("event rx"); + TestNode { node, rx } +} + +async fn saw(rx: &mut EventReceiver, mut pred: impl FnMut(&NodeEvent) -> bool) -> bool { + let deadline = tokio::time::Instant::now() + Duration::from_secs(10); + loop { + match tokio::time::timeout_at(deadline, rx.recv()).await { + Ok(Some(ev)) if pred(&ev) => return true, + Ok(Some(_)) => {} + _ => return false, + } + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn the_link_telemetry_follows_a_link_through_its_life() { + let port = next_port(); + let addr: SocketAddr = format!("127.0.0.1:{port}").parse().unwrap(); + let mut a = start( + ReticulumNodeBuilder::new() + .enable_transport(false) + .add_tcp_server(addr), + ) + .await; + let mut b = start( + ReticulumNodeBuilder::new() + .enable_transport(false) + .add_tcp_client(addr), + ) + .await; + tokio::time::sleep(Duration::from_millis(500)).await; + + let identity = Identity::generate(&mut rand_core::OsRng); + let signing_key: [u8; 32] = identity.public_key_bytes()[32..64].try_into().unwrap(); + let mut dest = Destination::new( + Some(identity), + Direction::In, + DestinationType::Single, + "telemetry", + &["e2e"], + ) + .unwrap(); + dest.set_accepts_links(true); + let hash = *dest.hash(); + a.node.register_destination(dest); + a.node.announce_destination(&hash, None).await.unwrap(); + assert!( + saw(&mut b.rx, |ev| matches!( + ev, + NodeEvent::AnnounceReceived { .. } + )) + .await, + "B hears A" + ); + + let handle = b.node.connect(&hash, &signing_key).await.expect("connect"); + let link_id = *handle.link_id(); + b.node + .await_link_established(&link_id) + .await + .expect("established"); + assert!( + saw(&mut a.rx, |ev| matches!( + ev, + NodeEvent::LinkEstablished { .. } + )) + .await, + "A sees the link" + ); + + // The list, from both ends. Every telemetry result is nameable from + // leviculum_std alone. + let _: LinkCensus = b.node.link_census(); + let _: LinkLifecycle = b.node.link_lifecycle(); + let _: NodeHeapCensus = b.node.heap_census(); + let _: LinkCountCheck = b.node.link_count_check(); + let list: Vec = b.node.link_list(); + assert_eq!(list.len(), 1); + assert_eq!(list[0].link_id, link_id, "the id the application was given"); + assert_eq!(list[0].role, LinkRole::Initiator); + assert_eq!(list[0].destination_hash, hash); + assert!(list[0].age_secs.is_some() && list[0].idle_secs.is_some()); + assert!( + list[0].rtt_ms.is_some(), + "an established link has a measured RTT" + ); + assert_eq!(a.node.link_list()[0].role, LinkRole::Responder); + + // The mirror agrees with the core. + let check = b.node.link_count_check(); + assert_eq!((check.mirror, check.core, check.alarms), (1, 1, 0)); + + // The metrics. + let capture = Capture::default(); + let m = capture.publish(&b.node); + assert_eq!(m["leviculum.link.established{role=initiator}"], 1.0); + assert_eq!(m["leviculum.link.live{role=initiator}"], 1.0); + assert_eq!(m["leviculum.link.live{role=responder}"], 0.0); + assert_eq!(m["leviculum.link.mirror"], 1.0); + assert_eq!(m["leviculum.link.mirror_skew"], 0.0); + assert_eq!(m["leviculum.link.max_per_destination"], 1.0); + assert!(m["leviculum.transport.packets{direction=sent}"] > 0.0); + assert!(m["leviculum.memory.bytes{component=links}"] > 0.0); + assert!(m.contains_key("leviculum.transport.dropped{reason=no-path}")); + let a_capture = Capture::default(); + let a_metrics = a_capture.publish(&a.node); + assert_eq!(a_metrics["leviculum.link.established{role=responder}"], 1.0); + // A recorder that arrives after earlier publishes still learns every + // unit and description. + assert_eq!( + a_capture.1.lock().unwrap().len(), + leviculum_std::telemetry::METRIC_CATALOG.len() + ); + + // Two nodes publishing into one recorder sum their counters, rather than + // the series holding only the larger total. + let both = Capture::default(); + both.publish(&a.node); + let m = both.publish(&b.node); + let sent = a.node.transport_stats().packets_sent() + b.node.transport_stats().packets_sent(); + assert!( + m["leviculum.transport.packets{direction=sent}"] >= sent as f64, + "summed: {} vs {sent}", + m["leviculum.transport.packets{direction=sent}"] + ); + assert_eq!(m["leviculum.link.established{role=initiator}"], 1.0); + assert_eq!(m["leviculum.link.established{role=responder}"], 1.0); + + // Close it: the counters move, the gauges fall, the ends agree on why. + b.node.close_link(&link_id).await.unwrap(); + assert!( + saw(&mut a.rx, |ev| matches!(ev, NodeEvent::LinkClosed { .. })).await, + "A sees the close" + ); + assert_eq!(b.node.link_lifecycle().closed(LinkCloseReason::Normal), 1); + assert_eq!( + a.node.link_lifecycle().closed(LinkCloseReason::PeerClosed), + 1 + ); + let m = capture.publish(&b.node); + assert_eq!(m["leviculum.link.closed{reason=normal}"], 1.0); + assert_eq!(m["leviculum.link.live{role=initiator}"], 0.0); + assert!(b.node.link_list().is_empty()); + let check = b.node.link_count_check(); + assert_eq!((check.mirror, check.core), (0, 0)); +} diff --git a/scripts/core-lock-census.txt b/scripts/core-lock-census.txt index 4eb6c3cde..dfc82ec5e 100644 --- a/scripts/core-lock-census.txt +++ b/scripts/core-lock-census.txt @@ -27,7 +27,7 @@ # # Baseline measured 2026-09-17 (Codeberg #199). -TOTAL 58 +TOTAL 65 driver::ReticulumNode::accept_resource driver::ReticulumNode::active_link_count @@ -43,14 +43,20 @@ driver::ReticulumNode::get_identity driver::ReticulumNode::get_path_clone driver::ReticulumNode::get_remote_identity driver::ReticulumNode::has_path +driver::ReticulumNode::heap_census driver::ReticulumNode::hops_to driver::ReticulumNode::identify_link driver::ReticulumNode::identity_hash driver::ReticulumNode::interface_stats driver::ReticulumNode::is_transport_enabled +driver::ReticulumNode::known_destination_evictions driver::ReticulumNode::known_remote_ratchet +driver::ReticulumNode::link_census +driver::ReticulumNode::link_count_check driver::ReticulumNode::link_destination driver::ReticulumNode::link_is_established +driver::ReticulumNode::link_lifecycle +driver::ReticulumNode::link_list driver::ReticulumNode::link_mdu driver::ReticulumNode::link_negotiated_mtu driver::ReticulumNode::link_stats @@ -59,6 +65,7 @@ driver::ReticulumNode::now_ms driver::ReticulumNode::path_count driver::ReticulumNode::path_table_entries driver::ReticulumNode::pending_link_count +driver::ReticulumNode::publish_metrics driver::ReticulumNode::rate_table_entries driver::ReticulumNode::register_destination driver::ReticulumNode::register_request_handler