diff --git a/crates/zonedata/src/lib.rs b/crates/zonedata/src/lib.rs index 2e62d1f3a..8d12071c6 100644 --- a/crates/zonedata/src/lib.rs +++ b/crates/zonedata/src/lib.rs @@ -180,6 +180,11 @@ pub use persister::{ LoadedZonePersisted, LoadedZonePersister, SignedZonePersisted, SignedZonePersister, }; +mod restorer; +pub use restorer::{ + LoadedZoneRestored, LoadedZoneRestorer, SignedZoneRestored, SignedZoneRestorer, +}; + mod data; use data::{Data, InstanceData}; diff --git a/crates/zonedata/src/persister.rs b/crates/zonedata/src/persister.rs index ae19663dc..d158acf6c 100644 --- a/crates/zonedata/src/persister.rs +++ b/crates/zonedata/src/persister.rs @@ -2,7 +2,7 @@ use std::sync::Arc; -use crate::{Data, DiffData}; +use crate::{Data, DiffData, LoadedZoneReader, SignedZoneReader}; //----------- LoadedZonePersister ---------------------------------------------- @@ -52,18 +52,33 @@ impl LoadedZonePersister { } impl LoadedZonePersister { - /// Perform the actual persisting. - pub fn persist(self) -> LoadedZonePersisted { - let LoadedZonePersister { - data, - loaded_index, - loaded_diff, - } = self; + /// Read the instance needing persistence (if it is non-empty). + pub fn read(&self) -> Option> { + let loaded = &self.data.loaded[self.loaded_index as usize]; + + // SAFETY: As per invariant 'loaded-access', 'loaded' will not be + // modified for the lifetime of 'self', and thus it is sound to access + // by shared reference. + let loaded = unsafe { &*loaded.get() }; - // TODO - let _ = (loaded_index, loaded_diff); + loaded.soa.as_ref()?; - LoadedZonePersisted { data } + // NOTE: As checked above, 'loaded' is complete (i.e. has a SOA record), + // so 'LoadedZoneReader::new()' will not panic. + Some(LoadedZoneReader::new(loaded)) + } + + /// The diff from the preceding instance to the current one. + pub fn loaded_diff(&self) -> &Arc { + &self.loaded_diff + } + + /// Mark persistence as complete. + /// + /// This should be called once the instance has been read and persisted to + /// disk. + pub fn mark_complete(self) -> LoadedZonePersisted { + LoadedZonePersisted { data: self.data } } } @@ -77,6 +92,14 @@ pub struct SignedZonePersister { /// The underlying data. data: Arc, + /// The index of the associated loaded instance, if any. + /// + /// ## Invariants + /// + /// - `loaded-access`: `data.loaded[loaded_index]` is sound to access + /// immutably for the lifetime of `self`. It will not be modified. + loaded_index: bool, + /// The index of the signed instance to persist, if any. /// /// ## Invariants @@ -85,6 +108,9 @@ pub struct SignedZonePersister { /// immutably for the lifetime of `self`. It will not be modified. signed_index: bool, + /// The diff of the loaded component from the prior instance, if any. + loaded_diff: Option>, + /// The diff of the signed component from the preceding instance. signed_diff: Arc, } @@ -101,32 +127,65 @@ impl SignedZonePersister { /// of `persister` (starting from this function call). pub(crate) unsafe fn new( data: Arc, + loaded_index: bool, signed_index: bool, + loaded_diff: Option>, signed_diff: Arc, ) -> Self { // Invariants: // - 'signed-access' is guaranteed by the caller. Self { data, + loaded_index, signed_index, + loaded_diff, signed_diff, } } } impl SignedZonePersister { - /// Perform the actual persisting. - pub fn persist(self) -> SignedZonePersisted { - let SignedZonePersister { - data, - signed_index, - signed_diff, - } = self; + /// Read the instance needing persistence (if it is non-empty). + pub fn read(&self) -> Option> { + let loaded = &self.data.loaded[self.loaded_index as usize]; + let signed = &self.data.signed[self.signed_index as usize]; + + // SAFETY: As per invariant 'loaded-access', 'loaded' will not be + // modified for the lifetime of 'self', and thus it is sound to access + // by shared reference. + let loaded = unsafe { &*loaded.get() }; + + // SAFETY: As per invariant 'signed-access', 'signed' will not be + // modified for the lifetime of 'self', and thus it is sound to access + // by shared reference. + let signed = unsafe { &*signed.get() }; + + signed.soa.as_ref()?; + + // NOTE: As checked above, 'signed' is complete (i.e. has a SOA record), + // and thus 'loaded' must also be complete, so 'SignedZoneReader::new()' + // will not panic. + Some(SignedZoneReader::new(loaded, signed)) + } - // TODO - let _ = (signed_index, signed_diff); + /// The diff from the preceding loaded instance to the current one. + /// + /// This is `None` iff a re-signing occurred. + pub fn loaded_diff(&self) -> Option<&Arc> { + self.loaded_diff.as_ref() + } - SignedZonePersisted { data } + /// The diff from the preceding signed instance to the current one. + pub fn signed_diff(&self) -> &Arc { + &self.signed_diff + } + + /// Mark persistence as complete. + /// + /// This should be called once the instance has been read and persisted to + /// disk. + pub fn mark_complete(self) -> SignedZonePersisted { + SignedZonePersisted { data: self.data } } } diff --git a/crates/zonedata/src/restorer.rs b/crates/zonedata/src/restorer.rs new file mode 100644 index 000000000..98f6d6974 --- /dev/null +++ b/crates/zonedata/src/restorer.rs @@ -0,0 +1,532 @@ +//! Restoring persisted instances of zones. + +use std::sync::Arc; + +use crate::{ + Data, DiffData, LoadedZonePatcher, LoadedZoneReader, LoadedZoneReplacer, SignedZonePatcher, + SignedZoneReader, SignedZoneReplacer, +}; + +//----------- LoadedZoneRestorer ----------------------------------------------- + +/// A restorer for a persisted loaded instance of a zone. +/// +/// A [`LoadedZoneRestorer`] restores a persisted instance of a zone from disk, +/// in order to resume serving the zone when Cascade starts up. +pub struct LoadedZoneRestorer { + /// The underlying data. + data: Arc, + + /// The index of the loaded instance to build into. + index: bool, + + /// The diff of the built loaded instance. + /// + /// If the new loaded instance has been built, this field is `Some`, and it + /// provides a diff mapping the current instance to the new one. + // + // TODO: It would be nice to use 'UniqueArc' here. + diff: Option>, +} + +impl LoadedZoneRestorer { + /// Construct a new [`LoadedZoneRestorer`]. + /// + /// ## Panics + /// + /// Panics if `data.loaded[]` is not empty. + /// + /// ## Safety + /// + /// `restorer = LoadedZoneRestorer::new(...)` is sound if and only if all + /// the following conditions are satisfied: + /// + /// - `data.loaded` will not be accessed outside of `restorer` (starting + /// from this function call). + pub(crate) unsafe fn new(data: Arc) -> Self { + // SAFETY: As per the caller, 'loaded[]' will not be accessed elsewhere, + // and so is sound to access immutably. + let target = unsafe { &*data.loaded[0].get() }; + assert!(target.soa.is_none(), "The target instance is not empty"); + let target = unsafe { &*data.loaded[1].get() }; + assert!(target.soa.is_none(), "The target instance is not empty"); + + // Invariants: + // + // - 'loaded-access' is guaranteed by the caller. + // - 'built': + // - 'loaded[index]' is empty as checked above. + // - 'diff' is 'None'. + Self { + data, + index: false, + diff: None, + } + } + + /// The underlying data. + pub(crate) const fn data(&self) -> &Arc { + &self.data + } +} + +impl LoadedZoneRestorer { + /// Build the new instance from an absolute source. + /// + /// A [`LoadedZoneReplacer`] is returned, which can be used to write the + /// records in the zone. + /// + /// If the instance has already been built, [`None`] is returned. + /// + /// Use [`Self::patch()`] to build the instance with diffs. + pub fn fill(&mut self) -> Option> { + // SAFETY: As per the caller, 'loaded[index]' will not be accessed + // elsewhere for the lifetime of 'self', and so is sound to access + // mutably. + let next = unsafe { &mut *self.data.loaded[self.index as usize].get() }; + + // SAFETY: As per the caller, 'loaded[!index]' will not be accessed + // elsewhere for the lifetime of 'self', and so is sound to access + // mutably. + let curr = unsafe { &*self.data.loaded[!self.index as usize].get() }; + + if next.soa.is_some() { + // Cannot build the initial instance again. + return None; + } + + // NOTE: + // - 'next' is empty following 'LoadedZoneRestorer::new()'. + // - 'next' may be modified by 'LoadedZoneReplacer' or + // 'LoadedZonePatcher', but they will set 'built' on success, and + // clean up 'next' on failure (in drop). + // - 'next' is only non-empty if a patcher/replacer was leaked, in which + // case a panic is appropriate. + // - 'diff' is empty, as checked by 'built()' above. + Some(LoadedZoneReplacer::new(curr, next, &mut self.diff)) + } + + /// Patch the instance being built. + /// + /// A [`LoadedZonePatcher`] is returned, which can be used to restore the + /// instance from a series of diffs. + /// + /// If an initial instance has not yet been built, [`None`] is returned. + /// + /// Use [`Self::fill()`] to build the initial instance. + pub fn patch(&mut self) -> Option> { + // SAFETY: As per the caller, 'loaded[index]' will not be + // accessed elsewhere for the lifetime of 'self', and so is sound to + // access mutably. + let mut curr = unsafe { &mut *self.data.loaded[self.index as usize].get() }; + + // SAFETY: As per the caller, 'loaded[!index]' will not be accessed + // elsewhere for the lifetime of 'self', and so is sound to access + // mutably. + let mut next = unsafe { &mut *self.data.loaded[!self.index as usize].get() }; + + // If nothing was previously built, stop. + curr.soa.as_ref()?; + + // If something new has already been built, switch to it. + if next.soa.is_some() { + std::mem::swap(&mut curr, &mut next); + self.index = !self.index; + next.soa = None; + next.records.clear(); + } + + self.diff = None; + + // NOTE: + // - 'curr' is complete, as checked above. + // - 'next' is empty following 'ZoneRestorer::new()'. + // - 'next' may be modified by 'LoadedZoneReplacer' or + // 'LoadedZonePatcher', but they will set 'built_loaded' on success, + // and clean up 'next' on failure (in drop). + // - 'next' is only non-empty if a patcher/replacer was leaked, in which + // case a panic is appropriate. + // - 'diff' is empty, as checked by 'built()' above. + Some(LoadedZonePatcher::new(curr, next, &mut self.diff)) + } + + /// Clear the instance. + /// + /// A new, empty instance is created. If a new instance was already built, + /// it will be overwritten. + pub fn clear(&mut self) { + // Initialize the absolute data. + + // SAFETY: As per the caller, 'loaded[index]' will not be + // accessed elsewhere for the lifetime of 'self', and so is sound to + // access mutably. + let next = unsafe { &mut *self.data.loaded[self.index as usize].get() }; + next.soa = None; + next.records.clear(); + + // SAFETY: As per the caller, 'loaded[!index]' will not be + // accessed elsewhere for the lifetime of 'self', and so is sound to + // access mutably. + let curr = unsafe { &mut *self.data.loaded[!self.index as usize].get() }; + curr.soa = None; + curr.records.clear(); + + // Set the diff. + self.diff = Some(Box::new(DiffData::new())); + } + + /// The new (loaded) instance. + /// + /// If a new instance has been built (with [`Self::fill()`] or + /// [`Self::patch()`]), it can be accessed here. Note that empty instances + /// (as built by [`Self::clear()`]) cannot be accessed. + pub fn next(&self) -> Option> { + // SAFETY: As per the caller, 'loaded[index]' will not be accessed + // elsewhere for the lifetime of 'self', and so is sound to access + // immutably. + let curr = unsafe { &*self.data.loaded[self.index as usize].get() }; + + // SAFETY: As per the caller, 'loaded[!index]' will not be accessed + // elsewhere for the lifetime of 'self', and so is sound to access + // immutably. + let next = unsafe { &*self.data.loaded[!self.index as usize].get() }; + + // Pick the first complete instance between 'next' and 'curr'. + [next, curr] + .into_iter() + .find(|inst| inst.soa.is_some()) + .map(LoadedZoneReader::new) + } +} + +impl LoadedZoneRestorer { + /// Finish restoring the instance. + /// + /// If a new instance has been built (with [`Self::fill()`], + /// [`Self::patch()`], or [`Self::clear()`]), a [`LoadedZoneRestored`] + /// marker is returned to prove it. Otherwise, `self` is returned to try + /// again. + pub fn finish(self) -> Result { + // SAFETY: As per the caller, 'loaded[index]' will not be accessed + // elsewhere for the lifetime of 'self', and so is sound to access + // mutably. + let curr = unsafe { &mut *self.data.loaded[self.index as usize].get() }; + + // SAFETY: As per the caller, 'loaded[!index]' will not be accessed + // elsewhere for the lifetime of 'self', and so is sound to access + // mutably. + let next = unsafe { &mut *self.data.loaded[!self.index as usize].get() }; + + if next.soa.is_some() { + // Make sure 'curr' is empty. + curr.soa = None; + curr.records.clear(); + + Ok(LoadedZoneRestored { + data: self.data, + index: !self.index, + }) + } else if curr.soa.is_some() { + Ok(LoadedZoneRestored { + data: self.data, + index: self.index, + }) + } else { + Err(self) + } + } +} + +//----------- SignedZoneRestorer ----------------------------------------------- + +/// A restorer for a persisted signed instance of a zone. +/// +/// A [`SignedZoneRestorer`] restores a persisted instance of a zone from disk, +/// in order to resume serving the zone when Cascade starts up. +pub struct SignedZoneRestorer { + /// The underlying data. + data: Arc, + + /// The index of the associated loaded instance. + /// + /// ## Invariants + /// + /// - `loaded-access`: `data.loaded[loaded_index]` is sound to access + /// immutably for the lifetime of `self`. It will not be modified. + loaded_index: bool, + + /// The index of the signed instance to build into. + index: bool, + + /// The diff of the built signed instance. + /// + /// If the new signed instance has been built, this field is `Some`, and it + /// provides a diff mapping the current instance to the new one. + // + // TODO: It would be nice to use 'UniqueArc' here. + diff: Option>, +} + +impl SignedZoneRestorer { + /// Construct a new [`SignedZoneRestorer`]. + /// + /// ## Panics + /// + /// Panics if `data.signed[]` is not empty. + /// + /// ## Safety + /// + /// `restorer = SignedZoneRestorer::new(...)` is sound if and only if all + /// the following conditions are satisfied: + /// + /// - `data.loaded[loaded_index]` will not be modified for the lifetime of + /// `restorer` (starting from this function call). + /// + /// - `data.signed` will not be accessed outside of `restorer` (starting + /// from this function call). + pub(crate) unsafe fn new(data: Arc, loaded_index: bool) -> Self { + // SAFETY: As per the caller, 'signed[]' will not be accessed elsewhere, + // and so is sound to access immutably. + let target = unsafe { &*data.signed[0].get() }; + assert!(target.soa.is_none(), "The target instance is not empty"); + let target = unsafe { &*data.signed[1].get() }; + assert!(target.soa.is_none(), "The target instance is not empty"); + + // Invariants: + // + // - 'loaded-access' is guaranteed by the caller. + // - 'signed-access' is guaranteed by the caller. + // - 'built': + // - 'signed[index]' is empty as checked above. + // - 'diff' is 'None'. + Self { + data, + loaded_index, + index: false, + diff: None, + } + } + + /// The underlying data. + pub(crate) const fn data(&self) -> &Arc { + &self.data + } +} + +impl SignedZoneRestorer { + /// Build the new instance from an absolute source. + /// + /// A [`SignedZoneReplacer`] is returned, which can be used to write the + /// records in the zone. + /// + /// If the instance has already been built, [`None`] is returned. + /// + /// Use [`Self::patch()`] to build the instance with diffs. + pub fn fill(&mut self) -> Option> { + // SAFETY: As per the caller, 'loaded[loaded_index]' will not be + // modified for the lifetime of 'self', and so is sound to access + // immutably. + let curr_loaded = unsafe { &*self.data.signed[self.loaded_index as usize].get() }; + + // SAFETY: As per the caller, 'signed[index]' will not be accessed + // elsewhere for the lifetime of 'self', and so is sound to access + // mutably. + let next = unsafe { &mut *self.data.signed[self.index as usize].get() }; + + // SAFETY: As per the caller, 'signed[!index]' will not be accessed + // elsewhere for the lifetime of 'self', and so is sound to access + // mutably. + let curr = unsafe { &*self.data.signed[!self.index as usize].get() }; + + if next.soa.is_some() { + // Cannot build the initial instance again. + return None; + } + + // NOTE: + // - 'next' is empty following 'SignedZoneRestorer::new()'. + // - 'next' may be modified by 'SignedZoneReplacer' or + // 'SignedZonePatcher', but they will set 'built' on success, and + // clean up 'next' on failure (in drop). + // - 'next' is only non-empty if a patcher/replacer was leaked, in which + // case a panic is appropriate. + // - 'diff' is empty, as checked by 'built()' above. + Some(SignedZoneReplacer::new( + curr_loaded, + None, + None, + curr, + next, + &mut self.diff, + )) + } + + /// Patch the instance being built. + /// + /// A [`SignedZonePatcher`] is returned, which can be used to restore the + /// instance from a series of diffs. + /// + /// If an initial instance has not yet been built, [`None`] is returned. + /// + /// Use [`Self::fill()`] to build the initial instance. + pub fn patch(&mut self) -> Option> { + // SAFETY: As per the caller, 'loaded[loaded_index]' will not be + // modified for the lifetime of 'self', and so is sound to access + // immutably. + let curr_loaded = unsafe { &*self.data.signed[self.loaded_index as usize].get() }; + + // SAFETY: As per the caller, 'signed[index]' will not be + // accessed elsewhere for the lifetime of 'self', and so is sound to + // access mutably. + let mut curr = unsafe { &mut *self.data.signed[self.index as usize].get() }; + + // SAFETY: As per the caller, 'signed[!index]' will not be accessed + // elsewhere for the lifetime of 'self', and so is sound to access + // mutably. + let mut next = unsafe { &mut *self.data.signed[!self.index as usize].get() }; + + // If nothing was previously built, stop. + curr.soa.as_ref()?; + + // If something new has already been built, switch to it. + if next.soa.is_some() { + std::mem::swap(&mut curr, &mut next); + self.index = !self.index; + next.soa = None; + next.records.clear(); + } + + self.diff = None; + + // NOTE: + // - 'curr' is complete, as checked above. + // - 'next' is empty following 'ZoneRestorer::new()'. + // - 'next' may be modified by 'SignedZoneReplacer' or + // 'SignedZonePatcher', but they will set 'built_signed' on success, + // and clean up 'next' on failure (in drop). + // - 'next' is only non-empty if a patcher/replacer was leaked, in which + // case a panic is appropriate. + // - 'diff' is empty, as checked by 'built()' above. + Some(SignedZonePatcher::new( + curr_loaded, + None, + None, + curr, + next, + &mut self.diff, + )) + } + + /// Clear the instance. + /// + /// A new, empty instance is created. If a new instance was already built, + /// it will be overwritten. + pub fn clear(&mut self) { + // Initialize the absolute data. + + // SAFETY: As per the caller, 'signed[index]' will not be + // accessed elsewhere for the lifetime of 'self', and so is sound to + // access mutably. + let next = unsafe { &mut *self.data.signed[self.index as usize].get() }; + next.soa = None; + next.records.clear(); + + // SAFETY: As per the caller, 'signed[!index]' will not be + // accessed elsewhere for the lifetime of 'self', and so is sound to + // access mutably. + let curr = unsafe { &mut *self.data.signed[!self.index as usize].get() }; + curr.soa = None; + curr.records.clear(); + + // Set the diff. + self.diff = Some(Box::new(DiffData::new())); + } + + /// The new (signed) instance. + /// + /// If a new instance has been built (with [`Self::fill()`] or + /// [`Self::patch()`]), it can be accessed here. Note that empty instances + /// (as built by [`Self::clear()`]) cannot be accessed. + pub fn next(&self) -> Option> { + // SAFETY: As per the caller, 'loaded[loaded_index]' will not be + // modified for the lifetime of 'self', and so is sound to access + // immutably. + let curr_loaded = unsafe { &*self.data.signed[self.loaded_index as usize].get() }; + + // SAFETY: As per the caller, 'signed[index]' will not be accessed + // elsewhere for the lifetime of 'self', and so is sound to access + // immutably. + let curr = unsafe { &*self.data.signed[self.index as usize].get() }; + + // SAFETY: As per the caller, 'signed[!index]' will not be accessed + // elsewhere for the lifetime of 'self', and so is sound to access + // immutably. + let next = unsafe { &*self.data.signed[!self.index as usize].get() }; + + // Pick the first complete instance between 'next' and 'curr'. + [next, curr] + .into_iter() + .find(|inst| inst.soa.is_some()) + .map(|inst| SignedZoneReader::new(curr_loaded, inst)) + } +} + +impl SignedZoneRestorer { + /// Finish restoring the instance. + /// + /// If a new instance has been built (with [`Self::fill()`], + /// [`Self::patch()`], or [`Self::clear()`]), a [`SignedZoneRestored`] + /// marker is returned to prove it. Otherwise, `self` is returned to try + /// again. + pub fn finish(self) -> Result { + // SAFETY: As per the caller, 'signed[index]' will not be accessed + // elsewhere for the lifetime of 'self', and so is sound to access + // mutably. + let curr = unsafe { &mut *self.data.signed[self.index as usize].get() }; + + // SAFETY: As per the caller, 'signed[!index]' will not be accessed + // elsewhere for the lifetime of 'self', and so is sound to access + // mutably. + let next = unsafe { &mut *self.data.signed[!self.index as usize].get() }; + + if next.soa.is_some() { + // Make sure 'curr' is empty. + curr.soa = None; + curr.records.clear(); + + Ok(SignedZoneRestored { + data: self.data, + index: !self.index, + }) + } else if curr.soa.is_some() { + Ok(SignedZoneRestored { + data: self.data, + index: self.index, + }) + } else { + Err(self) + } + } +} + +//----------- LoadedZoneRestored ----------------------------------------------- + +/// Proof that a loaded instance of a zone has been restored. +pub struct LoadedZoneRestored { + /// The underlying data. + pub(crate) data: Arc, + + /// The index of the restored instance. + pub(crate) index: bool, +} + +//----------- SignedZoneRestored ----------------------------------------------- + +/// Proof that a signed instance of a zone has been restored. +pub struct SignedZoneRestored { + /// The underlying data. + pub(crate) data: Arc, + + /// The index of the restored instance. + pub(crate) index: bool, +} diff --git a/crates/zonedata/src/storage/mod.rs b/crates/zonedata/src/storage/mod.rs index 1ec21d8bd..ea92d84d4 100644 --- a/crates/zonedata/src/storage/mod.rs +++ b/crates/zonedata/src/storage/mod.rs @@ -7,15 +7,15 @@ use std::sync::Arc; -use crate::{LoadedZoneReviewer, SignedZoneReviewer, ZoneViewer, data::Data}; +use crate::{LoadedZoneRestorer, LoadedZoneReviewer, SignedZoneReviewer, ZoneViewer, data::Data}; mod states; pub use states::{ CleanLoadedPendingStorage, CleanSignedPendingStorage, CleanWholePendingStorage, CleaningSignedStorage, CleaningStorage, LoadingStorage, PassiveStorage, - PersistingLoadedStorage, PersistingSignedStorage, ReviewLoadedPendingStorage, - ReviewSignedPendingStorage, ReviewingLoadedStorage, ReviewingSignedStorage, SigningStorage, - SwitchingStorage, + PersistingLoadedStorage, PersistingSignedStorage, RestoringLoadedStorage, + RestoringSignedStorage, ReviewLoadedPendingStorage, ReviewSignedPendingStorage, + ReviewingLoadedStorage, ReviewingSignedStorage, SigningStorage, SwitchingStorage, }; mod transitions; @@ -30,6 +30,12 @@ mod transitions; /// it is designed to live in a (synchronous) mutex -- expensive operations on /// the zone are always achievable without `&mut` access. pub enum ZoneDataStorage { + /// The loaded instance of the zone is being restored. + RestoringLoaded(RestoringLoadedStorage), + + /// The signed instance of the zone is being restored. + RestoringSigned(RestoringSignedStorage), + /// The zone is passive. Passive(PassiveStorage), @@ -85,36 +91,14 @@ pub enum ZoneDataStorage { impl ZoneDataStorage { /// Construct a new [`ZoneDataStorage`]. - pub fn new() -> (Self, LoadedZoneReviewer, SignedZoneReviewer, ZoneViewer) { - // TODO: When Cascade starts up, it should check for existing instances - // on disk. This might require a separate initialization function. - + pub fn new() -> (LoadedZoneRestorer, Self) { let data = Arc::new(Data::new()); - let curr_unsigned_index = false; - let curr_signed_index = false; - - let ureviewer = unsafe { LoadedZoneReviewer::new(data.clone(), curr_unsigned_index, None) }; - - let reviewer = unsafe { - SignedZoneReviewer::new( - data.clone(), - curr_unsigned_index, - curr_signed_index, - None, - None, - ) - }; - - let viewer = - unsafe { ZoneViewer::new(data.clone(), curr_unsigned_index, curr_signed_index) }; - - let storage = Self::Passive(PassiveStorage { - data, - curr_loaded_index: false, - curr_signed_index: false, - }); - - (storage, ureviewer, reviewer, viewer) + + let restorer = unsafe { LoadedZoneRestorer::new(data.clone()) }; + + let storage = RestoringLoadedStorage { data }; + + (restorer, Self::RestoringLoaded(storage)) } /// Extract the current state of the [`ZoneDataStorage`]. @@ -131,6 +115,8 @@ impl ZoneDataStorage { /// This is intended for logging and debugging. pub const fn as_str(&self) -> &'static str { match self { + ZoneDataStorage::RestoringLoaded(_) => "RestoringLoaded", + ZoneDataStorage::RestoringSigned(_) => "RestoringSigned", ZoneDataStorage::Passive(_) => "Passive", ZoneDataStorage::Loading(_) => "Loading", ZoneDataStorage::Signing(_) => "Signing", diff --git a/crates/zonedata/src/storage/states.rs b/crates/zonedata/src/storage/states.rs index a5cba9b63..0e21ca8bc 100644 --- a/crates/zonedata/src/storage/states.rs +++ b/crates/zonedata/src/storage/states.rs @@ -9,10 +9,57 @@ use super::ZoneDataStorage; #[cfg(doc)] use crate::{ - LoadedZoneBuilder, LoadedZonePersister, LoadedZoneReviewer, SignedZoneBuilder, - SignedZoneCleaner, SignedZonePersister, SignedZoneReviewer, ZoneCleaner, ZoneViewer, + LoadedZoneBuilder, LoadedZonePersister, LoadedZoneRestorer, LoadedZoneReviewer, + SignedZoneBuilder, SignedZoneCleaner, SignedZonePersister, SignedZoneRestorer, + SignedZoneReviewer, ZoneCleaner, ZoneViewer, }; +//----------- RestoringLoadedStorage ------------------------------------------- + +/// The [`ZoneDataStorage::RestoringLoaded`] state. +/// +/// This is the initial state, in which a persisted loaded instance of the zone +/// can be restored. +/// +/// ## Data +/// +/// There are no current loaded or signed instances of the zone. +/// +/// There is an upcoming loaded instance of the zone. +/// +/// ## Access +/// +/// The [`LoadedZoneRestorer`] points to the upcoming instance. +pub struct RestoringLoadedStorage { + /// The underlying data. + pub(super) data: Arc, +} + +//----------- RestoringSignedStorage ------------------------------------------- + +/// The [`ZoneDataStorage::RestoringSigned`] state. +/// +/// This state follows [`RestoringLoadedStorage`]; in it, a persisted signed +/// instance of the zone can be restored. +/// +/// ## Data +/// +/// There is a current loaded instance of the zone. There is no current signed +/// instance of the zone. +/// +/// There is an upcoming signed instance of the zone. +/// +/// ## Access +/// +/// The [`SignedZoneRestorer`] points to the upcoming instance. +pub struct RestoringSignedStorage { + /// The underlying data. + pub(super) data: Arc, + + /// The index of the current loaded instance. + pub(super) curr_loaded_index: bool, +} + //----------- PassiveStorage --------------------------------------------------- /// The [`ZoneDataStorage::Passive`] state. @@ -31,9 +78,6 @@ use crate::{ /// /// The [`LoadedZoneReviewer`], [`SignedZoneReviewer`], and [`ZoneViewer`] all /// point to the current instances. -/// -/// There is no [`LoadedZoneBuilder`], [`SignedZoneBuilder`], [`ZoneCleaner`], -/// [`SignedZoneCleaner`], [`LoadedZonePersister`], or [`SignedZonePersister`]. pub struct PassiveStorage { /// The underlying data. pub(super) data: Arc, diff --git a/crates/zonedata/src/storage/transitions.rs b/crates/zonedata/src/storage/transitions.rs index b2976ecfe..02831a430 100644 --- a/crates/zonedata/src/storage/transitions.rs +++ b/crates/zonedata/src/storage/transitions.rs @@ -4,18 +4,183 @@ use std::sync::Arc; use crate::{ LoadedZoneBuilder, LoadedZoneBuilt, LoadedZonePersisted, LoadedZonePersister, - SignedZoneBuilder, SignedZoneBuilt, SignedZoneCleaned, SignedZoneCleaner, SignedZonePersisted, - SignedZonePersister, ZoneCleaned, ZoneCleaner, + LoadedZoneRestored, LoadedZoneRestorer, SignedZoneBuilder, SignedZoneBuilt, SignedZoneCleaned, + SignedZoneCleaner, SignedZonePersisted, SignedZonePersister, SignedZoneRestored, + SignedZoneRestorer, ZoneCleaned, ZoneCleaner, }; use super::{ CleanLoadedPendingStorage, CleanSignedPendingStorage, CleanWholePendingStorage, CleaningSignedStorage, CleaningStorage, LoadedZoneReviewer, LoadingStorage, PassiveStorage, - PersistingLoadedStorage, PersistingSignedStorage, ReviewLoadedPendingStorage, - ReviewSignedPendingStorage, ReviewingLoadedStorage, ReviewingSignedStorage, SignedZoneReviewer, - SigningStorage, SwitchingStorage, ZoneViewer, + PersistingLoadedStorage, PersistingSignedStorage, RestoringLoadedStorage, + RestoringSignedStorage, ReviewLoadedPendingStorage, ReviewSignedPendingStorage, + ReviewingLoadedStorage, ReviewingSignedStorage, SignedZoneReviewer, SigningStorage, + SwitchingStorage, ZoneViewer, }; +//----------- RestoringLoadedStorage ------------------------------------------- + +impl RestoringLoadedStorage { + /// Finish restoring the loaded instance. + pub fn finish( + self, + restored: LoadedZoneRestored, + ) -> (SignedZoneRestorer, RestoringSignedStorage) { + assert!( + Arc::ptr_eq(&restored.data, &self.data), + "'restored' is for a different zone" + ); + + let restorer = unsafe { SignedZoneRestorer::new(self.data.clone(), restored.index) }; + + let storage = RestoringSignedStorage { + data: self.data, + curr_loaded_index: restored.index, + }; + + (restorer, storage) + } + + /// Abandon the restore operation. + pub fn abandon( + self, + mut restorer: LoadedZoneRestorer, + ) -> ( + LoadedZoneReviewer, + SignedZoneReviewer, + ZoneViewer, + PassiveStorage, + ) { + assert!( + Arc::ptr_eq(restorer.data(), &self.data), + "'restorer' is for a different zone" + ); + + restorer.clear(); + let Self { data } = self; + let curr_loaded_index = false; + let curr_signed_index = false; + + let loaded_reviewer = + unsafe { LoadedZoneReviewer::new(data.clone(), curr_loaded_index, None) }; + let signed_reviewer = unsafe { + SignedZoneReviewer::new( + data.clone(), + curr_loaded_index, + curr_signed_index, + None, + None, + ) + }; + let viewer = unsafe { ZoneViewer::new(data.clone(), curr_loaded_index, curr_signed_index) }; + + let storage = PassiveStorage { + data, + curr_loaded_index, + curr_signed_index, + }; + + (loaded_reviewer, signed_reviewer, viewer, storage) + } +} + +//----------- RestoringSignedStorage ------------------------------------------- + +impl RestoringSignedStorage { + /// Finish restoring the signed instance. + pub fn finish( + self, + restored: SignedZoneRestored, + ) -> ( + LoadedZoneReviewer, + SignedZoneReviewer, + ZoneViewer, + PassiveStorage, + ) { + assert!( + Arc::ptr_eq(&restored.data, &self.data), + "'restored' is for a different zone" + ); + + let Self { + data, + curr_loaded_index, + } = self; + let curr_signed_index = restored.index; + + let loaded_reviewer = + unsafe { LoadedZoneReviewer::new(data.clone(), curr_loaded_index, None) }; + let signed_reviewer = unsafe { + SignedZoneReviewer::new( + data.clone(), + curr_loaded_index, + curr_signed_index, + None, + None, + ) + }; + let viewer = unsafe { ZoneViewer::new(data.clone(), curr_loaded_index, curr_signed_index) }; + + let storage = PassiveStorage { + data, + curr_loaded_index, + curr_signed_index, + }; + + (loaded_reviewer, signed_reviewer, viewer, storage) + } + + /// Abandon the restore operation. + pub fn abandon( + self, + mut restorer: SignedZoneRestorer, + ) -> ( + LoadedZoneReviewer, + SignedZoneReviewer, + ZoneViewer, + SignedZoneBuilder, + SigningStorage, + ) { + assert!( + Arc::ptr_eq(restorer.data(), &self.data), + "'restorer' is for a different zone" + ); + + restorer.clear(); + + let Self { + data, + curr_loaded_index, + } = self; + let curr_signed_index = false; + + let loaded_reviewer = + unsafe { LoadedZoneReviewer::new(data.clone(), curr_loaded_index, None) }; + let signed_reviewer = unsafe { + SignedZoneReviewer::new( + data.clone(), + curr_loaded_index, + curr_signed_index, + None, + None, + ) + }; + let viewer = unsafe { ZoneViewer::new(data.clone(), curr_loaded_index, curr_signed_index) }; + let builder = unsafe { + SignedZoneBuilder::new(data.clone(), curr_loaded_index, !curr_signed_index, None) + }; + + let storage = SigningStorage { + data, + curr_loaded_index, + curr_signed_index, + loaded_diff: None, + }; + + (loaded_reviewer, signed_reviewer, viewer, builder, storage) + } +} + //----------- PassiveStorage --------------------------------------------------- impl PassiveStorage { @@ -294,7 +459,9 @@ impl ReviewingSignedStorage { let persister = unsafe { SignedZonePersister::new( self.data.clone(), + self.curr_loaded_index ^ self.loaded_diff.is_some(), !self.curr_signed_index, + self.loaded_diff.clone(), self.signed_diff.clone(), ) }; diff --git a/crates/zonedata/src/viewer.rs b/crates/zonedata/src/viewer.rs index ae37d6213..c5a9f604d 100644 --- a/crates/zonedata/src/viewer.rs +++ b/crates/zonedata/src/viewer.rs @@ -169,8 +169,8 @@ impl LoadedZoneReviewer { } impl LoadedZoneReviewer { - /// Read the loaded component, if there is one. - pub fn read_loaded(&self) -> Option> { + /// Read the instance, if it is non-empty. + pub fn read(&self) -> Option> { let instance = &self.data.loaded[self.loaded_index as usize]; // SAFETY: As per invariant 'loaded-access', 'instance' will not be diff --git a/src/center.rs b/src/center.rs index ac5e27875..d7f976a4b 100644 --- a/src/center.rs +++ b/src/center.rs @@ -16,7 +16,9 @@ use crate::api::KeyImport; use crate::config::RuntimeConfig; use crate::loader::Loader; use crate::loader::zone::LoaderZoneHandle; +use crate::persistence::{Persister, Restorer}; use crate::server::{LoadedReviewServer, PublicationServer, SignedReviewServer}; +use crate::state::PolicySpec; use crate::units::key_manager::KeyManager; use crate::units::zone_signer::ZoneSigner; use crate::zone::{HistoricalEvent, ZoneHandle}; @@ -46,12 +48,18 @@ pub struct Center { /// The zone loader. pub loader: Loader, - /// The zone signer + /// The zone signer. pub signer: ZoneSigner, - /// The key manager + /// The key manager. pub key_manager: KeyManager, + /// The zone data persister. + pub persister: Persister, + + /// The zone data restorer. + pub restorer: Restorer, + /// The review server for loaded instances of zones. pub loaded_review_server: LoadedReviewServer, @@ -97,14 +105,23 @@ pub async fn add_zone( // Create the zone and initialize its state. zone = Arc::new(Zone::new(name)); - let (loaded_reviewer, signed_reviewer, viewer); { let mut zone_state = zone.state.lock().unwrap(); + let restorer = zone_state.storage.restorer.take().unwrap(); zone_state.policy = Some(policy.latest.clone()); policy.zones.insert(zone.name.clone()); - loaded_reviewer = zone_state.storage.loaded_reviewer.take().unwrap(); - signed_reviewer = zone_state.storage.signed_reviewer.take().unwrap(); - viewer = zone_state.storage.viewer.take().unwrap(); + + // Don't try to restore zone data, since it's a completely new zone. + // + // This will clear the data for the zone and register it against the + // zone servers. + ZoneHandle { + zone: &zone, + state: &mut zone_state, + center, + } + .storage() + .abandon_loaded_restoration(restorer); } // Insert the zone in the global set. @@ -113,11 +130,6 @@ pub async fn add_zone( "Already checked that 'state.zones' does not contain 'name'" ); state.mark_dirty(center); - - // Update the zone servers. - LoadedReviewServer::add_zone(center, zone.clone(), loaded_reviewer); - SignedReviewServer::add_zone(center, zone.clone(), signed_reviewer); - PublicationServer::add_zone(center, zone.clone(), viewer); } // Send out a registration command so that prerequisites for zone setup @@ -279,10 +291,21 @@ pub struct State { impl State { /// Attempt to load the global state file. - pub fn init_from_file(config: &Config) -> io::Result { + /// + /// `zones` will be set to the names of zones that need to be loaded. + /// `policies` will be set to the set of policies from the global state + /// file, that need to be parsed and inserted in the state. + pub fn init_from_file( + config: &Config, + zones: &mut foldhash::HashSet>, + policies: &mut foldhash::HashMap, PolicySpec>, + ) -> io::Result { let path = config.daemon.state_file.value(); let spec = crate::state::Spec::load(path)?; - Ok(spec.parse()) + + info!("Loaded the global state file (from '{path}')"); + + Ok(spec.parse(zones, policies)) } /// Mark the global state as dirty. diff --git a/src/lib.rs b/src/lib.rs index e92e80b7b..ab5296921 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -10,6 +10,7 @@ pub mod loader; pub mod log; pub mod manager; pub mod metrics; +pub mod persistence; pub mod policy; pub mod server; pub mod signer; diff --git a/src/main.rs b/src/main.rs index dd88a6676..d6236347b 100644 --- a/src/main.rs +++ b/src/main.rs @@ -4,9 +4,11 @@ use cascaded::{ daemon::{PreBindError, SocketProvider, daemonize}, loader::Loader, manager::Manager, + persistence::{Persister, Restorer}, policy, server::{LoadedReviewServer, PublicationServer, SignedReviewServer}, units::{key_manager::KeyManager, zone_signer::ZoneSigner}, + zone::{Zone, ZoneByName}, }; use clap::{crate_authors, crate_description}; use std::{collections::HashMap, fs::create_dir_all}; @@ -71,7 +73,37 @@ fn main() -> ExitCode { } // Load the global state file or build one from scratch. - let mut state = match center::State::init_from_file(&config) { + let mut zones = Default::default(); + let mut policies = Default::default(); + let mut state = match center::State::init_from_file(&config, &mut zones, &mut policies) { + Ok(mut state) => { + // TODO: Restore the TSIG key store here, so that keys are available + // to the zones and policies being restored. + + // Restore pending zones. + for name in zones { + assert!( + !state.zones.contains(&name), + "Zone '{name}' was encountered twice" + ); + let zone = match Zone::restore(&config, name, &mut state.policies) { + Ok(zone) => zone, + Err(_) => return ExitCode::FAILURE, + }; + state.zones.insert(ZoneByName(Arc::new(zone))); + } + + // Restore pending policies. + state + .policies + .extend(policies.into_iter().map(|(name, spec)| { + let policy = spec.parse(&name); + (name, policy) + })); + + state + } + Err(err) => { if err.kind() != io::ErrorKind::NotFound { error!("Could not load the state file: {err}"); @@ -135,30 +167,6 @@ fn main() -> ExitCode { // TODO: Fail if any zone state files exist. state } - Ok(mut state) => { - info!("Successfully loaded the global state file"); - - let zone_state_dir = &config.zone_state_dir; - let policies = &mut state.policies; - for zone in &state.zones { - let name = &zone.0.name; - let path = zone_state_dir.join(format!("{name}.db")); - let spec = match cascaded::zone::state::Spec::load(&path) { - Ok(spec) => { - debug!("Loaded state of zone '{name}' (from {path})"); - spec - } - Err(err) => { - error!("Failed to load zone state '{name}' from '{path}': {err}"); - return ExitCode::FAILURE; - } - }; - let mut state = zone.0.state.lock().unwrap(); - *state = spec.parse(&zone.0, policies); - } - - state - } }; if config.loader.review.servers.is_empty() { @@ -204,6 +212,8 @@ fn main() -> ExitCode { logger, loader: Loader::new(), key_manager: KeyManager::new(), + persister: Persister::new(), + restorer: Restorer::new(), loaded_review_server: LoadedReviewServer::new(), signed_review_server: SignedReviewServer::new(), publication_server: PublicationServer::new(), @@ -241,6 +251,8 @@ fn main() -> ExitCode { } }; + info!("Cascade is fully initialized."); + let res = match tokio::signal::ctrl_c().await { Ok(_) => ExitCode::SUCCESS, Err(error) => { diff --git a/src/manager.rs b/src/manager.rs index e3eda35c1..4169ff057 100644 --- a/src/manager.rs +++ b/src/manager.rs @@ -6,6 +6,7 @@ use crate::center::Center; use crate::daemon::SocketProvider; use crate::loader::Loader; use crate::metrics::MetricsCollection; +use crate::persistence::Restorer; use crate::server::{LoadedReviewServer, PublicationServer, SignedReviewServer}; use crate::units::http_server::HTTP_UNIT_NAME; use crate::units::http_server::HttpServer; @@ -43,35 +44,36 @@ impl Manager { { let mut state = center.state.lock().unwrap(); Loader::init(¢er, &mut state); - LoadedReviewServer::init(¢er, &mut state); - SignedReviewServer::init(¢er, &mut state); - PublicationServer::init(¢er, &mut state); } let mut handles = Vec::new(); + // Spawn the zone data restorer. + debug!("Starting the zone data restorer"); + handles.push(Restorer::run(center.clone())); + // Spawn the zone loader. - info!("Starting unit 'ZL'"); + debug!("Starting the zone loader"); handles.push(Loader::run(center.clone())); - // Spawn the unsigned zone review server. - info!("Starting unit 'RS'"); + // Spawn the loaded zone review server. + debug!("Starting the loaded review server"); handles.extend(LoadedReviewServer::run(¢er, &mut socket_provider)?); // Spawn the key manager. - info!("Starting unit 'KM'"); + debug!("Starting the key manager"); handles.push(KeyManager::run(center.clone())); // Spawn the zone signer. - info!("Starting unit 'ZS'"); + debug!("Starting the zone signer"); handles.push(ZoneSigner::run(center.clone())); // Spawn the signed zone review server. - info!("Starting unit 'RS2'"); + debug!("Starting the signed review server"); handles.extend(SignedReviewServer::run(¢er, &mut socket_provider)?); - // Take out HTTP listen sockets before PS takes them all. - debug!("Pre-fetching listen sockets for 'HS'"); + // Take out HTTP listen sockets before the publication server takes them all. + debug!("Pre-fetching listen sockets for the remote-control server"); let http_sockets = center .config .remote_control @@ -89,21 +91,19 @@ impl Manager { // address. Otherwise this socket couldn't have been created. let addr = socket.local_addr().unwrap(); info!( - "Obtained TCP listener for HTTP server for remote-control and metrics on address {addr}" + "Obtained a TCP listener for the remote-control and metrics server on address {addr}" ); } - info!("Starting unit 'PS'"); + debug!("Starting the publication server"); handles.extend(PublicationServer::run(¢er, &mut socket_provider)?); - // Register any Manager metrics here, before giving the metrics to the HttpServer + // TODO: Register any `Manager` metrics here, before giving the metrics to `HttpServer`. - // Spawn the HTTP server. - info!("Starting unit 'HS'"); + // Spawn the remote-control server. + debug!("Starting the HTTP remote-control server"); let http_server = HttpServer::launch(center.clone(), http_sockets, metrics)?; - info!("All units report ready."); - Ok(Self { center, http_server, diff --git a/src/persistence/mod.rs b/src/persistence/mod.rs new file mode 100644 index 000000000..85311f245 --- /dev/null +++ b/src/persistence/mod.rs @@ -0,0 +1,105 @@ +//! Persisting zone data to and restoring from disk. +//! +//! The zone persister saves the data for loaded and signed zones to disk, so +//! that Cascade can seamlessly resume operation after a crash / restart. At +//! startup, it tries to restore data for all known zones. + +use std::sync::Arc; + +use crate::{ + center::Center, + util::AbortOnDrop, + zone::{ZoneByName, ZoneHandle}, +}; + +mod persist; +use persist::{persist_loaded, persist_signed}; + +mod restore; +use restore::{restore_loaded, restore_signed}; + +pub mod zone; + +//----------- Persister -------------------------------------------------------- + +/// The zone data persister. +/// +/// This component is responsible for persisting zone data, so it can be +/// restored (and Cascade can resume operation) after a crash / restart. +#[derive(Debug)] +pub struct Persister { + // TODO: Do we need any global state for persistence? +} + +impl Persister { + /// Construct a new [`Persister`]. + pub fn new() -> Self { + Self {} + } +} + +impl Default for Persister { + fn default() -> Self { + Self::new() + } +} + +//----------- Restorer --------------------------------------------------------- + +/// The zone data restorer. +/// +/// This component is responsible for restoring the data of persisted zones when +/// Cascade starts up. Its primary functionality is in [`Restorer::run()`]. +#[derive(Debug)] +pub struct Restorer {} + +impl Restorer { + /// Construct a new [`Restorer`]. + pub fn new() -> Self { + Self {} + } + + /// Drive this [`Restorer`]. + /// + /// At startup, the set of zones will be traversed, and for zones that were + /// restored from state files, restore operations for their zone data will + /// be initiated. + pub fn run(center: Arc
) -> AbortOnDrop { + AbortOnDrop::from(tokio::spawn(async move { + // Obtain a list of all zones (that need restoring). + let zones = { + let state = center.state.lock().unwrap(); + state + .zones + .iter() + .filter(|&z| z.0.restored) + .map(|ZoneByName(z)| z.clone()) + .collect::>() + }; + + // Attempt to restore data for every zone. + for zone in zones { + let mut state = zone.state.lock().unwrap(); + let mut handle = ZoneHandle { + zone: &zone, + state: &mut state, + center: ¢er, + }; + + // Zones that are _not_ restored from disk will move out the + // 'restorer' field and use it to initialize the zone data to + // an empty state. For zones that _are_ restored from disk, the + // 'restorer' field is moved out over here. + let restorer = handle.state.storage.restorer.take().unwrap(); + + handle.persistence().start_restore(restorer); + } + })) + } +} + +impl Default for Restorer { + fn default() -> Self { + Self::new() + } +} diff --git a/src/persistence/persist.rs b/src/persistence/persist.rs new file mode 100644 index 000000000..3981e81df --- /dev/null +++ b/src/persistence/persist.rs @@ -0,0 +1,41 @@ +//! Persisting zone data. + +use std::sync::Arc; + +use cascade_zonedata::{ + LoadedZonePersisted, LoadedZonePersister, SignedZonePersisted, SignedZonePersister, +}; + +use crate::{center::Center, zone::Zone}; + +/// Persist the data for a loaded instance of a zone. +#[tracing::instrument( + level = "trace", + skip_all, + fields(zone = %zone.name), +)] +pub fn persist_loaded( + zone: &Arc, + center: &Arc
, + persister: LoadedZonePersister, +) -> LoadedZonePersisted { + // TODO + let _ = (zone, center); + persister.mark_complete() +} + +/// Persist the data for a signed instance of a zone. +#[tracing::instrument( + level = "trace", + skip_all, + fields(zone = %zone.name), +)] +pub fn persist_signed( + zone: &Arc, + center: &Arc
, + persister: SignedZonePersister, +) -> SignedZonePersisted { + // TODO + let _ = (zone, center); + persister.mark_complete() +} diff --git a/src/persistence/restore.rs b/src/persistence/restore.rs new file mode 100644 index 000000000..bbfbb709d --- /dev/null +++ b/src/persistence/restore.rs @@ -0,0 +1,39 @@ +//! Restoring persisted zone data. + +use std::{io, sync::Arc}; + +use cascade_zonedata::{LoadedZoneRestorer, SignedZoneRestorer}; + +use crate::{center::Center, zone::Zone}; + +/// Restore the loaded instance data of a zone. +#[tracing::instrument( + level = "trace", + skip_all, + fields(zone = %zone.name), +)] +pub fn restore_loaded( + zone: &Arc, + center: &Arc
, + restorer: &mut LoadedZoneRestorer, +) -> io::Result<()> { + // TODO + let _ = (zone, center, restorer); + Err(io::Error::other("not yet implemented")) +} + +/// Restore the loaded instance data of a zone. +#[tracing::instrument( + level = "trace", + skip_all, + fields(zone = %zone.name), +)] +pub fn restore_signed( + zone: &Arc, + center: &Arc
, + restorer: &mut SignedZoneRestorer, +) -> io::Result<()> { + // TODO + let _ = (zone, center, restorer); + Err(io::Error::other("not yet implemented")) +} diff --git a/src/persistence/zone.rs b/src/persistence/zone.rs new file mode 100644 index 000000000..d5383c50d --- /dev/null +++ b/src/persistence/zone.rs @@ -0,0 +1,205 @@ +//! Zone-specific persistence management. + +use std::sync::Arc; + +use cascade_zonedata::{LoadedZonePersister, LoadedZoneRestorer, SignedZonePersister}; +use tracing::{debug, info, trace, trace_span}; + +use crate::{ + center::Center, + util::BackgroundTasks, + zone::{Zone, ZoneHandle, ZoneState}, +}; + +//----------- ZonePersistenceHandle -------------------------------------------- + +/// A handle for data persistence related operations on a [`Zone`]. +pub struct ZonePersistenceHandle<'a> { + /// The zone being operated on. + pub zone: &'a Arc, + + /// The locked zone state. + pub state: &'a mut ZoneState, + + /// Cascade's global state. + pub center: &'a Arc
, +} + +impl ZonePersistenceHandle<'_> { + /// Access the generic [`ZoneHandle`]. + pub const fn zone(&mut self) -> ZoneHandle<'_> { + ZoneHandle { + zone: self.zone, + state: self.state, + center: self.center, + } + } + + /// Begin restoring data for the zone. + /// + /// A background task will be spawned to restore the zone's data (for both + /// the loaded and signed instances). + #[tracing::instrument( + level = "trace", + skip_all, + fields(zone = %self.zone.name), + )] + pub fn start_restore(&mut self, restorer: LoadedZoneRestorer) { + let zone = self.zone.clone(); + let center = self.center.clone(); + let span = trace_span!("restore"); + self.state + .persistence + .ongoing + .spawn_blocking(span, move || { + debug!("Attempting to restore persisted zone data"); + + // Try to restore the loaded instance. + let mut restorer = restorer; + let restored = match super::restore_loaded(&zone, ¢er, &mut restorer) { + Ok(()) => restorer.finish().unwrap_or_else(|_| { + unreachable!( + "'restore_loaded()' always completes restoration on successful return" + ) + }), + Err(_) => { + trace!("Abandoning loaded restoration"); + let mut state = zone.state.lock().unwrap(); + let mut handle = ZoneHandle { + zone: &zone, + state: &mut state, + center: ¢er, + }; + handle.storage().abandon_loaded_restoration(restorer); + handle.state.persistence.ongoing.finish(); + return; + } + }; + + // Obtain the signed zone restorer. + let mut restorer = { + let mut state = zone.state.lock().unwrap(); + let mut handle = ZoneHandle { + zone: &zone, + state: &mut state, + center: ¢er, + }; + handle.storage().finish_loaded_restoration(restored) + }; + + // Try to restore the signed instance. + let restored = match super::restore_signed(&zone, ¢er, &mut restorer) { + Ok(()) => restorer.finish().unwrap_or_else(|_| { + unreachable!( + "'restore_signed()' always completes restoration on successful return" + ) + }), + Err(_) => { + trace!("Abandoning signed restoration"); + let mut state = zone.state.lock().unwrap(); + let mut handle = ZoneHandle { + zone: &zone, + state: &mut state, + center: ¢er, + }; + handle.storage().abandon_signed_restoration(restorer); + handle.state.persistence.ongoing.finish(); + return; + } + }; + + info!("Restored the zone's persisted data"); + let mut state = zone.state.lock().unwrap(); + let mut handle = ZoneHandle { + zone: &zone, + state: &mut state, + center: ¢er, + }; + handle.storage().finish_signed_restoration(restored); + handle.state.persistence.ongoing.finish(); + }); + } + + /// Begin persisting a loaded zone instance. + /// + /// A background task will be spawned to perform the provided zone + /// persistence and transition to the next state. + #[tracing::instrument( + level = "trace", + skip_all, + fields(zone = %self.zone.name), + )] + pub fn start_loaded_persistence(&mut self, persister: LoadedZonePersister) { + let zone = self.zone.clone(); + let center = self.center.clone(); + let span = trace_span!("loaded_persistence"); + self.state + .persistence + .ongoing + .spawn_blocking(span, move || { + debug!("Persisting the loaded instance"); + + let persisted = super::persist_loaded(&zone, ¢er, persister); + + // NOTE: The outer function, which is spawning the background + // task, has a lock of the zone state. Thus, the following lock + // cannot be taken until the outer function terminates. + let mut state = zone.state.lock().unwrap(); + let mut handle = ZoneHandle { + zone: &zone, + state: &mut state, + center: ¢er, + }; + + handle.start_new_sign(persisted); + + handle.state.persistence.ongoing.finish(); + }); + } + + /// Begin persisting a signed zone instance. + /// + /// A background task will be spawned to perform the provided zone + /// persistence and transition to the next state. + #[tracing::instrument( + level = "trace", + skip_all, + fields(zone = %self.zone.name), + )] + pub fn start_signed_persistence(&mut self, persister: SignedZonePersister) { + let zone = self.zone.clone(); + let center = self.center.clone(); + let span = trace_span!("signed_persistence"); + self.state + .persistence + .ongoing + .spawn_blocking(span, move || { + debug!("Persisting the signed instance"); + + let persisted = super::persist_signed(&zone, ¢er, persister); + + // NOTE: The outer function, which is spawning the background + // task, has a lock of the zone state. Thus, the following lock + // cannot be taken until the outer function terminates. + let mut state = zone.state.lock().unwrap(); + let mut handle = ZoneHandle { + zone: &zone, + state: &mut state, + center: ¢er, + }; + + handle.start_switch(persisted); + + handle.state.persistence.ongoing.finish(); + }); + } +} + +//----------- PersistenceState ------------------------------------------------- + +/// State related to data persistence for a zone. +#[derive(Debug, Default)] +pub struct PersistenceState { + /// Ongoing persist/restore operations. + pub ongoing: BackgroundTasks, +} diff --git a/src/server/mod.rs b/src/server/mod.rs index f8a181a67..0bc99fc64 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -7,12 +7,12 @@ use cascade_zonedata::{LoadedZoneReviewer, SignedZoneReviewer, ZoneViewer}; use domain::base::Serial; use crate::{ - center::{Center, State}, + center::Center, daemon::SocketProvider, manager::Terminated, units::zone_server::{Source, ZoneServer}, util::AbortOnDrop, - zone::{Zone, ZoneByName}, + zone::Zone, }; mod request; @@ -38,16 +38,6 @@ impl LoadedReviewServer { Self { service, handle } } - /// Initialize the server, synchronously. - pub fn init(center: &Arc
, state: &mut State) { - // Store the viewer for all known zones. - for ZoneByName(zone) in &state.zones { - let mut state = zone.state.lock().unwrap(); - let viewer = state.storage.loaded_reviewer.take().unwrap(); - Self::add_zone(center, zone.clone(), viewer); - } - } - /// Drive the server. pub fn run( center: &Arc
, @@ -137,16 +127,6 @@ impl SignedReviewServer { Self { service, handle } } - /// Initialize the server, synchronously. - pub fn init(center: &Arc
, state: &mut State) { - // Store the viewer for all known zones. - for ZoneByName(zone) in &state.zones { - let mut state = zone.state.lock().unwrap(); - let viewer = state.storage.signed_reviewer.take().unwrap(); - Self::add_zone(center, zone.clone(), viewer); - } - } - /// Drive the server. pub fn run( center: &Arc
, @@ -236,16 +216,6 @@ impl PublicationServer { Self { service, handle } } - /// Initialize the server, synchronously. - pub fn init(center: &Arc
, state: &mut State) { - // Store the viewer for all known zones. - for ZoneByName(zone) in &state.zones { - let mut state = zone.state.lock().unwrap(); - let viewer = state.storage.viewer.take().unwrap(); - Self::add_zone(center, zone.clone(), viewer); - } - } - /// Drive the server. pub fn run( center: &Arc
, diff --git a/src/server/service.rs b/src/server/service.rs index b6faecf37..8aaedf0a2 100644 --- a/src/server/service.rs +++ b/src/server/service.rs @@ -280,19 +280,15 @@ trait Viewer { impl Viewer for LoadedZoneReviewer { fn is_empty(&self) -> bool { - self.read_loaded().is_none() + self.read().is_none() } fn soa(&self) -> &SoaRecord { - self.read_loaded().unwrap().soa() + self.read().unwrap().soa() } fn non_soa_records(&self) -> impl Iterator + Send { - self.read_loaded() - .unwrap() - .regular_records() - .iter() - .cloned() + self.read().unwrap().regular_records().iter().cloned() } } diff --git a/src/state/mod.rs b/src/state/mod.rs index e5b896d75..a94ea1772 100644 --- a/src/state/mod.rs +++ b/src/state/mod.rs @@ -5,11 +5,16 @@ use std::{ io::{self, BufReader}, }; +use bytes::Bytes; use camino::Utf8Path; +use domain::base::Name; use serde::{Deserialize, Serialize}; use tracing::{debug, error}; -use crate::center::{Center, State}; +use crate::{ + center::{Center, State}, + policy::Policy, +}; pub mod v1; @@ -53,9 +58,26 @@ pub enum Spec { impl Spec { /// Parse from this specification. - pub fn parse(self) -> State { + /// + /// `zones` will be set to the names of zones that need to be loaded. + /// `policies` will be set to the set of policies from the global state + /// file, that need to be parsed and inserted in the state. + pub fn parse( + self, + zones: &mut foldhash::HashSet>, + policies: &mut foldhash::HashMap, PolicySpec>, + ) -> State { match self { - Self::V1(spec) => spec.parse(), + Self::V1(mut spec) => { + // Extract and write out 'zones' and 'policies'. + *zones = std::mem::take(&mut spec.zones); + *policies = std::mem::take(&mut spec.policies) + .into_iter() + .map(|(k, v)| (k, PolicySpec::V1(v))) + .collect(); + + spec.parse() + } } } @@ -85,3 +107,22 @@ impl Spec { crate::util::write_file(path, text.as_bytes()) } } + +//----------- PolicySpec ------------------------------------------------------- + +/// A policy serialized in global state. +#[derive(Clone, Debug, Serialize, Deserialize)] +#[serde(rename_all = "kebab-case", tag = "version")] +pub enum PolicySpec { + /// The version 1 format. + V1(v1::PolicySpec), +} + +impl PolicySpec { + /// Parse from this specification. + pub fn parse(self, name: &str) -> Policy { + match self { + Self::V1(spec) => spec.parse(name), + } + } +} diff --git a/src/state/v1.rs b/src/state/v1.rs index 137ad6f73..8c1e741ad 100644 --- a/src/state/v1.rs +++ b/src/state/v1.rs @@ -7,18 +7,15 @@ use bytes::Bytes; use domain::base::Name; use domain::base::Ttl; use serde::{Deserialize, Serialize}; -use tracing::info; use crate::policy::file::v1::OutboundSpec; use crate::policy::{AutoConfig, DsAlgorithm, KeyParameters}; -use crate::tsig::TsigStore; use crate::{ center::State, policy::{ KeyManagerPolicy, LoaderPolicy, Policy, PolicyVersion, ReviewPolicy, ServerPolicy, SignerDenialPolicy, SignerPolicy, SignerSerialPolicy, }, - zone::{Zone, ZoneByName}, }; //----------- Spec ------------------------------------------------------------- @@ -41,31 +38,19 @@ pub struct Spec { impl Spec { /// Parse from this specification. + /// + /// [`Self::zones`] and [`Self::policies`] are ignored; these should be + /// extracted from `self` before calling this function. pub fn parse(self) -> State { - let mut policies = foldhash::HashMap::default(); - for (name, spec) in self.policies { - info!("Adding policy '{name}' from global state"); - let policy = spec.parse(&name); - policies.insert(name, policy); - } - - #[allow(clippy::mutable_key_type)] - let zones = self - .zones - .into_iter() - .map(|name| { - info!("Adding zone '{name}' from global state"); - ZoneByName(Arc::new(Zone::new(name.clone()))) - }) - .collect(); - - State { - zones, - policies, - rt_config: cascade_cfg::RuntimeConfig::default(), - tsig_store: TsigStore::default(), - enqueued_save: None, - } + let Self { + // The caller will extract 'zones' and 'policies' beforehand. + zones: _, + policies: _, + // TODO: More fields. + }; + + // TODO: Initialize fields from 'Spec'. + State::default() } /// Build this state specification. diff --git a/src/zone/machine.rs b/src/zone/machine.rs index 0ff3aec31..4a4655acc 100644 --- a/src/zone/machine.rs +++ b/src/zone/machine.rs @@ -177,7 +177,10 @@ impl<'a> ZoneHandle<'a> { /// # Loaded Review operations impl<'a> ZoneHandle<'a> { + /// Approve the loaded instance currently under review. pub(crate) fn approve_loaded(&mut self) { + info!("The loaded instance has been approved"); + self.state.record_event( HistoricalEvent::UnsignedZoneReview { status: ZoneReviewStatus::Approved, @@ -185,15 +188,11 @@ impl<'a> ZoneHandle<'a> { None, // TODO ); - let (transition, state) = self.state.machine.transition(); - - let ZoneStateMachine::LoadedReview(loaded) = state else { - panic!("cannot approve loaded in this state"); - }; - - transition.move_to(ZoneStateMachine::Signing(loaded.approve())); - - self.storage().accept_loaded(); + // Persist the loaded instance while remaining in the 'LoadedReview' + // state. Once persistence is complete, 'begin_signing()' will be + // called, which will transition to the 'Signing' state. + let persister = self.storage().accept_loaded(); + self.persistence().start_loaded_persistence(persister); } #[expect(dead_code)] @@ -234,6 +233,44 @@ impl<'a> ZoneHandle<'a> { /// # Signing operations impl<'a> ZoneHandle<'a> { + /// Begin signing a new approved and persisted loaded instance. + pub(crate) fn start_new_sign(&mut self, persisted: cascade_zonedata::LoadedZonePersisted) { + // NOTE: The underlying state machine does not track persistence, so we + // only transition out of 'LoadedReview' once persistence is complete. + let (transition, state) = self.state.machine.transition(); + let ZoneStateMachine::LoadedReview(loaded) = state else { + unreachable!( + "A 'LoadedZonePersisted' can only exist when the zone is in loader review" + ); + }; + transition.move_to(ZoneStateMachine::Signing(loaded.approve())); + + let builder = self.storage().start_new_sign(persisted); + self.signer().enqueue_new_sign(builder); + } + + /// Begin signing a restored loaded instance. + /// + /// This is called when the zone is restored from persisted state, the data + /// for its last known loaded instance is restored successfully, but the + /// data for its last known signed instance could not be restored (or none + /// existed). It directly initiates signing. + pub(crate) fn start_sign_after_restore( + &mut self, + builder: cascade_zonedata::SignedZoneBuilder, + ) { + // Force the state machine to the signing state. + let (transition, state) = self.state.machine.transition(); + let ZoneStateMachine::Waiting(waiting) = state else { + panic!("'start_sign_after_restore()' called when the zone was not passive"); + }; + transition.move_to(ZoneStateMachine::Signing( + waiting.start_sign_after_restore(), + )); + + self.signer().enqueue_new_sign(builder); + } + pub(crate) fn finish_signing(&mut self, built: cascade_zonedata::SignedZoneBuilt) { self.state.record_event( // TODO: Get the right trigger. @@ -286,6 +323,8 @@ impl<'a> ZoneHandle<'a> { /// # Signed Review operations impl<'a> ZoneHandle<'a> { pub(crate) fn approve_signed(&mut self) { + info!("The signed instance has been approved"); + self.state.record_event( HistoricalEvent::SignedZoneReview { status: ZoneReviewStatus::Approved, @@ -293,15 +332,11 @@ impl<'a> ZoneHandle<'a> { None, // TODO ); - let (transition, state) = self.state.machine.transition(); - - let ZoneStateMachine::SignedReview(signed) = state else { - panic!("cannot approve signed in this state: {}", state.as_str()); - }; - - transition.move_to(ZoneStateMachine::Waiting(signed.approve())); - - self.storage().accept_signed(); + // Persist the signed instance while remaining in the 'SignedReview' + // state. Once persistence is complete, 'switch()' will be called, which + // will transition to the 'Waiting' state. + let persister = self.storage().accept_signed(); + self.persistence().start_signed_persistence(persister); } #[expect(dead_code)] @@ -340,6 +375,32 @@ impl<'a> ZoneHandle<'a> { } } +/// # Switching operations +impl<'a> ZoneHandle<'a> { + /// Begin switching to an approved instance of the zone. + pub(crate) fn start_switch(&mut self, persisted: cascade_zonedata::SignedZonePersisted) { + // Make sure the current state is 'SignedReview'. + assert!( + matches!(self.state.machine, ZoneStateMachine::SignedReview(_)), + "A 'SignedZonePersisted' exists but the zone is not in signer review" + ); + + self.storage().start_switch(persisted); + } + + /// Finish switching to a new instance of the zone. + pub(crate) fn finish_switch(&mut self, cleaner: cascade_zonedata::ZoneCleaner) { + // Move to the 'Waiting' state. + let (transition, state) = self.state.machine.transition(); + let ZoneStateMachine::SignedReview(signed) = state else { + panic!("The zone must be in signer review") + }; + transition.move_to(ZoneStateMachine::Waiting(signed.approve())); + + self.storage().start_cleanup(cleaner); + } +} + /// # Halted operations impl<'a> ZoneHandle<'a> { pub(crate) fn try_reset(&mut self) -> Result<(), ()> { @@ -474,6 +535,10 @@ impl Waiting { Loading {} } + fn start_sign_after_restore(self) -> Signing { + Signing {} + } + fn start_resign(self) -> Signing { Signing {} } diff --git a/src/zone/mod.rs b/src/zone/mod.rs index c2bf5071f..b2350569e 100644 --- a/src/zone/mod.rs +++ b/src/zone/mod.rs @@ -6,11 +6,13 @@ use std::{ cmp::Ordering, fmt, hash::{Hash, Hasher}, + io, sync::{Arc, Mutex}, time::{Duration, SystemTime}, }; use bytes::Bytes; +use cascade_cfg::Config; use domain::base::{Name, Rtype, Serial}; use domain::dnssec::sign::keys::keyset::UnixTime; use domain::rdata::dnssec::Timestamp; @@ -21,7 +23,8 @@ use crate::{ api::{self, ZoneReviewStatus}, center::Center, loader::zone::{LoaderState, LoaderZoneHandle}, - policy::PolicyVersion, + persistence::zone::{PersistenceState, ZonePersistenceHandle}, + policy::{Policy, PolicyVersion}, signer::zone::{SignerState, SignerZoneHandle}, util::{deserialize_duration_from_secs, serialize_duration_as_secs}, zone::machine::ZoneStateMachine, @@ -50,6 +53,69 @@ pub struct Zone { /// consistent with each other, and that changes to the zone happen in a /// single (sequentially consistent) order. pub state: Mutex, + + /// Whether the zone was restored from the state file. + /// + /// This is set if the zone originates from a previous execution of Cascade + /// and its state was loaded from a file (rather than being created in the + /// current execution). + pub restored: bool, +} + +impl Zone { + /// Construct a new zone. + /// + /// The zone is initialized to an empty state, where nothing is known about + /// it and Cascade won't act on it. + pub fn new(name: Name) -> Self { + Self { + name, + state: Default::default(), + restored: false, + } + } + + /// Restore a zone from a state file. + /// + /// A zone originating from a previous execution of Cascade is initialized, + /// by reading and parsing the appropriate state file. + /// + /// `policies` should contain the set of policies loaded from the global + /// state file. If the zone uses a policy that is not present in the global + /// state file, it will restore the last seen version of that policy. + /// + /// Persisted zone data will not be restored in this function, as it may + /// take a while (and should not block Cascade's initialization as a whole); + /// it will be handled by [`crate::persistence::Restorer::run()`]. + #[tracing::instrument( + level = "debug", + skip_all, + fields(%name), + )] + pub fn restore( + config: &Config, + name: Name, + policies: &mut foldhash::HashMap, Policy>, + ) -> io::Result { + let path = config.zone_state_dir.join(format!("{name}.db")); + + // Load the underlying state file. + let state = match state::Spec::load(&path) { + Ok(spec) => spec.parse(&name, policies), + Err(err) => { + error!("Failed to load the state of zone '{name}' from '{path}': {err}"); + return Err(err); + } + }; + + debug!("Restored the state of zone '{name}' (from '{path}')"); + + Ok(Self { + name, + state: Mutex::new(state), + restored: true, + }) + } } //----------- ZoneHandle ------------------------------------------------------- @@ -93,6 +159,15 @@ impl ZoneHandle<'_> { center: self.center, } } + + /// Consider data persistence specific operations. + pub const fn persistence(&mut self) -> ZonePersistenceHandle<'_> { + ZonePersistenceHandle { + zone: self.zone, + state: self.state, + center: self.center, + } + } } //----------- ZoneState -------------------------------------------------------- @@ -181,6 +256,9 @@ pub struct ZoneState { /// Data storage for the zone. pub storage: StorageState, + + /// Persisting zone data. + pub persistence: PersistenceState, // // TODO: // - A log? @@ -231,6 +309,7 @@ impl Default for ZoneState { loader: Default::default(), signer: Default::default(), storage: Default::default(), + persistence: Default::default(), } } } @@ -424,19 +503,6 @@ impl From for api::HistoricalEvent { } } -impl Zone { - /// Construct a new [`Zone`]. - /// - /// The zone is initialized to an empty state, where nothing is known about - /// it and Cascade won't act on it. - pub fn new(name: Name) -> Self { - Self { - name: name.clone(), - state: Default::default(), - } - } -} - //--- Loading / Saving impl Zone { diff --git a/src/zone/state/mod.rs b/src/zone/state/mod.rs index 24b00a264..8efe55379 100644 --- a/src/zone/state/mod.rs +++ b/src/zone/state/mod.rs @@ -7,14 +7,16 @@ use std::{ sync::Arc, }; +use bytes::Bytes; use camino::Utf8Path; +use domain::base::Name; use serde::{Deserialize, Serialize}; use tracing::warn; use crate::{ loader::zone::LoaderState, policy::{Policy, PolicyVersion}, - zone::{Zone, ZoneState}, + zone::ZoneState, }; pub mod v1; @@ -35,47 +37,43 @@ impl Spec { /// Merge this specification with an existing zone state. pub fn parse( self, - zone: &Arc, + zone_name: &Name, policies: &mut foldhash::HashMap, Policy>, ) -> ZoneState { /// Synchronize a loaded policy with global state. fn sync_policy( - policy: PolicyVersion, - zone: &Arc, + known_version: PolicyVersion, policies: &mut foldhash::HashMap, Policy>, - ) -> Arc { + ) -> &mut Policy { // Check whether a policy of this name exists. - match policies.entry(policy.name.clone()) { + match policies.entry(known_version.name.clone()) { hash_map::Entry::Occupied(entry) => { // A policy of this name exists. Compare to it. - let existing = &entry.get().latest; - if **existing == policy { - return existing.clone(); + let policy = entry.into_mut(); + if *policy.latest != known_version { + // TODO: Continue using the older version of the policy, and + // enqueue an explicit change to the zone, so that any + // necessary hooks (e.g. re-signing) can be activated. + warn!( + "Detected an inconsistency between the details of policy '{}' last used with the zone and the policy details in the global state; switching to the global state's details", + known_version.name + ); } - // TODO: Continue using the older version of the policy, and - // enqueue an explicit change to the zone, so that any - // necessary hooks (e.g. re-signing) can be activated. - warn!( - "Zone '{}' is using an older version of policy '{}'; it will be updated", - zone.name, policy.name - ); - existing.clone() + policy } hash_map::Entry::Vacant(entry) => { warn!( - "Zone '{}' is using an unknown policy '{}'; the policy has been restored", - zone.name, policy.name + "The policy '{}' used by the zone is not present in the global state; restoring it into the global state", + known_version.name ); - let policy = Arc::new(policy); entry.insert(Policy { - latest: policy.clone(), + latest: Arc::new(known_version), mid_deletion: false, zones: Default::default(), - }); - policy + }) } } } @@ -99,14 +97,14 @@ impl Spec { ..Default::default() }; - // This should always be some at this stage... - let policy = policy.map(|policy| sync_policy(policy.parse(), zone, policies)); - if let Some(policy) = &policy { - let p = policies - .get_mut(&*policy.name) - .expect("zone policy references should not be kept around"); - p.zones.insert(zone.name.clone()); + // TODO: Won't this always be a `Some`? + let mut policy = policy.map(|policy| sync_policy(policy.parse(), policies)); + if let Some(policy) = &mut policy { + // Register that the policy is in use by this zone. It might + // already be registered; that's fine. + policy.zones.insert(zone_name.clone()); } + let policy = policy.map(|p| p.latest.clone()); ZoneState { policy, diff --git a/src/zone/storage.rs b/src/zone/storage.rs index 5574d26ab..aa8013d51 100644 --- a/src/zone/storage.rs +++ b/src/zone/storage.rs @@ -26,9 +26,10 @@ use std::{fmt, sync::Arc}; use cascade_zonedata::{ - LoadedZoneBuilder, LoadedZoneBuilt, LoadedZonePersister, LoadedZoneReviewer, SignedZoneBuilder, - SignedZoneBuilt, SignedZonePersister, SignedZoneReviewer, SoaRecord, ZoneCleaner, - ZoneDataStorage, ZoneViewer, + LoadedZoneBuilder, LoadedZoneBuilt, LoadedZonePersisted, LoadedZonePersister, + LoadedZoneRestored, LoadedZoneRestorer, LoadedZoneReviewer, SignedZoneBuilder, SignedZoneBuilt, + SignedZonePersisted, SignedZonePersister, SignedZoneRestored, SignedZoneRestorer, + SignedZoneReviewer, SoaRecord, ZoneCleaner, ZoneDataStorage, }; use domain::base::Serial; use tracing::{info, trace, trace_span, warn}; @@ -129,7 +130,7 @@ impl StorageZoneHandle<'_> { // TODO: Use the instance ID here, which will not require // examining the zone contents. - let serial = loaded_reviewer.read_loaded().unwrap().soa().rdata.serial; + let serial = loaded_reviewer.read().unwrap().soa().rdata.serial; self.state.record_event( HistoricalEvent::NewVersionReceived, Some(domain::base::Serial(serial.into())), @@ -184,8 +185,7 @@ impl StorageZoneHandle<'_> { fields(zone = %self.zone.name), )] fn start_loaded_review(&mut self, loaded_reviewer: LoadedZoneReviewer) { - self.state.storage.loaded_review_soa = - loaded_reviewer.read_loaded().map(|r| r.soa().clone()); + self.state.storage.loaded_review_soa = loaded_reviewer.read().map(|r| r.soa().clone()); let zone = self.zone.clone(); let center = self.center.clone(); @@ -193,7 +193,7 @@ impl StorageZoneHandle<'_> { self.state.storage.background_tasks.spawn(span, async move { // Read the loaded instance. let reader = loaded_reviewer - .read_loaded() + .read() .unwrap_or_else(|| unreachable!("The loader never returns an empty instance")); let serial = reader.soa().rdata.serial; @@ -232,22 +232,23 @@ impl StorageZoneHandle<'_> { } /// Accept a loaded instance of a zone. + /// + /// A [`LoadedZonePersister`] is returned through which the instance must + /// be persisted. Once persistence is complete, the [`LoadedZonePersisted`] + /// should be passed to [`Self::start_new_sign()`]. #[tracing::instrument( level = "trace", skip_all, fields(zone = %self.zone.name), )] - pub fn accept_loaded(&mut self) { + pub fn accept_loaded(&mut self) -> LoadedZonePersister { // Examine the current state. let (transition, state) = transition(&mut self.state.storage.machine); match state { ZoneDataStorage::ReviewingLoaded(s) => { - // TODO: Specify the instance ID. - info!("The loaded instance has been approved; persisting it"); - let (s, persister) = s.mark_approved(); transition.move_to(ZoneDataStorage::PersistingLoaded(s)); - self.start_loaded_persistence(persister); + persister } _ => panic!("The zone is not undergoing loader review"), @@ -269,7 +270,7 @@ impl StorageZoneHandle<'_> { let (s, loaded_reviewer) = s.give_up(); self.state.storage.loaded_review_soa = - loaded_reviewer.read_loaded().map(|r| r.soa().clone()); + loaded_reviewer.read().map(|r| r.soa().clone()); transition.move_to(ZoneDataStorage::CleanLoadedPending(s)); loaded_reviewer } @@ -284,6 +285,26 @@ impl StorageZoneHandle<'_> { /// # Signer Operations impl StorageZoneHandle<'_> { + /// Start signing a new approved and persisted loaded instance. + #[tracing::instrument( + level = "trace", + skip_all, + fields(zone = %self.zone.name), + )] + pub fn start_new_sign(&mut self, persisted: LoadedZonePersisted) -> SignedZoneBuilder { + match transition(&mut self.state.storage.machine) { + (transition, ZoneDataStorage::PersistingLoaded(s)) => { + let (s, builder) = s.mark_complete(persisted); + transition.move_to(ZoneDataStorage::Signing(s)); + builder + } + + _ => unreachable!( + "'ZoneDataStorage::PersistingLoaded' is the only state where a 'LoadedZonePersisted' is available" + ), + } + } + /// Begin resigning the zone. /// /// If the zone data storage is not busy, a [`SignedZoneBuilder`] will be @@ -376,7 +397,7 @@ impl StorageZoneHandle<'_> { let (s, loaded_reviewer) = s.give_up(builder); self.state.storage.loaded_review_soa = - loaded_reviewer.read_loaded().map(|r| r.soa().clone()); + loaded_reviewer.read().map(|r| r.soa().clone()); transition.move_to(ZoneDataStorage::CleanLoadedPending(s)); loaded_reviewer } @@ -452,17 +473,14 @@ impl StorageZoneHandle<'_> { skip_all, fields(zone = %self.zone.name), )] - pub fn accept_signed(&mut self) { + pub fn accept_signed(&mut self) -> SignedZonePersister { // Examine the current state. let (transition, state) = transition(&mut self.state.storage.machine); match state { ZoneDataStorage::ReviewingSigned(s) => { - // TODO: Specify the instance ID. - info!("The signed instance has been approved; persisting it"); - let (s, persister) = s.mark_approved(); transition.move_to(ZoneDataStorage::PersistingSigned(s)); - self.start_signed_persistence(persister); + persister } _ => panic!("The zone is not undergoing signer review"), @@ -487,7 +505,7 @@ impl StorageZoneHandle<'_> { (new_s, loaded_reviewer, signed_reviewer) = s.give_up(); transition.move_to(ZoneDataStorage::CleanWholePending(new_s)); self.state.storage.loaded_review_soa = - loaded_reviewer.read_loaded().map(|r| r.soa().clone()); + loaded_reviewer.read().map(|r| r.soa().clone()); self.state.storage.signed_review_soa = signed_reviewer.read().map(|r| r.soa().clone()); } @@ -533,6 +551,117 @@ impl StorageZoneHandle<'_> { } } +/// # Persistence Tasks +impl StorageZoneHandle<'_> { + /// Successfully finish loaded-instance restoration. + /// + /// A [`SignedZoneRestorer`] is returned so the signed instance can be + /// restored afterwards. + pub fn finish_loaded_restoration( + &mut self, + restored: LoadedZoneRestored, + ) -> SignedZoneRestorer { + // Examine the current state. + match transition(&mut self.state.storage.machine) { + (transition, ZoneDataStorage::RestoringLoaded(s)) => { + let (restorer, s) = s.finish(restored); + transition.move_to(ZoneDataStorage::RestoringSigned(s)); + restorer + } + + _ => unreachable!( + "A 'LoadedZoneRestored' is only available in the 'RestoringLoaded' state" + ), + } + } + + /// Successfully finish signed-instance restoration. + /// + /// The zone is moved to the passive state, and it is registered against + /// Cascade's zone servers. + pub fn finish_signed_restoration(&mut self, restored: SignedZoneRestored) { + // Examine the current state. + let (loaded_reviewer, signed_reviewer, viewer); + match transition(&mut self.state.storage.machine) { + (transition, ZoneDataStorage::RestoringSigned(s)) => { + let new_s; + (loaded_reviewer, signed_reviewer, viewer, new_s) = s.finish(restored); + transition.move_to(ZoneDataStorage::Passive(new_s)); + } + + _ => unreachable!( + "A 'SignedZoneRestored' is only available in the 'RestoringSigned' state" + ), + } + + // Register the zone against the zone servers. + LoadedReviewServer::add_zone(self.center, self.zone.clone(), loaded_reviewer); + SignedReviewServer::add_zone(self.center, self.zone.clone(), signed_reviewer); + PublicationServer::add_zone(self.center, self.zone.clone(), viewer); + + // Send a notification that the state machine is now passive. + self.on_passive(); + } + + /// Abandon the ongoing loaded-instance restore. + /// + /// Any intermediate zone data is cleared and the zone is moved to the + /// passive state. It is registered against Cascade's zone servers. + pub fn abandon_loaded_restoration(&mut self, restorer: LoadedZoneRestorer) { + // Examine the current state. + let (loaded_reviewer, signed_reviewer, viewer); + match transition(&mut self.state.storage.machine) { + (transition, ZoneDataStorage::RestoringLoaded(s)) => { + let new_s; + (loaded_reviewer, signed_reviewer, viewer, new_s) = s.abandon(restorer); + transition.move_to(ZoneDataStorage::Passive(new_s)); + } + + _ => unreachable!( + "A 'LoadedZoneRestorer' is only available in the 'RestoringLoaded' state" + ), + }; + + // Update the zone servers. + LoadedReviewServer::add_zone(self.center, self.zone.clone(), loaded_reviewer); + SignedReviewServer::add_zone(self.center, self.zone.clone(), signed_reviewer); + PublicationServer::add_zone(self.center, self.zone.clone(), viewer); + + // Send a notification that the state machine is now passive. + self.on_passive(); + } + + /// Abandon the ongoing signed-instance restore. + /// + /// Any intermediate zone data for the signed instance is wiped. The + /// restored loaded instance is preserved. The zone is moved to the signing + /// state. It is registered against Cascade's zone servers and a signing + /// operation is enqueued. + pub fn abandon_signed_restoration(&mut self, restorer: SignedZoneRestorer) { + // Examine the current state. + let (loaded_reviewer, signed_reviewer, viewer, builder); + match transition(&mut self.state.storage.machine) { + (transition, ZoneDataStorage::RestoringSigned(s)) => { + let new_s; + (loaded_reviewer, signed_reviewer, viewer, builder, new_s) = s.abandon(restorer); + transition.move_to(ZoneDataStorage::Signing(new_s)); + } + + _ => unreachable!( + "A 'SignedZoneRestorer' is only available in the 'RestoringSigned' state" + ), + }; + + // Update the zone servers. + LoadedReviewServer::add_zone(self.center, self.zone.clone(), loaded_reviewer); + SignedReviewServer::add_zone(self.center, self.zone.clone(), signed_reviewer); + PublicationServer::add_zone(self.center, self.zone.clone(), viewer); + + // Initiate a new signing operation. + self.zone().start_sign_after_restore(builder); + } +} + /// # Background Tasks impl StorageZoneHandle<'_> { /// Run a cleanup of zone data. @@ -544,7 +673,7 @@ impl StorageZoneHandle<'_> { skip_all, fields(zone = %self.zone.name), )] - fn start_cleanup(&mut self, cleaner: ZoneCleaner) { + pub fn start_cleanup(&mut self, cleaner: ZoneCleaner) { let zone = self.zone.clone(); let center = self.center.clone(); let span = trace_span!("clean"); @@ -584,24 +713,39 @@ impl StorageZoneHandle<'_> { }); } - /// Begin persisting a loaded zone instance. + /// Start switching to an approved and persisted signed instance. /// - /// A background task will be spawned to perform the provided zone - /// persistence and transition to the next state. + /// A background task will be spawned to switch the publication server to + /// the newly persisted instance and transition to the next state. #[tracing::instrument( level = "trace", skip_all, fields(zone = %self.zone.name), )] - fn start_loaded_persistence(&mut self, persister: LoadedZonePersister) { + pub fn start_switch(&mut self, persisted: SignedZonePersisted) { + // Examine the current state. + let viewer = match transition(&mut self.state.storage.machine) { + (transition, ZoneDataStorage::PersistingSigned(s)) => { + let (s, viewer) = s.mark_complete(persisted); + transition.move_to(ZoneDataStorage::Switching(s)); + viewer + } + + _ => unreachable!( + "'ZoneDataStorage::PersistingSigned' is the only state where a 'SignedZonePersisted' is available" + ), + }; + + self.state.storage.published_soa = viewer.read().map(|r| r.soa().clone()); + self.state.storage.published_loaded_soa = viewer.read().map(|r| r.loaded().soa().clone()); + + // Spawn a background task to update the publication server. + let span = trace_span!("switch_publication_server"); let zone = self.zone.clone(); let center = self.center.clone(); - let span = trace_span!("persist_loaded"); - self.state.storage.background_tasks.spawn_blocking(span, move || { - trace!("Persisting the loaded instance"); - - // Perform the persisting. - let persisted = persister.persist(); + self.state.storage.background_tasks.spawn(span, async move { + // Update the publication server. + let old_viewer = PublicationServer::update_viewer(¢er, &zone, viewer).await; // NOTE: The outer function, which is spawning the background task, // has a lock of the zone state. Thus, the following lock cannot be @@ -613,68 +757,8 @@ impl StorageZoneHandle<'_> { center: ¢er, }; - // Transition the state machine. - let builder = match transition(&mut handle.state.storage.machine) { - (transition, ZoneDataStorage::PersistingLoaded(s)) => { - let (s, builder) = s.mark_complete(persisted); - transition.move_to(ZoneDataStorage::Signing(s)); - builder - } - - _ => unreachable!( - "'ZoneDataStorage::PersistingLoaded' is the only state where a 'LoadedZonePersister' is available" - ), - }; - handle.signer().enqueue_new_sign(builder); - - handle.state.storage.background_tasks.finish(); - }); - } - - /// Begin persisting a signed zone instance. - /// - /// A background task will be spawned to perform the provided zone - /// persistence and transition to the next state. - #[tracing::instrument( - level = "trace", - skip_all, - fields(zone = %self.zone.name), - )] - fn start_signed_persistence(&mut self, persister: SignedZonePersister) { - let zone = self.zone.clone(); - let center = self.center.clone(); - let span = trace_span!("persist_signed"); - self.state.storage.background_tasks.spawn(span, async move { - trace!("Persisting the signed instance"); - - // Perform the persisting. - let persisted = tokio::task::spawn_blocking(move || persister.persist()).await.unwrap(); - - // Mark persistence as completed. - let viewer = { - let mut state = zone.state.lock().unwrap(); - let state = &mut *state; - match transition(&mut state.storage.machine) { - (transition, ZoneDataStorage::PersistingSigned(s)) => { - let (s, viewer) = s.mark_complete(persisted); - transition.move_to(ZoneDataStorage::Switching(s)); - state.storage.published_soa = viewer.read().map(|r| r.soa().clone()); - state.storage.published_loaded_soa = viewer.read().map(|r| r.loaded().soa().clone()); - viewer - } - - _ => unreachable!( - "'ZoneDataStorage::PersistingSigned' is the only state where a 'SignedZonePersister' is available" - ), - } - }; - - // Update the publication server. - let old_viewer = PublicationServer::update_viewer(¢er, &zone, viewer).await; - - // Begin cleaning up the old instance. - let mut state = zone.state.lock().unwrap(); - let cleaner = match transition(&mut state.storage.machine) { + // Update the zone data storage state machine. + let cleaner = match transition(&mut handle.state.storage.machine) { (transition, ZoneDataStorage::Switching(s)) => { let (s, cleaner) = s.switch(old_viewer); transition.move_to(ZoneDataStorage::Cleaning(s)); @@ -684,14 +768,32 @@ impl StorageZoneHandle<'_> { _ => unreachable!("just transitioned to 'Switching'"), }; - state.last_published = Some(LastPublished { - loaded_serial: Serial(state.storage.published_loaded_soa.as_ref().unwrap().rdata.serial.into()), - signed_serial: Serial(state.storage.published_soa.as_ref().unwrap().rdata.serial.into()), + handle.state.last_published = Some(LastPublished { + loaded_serial: Serial( + handle + .state + .storage + .published_loaded_soa + .as_ref() + .unwrap() + .rdata + .serial + .into(), + ), + signed_serial: Serial( + handle + .state + .storage + .published_soa + .as_ref() + .unwrap() + .rdata + .serial + .into(), + ), }); - let mut handle = ZoneHandle { zone: &zone, state: &mut state, center: ¢er }; - - handle.storage().start_cleanup(cleaner); + handle.finish_switch(cleaner); handle.state.storage.background_tasks.finish(); }); @@ -788,26 +890,13 @@ pub struct StorageState { /// The underlying state machine. machine: ZoneDataStorage, - /// The current loaded zone reviewer. + /// A handle to restore the zone data at startup. /// - /// This is only used during initialization. - // - // TODO: Output it directly somehow? - pub loaded_reviewer: Option, - - /// The current zone reviewer. - /// - /// This is only used during initialization. - // - // TODO: Output it directly somehow? - pub signed_reviewer: Option, - - /// The current zone viewer. - /// - /// This is only used during initialization. - // - // TODO: Output it directly somehow? - pub viewer: Option, + /// This is only used during initialization. For zones restored from state + /// files at startup, this handle is used to initiate restores of their + /// (presumably persisted) data. For newly created zones, this handle is + /// passed to [`StorageZoneHandle::abandon_loaded_restoration()`]. + pub restorer: Option, /// The SOA record of the loaded instance of the zone being reviewed, if /// any. @@ -846,13 +935,11 @@ pub struct StorageState { impl StorageState { /// Construct a new [`StorageState`]. pub fn new() -> Self { - let (machine, loaded_reviewer, signed_reviewer, viewer) = ZoneDataStorage::new(); + let (restorer, machine) = ZoneDataStorage::new(); Self { machine, - loaded_reviewer: Some(loaded_reviewer), - signed_reviewer: Some(signed_reviewer), - viewer: Some(viewer), + restorer: Some(restorer), loaded_review_soa: None, signed_review_soa: None, published_soa: None,