gunnar-server 2.0.0

//! **A sold estate's paid data step, pulled** (G16-d, `grow_go_ahead`).
//!
//! The ONE monetize queues a signed `grow_go_ahead` to the estate's main AND its twin
//! once the step's money is captured; this box PULLS it like every other order. Here is
//! the box's half of [`gunnar_repair::exec::grow`]: verify monetize's `EstateGrowth`
//! (`monetize_embed::signing::verify_estate_growth`, its estate, not past `until`), say
//! whether the data disks already carry the step, and drive THIS box's own
//! [`GrowMachine`] — the same machine `Admin.GrowStart` drives. The order and the step
//! were verified under the monetize key before anything moves, so the machine is started
//! with an `operator` go-ahead minted here for exactly that size (decision G-D38).
//!
//! The twin first: a main grows only once its twin's `Bootstrap.GrowStatus` says every
//! twin disk carries main's whole set ([`twin_holds`]).

use std::sync::Arc;
use std::time::Duration;

use gunnar_repair::exec::{GrowCheck, GrowState};
use gunnar_repair::{Refusal, RefusalCode};

use crate::grow::GrowMachine;

/// 512-byte sectors in a GiB: the grow machine counts sectors, monetize GB a disk.
pub const SECTORS_PER_GIB: u64 = 2_097_152;

pub(crate) fn refuse(code: RefusalCode, message: impl Into<String>) -> Refusal {
    Refusal { code, message: message.into() }
}

/// The signer word on a paid step (`monetize_embed::signing::GROWTH_SIGNER_SHOP`).
pub const GROWTH_SIGNER_SHOP: &str = "shop";

/// **monetize's `EstateGrowth`**, field for field (`monetize_embed::signing::EstateGrowth`,
/// newer than the monetize-embed this build links): the paid step, signed by the shop key.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct EstateGrowth {
    pub v: u32,
    pub signer: String,
    pub estate: String,
    pub main_data_gib_per_member: u64,
    pub until_unix_ms: i64,
    pub reference: String,
    pub issued_unix_ms: i64,
    #[serde(default)]
    pub signature: String,
}

/// The bytes an [`EstateGrowth`]'s signature covers: monetize-embed's ONE canonical
/// JSON of every field but `signature`.
#[cfg(feature = "metered")]
pub fn growth_message(g: &EstateGrowth) -> Vec<u8> {
    let mut v = serde_json::to_value(g).expect("a growth go-ahead serializes");
    v.as_object_mut().expect("a growth go-ahead is an object").remove("signature");
    let mut s = String::new();
    monetize_embed::signing::canonical_json(&v, &mut s);
    s.into_bytes()
}

/// What monetize's verified step says.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PaidStep {
    /// monetize's order reference the step was paid on.
    pub reference: String,
    /// GB a data disk on main, as paid.
    pub main_gib: u64,
}

/// **Verify monetize's `EstateGrowth`** under `key` for `estate` at `now_unix_ms`, and
/// that `data_gib` is what THIS role is paid: the main its step, the twin at least that
/// (its disk holds main's whole set).
#[cfg(feature = "metered")]
pub fn verify_step(go_ahead: &str, key: Option<&[u8; 32]>, estate: &str, twin: bool, data_gib: u64, now_unix_ms: i64) -> Result<PaidStep, Refusal> {
    use ed25519_dalek::Verifier;
    let bad = |m: String| refuse(RefusalCode::BadOrder, m);
    let key = key.ok_or_else(|| bad("this box has no monetize key yet, so monetize's paid step cannot be verified: nothing moved".into()))?;
    let key = ed25519_dalek::VerifyingKey::from_bytes(key).map_err(|e| bad(format!("this box's monetize key is not an Ed25519 point ({e}): nothing moved")))?;
    let not_monetizes = |why: &str| bad(format!("the paid step does not verify under this box's monetize key ({why}): not monetize's, nothing moved"));
    let g: EstateGrowth = serde_json::from_str(go_ahead).map_err(|_| not_monetizes("not an EstateGrowth"))?;
    if g.v != 1 || g.signer != GROWTH_SIGNER_SHOP || g.estate.trim().is_empty() || g.main_data_gib_per_member == 0 {
        return Err(not_monetizes("malformed"));
    }
    let sig = monetize_embed::signing::base64_decode(&g.signature).ok_or_else(|| not_monetizes("no signature"))?;
    let sig = ed25519_dalek::Signature::from_slice(&sig).map_err(|_| not_monetizes("no signature"))?;
    key.verify(&growth_message(&g), &sig).map_err(|_| not_monetizes("forged"))?;
    if g.estate != estate {
        return Err(bad(format!("the paid step is estate {}'s; this box pulls as {estate:?}: nothing moved", g.estate)));
    }
    if g.until_unix_ms <= now_unix_ms {
        return Err(bad(format!("the paid step was good until {} (unix ms), its period is over: nothing moved", g.until_unix_ms)));
    }
    let main_gib = g.main_data_gib_per_member;
    if !twin && data_gib != main_gib {
        return Err(bad(format!("the order sizes main's disks to {data_gib} GiB, the paid step to {main_gib} GiB: nothing moved")));
    }
    if twin && data_gib < main_gib {
        return Err(bad(format!("the order sizes the twin's disk to {data_gib} GiB, under main's {main_gib} GiB a disk: it must hold main's whole set, nothing moved")));
    }
    Ok(PaidStep { reference: g.reference, main_gib })
}

