diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 23d3e83..741541a 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -110,6 +110,17 @@ jobs: restore-keys: | ${{ runner.os }}-cargo- - run: cargo clippy --all-targets --all-features -- -D warnings + # An optional feature means two source trees, and --all-features lints + # only one of them. The default build is what ships, so lint it + # explicitly: without this stage, code that compiles only with + # `profiling` enabled would pass CI while breaking every release build. + # Mirrored in testing/ci-local.sh — check-ci-parity.sh compares + # integration suites only and will not catch a stage added to one runner + # and not the other. + - name: Clippy (default features) + run: cargo clippy --all-targets -- -D warnings + - name: Build with the tick-body profiler enabled + run: cargo build --workspace --features profiling # ─────────────────────────────────────────────────────────────────────────── # Android cross-check @@ -293,6 +304,12 @@ jobs: check_name: Unit Tests Summary fail_on_failure: false + # The `profiling` feature adds a module, a recorder and a writer thread + # that the default-feature run above never compiles, so its own tests do + # not execute there. Mirrored in testing/ci-local.sh. + - name: Run library tests with the tick-body profiler enabled + run: cargo test --lib --features profiling + # ───────────────────────────────────────────────────────────────────────────── # Job 2b – Unit tests (macOS) # ───────────────────────────────────────────────────────────────────────────── diff --git a/CHANGELOG.md b/CHANGELOG.md index e34c0a2..9e176a3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -104,6 +104,17 @@ with v0.4.x or earlier peers. - The receive-path `RejectReason` classification (shipped in 0.4.0) is additionally wired into the Noise XX handshake cluster (msg1/msg2/msg3) and the rekey-initiator outbound sites on `next`. +- An optional tick-body profiler behind the new `profiling` Cargo feature, + **off by default**. When enabled, `fipsctl profile tick on [--dir PATH]` / + `off` / `status` starts and stops a capture at runtime with no restart. Each + capture writes one tab-separated file (default `/var/log/fips`, capped at + 32 MB) carrying, per ten-second interval, the exact count, max and total for + every step of the rx-loop tick arm, the whole-tick span, and gauges for ticks, + peer count, the gap between successive tick-arm entries and the resulting + arm-starvation delay. With the feature off the instrumentation macro is a pure + pass-through, so a default build contains no timing code on the tick path. + `LogsDirectory=fips` was added to the packaged systemd units so the capture + directory is created and cleaned up declaratively. ### Changed diff --git a/Cargo.toml b/Cargo.toml index 76e0b3d..e90d155 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -11,6 +11,14 @@ readme = "README.md" keywords = ["mesh", "p2p", "decentralized", "overlay-network", "nostr"] categories = ["network-programming", "command-line-utilities", "cryptography"] +[features] +default = [] +# Tick-body profiler (`src/instr`). Off by default: enabling it edits 26 call +# sites in the rx loop's hot tick arm, and only a compile-time gate makes the +# default build's neutrality a property of the generated code rather than of a +# runtime check. Build with `--features profiling` for a measurement run. +profiling = [] + [dependencies] ratatui = "0.30" secp256k1 = { version = "0.30", features = ["rand", "global-context"] } diff --git a/docs/reference/cli-fipsctl.md b/docs/reference/cli-fipsctl.md index feef6a0..3361e0a 100644 --- a/docs/reference/cli-fipsctl.md +++ b/docs/reference/cli-fipsctl.md @@ -115,6 +115,58 @@ Tell the daemon to drop a peer link. | -------- | ----------- | | `peer` | npub (bech32) or hostname from `/etc/fips/hosts`. | +### `profile tick ` + +> **Reading the output.** Step durations are wall clock measured across `await` +> points, not CPU time: a step that waits on I/O accrues that wait, and other +> tasks may run inside the span. That is the intended measure for head-of-line +> delay, and it means a large step is not necessarily an expensive one. +> `arm_starvation` is measured directly as the entry time minus the deadline +> the interval scheduled that tick for. It is not derived from +> `tick_entry_gap`, which carries no starvation signal on its own: under a +> steady delay every gap is exactly one tick period. + +Start, stop and inspect a capture of the rx-loop tick body. **Present +only when both `fipsctl` and the daemon are built with +`--features profiling`**; the feature is off by default, so a stock +package does not carry this subcommand and a stock daemon reports +`profile_tick_*` as an unknown command. + +| Subcommand | Control-socket command | Description | +| ---------- | ---------------------- | ----------- | +| `profile tick on` | `profile_tick_on` | Create the capture file and start recording. Fails if a capture is already running (naming the active file) or if the directory cannot be written. | +| `profile tick off` | `profile_tick_off` | Stop the capture. The writer is woken immediately, drains once more and is joined, so the command returns promptly. Succeeds, reporting nothing active, when no capture is running. | +| `profile tick status` | `profile_tick_status` | Report `idle`, `running`, `stopped_by_cap` or `stopped_by_error`, plus the active path, bytes written, flush interval and byte cap. | + +`profile tick on` options: + +| Flag | Argument | Default | Description | +| ---- | -------- | ------- | ----------- | +| `--dir` | directory path | `/var/log/fips` | Where to write the capture. Created if absent. Use it to profile a non-root `cargo run`, or on a platform whose log root differs. | + +One file is written per capture, named `profile-.tsv`. +It opens with a `#`-prefixed header block (node npub, build version, +platform, configured tick period, flush interval, byte cap, start +time), then a tab-separated column header, then one row per measured +step per flush interval: + +```text +ts_unix kind domain name count max total unit +``` + +`kind` is `step` for a timed span and `gauge` for a sampled scalar, so +a gauge value never lands under a duration column; `unit` names the +unit of `max` and `total` for that row. Every step present in the build +gets a row every interval, including zero-count rows. Gauges cover +ticks per interval, peer count, the wall gap between successive +tick-arm entries, and the arm-starvation delay, which is measured +against the deadline the tick was scheduled for rather than derived +from the gap. + +A capture stops itself on reaching 32 MB, appending a `#` line saying +so; `profile tick status` then reports `stopped_by_cap` until the next +`on` or `off` clears it. + ## Exit Codes | Code | Meaning | diff --git a/docs/reference/control-socket.md b/docs/reference/control-socket.md index 1d41a3e..2422704 100644 --- a/docs/reference/control-socket.md +++ b/docs/reference/control-socket.md @@ -159,6 +159,21 @@ not reproduced here to avoid duplicating the source. Both commands run on the daemon's main task and may block briefly while the node mutates its state. +#### Profiler toggle (`--features profiling` builds only) + +| Command | Params | Behaviour | +| ------- | ------ | --------- | +| `profile_tick_on` | `dir` (optional directory path; default `/var/log/fips`) | Creates the capture file, publishes its path, and starts the writer thread. `data`: `state`, `path`, `interval_secs`, `byte_cap`. Errors if a capture is already running (naming the active file) or the directory is unwritable. | +| `profile_tick_off` | — | Stops the capture, drains once more, joins the writer. `data`: `state`, `stopped`, `stopped_by_cap`, `stopped_by_error`, `path`, `bytes`. | +| `profile_tick_status` | — | `data`: `state` (`idle` / `running` / `stopped_by_cap` / `stopped_by_error`), `path`, `bytes`, `byte_cap`, `interval_secs`. | + +Unlike `connect` and `disconnect`, these three are served in the +control accept task rather than on the daemon's main task. All of their +state is process statics and none of them needs `&mut Node`, so +routing them through the main loop would only make the toggle queue +behind the tick body it exists to measure. They are absent from a +default build, where the daemon answers them as unknown commands. + ## Gateway Command Catalog `fips-gateway` exposes a separate control socket with its own command diff --git a/packaging/debian/fips.service b/packaging/debian/fips.service index d1b5616..0c131a4 100644 --- a/packaging/debian/fips.service +++ b/packaging/debian/fips.service @@ -19,6 +19,11 @@ RestartSec=5 RuntimeDirectory=fips RuntimeDirectoryMode=0750 +# Log directory (/var/log/fips/), where the built-in tick-body profiler writes +# its capture files. Declared so systemd creates it on start and removes it on +# purge; the daemon runs as root and already has access without it. +LogsDirectory=fips + # Security hardening (daemon runs as root for TUN and raw sockets) ProtectHome=yes PrivateTmp=yes diff --git a/packaging/systemd/fips.service b/packaging/systemd/fips.service index c58385f..fdda394 100644 --- a/packaging/systemd/fips.service +++ b/packaging/systemd/fips.service @@ -14,6 +14,11 @@ RestartSec=5 RuntimeDirectory=fips RuntimeDirectoryMode=0750 +# Log directory (/var/log/fips/), where the built-in tick-body profiler writes +# its capture files. Declared so systemd creates it on start and removes it on +# purge; the daemon runs as root and already has access without it. +LogsDirectory=fips + # Security hardening (daemon runs as root for TUN and raw sockets) ProtectHome=yes PrivateTmp=yes diff --git a/src/bin/fipsctl.rs b/src/bin/fipsctl.rs index 3adf155..6cd868d 100644 --- a/src/bin/fipsctl.rs +++ b/src/bin/fipsctl.rs @@ -76,6 +76,37 @@ enum Commands { #[command(subcommand)] what: StatsCommands, }, + /// Control the built-in profiler (requires a `--features profiling` build) + #[cfg(feature = "profiling")] + Profile { + #[command(subcommand)] + what: ProfileCommands, + }, +} + +#[cfg(feature = "profiling")] +#[derive(Subcommand, Debug)] +enum ProfileCommands { + /// Profile the rx-loop tick body + Tick { + #[command(subcommand)] + action: ProfileTickAction, + }, +} + +#[cfg(feature = "profiling")] +#[derive(Subcommand, Debug)] +enum ProfileTickAction { + /// Start a capture + On { + /// Directory for the capture file (default /var/log/fips) + #[arg(long)] + dir: Option, + }, + /// Stop the running capture + Off, + /// Report capture state + Status, } #[derive(Subcommand, Debug)] @@ -471,6 +502,20 @@ fn main() { build_command("show_stats_history", params) } }, + #[cfg(feature = "profiling")] + Commands::Profile { what } => match what { + ProfileCommands::Tick { action } => match action { + ProfileTickAction::On { dir } => match dir { + Some(dir) => build_command( + "profile_tick_on", + serde_json::json!({"dir": dir.display().to_string()}), + ), + None => build_query("profile_tick_on"), + }, + ProfileTickAction::Off => build_query("profile_tick_off"), + ProfileTickAction::Status => build_query("profile_tick_status"), + }, + }, Commands::Keygen { .. } => unreachable!(), }; diff --git a/src/control/read_handle.rs b/src/control/read_handle.rs index 22660f1..a2f66fa 100644 --- a/src/control/read_handle.rs +++ b/src/control/read_handle.rs @@ -118,10 +118,41 @@ impl ControlReadHandle { /// Cutover queries (R1) read only `NodeContext` / `MetricsRegistry` (the state /// the read handle already bundles) plus host-OS facts (`/proc`, nftables), so /// they render entirely in the control task without touching `Node`. +/// +/// **It now also carries mutating commands**, namely the `profile_tick_*` +/// family under the `profiling` feature. They are served here rather than on +/// the rx_loop deliberately: all of their state is process statics, they need +/// no `&mut Node`, and routing them through the loop would make the toggle +/// queue behind the very behavior it exists to measure. pub(crate) fn snapshot_dispatch(request: &Request, handle: &ControlReadHandle) -> Option { use crate::control::queries; match request.command.as_str() { + // Tick-body profiler toggle. Present only in a `--features profiling` + // build; otherwise these fall through to the rx_loop dispatch, which + // reports them as unknown commands. + #[cfg(feature = "profiling")] + "profile_tick_on" => { + let dir = request + .params + .as_ref() + .and_then(|p| p.get("dir")) + .and_then(|v| v.as_str()); + let context = handle.context(); + let npub = context.identity.npub(); + let period = context.config.node.tick_interval_secs; + Some(match crate::instr::capture::start(dir, &npub, period) { + Ok(value) => Response::ok(value), + Err(e) => Response::error(e), + }) + } + #[cfg(feature = "profiling")] + "profile_tick_off" => Some(match crate::instr::capture::stop() { + Ok(value) => Response::ok(value), + Err(e) => Response::error(e), + }), + #[cfg(feature = "profiling")] + "profile_tick_status" => Some(Response::ok(crate::instr::capture::status())), "show_listening_sockets" => Some(Response::ok( queries::show_listening_sockets_from_handle(handle), )), diff --git a/src/instr/capture.rs b/src/instr/capture.rs new file mode 100644 index 0000000..b07d346 --- /dev/null +++ b/src/instr/capture.rs @@ -0,0 +1,418 @@ +//! Capture lifecycle: the arm/disarm state machine, the sink file, and the +//! `fipsctl`-facing operations. +//! +//! The toggle — not the writer — creates and opens the sink and publishes its +//! path, so an unwritable directory fails the `on` command loudly instead of +//! being discovered later by a background thread with nobody to report to. +//! +//! Capture state is a single atomic state machine (`Idle`, `Running`, +//! `StoppedByCap`) transitioned by `compare_exchange`. Every accepted control +//! connection is served by its own spawned task, so two simultaneous `on` +//! requests are genuinely concurrent and must not both create a writer. + +use std::fs::File; +use std::io::Write; +use std::path::PathBuf; +use std::sync::Mutex; +use std::sync::atomic::{AtomicBool, AtomicU8, AtomicU64, Ordering}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use super::recorder; +use super::writer; + +/// Default sink directory. Overridable per capture with `--dir`. +pub(crate) const DEFAULT_DIR: &str = "/var/log/fips"; + +/// Writer flush interval. +pub(crate) const INTERVAL: Duration = Duration::from_secs(10); + +/// Size at which a capture stops itself. Reaching it stops the capture rather +/// than rotating: the point of a capture is a bounded, self-describing window. +pub(crate) const BYTE_CAP: u64 = 32 * 1024 * 1024; + +pub(crate) const IDLE: u8 = 0; +pub(crate) const RUNNING: u8 = 1; +pub(crate) const STOPPED_BY_CAP: u8 = 2; +/// The writer could not write and stopped itself. Distinct from a cap stop: +/// a capture that died on a full disk produced a truncated window, and calling +/// that "stopped_by_cap" tells the operator it ran to its limit when it did +/// not. The trailer line explaining it goes to the same failing file, so the +/// state is the only signal that survives. +pub(crate) const STOPPED_BY_ERROR: u8 = 3; + +static STATE: AtomicU8 = AtomicU8::new(IDLE); +static GATE: AtomicBool = AtomicBool::new(false); +static BYTES: AtomicU64 = AtomicU64::new(0); +static ACTIVE_PATH: Mutex> = Mutex::new(None); +static WRITER: Mutex> = Mutex::new(None); + +/// The per-tick gate. One relaxed load per tick when the feature is compiled in +/// and no capture is running. +#[inline] +pub(crate) fn gate() -> bool { + GATE.load(Ordering::Relaxed) +} + +pub(crate) fn bytes_written() -> u64 { + BYTES.load(Ordering::Relaxed) +} + +pub(crate) fn add_bytes(n: u64) -> u64 { + BYTES.fetch_add(n, Ordering::Relaxed) + n +} + +fn active_path() -> Option { + ACTIVE_PATH + .lock() + .unwrap_or_else(|e| e.into_inner()) + .clone() +} + +fn path_display() -> String { + active_path() + .map(|p| p.display().to_string()) + .unwrap_or_else(|| "".to_string()) +} + +fn state_name(state: u8) -> &'static str { + match state { + RUNNING => "running", + STOPPED_BY_CAP => "stopped_by_cap", + STOPPED_BY_ERROR => "stopped_by_error", + _ => "idle", + } +} + +/// Called by the writer when it stops itself. `terminal` is `STOPPED_BY_CAP` +/// or `STOPPED_BY_ERROR`. Returns true if this call is the one that stopped it. +pub(crate) fn mark_stopped(terminal: u8) -> bool { + debug_assert!(terminal == STOPPED_BY_CAP || terminal == STOPPED_BY_ERROR); + GATE.store(false, Ordering::Relaxed); + STATE + .compare_exchange(RUNNING, terminal, Ordering::AcqRel, Ordering::Acquire) + .is_ok() +} + +/// Join the writer thread, if one exists. Never called while holding another +/// lock the writer might want. +fn reap() { + let handle = WRITER.lock().unwrap_or_else(|e| e.into_inner()).take(); + if let Some(handle) = handle { + handle.stop_and_join(); + } +} + +/// Arm a capture. +/// +/// Opens the sink first and only then starts the writer, so a bad `--dir` is +/// reported to the caller rather than logged into the void. +pub(crate) fn start( + dir: Option<&str>, + node_npub: &str, + tick_period_secs: u64, +) -> Result { + claim()?; + + match open_sink(dir, node_npub, tick_period_secs) { + Ok((file, path, header_len)) => { + recorder::reset(); + BYTES.store(header_len, Ordering::Relaxed); + match writer::spawn(file) { + Ok(handle) => { + *WRITER.lock().unwrap_or_else(|e| e.into_inner()) = Some(handle); + *ACTIVE_PATH.lock().unwrap_or_else(|e| e.into_inner()) = Some(path.clone()); + GATE.store(true, Ordering::Release); + Ok(serde_json::json!({ + "state": "running", + "path": path.display().to_string(), + "interval_secs": INTERVAL.as_secs(), + "byte_cap": BYTE_CAP, + })) + } + Err(e) => { + let _ = std::fs::remove_file(&path); + BYTES.store(0, Ordering::Relaxed); + STATE.store(IDLE, Ordering::Release); + Err(format!("cannot start profile writer thread: {e}")) + } + } + } + Err(e) => { + BYTES.store(0, Ordering::Relaxed); + STATE.store(IDLE, Ordering::Release); + Err(e) + } + } +} + +/// Take the capture slot, reaping a cap-stopped predecessor if that is what is +/// in the way. +fn claim() -> Result<(), String> { + match STATE.compare_exchange(IDLE, RUNNING, Ordering::AcqRel, Ordering::Acquire) { + Ok(_) => Ok(()), + Err(RUNNING) => Err(format!("capture already running: {}", path_display())), + Err(stopped @ (STOPPED_BY_CAP | STOPPED_BY_ERROR)) => { + reap(); + *ACTIVE_PATH.lock().unwrap_or_else(|e| e.into_inner()) = None; + STATE + .compare_exchange(stopped, RUNNING, Ordering::AcqRel, Ordering::Acquire) + .map(|_| ()) + .map_err(|_| "capture state changed concurrently; retry".to_string()) + } + Err(_) => Err("capture in an unexpected state".to_string()), + } +} + +/// Disarm the capture. Succeeds when nothing is running, reporting so. +pub(crate) fn stop() -> Result { + let previous = STATE.load(Ordering::Acquire); + if previous == IDLE { + return Ok(serde_json::json!({"state": "idle", "stopped": false})); + } + GATE.store(false, Ordering::Release); + // The writer wakes on the stop message rather than after the interval, so + // this join returns promptly instead of parking the caller for up to one + // flush interval. + reap(); + let path = path_display(); + *ACTIVE_PATH.lock().unwrap_or_else(|e| e.into_inner()) = None; + let bytes = bytes_written(); + // Clear the counter with the slot: a later `status` while idle must not + // report the previous capture's byte total as though a capture were live. + BYTES.store(0, Ordering::Relaxed); + STATE.store(IDLE, Ordering::Release); + Ok(serde_json::json!({ + "state": "idle", + "stopped": true, + "stopped_by_cap": previous == STOPPED_BY_CAP, + "stopped_by_error": previous == STOPPED_BY_ERROR, + "path": path, + "bytes": bytes, + })) +} + +/// Report capture state. Distinguishes all four states. +pub(crate) fn status() -> serde_json::Value { + let state = STATE.load(Ordering::Acquire); + serde_json::json!({ + "state": state_name(state), + "path": active_path().map(|p| p.display().to_string()), + "bytes": bytes_written(), + "byte_cap": BYTE_CAP, + "interval_secs": INTERVAL.as_secs(), + }) +} + +/// Stop and reap at daemon teardown. Idempotent. +pub(crate) fn shutdown() { + if STATE.load(Ordering::Acquire) != IDLE { + let _ = stop(); + } +} + +/// Create the sink file and write its header block. Returns the open file, its +/// path, and the number of header bytes written. +fn open_sink( + dir: Option<&str>, + node_npub: &str, + tick_period_secs: u64, +) -> Result<(File, PathBuf, u64), String> { + let dir = PathBuf::from(dir.unwrap_or(DEFAULT_DIR)); + std::fs::create_dir_all(&dir) + .map_err(|e| format!("cannot use profile directory {}: {e}", dir.display()))?; + + let start_unix = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0); + let path = dir.join(format!("profile-{}.tsv", compact_utc(start_unix))); + + let mut file = File::create(&path) + .map_err(|e| format!("cannot create profile file {}: {e}", path.display()))?; + + let header = format!( + "# fips tick profile\n\ + # node\t{node}\n\ + # build\t{build}\n\ + # platform\t{platform}\n\ + # tick_period_secs\t{period}\n\ + # interval_secs\t{interval}\n\ + # byte_cap\t{cap}\n\ + # start_utc\t{start_utc}\n\ + # start_unix\t{start_unix}\n\ + # NOTE\tstep durations are WALL CLOCK across await points, not CPU time:\n\ + # NOTE\ta step that awaits I/O accrues the wait, and other tasks may run\n\ + # NOTE\tinside that span. That is the intended measure for head-of-line\n\ + # NOTE\tdelay; do not read a large step as CPU cost.\n\ + # NOTE\tarm_starvation is measured directly as (entry time - the deadline\n\ + # NOTE\tthe interval scheduled the tick for). It is NOT derived from\n\ + # NOTE\ttick_entry_gap, which carries no starvation signal by itself:\n\ + # NOTE\tunder a steady delay every gap is exactly one tick period.\n\ + ts_unix\tkind\tdomain\tname\tcount\tmax\ttotal\tunit\n", + node = node_npub, + build = crate::version::short_version(), + platform = std::env::consts::OS, + period = tick_period_secs, + interval = INTERVAL.as_secs(), + cap = BYTE_CAP, + start_utc = iso_utc(start_unix), + start_unix = start_unix, + ); + file.write_all(header.as_bytes()) + .map_err(|e| format!("cannot write profile header to {}: {e}", path.display()))?; + + Ok((file, path, header.len() as u64)) +} + +/// Break a Unix timestamp into UTC `(year, month, day, hour, minute, second)`. +/// +/// Hinnant's `civil_from_days`, era-based. No date crate is in the dependency +/// set and one filename stamp does not justify adding one. +fn utc_parts(unix: u64) -> (i64, u32, u32, u32, u32, u32) { + let days = (unix / 86_400) as i64; + let secs = unix % 86_400; + let z = days + 719_468; + let era = z.div_euclid(146_097); + let doe = z.rem_euclid(146_097); + let yoe = (doe - doe / 1_460 + doe / 36_524 - doe / 146_096) / 365; + let y = yoe + era * 400; + let doy = doe - (365 * yoe + yoe / 4 - yoe / 100); + let mp = (5 * doy + 2) / 153; + let d = (doy - (153 * mp + 2) / 5 + 1) as u32; + let m = (if mp < 10 { mp + 3 } else { mp - 9 }) as u32; + let y = if m <= 2 { y + 1 } else { y }; + ( + y, + m, + d, + (secs / 3_600) as u32, + ((secs % 3_600) / 60) as u32, + (secs % 60) as u32, + ) +} + +/// `20260727T191500Z` — filename-safe. +fn compact_utc(unix: u64) -> String { + let (y, mo, d, h, mi, s) = utc_parts(unix); + format!("{y:04}{mo:02}{d:02}T{h:02}{mi:02}{s:02}Z") +} + +/// `2026-07-27T19:15:00Z` — for the header block. +fn iso_utc(unix: u64) -> String { + let (y, mo, d, h, mi, s) = utc_parts(unix); + format!("{y:04}-{mo:02}-{d:02}T{h:02}:{mi:02}:{s:02}Z") +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn utc_parts_matches_known_instants() { + assert_eq!(utc_parts(0), (1970, 1, 1, 0, 0, 0)); + assert_eq!(utc_parts(946_684_800), (2000, 1, 1, 0, 0, 0)); + // 2026-07-27T19:15:00Z + assert_eq!(utc_parts(1_785_179_700), (2026, 7, 27, 19, 15, 0)); + // Leap day. + assert_eq!(utc_parts(1_709_164_800), (2024, 2, 29, 0, 0, 0)); + } + + #[test] + fn stamps_render_expected_shapes() { + assert_eq!(compact_utc(1_785_179_700), "20260727T191500Z"); + assert_eq!(iso_utc(1_785_179_700), "2026-07-27T19:15:00Z"); + } + + // The lock these tests take is shared with the recorder tests, which + // mutate the same statics. See `crate::instr::test_serial`. + + #[test] + fn capture_round_trip_writes_header_and_rows() { + let _guard = crate::instr::test_serial(); + let dir = tempfile::tempdir().expect("tempdir"); + let dir_str = dir.path().to_str().unwrap().to_string(); + + let started = start(Some(&dir_str), "npub1test", 1).expect("start"); + assert_eq!(started["state"], "running"); + assert!(gate(), "gate must be armed while running"); + let path = PathBuf::from(started["path"].as_str().unwrap()); + + // A second `on` is refused while one is running, and names the file. + let refused = start(Some(&dir_str), "npub1test", 1).unwrap_err(); + assert!(refused.contains(&path.display().to_string()), "{refused}"); + + // Feed one observation so the drained rows are not all zero. + recorder::record( + recorder::Domain::Tick, + recorder::Step::WholeTick, + Duration::from_millis(7), + ); + + // Stopping wakes the writer immediately; it drains once more and joins. + let stopped = stop().expect("stop"); + assert_eq!(stopped["stopped"], true); + assert_eq!(stopped["stopped_by_cap"], false); + assert!(!gate(), "gate must be clear after stop"); + + let text = std::fs::read_to_string(&path).expect("read capture"); + assert!(text.starts_with("# fips tick profile\n"), "{text}"); + assert!(text.contains("# node\tnpub1test\n"), "{text}"); + assert!( + text.contains("ts_unix\tkind\tdomain\tname\tcount\tmax\ttotal\tunit\n"), + "{text}" + ); + // The final drain emitted one row per emitted step, plus the gauges. + let rows: Vec<&str> = text + .lines() + .filter(|l| l.starts_with(|c: char| c.is_ascii_digit())) + .collect(); + let expected_steps = recorder::STEPS.iter().filter(|s| s.emitted()).count(); + assert_eq!(rows.len(), expected_steps + recorder::N_GAUGES); + // The 7 ms observation above is in the whole-tick row, converted to + // microseconds. Bounds rather than equality: the gate is process-wide, + // so a node under test elsewhere in this binary may have ticked into + // the same capture window. + let whole_tick = rows + .iter() + .find(|r| r.contains("\tstep\ttick\twhole_tick\t")) + .expect("whole_tick row"); + let fields: Vec<&str> = whole_tick.split('\t').collect(); + assert_eq!(fields.last(), Some(&"us"), "{whole_tick}"); + assert!( + fields[4].parse::().unwrap() >= 1, + "count: {whole_tick}" + ); + assert!( + fields[5].parse::().unwrap() >= 7_000, + "max: {whole_tick}" + ); + assert!( + rows.iter() + .any(|r| r.contains("\tgauge\ttick\tarm_starvation\t")), + "{text}" + ); + + // A stop with nothing running is not an error. + let again = stop().expect("second stop"); + assert_eq!(again["stopped"], false); + } + + #[test] + fn start_fails_loudly_on_an_unwritable_directory() { + let _guard = crate::instr::test_serial(); + let err = start(Some("/proc/fips-profile-should-not-exist"), "npub1test", 1) + .expect_err("must fail"); + assert!(err.contains("profile directory"), "{err}"); + // The failed attempt must leave the slot free for the next try. + assert_eq!(STATE.load(Ordering::Acquire), IDLE); + assert!(!gate()); + } + + #[test] + fn status_reports_the_bounds_it_is_enforcing() { + let _guard = crate::instr::test_serial(); + let value = status(); + assert_eq!(value["byte_cap"], BYTE_CAP); + assert_eq!(value["interval_secs"], INTERVAL.as_secs()); + } +} diff --git a/src/instr/mod.rs b/src/instr/mod.rs new file mode 100644 index 0000000..7b7ab36 --- /dev/null +++ b/src/instr/mod.rs @@ -0,0 +1,171 @@ +//! Tick-body instrumentation. +//! +//! A purpose-built, feature-gated profiler for the rx-loop tick arm. It exists +//! to answer one question with field data: which subsystem step dominates the +//! tick body, and how long does the tick arm wait behind the other `select!` +//! arms before it runs at all. +//! +//! # Shape +//! +//! - Everything that costs anything at runtime is behind the `profiling` Cargo +//! feature, which is **off by default**. The default build's neutrality is a +//! property of the generated code, not of a runtime check. +//! - The instrumentation macro is defined twice, once per feature state. The +//! feature-off definition is a pure pass-through: it expands to the measured +//! expression and nothing else, so no timing code exists in a default build. +//! - The always-present surface — [`gate`], [`tick_entry`], [`tick_gauges`], +//! [`shutdown`] — exists in both feature states because the call sites in +//! `rx_loop.rs` and the lifecycle teardown must compile either way. Their +//! feature-off forms are empty (and [`gate`] is a `const fn` returning +//! `false`), so they cost nothing. +//! - The module is named `instr` rather than `profiling` so that it sorts +//! before `node` in `lib.rs`'s alphabetical module list: a `#[macro_use]` +//! module must be declared before the modules that use its macros. +//! +//! # Data model +//! +//! Domain above step: [`Domain`] carries exactly one variant today +//! (`Domain::Tick`). The primitive, the recorder, the writer and the `fipsctl` +//! surface all take a domain, so adding a data-path domain later is additive. +//! No second domain is declared until something records into it. +//! +//! Per (domain, step) the recorder keeps an exact count, max and total in fixed +//! static `AtomicU64` arrays — no histogram, no accumulation, a fixed footprint +//! regardless of run length. Gauges (ticks per interval, peer count, and the +//! arm-starvation figures) live in a parallel array and are emitted with an +//! explicit row kind so a gauge value never lands under a duration column. + +#[cfg(feature = "profiling")] +pub(crate) mod capture; +#[cfg(feature = "profiling")] +mod recorder; +#[cfg(feature = "profiling")] +mod writer; + +#[cfg(feature = "profiling")] +pub(crate) use recorder::{Domain, Step, now, record}; + +// --------------------------------------------------------------------------- +// The macro pair. +// +// Every path in the body is `$crate::`-qualified. `macro_rules!` bodies are not +// path-hygienic: an unqualified `Instant::now()` or `record(..)` would resolve +// at the *call site* (`rx_loop.rs`), where neither name is in scope. Importing +// them there is worse still, because the imports would be unused in the +// feature-off build and red it under `-D warnings`. +// +// `$e` is evaluated exactly once in both forms, which is what makes nesting the +// whole-tick span around the per-step spans safe. +// --------------------------------------------------------------------------- + +/// Time `$e` as one step of `$domain`, when `$on` is true. +/// +/// `$on` is the per-tick gate hoist: the enable flag is read once at the top of +/// the tick arm into a local, and that local is passed explicitly to every +/// invocation, because macro hygiene makes a call-site local invisible inside +/// the macro body. +#[cfg(feature = "profiling")] +macro_rules! instr_step { + ($on:expr, $domain:expr, $step:expr, $e:expr) => {{ + let t0 = if $on { + Some($crate::instr::now()) + } else { + None + }; + let r = $e; + if let Some(t) = t0 { + $crate::instr::record($domain, $step, t.elapsed()); + } + r + }}; +} + +/// Feature-off form: a pure pass-through. The expansion contains no clock read, +/// no counter update and no reference to the recorder — only the measured +/// expression, plus a discard of the gate local so it is not unused. +#[cfg(not(feature = "profiling"))] +macro_rules! instr_step { + ($on:expr, $domain:expr, $step:expr, $e:expr) => {{ + let _ = &$on; + $e + }}; +} + +// --------------------------------------------------------------------------- +// Always-present surface. +// --------------------------------------------------------------------------- + +/// Whether a capture is armed. Read **once per tick** into a local that is then +/// passed to each `instr_step!` invocation, so the feature-on-but-idle cost of +/// the whole tick arm is a single relaxed load. +#[cfg(feature = "profiling")] +#[inline] +pub(crate) fn gate() -> bool { + capture::gate() +} + +/// Feature-off gate: a `const fn` returning `false`, so the whole tick arm +/// folds to the uninstrumented sequence at compile time. +#[cfg(not(feature = "profiling"))] +#[inline] +pub(crate) const fn gate() -> bool { + false +} + +/// Record how late this tick-arm entry is against its scheduled deadline. +/// +/// The arm is polled **last** under `biased;`, so its lateness is the time it +/// spent waiting behind the packet, TUN and control arms. `tokio::time:: +/// interval::tick` returns the deadline it was scheduled for, so this is a +/// direct subtraction rather than a model. Two earlier designs derived it from +/// the inter-entry gap instead and both under-reported: one by the previous +/// body, the other by reporting only the first difference of the delay, so a +/// sustained stall read as zero. The inter-entry gap is still recorded as its +/// own gauge, but it carries no starvation signal on its own. +#[cfg(feature = "profiling")] +#[inline] +pub(crate) fn tick_entry(on: bool, deadline: std::time::Instant, now: std::time::Instant) { + recorder::tick_entry(on, deadline, now); +} + +#[cfg(not(feature = "profiling"))] +#[inline] +pub(crate) fn tick_entry(_on: bool, _deadline: std::time::Instant, _now: std::time::Instant) {} + +/// Sample the per-tick gauges taken from node state. +#[cfg(feature = "profiling")] +#[inline] +pub(crate) fn tick_gauges(on: bool, peers: u64) { + recorder::tick_gauges(on, peers); +} + +#[cfg(not(feature = "profiling"))] +#[inline] +pub(crate) fn tick_gauges(_on: bool, _peers: u64) {} + +/// Stop and reap any running capture at daemon teardown. Idempotent. +#[cfg(feature = "profiling")] +pub(crate) fn shutdown() { + capture::shutdown(); +} + +#[cfg(not(feature = "profiling"))] +pub(crate) fn shutdown() {} + +/// One serialization lock for every test in this module tree. +/// +/// The recorder counters and the capture state machine are the *same* process +/// statics: `capture::start` calls `recorder::reset`, and `capture::stop` +/// drains every slot. Two suites with their own locks therefore do not +/// serialize against each other, and the feature-on stage runs tests as +/// threads in one process, so a capture round-trip can zero the counters a +/// recorder test is mid-way through asserting on. One lock for both. +#[cfg(all(test, feature = "profiling"))] +pub(crate) static TEST_SERIAL: std::sync::Mutex<()> = std::sync::Mutex::new(()); + +/// Take the shared test lock, recovering from a poisoned mutex so one failing +/// test does not cascade into every other one. +#[cfg(all(test, feature = "profiling"))] +pub(crate) fn test_serial() -> std::sync::MutexGuard<'static, ()> { + TEST_SERIAL.lock().unwrap_or_else(|e| e.into_inner()) +} diff --git a/src/instr/recorder.rs b/src/instr/recorder.rs new file mode 100644 index 0000000..10a742a --- /dev/null +++ b/src/instr/recorder.rs @@ -0,0 +1,487 @@ +//! Fixed-footprint recorder: exact count / max / total per (domain, step). +//! +//! All state is process statics, not `Node` state, because the `fipsctl` +//! handler that arms and disarms a capture runs in the control accept task and +//! has no `&Node` — that is the whole point of serving it off-loop, so it +//! cannot queue behind the behavior it is measuring. +//! +//! The writer thread is the only reader. It takes each interval's figures with +//! `swap(0)`, so there are no "previous value" arrays to carry and the counters +//! are per-interval by construction. + +use std::sync::LazyLock; +use std::sync::atomic::{AtomicU64, Ordering::Relaxed}; +use std::time::{Duration, Instant}; + +/// Measurement domain. Structural only: one variant today. +/// +/// A data-path domain is deliberately **not** declared until something records +/// into it. What generalizes here is the enum, the counter table and the +/// writer; the per-tick gate hoist does not, so a data-path domain will need +/// its own gate strategy. +#[derive(Copy, Clone, Debug, PartialEq, Eq)] +#[repr(usize)] +pub(crate) enum Domain { + Tick = 0, +} + +pub(crate) const N_DOMAINS: usize = 1; +pub(crate) const DOMAINS: [Domain; N_DOMAINS] = [Domain::Tick]; + +impl Domain { + pub(crate) const fn name(self) -> &'static str { + match self { + Domain::Tick => "tick", + } + } +} + +/// One measured step of the rx-loop tick arm, in call order, plus the +/// whole-body span. +/// +/// `as usize` indexes the counter arrays, so the discriminants are dense and +/// `WholeTick` is last (it defines `N_STEPS`). Variants are declared +/// unconditionally — see [`Step::emitted`] for how the two platform- and +/// profile-conditional steps are kept out of the emitted table. +#[derive(Copy, Clone, Debug, PartialEq, Eq)] +#[repr(usize)] +pub(crate) enum Step { + CheckTimeouts = 0, + ReloadPeerAcl, + ReloadHostMap, + PollPendingConnects, + PollNostrRendezvous, + PollLanRendezvous, + DrivePeerTimers, + ResendPendingRekeys, + /// `next`-only: the FMP rekey msg3 resend driver has no master-line + /// counterpart, so this variant exists on this line alone. + ResendPendingFmpRekeyMsg3, + ResendPendingSessionHandshakes, + ResendPendingSessionMsg3, + PurgeIdleSessions, + ProcessPendingRetries, + CheckTreeState, + CheckBloomState, + ComputeMeshSize, + RecordStatsHistory, + CheckMmpReports, + CheckSessionMmpReports, + CheckLinkHeartbeats, + CheckRekey, + CheckSessionRekey, + CheckPendingLookups, + PollTransportDiscovery, + SampleTransportCongestion, + ActivateConnectedUdpSessions, + DebugAssertPeerMapsCoherent, + /// The whole tick-arm body, from before `check_timeouts` to after the last + /// step. Composes safely with the per-step spans because the macro + /// evaluates its measured expression exactly once. + WholeTick, +} + +pub(crate) const N_STEPS: usize = Step::WholeTick as usize + 1; + +/// Every step, in emission order. Index `i` of this table is `STEPS[i] as +/// usize`; `steps_table_is_dense` asserts it. +pub(crate) const STEPS: [Step; N_STEPS] = [ + Step::CheckTimeouts, + Step::ReloadPeerAcl, + Step::ReloadHostMap, + Step::PollPendingConnects, + Step::PollNostrRendezvous, + Step::PollLanRendezvous, + Step::DrivePeerTimers, + Step::ResendPendingRekeys, + Step::ResendPendingFmpRekeyMsg3, + Step::ResendPendingSessionHandshakes, + Step::ResendPendingSessionMsg3, + Step::PurgeIdleSessions, + Step::ProcessPendingRetries, + Step::CheckTreeState, + Step::CheckBloomState, + Step::ComputeMeshSize, + Step::RecordStatsHistory, + Step::CheckMmpReports, + Step::CheckSessionMmpReports, + Step::CheckLinkHeartbeats, + Step::CheckRekey, + Step::CheckSessionRekey, + Step::CheckPendingLookups, + Step::PollTransportDiscovery, + Step::SampleTransportCongestion, + Step::ActivateConnectedUdpSessions, + Step::DebugAssertPeerMapsCoherent, + Step::WholeTick, +]; + +impl Step { + pub(crate) const fn name(self) -> &'static str { + match self { + Step::CheckTimeouts => "check_timeouts", + Step::ReloadPeerAcl => "reload_peer_acl", + Step::ReloadHostMap => "reload_host_map", + Step::PollPendingConnects => "poll_pending_connects", + Step::PollNostrRendezvous => "poll_nostr_rendezvous", + Step::PollLanRendezvous => "poll_lan_rendezvous", + Step::DrivePeerTimers => "drive_peer_timers", + Step::ResendPendingRekeys => "resend_pending_rekeys", + Step::ResendPendingFmpRekeyMsg3 => "resend_pending_fmp_rekey_msg3", + Step::ResendPendingSessionHandshakes => "resend_pending_session_handshakes", + Step::ResendPendingSessionMsg3 => "resend_pending_session_msg3", + Step::PurgeIdleSessions => "purge_idle_sessions", + Step::ProcessPendingRetries => "process_pending_retries", + Step::CheckTreeState => "check_tree_state", + Step::CheckBloomState => "check_bloom_state", + Step::ComputeMeshSize => "compute_mesh_size", + Step::RecordStatsHistory => "record_stats_history", + Step::CheckMmpReports => "check_mmp_reports", + Step::CheckSessionMmpReports => "check_session_mmp_reports", + Step::CheckLinkHeartbeats => "check_link_heartbeats", + Step::CheckRekey => "check_rekey", + Step::CheckSessionRekey => "check_session_rekey", + Step::CheckPendingLookups => "check_pending_lookups", + Step::PollTransportDiscovery => "poll_transport_discovery", + Step::SampleTransportCongestion => "sample_transport_congestion", + Step::ActivateConnectedUdpSessions => "activate_connected_udp_sessions", + Step::DebugAssertPeerMapsCoherent => "debug_assert_peer_maps_coherent", + Step::WholeTick => "whole_tick", + } + } + + /// Whether this step gets a row in this build. + /// + /// Two steps are conditionally compiled at their call sites. Emitting a row + /// for them in a build where the call site does not exist would publish a + /// count that is structurally zero forever, which reads as "this step never + /// runs" rather than "this step is not in this build". The predicates below + /// are the same `cfg` expressions that gate the call sites in + /// `node::dataplane::rx_loop`; keep them in step. + pub(crate) const fn emitted(self) -> bool { + match self { + Step::ActivateConnectedUdpSessions => { + cfg!(any(target_os = "linux", target_os = "macos")) + } + Step::DebugAssertPeerMapsCoherent => cfg!(debug_assertions), + _ => true, + } + } +} + +/// A scalar sampled once per tick, as opposed to a duration. +/// +/// Gauges carry their own row kind and their own unit in the output so a gauge +/// value can never be read as a duration. +#[derive(Copy, Clone, Debug, PartialEq, Eq)] +#[repr(usize)] +pub(crate) enum Gauge { + Ticks = 0, + Peers, + TickGap, + ArmStarvation, +} + +pub(crate) const N_GAUGES: usize = Gauge::ArmStarvation as usize + 1; + +pub(crate) const GAUGES: [Gauge; N_GAUGES] = [ + Gauge::Ticks, + Gauge::Peers, + Gauge::TickGap, + Gauge::ArmStarvation, +]; + +impl Gauge { + pub(crate) const fn name(self) -> &'static str { + match self { + Gauge::Ticks => "ticks", + Gauge::Peers => "peers", + Gauge::TickGap => "tick_entry_gap", + Gauge::ArmStarvation => "arm_starvation", + } + } + + /// Unit of the `max` and `total` columns for this gauge. + pub(crate) const fn unit(self) -> &'static str { + match self { + Gauge::Ticks => "ticks", + Gauge::Peers => "peers", + Gauge::TickGap | Gauge::ArmStarvation => "us", + } + } + + /// Whether the gauge's stored values are nanosecond durations that the + /// writer converts to microseconds. + pub(crate) const fn is_duration(self) -> bool { + matches!(self, Gauge::TickGap | Gauge::ArmStarvation) + } +} + +const N_SLOTS: usize = N_DOMAINS * N_STEPS; + +static COUNT: [AtomicU64; N_SLOTS] = [const { AtomicU64::new(0) }; N_SLOTS]; +static MAX_NS: [AtomicU64; N_SLOTS] = [const { AtomicU64::new(0) }; N_SLOTS]; +static TOTAL_NS: [AtomicU64; N_SLOTS] = [const { AtomicU64::new(0) }; N_SLOTS]; + +static G_COUNT: [AtomicU64; N_GAUGES] = [const { AtomicU64::new(0) }; N_GAUGES]; +static G_MAX: [AtomicU64; N_GAUGES] = [const { AtomicU64::new(0) }; N_GAUGES]; +static G_TOTAL: [AtomicU64; N_GAUGES] = [const { AtomicU64::new(0) }; N_GAUGES]; + +/// Monotonic baseline so tick-arm entry times fit in an atomic. Offset by one +/// on store so that zero can mean "no previous entry". +static BASE: LazyLock = LazyLock::new(Instant::now); +static PREV_ENTRY_NS: AtomicU64 = AtomicU64::new(0); + +#[inline] +const fn slot(domain: Domain, step: Step) -> usize { + (domain as usize * N_STEPS) + step as usize +} + +/// Read the clock for a step span. +#[inline] +pub(crate) fn now() -> Instant { + Instant::now() +} + +/// Record one observation of `step`. +#[inline] +pub(crate) fn record(domain: Domain, step: Step, elapsed: Duration) { + let ns = elapsed.as_nanos() as u64; + let idx = slot(domain, step); + COUNT[idx].fetch_add(1, Relaxed); + TOTAL_NS[idx].fetch_add(ns, Relaxed); + MAX_NS[idx].fetch_max(ns, Relaxed); +} + +#[inline] +fn record_gauge(gauge: Gauge, value: u64) { + let idx = gauge as usize; + G_COUNT[idx].fetch_add(1, Relaxed); + G_TOTAL[idx].fetch_add(value, Relaxed); + G_MAX[idx].fetch_max(value, Relaxed); +} + +/// Sample the inter-entry gap and the measured arm-starvation delay. +pub(crate) fn tick_entry(on: bool, deadline: Instant, now: Instant) { + if !on { + return; + } + // Offset by one so that a stored zero unambiguously means "no previous + // entry", even for an entry that lands on the baseline instant. + let stamp = BASE.elapsed().as_nanos() as u64 + 1; + let late = now.saturating_duration_since(deadline).as_nanos() as u64; + tick_entry_at(stamp, late); +} + +/// The clock-free half of [`tick_entry`]: both times are inputs, so the +/// arithmetic can be driven with synthetic stamps in a test. +/// +/// `late_ns` is how far past its scheduled deadline this entry was, measured +/// directly rather than derived. Two earlier designs derived it from the +/// inter-entry gap and were both wrong: subtracting the previous body +/// understated it by exactly the body, and subtracting `max(period, body)` +/// reported the *first difference* of the delay, so a sustained stall — the +/// overload regime this measurement exists to characterize — read as zero +/// forever. `tokio::time::interval::tick` hands back the deadline it was +/// scheduled for, so the delay is a subtraction with no model behind it. +fn tick_entry_at(stamp: u64, late_ns: u64) { + let prev = PREV_ENTRY_NS.swap(stamp, Relaxed); + record_gauge(Gauge::Ticks, 1); + record_gauge(Gauge::ArmStarvation, late_ns); + if prev == 0 { + // First entry of this capture: there is no previous entry to measure a + // gap against, and the idle interval before arming is not a gap. The + // lateness above does not depend on a previous entry, so it still counts. + return; + } + record_gauge(Gauge::TickGap, stamp.saturating_sub(prev)); +} + +/// Sample the gauges that come from node state. +pub(crate) fn tick_gauges(on: bool, peers: u64) { + if !on { + return; + } + record_gauge(Gauge::Peers, peers); +} + +/// Take (and clear) this interval's figures for one step: count, max ns, total +/// ns. Called only by the writer thread. +pub(crate) fn take_step(domain: Domain, step: Step) -> (u64, u64, u64) { + let idx = slot(domain, step); + ( + COUNT[idx].swap(0, Relaxed), + MAX_NS[idx].swap(0, Relaxed), + TOTAL_NS[idx].swap(0, Relaxed), + ) +} + +/// Take (and clear) this interval's figures for one gauge. +pub(crate) fn take_gauge(gauge: Gauge) -> (u64, u64, u64) { + let idx = gauge as usize; + ( + G_COUNT[idx].swap(0, Relaxed), + G_MAX[idx].swap(0, Relaxed), + G_TOTAL[idx].swap(0, Relaxed), + ) +} + +/// Zero every counter so a capture starts from a clean slate. +pub(crate) fn reset() { + for i in 0..N_SLOTS { + COUNT[i].store(0, Relaxed); + MAX_NS[i].store(0, Relaxed); + TOTAL_NS[i].store(0, Relaxed); + } + for i in 0..N_GAUGES { + G_COUNT[i].store(0, Relaxed); + G_MAX[i].store(0, Relaxed); + G_TOTAL[i].store(0, Relaxed); + } + PREV_ENTRY_NS.store(0, Relaxed); +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::MutexGuard; + + /// Shared with the capture tests: they mutate the same statics. See + /// `crate::instr::test_serial`. + fn serial() -> MutexGuard<'static, ()> { + crate::instr::test_serial() + } + + #[test] + fn steps_table_is_dense() { + for (i, step) in STEPS.iter().enumerate() { + assert_eq!(*step as usize, i, "step {} is out of order", step.name()); + } + assert_eq!(STEPS.len(), N_STEPS); + } + + #[test] + fn gauges_table_is_dense() { + for (i, gauge) in GAUGES.iter().enumerate() { + assert_eq!(*gauge as usize, i, "gauge {} is out of order", gauge.name()); + } + assert_eq!(GAUGES.len(), N_GAUGES); + } + + #[test] + fn step_names_are_unique() { + let mut names: Vec<&str> = STEPS.iter().map(|s| s.name()).collect(); + names.sort_unstable(); + let before = names.len(); + names.dedup(); + assert_eq!(before, names.len(), "duplicate step name"); + } + + #[test] + fn emitted_row_count_matches_build() { + let emitted = STEPS.iter().filter(|s| s.emitted()).count(); + // 25 unconditional subsystem steps on this line (24 shared with the + // master line, plus `resend_pending_fmp_rekey_msg3`, which exists only + // here) + the whole-tick span, plus the two conditionally-compiled + // steps where this build has them. The count is pinned deliberately: it + // is what caught the extra step when the master-line instrumentation + // was merged up, rather than letting the tables silently disagree. + let mut expected = 26; + if cfg!(any(target_os = "linux", target_os = "macos")) { + expected += 1; + } + if cfg!(debug_assertions) { + expected += 1; + } + assert_eq!(emitted, expected); + } + + #[test] + fn record_accumulates_count_max_and_total() { + let _guard = serial(); + reset(); + record(Domain::Tick, Step::CheckRekey, Duration::from_nanos(10)); + record(Domain::Tick, Step::CheckRekey, Duration::from_nanos(30)); + let (count, max, total) = take_step(Domain::Tick, Step::CheckRekey); + assert_eq!((count, max, total), (2, 30, 40)); + // Taking clears the slot. + assert_eq!(take_step(Domain::Tick, Step::CheckRekey), (0, 0, 0)); + } + + #[test] + fn starvation_is_the_measured_lateness_of_the_entry() { + let _guard = serial(); + reset(); + // The interval hands back the deadline it was scheduled for, so the + // delay is `now - deadline` and nothing is derived from the period, the + // previous entry, or the previous body. + PREV_ENTRY_NS.store(1_000_000_000, Relaxed); + tick_entry_at(1_100_000_000, 50_000_000); + assert_eq!(take_gauge(Gauge::TickGap), (1, 100_000_000, 100_000_000)); + assert_eq!( + take_gauge(Gauge::ArmStarvation), + (1, 50_000_000, 50_000_000) + ); + assert_eq!(take_gauge(Gauge::Ticks).0, 1); + reset(); + } + + #[test] + fn sustained_lateness_is_reported_on_every_tick() { + let _guard = serial(); + reset(); + // The regime the two earlier designs both hid. Three consecutive entries + // each 50 ms past their deadline, one period apart, i.e. the arm waiting + // a constant amount behind the other select arms every round. The gaps + // are all exactly one period, so any formula derived from the + // inter-entry gap reports zero here; measured lateness reports 50 ms + // three times, which is the truth. + PREV_ENTRY_NS.store(1_000_000_000, Relaxed); + tick_entry_at(1_050_000_000, 50_000_000); + tick_entry_at(1_100_000_000, 50_000_000); + tick_entry_at(1_150_000_000, 50_000_000); + let (count, max, total) = take_gauge(Gauge::ArmStarvation); + assert_eq!(count, 3); + assert_eq!(max, 50_000_000); + assert_eq!(total, 150_000_000); + // ...and the gap alone carries no signal about it: every gap is one + // period, exactly as it would be on a perfectly healthy node. + assert_eq!(take_gauge(Gauge::TickGap), (3, 50_000_000, 150_000_000)); + reset(); + } + + #[test] + fn first_entry_of_a_capture_records_no_gap_but_still_records_lateness() { + let _guard = serial(); + reset(); + tick_entry_at(500, 7_000_000); + assert_eq!(take_gauge(Gauge::Ticks).0, 1); + assert_eq!(take_gauge(Gauge::TickGap), (0, 0, 0)); + // Lateness does not depend on a previous entry, so the first tick of a + // capture still contributes one. + assert_eq!(take_gauge(Gauge::ArmStarvation), (1, 7_000_000, 7_000_000)); + reset(); + } + + #[test] + fn an_on_schedule_entry_reports_no_starvation() { + let _guard = serial(); + reset(); + PREV_ENTRY_NS.store(1_000_000_000, Relaxed); + tick_entry_at(1_050_000_000, 0); + assert_eq!(take_gauge(Gauge::ArmStarvation), (1, 0, 0)); + assert_eq!(take_gauge(Gauge::TickGap), (1, 50_000_000, 50_000_000)); + reset(); + } + + #[test] + fn gate_off_records_nothing() { + let _guard = serial(); + reset(); + let t = Instant::now(); + tick_entry(false, t, t); + tick_gauges(false, 42); + assert_eq!(take_gauge(Gauge::Ticks), (0, 0, 0)); + assert_eq!(take_gauge(Gauge::Peers), (0, 0, 0)); + } +} diff --git a/src/instr/writer.rs b/src/instr/writer.rs new file mode 100644 index 0000000..0f11527 --- /dev/null +++ b/src/instr/writer.rs @@ -0,0 +1,218 @@ +//! The capture writer: a dedicated, named OS thread with an explicit +//! lifecycle. +//! +//! There is no worker-thread lifecycle in this codebase to copy — the crypto +//! worker pools are never torn down and drop their join handles at spawn — so +//! this one is designed here. +//! +//! Two properties matter: +//! +//! - **It is an OS thread, not a spawned task.** The runtime is +//! `current_thread`, so file I/O on a task would run on the rx loop's own +//! thread and stall every `select!` arm, including the one being measured. +//! - **It waits on a channel with a timeout, not on a sleep.** `recv_timeout` +//! returns immediately when the toggle sends stop, so `off` performs a final +//! drain and joins promptly instead of parking the caller for up to a full +//! flush interval. +//! +//! The thread is created lazily when a capture starts, so a node that never +//! arms one never has the thread. + +use std::fs::File; +use std::io::Write; +use std::sync::mpsc::{self, Receiver, RecvTimeoutError, Sender}; +use std::thread::{self, JoinHandle}; +use std::time::{SystemTime, UNIX_EPOCH}; + +use super::capture::{self, BYTE_CAP, INTERVAL}; +use super::recorder::{self, DOMAINS, GAUGES, STEPS}; + +/// Owner-side handle to the writer thread. +pub(crate) struct Handle { + stop_tx: Sender<()>, + join: JoinHandle<()>, +} + +impl Handle { + /// Wake the writer, let it drain once more, and join it. + pub(crate) fn stop_and_join(self) { + // A send error means the thread already exited (cap stop); joining is + // still correct and returns at once. + let _ = self.stop_tx.send(()); + let _ = self.join.join(); + } +} + +/// Start the writer thread on an already-open sink. +pub(crate) fn spawn(file: File) -> std::io::Result { + let (stop_tx, stop_rx) = mpsc::channel(); + let join = thread::Builder::new() + .name("fips-profile".to_string()) + .spawn(move || run(file, stop_rx))?; + Ok(Handle { stop_tx, join }) +} + +/// What one flush cycle decided. Separated from [`run`] so the terminal paths +/// can be driven in a test with a failing sink: the loop below owns the waiting +/// and the state transition, this owns the decision. +#[derive(Debug, PartialEq, Eq)] +enum Cycle { + Continue, + CapReached, + WriteFailed, +} + +/// Drain one interval into the sink and decide whether the capture goes on. +fn flush_cycle(file: &mut W) -> Cycle { + if flush(file).is_err() { + // The sink is gone or full; stop rather than spinning on a broken file + // for the rest of the run. + let _ = note(file, "capture stopped: write error"); + return Cycle::WriteFailed; + } + if capture::bytes_written() >= BYTE_CAP { + let _ = note( + file, + &format!("capture stopped: byte cap {BYTE_CAP} reached"), + ); + let _ = file.flush(); + return Cycle::CapReached; + } + Cycle::Continue +} + +fn run(mut file: W, stop_rx: Receiver<()>) { + loop { + match stop_rx.recv_timeout(INTERVAL) { + // Stop requested, or the owner went away: final drain, then exit. + Ok(()) | Err(RecvTimeoutError::Disconnected) => { + let _ = flush(&mut file); + let _ = file.flush(); + return; + } + Err(RecvTimeoutError::Timeout) => match flush_cycle(&mut file) { + Cycle::Continue => {} + Cycle::WriteFailed => { + // The trailer went to the same failing file, so it is not a + // signal that survives. `stop` and a subsequent `on` both + // clear the state without surfacing it, so an operator would + // otherwise never learn the window was truncated. + tracing::warn!( + target: "fips::instr", + "profile capture stopped: write error on the sink" + ); + capture::mark_stopped(capture::STOPPED_BY_ERROR); + return; + } + Cycle::CapReached => { + capture::mark_stopped(capture::STOPPED_BY_CAP); + return; + } + }, + } + } +} + +/// Append a `#`-prefixed trailer line. +fn note(file: &mut W, text: &str) -> std::io::Result<()> { + let line = format!("# {text}\n"); + file.write_all(line.as_bytes())?; + capture::add_bytes(line.len() as u64); + Ok(()) +} + +/// Emit one interval: every step of every domain, then the gauges. +/// +/// Every emitted step gets a row every interval, including zero-count rows, so +/// "this step did not run" is visible rather than absent. The two steps whose +/// call sites are conditionally compiled are excluded in builds that do not +/// have them, so no row is structurally zero forever. +fn flush(file: &mut W) -> std::io::Result<()> { + let ts = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0); + + let mut out = String::with_capacity(4096); + for domain in DOMAINS { + for step in STEPS { + if !step.emitted() { + continue; + } + let (count, max_ns, total_ns) = recorder::take_step(domain, step); + out.push_str(&format!( + "{ts}\tstep\t{domain}\t{name}\t{count}\t{max}\t{total}\tus\n", + domain = domain.name(), + name = step.name(), + max = max_ns / 1_000, + total = total_ns / 1_000, + )); + } + } + // Gauges carry the tick domain today; the row kind and the unit column keep + // them distinguishable from the duration rows above. + for gauge in GAUGES { + let (count, mut max, mut total) = recorder::take_gauge(gauge); + if gauge.is_duration() { + max /= 1_000; + total /= 1_000; + } + out.push_str(&format!( + "{ts}\tgauge\t{domain}\t{name}\t{count}\t{max}\t{total}\t{unit}\n", + domain = recorder::Domain::Tick.name(), + name = gauge.name(), + unit = gauge.unit(), + )); + } + + file.write_all(out.as_bytes())?; + capture::add_bytes(out.len() as u64); + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + /// A sink that fails every write, so the writer's error path is driven by a + /// real `Err` rather than asserted about. + struct AlwaysFails; + + impl Write for AlwaysFails { + fn write(&mut self, _buf: &[u8]) -> std::io::Result { + Err(std::io::Error::new( + std::io::ErrorKind::StorageFull, + "no space left on device", + )) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + + /// A failing sink ends the capture as a write error, which `run` turns into + /// `STOPPED_BY_ERROR` — not into a byte-cap stop. The distinction is what + /// tells an operator that a window is truncated rather than complete. + /// + /// **Coverage note.** This drives the decision, not the filesystem + /// condition. The mesh rehearsal cannot produce one: removing the file + /// leaves the writer's descriptor valid and writes keep succeeding into the + /// unlinked inode, and mounting a tiny filesystem inside the test container + /// is refused. So "a real ENOSPC reaches this branch" stays unexercised; + /// what is covered is that an `Err` from the sink produces the error + /// outcome and not the cap outcome. + #[test] + fn a_failing_sink_ends_the_cycle_as_an_error_not_a_cap() { + let _guard = crate::instr::test_serial(); + assert_eq!(flush_cycle(&mut AlwaysFails), Cycle::WriteFailed); + } + + /// The healthy path must not be reported as either terminal state, or the + /// test above would pass for a writer that always stops. + #[test] + fn a_working_sink_continues() { + let _guard = crate::instr::test_serial(); + let mut sink: Vec = Vec::new(); + assert_eq!(flush_cycle(&mut sink), Cycle::Continue); + } +} diff --git a/src/lib.rs b/src/lib.rs index 4624828..1ca5562 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -15,6 +15,11 @@ pub mod control; #[cfg(target_os = "linux")] pub mod gateway; pub mod identity; +// Declared before `node` (and named to sort there) because it carries +// `#[macro_use]`: the tick instrumentation macro must be in scope for the +// modules that follow. +#[macro_use] +pub(crate) mod instr; pub mod mdns; pub mod node; pub mod noise; diff --git a/src/node/dataplane/rx_loop.rs b/src/node/dataplane/rx_loop.rs index e9fb5de..109d8c4 100644 --- a/src/node/dataplane/rx_loop.rs +++ b/src/node/dataplane/rx_loop.rs @@ -109,8 +109,8 @@ impl Node { } }; - let mut tick = - tokio::time::interval(Duration::from_secs(self.config().node.tick_interval_secs)); + let tick_period = Duration::from_secs(self.config().node.tick_interval_secs); + let mut tick = tokio::time::interval(tick_period); // Set up control socket channel let (control_tx, mut control_rx) = @@ -321,47 +321,95 @@ impl Node { ).await; let _ = response_tx.send(response); } - _ = tick.tick() => { - self.check_timeouts(); - let now_ms = Self::now_ms(); - self.reload_peer_acl().await; - // The host map hot-reloads on the same tick as the ACL. It - // is polled separately from `reload_peer_acl` because the - // ACL's embedded alias reloader and this snapshot are - // distinct resources; the `path_mtu_lookup` cache and the - // `nostr_rendezvous` subsystem are deliberately excluded - // from `Reloadable` since neither reloads from a backing - // file (see `node::reloadable`). - self.reload_host_map().await; - self.poll_pending_connects().await; - self.poll_nostr_rendezvous().await; - self.poll_lan_rendezvous().await; - self.drive_peer_timers(now_ms).await; - self.resend_pending_rekeys(now_ms).await; - self.resend_pending_fmp_rekey_msg3(now_ms).await; - self.resend_pending_session_handshakes(now_ms).await; - self.resend_pending_session_msg3(now_ms).await; - self.purge_idle_sessions(now_ms); - self.process_pending_retries(now_ms).await; - self.check_tree_state().await; - self.check_bloom_state().await; - self.compute_mesh_size(); - self.record_stats_history(); - self.check_mmp_reports().await; - self.check_session_mmp_reports().await; - self.check_link_heartbeats().await; - self.check_rekey().await; - self.check_session_rekey().await; - self.check_pending_lookups(now_ms).await; - self.poll_transport_discovery().await; - self.sample_transport_congestion(); - #[cfg(any(target_os = "linux", target_os = "macos"))] - self.activate_connected_udp_sessions().await; - // Debug-build sweep of the peer-lifecycle map invariant - // (leaked machines / machine-less legs); two map scans, - // compiled out of release builds. - #[cfg(debug_assertions)] - self.debug_assert_peer_maps_coherent(); + deadline = tick.tick() => { + // Tick-body instrumentation. The gate is read ONCE per tick + // into `instr_on`, which is then passed explicitly to every + // `instr_step!` invocation — macro hygiene makes a call-site + // local invisible inside the macro body. With the + // `profiling` feature off, `gate()` is a `const fn` + // returning false and the macro is a pure pass-through, so + // the whole arm compiles to the uninstrumented sequence. + // + // `tick_entry` records how late this entry is against the + // deadline the interval scheduled it for. That is the + // measurement this instrumentation exists for: the arm is + // polled LAST under `biased;`, so the + // lateness IS the time it spent waiting behind the packet, + // TUN and control arms. `tick()` hands back its scheduled + // deadline, so this is a subtraction rather than a model. + // The whole-tick span below measures the body alone. + let instr_on = crate::instr::gate(); + crate::instr::tick_entry(instr_on, deadline.into_std(), std::time::Instant::now()); + instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::WholeTick, { + instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::CheckTimeouts, + self.check_timeouts()); + let now_ms = Self::now_ms(); + instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::ReloadPeerAcl, + self.reload_peer_acl().await); + // The host map hot-reloads on the same tick as the ACL. It + // is polled separately from `reload_peer_acl` because the + // ACL's embedded alias reloader and this snapshot are + // distinct resources; the `path_mtu_lookup` cache and the + // `nostr_rendezvous` subsystem are deliberately excluded + // from `Reloadable` since neither reloads from a backing + // file (see `node::reloadable`). + instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::ReloadHostMap, + self.reload_host_map().await); + instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::PollPendingConnects, + self.poll_pending_connects().await); + instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::PollNostrRendezvous, + self.poll_nostr_rendezvous().await); + instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::PollLanRendezvous, + self.poll_lan_rendezvous().await); + instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::DrivePeerTimers, + self.drive_peer_timers(now_ms).await); + instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::ResendPendingRekeys, + self.resend_pending_rekeys(now_ms).await); + instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::ResendPendingFmpRekeyMsg3, + self.resend_pending_fmp_rekey_msg3(now_ms).await); + instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::ResendPendingSessionHandshakes, + self.resend_pending_session_handshakes(now_ms).await); + instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::ResendPendingSessionMsg3, + self.resend_pending_session_msg3(now_ms).await); + instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::PurgeIdleSessions, + self.purge_idle_sessions(now_ms)); + instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::ProcessPendingRetries, + self.process_pending_retries(now_ms).await); + instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::CheckTreeState, + self.check_tree_state().await); + instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::CheckBloomState, + self.check_bloom_state().await); + instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::ComputeMeshSize, + self.compute_mesh_size()); + instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::RecordStatsHistory, + self.record_stats_history()); + instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::CheckMmpReports, + self.check_mmp_reports().await); + instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::CheckSessionMmpReports, + self.check_session_mmp_reports().await); + instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::CheckLinkHeartbeats, + self.check_link_heartbeats().await); + instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::CheckRekey, + self.check_rekey().await); + instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::CheckSessionRekey, + self.check_session_rekey().await); + instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::CheckPendingLookups, + self.check_pending_lookups(now_ms).await); + instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::PollTransportDiscovery, + self.poll_transport_discovery().await); + instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::SampleTransportCongestion, + self.sample_transport_congestion()); + #[cfg(any(target_os = "linux", target_os = "macos"))] + instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::ActivateConnectedUdpSessions, + self.activate_connected_udp_sessions().await); + // Debug-build sweep of the peer-lifecycle map invariant + // (leaked machines / machine-less legs); two map scans, + // compiled out of release builds. + #[cfg(debug_assertions)] + instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::DebugAssertPeerMapsCoherent, + self.debug_assert_peer_maps_coherent()); + }); + crate::instr::tick_gauges(instr_on, self.peers.len() as u64); } // Shutdown signal → enter the bounded drain in place, ONCE. // Gated on `is_none()` so it only fires while serving; after diff --git a/src/node/lifecycle/mod.rs b/src/node/lifecycle/mod.rs index 37c99ed..c717440 100644 --- a/src/node/lifecycle/mod.rs +++ b/src/node/lifecycle/mod.rs @@ -2221,6 +2221,10 @@ impl Node { // which the helper emits at the Dns→rest seam. self.execute_teardown(actions, true).await; + // Stop and join the profile writer thread if a capture is armed. + // Idempotent, and compiled away in a default build. + crate::instr::shutdown(); + self.supervisor.state = NodeState::Stopped; info!(state = %self.supervisor.state, "Node stopped"); Ok(()) @@ -2475,6 +2479,7 @@ impl Node { info!(state = %self.supervisor.state, "Node stopping (drain complete)"); let stop_actions = self.supervisor.fsm.step(Event::DrainDeadlineElapsed); self.execute_teardown(stop_actions, false).await; + crate::instr::shutdown(); self.supervisor.state = NodeState::Stopped; info!(state = %self.supervisor.state, "Node stopped"); } else if let Err(e) = self.stop().await { diff --git a/testing/ci-local.sh b/testing/ci-local.sh index f32c861..37b85eb 100755 --- a/testing/ci-local.sh +++ b/testing/ci-local.sh @@ -535,6 +535,30 @@ run_build() { return 1 fi + # An optional feature means two source trees, and --all-features lints only + # one of them. The default build is what ships, so lint it explicitly: + # without this stage, code that compiles only with `profiling` enabled would + # pass CI while breaking every release build. + info "cargo clippy --all-targets -- -D warnings (default features)" + if cargo clippy --all-targets -- -D warnings 2>&1; then + record "clippy-default-features" 0 + else + record "clippy-default-features" 1 + return 1 + fi + + # Tick-body profiler: the feature-on tree must build too. Added alongside + # the default-feature stages rather than replacing them. Mirrored in + # .github/workflows/ci.yml — check-ci-parity.sh compares integration suites + # only and will not catch a stage added to one runner and not the other. + info "cargo build --workspace --features profiling" + if cargo build --workspace --features profiling 2>&1; then + record "build-profiling" 0 + else + record "build-profiling" 1 + return 1 + fi + # Guard: the effectively-immutable state lives solely in NodeContext. The # Node struct must not re-declare a bundled field (config/identity/ # startup_epoch/started_at/is_leaf_only/node_profile/max_*) — a shadow field @@ -574,6 +598,17 @@ run_tests() { record "unit-tests" 1 fi fi + + # The `profiling` feature adds a module, a recorder and a writer thread that + # the default-feature run above never compiles. Run the library tests once + # more with it on so its own tests execute at all. Mirrored in + # .github/workflows/ci.yml. + info "cargo test --lib --features profiling" + if cargo test --lib --features profiling 2>&1; then + record "unit-tests-profiling" 0 + else + record "unit-tests-profiling" 1 + fi } # ── Stage 3: Integration Tests ─────────────────────────────────────────────