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
186pub type FollowProgressCallback = Arc<dyn Fn(&FollowIteration) + Send + Sync>;
188
189#[derive(Clone)]
191pub struct DnsFollower {
192 resolver: DnsResolver,
193}
194
195impl Default for DnsFollower {
196 fn default() -> Self {
197 Self::new()
198 }
199}
200
201impl DnsFollower {
202 pub fn new() -> Self {
203 Self {
204 resolver: DnsResolver::new(),
205 }
206 }
207
208 pub fn with_resolver(resolver: DnsResolver) -> Self {
209 Self { resolver }
210 }
211
212 pub fn from_config(config: &crate::config::SeerConfig) -> Self {
216 Self::with_resolver(DnsResolver::from_config(config))
217 }
218
219 #[instrument(skip(self, config, callback, cancel_rx))]
221 pub async fn follow(
222 &self,
223 domain: &str,
224 record_type: RecordType,
225 nameserver: Option<&str>,
226 config: FollowConfig,
227 callback: Option<FollowProgressCallback>,
228 cancel_rx: Option<watch::Receiver<bool>>,
229 ) -> Result<FollowResult> {
230 config.validate()?;
231 let domain = crate::dns::resolver::prepare_query(domain, record_type)?;
234 let started_at = Utc::now();
235 let mut iterations: Vec<FollowIteration> = Vec::with_capacity(config.iterations);
237 let mut baseline: Option<Observation> = None;
242 let mut total_changes = 0;
243 let mut interrupted = false;
244
245 debug!(
246 domain = %domain,
247 record_type = %record_type,
248 iterations = config.iterations,
249 interval_secs = config.interval_secs,
250 "Starting DNS follow"
251 );
252
253 for i in 0..config.iterations {
254 if let Some(ref rx) = cancel_rx {
256 if *rx.borrow() {
257 debug!("Follow operation cancelled");
258 interrupted = true;
259 break;
260 }
261 }
262
263 let timestamp = Utc::now();
264 let iteration_num = i + 1;
265
266 let (records, error) = match self
268 .resolver
269 .resolve(&domain, record_type, nameserver)
270 .await
271 {
272 Ok(records) => (records, None),
273 Err(e) => {
274 debug!(domain = %domain, error = %e, "DNS follow query failed");
275 (Vec::new(), Some(e.sanitized_message()))
277 }
278 };
279
280 let current = observe(&records);
281
282 let (changed, added, removed) = match (&baseline, &error) {
288 (Some(previous), None) => diff_observations(¤t, previous),
289 _ => (false, Vec::new(), Vec::new()),
295 };
296
297 if changed {
298 total_changes += 1;
299 }
300
301 let error_is_none = error.is_none();
303
304 let iteration = FollowIteration {
305 iteration: iteration_num,
306 total_iterations: config.iterations,
307 timestamp,
308 records,
309 changed,
310 added,
311 removed,
312 error,
313 };
314
315 if let Some(ref cb) = callback {
317 if !config.changes_only || iteration_num == 1 || changed {
319 cb(&iteration);
320 }
321 }
322
323 iterations.push(iteration);
324 if error_is_none {
329 baseline = Some(current);
330 }
331
332 if i < config.iterations - 1 {
334 let sleep_duration = Duration::from_secs(config.interval_secs);
335
336 if let Some(ref rx) = cancel_rx {
338 let mut rx_clone = rx.clone();
339 tokio::select! {
340 _ = tokio::time::sleep(sleep_duration) => {}
341 _ = rx_clone.changed() => {
342 if *rx_clone.borrow() {
343 debug!("Follow operation cancelled during sleep");
344 interrupted = true;
345 break;
346 }
347 }
348 }
349 } else {
350 tokio::time::sleep(sleep_duration).await;
351 }
352 }
353 }
354
355 let ended_at = Utc::now();
356
357 Ok(FollowResult {
358 domain: domain.to_string(),
359 record_type,
360 nameserver: nameserver.map(|s| s.to_string()),
361 iterations_requested: config.iterations,
362 interval_secs: config.interval_secs,
363 iterations,
364 interrupted,
365 total_changes,
366 started_at,
367 ended_at,
368 })
369 }
370}
371
372type Observation = BTreeMap<String, String>;
376
377fn observe(records: &[DnsRecord]) -> Observation {
378 records
379 .iter()
380 .map(|r| (r.data.comparison_key(), r.data.to_string()))
381 .collect()
382}
383
384fn diff_observations(
390 current: &Observation,
391 previous: &Observation,
392) -> (bool, Vec<String>, Vec<String>) {
393 let mut added: Vec<String> = current
394 .iter()
395 .filter(|(key, _)| !previous.contains_key(*key))
396 .map(|(_, value)| value.clone())
397 .collect();
398 let mut removed: Vec<String> = previous
399 .iter()
400 .filter(|(key, _)| !current.contains_key(*key))
401 .map(|(_, value)| value.clone())
402 .collect();
403 added.sort();
404 added.dedup();
405 removed.sort();
406 removed.dedup();
407
408 let changed = !added.is_empty() || !removed.is_empty();
409 (changed, added, removed)
410}
411
412#[cfg(test)]
413mod tests {
414 use super::*;
415
416 use super::super::records::RecordData;
417
418 #[test]
422 fn from_config_applies_dns_timeout() {
423 let mut config = crate::config::SeerConfig::default();
424 config.timeouts.dns_secs = 9;
425 let follower = DnsFollower::from_config(&config);
426 assert_eq!(follower.resolver.timeout(), Duration::from_secs(9));
427 }
428
429 fn record(data: RecordData) -> DnsRecord {
430 DnsRecord {
431 name: "example.com".to_string(),
432 record_type: RecordType::NS,
433 ttl: 300,
434 data,
435 }
436 }
437
438 fn ns_observation<const N: usize>(names: [&str; N]) -> Observation {
440 let records: Vec<DnsRecord> = names
441 .iter()
442 .map(|n| {
443 record(RecordData::NS {
444 nameserver: n.to_string(),
445 })
446 })
447 .collect();
448 observe(&records)
449 }
450
451 #[tokio::test]
452 async fn test_follow_config_default() {
453 let config = FollowConfig::default();
454 assert_eq!(config.iterations, 10);
455 assert_eq!(config.interval_secs, 60);
456 assert!(!config.changes_only);
457 }
458
459 #[test]
463 fn diff_values_ignores_case_only_differences() {
464 let previous = ns_observation(["NS1.EXAMPLE.COM.", "ns2.example.com."]);
465 let current = ns_observation(["ns1.example.com.", "NS2.EXAMPLE.COM."]);
466 let (changed, added, removed) = diff_observations(¤t, &previous);
467 assert!(!changed, "case-only differences must not count as a change");
468 assert!(added.is_empty(), "no added values: {added:?}");
469 assert!(removed.is_empty(), "no removed values: {removed:?}");
470 }
471
472 #[test]
476 fn diff_values_detects_real_change_preserving_case() {
477 let previous = ns_observation(["NS1.Example.com."]);
478 let current = ns_observation(["NS2.Example.Com."]);
479 let (changed, added, removed) = diff_observations(¤t, &previous);
480 assert!(changed, "a different nameserver is a real change");
481 assert_eq!(added, vec!["NS2.Example.Com.".to_string()]);
482 assert_eq!(removed, vec!["NS1.Example.com.".to_string()]);
483 }
484
485 #[test]
488 fn diff_values_reports_case_change_in_txt() {
489 let txt = |t: &str| {
490 observe(&[record(RecordData::TXT {
491 text: t.to_string(),
492 })])
493 };
494 let (changed, added, removed) = diff_observations(&txt("token=abc"), &txt("token=AbC"));
495 assert!(changed);
496 assert_eq!(added, vec!["\"token=abc\"".to_string()]);
497 assert_eq!(removed, vec!["\"token=AbC\"".to_string()]);
498 }
499
500 #[test]
501 fn follow_config_rounds_interval_to_nearest_second() {
502 assert_eq!(FollowConfig::new(2, 2.05).unwrap().interval_secs, 123);
504 assert_eq!(FollowConfig::new(2, 0.5).unwrap().interval_secs, 30);
505 }
506
507 #[tokio::test]
511 async fn follow_rejects_out_of_range_literal_config() {
512 let follower = DnsFollower::new();
513 for config in [
514 FollowConfig {
515 iterations: usize::MAX,
516 interval_secs: 1,
517 changes_only: false,
518 },
519 FollowConfig {
520 iterations: 0,
521 interval_secs: 1,
522 changes_only: false,
523 },
524 FollowConfig {
525 iterations: 2,
526 interval_secs: MAX_FOLLOW_INTERVAL_SECS + 1,
527 changes_only: false,
528 },
529 ] {
530 let err = follower
531 .follow("example.com", RecordType::A, None, config, None, None)
532 .await
533 .expect_err("out-of-range config must be rejected before any query");
534 assert!(matches!(err, SeerError::InvalidInput(_)), "{err:?}");
535 }
536 }
537
538 #[tokio::test]
543 async fn follow_first_iteration_error_is_not_a_baseline() {
544 use std::sync::atomic::{AtomicBool, Ordering};
545
546 use hickory_resolver::proto::rr::rdata as wire;
547 use hickory_resolver::proto::rr::{RData as HickoryRData, RecordType as WireType};
548
549 use crate::dns::test_support::{mock_dns_resolver, spawn_mock_dns_fn, MockReply};
550
551 let healthy = Arc::new(AtomicBool::new(false));
554 let port = spawn_mock_dns_fn({
555 let healthy = Arc::clone(&healthy);
556 move |_, qtype| {
557 if healthy.load(Ordering::SeqCst) && qtype == WireType::A {
558 MockReply::Answer(vec![HickoryRData::A(wire::A(std::net::Ipv4Addr::new(
559 192, 0, 2, 7,
560 )))])
561 } else {
562 MockReply::ServFail
563 }
564 }
565 })
566 .await;
567 let callback: FollowProgressCallback = {
568 let healthy = Arc::clone(&healthy);
569 Arc::new(move |_: &FollowIteration| healthy.store(true, Ordering::SeqCst))
570 };
571
572 let config = FollowConfig {
573 iterations: 3,
574 interval_secs: 0,
575 changes_only: false,
576 };
577 let result = DnsFollower::with_resolver(mock_dns_resolver(port))
578 .follow(
579 "seer.test",
580 RecordType::A,
581 Some("127.0.0.1"),
582 config,
583 Some(callback),
584 None,
585 )
586 .await
587 .expect("follow against the mock fixture");
588
589 assert!(!result.iterations[0].success(), "iteration 1 must error");
590 let second = &result.iterations[1];
591 assert!(second.success(), "{second:?}");
592 assert!(
593 !second.changed && second.added.is_empty() && second.removed.is_empty(),
594 "the first success establishes the baseline, it is not a change: {second:?}"
595 );
596 assert!(!result.iterations[2].changed);
597 assert_eq!(result.total_changes, 0);
598 }
599
600 #[tokio::test]
602 async fn follow_uses_per_name_normalization() {
603 use crate::dns::test_support::{mock_dns_resolver, spawn_mock_dns, MockMode};
604
605 let port = spawn_mock_dns(MockMode::Zone).await;
606 let follower = DnsFollower::with_resolver(mock_dns_resolver(port));
607 let one_shot = || FollowConfig::new(1, 0.0).expect("valid config");
608
609 let result = follower
610 .follow(
611 "www.seer.test",
612 RecordType::CNAME,
613 Some("127.0.0.1"),
614 one_shot(),
615 None,
616 None,
617 )
618 .await
619 .expect("follow www");
620 assert_eq!(result.domain, "www.seer.test");
621 assert_eq!(result.iterations[0].record_count(), 1);
622
623 let result = follower
624 .follow(
625 "2606:4700:4700::1111",
626 RecordType::PTR,
627 Some("127.0.0.1"),
628 one_shot(),
629 None,
630 None,
631 )
632 .await
633 .expect("IPv6 PTR literal must be accepted");
634 assert_eq!(result.domain, "2606:4700:4700::1111");
635 assert_eq!(result.iterations[0].record_count(), 1);
636 }
637
638 #[test]
639 fn follow_config_rejects_unbounded_iterations() {
640 assert!(FollowConfig::new(MAX_FOLLOW_ITERATIONS, 1.0).is_ok());
641 let err = FollowConfig::new(MAX_FOLLOW_ITERATIONS + 1, 1.0).unwrap_err();
642 assert!(matches!(err, SeerError::InvalidInput(_)));
643 assert!(FollowConfig::new(usize::MAX, 1.0).is_err());
644 }
645
646 #[tokio::test]
647 async fn test_follow_config_new() {
648 let config = FollowConfig::new(5, 0.5).unwrap();
649 assert_eq!(config.iterations, 5);
650 assert_eq!(config.interval_secs, 30);
651 }
652
653 #[tokio::test]
659 async fn follow_loop_detects_change_against_mock_dns() {
660 use std::sync::Mutex;
661
662 use hickory_resolver::proto::rr::rdata as wire;
663 use hickory_resolver::proto::rr::RData as HickoryRData;
664
665 use crate::dns::test_support::{mock_dns_resolver, spawn_mock_dns_sequence};
666
667 let a = |last_octet: u8| {
668 HickoryRData::A(wire::A(std::net::Ipv4Addr::new(192, 0, 2, last_octet)))
669 };
670 let port = spawn_mock_dns_sequence(vec![vec![a(1)], vec![a(2)], vec![a(2)]]).await;
673
674 let config = FollowConfig {
678 iterations: 3,
679 interval_secs: 0,
680 changes_only: false,
681 };
682
683 let seen: Arc<Mutex<Vec<(usize, bool)>>> = Arc::new(Mutex::new(Vec::new()));
684 let callback: FollowProgressCallback = {
685 let seen = Arc::clone(&seen);
686 Arc::new(move |it: &FollowIteration| {
687 seen.lock()
688 .expect("callback mutex")
689 .push((it.iteration, it.changed));
690 })
691 };
692
693 let follower = DnsFollower::with_resolver(mock_dns_resolver(port));
694 let result = follower
695 .follow(
696 "seer.test",
697 RecordType::A,
698 Some("127.0.0.1"),
699 config,
700 Some(callback),
701 None,
702 )
703 .await
704 .expect("follow against the mock fixture must succeed");
705
706 assert_eq!(result.completed_iterations(), 3);
708 assert!(result.iterations.iter().all(|i| i.success()));
709 assert!(!result.interrupted);
710 assert_eq!(result.domain, "seer.test");
711 assert_eq!(result.record_type, RecordType::A);
712
713 let first = &result.iterations[0];
715 assert_eq!(first.record_count(), 1);
716 assert_eq!(first.records[0].data.to_string(), "192.0.2.1");
717 assert!(!first.changed, "first iteration is never a change");
718 assert!(first.added.is_empty() && first.removed.is_empty());
719
720 let second = &result.iterations[1];
722 assert!(second.changed, "record change must be detected");
723 assert_eq!(second.added, vec!["192.0.2.2".to_string()]);
724 assert_eq!(second.removed, vec!["192.0.2.1".to_string()]);
725
726 let third = &result.iterations[2];
728 assert!(!third.changed, "unchanged records must not be a change");
729 assert!(third.added.is_empty() && third.removed.is_empty());
730
731 assert_eq!(result.total_changes, 1);
732
733 let seen = seen.lock().expect("callback mutex");
735 assert_eq!(*seen, vec![(1, false), (2, true), (3, false)]);
736 }
737
738 #[tokio::test]
739 #[ignore = "live network; run with --ignored or SEER_LIVE_TESTS=1"]
740 async fn test_follow_single_iteration() {
741 let follower = DnsFollower::new();
742 let config = FollowConfig::new(1, 0.0).unwrap();
743
744 let result = follower
745 .follow("example.com", RecordType::A, None, config, None, None)
746 .await;
747
748 assert!(result.is_ok());
749 let result = result.unwrap();
750 assert_eq!(result.completed_iterations(), 1);
751 assert!(!result.interrupted);
752 }
753
754 #[test]
755 fn follow_config_rejects_zero_iterations() {
756 assert!(FollowConfig::new(0, 1.0).is_err());
757 }
758
759 #[test]
760 fn follow_config_rejects_infinite_interval() {
761 assert!(FollowConfig::new(10, f64::INFINITY).is_err());
762 assert!(FollowConfig::new(10, f64::NEG_INFINITY).is_err());
763 }
764
765 #[test]
766 fn follow_config_rejects_nan_interval() {
767 assert!(FollowConfig::new(10, f64::NAN).is_err());
768 }
769
770 #[test]
771 fn follow_config_rejects_negative_interval() {
772 assert!(FollowConfig::new(10, -1.0).is_err());
773 }
774
775 #[test]
776 fn follow_config_rejects_interval_above_cap() {
777 assert!(FollowConfig::new(10, 60.1).is_err());
778 }
779
780 #[test]
781 fn follow_config_accepts_valid() {
782 assert!(FollowConfig::new(10, 1.5).is_ok());
783 assert!(FollowConfig::new(1, 0.0).is_ok());
784 assert!(FollowConfig::new(1, 60.0).is_ok());
785 }
786
787 #[test]
788 fn follow_config_floors_subsecond_interval_for_multi_iteration() {
789 let config = FollowConfig::new(10_000, 0.001).unwrap();
793 assert!(
794 config.interval_secs >= 1,
795 "multi-iteration interval must be floored to >= 1s, got {}",
796 config.interval_secs
797 );
798 }
799
800 #[test]
801 fn follow_config_allows_zero_interval_for_single_iteration() {
802 let config = FollowConfig::new(1, 0.0).unwrap();
805 assert_eq!(config.interval_secs, 0);
806 }
807
808 #[tokio::test]
809 #[ignore = "live network; run with --ignored or SEER_LIVE_TESTS=1"]
810 async fn follow_honors_cancel() {
811 use tokio::sync::watch;
812
813 let (tx, rx) = watch::channel(false);
814 let config = FollowConfig::new(100, 0.5).unwrap();
816 let follower = DnsFollower::new();
817
818 let handle = tokio::spawn(async move {
819 follower
820 .follow("example.com", RecordType::A, None, config, None, Some(rx))
821 .await
822 });
823
824 tokio::time::sleep(Duration::from_millis(200)).await;
826 tx.send(true).unwrap();
827
828 let joined = tokio::time::timeout(Duration::from_secs(10), handle)
829 .await
830 .expect("follow should return promptly after cancel");
831 let result = joined.expect("join").expect("follow result");
832 assert!(result.interrupted, "follow should be interrupted");
833 assert!(
834 result.completed_iterations() < 100,
835 "should not complete all iterations"
836 );
837 }
838}