/// A community build links no monetize verifier: a paid step cannot be checked.
#[cfg(not(feature = "metered"))]
pub fn verify_step(go_ahead: &str, key: Option<&[u8; 32]>, estate: &str, twin: bool, data_gib: u64, now_unix_ms: i64) -> Result<PaidStep, Refusal> {
    let _ = (go_ahead, key, estate, twin, now_unix_ms);
    Err(refuse(RefusalCode::BadOrder, format!("this community build carries no monetize verifier, so monetize's signed grow step to {data_gib} GiB cannot be checked: refused")))
}

/// **Every data disk already carries `data_gib`**: the set is complete and each
/// member spans it (within the grow machine's own 1 %).
pub fn carried(status: &gunnar_proto::GrowStatusResponse, data_gib: u64) -> bool {
    let target = data_gib.saturating_mul(SECTORS_PER_GIB);
    status.members_verdict == "complete" && !status.members.is_empty() && status.members.iter().all(|m| crate::grow::target_reached(m.set_bytes / 512, target))
}

/// **The twin holds main's set** at `main_gib` a disk over `main_members` disks: its
/// own growth is not under way and every twin disk spans the whole set. Otherwise
/// refused [`RefusalCode::NoRoom`], by name.
pub fn twin_holds(twin: &gunnar_proto::GrowStatusResponse, main_members: usize, main_gib: u64) -> Result<String, Refusal> {
    let whole = (main_members as u64).max(1).saturating_mul(main_gib);
    let no = |why: String| Err(refuse(RefusalCode::NoRoom, format!("main grows after its twin, and the twin {why}; main's set is {main_members} × {main_gib} GiB: nothing moved")));
    if !matches!(twin.phase.as_str(), "idle" | "done" | "failed") {
        return no(format!("is still growing ({}: {})", twin.phase, twin.detail));
    }
    if twin.members.is_empty() {
        return no(format!("reports no data disk ({})", twin.members_verdict));
    }
    if twin.members_verdict != "complete" {
        return no(format!("has an {} set", twin.members_verdict));
    }
    let short = twin.members.iter().map(|m| m.set_bytes >> 30).min().unwrap_or(0);
    if !carried(twin, whole) {
        return no(format!("holds {short} GiB a disk, under the {whole} GiB main's set needs"));
    }
    Ok(format!("holds {short} GiB a disk, main's whole set of {whole} GiB"))
}

/// The machine's status as the pulled growth reads it.
pub fn state_of(status: &gunnar_proto::GrowStatusResponse) -> GrowState {
    let said = if status.detail.is_empty() { status.phase.clone() } else { status.detail.clone() };
    match status.phase.as_str() {
        "waiting-members" => GrowState::WaitingMembers(said),
        "done" => GrowState::Done(said),
        "failed" => GrowState::Failed(said),
        "idle" => GrowState::Failed(format!("the grow machine is idle: {said}")),
        _ => GrowState::Running(said),
    }
}

/// **This box's grow machine, as a pulled order drives it.**
#[derive(Clone)]
pub struct PaidGrowth {
    machine: Arc<GrowMachine>,
    poll: Duration,
    ceiling: Duration,
}

impl PaidGrowth {
    pub fn new(machine: Arc<GrowMachine>) -> PaidGrowth {
        PaidGrowth {
            machine,
            poll: Duration::from_secs(2),
            ceiling: crate::grow::DEFAULT_DRAIN_TIMEOUT + crate::grow::DEFAULT_WAIT_MEMBERS_TIMEOUT + Duration::from_secs(300),
        }
    }

    /// How often [`PaidGrowth::settle`] looks, and for how long at most.
    pub fn pacing(mut self, poll: Duration, ceiling: Duration) -> PaidGrowth {
        self.poll = poll;
        self.ceiling = ceiling;
        self
    }

    /// Step 1: [`verify_step`], then whether the disks already carry it.
    pub fn check(&self, go_ahead: &str, data_gib: u64, estate: &str, key: Option<&[u8; 32]>, twin: bool, now_unix_ms: i64) -> Result<GrowCheck, Refusal> {
        let step = verify_step(go_ahead, key, estate, twin, data_gib, now_unix_ms)?;
        Ok(GrowCheck { twin, data_gib, reference: step.reference, carried: carried(&self.machine.status(), data_gib) })
    }

    /// The data disks of this box's own set (what main's twin must hold).
    pub fn members(&self) -> usize {
        self.machine.status().members.len()
    }

    /// Start the machine to `check.data_gib` a disk under a one-time `nonce`. The step
    /// was verified under the monetize key, so the go-ahead is minted here (G-D38).
    pub fn start(&self, check: &GrowCheck, nonce: &str, now_unix_ms: i64) -> Result<GrowState, Refusal> {
        let target = check.data_gib.saturating_mul(SECTORS_PER_GIB);
        let request = gunnar_proto::GrowStartRequest {
            target_sectors_per_member: target,
            go_ahead: crate::grow::GoAhead::operator(target, nonce, now_unix_ms),
            drain_timeout_secs: 0,
            wait_members_timeout_secs: 0,
        };
        let code = if self.machine.phase().is_active() { RefusalCode::AlreadyRunning } else { RefusalCode::BadOrder };
        self.machine.start(&request).map(|s| state_of(&s)).map_err(|why| refuse(code, format!("the grow machine refused monetize's signed grow step: {why}")))
    }

    /// Wait for the machine to stop moving: `waiting-members` (unless `past_waiting`),
    /// `done`, `failed` — or the ceiling, as it stands.
    pub async fn settle(&self, past_waiting: bool) -> GrowState {
        let until = tokio::time::Instant::now() + self.ceiling;
        loop {
            let state = state_of(&self.machine.status());
            match &state {
                GrowState::Done(_) | GrowState::Failed(_) => return state,
                GrowState::WaitingMembers(_) if !past_waiting => return state,
                _ if tokio::time::Instant::now() >= until => return state,
                _ => tokio::time::sleep(self.poll).await,
            }
        }
    }

    /// `GrowResume`: look at the members again (`give_up`: mount back what is there).
    pub fn look_again(&self, give_up: bool) -> Result<(), Refusal> {
        self.machine.resume(give_up).map(|_| ()).map_err(|why| refuse(RefusalCode::BadOrder, format!("the grow machine would not look again: {why}")))
    }
}

#[cfg(test)]
pub(crate) mod tests {
    use super::*;
    #[cfg(feature = "metered")]
    use crate::grow::tests::{carrying, Fake, Script};
    #[cfg(feature = "metered")]
    use crate::grow::world::{Grown, SurveyFacts};
    #[cfg(feature = "metered")]
    use std::collections::VecDeque;

    const GIB: u64 = 1 << 30;

    fn status(phase: &str, verdict: &str, set_gib: &[u64]) -> gunnar_proto::GrowStatusResponse {
        gunnar_proto::GrowStatusResponse {
            phase: phase.into(),
            members_verdict: verdict.into(),
            detail: format!("{phase}: as scripted"),
            members: set_gib.iter().enumerate().map(|(i, g)| gunnar_proto::GrowMember { index: i as u32, set_bytes: g * GIB, disk_bytes: g * GIB, ..Default::default() }).collect(),
            ..Default::default()
        }
    }

    #[cfg(feature = "metered")]
    pub(crate) mod signed {
        use ed25519_dalek::Signer;
        use super::super::{growth_message, EstateGrowth, GROWTH_SIGNER_SHOP};
        use monetize_embed::signing::base64_encode;

        pub const SHOP: [u8; 32] = [11u8; 32];

        pub fn shop_pub() -> [u8; 32] {
            ed25519_dalek::SigningKey::from_bytes(&SHOP).verifying_key().to_bytes()
        }

        pub fn step(seed: [u8; 32], estate: &str, gib: u64, until: i64) -> String {
            let mut g = EstateGrowth {
                v: 1,
                signer: GROWTH_SIGNER_SHOP.into(),
                estate: estate.into(),
                main_data_gib_per_member: gib,
                until_unix_ms: until,
                reference: "estate/acme/change-2".into(),
                issued_unix_ms: 1_000,
                signature: String::new(),
            };
            g.signature = base64_encode(&ed25519_dalek::SigningKey::from_bytes(&seed).sign(&growth_message(&g)).to_bytes());
            serde_json::to_string(&g).unwrap()
        }
    }

    /// G16-d: monetize's step verifies under the box's monetize key, for THIS estate,
    /// until its period ends, at the size THIS role is paid — or nothing moves.
    #[cfg(feature = "metered")]
    #[test]
    fn a_paid_step_verifies_under_the_monetize_key_for_this_estate_until_its_period_ends() {
        use signed::*;
        let key = shop_pub();
        let go = step(SHOP, "acme", 120, 50_000);
        let s = verify_step(&go, Some(&key), "acme", false, 120, 10_000).expect("main, its step");
        assert_eq!(s, PaidStep { reference: "estate/acme/change-2".into(), main_gib: 120 });
        verify_step(&go, Some(&key), "acme", true, 480, 10_000).expect("the twin, main's whole set");

        let no = |r: Result<PaidStep, Refusal>, says: &str| {
            let r = r.expect_err(says);
            assert_eq!(r.code, RefusalCode::BadOrder, "{r}");
            assert!(r.message.contains(says), "{says:?} in {r}");
        };
        no(verify_step(&step([12u8; 32], "acme", 120, 50_000), Some(&key), "acme", false, 120, 10_000), "not monetize's");
        no(verify_step(&step(SHOP, "globex", 120, 50_000), Some(&key), "acme", false, 120, 10_000), "globex");
        no(verify_step(&go, Some(&key), "acme", false, 120, 50_001), "until");
        no(verify_step(&go, Some(&key), "acme", false, 240, 10_000), "120");
        no(verify_step(&go, Some(&key), "acme", true, 60, 10_000), "whole set");
        no(verify_step(&go, None, "acme", false, 120, 10_000), "no monetize key");
        no(verify_step("{not json", Some(&key), "acme", false, 120, 10_000), "not monetize's");
    }

    /// The signed bytes are monetize's: sorted keys, no `signature`, no spaces.
    #[cfg(feature = "metered")]
    #[test]
    fn the_paid_steps_signed_bytes_are_monetizes_canonical_form() {
        let g = EstateGrowth { v: 1, signer: "shop".into(), estate: "acme".into(), main_data_gib_per_member: 120, until_unix_ms: 9_000, reference: "estate/acme/change-2".into(), issued_unix_ms: 7, signature: "x".into() };
        assert_eq!(
            String::from_utf8(growth_message(&g)).unwrap(),
            r#"{"estate":"acme","issued_unix_ms":7,"main_data_gib_per_member":120,"reference":"estate/acme/change-2","signer":"shop","until_unix_ms":9000,"v":1}"#
        );
    }

    /// A community build carries no monetize verifier: a paid step is refused by name.
    #[cfg(not(feature = "metered"))]
    #[test]
    fn a_community_build_refuses_a_paid_step_by_name() {
        let r = verify_step("{}", Some(&[1u8; 32]), "acme", false, 120, 0).unwrap_err();
        assert_eq!(r.code, RefusalCode::BadOrder);
        assert!(r.message.contains("community build"), "{r}");
    }

    /// Carried = a complete set whose every member spans the step (a hair under counts).
    #[test]
    fn a_step_is_carried_only_when_every_member_of_a_complete_set_spans_it() {
        assert!(carried(&status("idle", "complete", &[120, 120]), 120));
        let mut hair = status("idle", "complete", &[120, 120]);
        hair.members[1].set_bytes -= 67 * 512;
        assert!(carried(&hair, 120), "a partition start under is carried");
        assert!(!carried(&status("idle", "complete", &[120, 60]), 120));
        assert!(!carried(&status("idle", "complete", &[120, 120]), 240));
        assert!(!carried(&status("idle", "incomplete: member 1 of 2 is missing", &[120, 120]), 120));
        assert!(!carried(&status("idle", "none", &[]), 120), "no members, no claim");
    }

    /// The twin holds main's set when its own growth is not running and each twin disk
    /// spans main's members × main's step; otherwise NoRoom, saying what it has.
    #[test]
    fn the_twin_holds_mains_set_only_once_its_own_disks_span_it() {
        let ok = twin_holds(&status("done", "complete", &[480]), 4, 120).expect("480 GiB holds 4 × 120");
        assert!(ok.contains("480"), "{ok}");
        twin_holds(&status("idle", "complete", &[480]), 4, 120).expect("an idle twin at size");
        for (twin, says) in [
            (status("done", "complete", &[240]), "240"),
            (status("waiting-members", "complete", &[480]), "waiting-members"),
            (status("idle", "none", &[]), "no data disk"),
            (status("idle", "incomplete: member 1 of 2 is missing", &[480]), "incomplete"),
        ] {
            let r = twin_holds(&twin, 4, 120).expect_err(says);
            assert_eq!(r.code, RefusalCode::NoRoom, "{r}");
            assert!(r.message.contains(says) && r.message.contains("twin"), "{says:?} in {r}");
        }
    }

    #[test]
    fn the_machines_phase_reads_as_the_pulled_growth_needs_it() {
        assert_eq!(state_of(&status("draining", "", &[])), GrowState::Running("draining: as scripted".into()));
        assert_eq!(state_of(&status("waiting-members", "", &[])), GrowState::WaitingMembers("waiting-members: as scripted".into()));
        assert_eq!(state_of(&status("done", "", &[])), GrowState::Done("done: as scripted".into()));
        assert_eq!(state_of(&status("failed", "", &[])), GrowState::Failed("failed: as scripted".into()));
        assert!(matches!(state_of(&status("idle", "", &[])), GrowState::Failed(d) if d.contains("idle")), "a machine that stopped moving is not running");
    }

    #[cfg(feature = "metered")]
    fn larger_to(sectors: u64) -> SurveyFacts {
        let mut s = carrying(sectors);
        for m in &mut s.members {
            m.larger = true;
        }
        s.eligible = true;
        s.target_sectors = Some(sectors);
        s
    }

    /// End to end over the real grow machine: start (drain, unmount), wait for the
    /// members, monetize sizes them, look again, done — and a step already carried
    /// is done at once with nothing unmounted.
    #[cfg(feature = "metered")]
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn the_paid_step_drives_this_boxs_own_grow_machine_to_done() {
        use signed::*;
        let script = Script {
            in_flight: VecDeque::from([0]),
            surveys: VecDeque::from([carrying(60 * SECTORS_PER_GIB)]),
            grown: VecDeque::from([Ok(Grown { grew: true, array_bytes: 480 * GIB, refusal: None, xfs: "grown".into() })]),
            ..Script::default()
        };
        let world = Fake::new(script);
        let g = PaidGrowth::new(GrowMachine::new(world.clone())).pacing(Duration::from_millis(20), Duration::from_secs(20));
        assert_eq!(g.members(), 4);

        let now = crate::twin::now_ms();
        let go = step(SHOP, "acme", 120, now + 60_000);
        let c = g.check(&go, 120, "acme", Some(&shop_pub()), false, now).expect("verified");
        assert_eq!(c, GrowCheck { twin: false, data_gib: 120, reference: "estate/acme/change-2".into(), carried: false });
        assert!(matches!(g.start(&c, "pull-6-estate/acme/change-2", now).expect("started"), GrowState::Running(_)));
        assert!(matches!(g.settle(false).await, GrowState::WaitingMembers(_)), "/data is off, the members are awaited");
        assert!(g.start(&c, "pull-7-estate/acme/change-2", now).is_err(), "one growth at a time");

        world.script.lock().unwrap().surveys = VecDeque::from([larger_to(120 * SECTORS_PER_GIB)]);
        g.look_again(false).expect("resume applies while waiting");
        let end = g.settle(true).await;
        assert!(matches!(&end, GrowState::Done(d) if d.starts_with("done")), "{end:?}");

        // Carried now: the next go-ahead moves nothing.
        let c = g.check(&go, 120, "acme", Some(&shop_pub()), false, now).expect("verified");
        assert!(c.carried, "{c:?}");
    }
}