diff --git a/beyond-pg-init/src/bootsetup.rs b/beyond-pg-init/src/bootsetup.rs index 462dde9..27ed3d0 100644 --- a/beyond-pg-init/src/bootsetup.rs +++ b/beyond-pg-init/src/bootsetup.rs @@ -195,7 +195,15 @@ fn setup_network() { } if let Some(ref gua) = parse_cmdline_ipv6_ext(&cmdline) { run_ip( - &["-6", "addr", "add", &format!("{gua}/128"), "dev", "eth0", "nodad"], + &[ + "-6", + "addr", + "add", + &format!("{gua}/128"), + "dev", + "eth0", + "nodad", + ], &format!("add IPv6 GUA {gua}/128"), ); } diff --git a/beyond-pg-init/src/substrate.rs b/beyond-pg-init/src/substrate.rs index 128a8b0..2202d78 100644 --- a/beyond-pg-init/src/substrate.rs +++ b/beyond-pg-init/src/substrate.rs @@ -70,6 +70,35 @@ struct GuestResourceStatsPayload { psi_mem_some_avg10: Option, #[serde(skip_serializing_if = "Option::is_none")] psi_mem_full_avg10: Option, + /// Cumulative `workingset_refault_file` from `/proc/vmstat`: file pages + /// evicted from the page cache and then read back in. + /// + /// Reported so the host can right-size this VM's memory. Postgres is + /// page-cache-bound and reads with `read()`, not `mmap`, so when its working + /// set outgrows RAM it does not swap, takes no major faults and never stalls + /// on reclaim — measured on a real VM (1.8 GB working set, 735 MiB RAM): + /// PSI 0.00, swap 0, major faults 0/tick. None of the usual memory-pressure + /// signals move. Refaults are the one counter that does: re-reading a page we + /// just evicted is precisely "the working set does not fit". + #[serde(skip_serializing_if = "Option::is_none")] + workingset_refault_file: Option, +} + +/// Read cumulative `workingset_refault_file` from `/proc/vmstat`. +fn read_workingset_refault_file() -> Option { + let raw = std::fs::read_to_string("/proc/vmstat").ok()?; + parse_workingset_refault_file(&raw) +} + +/// Parse `workingset_refault_file` from `/proc/vmstat` text. Split out from the +/// file read so it's testable without `/proc`. +fn parse_workingset_refault_file(raw: &str) -> Option { + raw.lines().find_map(|l| { + l.strip_prefix("workingset_refault_file ")? + .trim() + .parse() + .ok() + }) } /// Read Linux PSI memory pressure `(some.avg10, full.avg10)` from @@ -107,6 +136,7 @@ fn encode_resource_stats_frame() -> Option> { disk_used_bytes: 0, psi_mem_some_avg10: Some(some), psi_mem_full_avg10: Some(full), + workingset_refault_file: read_workingset_refault_file(), }; let body = rmp_serde::to_vec_named(&payload).ok()?; Some(encode_frame(MSG_GUEST_RESOURCE_STATS, &body)) @@ -313,7 +343,9 @@ fn spawn_log_sink(log_tx: tokio::sync::mpsc::Sender) { let listener = match UnixListener::bind(path) { Ok(l) => l, Err(e) => { - eprintln!("[init] WARNING: log sink bind {path} failed: {e}; workload logs disabled"); + eprintln!( + "[init] WARNING: log sink bind {path} failed: {e}; workload logs disabled" + ); return; } }; @@ -404,7 +436,7 @@ mod tests { /// `vsock_protocol::GuestResourceStatsPayload` decoder: BE length, the type /// byte, then a MessagePack map whose keys match the host struct. `seq` and /// `disk_used_bytes` are required host fields; `disk_total_bytes` is omitted - /// so the host skips disk billing; the PSI keys carry the signal. + /// so the host skips disk billing; the PSI and refault keys carry the signal. #[test] fn resource_stats_frame_is_stable() { let payload = GuestResourceStatsPayload { @@ -412,6 +444,7 @@ mod tests { disk_used_bytes: 0, psi_mem_some_avg10: Some(12.34), psi_mem_full_avg10: Some(0.5), + workingset_refault_file: Some(123_456), }; let body = rmp_serde::to_vec_named(&payload).unwrap(); let frame = encode_frame(MSG_GUEST_RESOURCE_STATS, &body); @@ -426,12 +459,47 @@ mod tests { assert!(obj.contains_key("disk_used_bytes")); assert!(obj.contains_key("psi_mem_some_avg10")); assert!(obj.contains_key("psi_mem_full_avg10")); + assert!( + obj.contains_key("workingset_refault_file"), + "the refault counter must reach the host — it is the ONLY signal that \ + sees this VM outgrow its RAM (PSI, swap and major faults all read \ + zero on a page-cache-bound workload)" + ); assert!( !obj.contains_key("disk_total_bytes"), "disk_total omitted so the host skips disk billing" ); } + #[test] + fn refault_frame_omits_the_key_when_unavailable() { + // The field is Option: a kernel without the counter must simply omit it, + // not send a bogus 0 that the host would read as "no refaults". + let payload = GuestResourceStatsPayload { + seq: 0, + disk_used_bytes: 0, + psi_mem_some_avg10: Some(1.0), + psi_mem_full_avg10: Some(0.0), + workingset_refault_file: None, + }; + let body = rmp_serde::to_vec_named(&payload).unwrap(); + let v: serde_json::Value = rmp_serde::from_slice(&body).unwrap(); + assert!( + !v.as_object() + .unwrap() + .contains_key("workingset_refault_file") + ); + } + + #[test] + fn parses_workingset_refault_file() { + let raw = "nr_free_pages 12345\nworkingset_refault_anon 0\nworkingset_refault_file 297848\npgmajfault 8\n"; + assert_eq!(parse_workingset_refault_file(raw), Some(297_848)); + // Absent on kernels that don't export it — must be None, not 0. + assert_eq!(parse_workingset_refault_file("nr_free_pages 1\n"), None); + assert_eq!(parse_workingset_refault_file(""), None); + } + #[test] fn parses_psi_memory() { let raw = "\ diff --git a/src/boot.rs b/src/boot.rs index 5b30762..c13fc27 100644 --- a/src/boot.rs +++ b/src/boot.rs @@ -289,7 +289,12 @@ fn materialize_template_into( // Idempotent clean slate. `pgdata` may exist empty (partial-PGDATA cleanup // recreated it above); a prior interrupted materialize may have left a wal // dir or *.staging dirs. - for p in [&main_staging, &wal_staging, &pgdata.to_string(), &wal_target.to_string()] { + for p in [ + &main_staging, + &wal_staging, + &pgdata.to_string(), + &wal_target.to_string(), + ] { rm_rf(p)?; } @@ -373,7 +378,6 @@ fn cp_a(src: &str, dst: &str) -> Result<(), BootError> { } } - /// chown `/var/lib/postgresql` → postgres recursively. A fresh durable volume /// mounts root-owned over the image dir, and a root-run `pg_basebackup` (replica /// seeding) writes a root-owned PGDATA — either way postgres dies at startup with @@ -737,7 +741,10 @@ mod tests { .unwrap(); assert!(pgdata.join("PG_VERSION").exists()); - assert!(!pgdata.join("garbage").exists(), "stale staging must not leak in"); + assert!( + !pgdata.join("garbage").exists(), + "stale staging must not leak in" + ); } // ----------------------------------------------------------------------- diff --git a/src/init.rs b/src/init.rs index 5463523..e207bf7 100644 --- a/src/init.rs +++ b/src/init.rs @@ -121,7 +121,15 @@ mod linux { } if let Some(ref gua) = read_cmdline_ipv6_ext() { run_ip_cmd( - &["-6", "addr", "add", &format!("{gua}/128"), "dev", "eth0", "nodad"], + &[ + "-6", + "addr", + "add", + &format!("{gua}/128"), + "dev", + "eth0", + "nodad", + ], "add IPv6 GUA address", ); } diff --git a/src/supervisor.rs b/src/supervisor.rs index f3a5be7..431cbc3 100644 --- a/src/supervisor.rs +++ b/src/supervisor.rs @@ -740,11 +740,8 @@ async fn run_inner_inner(role: MaybeRole) -> Result<(), Box(1); - let memory_watcher_handle = tokio::spawn(memory_watcher_task( - cfg.vcpus, - cfg.ram_bytes, - retune_tx, - )); + let memory_watcher_handle = + tokio::spawn(memory_watcher_task(cfg.vcpus, cfg.ram_bytes, retune_tx)); // Background: watch the platform TLS cert for rotation, if applicable. // The guest agent atomically replaces /run/beyond/tls/cert.pem every ~22h. @@ -1840,7 +1837,12 @@ async fn post_start(cfg: &MmdsConfig) -> Result<(), Box> ); } } - extensions.extend(OPTIONAL_EXTENSIONS.iter().copied().filter(|e| extension_installed(e))); + extensions.extend( + OPTIONAL_EXTENSIONS + .iter() + .copied() + .filter(|e| extension_installed(e)), + ); if cfg.cdc_enabled { warn!("CDC enabled: replication slot 'cdc' will accumulate WAL until a consumer connects"); @@ -1871,7 +1873,10 @@ fn build_post_start_script(cfg: &MmdsConfig, extensions: &[&str]) -> String { let mut s = String::new(); // 1. Reset the superuser password from the per-instance MMDS secret. - s.push_str(&pg::alter_role_password_sql("postgres", &cfg.postgres_password)); + s.push_str(&pg::alter_role_password_sql( + "postgres", + &cfg.postgres_password, + )); s.push_str(";\n"); // 2. pgbouncer auth (role + SECURITY DEFINER lookup function + grants). diff --git a/src/vsock.rs b/src/vsock.rs index d430d45..18eaf76 100644 --- a/src/vsock.rs +++ b/src/vsock.rs @@ -205,8 +205,7 @@ mod tests { #[serde(with = "serde_bytes")] bytes: Vec, } - let outer: OuterIn = - rmp_serde::from_slice(&frame[5..]).expect("decode AppMessagePayload"); + let outer: OuterIn = rmp_serde::from_slice(&frame[5..]).expect("decode AppMessagePayload"); // 3. Inner envelope: [ServiceKind::Ready=12][msgpack_named(ReadyPayload)]. let (kind, rest) = outer.bytes.split_first().expect("non-empty inner bytes"); @@ -219,7 +218,8 @@ mod tests { // omitted, so the driver's `ReadyPayload` decode yields // `http_port: None`. assert_eq!( - rest, &[0x80], + rest, + &[0x80], "ReadyPayload must serialize as an empty msgpack map (http_port omitted)" ); }