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