Integrate the zone server(s) with the new zone storage

crates/zonedata/src/reader.rs

@@ -64,7 +64,7 @@ impl<'d> LoadedZoneReader<'d> {
     ///
     /// Records are sorted in DNSSEC canonical order. The SOA record **is**
     /// included.
-    pub fn all_records(&self) -> impl IntoIterator<Item = RegularRecord> + use<'d> {
+    pub fn all_records(&self) -> impl Iterator<Item = RegularRecord> + Send + use<'d> {
         let (soa, records) = (self.soa(), self.regular_records());
         let soa = RegularRecord::from(soa.clone());
 
@@ -86,7 +86,7 @@ impl<'d> LoadedZoneReader<'d> {
     /// DNSSEC related records that would be produced by Cascade's signer (e.g.
     /// RRSIGs, NSEC/NSEC3, etc.) are stripped. The records are sorted in DNSSEC
     /// canonical order. The SOA record **is not** included.
-    pub fn unsigned_records(&self) -> impl IntoIterator<Item = RegularRecord> + use<'d> {
+    pub fn unsigned_records(&self) -> impl Iterator<Item = RegularRecord> + Send + use<'d> {
         // Filter out records that would be generated during signing.
         //
         // TODO: 'RType::{CDS, CDNSKEY, ZONEMD}'.
@@ -177,15 +177,15 @@ impl<'d> SignedZoneReader<'d> {
     /// Records are sorted in DNSSEC canonical order. Only records also present
     /// in the signed instance are included (the loaded SOA record, and loaded
     /// DNSKEY, RRSIG, CDS, CDNSKEY, ZONEMD records are excluded).
-    pub fn loaded_records(&self) -> impl IntoIterator<Item = RegularRecord> + use<'d> {
+    pub fn loaded_records(&self) -> impl Iterator<Item = RegularRecord> + Send + use<'d> {
         LoadedZoneReader::new(self.loaded_instance).unsigned_records()
     }
 
     /// All records in the zone.
     ///
     /// Records are **unsorted**. The SOA record and records from the loaded
     /// instance **are** included.
-    pub fn all_records(&self) -> impl IntoIterator<Item = RegularRecord> + use<'d> {
+    pub fn all_records(&self) -> impl Iterator<Item = RegularRecord> + Send + use<'d> {
         [self.soa().clone().into()]
             .into_iter()
             .chain(self.loaded_records())

src/center.rs

@@ -19,9 +19,8 @@ use crate::api::KeyImport;
 use crate::config::RuntimeConfig;
 use crate::loader::Loader;
 use crate::loader::zone::LoaderZoneHandle;
-use crate::manager::record_zone_event;
+use crate::server::{LoadedReviewServer, PublicationServer, SignedReviewServer};
 use crate::units::key_manager::KeyManager;
-use crate::units::zone_server::ZoneServer;
 use crate::units::zone_signer::ZoneSigner;
 use crate::zone::{HistoricalEvent, ZoneHandle};
 use crate::{
@@ -56,14 +55,14 @@ pub struct Center {
     /// The key manager
     pub key_manager: KeyManager,
 
-    /// The review server for unsigned zones.
-    pub unsigned_review_server: ZoneServer,
+    /// The review server for loaded instances of zones.
+    pub loaded_review_server: LoadedReviewServer,
 
-    /// The review server for signed zones.
-    pub signed_review_server: ZoneServer,
+    /// The review server for signed instances of zones.
+    pub signed_review_server: SignedReviewServer,
 
-    /// The zone server.
-    pub publication_server: ZoneServer,
+    /// The server for published instances of zones.
+    pub publication_server: PublicationServer,
 
     /// The latest unsigned contents of all zones.
     pub unsigned_zones: Arc<ArcSwap<ZoneTree>>,
@@ -88,63 +87,73 @@ pub async fn add_zone(
     source: api::ZoneSource,
     key_imports: Vec<KeyImport>,
 ) -> Result<(), ZoneAddError> {
-    let zone = Arc::new(Zone::new(name.clone()));
-
+    // Create and insert the zone.
+    let zone;
     {
+        // Lock the global state to check consistency and insert the zone.
         let mut state = center.state.lock().unwrap();
 
-        // We check whether the state contains this zone, because
-        // this is the most useful error to report.
-        let zone_by_name = ZoneByName(zone.clone());
-        if state.zones.contains(&zone_by_name) {
+        // Prioritize 'AlreadyExists' over other kinds of errors.
+        if state.zones.contains(&name) {
             return Err(ZoneAddError::AlreadyExists);
         }
 
-        // Do this inside a block to prevent holding a mutable reference to
-        // state.
-        {
-            let policy = state
-                .policies
-                .get_mut(&policy_name)
-                .ok_or(ZoneAddError::NoSuchPolicy)?;
-            if policy.mid_deletion {
-                return Err(ZoneAddError::PolicyMidDeletion);
-            }
+        // Look up the requested policy.
+        let policy = state
+            .policies
+            .get_mut(&policy_name)
+            .ok_or(ZoneAddError::NoSuchPolicy)?;
+        if policy.mid_deletion {
+            return Err(ZoneAddError::PolicyMidDeletion);
+        }
 
+        // 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();
             zone_state.policy = Some(policy.latest.clone());
-            policy.zones.insert(name.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();
         }
 
-        // Actually insert the zone now. This shouldn't fail since we've done
-        // the `contains` check above and we hold a lock to the state, but it
-        // doesn't hurt to have proper error handling here just in case.
-        if !state.zones.insert(zone_by_name.clone()) {
-            return Err(ZoneAddError::AlreadyExists);
-        }
+        // Insert the zone in the global set.
+        assert!(
+            state.zones.insert(ZoneByName(zone.clone())),
+            "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
     // (such as invoking dnst keyset create, ..., init) can be done _before_
     // the pipeline for the zone starts. We do this _after_ adding the zone
     // because otherwise updating zone history will fail. If registration
     // fails we will have to remove the added zone.
-    if let Err(err) = register_zone(center, name.clone(), policy_name.clone(), key_imports).await {
+    if let Err(err) =
+        register_zone(center, zone.name.clone(), policy_name.clone(), key_imports).await
+    {
         // Remove in reverse order what was added above.
         let mut state = center.state.lock().unwrap();
-        let zone_by_name = ZoneByName(zone);
-        state.zones.remove(&zone_by_name);
+        state.zones.remove(&zone.name);
         if let Some(policy) = state.policies.get_mut(&policy_name) {
-            policy.zones.remove(&name);
+            policy.zones.remove(&zone.name);
         }
         return Err(err);
     }
 
-    record_zone_event(center, &zone, HistoricalEvent::Added, None);
-
     {
         let mut state = zone.state.lock().unwrap();
 
+        state.record_event(HistoricalEvent::Added, None);
+
         let source = match source {
             cascade_api::ZoneSource::None => crate::loader::Source::None,
             cascade_api::ZoneSource::Zonefile { path } => crate::loader::Source::Zonefile { path },
@@ -173,12 +182,7 @@ pub async fn add_zone(
         // NOTE: The zone is marked as dirty by the above operation.
     }
 
-    {
-        let mut state = center.state.lock().unwrap();
-        state.mark_dirty(center);
-    }
-
-    info!("Added zone '{name}'");
+    info!("Added zone '{}'", zone.name);
     Ok(())
 }
 
@@ -198,7 +202,7 @@ async fn register_zone(
 /// Remove a zone.
 pub fn remove_zone(center: &Arc<Center>, name: Name<Bytes>) -> Result<(), ZoneRemoveError> {
     let mut state = center.state.lock().unwrap();
-    let zone = state.zones.take(&name).ok_or(ZoneRemoveError::NotFound)?;
+    let zone = state.zones.take(&name).ok_or(ZoneRemoveError::NotFound)?.0;
 
     // Remove the zone from all the places it might be stored.
     // The zone might not have made it to these places, but that's not an issue
@@ -222,10 +226,14 @@ pub fn remove_zone(center: &Arc<Center>, name: Name<Bytes>) -> Result<(), ZoneRe
         z
     });
 
-    let mut zone_state = zone.0.state.lock().unwrap();
+    LoadedReviewServer::remove_zone(center, &zone);
+    SignedReviewServer::remove_zone(center, &zone);
+    PublicationServer::remove_zone(center, &zone);
+
+    let mut zone_state = zone.state.lock().unwrap();
 
     ZoneHandle {
-        zone: &zone.0,
+        zone: &zone,
         state: &mut zone_state,
         center,
     }
@@ -245,7 +253,7 @@ pub fn remove_zone(center: &Arc<Center>, name: Name<Bytes>) -> Result<(), ZoneRe
 
     info!("Removed zone '{name}'");
     zone_state.record_event(HistoricalEvent::Removed, None);
-    zone.0.mark_dirty(&mut zone_state, center);
+    zone.mark_dirty(&mut zone_state, center);
     Ok(())
 }
 

src/lib.rs

@@ -11,6 +11,7 @@ pub mod log;
 pub mod manager;
 pub mod metrics;
 pub mod policy;
+pub mod server;
 pub mod signer;
 pub mod state;
 pub mod tsig;

src/main.rs

@@ -5,11 +5,8 @@ use cascaded::{
     loader::Loader,
     manager::Manager,
     policy,
-    units::{
-        key_manager::KeyManager,
-        zone_server::{Source, ZoneServer},
-        zone_signer::ZoneSigner,
-    },
+    server::{LoadedReviewServer, PublicationServer, SignedReviewServer},
+    units::{key_manager::KeyManager, zone_signer::ZoneSigner},
 };
 use clap::{crate_authors, crate_description};
 use std::{collections::HashMap, fs::create_dir_all};
@@ -207,9 +204,9 @@ fn main() -> ExitCode {
         logger,
         loader: Loader::new(),
         key_manager: KeyManager::new(),
-        unsigned_review_server: ZoneServer::new(Source::Unsigned),
-        signed_review_server: ZoneServer::new(Source::Signed),
-        publication_server: ZoneServer::new(Source::Published),
+        loaded_review_server: LoadedReviewServer::new(),
+        signed_review_server: SignedReviewServer::new(),
+        publication_server: PublicationServer::new(),
         signer: ZoneSigner::new(),
         unsigned_zones: Default::default(),
         signed_zones: Default::default(),

src/manager.rs

@@ -6,10 +6,10 @@ use crate::center::Center;
 use crate::daemon::SocketProvider;
 use crate::loader::Loader;
 use crate::metrics::MetricsCollection;
+use crate::server::{LoadedReviewServer, PublicationServer, SignedReviewServer};
 use crate::units::http_server::HTTP_UNIT_NAME;
 use crate::units::http_server::HttpServer;
 use crate::units::key_manager::KeyManager;
-use crate::units::zone_server::{self, ZoneServer};
 use crate::units::zone_signer::ZoneSigner;
 use crate::util::AbortOnDrop;
 use crate::zone::{HistoricalEvent, Zone};
@@ -43,6 +43,9 @@ impl Manager {
         {
             let mut state = center.state.lock().unwrap();
             Loader::init(&center, &mut state);
+            LoadedReviewServer::init(&center, &mut state);
+            SignedReviewServer::init(&center, &mut state);
+            PublicationServer::init(&center, &mut state);
         }
 
         let mut handles = Vec::new();
@@ -53,11 +56,7 @@ impl Manager {
 
         // Spawn the unsigned zone review server.
         info!("Starting unit 'RS'");
-        handles.extend(ZoneServer::run(
-            &center,
-            zone_server::Source::Unsigned,
-            &mut socket_provider,
-        )?);
+        handles.extend(LoadedReviewServer::run(&center, &mut socket_provider)?);
 
         // Spawn the key manager.
         info!("Starting unit 'KM'");
@@ -69,11 +68,7 @@ impl Manager {
 
         // Spawn the signed zone review server.
         info!("Starting unit 'RS2'");
-        handles.extend(ZoneServer::run(
-            &center,
-            zone_server::Source::Signed,
-            &mut socket_provider,
-        )?);
+        handles.extend(SignedReviewServer::run(&center, &mut socket_provider)?);
 
         // Take out HTTP listen sockets before PS takes them all.
         debug!("Pre-fetching listen sockets for 'HS'");
@@ -99,11 +94,7 @@ impl Manager {
         }
 
         info!("Starting unit 'PS'");
-        handles.extend(ZoneServer::run(
-            &center,
-            zone_server::Source::Published,
-            &mut socket_provider,
-        )?);
+        handles.extend(PublicationServer::run(&center, &mut socket_provider)?);
 
         // Register any Manager metrics here, before giving the metrics to the HttpServer