1use std::collections::BTreeMap;
2use std::sync::Arc;
3use std::time::Duration;
4
5use chrono::{DateTime, Utc};
6use serde::{Deserialize, Serialize};
7use tokio::sync::watch;
8use tracing::{debug, instrument};
9
10use super::records::{DnsRecord, RecordType};
11use super::resolver::DnsResolver;
12use crate::error::{Result, SeerError};
13
14pub const MAX_FOLLOW_ITERATIONS: usize = 10_000;
18
19pub const MAX_FOLLOW_INTERVAL_SECS: u64 = 3600;
24
25#[derive(Debug, Clone)]
27pub struct FollowConfig {
28 pub iterations: usize,
30 pub interval_secs: u64,
32 pub changes_only: bool,
34}
35
36impl Default for FollowConfig {
37 fn default() -> Self {
38 Self {
39 iterations: 10,
40 interval_secs: 60,
41 changes_only: false,
42 }
43 }
44}
45
46impl FollowConfig {
47 pub fn new(iterations: usize, interval_minutes: f64) -> Result<Self> {
55 if iterations == 0 {
56 return Err(SeerError::InvalidInput(
57 "iterations must be at least 1".into(),
58 ));
59 }
60 if iterations > MAX_FOLLOW_ITERATIONS {
61 return Err(SeerError::InvalidInput(format!(
62 "iterations must be at most {MAX_FOLLOW_ITERATIONS}"
63 )));
64 }
65 if !interval_minutes.is_finite() {
66 return Err(SeerError::InvalidInput(
67 "interval_minutes must be a finite number".into(),
68 ));
69 }
70 if interval_minutes < 0.0 {
71 return Err(SeerError::InvalidInput(
72 "interval_minutes must be non-negative".into(),
73 ));
74 }
75 if interval_minutes > MAX_FOLLOW_INTERVAL_SECS as f64 / 60.0 {
76 return Err(SeerError::InvalidInput(
77 "interval_minutes must be at most 60".into(),
78 ));
79 }
80 let mut interval_secs = (interval_minutes * 60.0).round() as u64;
90 if iterations > 1 {
91 interval_secs = interval_secs.max(1);
92 }
93 Ok(Self {
94 iterations,
95 interval_secs,
96 changes_only: false,
97 })
98 }
99
100 pub fn with_changes_only(mut self, changes_only: bool) -> Self {
101 self.changes_only = changes_only;
102 self
103 }
104
105 fn validate(&self) -> Result<()> {
111 if self.iterations == 0 || self.iterations > MAX_FOLLOW_ITERATIONS {
112 return Err(SeerError::InvalidInput(format!(
113 "iterations must be between 1 and {MAX_FOLLOW_ITERATIONS}"
114 )));
115 }
116 if self.interval_secs > MAX_FOLLOW_INTERVAL_SECS {
117 return Err(SeerError::InvalidInput(format!(
118 "interval must be at most {MAX_FOLLOW_INTERVAL_SECS} seconds"
119 )));
120 }
121 Ok(())
122 }
123}
124
125#[derive(Debug, Clone, Serialize, Deserialize)]
127pub struct FollowIteration {
128 pub iteration: usize,
130 pub total_iterations: usize,
132 pub timestamp: DateTime<Utc>,
134 pub records: Vec<DnsRecord>,
136 pub changed: bool,
138 pub added: Vec<String>,
140 pub removed: Vec<String>,
142 pub error: Option<String>,
144}
145
146impl FollowIteration {
147 pub fn success(&self) -> bool {
148 self.error.is_none()
149 }
150
151 pub fn record_count(&self) -> usize {
152 self.records.len()
153 }
154}
155
156#[derive(Debug, Clone, Serialize, Deserialize)]
158pub struct FollowResult {
159 pub domain: String,
161 pub record_type: RecordType,
163 pub nameserver: Option<String>,
165 pub iterations_requested: usize,
167 pub interval_secs: u64,
168 pub iterations: Vec<FollowIteration>,
170 pub interrupted: bool,
172 pub total_changes: usize,
174 pub started_at: DateTime<Utc>,
176 pub ended_at: DateTime<Utc>,
178}
179
180impl FollowResult {
181 pub fn completed_iterations(&self) -> usize {
182 self.iterations.len()
183 }
184
185 pub fn successful_iterations(&self) -> usize {
186 self.iterations.iter().filter(|i| i.success()).count()
187 }
188
189 pub fn failed_iterations(&self) -> usize {
190 self.iterations.iter().filter(|i| !i.success()).count()
191 }
192}
193
194pub type FollowProgressCallback = Arc<dyn Fn(&FollowIteration) + Send + Sync>;
196
197#[derive(Clone)]
199pub struct DnsFollower {
200 resolver: DnsResolver,
201}
202
203impl Default for DnsFollower {
204 fn default() -> Self {
205 Self::new()
206 }
207}
208
209impl DnsFollower {
210 pub fn new() -> Self {
211 Self {
212 resolver: DnsResolver::new(),
213 }
214 }
215
216 pub fn with_resolver(resolver: DnsResolver) -> Self {
217 Self { resolver }
218 }
219
220 #[instrument(skip(self, config, callback, cancel_rx))]
222 pub async fn follow(
223 &self,
224 domain: &str,
225 record_type: RecordType,
226 nameserver: Option<&str>,
227 config: FollowConfig,
228 callback: Option<FollowProgressCallback>,
229 cancel_rx: Option<watch::Receiver<bool>>,
230 ) -> Result<FollowResult> {
231 config.validate()?;
232 let domain = crate::dns::resolver::prepare_query(domain, record_type)?;
235 let started_at = Utc::now();
236 let mut iterations: Vec<FollowIteration> = Vec::with_capacity(config.iterations);
238 let mut baseline: Option<Observation> = None;
243 let mut total_changes = 0;
244 let mut interrupted = false;
245
246 debug!(
247 domain = %domain,
248 record_type = %record_type,
249 iterations = config.iterations,
250 interval_secs = config.interval_secs,
251 "Starting DNS follow"
252 );
253
254 for i in 0..config.iterations {
255 if let Some(ref rx) = cancel_rx {
257 if *rx.borrow() {
258 debug!("Follow operation cancelled");
259 interrupted = true;
260 break;
261 }
262 }
263
264 let timestamp = Utc::now();
265 let iteration_num = i + 1;
266
267 let (records, error) = match self
269 .resolver
270 .resolve(&domain, record_type, nameserver)
271 .await
272 {
273 Ok(records) => (records, None),
274 Err(e) => {
275 debug!(domain = %domain, error = %e, "DNS follow query failed");
276 (Vec::new(), Some(e.sanitized_message()))
278 }
279 };
280
281 let current = observe(&records);
282
283 let (changed, added, removed) = match (&baseline, &error) {
289 (Some(previous), None) => diff_observations(¤t, previous),
290 _ => (false, Vec::new(), Vec::new()),
296 };
297
298 if changed {
299 total_changes += 1;
300 }
301
302 let error_is_none = error.is_none();
304
305 let iteration = FollowIteration {
306 iteration: iteration_num,
307 total_iterations: config.iterations,
308 timestamp,
309 records,
310 changed,
311 added,
312 removed,
313 error,
314 };
315
316 if let Some(ref cb) = callback {
318 if !config.changes_only || iteration_num == 1 || changed {
320 cb(&iteration);
321 }
322 }
323
324 iterations.push(iteration);
325 if error_is_none {
330 baseline = Some(current);
331 }
332
333 if i < config.iterations - 1 {
335 let sleep_duration = Duration::from_secs(config.interval_secs);
336
337 if let Some(ref rx) = cancel_rx {
339 let mut rx_clone = rx.clone();
340 tokio::select! {
341 _ = tokio::time::sleep(sleep_duration) => {}
342 _ = rx_clone.changed() => {
343 if *rx_clone.borrow() {
344 debug!("Follow operation cancelled during sleep");
345 interrupted = true;
346 break;
347 }
348 }
349 }
350 } else {
351 tokio::time::sleep(sleep_duration).await;
352 }
353 }
354 }
355
356 let ended_at = Utc::now();
357
358 Ok(FollowResult {
359 domain: domain.to_string(),
360 record_type,
361 nameserver: nameserver.map(|s| s.to_string()),
362 iterations_requested: config.iterations,
363 interval_secs: config.interval_secs,
364 iterations,
365 interrupted,
366 total_changes,
367 started_at,
368 ended_at,
369 })
370 }
371
372 #[instrument(skip(self, config), fields(domain = %domain, record_type = ?record_type))]
374 pub async fn follow_simple(
375 &self,
376 domain: &str,
377 record_type: RecordType,
378 nameserver: Option<&str>,
379 config: FollowConfig,
380 ) -> Result<FollowResult> {
381 self.follow(domain, record_type, nameserver, config, None, None)
382 .await
383 }
384}
385
386type Observation = BTreeMap<String, String>;
390
391fn observe(records: &[DnsRecord]) -> Observation {
392 records
393 .iter()
394 .map(|r| (r.data.comparison_key(), r.data.to_string()))
395 .collect()
396}
397
398fn diff_observations(
404 current: &Observation,
405 previous: &Observation,
406) -> (bool, Vec<String>, Vec<String>) {
407 let mut added: Vec<String> = current
408 .iter()
409 .filter(|(key, _)| !previous.contains_key(*key))
410 .map(|(_, value)| value.clone())
411 .collect();
412 let mut removed: Vec<String> = previous
413 .iter()
414 .filter(|(key, _)| !current.contains_key(*key))
415 .map(|(_, value)| value.clone())
416 .collect();
417 added.sort();
418 added.dedup();
419 removed.sort();
420 removed.dedup();
421
422 let changed = !added.is_empty() || !removed.is_empty();
423 (changed, added, removed)
424}
425
426#[cfg(test)]
427mod tests {
428 use super::*;
429
430 use super::super::records::RecordData;
431
432 fn record(data: RecordData) -> DnsRecord {
433 DnsRecord {
434 name: "example.com".to_string(),
435 record_type: RecordType::NS,
436 ttl: 300,
437 data,
438 }
439 }
440
441 fn ns_observation<const N: usize>(names: [&str; N]) -> Observation {
443 let records: Vec<DnsRecord> = names
444 .iter()
445 .map(|n| {
446 record(RecordData::NS {
447 nameserver: n.to_string(),
448 })
449 })
450 .collect();
451 observe(&records)
452 }
453
454 #[tokio::test]
455 async fn test_follow_config_default() {
456 let config = FollowConfig::default();
457 assert_eq!(config.iterations, 10);
458 assert_eq!(config.interval_secs, 60);
459 assert!(!config.changes_only);
460 }
461
462 #[test]
466 fn diff_values_ignores_case_only_differences() {
467 let previous = ns_observation(["NS1.EXAMPLE.COM.", "ns2.example.com."]);
468 let current = ns_observation(["ns1.example.com.", "NS2.EXAMPLE.COM."]);
469 let (changed, added, removed) = diff_observations(¤t, &previous);
470 assert!(!changed, "case-only differences must not count as a change");
471 assert!(added.is_empty(), "no added values: {added:?}");
472 assert!(removed.is_empty(), "no removed values: {removed:?}");
473 }
474
475 #[test]
479 fn diff_values_detects_real_change_preserving_case() {
480 let previous = ns_observation(["NS1.Example.com."]);
481 let current = ns_observation(["NS2.Example.Com."]);
482 let (changed, added, removed) = diff_observations(¤t, &previous);
483 assert!(changed, "a different nameserver is a real change");
484 assert_eq!(added, vec!["NS2.Example.Com.".to_string()]);
485 assert_eq!(removed, vec!["NS1.Example.com.".to_string()]);
486 }
487
488 #[test]
491 fn diff_values_reports_case_change_in_txt() {
492 let txt = |t: &str| {
493 observe(&[record(RecordData::TXT {
494 text: t.to_string(),
495 })])
496 };
497 let (changed, added, removed) = diff_observations(&txt("token=abc"), &txt("token=AbC"));
498 assert!(changed);
499 assert_eq!(added, vec!["\"token=abc\"".to_string()]);
500 assert_eq!(removed, vec!["\"token=AbC\"".to_string()]);
501 }
502
503 #[test]
504 fn follow_config_rounds_interval_to_nearest_second() {
505 assert_eq!(FollowConfig::new(2, 2.05).unwrap().interval_secs, 123);
507 assert_eq!(FollowConfig::new(2, 0.5).unwrap().interval_secs, 30);
508 }
509
510 #[tokio::test]
514 async fn follow_rejects_out_of_range_literal_config() {
515 let follower = DnsFollower::new();
516 for config in [
517 FollowConfig {
518 iterations: usize::MAX,
519 interval_secs: 1,
520 changes_only: false,
521 },
522 FollowConfig {
523 iterations: 0,
524 interval_secs: 1,
525 changes_only: false,
526 },
527 FollowConfig {
528 iterations: 2,
529 interval_secs: MAX_FOLLOW_INTERVAL_SECS + 1,
530 changes_only: false,
531 },
532 ] {
533 let err = follower
534 .follow_simple("example.com", RecordType::A, None, config)
535 .await
536 .expect_err("out-of-range config must be rejected before any query");
537 assert!(matches!(err, SeerError::InvalidInput(_)), "{err:?}");
538 }
539 }
540
541 #[tokio::test]
546 async fn follow_first_iteration_error_is_not_a_baseline() {
547 use std::sync::atomic::{AtomicBool, Ordering};
548
549 use hickory_resolver::proto::rr::rdata as wire;
550 use hickory_resolver::proto::rr::{RData as HickoryRData, RecordType as WireType};
551
552 use crate::dns::test_support::{mock_dns_resolver, spawn_mock_dns_fn, MockReply};
553
554 let healthy = Arc::new(AtomicBool::new(false));
557 let port = spawn_mock_dns_fn({
558 let healthy = Arc::clone(&healthy);
559 move |_, qtype| {
560 if healthy.load(Ordering::SeqCst) && qtype == WireType::A {
561 MockReply::Answer(vec![HickoryRData::A(wire::A(std::net::Ipv4Addr::new(
562 192, 0, 2, 7,
563 )))])
564 } else {
565 MockReply::ServFail
566 }
567 }
568 })
569 .await;
570 let callback: FollowProgressCallback = {
571 let healthy = Arc::clone(&healthy);
572 Arc::new(move |_: &FollowIteration| healthy.store(true, Ordering::SeqCst))
573 };
574
575 let config = FollowConfig {
576 iterations: 3,
577 interval_secs: 0,
578 changes_only: false,
579 };
580 let result = DnsFollower::with_resolver(mock_dns_resolver(port))
581 .follow(
582 "seer.test",
583 RecordType::A,
584 Some("127.0.0.1"),
585 config,
586 Some(callback),
587 None,
588 )
589 .await
590 .expect("follow against the mock fixture");
591
592 assert!(!result.iterations[0].success(), "iteration 1 must error");
593 let second = &result.iterations[1];
594 assert!(second.success(), "{second:?}");
595 assert!(
596 !second.changed && second.added.is_empty() && second.removed.is_empty(),
597 "the first success establishes the baseline, it is not a change: {second:?}"
598 );
599 assert!(!result.iterations[2].changed);
600 assert_eq!(result.total_changes, 0);
601 }
602
603 #[tokio::test]
605 async fn follow_uses_per_name_normalization() {
606 use crate::dns::test_support::{mock_dns_resolver, spawn_mock_dns, MockMode};
607
608 let port = spawn_mock_dns(MockMode::Zone).await;
609 let follower = DnsFollower::with_resolver(mock_dns_resolver(port));
610 let one_shot = || FollowConfig::new(1, 0.0).expect("valid config");
611
612 let result = follower
613 .follow_simple(
614 "www.seer.test",
615 RecordType::CNAME,
616 Some("127.0.0.1"),
617 one_shot(),
618 )
619 .await
620 .expect("follow www");
621 assert_eq!(result.domain, "www.seer.test");
622 assert_eq!(result.iterations[0].record_count(), 1);
623
624 let result = follower
625 .follow_simple(
626 "2606:4700:4700::1111",
627 RecordType::PTR,
628 Some("127.0.0.1"),
629 one_shot(),
630 )
631 .await
632 .expect("IPv6 PTR literal must be accepted");
633 assert_eq!(result.domain, "2606:4700:4700::1111");
634 assert_eq!(result.iterations[0].record_count(), 1);
635 }
636
637 #[test]
638 fn follow_config_rejects_unbounded_iterations() {
639 assert!(FollowConfig::new(MAX_FOLLOW_ITERATIONS, 1.0).is_ok());
640 let err = FollowConfig::new(MAX_FOLLOW_ITERATIONS + 1, 1.0).unwrap_err();
641 assert!(matches!(err, SeerError::InvalidInput(_)));
642 assert!(FollowConfig::new(usize::MAX, 1.0).is_err());
643 }
644
645 #[tokio::test]
646 async fn test_follow_config_new() {
647 let config = FollowConfig::new(5, 0.5).unwrap();
648 assert_eq!(config.iterations, 5);
649 assert_eq!(config.interval_secs, 30);
650 }
651
652 #[tokio::test]
658 async fn follow_loop_detects_change_against_mock_dns() {
659 use std::sync::Mutex;
660
661 use hickory_resolver::proto::rr::rdata as wire;
662 use hickory_resolver::proto::rr::RData as HickoryRData;
663
664 use crate::dns::test_support::{mock_dns_resolver, spawn_mock_dns_sequence};
665
666 let a = |last_octet: u8| {
667 HickoryRData::A(wire::A(std::net::Ipv4Addr::new(192, 0, 2, last_octet)))
668 };
669 let port = spawn_mock_dns_sequence(vec![vec![a(1)], vec![a(2)], vec![a(2)]]).await;
672
673 let config = FollowConfig {
677 iterations: 3,
678 interval_secs: 0,
679 changes_only: false,
680 };
681
682 let seen: Arc<Mutex<Vec<(usize, bool)>>> = Arc::new(Mutex::new(Vec::new()));
683 let callback: FollowProgressCallback = {
684 let seen = Arc::clone(&seen);
685 Arc::new(move |it: &FollowIteration| {
686 seen.lock()
687 .expect("callback mutex")
688 .push((it.iteration, it.changed));
689 })
690 };
691
692 let follower = DnsFollower::with_resolver(mock_dns_resolver(port));
693 let result = follower
694 .follow(
695 "seer.test",
696 RecordType::A,
697 Some("127.0.0.1"),
698 config,
699 Some(callback),
700 None,
701 )
702 .await
703 .expect("follow against the mock fixture must succeed");
704
705 assert_eq!(result.completed_iterations(), 3);
707 assert_eq!(result.successful_iterations(), 3);
708 assert!(!result.interrupted);
709 assert_eq!(result.domain, "seer.test");
710 assert_eq!(result.record_type, RecordType::A);
711
712 let first = &result.iterations[0];
714 assert_eq!(first.record_count(), 1);
715 assert_eq!(first.records[0].data.to_string(), "192.0.2.1");
716 assert!(!first.changed, "first iteration is never a change");
717 assert!(first.added.is_empty() && first.removed.is_empty());
718
719 let second = &result.iterations[1];
721 assert!(second.changed, "record change must be detected");
722 assert_eq!(second.added, vec!["192.0.2.2".to_string()]);
723 assert_eq!(second.removed, vec!["192.0.2.1".to_string()]);
724
725 let third = &result.iterations[2];
727 assert!(!third.changed, "unchanged records must not be a change");
728 assert!(third.added.is_empty() && third.removed.is_empty());
729
730 assert_eq!(result.total_changes, 1);
731
732 let seen = seen.lock().expect("callback mutex");
734 assert_eq!(*seen, vec![(1, false), (2, true), (3, false)]);
735 }
736
737 #[tokio::test]
738 #[ignore = "live network; run with --ignored or SEER_LIVE_TESTS=1"]
739 async fn test_follow_single_iteration() {
740 let follower = DnsFollower::new();
741 let config = FollowConfig::new(1, 0.0).unwrap();
742
743 let result = follower
744 .follow_simple("example.com", RecordType::A, None, config)
745 .await;
746
747 assert!(result.is_ok());
748 let result = result.unwrap();
749 assert_eq!(result.completed_iterations(), 1);
750 assert!(!result.interrupted);
751 }
752
753 #[test]
754 fn follow_config_rejects_zero_iterations() {
755 assert!(FollowConfig::new(0, 1.0).is_err());
756 }
757
758 #[test]
759 fn follow_config_rejects_infinite_interval() {
760 assert!(FollowConfig::new(10, f64::INFINITY).is_err());
761 assert!(FollowConfig::new(10, f64::NEG_INFINITY).is_err());
762 }
763
764 #[test]
765 fn follow_config_rejects_nan_interval() {
766 assert!(FollowConfig::new(10, f64::NAN).is_err());
767 }
768
769 #[test]
770 fn follow_config_rejects_negative_interval() {
771 assert!(FollowConfig::new(10, -1.0).is_err());
772 }
773
774 #[test]
775 fn follow_config_rejects_interval_above_cap() {
776 assert!(FollowConfig::new(10, 60.1).is_err());
777 }
778
779 #[test]
780 fn follow_config_accepts_valid() {
781 assert!(FollowConfig::new(10, 1.5).is_ok());
782 assert!(FollowConfig::new(1, 0.0).is_ok());
783 assert!(FollowConfig::new(1, 60.0).is_ok());
784 }
785
786 #[test]
787 fn follow_config_floors_subsecond_interval_for_multi_iteration() {
788 let config = FollowConfig::new(10_000, 0.001).unwrap();
792 assert!(
793 config.interval_secs >= 1,
794 "multi-iteration interval must be floored to >= 1s, got {}",
795 config.interval_secs
796 );
797 }
798
799 #[test]
800 fn follow_config_allows_zero_interval_for_single_iteration() {
801 let config = FollowConfig::new(1, 0.0).unwrap();
804 assert_eq!(config.interval_secs, 0);
805 }
806
807 #[tokio::test]
808 #[ignore = "live network; run with --ignored or SEER_LIVE_TESTS=1"]
809 async fn follow_honors_cancel() {
810 use tokio::sync::watch;
811
812 let (tx, rx) = watch::channel(false);
813 let config = FollowConfig::new(100, 0.5).unwrap();
815 let follower = DnsFollower::new();
816
817 let handle = tokio::spawn(async move {
818 follower
819 .follow("example.com", RecordType::A, None, config, None, Some(rx))
820 .await
821 });
822
823 tokio::time::sleep(Duration::from_millis(200)).await;
825 tx.send(true).unwrap();
826
827 let joined = tokio::time::timeout(Duration::from_secs(10), handle)
828 .await
829 .expect("follow should return promptly after cancel");
830 let result = joined.expect("join").expect("follow result");
831 assert!(result.interrupted, "follow should be interrupted");
832 assert!(
833 result.completed_iterations() < 100,
834 "should not complete all iterations"
835 );
836 }
837}