rpki/src/parallel/dead_repo_blacklist_tests.rs

use super::*;
use crate::parallel::types::RepoTransportMode;

const RRDP_URI: &str = "https://dead.example.com/notification.xml";
const RSYNC_URI: &str = "rsync://dead.example.com/repo/";

fn now() -> u64 {
    1_700_000_000
}

#[test]
fn admission_requires_threshold_consecutive_failures() {
    let mut bl = DeadRepoBlacklist::new();
    assert!(!bl.is_blacklisted(RepoTransportMode::Rrdp, RRDP_URI));

    let out1 = bl.record_transport_failure(RepoTransportMode::Rrdp, RRDP_URI, now(), 3, 256);
    assert_eq!(
        out1,
        DeadRepoFailureOutcome::Counted {
            consecutive_failures: 1
        }
    );
    assert!(!bl.is_blacklisted(RepoTransportMode::Rrdp, RRDP_URI));

    let out2 = bl.record_transport_failure(RepoTransportMode::Rrdp, RRDP_URI, now() + 1, 3, 256);
    assert_eq!(
        out2,
        DeadRepoFailureOutcome::Counted {
            consecutive_failures: 2
        }
    );
    assert!(!bl.is_blacklisted(RepoTransportMode::Rrdp, RRDP_URI));

    let out3 = bl.record_transport_failure(RepoTransportMode::Rrdp, RRDP_URI, now() + 2, 3, 256);
    assert_eq!(out3, DeadRepoFailureOutcome::Admitted);
    assert!(bl.is_blacklisted(RepoTransportMode::Rrdp, RRDP_URI));

    // Further failures keep counting but do not re-admit.
    let out4 = bl.record_transport_failure(RepoTransportMode::Rrdp, RRDP_URI, now() + 3, 3, 256);
    assert!(matches!(
        out4,
        DeadRepoFailureOutcome::Counted {
            consecutive_failures: 4
        }
    ));
}

#[test]
fn threshold_of_one_admits_immediately() {
    let mut bl = DeadRepoBlacklist::new();
    let out = bl.record_transport_failure(RepoTransportMode::Rsync, RSYNC_URI, now(), 1, 256);
    assert_eq!(out, DeadRepoFailureOutcome::Admitted);
}

#[test]
fn success_resets_counting_entry() {
    let mut bl = DeadRepoBlacklist::new();
    bl.record_transport_failure(RepoTransportMode::Rrdp, RRDP_URI, now(), 3, 256);
    assert_eq!(bl.len(), 1);
    let out = bl.record_transport_success(RepoTransportMode::Rrdp, RRDP_URI);
    assert_eq!(out, Some(DeadRepoSuccessOutcome::CounterReset));
    assert!(bl.is_empty());
    // Next failure starts from zero again.
    let out = bl.record_transport_failure(RepoTransportMode::Rrdp, RRDP_URI, now(), 3, 256);
    assert_eq!(
        out,
        DeadRepoFailureOutcome::Counted {
            consecutive_failures: 1
        }
    );
}

#[test]
fn success_on_blacklisted_entry_self_heals() {
    let mut bl = DeadRepoBlacklist::new();
    for i in 0..3 {
        bl.record_transport_failure(RepoTransportMode::Rsync, RSYNC_URI, now() + i, 3, 256);
    }
    assert!(bl.is_blacklisted(RepoTransportMode::Rsync, RSYNC_URI));
    let out = bl.record_transport_success(RepoTransportMode::Rsync, RSYNC_URI);
    assert_eq!(out, Some(DeadRepoSuccessOutcome::BlacklistRemoved));
    assert!(!bl.is_blacklisted(RepoTransportMode::Rsync, RSYNC_URI));
    assert!(bl.is_empty());
}

#[test]
fn success_on_unknown_entry_is_none() {
    let mut bl = DeadRepoBlacklist::new();
    assert_eq!(
        bl.record_transport_success(RepoTransportMode::Rrdp, RRDP_URI),
        None
    );
}

