Reduce MR + ME Routing hot-path contention

docs/Config_params/CONFIG_PARAMS.en.md

@@ -224,6 +224,151 @@ This document lists all configuration keys accepted by `config.toml`.
 | [`auto_degradation_enabled`](#auto_degradation_enabled) | `bool` | `true` |
 | [`degradation_min_unavailable_dc_groups`](#degradation_min_unavailable_dc_groups) | `u8` | `2` |
 | [`rst_on_close`](#rst_on_close) | `"off"`, `"errors"`, or `"always"` | `"off"` |
+| Key | Type | Default | Hot-Reload |
+| --- | ---- | ------- | ---------- |
+| [`data_path`](#data_path) | `String` | — | `✘` |
+| [`quota_state_path`](#quota_state_path) | `Path` | `"telemt.limit.json"` | `✘` |
+| [`config_strict`](#config_strict) | `bool` | `false` | `✘` |
+| [`prefer_ipv6`](#prefer_ipv6) | `bool` | `false` | `✘` |
+| [`fast_mode`](#fast_mode) | `bool` | `true` | `✘` |
+| [`use_middle_proxy`](#use_middle_proxy) | `bool` | `true` | `✘` |
+| [`proxy_secret_path`](#proxy_secret_path) | `String` | `"proxy-secret"` | `✘` |
+| [`proxy_secret_url`](#proxy_secret_url) | `String` | `"https://core.telegram.org/getProxySecret"` | `✘` |
+| [`proxy_config_v4_cache_path`](#proxy_config_v4_cache_path) | `String` | `"cache/proxy-config-v4.txt"` | `✘` |
+| [`proxy_config_v4_url`](#proxy_config_v4_url) | `String` | `"https://core.telegram.org/getProxyConfig"` | `✘` |
+| [`proxy_config_v6_cache_path`](#proxy_config_v6_cache_path) | `String` | `"cache/proxy-config-v6.txt"` | `✘` |
+| [`proxy_config_v6_url`](#proxy_config_v6_url) | `String` | `"https://core.telegram.org/getProxyConfigV6"` | `✘` |
+| [`ad_tag`](#ad_tag) | `String` | — | `✔` |
+| [`middle_proxy_nat_ip`](#middle_proxy_nat_ip) | `IpAddr` | — | `✘` |
+| [`middle_proxy_nat_probe`](#middle_proxy_nat_probe) | `bool` | `true` | `✘` |
+| [`middle_proxy_nat_stun`](#middle_proxy_nat_stun) | `String` | — | `✘` |
+| [`middle_proxy_nat_stun_servers`](#middle_proxy_nat_stun_servers) | `String[]` | `[]` | `✘` |
+| [`stun_nat_probe_concurrency`](#stun_nat_probe_concurrency) | `usize` | `8` | `✘` |
+| [`middle_proxy_pool_size`](#middle_proxy_pool_size) | `usize` | `8` | `✘` |
+| [`middle_proxy_warm_standby`](#middle_proxy_warm_standby) | `usize` | `16` | `✘` |
+| [`me_init_retry_attempts`](#me_init_retry_attempts) | `u32` | `0` | `✘` |
+| [`me2dc_fallback`](#me2dc_fallback) | `bool` | `true` | `✘` |
+| [`me2dc_fast`](#me2dc_fast) | `bool` | `false` | `✘` |
+| [`me_keepalive_enabled`](#me_keepalive_enabled) | `bool` | `true` | `✘` |
+| [`me_keepalive_interval_secs`](#me_keepalive_interval_secs) | `u64` | `8` | `✘` |
+| [`me_keepalive_jitter_secs`](#me_keepalive_jitter_secs) | `u64` | `2` | `✘` |
+| [`me_keepalive_payload_random`](#me_keepalive_payload_random) | `bool` | `true` | `✘` |
+| [`rpc_proxy_req_every`](#rpc_proxy_req_every) | `u64` | `0` | `✘` |
+| [`me_writer_cmd_channel_capacity`](#me_writer_cmd_channel_capacity) | `usize` | `4096` | `✘` |
+| [`me_route_channel_capacity`](#me_route_channel_capacity) | `usize` | `768` | `✘` |
+| [`me_c2me_channel_capacity`](#me_c2me_channel_capacity) | `usize` | `1024` | `✘` |
+| [`me_c2me_send_timeout_ms`](#me_c2me_send_timeout_ms) | `u64` | `4000` | `✘` |
+| [`me_reader_route_data_wait_ms`](#me_reader_route_data_wait_ms) | `u64` | `2` | `✔` |
+| [`me_d2c_flush_batch_max_frames`](#me_d2c_flush_batch_max_frames) | `usize` | `32` | `✔` |
+| [`me_d2c_flush_batch_max_bytes`](#me_d2c_flush_batch_max_bytes) | `usize` | `131072` | `✔` |
+| [`me_d2c_flush_batch_max_delay_us`](#me_d2c_flush_batch_max_delay_us) | `u64` | `500` | `✔` |
+| [`me_d2c_ack_flush_immediate`](#me_d2c_ack_flush_immediate) | `bool` | `true` | `✔` |
+| [`me_quota_soft_overshoot_bytes`](#me_quota_soft_overshoot_bytes) | `u64` | `65536` | `✔` |
+| [`me_d2c_frame_buf_shrink_threshold_bytes`](#me_d2c_frame_buf_shrink_threshold_bytes) | `usize` | `262144` | `✔` |
+| [`direct_relay_copy_buf_c2s_bytes`](#direct_relay_copy_buf_c2s_bytes) | `usize` | `65536` | `✔` |
+| [`direct_relay_copy_buf_s2c_bytes`](#direct_relay_copy_buf_s2c_bytes) | `usize` | `262144` | `✔` |
+| [`crypto_pending_buffer`](#crypto_pending_buffer) | `usize` | `262144` | `✘` |
+| [`max_client_frame`](#max_client_frame) | `usize` | `16777216` | `✘` |
+| [`desync_all_full`](#desync_all_full) | `bool` | `false` | `✔` |
+| [`beobachten`](#beobachten) | `bool` | `true` | `✘` |
+| [`beobachten_minutes`](#beobachten_minutes) | `u64` | `10` | `✘` |
+| [`beobachten_flush_secs`](#beobachten_flush_secs) | `u64` | `15` | `✘` |
+| [`beobachten_file`](#beobachten_file) | `String` | `"cache/beobachten.txt"` | `✘` |
+| [`hardswap`](#hardswap) | `bool` | `true` | `✔` |
+| [`me_warmup_stagger_enabled`](#me_warmup_stagger_enabled) | `bool` | `true` | `✘` |
+| [`me_warmup_step_delay_ms`](#me_warmup_step_delay_ms) | `u64` | `500` | `✘` |
+| [`me_warmup_step_jitter_ms`](#me_warmup_step_jitter_ms) | `u64` | `300` | `✘` |
+| [`me_reconnect_max_concurrent_per_dc`](#me_reconnect_max_concurrent_per_dc) | `u32` | `8` | `✘` |
+| [`me_reconnect_backoff_base_ms`](#me_reconnect_backoff_base_ms) | `u64` | `500` | `✘` |
+| [`me_reconnect_backoff_cap_ms`](#me_reconnect_backoff_cap_ms) | `u64` | `30000` | `✘` |
+| [`me_reconnect_fast_retry_count`](#me_reconnect_fast_retry_count) | `u32` | `16` | `✘` |
+| [`me_single_endpoint_shadow_writers`](#me_single_endpoint_shadow_writers) | `u8` | `2` | `✔` |
+| [`me_single_endpoint_outage_mode_enabled`](#me_single_endpoint_outage_mode_enabled) | `bool` | `true` | `✔` |
+| [`me_single_endpoint_outage_disable_quarantine`](#me_single_endpoint_outage_disable_quarantine) | `bool` | `true` | `✔` |
+| [`me_single_endpoint_outage_backoff_min_ms`](#me_single_endpoint_outage_backoff_min_ms) | `u64` | `250` | `✔` |
+| [`me_single_endpoint_outage_backoff_max_ms`](#me_single_endpoint_outage_backoff_max_ms) | `u64` | `3000` | `✔` |
+| [`me_single_endpoint_shadow_rotate_every_secs`](#me_single_endpoint_shadow_rotate_every_secs) | `u64` | `900` | `✔` |
+| [`me_floor_mode`](#me_floor_mode) | `"static"` or `"adaptive"` | `"adaptive"` | `✔` |
+| [`me_adaptive_floor_idle_secs`](#me_adaptive_floor_idle_secs) | `u64` | `90` | `✔` |
+| [`me_adaptive_floor_min_writers_single_endpoint`](#me_adaptive_floor_min_writers_single_endpoint) | `u8` | `1` | `✔` |
+| [`me_adaptive_floor_min_writers_multi_endpoint`](#me_adaptive_floor_min_writers_multi_endpoint) | `u8` | `1` | `✔` |
+| [`me_adaptive_floor_recover_grace_secs`](#me_adaptive_floor_recover_grace_secs) | `u64` | `180` | `✔` |
+| [`me_adaptive_floor_writers_per_core_total`](#me_adaptive_floor_writers_per_core_total) | `u16` | `48` | `✔` |
+| [`me_adaptive_floor_cpu_cores_override`](#me_adaptive_floor_cpu_cores_override) | `u16` | `0` | `✔` |
+| [`me_adaptive_floor_max_extra_writers_single_per_core`](#me_adaptive_floor_max_extra_writers_single_per_core) | `u16` | `1` | `✔` |
+| [`me_adaptive_floor_max_extra_writers_multi_per_core`](#me_adaptive_floor_max_extra_writers_multi_per_core) | `u16` | `2` | `✔` |
+| [`me_adaptive_floor_max_active_writers_per_core`](#me_adaptive_floor_max_active_writers_per_core) | `u16` | `64` | `✔` |
+| [`me_adaptive_floor_max_warm_writers_per_core`](#me_adaptive_floor_max_warm_writers_per_core) | `u16` | `64` | `✔` |
+| [`me_adaptive_floor_max_active_writers_global`](#me_adaptive_floor_max_active_writers_global) | `u32` | `256` | `✔` |
+| [`me_adaptive_floor_max_warm_writers_global`](#me_adaptive_floor_max_warm_writers_global) | `u32` | `256` | `✔` |
+| [`upstream_connect_retry_attempts`](#upstream_connect_retry_attempts) | `u32` | `2` | `✘` |
+| [`upstream_connect_retry_backoff_ms`](#upstream_connect_retry_backoff_ms) | `u64` | `100` | `✘` |
+| [`upstream_connect_budget_ms`](#upstream_connect_budget_ms) | `u64` | `3000` | `✘` |
+| [`tg_connect`](#tg_connect) | `u64` | `10` | `✘` |
+| [`upstream_unhealthy_fail_threshold`](#upstream_unhealthy_fail_threshold) | `u32` | `5` | `✘` |
+| [`upstream_connect_failfast_hard_errors`](#upstream_connect_failfast_hard_errors) | `bool` | `false` | `✘` |
+| [`stun_iface_mismatch_ignore`](#stun_iface_mismatch_ignore) | `bool` | `false` | `✘` |
+| [`unknown_dc_log_path`](#unknown_dc_log_path) | `String` | `"unknown-dc.txt"` | `✘` |
+| [`unknown_dc_file_log_enabled`](#unknown_dc_file_log_enabled) | `bool` | `false` | `✘` |
+| [`log_level`](#log_level) | `"debug"`, `"verbose"`, `"normal"`, or `"silent"` | `"normal"` | `✔` |
+| [`disable_colors`](#disable_colors) | `bool` | `false` | `✘` |
+| [`me_socks_kdf_policy`](#me_socks_kdf_policy) | `"strict"` or `"compat"` | `"strict"` | `✔` |
+| [`me_route_backpressure_enabled`](#me_route_backpressure_enabled) | `bool` | `false` | `✔` |
+| [`me_route_fairshare_enabled`](#me_route_fairshare_enabled) | `bool` | `false` | `✔` |
+| [`me_route_backpressure_base_timeout_ms`](#me_route_backpressure_base_timeout_ms) | `u64` | `25` | `✔` |
+| [`me_route_backpressure_high_timeout_ms`](#me_route_backpressure_high_timeout_ms) | `u64` | `120` | `✔` |
+| [`me_route_backpressure_high_watermark_pct`](#me_route_backpressure_high_watermark_pct) | `u8` | `80` | `✔` |
+| [`me_health_interval_ms_unhealthy`](#me_health_interval_ms_unhealthy) | `u64` | `1000` | `✔` |
+| [`me_health_interval_ms_healthy`](#me_health_interval_ms_healthy) | `u64` | `3000` | `✔` |
+| [`me_admission_poll_ms`](#me_admission_poll_ms) | `u64` | `1000` | `✔` |
+| [`me_warn_rate_limit_ms`](#me_warn_rate_limit_ms) | `u64` | `5000` | `✔` |
+| [`me_route_no_writer_mode`](#me_route_no_writer_mode) | `"async_recovery_failfast"`, `"inline_recovery_legacy"`, or `"hybrid_async_persistent"` | `"hybrid_async_persistent"` | `✘` |
+| [`me_route_no_writer_wait_ms`](#me_route_no_writer_wait_ms) | `u64` | `250` | `✘` |
+| [`me_route_hybrid_max_wait_ms`](#me_route_hybrid_max_wait_ms) | `u64` | `3000` | `✘` |
+| [`me_route_blocking_send_timeout_ms`](#me_route_blocking_send_timeout_ms) | `u64` | `250` | `✘` |
+| [`me_route_inline_recovery_attempts`](#me_route_inline_recovery_attempts) | `u32` | `3` | `✘` |
+| [`me_route_inline_recovery_wait_ms`](#me_route_inline_recovery_wait_ms) | `u64` | `3000` | `✘` |
+| [`fast_mode_min_tls_record`](#fast_mode_min_tls_record) | `usize` | `0` | `✘` |
+| [`update_every`](#update_every) | `u64` | `300` | `✔` |
+| [`me_reinit_every_secs`](#me_reinit_every_secs) | `u64` | `900` | `✔` |
+| [`me_hardswap_warmup_delay_min_ms`](#me_hardswap_warmup_delay_min_ms) | `u64` | `1000` | `✔` |
+| [`me_hardswap_warmup_delay_max_ms`](#me_hardswap_warmup_delay_max_ms) | `u64` | `2000` | `✔` |
+| [`me_hardswap_warmup_extra_passes`](#me_hardswap_warmup_extra_passes) | `u8` | `3` | `✔` |
+| [`me_hardswap_warmup_pass_backoff_base_ms`](#me_hardswap_warmup_pass_backoff_base_ms) | `u64` | `500` | `✔` |
+| [`me_config_stable_snapshots`](#me_config_stable_snapshots) | `u8` | `2` | `✔` |
+| [`me_config_apply_cooldown_secs`](#me_config_apply_cooldown_secs) | `u64` | `300` | `✔` |
+| [`me_snapshot_require_http_2xx`](#me_snapshot_require_http_2xx) | `bool` | `true` | `✔` |
+| [`me_snapshot_reject_empty_map`](#me_snapshot_reject_empty_map) | `bool` | `true` | `✔` |
+| [`me_snapshot_min_proxy_for_lines`](#me_snapshot_min_proxy_for_lines) | `u32` | `1` | `✔` |
+| [`proxy_secret_stable_snapshots`](#proxy_secret_stable_snapshots) | `u8` | `2` | `✔` |
+| [`proxy_secret_rotate_runtime`](#proxy_secret_rotate_runtime) | `bool` | `true` | `✔` |
+| [`me_secret_atomic_snapshot`](#me_secret_atomic_snapshot) | `bool` | `true` | `✔` |
+| [`proxy_secret_len_max`](#proxy_secret_len_max) | `usize` | `256` | `✔` |
+| [`me_pool_drain_ttl_secs`](#me_pool_drain_ttl_secs) | `u64` | `90` | `✔` |
+| [`me_instadrain`](#me_instadrain) | `bool` | `false` | `✔` |
+| [`me_pool_drain_threshold`](#me_pool_drain_threshold) | `u64` | `32` | `✔` |
+| [`me_pool_drain_soft_evict_enabled`](#me_pool_drain_soft_evict_enabled) | `bool` | `true` | `✘` |
+| [`me_pool_drain_soft_evict_grace_secs`](#me_pool_drain_soft_evict_grace_secs) | `u64` | `10` | `✘` |
+| [`me_pool_drain_soft_evict_per_writer`](#me_pool_drain_soft_evict_per_writer) | `u8` | `2` | `✘` |
+| [`me_pool_drain_soft_evict_budget_per_core`](#me_pool_drain_soft_evict_budget_per_core) | `u16` | `16` | `✘` |
+| [`me_pool_drain_soft_evict_cooldown_ms`](#me_pool_drain_soft_evict_cooldown_ms) | `u64` | `1000` | `✘` |
+| [`me_bind_stale_mode`](#me_bind_stale_mode) | `"never"`, `"ttl"`, or `"always"` | `"ttl"` | `✔` |
+| [`me_bind_stale_ttl_secs`](#me_bind_stale_ttl_secs) | `u64` | `90` | `✔` |
+| [`me_pool_min_fresh_ratio`](#me_pool_min_fresh_ratio) | `f32` | `0.8` | `✔` |
+| [`me_reinit_drain_timeout_secs`](#me_reinit_drain_timeout_secs) | `u64` | `90` | `✔` |
+| [`proxy_secret_auto_reload_secs`](#proxy_secret_auto_reload_secs) | `u64` | `3600` | `✔` |
+| [`proxy_config_auto_reload_secs`](#proxy_config_auto_reload_secs) | `u64` | `3600` | `✔` |
+| [`me_reinit_singleflight`](#me_reinit_singleflight) | `bool` | `true` | `✔` |
+| [`me_reinit_trigger_channel`](#me_reinit_trigger_channel) | `usize` | `64` | `✘` |
+| [`me_reinit_coalesce_window_ms`](#me_reinit_coalesce_window_ms) | `u64` | `200` | `✔` |
+| [`me_deterministic_writer_sort`](#me_deterministic_writer_sort) | `bool` | `true` | `✔` |
+| [`me_writer_pick_mode`](#me_writer_pick_mode) | `"sorted_rr"` or `"p2c"` | `"p2c"` | `✔` |
+| [`me_writer_pick_sample_size`](#me_writer_pick_sample_size) | `u8` | `3` | `✔` |
+| [`ntp_check`](#ntp_check) | `bool` | `true` | `✘` |
+| [`ntp_servers`](#ntp_servers) | `String[]` | `["pool.ntp.org"]` | `✘` |
+| [`auto_degradation_enabled`](#auto_degradation_enabled) | `bool` | `true` | `✘` |
+| [`degradation_min_unavailable_dc_groups`](#degradation_min_unavailable_dc_groups) | `u8` | `2` | `✘` |
+| [`rst_on_close`](#rst_on_close) | `"off"`, `"errors"`, or `"always"` | `"off"` | `✘` |
 
 ## data_path
   - **Constraints / validation**: `String` (optional).

docs/Config_params/CONFIG_PARAMS.ru.md

@@ -224,6 +224,151 @@
 | [`auto_degradation_enabled`](#auto_degradation_enabled) | `bool` | `true` |
 | [`degradation_min_unavailable_dc_groups`](#degradation_min_unavailable_dc_groups) | `u8` | `2` |
 | [`rst_on_close`](#rst_on_close) | `"off"`, `"errors"` или `"always"` | `"off"` |
+| Ключ | Тип | По умолчанию | Hot-Reload |
+| --- | ---- | ------- | ---------- |
+| [`data_path`](#data_path) | `String` | — | `✘` |
+| [`quota_state_path`](#quota_state_path) | `Path` | `"telemt.limit.json"` | `✘` |
+| [`config_strict`](#config_strict) | `bool` | `false` | `✘` |
+| [`prefer_ipv6`](#prefer_ipv6) | `bool` | `false` | `✘` |
+| [`fast_mode`](#fast_mode) | `bool` | `true` | `✘` |
+| [`use_middle_proxy`](#use_middle_proxy) | `bool` | `true` | `✘` |
+| [`proxy_secret_path`](#proxy_secret_path) | `String` | `"proxy-secret"` | `✘` |
+| [`proxy_secret_url`](#proxy_secret_url) | `String` | `"https://core.telegram.org/getProxySecret"` | `✘` |
+| [`proxy_config_v4_cache_path`](#proxy_config_v4_cache_path) | `String` | `"cache/proxy-config-v4.txt"` | `✘` |
+| [`proxy_config_v4_url`](#proxy_config_v4_url) | `String` | `"https://core.telegram.org/getProxyConfig"` | `✘` |
+| [`proxy_config_v6_cache_path`](#proxy_config_v6_cache_path) | `String` | `"cache/proxy-config-v6.txt"` | `✘` |
+| [`proxy_config_v6_url`](#proxy_config_v6_url) | `String` | `"https://core.telegram.org/getProxyConfigV6"` | `✘` |
+| [`ad_tag`](#ad_tag) | `String` | — | `✔` |
+| [`middle_proxy_nat_ip`](#middle_proxy_nat_ip) | `IpAddr` | — | `✘` |
+| [`middle_proxy_nat_probe`](#middle_proxy_nat_probe) | `bool` | `true` | `✘` |
+| [`middle_proxy_nat_stun`](#middle_proxy_nat_stun) | `String` | — | `✘` |
+| [`middle_proxy_nat_stun_servers`](#middle_proxy_nat_stun_servers) | `String[]` | `[]` | `✘` |
+| [`stun_nat_probe_concurrency`](#stun_nat_probe_concurrency) | `usize` | `8` | `✘` |
+| [`middle_proxy_pool_size`](#middle_proxy_pool_size) | `usize` | `8` | `✘` |
+| [`middle_proxy_warm_standby`](#middle_proxy_warm_standby) | `usize` | `16` | `✘` |
+| [`me_init_retry_attempts`](#me_init_retry_attempts) | `u32` | `0` | `✘` |
+| [`me2dc_fallback`](#me2dc_fallback) | `bool` | `true` | `✘` |
+| [`me2dc_fast`](#me2dc_fast) | `bool` | `false` | `✘` |
+| [`me_keepalive_enabled`](#me_keepalive_enabled) | `bool` | `true` | `✘` |
+| [`me_keepalive_interval_secs`](#me_keepalive_interval_secs) | `u64` | `8` | `✘` |
+| [`me_keepalive_jitter_secs`](#me_keepalive_jitter_secs) | `u64` | `2` | `✘` |
+| [`me_keepalive_payload_random`](#me_keepalive_payload_random) | `bool` | `true` | `✘` |
+| [`rpc_proxy_req_every`](#rpc_proxy_req_every) | `u64` | `0` | `✘` |
+| [`me_writer_cmd_channel_capacity`](#me_writer_cmd_channel_capacity) | `usize` | `4096` | `✘` |
+| [`me_route_channel_capacity`](#me_route_channel_capacity) | `usize` | `768` | `✘` |
+| [`me_c2me_channel_capacity`](#me_c2me_channel_capacity) | `usize` | `1024` | `✘` |
+| [`me_c2me_send_timeout_ms`](#me_c2me_send_timeout_ms) | `u64` | `4000` | `✘` |
+| [`me_reader_route_data_wait_ms`](#me_reader_route_data_wait_ms) | `u64` | `2` | `✔` |
+| [`me_d2c_flush_batch_max_frames`](#me_d2c_flush_batch_max_frames) | `usize` | `32` | `✔` |
+| [`me_d2c_flush_batch_max_bytes`](#me_d2c_flush_batch_max_bytes) | `usize` | `131072` | `✔` |
+| [`me_d2c_flush_batch_max_delay_us`](#me_d2c_flush_batch_max_delay_us) | `u64` | `500` | `✔` |
+| [`me_d2c_ack_flush_immediate`](#me_d2c_ack_flush_immediate) | `bool` | `true` | `✔` |
+| [`me_quota_soft_overshoot_bytes`](#me_quota_soft_overshoot_bytes) | `u64` | `65536` | `✔` |
+| [`me_d2c_frame_buf_shrink_threshold_bytes`](#me_d2c_frame_buf_shrink_threshold_bytes) | `usize` | `262144` | `✔` |
+| [`direct_relay_copy_buf_c2s_bytes`](#direct_relay_copy_buf_c2s_bytes) | `usize` | `65536` | `✔` |
+| [`direct_relay_copy_buf_s2c_bytes`](#direct_relay_copy_buf_s2c_bytes) | `usize` | `262144` | `✔` |
+| [`crypto_pending_buffer`](#crypto_pending_buffer) | `usize` | `262144` | `✘` |
+| [`max_client_frame`](#max_client_frame) | `usize` | `16777216` | `✘` |
+| [`desync_all_full`](#desync_all_full) | `bool` | `false` | `✔` |
+| [`beobachten`](#beobachten) | `bool` | `true` | `✘` |
+| [`beobachten_minutes`](#beobachten_minutes) | `u64` | `10` | `✘` |
+| [`beobachten_flush_secs`](#beobachten_flush_secs) | `u64` | `15` | `✘` |
+| [`beobachten_file`](#beobachten_file) | `String` | `"cache/beobachten.txt"` | `✘` |
+| [`hardswap`](#hardswap) | `bool` | `true` | `✔` |
+| [`me_warmup_stagger_enabled`](#me_warmup_stagger_enabled) | `bool` | `true` | `✘` |
+| [`me_warmup_step_delay_ms`](#me_warmup_step_delay_ms) | `u64` | `500` | `✘` |
+| [`me_warmup_step_jitter_ms`](#me_warmup_step_jitter_ms) | `u64` | `300` | `✘` |
+| [`me_reconnect_max_concurrent_per_dc`](#me_reconnect_max_concurrent_per_dc) | `u32` | `8` | `✘` |
+| [`me_reconnect_backoff_base_ms`](#me_reconnect_backoff_base_ms) | `u64` | `500` | `✘` |
+| [`me_reconnect_backoff_cap_ms`](#me_reconnect_backoff_cap_ms) | `u64` | `30000` | `✘` |
+| [`me_reconnect_fast_retry_count`](#me_reconnect_fast_retry_count) | `u32` | `16` | `✘` |
+| [`me_single_endpoint_shadow_writers`](#me_single_endpoint_shadow_writers) | `u8` | `2` | `✔` |
+| [`me_single_endpoint_outage_mode_enabled`](#me_single_endpoint_outage_mode_enabled) | `bool` | `true` | `✔` |
+| [`me_single_endpoint_outage_disable_quarantine`](#me_single_endpoint_outage_disable_quarantine) | `bool` | `true` | `✔` |
+| [`me_single_endpoint_outage_backoff_min_ms`](#me_single_endpoint_outage_backoff_min_ms) | `u64` | `250` | `✔` |
+| [`me_single_endpoint_outage_backoff_max_ms`](#me_single_endpoint_outage_backoff_max_ms) | `u64` | `3000` | `✔` |
+| [`me_single_endpoint_shadow_rotate_every_secs`](#me_single_endpoint_shadow_rotate_every_secs) | `u64` | `900` | `✔` |
+| [`me_floor_mode`](#me_floor_mode) | `"static"` or `"adaptive"` | `"adaptive"` | `✔` |
+| [`me_adaptive_floor_idle_secs`](#me_adaptive_floor_idle_secs) | `u64` | `90` | `✔` |
+| [`me_adaptive_floor_min_writers_single_endpoint`](#me_adaptive_floor_min_writers_single_endpoint) | `u8` | `1` | `✔` |
+| [`me_adaptive_floor_min_writers_multi_endpoint`](#me_adaptive_floor_min_writers_multi_endpoint) | `u8` | `1` | `✔` |
+| [`me_adaptive_floor_recover_grace_secs`](#me_adaptive_floor_recover_grace_secs) | `u64` | `180` | `✔` |
+| [`me_adaptive_floor_writers_per_core_total`](#me_adaptive_floor_writers_per_core_total) | `u16` | `48` | `✔` |
+| [`me_adaptive_floor_cpu_cores_override`](#me_adaptive_floor_cpu_cores_override) | `u16` | `0` | `✔` |
+| [`me_adaptive_floor_max_extra_writers_single_per_core`](#me_adaptive_floor_max_extra_writers_single_per_core) | `u16` | `1` | `✔` |
+| [`me_adaptive_floor_max_extra_writers_multi_per_core`](#me_adaptive_floor_max_extra_writers_multi_per_core) | `u16` | `2` | `✔` |
+| [`me_adaptive_floor_max_active_writers_per_core`](#me_adaptive_floor_max_active_writers_per_core) | `u16` | `64` | `✔` |
+| [`me_adaptive_floor_max_warm_writers_per_core`](#me_adaptive_floor_max_warm_writers_per_core) | `u16` | `64` | `✔` |
+| [`me_adaptive_floor_max_active_writers_global`](#me_adaptive_floor_max_active_writers_global) | `u32` | `256` | `✔` |
+| [`me_adaptive_floor_max_warm_writers_global`](#me_adaptive_floor_max_warm_writers_global) | `u32` | `256` | `✔` |
+| [`upstream_connect_retry_attempts`](#upstream_connect_retry_attempts) | `u32` | `2` | `✘` |
+| [`upstream_connect_retry_backoff_ms`](#upstream_connect_retry_backoff_ms) | `u64` | `100` | `✘` |
+| [`upstream_connect_budget_ms`](#upstream_connect_budget_ms) | `u64` | `3000` | `✘` |
+| [`tg_connect`](#tg_connect) | `u64` | `10` | `✘` |
+| [`upstream_unhealthy_fail_threshold`](#upstream_unhealthy_fail_threshold) | `u32` | `5` | `✘` |
+| [`upstream_connect_failfast_hard_errors`](#upstream_connect_failfast_hard_errors) | `bool` | `false` | `✘` |
+| [`stun_iface_mismatch_ignore`](#stun_iface_mismatch_ignore) | `bool` | `false` | `✘` |
+| [`unknown_dc_log_path`](#unknown_dc_log_path) | `String` | `"unknown-dc.txt"` | `✘` |
+| [`unknown_dc_file_log_enabled`](#unknown_dc_file_log_enabled) | `bool` | `false` | `✘` |
+| [`log_level`](#log_level) | `"debug"`, `"verbose"`, `"normal"`, or `"silent"` | `"normal"` | `✔` |
+| [`disable_colors`](#disable_colors) | `bool` | `false` | `✘` |
+| [`me_socks_kdf_policy`](#me_socks_kdf_policy) | `"strict"` or `"compat"` | `"strict"` | `✔` |
+| [`me_route_backpressure_enabled`](#me_route_backpressure_enabled) | `bool` | `false` | `✔` |
+| [`me_route_fairshare_enabled`](#me_route_fairshare_enabled) | `bool` | `false` | `✔` |
+| [`me_route_backpressure_base_timeout_ms`](#me_route_backpressure_base_timeout_ms) | `u64` | `25` | `✔` |
+| [`me_route_backpressure_high_timeout_ms`](#me_route_backpressure_high_timeout_ms) | `u64` | `120` | `✔` |
+| [`me_route_backpressure_high_watermark_pct`](#me_route_backpressure_high_watermark_pct) | `u8` | `80` | `✔` |
+| [`me_health_interval_ms_unhealthy`](#me_health_interval_ms_unhealthy) | `u64` | `1000` | `✔` |
+| [`me_health_interval_ms_healthy`](#me_health_interval_ms_healthy) | `u64` | `3000` | `✔` |
+| [`me_admission_poll_ms`](#me_admission_poll_ms) | `u64` | `1000` | `✔` |
+| [`me_warn_rate_limit_ms`](#me_warn_rate_limit_ms) | `u64` | `5000` | `✔` |
+| [`me_route_no_writer_mode`](#me_route_no_writer_mode) | `"async_recovery_failfast"`, `"inline_recovery_legacy"`, or `"hybrid_async_persistent"` | `"hybrid_async_persistent"` | `✘` |
+| [`me_route_no_writer_wait_ms`](#me_route_no_writer_wait_ms) | `u64` | `250` | `✘` |
+| [`me_route_hybrid_max_wait_ms`](#me_route_hybrid_max_wait_ms) | `u64` | `3000` | `✘` |
+| [`me_route_blocking_send_timeout_ms`](#me_route_blocking_send_timeout_ms) | `u64` | `250` | `✘` |
+| [`me_route_inline_recovery_attempts`](#me_route_inline_recovery_attempts) | `u32` | `3` | `✘` |
+| [`me_route_inline_recovery_wait_ms`](#me_route_inline_recovery_wait_ms) | `u64` | `3000` | `✘` |
+| [`fast_mode_min_tls_record`](#fast_mode_min_tls_record) | `usize` | `0` | `✘` |
+| [`update_every`](#update_every) | `u64` | `300` | `✔` |
+| [`me_reinit_every_secs`](#me_reinit_every_secs) | `u64` | `900` | `✔` |
+| [`me_hardswap_warmup_delay_min_ms`](#me_hardswap_warmup_delay_min_ms) | `u64` | `1000` | `✔` |
+| [`me_hardswap_warmup_delay_max_ms`](#me_hardswap_warmup_delay_max_ms) | `u64` | `2000` | `✔` |
+| [`me_hardswap_warmup_extra_passes`](#me_hardswap_warmup_extra_passes) | `u8` | `3` | `✔` |
+| [`me_hardswap_warmup_pass_backoff_base_ms`](#me_hardswap_warmup_pass_backoff_base_ms) | `u64` | `500` | `✔` |
+| [`me_config_stable_snapshots`](#me_config_stable_snapshots) | `u8` | `2` | `✔` |
+| [`me_config_apply_cooldown_secs`](#me_config_apply_cooldown_secs) | `u64` | `300` | `✔` |
+| [`me_snapshot_require_http_2xx`](#me_snapshot_require_http_2xx) | `bool` | `true` | `✔` |
+| [`me_snapshot_reject_empty_map`](#me_snapshot_reject_empty_map) | `bool` | `true` | `✔` |
+| [`me_snapshot_min_proxy_for_lines`](#me_snapshot_min_proxy_for_lines) | `u32` | `1` | `✔` |
+| [`proxy_secret_stable_snapshots`](#proxy_secret_stable_snapshots) | `u8` | `2` | `✔` |
+| [`proxy_secret_rotate_runtime`](#proxy_secret_rotate_runtime) | `bool` | `true` | `✔` |
+| [`me_secret_atomic_snapshot`](#me_secret_atomic_snapshot) | `bool` | `true` | `✔` |
+| [`proxy_secret_len_max`](#proxy_secret_len_max) | `usize` | `256` | `✔` |
+| [`me_pool_drain_ttl_secs`](#me_pool_drain_ttl_secs) | `u64` | `90` | `✔` |
+| [`me_instadrain`](#me_instadrain) | `bool` | `false` | `✔` |
+| [`me_pool_drain_threshold`](#me_pool_drain_threshold) | `u64` | `32` | `✔` |
+| [`me_pool_drain_soft_evict_enabled`](#me_pool_drain_soft_evict_enabled) | `bool` | `true` | `✘` |
+| [`me_pool_drain_soft_evict_grace_secs`](#me_pool_drain_soft_evict_grace_secs) | `u64` | `10` | `✘` |
+| [`me_pool_drain_soft_evict_per_writer`](#me_pool_drain_soft_evict_per_writer) | `u8` | `2` | `✘` |
+| [`me_pool_drain_soft_evict_budget_per_core`](#me_pool_drain_soft_evict_budget_per_core) | `u16` | `16` | `✘` |
+| [`me_pool_drain_soft_evict_cooldown_ms`](#me_pool_drain_soft_evict_cooldown_ms) | `u64` | `1000` | `✘` |
+| [`me_bind_stale_mode`](#me_bind_stale_mode) | `"never"`, `"ttl"`, or `"always"` | `"ttl"` | `✔` |
+| [`me_bind_stale_ttl_secs`](#me_bind_stale_ttl_secs) | `u64` | `90` | `✔` |
+| [`me_pool_min_fresh_ratio`](#me_pool_min_fresh_ratio) | `f32` | `0.8` | `✔` |
+| [`me_reinit_drain_timeout_secs`](#me_reinit_drain_timeout_secs) | `u64` | `90` | `✔` |
+| [`proxy_secret_auto_reload_secs`](#proxy_secret_auto_reload_secs) | `u64` | `3600` | `✔` |
+| [`proxy_config_auto_reload_secs`](#proxy_config_auto_reload_secs) | `u64` | `3600` | `✔` |
+| [`me_reinit_singleflight`](#me_reinit_singleflight) | `bool` | `true` | `✔` |
+| [`me_reinit_trigger_channel`](#me_reinit_trigger_channel) | `usize` | `64` | `✘` |
+| [`me_reinit_coalesce_window_ms`](#me_reinit_coalesce_window_ms) | `u64` | `200` | `✔` |
+| [`me_deterministic_writer_sort`](#me_deterministic_writer_sort) | `bool` | `true` | `✔` |
+| [`me_writer_pick_mode`](#me_writer_pick_mode) | `"sorted_rr"` or `"p2c"` | `"p2c"` | `✔` |
+| [`me_writer_pick_sample_size`](#me_writer_pick_sample_size) | `u8` | `3` | `✔` |
+| [`ntp_check`](#ntp_check) | `bool` | `true` | `✘` |
+| [`ntp_servers`](#ntp_servers) | `String[]` | `["pool.ntp.org"]` | `✘` |
+| [`auto_degradation_enabled`](#auto_degradation_enabled) | `bool` | `true` | `✘` |
+| [`degradation_min_unavailable_dc_groups`](#degradation_min_unavailable_dc_groups) | `u8` | `2` | `✘` |
+| [`rst_on_close`](#rst_on_close) | `"off"`, `"errors"` или `"always"` | `"off"` | `✘` |
 
 ## data_path
   - **Ограничения / валидация**: `String` (необязательный параметр).

docs/FAQ.en.md

@@ -172,7 +172,7 @@ Those cross-DC requests are normal and happen constantly.
 > If your home DC is DC2 and DC2 goes down, you **cannot** reach DC5 even though DC5 itself is perfectly healthy.  
 > The client has no valid session to route the request through.  
 
-This is also why it is required for MTProxy to reach Telegram's DC infrastructure as a whole.  
+This is also why an MTProxy only needs to reach Telegram's DC infrastructure as a whole.  
 The proxy itself doesn't care which DC your account lives on. The client negotiates the correct DC through the proxy after connecting.
 
 ### How many people can use one link

docs/FAQ.ru.md

@@ -157,7 +157,7 @@ https://github.com/telemt/telemt/discussions/167
 
 ## Как клиенты взаимодействуют с дата-центрами Telegram
 При регистрации аккаунта Telegram он навсегда привязывается к одному из дата-центров (DC).  
-Telegram заранее определяет к какому DC привязать аккаунт исходя из региона, к которому относится номер телефона.  
+Telegram заранее определяет к какому DC привязать аккаунт исходя из региона, к которому относиться номер телефона.  
 Этот DC становится вашим **домашним**: именно там хранится весь контент, который вы загружаете (фото, видео, файлы, сообщения).  
 И именно на нем клиент авторизуется при каждом подключении.  
 
@@ -170,7 +170,7 @@ Telegram заранее определяет к какому DC привязат
 > Если ваш домашний DC — DC2, и DC2 лежит, вы **не сможете** достучаться  и до DC5, даже если сам DC5 полностью исправен.  
 > У клиента просто нет валидной сессии, через которую можно было бы направить запрос.  
 
-По той же причине MTProxy необходимо иметь доступ к инфраструктуре Telegram целиком, а не частично.  
+По той же причине MTProxy достаточно иметь доступ к инфраструктуре Telegram в целом.  
 Cамому MTProxy всё равно, на каком DC живёт ваш аккаунт. Клиент cам договаривается о нужном DC через прокси уже после подключения.  
 
 ## Что такое dd и ee в контексте MTProxy?

src/proxy/middle_relay/idle.rs

@@ -1,4 +1,5 @@
 use super::*;
+use dashmap::DashMap;
 
 mod read;
 
@@ -10,10 +11,10 @@ pub(crate) use self::read::{
 
 #[derive(Default)]
 pub(crate) struct RelayIdleCandidateRegistry {
-    pub(in crate::proxy::middle_relay) by_conn_id: HashMap<u64, RelayIdleCandidateMeta>,
+    pub(in crate::proxy::middle_relay) by_conn_id: DashMap<u64, RelayIdleCandidateMeta>,
-    pub(in crate::proxy::middle_relay) ordered: BTreeSet<(u64, u64)>,
+    pub(in crate::proxy::middle_relay) ordered: parking_lot::Mutex<BTreeSet<(u64, u64)>>,
-    pressure_event_seq: u64,
+    pressure_event_seq: AtomicU64,
-    pressure_consumed_seq: u64,
+    pressure_consumed_seq: AtomicU64,
 }
 
 /// Queue metadata used to preserve FIFO ordering for idle relay eviction.
@@ -23,25 +24,10 @@ pub(in crate::proxy::middle_relay) struct RelayIdleCandidateMeta {
     pub(in crate::proxy::middle_relay) mark_pressure_seq: u64,
 }
 
-pub(super) fn relay_idle_candidate_registry_lock_in(
-    shared: &ProxySharedState,
-) -> std::sync::MutexGuard<'_, RelayIdleCandidateRegistry> {
-    let registry = &shared.middle_relay.relay_idle_registry;
-    match registry.lock() {
-        Ok(guard) => guard,
-        Err(poisoned) => {
-            let mut guard = poisoned.into_inner();
-            *guard = RelayIdleCandidateRegistry::default();
-            registry.clear_poison();
-            guard
-        }
-    }
-}
-
 pub(super) fn mark_relay_idle_candidate_in(shared: &ProxySharedState, conn_id: u64) -> bool {
-    let mut guard = relay_idle_candidate_registry_lock_in(shared);
+    let registry = &shared.middle_relay.relay_idle_registry;
 
-    if guard.by_conn_id.contains_key(&conn_id) {
+    if registry.by_conn_id.contains_key(&conn_id) {
         return false;
     }
 
@@ -52,24 +38,35 @@ pub(super) fn mark_relay_idle_candidate_in(shared: &ProxySharedState, conn_id: u
         .saturating_add(1);
     let meta = RelayIdleCandidateMeta {
         mark_order_seq,
-        mark_pressure_seq: guard.pressure_event_seq,
+        mark_pressure_seq: registry.pressure_event_seq.load(Ordering::Relaxed),
     };
-    guard.by_conn_id.insert(conn_id, meta);
+    match registry.by_conn_id.entry(conn_id) {
-    guard.ordered.insert((meta.mark_order_seq, conn_id));
+        dashmap::mapref::entry::Entry::Occupied(_) => false,
-    true
+        dashmap::mapref::entry::Entry::Vacant(entry) => {
+            entry.insert(meta);
+            registry.ordered.lock().insert((meta.mark_order_seq, conn_id));
+            true
+        }
+    }
 }
 
 pub(super) fn clear_relay_idle_candidate_in(shared: &ProxySharedState, conn_id: u64) {
-    let mut guard = relay_idle_candidate_registry_lock_in(shared);
+    let registry = &shared.middle_relay.relay_idle_registry;
 
-    if let Some(meta) = guard.by_conn_id.remove(&conn_id) {
+    if let Some((_, meta)) = registry.by_conn_id.remove(&conn_id) {
-        guard.ordered.remove(&(meta.mark_order_seq, conn_id));
+        registry
+            .ordered
+            .lock()
+            .remove(&(meta.mark_order_seq, conn_id));
     }
 }
 
 pub(super) fn note_relay_pressure_event_in(shared: &ProxySharedState) {
-    let mut guard = relay_idle_candidate_registry_lock_in(shared);
+    shared
-    guard.pressure_event_seq = guard.pressure_event_seq.wrapping_add(1);
+        .middle_relay
+        .relay_idle_registry
+        .pressure_event_seq
+        .fetch_add(1, Ordering::Relaxed);
 }
 
 pub(crate) fn note_global_relay_pressure(shared: &ProxySharedState) {
@@ -77,8 +74,11 @@ pub(crate) fn note_global_relay_pressure(shared: &ProxySharedState) {
 }
 
 pub(super) fn relay_pressure_event_seq_in(shared: &ProxySharedState) -> u64 {
-    let guard = relay_idle_candidate_registry_lock_in(shared);
+    shared
-    guard.pressure_event_seq
+        .middle_relay
+        .relay_idle_registry
+        .pressure_event_seq
+        .load(Ordering::Relaxed)
 }
 
 pub(super) fn maybe_evict_idle_candidate_on_pressure_in(
@@ -87,33 +87,43 @@ pub(super) fn maybe_evict_idle_candidate_on_pressure_in(
     seen_pressure_seq: &mut u64,
     stats: &Stats,
 ) -> bool {
-    let mut guard = relay_idle_candidate_registry_lock_in(shared);
+    let registry = &shared.middle_relay.relay_idle_registry;
 
-    let latest_pressure_seq = guard.pressure_event_seq;
+    let latest_pressure_seq = registry.pressure_event_seq.load(Ordering::Relaxed);
     if latest_pressure_seq == *seen_pressure_seq {
         return false;
     }
     *seen_pressure_seq = latest_pressure_seq;
 
-    if latest_pressure_seq == guard.pressure_consumed_seq {
+    if latest_pressure_seq == registry.pressure_consumed_seq.load(Ordering::Relaxed) {
         return false;
     }
 
-    if guard.ordered.is_empty() {
+    let oldest = {
-        guard.pressure_consumed_seq = latest_pressure_seq;
+        let mut ordered = registry.ordered.lock();
-        return false;
+        loop {
-    }
+            let Some((mark_order_seq, candidate_conn_id)) = ordered.iter().next().copied() else {
-
+                registry
-    let oldest = guard
+                    .pressure_consumed_seq
-        .ordered
+                    .store(latest_pressure_seq, Ordering::Relaxed);
-        .iter()
+                return false;
-        .next()
+            };
-        .map(|(_, candidate_conn_id)| *candidate_conn_id);
+            let Some(candidate_meta) = registry.by_conn_id.get(&candidate_conn_id) else {
+                ordered.remove(&(mark_order_seq, candidate_conn_id));
+                continue;
+            };
+            if candidate_meta.mark_order_seq != mark_order_seq {
+                ordered.remove(&(mark_order_seq, candidate_conn_id));
+                continue;
+            }
+            break Some(candidate_conn_id);
+        }
+    };
     if oldest != Some(conn_id) {
         return false;
     }
 
-    let Some(candidate_meta) = guard.by_conn_id.get(&conn_id).copied() else {
+    let Some(candidate_meta) = registry.by_conn_id.get(&conn_id).map(|entry| *entry.value()) else {
         return false;
     };
 
@@ -121,10 +131,15 @@ pub(super) fn maybe_evict_idle_candidate_on_pressure_in(
         return false;
     }
 
-    if let Some(meta) = guard.by_conn_id.remove(&conn_id) {
+    if let Some((_, meta)) = registry.by_conn_id.remove(&conn_id) {
-        guard.ordered.remove(&(meta.mark_order_seq, conn_id));
+        registry
+            .ordered
+            .lock()
+            .remove(&(meta.mark_order_seq, conn_id));
     }
-    guard.pressure_consumed_seq = latest_pressure_seq;
+    registry
+        .pressure_consumed_seq
+        .store(latest_pressure_seq, Ordering::Relaxed);
     stats.increment_relay_pressure_evict_total();
     true
 }
@@ -220,72 +235,32 @@ pub(crate) fn mark_relay_idle_candidate_for_testing(
     shared: &ProxySharedState,
     conn_id: u64,
 ) -> bool {
-    let registry = &shared.middle_relay.relay_idle_registry;
+    mark_relay_idle_candidate_in(shared, conn_id)
-    let mut guard = match registry.lock() {
-        Ok(guard) => guard,
-        Err(poisoned) => {
-            let mut guard = poisoned.into_inner();
-            *guard = RelayIdleCandidateRegistry::default();
-            registry.clear_poison();
-            guard
-        }
-    };
-
-    if guard.by_conn_id.contains_key(&conn_id) {
-        return false;
-    }
-
-    let mark_order_seq = shared
-        .middle_relay
-        .relay_idle_mark_seq
-        .fetch_add(1, Ordering::Relaxed);
-    let mark_pressure_seq = guard.pressure_event_seq;
-    let meta = RelayIdleCandidateMeta {
-        mark_order_seq,
-        mark_pressure_seq,
-    };
-    guard.by_conn_id.insert(conn_id, meta);
-    guard.ordered.insert((mark_order_seq, conn_id));
-    true
 }
 
 #[cfg(test)]
 pub(crate) fn oldest_relay_idle_candidate_for_testing(shared: &ProxySharedState) -> Option<u64> {
     let registry = &shared.middle_relay.relay_idle_registry;
-    let guard = match registry.lock() {
+    registry
-        Ok(guard) => guard,
+        .ordered
-        Err(poisoned) => {
+        .lock()
-            let mut guard = poisoned.into_inner();
+        .iter()
-            *guard = RelayIdleCandidateRegistry::default();
+        .next()
-            registry.clear_poison();
+        .map(|(_, conn_id)| *conn_id)
-            guard
-        }
-    };
-    guard.ordered.iter().next().map(|(_, conn_id)| *conn_id)
 }
 
 #[cfg(test)]
 pub(crate) fn clear_relay_idle_candidate_for_testing(shared: &ProxySharedState, conn_id: u64) {
-    let registry = &shared.middle_relay.relay_idle_registry;
+    clear_relay_idle_candidate_in(shared, conn_id);
-    let mut guard = match registry.lock() {
-        Ok(guard) => guard,
-        Err(poisoned) => {
-            let mut guard = poisoned.into_inner();
-            *guard = RelayIdleCandidateRegistry::default();
-            registry.clear_poison();
-            guard
-        }
-    };
-    if let Some(meta) = guard.by_conn_id.remove(&conn_id) {
-        guard.ordered.remove(&(meta.mark_order_seq, conn_id));
-    }
 }
 
 #[cfg(test)]
 pub(crate) fn clear_relay_idle_pressure_state_for_testing_in_shared(shared: &ProxySharedState) {
-    if let Ok(mut guard) = shared.middle_relay.relay_idle_registry.lock() {
+    let registry = &shared.middle_relay.relay_idle_registry;
-        *guard = RelayIdleCandidateRegistry::default();
+    registry.by_conn_id.clear();
-    }
+    registry.ordered.lock().clear();
+    registry.pressure_event_seq.store(0, Ordering::Relaxed);
+    registry.pressure_consumed_seq.store(0, Ordering::Relaxed);
     shared
         .middle_relay
         .relay_idle_mark_seq
@@ -327,15 +302,10 @@ pub(crate) fn set_relay_pressure_state_for_testing(
     pressure_consumed_seq: u64,
 ) {
     let registry = &shared.middle_relay.relay_idle_registry;
-    let mut guard = match registry.lock() {
+    registry
-        Ok(guard) => guard,
+        .pressure_event_seq
-        Err(poisoned) => {
+        .store(pressure_event_seq, Ordering::Relaxed);
-            let mut guard = poisoned.into_inner();
+    registry
-            *guard = RelayIdleCandidateRegistry::default();
+        .pressure_consumed_seq
-            registry.clear_poison();
+        .store(pressure_consumed_seq, Ordering::Relaxed);
-            guard
-        }
-    };
-    guard.pressure_event_seq = pressure_event_seq;
-    guard.pressure_consumed_seq = pressure_consumed_seq;
 }

src/proxy/middle_relay/quota.rs

@@ -41,11 +41,12 @@ pub(super) async fn reserve_user_quota_with_yield(
             return Err(MiddleQuotaReserveError::DeadlineExceeded);
         }
         tokio::select! {
-            _ = tokio::time::sleep(Duration::from_millis(backoff_ms)) => {}
+            biased;
             _ = cancel.cancelled() => {
                 stats.increment_quota_acquire_cancelled_total();
                 return Err(MiddleQuotaReserveError::Cancelled);
             }
+            _ = tokio::time::sleep(Duration::from_millis(backoff_ms)) => {}
         }
         backoff_rounds = backoff_rounds.saturating_add(1);
         if backoff_rounds >= QUOTA_RESERVE_MAX_BACKOFF_ROUNDS {
@@ -128,11 +129,12 @@ pub(super) async fn wait_for_traffic_budget_or_cancel(
             return Err(ProxyError::TrafficBudgetWaitDeadlineExceeded);
         }
         tokio::select! {
-            _ = tokio::time::sleep(next_refill_delay()) => {}
+            biased;
             _ = cancel.cancelled() => {
                 stats.increment_flow_wait_middle_rate_limit_cancelled_total();
                 return Err(ProxyError::TrafficBudgetWaitCancelled);
             }
+            _ = tokio::time::sleep(next_refill_delay()) => {}
         }
         let wait_ms = wait_started_at
             .elapsed()

src/proxy/shared_state.rs

@@ -59,7 +59,7 @@ pub(crate) struct MiddleRelaySharedState {
     pub(crate) desync_hasher: RandomState,
     pub(crate) desync_full_cache_last_emit_at: Mutex<Option<Instant>>,
     pub(crate) desync_dedup_rotation_state: Mutex<DesyncDedupRotationState>,
-    pub(crate) relay_idle_registry: Mutex<RelayIdleCandidateRegistry>,
+    pub(crate) relay_idle_registry: RelayIdleCandidateRegistry,
     pub(crate) relay_idle_mark_seq: AtomicU64,
 }
 
@@ -97,7 +97,7 @@ impl ProxySharedState {
                 desync_hasher: RandomState::new(),
                 desync_full_cache_last_emit_at: Mutex::new(None),
                 desync_dedup_rotation_state: Mutex::new(DesyncDedupRotationState::default()),
-                relay_idle_registry: Mutex::new(RelayIdleCandidateRegistry::default()),
+                relay_idle_registry: RelayIdleCandidateRegistry::default(),
                 relay_idle_mark_seq: AtomicU64::new(0),
             },
             traffic_limiter: TrafficLimiter::new(),

src/proxy/tests/middle_relay_idle_registry_poison_security_tests.rs

@@ -1,33 +1,21 @@
 use super::*;
-use std::panic::{AssertUnwindSafe, catch_unwind};
 
 #[test]
-fn blackhat_registry_poison_recovers_with_fail_closed_reset_and_pressure_accounting() {
+fn blackhat_registry_stale_order_entry_is_skipped_and_pressure_accounting_continues() {
     let shared = ProxySharedState::new();
     clear_relay_idle_pressure_state_for_testing_in_shared(shared.as_ref());
 
-    let _ = catch_unwind(AssertUnwindSafe(|| {
+    shared
-        let mut guard = shared
+        .middle_relay
-            .middle_relay
+        .relay_idle_registry
-            .relay_idle_registry
+        .ordered
-            .lock()
+        .lock()
-            .expect("registry lock must be acquired before poison");
+        .insert((0, 999));
-        guard.by_conn_id.insert(
-            999,
-            RelayIdleCandidateMeta {
-                mark_order_seq: 1,
-                mark_pressure_seq: 0,
-            },
-        );
-        guard.ordered.insert((1, 999));
-        panic!("intentional poison for idle-registry recovery");
-    }));
 
-    // Helper lock must recover from poison, reset stale state, and continue.
     assert!(mark_relay_idle_candidate_for_testing(shared.as_ref(), 42));
     assert_eq!(
         oldest_relay_idle_candidate_for_testing(shared.as_ref()),
-        Some(42)
+        Some(999)
     );
 
     let before = relay_pressure_event_seq_for_testing(shared.as_ref());
@@ -35,25 +23,43 @@ fn blackhat_registry_poison_recovers_with_fail_closed_reset_and_pressure_account
     let after = relay_pressure_event_seq_for_testing(shared.as_ref());
     assert!(
         after > before,
-        "pressure accounting must still advance after poison"
+        "pressure accounting must still advance with stale ordered entries"
+    );
+
+    let mut seen_pressure_seq = before;
+    assert!(maybe_evict_idle_candidate_on_pressure_for_testing(
+        shared.as_ref(),
+        42,
+        &mut seen_pressure_seq,
+        &Stats::new()
+    ));
+    assert_eq!(
+        oldest_relay_idle_candidate_for_testing(shared.as_ref()),
+        None
     );
 
     clear_relay_idle_pressure_state_for_testing_in_shared(shared.as_ref());
 }
 
 #[test]
-fn clear_state_helper_must_reset_poisoned_registry_for_deterministic_fifo_tests() {
+fn clear_state_helper_must_reset_split_registry_for_deterministic_fifo_tests() {
     let shared = ProxySharedState::new();
     clear_relay_idle_pressure_state_for_testing_in_shared(shared.as_ref());
 
-    let _ = catch_unwind(AssertUnwindSafe(|| {
+    shared.middle_relay.relay_idle_registry.by_conn_id.insert(
-        let _guard = shared
+        999,
-            .middle_relay
+        RelayIdleCandidateMeta {
-            .relay_idle_registry
+            mark_order_seq: 1,
-            .lock()
+            mark_pressure_seq: 0,
-            .expect("registry lock must be acquired before poison");
+        },
-        panic!("intentional poison while lock held");
+    );
-    }));
+    shared
+        .middle_relay
+        .relay_idle_registry
+        .ordered
+        .lock()
+        .insert((1, 999));
+    set_relay_pressure_state_for_testing(shared.as_ref(), 7, 6);
 
     clear_relay_idle_pressure_state_for_testing_in_shared(shared.as_ref());
 

src/transport/middle_proxy/pool_writer.rs

@@ -52,6 +52,8 @@ async fn writer_command_loop(
 ) -> Result<()> {
     loop {
         tokio::select! {
+            biased;
+            _ = cancel.cancelled() => return Ok(()),
             cmd = rx.recv() => {
                 match cmd {
                     Some(WriterCommand::Data(payload)) => {
@@ -69,7 +71,6 @@ async fn writer_command_loop(
                     Some(WriterCommand::Close) | None => return Ok(()),
                 }
             }
-            _ = cancel.cancelled() => return Ok(()),
         }
     }
 }
@@ -108,6 +109,7 @@ async fn ping_loop(
         Duration::from_secs(wait)
     };
     tokio::select! {
+        biased;
         _ = cancel_ping_token.cancelled() => return,
         _ = tokio::time::sleep(startup_jitter) => {}
     }
@@ -131,6 +133,7 @@ async fn ping_loop(
             Duration::from_secs(secs)
         };
         tokio::select! {
+            biased;
             _ = cancel_ping_token.cancelled() => return,
             _ = tokio::time::sleep(wait) => {}
         }
@@ -191,6 +194,7 @@ async fn rpc_proxy_req_signal_loop(
     };
 
     tokio::select! {
+        biased;
         _ = cancel_signal.cancelled() => return,
         _ = tokio::time::sleep(Duration::from_millis(startup_jitter_ms)) => {}
     }
@@ -207,6 +211,7 @@ async fn rpc_proxy_req_signal_loop(
         };
 
         tokio::select! {
+            biased;
             _ = cancel_signal.cancelled() => return,
             _ = tokio::time::sleep(wait) => {}
         }

src/transport/middle_proxy/reader.rs

@@ -242,6 +242,7 @@ pub(crate) async fn reader_loop(
     let mut raw = enc_leftover;
     let mut expected_seq: i32 = 0;
     let mut data_route_queue_full_streak = HashMap::<u64, u8>::new();
+    let mut tmp = [0u8; 65_536];
     let mut fairness = WorkerFairnessState::new(
         WorkerFairnessConfig {
             worker_id: (writer_id as u16).saturating_add(1),
@@ -263,18 +264,18 @@ pub(crate) async fn reader_loop(
         let fairshare_enabled = route_fairshare_enabled.load(Ordering::Relaxed);
         fairness.set_backpressure_enabled(backpressure_enabled);
         let fairness_has_backlog = should_schedule_fairness_retry(&fairness_snapshot);
-        let mut tmp = [0u8; 65_536];
         let backlog_retry_enabled = fairness_has_backlog;
         let backlog_retry_delay =
             fairness_retry_delay(reader_route_data_wait_ms.load(Ordering::Relaxed));
         let mut retry_only = false;
         let n = tokio::select! {
+            biased;
+            _ = cancel.cancelled() => return Ok(()),
             res = rd.read(&mut tmp) => res.map_err(ProxyError::Io)?,
             _ = tokio::time::sleep(backlog_retry_delay), if backlog_retry_enabled => {
                 retry_only = true;
                 0usize
             },
-            _ = cancel.cancelled() => return Ok(()),
         };
         if retry_only {
             let route_wait_ms = reader_route_data_wait_ms.load(Ordering::Relaxed);

src/transport/middle_proxy/registry.rs

@@ -77,26 +77,24 @@ struct HotBindingTable {
 
 struct BindingState {
     inner: Mutex<BindingInner>,
+    writer_idle_since_epoch_secs: DashMap<u64, u64>,
+    bound_clients_by_writer: DashMap<u64, usize>,
+    active_sessions_by_target_dc: DashMap<i16, usize>,
+    last_meta_for_writer: DashMap<u64, ConnMeta>,
 }
 
 struct BindingInner {
-    writers: HashMap<u64, mpsc::Sender<WriterCommand>>,
     writer_for_conn: HashMap<u64, u64>,
     conns_for_writer: HashMap<u64, HashSet<u64>>,
     meta: HashMap<u64, ConnMeta>,
-    last_meta_for_writer: HashMap<u64, ConnMeta>,
-    writer_idle_since_epoch_secs: HashMap<u64, u64>,
 }
 
 impl BindingInner {
     fn new() -> Self {
         Self {
-            writers: HashMap::new(),
             writer_for_conn: HashMap::new(),
             conns_for_writer: HashMap::new(),
             meta: HashMap::new(),
-            last_meta_for_writer: HashMap::new(),
-            writer_idle_since_epoch_secs: HashMap::new(),
         }
     }
 }
@@ -149,6 +147,10 @@ impl ConnRegistry {
             },
             binding: BindingState {
                 inner: Mutex::new(BindingInner::new()),
+                writer_idle_since_epoch_secs: DashMap::new(),
+                bound_clients_by_writer: DashMap::new(),
+                active_sessions_by_target_dc: DashMap::new(),
+                last_meta_for_writer: DashMap::new(),
             },
             next_id: AtomicU64::new(start),
             route_channel_capacity,

src/transport/middle_proxy/registry/writer.rs

@@ -13,13 +13,55 @@ use super::{
 };
 
 impl ConnRegistry {
+    fn set_writer_bound_count(&self, writer_id: u64, count: usize) {
+        self.binding.bound_clients_by_writer.insert(writer_id, count);
+        if count == 0 {
+            self.binding
+                .writer_idle_since_epoch_secs
+                .entry(writer_id)
+                .or_insert_with(Self::now_epoch_secs);
+        } else {
+            self.binding.writer_idle_since_epoch_secs.remove(&writer_id);
+        }
+    }
+
+    fn adjust_active_target_dc(&self, target_dc: i16, delta: isize) {
+        if target_dc == 0 || delta == 0 {
+            return;
+        }
+        if delta > 0 {
+            self.binding
+                .active_sessions_by_target_dc
+                .entry(target_dc)
+                .and_modify(|count| *count = count.saturating_add(delta as usize))
+                .or_insert(delta as usize);
+            return;
+        }
+
+        let remove =
+            if let Some(mut count) = self.binding.active_sessions_by_target_dc.get_mut(&target_dc) {
+            let decrement = delta.unsigned_abs();
+            *count = count.saturating_sub(decrement);
+            *count == 0
+        } else {
+            false
+        };
+        if remove {
+            self.binding.active_sessions_by_target_dc.remove(&target_dc);
+        }
+    }
+
     pub async fn register_writer(&self, writer_id: u64, tx: mpsc::Sender<WriterCommand>) {
         let mut binding = self.binding.inner.lock().await;
-        binding.writers.insert(writer_id, tx.clone());
         binding
             .conns_for_writer
             .entry(writer_id)
             .or_insert_with(HashSet::new);
+        self.binding.bound_clients_by_writer.entry(writer_id).or_insert(0);
+        self.binding
+            .writer_idle_since_epoch_secs
+            .entry(writer_id)
+            .or_insert_with(Self::now_epoch_secs);
         self.writers.map.insert(writer_id, tx);
     }
 
@@ -29,19 +71,18 @@ impl ConnRegistry {
         self.routing.byte_budget.remove(&id);
         self.hot_binding.map.remove(&id);
         let mut binding = self.binding.inner.lock().await;
-        binding.meta.remove(&id);
+        let previous_meta = binding.meta.remove(&id);
+        if let Some(meta) = previous_meta.as_ref() {
+            self.adjust_active_target_dc(meta.target_dc, -1);
+        }
         if let Some(writer_id) = binding.writer_for_conn.remove(&id) {
-            let became_empty = if let Some(set) = binding.conns_for_writer.get_mut(&writer_id) {
+            let next_count = if let Some(set) = binding.conns_for_writer.get_mut(&writer_id) {
                 set.remove(&id);
-                set.is_empty()
+                set.len()
             } else {
-                false
+                0
             };
-            if became_empty {
+            self.set_writer_bound_count(writer_id, next_count);
-                binding
-                    .writer_idle_since_epoch_secs
-                    .insert(writer_id, Self::now_epoch_secs());
-            }
             return Some(writer_id);
         }
         None
@@ -248,7 +289,7 @@ impl ConnRegistry {
         if !self.routing.map.contains_key(&conn_id) {
             return false;
         }
-        if !binding.writers.contains_key(&writer_id) {
+        if !self.writers.map.contains_key(&writer_id) {
             return false;
         }
 
@@ -256,28 +297,32 @@ impl ConnRegistry {
         if let Some(previous_writer_id) = previous_writer_id
             && previous_writer_id != writer_id
         {
-            let became_empty =
+            let next_count =
                 if let Some(set) = binding.conns_for_writer.get_mut(&previous_writer_id) {
                     set.remove(&conn_id);
-                    set.is_empty()
+                    set.len()
                 } else {
-                    false
+                    0
                 };
-            if became_empty {
+            self.set_writer_bound_count(previous_writer_id, next_count);
-                binding
-                    .writer_idle_since_epoch_secs
-                    .insert(previous_writer_id, Self::now_epoch_secs());
-            }
         }
 
-        binding.meta.insert(conn_id, meta.clone());
+        if let Some(previous_meta) = binding.meta.insert(conn_id, meta.clone()) {
-        binding.last_meta_for_writer.insert(writer_id, meta.clone());
+            self.adjust_active_target_dc(previous_meta.target_dc, -1);
-        binding.writer_idle_since_epoch_secs.remove(&writer_id);
+        }
-        binding
+        self.adjust_active_target_dc(meta.target_dc, 1);
-            .conns_for_writer
+        self.binding
-            .entry(writer_id)
+            .last_meta_for_writer
-            .or_insert_with(HashSet::new)
+            .insert(writer_id, meta.clone());
-            .insert(conn_id);
+        let next_count = {
+            let set = binding
+                .conns_for_writer
+                .entry(writer_id)
+                .or_insert_with(HashSet::new);
+            set.insert(conn_id);
+            set.len()
+        };
+        self.set_writer_bound_count(writer_id, next_count);
         self.hot_binding
             .map
             .insert(conn_id, HotConnBinding { writer_id, meta });
@@ -290,27 +335,38 @@ impl ConnRegistry {
             .conns_for_writer
             .entry(writer_id)
             .or_insert_with(HashSet::new);
-        binding
+        let count = binding
-            .writer_idle_since_epoch_secs
+            .conns_for_writer
-            .entry(writer_id)
+            .get(&writer_id)
-            .or_insert(Self::now_epoch_secs());
+            .map(|set| set.len())
+            .unwrap_or(0);
+        self.set_writer_bound_count(writer_id, count);
     }
 
     pub async fn get_last_writer_meta(&self, writer_id: u64) -> Option<ConnMeta> {
-        let binding = self.binding.inner.lock().await;
+        self.binding
-        binding.last_meta_for_writer.get(&writer_id).cloned()
+            .last_meta_for_writer
+            .get(&writer_id)
+            .map(|entry| entry.value().clone())
     }
 
     pub async fn writer_idle_since_snapshot(&self) -> HashMap<u64, u64> {
-        let binding = self.binding.inner.lock().await;
+        self.binding
-        binding.writer_idle_since_epoch_secs.clone()
+            .writer_idle_since_epoch_secs
+            .iter()
+            .map(|entry| (*entry.key(), *entry.value()))
+            .collect()
     }
 
     pub async fn writer_idle_since_for_writer_ids(&self, writer_ids: &[u64]) -> HashMap<u64, u64> {
-        let binding = self.binding.inner.lock().await;
         let mut out = HashMap::<u64, u64>::with_capacity(writer_ids.len());
         for writer_id in writer_ids {
-            if let Some(idle_since) = binding.writer_idle_since_epoch_secs.get(writer_id).copied() {
+            if let Some(idle_since) = self
+                .binding
+                .writer_idle_since_epoch_secs
+                .get(writer_id)
+                .map(|entry| *entry.value())
+            {
                 out.insert(*writer_id, idle_since);
             }
         }
@@ -320,25 +376,19 @@ impl ConnRegistry {
     pub(in crate::transport::middle_proxy) async fn writer_activity_snapshot(
         &self,
     ) -> WriterActivitySnapshot {
-        let binding = self.binding.inner.lock().await;
-        let mut bound_clients_by_writer = HashMap::<u64, usize>::new();
-        let mut active_sessions_by_target_dc = HashMap::<i16, usize>::new();
-
-        for (writer_id, conn_ids) in &binding.conns_for_writer {
-            bound_clients_by_writer.insert(*writer_id, conn_ids.len());
-        }
-        for conn_meta in binding.meta.values() {
-            if conn_meta.target_dc == 0 {
-                continue;
-            }
-            *active_sessions_by_target_dc
-                .entry(conn_meta.target_dc)
-                .or_insert(0) += 1;
-        }
-
         WriterActivitySnapshot {
-            bound_clients_by_writer,
+            bound_clients_by_writer: self
-            active_sessions_by_target_dc,
+                .binding
+                .bound_clients_by_writer
+                .iter()
+                .map(|entry| (*entry.key(), *entry.value()))
+                .collect(),
+            active_sessions_by_target_dc: self
+                .binding
+                .active_sessions_by_target_dc
+                .iter()
+                .map(|entry| (*entry.key(), *entry.value()))
+                .collect(),
         }
     }
 
@@ -393,10 +443,10 @@ impl ConnRegistry {
 
     pub async fn writer_lost(&self, writer_id: u64) -> Vec<BoundConn> {
         let mut binding = self.binding.inner.lock().await;
-        binding.writers.remove(&writer_id);
         self.writers.map.remove(&writer_id);
-        binding.last_meta_for_writer.remove(&writer_id);
+        self.binding.last_meta_for_writer.remove(&writer_id);
-        binding.writer_idle_since_epoch_secs.remove(&writer_id);
+        self.binding.writer_idle_since_epoch_secs.remove(&writer_id);
+        self.binding.bound_clients_by_writer.remove(&writer_id);
         let conns = binding
             .conns_for_writer
             .remove(&writer_id)
@@ -410,6 +460,10 @@ impl ConnRegistry {
                 continue;
             }
             binding.writer_for_conn.remove(&conn_id);
+            let meta = binding.meta.remove(&conn_id);
+            if let Some(meta) = meta.as_ref() {
+                self.adjust_active_target_dc(meta.target_dc, -1);
+            }
             let remove_hot = self
                 .hot_binding
                 .map
@@ -419,10 +473,10 @@ impl ConnRegistry {
             if remove_hot {
                 self.hot_binding.map.remove(&conn_id);
             }
-            if let Some(m) = binding.meta.get(&conn_id) {
+            if let Some(m) = meta {
                 out.push(BoundConn {
                     conn_id,
-                    meta: m.clone(),
+                    meta: m,
                 });
             }
         }
@@ -438,11 +492,10 @@ impl ConnRegistry {
     }
 
     pub async fn is_writer_empty(&self, writer_id: u64) -> bool {
-        let binding = self.binding.inner.lock().await;
+        self.binding
-        binding
+            .bound_clients_by_writer
-            .conns_for_writer
             .get(&writer_id)
-            .map(|s| s.is_empty())
+            .map(|count| *count.value() == 0)
             .unwrap_or(true)
     }
 
@@ -457,21 +510,20 @@ impl ConnRegistry {
             return false;
         }
 
-        binding.writers.remove(&writer_id);
         self.writers.map.remove(&writer_id);
-        binding.last_meta_for_writer.remove(&writer_id);
+        self.binding.last_meta_for_writer.remove(&writer_id);
-        binding.writer_idle_since_epoch_secs.remove(&writer_id);
+        self.binding.writer_idle_since_epoch_secs.remove(&writer_id);
+        self.binding.bound_clients_by_writer.remove(&writer_id);
         binding.conns_for_writer.remove(&writer_id);
         true
     }
 
     #[allow(dead_code)]
     pub(super) async fn non_empty_writer_ids(&self, writer_ids: &[u64]) -> HashSet<u64> {
-        let binding = self.binding.inner.lock().await;
         let mut out = HashSet::<u64>::with_capacity(writer_ids.len());
         for writer_id in writer_ids {
-            if let Some(conns) = binding.conns_for_writer.get(writer_id)
+            if let Some(count) = self.binding.bound_clients_by_writer.get(writer_id)
-                && !conns.is_empty()
+                && *count.value() > 0
             {
                 out.insert(*writer_id);
             }

src/transport/middle_proxy/send.rs

@@ -15,7 +15,6 @@ use super::registry::ConnMeta;
 use super::wire::build_proxy_req_payload;
 use crate::config::{MeRouteNoWriterMode, MeWriterPickMode};
 use crate::error::{ProxyError, Result};
-use crate::network::IpFamily;
 use crate::stream::PooledBuffer;
 use rand::seq::SliceRandom;
 
@@ -124,9 +123,8 @@ impl MePool {
             }
 
             let mut writers_snapshot = {
-                let ws = self.writers.read().await;
+                let ws = self.writers.snapshot();
                 if ws.is_empty() {
-                    drop(ws);
                     match no_writer_mode {
                         MeRouteNoWriterMode::AsyncRecoveryFailfast => {
                             let deadline = *no_writer_deadline.get_or_insert_with(|| {
@@ -154,38 +152,32 @@ impl MePool {
                                 for _ in
                                     0..self.route_runtime.me_route_inline_recovery_attempts.max(1)
                                 {
-                                    for family in self.family_order() {
+                                    let preferred = self.preferred_endpoints_by_dc.load_full();
-                                        let map = match family {
+                                    for (dc, addrs) in preferred.iter() {
-                                            IpFamily::V4 => self.proxy_map_v4.read().await.clone(),
+                                        for addr in addrs {
-                                            IpFamily::V6 => self.proxy_map_v6.read().await.clone(),
+                                            let _ = self
-                                        };
+                                                .connect_one_for_dc(
-                                        for (dc, addrs) in &map {
+                                                    *addr,
-                                            for (ip, port) in addrs {
+                                                    *dc,
-                                                let addr = SocketAddr::new(*ip, *port);
+                                                    self.rng.as_ref(),
-                                                let _ = self
+                                                )
-                                                    .connect_one_for_dc(
+                                                .await;
-                                                        addr,
-                                                        *dc,
-                                                        self.rng.as_ref(),
-                                                    )
-                                                    .await;
-                                            }
                                         }
                                     }
-                                    if !self.writers.read().await.is_empty() {
+                                    if !self.writers.snapshot().is_empty() {
                                         break;
                                     }
                                 }
                             }
 
-                            if !self.writers.read().await.is_empty() {
+                            if !self.writers.snapshot().is_empty() {
                                 continue;
                             }
                             let deadline = *no_writer_deadline.get_or_insert_with(|| {
                                 Instant::now() + self.route_runtime.me_route_inline_recovery_wait
                             });
                             if !self.wait_for_writer_until(deadline).await {
-                                if !self.writers.read().await.is_empty() {
+                                if !self.writers.snapshot().is_empty() {
                                     continue;
                                 }
                                 self.stats.increment_me_no_writer_failfast_total();
@@ -222,7 +214,7 @@ impl MePool {
                         }
                     }
                 }
-                ws.clone()
+                ws
             };
 
             let mut candidate_indices = self
@@ -285,7 +277,12 @@ impl MePool {
                             ));
                         }
                         emergency_attempts += 1;
-                        let mut endpoints = self.endpoint_candidates_for_target_dc(routed_dc).await;
+                        let mut endpoints = self
+                            .preferred_endpoints_by_dc
+                            .load()
+                            .get(&routed_dc)
+                            .cloned()
+                            .unwrap_or_default();
                         endpoints.shuffle(&mut rand::rng());
                         for addr in endpoints {
                             if self
@@ -298,9 +295,7 @@ impl MePool {
                         }
                         tokio::time::sleep(Duration::from_millis(100 * emergency_attempts as u64))
                             .await;
-                        let ws2 = self.writers.read().await;
+                        writers_snapshot = self.writers.snapshot();
-                        writers_snapshot = ws2.clone();
-                        drop(ws2);
                         candidate_indices = self
                             .candidate_indices_for_dc(&writers_snapshot, routed_dc, false)
                             .await;

src/transport/middle_proxy/send/recovery.rs

@@ -1,13 +1,9 @@
-use std::collections::HashSet;
-use std::net::SocketAddr;
 use std::sync::Arc;
 use std::sync::atomic::Ordering;
 use std::time::{Duration, Instant};
 
 use tracing::warn;
 
-use crate::network::IpFamily;
-
 use super::super::MePool;
 use super::{
     HYBRID_GLOBAL_BURST_PERIOD_ROUNDS, HYBRID_RECENT_SUCCESS_WINDOW_MS,
@@ -17,18 +13,18 @@ use super::{
 impl MePool {
     pub(super) async fn wait_for_writer_until(&self, deadline: Instant) -> bool {
         let mut rx = self.writer_epoch.subscribe();
-        if !self.writers.read().await.is_empty() {
+        if !self.writers.snapshot().is_empty() {
             return true;
         }
         let now = Instant::now();
         if now >= deadline {
-            return !self.writers.read().await.is_empty();
+            return !self.writers.snapshot().is_empty();
         }
         let timeout = deadline.saturating_duration_since(now);
         if tokio::time::timeout(timeout, rx.changed()).await.is_ok() {
-            return !self.writers.read().await.is_empty();
+            return !self.writers.snapshot().is_empty();
         }
-        !self.writers.read().await.is_empty()
+        !self.writers.snapshot().is_empty()
     }
 
     pub(super) async fn wait_for_candidate_until(&self, routed_dc: i32, deadline: Instant) -> bool {
@@ -58,11 +54,11 @@ impl MePool {
 
     pub(super) async fn has_candidate_for_target_dc(&self, routed_dc: i32) -> bool {
         let writers_snapshot = {
-            let ws = self.writers.read().await;
+            let ws = self.writers.snapshot();
             if ws.is_empty() {
                 return false;
             }
-            ws.clone()
+            ws
         };
         let mut candidate_indices = self
             .candidate_indices_for_dc(&writers_snapshot, routed_dc, false)
@@ -79,7 +75,7 @@ impl MePool {
         self: &Arc<Self>,
         routed_dc: i32,
     ) -> bool {
-        let endpoints = self.endpoint_candidates_for_target_dc(routed_dc).await;
+        let endpoints = self.preferred_endpoints_for_dc(routed_dc).await;
         if endpoints.is_empty() {
             return false;
         }
@@ -92,33 +88,19 @@ impl MePool {
 
     pub(super) async fn trigger_async_recovery_global(self: &Arc<Self>) {
         self.stats.increment_me_async_recovery_trigger_total();
-        let mut seen = HashSet::<(i32, SocketAddr)>::new();
+        let preferred = self.preferred_endpoints_by_dc.load();
-        for family in self.family_order() {
+        let mut triggered = 0usize;
-            let map_guard = match family {
+        for (dc, addrs) in preferred.iter() {
-                IpFamily::V4 => self.proxy_map_v4.read().await,
+            for addr in addrs {
-                IpFamily::V6 => self.proxy_map_v6.read().await,
+                self.trigger_immediate_refill_for_dc(*addr, *dc);
-            };
+                triggered = triggered.saturating_add(1);
-            for (dc, addrs) in map_guard.iter() {
+                if triggered >= 8 {
-                for (ip, port) in addrs {
+                    return;
-                    let addr = SocketAddr::new(*ip, *port);
-                    if seen.insert((*dc, addr)) {
-                        self.trigger_immediate_refill_for_dc(addr, *dc);
-                    }
-                    if seen.len() >= 8 {
-                        return;
-                    }
                 }
             }
         }
     }
 
-    pub(super) async fn endpoint_candidates_for_target_dc(
-        &self,
-        routed_dc: i32,
-    ) -> Vec<SocketAddr> {
-        self.preferred_endpoints_for_dc(routed_dc).await
-    }
-
     pub(super) async fn maybe_trigger_hybrid_recovery(
         self: &Arc<Self>,
         routed_dc: i32,

src/transport/middle_proxy/send/selection.rs

@@ -15,7 +15,10 @@ impl MePool {
         routed_dc: i32,
         include_warm: bool,
     ) -> Vec<usize> {
-        let preferred = self.preferred_endpoints_for_dc(routed_dc).await;
+        let preferred_snapshot = self.preferred_endpoints_by_dc.load();
+        let Some(preferred) = preferred_snapshot.get(&routed_dc) else {
+            return Vec::new();
+        };
         if preferred.is_empty() {
             return Vec::new();
         }
@@ -25,7 +28,7 @@ impl MePool {
             if !self.writer_eligible_for_selection(w, include_warm) {
                 continue;
             }
-            if w.writer_dc == routed_dc && preferred.contains(&w.addr) {
+            if w.writer_dc == routed_dc && preferred.binary_search(&w.addr).is_ok() {
                 out.push(idx);
             }
         }