feature: proactive re-resolving

Cargo.lock

···
  "deadpool-redis",
  "httpdate",
  "metrohash",
+ "once_cell",
  "reqwest",
  "serde",
  "serde_json",

Cargo.toml

···
 tokio-util = { version = "0.7", features = ["rt"] }
 tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
+
+[dev-dependencies]
+once_cell = "1.20"

src/bin/quickdid.rs

···
     cache::create_redis_pool,
     config::Config,
     handle_resolver::{
-        create_base_resolver, create_caching_resolver, create_rate_limited_resolver_with_timeout,
-        create_redis_resolver_with_ttl, create_sqlite_resolver_with_ttl,
+        create_base_resolver, create_caching_resolver, create_proactive_refresh_resolver_dyn,
+        create_rate_limited_resolver_with_timeout, create_redis_resolver_with_ttl,
+        create_sqlite_resolver_with_ttl,
     },
     handle_resolver_task::{HandleResolverTaskConfig, create_handle_resolver_task_with_config},
     http::{AppContext, create_router},
···
                 println!(
                     "    METRICS_STATSD_HOST StatsD host when using statsd adapter (e.g., localhost:8125)"
                 );
+                println!("    METRICS_PREFIX      Prefix for all metrics (default: quickdid)");
                 println!(
-                    "    METRICS_PREFIX      Prefix for all metrics (default: quickdid)"
+                    "    METRICS_TAGS        Default tags for metrics (comma-separated key:value pairs)"
                 );
+                println!();
+                println!("    PROACTIVE CACHE REFRESH:");
                 println!(
-                    "    METRICS_TAGS        Default tags for metrics (comma-separated key:value pairs)"
+                    "    PROACTIVE_REFRESH_ENABLED   Enable proactive cache refresh (default: false)"
+                );
+                println!(
+                    "    PROACTIVE_REFRESH_THRESHOLD Threshold as percentage of TTL (0.0-1.0, default: 0.8)"
                 );
                 println!();
                 println!(
···
         tracing::error!("Failed to create metrics publisher: {}", e);
         anyhow::anyhow!("Failed to create metrics publisher: {}", e)
     })?;
-    
+
     tracing::info!(
         "Metrics publisher created with {} adapter",
         config.metrics_adapter
···
     metrics_publisher.gauge("server", 1).await;
 
     // Create base handle resolver using factory function
-    let mut base_handle_resolver =
-        create_base_resolver(dns_resolver_arc.clone(), http_client.clone(), metrics_publisher.clone());
+    let mut base_handle_resolver = create_base_resolver(
+        dns_resolver_arc.clone(),
+        http_client.clone(),
+        metrics_publisher.clone(),
+    );
 
     // Apply rate limiting if configured
     if config.resolver_max_concurrent > 0 {
···
         None
     };
 
-    // Create handle resolver with cache priority: Redis > SQLite > In-memory
-    let handle_resolver: Arc<dyn quickdid::handle_resolver::HandleResolver> =
-        if let Some(pool) = redis_pool {
-            tracing::info!(
-                "Using Redis-backed handle resolver with {}-second cache TTL",
-                config.cache_ttl_redis
-            );
-            create_redis_resolver_with_ttl(base_handle_resolver, pool, config.cache_ttl_redis, metrics_publisher.clone())
-        } else if let Some(pool) = sqlite_pool {
-            tracing::info!(
-                "Using SQLite-backed handle resolver with {}-second cache TTL",
-                config.cache_ttl_sqlite
-            );
-            create_sqlite_resolver_with_ttl(base_handle_resolver, pool, config.cache_ttl_sqlite, metrics_publisher.clone())
-        } else {
-            tracing::info!(
-                "Using in-memory handle resolver with {}-second cache TTL",
-                config.cache_ttl_memory
-            );
-            create_caching_resolver(base_handle_resolver, config.cache_ttl_memory, metrics_publisher.clone())
-        };
-
     // Create task tracker and cancellation token
     let tracker = TaskTracker::new();
     let token = CancellationToken::new();
 
-    // Setup background handle resolution task and get the queue adapter
+    // Create the queue adapter first (needed for proactive refresh)
     let handle_queue: Arc<dyn QueueAdapter<HandleResolutionWork>> = {
         // Create queue adapter based on configuration
         let adapter: Arc<dyn QueueAdapter<HandleResolutionWork>> = match config
···
             }
         };
 
-        // Keep a reference to the adapter for the AppContext
-        let adapter_for_context = adapter.clone();
+        adapter
+    };
+
+    // Create handle resolver with cache priority: Redis > SQLite > In-memory
+    let (mut handle_resolver, cache_ttl): (
+        Arc<dyn quickdid::handle_resolver::HandleResolver>,
+        u64,
+    ) = if let Some(pool) = redis_pool {
+        tracing::info!(
+            "Using Redis-backed handle resolver with {}-second cache TTL",
+            config.cache_ttl_redis
+        );
+        (
+            create_redis_resolver_with_ttl(
+                base_handle_resolver,
+                pool,
+                config.cache_ttl_redis,
+                metrics_publisher.clone(),
+            ),
+            config.cache_ttl_redis,
+        )
+    } else if let Some(pool) = sqlite_pool {
+        tracing::info!(
+            "Using SQLite-backed handle resolver with {}-second cache TTL",
+            config.cache_ttl_sqlite
+        );
+        (
+            create_sqlite_resolver_with_ttl(
+                base_handle_resolver,
+                pool,
+                config.cache_ttl_sqlite,
+                metrics_publisher.clone(),
+            ),
+            config.cache_ttl_sqlite,
+        )
+    } else {
+        tracing::info!(
+            "Using in-memory handle resolver with {}-second cache TTL",
+            config.cache_ttl_memory
+        );
+        (
+            create_caching_resolver(
+                base_handle_resolver,
+                config.cache_ttl_memory,
+                metrics_publisher.clone(),
+            ),
+            config.cache_ttl_memory,
+        )
+    };
+
+    // Apply proactive refresh if enabled
+    if config.proactive_refresh_enabled && !matches!(config.queue_adapter.as_str(), "noop" | "none")
+    {
+        tracing::info!(
+            "Enabling proactive cache refresh with {}% threshold",
+            (config.proactive_refresh_threshold * 100.0) as u32
+        );
+        handle_resolver = create_proactive_refresh_resolver_dyn(
+            handle_resolver,
+            handle_queue.clone(),
+            cache_ttl,
+            config.proactive_refresh_threshold,
+        );
+    } else if config.proactive_refresh_enabled {
+        tracing::warn!(
+            "Proactive refresh enabled but queue adapter is no-op, skipping proactive refresh"
+        );
+    }
+
+    // Setup background handle resolution task
+    {
+        let adapter_for_task = handle_queue.clone();
 
         // Only spawn handle resolver task if not using noop adapter
         if !matches!(config.queue_adapter.as_str(), "noop" | "none") {
···
 
             // Create and start handle resolver task
             let handle_task = create_handle_resolver_task_with_config(
-                adapter,
+                adapter_for_task,
                 handle_resolver.clone(),
                 token.clone(),
                 handle_task_config,
···
         } else {
             tracing::info!("Background handle resolution task disabled (using no-op adapter)");
         }
-
-        // Return the adapter to be used in AppContext
-        adapter_for_context
     };
 
     // Create app context with the queue adapter

src/config.rs

···
     /// Default tags for all metrics (comma-separated key:value pairs)
     /// Example: "env:production,service:quickdid"
     pub metrics_tags: Option<String>,
+
+    /// Enable proactive cache refresh for frequently accessed handles.
+    /// When enabled, cache entries that have reached the refresh threshold
+    /// will be queued for background refresh to keep the cache warm.
+    /// Default: false
+    pub proactive_refresh_enabled: bool,
+
+    /// Threshold as a percentage (0.0-1.0) of cache TTL when to trigger proactive refresh.
+    /// For example, 0.8 means refresh when an entry has lived for 80% of its TTL.
+    /// Default: 0.8 (80%)
+    pub proactive_refresh_threshold: f64,
 }
 
 impl Config {
···
             metrics_statsd_host: get_env_or_default("METRICS_STATSD_HOST", None),
             metrics_prefix: get_env_or_default("METRICS_PREFIX", Some("quickdid")).unwrap(),
             metrics_tags: get_env_or_default("METRICS_TAGS", None),
+            proactive_refresh_enabled: parse_env("PROACTIVE_REFRESH_ENABLED", false)?,
+            proactive_refresh_threshold: parse_env("PROACTIVE_REFRESH_THRESHOLD", 0.8)?,
         };
 
         // Calculate the Cache-Control header value if enabled