#[test]
fn entries_are_per_transport() {
    let mut bl = DeadRepoBlacklist::new();
    for i in 0..3 {
        bl.record_transport_failure(RepoTransportMode::Rrdp, RRDP_URI, now() + i, 3, 256);
    }
    assert!(bl.is_blacklisted(RepoTransportMode::Rrdp, RRDP_URI));
    assert!(!bl.is_blacklisted(RepoTransportMode::Rsync, RRDP_URI));
}

#[test]
fn capacity_evicts_stale_counting_entries_first() {
    let mut bl = DeadRepoBlacklist::new();
    // One blacklisted entry (old) and two counting entries.
    for i in 0..3 {
        bl.record_transport_failure(RepoTransportMode::Rrdp, "https://a.example/n.xml", 100 + i, 1, 3);
    }
    bl.record_transport_failure(RepoTransportMode::Rrdp, "https://b.example/n.xml", 200, 3, 3);
    bl.record_transport_failure(RepoTransportMode::Rrdp, "https://c.example/n.xml", 300, 3, 3);
    assert_eq!(bl.len(), 3);
    assert_eq!(bl.blacklisted_len(), 1);

    // Inserting a fourth entry evicts the stalest counting entry (b).
    bl.record_transport_failure(RepoTransportMode::Rrdp, "https://d.example/n.xml", 400, 3, 3);
    assert_eq!(bl.len(), 3);
    assert!(bl.entries_sorted().iter().any(|e| e.uri == "https://a.example/n.xml"));
    assert!(!bl.entries_sorted().iter().any(|e| e.uri == "https://b.example/n.xml"));
    assert!(bl.entries_sorted().iter().any(|e| e.uri == "https://c.example/n.xml"));
    assert!(bl.entries_sorted().iter().any(|e| e.uri == "https://d.example/n.xml"));
}

#[test]
fn capacity_evicts_oldest_blacklisted_when_no_counting_entries() {
    let mut bl = DeadRepoBlacklist::new();
    bl.record_transport_failure(RepoTransportMode::Rrdp, "https://a.example/n.xml", 100, 1, 2);
    bl.record_transport_failure(RepoTransportMode::Rrdp, "https://b.example/n.xml", 200, 1, 2);
    assert_eq!(bl.blacklisted_len(), 2);
    bl.record_transport_failure(RepoTransportMode::Rrdp, "https://c.example/n.xml", 300, 1, 2);
    assert_eq!(bl.len(), 2);
    assert!(!bl.entries_sorted().iter().any(|e| e.uri == "https://a.example/n.xml"));
    assert!(bl.entries_sorted().iter().any(|e| e.uri == "https://b.example/n.xml"));
    assert!(bl.entries_sorted().iter().any(|e| e.uri == "https://c.example/n.xml"));
}

#[test]
fn store_load_roundtrip() {
    let dir = std::env::temp_dir().join(format!(
        "dead_repo_blacklist_test_roundtrip_{}_{}",
        std::process::id(),
        now()
    ));
    let path = dir.join("state").join("dead-repo-blacklist.json");
    let mut bl = DeadRepoBlacklist::new();
    for i in 0..3 {
        bl.record_transport_failure(RepoTransportMode::Rrdp, RRDP_URI, now() + i, 3, 256);
    }
    bl.record_transport_failure(RepoTransportMode::Rsync, RSYNC_URI, now(), 3, 256);
    bl.store_atomic(&path, now() + 10).unwrap();

    let (loaded, warning) = DeadRepoBlacklist::load(&path);
    assert!(warning.is_none());
    assert!(loaded.is_blacklisted(RepoTransportMode::Rrdp, RRDP_URI));
    assert!(!loaded.is_blacklisted(RepoTransportMode::Rsync, RSYNC_URI));
    assert_eq!(loaded.len(), 2);
    let entry = loaded
        .entries_sorted()
        .into_iter()
        .find(|e| e.uri == RRDP_URI)
        .unwrap();
    assert_eq!(entry.consecutive_failures, 3);
    assert!(entry.blacklisted);
    assert_eq!(entry.first_failure_at_unix, now());

    let _ = std::fs::remove_dir_all(&dir);
}

