From cecc8df758bae0539a968efc0bc725711f7a929e Mon Sep 17 00:00:00 2001 From: Alexey <247128645+axkurcom@users.noreply.github.com> Date: Sat, 31 Oct 2026 12:19:16 +0300 Subject: [PATCH] Dead-Code Cleaning --- src/api/config_store.rs | 4 +- src/api/config_store/persistence.rs | 1 + src/config/load.rs | 8 +- src/config/load/pipeline.rs | 1 + src/config/load/runtime_web/build.rs | 2 + src/config/types/web/runtime.rs | 1 + src/daemon/pid_file.rs | 1 + src/logging/file.rs | 4 +- src/maestro/generation.rs | 1 + src/proxy/authenticated.rs | 2 + src/proxy/client.rs | 6 +- src/proxy/client/stream_entry.rs | 3 +- src/proxy/direct_buffer_budget.rs | 1 + src/proxy/direct_relay.rs | 6 +- src/proxy/direct_relay/relay.rs | 3 +- src/proxy/masking.rs | 7 +- src/proxy/masking/handler.rs | 1 + src/proxy/route_mode.rs | 1 - src/proxy/shared_state.rs | 14 +- src/proxy/user_admission.rs | 14 ++ src/quota_state.rs | 11 +- src/slot_budget.rs | 1 + src/transport/middle_proxy/health.rs | 33 ++-- .../middle_proxy/health/floor_plan.rs | 54 ------- .../middle_proxy/health/idle_refresh.rs | 141 ------------------ src/transport/middle_proxy/health/recovery.rs | 1 + src/transport/middle_proxy/pool_nat.rs | 1 + src/transport/middle_proxy/pool_refill.rs | 16 -- .../middle_proxy/pool_reinit/coordination.rs | 1 + .../middle_proxy/pool_reinit/reconcile.rs | 12 -- src/transport/middle_proxy/pool_writer.rs | 4 + .../middle_proxy/pool_writer/publication.rs | 1 + .../middle_proxy/pool_writer/runtime.rs | 35 ----- src/transport/middle_proxy/registry.rs | 4 + src/transport/middle_proxy/registry/writer.rs | 2 + src/transport/upstream/manager_config.rs | 2 + src/util/secure_fs/mod.rs | 4 +- src/util/secure_fs/write.rs | 1 + src/web/manager/carrier_learning.rs | 3 + src/web/manager/lifecycle.rs | 2 + 40 files changed, 104 insertions(+), 306 deletions(-) diff --git a/src/api/config_store.rs b/src/api/config_store.rs index 7813859..85feb2e 100644 --- a/src/api/config_store.rs +++ b/src/api/config_store.rs @@ -16,8 +16,10 @@ mod atomic; pub(in crate::api) use atomic::{write_atomic, write_atomic_if_unchanged}; #[cfg(test)] use persistence::{find_toml_table_bounds, render_access_section, save_sections_to_disk}; +#[cfg(test)] +pub(in crate::api) use persistence::save_access_sections_to_disk; pub(in crate::api) use persistence::{ - render_server_listeners, render_top_level_section, save_access_sections_to_disk, + render_server_listeners, render_top_level_section, save_access_sections_to_disk_if_revision, upsert_toml_table, }; diff --git a/src/api/config_store/persistence.rs b/src/api/config_store/persistence.rs index c49b598..980b2c6 100644 --- a/src/api/config_store/persistence.rs +++ b/src/api/config_store/persistence.rs @@ -97,6 +97,7 @@ pub(in crate::api) fn render_server_listeners(cfg: &ProxyConfig) -> Result>(path: P) -> Result { Self::load_with_metadata(path).map(|loaded| loaded.config) } /// Loads typed configuration together with its source and rendered metadata. + #[cfg(test)] pub(crate) fn load_with_metadata>(path: P) -> Result { Self::load_with_source_overrides(path, &BTreeMap::new()) } /// Loads a typed snapshot while replacing selected captured source documents. + #[cfg(test)] pub(crate) fn load_with_source_overrides>( path: P, source_overrides: &BTreeMap, @@ -247,6 +252,7 @@ impl ProxyConfig { }) } + #[cfg(test)] fn load_source_graph(graph: ConfigSourceGraph) -> Result { pipeline::load_source_graph(graph) } diff --git a/src/config/load/pipeline.rs b/src/config/load/pipeline.rs index c578cdf..9f70f2d 100644 --- a/src/config/load/pipeline.rs +++ b/src/config/load/pipeline.rs @@ -1,6 +1,7 @@ use super::*; /// Completes legacy synchronous loads without performing hostname resolution. +#[cfg(test)] pub(super) fn load_source_graph(graph: ConfigSourceGraph) -> Result { finish(parse_source_graph(graph)?) } diff --git a/src/config/load/runtime_web/build.rs b/src/config/load/runtime_web/build.rs index 94f82fa..720d417 100644 --- a/src/config/load/runtime_web/build.rs +++ b/src/config/load/runtime_web/build.rs @@ -128,6 +128,7 @@ pub(in crate::config::load) fn rebuild(config: &mut ProxyConfig) -> Result<()> { group_id: vhost_idx, profiles: shared_profiles, }), + #[cfg(test)] host: String::new(), base: if vhost.base_path.is_empty() { "/".to_string() @@ -155,6 +156,7 @@ pub(in crate::config::load) fn rebuild(config: &mut ProxyConfig) -> Result<()> { host.clone(), Arc::new(WebRuntimeVhost { wildcard: None, + #[cfg(test)] host: host.clone(), base: if vhost.base_path.is_empty() { "/".to_string() diff --git a/src/config/types/web/runtime.rs b/src/config/types/web/runtime.rs index b6e7071..731ef15 100644 --- a/src/config/types/web/runtime.rs +++ b/src/config/types/web/runtime.rs @@ -23,6 +23,7 @@ pub(crate) struct WebRuntimeVhost { /// Shared automatic-host policy; exact hosts keep their precomputed bindings. pub(crate) wildcard: Option, /// Canonical lowercase ACE hostname; empty for a wildcard template. + #[cfg(test)] pub(crate) host: String, /// Exact slash-delimited endpoint base, including the trailing slash. pub(crate) base: String, diff --git a/src/daemon/pid_file.rs b/src/daemon/pid_file.rs index c9fd035..2049fe6 100644 --- a/src/daemon/pid_file.rs +++ b/src/daemon/pid_file.rs @@ -57,6 +57,7 @@ impl PidFile { } /// Checks whether the PID file names a running process without modifying either file. + #[cfg(test)] pub fn check_running(&self) -> Result, DaemonError> { let Some(pid) = read_pid_file_if_exists(&self.path, self.strict_runtime_paths)? else { return Ok(None); diff --git a/src/logging/file.rs b/src/logging/file.rs index 8b1de52..13f0fa7 100644 --- a/src/logging/file.rs +++ b/src/logging/file.rs @@ -1,6 +1,8 @@ #[cfg(not(unix))] use std::fs::OpenOptions; -use std::fs::{self, File}; +use std::fs::File; +#[cfg(any(test, not(unix)))] +use std::fs; use std::io::{self, Write}; use std::path::{Path, PathBuf}; use std::time::{Duration, SystemTime, UNIX_EPOCH}; diff --git a/src/maestro/generation.rs b/src/maestro/generation.rs index 10da12f..4db38aa 100644 --- a/src/maestro/generation.rs +++ b/src/maestro/generation.rs @@ -74,6 +74,7 @@ impl SessionAdmission { .fetch_or(SESSION_ADMISSION_CLOSED, Ordering::AcqRel); } + #[cfg(test)] fn reopen(&self) { self.state .fetch_and(!SESSION_ADMISSION_CLOSED, Ordering::AcqRel); diff --git a/src/proxy/authenticated.rs b/src/proxy/authenticated.rs index a62ce25..48e66f6 100644 --- a/src/proxy/authenticated.rs +++ b/src/proxy/authenticated.rs @@ -297,6 +297,7 @@ impl Drop for UserIpPermit { impl UserConnectionReservation { /// Creates an active reservation after both admission counters were acquired. + #[cfg(test)] pub(crate) fn new( stats: Arc, ip_tracker: Arc, @@ -375,6 +376,7 @@ impl Drop for UserConnectionReservation { } /// Applies user quota, connection, and source-IP admission atomically. +#[cfg(test)] pub(crate) async fn acquire_user_connection_reservation( user: &str, config: &ProxyConfig, diff --git a/src/proxy/client.rs b/src/proxy/client.rs index bfe6469..5273402 100644 --- a/src/proxy/client.rs +++ b/src/proxy/client.rs @@ -71,9 +71,9 @@ mod authenticated; use handshake_support::*; #[cfg(test)] pub use stream_entry::handle_client_stream; -pub use stream_entry::{ - handle_client_stream_with_shared, handle_client_stream_with_shared_and_pool_runtime, -}; +#[cfg(test)] +pub use stream_entry::handle_client_stream_with_shared; +pub use stream_entry::handle_client_stream_with_shared_and_pool_runtime; pub struct ClientHandler; diff --git a/src/proxy/client/stream_entry.rs b/src/proxy/client/stream_entry.rs index 34f68d8..fdb9848 100644 --- a/src/proxy/client/stream_entry.rs +++ b/src/proxy/client/stream_entry.rs @@ -41,7 +41,8 @@ where } #[allow(clippy::too_many_arguments)] -#[allow(dead_code)] +/// Runs a client stream with shared process state in tests. +#[cfg(test)] pub async fn handle_client_stream_with_shared( stream: S, peer: SocketAddr, diff --git a/src/proxy/direct_buffer_budget.rs b/src/proxy/direct_buffer_budget.rs index 8204351..39f6374 100644 --- a/src/proxy/direct_buffer_budget.rs +++ b/src/proxy/direct_buffer_budget.rs @@ -269,6 +269,7 @@ impl DirectBufferBudget { } /// Returns the conservative ceiling used when memory discovery is unavailable. +#[cfg(test)] pub(crate) fn fallback_direct_buffer_hard_limit() -> usize { AUTO_HARD_FALLBACK_BYTES } diff --git a/src/proxy/direct_relay.rs b/src/proxy/direct_relay.rs index 45a1b8b..69808c1 100644 --- a/src/proxy/direct_relay.rs +++ b/src/proxy/direct_relay.rs @@ -43,9 +43,9 @@ mod relay; // Telegram DC resolution and upstream handshake. mod routing; -pub(crate) use relay::{ - handle_via_direct, handle_via_direct_with_shared, handle_via_direct_with_shared_and_conntrack, -}; +#[cfg(test)] +pub(crate) use relay::handle_via_direct; +pub(crate) use relay::handle_via_direct_with_shared_and_conntrack; use routing::*; const UNKNOWN_DC_LOG_DISTINCT_LIMIT: usize = 1024; static LOGGED_UNKNOWN_DCS: OnceLock>> = OnceLock::new(); diff --git a/src/proxy/direct_relay/relay.rs b/src/proxy/direct_relay/relay.rs index 298ff5d..ba8c019 100644 --- a/src/proxy/direct_relay/relay.rs +++ b/src/proxy/direct_relay/relay.rs @@ -1,7 +1,7 @@ use super::*; -#[allow(dead_code)] /// Runs Direct relay with standalone cancellation and shared-state defaults. +#[cfg(test)] pub(crate) async fn handle_via_direct( client_reader: CryptoReader, client_writer: CryptoWriter, @@ -39,6 +39,7 @@ where } /// Runs Direct relay for a kernel-backed TCP client tuple. +#[cfg(test)] pub(crate) async fn handle_via_direct_with_shared( client_reader: CryptoReader, client_writer: CryptoWriter, diff --git a/src/proxy/masking.rs b/src/proxy/masking.rs index e9c3284..aa11fb7 100644 --- a/src/proxy/masking.rs +++ b/src/proxy/masking.rs @@ -77,15 +77,16 @@ mod relay; pub use backend_setup::handle_bad_client; use backend_setup::*; use copy::*; +#[cfg(test)] pub(crate) use handler::handle_bad_client_with_shared; pub(in crate::proxy) use handler::handle_bad_client_with_shared_resolver; use interfaces::*; use relay::*; use target::*; pub(crate) use timing::sample_lognormal_percentile_bounded; -use timing::{ - mask_outcome_target_budget, wait_mask_connect_budget_if_needed, wait_mask_outcome_budget, -}; +#[cfg(test)] +use timing::mask_outcome_target_budget; +use timing::{wait_mask_connect_budget_if_needed, wait_mask_outcome_budget}; #[cfg(test)] #[path = "tests/masking_security_tests.rs"] diff --git a/src/proxy/masking/handler.rs b/src/proxy/masking/handler.rs index 1c1e3a6..f0c1ea4 100644 --- a/src/proxy/masking/handler.rs +++ b/src/proxy/masking/handler.rs @@ -1,6 +1,7 @@ use super::*; /// Handles a bad client with shared pre-auth fallback admission state. +#[cfg(test)] pub(crate) async fn handle_bad_client_with_shared( reader: R, writer: W, diff --git a/src/proxy/route_mode.rs b/src/proxy/route_mode.rs index d5324fd..312dd51 100644 --- a/src/proxy/route_mode.rs +++ b/src/proxy/route_mode.rs @@ -1,4 +1,3 @@ -use std::sync::Arc; use std::time::{Duration, SystemTime, UNIX_EPOCH}; use tokio::sync::watch; diff --git a/src/proxy/shared_state.rs b/src/proxy/shared_state.rs index c532a0b..78538f4 100644 --- a/src/proxy/shared_state.rs +++ b/src/proxy/shared_state.rs @@ -8,7 +8,9 @@ use std::time::Instant; use dashmap::DashMap; use tokio::sync::{OwnedSemaphorePermit, Semaphore, mpsc}; -use crate::proxy::direct_buffer_budget::{DirectBufferBudget, fallback_direct_buffer_hard_limit}; +use crate::proxy::direct_buffer_budget::DirectBufferBudget; +#[cfg(test)] +use crate::proxy::direct_buffer_budget::fallback_direct_buffer_hard_limit; use crate::proxy::handshake::{AuthProbeSaturationState, AuthProbeState}; use crate::proxy::middle_relay::{DesyncDedupRotationState, RelayIdleCandidateRegistry}; use crate::proxy::traffic_limiter::TrafficLimiter; @@ -103,6 +105,8 @@ pub(crate) struct ProxySharedState { } impl ProxySharedState { + /// Creates isolated process authorities for tests. + #[cfg(test)] pub(crate) fn new() -> Arc { Self::new_with_direct_buffer_budget(DirectBufferBudget::new( fallback_direct_buffer_hard_limit(), @@ -110,6 +114,7 @@ impl ProxySharedState { } /// Creates process state with the startup-resolved Direct buffer envelope. + #[cfg(test)] pub(crate) fn new_with_direct_buffer_budget( direct_buffer_budget: Arc, ) -> Arc { @@ -120,6 +125,7 @@ impl ProxySharedState { } /// Creates generation state around one process-owned user authority. + #[cfg(test)] pub(crate) fn new_with_direct_buffer_budget_and_user_admission( direct_buffer_budget: Arc, user_admission: Arc, @@ -203,6 +209,7 @@ impl ProxySharedState { } /// Reconciles the complete user authentication policy from configuration. + #[cfg(test)] pub(crate) fn apply_user_config( &self, users: &HashMap, @@ -255,6 +262,7 @@ impl ProxySharedState { } /// Applies one persisted user mutation before asynchronous config reload. + #[cfg(test)] pub(crate) fn stage_user( &self, user: &str, @@ -299,6 +307,8 @@ impl ProxySharedState { self.user_admission.claim_authenticated(user, credential_id) } + /// Registers a test session without an authenticated credential. + #[cfg(test)] pub(crate) fn register_user_session( self: &Arc, user: &str, @@ -319,6 +329,8 @@ impl ProxySharedState { Some(registration) } + /// Cancels all owners of a user in lifecycle tests. + #[cfg(test)] pub(crate) fn cancel_user_sessions(&self, user: &str) -> usize { self.user_admission.cancel_user_owners(user) } diff --git a/src/proxy/user_admission.rs b/src/proxy/user_admission.rs index bcb73c7..773b5e4 100644 --- a/src/proxy/user_admission.rs +++ b/src/proxy/user_admission.rs @@ -96,8 +96,16 @@ pub(crate) struct UserMutationResult { /// Incarnation invalidated or created by the mutation. pub(crate) incarnation: UserIncarnation, /// Number of live owners cancelled by the mutation. + #[cfg_attr( + not(test), + expect(dead_code, reason = "Mutation results are asserted by admission regression tests") + )] pub(crate) cancelled: usize, /// Whether the effective enabled state changed from enabled to disabled. + #[cfg_attr( + not(test), + expect(dead_code, reason = "Mutation results are asserted by admission regression tests") + )] pub(crate) newly_disabled: bool, } @@ -109,6 +117,7 @@ pub(crate) struct UserAdmissionAuthority { impl UserAdmissionAuthority { /// Creates an uninitialized authority for isolated tests and startup wiring. + #[cfg(test)] pub(crate) fn new() -> Arc { Self::new_with_quota_store(Arc::new(QuotaStore::default())) } @@ -173,6 +182,7 @@ impl UserAdmissionAuthority { } /// Reconciles the complete configured user set into the process authority. + #[cfg(test)] pub(crate) fn apply_config( &self, users: &HashMap, @@ -204,6 +214,7 @@ impl UserAdmissionAuthority { } /// Returns the number of rejected updates from non-owning generations. + #[cfg(test)] pub(crate) fn stale_config_source_rejections(&self) -> u64 { self.state.lock().stale_config_source_rejections } @@ -338,6 +349,7 @@ impl UserAdmissionAuthority { } /// Applies one persisted user value ahead of asynchronous runtime reload. + #[cfg(test)] pub(crate) fn stage_user( &self, user: &str, @@ -502,6 +514,7 @@ impl UserAdmissionAuthority { } /// Registers a legacy owner when no credential snapshot is available. + #[cfg(test)] pub(crate) fn register_legacy(self: &Arc, user: &str) -> Option { let credential_id = { let state = self.state.lock(); @@ -518,6 +531,7 @@ impl UserAdmissionAuthority { } /// Cancels all current owners without changing admission policy. + #[cfg(test)] pub(crate) fn cancel_user_owners(&self, user: &str) -> usize { let tokens = self.state.lock().owner_tokens(user); let count = tokens.len(); diff --git a/src/quota_state.rs b/src/quota_state.rs index 58af1a5..1122847 100644 --- a/src/quota_state.rs +++ b/src/quota_state.rs @@ -63,6 +63,7 @@ impl QuotaStateOwner { } /// Loads only quota entries owned by currently configured users. + #[cfg(test)] pub(crate) async fn load(&self, configured_users: &BTreeSet) { let _ = self.load_inner(configured_users, false).await; } @@ -179,15 +180,6 @@ impl QuotaStateOwner { wait_for_blocking_io(task).await } - /// Removes a deleted user's persisted quota checkpoint. - pub(crate) async fn remove_user( - &self, - configured_users: &BTreeSet, - user: &str, - ) -> std::io::Result<()> { - let guard = Arc::clone(&self.mutation).lock_owned().await; - self.remove_user_locked(configured_users, user, guard).await - } /// Persists deletion while retaining the already-owned lifecycle gate. pub(crate) async fn remove_user_locked( @@ -323,6 +315,7 @@ async fn read_state_file(path: &Path) -> std::io::Result> Ok(Some(state)) } +#[cfg(test)] async fn write_state_file(path: &Path, state: QuotaStateFile) -> std::io::Result<()> { let path = path.to_path_buf(); let task = tokio::task::spawn_blocking(move || write_state_file_blocking(&path, &state)); diff --git a/src/slot_budget.rs b/src/slot_budget.rs index 737b316..ec12aad 100644 --- a/src/slot_budget.rs +++ b/src/slot_budget.rs @@ -61,6 +61,7 @@ impl SlotBudget { } /// Returns the exact number of currently committed or reserved slots. + #[cfg(test)] pub(crate) fn used(&self) -> usize { self.used.load(Ordering::Acquire) } diff --git a/src/transport/middle_proxy/health.rs b/src/transport/middle_proxy/health.rs index f71b9cf..e5d9898 100644 --- a/src/transport/middle_proxy/health.rs +++ b/src/transport/middle_proxy/health.rs @@ -7,6 +7,7 @@ use std::sync::Arc; use std::time::{Duration, Instant}; use rand::RngExt; +#[cfg(test)] use tokio::sync::Semaphore; use tokio::task::JoinSet; use tracing::{debug, info, warn}; @@ -65,27 +66,17 @@ struct FamilyFloorPlan { warm_cap_effective_total: usize, active_writers_current: usize, warm_writers_current: usize, - target_writers_total: usize, } -#[derive(Debug)] -struct FamilyReconnectOutcome { - key: (i32, IpFamily), - dc: i32, - family: IpFamily, - required: usize, - endpoint_count: usize, -} +#[cfg(test)] struct ScheduledReconnects<'a> { inflight: &'a mut HashMap<(i32, IpFamily), usize>, keys: Vec<(i32, IpFamily)>, } +#[cfg(test)] impl ScheduledReconnects<'_> { - fn current(&self, key: &(i32, IpFamily)) -> usize { - self.inflight.get(key).copied().unwrap_or(0) - } fn reserve(&mut self, key: (i32, IpFamily)) { self.keys.push(key); @@ -93,6 +84,7 @@ impl ScheduledReconnects<'_> { } } +#[cfg(test)] impl Drop for ScheduledReconnects<'_> { fn drop(&mut self) { for key in self.keys.drain(..) { @@ -118,9 +110,9 @@ mod scheduler; mod drain; // Per-family reconnect scheduling and health state. mod family; -// Adaptive-floor planning and warning rate limits. +// Adaptive-floor planning. mod floor_plan; -// Idle-writer refresh and cap-aware swaps. +// Idle-writer refresh under the current capacity policy. mod idle_refresh; // Single-endpoint recovery and shadow rotation. mod recovery; @@ -128,18 +120,15 @@ mod recovery; mod zombie_watchdog; pub use drain::me_drain_timeout_enforcer; -pub(in crate::transport::middle_proxy) use drain::{ - health_drain_close_budget, reap_draining_writers, -}; +#[cfg(test)] +pub(in crate::transport::middle_proxy) use drain::health_drain_close_budget; +pub(in crate::transport::middle_proxy) use drain::reap_draining_writers; use family::{check_family, update_family_runtime_state}; -use floor_plan::{ - build_family_floor_plan, live_active_writers_for_dc_family, should_emit_rate_limited_warn, -}; -use idle_refresh::{maybe_refresh_idle_writer_for_dc, maybe_swap_idle_writer_for_cap}; +use floor_plan::build_family_floor_plan; +use idle_refresh::maybe_refresh_idle_writer_for_dc; pub use monitor::me_health_monitor; use recovery::{ has_bound_clients_on_endpoint, maybe_rotate_single_endpoint_shadow, - recover_single_endpoint_outage, }; pub use zombie_watchdog::me_zombie_writer_watchdog; diff --git a/src/transport/middle_proxy/health/floor_plan.rs b/src/transport/middle_proxy/health/floor_plan.rs index 01b41b3..6599654 100644 --- a/src/transport/middle_proxy/health/floor_plan.rs +++ b/src/transport/middle_proxy/health/floor_plan.rs @@ -1,56 +1,6 @@ use super::*; -pub(super) fn should_emit_rate_limited_warn( - next_allowed: &mut HashMap<(i32, IpFamily), Instant>, - key: (i32, IpFamily), - now: Instant, - cooldown: Duration, -) -> bool { - let Some(ready_at) = next_allowed.get(&key).copied() else { - next_allowed.insert(key, now + cooldown); - return true; - }; - if now >= ready_at { - next_allowed.insert(key, now + cooldown); - return true; - } - false -} -pub(super) async fn live_active_writers_for_dc_family( - pool: &Arc, - dc: i32, - family: IpFamily, -) -> usize { - let active_generation = pool.current_generation(); - let writers = pool.writers.read().await; - writers - .iter() - .filter(|writer| { - if writer.draining.load(std::sync::atomic::Ordering::Relaxed) { - return false; - } - if writer.writer_dc != dc { - return false; - } - if writer.generation != active_generation { - return false; - } - if !matches!( - crate::transport::middle_proxy::pool::WriterContour::from_u8( - writer.contour.load(std::sync::atomic::Ordering::Relaxed), - ), - crate::transport::middle_proxy::pool::WriterContour::Active - ) { - return false; - } - match family { - IpFamily::V4 => writer.addr.is_ipv4(), - IpFamily::V6 => writer.addr.is_ipv6(), - } - }) - .count() -} pub(super) fn adaptive_floor_class_min( pool: &Arc, @@ -177,7 +127,6 @@ pub(super) async fn build_family_floor_plan( warm_cap_effective_total: warm_cap_configured_total, active_writers_current, warm_writers_current, - target_writers_total: 0, }; } @@ -199,7 +148,6 @@ pub(super) async fn build_family_floor_plan( warm_cap_effective_total: warm_cap_configured_total, active_writers_current, warm_writers_current, - target_writers_total: target_total, }; } @@ -251,7 +199,6 @@ pub(super) async fn build_family_floor_plan( } let active_cap_effective_total = active_cap_configured_total.max(other_active.saturating_add(min_sum)); - let target_writers_total = other_active.saturating_add(target_sum); FamilyFloorPlan { by_dc, active_cap_configured_total, @@ -260,6 +207,5 @@ pub(super) async fn build_family_floor_plan( warm_cap_effective_total: warm_cap_configured_total, active_writers_current, warm_writers_current, - target_writers_total, } } diff --git a/src/transport/middle_proxy/health/idle_refresh.rs b/src/transport/middle_proxy/health/idle_refresh.rs index 53054f1..78fddc4 100644 --- a/src/transport/middle_proxy/health/idle_refresh.rs +++ b/src/transport/middle_proxy/health/idle_refresh.rs @@ -1,146 +1,5 @@ use super::*; -pub(super) async fn maybe_swap_idle_writer_for_cap( - pool: &Arc, - rng: &Arc, - dc: i32, - family: IpFamily, - endpoints: &[SocketAddr], - required: usize, - floor_targets_by_dc: &HashMap, - live_writer_ids_by_addr: &HashMap<(i32, SocketAddr), Vec>, - writer_idle_since: &HashMap, - bound_clients_by_writer: &HashMap, -) -> bool { - let Some(replacement_endpoint) = endpoints - .iter() - .min_by_key(|endpoint| { - live_writer_ids_by_addr - .get(&(dc, **endpoint)) - .map_or(0, Vec::len) - }) - .copied() - else { - return false; - }; - - let mut alive_by_dc = HashMap::::new(); - for ((writer_dc, endpoint), writer_ids) in live_writer_ids_by_addr { - if endpoint.is_ipv4() == matches!(family, IpFamily::V4) { - *alive_by_dc.entry(*writer_dc).or_insert(0) += writer_ids.len(); - } - } - - let now_epoch_secs = MePool::now_epoch_secs(); - let mut candidates = Vec::<(u64, SocketAddr, i32, u64, usize)>::new(); - for ((writer_dc, endpoint), writer_ids) in live_writer_ids_by_addr { - if endpoint.is_ipv4() != matches!(family, IpFamily::V4) { - continue; - } - let Some(donor_floor) = floor_targets_by_dc.get(writer_dc).copied() else { - continue; - }; - if alive_by_dc.get(writer_dc).copied().unwrap_or(0) <= donor_floor { - continue; - } - for writer_id in writer_ids { - if bound_clients_by_writer.get(writer_id).copied().unwrap_or(0) > 0 { - continue; - } - let Some(idle_since_epoch_secs) = writer_idle_since.get(writer_id).copied() else { - continue; - }; - candidates.push(( - *writer_id, - *endpoint, - *writer_dc, - now_epoch_secs.saturating_sub(idle_since_epoch_secs), - donor_floor, - )); - } - } - candidates.sort_unstable_by(|left, right| right.3.cmp(&left.3)); - - for (old_writer_id, donor_endpoint, donor_dc, idle_age_secs, donor_floor) in candidates { - let expected_role = { - let writers = pool.writers.read().await; - writers - .iter() - .find(|writer| writer.id == old_writer_id) - .map(WriterRole::from_writer) - }; - let Some(expected_role) = expected_role else { - continue; - }; - if expected_role.dc != donor_dc - || expected_role.family != family - || expected_role.contour != WriterContour::Active - { - continue; - } - let Some(mut reservation) = pool - .registry - .try_reserve_writer_replacement(old_writer_id) - .await - else { - continue; - }; - let replace = pool.replace_writer_with_generation_contour_for_dc( - replacement_endpoint, - rng.as_ref(), - pool.current_generation(), - WriterContour::Active, - dc, - expected_role, - WriterReplacementPurpose::FloorRebalance { - donor_floor, - receiver_floor: required, - }, - &mut reservation, - ); - match tokio::time::timeout(pool.reconnect_runtime.me_one_timeout, replace).await { - Ok(Ok(())) => { - info!( - dc = %dc, - ?family, - %replacement_endpoint, - donor_dc = %donor_dc, - %donor_endpoint, - old_writer_id, - idle_age_secs, - "Adaptive floor cap rebalance committed" - ); - return true; - } - Ok(Err(error)) => { - debug!( - dc = %dc, - ?family, - %replacement_endpoint, - donor_dc = %donor_dc, - old_writer_id, - idle_age_secs, - %error, - "Adaptive floor cap rebalance failed" - ); - return false; - } - Err(_) => { - debug!( - dc = %dc, - ?family, - %replacement_endpoint, - donor_dc = %donor_dc, - old_writer_id, - idle_age_secs, - "Adaptive floor cap rebalance timed out" - ); - return false; - } - } - } - false -} pub(super) async fn maybe_refresh_idle_writer_for_dc( pool: &Arc, diff --git a/src/transport/middle_proxy/health/recovery.rs b/src/transport/middle_proxy/health/recovery.rs index b9455ce..d6c893c 100644 --- a/src/transport/middle_proxy/health/recovery.rs +++ b/src/transport/middle_proxy/health/recovery.rs @@ -15,6 +15,7 @@ pub(super) fn has_bound_clients_on_endpoint( } /// Restores missing single-endpoint coverage under the bounded outage retry policy. +#[cfg(test)] pub(super) async fn recover_single_endpoint_outage( pool: &Arc, rng: &Arc, diff --git a/src/transport/middle_proxy/pool_nat.rs b/src/transport/middle_proxy/pool_nat.rs index d50396d..efe9551 100644 --- a/src/transport/middle_proxy/pool_nat.rs +++ b/src/transport/middle_proxy/pool_nat.rs @@ -169,6 +169,7 @@ impl MePool { } /// Applies the configured or HTTP-detected NAT IP to non-public addresses. + #[cfg(test)] pub(super) fn translate_ip_for_nat(&self, ip: IpAddr) -> IpAddr { Self::translate_ip_with_nat( ip, diff --git a/src/transport/middle_proxy/pool_refill.rs b/src/transport/middle_proxy/pool_refill.rs index 2f9e0f5..ad78bfd 100644 --- a/src/transport/middle_proxy/pool_refill.rs +++ b/src/transport/middle_proxy/pool_refill.rs @@ -193,22 +193,6 @@ impl MePool { self.refill_states.lock().contains_key(&key) } - pub(super) async fn connect_endpoints_round_robin( - self: &Arc, - dc: i32, - endpoints: &[SocketAddr], - rng: &SecureRandom, - ) -> bool { - self.connect_endpoints_round_robin_with_generation_contour( - dc, - endpoints, - rng, - self.current_generation(), - WriterContour::Active, - WriterOpenIntent::Normal, - ) - .await - } pub(super) async fn connect_endpoints_round_robin_with_generation_contour( self: &Arc, diff --git a/src/transport/middle_proxy/pool_reinit/coordination.rs b/src/transport/middle_proxy/pool_reinit/coordination.rs index 876445c..11d58e0 100644 --- a/src/transport/middle_proxy/pool_reinit/coordination.rs +++ b/src/transport/middle_proxy/pool_reinit/coordination.rs @@ -378,6 +378,7 @@ impl MePool { } /// Returns the currently authoritative endpoint set for drain and coverage decisions. + #[cfg(test)] pub(in crate::transport::middle_proxy) async fn desired_dc_endpoints( &self, ) -> HashMap> { diff --git a/src/transport/middle_proxy/pool_reinit/reconcile.rs b/src/transport/middle_proxy/pool_reinit/reconcile.rs index f22c05a..e6efa13 100644 --- a/src/transport/middle_proxy/pool_reinit/reconcile.rs +++ b/src/transport/middle_proxy/pool_reinit/reconcile.rs @@ -169,14 +169,6 @@ impl MePool { } } - /// Reconciles one immutable endpoint snapshot without draining uncovered groups. - pub async fn zero_downtime_reinit_after_map_change( - self: &Arc, - rng: &SecureRandom, - ) -> bool { - let endpoint_snapshot = self.endpoint_snapshot.load_full(); - self.reinit_endpoint_snapshot(rng, endpoint_snapshot).await - } /// Reconciles exactly the revision owned by the scheduler's retry objective. pub(in crate::transport::middle_proxy) async fn reinit_endpoint_snapshot( @@ -417,8 +409,4 @@ impl MePool { true } - /// Applies periodic rotation using the same coverage and generation fences as map changes. - pub async fn zero_downtime_reinit_periodic(self: &Arc, rng: &SecureRandom) -> bool { - self.zero_downtime_reinit_after_map_change(rng).await - } } diff --git a/src/transport/middle_proxy/pool_writer.rs b/src/transport/middle_proxy/pool_writer.rs index b9298d8..f208804 100644 --- a/src/transport/middle_proxy/pool_writer.rs +++ b/src/transport/middle_proxy/pool_writer.rs @@ -55,6 +55,10 @@ pub(super) enum WriterReplacementPurpose { /// Replaces an idle writer before the upstream idle lifetime expires. IdleRefresh, /// Moves one idle slot from a donor above floor to a receiver below floor. + #[cfg_attr( + not(test), + expect(dead_code, reason = "Legacy purpose is exercised by commit-fence regression tests") + )] FloorRebalance { /// Minimum writer count that must remain in the donor group. donor_floor: usize, diff --git a/src/transport/middle_proxy/pool_writer/publication.rs b/src/transport/middle_proxy/pool_writer/publication.rs index d148ca9..6eda06a 100644 --- a/src/transport/middle_proxy/pool_writer/publication.rs +++ b/src/transport/middle_proxy/pool_writer/publication.rs @@ -152,6 +152,7 @@ impl MePool { /// Commits writer visibility and lifecycle ownership after all cancellation points. #[allow(clippy::too_many_arguments)] + #[cfg(test)] pub(in crate::transport::middle_proxy) async fn publish_prepared_writer( self: &Arc, writer: MeWriter, diff --git a/src/transport/middle_proxy/pool_writer/runtime.rs b/src/transport/middle_proxy/pool_writer/runtime.rs index d15824a..b31fbb8 100644 --- a/src/transport/middle_proxy/pool_writer/runtime.rs +++ b/src/transport/middle_proxy/pool_writer/runtime.rs @@ -407,37 +407,6 @@ impl MePool { removed } - pub(crate) async fn mark_writer_draining_with_timeout( - self: &Arc, - writer_id: u64, - timeout: Option, - allow_drain_fallback: bool, - ) { - let timeout = timeout.filter(|d| !d.is_zero()); - let found = { - let ws = self.writers.write().await; - let mut registry_registration = self.registry.prepare_writer_registration().await; - if let Some(w) = ws.iter().find(|w| w.id == writer_id) { - if !allow_drain_fallback { - registry_registration.retire(writer_id); - } - self.apply_writer_draining_state(w, timeout, allow_drain_fallback); - true - } else { - false - } - }; - - if !found { - return; - } - - let timeout_secs = timeout.map(|d| d.as_secs()).unwrap_or(0); - debug!( - writer_id, - timeout_secs, allow_drain_fallback, "ME writer marked draining" - ); - } /// Publishes drain metadata before the release-store that makes draining observable. pub(in crate::transport::middle_proxy) fn apply_writer_draining_state( @@ -472,10 +441,6 @@ impl MePool { !already_draining } - pub(crate) async fn mark_writer_draining(self: &Arc, writer_id: u64) { - self.mark_writer_draining_with_timeout(writer_id, Some(Duration::from_secs(300)), false) - .await; - } pub(in crate::transport::middle_proxy) fn writer_accepts_new_binding( &self, diff --git a/src/transport/middle_proxy/registry.rs b/src/transport/middle_proxy/registry.rs index 24493d3..18ce532 100644 --- a/src/transport/middle_proxy/registry.rs +++ b/src/transport/middle_proxy/registry.rs @@ -180,6 +180,8 @@ impl ConnRegistry { .as_secs() } + /// Creates a test registry with the requested route channel capacity. + #[cfg(test)] pub fn with_route_channel_capacity(route_channel_capacity: usize) -> Self { let route_channel_capacity = route_channel_capacity.max(1); Self::with_route_limits( @@ -288,6 +290,8 @@ impl ConnRegistry { .store(watermark, Ordering::Relaxed); } + /// Registers a test route without a lifetime cleanup lease. + #[cfg(test)] pub async fn register(&self) -> (u64, mpsc::Receiver) { self.register_route() } diff --git a/src/transport/middle_proxy/registry/writer.rs b/src/transport/middle_proxy/registry/writer.rs index efe3c34..8dff646 100644 --- a/src/transport/middle_proxy/registry/writer.rs +++ b/src/transport/middle_proxy/registry/writer.rs @@ -348,6 +348,8 @@ impl ConnRegistry { WriterBindOutcome::Bound } + /// Binds a test route and reports whether publication succeeded. + #[cfg(test)] pub async fn bind_writer(&self, conn_id: u64, writer_id: u64, meta: ConnMeta) -> bool { matches!( self.bind_writer_with_outcome(conn_id, writer_id, meta) diff --git a/src/transport/upstream/manager_config.rs b/src/transport/upstream/manager_config.rs index bb47662..478392b 100644 --- a/src/transport/upstream/manager_config.rs +++ b/src/transport/upstream/manager_config.rs @@ -48,6 +48,8 @@ impl UpstreamManager { Ok(self) } + /// Replaces DNS overrides for isolated upstream tests. + #[cfg(test)] pub(crate) fn update_dns_overrides(&self, entries: &[String]) -> Result<()> { self.dns_resolver.apply_entries(entries) } diff --git a/src/util/secure_fs/mod.rs b/src/util/secure_fs/mod.rs index 68e1bce..56ce4d1 100644 --- a/src/util/secure_fs/mod.rs +++ b/src/util/secure_fs/mod.rs @@ -11,8 +11,10 @@ pub(crate) use path::{ AnchoredPath, chdir_nofollow_or_create, open_compatible_dir, open_dir_nofollow, open_trusted_dir_nofollow_or_create, }; +#[cfg(test)] +pub(crate) use write::open_append_regular; pub(crate) use write::{ - atomic_replace, atomic_replace_async, open_append_regular, open_append_regular_at, + atomic_replace, atomic_replace_async, open_append_regular_at, read_regular_limited, read_regular_limited_async, }; diff --git a/src/util/secure_fs/write.rs b/src/util/secure_fs/write.rs index 72e0a39..08423dc 100644 --- a/src/util/secure_fs/write.rs +++ b/src/util/secure_fs/write.rs @@ -91,6 +91,7 @@ fn same_file_version(left: &std::fs::Metadata, right: &std::fs::Metadata) -> boo } /// Opens an append-only regular file without following path components or hard links. +#[cfg(test)] pub(crate) fn open_append_regular(path: &Path, mode: u32) -> io::Result { let anchored = AnchoredPath::open_creating_parents(path, 0o750)?; open_regular_at( diff --git a/src/web/manager/carrier_learning.rs b/src/web/manager/carrier_learning.rs index 0579e7f..508cf56 100644 --- a/src/web/manager/carrier_learning.rs +++ b/src/web/manager/carrier_learning.rs @@ -133,6 +133,7 @@ pub(super) struct DetachedCarrierEvidence { /// Result of one generation-fenced policy reconciliation. pub(super) struct CarrierLearningPolicyOutcome { /// Current epoch after the reconciliation. + #[cfg(test)] pub(super) epoch: Option, /// Whether this generation was current enough to apply. pub(super) applied: bool, @@ -215,6 +216,7 @@ impl CarrierLearning { .is_some_and(|applied| generation < applied) { return CarrierLearningPolicyOutcome { + #[cfg(test)] epoch: self.epoch, applied: false, detached: None, @@ -239,6 +241,7 @@ impl CarrierLearning { self.policy_started_at = now; } CarrierLearningPolicyOutcome { + #[cfg(test)] epoch: self.epoch, applied: true, detached, diff --git a/src/web/manager/lifecycle.rs b/src/web/manager/lifecycle.rs index 510d6c2..623cc3a 100644 --- a/src/web/manager/lifecycle.rs +++ b/src/web/manager/lifecycle.rs @@ -160,6 +160,7 @@ impl WebProcessRuntime { } /// Stops issuance and drains all WEB work until one absolute deadline. + #[cfg(test)] pub(crate) async fn shutdown_until( self: &std::sync::Arc, deadline: TokioInstant, @@ -168,6 +169,7 @@ impl WebProcessRuntime { } /// Stops issuance and drains WEB work under the currently configured budget. + #[cfg(test)] pub(crate) async fn shutdown(self: &std::sync::Arc) -> WebShutdownOutcome { let timeout_secs = self .active_runtime