From 171db535bed67e26a1efdedc56ef7da7f75b8d0b Mon Sep 17 00:00:00 2001 From: Nick Potts Date: Tue, 10 Nov 2026 12:22:11 +0000 Subject: [PATCH 1/3] Add the Volume approval, its verifier and the Machine's operation records (E2) A Device signs one VolumeApproval for a Move, Mirror, Mirror removal, Restore, resize or seed, under a new volume purpose that no session grant can sign. VolumeWorkflow::permits is the transition graph each Machine checks a step against. Each Machine keeps its Volume operations in the approvals log: open after a first step verifies inside its window, ended for good after, with recipient grants and withdrawals. A new owner of a copy ends the old one, and a Move's source with no decision records Resumed. Nothing calls these yet. --- core/crates/ployz-core/src/authority/keys.rs | 4 + core/crates/ployz-core/src/authority/mod.rs | 2 + .../ployz-core/src/authority/refusal.rs | 2 +- .../ployz-core/src/authority/restart.rs | 2 +- .../crates/ployz-core/src/authority/volume.rs | 327 +++++++++ .../ployz-core/src/authority/volume_tests.rs | 511 ++++++++++++++ core/crates/ployz-sdk/authority.d.ts | 2 + core/crates/ployz/src/sdk/payloads.rs | 1 + .../ployzd/src/machine/local_machine.rs | 3 + .../machine/local_machine/restart_approval.rs | 36 +- .../machine/local_machine/volume_approval.rs | 390 +++++++++- .../local_machine/volume_approval_tests.rs | 667 ++++++++++++++++++ core/crates/ployzd/src/machine_api/local.rs | 7 + .../approval-session/volume-approval.test.ts | 49 ++ 14 files changed, 1996 insertions(+), 7 deletions(-) create mode 100644 core/crates/ployz-core/src/authority/volume.rs create mode 100644 core/crates/ployz-core/src/authority/volume_tests.rs create mode 100644 core/crates/ployzd/src/machine/local_machine/volume_approval_tests.rs create mode 100644 dashboard/src/modules/approval-session/volume-approval.test.ts diff --git a/core/crates/ployz-core/src/authority/keys.rs b/core/crates/ployz-core/src/authority/keys.rs index c320f76fe7..45d632e5b5 100644 --- a/core/crates/ployz-core/src/authority/keys.rs +++ b/core/crates/ployz-core/src/authority/keys.rs @@ -27,6 +27,9 @@ pub enum Purpose { /// Standing authority over every future push, such as an automatic deploy rule. Only a /// Device signs one: a session grant covers workflows only. Rule, + /// One Volume workflow, which can stop a Service, start one elsewhere and send a Volume's + /// data to another Machine. Only a Device signs one: a session grant covers none. + Volume, } impl Purpose { @@ -37,6 +40,7 @@ impl Purpose { Self::Join => "join", Self::Enforce => "enforce", Self::Rule => "rule", + Self::Volume => "volume", } } } diff --git a/core/crates/ployz-core/src/authority/mod.rs b/core/crates/ployz-core/src/authority/mod.rs index ec6adb2736..52d52a90cb 100644 --- a/core/crates/ployz-core/src/authority/mod.rs +++ b/core/crates/ployz-core/src/authority/mod.rs @@ -20,6 +20,7 @@ pub mod session; pub mod starting_set; pub mod trusted_key; pub mod validity; +pub mod volume; pub use canonical::{CanonicalError, Timestamp}; pub use deploy::{ @@ -42,3 +43,4 @@ pub use trusted_key::{ Passkey, PasskeyAccount, PasskeyApprover, PasskeyAssertion, PasskeyPublicKey, TrustedKeys, TrustedPasskey, }; +pub use volume::{VolumeApproval, VolumeApproved, VolumeFacts, VolumeStep, VolumeWorkflow}; diff --git a/core/crates/ployz-core/src/authority/refusal.rs b/core/crates/ployz-core/src/authority/refusal.rs index c639175b8b..2c47d46bf9 100644 --- a/core/crates/ployz-core/src/authority/refusal.rs +++ b/core/crates/ployz-core/src/authority/refusal.rs @@ -25,7 +25,7 @@ pub enum AuthorityRefusal { Expired, #[error("this Machine holds no Sealing Key, so it cannot tell which Cluster was approved")] NoClusterIdentity, - #[error("operation {operation} already ran this step; a new restart signs a new approval")] + #[error("operation {operation} already ran this step; it needs a new approval")] Replayed { operation: OperationId }, #[error("this GitHub token already minted a Build Grant; a build asks for a new one")] ProofReplayed, diff --git a/core/crates/ployz-core/src/authority/restart.rs b/core/crates/ployz-core/src/authority/restart.rs index 8f3ed55da4..e2a6f4efa4 100644 --- a/core/crates/ployz-core/src/authority/restart.rs +++ b/core/crates/ployz-core/src/authority/restart.rs @@ -144,7 +144,7 @@ impl RestartApproval { pub use super::validity::{CLOCK_SKEW_MILLIS, MAX_LIFETIME_MILLIS}; -fn ordered_machines<'de, D: serde::Deserializer<'de>>( +pub(super) fn ordered_machines<'de, D: serde::Deserializer<'de>>( deserializer: D, ) -> Result, D::Error> { let machines = Vec::::deserialize(deserializer)?; diff --git a/core/crates/ployz-core/src/authority/volume.rs b/core/crates/ployz-core/src/authority/volume.rs new file mode 100644 index 0000000000..b7fa6ace7d --- /dev/null +++ b/core/crates/ployz-core/src/authority/volume.rs @@ -0,0 +1,327 @@ +//! The approval a person's Device signs for one Volume workflow, the steps each workflow +//! lets each Machine take, and the checks a Machine runs before it admits the first step. +//! +//! A workflow is a graph, not a list of routes: [`VolumeWorkflow::permits`] answers for +//! one Machine and one step. A Machine keeps its own record of the operations it admitted, +//! so a later step of an open operation needs no fresh window and an ended one never runs +//! again. That record is the Machine's, not this module's. + +use serde::{Deserialize, Serialize}; +use ts_rs::TS; + +use super::{ + canonical::Timestamp, + deploy::ApprovedService, + keys::{Approver, Purpose, Signed, SigningPublicKey, Statement}, + proposal::StepDigest, + refusal::AuthorityRefusal, + restart::{SignerKey, ordered_machines}, + trusted_key::TrustedKeys, +}; +use crate::{ + CopyTransition, DockerVolumeName, MachineId, Namespace, OperationId, + ProvisionedVolumeMaximumBytes, SealingPublicKey, SlotRole, SnapshotGuid, +}; + +/// A person approves one workflow on one Volume in one Cluster, under one operation whose +/// window bounds when its first step may start on each Machine. +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize, TS)] +#[serde(deny_unknown_fields)] +pub struct VolumeApproval { + /// The Cluster's Sealing public key, which every Machine holds and Cloud cannot mint. + pub cluster: SealingPublicKey, + pub operation: OperationId, + pub namespace: Namespace, + /// The Docker Volume the workflow acts on. A seed names the Volume it creates. + pub volume: DockerVolumeName, + pub workflow: VolumeWorkflow, + pub issued_at: Timestamp, + pub expires_at: Timestamp, + /// The Device that signed, which the signature must come from. + pub signer: SignerKey, +} + +impl Statement for VolumeApproval { + const PURPOSE: Purpose = Purpose::Volume; +} + +/// What a person approves, with every Machine it reaches by ID. +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize, TS)] +#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] +pub enum VolumeWorkflow { + /// The Writer moves from `from` to `to`, which starts `service`. The Mirrors in + /// `mirrors` keep following the new Writer. + Move { + from: MachineId, + to: MachineId, + service: ApprovedService, + /// Sorted and distinct. + #[serde(deserialize_with = "ordered_machines")] + mirrors: Vec, + }, + /// `mirror` keeps a copy of the Writer on `writer`, and every later sync of it, until + /// the Mirror is removed. With `replace` it first destroys the copy it holds. + Mirror { + writer: MachineId, + mirror: MachineId, + replace: bool, + }, + /// `mirror` destroys its copy, held in `role`. `last_copy` says no other copy remains + /// beside the Writer's. + MirrorRemoval { + mirror: MachineId, + writer: MachineId, + role: SlotRole, + last_copy: bool, + }, + /// `machine` promotes its copy at `at` with no start. Writes after `loses_after` are + /// lost. + Restore { + machine: MachineId, + at: SnapshotGuid, + loses_after: Timestamp, + }, + /// Each of `machines` bounds the Volume at `maximum_bytes`, below the bound it holds + /// only when `shrink`. + Resize { + /// Sorted and distinct. + #[serde(deserialize_with = "ordered_machines")] + machines: Vec, + maximum_bytes: ProvisionedVolumeMaximumBytes, + shrink: bool, + }, + /// `machine` creates the approval's Volume from `source` at snapshot `at`. + Seed { + machine: MachineId, + source: DockerVolumeName, + at: SnapshotGuid, + }, +} + +/// One step a Machine is asked to take on a Volume, as its request reads. +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum VolumeStep { + /// The `SetVolumeCopy` transition the planner chose for the copy this Machine holds. + Copy(CopyTransition), + /// Starting a Container on the Volume, after a Writer want's transition. + Start, + Seed { + source: DockerVolumeName, + at: SnapshotGuid, + }, + Resize { + maximum_bytes: ProvisionedVolumeMaximumBytes, + allow_shrink: bool, + }, +} + +impl VolumeWorkflow { + /// Whether this workflow lets `machine` take `step`. A step outside the graph is + /// refused however valid the signature. + #[must_use] + pub fn permits(&self, machine: &MachineId, step: &VolumeStep) -> bool { + use CopyTransition::{ + MirrorToForget, MirrorToMirror, MirrorToWriter, NoneToMirror, WriterToMirror, + WriterToSnapshot, WriterToWriter, + }; + match (self, step) { + ( + Self::Move { + from, to, mirrors, .. + }, + VolumeStep::Copy(transition), + ) => { + (machine == from + && matches!( + transition, + WriterToSnapshot | WriterToMirror | WriterToWriter + )) + || (machine == to + && matches!(transition, NoneToMirror | MirrorToMirror | MirrorToWriter)) + || (mirrors.contains(machine) && *transition == MirrorToMirror) + } + (Self::Move { from, to, .. }, VolumeStep::Start) => machine == from || machine == to, + ( + Self::Mirror { + writer, + mirror, + replace, + }, + VolumeStep::Copy(transition), + ) => { + (machine == writer && *transition == WriterToSnapshot) + || (machine == mirror + && (matches!(transition, NoneToMirror | MirrorToMirror) + || (*replace && *transition == MirrorToForget))) + } + (Self::MirrorRemoval { mirror, .. }, VolumeStep::Copy(transition)) => { + machine == mirror && *transition == MirrorToForget + } + (Self::Restore { machine: owner, .. }, VolumeStep::Copy(transition)) => { + machine == owner && *transition == MirrorToWriter + } + ( + Self::Resize { + machines, + maximum_bytes, + shrink, + }, + VolumeStep::Resize { + maximum_bytes: wanted, + allow_shrink, + }, + ) => { + machines.contains(machine) && wanted == maximum_bytes && (*shrink || !allow_shrink) + } + ( + Self::Seed { + machine: owner, + source, + at, + }, + VolumeStep::Seed { + source: wanted, + at: wanted_at, + }, + ) => machine == owner && wanted == source && wanted_at == at, + _ => false, + } + } + + /// Whether this workflow changes the role of `machine`'s copy, or destroys it. A + /// Machine holds at most one open owner of a Volume's copy. + #[must_use] + pub fn owns(&self, machine: &MachineId) -> bool { + match self { + Self::Move { from, to, .. } => machine == from || machine == to, + Self::Mirror { mirror, .. } | Self::MirrorRemoval { mirror, .. } => machine == mirror, + Self::Restore { machine: owner, .. } => machine == owner, + Self::Resize { .. } | Self::Seed { .. } => false, + } + } + + /// Whether this workflow sends the Volume to `recipient` as a Mirror: a Mirror's own, + /// or one a Move keeps following. + #[must_use] + pub fn mirrors_to(&self, recipient: &MachineId) -> bool { + match self { + Self::Mirror { mirror, .. } => mirror == recipient, + Self::Move { mirrors, .. } => mirrors.contains(recipient), + Self::MirrorRemoval { .. } + | Self::Restore { .. } + | Self::Resize { .. } + | Self::Seed { .. } => false, + } + } + + /// Whether `machine` is the source of a Move under this workflow. + #[must_use] + pub fn moves_from(&self, machine: &MachineId) -> bool { + matches!(self, Self::Move { from, .. } if from == machine) + } + + fn names(&self, machine: &MachineId) -> bool { + match self { + Self::Move { + from, to, mirrors, .. + } => machine == from || machine == to || mirrors.contains(machine), + Self::Mirror { writer, mirror, .. } | Self::MirrorRemoval { mirror, writer, .. } => { + machine == writer || machine == mirror + } + Self::Restore { machine: owner, .. } | Self::Seed { machine: owner, .. } => { + machine == owner + } + Self::Resize { machines, .. } => machines.contains(machine), + } + } + + /// A Move to itself, or a Mirror of its own Writer, would let one Machine take both + /// sides of the graph. + fn distinct(&self) -> bool { + match self { + Self::Move { + from, to, mirrors, .. + } => from != to && !mirrors.contains(from) && !mirrors.contains(to), + Self::Mirror { writer, mirror, .. } | Self::MirrorRemoval { mirror, writer, .. } => { + writer != mirror + } + Self::Restore { .. } | Self::Resize { .. } | Self::Seed { .. } => true, + } + } +} + +/// What a Machine knows when it checks a Volume approval against one step. +#[derive(Clone, Debug)] +pub struct VolumeFacts<'a> { + /// The Sealing public key this Machine holds, or `None` while it holds none. + pub cluster: Option, + pub machine: &'a MachineId, + pub trusted: &'a TrustedKeys, + pub volume: &'a DockerVolumeName, + pub step: &'a VolumeStep, + pub now: Timestamp, +} + +/// A Volume approval a Machine checked: its body, the Device that signed it, and the +/// digest of the body as signed, which the Machine's record keeps. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct VolumeApproved { + pub body: VolumeApproval, + pub by: SigningPublicKey, + pub digest: StepDigest, +} + +/// Check `approval` lets this Machine take `facts.step` now, signed by a Device this +/// Machine trusts. A browser key is refused even under a valid session grant, and so is an +/// envelope that rides a proposal. +/// +/// # Errors +/// Returns the first [`AuthorityRefusal`] found: a signature problem, a malformed body, a +/// signer that is not a trusted Device, a body naming another Cluster or Volume, a +/// workflow that does not name this Machine or does not permit the step, an invalid +/// lifetime, or an expired window. An ended operation is the Machine's record to refuse. +pub fn verify( + approval: &Signed, + facts: &VolumeFacts<'_>, +) -> Result { + let approvers = approval.approvers(facts.trusted, facts.now)?; + let body = approval.open()?; + if approval.proposal.is_some() { + return Err(AuthorityRefusal::Mismatch { + field: "a proposal", + }); + } + let by = approvers + .into_iter() + .find_map(|approver| match approver { + Approver::Device(key) if approver.signs_as(body.signer) => Some(key), + Approver::Device(_) | Approver::Browser { .. } => None, + }) + .ok_or(AuthorityRefusal::Mismatch { + field: "the signer", + })?; + let cluster = facts.cluster.ok_or(AuthorityRefusal::NoClusterIdentity)?; + let checks = [ + ("a Cluster", body.cluster == cluster), + ("a Volume", &body.volume == facts.volume), + ("Machines", body.workflow.distinct()), + ("this Server", body.workflow.names(facts.machine)), + ( + "a Volume step", + body.workflow.permits(facts.machine, facts.step), + ), + ]; + if let Some((field, _)) = checks.iter().find(|(_, holds)| !holds) { + return Err(AuthorityRefusal::Mismatch { field }); + } + super::validity::verify(body.issued_at, body.expires_at, facts.now)?; + Ok(VolumeApproved { + body, + by, + digest: StepDigest::of(&approval.body), + }) +} + +#[cfg(test)] +#[path = "volume_tests.rs"] +mod tests; diff --git a/core/crates/ployz-core/src/authority/volume_tests.rs b/core/crates/ployz-core/src/authority/volume_tests.rs new file mode 100644 index 0000000000..24ffed957c --- /dev/null +++ b/core/crates/ployz-core/src/authority/volume_tests.rs @@ -0,0 +1,511 @@ +use std::num::NonZeroU64; + +use super::*; +use crate::{ + DeviceKey, ServiceName, + authority::{ + canonical, + deploy::PayloadDigest, + keys::Signer, + proposal::{ProposalApproval, ProposalStep, StepKind}, + session::{ActionFamily, Delegation, GrantBody, SessionGrant}, + trusted_key::tests::{Answer, Authenticator, HONEST, trusting}, + }, +}; + +const OPERATION: &str = "0f8fad5bd9cb469fa16570867728950e"; + +fn signer() -> Signer { + Signer::from_seed(&[7; 32]) +} + +fn id(fill: char) -> MachineId { + MachineId::parse(fill.to_string().repeat(32)).unwrap() +} + +fn a() -> MachineId { + id('a') +} + +fn b() -> MachineId { + id('b') +} + +fn c() -> MachineId { + id('c') +} + +fn volume() -> DockerVolumeName { + DockerVolumeName::parse("app_data").unwrap() +} + +fn cluster() -> SealingPublicKey { + SealingPublicKey::from_bytes([9; 32]) +} + +fn bytes(gb: u64) -> ProvisionedVolumeMaximumBytes { + ProvisionedVolumeMaximumBytes::new(NonZeroU64::new(gb * 1_000_000_000).unwrap()) +} + +fn at(millis: u64) -> Timestamp { + Timestamp::from_millis(millis).unwrap() +} + +fn moving() -> VolumeWorkflow { + VolumeWorkflow::Move { + from: a(), + to: b(), + service: ApprovedService { + name: ServiceName::parse("db").unwrap(), + digest: PayloadDigest::try_from("d".repeat(64)).unwrap(), + image: None, + }, + mirrors: vec![c()], + } +} + +fn workflows() -> [VolumeWorkflow; 6] { + [ + moving(), + VolumeWorkflow::Mirror { + writer: a(), + mirror: b(), + replace: false, + }, + VolumeWorkflow::MirrorRemoval { + mirror: b(), + writer: a(), + role: SlotRole::Mirror, + last_copy: true, + }, + VolumeWorkflow::Restore { + machine: b(), + at: SnapshotGuid::new(18_446_744_073_709_551_615), + loses_after: at(500_000), + }, + VolumeWorkflow::Resize { + machines: vec![a(), b()], + maximum_bytes: bytes(20), + shrink: false, + }, + VolumeWorkflow::Seed { + machine: a(), + source: DockerVolumeName::parse("app_old").unwrap(), + at: SnapshotGuid::new(42), + }, + ] +} + +fn approval(workflow: VolumeWorkflow) -> VolumeApproval { + VolumeApproval { + cluster: cluster(), + operation: OperationId::parse(OPERATION).unwrap(), + namespace: Namespace::parse("app").unwrap(), + volume: volume(), + workflow, + issued_at: at(100_000), + expires_at: at(1_000_000), + signer: SignerKey::Device(DeviceKey::from(signer().public())), + } +} + +fn trusted() -> TrustedKeys { + TrustedKeys { + devices: vec![DeviceKey::from(signer().public())], + passkeys: Vec::new(), + } +} + +fn check( + signed: &Signed, + machine: &MachineId, + step: &VolumeStep, + now: u64, +) -> Result { + verify( + signed, + &VolumeFacts { + cluster: Some(cluster()), + machine, + trusted: &trusted(), + volume: &volume(), + step, + now: at(now), + }, + ) +} + +fn cut() -> VolumeStep { + VolumeStep::Copy(CopyTransition::WriterToMirror) +} + +fn mismatch(field: &'static str) -> Result { + Err(AuthorityRefusal::Mismatch { field }) +} + +#[test] +fn each_workflow_signs_its_golden_canonical_body() { + let head = r#"{"cluster":"CQkJCQkJCQkJCQkJCQkJCQkJCQkJCQkJCQkJCQkJCQk=","expires_at":1000000,"issued_at":100000,"namespace":"app","operation":"0f8fad5bd9cb469fa16570867728950e","signer":"ea4a6c63e29c520abef5507b132ec5f9954776aebebe7b92421eea691446d22c","volume":"app_data","workflow":"#; + let (a, b, c) = ("a".repeat(32), "b".repeat(32), "c".repeat(32)); + let d = "d".repeat(64); + let golden = [ + format!( + r#"{{"from":"{a}","kind":"move","mirrors":["{c}"],"service":{{"digest":"{d}","name":"db"}},"to":"{b}"}}"# + ), + format!(r#"{{"kind":"mirror","mirror":"{b}","replace":false,"writer":"{a}"}}"#), + format!( + r#"{{"kind":"mirror_removal","last_copy":true,"mirror":"{b}","role":"mirror","writer":"{a}"}}"# + ), + format!( + r#"{{"at":"18446744073709551615","kind":"restore","loses_after":500000,"machine":"{b}"}}"# + ), + format!( + r#"{{"kind":"resize","machines":["{a}","{b}"],"maximum_bytes":20000000000,"shrink":false}}"# + ), + format!(r#"{{"at":"42","kind":"seed","machine":"{a}","source":"app_old"}}"#), + ]; + for (workflow, golden) in workflows().into_iter().zip(golden) { + let signed = Signed::sign(&approval(workflow.clone()), &signer()).unwrap(); + assert_eq!(signed.body, format!("{head}{golden}}}"), "{workflow:?}"); + assert_eq!(signed.open().unwrap(), approval(workflow)); + assert!( + signed + .signing_input() + .starts_with(b"ployz-authority/v1\nvolume\n") + ); + } +} + +#[test] +fn a_device_approval_admits_each_step_its_graph_names_on_each_machine() { + use CopyTransition::*; + let moved = Signed::sign(&approval(moving()), &signer()).unwrap(); + for (machine, step) in [ + (a(), VolumeStep::Copy(WriterToSnapshot)), + (a(), cut()), + (a(), VolumeStep::Copy(WriterToWriter)), + (a(), VolumeStep::Start), + (b(), VolumeStep::Copy(NoneToMirror)), + (b(), VolumeStep::Copy(MirrorToMirror)), + (b(), VolumeStep::Copy(MirrorToWriter)), + (b(), VolumeStep::Start), + (c(), VolumeStep::Copy(MirrorToMirror)), + ] { + let approved = check(&moved, &machine, &step, 200_000).unwrap(); + assert_eq!(approved.by, signer().public()); + assert_eq!(approved.digest, StepDigest::of(&moved.body)); + } + for (machine, step) in [ + (a(), VolumeStep::Copy(MirrorToWriter)), + (a(), VolumeStep::Copy(WriterToForget)), + (b(), cut()), + (b(), VolumeStep::Copy(MirrorToForget)), + (c(), VolumeStep::Copy(MirrorToWriter)), + (c(), VolumeStep::Start), + ( + a(), + VolumeStep::Resize { + maximum_bytes: bytes(20), + allow_shrink: false, + }, + ), + ] { + assert_eq!( + check(&moved, &machine, &step, 200_000), + mismatch("a Volume step"), + "{machine} {step:?}" + ); + } +} + +#[test] +fn the_other_workflows_permit_only_their_own_steps() { + use CopyTransition::*; + let [_, mirror, removal, restore, resize, seed] = workflows(); + let shrink = |allow_shrink| VolumeStep::Resize { + maximum_bytes: bytes(20), + allow_shrink, + }; + let seeding = |source: &str, guid| VolumeStep::Seed { + source: DockerVolumeName::parse(source).unwrap(), + at: SnapshotGuid::new(guid), + }; + for (workflow, machine, step, permitted) in [ + (&mirror, a(), VolumeStep::Copy(WriterToSnapshot), true), + (&mirror, b(), VolumeStep::Copy(NoneToMirror), true), + (&mirror, b(), VolumeStep::Copy(MirrorToMirror), true), + (&mirror, b(), VolumeStep::Copy(MirrorToForget), false), + (&mirror, b(), VolumeStep::Copy(MirrorToWriter), false), + (&mirror, a(), cut(), false), + (&mirror, b(), VolumeStep::Start, false), + (&removal, b(), VolumeStep::Copy(MirrorToForget), true), + (&removal, a(), VolumeStep::Copy(WriterToForget), false), + (&restore, b(), VolumeStep::Copy(MirrorToWriter), true), + (&restore, b(), VolumeStep::Start, false), + (&resize, a(), shrink(false), true), + (&resize, b(), shrink(true), false), + (&seed, a(), seeding("app_old", 42), true), + (&seed, a(), seeding("other", 42), false), + (&seed, a(), seeding("app_old", 43), false), + ] { + assert_eq!( + workflow.permits(&machine, &step), + permitted, + "{workflow:?} {machine} {step:?}" + ); + } + let replacing = VolumeWorkflow::Mirror { + writer: a(), + mirror: b(), + replace: true, + }; + assert!(replacing.permits(&b(), &VolumeStep::Copy(MirrorToForget))); + let resize_other = VolumeStep::Resize { + maximum_bytes: bytes(30), + allow_shrink: false, + }; + assert!(!resize.permits(&a(), &resize_other)); +} + +#[test] +fn each_refusal_names_what_the_approval_lacks() { + let signed = Signed::sign(&approval(moving()), &signer()).unwrap(); + let other_volume = DockerVolumeName::parse("app_other").unwrap(); + assert_eq!( + verify( + &signed, + &VolumeFacts { + cluster: Some(cluster()), + machine: &a(), + trusted: &trusted(), + volume: &other_volume, + step: &cut(), + now: at(200_000), + }, + ), + mismatch("a Volume") + ); + assert_eq!( + verify( + &signed, + &VolumeFacts { + cluster: Some(SealingPublicKey::from_bytes([8; 32])), + machine: &a(), + trusted: &trusted(), + volume: &volume(), + step: &cut(), + now: at(200_000), + }, + ), + mismatch("a Cluster") + ); + assert_eq!( + check(&signed, &id('e'), &cut(), 200_000), + mismatch("this Server") + ); + assert_eq!( + check(&signed, &b(), &cut(), 200_000), + mismatch("a Volume step") + ); + + let mut changed = signed.clone(); + changed.body = changed.body.replace(&"b".repeat(32), &"e".repeat(32)); + assert_eq!( + check(&changed, &id('e'), &VolumeStep::Start, 200_000), + Err(AuthorityRefusal::BadSignature { + key: signer().public() + }) + ); + + let stranger = Signer::from_seed(&[3; 32]); + assert_eq!( + check( + &Signed::sign(&approval(moving()), &stranger).unwrap(), + &a(), + &cut(), + 200_000 + ), + Err(AuthorityRefusal::UnknownKey { + key: stranger.public() + }) + ); + + let mut forged = signed.clone(); + let [signature] = forged.signatures.as_mut_slice() else { + panic!("one signature"); + }; + signature.signature[0] ^= 1; + assert_eq!( + check(&forged, &a(), &cut(), 200_000), + Err(AuthorityRefusal::BadSignature { + key: signer().public() + }) + ); + + let mut unsigned = signed.clone(); + unsigned.signatures.clear(); + assert_eq!( + check(&unsigned, &a(), &cut(), 200_000), + Err(AuthorityRefusal::Unsigned { + purpose: Purpose::Volume + }) + ); + + let mut to_itself = approval(moving()); + let VolumeWorkflow::Move { to, .. } = &mut to_itself.workflow else { + unreachable!("a Move") + }; + *to = a(); + assert_eq!( + check( + &Signed::sign(&to_itself, &signer()).unwrap(), + &a(), + &cut(), + 200_000 + ), + mismatch("Machines") + ); +} + +#[test] +fn the_window_bounds_the_first_step_with_clock_skew() { + let signed = Signed::sign(&approval(moving()), &signer()).unwrap(); + let skew = crate::authority::validity::CLOCK_SKEW_MILLIS; + assert!(check(&signed, &a(), &cut(), 1_000_000 + skew).is_ok()); + assert_eq!( + check(&signed, &a(), &cut(), 1_000_001 + skew), + Err(AuthorityRefusal::Expired) + ); + let early = VolumeApproval { + issued_at: at(300_000), + ..approval(moving()) + }; + assert_eq!( + check( + &Signed::sign(&early, &signer()).unwrap(), + &a(), + &cut(), + 299_999 - skew + ), + mismatch("an issue time") + ); + let long = VolumeApproval { + expires_at: at(100_000 + crate::authority::validity::MAX_LIFETIME_MILLIS + 1), + ..approval(moving()) + }; + assert_eq!( + check( + &Signed::sign(&long, &signer()).unwrap(), + &a(), + &cut(), + 200_000 + ), + mismatch("a lifetime") + ); +} + +#[test] +fn a_browser_key_under_a_grant_of_every_family_cannot_sign_a_volume_approval() { + let browser = Signer::from_seed(&[4; 32]); + let passkey = Authenticator::es256(4); + let grant = GrantBody { + browser_key: browser.public(), + cluster: cluster(), + environment: Namespace::parse("app").unwrap(), + environment_id: uuid::Uuid::from_u128(1), + machines: vec![a(), b(), c()], + families: vec![ + ActionFamily::Deploy, + ActionFamily::Restart, + ActionFamily::Settings, + ActionFamily::Start, + ActionFamily::Stop, + ], + issued_at: at(0), + expires_at: at(3_000_000), + delegation: Delegation::None, + }; + let body = canonical::to_canonical(&grant).unwrap(); + let assertion = passkey.answer(&Answer { + challenge: &GrantBody::challenge(&body), + ..HONEST + }); + let browser_approval = VolumeApproval { + signer: SignerKey::Browser { + browser: browser.public(), + }, + ..approval(moving()) + }; + let signed = Signed::sign_under( + &browser_approval, + &browser, + SessionGrant { body, assertion }, + ) + .unwrap(); + let trusted = trusting(&passkey); + assert!( + signed.approvers(&trusted, at(200_000)).is_ok(), + "the grant itself verifies" + ); + assert_eq!( + verify( + &signed, + &VolumeFacts { + cluster: Some(cluster()), + machine: &a(), + trusted: &trusted, + volume: &volume(), + step: &cut(), + now: at(200_000), + }, + ), + mismatch("the signer") + ); +} + +#[test] +fn a_signed_approval_riding_a_proposal_is_refused() { + let mut signed = Signed::sign(&approval(moving()), &signer()).unwrap(); + let bundle = ProposalApproval { + cluster: cluster(), + environment: Namespace::parse("app").unwrap(), + operation: OperationId::parse(OPERATION).unwrap(), + steps: vec![ProposalStep { + index: 0, + kind: StepKind::Deploy, + digest: StepDigest::of(&signed.body), + }], + issued_at: at(100_000), + expires_at: at(1_000_000), + signer: SignerKey::Device(DeviceKey::from(signer().public())), + }; + signed.proposal = Some(Box::new(Signed::sign(&bundle, &signer()).unwrap())); + assert_eq!( + check(&signed, &a(), &cut(), 200_000), + mismatch("a proposal") + ); +} + +#[test] +fn a_field_this_machine_cannot_read_is_malformed() { + let mut signed = Signed::sign(&approval(moving()), &signer()).unwrap(); + signed.body = signed + .body + .replace(r#""kind":"move","#, r#""kind":"move","keep":true,"#); + assert!(matches!( + signed.open(), + Err(AuthorityRefusal::Malformed { + purpose: Purpose::Volume, + .. + }) + )); + let mut unsorted = Signed::sign(&approval(workflows()[4].clone()), &signer()).unwrap(); + let (a, b) = ("a".repeat(32), "b".repeat(32)); + unsorted.body = unsorted + .body + .replace(&format!(r#"["{a}","{b}"]"#), &format!(r#"["{b}","{a}"]"#)); + assert!(matches!( + unsorted.open(), + Err(AuthorityRefusal::Malformed { .. }) + )); +} diff --git a/core/crates/ployz-sdk/authority.d.ts b/core/crates/ployz-sdk/authority.d.ts index 6dfe0908e9..d13ade1aae 100644 --- a/core/crates/ployz-sdk/authority.d.ts +++ b/core/crates/ployz-sdk/authority.d.ts @@ -18,6 +18,8 @@ export type { SignerKey, StepDigest, StepKind, + VolumeApproval, + VolumeWorkflow, } from "./generated/payloads"; import type { ApprovedChange, DeployTerms, PlanView } from "./generated/payloads"; diff --git a/core/crates/ployz/src/sdk/payloads.rs b/core/crates/ployz/src/sdk/payloads.rs index d146787e2d..d82f2c0ba1 100644 --- a/core/crates/ployz/src/sdk/payloads.rs +++ b/core/crates/ployz/src/sdk/payloads.rs @@ -93,6 +93,7 @@ fn typescript_declarations() -> String { declarations.add::(); declarations.add::(); declarations.add::(); + declarations.add::(); declarations.add::(); declarations.add::(); declarations.add::(); diff --git a/core/crates/ployzd/src/machine/local_machine.rs b/core/crates/ployzd/src/machine/local_machine.rs index 863e3b45bd..a6dd08ece6 100644 --- a/core/crates/ployzd/src/machine/local_machine.rs +++ b/core/crates/ployzd/src/machine/local_machine.rs @@ -216,6 +216,9 @@ pub enum Error { /// or its `SeedVolume`. The refusal names `capability` in its details. #[error("{}", volume_approval::refusal_text(.capability))] VolumeRefused { capability: &'static str }, + /// A Volume step that no Volume approval admits on this Machine. + #[error("This Server refused the Volume step: {0}")] + VolumeStepRefused(ployz_core::authority::AuthorityRefusal), #[error("ployz.machine.join.device-approval.v1 needs a Device-signed join approval: {0}")] JoinRefused(ployz_core::authority::AuthorityRefusal), #[error("{SET_DEVICE_ENFORCEMENT_CAPABILITY} needs a Device-signed starting set: {0}")] diff --git a/core/crates/ployzd/src/machine/local_machine/restart_approval.rs b/core/crates/ployzd/src/machine/local_machine/restart_approval.rs index 41174894f5..a62901e29c 100644 --- a/core/crates/ployzd/src/machine/local_machine/restart_approval.rs +++ b/core/crates/ployzd/src/machine/local_machine/restart_approval.rs @@ -22,7 +22,7 @@ use ployz_core::{ use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; -use super::{Error, LocalMachine}; +use super::{Error, LocalMachine, volume_approval::VolumeRecords}; const FILE_NAME: &str = "restart-approvals.json"; @@ -242,6 +242,9 @@ struct Consumption { /// The hook each rule's Service last ran here, one per push. #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] rule_hooks: BTreeMap, + /// Volume operations, grants and withdrawals, none ever pruned. + #[serde(default, skip_serializing_if = "VolumeRecords::is_empty")] + volumes: VolumeRecords, #[serde(flatten)] operations: BTreeMap, } @@ -856,6 +859,37 @@ impl ConsumptionLog { Ok(guard) } + /// Apply `change` to this Machine's Volume records at `now`, raised to the log's time + /// floor, and keep the result, synced, before returning its answer. A refusal keeps the + /// records as they were, and an unchanged record is not written. + pub(super) fn change_volumes( + &self, + now: Timestamp, + change: impl FnOnce(&mut VolumeRecords, Timestamp) -> Result, + ) -> Result { + let mut guard = self.loaded()?; + let consumption = guard.as_ref().expect("loaded above"); + let now = consumption + .expiry_time_floor + .map_or(now, |floor| now.max(floor)); + let mut updated = consumption.clone(); + let answer = change(&mut updated.volumes, now)?; + if updated.volumes == consumption.volumes { + return Ok(answer); + } + updated.expiry_time_floor = Some(now); + self.write(&mut guard, updated)?; + Ok(answer) + } + + pub(super) fn read_volumes( + &self, + read: impl FnOnce(&VolumeRecords) -> R, + ) -> Result { + let guard = self.loaded()?; + Ok(read(&guard.as_ref().expect("loaded above").volumes)) + } + fn write( &self, guard: &mut std::sync::MutexGuard<'_, Option>, diff --git a/core/crates/ployzd/src/machine/local_machine/volume_approval.rs b/core/crates/ployzd/src/machine/local_machine/volume_approval.rs index 59808a5e3c..fbe4ca8785 100644 --- a/core/crates/ployzd/src/machine/local_machine/volume_approval.rs +++ b/core/crates/ployzd/src/machine/local_machine/volume_approval.rs @@ -1,14 +1,136 @@ -//! The Volume verbs an enforcing Machine refuses from an Operator caller. A Volume -//! approval a Device signs replaces each refusal; until then a Full caller still runs them. +//! The Volume verbs an enforcing Machine refuses from an Operator caller, and this +//! Machine's own record of the Volume operations a Device approved: which are open, which +//! ended, who may pull what, and which recipients were withdrawn. +//! +//! An operation is admitted once inside its window and then runs on its record, so the +//! work a person started finishes after the window. An ended operation never runs again. +//! Each record is synced before the effect it guards, and none reads a clock again. + +use std::collections::BTreeMap; use ployz_core::{ - SEED_REFUSED_BY_ENFORCEMENT, SEED_VOLUME_CAPABILITY, SET_VOLUME_COPY_CAPABILITY, - VOLUME_START_REFUSED_BY_ENFORCEMENT, + DockerVolumeName, MachineId, OperationId, SEED_REFUSED_BY_ENFORCEMENT, SEED_VOLUME_CAPABILITY, + SET_VOLUME_COPY_CAPABILITY, Sealing, SnapshotName, VOLUME_START_REFUSED_BY_ENFORCEMENT, + authority::{ + AuthorityRefusal, Signed, StepDigest, Timestamp, VolumeApproval, VolumeFacts, VolumeStep, + VolumeWorkflow, volume, + }, }; +use serde::{Deserialize, Serialize}; use super::{Error, LocalMachine}; impl LocalMachine { + /// Admit `step` on `volume` under `approval`. An operation this Machine ended is + /// refused for good. An operation open here passes on its record, at any time, for any + /// step its workflow permits this Machine. Otherwise the approval must verify inside its + /// window, and the operation is recorded open, synced, before this returns. Admitting + /// an operation that owns this Machine's copy ends the one that owned it before. + /// + /// # Errors + /// Returns [`Error::VolumeStepRefused`] naming what the approval lacks, an operation + /// already ended here as [`AuthorityRefusal::Replayed`], and [`Error::ApprovalRecord`] + /// when the record cannot be read or written. + pub fn admit_volume_step( + &self, + approval: &Signed, + volume: &DockerVolumeName, + step: &VolumeStep, + ) -> Result { + let record = self.record(); + let machine = record.id(); + let trusted = record.trusted_keys(); + let cluster = match self.held_sealing(record.phase()) { + Some(Sealing::Held { public_key }) => Some(public_key), + _ => None, + }; + let operation = approval.open().map_err(Error::VolumeStepRefused)?.operation; + let digest = StepDigest::of(&approval.body); + self.approvals + .change_volumes(Timestamp::now(), |records, now| { + records.admit(operation, digest, &machine, volume, step, || { + volume::verify( + approval, + &VolumeFacts { + cluster, + machine: &machine, + trusted: &trusted, + volume, + step, + now, + }, + ) + .map(|approved| approved.body.workflow) + }) + }) + .inspect(|()| tracing::info!(%operation, %volume, ?step, "Volume step admitted"))?; + Ok(operation) + } + + /// End `operation` on this Machine, with its grants. It is refused from then on. + /// + /// # Errors + /// Returns [`Error::ApprovalRecord`] when the record cannot be read or written. + pub fn end_volume_operation(&self, operation: OperationId) -> Result<(), Error> { + self.approvals + .change_volumes(Timestamp::now(), |records, _| { + records.end(operation); + Ok(()) + }) + } + + /// Let `recipient` pull `scope` of `operation`'s Volume from this Machine while the + /// operation is open. A grant merges into the one the operation already holds there. + /// + /// # Errors + /// Returns [`Error::VolumeStepRefused`] for an operation not open here, a recipient + /// its workflow does not send to, or one withdrawn since this Machine admitted it, and + /// [`Error::ApprovalRecord`] when the record cannot be read or written. + pub fn grant_volume( + &self, + operation: OperationId, + recipient: MachineId, + scope: GrantScope, + ) -> Result<(), Error> { + self.approvals + .change_volumes(Timestamp::now(), |records, _| { + records.grant(operation, recipient, scope) + }) + } + + /// Whether a live grant lets `recipient` pull `snapshot` of `volume` from here. + /// + /// # Errors + /// Returns [`Error::ApprovalRecord`] when the record cannot be read. + pub fn volume_grant_live( + &self, + volume: &DockerVolumeName, + recipient: &MachineId, + snapshot: &SnapshotName, + ) -> Result { + self.approvals + .read_volumes(|records| records.live(volume, recipient, snapshot)) + } + + /// Withdraw `recipient` from `volume` here, before anything is destroyed: every Mirror + /// operation this Machine admitted for it ends, and so does every grant to it. It needs + /// no approval, because it only narrows what a driver can do. An approval this Machine + /// never admitted is still new work while its window lasts. + /// + /// # Errors + /// Returns [`Error::ApprovalRecord`] when the record cannot be read or written. + pub fn withdraw_volume( + &self, + volume: &DockerVolumeName, + recipient: &MachineId, + ) -> Result<(), Error> { + let machine = self.record().id(); + self.approvals + .change_volumes(Timestamp::now(), |records, _| { + records.withdraw(&machine, volume, recipient); + Ok(()) + }) + } /// Under Device enforcement an Operator's `SetVolumeCopy` that starts the Writer's /// Container is refused before the switch lock. A Create runs a spec the caller /// chose. A Kept start names a stopped Container that nothing on this Machine ties @@ -37,6 +159,262 @@ impl LocalMachine { } } +/// What a grant lets its recipient pull. +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum GrantScope { + /// Every snapshot of the Volume: a Mirror's, until it is withdrawn. + Standing, + /// Only these snapshots, which this Machine took under the operation. + Snapshots(Vec), +} + +/// This Machine's Volume operations, grants and withdrawals, kept in the approvals log. +/// No record here is pruned. +#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize)] +pub(super) struct VolumeRecords { + /// Raised by each admission and each withdrawal, so a withdrawal outranks exactly the + /// operations admitted before it. It orders events on this Machine and reads no clock. + sequence: u64, + #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] + operations: BTreeMap, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + grants: Vec, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + withdrawals: Vec, +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +struct OperationRecord { + /// The approval's body as signed, so another body under the same ID is refused. + digest: StepDigest, + volume: DockerVolumeName, + workflow: VolumeWorkflow, + admitted: u64, + state: OperationState, + #[serde(default, skip_serializing_if = "Option::is_none")] + decision: Option, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +enum OperationState { + Open, + Ended, +} + +/// What the source of a Move did after its cut. Written once. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +enum Decision { + Resumed, +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +struct Grant { + operation: OperationId, + volume: DockerVolumeName, + recipient: MachineId, + scope: GrantScope, +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +struct Withdrawal { + volume: DockerVolumeName, + recipient: MachineId, + at: u64, +} + +impl VolumeRecords { + pub(super) fn is_empty(&self) -> bool { + *self == Self::default() + } + + fn admit( + &mut self, + operation: OperationId, + digest: StepDigest, + machine: &MachineId, + volume: &DockerVolumeName, + step: &VolumeStep, + verify: impl FnOnce() -> Result, + ) -> Result<(), Error> { + let refused = |field| { + Err(Error::VolumeStepRefused(AuthorityRefusal::Mismatch { + field, + })) + }; + if let Some(record) = self.operations.get(&operation) { + if record.state == OperationState::Ended { + return Err(Error::VolumeStepRefused(AuthorityRefusal::Replayed { + operation, + })); + } + if record.digest != digest { + return refused("an operation"); + } + if record.volume != *volume { + return refused("a Volume"); + } + if !record.workflow.permits(machine, step) { + return refused("a Volume step"); + } + return Ok(()); + } + let workflow = verify().map_err(Error::VolumeStepRefused)?; + if workflow.owns(machine) { + let owners: Vec = self + .operations + .iter() + .filter(|(_, record)| { + record.state == OperationState::Open + && record.volume == *volume + && record.workflow.owns(machine) + }) + .map(|(operation, _)| *operation) + .collect(); + for owner in owners { + self.end_owner(owner, machine); + } + } + self.sequence += 1; + self.operations.insert( + operation, + OperationRecord { + digest, + volume: volume.clone(), + workflow, + admitted: self.sequence, + state: OperationState::Open, + decision: None, + }, + ); + Ok(()) + } + + /// End an earlier owner of this Machine's copy. A Move this Machine is the source of, + /// and that has no decision yet, records `Resumed`, so no handoff can follow. + fn end_owner(&mut self, operation: OperationId, machine: &MachineId) { + if let Some(record) = self.operations.get_mut(&operation) + && record.workflow.moves_from(machine) + && record.decision.is_none() + { + record.decision = Some(Decision::Resumed); + } + self.end(operation); + } + + fn end(&mut self, operation: OperationId) { + if let Some(record) = self.operations.get_mut(&operation) { + record.state = OperationState::Ended; + } + self.grants.retain(|grant| grant.operation != operation); + } + + fn grant( + &mut self, + operation: OperationId, + recipient: MachineId, + scope: GrantScope, + ) -> Result<(), Error> { + let refused = |refusal| Err(Error::VolumeStepRefused(refusal)); + let Some(record) = self.operations.get(&operation) else { + return refused(AuthorityRefusal::Mismatch { + field: "an operation", + }); + }; + if record.state == OperationState::Ended { + return refused(AuthorityRefusal::Replayed { operation }); + } + let target = + matches!(&record.workflow, VolumeWorkflow::Move { to, .. } if *to == recipient); + let sends = record.workflow.mirrors_to(&recipient) + || (target && matches!(scope, GrantScope::Snapshots(_))); + if !sends { + return refused(AuthorityRefusal::Mismatch { + field: "a recipient", + }); + } + let outranked = self.withdrawals.iter().any(|withdrawal| { + withdrawal.volume == record.volume + && withdrawal.recipient == recipient + && withdrawal.at > record.admitted + }); + if outranked { + return refused(AuthorityRefusal::Replayed { operation }); + } + let volume = record.volume.clone(); + let held = self + .grants + .iter_mut() + .find(|grant| grant.operation == operation && grant.recipient == recipient); + match (held, scope) { + (Some(held), GrantScope::Standing) => held.scope = GrantScope::Standing, + (Some(held), GrantScope::Snapshots(added)) => { + if let GrantScope::Snapshots(names) = &mut held.scope { + for name in added { + if !names.contains(&name) { + names.push(name); + } + } + } + } + (None, scope) => self.grants.push(Grant { + operation, + volume, + recipient, + scope, + }), + } + Ok(()) + } + + fn live( + &self, + volume: &DockerVolumeName, + recipient: &MachineId, + snapshot: &SnapshotName, + ) -> bool { + self.grants.iter().any(|grant| { + grant.volume == *volume + && grant.recipient == *recipient + && match &grant.scope { + GrantScope::Standing => true, + GrantScope::Snapshots(names) => names.contains(snapshot), + } + }) + } + + /// End every open operation here that mirrors `volume` to `recipient`: a Mirror's own, + /// on either Machine, and a Move's following on the recipient itself. A Move between + /// two other Machines goes on, but can no longer grant `recipient` anything. + fn withdraw(&mut self, machine: &MachineId, volume: &DockerVolumeName, recipient: &MachineId) { + self.sequence += 1; + self.withdrawals.push(Withdrawal { + volume: volume.clone(), + recipient: *recipient, + at: self.sequence, + }); + let mirrored: Vec = self + .operations + .iter() + .filter(|(_, record)| { + record.state == OperationState::Open + && record.volume == *volume + && record.workflow.mirrors_to(recipient) + && (matches!(record.workflow, VolumeWorkflow::Mirror { .. }) + || machine == recipient) + }) + .map(|(operation, _)| *operation) + .collect(); + for operation in mirrored { + self.end(operation); + } + self.grants + .retain(|grant| !(grant.volume == *volume && grant.recipient == *recipient)); + } +} + /// What a refused Volume verb answers, by the capability refused. pub(super) fn refusal_text(capability: &str) -> &'static str { if capability == SEED_VOLUME_CAPABILITY { @@ -45,3 +423,7 @@ pub(super) fn refusal_text(capability: &str) -> &'static str { VOLUME_START_REFUSED_BY_ENFORCEMENT } } + +#[cfg(test)] +#[path = "volume_approval_tests.rs"] +mod tests; diff --git a/core/crates/ployzd/src/machine/local_machine/volume_approval_tests.rs b/core/crates/ployzd/src/machine/local_machine/volume_approval_tests.rs new file mode 100644 index 0000000000..f4b3503dcb --- /dev/null +++ b/core/crates/ployzd/src/machine/local_machine/volume_approval_tests.rs @@ -0,0 +1,667 @@ +use std::path::Path; + +use ployz_core::{ + CopyTransition, DeviceKey, MachineName, Namespace, ServiceName, SnapshotGuid, + authority::{ApprovedService, PayloadDigest, Signer, SignerKey, validity::MAX_LIFETIME_MILLIS}, +}; + +use super::*; +use crate::machine::{ + LocalMachineStore, RecordOwner, local_machine::restart_approval::ConsumptionLog, +}; + +fn id(fill: char) -> MachineId { + MachineId::parse(fill.to_string().repeat(32)).unwrap() +} + +fn data() -> DockerVolumeName { + DockerVolumeName::parse("app_data").unwrap() +} + +fn other_volume() -> DockerVolumeName { + DockerVolumeName::parse("app_other").unwrap() +} + +fn operation(fill: char) -> OperationId { + OperationId::parse(fill.to_string().repeat(32)).unwrap() +} + +fn digest(fill: char) -> StepDigest { + StepDigest::of(&fill.to_string()) +} + +fn at(millis: u64) -> Timestamp { + Timestamp::from_millis(millis).unwrap() +} + +fn mirror(writer: char, mirror: char) -> VolumeWorkflow { + VolumeWorkflow::Mirror { + writer: id(writer), + mirror: id(mirror), + replace: false, + } +} + +fn moving(from: char, to: char, mirrors: &[char]) -> VolumeWorkflow { + moving_from(id(from), to, mirrors) +} + +fn moving_from(from: MachineId, to: char, mirrors: &[char]) -> VolumeWorkflow { + VolumeWorkflow::Move { + from, + to: id(to), + service: ApprovedService { + name: ServiceName::parse("db").unwrap(), + digest: PayloadDigest::parse(&"d".repeat(64)).unwrap(), + image: None, + }, + mirrors: mirrors.iter().map(|fill| id(*fill)).collect(), + } +} + +fn restore(machine: char) -> VolumeWorkflow { + VolumeWorkflow::Restore { + machine: id(machine), + at: SnapshotGuid::new(7), + loses_after: at(1), + } +} + +fn snapshot() -> VolumeStep { + VolumeStep::Copy(CopyTransition::WriterToSnapshot) +} + +fn pull() -> VolumeStep { + VolumeStep::Copy(CopyTransition::MirrorToMirror) +} + +fn name(value: &str) -> SnapshotName { + value.parse().unwrap() +} + +/// One Machine's view of the approvals log at `dir`, opened afresh for each call, so every +/// answer comes from what reached the disk. +struct Records<'a> { + dir: &'a Path, + machine: MachineId, +} + +impl Records<'_> { + fn log(&self) -> std::sync::Arc { + ConsumptionLog::new(self.dir) + } + + /// Admit `step` of operation `op` with `body` as its digest. `verified` stands in for the + /// signed approval's check, which only a first step reaches. + fn admit( + &self, + op: char, + body: char, + volume: &DockerVolumeName, + step: &VolumeStep, + verified: Result, + ) -> Result<(), Error> { + self.log().change_volumes(at(5_000), |records, _| { + records.admit( + operation(op), + digest(body), + &self.machine, + volume, + step, + || verified, + ) + }) + } + + fn admit_new(&self, op: char, workflow: VolumeWorkflow, step: &VolumeStep) { + self.admit(op, op, &data(), step, Ok(workflow)).unwrap(); + } + + /// A later step of an open operation, whose window has passed. + fn admit_late(&self, op: char, step: &VolumeStep) -> Result<(), Error> { + self.admit(op, op, &data(), step, Err(AuthorityRefusal::Expired)) + } + + fn grant(&self, op: char, recipient: char, scope: GrantScope) -> Result<(), Error> { + self.log().change_volumes(at(5_000), |records, _| { + records.grant(operation(op), id(recipient), scope) + }) + } + + fn live(&self, volume: &DockerVolumeName, recipient: char, snapshot: &str) -> bool { + self.log() + .read_volumes(|records| records.live(volume, &id(recipient), &name(snapshot))) + .unwrap() + } + + fn withdraw(&self, volume: &DockerVolumeName, recipient: char) { + self.log() + .change_volumes(at(5_000), |records, _| { + records.withdraw(&self.machine, volume, &id(recipient)); + Ok(()) + }) + .unwrap(); + } + + fn end(&self, op: char) { + self.log() + .change_volumes(at(5_000), |records, _| { + records.end(operation(op)); + Ok(()) + }) + .unwrap(); + } + + fn decision(&self, op: char) -> Option { + self.log() + .read_volumes(|records| { + records + .operations + .get(&operation(op)) + .and_then(|record| record.decision) + }) + .unwrap() + } +} + +fn replayed(result: Result<(), Error>, op: char) -> bool { + matches!( + result, + Err(Error::VolumeStepRefused(AuthorityRefusal::Replayed { operation: refused })) + if refused == operation(op) + ) +} + +fn mismatch(result: Result<(), Error>) -> Option<&'static str> { + match result { + Err(Error::VolumeStepRefused(AuthorityRefusal::Mismatch { field })) => Some(field), + _ => None, + } +} + +#[test] +fn an_open_operation_finishes_after_its_window_and_an_ended_one_never_runs_again() { + let dir = tempfile::tempdir().unwrap(); + let writer = Records { + dir: dir.path(), + machine: id('a'), + }; + + assert!(matches!( + writer.admit( + '1', + '1', + &data(), + &snapshot(), + Err(AuthorityRefusal::Expired) + ), + Err(Error::VolumeStepRefused(AuthorityRefusal::Expired)) + )); + assert!( + !dir.path().join("restart-approvals.json").exists(), + "a refused first step writes nothing" + ); + + writer.admit_new('1', mirror('a', 'b'), &snapshot()); + writer.admit_late('1', &snapshot()).unwrap(); + writer + .log() + .change_volumes(at(1), |records, _| { + records.admit( + operation('1'), + digest('1'), + &id('a'), + &data(), + &snapshot(), + || unreachable!("an open operation is not checked against the clock"), + ) + }) + .unwrap(); + writer.grant('1', 'b', GrantScope::Standing).unwrap(); + assert!(writer.live(&data(), 'b', "w-1-1")); + + writer.end('1'); + assert!(replayed(writer.admit_late('1', &snapshot()), '1')); + assert!(replayed( + writer.admit('1', '1', &data(), &snapshot(), Ok(mirror('a', 'b'))), + '1' + )); + assert!(!writer.live(&data(), 'b', "w-1-1")); + assert!(replayed(writer.grant('1', 'b', GrantScope::Standing), '1')); +} + +#[test] +fn an_open_operation_still_admits_only_its_own_body_volume_and_graph() { + let dir = tempfile::tempdir().unwrap(); + let writer = Records { + dir: dir.path(), + machine: id('a'), + }; + writer.admit_new('1', mirror('a', 'b'), &snapshot()); + assert_eq!( + mismatch(writer.admit('1', '2', &data(), &snapshot(), Ok(mirror('a', 'b')))), + Some("an operation") + ); + assert_eq!( + mismatch(writer.admit('1', '1', &other_volume(), &snapshot(), Ok(mirror('a', 'b')))), + Some("a Volume") + ); + assert_eq!( + mismatch(writer.admit_late('1', &VolumeStep::Copy(CopyTransition::WriterToMirror))), + Some("a Volume step") + ); + assert_eq!( + mismatch(writer.grant('1', 'c', GrantScope::Standing)), + Some("a recipient") + ); + assert_eq!( + mismatch(writer.grant('9', 'b', GrantScope::Standing)), + Some("an operation") + ); +} + +#[test] +fn a_bounded_grant_serves_only_its_snapshots_and_ends_with_its_move() { + let dir = tempfile::tempdir().unwrap(); + let source = Records { + dir: dir.path(), + machine: id('a'), + }; + source.admit_new('1', moving('a', 'b', &[]), &snapshot()); + assert_eq!( + mismatch(source.grant('1', 'b', GrantScope::Standing)), + Some("a recipient"), + "a Move's target never holds every snapshot" + ); + source + .grant('1', 'b', GrantScope::Snapshots(vec![name("w-1-1")])) + .unwrap(); + source + .grant('1', 'b', GrantScope::Snapshots(vec![name("f-1")])) + .unwrap(); + assert!(source.live(&data(), 'b', "w-1-1")); + assert!(source.live(&data(), 'b', "f-1")); + assert!(!source.live(&data(), 'b', "w-2-1")); + assert!(!source.live(&data(), 'c', "f-1")); + assert!(!source.live(&other_volume(), 'b', "f-1")); + + source.end('1'); + assert!(!source.live(&data(), 'b', "f-1")); +} + +#[test] +fn a_withdrawal_ends_only_what_mirrored_that_volume_to_that_machine() { + let dir = tempfile::tempdir().unwrap(); + let writer = Records { + dir: dir.path(), + machine: id('a'), + }; + writer.admit_new('1', mirror('a', 'b'), &snapshot()); + writer.admit_new('2', mirror('a', 'c'), &snapshot()); + writer + .admit('3', '3', &other_volume(), &snapshot(), Ok(mirror('a', 'b'))) + .unwrap(); + for (op, recipient) in [('1', 'b'), ('2', 'c'), ('3', 'b')] { + writer.grant(op, recipient, GrantScope::Standing).unwrap(); + } + + writer.withdraw(&data(), 'b'); + + assert!(replayed(writer.admit_late('1', &snapshot()), '1')); + assert!(replayed( + writer.admit('1', '1', &data(), &snapshot(), Ok(mirror('a', 'b'))), + '1' + )); + assert!(!writer.live(&data(), 'b', "w-1-1")); + writer.admit_late('2', &snapshot()).unwrap(); + assert!(writer.live(&data(), 'c', "w-1-1")); + writer + .admit( + '3', + '3', + &other_volume(), + &snapshot(), + Err(AuthorityRefusal::Expired), + ) + .unwrap(); + assert!(writer.live(&other_volume(), 'b', "w-1-1")); + + writer.admit_new('4', mirror('a', 'b'), &snapshot()); + writer.grant('4', 'b', GrantScope::Standing).unwrap(); + assert!(writer.live(&data(), 'b', "w-1-1")); +} + +#[test] +fn a_mirror_that_withdraws_itself_ends_its_own_pulls_and_a_followed_move() { + let dir = tempfile::tempdir().unwrap(); + let removed = Records { + dir: dir.path(), + machine: id('c'), + }; + removed.admit_new('1', mirror('a', 'c'), &pull()); + removed.admit_new('2', moving('a', 'b', &['c']), &pull()); + + removed.withdraw(&data(), 'c'); + + assert!(replayed(removed.admit_late('1', &pull()), '1')); + assert!(replayed(removed.admit_late('2', &pull()), '2')); +} + +#[test] +fn a_withdrawal_outranks_a_grant_under_a_move_admitted_before_it() { + let dir = tempfile::tempdir().unwrap(); + let source = Records { + dir: dir.path(), + machine: id('a'), + }; + source.admit_new('1', moving('a', 'b', &['c']), &snapshot()); + source.withdraw(&data(), 'c'); + + source.admit_late('1', &snapshot()).unwrap(); + assert!(replayed(source.grant('1', 'c', GrantScope::Standing), '1')); + source + .grant('1', 'b', GrantScope::Snapshots(vec![name("f-1")])) + .unwrap(); + + source.admit_new('2', moving('a', 'b', &['c']), &snapshot()); + source.grant('2', 'c', GrantScope::Standing).unwrap(); +} + +#[test] +fn a_new_owner_of_the_copy_ends_the_old_one_and_a_source_without_a_decision_resumes() { + let dir = tempfile::tempdir().unwrap(); + let source = Records { + dir: dir.path(), + machine: id('a'), + }; + source.admit_new('1', moving('a', 'b', &[]), &snapshot()); + source + .grant('1', 'b', GrantScope::Snapshots(vec![name("f-1")])) + .unwrap(); + source.admit_new('2', mirror('e', 'a'), &pull()); + source.admit_new( + '3', + VolumeWorkflow::Resize { + machines: vec![id('a')], + maximum_bytes: ployz_core::ProvisionedVolumeMaximumBytes::new( + std::num::NonZeroU64::new(1_000_000_000).unwrap(), + ), + shrink: false, + }, + &VolumeStep::Resize { + maximum_bytes: ployz_core::ProvisionedVolumeMaximumBytes::new( + std::num::NonZeroU64::new(1_000_000_000).unwrap(), + ), + allow_shrink: false, + }, + ); + assert!(replayed(source.admit_late('1', &snapshot()), '1')); + assert_eq!(source.decision('1'), Some(Decision::Resumed)); + assert!(!source.live(&data(), 'b', "f-1")); + + source.admit_new( + '4', + restore('a'), + &VolumeStep::Copy(CopyTransition::MirrorToWriter), + ); + assert!(replayed(source.admit_late('2', &pull()), '2')); + assert_eq!(source.decision('2'), None); + source + .admit_late( + '3', + &VolumeStep::Resize { + maximum_bytes: ployz_core::ProvisionedVolumeMaximumBytes::new( + std::num::NonZeroU64::new(1_000_000_000).unwrap(), + ), + allow_shrink: false, + }, + ) + .unwrap(); + + let target = Records { + dir: dir.path(), + machine: id('b'), + }; + target + .admit( + '5', + '5', + &other_volume(), + &pull(), + Ok(moving('a', 'b', &[])), + ) + .unwrap(); + target + .admit( + '6', + '6', + &other_volume(), + &VolumeStep::Copy(CopyTransition::MirrorToWriter), + Ok(restore('b')), + ) + .unwrap(); + assert!(replayed( + target.admit( + '5', + '5', + &other_volume(), + &pull(), + Err(AuthorityRefusal::Expired) + ), + '5' + )); + assert_eq!(target.decision('5'), None, "b is not the Move's source"); +} + +#[test] +fn records_written_before_this_format_existed_read_back_with_no_volume_records() { + let dir = tempfile::tempdir().unwrap(); + std::fs::write( + dir.path().join("restart-approvals.json"), + br#"{"expiry_time_floor":1000}"#, + ) + .unwrap(); + let writer = Records { + dir: dir.path(), + machine: id('a'), + }; + assert!(!writer.live(&data(), 'b', "w-1-1")); + writer.admit_new('1', mirror('a', 'b'), &snapshot()); + let stored = std::fs::read_to_string(dir.path().join("restart-approvals.json")).unwrap(); + assert!( + stored.contains(r#""volumes":{"sequence":1,"operations""#), + "{stored}" + ); +} + +struct Enforcing { + machine: LocalMachine, + signer: Signer, + cluster: ployz_core::SealingPublicKey, + data_dir: crate::test_dir::TestDir, +} + +async fn enforcing() -> Enforcing { + let sealing = crate::sealed_env::test_support::Cluster::new(); + let data_dir = crate::test_dir::TestDir::new("volume-approval"); + let mut store = LocalMachineStore::open(&data_dir.0).unwrap(); + store + .enroll(ployz_core::EnrollRequest { + role: ployz_core::EnrollAs::Founder { + initial_policy: Default::default(), + name: MachineName::parse("local").unwrap(), + cluster_network: "10.210.0.0/16".parse().unwrap(), + public_ip: None, + advertised_endpoints: vec![ployz_core::AdvertisedEndpoint( + "192.0.2.1:51820".parse().unwrap(), + )], + grant: None, + }, + wireguard_mtu: None, + pinned: Default::default(), + }) + .unwrap(); + std::fs::copy( + sealing.dir.0.join("sealing.key"), + data_dir.0.join("sealing.key"), + ) + .unwrap(); + let owner = RecordOwner::spawn(store).unwrap(); + let signer = Signer::from_seed(&[7; 32]); + let device = DeviceKey::from(signer.public()); + owner + .mutate(move |store| { + store.approve_device(device).unwrap(); + store.set_device_enforcement(true, Vec::new()).unwrap(); + }) + .await + .unwrap(); + let cluster = crate::sealing_key::SealingKey::load(&data_dir.0) + .unwrap() + .unwrap() + .public_key(); + Enforcing { + machine: LocalMachine::new(owner), + signer, + cluster, + data_dir, + } +} + +impl Enforcing { + /// A Move of `app_data` from this Machine to another, signed now by `signer`, open + /// from `issued` milliseconds ago for `lifetime` milliseconds. + fn approval(&self, signer: &Signer, issued: u64, lifetime: u64) -> Signed { + let issued_at = at(Timestamp::now().as_millis() - issued); + let body = VolumeApproval { + cluster: self.cluster, + operation: operation('1'), + namespace: Namespace::parse("app").unwrap(), + volume: data(), + workflow: moving_from(self.machine.record().id(), 'b', &[]), + issued_at, + expires_at: issued_at.plus(lifetime), + signer: SignerKey::Device(DeviceKey::from(signer.public())), + }; + Signed::sign(&body, signer).unwrap() + } +} + +#[tokio::test] +async fn a_device_approval_opens_the_operation_on_disk_and_a_late_or_foreign_one_is_refused() { + let fixture = enforcing().await; + let late = fixture.approval( + &fixture.signer, + MAX_LIFETIME_MILLIS + 600_000, + MAX_LIFETIME_MILLIS, + ); + assert!(matches!( + fixture + .machine + .admit_volume_step(&late, &data(), &snapshot()), + Err(Error::VolumeStepRefused(AuthorityRefusal::Expired)) + )); + let stranger = Signer::from_seed(&[3; 32]); + assert!(matches!( + fixture.machine.admit_volume_step( + &fixture.approval(&stranger, 0, MAX_LIFETIME_MILLIS), + &data(), + &snapshot() + ), + Err(Error::VolumeStepRefused( + AuthorityRefusal::UnknownKey { .. } + )) + )); + + let approval = fixture.approval(&fixture.signer, 0, MAX_LIFETIME_MILLIS); + assert_eq!( + fixture + .machine + .admit_volume_step(&approval, &data(), &snapshot()) + .unwrap(), + operation('1') + ); + assert!(matches!( + fixture + .machine + .admit_volume_step(&approval, &data(), &pull()), + Err(Error::VolumeStepRefused(AuthorityRefusal::Mismatch { + field: "a Volume step" + })) + )); + fixture + .machine + .grant_volume( + operation('1'), + id('b'), + GrantScope::Snapshots(vec![name("f-1")]), + ) + .unwrap(); + assert!( + fixture + .machine + .volume_grant_live(&data(), &id('b'), &name("f-1")) + .unwrap() + ); + + fixture + .machine + .end_volume_operation(operation('1')) + .unwrap(); + let reopened = ConsumptionLog::new(&fixture.data_dir.0); + assert!( + !reopened + .read_volumes(|records| records.live(&data(), &id('b'), &name("f-1"))) + .unwrap() + ); + assert!(matches!( + reopened.change_volumes(Timestamp::now(), |records, _| { + records.admit( + operation('1'), + StepDigest::of(&approval.body), + &fixture.machine.record().id(), + &data(), + &snapshot(), + || unreachable!("an ended operation is refused before any check"), + ) + }), + Err(Error::VolumeStepRefused(AuthorityRefusal::Replayed { .. })) + )); + assert!(matches!( + fixture + .machine + .admit_volume_step(&approval, &data(), &snapshot()), + Err(Error::VolumeStepRefused(AuthorityRefusal::Replayed { .. })) + )); +} + +#[tokio::test] +async fn a_withdrawal_by_this_machine_survives_a_reload() { + let fixture = enforcing().await; + let approval = fixture.approval(&fixture.signer, 0, MAX_LIFETIME_MILLIS); + fixture + .machine + .admit_volume_step(&approval, &data(), &snapshot()) + .unwrap(); + fixture.machine.withdraw_volume(&data(), &id('b')).unwrap(); + assert!(matches!( + fixture.machine.grant_volume( + operation('1'), + id('b'), + GrantScope::Snapshots(vec![name("f-1")]) + ), + Err(Error::VolumeStepRefused(AuthorityRefusal::Replayed { .. })) + )); + let reopened = ConsumptionLog::new(&fixture.data_dir.0); + let withdrawn = reopened + .read_volumes(|records| records.withdrawals.clone()) + .unwrap(); + assert_eq!( + withdrawn, + vec![Withdrawal { + volume: data(), + recipient: id('b'), + at: 2, + }] + ); +} diff --git a/core/crates/ployzd/src/machine_api/local.rs b/core/crates/ployzd/src/machine_api/local.rs index 2a8005d3c5..ecf88538e4 100644 --- a/core/crates/ployzd/src/machine_api/local.rs +++ b/core/crates/ployzd/src/machine_api/local.rs @@ -1288,6 +1288,13 @@ fn local_failure(error: LocalMachineError) -> Result { "next": "ployz device enforce --off from an approved device", }), )), + LocalMachineError::VolumeStepRefused(_) => Err(ployz_core::rpc::hinted_status( + tonic::Code::PermissionDenied, + &error, + serde_json::json!({ + "next": "approve the Volume change again from an approved device", + }), + )), LocalMachineError::RuleGrantRefused(_) | LocalMachineError::RuleProofRefused(_) | LocalMachineError::RuleHeadUnread(_) => Err(build_refused( diff --git a/dashboard/src/modules/approval-session/volume-approval.test.ts b/dashboard/src/modules/approval-session/volume-approval.test.ts new file mode 100644 index 0000000000..c3eeeab997 --- /dev/null +++ b/dashboard/src/modules/approval-session/volume-approval.test.ts @@ -0,0 +1,49 @@ +import type { MachineId, OperationId } from "@ployz/sdk"; +import type { VolumeApproval, VolumeWorkflow } from "@ployz/sdk/authority"; +import { describe, expect, it } from "vitest"; +import { canonicalJson } from "./approval-session"; + +const a = "a".repeat(32) as MachineId; +const b = "b".repeat(32) as MachineId; +const c = "c".repeat(32) as MachineId; +const d = "d".repeat(64); + +const head = `{"cluster":"CQkJCQkJCQkJCQkJCQkJCQkJCQkJCQkJCQkJCQkJCQk=","expires_at":1000000,"issued_at":100000,"namespace":"app","operation":"0f8fad5bd9cb469fa16570867728950e","signer":"ea4a6c63e29c520abef5507b132ec5f9954776aebebe7b92421eea691446d22c","volume":"app_data","workflow":`; + +// The bodies `ployz_core::authority::volume`'s golden test pins, one per workflow. +const workflows: [VolumeWorkflow, string][] = [ + [ + { kind: "move", from: a, to: b, service: { name: "db", digest: d }, mirrors: [c] }, + `{"from":"${a}","kind":"move","mirrors":["${c}"],"service":{"digest":"${d}","name":"db"},"to":"${b}"}`, + ], + [{ kind: "mirror", writer: a, mirror: b, replace: false }, `{"kind":"mirror","mirror":"${b}","replace":false,"writer":"${a}"}`], + [ + { kind: "mirror_removal", mirror: b, writer: a, role: "mirror", last_copy: true }, + `{"kind":"mirror_removal","last_copy":true,"mirror":"${b}","role":"mirror","writer":"${a}"}`, + ], + [ + { kind: "restore", machine: b, at: "18446744073709551615", loses_after: 500000 }, + `{"at":"18446744073709551615","kind":"restore","loses_after":500000,"machine":"${b}"}`, + ], + [ + { kind: "resize", machines: [a, b], maximum_bytes: 20_000_000_000, shrink: false }, + `{"kind":"resize","machines":["${a}","${b}"],"maximum_bytes":20000000000,"shrink":false}`, + ], + [{ kind: "seed", machine: a, source: "app_old", at: "42" }, `{"at":"42","kind":"seed","machine":"${a}","source":"app_old"}`], +]; + +describe("a Volume approval's canonical body", () => { + it.each(workflows)("matches ployz-core's bytes for %j", (workflow, golden) => { + const approval = { + cluster: "CQkJCQkJCQkJCQkJCQkJCQkJCQkJCQkJCQkJCQkJCQk=", + operation: "0f8fad5bd9cb469fa16570867728950e" as OperationId, + namespace: "app", + volume: "app_data", + workflow, + issued_at: 100000, + expires_at: 1000000, + signer: "ea4a6c63e29c520abef5507b132ec5f9954776aebebe7b92421eea691446d22c", + } satisfies VolumeApproval; + expect(canonicalJson(approval)).toBe(`${head}${golden}}`); + }); +}); From f2a14bb0da33ee36d5e520b8b1221274b924db95 Mon Sep 17 00:00:00 2001 From: Nick Potts Date: Tue, 10 Nov 2026 12:42:29 +0000 Subject: [PATCH 2/3] End a following Mirror's Move when a later operation takes its copy, and keep a replacing Mirror through its own withdrawal A Move now owns the copy of each Mirror it keeps following, so admitting a later owner there ends the Move on that Machine. withdraw_volume takes the replacing Mirror operation whose Forget writes the withdrawal. That operation stays open and the withdrawal does not outrank its grants. It ends every other Mirror operation as before. --- .../crates/ployz-core/src/authority/volume.rs | 9 +- .../machine/local_machine/volume_approval.rs | 34 +++++-- .../local_machine/volume_approval_tests.rs | 99 ++++++++++++++++++- 3 files changed, 126 insertions(+), 16 deletions(-) diff --git a/core/crates/ployz-core/src/authority/volume.rs b/core/crates/ployz-core/src/authority/volume.rs index b7fa6ace7d..ba3ca9fa70 100644 --- a/core/crates/ployz-core/src/authority/volume.rs +++ b/core/crates/ployz-core/src/authority/volume.rs @@ -188,12 +188,15 @@ impl VolumeWorkflow { } } - /// Whether this workflow changes the role of `machine`'s copy, or destroys it. A - /// Machine holds at most one open owner of a Volume's copy. + /// Whether this workflow drives `machine`'s copy: changes its role, destroys it, or, + /// for a Mirror a Move keeps following, says where it pulls from. A Machine holds at + /// most one open owner of a Volume's copy. #[must_use] pub fn owns(&self, machine: &MachineId) -> bool { match self { - Self::Move { from, to, .. } => machine == from || machine == to, + Self::Move { + from, to, mirrors, .. + } => machine == from || machine == to || mirrors.contains(machine), Self::Mirror { mirror, .. } | Self::MirrorRemoval { mirror, .. } => machine == mirror, Self::Restore { machine: owner, .. } => machine == owner, Self::Resize { .. } | Self::Seed { .. } => false, diff --git a/core/crates/ployzd/src/machine/local_machine/volume_approval.rs b/core/crates/ployzd/src/machine/local_machine/volume_approval.rs index fbe4ca8785..15c15a650a 100644 --- a/core/crates/ployzd/src/machine/local_machine/volume_approval.rs +++ b/core/crates/ployzd/src/machine/local_machine/volume_approval.rs @@ -115,7 +115,9 @@ impl LocalMachine { /// Withdraw `recipient` from `volume` here, before anything is destroyed: every Mirror /// operation this Machine admitted for it ends, and so does every grant to it. It needs /// no approval, because it only narrows what a driver can do. An approval this Machine - /// never admitted is still new work while its window lasts. + /// never admitted is still new work while its window lasts. `replacing` names the + /// Mirror operation whose own `Forget` writes this withdrawal: it stays open, and this + /// withdrawal does not outrank its grants. /// /// # Errors /// Returns [`Error::ApprovalRecord`] when the record cannot be read or written. @@ -123,11 +125,12 @@ impl LocalMachine { &self, volume: &DockerVolumeName, recipient: &MachineId, + replacing: Option, ) -> Result<(), Error> { let machine = self.record().id(); self.approvals .change_volumes(Timestamp::now(), |records, _| { - records.withdraw(&machine, volume, recipient); + records.withdraw(&machine, volume, recipient, replacing); Ok(()) }) } @@ -223,6 +226,9 @@ struct Withdrawal { volume: DockerVolumeName, recipient: MachineId, at: u64, + /// The replacing Mirror operation this withdrawal leaves open. + #[serde(default, skip_serializing_if = "Option::is_none")] + replacing: Option, } impl VolumeRecords { @@ -339,6 +345,7 @@ impl VolumeRecords { withdrawal.volume == record.volume && withdrawal.recipient == recipient && withdrawal.at > record.admitted + && withdrawal.replacing != Some(operation) }); if outranked { return refused(AuthorityRefusal::Replayed { operation }); @@ -387,19 +394,28 @@ impl VolumeRecords { /// End every open operation here that mirrors `volume` to `recipient`: a Mirror's own, /// on either Machine, and a Move's following on the recipient itself. A Move between - /// two other Machines goes on, but can no longer grant `recipient` anything. - fn withdraw(&mut self, machine: &MachineId, volume: &DockerVolumeName, recipient: &MachineId) { + /// two other Machines goes on, but can no longer grant `recipient` anything. The + /// `replacing` operation, and its grants, stay. + fn withdraw( + &mut self, + machine: &MachineId, + volume: &DockerVolumeName, + recipient: &MachineId, + replacing: Option, + ) { self.sequence += 1; self.withdrawals.push(Withdrawal { volume: volume.clone(), recipient: *recipient, at: self.sequence, + replacing, }); let mirrored: Vec = self .operations .iter() - .filter(|(_, record)| { - record.state == OperationState::Open + .filter(|(operation, record)| { + Some(**operation) != replacing + && record.state == OperationState::Open && record.volume == *volume && record.workflow.mirrors_to(recipient) && (matches!(record.workflow, VolumeWorkflow::Mirror { .. }) @@ -410,8 +426,10 @@ impl VolumeRecords { for operation in mirrored { self.end(operation); } - self.grants - .retain(|grant| !(grant.volume == *volume && grant.recipient == *recipient)); + self.grants.retain(|grant| { + Some(grant.operation) == replacing + || !(grant.volume == *volume && grant.recipient == *recipient) + }); } } diff --git a/core/crates/ployzd/src/machine/local_machine/volume_approval_tests.rs b/core/crates/ployzd/src/machine/local_machine/volume_approval_tests.rs index f4b3503dcb..aa650819ef 100644 --- a/core/crates/ployzd/src/machine/local_machine/volume_approval_tests.rs +++ b/core/crates/ployzd/src/machine/local_machine/volume_approval_tests.rs @@ -135,9 +135,23 @@ impl Records<'_> { } fn withdraw(&self, volume: &DockerVolumeName, recipient: char) { + self.withdraw_replacing(volume, recipient, None); + } + + fn withdraw_replacing( + &self, + volume: &DockerVolumeName, + recipient: char, + replacing: Option, + ) { self.log() .change_volumes(at(5_000), |records, _| { - records.withdraw(&self.machine, volume, &id(recipient)); + records.withdraw( + &self.machine, + volume, + &id(recipient), + replacing.map(operation), + ); Ok(()) }) .unwrap(); @@ -339,11 +353,11 @@ fn a_mirror_that_withdraws_itself_ends_its_own_pulls_and_a_followed_move() { machine: id('c'), }; removed.admit_new('1', mirror('a', 'c'), &pull()); - removed.admit_new('2', moving('a', 'b', &['c']), &pull()); - removed.withdraw(&data(), 'c'); - assert!(replayed(removed.admit_late('1', &pull()), '1')); + + removed.admit_new('2', moving('a', 'b', &['c']), &pull()); + removed.withdraw(&data(), 'c'); assert!(replayed(removed.admit_late('2', &pull()), '2')); } @@ -643,7 +657,10 @@ async fn a_withdrawal_by_this_machine_survives_a_reload() { .machine .admit_volume_step(&approval, &data(), &snapshot()) .unwrap(); - fixture.machine.withdraw_volume(&data(), &id('b')).unwrap(); + fixture + .machine + .withdraw_volume(&data(), &id('b'), None) + .unwrap(); assert!(matches!( fixture.machine.grant_volume( operation('1'), @@ -662,6 +679,78 @@ async fn a_withdrawal_by_this_machine_survives_a_reload() { volume: data(), recipient: id('b'), at: 2, + replacing: None, }] ); } + +#[test] +fn a_later_owner_of_a_following_mirrors_copy_ends_the_move_there() { + let dir = tempfile::tempdir().unwrap(); + let follower = Records { + dir: dir.path(), + machine: id('c'), + }; + follower.admit_new('1', moving('a', 'b', &['c']), &pull()); + follower.admit_new('2', mirror('b', 'c'), &pull()); + + assert!(replayed(follower.admit_late('1', &pull()), '1')); + follower.admit_late('2', &pull()).unwrap(); +} + +#[test] +fn a_replacing_mirror_survives_the_withdrawal_its_own_forget_writes() { + let dir = tempfile::tempdir().unwrap(); + let replaced = Records { + dir: dir.path(), + machine: id('b'), + }; + let replacing = VolumeWorkflow::Mirror { + writer: id('a'), + mirror: id('b'), + replace: true, + }; + replaced.admit_new('1', mirror('a', 'b'), &pull()); + replaced.admit_new( + '2', + replacing.clone(), + &VolumeStep::Copy(CopyTransition::MirrorToForget), + ); + replaced.withdraw_replacing(&data(), 'b', Some('2')); + + assert!(replayed(replaced.admit_late('1', &pull()), '1')); + replaced + .admit_late('2', &VolumeStep::Copy(CopyTransition::NoneToMirror)) + .unwrap(); + + let writer_dir = tempfile::tempdir().unwrap(); + let writer = Records { + dir: writer_dir.path(), + machine: id('a'), + }; + writer.admit_new('1', mirror('a', 'b'), &snapshot()); + writer.grant('1', 'b', GrantScope::Standing).unwrap(); + writer.admit_new('2', replacing, &snapshot()); + writer.grant('2', 'b', GrantScope::Standing).unwrap(); + writer.withdraw_replacing(&data(), 'b', Some('2')); + assert!(replayed(writer.admit_late('1', &snapshot()), '1')); + assert!(replayed(writer.grant('1', 'b', GrantScope::Standing), '1')); + writer.admit_late('2', &snapshot()).unwrap(); + assert!(writer.live(&data(), 'b', "w-1-1")); + writer.grant('2', 'b', GrantScope::Standing).unwrap(); +} + +#[test] +fn a_replacement_spares_its_operation_only_from_its_own_withdrawal() { + let dir = tempfile::tempdir().unwrap(); + let source = Records { + dir: dir.path(), + machine: id('a'), + }; + source.admit_new('1', moving('a', 'b', &['c', 'e']), &snapshot()); + source.withdraw(&data(), 'c'); + source.withdraw_replacing(&data(), 'e', Some('1')); + + assert!(replayed(source.grant('1', 'c', GrantScope::Standing), '1')); + source.grant('1', 'e', GrantScope::Standing).unwrap(); +} From 08ac63c3ffa2e110ce65b03b6b3dac40b67a5277 Mon Sep 17 00:00:00 2001 From: Nick Potts Date: Tue, 10 Nov 2026 12:56:08 +0000 Subject: [PATCH 3/3] Withdraw a replaced Mirror inside the replacement's own opening Forget withdraw_volume no longer takes an operation to spare. Admitting the opening MirrorToForget of a Mirror approval that replaces this Machine's copy writes the withdrawal of this Volume and Machine in the same synced write, just before the replacement opens. The replacement then ranks after its own withdrawal and needs no exemption. Nothing else can withdraw that way: a step's verifier refuses an ordinary Mirror's Forget, a Move's, and a replacement naming another Volume or Machine, and a resent Forget on the open record withdraws nothing. --- .../machine/local_machine/volume_approval.rs | 51 ++--- .../local_machine/volume_approval_tests.rs | 204 +++++++++++++----- 2 files changed, 172 insertions(+), 83 deletions(-) diff --git a/core/crates/ployzd/src/machine/local_machine/volume_approval.rs b/core/crates/ployzd/src/machine/local_machine/volume_approval.rs index 15c15a650a..30a00ccf68 100644 --- a/core/crates/ployzd/src/machine/local_machine/volume_approval.rs +++ b/core/crates/ployzd/src/machine/local_machine/volume_approval.rs @@ -9,8 +9,9 @@ use std::collections::BTreeMap; use ployz_core::{ - DockerVolumeName, MachineId, OperationId, SEED_REFUSED_BY_ENFORCEMENT, SEED_VOLUME_CAPABILITY, - SET_VOLUME_COPY_CAPABILITY, Sealing, SnapshotName, VOLUME_START_REFUSED_BY_ENFORCEMENT, + CopyTransition, DockerVolumeName, MachineId, OperationId, SEED_REFUSED_BY_ENFORCEMENT, + SEED_VOLUME_CAPABILITY, SET_VOLUME_COPY_CAPABILITY, Sealing, SnapshotName, + VOLUME_START_REFUSED_BY_ENFORCEMENT, authority::{ AuthorityRefusal, Signed, StepDigest, Timestamp, VolumeApproval, VolumeFacts, VolumeStep, VolumeWorkflow, volume, @@ -26,6 +27,10 @@ impl LocalMachine { /// step its workflow permits this Machine. Otherwise the approval must verify inside its /// window, and the operation is recorded open, synced, before this returns. Admitting /// an operation that owns this Machine's copy ends the one that owned it before. + /// Admitting a replacing Mirror's opening `MirrorToForget` also withdraws this Machine + /// from the Volume in that same write, just before the replacement opens, so the + /// withdrawal ends every other Mirror operation here and none of the replacement's. The + /// caller writes no other withdrawal for that `Forget`. /// /// # Errors /// Returns [`Error::VolumeStepRefused`] naming what the approval lacks, an operation @@ -115,9 +120,7 @@ impl LocalMachine { /// Withdraw `recipient` from `volume` here, before anything is destroyed: every Mirror /// operation this Machine admitted for it ends, and so does every grant to it. It needs /// no approval, because it only narrows what a driver can do. An approval this Machine - /// never admitted is still new work while its window lasts. `replacing` names the - /// Mirror operation whose own `Forget` writes this withdrawal: it stays open, and this - /// withdrawal does not outrank its grants. + /// never admitted is still new work while its window lasts. /// /// # Errors /// Returns [`Error::ApprovalRecord`] when the record cannot be read or written. @@ -125,15 +128,15 @@ impl LocalMachine { &self, volume: &DockerVolumeName, recipient: &MachineId, - replacing: Option, ) -> Result<(), Error> { let machine = self.record().id(); self.approvals .change_volumes(Timestamp::now(), |records, _| { - records.withdraw(&machine, volume, recipient, replacing); + records.withdraw(&machine, volume, recipient); Ok(()) }) } + /// Under Device enforcement an Operator's `SetVolumeCopy` that starts the Writer's /// Container is refused before the switch lock. A Create runs a spec the caller /// chose. A Kept start names a stopped Container that nothing on this Machine ties @@ -226,9 +229,6 @@ struct Withdrawal { volume: DockerVolumeName, recipient: MachineId, at: u64, - /// The replacing Mirror operation this withdrawal leaves open. - #[serde(default, skip_serializing_if = "Option::is_none")] - replacing: Option, } impl VolumeRecords { @@ -283,6 +283,13 @@ impl VolumeRecords { self.end_owner(owner, machine); } } + let replaces_here = matches!( + &workflow, + VolumeWorkflow::Mirror { mirror, replace: true, .. } if mirror == machine + ) && *step == VolumeStep::Copy(CopyTransition::MirrorToForget); + if replaces_here { + self.withdraw(machine, volume, machine); + } self.sequence += 1; self.operations.insert( operation, @@ -345,7 +352,6 @@ impl VolumeRecords { withdrawal.volume == record.volume && withdrawal.recipient == recipient && withdrawal.at > record.admitted - && withdrawal.replacing != Some(operation) }); if outranked { return refused(AuthorityRefusal::Replayed { operation }); @@ -394,28 +400,19 @@ impl VolumeRecords { /// End every open operation here that mirrors `volume` to `recipient`: a Mirror's own, /// on either Machine, and a Move's following on the recipient itself. A Move between - /// two other Machines goes on, but can no longer grant `recipient` anything. The - /// `replacing` operation, and its grants, stay. - fn withdraw( - &mut self, - machine: &MachineId, - volume: &DockerVolumeName, - recipient: &MachineId, - replacing: Option, - ) { + /// two other Machines goes on, but can no longer grant `recipient` anything. + fn withdraw(&mut self, machine: &MachineId, volume: &DockerVolumeName, recipient: &MachineId) { self.sequence += 1; self.withdrawals.push(Withdrawal { volume: volume.clone(), recipient: *recipient, at: self.sequence, - replacing, }); let mirrored: Vec = self .operations .iter() - .filter(|(operation, record)| { - Some(**operation) != replacing - && record.state == OperationState::Open + .filter(|(_, record)| { + record.state == OperationState::Open && record.volume == *volume && record.workflow.mirrors_to(recipient) && (matches!(record.workflow, VolumeWorkflow::Mirror { .. }) @@ -426,10 +423,8 @@ impl VolumeRecords { for operation in mirrored { self.end(operation); } - self.grants.retain(|grant| { - Some(grant.operation) == replacing - || !(grant.volume == *volume && grant.recipient == *recipient) - }); + self.grants + .retain(|grant| !(grant.volume == *volume && grant.recipient == *recipient)); } } diff --git a/core/crates/ployzd/src/machine/local_machine/volume_approval_tests.rs b/core/crates/ployzd/src/machine/local_machine/volume_approval_tests.rs index aa650819ef..6bbfa4bfad 100644 --- a/core/crates/ployzd/src/machine/local_machine/volume_approval_tests.rs +++ b/core/crates/ployzd/src/machine/local_machine/volume_approval_tests.rs @@ -135,28 +135,20 @@ impl Records<'_> { } fn withdraw(&self, volume: &DockerVolumeName, recipient: char) { - self.withdraw_replacing(volume, recipient, None); - } - - fn withdraw_replacing( - &self, - volume: &DockerVolumeName, - recipient: char, - replacing: Option, - ) { self.log() .change_volumes(at(5_000), |records, _| { - records.withdraw( - &self.machine, - volume, - &id(recipient), - replacing.map(operation), - ); + records.withdraw(&self.machine, volume, &id(recipient)); Ok(()) }) .unwrap(); } + fn withdrawals(&self) -> Vec { + self.log() + .read_volumes(|records| records.withdrawals.clone()) + .unwrap() + } + fn end(&self, op: char) { self.log() .change_volumes(at(5_000), |records, _| { @@ -559,6 +551,88 @@ impl Enforcing { }; Signed::sign(&body, signer).unwrap() } + + /// `workflow` on `volume`, signed now by this Machine's Device. + fn signed(&self, volume: DockerVolumeName, workflow: VolumeWorkflow) -> Signed { + let issued_at = Timestamp::now(); + let body = VolumeApproval { + cluster: self.cluster, + operation: operation('1'), + namespace: Namespace::parse("app").unwrap(), + volume, + workflow, + issued_at, + expires_at: issued_at.plus(MAX_LIFETIME_MILLIS), + signer: SignerKey::Device(DeviceKey::from(self.signer.public())), + }; + Signed::sign(&body, &self.signer).unwrap() + } + + fn records_empty(&self) -> bool { + ConsumptionLog::new(&self.data_dir.0) + .read_volumes(VolumeRecords::is_empty) + .unwrap() + } +} + +#[tokio::test] +async fn only_a_replacing_mirror_of_this_volume_and_machine_withdraws_with_its_forget() { + let fixture = enforcing().await; + let here = fixture.machine.record().id(); + let mirror_here = |replace| VolumeWorkflow::Mirror { + writer: id('a'), + mirror: here, + replace, + }; + let refused = |volume: DockerVolumeName, workflow: VolumeWorkflow| match fixture + .machine + .admit_volume_step(&fixture.signed(volume, workflow), &data(), &forget()) + { + Err(Error::VolumeStepRefused(AuthorityRefusal::Mismatch { field })) => Some(field), + _ => None, + }; + + assert_eq!(refused(data(), mirror_here(false)), Some("a Volume step")); + assert_eq!( + refused( + data(), + VolumeWorkflow::Move { + from: id('a'), + to: id('b'), + service: ApprovedService { + name: ServiceName::parse("db").unwrap(), + digest: PayloadDigest::parse(&"d".repeat(64)).unwrap(), + image: None, + }, + mirrors: vec![here], + } + ), + Some("a Volume step") + ); + assert_eq!(refused(data(), replacing('a', 'c')), Some("this Server")); + assert_eq!(refused(other_volume(), mirror_here(true)), Some("a Volume")); + assert!(fixture.records_empty(), "a refused Forget writes nothing"); + + let replacement = fixture.signed(data(), mirror_here(true)); + fixture + .machine + .admit_volume_step(&replacement, &data(), &forget()) + .unwrap(); + fixture + .machine + .admit_volume_step(&replacement, &data(), &forget()) + .unwrap(); + let withdrawals = ConsumptionLog::new(&fixture.data_dir.0) + .read_volumes(|records| records.withdrawals.clone()) + .unwrap(); + assert_eq!( + withdrawals, + vec![Withdrawal { + volume: data(), + recipient: here, + at: 1, + }] + ); } #[tokio::test] @@ -657,10 +731,7 @@ async fn a_withdrawal_by_this_machine_survives_a_reload() { .machine .admit_volume_step(&approval, &data(), &snapshot()) .unwrap(); - fixture - .machine - .withdraw_volume(&data(), &id('b'), None) - .unwrap(); + fixture.machine.withdraw_volume(&data(), &id('b')).unwrap(); assert!(matches!( fixture.machine.grant_volume( operation('1'), @@ -679,7 +750,6 @@ async fn a_withdrawal_by_this_machine_survives_a_reload() { volume: data(), recipient: id('b'), at: 2, - replacing: None, }] ); } @@ -698,59 +768,83 @@ fn a_later_owner_of_a_following_mirrors_copy_ends_the_move_there() { follower.admit_late('2', &pull()).unwrap(); } +fn replacing(writer: char, mirror: char) -> VolumeWorkflow { + VolumeWorkflow::Mirror { + writer: id(writer), + mirror: id(mirror), + replace: true, + } +} + +fn forget() -> VolumeStep { + VolumeStep::Copy(CopyTransition::MirrorToForget) +} + #[test] -fn a_replacing_mirror_survives_the_withdrawal_its_own_forget_writes() { +fn a_replacing_mirrors_opening_forget_withdraws_every_other_mirror_here_and_keeps_itself() { let dir = tempfile::tempdir().unwrap(); let replaced = Records { dir: dir.path(), machine: id('b'), }; - let replacing = VolumeWorkflow::Mirror { - writer: id('a'), - mirror: id('b'), - replace: true, - }; replaced.admit_new('1', mirror('a', 'b'), &pull()); - replaced.admit_new( - '2', - replacing.clone(), - &VolumeStep::Copy(CopyTransition::MirrorToForget), - ); - replaced.withdraw_replacing(&data(), 'b', Some('2')); + replaced + .admit('3', '3', &other_volume(), &pull(), Ok(mirror('a', 'b'))) + .unwrap(); + replaced.admit_new('2', replacing('a', 'b'), &forget()); + + assert_eq!( + replaced.withdrawals(), + vec![Withdrawal { + volume: data(), + recipient: id('b'), + at: 3, + }] + ); assert!(replayed(replaced.admit_late('1', &pull()), '1')); replaced .admit_late('2', &VolumeStep::Copy(CopyTransition::NoneToMirror)) .unwrap(); + replaced + .admit( + '3', + '3', + &other_volume(), + &pull(), + Err(AuthorityRefusal::Expired), + ) + .unwrap(); +} - let writer_dir = tempfile::tempdir().unwrap(); - let writer = Records { - dir: writer_dir.path(), - machine: id('a'), +#[test] +fn a_replacement_withdraws_once_and_a_later_withdrawal_ends_it() { + let dir = tempfile::tempdir().unwrap(); + let replaced = Records { + dir: dir.path(), + machine: id('b'), }; - writer.admit_new('1', mirror('a', 'b'), &snapshot()); - writer.grant('1', 'b', GrantScope::Standing).unwrap(); - writer.admit_new('2', replacing, &snapshot()); - writer.grant('2', 'b', GrantScope::Standing).unwrap(); - writer.withdraw_replacing(&data(), 'b', Some('2')); - assert!(replayed(writer.admit_late('1', &snapshot()), '1')); - assert!(replayed(writer.grant('1', 'b', GrantScope::Standing), '1')); - writer.admit_late('2', &snapshot()).unwrap(); - assert!(writer.live(&data(), 'b', "w-1-1")); - writer.grant('2', 'b', GrantScope::Standing).unwrap(); + replaced.admit_new('2', replacing('a', 'b'), &forget()); + replaced.admit_late('2', &forget()).unwrap(); + assert_eq!( + replaced.withdrawals().len(), + 1, + "a resent Forget withdraws nothing" + ); + replaced.admit_late('2', &pull()).unwrap(); + + replaced.withdraw(&data(), 'b'); + assert!(replayed(replaced.admit_late('2', &pull()), '2')); } #[test] -fn a_replacement_spares_its_operation_only_from_its_own_withdrawal() { +fn a_replacement_opened_by_a_pull_withdraws_nothing() { let dir = tempfile::tempdir().unwrap(); - let source = Records { + let mirror_machine = Records { dir: dir.path(), - machine: id('a'), + machine: id('b'), }; - source.admit_new('1', moving('a', 'b', &['c', 'e']), &snapshot()); - source.withdraw(&data(), 'c'); - source.withdraw_replacing(&data(), 'e', Some('1')); - - assert!(replayed(source.grant('1', 'c', GrantScope::Standing), '1')); - source.grant('1', 'e', GrantScope::Standing).unwrap(); + mirror_machine.admit_new('2', replacing('a', 'b'), &pull()); + assert!(mirror_machine.withdrawals().is_empty()); + mirror_machine.admit_late('2', &pull()).unwrap(); }