#[test]
fn load_missing_file_is_empty_without_warning() {
    let path = std::env::temp_dir().join(format!(
        "dead_repo_blacklist_test_missing_{}_{}.json",
        std::process::id(),
        now()
    ));
    let (bl, warning) = DeadRepoBlacklist::load(&path);
    assert!(bl.is_empty());
    assert!(warning.is_none());
}

#[test]
fn load_corrupted_file_degrades_to_empty_with_warning() {
    let path = std::env::temp_dir().join(format!(
        "dead_repo_blacklist_test_corrupt_{}_{}.json",
        std::process::id(),
        now()
    ));
    std::fs::write(&path, b"{not json").unwrap();
    let (bl, warning) = DeadRepoBlacklist::load(&path);
    assert!(bl.is_empty());
    assert!(warning.is_some());
    let _ = std::fs::remove_file(&path);
}

#[test]
fn load_wrong_schema_version_degrades_to_empty() {
    let path = std::env::temp_dir().join(format!(
        "dead_repo_blacklist_test_schema_{}_{}.json",
        std::process::id(),
        now()
    ));
    std::fs::write(
        &path,
        r#"{"schema_version": 99, "updated_at_unix": 0, "entries": []}"#,
    )
    .unwrap();
    let (bl, warning) = DeadRepoBlacklist::load(&path);
    assert!(bl.is_empty());
    assert!(warning.unwrap().contains("schema_version"));
    let _ = std::fs::remove_file(&path);
}

#[test]
fn probe_due_uses_exponential_backoff() {
    let mut bl = DeadRepoBlacklist::new();
    for i in 0..3 {
        bl.record_transport_failure(RepoTransportMode::Rrdp, RRDP_URI, now() + i, 3, 256);
    }
    // Never probed: due immediately.
    let due = bl.probe_due_entries(now() + 100, 600);
    assert_eq!(due.len(), 1);

    // Probe failure: not due until base interval has passed.
    bl.record_probe_failure(RepoTransportMode::Rrdp, RRDP_URI, now() + 100);
    assert!(bl.probe_due_entries(now() + 100 + 599, 600).is_empty());
    assert_eq!(bl.probe_due_entries(now() + 100 + 600, 600).len(), 1);

    // Second probe failure: backoff doubles to 1200s.
    bl.record_probe_failure(RepoTransportMode::Rrdp, RRDP_URI, now() + 700);
    assert!(bl.probe_due_entries(now() + 700 + 1199, 600).is_empty());
    assert_eq!(bl.probe_due_entries(now() + 700 + 1200, 600).len(), 1);
}

#[test]
fn probe_success_removes_entry() {
    let mut bl = DeadRepoBlacklist::new();
    for i in 0..3 {
        bl.record_transport_failure(RepoTransportMode::Rsync, RSYNC_URI, now() + i, 3, 256);
    }
    assert!(bl.record_probe_success(RepoTransportMode::Rsync, RSYNC_URI));
    assert!(bl.is_empty());
    assert!(!bl.record_probe_success(RepoTransportMode::Rsync, RSYNC_URI));
}

#[test]
fn counting_entries_are_not_probed() {
    let mut bl = DeadRepoBlacklist::new();
    bl.record_transport_failure(RepoTransportMode::Rrdp, RRDP_URI, now(), 3, 256);
    assert!(bl.probe_due_entries(now() + 10_000, 600).is_empty());
}

#[test]
fn probe_backoff_caps_at_max_shift() {
    assert_eq!(probe_backoff_secs(600, 0), 600);
    assert_eq!(probe_backoff_secs(600, 1), 600);
    assert_eq!(probe_backoff_secs(600, 2), 1200);
    assert_eq!(probe_backoff_secs(600, 5), 9600);
    assert_eq!(probe_backoff_secs(600, 100), 9600);
}

