nordisk/gunnar

crates/gunnar-repair/src/fence.rs

//! **`fence_primary`**: the old primary takes no client write and ships no copy,
//! so there are never two writable boxes (the split-brain a recharge would
//! otherwise open). Two existing, idempotent RPCs, in the order abdication uses
//! them: `Mirror.SetRole(STANDBY)` (refuse client pushes), then
//! `Mirror.DemoteReplication` (clear the roster), then the role read back.

#![allow(async_fn_in_trait)]

use serde::{Deserialize, Serialize};

use crate::plan::{refuse, Refusal, RefusalCode};
use crate::pump::Fault;

/// The role word a fenced box answers.
pub const STANDBY: &str = "standby";

/// **The box to fence**, as the host reaches it (signed with an admin key).
pub trait Fenceable {
    fn addr(&self) -> String;
    /// `Mirror.SetRole(STANDBY)`: the role the box answers after it.
    async fn set_standby(&self) -> Result<String, Fault>;
    /// `Mirror.DemoteReplication`: the targets left (0 = demoted).
    async fn demote(&self) -> Result<usize, Fault>;
}

/// What a fence did.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Fenced {
    pub primary: String,
    /// The box answered and is now a standby with no roster.
    pub fenced: bool,
    pub role: String,
    pub targets_left: usize,
    /// The epoch the order named (0 = a local fence).
    pub epoch: u64,
    pub detail: String,
}

/// **Fence `primary`.** `Err(Unreachable)` when the box does not answer: a box that
/// answers nothing takes no write either, but only a cloud stop PROVES it — the
/// caller says so (`CloudProvider.stop` is monetize's lever, never this crate's).
pub async fn fence_primary<F: Fenceable>(primary: &F, epoch: u64) -> Result<Fenced, Refusal> {
    let addr = primary.addr();
    let unreachable = |f: Fault| {
        refuse(
            RefusalCode::Unreachable,
            format!(
                "the primary at {addr} could not be fenced ({f}); if it is down it takes no write, but only stopping the machine proves it — stop it before a recharge"
            ),
        )
    };
    let role = primary.set_standby().await.map_err(unreachable)?;
    let targets_left = primary.demote().await.map_err(unreachable)?;
    let fenced = role.eq_ignore_ascii_case(STANDBY) && targets_left == 0;
    let detail = if fenced {
        format!("the box at {addr} is a standby with no replication roster: it refuses client pushes and ships nothing")
    } else {
        format!("the box at {addr} answers role `{role}` with {targets_left} target(s) left after the fence: NOT fenced")
    };
    if !fenced {
        return Err(refuse(RefusalCode::NotFenced, detail));
    }
    Ok(Fenced {
        primary: addr,
        fenced,
        role,
        targets_left,
        epoch,
        detail,
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::pump::block_on_ready;
    use std::sync::Mutex;

    struct Box_ {
        up: bool,
        role: Mutex<String>,
        calls: Mutex<Vec<&'static str>>,
    }

    impl Fenceable for Box_ {
        fn addr(&self) -> String {
            "main:1".into()
        }
        async fn set_standby(&self) -> Result<String, Fault> {
            self.calls.lock().unwrap().push("set_role");
            if !self.up {
                return Err(Fault::new("unavailable", "connection refused"));
            }
            *self.role.lock().unwrap() = STANDBY.into();
            Ok(STANDBY.into())
        }
        async fn demote(&self) -> Result<usize, Fault> {
            self.calls.lock().unwrap().push("demote");
            Ok(0)
        }
    }

    #[test]
    fn a_fence_freezes_writes_first_then_clears_the_roster() {
        let b = Box_ {
            up: true,
            role: Mutex::new("primary".into()),
            calls: Mutex::new(vec![]),
        };
        let f = block_on_ready(fence_primary(&b, 5)).unwrap();
        assert!(f.fenced && f.role == STANDBY && f.epoch == 5);
        assert_eq!(*b.calls.lock().unwrap(), vec!["set_role", "demote"]);
    }

    #[test]
    fn an_unreachable_primary_is_not_called_fenced() {
        let b = Box_ {
            up: false,
            role: Mutex::new("primary".into()),
            calls: Mutex::new(vec![]),
        };
        let e = block_on_ready(fence_primary(&b, 1)).unwrap_err();
        assert_eq!(e.code, RefusalCode::Unreachable);
        assert!(e.message.contains("stop it"), "{e}");
    }
}