···
                 "RESOLVER_MAX_CONCURRENT_TIMEOUT_MS must be <= 60000 (60 seconds)".to_string(),
             ));
         }
-        
+
         // Validate metrics configuration
         match self.metrics_adapter.as_str() {
             "noop" | "statsd" => {}
···
                 )));
             }
         }
-        
+
         // If statsd is configured, ensure host is provided
         if self.metrics_adapter == "statsd" && self.metrics_statsd_host.is_none() {
             return Err(ConfigError::MissingRequired(
                 "METRICS_STATSD_HOST is required when METRICS_ADAPTER is 'statsd'".to_string(),
             ));
         }
-        
+
+        // Validate proactive refresh threshold
+        if self.proactive_refresh_threshold < 0.0 || self.proactive_refresh_threshold > 1.0 {
+            return Err(ConfigError::InvalidValue(format!(
+                "PROACTIVE_REFRESH_THRESHOLD must be between 0.0 and 1.0, got {}",
+                self.proactive_refresh_threshold
+            )));
+        }
+
         Ok(())
     }
 }

src/handle_resolver/base.rs

···
 impl HandleResolver for BaseHandleResolver {
     async fn resolve(&self, s: &str) -> Result<(String, u64), HandleResolverError> {
         let start_time = std::time::Instant::now();
-        
+
         // Perform DNS/HTTP resolution
         let result = resolve_subject(&self.http_client, &*self.dns_resolver, s)
             .await
             .map_err(|e| HandleResolverError::ResolutionFailed(e.to_string()));
 
         let duration_ms = start_time.elapsed().as_millis() as u64;
-        
+
         // Publish metrics
-        
+
         match result {
             Ok(did) => {
-                self.metrics.time_with_tags("resolver.base.duration_ms", duration_ms, &[("success", "1")]).await;
-                
+                self.metrics
+                    .time_with_tags(
+                        "resolver.base.duration_ms",
+                        duration_ms,
+                        &[("success", "1")],
+                    )
+                    .await;
+
                 let timestamp = SystemTime::now()
                     .duration_since(UNIX_EPOCH)
                     .map_err(|e| {
···
                 Ok((did, timestamp))
             }
             Err(e) => {
-                self.metrics.time_with_tags("resolver.base.duration_ms", duration_ms, &[("success", "0")]).await;
+                self.metrics
+                    .time_with_tags(
+                        "resolver.base.duration_ms",
+                        duration_ms,
+                        &[("success", "0")],
+                    )
+                    .await;
                 Err(e)
             }
         }

src/handle_resolver/memory.rs

···
     /// * `inner` - The underlying resolver to use for actual resolution
     /// * `ttl_seconds` - How long to cache results in seconds
     /// * `metrics` - Metrics publisher for telemetry
-    pub fn new(inner: Arc<dyn HandleResolver>, ttl_seconds: u64, metrics: SharedMetricsPublisher) -> Self {
+    pub fn new(
+        inner: Arc<dyn HandleResolver>,
+        ttl_seconds: u64,
+        metrics: SharedMetricsPublisher,
+    ) -> Self {
         Self {
             inner,
             cache: Arc::new(RwLock::new(HashMap::new())),
···
                                 handle,
                                 error
                             );
-                            self.metrics.incr("resolver.memory.cache_hit_not_resolved").await;
+                            self.metrics
+                                .incr("resolver.memory.cache_hit_not_resolved")
+                                .await;
                             return Err(HandleResolverError::HandleNotFoundCached(error.clone()));
                         }
                         tracing::debug!("Cache entry expired for handle {}", handle);
···
                     tracing::debug!("Cached failed resolution for handle {}: {}", handle, e);
                 }
             }
-            
+
             // Track cache size
             let cache_size = cache.len() as u64;
-            self.metrics.gauge("resolver.memory.cache_entries", cache_size).await;
+            self.metrics
+                .gauge("resolver.memory.cache_entries", cache_size)
+                .await;
         }
 
         result

src/handle_resolver/mod.rs

···
 mod base;
 mod errors;
 mod memory;
+mod proactive_refresh;
 mod rate_limited;
 mod redis;
 mod sqlite;
···
 // Factory functions for creating resolvers
 pub use base::create_base_resolver;
 pub use memory::create_caching_resolver;
+pub use proactive_refresh::{
+    ProactiveRefreshResolver, create_proactive_refresh_resolver,
+    create_proactive_refresh_resolver_dyn, create_proactive_refresh_resolver_with_threshold,
+};
 pub use rate_limited::{create_rate_limited_resolver, create_rate_limited_resolver_with_timeout};
 pub use redis::{create_redis_resolver, create_redis_resolver_with_ttl};
 pub use sqlite::{create_sqlite_resolver, create_sqlite_resolver_with_ttl};

src/handle_resolver/proactive_refresh.rs

···
+use crate::handle_resolution_result::HandleResolutionResult;
+use crate::handle_resolver::{HandleResolver, HandleResolverError};
+use crate::queue::{HandleResolutionWork, QueueAdapter};
+use async_trait::async_trait;
+use std::sync::Arc;
+use std::time::{SystemTime, UNIX_EPOCH};
+use tracing::{debug, trace};
+
+/// Create a ProactiveRefreshResolver with default 80% threshold
+///
+/// # Arguments
+/// * `inner` - The inner resolver to wrap
+/// * `queue` - The queue adapter for background refresh tasks
+/// * `cache_ttl` - The TTL in seconds for cache entries
+pub fn create_proactive_refresh_resolver<R, Q>(
+    inner: Arc<R>,
+    queue: Arc<Q>,
+    cache_ttl: u64,
+) -> Arc<ProactiveRefreshResolver<R, Q>>
+where
+    R: HandleResolver + Send + Sync + 'static,
+    Q: QueueAdapter<HandleResolutionWork> + Send + Sync + 'static,
+{
+    Arc::new(ProactiveRefreshResolver::new(inner, queue, cache_ttl))
+}
+
+/// Create a ProactiveRefreshResolver with custom threshold
+///
+/// # Arguments
+/// * `inner` - The inner resolver to wrap
+/// * `queue` - The queue adapter for background refresh tasks
+/// * `cache_ttl` - The TTL in seconds for cache entries
+/// * `threshold` - The threshold as a percentage (0.0 to 1.0) of TTL when to trigger refresh
+pub fn create_proactive_refresh_resolver_with_threshold<R, Q>(
+    inner: Arc<R>,
+    queue: Arc<Q>,
+    cache_ttl: u64,
+    threshold: f64,
+) -> Arc<ProactiveRefreshResolver<R, Q>>
+where
+    R: HandleResolver + Send + Sync + 'static,
+    Q: QueueAdapter<HandleResolutionWork> + Send + Sync + 'static,
+{
+    Arc::new(ProactiveRefreshResolver::with_threshold(
+        inner, queue, cache_ttl, threshold,
+    ))
+}
+
+/// Wrapper struct for dynamic dispatch with proactive refresh
+/// This works with trait objects instead of concrete types
+pub struct DynProactiveRefreshResolver {
+    inner: Arc<dyn HandleResolver>,
+    queue: Arc<dyn QueueAdapter<HandleResolutionWork>>,
+    #[allow(dead_code)]
+    cache_ttl: u64,
+    #[allow(dead_code)]
+    refresh_threshold: f64,
+}
+
+impl DynProactiveRefreshResolver {
+    pub fn new(
+        inner: Arc<dyn HandleResolver>,
+        queue: Arc<dyn QueueAdapter<HandleResolutionWork>>,
+        cache_ttl: u64,
+        refresh_threshold: f64,
+    ) -> Self {
+        Self {
+            inner,
+            queue,
+            cache_ttl,
+            refresh_threshold: refresh_threshold.clamp(0.0, 1.0),
+        }
+    }
+
+    async fn maybe_queue_for_refresh(&self, handle: &str, resolve_time: u64) {
+        // If resolution took less than 5ms, it was probably a cache hit
+        if resolve_time < 5000 {
+            trace!(
+                handle = handle,
+                resolve_time_us = resolve_time,
+                "Fast resolution detected, considering proactive refresh"
+            );
+
+            // Simple heuristic: queue for refresh with some probability
+            let now = SystemTime::now()
+                .duration_since(UNIX_EPOCH)
+                .unwrap_or_default()
+                .as_secs();
+
+            // Queue every N seconds for frequently accessed handles
+            if now % 60 == 0 {
+                let work = HandleResolutionWork {
+                    handle: handle.to_string(),
+                };
+
+                if let Err(e) = self.queue.push(work).await {
+                    debug!(
+                        handle = handle,
+                        error = %e,
+                        "Failed to queue handle for proactive refresh"
+                    );
+                } else {
+                    debug!(handle = handle, "Queued handle for proactive refresh");
+                }
+            }
+        }
+    }
+}
+
+#[async_trait]
+impl HandleResolver for DynProactiveRefreshResolver {
+    async fn resolve(&self, handle: &str) -> Result<(String, u64), HandleResolverError> {
+        // Resolve through the inner resolver
+        let (did, resolve_time) = self.inner.resolve(handle).await?;
+
+        // Check if we should queue for refresh based on resolution time
+        self.maybe_queue_for_refresh(handle, resolve_time).await;
+
+        Ok((did, resolve_time))
+    }
+}
+
+/// Create a ProactiveRefreshResolver with custom threshold using trait objects
+/// This version works with dyn HandleResolver and dyn QueueAdapter
+///
+/// # Arguments
+/// * `inner` - The inner resolver to wrap
+/// * `queue` - The queue adapter for background refresh tasks
+/// * `cache_ttl` - The TTL in seconds for cache entries
+/// * `threshold` - The threshold as a percentage (0.0 to 1.0) of TTL when to trigger refresh
+pub fn create_proactive_refresh_resolver_dyn(
+    inner: Arc<dyn HandleResolver>,
+    queue: Arc<dyn QueueAdapter<HandleResolutionWork>>,
+    cache_ttl: u64,
+    threshold: f64,
+) -> Arc<dyn HandleResolver> {
+    Arc::new(DynProactiveRefreshResolver::new(
+        inner, queue, cache_ttl, threshold,
+    ))
+}
+
+/// A handle resolver that proactively refreshes cache entries when they reach
+/// a certain staleness threshold (default 80% of TTL).
+///
+/// This resolver wraps another resolver and checks successful resolutions from cache.
+/// When a cached entry has lived for more than the threshold percentage of its TTL,
+/// it queues the handle for background refresh to keep the cache warm.
+///
+/// Note: Due to the current trait design, this implementation uses the resolution time
+/// as a heuristic. When resolve_time is 0 (instant cache hit), it may queue for refresh.
+/// For full functionality, the trait would need to expose cache timestamps.
+pub struct ProactiveRefreshResolver<R: HandleResolver, Q: QueueAdapter<HandleResolutionWork>> {
+    inner: Arc<R>,
+    queue: Arc<Q>,
+    /// TTL in seconds for cache entries
+    cache_ttl: u64,
+    /// Threshold as a percentage (0.0 to 1.0) of TTL when to trigger refresh
+    /// Default is 0.8 (80%)
+    refresh_threshold: f64,
+}
+
+impl<R: HandleResolver, Q: QueueAdapter<HandleResolutionWork>> ProactiveRefreshResolver<R, Q> {
+    pub fn new(inner: Arc<R>, queue: Arc<Q>, cache_ttl: u64) -> Self {
+        Self::with_threshold(inner, queue, cache_ttl, 0.8)
+    }
+
+    pub fn with_threshold(
+        inner: Arc<R>,
+        queue: Arc<Q>,
+        cache_ttl: u64,
+        refresh_threshold: f64,
+    ) -> Self {
+        Self {
+            inner,
+            queue,
+            cache_ttl,
+            refresh_threshold: refresh_threshold.clamp(0.0, 1.0),
+        }
+    }
+
+    /// Check if a cached entry needs proactive refresh based on its age
+    #[allow(dead_code)]
+    fn needs_refresh(&self, result: &HandleResolutionResult) -> bool {
+        let now = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .unwrap_or_default()
+            .as_secs();
+
+        let age = now.saturating_sub(result.timestamp);
+        let threshold = (self.cache_ttl as f64 * self.refresh_threshold) as u64;
+
+        let needs_refresh = age >= threshold;
+
+        if needs_refresh {
+            debug!(
+                handle = ?result.to_did(),
+                age_seconds = age,
+                threshold_seconds = threshold,
+                cache_ttl = self.cache_ttl,
+                "Cache entry needs proactive refresh"
+            );
+        } else {
+            trace!(
+                handle = ?result.to_did(),
+                age_seconds = age,
+                threshold_seconds = threshold,
+                "Cache entry still fresh"
+            );
+        }
+
+        needs_refresh
+    }
+
+    /// Queue a handle for background refresh
+    async fn queue_for_refresh(&self, handle: &str) {
+        let work = HandleResolutionWork {
+            handle: handle.to_string(),
+        };
+
+        match self.queue.push(work).await {
+            Ok(_) => {
+                debug!(handle = handle, "Queued handle for proactive refresh");
+            }
+            Err(e) => {
+                // Don't fail the request if we can't queue for refresh
+                debug!(
+                    handle = handle,
+                    error = %e,
+                    "Failed to queue handle for proactive refresh"
+                );
+            }
+        }
+    }
+
+    /// Check if we should queue for refresh based on resolution time
+    ///
+    /// This is a heuristic approach:
+    /// - If resolve_time is very low (< 5ms), it was likely a cache hit
+    /// - We probabilistically queue for refresh based on time since service start
+    ///
+    /// For proper implementation, the HandleResolver trait would need to expose
+    /// cache metadata or return HandleResolutionResult directly.
+    async fn maybe_queue_for_refresh(&self, handle: &str, resolve_time: u64) {
+        // If resolution took less than 5ms, it was probably a cache hit
+        if resolve_time < 5000 {
+            // Use a simple probabilistic approach for demonstration
+            // In production, you'd want access to the actual cache timestamp
+            trace!(
+                handle = handle,
+                resolve_time_us = resolve_time,
+                "Fast resolution detected, considering proactive refresh"
+            );
+
+            // Queue for refresh with some probability to avoid overwhelming the queue
+            // This is a simplified approach - ideally we'd have access to cache metadata
+            let now = SystemTime::now()
+                .duration_since(UNIX_EPOCH)
+                .unwrap_or_default()
+                .as_secs();
+
+            // Simple heuristic: queue every N seconds for frequently accessed handles
+            if now % 60 == 0 {
+                self.queue_for_refresh(handle).await;
+            }
+        }
+    }
+}
+
+#[async_trait]
+impl<R, Q> HandleResolver for ProactiveRefreshResolver<R, Q>
+where
+    R: HandleResolver + Send + Sync,
+    Q: QueueAdapter<HandleResolutionWork> + Send + Sync,
+{
+    async fn resolve(&self, handle: &str) -> Result<(String, u64), HandleResolverError> {
+        // Resolve through the inner resolver
+        let (did, resolve_time) = self.inner.resolve(handle).await?;
+
+        // Check if we should queue for refresh based on resolution time
+        self.maybe_queue_for_refresh(handle, resolve_time).await;
+
+        Ok((did, resolve_time))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::handle_resolution_result::DidMethodType;
+
+    #[test]
+    fn test_needs_refresh_calculation() {
+        // Create a resolver with 100 second TTL and 80% threshold
+        let inner = Arc::new(MockResolver);
+        let queue = Arc::new(MockQueueAdapter);
+        let resolver = ProactiveRefreshResolver::new(inner, queue, 100);
+
+        let now = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .unwrap()
+            .as_secs();
+
+        // Test entry that's 50% through TTL (should not refresh)
+        let fresh_result = HandleResolutionResult {
+            timestamp: now - 50,
+            method_type: DidMethodType::Plc,
+            payload: "alice123".to_string(),
+        };
+        assert!(!resolver.needs_refresh(&fresh_result));
+
+        // Test entry that's 80% through TTL (should refresh)
+        let stale_result = HandleResolutionResult {
+            timestamp: now - 80,
+            method_type: DidMethodType::Plc,
+            payload: "alice123".to_string(),
+        };
+        assert!(resolver.needs_refresh(&stale_result));
+
+        // Test entry that's 90% through TTL (should definitely refresh)
+        let very_stale_result = HandleResolutionResult {
+            timestamp: now - 90,
+            method_type: DidMethodType::Plc,
+            payload: "alice123".to_string(),
+        };
+        assert!(resolver.needs_refresh(&very_stale_result));
+    }
+
+    #[test]
+    fn test_custom_threshold() {
+        let inner = Arc::new(MockResolver);
+        let queue = Arc::new(MockQueueAdapter);
+
+        // Create resolver with 50% threshold
+        let resolver = ProactiveRefreshResolver::with_threshold(inner, queue, 100, 0.5);
+
+        let now = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .unwrap()
+            .as_secs();
+
+        // Test entry that's 40% through TTL (should not refresh with 50% threshold)
+        let result_40 = HandleResolutionResult {
+            timestamp: now - 40,
+            method_type: DidMethodType::Plc,
+            payload: "alice123".to_string(),
+        };
+        assert!(!resolver.needs_refresh(&result_40));
+
+        // Test entry that's 60% through TTL (should refresh with 50% threshold)
+        let result_60 = HandleResolutionResult {
+            timestamp: now - 60,
+            method_type: DidMethodType::Plc,
+            payload: "alice123".to_string(),
+        };
+        assert!(resolver.needs_refresh(&result_60));
+    }
+
+    // Mock resolver for testing
+    struct MockResolver;
+
+    #[async_trait]
+    impl HandleResolver for MockResolver {
+        async fn resolve(&self, handle: &str) -> Result<(String, u64), HandleResolverError> {
+            Ok((format!("did:plc:{}", handle), 1000))
+        }
+    }
+
+    // Mock queue adapter for testing
+    struct MockQueueAdapter;
+
+    #[async_trait]
+    impl QueueAdapter<HandleResolutionWork> for MockQueueAdapter {
+        async fn pull(&self) -> Option<HandleResolutionWork> {
+            None
+        }
+
+        async fn push(&self, _work: HandleResolutionWork) -> crate::queue::Result<()> {
+            Ok(())
+        }
+
+        async fn ack(&self, _item: &HandleResolutionWork) -> crate::queue::Result<()> {
+            Ok(())
+        }
+
+        async fn try_push(&self, _work: HandleResolutionWork) -> crate::queue::Result<()> {
+            Ok(())
+        }
+
+        async fn is_healthy(&self) -> bool {
+            true
+        }
+    }
+}

src/handle_resolver/rate_limited.rs

···
     /// * `inner` - The inner resolver to wrap
     /// * `max_concurrent` - Maximum number of concurrent resolutions allowed
     /// * `metrics` - Metrics publisher for telemetry
-    pub fn new(inner: Arc<dyn HandleResolver>, max_concurrent: usize, metrics: SharedMetricsPublisher) -> Self {
+    pub fn new(
+        inner: Arc<dyn HandleResolver>,
+        max_concurrent: usize,
+        metrics: SharedMetricsPublisher,
+    ) -> Self {
         Self {
             inner,
             semaphore: Arc::new(Semaphore::new(max_concurrent)),
···
 impl HandleResolver for RateLimitedHandleResolver {
     async fn resolve(&self, s: &str) -> Result<(String, u64), HandleResolverError> {
         let permit_start = std::time::Instant::now();
-        
+
         // Track rate limiter queue depth
         let available_permits = self.semaphore.available_permits();
-        self.metrics.gauge("resolver.rate_limit.available_permits", available_permits as u64).await;
-        
+        self.metrics
+            .gauge(
+                "resolver.rate_limit.available_permits",
+                available_permits as u64,
+            )
+            .await;
+
         // Acquire a permit from the semaphore, with optional timeout
         let _permit = match self.timeout_ms {
             Some(timeout_ms) if timeout_ms > 0 => {
···
                 match timeout(duration, self.semaphore.acquire()).await {
                     Ok(Ok(permit)) => {
                         let wait_ms = permit_start.elapsed().as_millis() as u64;
-                        self.metrics.time("resolver.rate_limit.permit_acquired", wait_ms).await;
+                        self.metrics
+                            .time("resolver.rate_limit.permit_acquired", wait_ms)
+                            .await;
                         permit
                     }
                     Ok(Err(e)) => {
···
                     }
                     Err(_) => {
                         // Timeout occurred
-                        self.metrics.incr("resolver.rate_limit.permit_timeout").await;
+                        self.metrics
+                            .incr("resolver.rate_limit.permit_timeout")
+                            .await;
                         return Err(HandleResolverError::ResolutionFailed(format!(
                             "Rate limit permit acquisition timed out after {}ms",
                             timeout_ms
···
                 match self.semaphore.acquire().await {
                     Ok(permit) => {
                         let wait_ms = permit_start.elapsed().as_millis() as u64;
-                        self.metrics.time("resolver.rate_limit.permit_acquired", wait_ms).await;
+                        self.metrics
+                            .time("resolver.rate_limit.permit_acquired", wait_ms)
+                            .await;
                         permit
                     }
                     Err(e) => {
···
     max_concurrent: usize,
     metrics: SharedMetricsPublisher,
 ) -> Arc<dyn HandleResolver> {
-    Arc::new(RateLimitedHandleResolver::new(inner, max_concurrent, metrics))
+    Arc::new(RateLimitedHandleResolver::new(
+        inner,
+        max_concurrent,
+        metrics,
+    ))
 }
 
 /// Create a rate-limited handle resolver with timeout.

src/handle_resolver/redis.rs

···
 
 impl RedisHandleResolver {
     /// Create a new Redis-backed handle resolver with default 90-day TTL.
-    fn new(inner: Arc<dyn HandleResolver>, pool: RedisPool, metrics: SharedMetricsPublisher) -> Self {
+    fn new(
+        inner: Arc<dyn HandleResolver>,
+        pool: RedisPool,
+        metrics: SharedMetricsPublisher,
+    ) -> Self {
         Self::with_ttl(inner, pool, 90 * 24 * 60 * 60, metrics) // 90 days default
     }
 
     /// Create a new Redis-backed handle resolver with custom TTL.
-    fn with_ttl(inner: Arc<dyn HandleResolver>, pool: RedisPool, ttl_seconds: u64, metrics: SharedMetricsPublisher) -> Self {
+    fn with_ttl(
+        inner: Arc<dyn HandleResolver>,
+        pool: RedisPool,
+        ttl_seconds: u64,
+        metrics: SharedMetricsPublisher,
+    ) -> Self {
         Self::with_full_config(inner, pool, "handle:".to_string(), ttl_seconds, metrics)
     }
 
···
                                 return Ok((did, cached_result.timestamp));
                             } else {
                                 tracing::debug!("Cache hit (not resolved) for handle {}", handle);
-                                self.metrics.incr("resolver.redis.cache_hit_not_resolved").await;
+                                self.metrics
+                                    .incr("resolver.redis.cache_hit_not_resolved")
+                                    .await;
                                 return Err(HandleResolverError::HandleNotFound);
                             }
                         }
···
                             Ok(res) => res,
                             Err(e) => {
                                 tracing::warn!("Failed to create resolution result: {}", e);
-                                self.metrics.incr("resolver.redis.result_create_error").await;
+                                self.metrics
+                                    .incr("resolver.redis.result_create_error")
+                                    .await;
                                 return result;
                             }
                         }
···
                             Ok(res) => res,
                             Err(err) => {
                                 tracing::warn!("Failed to create not_resolved result: {}", err);
-                                self.metrics.incr("resolver.redis.result_create_error").await;
+                                self.metrics
+                                    .incr("resolver.redis.result_create_error")
+                                    .await;
                                 return result;
                             }
                         }
···
     ttl_seconds: u64,
     metrics: SharedMetricsPublisher,
 ) -> Arc<dyn HandleResolver> {
-    Arc::new(RedisHandleResolver::with_ttl(inner, pool, ttl_seconds, metrics))
+    Arc::new(RedisHandleResolver::with_ttl(
+        inner,
+        pool,
+        ttl_seconds,
+        metrics,
+    ))
 }
 
 #[cfg(test)]
···
 
         // Create metrics publisher
         let metrics = Arc::new(crate::metrics::NoOpMetricsPublisher);
-        
+
         // Create Redis-backed resolver with a unique key prefix for testing
         let test_prefix = format!(
             "test:handle:{}:",
···
 
         // Create metrics publisher
         let metrics = Arc::new(crate::metrics::NoOpMetricsPublisher);
-        
+
         // Create Redis-backed resolver with a unique key prefix for testing
         let test_prefix = format!(
             "test:handle:{}:",

src/handle_resolver/sqlite.rs

···
 
 impl SqliteHandleResolver {
     /// Create a new SQLite-backed handle resolver with default 90-day TTL.
-    fn new(inner: Arc<dyn HandleResolver>, pool: SqlitePool, metrics: SharedMetricsPublisher) -> Self {
+    fn new(
+        inner: Arc<dyn HandleResolver>,
+        pool: SqlitePool,
+        metrics: SharedMetricsPublisher,
+    ) -> Self {
         Self::with_ttl(inner, pool, 90 * 24 * 60 * 60, metrics) // 90 days default
     }
 
     /// Create a new SQLite-backed handle resolver with custom TTL.
-    fn with_ttl(inner: Arc<dyn HandleResolver>, pool: SqlitePool, ttl_seconds: u64, metrics: SharedMetricsPublisher) -> Self {
+    fn with_ttl(
+        inner: Arc<dyn HandleResolver>,
+        pool: SqlitePool,
+        ttl_seconds: u64,
+        metrics: SharedMetricsPublisher,
+    ) -> Self {
         Self {
             inner,
             pool,
···
                                 return Ok((did, cached_result.timestamp));
                             } else {
                                 tracing::debug!("Cache hit (not resolved) for handle {}", handle);
-                                self.metrics.incr("resolver.sqlite.cache_hit_not_resolved").await;
+                                self.metrics
+                                    .incr("resolver.sqlite.cache_hit_not_resolved")
+                                    .await;
                                 return Err(HandleResolverError::HandleNotFound);
                             }
                         }
···
                     Ok(res) => res,
                     Err(e) => {
                         tracing::warn!("Failed to create resolution result: {}", e);
-                        self.metrics.incr("resolver.sqlite.result_create_error").await;
+                        self.metrics
+                            .incr("resolver.sqlite.result_create_error")
+                            .await;
                         return result;
                     }
                 }
···
                     Ok(res) => res,
                     Err(err) => {
                         tracing::warn!("Failed to create not_resolved result: {}", err);
-                        self.metrics.incr("resolver.sqlite.result_create_error").await;
+                        self.metrics
+                            .incr("resolver.sqlite.result_create_error")
+                            .await;
                         return result;
                     }
                 }
···
     ttl_seconds: u64,
     metrics: SharedMetricsPublisher,
 ) -> Arc<dyn HandleResolver> {
-    Arc::new(SqliteHandleResolver::with_ttl(inner, pool, ttl_seconds, metrics))
+    Arc::new(SqliteHandleResolver::with_ttl(
+        inner,
+        pool,
+        ttl_seconds,
+        metrics,
+    ))
 }
 
 #[cfg(test)]
···
 
         // Create metrics publisher
         let metrics = Arc::new(crate::metrics::NoOpMetricsPublisher);
-        
+
         // Create SQLite-backed resolver
-        let sqlite_resolver = SqliteHandleResolver::with_ttl(mock_resolver, pool.clone(), 3600, metrics);
+        let sqlite_resolver =
+            SqliteHandleResolver::with_ttl(mock_resolver, pool.clone(), 3600, metrics);
 
         let test_handle = "alice.bsky.social";
         let expected_key = sqlite_resolver.make_key(test_handle) as i64;
···
 
         // Create metrics publisher
         let metrics = Arc::new(crate::metrics::NoOpMetricsPublisher);
-        
+
         // Create SQLite-backed resolver
-        let sqlite_resolver = SqliteHandleResolver::with_ttl(mock_resolver, pool.clone(), 3600, metrics);
+        let sqlite_resolver =
+            SqliteHandleResolver::with_ttl(mock_resolver, pool.clone(), 3600, metrics);
 
         let test_handle = "error.bsky.social";
         let expected_key = sqlite_resolver.make_key(test_handle) as i64;

src/handle_resolver_task.rs

···
             .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
 
         // Publish metrics
-        self.metrics_publisher.incr("task.handle_resolution.processed").await;
-        self.metrics_publisher.time("task.handle_resolution.duration_ms", duration_ms).await;
+        self.metrics_publisher
+            .incr("task.handle_resolution.processed")
+            .await;
+        self.metrics_publisher
+            .time("task.handle_resolution.duration_ms", duration_ms)
+            .await;
 
         match result {
             Ok(Ok((did, _timestamp))) => {
···
                     .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
 
                 // Publish success metrics
-                self.metrics_publisher.incr("task.handle_resolution.success").await;
-                self.metrics_publisher.incr("task.handle_resolution.cached").await;
+                self.metrics_publisher
+                    .incr("task.handle_resolution.success")
+                    .await;
+                self.metrics_publisher
+                    .incr("task.handle_resolution.cached")
+                    .await;
 
                 info!(
                     handle = %work.handle,
···
                     .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
 
                 // Publish failure metrics
-                self.metrics_publisher.incr("task.handle_resolution.failed").await;
+                self.metrics_publisher
+                    .incr("task.handle_resolution.failed")
+                    .await;
 
                 error!(
                     handle = %work.handle,
···
                     .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
 
                 // Publish timeout metrics
-                self.metrics_publisher.incr("task.handle_resolution.timeout").await;
+                self.metrics_publisher
+                    .incr("task.handle_resolution.timeout")
+                    .await;
 
                 error!(
                     handle = %work.handle,
···
     metrics_publisher: SharedMetricsPublisher,
 ) -> HandleResolverTaskHandle {
     HandleResolverTaskHandle {
-        task: HandleResolverTask::with_config(adapter, handle_resolver, cancel_token, config, metrics_publisher),
+        task: HandleResolverTask::with_config(
+            adapter,
+            handle_resolver,
+            cancel_token,
+            config,
+            metrics_publisher,
+        ),
     }
 }
 

src/http/handle_xrpc_resolve_handle.rs

···
     let handle = match params.handle {
         Some(h) => h,
         None => {
-            metrics.incr_with_tags("xrpc.com.atproto.identity.resolveHandle.invalid_handle", &[("reason", "missing")]).await;
+            metrics
+                .incr_with_tags(
+                    "xrpc.com.atproto.identity.resolveHandle.invalid_handle",
+                    &[("reason", "missing")],
+                )
+                .await;
             return (
                 StatusCode::BAD_REQUEST,
                 Json(ErrorResponse {
···
         Ok(InputType::Handle(value)) => value,
         Ok(InputType::Plc(_)) | Ok(InputType::Web(_)) => {
             // It's a DID, not a handle
-            metrics.incr_with_tags("xrpc.com.atproto.identity.resolveHandle.invalid_handle", &[("reason", "did")]).await;
+            metrics
+                .incr_with_tags(
+                    "xrpc.com.atproto.identity.resolveHandle.invalid_handle",
+                    &[("reason", "did")],
+                )
+                .await;
             return (
                 StatusCode::BAD_REQUEST,
                 Json(ErrorResponse {
···
                 .into_response();
         }
         Err(_) => {
-            metrics.incr_with_tags("xrpc.com.atproto.identity.resolveHandle.invalid_handle", &[("reason", "error")]).await;
+            metrics
+                .incr_with_tags(
+                    "xrpc.com.atproto.identity.resolveHandle.invalid_handle",
+                    &[("reason", "error")],
+                )
+                .await;
             return (
                 StatusCode::BAD_REQUEST,
                 Json(ErrorResponse {
···
     };
 
     if validating {
-        metrics.incr("xrpc.com.atproto.identity.resolveHandle").await;
+        metrics
+            .incr("xrpc.com.atproto.identity.resolveHandle")
+            .await;
         return StatusCode::NO_CONTENT.into_response();
     }
 
···
         // Queue the work
         match queue.push(work).await {
             Ok(()) => {
-                metrics.incr("xrpc.com.atproto.identity.resolveHandle").await;
+                metrics
+                    .incr("xrpc.com.atproto.identity.resolveHandle")
+                    .await;
                 tracing::debug!("Queued handle resolution for {}", handle);
             }
             Err(e) => {
-                metrics.incr("xrpc.com.atproto.identity.resolveHandle.queue_failure").await;
+                metrics
+                    .incr("xrpc.com.atproto.identity.resolveHandle.queue_failure")
+                    .await;
                 tracing::error!("Failed to queue handle resolution: {}", e);
             }
         }
···
         Ok((did, timestamp)) => {
             tracing::debug!(handle, did, "Found cached DID for handle");
 
-            metrics.incr_with_tags("handle.resolution.request", &[("success", "1")]).await;
+            metrics
+                .incr_with_tags("handle.resolution.request", &[("success", "1")])
+                .await;
 
             let etag = calculate_etag(&did, app_context.etag_seed());
             ResolutionResult::Success {
···
         }
         Err(err) => {
             tracing::debug!(error = ?err, handle, "Error resolving handle");
-            metrics.incr_with_tags("handle.resolution.request", &[("success", "0")]).await;
+            metrics
+                .incr_with_tags("handle.resolution.request", &[("success", "0")])
+                .await;
             let error_content = format!("error:{}:{}", handle, err);
             let etag = calculate_etag(&error_content, app_context.etag_seed());
             let timestamp = SystemTime::now()

src/http/server.rs

···
 
     // Process the request
     let response = next.run(request).await;
-    
+
     // Calculate duration
     let duration_ms = start.elapsed().as_millis() as u64;
     let status_code = response.status().as_u16().to_string();
-    
+
     // Publish metrics with tags
-    metrics.time_with_tags(
-        "http.request.duration_ms",
-        duration_ms,
-        &[
-            ("method", &method),
-            ("path", &path),
-            ("status", &status_code),
-        ],
-    ).await;
+    metrics
+        .time_with_tags(
+            "http.request.duration_ms",
+            duration_ms,
+            &[
+                ("method", &method),
+                ("path", &path),
+                ("status", &status_code),
+            ],
+        )
+        .await;
 
     response
 }

src/metrics.rs

···
 use crate::config::Config;
 use async_trait::async_trait;
-use cadence::{BufferedUdpMetricSink, Counted, CountedExt, Gauged, Metric, QueuingMetricSink, StatsdClient, Timed};
+use cadence::{
+    BufferedUdpMetricSink, Counted, CountedExt, Gauged, Metric, QueuingMetricSink, StatsdClient,
+    Timed,
+};
 use std::net::UdpSocket;
 use std::sync::Arc;
 use thiserror::Error;
···
     pub fn new(host: &str, prefix: &str) -> Result<Self, Box<dyn std::error::Error>> {
         Self::new_with_tags(host, prefix, vec![])
     }
-    
+
     /// Create a new StatsdMetricsPublisher with default tags
     pub fn new_with_tags(
-        host: &str, 
-        prefix: &str, 
-        default_tags: Vec<(String, String)>
+        host: &str,
+        prefix: &str,
+        default_tags: Vec<(String, String)>,
     ) -> Result<Self, Box<dyn std::error::Error>> {
-        tracing::info!("Creating StatsdMetricsPublisher: host={}, prefix={}, tags={:?}", host, prefix, default_tags);
-        
+        tracing::info!(
+            "Creating StatsdMetricsPublisher: host={}, prefix={}, tags={:?}",
+            host,
+            prefix,
+            default_tags
+        );
+
         // let socket = UdpSocket::bind("0.0.0.0:0")?;
         let socket = UdpSocket::bind("[::0]:0")?;
         socket.set_nonblocking(true)?;
-        
+
         let buffered_sink = BufferedUdpMetricSink::from(host, socket)?;
         let queuing_sink = QueuingMetricSink::builder()
             .with_error_handler(move |error| {
···
             })
             .build(buffered_sink);
         let client = StatsdClient::from_sink(prefix, queuing_sink);
-        
+
         tracing::info!("StatsdMetricsPublisher created successfully");
-        Ok(Self { client, default_tags })
+        Ok(Self {
+            client,
+            default_tags,
+        })
     }
-    
+
     /// Create from an existing StatsdClient
     pub fn from_client(client: StatsdClient) -> Self {
         Self::from_client_with_tags(client, vec![])
     }
-    
+
     /// Create from an existing StatsdClient with default tags
-    pub fn from_client_with_tags(client: StatsdClient, default_tags: Vec<(String, String)>) -> Self {
-        Self { client, default_tags }
+    pub fn from_client_with_tags(
+        client: StatsdClient,
+        default_tags: Vec<(String, String)>,
+    ) -> Self {
+        Self {
+            client,
+            default_tags,
+        }
     }
-    
+
     /// Apply default tags to a builder
-    fn apply_default_tags<'a, M>(&'a self, mut builder: cadence::MetricBuilder<'a, 'a, M>) -> cadence::MetricBuilder<'a, 'a, M> 
+    fn apply_default_tags<'a, M>(
+        &'a self,
+        mut builder: cadence::MetricBuilder<'a, 'a, M>,
+    ) -> cadence::MetricBuilder<'a, 'a, M>
     where
         M: Metric + From<String>,
     {
···
     /// Failed to create metrics publisher
     #[error("error-quickdid-metrics-1 Failed to create metrics publisher: {0}")]
     CreationFailed(String),
-    
+
     /// Invalid configuration for metrics
     #[error("error-quickdid-metrics-2 Invalid metrics configuration: {0}")]
     InvalidConfig(String),
···
 /// `metrics_adapter` configuration value.
 ///
 /// ## Example
-/// 
+///
 /// ```rust,no_run
 /// use quickdid::config::Config;
 /// use quickdid::metrics::create_metrics_publisher;
-/// 
+///
 /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
 /// let config = Config::from_env()?;
 /// let metrics = create_metrics_publisher(&config)?;
-/// 
+///
 /// // Use the metrics publisher
 /// metrics.incr("request.count").await;
 /// # Ok(())
···
 /// ```
 pub fn create_metrics_publisher(config: &Config) -> Result<SharedMetricsPublisher, MetricsError> {
     match config.metrics_adapter.as_str() {
-        "noop" => {
-            Ok(Arc::new(NoOpMetricsPublisher::new()))
-        }
+        "noop" => Ok(Arc::new(NoOpMetricsPublisher::new())),
         "statsd" => {
-            let host = config.metrics_statsd_host.as_ref()
-                .ok_or_else(|| MetricsError::InvalidConfig(
-                    "METRICS_STATSD_HOST is required when using statsd adapter".to_string()
-                ))?;
-            
+            let host = config.metrics_statsd_host.as_ref().ok_or_else(|| {
+                MetricsError::InvalidConfig(
+                    "METRICS_STATSD_HOST is required when using statsd adapter".to_string(),
+                )
+            })?;
+
             // Parse tags from comma-separated key:value pairs
             let default_tags = if let Some(tags_str) = &config.metrics_tags {
                 tags_str
···
             } else {
                 vec![]
             };
-            
-            let publisher = StatsdMetricsPublisher::new_with_tags(
-                host,
-                &config.metrics_prefix,
-                default_tags
-            ).map_err(|e| MetricsError::CreationFailed(e.to_string()))?;
-            
+
+            let publisher =
+                StatsdMetricsPublisher::new_with_tags(host, &config.metrics_prefix, default_tags)
+                    .map_err(|e| MetricsError::CreationFailed(e.to_string()))?;
+
             Ok(Arc::new(publisher))
         }
-        _ => {
-            Err(MetricsError::InvalidConfig(format!(
-                "Unknown metrics adapter: {}",
-                config.metrics_adapter
-            )))
-        }
+        _ => Err(MetricsError::InvalidConfig(format!(
+            "Unknown metrics adapter: {}",
+            config.metrics_adapter
+        ))),
     }
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
+    use once_cell::sync::Lazy;
+    use std::sync::Mutex;
+
+    // Use a mutex to serialize tests that modify environment variables
+    static ENV_MUTEX: Lazy<Mutex<()>> = Lazy::new(|| Mutex::new(()));
 
     #[tokio::test]
     async fn test_noop_metrics() {
         let metrics = NoOpMetricsPublisher::new();
-        
+
         // These should all be no-ops and not panic
         metrics.incr("test.counter").await;
         metrics.count("test.counter", 5).await;
-        metrics.incr_with_tags("test.counter", &[("env", "test")]).await;
-        metrics.count_with_tags("test.counter", 10, &[("env", "test"), ("service", "quickdid")]).await;
+        metrics
+            .incr_with_tags("test.counter", &[("env", "test")])
+            .await;
+        metrics
+            .count_with_tags(
+                "test.counter",
+                10,
+                &[("env", "test"), ("service", "quickdid")],
+            )
+            .await;
         metrics.gauge("test.gauge", 100).await;
-        metrics.gauge_with_tags("test.gauge", 200, &[("host", "localhost")]).await;
+        metrics
+            .gauge_with_tags("test.gauge", 200, &[("host", "localhost")])
+            .await;
         metrics.time("test.timing", 42).await;
-        metrics.time_with_tags("test.timing", 84, &[("endpoint", "/resolve")]).await;
+        metrics
+            .time_with_tags("test.timing", 84, &[("endpoint", "/resolve")])
+            .await;
     }
 
     #[tokio::test]
     async fn test_shared_metrics() {
         let metrics: SharedMetricsPublisher = Arc::new(NoOpMetricsPublisher::new());
-        
+
         // Verify it can be used as a shared reference
         metrics.incr("shared.counter").await;
         metrics.gauge("shared.gauge", 50).await;
-        
+
         // Verify it can be cloned
         let metrics2 = Arc::clone(&metrics);
         metrics2.count("cloned.counter", 3).await;
     }
-    
+
     #[test]
     fn test_create_noop_publisher() {
         use std::env;
-        
+
+        // Lock mutex to prevent concurrent environment variable modification
+        let _guard = ENV_MUTEX.lock().unwrap();
+
         // Clean up any existing environment variables first
         unsafe {
             env::remove_var("METRICS_ADAPTER");
···
             env::remove_var("METRICS_PREFIX");
             env::remove_var("METRICS_TAGS");
         }
-        
+
         // Set up environment for noop adapter
         unsafe {
             env::set_var("HTTP_EXTERNAL", "test.example.com");
             env::set_var("SERVICE_KEY", "did:key:test");
             env::set_var("METRICS_ADAPTER", "noop");
         }
-        
+
         let config = Config::from_env().unwrap();
         let metrics = create_metrics_publisher(&config).unwrap();
-        
+
         // Should create successfully - actual type checking happens at compile time
         assert!(Arc::strong_count(&metrics) == 1);
-        
+
         // Clean up
         unsafe {
             env::remove_var("METRICS_ADAPTER");
+            env::remove_var("HTTP_EXTERNAL");
+            env::remove_var("SERVICE_KEY");
         }
     }
-    
+
     #[test]
     fn test_create_statsd_publisher() {
         use std::env;
-        
+
+        // Lock mutex to prevent concurrent environment variable modification
+        let _guard = ENV_MUTEX.lock().unwrap();
+
         // Clean up any existing environment variables first
         unsafe {
             env::remove_var("METRICS_ADAPTER");
···
             env::remove_var("METRICS_PREFIX");
             env::remove_var("METRICS_TAGS");
         }
-        
+
         // Set up environment for statsd adapter
         unsafe {
             env::set_var("HTTP_EXTERNAL", "test.example.com");
···
             env::set_var("METRICS_PREFIX", "test");
             env::set_var("METRICS_TAGS", "env:test,service:quickdid");
         }
-        
+
         let config = Config::from_env().unwrap();
         let metrics = create_metrics_publisher(&config).unwrap();
-        
+
         // Should create successfully
         assert!(Arc::strong_count(&metrics) == 1);
-        
+
         // Clean up
         unsafe {
             env::remove_var("METRICS_ADAPTER");
             env::remove_var("METRICS_STATSD_HOST");
             env::remove_var("METRICS_PREFIX");
             env::remove_var("METRICS_TAGS");
+            env::remove_var("HTTP_EXTERNAL");
+            env::remove_var("SERVICE_KEY");
         }
     }
-    
+
     #[test]
     fn test_missing_statsd_host() {
         use std::env;
-        
+
+        // Lock mutex to prevent concurrent environment variable modification
+        let _guard = ENV_MUTEX.lock().unwrap();
+
         // Clean up any existing environment variables first
         unsafe {
             env::remove_var("METRICS_ADAPTER");
···
             env::remove_var("METRICS_PREFIX");
             env::remove_var("METRICS_TAGS");
         }
-        
+
         // Set up environment for statsd adapter without host
         unsafe {
             env::set_var("HTTP_EXTERNAL", "test.example.com");
···
             env::set_var("METRICS_ADAPTER", "statsd");
             env::remove_var("METRICS_STATSD_HOST");
         }
-        
+
         let config = Config::from_env().unwrap();
         let result = create_metrics_publisher(&config);
-        
+
         // Should fail with invalid config error
         assert!(result.is_err());
         if let Err(e) = result {
             assert!(matches!(e, MetricsError::InvalidConfig(_)));
         }
-        
+
         // Clean up
         unsafe {
             env::remove_var("METRICS_ADAPTER");
+            env::remove_var("HTTP_EXTERNAL");
+            env::remove_var("SERVICE_KEY");
         }
     }
-    
+
     #[test]
     fn test_invalid_adapter() {
         use std::env;
-        
+
+        // Lock mutex to prevent concurrent environment variable modification
+        let _guard = ENV_MUTEX.lock().unwrap();
+
         // Clean up any existing environment variables first
         unsafe {
             env::remove_var("METRICS_ADAPTER");
···
             env::remove_var("METRICS_PREFIX");
             env::remove_var("METRICS_TAGS");
         }
-        
+
         // Set up environment with invalid adapter
         unsafe {
             env::set_var("HTTP_EXTERNAL", "test.example.com");
···
             env::set_var("METRICS_ADAPTER", "invalid");
             env::remove_var("METRICS_STATSD_HOST"); // Clean up from other tests
         }
-        
+
         let config = Config::from_env().unwrap();
-        
+
         // Config validation should catch this
         let validation_result = config.validate();
         assert!(validation_result.is_err());
-        
+
         // Clean up
         unsafe {
             env::remove_var("METRICS_ADAPTER");
+            env::remove_var("HTTP_EXTERNAL");
+            env::remove_var("SERVICE_KEY");
         }
     }
-}
+}