From 7d2ae854fc9b05cc783c9daa3d3736a459cf734e Mon Sep 17 00:00:00 2001 From: Eric Moore Date: Wed, 7 Oct 2026 10:22:53 -0500 Subject: [PATCH 1/5] feat(node): link census, live links by role and destination (CIRISEdge#819, leviculum#57) A field gauge for link accumulation. NodeCore::link_census, and ReticulumNode::link_census in the driver, return the established links (Active or Stale) split into initiator and responder, a count of pending ones, and the same split per destination, most links first. It names the owner of a growing count. The initiator keeps a link alive with its keepalives, so a rising initiator count toward one destination means this node is not closing what it dialled. A rising responder count under its own destination means the peers are not closing theirs. CIRISEdge#819 traced a canonical's 1024 links per 35 min to the first case: its dial pool grows to demand and never shrinks. Before that, diagnosing it took a reading of the reap rules from the source. One walk of the link table under a brief core lock. Added to the lock census, together with known_destination_evictions (leviculum#49), which should have been added when that method landed. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01BXj7C4ZeFuczxxDVadAHZ5 --- leviculum-core/src/node/link_management.rs | 49 +++++++++ leviculum-core/src/node/mod.rs | 27 +++++ leviculum-core/src/node/mvr_link_census.rs | 110 +++++++++++++++++++++ leviculum-std/src/driver/mod.rs | 8 ++ scripts/core-lock-census.txt | 4 +- 5 files changed, 197 insertions(+), 1 deletion(-) create mode 100644 leviculum-core/src/node/mvr_link_census.rs diff --git a/leviculum-core/src/node/link_management.rs b/leviculum-core/src/node/link_management.rs index cf80140fb..65dd502f6 100644 --- a/leviculum-core/src/node/link_management.rs +++ b/leviculum-core/src/node/link_management.rs @@ -636,6 +636,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, diff --git a/leviculum-core/src/node/mod.rs b/leviculum-core/src/node/mod.rs index 2e9918eda..c51023c03 100644 --- a/leviculum-core/src/node/mod.rs +++ b/leviculum-core/src/node/mod.rs @@ -12825,3 +12825,30 @@ 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; 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..e03c0241c --- /dev/null +++ b/leviculum-core/src/node/mvr_link_census.rs @@ -0,0 +1,110 @@ +//! 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::node::{LinkCensus, 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); + + // A closed link leaves the census. + let _ = a.close_link(&ids[0]); + assert_eq!(a.link_census().initiator, 1); +} diff --git a/leviculum-std/src/driver/mod.rs b/leviculum-std/src/driver/mod.rs index 264ed3e24..641f9e42b 100644 --- a/leviculum-std/src/driver/mod.rs +++ b/leviculum-std/src/driver/mod.rs @@ -2624,6 +2624,14 @@ impl ReticulumNode { self.inner.lock_recover().storage().identity_evictions() } + /// 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. diff --git a/scripts/core-lock-census.txt b/scripts/core-lock-census.txt index 4eb6c3cde..26370f7ea 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 60 driver::ReticulumNode::accept_resource driver::ReticulumNode::active_link_count @@ -48,7 +48,9 @@ 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_destination driver::ReticulumNode::link_is_established driver::ReticulumNode::link_mdu From 5748fd3dcfe9cbe2f63da12cb904a32db799a55e Mon Sep 17 00:00:00 2001 From: Eric Moore Date: Wed, 7 Oct 2026 11:07:18 -0500 Subject: [PATCH 2/5] feat(telemetry): link list, lifecycle counters, mirror alarm, metrics facade (leviculum#77) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The P0 and P1 leviculum asks of CIRISServer's unified-telemetry FSD (PR #746, §4). The maintainer has made instrumentation the priority before the next scale step. P0 - ReticulumNode::link_list (NodeCore::link_list): every link with its caller-visible id, destination, role, state, age, idle seconds, RTT and interface. - Cumulative lifecycle counters, link_lifecycle: established by role, closed by LinkCloseReason, and handshake_failed by reason for links that never established. remove_link notes whether the departing link had established, and emit_link_closed counts it. Every close follows its removal, so established - closed equals the live census. - Divergence alarm. Every 10 s the event loop compares the completion mirror with the core's established count (MirrorWatch). A disagreement that outlasts 30 s raises one LINK_MIRROR_DIVERGED event (catalogued) per episode, counted in link_count_check().alarms. - The per-destination census from the previous commit. P1, facade per FSD §5: `metrics` in the library, the exporter in the host - The `metrics` crate (default-features off; its one dependency is rapidhash). ReticulumNode::publish_metrics records the lifecycle, the census, the mirror check, TransportStats (drops by every DropReason, through the new TransportStats::drops_for), PlaneStats, the known- destination evictions and the core's heap census. That last one gives std hosts per-component memory and a mean bytes-per-link. It is pull based: the host calls it on its scrape cadence. Counters use absolute(), so a scrape reads the node's own totals. - telemetry::METRIC_CATALOG lists 24 metrics, each with a kind, a unit, its allowed label keys and a description; describe_* is registered from it once. Every label value is a bounded enumeration. No link, destination or peer id is ever a label. - event_catalog_completeness now scans metric sites too. Every recorded name must be catalogued, every catalogued name recorded, and every label key allowed. The source walk is shared with the event scan. Housekeeping - THIRD-PARTY-NOTICES regenerated for metrics and rapidhash. This also picks up a stale rustls version line that was already wrong on main. - leviculum-nrf/Cargo.lock now carries the fork's version string, which the release bumps never updated. cargo-about --frozen failed on it. - The lock census adds the new methods (TOTAL 65). - event_log.rs's apply_inbound citation already pointed at unrelated code on main; it is corrected. - The doc-citation guard reports no drift beyond what main already has. Tests: mvr_link_census (list, counters by reason, handshake failure, and the counters agreeing with the census); drops_for over every reason; MirrorWatch (grace, one alarm per episode, re-arm); the metric scanner; and telemetry_publish, which runs two real nodes over TCP through the whole life of a link with a capturing recorder. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01BXj7C4ZeFuczxxDVadAHZ5 --- Cargo.lock | 32 +- THIRD-PARTY-NOTICES | 248 ++++++++- leviculum-core/src/node/link_management.rs | 10 + leviculum-core/src/node/mod.rs | 10 + leviculum-core/src/node/mvr_link_census.rs | 43 +- leviculum-core/src/node/telemetry.rs | 267 ++++++++++ leviculum-nrf/Cargo.lock | 4 +- leviculum-std/Cargo.toml | 5 + leviculum-std/src/driver/mod.rs | 93 ++++ .../src/event_catalog_completeness.rs | 145 +++++- leviculum-std/src/event_log.rs | 8 +- leviculum-std/src/lib.rs | 5 + leviculum-std/src/telemetry.rs | 470 ++++++++++++++++++ leviculum-std/tests/telemetry_publish.rs | 228 +++++++++ scripts/core-lock-census.txt | 7 +- 15 files changed, 1553 insertions(+), 22 deletions(-) create mode 100644 leviculum-core/src/node/telemetry.rs create mode 100644 leviculum-std/src/telemetry.rs create mode 100644 leviculum-std/tests/telemetry_publish.rs 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 65dd502f6..fc106e34a 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 @@ -1284,6 +1290,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, @@ -1299,6 +1306,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, @@ -1529,6 +1537,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, @@ -4271,6 +4280,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 c51023c03..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, @@ -12852,3 +12855,10 @@ pub struct DestinationLinks { } #[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 index e03c0241c..5ee3b368d 100644 --- a/leviculum-core/src/node/mvr_link_census.rs +++ b/leviculum-core/src/node/mvr_link_census.rs @@ -12,7 +12,8 @@ use rand_core::OsRng; use crate::destination::{Destination, DestinationType, Direction}; use crate::identity::Identity; -use crate::node::{LinkCensus, NodeCore, NodeCoreBuilder}; +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}; @@ -104,7 +105,43 @@ fn the_census_counts_live_links_by_role_and_destination() { assert_eq!((census.initiator, census.responder), (0, 2)); assert_eq!(census.by_destination[0].responder, 2); - // A closed link leaves the census. - let _ = a.close_link(&ids[0]); + // 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 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), 1); + 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..117ba6ee3 --- /dev/null +++ b/leviculum-core/src/node/telemetry.rs @@ -0,0 +1,267 @@ +//! 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. + pub rtt_ms: u64, + /// 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. 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, + 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_ms(), + interface_index: link.attached_interface(), + } + }) + .collect() + } + + /// Cumulative link lifecycle counters (leviculum#77). + pub fn link_lifecycle(&self) -> LinkLifecycle { + self.link_telemetry.lifecycle + } + + /// 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 641f9e42b..94904994b 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). @@ -2624,6 +2625,72 @@ 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`. + /// 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, + }); + } + /// 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. @@ -4680,6 +4747,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); @@ -5326,6 +5400,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..5218b356d 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:5172`), 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..05d04c359 100644 --- a/leviculum-std/src/lib.rs +++ b/leviculum-std/src/lib.rs @@ -95,3 +95,8 @@ 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; diff --git a/leviculum-std/src/telemetry.rs b/leviculum-std/src/telemetry.rs new file mode 100644 index 000000000..e6b7d825f --- /dev/null +++ b/leviculum-std/src/telemetry.rs @@ -0,0 +1,470 @@ +//! 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 set with `absolute`, so a +//! scrape always reads the node's own cumulative total. +//! +//! [`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 std::sync::Once; + +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.", + ), + 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, +} + +static DESCRIBED: Once = Once::new(); + +/// Register every catalogue entry's unit and description with the +/// installed recorder, once. +fn describe_all() { + DESCRIBED.call_once(|| { + 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), + } + } + }); +} + +/// Record a snapshot onto whatever recorder is installed. +pub(crate) fn record(s: &Snapshot) { + describe_all(); + + // Link lifecycle + let lc = &s.lifecycle; + counter!("leviculum.link.established", "role" => "initiator") + .absolute(lc.established_initiator); + counter!("leviculum.link.established", "role" => "responder") + .absolute(lc.established_responder); + for reason in LINK_CLOSE_REASONS { + let name = link_close_reason_name(reason); + counter!("leviculum.link.closed", "reason" => name).absolute(lc.closed(reason)); + counter!("leviculum.link.handshake_failed", "reason" => name) + .absolute(lc.handshake_failed(reason)); + } + 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(s.mirror_alarms); + + // Transport + let t = &s.transport; + counter!("leviculum.transport.packets", "direction" => "sent").absolute(t.packets_sent()); + counter!("leviculum.transport.packets", "direction" => "received") + .absolute(t.packets_received()); + counter!("leviculum.transport.packets", "direction" => "forwarded") + .absolute(t.packets_forwarded()); + counter!("leviculum.transport.packets", "direction" => "forwarded_link") + .absolute(t.packets_forwarded_link()); + counter!("leviculum.transport.announces_processed").absolute(t.announces_processed()); + for reason in DropReason::ALL { + counter!("leviculum.transport.dropped", "reason" => reason.kebab()) + .absolute(t.drops_for(reason)); + } + counter!("leviculum.transport.suppressed", "kind" => "lr_local_client_redirect") + .absolute(t.lr_local_client_redirects()); + counter!("leviculum.transport.suppressed", "kind" => "path_request_pending") + .absolute(t.path_request_pending_suppressions()); + counter!("leviculum.transport.suppressed", "kind" => "path_request_retry") + .absolute(t.path_request_retry_withholds()); + counter!("leviculum.known_destination.evictions").absolute(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(p.control_dropped_total); + counter!("leviculum.events.dropped", "plane" => "data").absolute(p.data_dropped_total); + gauge!("leviculum.retry.queued").set(p.retry_queued as f64); + counter!("leviculum.retry.dropped").absolute(p.retry_dropped_total); + counter!("leviculum.shed.packets").absolute(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..d48333e8a --- /dev/null +++ b/leviculum-std/tests/telemetry_publish.rs @@ -0,0 +1,228 @@ +//! 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::link::LinkCloseReason; +use leviculum_core::node::LinkRole; +use leviculum_core::{Destination, DestinationType, Direction, Identity}; +use leviculum_std::driver::{ReticulumNode, ReticulumNodeBuilder}; +use leviculum_std::{EventReceiver, NodeEvent}; +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>>); + +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, _: KeyName, _: Option, _: SharedString) {} + fn describe_gauge(&self, _: KeyName, _: Option, _: SharedString) {} + 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. + let list = 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_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_metrics = Capture::default().publish(&a.node); + assert_eq!(a_metrics["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 26370f7ea..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 60 +TOTAL 65 driver::ReticulumNode::accept_resource driver::ReticulumNode::active_link_count @@ -43,6 +43,7 @@ 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 @@ -51,8 +52,11 @@ 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 @@ -61,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 From 24dca1c77045df16506b000839ad370dd59ed959 Mon Sep 17 00:00:00 2001 From: Eric Moore Date: Wed, 7 Oct 2026 11:09:05 -0500 Subject: [PATCH 3/5] fix(telemetry): re-export the telemetry result types from leviculum-std (Codex on #76) LinkCensus, DestinationLinks, LinkInfo, LinkLifecycle, LinkRole, LINK_CLOSE_REASONS, link_close_reason_name, LinkState and NodeHeapCensus are now nameable from leviculum_std, as LinkStats and TransportStats already were. A consumer without its own leviculum-core dependency can use them in annotations and fields. telemetry_publish names every result through leviculum_std alone. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01BXj7C4ZeFuczxxDVadAHZ5 --- leviculum-std/src/lib.rs | 10 ++++++++++ leviculum-std/tests/telemetry_publish.rs | 18 +++++++++++++----- 2 files changed, 23 insertions(+), 5 deletions(-) diff --git a/leviculum-std/src/lib.rs b/leviculum-std/src/lib.rs index 05d04c359..1e8acdd60 100644 --- a/leviculum-std/src/lib.rs +++ b/leviculum-std/src/lib.rs @@ -100,3 +100,13 @@ pub use rpc::{ /// 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/tests/telemetry_publish.rs b/leviculum-std/tests/telemetry_publish.rs index d48333e8a..310c57d4f 100644 --- a/leviculum-std/tests/telemetry_publish.rs +++ b/leviculum-std/tests/telemetry_publish.rs @@ -8,11 +8,14 @@ use std::sync::atomic::{AtomicU16, Ordering}; use std::sync::{Arc, Mutex}; use std::time::Duration; -use leviculum_core::link::LinkCloseReason; -use leviculum_core::node::LinkRole; use leviculum_core::{Destination, DestinationType, Direction, Identity}; use leviculum_std::driver::{ReticulumNode, ReticulumNodeBuilder}; -use leviculum_std::{EventReceiver, NodeEvent}; +// 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, @@ -180,8 +183,13 @@ async fn the_link_telemetry_follows_a_link_through_its_life() { "A sees the link" ); - // The list, from both ends. - let list = b.node.link_list(); + // 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); From 7e33a815780dbfddd5c176e4e2a408a169f874f5 Mon Sep 17 00:00:00 2001 From: Eric Moore Date: Wed, 7 Oct 2026 11:44:43 -0500 Subject: [PATCH 4/5] fix(telemetry): Codex on #76: rejected links counted, unmeasured RTT is None, describes per publish, process-total counters - reject_link counts the refused request, as LinkLifecycle::rejected and as leviculum.link.handshake_failed{reason="rejected"}. remove_link noted the departing link but reject_link reports no close, so refused handshakes vanished from the metrics. - LinkInfo::rtt_ms is Option: None until a sample exists, as in LinkStats. Before, a pending link reported the 500 ms working fallback as though it had been measured. - Units and descriptions are registered on every publish instead of once per process, so a recorder installed (or swapped for a local one) after the first publish still learns them. - Counters are process totals. Each node keeps the totals it last published (CounterBaseline, held by ReticulumNode) and adds its growth to one process-wide sum per series, which absolute() records. Two nodes in one process now sum, and a replacement node carries on instead of sitting under its predecessor's total, which absolute's keep-the- maximum contract had caused. A recorder installed late still reads full totals. Plain per-node increments would have lost exactly that. Tests: the census test covers a refused request. telemetry_publish checks that a late recorder receives every description, that two nodes sum into one recorder, and that an established link has a measured RTT. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01BXj7C4ZeFuczxxDVadAHZ5 --- leviculum-core/src/node/link_management.rs | 4 + leviculum-core/src/node/mvr_link_census.rs | 15 +- leviculum-core/src/node/telemetry.rs | 19 +- leviculum-std/src/driver/mod.rs | 33 ++-- leviculum-std/src/event_log.rs | 2 +- leviculum-std/src/telemetry.rs | 211 ++++++++++++++++----- leviculum-std/tests/telemetry_publish.rs | 37 +++- 7 files changed, 252 insertions(+), 69 deletions(-) diff --git a/leviculum-core/src/node/link_management.rs b/leviculum-core/src/node/link_management.rs index fc106e34a..a678c50b6 100644 --- a/leviculum-core/src/node/link_management.rs +++ b/leviculum-core/src/node/link_management.rs @@ -579,7 +579,11 @@ impl NodeCore { /// there is nothing to resolve. If responder-side re-keying is ever /// added, this becomes the next stale-id bug. pub fn reject_link(&mut self, link_id: &LinkId) { + let existed = self.links.contains_key(link_id); self.remove_link(link_id); + if existed { + self.note_link_rejected(); + } } /// Close a link gracefully diff --git a/leviculum-core/src/node/mvr_link_census.rs b/leviculum-core/src/node/mvr_link_census.rs index 5ee3b368d..25cdb48fe 100644 --- a/leviculum-core/src/node/mvr_link_census.rs +++ b/leviculum-core/src/node/mvr_link_census.rs @@ -133,10 +133,23 @@ fn the_census_counts_live_links_by_role_and_destination() { 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); + 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), 1); + 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(); diff --git a/leviculum-core/src/node/telemetry.rs b/leviculum-core/src/node/telemetry.rs index 117ba6ee3..cfb2f201d 100644 --- a/leviculum-core/src/node/telemetry.rs +++ b/leviculum-core/src/node/telemetry.rs @@ -42,8 +42,9 @@ pub struct LinkInfo { /// 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. - pub rtt_ms: u64, + /// 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, } @@ -90,7 +91,8 @@ pub fn link_close_reason_name(reason: LinkCloseReason) -> &'static str { /// 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. So +/// 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)] @@ -99,6 +101,8 @@ pub struct LinkLifecycle { 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], } @@ -165,7 +169,7 @@ impl NodeCore { 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_ms(), + rtt_ms: link.rtt_us().map(|us| us / 1000), interface_index: link.attached_interface(), } }) @@ -177,6 +181,13 @@ impl NodeCore { 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); diff --git a/leviculum-std/src/driver/mod.rs b/leviculum-std/src/driver/mod.rs index 94904994b..cd3424192 100644 --- a/leviculum-std/src/driver/mod.rs +++ b/leviculum-std/src/driver/mod.rs @@ -1512,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. @@ -1725,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, @@ -2664,6 +2666,8 @@ impl ReticulumNode { /// (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) = { @@ -2676,19 +2680,22 @@ impl ReticulumNode { 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, - }); + 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, diff --git a/leviculum-std/src/event_log.rs b/leviculum-std/src/event_log.rs index 5218b356d..b4fc69c02 100644 --- a/leviculum-std/src/event_log.rs +++ b/leviculum-std/src/event_log.rs @@ -1323,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:5172`), 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/telemetry.rs b/leviculum-std/src/telemetry.rs index e6b7d825f..11fff5c42 100644 --- a/leviculum-std/src/telemetry.rs +++ b/leviculum-std/src/telemetry.rs @@ -8,8 +8,16 @@ //! 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 set with `absolute`, so a -//! scrape always reads the node's own cumulative total. +//! 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 @@ -18,8 +26,6 @@ //! enumerations (role, reason, plane, component): no link id, destination or //! peer key ever becomes a label. -use std::sync::Once; - 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}; @@ -99,7 +105,8 @@ pub const METRIC_CATALOG: &[MetricSpec] = &[ "leviculum.link.handshake_failed", Unit::Count, &["reason"], - "Links that ended before they were established, by close reason.", + "Links that ended before they were established, by close reason, plus \ + reason=rejected for requests the application refused.", ), gauge_spec( "leviculum.link.live", @@ -252,37 +259,88 @@ pub(crate) struct Snapshot { pub heap: NodeHeapCensus, } -static DESCRIBED: Once = Once::new(); - -/// Register every catalogue entry's unit and description with the -/// installed recorder, once. +/// 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() { - DESCRIBED.call_once(|| { - 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), - } + 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) { +pub(crate) fn record(s: &Snapshot, b: &mut CounterBaseline) { describe_all(); // Link lifecycle let lc = &s.lifecycle; - counter!("leviculum.link.established", "role" => "initiator") - .absolute(lc.established_initiator); - counter!("leviculum.link.established", "role" => "responder") - .absolute(lc.established_responder); + 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(lc.closed(reason)); - counter!("leviculum.link.handshake_failed", "reason" => name) - .absolute(lc.handshake_failed(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); @@ -299,29 +357,74 @@ pub(crate) fn record(s: &Snapshot) { 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(s.mirror_alarms); + 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(t.packets_sent()); - counter!("leviculum.transport.packets", "direction" => "received") - .absolute(t.packets_received()); - counter!("leviculum.transport.packets", "direction" => "forwarded") - .absolute(t.packets_forwarded()); - counter!("leviculum.transport.packets", "direction" => "forwarded_link") - .absolute(t.packets_forwarded_link()); - counter!("leviculum.transport.announces_processed").absolute(t.announces_processed()); + 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(t.drops_for(reason)); + 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(t.lr_local_client_redirects()); - counter!("leviculum.transport.suppressed", "kind" => "path_request_pending") - .absolute(t.path_request_pending_suppressions()); - counter!("leviculum.transport.suppressed", "kind" => "path_request_retry") - .absolute(t.path_request_retry_withholds()); - counter!("leviculum.known_destination.evictions").absolute(s.known_evictions); + 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; @@ -329,11 +432,27 @@ pub(crate) fn record(s: &Snapshot) { 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(p.control_dropped_total); - counter!("leviculum.events.dropped", "plane" => "data").absolute(p.data_dropped_total); + 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(p.retry_dropped_total); - counter!("leviculum.shed.packets").absolute(p.shed_packets_total); + 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 diff --git a/leviculum-std/tests/telemetry_publish.rs b/leviculum-std/tests/telemetry_publish.rs index 310c57d4f..f44441573 100644 --- a/leviculum-std/tests/telemetry_publish.rs +++ b/leviculum-std/tests/telemetry_publish.rs @@ -39,7 +39,7 @@ fn next_port() -> u16 { /// A recorder that keeps the last value written to every series, keyed /// `name{label=value,...}`. #[derive(Default, Clone)] -struct Capture(Arc>>); +struct Capture(Arc>>, Arc>>); struct Series(String, Arc>>); @@ -77,8 +77,12 @@ fn series_name(key: &Key) -> String { } impl Recorder for Capture { - fn describe_counter(&self, _: KeyName, _: Option, _: SharedString) {} - fn describe_gauge(&self, _: KeyName, _: Option, _: SharedString) {} + 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()))) @@ -195,6 +199,10 @@ async fn the_link_telemetry_follows_a_link_through_its_life() { 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. @@ -213,8 +221,29 @@ async fn the_link_telemetry_follows_a_link_through_its_life() { 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_metrics = Capture::default().publish(&a.node); + 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(); From a4b2554705fea24111edcd5b46f9cf92267d086b Mon Sep 17 00:00:00 2001 From: Eric Moore Date: Wed, 7 Oct 2026 11:55:29 -0500 Subject: [PATCH 5/5] fix(node): reject_link refuses only a pending inbound request; anything else is closed (Codex on #76) reject_link removed whatever link it was given, without a close. On an established or locally dialled link, that left the completion mirror and the lifecycle counters believing the link alive, and broke established - closed == live. Now: - a pending inbound request is refused and counted as rejected; - any other link is closed the way close_link closes it (close packet, LinkClosed with reason Normal, counted as a close). Its events go out with the node's next output. Test: rejecting a live link counts a close, not a refusal, and the counters still agree with the census. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01BXj7C4ZeFuczxxDVadAHZ5 --- leviculum-core/src/node/link_management.rs | 31 +++++++++++++++++++--- leviculum-core/src/node/mvr_link_census.rs | 16 +++++++++++ 2 files changed, 44 insertions(+), 3 deletions(-) diff --git a/leviculum-core/src/node/link_management.rs b/leviculum-core/src/node/link_management.rs index a678c50b6..6ec431be1 100644 --- a/leviculum-core/src/node/link_management.rs +++ b/leviculum-core/src/node/link_management.rs @@ -578,12 +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 existed = self.links.contains_key(link_id); - self.remove_link(link_id); - if existed { + 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 diff --git a/leviculum-core/src/node/mvr_link_census.rs b/leviculum-core/src/node/mvr_link_census.rs index 25cdb48fe..711692242 100644 --- a/leviculum-core/src/node/mvr_link_census.rs +++ b/leviculum-core/src/node/mvr_link_census.rs @@ -144,6 +144,22 @@ fn the_census_counts_live_links_by_role_and_destination() { 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.