From 7d16a64fbe498a2163adfede5aa23608c45cd6a5 Mon Sep 17 00:00:00 2001 From: arya dradjica Date: Fri, 20 Mar 2026 10:21:13 +0100 Subject: [PATCH 01/10] Add module 'server' --- src/lib.rs | 1 + src/server/mod.rs | 1 + 2 files changed, 2 insertions(+) create mode 100644 src/server/mod.rs diff --git a/src/lib.rs b/src/lib.rs index 3d59a56d5..e92e80b7b 100644 --- a/src/lib.rs +++ b/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; diff --git a/src/server/mod.rs b/src/server/mod.rs new file mode 100644 index 000000000..2494d9851 --- /dev/null +++ b/src/server/mod.rs @@ -0,0 +1 @@ +//! Serving zone data. From 4e8994f8d672036e884e629fe5ebe58bda63572f Mon Sep 17 00:00:00 2001 From: arya dradjica Date: Tue, 24 Mar 2026 07:06:16 +0100 Subject: [PATCH 02/10] [server] Add '{LoadedReview,SignedReview,Publication}Server' These are specialized versions of 'ZoneServer'. They provide some methods that forward to 'ZoneServer' for now; these will gradually get inlined. Having a dedicated type for each server allows storing type-specific data, prevents methods from being called on the wrong types, and helps simplify the code (removing 'match' statements). --- src/center.rs | 14 ++-- src/main.rs | 13 ++-- src/manager.rs | 20 ++---- src/server/mod.rs | 145 +++++++++++++++++++++++++++++++++++++++ src/units/http_server.rs | 10 +-- src/units/zone_server.rs | 29 +++++--- src/zone/storage.rs | 5 +- 7 files changed, 191 insertions(+), 45 deletions(-) diff --git a/src/center.rs b/src/center.rs index 67a799684..8a8414e77 100644 --- a/src/center.rs +++ b/src/center.rs @@ -20,8 +20,8 @@ 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 +56,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>, diff --git a/src/main.rs b/src/main.rs index ec0823f5e..258aad356 100644 --- a/src/main.rs +++ b/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(), diff --git a/src/manager.rs b/src/manager.rs index ebb53cb39..2c12c5b96 100644 --- a/src/manager.rs +++ b/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}; @@ -53,11 +53,7 @@ impl Manager { // Spawn the unsigned zone review server. info!("Starting unit 'RS'"); - handles.extend(ZoneServer::run( - ¢er, - zone_server::Source::Unsigned, - &mut socket_provider, - )?); + handles.extend(LoadedReviewServer::run(¢er, &mut socket_provider)?); // Spawn the key manager. info!("Starting unit 'KM'"); @@ -69,11 +65,7 @@ impl Manager { // Spawn the signed zone review server. info!("Starting unit 'RS2'"); - handles.extend(ZoneServer::run( - ¢er, - zone_server::Source::Signed, - &mut socket_provider, - )?); + 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'"); @@ -99,11 +91,7 @@ impl Manager { } info!("Starting unit 'PS'"); - handles.extend(ZoneServer::run( - ¢er, - zone_server::Source::Published, - &mut socket_provider, - )?); + handles.extend(PublicationServer::run(¢er, &mut socket_provider)?); // Register any Manager metrics here, before giving the metrics to the HttpServer diff --git a/src/server/mod.rs b/src/server/mod.rs index 2494d9851..a8e3c4fe0 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -1 +1,146 @@ //! Serving zone data. + +use std::sync::Arc; + +use cascade_api::{ZoneReviewDecision, ZoneReviewResult}; +use domain::base::Serial; + +use crate::{ + center::Center, + daemon::SocketProvider, + manager::Terminated, + units::zone_server::{Source, ZoneServer}, + util::AbortOnDrop, + zone::Zone, +}; + +//----------- LoadedReviewServer ----------------------------------------------- + +/// The review server for loaded instances of zones. +#[derive(Debug)] +pub struct LoadedReviewServer {} + +impl LoadedReviewServer { + /// Construct a new [`LoadedReviewServer`]. + pub fn new() -> Self { + Self {} + } + + /// Drive the server. + pub fn run( + center: &Arc
, + socket_provider: &mut SocketProvider, + ) -> Result, Terminated> { + // TODO: Inline. + ZoneServer::run(center, Source::Unsigned, socket_provider) + } + + /// Start reviewing a newly loaded instance. + pub fn start_review( + center: &Arc
, + zone: &Arc, + zone_serial: Serial, + ) -> Option> { + // TODO: Inline. + ZoneServer::new(Source::Unsigned).on_seek_approval_for_zone(center, zone, zone_serial) + } + + /// Process a review of a served instance. + pub fn process_review( + center: &Arc
, + zone: &Arc, + zone_serial: Serial, + decision: ZoneReviewDecision, + ) -> ZoneReviewResult { + // TODO: Inline. + ZoneServer::new(Source::Unsigned).on_zone_review(center, zone, zone_serial, decision) + } +} + +impl Default for LoadedReviewServer { + fn default() -> Self { + Self::new() + } +} + +//----------- SignedReviewServer ----------------------------------------------- + +/// The review server for signed instances of zones. +#[derive(Debug)] +pub struct SignedReviewServer {} + +impl SignedReviewServer { + /// Construct a new [`SignedReviewServer`]. + pub fn new() -> Self { + Self {} + } + + /// Drive the server. + pub fn run( + center: &Arc
, + socket_provider: &mut SocketProvider, + ) -> Result, Terminated> { + // TODO: Inline. + ZoneServer::run(center, Source::Signed, socket_provider) + } + + /// Start reviewing a newly signed instance. + pub fn start_review( + center: &Arc
, + zone: &Arc, + zone_serial: Serial, + ) -> Option> { + // TODO: Inline. + ZoneServer::new(Source::Signed).on_seek_approval_for_zone(center, zone, zone_serial) + } + + /// Process a review of a served instance. + pub fn process_review( + center: &Arc
, + zone: &Arc, + zone_serial: Serial, + decision: ZoneReviewDecision, + ) -> ZoneReviewResult { + // TODO: Inline. + ZoneServer::new(Source::Signed).on_zone_review(center, zone, zone_serial, decision) + } +} + +impl Default for SignedReviewServer { + fn default() -> Self { + Self::new() + } +} + +//----------- PublicationServer ------------------------------------------------ + +/// The server for published instances of zones. +#[derive(Debug)] +pub struct PublicationServer {} + +impl PublicationServer { + /// Construct a new [`PublicationServer`]. + pub fn new() -> Self { + Self {} + } + + /// Drive the server. + pub fn run( + center: &Arc
, + socket_provider: &mut SocketProvider, + ) -> Result, Terminated> { + ZoneServer::run(center, Source::Published, socket_provider) + } + + /// Publish an instance. + pub fn publish(center: &Arc
, zone: &Arc, zone_serial: Serial) { + // TODO: Inline. + ZoneServer::new(Source::Published).on_publish_signed_zone(center, zone, zone_serial) + } +} + +impl Default for PublicationServer { + fn default() -> Self { + Self::new() + } +} diff --git a/src/units/http_server.rs b/src/units/http_server.rs index 0d223b314..85220678a 100644 --- a/src/units/http_server.rs +++ b/src/units/http_server.rs @@ -42,6 +42,8 @@ use crate::manager::Terminated; use crate::metrics::MetricsCollection; use crate::policy::SignerDenialPolicy; use crate::policy::SignerSerialPolicy; +use crate::server::LoadedReviewServer; +use crate::server::SignedReviewServer; use crate::units::key_manager::KmipClientCredentials; use crate::units::key_manager::KmipClientCredentialsFile; use crate::units::key_manager::KmipServerCredentialsFileMode; @@ -636,7 +638,7 @@ impl HttpServer { ); return Json(Err(ZoneReviewError::NoSuchZone)); }; - let result = center.unsigned_review_server.on_zone_review( + let result = LoadedReviewServer::process_review( center, &zone, zone_serial, @@ -658,7 +660,7 @@ impl HttpServer { ); return Json(Err(ZoneReviewError::NoSuchZone)); }; - let result = center.unsigned_review_server.on_zone_review( + let result = LoadedReviewServer::process_review( center, &zone, zone_serial, @@ -709,7 +711,7 @@ impl HttpServer { ); return Json(Err(ZoneReviewError::NoSuchZone)); }; - let result = center.signed_review_server.on_zone_review( + let result = SignedReviewServer::process_review( center, &zone, zone_serial, @@ -731,7 +733,7 @@ impl HttpServer { ); return Json(Err(ZoneReviewError::NoSuchZone)); }; - let result = center.signed_review_server.on_zone_review( + let result = SignedReviewServer::process_review( center, &zone, zone_serial, diff --git a/src/units/zone_server.rs b/src/units/zone_server.rs index 9dd3f3ccc..64affb9b5 100644 --- a/src/units/zone_server.rs +++ b/src/units/zone_server.rs @@ -43,6 +43,7 @@ use crate::config::SocketConfig; use crate::daemon::SocketProvider; use crate::manager::Terminated; use crate::manager::record_zone_event; +use crate::server::{LoadedReviewServer, PublicationServer, SignedReviewServer}; use crate::util::AbortOnDrop; use crate::zone::{ HistoricalEvent, SignedZoneVersionState, UnsignedZoneVersionState, Zone, ZoneHandle, @@ -464,6 +465,7 @@ impl ZoneServer { let _: Result<_, _> = Self::process_output(stderr, true).await; }); let zone = zone.clone(); + let source = self.source; tokio::spawn(async move { let status = match child.wait().await { Ok(status) => status, @@ -480,12 +482,25 @@ impl ZoneServer { false => ZoneReviewDecision::Reject, }; - let server = match zone_type { - "unsigned" => ¢er.unsigned_review_server, - "signed" => ¢er.signed_review_server, - _ => unreachable!(), + match source { + Source::Unsigned => { + let _ = LoadedReviewServer::process_review( + ¢er, + &zone, + zone_serial, + decision, + ); + } + Source::Signed => { + let _ = SignedReviewServer::process_review( + ¢er, + &zone, + zone_serial, + decision, + ); + } + Source::Published => unreachable!(), }; - let _ = server.on_zone_review(¢er, &zone, zone_serial, decision); }); } Err(err) => { @@ -545,9 +560,7 @@ impl ZoneServer { center.signer.on_publish_signed_zone(center); info!("[CC]: Instructing publication server to publish the signed zone"); - center - .publication_server - .on_publish_signed_zone(center, zone, zone_serial); + PublicationServer::publish(center, zone, zone_serial); } async fn process_output( diff --git a/src/zone/storage.rs b/src/zone/storage.rs index ec13ed08c..05f5be626 100644 --- a/src/zone/storage.rs +++ b/src/zone/storage.rs @@ -36,6 +36,7 @@ use tracing::{info, trace, trace_span, warn}; use crate::{ center::Center, common::light_weight_zone::LightWeightZone, + server::{LoadedReviewServer, SignedReviewServer}, util::{BackgroundTasks, force_future}, zone::{HistoricalEvent, Zone, ZoneHandle, ZoneState}, }; @@ -237,7 +238,7 @@ impl StorageZoneHandle<'_> { // TODO: 'on_seek_approval_for_zone' tries to lock zone state. std::mem::drop(state); - center.unsigned_review_server.on_seek_approval_for_zone( + LoadedReviewServer::start_review( ¢er, &zone, domain::base::Serial(serial.into()), @@ -555,7 +556,7 @@ impl StorageZoneHandle<'_> { // TODO: 'on_seek_approval_for_zone' tries to lock zone state. std::mem::drop(state); - center.signed_review_server.on_seek_approval_for_zone( + SignedReviewServer::start_review( ¢er, &zone, domain::base::Serial(serial.into()), From 65bd2bb076e47726448a6fc52730a95452f4a6e3 Mon Sep 17 00:00:00 2001 From: arya dradjica Date: Fri, 20 Mar 2026 11:00:37 +0100 Subject: [PATCH 03/10] [server] Add module 'request' This performs basic parsing of request messages. In the future, it will collect data across the whole DNS message, e.g. EDNS options and TSIG state. --- src/server/mod.rs | 2 + src/server/request.rs | 161 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 163 insertions(+) create mode 100644 src/server/request.rs diff --git a/src/server/mod.rs b/src/server/mod.rs index a8e3c4fe0..01445d4a2 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -14,6 +14,8 @@ use crate::{ zone::Zone, }; +mod request; + //----------- LoadedReviewServer ----------------------------------------------- /// The review server for loaded instances of zones. diff --git a/src/server/request.rs b/src/server/request.rs new file mode 100644 index 000000000..44c4a248a --- /dev/null +++ b/src/server/request.rs @@ -0,0 +1,161 @@ +//! Parsing DNS requests. +//! +//! This module provides [`Request`], which DNS request messages can be parsed +//! into. It collates all relevant information from the request message and +//! limits itself to the kinds of requests supported by Cascade. + +use domain::{ + new::{ + base::{ + Message, MessageItem, QType, RClass, RType, Record, + name::{Name, RevName}, + wire, + }, + rdata::{RecordData, Soa}, + }, + utils::dst::UnsizedCopy, +}; + +/// Parse a DNS request message into a [`Request`]. +pub fn parse(message: &Message) -> Result { + let mut parser = message.parse(); + + // TODO: Store domain names in a bump allocator. + // TODO: Check the message header. + // TODO: Support EDNS cookie requests with QCOUNT=0. + // TODO: Check for NOTIFY messages. + + // Extract the question in the message. + let Some(MessageItem::Question(question)) = parser.next().transpose()? else { + return Err(RequestParseError::Wire(wire::ParseError)); + }; + // TODO: Check all fields of 'question'. + + match question.qtype { + QType::SOA => { + // This is a zone-related query for a SOA record. + // + // TODO: Check for later records. + Ok(Request { + kind: RequestKind::Zone(ZoneRequest { + name: question.qname.unsized_copy_into(), + kind: ZoneRequestKind::Soa, + }), + }) + } + + qt if qt.code == 252 => { + // TODO: 'QType::AXFR' + // This is a zone-related request for an AXFR. + // + // TODO: Check for later records. + Ok(Request { + kind: RequestKind::Zone(ZoneRequest { + name: question.qname.unsized_copy_into(), + kind: ZoneRequestKind::Axfr, + }), + }) + } + + qt if qt.code == 251 => { + // TODO: 'QType::IXFR' + // This is a zone-related request for an IXFR. + + // Parse the SOA record known to the client. + let Some(MessageItem::Authority(Record { + rname, + rtype: rtype @ RType::SOA, + rclass: rclass @ RClass::IN, + ttl, + rdata: RecordData::Soa(rdata), + })) = parser.next().transpose()? + else { + return Err(RequestParseError::Wire(wire::ParseError)); + }; + if rname != question.qname { + return Err(RequestParseError::Wire(wire::ParseError)); + } + let known_soa = Record { + rname: (), + rtype, + rclass, + ttl, + rdata: rdata.map_names(|n| n.unsized_copy_into()), + }; + + // TODO: Check for later records. + + Ok(Request { + kind: RequestKind::Zone(ZoneRequest { + name: question.qname.unsized_copy_into(), + kind: ZoneRequestKind::Ixfr { known_soa }, + }), + }) + } + + // TODO: Return a more appropriate error type? + _ => Err(RequestParseError::Wire(wire::ParseError)), + } +} + +/// A DNS request. +// +// TODO: Borrow from the original message. +pub struct Request { + /// The kind of request. + pub kind: RequestKind, + // + // TODO: + // - EDNS cookie state. + // - Other EDNS options? + // - TSIG state. +} + +/// A kind of DNS request. +pub enum RequestKind { + /// A zone related request. + Zone(ZoneRequest), + // + // TODO: Generic queries. +} + +/// A DNS request related to a zone. +pub struct ZoneRequest { + /// The name of the relevant zone. + // + // TODO: Borrow this data. + pub name: Box, + + /// The kind of zone request. + pub kind: ZoneRequestKind, +} + +/// A kind of zone-related DNS request. +pub enum ZoneRequestKind { + /// A query for the SOA record. + Soa, + + /// An AXFR request. + Axfr, + + /// An IXFR request. + #[expect(dead_code)] + Ixfr { + /// The SOA record known to the client. + known_soa: Record<(), Soa>>, + }, + // + // TODO: NOTIFY messages. +} + +/// An error parsing a DNS request. +pub enum RequestParseError { + /// A low-level wire format parsing error. + Wire(wire::ParseError), +} + +impl From for RequestParseError { + fn from(error: wire::ParseError) -> Self { + Self::Wire(error) + } +} From f513728410fc73548031c7820c800b65e1bee896 Mon Sep 17 00:00:00 2001 From: arya dradjica Date: Fri, 20 Mar 2026 11:00:37 +0100 Subject: [PATCH 04/10] [server] Add module 'service' The newly introduced 'ZoneService' will serve as the basis for the zone server. It integrates with 'domain::net::server' and so should be a drop-in replacement for the existing query service. At the moment, 'ZoneService' only supports SOA queries. Support for AXFR and IXFR will follow; NOTIFY messages and extensions like EDNS and TSIG are left to the existing middleware services for now. --- src/server/mod.rs | 1 + src/server/service.rs | 319 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 320 insertions(+) create mode 100644 src/server/service.rs diff --git a/src/server/mod.rs b/src/server/mod.rs index 01445d4a2..210f1f50f 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -15,6 +15,7 @@ use crate::{ }; mod request; +mod service; //----------- LoadedReviewServer ----------------------------------------------- diff --git a/src/server/service.rs b/src/server/service.rs new file mode 100644 index 000000000..039bc9410 --- /dev/null +++ b/src/server/service.rs @@ -0,0 +1,319 @@ +//! Servicing DNS requests. + +use std::sync::{Arc, RwLock}; + +use cascade_zonedata::{LoadedZoneReviewer, SignedZoneReviewer, SoaRecord, ZoneViewer}; +use domain::{ + new::base::{ + name::{RevName, RevNameBuf}, + wire::ParseBytes, + }, + utils::dst::UnsizedCopy, +}; + +use crate::zone::Zone; + +//----------- ZoneService ------------------------------------------------------ + +/// Cascade's DNS service. +/// +/// This type is responsible for answering DNS queries received by Cascade, +/// using zone viewer objects. When a query is received, the corresponding zone +/// is looked up, the appropriate information is retrieved from its viewer, and +/// the appropriate DNS answer is synthesized and returned. +/// +/// The viewer type has to implement [`Viewer`]. It should only be +/// [`LoadedZoneReviewer`], [`SignedZoneReviewer`], or [`ZoneViewer`]. +/// +/// The service can be interacted with through a [`ZoneServiceHandle`], which +/// is also returned by [`ZoneService::new()`]. +pub struct ZoneService { + /// The underlying state. + // + // TODO: The state is currently wrapped in an 'RwLock'. This is necessary + // as the state might have to change, e.g. in response to changes to zones. + // It would be preferable to hold the state directly, and use channels to + // communicate the need for changes. This is currently impossible because + // of the limitations of 'domain::net::server'; that architecture should be + // gradually replaced locally for better flexibility and efficiency. + state: Arc>>, +} + +impl ZoneService { + /// Construct a new [`ZoneService`]. + /// + /// In addition to the service, a [`ZoneServiceHandle`] is returned through + /// which the service can be interacted with. + pub fn new() -> (ZoneService, ZoneServiceHandle) { + let state = Arc::new(std::sync::RwLock::default()); + let service = ZoneService { + state: state.clone(), + }; + let handle = ZoneServiceHandle { state }; + (service, handle) + } +} + +impl Clone for ZoneService { + fn clone(&self) -> Self { + Self { + state: self.state.clone(), + } + } +} + +/// A compatibility layer to [`domain::net::server`]. +/// +/// In the future, the network server stack should be gradually inlined here, +/// so it can use [`domain::new`] and support more functionality (e.g. handling +/// XFRs by spawning OS threads). +mod compat { + use std::{pin::Pin, sync::Arc}; + + use cascade_zonedata::OldRecord; + use domain::{ + base::{Message, MessageBuilder, iana::Rcode}, + net::server::{ + message::Request, + service::{CallResult, Service, ServiceResult}, + }, + new::base::wire::ParseBytesZC, + tsig, + }; + use futures::Stream; + + use crate::server::request::{RequestKind, ZoneRequestKind}; + + use super::{Viewer, ZoneService}; + + impl Service, Option>> for ZoneService + where + V: Viewer + Send + Sync + 'static, + { + type Target = Vec; + type Stream = ResponseStream; + type Future = Response; + + fn call(&self, old_request: Request, Option>>) -> Response { + // Parse the request. + let message = old_request.message().as_slice(); + let message = domain::new::base::Message::parse_bytes_by_ref(message) + .expect("'message' was already checked to be a valid DNS message"); + let request = match crate::server::request::parse(message) { + Ok(request) => request, + Err(_error) => { + // TODO: Generate the response using 'error'. + return Box::pin(std::future::ready(error( + old_request.message(), + Rcode::FORMERR, + ))); + } + }; + + // Determine how to handle the request. + match request.kind { + RequestKind::Zone(zone_request) => { + // Look up the relevant zone. + let state = self.state.read().unwrap(); + let Some(zone) = state.zones.get(&*zone_request.name) else { + // No such zone could be found. + let rcode = match zone_request.kind { + // Return NXDOMAIN for normal queries. + ZoneRequestKind::Soa => Rcode::NXDOMAIN, + // Return NOTAUTH for zone transfers. + ZoneRequestKind::Axfr | ZoneRequestKind::Ixfr { .. } => Rcode::NOTAUTH, + }; + return Box::pin(std::future::ready(error(old_request.message(), rcode))); + }; + + match zone_request.kind { + ZoneRequestKind::Soa => Box::pin({ + let viewer = zone.viewer.clone(); + async move { + let viewer = viewer.read_owned().await; + soa(old_request.message(), &*viewer) + } + }) as Response, + + _ => todo!(), + } + } + } + } + } + + fn soa(request: &Message>, viewer: &V) -> ResponseStream { + if viewer.is_empty() { + // The zone is known to exist, but we don't have any data for it. + return error(request, Rcode::NXDOMAIN); + } + let soa = viewer.soa().clone(); + + let builder = MessageBuilder::new_stream_vec(); + let mut builder = builder.start_answer(request, Rcode::NOERROR).unwrap(); + builder.header_mut().set_aa(true); + builder.push(OldRecord::from(soa)).unwrap(); + + let response = builder.additional(); + let result = Ok(CallResult::new(response)); + Box::new(futures::stream::once(std::future::ready(result))) as _ + } + + fn error(request: &Message>, rcode: Rcode) -> ResponseStream { + let response = MessageBuilder::new_stream_vec() + .start_error(request, rcode) + .additional(); + let result = Ok(CallResult::new(response)); + Box::new(futures::stream::once(std::future::ready(result))) as _ + } + + type ResponseStream = Box>> + Unpin + Send + Sync>; + type Response = Pin + Send + Sync>>; +} + +//----------- Viewer ----------------------------------------------------------- + +/// A viewer through which zone data can be served. +trait Viewer { + /// Whether the zone instance is empty. + fn is_empty(&self) -> bool; + + /// Return the SOA record. + fn soa(&self) -> &SoaRecord; +} + +impl Viewer for LoadedZoneReviewer { + fn is_empty(&self) -> bool { + self.read_loaded().is_none() + } + + fn soa(&self) -> &SoaRecord { + self.read_loaded().unwrap().soa() + } +} + +impl Viewer for SignedZoneReviewer { + fn is_empty(&self) -> bool { + self.read().is_none() + } + + fn soa(&self) -> &SoaRecord { + self.read().unwrap().soa() + } +} + +impl Viewer for ZoneViewer { + fn is_empty(&self) -> bool { + self.read().is_none() + } + + fn soa(&self) -> &SoaRecord { + self.read().unwrap().soa() + } +} + +//----------- ZoneServiceHandle ------------------------------------------------ + +/// A handle for controlling a [`ZoneService`]. +pub struct ZoneServiceHandle { + /// The underlying state. + state: Arc>>, +} + +impl ZoneServiceHandle { + /// Register a new zone. + /// + /// ## Panics + /// + /// Panics if the zone is already registered. + pub fn add_zone(&self, zone: Arc, viewer: V) { + let mut state = self.state.write().unwrap(); + let name = RevNameBuf::parse_bytes(zone.name.as_slice()).unwrap(); + let zone = ServedZone { + handle: zone, + viewer: Arc::new(tokio::sync::RwLock::new(viewer)), + }; + let previous = state.zones.insert(name.unsized_copy_into(), zone); + assert!(previous.is_none(), "the zone is already registered"); + } + + /// Update the viewer of a zone. + /// + /// The old viewer is returned. + /// + /// ## Panics + /// + /// Panics if the zone has not been registered. + pub async fn update_viewer(&self, zone: &Arc, viewer: V) -> V { + // Locate the slot for the viewer. + let slot = { + let state = self.state.read().unwrap(); + let name = RevNameBuf::parse_bytes(zone.name.as_slice()).unwrap(); + let zone = state + .zones + .get(&*name) + .expect("the zone has been registered"); + zone.viewer.clone() + }; + let mut slot = slot.write().await; + std::mem::replace(&mut *slot, viewer) + } + + /// Remove a zone. + /// + /// ## Panics + /// + /// Panics if the zone is not known to the service. + pub fn remove_zone(&self, zone: &Arc) { + let mut state = self.state.write().unwrap(); + let name = RevNameBuf::parse_bytes(zone.name.as_slice()).unwrap(); + let existed = state.zones.remove(&*name); + let ServedZone { handle, viewer } = existed.expect("the zone exists"); + assert!( + Arc::ptr_eq(&handle, zone), + "distinct 'Arc's had the same name" + ); + let _ = viewer; + } +} + +//----------- ZoneServiceState ------------------------------------------------- + +/// State for serving zone data. +struct ZoneServiceState { + /// Zones being served. + zones: foldhash::HashMap, ServedZone>, +} + +impl Default for ZoneServiceState { + fn default() -> Self { + Self { + zones: Default::default(), + } + } +} + +/// A zone being served. +struct ServedZone { + /// The zone handle. + handle: Arc, + + /// The viewer for the zone. + /// + /// [`tokio::sync::RwLock`] is used as viewers might be held for extended + /// periods of time (e.g. across whole AXFRs). + // + // TODO: Use a more advanced double-buffering mechanism that allows new + // readers to use the updated value, while old readers can retain their + // lock. + viewer: Arc>, +} + +impl Clone for ServedZone { + fn clone(&self) -> Self { + Self { + handle: self.handle.clone(), + viewer: self.viewer.clone(), + } + } +} From 32a88b3451bf023b9c79080bd05322e460a56120 Mon Sep 17 00:00:00 2001 From: arya dradjica Date: Tue, 24 Mar 2026 09:10:40 +0100 Subject: [PATCH 05/10] [server] Store 'ZoneService's and their handles These are not used yet. --- src/server/mod.rs | 149 +++++++++++++++++++++++++++++++++++---- src/units/zone_server.rs | 13 +++- 2 files changed, 146 insertions(+), 16 deletions(-) diff --git a/src/server/mod.rs b/src/server/mod.rs index 210f1f50f..0bc99fc64 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -1,8 +1,9 @@ //! Serving zone data. -use std::sync::Arc; +use std::{fmt, sync::Arc}; use cascade_api::{ZoneReviewDecision, ZoneReviewResult}; +use cascade_zonedata::{LoadedZoneReviewer, SignedZoneReviewer, ZoneViewer}; use domain::base::Serial; use crate::{ @@ -17,16 +18,24 @@ use crate::{ mod request; mod service; +use service::{ZoneService, ZoneServiceHandle}; + //----------- LoadedReviewServer ----------------------------------------------- /// The review server for loaded instances of zones. -#[derive(Debug)] -pub struct LoadedReviewServer {} +pub struct LoadedReviewServer { + /// The underlying service. + service: ZoneService, + + /// A handle for controlling the service. + handle: ZoneServiceHandle, +} impl LoadedReviewServer { /// Construct a new [`LoadedReviewServer`]. pub fn new() -> Self { - Self {} + let (service, handle) = ZoneService::new(); + Self { service, handle } } /// Drive the server. @@ -35,7 +44,12 @@ impl LoadedReviewServer { socket_provider: &mut SocketProvider, ) -> Result, Terminated> { // TODO: Inline. - ZoneServer::run(center, Source::Unsigned, socket_provider) + ZoneServer::run( + center, + Source::Unsigned, + socket_provider, + center.loaded_review_server.service.clone(), + ) } /// Start reviewing a newly loaded instance. @@ -58,6 +72,29 @@ impl LoadedReviewServer { // TODO: Inline. ZoneServer::new(Source::Unsigned).on_zone_review(center, zone, zone_serial, decision) } + + /// Register a new zone. + pub fn add_zone(center: &Arc
, zone: Arc, viewer: LoadedZoneReviewer) { + let handle = ¢er.loaded_review_server.handle; + handle.add_zone(zone, viewer) + } + + /// Update the viewer of a zone. + #[tracing::instrument(level = "trace", skip_all, fields(zone = %zone.name))] + pub async fn update_viewer( + center: &Arc
, + zone: &Arc, + viewer: LoadedZoneReviewer, + ) -> LoadedZoneReviewer { + let handle = ¢er.loaded_review_server.handle; + handle.update_viewer(zone, viewer).await + } + + /// Remove a zone. + pub fn remove_zone(center: &Arc
, zone: &Arc) { + let handle = ¢er.loaded_review_server.handle; + handle.remove_zone(zone); + } } impl Default for LoadedReviewServer { @@ -66,16 +103,28 @@ impl Default for LoadedReviewServer { } } +impl fmt::Debug for LoadedReviewServer { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("LoadedReviewServer").finish_non_exhaustive() + } +} + //----------- SignedReviewServer ----------------------------------------------- /// The review server for signed instances of zones. -#[derive(Debug)] -pub struct SignedReviewServer {} +pub struct SignedReviewServer { + /// The underlying service. + service: ZoneService, + + /// A handle for controlling the service. + handle: ZoneServiceHandle, +} impl SignedReviewServer { /// Construct a new [`SignedReviewServer`]. pub fn new() -> Self { - Self {} + let (service, handle) = ZoneService::new(); + Self { service, handle } } /// Drive the server. @@ -84,7 +133,12 @@ impl SignedReviewServer { socket_provider: &mut SocketProvider, ) -> Result, Terminated> { // TODO: Inline. - ZoneServer::run(center, Source::Signed, socket_provider) + ZoneServer::run( + center, + Source::Signed, + socket_provider, + center.signed_review_server.service.clone(), + ) } /// Start reviewing a newly signed instance. @@ -107,6 +161,29 @@ impl SignedReviewServer { // TODO: Inline. ZoneServer::new(Source::Signed).on_zone_review(center, zone, zone_serial, decision) } + + /// Register a new zone. + pub fn add_zone(center: &Arc
, zone: Arc, viewer: SignedZoneReviewer) { + let handle = ¢er.signed_review_server.handle; + handle.add_zone(zone, viewer) + } + + /// Update the viewer of a zone. + #[tracing::instrument(level = "trace", skip_all, fields(zone = %zone.name))] + pub async fn update_viewer( + center: &Arc
, + zone: &Arc, + viewer: SignedZoneReviewer, + ) -> SignedZoneReviewer { + let handle = ¢er.signed_review_server.handle; + handle.update_viewer(zone, viewer).await + } + + /// Remove a zone. + pub fn remove_zone(center: &Arc
, zone: &Arc) { + let handle = ¢er.signed_review_server.handle; + handle.remove_zone(zone); + } } impl Default for SignedReviewServer { @@ -115,16 +192,28 @@ impl Default for SignedReviewServer { } } +impl fmt::Debug for SignedReviewServer { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("SignedReviewServer").finish_non_exhaustive() + } +} + //----------- PublicationServer ------------------------------------------------ /// The server for published instances of zones. -#[derive(Debug)] -pub struct PublicationServer {} +pub struct PublicationServer { + /// The underlying service. + service: ZoneService, + + /// A handle for controlling the service. + handle: ZoneServiceHandle, +} impl PublicationServer { /// Construct a new [`PublicationServer`]. pub fn new() -> Self { - Self {} + let (service, handle) = ZoneService::new(); + Self { service, handle } } /// Drive the server. @@ -132,7 +221,12 @@ impl PublicationServer { center: &Arc
, socket_provider: &mut SocketProvider, ) -> Result, Terminated> { - ZoneServer::run(center, Source::Published, socket_provider) + ZoneServer::run( + center, + Source::Published, + socket_provider, + center.publication_server.service.clone(), + ) } /// Publish an instance. @@ -140,6 +234,29 @@ impl PublicationServer { // TODO: Inline. ZoneServer::new(Source::Published).on_publish_signed_zone(center, zone, zone_serial) } + + /// Register a new zone. + pub fn add_zone(center: &Arc
, zone: Arc, viewer: ZoneViewer) { + let handle = ¢er.publication_server.handle; + handle.add_zone(zone, viewer) + } + + /// Update the viewer of a zone. + #[tracing::instrument(level = "trace", skip_all, fields(zone = %zone.name))] + pub async fn update_viewer( + center: &Arc
, + zone: &Arc, + viewer: ZoneViewer, + ) -> ZoneViewer { + let handle = ¢er.publication_server.handle; + handle.update_viewer(zone, viewer).await + } + + /// Remove a zone. + pub fn remove_zone(center: &Arc
, zone: &Arc) { + let handle = ¢er.publication_server.handle; + handle.remove_zone(zone); + } } impl Default for PublicationServer { @@ -147,3 +264,9 @@ impl Default for PublicationServer { Self::new() } } + +impl fmt::Debug for PublicationServer { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("PublicationServer").finish_non_exhaustive() + } +} diff --git a/src/units/zone_server.rs b/src/units/zone_server.rs index 64affb9b5..0e4002156 100644 --- a/src/units/zone_server.rs +++ b/src/units/zone_server.rs @@ -163,11 +163,17 @@ impl ZoneServer { } /// Launch a zone server. - pub fn run( + pub fn run( center: &Arc
, source: Source, socket_provider: &mut SocketProvider, - ) -> Result, Terminated> { + service: S, + ) -> Result, Terminated> + where + S: Service, Option>> + Unpin + Clone, + S::Future: Unpin + Sync, + S::Stream: Sync, + { let unit_name = match source { Source::Unsigned => "RS", Source::Signed => "RS2", @@ -203,7 +209,8 @@ impl ZoneServer { center: center.clone(), }; - // let svc = ZoneServerService::new(zones.clone()); + let _ = service; + // let svc = service; let svc = service_fn(zone_server_service, zones.clone()); let svc = XfrMiddlewareSvc::new(svc, zones.clone(), max_concurrency); let svc = NotifyMiddlewareSvc::new(svc, notifier); From 2bfe88b10de1ca3f94de348b792f4c2f24a8dbeb Mon Sep 17 00:00:00 2001 From: arya dradjica Date: Tue, 24 Mar 2026 09:10:40 +0100 Subject: [PATCH 06/10] [storage] Pass (re)viewers to the zone servers --- src/center.rs | 88 +++++++------ src/manager.rs | 3 + src/server/mod.rs | 34 ++++- src/zone/storage.rs | 304 ++++++++++++++++++++++++++++++-------------- 4 files changed, 293 insertions(+), 136 deletions(-) diff --git a/src/center.rs b/src/center.rs index 8a8414e77..f48330aeb 100644 --- a/src/center.rs +++ b/src/center.rs @@ -19,7 +19,6 @@ 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_signer::ZoneSigner; @@ -88,40 +87,49 @@ pub async fn add_zone( source: api::ZoneSource, key_imports: Vec, ) -> 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 @@ -129,22 +137,23 @@ pub async fn add_zone( // 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
, name: Name) -> 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
, name: Name) -> 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
, name: Name) -> 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(()) } diff --git a/src/manager.rs b/src/manager.rs index 2c12c5b96..e3eda35c1 100644 --- a/src/manager.rs +++ b/src/manager.rs @@ -43,6 +43,9 @@ 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(); diff --git a/src/server/mod.rs b/src/server/mod.rs index 0bc99fc64..f8a181a67 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, + center::{Center, State}, daemon::SocketProvider, manager::Terminated, units::zone_server::{Source, ZoneServer}, util::AbortOnDrop, - zone::Zone, + zone::{Zone, ZoneByName}, }; mod request; @@ -38,6 +38,16 @@ 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
, @@ -127,6 +137,16 @@ 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
, @@ -216,6 +236,16 @@ 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/zone/storage.rs b/src/zone/storage.rs index 05f5be626..83e70f021 100644 --- a/src/zone/storage.rs +++ b/src/zone/storage.rs @@ -36,7 +36,7 @@ use tracing::{info, trace, trace_span, warn}; use crate::{ center::Center, common::light_weight_zone::LightWeightZone, - server::{LoadedReviewServer, SignedReviewServer}, + server::{LoadedReviewServer, PublicationServer, SignedReviewServer}, util::{BackgroundTasks, force_future}, zone::{HistoricalEvent, Zone, ZoneHandle, ZoneState}, }; @@ -193,7 +193,7 @@ impl StorageZoneHandle<'_> { let zone = self.zone.clone(); let center = self.center.clone(); let span = trace_span!("start_loaded_review"); - self.state.storage.background_tasks.spawn_blocking(span, move || { + self.state.storage.background_tasks.spawn(span, async move { trace!("Converting the loaded instance to 'zonetree'"); // Read the loaded instance. @@ -202,26 +202,37 @@ impl StorageZoneHandle<'_> { .unwrap_or_else(|| unreachable!("The loader never returns an empty instance")); let serial = reader.soa().rdata.serial; - // Build a compatibility shim for the new instance. - let zonetree_zone = Self::build_compat_for_loaded(&zone, &reader); + let loaded_reviewer = tokio::task::spawn_blocking({ + let zone = zone.clone(); + let center = center.clone(); + move || { + // Read the loaded instance. + let reader = loaded_reviewer + .read_loaded() + .unwrap_or_else(|| unreachable!("The loader never returns an empty instance")); + + // Build a compatibility shim for the new instance. + let zonetree_zone = Self::build_compat_for_loaded(&zone, &reader); + + // Insert the compatibility shim in the global view (possibly + // replacing a previous one). + center.unsigned_zones.rcu(|tree| { + let mut tree = Arc::unwrap_or_clone(tree.clone()); + let _ = tree.remove_zone(&zone.name, domain::base::iana::Class::IN); + tree.insert_zone(zonetree_zone.clone()).unwrap(); + tree + }); + + loaded_reviewer + } + }).await.unwrap(); - // Insert the compatibility shim in the global view (possibly - // replacing a previous one). - center.unsigned_zones.rcu(|tree| { - let mut tree = Arc::unwrap_or_clone(tree.clone()); - let _ = tree.remove_zone(&zone.name, domain::base::iana::Class::IN); - tree.insert_zone(zonetree_zone.clone()).unwrap(); - tree - }); + trace!("Updating the viewer in 'LoadedReviewServer'"); + let old_loaded_reviewer = LoadedReviewServer::update_viewer(¢er, &zone, loaded_reviewer).await; let mut state = zone.state.lock().unwrap(); - // TODO: Pass on the reviewer to the zone server. - let old_loaded_reviewer = - std::mem::replace(&mut state.storage.loaded_reviewer, loaded_reviewer); - // Transition into the reviewing state. - trace!("Initiating loaded review"); match transition(&mut state.storage.machine) { (transition, ZoneDataStorage::ReviewLoadedPending(s)) => { let s = s.start(old_loaded_reviewer); @@ -304,25 +315,23 @@ impl StorageZoneHandle<'_> { )] pub fn abandon_loaded_review(&mut self) { // Examine the current state. - let (transition, state) = transition(&mut self.state.storage.machine); - match state { - ZoneDataStorage::ReviewingLoaded(s) => { + let loaded_reviewer = match transition(&mut self.state.storage.machine) { + (transition, ZoneDataStorage::ReviewingLoaded(s)) => { // TODO: Specify the instance ID. info!("The loaded instance has been rejected; cleaning it up"); let (s, loaded_reviewer) = s.give_up(); self.state.storage.loaded_review_soa = loaded_reviewer.read_loaded().map(|r| r.soa().clone()); - // TODO: Communicate the new reviewer handle to the zone server. - let old_loaded_reviewer = - std::mem::replace(&mut self.state.storage.loaded_reviewer, loaded_reviewer); - let (s, cleaner) = s.stop_review(old_loaded_reviewer); - transition.move_to(ZoneDataStorage::Cleaning(s)); - self.start_cleanup(cleaner); + transition.move_to(ZoneDataStorage::CleanLoadedPending(s)); + loaded_reviewer } _ => panic!("The zone is not undergoing loader review"), - } + }; + + // Stop serving the abandoned instance. + self.start_rewinding_loaded_review(loaded_reviewer); } } @@ -414,26 +423,24 @@ impl StorageZoneHandle<'_> { )] pub fn abandon_sign(&mut self, builder: SignedZoneBuilder) { // Examine the current state. - let (transition, state) = transition(&mut self.state.storage.machine); - match state { - ZoneDataStorage::Signing(s) => { + let loaded_reviewer = match transition(&mut self.state.storage.machine) { + (transition, ZoneDataStorage::Signing(s)) => { trace!("Abandoning the ongoing sign operation"); let (s, loaded_reviewer) = s.give_up(builder); self.state.storage.loaded_review_soa = loaded_reviewer.read_loaded().map(|r| r.soa().clone()); - // TODO: Communicate the new reviewer handle to the zone server. - let old_loaded_reviewer = - std::mem::replace(&mut self.state.storage.loaded_reviewer, loaded_reviewer); - let (s, cleaner) = s.stop_review(old_loaded_reviewer); - transition.move_to(ZoneDataStorage::Cleaning(s)); - self.start_cleanup(cleaner); + transition.move_to(ZoneDataStorage::CleanLoadedPending(s)); + loaded_reviewer } _ => unreachable!( "'ZoneDataStorage::Signing' is the only state where a 'SignedZoneBuilder' is available" ), - } + }; + + // Stop serving the abandoned instance. + self.start_rewinding_loaded_review(loaded_reviewer); } /// Accept a signed instance of a zone. @@ -467,9 +474,8 @@ impl StorageZoneHandle<'_> { )] pub fn abandon_signed_review(&mut self) { // Examine the current state. - let (transition, state) = transition(&mut self.state.storage.machine); - match state { - ZoneDataStorage::ReviewingSigned(s) => { + let viewers = match transition(&mut self.state.storage.machine) { + (transition, ZoneDataStorage::ReviewingSigned(s)) => { // TODO: Specify the instance ID. info!("The signed instance has been rejected; cleaning it up"); @@ -478,23 +484,48 @@ impl StorageZoneHandle<'_> { loaded_reviewer.read_loaded().map(|r| r.soa().clone()); self.state.storage.signed_review_soa = signed_reviewer.read().map(|r| r.soa().clone()); + transition.move_to(ZoneDataStorage::CleanWholePending(s)); + (loaded_reviewer, signed_reviewer) + } + + _ => panic!("The zone is not undergoing signer review"), + }; + + let span = trace_span!("reset_review_servers"); + let zone = self.zone.clone(); + let center = self.center.clone(); + self.state.storage.background_tasks.spawn(span, async move { + trace!("Resetting the loaded review server"); + let old_loaded_reviewer = + LoadedReviewServer::update_viewer(¢er, &zone, viewers.0).await; - // TODO: Communicate the new reviewer handle to the zone server. - let old_signed_reviewer = - std::mem::replace(&mut self.state.storage.signed_reviewer, signed_reviewer); - let s = s.stop_review(old_signed_reviewer); + trace!("Resetting the signed review server"); + let old_signed_reviewer = + SignedReviewServer::update_viewer(¢er, &zone, viewers.1).await; - // TODO: Communicate the new reviewer handle to the zone server. - let old_loaded_reviewer = - std::mem::replace(&mut self.state.storage.loaded_reviewer, loaded_reviewer); - let (s, cleaner) = s.stop_review(old_loaded_reviewer); + // Examine the current state. + let mut state = zone.state.lock().unwrap(); + let mut handle = ZoneHandle { + zone: &zone, + state: &mut state, + center: ¢er, + }; + let cleaner = match transition(&mut handle.state.storage.machine) { + (transition, ZoneDataStorage::CleanWholePending(s)) => { + let (s, cleaner) = s + .stop_review(old_signed_reviewer) + .stop_review(old_loaded_reviewer); + transition.move_to(ZoneDataStorage::Cleaning(s)); + cleaner + } - transition.move_to(ZoneDataStorage::Cleaning(s)); - self.start_cleanup(cleaner); - } + _ => panic!("The zone was left in 'CleanWholePending' state"), + }; - _ => panic!("The zone is not undergoing signer review"), - } + handle.storage().start_cleanup(cleaner); + + handle.state.storage.background_tasks.finish(); + }); } } @@ -514,31 +545,45 @@ impl StorageZoneHandle<'_> { let zone = self.zone.clone(); let center = self.center.clone(); let span = trace_span!("start_signed_review"); - self.state.storage.background_tasks.spawn_blocking(span, move || { + self.state.storage.background_tasks.spawn(span, async move { + trace!("Converting the signed instance to 'zonetree'"); + // Read the instance. let reader = signed_reviewer .read() .unwrap_or_else(|| unreachable!("The signer never returns an empty instance")); let serial = reader.soa().rdata.serial; - // Build a compatibility shim for the new instance. - let zonetree_zone = Self::build_compat_for_signed(&zone, &reader); + let signed_reviewer = tokio::task::spawn_blocking({ + let zone = zone.clone(); + let center = center.clone(); + move || { + // Read the loaded instance. + let reader = signed_reviewer + .read() + .unwrap_or_else(|| unreachable!("The signer never returns an empty instance")); + + // Build a compatibility shim for the new instance. + let zonetree_zone = Self::build_compat_for_signed(&zone, &reader); + + // Insert the compatibility shim in the global view (possibly + // replacing a previous one). + center.signed_zones.rcu(|tree| { + let mut tree = Arc::unwrap_or_clone(tree.clone()); + let _ = tree.remove_zone(&zone.name, domain::base::iana::Class::IN); + tree.insert_zone(zonetree_zone.clone()).unwrap(); + tree + }); + + signed_reviewer + } + }).await.unwrap(); - // Insert the compatibility shim in the global view (possibly - // replacing a previous one). - center.signed_zones.rcu(|tree| { - let mut tree = Arc::unwrap_or_clone(tree.clone()); - let _ = tree.remove_zone(&zone.name, domain::base::iana::Class::IN); - tree.insert_zone(zonetree_zone.clone()).unwrap(); - tree - }); + trace!("Updating the viewer in 'SignedReviewServer'"); + let old_signed_reviewer = SignedReviewServer::update_viewer(¢er, &zone, signed_reviewer).await; let mut state = zone.state.lock().unwrap(); - // TODO: Pass on the reviewer to the zone server. - let old_signed_reviewer = - std::mem::replace(&mut state.storage.signed_reviewer, signed_reviewer); - // Transition into the reviewing state. match transition(&mut state.storage.machine) { (transition, ZoneDataStorage::ReviewSignedPending(s)) => { @@ -712,39 +757,104 @@ impl StorageZoneHandle<'_> { let zone = self.zone.clone(); let center = self.center.clone(); let span = trace_span!("persist_signed"); - self.state.storage.background_tasks.spawn_blocking(span, move || { + self.state.storage.background_tasks.spawn(span, async move { trace!("Persisting the signed instance"); // Perform the persisting. - let persisted = persister.persist(); + 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()); + viewer + } + + _ => unreachable!( + "'ZoneDataStorage::PersistingSigned' is the only state where a 'SignedZonePersister' is available" + ), + } + }; - // 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. + // 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) { + (transition, ZoneDataStorage::Switching(s)) => { + let (s, cleaner) = s.switch(old_viewer); + transition.move_to(ZoneDataStorage::Cleaning(s)); + cleaner + } + + _ => unreachable!("just transitioned to 'Switching'"), + }; + + let mut handle = ZoneHandle { zone: &zone, state: &mut state, center: ¢er }; + + handle.storage().start_cleanup(cleaner); + + handle.state.storage.background_tasks.finish(); + }); + } + + /// Rewind the loaded review server. + /// + /// When an upcoming loaded instance is under review and is abandoned, the + /// loaded review server must be updated to stop serving it. A background + /// task will be started to achieve this. + /// + /// The loaded reviewer object for the current instance (not the one being + /// abandoned) is received. The old reviewer will be returned to the state + /// machine and the old instance will be cleaned up. + #[tracing::instrument( + level = "trace", + skip_all, + fields(zone = %self.zone.name), + )] + fn start_rewinding_loaded_review(&mut self, loaded_reviewer: LoadedZoneReviewer) { + assert!( + matches!( + self.state.storage.machine, + ZoneDataStorage::CleanLoadedPending(_) + ), + "The zone is not in the 'CleanLoadedPending' state" + ); + + let span = trace_span!("rewind_loaded_review_server"); + let zone = self.zone.clone(); + let center = self.center.clone(); + self.state.storage.background_tasks.spawn(span, async move { + trace!("Rewinding the loaded review server"); + + // Rewind the loaded review server. + let old_loaded_reviewer = + LoadedReviewServer::update_viewer(¢er, &zone, loaded_reviewer).await; + + // Examine the current state. let mut state = zone.state.lock().unwrap(); let mut handle = ZoneHandle { zone: &zone, state: &mut state, center: ¢er, }; - - // Transition the state machine. let cleaner = match transition(&mut handle.state.storage.machine) { - (transition, ZoneDataStorage::PersistingSigned(s)) => { - let (s, viewer) = s.mark_complete(persisted); - handle.state.storage.published_soa = viewer.read().map(|r| r.soa().clone()); - // TODO: Pass on the viewer to the zone server. - let old_viewer = - std::mem::replace(&mut handle.state.storage.viewer, viewer); - let (s, cleaner) = s.switch(old_viewer); + (transition, ZoneDataStorage::CleanLoadedPending(s)) => { + let (s, cleaner) = s.stop_review(old_loaded_reviewer); transition.move_to(ZoneDataStorage::Cleaning(s)); cleaner } - _ => unreachable!( - "'ZoneDataStorage::PersistingSigned' is the only state where a 'SignedZonePersister' is available" - ), + _ => unreachable!("The zone was in the 'CleanLoadedPending' state"), }; + + // Initiate cleanup of the abandoned instance. handle.storage().start_cleanup(cleaner); handle.state.storage.background_tasks.finish(); @@ -786,19 +896,25 @@ pub struct StorageState { machine: ZoneDataStorage, /// The current loaded zone reviewer. + /// + /// This is only used during initialization. // - // TODO: Move into the zone server unit. - loaded_reviewer: LoadedZoneReviewer, + // TODO: Output it directly somehow? + pub loaded_reviewer: Option, /// The current zone reviewer. + /// + /// This is only used during initialization. // - // TODO: Move into the zone server unit. - signed_reviewer: SignedZoneReviewer, + // TODO: Output it directly somehow? + pub signed_reviewer: Option, /// The current zone viewer. + /// + /// This is only used during initialization. // - // TODO: Move into the zone server unit. - viewer: ZoneViewer, + // TODO: Output it directly somehow? + pub viewer: Option, /// The SOA record of the loaded instance of the zone being reviewed, if /// any. @@ -834,9 +950,9 @@ impl StorageState { Self { machine, - loaded_reviewer, - signed_reviewer, - viewer, + loaded_reviewer: Some(loaded_reviewer), + signed_reviewer: Some(signed_reviewer), + viewer: Some(viewer), loaded_review_soa: None, signed_review_soa: None, published_soa: None, From 41262e2901056f824a7281317ce12ee32925fc0b Mon Sep 17 00:00:00 2001 From: arya dradjica Date: Tue, 24 Mar 2026 17:29:25 +0100 Subject: [PATCH 07/10] [zone_server] Serve 'ZoneService' --- src/units/zone_server.rs | 50 ++++------------------------------------ 1 file changed, 4 insertions(+), 46 deletions(-) diff --git a/src/units/zone_server.rs b/src/units/zone_server.rs index 0e4002156..21d5561c9 100644 --- a/src/units/zone_server.rs +++ b/src/units/zone_server.rs @@ -8,7 +8,7 @@ use std::time::Duration; use arc_swap::ArcSwap; use bytes::Bytes; -use domain::base::iana::{Class, Opcode, Rcode}; +use domain::base::iana::{Class, Opcode}; use domain::base::{MessageBuilder, Name, Rtype, Serial, ToName}; use domain::net::client::dgram::Connection; use domain::net::client::protocol::UdpConnect; @@ -24,13 +24,9 @@ use domain::net::server::middleware::notify::{Notifiable, NotifyError, NotifyMid use domain::net::server::middleware::tsig::TsigMiddlewareSvc; use domain::net::server::middleware::xfr::XfrMiddlewareSvc; use domain::net::server::middleware::xfr::{XfrData, XfrDataProvider, XfrDataProviderError}; -use domain::net::server::service::{CallResult, Service, ServiceResult}; +use domain::net::server::service::Service; use domain::net::server::stream::{self, StreamServer}; -use domain::net::server::util::mk_builder_for_target; -use domain::net::server::util::service_fn; -use domain::tsig::Algorithm; -use domain::tsig::KeyStore; -use domain::zonetree::Answer; +use domain::tsig::{Algorithm, KeyStore}; use domain::zonetree::types::EmptyZoneDiff; use domain::zonetree::{StoredName, ZoneTree}; use tracing::{debug, error, info, trace, warn}; @@ -209,9 +205,7 @@ impl ZoneServer { center: center.clone(), }; - let _ = service; - // let svc = service; - let svc = service_fn(zone_server_service, zones.clone()); + let svc = service; let svc = XfrMiddlewareSvc::new(svc, zones.clone(), max_concurrency); let svc = NotifyMiddlewareSvc::new(svc, notifier); let svc = TsigMiddlewareSvc::new(svc, CenterKeyStore(center.clone())); @@ -828,42 +822,6 @@ impl Notifiable for LoaderNotifier { } } -fn zone_server_service( - request: Request, Option>>, - zones: XfrDataProvidingZonesWrapper, -) -> ServiceResult> { - let question = request.message().sole_question().unwrap(); - let zone = zones - .zones - .load() - .find_zone(question.qname(), question.qclass()) - .map(|zone| zone.read()); - let answer = match zone { - Some(zone) => { - let qname = question.qname().to_bytes(); - let qtype = question.qtype(); - let mut answer = zone.query(qname, qtype).unwrap(); - - // https://github.com/NLnetLabs/cascade/issues/435 - // Set the AA flag on responses to workaround the scenario where - // BIND refuses to fetch the zone via XFR, because after receiving - // a NOTIFY from Cascade it issues a SOA query and is not happy - // with the SOA response containing an unset AA flag. This is a - // temporary "fix", strictly speaking this is incorrect as not all - // queries should be responded to with the AA flag set, e.g. we - // cannot respond authoritatively for glue records. - // TODO: Implement a proper fix. - answer.set_authoritative(true); - answer - } - None => Answer::new(Rcode::NXDOMAIN), - }; - - let builder = mk_builder_for_target(); - let additional = answer.to_message(request.message(), builder); - Ok(CallResult::new(additional)) -} - pub fn send_notify_to_addrs( apex_name: StoredName, notify_set: impl Iterator, From 29414c114c9fb1ba7d2476be68e43c613ef24a6a Mon Sep 17 00:00:00 2001 From: arya dradjica Date: Thu, 26 Mar 2026 12:12:27 +0100 Subject: [PATCH 08/10] [zone/service] Implement AXFR --- crates/zonedata/src/reader.rs | 8 +-- src/server/service.rs | 124 +++++++++++++++++++++++++++++++++- src/units/zone_signer.rs | 1 - 3 files changed, 125 insertions(+), 8 deletions(-) diff --git a/crates/zonedata/src/reader.rs b/crates/zonedata/src/reader.rs index 12119774e..416a66dc4 100644 --- a/crates/zonedata/src/reader.rs +++ b/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 + use<'d> { + pub fn all_records(&self) -> impl Iterator + 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 + use<'d> { + pub fn unsigned_records(&self) -> impl Iterator + Send + use<'d> { // Filter out records that would be generated during signing. // // TODO: 'RType::{CDS, CDNSKEY, ZONEMD}'. @@ -177,7 +177,7 @@ 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 + use<'d> { + pub fn loaded_records(&self) -> impl Iterator + Send + use<'d> { LoadedZoneReader::new(self.loaded_instance).unsigned_records() } @@ -185,7 +185,7 @@ impl<'d> SignedZoneReader<'d> { /// /// Records are **unsorted**. The SOA record and records from the loaded /// instance **are** included. - pub fn all_records(&self) -> impl IntoIterator + use<'d> { + pub fn all_records(&self) -> impl Iterator + Send + use<'d> { [self.soa().clone().into()] .into_iter() .chain(self.loaded_records()) diff --git a/src/server/service.rs b/src/server/service.rs index 039bc9410..b6faecf37 100644 --- a/src/server/service.rs +++ b/src/server/service.rs @@ -2,7 +2,9 @@ use std::sync::{Arc, RwLock}; -use cascade_zonedata::{LoadedZoneReviewer, SignedZoneReviewer, SoaRecord, ZoneViewer}; +use cascade_zonedata::{ + LoadedZoneReviewer, RegularRecord, SignedZoneReviewer, SoaRecord, ZoneViewer, +}; use domain::{ new::base::{ name::{RevName, RevNameBuf}, @@ -84,7 +86,7 @@ mod compat { use crate::server::request::{RequestKind, ZoneRequestKind}; - use super::{Viewer, ZoneService}; + use super::{ServedZone, Viewer, ZoneService}; impl Service, Option>> for ZoneService where @@ -135,7 +137,15 @@ mod compat { } }) as Response, - _ => todo!(), + ZoneRequestKind::Axfr => { + Box::pin(axfr(old_request, zone.clone())) as Response + } + + // TODO: Support IXFR. + ZoneRequestKind::Ixfr { .. } => Box::pin(std::future::ready(error( + old_request.message(), + Rcode::NOTIMP, + ))), } } } @@ -159,6 +169,89 @@ mod compat { Box::new(futures::stream::once(std::future::ready(result))) as _ } + async fn axfr( + request: Request, Option>>, + zone: ServedZone, + ) -> ResponseStream { + // Refuse AXFR requests over UDP. + if request.transport_ctx().is_udp() { + return error(request.message(), Rcode::NOTIMP); + } + + // Obtain a read lock to read the zone for an extended duration. + let viewer = zone.viewer.read_owned().await; + + if viewer.is_empty() { + // The zone is known to exist, but we don't have any data for it. + return error(request.message(), Rcode::NOTAUTH); + } + + // NOTE: The following code is a bit tricky. Ideally, we would elide + // the channel and return the `messages` iterator as an async `Stream`; + // but the iterator borrows from `viewer` via `.non_soa_records()`, and + // this prevents the iterator from satisfying `'static`. Rust actually + // _does_ have machinery to work around this, in async functions, so we + // prepare the messages in an async function (as a Tokio task) and send + // them over a channel from there. + // + // In the future, AXFRs could be implemented by spawning an OS thread + // and doing all the work there. This is incompatible with the API of + // `domain::net::server`, as the underlying TCP connection cannot be + // extracted, but we plan to stop using that API anyway. + + let (tx, mut rx) = tokio::sync::mpsc::channel(1024); + + tokio::task::spawn(async move { + // Extract the records to serve. + let soa = viewer.soa().clone(); + let mut records = [soa.clone().into()] + .into_iter() + .chain(viewer.non_soa_records()) + .chain([soa.into()]) + .peekable(); + + // Divide the records into DNS messages. + let mut max_message_size = u16::MAX; // TCP + max_message_size -= request.num_reserved_bytes(); + let messages = std::iter::from_fn(move || { + records.peek()?; + + let mut builder = MessageBuilder::new_stream_vec(); + builder.set_push_limit(max_message_size as usize); + let mut builder = builder + .start_answer(request.message(), Rcode::NOERROR) + .unwrap(); + builder.header_mut().set_aa(true); + + while let Some(record) = records.peek() { + match builder.push(OldRecord::from(record.clone())) { + // On success, consume the record. + Ok(()) => { + let _ = records.next(); + } + + // Once the message runs out of space, stop. + Err(_) => break, + } + } + + let response = builder.additional(); + Some(CallResult::new(response)) + }); + + for message in messages { + if tx.send(message).await.is_err() { + // The channel has closed; stop. + break; + } + } + }); + + let stream = futures::stream::poll_fn(move |cx| rx.poll_recv(cx).map(|m| m.map(Ok))); + + Box::new(stream) as _ + } + fn error(request: &Message>, rcode: Rcode) -> ResponseStream { let response = MessageBuilder::new_stream_vec() .start_error(request, rcode) @@ -180,6 +273,9 @@ trait Viewer { /// Return the SOA record. fn soa(&self) -> &SoaRecord; + + /// Return all records in the zone (excluding SOA). + fn non_soa_records(&self) -> impl Iterator + Send; } impl Viewer for LoadedZoneReviewer { @@ -190,6 +286,14 @@ impl Viewer for LoadedZoneReviewer { fn soa(&self) -> &SoaRecord { self.read_loaded().unwrap().soa() } + + fn non_soa_records(&self) -> impl Iterator + Send { + self.read_loaded() + .unwrap() + .regular_records() + .iter() + .cloned() + } } impl Viewer for SignedZoneReviewer { @@ -200,6 +304,13 @@ impl Viewer for SignedZoneReviewer { fn soa(&self) -> &SoaRecord { self.read().unwrap().soa() } + + fn non_soa_records(&self) -> impl Iterator + Send { + let reader = self.read().unwrap(); + reader + .loaded_records() + .chain(reader.generated_records().iter().cloned()) + } } impl Viewer for ZoneViewer { @@ -210,6 +321,13 @@ impl Viewer for ZoneViewer { fn soa(&self) -> &SoaRecord { self.read().unwrap().soa() } + + fn non_soa_records(&self) -> impl Iterator + Send { + let reader = self.read().unwrap(); + reader + .loaded_records() + .chain(reader.generated_records().iter().cloned()) + } } //----------- ZoneServiceHandle ------------------------------------------------ diff --git a/src/units/zone_signer.rs b/src/units/zone_signer.rs index aca93f40d..3aa87479a 100644 --- a/src/units/zone_signer.rs +++ b/src/units/zone_signer.rs @@ -441,7 +441,6 @@ impl ZoneSigner { // TODO: Filter out DNSSEC records from the loaded instance. let mut records = loaded .unsigned_records() - .into_iter() .map(OldRecord::from) .collect::>(); records.push(new_soa.clone().into()); From 22b9ed7a36fbef9664635457073cb57eb7d7d851 Mon Sep 17 00:00:00 2001 From: arya dradjica Date: Tue, 7 Apr 2026 13:55:07 +0200 Subject: [PATCH 09/10] [zone_server] Stop using 'XfrMiddlewareSvc' --- src/units/zone_server.rs | 93 +--------------------------------------- 1 file changed, 2 insertions(+), 91 deletions(-) diff --git a/src/units/zone_server.rs b/src/units/zone_server.rs index 21d5561c9..2fe7a87ac 100644 --- a/src/units/zone_server.rs +++ b/src/units/zone_server.rs @@ -1,4 +1,4 @@ -use std::future::{Future, ready}; +use std::future::Future; use std::marker::Sync; use std::net::{IpAddr, SocketAddr}; use std::pin::Pin; @@ -6,7 +6,6 @@ use std::process::Stdio; use std::sync::Arc; use std::time::Duration; -use arc_swap::ArcSwap; use bytes::Bytes; use domain::base::iana::{Class, Opcode}; use domain::base::{MessageBuilder, Name, Rtype, Serial, ToName}; @@ -16,19 +15,15 @@ use domain::net::client::request::{RequestMessage, SendRequest}; use domain::net::server::ConnectionConfig; use domain::net::server::buf::VecBufSource; use domain::net::server::dgram::{self, DgramServer}; -use domain::net::server::message::Request; use domain::net::server::middleware::cookies::CookiesMiddlewareSvc; use domain::net::server::middleware::edns::EdnsMiddlewareSvc; use domain::net::server::middleware::mandatory::MandatoryMiddlewareSvc; use domain::net::server::middleware::notify::{Notifiable, NotifyError, NotifyMiddlewareSvc}; use domain::net::server::middleware::tsig::TsigMiddlewareSvc; -use domain::net::server::middleware::xfr::XfrMiddlewareSvc; -use domain::net::server::middleware::xfr::{XfrData, XfrDataProvider, XfrDataProviderError}; use domain::net::server::service::Service; use domain::net::server::stream::{self, StreamServer}; use domain::tsig::{Algorithm, KeyStore}; -use domain::zonetree::types::EmptyZoneDiff; -use domain::zonetree::{StoredName, ZoneTree}; +use domain::zonetree::StoredName; use tracing::{debug, error, info, trace, warn}; use crate::api::{ @@ -178,27 +173,6 @@ impl ZoneServer { // TODO: metrics and status reporting - // TODO: This will just choose all current zones to be served. For signed and published - // zones this doesn't matter so much as they only exist while being and once approved. - // But for unsigned zones the zone could be updated whilst being reviewed and we only - // serve the latest version of the zone, not the specific serial being reviewed! - let zones = match source { - Source::Unsigned => center.unsigned_zones.clone(), - Source::Signed => center.signed_zones.clone(), - Source::Published => center.published_zones.clone(), - }; - - let max_concurrency = std::thread::available_parallelism() - .unwrap() - .get() - .div_ceil(2); - - // TODO: Pass xfr_out to XfrDataProvidingZonesWrapper for enforcement. - let zones = XfrDataProvidingZonesWrapper { - zones, - center: center.clone(), - }; - // Propagate NOTIFY messages if this is the publication server. let notifier = LoaderNotifier { enabled: matches!(source, Source::Published), @@ -206,7 +180,6 @@ impl ZoneServer { }; let svc = service; - let svc = XfrMiddlewareSvc::new(svc, zones.clone(), max_concurrency); let svc = NotifyMiddlewareSvc::new(svc, notifier); let svc = TsigMiddlewareSvc::new(svc, CenterKeyStore(center.clone())); let svc = CookiesMiddlewareSvc::with_random_secret(svc); @@ -722,68 +695,6 @@ impl KeyStore for CenterKeyStore { } } -//----------- XfrDataProvidingZonesWrapper ----------------------------------- - -#[derive(Clone)] -struct XfrDataProvidingZonesWrapper { - zones: Arc>, - - /// Access to Center for TSIG key lookup. - center: Arc
, -} - -impl XfrDataProvider>> for XfrDataProvidingZonesWrapper { - type Diff = EmptyZoneDiff; - - fn request( - &self, - req: &Request>>, - _diff_from: Option, - ) -> Pin< - Box< - dyn Future< - Output = Result< - domain::net::server::middleware::xfr::XfrData, - domain::net::server::middleware::xfr::XfrDataProviderError, - >, - > + Sync - + Send - + '_, - >, - > - where - Octs: octseq::Octets + Send + Sync, - { - let res = req - .message() - .sole_question() - .map_err(XfrDataProviderError::ParseError) - .and_then(|q| { - if let Some(zone) = self.zones.load().find_zone(q.qname(), q.qclass()) { - Ok(XfrData::new(zone.clone(), vec![], false)) - } else { - Err(XfrDataProviderError::UnknownZone) - } - }); - - Box::pin(ready(res)) - } -} - -impl KeyStore for XfrDataProvidingZonesWrapper { - type Key = Arc; - - fn get_key(&self, name: &N, algorithm: Algorithm) -> Option> { - let tsig_store = &self.center.state.lock().unwrap().tsig_store; - let key_name: domain::tsig::KeyName = name.try_to_name().ok()?; - tsig_store - .map - .get(&key_name) - .map(|k| k.inner.clone()) - .filter(|k| k.algorithm() == algorithm) - } -} - //----------- LoaderNotifier --------------------------------------------------- /// A forwarder of NOTIFY messages to the zone loader. From 176097facf30064f791be31c816e15d6cfcc8441 Mon Sep 17 00:00:00 2001 From: arya dradjica Date: Thu, 9 Apr 2026 09:26:26 +0200 Subject: [PATCH 10/10] [zone/storage] Move signed review methods --- src/zone/storage.rs | 171 ++++++++++++++++++++++---------------------- 1 file changed, 85 insertions(+), 86 deletions(-) diff --git a/src/zone/storage.rs b/src/zone/storage.rs index 83e70f021..9443e9954 100644 --- a/src/zone/storage.rs +++ b/src/zone/storage.rs @@ -442,91 +442,6 @@ impl StorageZoneHandle<'_> { // Stop serving the abandoned instance. self.start_rewinding_loaded_review(loaded_reviewer); } - - /// Accept a signed instance of a zone. - #[tracing::instrument( - level = "trace", - skip_all, - fields(zone = %self.zone.name), - )] - pub fn accept_signed(&mut self) { - // 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); - } - - _ => panic!("The zone is not undergoing signer review"), - } - } - - /// Give up on a signed instance undergoing review. - #[tracing::instrument( - level = "trace", - skip_all, - fields(zone = %self.zone.name), - )] - pub fn abandon_signed_review(&mut self) { - // Examine the current state. - let viewers = match transition(&mut self.state.storage.machine) { - (transition, ZoneDataStorage::ReviewingSigned(s)) => { - // TODO: Specify the instance ID. - info!("The signed instance has been rejected; cleaning it up"); - - let (s, loaded_reviewer, signed_reviewer) = s.give_up(); - self.state.storage.loaded_review_soa = - loaded_reviewer.read_loaded().map(|r| r.soa().clone()); - self.state.storage.signed_review_soa = - signed_reviewer.read().map(|r| r.soa().clone()); - transition.move_to(ZoneDataStorage::CleanWholePending(s)); - (loaded_reviewer, signed_reviewer) - } - - _ => panic!("The zone is not undergoing signer review"), - }; - - let span = trace_span!("reset_review_servers"); - let zone = self.zone.clone(); - let center = self.center.clone(); - self.state.storage.background_tasks.spawn(span, async move { - trace!("Resetting the loaded review server"); - let old_loaded_reviewer = - LoadedReviewServer::update_viewer(¢er, &zone, viewers.0).await; - - trace!("Resetting the signed review server"); - let old_signed_reviewer = - SignedReviewServer::update_viewer(¢er, &zone, viewers.1).await; - - // Examine the current state. - let mut state = zone.state.lock().unwrap(); - let mut handle = ZoneHandle { - zone: &zone, - state: &mut state, - center: ¢er, - }; - let cleaner = match transition(&mut handle.state.storage.machine) { - (transition, ZoneDataStorage::CleanWholePending(s)) => { - let (s, cleaner) = s - .stop_review(old_signed_reviewer) - .stop_review(old_loaded_reviewer); - transition.move_to(ZoneDataStorage::Cleaning(s)); - cleaner - } - - _ => panic!("The zone was left in 'CleanWholePending' state"), - }; - - handle.storage().start_cleanup(cleaner); - - handle.state.storage.background_tasks.finish(); - }); - } } /// # Signer Review Operations @@ -643,7 +558,91 @@ impl StorageZoneHandle<'_> { zone } - // TODO: approve_signed() + /// Accept a signed instance of a zone. + #[tracing::instrument( + level = "trace", + skip_all, + fields(zone = %self.zone.name), + )] + pub fn accept_signed(&mut self) { + // 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); + } + + _ => panic!("The zone is not undergoing signer review"), + } + } + + /// Give up on a signed instance undergoing review. + #[tracing::instrument( + level = "trace", + skip_all, + fields(zone = %self.zone.name), + )] + pub fn abandon_signed_review(&mut self) { + // Examine the current state. + let (loaded_reviewer, signed_reviewer); + match transition(&mut self.state.storage.machine) { + (transition, ZoneDataStorage::ReviewingSigned(s)) => { + // TODO: Specify the instance ID. + info!("The signed instance has been rejected; cleaning it up"); + + let new_s; + (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()); + self.state.storage.signed_review_soa = + signed_reviewer.read().map(|r| r.soa().clone()); + } + + _ => panic!("The zone is not undergoing signer review"), + }; + + let span = trace_span!("reset_review_servers"); + let zone = self.zone.clone(); + let center = self.center.clone(); + self.state.storage.background_tasks.spawn(span, async move { + trace!("Resetting the signed review server"); + let old_signed_reviewer = + SignedReviewServer::update_viewer(¢er, &zone, signed_reviewer).await; + + trace!("Resetting the loaded review server"); + let old_loaded_reviewer = + LoadedReviewServer::update_viewer(¢er, &zone, loaded_reviewer).await; + + // Examine the current state. + let mut state = zone.state.lock().unwrap(); + let mut handle = ZoneHandle { + zone: &zone, + state: &mut state, + center: ¢er, + }; + let cleaner = match transition(&mut handle.state.storage.machine) { + (transition, ZoneDataStorage::CleanWholePending(s)) => { + let (s, cleaner) = s + .stop_review(old_signed_reviewer) + .stop_review(old_loaded_reviewer); + transition.move_to(ZoneDataStorage::Cleaning(s)); + cleaner + } + + _ => unreachable!("The zone was left in 'CleanWholePending' state"), + }; + + handle.storage().start_cleanup(cleaner); + + handle.state.storage.background_tasks.finish(); + }); + } } /// # Background Tasks