use std::path::{Path, PathBuf};
use std::sync::atomic::AtomicU64;
use std::sync::Mutex;
use gunnar_grpc::ControlPlane as _;
use snapshot::Volume as _;
use super::*;
const MIB: u64 = 1 << 20;
const DROPPED_MARK: &str = "txlog-snapshot-dropped";
fn bytes_under(root: &Path) -> u64 {
let mut total = 0u64;
let mut stack = vec![root.to_path_buf()];
while let Some(d) = stack.pop() {
let Ok(rd) = std::fs::read_dir(&d) else { continue };
for e in rd.flatten() {
let Ok(ft) = e.file_type() else { continue };
if ft.is_dir() {
stack.push(e.path());
} else if ft.is_file() {
total += e.metadata().map(|m| m.len()).unwrap_or(0);
}
}
}
total
}
struct Capped {
root: PathBuf,
cap: AtomicU64,
peak: AtomicU64,
watch: Mutex<Option<Box<dyn FnMut() + Send>>>,
}
impl Capped {
fn new(root: &Path, cap: u64) -> Arc<Self> {
Arc::new(Self { root: root.to_path_buf(), cap: AtomicU64::new(cap), peak: AtomicU64::new(0), watch: Mutex::new(None) })
}
}
impl snapshot::Volume for Capped {
fn space(&self, _path: &Path) -> Option<(u64, u64)> {
let cap = self.cap.load(Ordering::SeqCst);
Some((cap, cap.saturating_sub(bytes_under(&self.root))))
}
fn write_at(&self, file: &std::fs::File, bytes: &[u8], off: u64) -> std::io::Result<()> {
use std::os::unix::fs::FileExt as _;
let grows = (off + bytes.len() as u64).saturating_sub(file.metadata()?.len());
let used = bytes_under(&self.root);
if used + grows > self.cap.load(Ordering::SeqCst) {
return Err(std::io::Error::from_raw_os_error(rustix::io::Errno::NOSPC.raw_os_error()));
}
file.write_all_at(bytes, off)?;
self.peak.fetch_max(used + grows, Ordering::SeqCst);
if let Some(w) = self.watch.lock().unwrap().as_mut() {
w();
}
Ok(())
}
}
fn push_noise(primary: &ServerStores, what: &str, n: u64, bytes: u64) {
let app = primary.catalog().get(&RepoRef::new("ns", "app").unwrap()).unwrap().unwrap();
let store = primary.repo_store(&app).unwrap();
for i in 0..n {
let seed = what.bytes().fold(0x5EED_u64, |h, b| h.wrapping_mul(131).wrapping_add(u64::from(b)));
let (p, o) = one_blob_pack(store.hash_kind(), &noise(seed.wrapping_add(i), bytes as usize));
store.serve().put_pack(&p).unwrap();
store.serve().update_ref(&format!("refs/heads/{what}-{i}"), None, Some(&o)).unwrap();
}
}
struct Away {
pd: tempfile::TempDir,
primary: Arc<ServerStores>,
twin: Arc<ServerStores>,
head: u64,
own: u64,
stale: u64,
snapshot: u64,
}
fn settled<T>(primary: &ServerStores, what: &str, mut f: impl FnMut() -> anyhow::Result<T>) -> T {
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(60);
loop {
wait_until_indexed(primary, what);
match f() {
Ok(v) => return v,
Err(e) if format!("{e:#}").contains("index in progress") && std::time::Instant::now() < deadline => {
std::thread::sleep(std::time::Duration::from_millis(50));
}
Err(e) => panic!("{what}: {e:#}"),
}
}
}
impl Away {
fn new(twin_dir: &Path, pack: u64) -> Self {
let pd = tempfile::tempdir().unwrap();
let (primary, _facts) = primary(pd.path());
push_noise(&primary, "base", 3, pack);
let twin = open_twin(twin_dir);
let joined_at = primary.txlog().head_seq();
let join = Follower::new(Arc::clone(&twin), Box::new(LocalSource { stores: Arc::clone(&primary), retained_from: joined_at }));
assert_eq!(settled(&primary, "the twin joins behind a hole", || join.step()), Progress::SnapshotInstalled { at: joined_at }, "the twin joined through a snapshot: it keeps a base");
push_noise(&primary, "followed", 3, pack);
let follow = Follower::new(Arc::clone(&twin), Box::new(LocalSource { stores: Arc::clone(&primary), retained_from: 0 }));
while !matches!(follow.step().unwrap(), Progress::CaughtUp { .. }) {}
let own = twin.txlog().head_seq();
assert_eq!(own, primary.txlog().head_seq());
push_noise(&primary, "while-silent", 4, pack);
let head = primary.txlog().head_seq();
let snapshot = settled(&primary, "measuring main's snapshot", || {
let mut bytes = 0u64;
snapshot::produce(&primary, &mut |p| {
bytes += p.blob.len() as u64;
Ok(())
})?;
Ok(bytes)
});
let stale = bytes_under(&twin_dir.join(snapshot::BASE)) + twin.txlog().bytes();
Away { pd, primary, twin, head, own, stale, snapshot }
}
fn hole(&self) -> LocalSource {
LocalSource { stores: Arc::clone(&self.primary), retained_from: self.head }
}
fn mains_set(&self) -> u64 {
bytes_under(self.pd.path()) - self.primary.txlog().bytes() + 2 * MIB
}
}
#[derive(Debug)]
struct Window {
state: String,
why: String,
serve: Result<u64, String>,
alarms: Vec<String>,
}
fn look(twin: &Arc<ServerStores>) -> Window {
let st = status(twin.root()).expect("a status");
let serve = snapshot::serve(twin, &mut |_| Ok(())).map_err(|e| format!("{e:#}"));
Window { state: st.state.to_string(), why: st.why, serve, alarms: alarms_of(twin) }
}
fn alarms_of(stores: &Arc<ServerStores>) -> Vec<String> {
crate::CoreControlPlane::new(Arc::clone(stores)).health_alarms().into_iter().map(|a| format!("{}: {}", a.event, a.detail)).collect()
}
fn said<T: std::fmt::Debug>(r: &anyhow::Result<T>) -> String {
match r {
Ok(v) => format!("Ok({v:?})"),
Err(e) => format!("Err({e:#})"),
}
}
fn recharge_refused(twin: &Arc<ServerStores>) -> String {
let fd = tempfile::tempdir().unwrap();
let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
let r = restore_from(Box::new(LocalSource { stores: Arc::clone(twin), retained_from: 0 }), fresh, "");
format!("{:#}", r.expect_err("a recharge from a twin that holds no whole copy must be refused"))
}
fn resynced_and_following(a: &Away, twin: &Arc<ServerStores>, f: &Follower, who: &str) {
assert_eq!((twin.txlog().first_seq(), twin.txlog().head_seq()), (a.head + 1, a.head), "{who}: the twin's log resumes after the snapshot");
assert_eq!(snapshot::base_at(twin.root()).unwrap(), Some(a.head), "{who}: the twin keeps the new snapshot as its base");
for gone in [DROPPED_MARK, snapshot::STAGING, snapshot::INSTALLING] {
assert!(!twin.root().join(gone).exists(), "{who}: {gone} is left on the twin after the resync");
}
assert!(!alarms_of(twin).iter().any(|a| a.starts_with("twin.resync")), "{who}: the alarm is cleared: {:?}", alarms_of(twin));
{
let (_d, main) = recharged(twin);
same_state(&a.primary, &main, who);
for i in 0..4 {
assert!(refs_of(&main).iter().any(|(n, _, _)| n == &format!("refs/heads/while-silent-{i}")), "{who}: the push made while the twin was away is on the recharged main: {i}");
}
}
push_noise(&a.primary, "after-resync", 1, 64 << 10);
assert!(matches!(f.step().unwrap(), Progress::Applied { count, .. } if count >= 1), "{who}: the twin follows on after the resync");
let st = status(twin.root()).unwrap();
assert!(st.caught_up(), "{who}: caught up after the resync: {st:?}");
}
#[test]
fn resync_space_a_twin_the_size_of_mains_set_takes_a_full_resync_and_never_fills() {
let td = tempfile::tempdir().unwrap();
let a = Away::new(td.path(), 2 * MIB);
let cap = a.mains_set();
let rest = bytes_under(td.path()) - a.stale;
assert!(rest + a.stale + a.snapshot > cap, "the geometry: staged BESIDE its old copy ({} + {} + {}) the resync must not fit the twin's volume ({cap})", rest, a.stale, a.snapshot);
assert!(rest + a.snapshot + 2 * MIB <= cap, "the geometry: WITHOUT its old copy the resync fits ({} + {} + 2 MiB <= {cap})", rest, a.snapshot);
let vol = Capped::new(td.path(), cap);
let seen: Arc<Mutex<Option<Window>>> = Arc::new(Mutex::new(None));
*vol.watch.lock().unwrap() = Some(Box::new({
let (twin, seen, root) = (Arc::clone(&a.twin), Arc::clone(&seen), td.path().to_path_buf());
move || {
let mut s = seen.lock().unwrap();
if s.is_none() && root.join(DROPPED_MARK).exists() {
*s = Some(look(&twin));
}
}
}));
let f = Follower::new(Arc::clone(&a.twin), Box::new(a.hole())).with_volume(vol.clone());
let got = f.step();
assert!(matches!(got, Ok(Progress::SnapshotInstalled { at }) if at == a.head), "the resync landed on a volume the size of main's set: {}", said(&got));
assert!(vol.peak.load(Ordering::SeqCst) <= cap);
let w = seen.lock().unwrap().take().expect("the old copy was given up mid-resync: the volume holds one copy, not two");
assert_eq!(w.state, "resyncing", "{w:?}");
assert!(w.why.contains("NO copy"), "the status says the twin holds no copy meanwhile: {w:?}");
assert!(w.serve.as_ref().is_err_and(|e| e.contains("no whole copy")), "a twin without its copy serves none: {w:?}");
assert!(w.alarms.iter().any(|a| a.starts_with("twin.resync_without_copy")), "the twin raises an alarm: {w:?}");
resynced_and_following(&a, &a.twin, &f, "after a resync on a volume the size of main's set");
row(
"twin-resync-space: a twin whose volume is main's data set takes a full resync — its stale copy given up, alarmed — and a main recharged from it is main",
true,
&format!("cap {cap}, old copy {}, snapshot {}, peak {}", a.stale, a.snapshot, vol.peak.load(Ordering::SeqCst)),
);
}
#[test]
fn resync_space_a_crash_mid_resync_restarts_it_and_the_twin_is_whole_and_caught_up() {
struct CutAfterTheDrop {
inner: LocalSource,
twin_root: PathBuf,
after: AtomicU64,
}
impl LogSource for CutAfterTheDrop {
fn head(&self) -> anyhow::Result<(u64, u64)> {
self.inner.head()
}
fn fetch(&self, from: u64, to: u64, max_bytes: u64) -> anyhow::Result<Vec<Entry>> {
self.inner.fetch(from, to, max_bytes)
}
fn snapshot(&self, into: &mut snapshot::Stager) -> anyhow::Result<bool> {
snapshot::serve(&self.inner.stores, &mut |p| {
if self.twin_root.join(DROPPED_MARK).exists() && self.after.fetch_add(1, Ordering::SeqCst) >= 2 {
anyhow::bail!("the link to the primary died mid-resync (test)");
}
into.feed(p)
})?;
Ok(true)
}
}
let td = tempfile::tempdir().unwrap();
let a = Away::new(td.path(), 2 * MIB);
let cap = a.mains_set();
let f = Follower::new(Arc::clone(&a.twin), Box::new(CutAfterTheDrop { inner: a.hole(), twin_root: td.path().to_path_buf(), after: AtomicU64::new(0) }))
.with_volume(Capped::new(td.path(), cap));
let got = f.step();
assert!(got.as_ref().is_err_and(|e| format!("{e:#}").contains("died mid-resync")), "the link died after the old copy was given up: {}", said(&got));
assert!(td.path().join(DROPPED_MARK).exists());
assert!(!td.path().join(snapshot::BASE).exists(), "the stale base was given up");
assert_eq!((a.twin.txlog().first_seq(), a.twin.txlog().head_seq()), (a.own + 1, a.own), "the stale log was given up; the head stays the twin's own");
let st = status(td.path()).unwrap();
assert!(!st.caught_up() && st.state != State::Following && st.why.contains("NO copy"), "{st:?}");
let refused = recharge_refused(&a.twin);
assert!(refused.contains("no whole copy"), "{refused}");
assert!(alarms_of(&a.twin).iter().any(|x| x.starts_with("twin.resync_without_copy")), "{:?}", alarms_of(&a.twin));
drop(f);
let Away { pd, primary, twin, head, own, stale, snapshot } = a;
drop(twin);
status_slots().remove(td.path());
let twin = Arc::new(ServerStores::open(td.path()).unwrap());
assert!(twin.is_twin());
let a = Away { pd, primary, twin, head, own, stale, snapshot };
let refused = recharge_refused(&a.twin);
assert!(refused.contains("no whole copy"), "after the restart, before any poll: {refused}");
assert!(alarms_of(&a.twin).iter().any(|x| x.starts_with("twin.resync_without_copy")), "the alarm survives the restart");
let f = Follower::new(Arc::clone(&a.twin), Box::new(a.hole())).with_volume(Capped::new(td.path(), cap));
let got = f.step();
assert!(matches!(got, Ok(Progress::SnapshotInstalled { at }) if at == a.head), "the first poll after the restart takes the resync again: {}", said(&got));
resynced_and_following(&a, &a.twin, &f, "after a crash mid-resync and a restart");
row("twin-resync-space: a crash mid-resync (old copy already given up) is refused as a source, alarmed across the restart, and the next poll resyncs whole", true, &format!("cap {cap}, snapshot at {}", a.head));
}
#[test]
fn resync_space_a_give_up_cut_short_by_a_crash_is_finished_before_the_resync() {
let td = tempfile::tempdir().unwrap();
let a = Away::new(td.path(), 2 * MIB);
let cap = a.mains_set();
std::fs::write(td.path().join(DROPPED_MARK), format!("dropped_unix_ms 1\nhead {}\n", a.own)).unwrap();
assert!(td.path().join(snapshot::BASE).exists() && a.twin.txlog().bytes() > 0, "the old copy is still on the disk");
let refused = recharge_refused(&a.twin);
assert!(refused.contains("no whole copy"), "{refused}");
let f = Follower::new(Arc::clone(&a.twin), Box::new(a.hole())).with_volume(Capped::new(td.path(), cap));
let got = f.step();
assert!(matches!(got, Ok(Progress::SnapshotInstalled { at }) if at == a.head), "{}", said(&got));
resynced_and_following(&a, &a.twin, &f, "after a give-up cut short by a crash");
}
#[test]
fn resync_space_a_volume_too_small_even_alone_waits_in_recovering_until_grown_and_never_hammers_main() {
struct Counting {
inner: LocalSource,
asked: AtomicU64,
}
impl LogSource for Counting {
fn head(&self) -> anyhow::Result<(u64, u64)> {
self.inner.head()
}
fn fetch(&self, from: u64, to: u64, max_bytes: u64) -> anyhow::Result<Vec<Entry>> {
self.inner.fetch(from, to, max_bytes)
}
fn snapshot(&self, into: &mut snapshot::Stager) -> anyhow::Result<bool> {
self.asked.fetch_add(1, Ordering::SeqCst);
self.inner.snapshot(into)
}
}
let td = tempfile::tempdir().unwrap();
let a = Away::new(td.path(), 2 * MIB);
let rest = bytes_under(td.path()) - a.stale;
let cap = rest + a.snapshot - 2 * MIB;
assert!(rest + a.stale <= cap, "the geometry: the twin's old copy fits ({} + {} <= {cap})", rest, a.stale);
let vol = Capped::new(td.path(), cap);
let source = Arc::new(Counting { inner: a.hole(), asked: AtomicU64::new(0) });
struct Shared(Arc<Counting>);
impl LogSource for Shared {
fn head(&self) -> anyhow::Result<(u64, u64)> {
self.0.head()
}
fn fetch(&self, from: u64, to: u64, max_bytes: u64) -> anyhow::Result<Vec<Entry>> {
self.0.fetch(from, to, max_bytes)
}
fn snapshot(&self, into: &mut snapshot::Stager) -> anyhow::Result<bool> {
self.0.snapshot(into)
}
}
let f = Follower::new(Arc::clone(&a.twin), Box::new(Shared(Arc::clone(&source)))).with_volume(vol.clone());
let got = f.step();
let err = format!("{:#}", got.as_ref().map(drop).expect_err("a volume smaller than main's data cannot take the resync"));
assert!(err.contains(gunnar_grpc::TWIN_TOO_SMALL), "named as too small, not a bare ENOSPC: {err}");
let st = status(td.path()).unwrap();
assert_eq!(st.state, State::Recovering, "{st:?}");
assert!(st.why.contains("NO copy") && st.why.contains(gunnar_grpc::TWIN_TOO_SMALL), "{st:?}");
assert!(alarms_of(&a.twin).iter().any(|x| x.starts_with("twin.resync_without_copy")), "{:?}", alarms_of(&a.twin));
assert_eq!(source.asked.load(Ordering::SeqCst), 1);
assert!(!td.path().join(snapshot::STAGING).exists(), "the half staged is not left taking the room");
let again = f.step();
assert!(again.as_ref().is_err_and(|e| format!("{e:#}").contains(gunnar_grpc::TWIN_TOO_SMALL)), "{}", said(&again));
assert_eq!(source.asked.load(Ordering::SeqCst), 1, "a twin too small for main's data does not freeze main's writes with a resync it cannot hold");
vol.cap.store(cap * 2, Ordering::SeqCst);
let got = f.step();
assert!(matches!(got, Ok(Progress::SnapshotInstalled { at }) if at == a.head), "{}", said(&got));
assert_eq!(source.asked.load(Ordering::SeqCst), 2);
assert!(!td.path().join("txlog-snapshot-too-small").exists());
resynced_and_following(&a, &a.twin, &f, "after the volume was grown");
}
#[test]
fn resync_space_a_twin_mid_install_serves_no_snapshot_and_no_recharge() {
let td = tempfile::tempdir().unwrap();
let a = Away::new(td.path(), 256 << 10);
assert!(snapshot::base_at(td.path()).unwrap().is_some());
crate::durable::write_atomic_unframed(td.path(), snapshot::INSTALLING, format!("{}\n", a.head).as_bytes()).unwrap();
let served = snapshot::serve(&a.twin, &mut |_| Ok(()));
assert!(served.as_ref().is_err_and(|e| format!("{e:#}").contains("no whole copy")), "{}", said(&served));
let refused = recharge_refused(&a.twin);
assert!(refused.contains("no whole copy"), "{refused}");
}
#[test]
#[ignore = "needs GUNNAR_TWIN_CAPPED_DIR: an empty directory on a small filesystem of its own"]
fn resync_space_on_a_real_filesystem_the_size_of_mains_set() {
let dir = PathBuf::from(std::env::var_os("GUNNAR_TWIN_CAPPED_DIR").expect("GUNNAR_TWIN_CAPPED_DIR"));
let root = dir.join(format!("twin-{}", std::process::id()));
std::fs::create_dir_all(&root).unwrap();
let (total, free) = snapshot::Disk.space(&root).expect("a measurable filesystem");
let pack = total / 12;
let a = Away::new(&root, pack);
let (_, free_before) = snapshot::Disk.space(&root).unwrap();
assert!(a.snapshot > free_before, "the geometry: staged beside its old copy the resync cannot fit ({} > {free_before} free of {total}; {free} at the start)", a.snapshot);
let f = Follower::new(Arc::clone(&a.twin), Box::new(a.hole()));
let got = f.step();
assert!(matches!(got, Ok(Progress::SnapshotInstalled { at }) if at == a.head), "the resync landed on a real {total}-byte filesystem: {}", said(&got));
resynced_and_following(&a, &a.twin, &f, "on a real filesystem the size of main's set");
row("twin-resync-space: the resync on a REAL filesystem the size of main's set", true, &format!("filesystem {total}, packs {pack}, snapshot {}", a.snapshot));
drop(f);
drop(a);
let _ = std::fs::remove_dir_all(&root);
}