#[test]
fn http_fetch_error_classification() {
    use crate::parallel::types::RepoTransportErrorClass;
    assert_eq!(
        classify_http_fetch_error("http request failed: reqwest::Error { kind: Connect, .. }"),
        RepoTransportErrorClass::TransportFetch
    );
    assert_eq!(
        classify_http_fetch_error("http read body failed: connection closed; status=200"),
        RepoTransportErrorClass::TransportFetch
    );
    assert_eq!(
        classify_http_fetch_error("http status 404 Not Found; content_type=; .."),
        RepoTransportErrorClass::Protocol
    );
    assert_eq!(
        classify_http_fetch_error("something unexpected"),
        RepoTransportErrorClass::Unknown
    );
}

#[test]
fn rsync_fetch_error_classification() {
    use crate::parallel::types::RepoTransportErrorClass;
    assert_eq!(
        classify_rsync_fetch_error("rsync error: timeout waiting for daemon connection"),
        RepoTransportErrorClass::TransportFetch
    );
    assert_eq!(
        classify_rsync_fetch_error("rsync: failed to connect to host: Connection refused"),
        RepoTransportErrorClass::TransportFetch
    );
    assert_eq!(
        classify_rsync_fetch_error("temporary failure in name resolution"),
        RepoTransportErrorClass::TransportFetch
    );
    // Content-level failures must not count.
    assert_eq!(
        classify_rsync_fetch_error("rsync file digest mismatch after download"),
        RepoTransportErrorClass::Unknown
    );
    // Stall failures (#141 M9): fail-fast give-up means the host delivered no
    // bytes inside the wall-clock window — transport death, must count.
    assert_eq!(
        classify_rsync_fetch_error(
            "rsync fail-fast gave up after 1 attempts with no progress: rsync wall-clock window expired"
        ),
        RepoTransportErrorClass::TransportFetch
    );
    assert_eq!(
        classify_rsync_fetch_error(
            "rsync fail-fast gave up after 3 attempts with no additional progress: rsync wall-clock window expired"
        ),
        RepoTransportErrorClass::TransportFetch
    );
    // Content-level give-up text without the fail-fast marker stays unknown.
    assert_eq!(
        classify_rsync_fetch_error("rsync error: some files vanished before transfer"),
        RepoTransportErrorClass::Unknown
    );
}

#[test]
fn rrdp_sync_error_classification() {
    use crate::parallel::types::RepoTransportErrorClass;
    use crate::sync::rrdp::RrdpSyncError;
    assert_eq!(
        classify_rrdp_sync_error(&RrdpSyncError::Fetch(
            "http request failed: connect timeout".to_string()
        )),
        RepoTransportErrorClass::TransportFetch
    );
    assert_eq!(
        classify_rrdp_sync_error(&RrdpSyncError::Fetch("http status 500".to_string())),
        RepoTransportErrorClass::Protocol
    );
    assert_eq!(
        classify_rrdp_sync_error(&RrdpSyncError::Storage("db full".to_string())),
        RepoTransportErrorClass::Storage
    );
}

#[test]
fn readmission_after_probe_removal_starts_fresh() {
    let mut bl = DeadRepoBlacklist::new();
    for i in 0..3 {
        bl.record_transport_failure(RepoTransportMode::Rrdp, RRDP_URI, now() + i, 3, 256);
    }
    assert!(bl.record_probe_success(RepoTransportMode::Rrdp, RRDP_URI));
    // No half-open: needs the full threshold again.
    let out = bl.record_transport_failure(RepoTransportMode::Rrdp, RRDP_URI, now() + 10, 3, 256);
    assert_eq!(
        out,
        DeadRepoFailureOutcome::Counted {
            consecutive_failures: 1
        }
    );
}