use std::collections::HashMap;
use std::panic::{AssertUnwindSafe, catch_unwind, resume_unwind};
use std::path::Path;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use crate::error::{Error, Result};
use crate::fault::{CrashSignal, FaultAction, FaultHook};
use crate::fsutil::ensure_private_dir;
use crate::journal::JournalStore;
use crate::manager::{ConflictPolicy, DnsManager, Inner};
use crate::ownership::ResourceLockManager;
use crate::platform::Backend;
use crate::platform::fake::FakeBackend;
use crate::watch::{DnsEvent, SuppressionRegistry};
pub use crate::fault::TxPoint;
pub use crate::platform::fake::{FakeOp, FakeState};
#[derive(Clone)]
pub struct FakeDns {
backend: Arc<FakeBackend>,
}
impl std::fmt::Debug for FakeDns {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("FakeDns").finish()
}
}
impl FakeDns {
pub fn new() -> Self {
Self {
backend: Arc::new(FakeBackend::new()),
}
}
pub fn with_capabilities(caps: crate::capability::Capabilities) -> Self {
Self {
backend: Arc::new(FakeBackend::with_capabilities(caps)),
}
}
pub fn with_multi_resource(caps: crate::capability::Capabilities) -> Self {
Self {
backend: Arc::new(FakeBackend::with_multi_resource(caps)),
}
}
pub fn external_change(&self, resource: &str, state: FakeState) -> Result<()> {
let id: crate::ResourceId = resource.parse().map_err(|e| {
Error::invalid_config(format_args!("invalid resource id {resource:?}: {e}"))
})?;
self.backend.external_change(&id, state);
Ok(())
}
pub fn external_remove(&self, resource: &str) -> Result<bool> {
let id: crate::ResourceId = resource.parse().map_err(|e| {
Error::invalid_config(format_args!("invalid resource id {resource:?}: {e}"))
})?;
Ok(self.backend.external_remove(&id))
}
pub fn current_state(&self, resource: &str) -> Result<Option<FakeState>> {
let id: crate::ResourceId = resource.parse().map_err(|e| {
Error::invalid_config(format_args!("invalid resource id {resource:?}: {e}"))
})?;
Ok(self.backend.state_of(&id))
}
pub fn emit_event(&self, event: DnsEvent) {
self.backend.notify(event);
}
pub fn inject_backend_failure(&self, op: FakeOp, times: u32, message: impl Into<String>) {
self.backend.inject_failure(op, times, message);
}
pub fn inject_backend_failure_after(
&self,
op: FakeOp,
skip: u32,
times: u32,
message: impl Into<String>,
) {
self.backend.inject_failure_after(op, skip, times, message);
}
pub fn lie_once_on_readback(&self, state: FakeState) {
self.backend.lie_once_on_readback(state);
}
}
impl Default for FakeDns {
fn default() -> Self {
Self::new()
}
}
pub fn manager_for_testing(
owner: &str,
state_dir: &Path,
fake: &FakeDns,
lock_timeout: Duration,
) -> Result<DnsManager> {
manager_for_testing_with_policy(
owner,
state_dir,
fake,
lock_timeout,
ConflictPolicy::Cooperative,
)
}
pub fn manager_for_testing_with_policy(
owner: &str,
state_dir: &Path,
fake: &FakeDns,
lock_timeout: Duration,
conflict_policy: ConflictPolicy,
) -> Result<DnsManager> {
use std::collections::HashMap;
ensure_private_dir(state_dir)?;
let locks = ResourceLockManager::new(state_dir.join("locks"), lock_timeout);
locks.ensure_dir()?;
let journal = JournalStore::open(state_dir.join("journal"))?;
let backend: Arc<dyn Backend> = fake.backend.clone();
if conflict_policy == ConflictPolicy::Enforce && !backend.capabilities().watch {
return Err(Error::unsupported(
backend.capabilities().backend,
"ConflictPolicy::Enforce requires change notifications, which this backend does not support",
));
}
Ok(DnsManager::from_inner(Arc::new(Inner {
owner: owner.to_string(),
backend,
locks,
journal,
conflict_policy,
hook: Mutex::new(None),
suppressions: std::sync::Arc::new(SuppressionRegistry::new()),
active: Mutex::new(HashMap::new()),
lease_tokens: Mutex::new(HashMap::new()),
reconciler: crate::reconciliation::Reconciler::default(),
enforce: Mutex::new(crate::manager::EnforceState::default()),
})))
}
pub fn manager_for_backend(
owner: &str,
state_dir: &Path,
kind: crate::capability::BackendKind,
lock_timeout: Duration,
) -> Result<DnsManager> {
use std::collections::HashMap;
ensure_private_dir(state_dir)?;
let locks = ResourceLockManager::new(state_dir.join("locks"), lock_timeout);
locks.ensure_dir()?;
let journal = JournalStore::open(state_dir.join("journal"))?;
let backend = crate::platform::construct_backend(kind, owner)?;
Ok(DnsManager::from_inner(Arc::new(Inner {
owner: owner.to_string(),
backend,
locks,
journal,
conflict_policy: ConflictPolicy::Cooperative,
hook: Mutex::new(None),
suppressions: std::sync::Arc::new(SuppressionRegistry::new()),
active: Mutex::new(HashMap::new()),
lease_tokens: Mutex::new(HashMap::new()),
reconciler: crate::reconciliation::Reconciler::default(),
enforce: Mutex::new(crate::manager::EnforceState::default()),
})))
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DebugReconcile {
NotOwned,
StillOurs,
Rebased,
Deferred,
Failed,
}
#[derive(Debug, Clone)]
enum FaultSpec {
Crash,
Fail(String),
}
#[derive(Debug, Default)]
pub struct FaultInjector {
actions: Mutex<HashMap<TxPoint, FaultSpec>>,
}
impl FaultInjector {
pub fn new() -> Arc<Self> {
Arc::new(Self::default())
}
pub fn crash_at(&self, point: TxPoint) -> &Self {
self.actions
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.insert(point, FaultSpec::Crash);
self
}
pub fn fail_at(&self, point: TxPoint, message: impl Into<String>) -> &Self {
self.actions
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.insert(point, FaultSpec::Fail(message.into()));
self
}
pub fn disarm(&self, point: TxPoint) {
self.actions
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.remove(&point);
}
pub fn clear(&self) {
self.actions
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.clear();
}
}
impl FaultHook for FaultInjector {
fn on_point(&self, point: TxPoint) -> FaultAction {
match self
.actions
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.get(&point)
{
Some(FaultSpec::Crash) => FaultAction::Crash,
Some(FaultSpec::Fail(message)) => FaultAction::Fail(message.clone()),
None => FaultAction::Continue,
}
}
}
#[derive(Debug)]
pub enum CrashOutcome<T> {
Completed(Result<T>),
Crashed,
}
pub fn catch_crash<T>(f: impl FnOnce() -> Result<T>) -> CrashOutcome<T> {
match catch_unwind(AssertUnwindSafe(f)) {
Ok(result) => CrashOutcome::Completed(result),
Err(payload) => {
if payload.downcast_ref::<CrashSignal>().is_some() {
CrashOutcome::Crashed
} else {
resume_unwind(payload)
}
}
}
}