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 mut 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 mut rx) = cancel_rx {
337 if cancelled_during(rx, sleep_duration).await {
338 debug!("Follow operation cancelled during sleep");
339 interrupted = true;
340 break;
341 }
342 } else {
343 tokio::time::sleep(sleep_duration).await;
344 }
345 }
346 }
347
348 let ended_at = Utc::now();
349
350 Ok(FollowResult {
351 domain,
352 record_type,
353 nameserver: nameserver.map(|s| s.to_string()),
354 iterations_requested: config.iterations,
355 interval_secs: config.interval_secs,
356 iterations,
357 interrupted,
358 total_changes,
359 started_at,
360 ended_at,
361 })
362 }
363}
364
365async fn cancelled_during(rx: &mut watch::Receiver<bool>, duration: Duration) -> bool {
371 let cancel = async {
372 if rx.wait_for(|cancelled| *cancelled).await.is_err() {
373 std::future::pending::<()>().await;
374 }
375 };
376 tokio::select! {
377 _ = tokio::time::sleep(duration) => false,
378 _ = cancel => true,
379 }
380}
381
382type Observation = BTreeMap<String, String>;
386
387fn observe(records: &[DnsRecord]) -> Observation {
388 records
389 .iter()
390 .map(|r| (r.data.comparison_key(), r.data.to_string()))
391 .collect()
392}
393
394fn diff_observations(
400 current: &Observation,
401 previous: &Observation,
402) -> (bool, Vec<String>, Vec<String>) {
403 let mut added: Vec<String> = current
404 .iter()
405 .filter(|(key, _)| !previous.contains_key(*key))
406 .map(|(_, value)| value.clone())
407 .collect();
408 let mut removed: Vec<String> = previous
409 .iter()
410 .filter(|(key, _)| !current.contains_key(*key))
411 .map(|(_, value)| value.clone())
412 .collect();
413 added.sort();
414 added.dedup();
415 removed.sort();
416 removed.dedup();
417
418 let changed = !added.is_empty() || !removed.is_empty();
419 (changed, added, removed)
420}
421
422#[cfg(test)]
423mod tests {
424 use super::*;
425
426 use super::super::records::RecordData;
427
428 #[test]
432 fn from_config_applies_dns_timeout() {
433 let mut config = crate::config::SeerConfig::default();
434 config.timeouts.dns_secs = 9;
435 let follower = DnsFollower::from_config(&config);
436 assert_eq!(follower.resolver.timeout(), Duration::from_secs(9));
437 }
438
439 fn record(data: RecordData) -> DnsRecord {
440 DnsRecord {
441 name: "example.com".to_string(),
442 record_type: RecordType::NS,
443 ttl: 300,
444 data,
445 }
446 }
447
448 fn ns_observation<const N: usize>(names: [&str; N]) -> Observation {
450 let records: Vec<DnsRecord> = names
451 .iter()
452 .map(|n| {
453 record(RecordData::NS {
454 nameserver: n.to_string(),
455 })
456 })
457 .collect();
458 observe(&records)
459 }
460
461 #[tokio::test]
462 async fn test_follow_config_default() {
463 let config = FollowConfig::default();
464 assert_eq!(config.iterations, 10);
465 assert_eq!(config.interval_secs, 60);
466 assert!(!config.changes_only);
467 }
468
469 #[test]
473 fn diff_values_ignores_case_only_differences() {
474 let previous = ns_observation(["NS1.EXAMPLE.COM.", "ns2.example.com."]);
475 let current = ns_observation(["ns1.example.com.", "NS2.EXAMPLE.COM."]);
476 let (changed, added, removed) = diff_observations(¤t, &previous);
477 assert!(!changed, "case-only differences must not count as a change");
478 assert!(added.is_empty(), "no added values: {added:?}");
479 assert!(removed.is_empty(), "no removed values: {removed:?}");
480 }
481
482 #[test]
486 fn diff_values_detects_real_change_preserving_case() {
487 let previous = ns_observation(["NS1.Example.com."]);
488 let current = ns_observation(["NS2.Example.Com."]);
489 let (changed, added, removed) = diff_observations(¤t, &previous);
490 assert!(changed, "a different nameserver is a real change");
491 assert_eq!(added, vec!["NS2.Example.Com.".to_string()]);
492 assert_eq!(removed, vec!["NS1.Example.com.".to_string()]);
493 }
494
495 #[test]
498 fn diff_values_reports_case_change_in_txt() {
499 let txt = |t: &str| observe(&[record(RecordData::txt(vec![t.to_string()]))]);
500 let (changed, added, removed) = diff_observations(&txt("token=abc"), &txt("token=AbC"));
501 assert!(changed);
502 assert_eq!(added, vec!["\"token=abc\"".to_string()]);
503 assert_eq!(removed, vec!["\"token=AbC\"".to_string()]);
504 }
505
506 #[test]
507 fn follow_config_rounds_interval_to_nearest_second() {
508 assert_eq!(FollowConfig::new(2, 2.05).unwrap().interval_secs, 123);
510 assert_eq!(FollowConfig::new(2, 0.5).unwrap().interval_secs, 30);
511 }
512
513 #[tokio::test]
517 async fn follow_rejects_out_of_range_literal_config() {
518 let follower = DnsFollower::new();
519 for config in [
520 FollowConfig {
521 iterations: usize::MAX,
522 interval_secs: 1,
523 changes_only: false,
524 },
525 FollowConfig {
526 iterations: 0,
527 interval_secs: 1,
528 changes_only: false,
529 },
530 FollowConfig {
531 iterations: 2,
532 interval_secs: MAX_FOLLOW_INTERVAL_SECS + 1,
533 changes_only: false,
534 },
535 ] {
536 let err = follower
537 .follow("example.com", RecordType::A, None, config, None, None)
538 .await
539 .expect_err("out-of-range config must be rejected before any query");
540 assert!(matches!(err, SeerError::InvalidInput(_)), "{err:?}");
541 }
542 }
543
544 #[tokio::test]
549 async fn follow_first_iteration_error_is_not_a_baseline() {
550 use std::sync::atomic::{AtomicBool, Ordering};
551
552 use hickory_resolver::proto::op::ResponseCode;
553 use hickory_resolver::proto::rr::rdata as wire;
554 use hickory_resolver::proto::rr::{RData as HickoryRData, RecordType as WireType};
555
556 use crate::dns::test_support::{mock_dns_resolver, spawn_mock_dns_fn, MockReply};
557
558 let healthy = Arc::new(AtomicBool::new(false));
561 let port = spawn_mock_dns_fn({
562 let healthy = Arc::clone(&healthy);
563 move |_, qtype| {
564 if healthy.load(Ordering::SeqCst) && qtype == WireType::A {
565 MockReply::Answer(vec![HickoryRData::A(wire::A(std::net::Ipv4Addr::new(
566 192, 0, 2, 7,
567 )))])
568 } else {
569 MockReply::Rcode(ResponseCode::ServFail)
570 }
571 }
572 })
573 .await;
574 let callback: FollowProgressCallback = {
575 let healthy = Arc::clone(&healthy);
576 Arc::new(move |_: &FollowIteration| healthy.store(true, Ordering::SeqCst))
577 };
578
579 let config = FollowConfig {
580 iterations: 3,
581 interval_secs: 0,
582 changes_only: false,
583 };
584 let result = DnsFollower::with_resolver(mock_dns_resolver(port))
585 .follow(
586 "seer.test",
587 RecordType::A,
588 Some("127.0.0.1"),
589 config,
590 Some(callback),
591 None,
592 )
593 .await
594 .expect("follow against the mock fixture");
595
596 assert!(!result.iterations[0].success(), "iteration 1 must error");
597 let second = &result.iterations[1];
598 assert!(second.success(), "{second:?}");
599 assert!(
600 !second.changed && second.added.is_empty() && second.removed.is_empty(),
601 "the first success establishes the baseline, it is not a change: {second:?}"
602 );
603 assert!(!result.iterations[2].changed);
604 assert_eq!(result.total_changes, 0);
605 }
606
607 #[tokio::test]
609 async fn follow_uses_per_name_normalization() {
610 use crate::dns::test_support::{mock_dns_resolver, spawn_mock_dns, MockMode};
611
612 let port = spawn_mock_dns(MockMode::Zone).await;
613 let follower = DnsFollower::with_resolver(mock_dns_resolver(port));
614 let one_shot = || FollowConfig::new(1, 0.0).expect("valid config");
615
616 let result = follower
617 .follow(
618 "www.seer.test",
619 RecordType::CNAME,
620 Some("127.0.0.1"),
621 one_shot(),
622 None,
623 None,
624 )
625 .await
626 .expect("follow www");
627 assert_eq!(result.domain, "www.seer.test");
628 assert_eq!(result.iterations[0].record_count(), 1);
629
630 let result = follower
631 .follow(
632 "2606:4700:4700::1111",
633 RecordType::PTR,
634 Some("127.0.0.1"),
635 one_shot(),
636 None,
637 None,
638 )
639 .await
640 .expect("IPv6 PTR literal must be accepted");
641 assert_eq!(result.domain, "2606:4700:4700::1111");
642 assert_eq!(result.iterations[0].record_count(), 1);
643 }
644
645 #[test]
646 fn follow_config_rejects_unbounded_iterations() {
647 assert!(FollowConfig::new(MAX_FOLLOW_ITERATIONS, 1.0).is_ok());
648 let err = FollowConfig::new(MAX_FOLLOW_ITERATIONS + 1, 1.0).unwrap_err();
649 assert!(matches!(err, SeerError::InvalidInput(_)));
650 assert!(FollowConfig::new(usize::MAX, 1.0).is_err());
651 }
652
653 #[tokio::test]
654 async fn test_follow_config_new() {
655 let config = FollowConfig::new(5, 0.5).unwrap();
656 assert_eq!(config.iterations, 5);
657 assert_eq!(config.interval_secs, 30);
658 }
659
660 #[tokio::test]
666 async fn follow_loop_detects_change_against_mock_dns() {
667 use std::sync::Mutex;
668
669 use hickory_resolver::proto::rr::rdata as wire;
670 use hickory_resolver::proto::rr::RData as HickoryRData;
671
672 use crate::dns::test_support::{mock_dns_resolver, spawn_mock_dns_sequence};
673
674 let a = |last_octet: u8| {
675 HickoryRData::A(wire::A(std::net::Ipv4Addr::new(192, 0, 2, last_octet)))
676 };
677 let port = spawn_mock_dns_sequence(vec![vec![a(1)], vec![a(2)], vec![a(2)]]).await;
680
681 let config = FollowConfig {
685 iterations: 3,
686 interval_secs: 0,
687 changes_only: false,
688 };
689
690 let seen: Arc<Mutex<Vec<(usize, bool)>>> = Arc::new(Mutex::new(Vec::new()));
691 let callback: FollowProgressCallback = {
692 let seen = Arc::clone(&seen);
693 Arc::new(move |it: &FollowIteration| {
694 seen.lock()
695 .expect("callback mutex")
696 .push((it.iteration, it.changed));
697 })
698 };
699
700 let follower = DnsFollower::with_resolver(mock_dns_resolver(port));
701 let result = follower
702 .follow(
703 "seer.test",
704 RecordType::A,
705 Some("127.0.0.1"),
706 config,
707 Some(callback),
708 None,
709 )
710 .await
711 .expect("follow against the mock fixture must succeed");
712
713 assert_eq!(result.completed_iterations(), 3);
715 assert!(result.iterations.iter().all(|i| i.success()));
716 assert!(!result.interrupted);
717 assert_eq!(result.domain, "seer.test");
718 assert_eq!(result.record_type, RecordType::A);
719
720 let first = &result.iterations[0];
722 assert_eq!(first.record_count(), 1);
723 assert_eq!(first.records[0].data.to_string(), "192.0.2.1");
724 assert!(!first.changed, "first iteration is never a change");
725 assert!(first.added.is_empty() && first.removed.is_empty());
726
727 let second = &result.iterations[1];
729 assert!(second.changed, "record change must be detected");
730 assert_eq!(second.added, vec!["192.0.2.2".to_string()]);
731 assert_eq!(second.removed, vec!["192.0.2.1".to_string()]);
732
733 let third = &result.iterations[2];
735 assert!(!third.changed, "unchanged records must not be a change");
736 assert!(third.added.is_empty() && third.removed.is_empty());
737
738 assert_eq!(result.total_changes, 1);
739
740 let seen = seen.lock().expect("callback mutex");
742 assert_eq!(*seen, vec![(1, false), (2, true), (3, false)]);
743 }
744
745 #[tokio::test]
746 #[ignore = "live network; run with --ignored or SEER_LIVE_TESTS=1"]
747 async fn test_follow_single_iteration() {
748 let follower = DnsFollower::new();
749 let config = FollowConfig::new(1, 0.0).unwrap();
750
751 let result = follower
752 .follow("example.com", RecordType::A, None, config, None, None)
753 .await;
754
755 assert!(result.is_ok());
756 let result = result.unwrap();
757 assert_eq!(result.completed_iterations(), 1);
758 assert!(!result.interrupted);
759 }
760
761 #[test]
762 fn follow_config_rejects_zero_iterations() {
763 assert!(FollowConfig::new(0, 1.0).is_err());
764 }
765
766 #[test]
767 fn follow_config_rejects_infinite_interval() {
768 assert!(FollowConfig::new(10, f64::INFINITY).is_err());
769 assert!(FollowConfig::new(10, f64::NEG_INFINITY).is_err());
770 }
771
772 #[test]
773 fn follow_config_rejects_nan_interval() {
774 assert!(FollowConfig::new(10, f64::NAN).is_err());
775 }
776
777 #[test]
778 fn follow_config_rejects_negative_interval() {
779 assert!(FollowConfig::new(10, -1.0).is_err());
780 }
781
782 #[test]
783 fn follow_config_rejects_interval_above_cap() {
784 assert!(FollowConfig::new(10, 60.1).is_err());
785 }
786
787 #[test]
788 fn follow_config_accepts_valid() {
789 assert!(FollowConfig::new(10, 1.5).is_ok());
790 assert!(FollowConfig::new(1, 0.0).is_ok());
791 assert!(FollowConfig::new(1, 60.0).is_ok());
792 }
793
794 #[test]
795 fn follow_config_floors_subsecond_interval_for_multi_iteration() {
796 let config = FollowConfig::new(10_000, 0.001).unwrap();
800 assert!(
801 config.interval_secs >= 1,
802 "multi-iteration interval must be floored to >= 1s, got {}",
803 config.interval_secs
804 );
805 }
806
807 #[test]
808 fn follow_config_allows_zero_interval_for_single_iteration() {
809 let config = FollowConfig::new(1, 0.0).unwrap();
812 assert_eq!(config.interval_secs, 0);
813 }
814
815 #[tokio::test]
816 #[ignore = "live network; run with --ignored or SEER_LIVE_TESTS=1"]
817 async fn follow_honors_cancel() {
818 use tokio::sync::watch;
819
820 let (tx, rx) = watch::channel(false);
821 let config = FollowConfig::new(100, 0.5).unwrap();
823 let follower = DnsFollower::new();
824
825 let handle = tokio::spawn(async move {
826 follower
827 .follow("example.com", RecordType::A, None, config, None, Some(rx))
828 .await
829 });
830
831 tokio::time::sleep(Duration::from_millis(200)).await;
833 tx.send(true).unwrap();
834
835 let joined = tokio::time::timeout(Duration::from_secs(10), handle)
836 .await
837 .expect("follow should return promptly after cancel");
838 let result = joined.expect("join").expect("follow result");
839 assert!(result.interrupted, "follow should be interrupted");
840 assert!(
841 result.completed_iterations() < 100,
842 "should not complete all iterations"
843 );
844 }
845
846 #[tokio::test(start_paused = true)]
850 async fn cancel_sleep_runs_full_length_when_sender_dropped() {
851 let (tx, mut rx) = watch::channel(false);
852 drop(tx);
853 let start = tokio::time::Instant::now();
854 assert!(!cancelled_during(&mut rx, Duration::from_secs(30)).await);
855 assert_eq!(start.elapsed(), Duration::from_secs(30));
856 }
857
858 #[tokio::test(start_paused = true)]
861 async fn cancel_sleep_only_true_cancels() {
862 let (tx, mut rx) = watch::channel(false);
863 tx.send(false).expect("receiver alive");
864 let start = tokio::time::Instant::now();
865 assert!(!cancelled_during(&mut rx, Duration::from_secs(5)).await);
866 assert_eq!(start.elapsed(), Duration::from_secs(5));
867
868 tx.send(true).expect("receiver alive");
869 let start = tokio::time::Instant::now();
870 assert!(cancelled_during(&mut rx, Duration::from_secs(5)).await);
871 assert_eq!(start.elapsed(), Duration::ZERO);
872 }
873
874 #[tokio::test(start_paused = true)]
875 async fn cancel_sleep_wakes_on_cancel_mid_sleep() {
876 let (tx, mut rx) = watch::channel(false);
877 tokio::spawn(async move {
878 tokio::time::sleep(Duration::from_secs(2)).await;
879 tx.send(true).expect("receiver alive");
880 });
881 let start = tokio::time::Instant::now();
882 assert!(cancelled_during(&mut rx, Duration::from_secs(60)).await);
883 assert_eq!(start.elapsed(), Duration::from_secs(2));
884 }
885}