feat: add debug-level per-op, session and link metrics
This commit is contained in:
parent
0ba9f32837
commit
07f064460f
5 changed files with 76 additions and 0 deletions
|
|
@ -77,6 +77,8 @@ bunshin serve --key server.key --db mail.db --host 0.0.0.0 --port 1961
|
||||||
|
|
||||||
`serve` accepts `--max-envelope`, `--quota`, `--requests-quota`, `--retention-days`, `--requests-retention-days`, `--max-tokens`, `--invite-token`, `--rate-connections`, `--rate-sends` and `--rate-tokens` to control size limits, the mailbox's two quota tiers, their retention, the accept-token cap, registration gating and abuse control. Run `bunshin serve --help` for defaults.
|
`serve` accepts `--max-envelope`, `--quota`, `--requests-quota`, `--retention-days`, `--requests-retention-days`, `--max-tokens`, `--invite-token`, `--rate-connections`, `--rate-sends` and `--rate-tokens` to control size limits, the mailbox's two quota tiers, their retention, the accept-token cap, registration gating and abuse control. Run `bunshin serve --help` for defaults.
|
||||||
|
|
||||||
|
`--verbose` (or `services.bunshin.verbose` in the NixOS module) raises logging to debug level, which adds server-side metrics on both carriers: per-operation timing, request and response sizes, handshake duration, per-session summaries, and RNS link events. The default info level stays quiet on success.
|
||||||
|
|
||||||
## Status
|
## Status
|
||||||
|
|
||||||
Implements SPEC.md version 1.1 in full: `AUTH`, `RESOLVE`, `SEND`, `FETCH`, `DELETE` and `REGISTER`, including accept tokens and the main/requests tier split, fetch cursors, 32-byte message ids, `REGISTER` proof of possession, and dual-signed key rotation chains.
|
Implements SPEC.md version 1.1 in full: `AUTH`, `RESOLVE`, `SEND`, `FETCH`, `DELETE` and `REGISTER`, including accept tokens and the main/requests tier split, fetch cursors, 32-byte message ids, `REGISTER` proof of possession, and dual-signed key rotation chains.
|
||||||
|
|
|
||||||
11
flake.nix
11
flake.nix
|
|
@ -220,6 +220,15 @@
|
||||||
description = "Open the configured TCP port in the firewall.";
|
description = "Open the configured TCP port in the firewall.";
|
||||||
};
|
};
|
||||||
|
|
||||||
|
verbose = mkOption {
|
||||||
|
type = types.bool;
|
||||||
|
default = false;
|
||||||
|
description = ''
|
||||||
|
Enable debug-level logging (--verbose): per-op timing,
|
||||||
|
session summaries and RNS link events on both carriers.
|
||||||
|
'';
|
||||||
|
};
|
||||||
|
|
||||||
rns = {
|
rns = {
|
||||||
enable = mkEnableOption "the RNS carrier alongside TCP (needs an rns-built package)";
|
enable = mkEnableOption "the RNS carrier alongside TCP (needs an rns-built package)";
|
||||||
|
|
||||||
|
|
@ -333,6 +342,8 @@
|
||||||
''}
|
''}
|
||||||
${lib.optionalString (cfg.rns.enable && cfg.rns.udpForward != null)
|
${lib.optionalString (cfg.rns.enable && cfg.rns.udpForward != null)
|
||||||
''args+=(--rns-udp-forward ${lib.escapeShellArg cfg.rns.udpForward})''}
|
''args+=(--rns-udp-forward ${lib.escapeShellArg cfg.rns.udpForward})''}
|
||||||
|
${lib.optionalString cfg.verbose
|
||||||
|
''args+=(--verbose)''}
|
||||||
${lib.optionalString (cfg.inviteToken != null)
|
${lib.optionalString (cfg.inviteToken != null)
|
||||||
''args+=(--invite-token ${lib.escapeShellArg cfg.inviteToken})''}
|
''args+=(--invite-token ${lib.escapeShellArg cfg.inviteToken})''}
|
||||||
${lib.optionalString (cfg.inviteTokenFile != null)
|
${lib.optionalString (cfg.inviteTokenFile != null)
|
||||||
|
|
|
||||||
|
|
@ -280,6 +280,7 @@ extern "C" fn smolmail_rns_on_link_opened(link_id: *const u8) -> i32 {
|
||||||
return 1;
|
return 1;
|
||||||
}
|
}
|
||||||
links.insert(link);
|
links.insert(link);
|
||||||
|
log::debug!("link {} opened ({} open)", data_encoding::HEXLOWER.encode(&link), links.len());
|
||||||
0
|
0
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -289,6 +290,7 @@ extern "C" fn smolmail_rns_on_link_closed(link_id: *const u8) {
|
||||||
let link: [u8; 16] = unsafe { std::slice::from_raw_parts(link_id, 16).try_into().unwrap() };
|
let link: [u8; 16] = unsafe { std::slice::from_raw_parts(link_id, 16).try_into().unwrap() };
|
||||||
state.links.lock().unwrap().remove(&link);
|
state.links.lock().unwrap().remove(&link);
|
||||||
state.sessions.lock().unwrap().remove(&link);
|
state.sessions.lock().unwrap().remove(&link);
|
||||||
|
log::debug!("link {} closed", data_encoding::HEXLOWER.encode(&link));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -103,6 +103,7 @@ fn handle_connection(
|
||||||
) -> anyhow::Result<()> {
|
) -> anyhow::Result<()> {
|
||||||
stream.set_read_timeout(Some(Duration::from_secs(IDLE_TIMEOUT_SECS)))?;
|
stream.set_read_timeout(Some(Duration::from_secs(IDLE_TIMEOUT_SECS)))?;
|
||||||
|
|
||||||
|
let handshake_started = std::time::Instant::now();
|
||||||
let (transport, handshake_hash) = match handshake(&mut stream, static_key) {
|
let (transport, handshake_hash) = match handshake(&mut stream, static_key) {
|
||||||
Ok(v) => v,
|
Ok(v) => v,
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
|
|
@ -110,6 +111,17 @@ fn handle_connection(
|
||||||
return Ok(());
|
return Ok(());
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
log::debug!(
|
||||||
|
"handshake from {peer_ip} in {:?}",
|
||||||
|
handshake_started.elapsed()
|
||||||
|
);
|
||||||
|
|
||||||
|
// Drop logs the summary once, whatever return path closes the session.
|
||||||
|
let summary = SessionSummary {
|
||||||
|
peer_ip,
|
||||||
|
started: std::time::Instant::now(),
|
||||||
|
ops: std::cell::Cell::new(0),
|
||||||
|
};
|
||||||
|
|
||||||
let store = Store::open(db_path)?;
|
let store = Store::open(db_path)?;
|
||||||
let server_static = derive_public(static_key);
|
let server_static = derive_public(static_key);
|
||||||
|
|
@ -129,6 +141,7 @@ fn handle_connection(
|
||||||
return Ok(());
|
return Ok(());
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
summary.ops.set(summary.ops.get() + 1);
|
||||||
|
|
||||||
let (status, payload) = session.dispatch(op, &body);
|
let (status, payload) = session.dispatch(op, &body);
|
||||||
let mut response = Vec::with_capacity(1 + payload.len());
|
let mut response = Vec::with_capacity(1 + payload.len());
|
||||||
|
|
@ -140,6 +153,23 @@ fn handle_connection(
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
struct SessionSummary<'a> {
|
||||||
|
peer_ip: &'a str,
|
||||||
|
started: std::time::Instant,
|
||||||
|
ops: std::cell::Cell<u64>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Drop for SessionSummary<'_> {
|
||||||
|
fn drop(&mut self) {
|
||||||
|
log::debug!(
|
||||||
|
"session {} closed: {} op(s) in {:?}",
|
||||||
|
self.peer_ip,
|
||||||
|
self.ops.get(),
|
||||||
|
self.started.elapsed()
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
fn is_eof_like(e: &anyhow::Error) -> bool {
|
fn is_eof_like(e: &anyhow::Error) -> bool {
|
||||||
if let Some(io_err) = e.downcast_ref::<std::io::Error>() {
|
if let Some(io_err) = e.downcast_ref::<std::io::Error>() {
|
||||||
return matches!(
|
return matches!(
|
||||||
|
|
|
||||||
|
|
@ -44,6 +44,19 @@ impl From<rusqlite::Error> for HandlerError {
|
||||||
|
|
||||||
type OpResult = Result<(u8, Vec<u8>), HandlerError>;
|
type OpResult = Result<(u8, Vec<u8>), HandlerError>;
|
||||||
|
|
||||||
|
/// Wire op byte to log name; unknown bytes are logged as dispatched too.
|
||||||
|
fn op_name(op: u8) -> &'static str {
|
||||||
|
match op {
|
||||||
|
OP_AUTH => "AUTH",
|
||||||
|
OP_RESOLVE => "RESOLVE",
|
||||||
|
OP_SEND => "SEND",
|
||||||
|
OP_FETCH => "FETCH",
|
||||||
|
OP_DELETE => "DELETE",
|
||||||
|
OP_REGISTER => "REGISTER",
|
||||||
|
_ => "UNKNOWN",
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
pub struct Session<'a> {
|
pub struct Session<'a> {
|
||||||
config: &'a ServerConfig,
|
config: &'a ServerConfig,
|
||||||
store: Store,
|
store: Store,
|
||||||
|
|
@ -72,7 +85,25 @@ impl<'a> Session<'a> {
|
||||||
/// panicking or propagating errors to the caller: a bad frame or a
|
/// panicking or propagating errors to the caller: a bad frame or a
|
||||||
/// storage error both become a response, and the caller decides
|
/// storage error both become a response, and the caller decides
|
||||||
/// separately whether to keep the connection open.
|
/// separately whether to keep the connection open.
|
||||||
|
///
|
||||||
|
/// Timing lives in this wrapper because both carriers dispatch through
|
||||||
|
/// it, so the metrics flag covers TCP and RNS from one place.
|
||||||
pub fn dispatch(&mut self, op: u8, body: &[u8]) -> (u8, Vec<u8>) {
|
pub fn dispatch(&mut self, op: u8, body: &[u8]) -> (u8, Vec<u8>) {
|
||||||
|
let started = std::time::Instant::now();
|
||||||
|
let (status, payload) = self.dispatch_op(op, body);
|
||||||
|
log::debug!(
|
||||||
|
"op {} from {} in {:?}: status {}, {}B request, {}B response",
|
||||||
|
op_name(op),
|
||||||
|
self.peer_ip,
|
||||||
|
started.elapsed(),
|
||||||
|
status,
|
||||||
|
body.len(),
|
||||||
|
payload.len()
|
||||||
|
);
|
||||||
|
(status, payload)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn dispatch_op(&mut self, op: u8, body: &[u8]) -> (u8, Vec<u8>) {
|
||||||
if matches!(op, OP_FETCH | OP_DELETE) && self.username.is_none() {
|
if matches!(op, OP_FETCH | OP_DELETE) && self.username.is_none() {
|
||||||
return (AUTH_REQUIRED, Vec::new());
|
return (AUTH_REQUIRED, Vec::new());
|
||||||
}
|
}
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue