gunnar-server 3.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 super::super::{growth_message, EstateGrowth, GROWTH_SIGNER_SHOP};
        use ed25519_dalek::Signer;
        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:?}");
    }
}