Merge pull request #350 from telemt/bump

Update Cargo.toml

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "telemt"
-version = "3.3.5"
+version = "3.3.6"
 edition = "2024"
 
 [dependencies]

README.md

@@ -19,18 +19,24 @@
 
 ### 🇷🇺 RU
 
-#### Релиз 3.3.3 LTS - 6 марта
+#### Релиз 3.3.5 LTS - 6 марта
 
-6 марта мы выпустили Telemt **3.3.3**
+6 марта мы выпустили Telemt **3.3.5**
 
-Это первая версия telemt работающая в комплексных условиях и при этом предоставляющая API
+Это [3.3.5 - первая LTS-версия telemt](https://github.com/telemt/telemt/releases/tag/3.3.5)!
 
-В ней используется новый алгоритм - ME NoWait, который вместе с Adaptive Floor и моделью усовершенствованного доступа к KDF Fingerprint на RwLock позволяет достигать максимальную производительность, даже в условиях lossy-сети
+В ней используется:
+- новый алгоритм ME NoWait для непревзойдённо быстрого восстановления пула
+- Adaptive Floor, поддерживающий количество ME Writer на оптимальном уровне
+- модель усовершенствованного доступа к KDF Fingerprint на RwLock
+- строгая привязка Middle-End к DC-ID с предсказуемым алгоритмом деградации и самовосстановления
 
-Будем рады вашему фидбеку и предложениям по улучшению — особенно в части **статистики** и **UX**
+Telemt Control API V1 в 3.3.5 включает:
+- несколько режимов работы в зависимости от доступных ресурсов
+- снапшот-модель для живых метрик без вмешательства в hot-path
+- минималистичный набор запросов для управления пользователями
 
-Релиз:  
-[3.3.3](https://github.com/telemt/telemt/releases/tag/3.3.3)
+Будем рады вашему фидбеку и предложениям по улучшению — особенно в части **API**, **статистики**, **UX**
 
 ---
 
@@ -47,18 +53,24 @@
 
 ### 🇬🇧 EN
 
-#### Release 3.3.3 LTS - March 6
+#### Release 3.3.5 LTS - March 6
 
 On March 6, we released Telemt **3.3.3**
 
-This is the first telemt's version designed to operate reliably in complex network conditions while also providing a runtime API!
+This is [3.3.5 - the first LTS release of telemt](https://github.com/telemt/telemt/releases/tag/3.3.5)
 
-The release introduces a new algorithm — ME NoWait, which combined with Adaptive Floor and an improved KDF Fingerprint access model based on RwLock, it enables the system to achieve maximum performance even in lossy network environments
+It introduces:
+- the new ME NoWait algorithm for exceptionally fast pool recovery
+- Adaptive Floor, which maintains the number of ME Writers at an optimal level
+- an improved KDF Fingerprint access model based on RwLock
+- strict binding of Middle-End instances to DC-ID with a predictable degradation and self-recovery algorithm
 
-We are looking forward to your feedback and improvement proposals — especially regarding **statistics** and **UX**
+Telemt Control API V1 in version 3.3.5 includes:
+- multiple operating modes depending on available resources
+- a snapshot-based model for live metrics without interfering with the hot path
+- a minimalistic request set for user management
 
-Release:  
-[3.3.3](https://github.com/telemt/telemt/releases/tag/3.3.3)
+We are looking forward to your feedback and improvement proposals — especially regarding **API**, **statistics**, **UX**
 
 ---
 
@@ -81,31 +93,6 @@ We welcome ideas, architectural feedback, and pull requests.
 
 ⚓ Our ***Middle-End Pool*** is fastest by design in standard scenarios, compared to other implementations of connecting to the Middle-End Proxy: non dramatically, but usual
 
-# GOTO
-- [Features](#features)
-- [Quick Start Guide](#quick-start-guide)
-- [How to use?](#how-to-use)
-  - [Systemd Method](#telemt-via-systemd)
-- [Configuration](#configuration)
-  - [Minimal Configuration](#minimal-configuration-for-first-start)
-  - [Advanced](#advanced)
-    - [Adtag](#adtag)
-    - [Listening and Announce IPs](#listening-and-announce-ips)
-    - [Upstream Manager](#upstream-manager)
-      - [IP](#bind-on-ip)
-      - [SOCKS](#socks45-as-upstream)
-- [FAQ](#faq)
-  - [Recognizability for DPI + crawler](#recognizability-for-dpi-and-crawler)
-  - [Telegram Calls](#telegram-calls-via-mtproxy)
-  - [DPI](#how-does-dpi-see-mtproxy-tls)
-  - [Whitelist on Network Level](#whitelist-on-ip)
-  - [Too many open files](#too-many-open-files)
-- [Build](#build)
-- [Docker](#docker)
-- [Why Rust?](#why-rust)
-
-## Features
-
 - Full support for all official MTProto proxy modes:
   - Classic
   - Secure - with `dd` prefix
@@ -116,59 +103,40 @@ We welcome ideas, architectural feedback, and pull requests.
 - Graceful shutdown on Ctrl+C
 - Extensive logging via `trace` and `debug` with `RUST_LOG` method
 
+# GOTO
+- [Telemt - MTProxy on Rust + Tokio](#telemt---mtproxy-on-rust--tokio)
+  - [NEWS and EMERGENCY](#news-and-emergency)
+    - [✈️ Telemt 3 is released!](#️-telemt-3-is-released)
+    - [🇷🇺 RU](#-ru)
+      - [Релиз 3.3.5 LTS - 6 марта](#релиз-335-lts---6-марта)
+    - [🇬🇧 EN](#-en)
+      - [Release 3.3.5 LTS - March 6](#release-335-lts---march-6)
+- [Features](#features)
+- [GOTO](#goto)
+  - [Quick Start Guide](#quick-start-guide)
+  - [FAQ](#faq)
+    - [Recognizability for DPI and crawler](#recognizability-for-dpi-and-crawler)
+      - [Client WITH secret-key accesses the MTProxy resource:](#client-with-secret-key-accesses-the-mtproxy-resource)
+      - [Client WITHOUT secret-key gets transparent access to the specified resource:](#client-without-secret-key-gets-transparent-access-to-the-specified-resource)
+    - [Telegram Calls via MTProxy](#telegram-calls-via-mtproxy)
+    - [How does DPI see MTProxy TLS?](#how-does-dpi-see-mtproxy-tls)
+    - [Whitelist on IP](#whitelist-on-ip)
+    - [Too many open files](#too-many-open-files)
+  - [Build](#build)
+  - [Why Rust?](#why-rust)
+  - [Issues](#issues)
+  - [Roadmap](#roadmap)
+
+
 ## Quick Start Guide
-
-### [Quick Start Guide RU](docs/QUICK_START_GUIDE.ru.md)  
-### [Quick Start Guide EN](docs/QUICK_START_GUIDE.en.md)
-
-
-### Advanced
-#### Adtag (per-user)
-To use channel advertising and usage statistics from Telegram, get an Adtag from [@mtproxybot](https://t.me/mtproxybot). Set it per user in `[access.user_ad_tags]` (32 hex chars):
-```toml
-[access.user_ad_tags]
-username1 = "11111111111111111111111111111111"  # Replace with your tag from @mtproxybot
-username2 = "22222222222222222222222222222222"
-```
-#### Listening and Announce IPs
-To specify listening address and/or address in links, add to section `[[server.listeners]]` of config.toml:
-```toml
-[[server.listeners]]
-ip = "0.0.0.0"          # 0.0.0.0 = all IPs; your IP = specific listening
-announce_ip = "1.2.3.4" # IP in links; comment with # if not used
-```
-#### Upstream Manager
-To specify upstream, add to section `[[upstreams]]` of config.toml:
-##### Bind on IP
-```toml
-[[upstreams]]
-type = "direct"
-weight = 1
-enabled = true
-interface = "192.168.1.100" # Change to your outgoing IP
-```
-##### SOCKS4/5 as Upstream
-- Without Auth:
-```toml
-[[upstreams]]
-type = "socks5"            # Specify SOCKS4 or SOCKS5
-address = "1.2.3.4:1234"   # SOCKS-server Address
-weight = 1                 # Set Weight for Scenarios
-enabled = true
-```
-
-- With Auth:
-```toml
-[[upstreams]]
-type = "socks5"            # Specify SOCKS4 or SOCKS5
-address = "1.2.3.4:1234"   # SOCKS-server Address
-username = "user"          # Username for Auth on SOCKS-server
-password = "pass"          # Password for Auth on SOCKS-server
-weight = 1                 # Set Weight for Scenarios
-enabled = true
-```
+- [Quick Start Guide RU](docs/QUICK_START_GUIDE.ru.md)
+- [Quick Start Guide EN](docs/QUICK_START_GUIDE.en.md)
 
 ## FAQ
+
+- [FAQ RU](docs/FAQ.ru.md)
+- [FAQ EN](docs/FAQ.en.md)
+
 ### Recognizability for DPI and crawler
 Since version 1.1.0.0, we have debugged masking perfectly: for all clients without "presenting" a key, 
 we transparently direct traffic to the target host!
@@ -313,41 +281,6 @@ chmod +x /bin/telemt
 telemt config.toml
 ```
 
-## Docker
-**Quick start (Docker Compose)**
-
-1. Edit `config.toml` in repo root (at least: port, users secrets, tls_domain)
-2. Start container:
-```bash
-docker compose up -d --build
-```
-3. Check logs:
-```bash
-docker compose logs -f telemt
-```
-4. Stop:
-```bash
-docker compose down
-```
-
-**Notes**
-- `docker-compose.yml` maps `./config.toml` to `/app/config.toml` (read-only)
-- By default it publishes `443:443` and runs with dropped capabilities (only `NET_BIND_SERVICE` is added)
-- If you really need host networking (usually only for some IPv6 setups) uncomment `network_mode: host`
-
-**Run without Compose**
-```bash
-docker build -t telemt:local .
-docker run --name telemt --restart unless-stopped \
-  -p 443:443 \
-  -e RUST_LOG=info \
-  -v "$PWD/config.toml:/app/config.toml:ro" \
-  --read-only \
-  --cap-drop ALL --cap-add NET_BIND_SERVICE \
-  --ulimit nofile=65536:65536 \
-  telemt:local
-```
-
 ## Why Rust?
 - Long-running reliability and idempotent behavior
 - Rust's deterministic resource management - RAII 

docs/FAQ.en.md

@@ -0,0 +1,112 @@
+## How to set up "proxy sponsor" channel and statistics via @MTProxybot bot
+
+1. Go to @MTProxybot bot.
+2. Enter the command `/newproxy`
+3. Send the server IP and port. For example: 1.2.3.4:443
+4. Open the config `nano /etc/telemt.toml`.
+5. Copy and send the user secret from the [access.users] section to the bot.
+6. Copy the tag received from the bot. For example 1234567890abcdef1234567890abcdef.
+> [!WARNING]
+> The link provided by the bot will not work. Do not copy or use it!
+7. Uncomment the ad_tag parameter and enter the tag received from the bot.
+8. Uncomment/add the parameter `use_middle_proxy = true`.
+
+Config example:
+```toml
+[general]
+ad_tag = "1234567890abcdef1234567890abcdef"
+use_middle_proxy = true
+```
+9. Save the config. Ctrl+S -> Ctrl+X.
+10. Restart telemt `systemctl restart telemt`.
+11. In the bot, send the command /myproxies and select the added server.
+12. Click the "Set promotion" button.
+13. Send a **public link** to the channel. Private channels cannot be added!
+14. Wait approximately 1 hour for the information to update on Telegram servers.
+> [!WARNING]
+> You will not see the "proxy sponsor" if you are already subscribed to the channel.
+
+**You can also set up different channels for different users.**
+```toml
+[access.user_ad_tags]
+hello = "ad_tag"
+hello2 = "ad_tag2"
+```
+
+## How many people can use 1 link
+
+By default, 1 link can be used by any number of people.  
+You can limit the number of IPs using the proxy.
+```toml
+[access.user_max_unique_ips]
+hello = 1
+```
+This parameter limits how many unique IPs can use 1 link simultaneously. If one user disconnects, a second user can connect. Also, multiple users can sit behind the same IP.
+
+## How to create multiple different links
+
+1. Generate the required number of secrets `openssl rand -hex 16`
+2. Open the config `nano /etc/telemt.toml`
+3. Add new users.
+```toml
+[access.users]
+user1 = "00000000000000000000000000000001"
+user2 = "00000000000000000000000000000002"
+user3 = "00000000000000000000000000000003"
+```
+4. Save the config. Ctrl+S -> Ctrl+X. You don't need to restart telemt.
+5. Get the links via `journalctl -u telemt -n -g "links" --no-pager -o cat | tac`
+
+## How to view metrics
+
+1. Open the config `nano /etc/telemt.toml`
+2. Add the following parameters
+```toml
+[server]
+metrics_port = 9090
+metrics_whitelist = ["127.0.0.1/32", "::1/128", "0.0.0.0/0"]
+```
+3. Save the config. Ctrl+S -> Ctrl+X.
+4. Metrics are available at SERVER_IP:9090/metrics.
+> [!WARNING]
+> "0.0.0.0/0" in metrics_whitelist opens access from any IP. Replace with your own IP. For example "1.2.3.4"
+
+## Additional parameters
+
+### Domain in link instead of IP
+To specify a domain in the links, add to the `[general.links]` section of the config file.
+```toml
+[general.links]
+public_host = "proxy.example.com"
+```
+
+### Upstream Manager
+To specify an upstream, add to the `[[upstreams]]` section of the config.toml file:
+#### Binding to IP
+```toml
+[[upstreams]]
+type = "direct"
+weight = 1
+enabled = true
+interface = "192.168.1.100" # Change to your outgoing IP
+```
+#### SOCKS4/5 as Upstream
+- Without authentication:
+```toml
+[[upstreams]]
+type = "socks5"            # Specify SOCKS4 or SOCKS5
+address = "1.2.3.4:1234"   # SOCKS-server Address
+weight = 1                 # Set Weight for Scenarios
+enabled = true
+```
+
+- With authentication:
+```toml
+[[upstreams]]
+type = "socks5"            # Specify SOCKS4 or SOCKS5
+address = "1.2.3.4:1234"   # SOCKS-server Address
+username = "user"          # Username for Auth on SOCKS-server
+password = "pass"          # Password for Auth on SOCKS-server
+weight = 1                 # Set Weight for Scenarios
+enabled = true
+```

docs/FAQ.ru.md

@@ -1,4 +1,4 @@
-## Как настроить канал "спонсор прокси"
+## Как настроить канал "спонсор прокси" и статистику через бота @MTProxybot
 
 1. Зайти в бота @MTProxybot.
 2. Ввести команду `/newproxy`
@@ -70,3 +70,43 @@ metrics_whitelist = ["127.0.0.1/32", "::1/128", "0.0.0.0/0"]
 4. Метрики доступны по адресу SERVER_IP:9090/metrics. 
 > [!WARNING]
 > "0.0.0.0/0" в metrics_whitelist открывает доступ с любого IP. Замените на свой ip. Например "1.2.3.4"
+
+## Дополнительные параметры
+
+### Домен в ссылке вместо IP
+Чтобы указать домен в ссылках, добавьте в секцию `[general.links]` файла config.
+```toml
+[general.links]
+public_host = "proxy.example.com"
+```
+
+### Upstream Manager
+Чтобы указать апстрим, добавьте в секцию `[[upstreams]]` файла config.toml:
+#### Привязка к IP
+```toml
+[[upstreams]]
+type = "direct"
+weight = 1
+enabled = true
+interface = "192.168.1.100" # Change to your outgoing IP
+```
+#### SOCKS4/5 как Upstream
+- Без авторизации:
+```toml
+[[upstreams]]
+type = "socks5"            # Specify SOCKS4 or SOCKS5
+address = "1.2.3.4:1234"   # SOCKS-server Address
+weight = 1                 # Set Weight for Scenarios
+enabled = true
+```
+
+- С авторизацией:
+```toml
+[[upstreams]]
+type = "socks5"            # Specify SOCKS4 or SOCKS5
+address = "1.2.3.4:1234"   # SOCKS-server Address
+username = "user"          # Username for Auth on SOCKS-server
+password = "pass"          # Password for Auth on SOCKS-server
+weight = 1                 # Set Weight for Scenarios
+enabled = true
+```

docs/QUICK_START_GUIDE.en.md

@@ -67,6 +67,12 @@ classic = false
 secure = false
 tls = true
 
+[server.api]
+enabled = true
+# listen = "127.0.0.1:9091"
+# whitelist = ["127.0.0.1/32"]
+# read_only = true
+
 # === Anti-Censorship & Masking ===
 [censorship]
 tls_domain = "petrovich.ru"
@@ -75,6 +81,7 @@ tls_domain = "petrovich.ru"
 # format: "username" = "32_hex_chars_secret"
 hello = "00000000000000000000000000000000"
 ```
+
 then Ctrl+S -> Ctrl+X to save
 
 > [!WARNING]
@@ -115,7 +122,12 @@ then Ctrl+S -> Ctrl+X to save
 
 **5.** For automatic startup at system boot, enter `systemctl enable telemt`
 
-**6.** To get the links, enter `journalctl -u telemt -n -g "links" --no-pager -o cat | tac`
+**6.** To get the link(s), enter
+```bash
+curl -s http://127.0.0.1:9091/v1/users | jq
+```
+
+> Any number of people can use one link.
 
 ---
 

docs/QUICK_START_GUIDE.ru.md

@@ -67,6 +67,12 @@ classic = false
 secure = false
 tls = true
 
+[server.api]
+enabled = true
+# listen = "127.0.0.1:9091"
+# whitelist = ["127.0.0.1/32"]
+# read_only = true
+
 # === Anti-Censorship & Masking ===
 [censorship]
 tls_domain = "petrovich.ru"
@@ -75,6 +81,7 @@ tls_domain = "petrovich.ru"
 # format: "username" = "32_hex_chars_secret"
 hello = "00000000000000000000000000000000"
 ```
+
 Затем нажмите Ctrl+S -> Ctrl+X, чтобы сохранить
 
 > [!WARNING]
@@ -115,9 +122,14 @@ WantedBy=multi-user.target
 
 **5.** Для автоматического запуска при запуске системы в введите `systemctl enable telemt`
 
-**6.** Для получения ссылки введите `journalctl -u telemt -n -g "links" --no-pager -o cat | tac`
+**6.** Для получения ссылки/ссылок введите 
+```bash
+curl -s http://127.0.0.1:9091/v1/users | jq
+```
+> Одной ссылкой модет пользоваться сколько угодно человек.
+
 > [!WARNING]
-> Рабочую ссылку может выдать только команда из 6 пункта. Не пытайтесь делать ее самостоятельно или копировать откуда-либо!
+> Рабочую ссылку может выдать только команда из 6 пункта. Не пытайтесь делать ее самостоятельно или копировать откуда-либо если вы не уверены в том, что делаете!
 
 ---
 

src/api/model.rs

@@ -269,6 +269,10 @@ pub(super) struct DcStatus {
     pub(super) available_endpoints: usize,
     pub(super) available_pct: f64,
     pub(super) required_writers: usize,
+    pub(super) floor_min: usize,
+    pub(super) floor_target: usize,
+    pub(super) floor_max: usize,
+    pub(super) floor_capped: bool,
     pub(super) alive_writers: usize,
     pub(super) coverage_pct: f64,
     pub(super) rtt_ms: Option<f64>,
@@ -308,7 +312,17 @@ pub(super) struct MinimalMeRuntimeData {
     pub(super) floor_mode: &'static str,
     pub(super) adaptive_floor_idle_secs: u64,
     pub(super) adaptive_floor_min_writers_single_endpoint: u8,
+    pub(super) adaptive_floor_min_writers_multi_endpoint: u8,
     pub(super) adaptive_floor_recover_grace_secs: u64,
+    pub(super) adaptive_floor_writers_per_core_total: u16,
+    pub(super) adaptive_floor_cpu_cores_override: u16,
+    pub(super) adaptive_floor_max_extra_writers_single_per_core: u16,
+    pub(super) adaptive_floor_max_extra_writers_multi_per_core: u16,
+    pub(super) adaptive_floor_cpu_cores_detected: u32,
+    pub(super) adaptive_floor_cpu_cores_effective: u32,
+    pub(super) adaptive_floor_global_cap_raw: u64,
+    pub(super) adaptive_floor_global_cap_effective: u64,
+    pub(super) adaptive_floor_target_writers_total: u64,
     pub(super) me_keepalive_enabled: bool,
     pub(super) me_keepalive_interval_secs: u64,
     pub(super) me_keepalive_jitter_secs: u64,

src/api/runtime_stats.rs

@@ -349,6 +349,10 @@ async fn get_minimal_payload_cached(
                 available_endpoints: entry.available_endpoints,
                 available_pct: entry.available_pct,
                 required_writers: entry.required_writers,
+                floor_min: entry.floor_min,
+                floor_target: entry.floor_target,
+                floor_max: entry.floor_max,
+                floor_capped: entry.floor_capped,
                 alive_writers: entry.alive_writers,
                 coverage_pct: entry.coverage_pct,
                 rtt_ms: entry.rtt_ms,
@@ -366,7 +370,21 @@ async fn get_minimal_payload_cached(
         adaptive_floor_idle_secs: runtime.adaptive_floor_idle_secs,
         adaptive_floor_min_writers_single_endpoint: runtime
             .adaptive_floor_min_writers_single_endpoint,
+        adaptive_floor_min_writers_multi_endpoint: runtime
+            .adaptive_floor_min_writers_multi_endpoint,
         adaptive_floor_recover_grace_secs: runtime.adaptive_floor_recover_grace_secs,
+        adaptive_floor_writers_per_core_total: runtime
+            .adaptive_floor_writers_per_core_total,
+        adaptive_floor_cpu_cores_override: runtime.adaptive_floor_cpu_cores_override,
+        adaptive_floor_max_extra_writers_single_per_core: runtime
+            .adaptive_floor_max_extra_writers_single_per_core,
+        adaptive_floor_max_extra_writers_multi_per_core: runtime
+            .adaptive_floor_max_extra_writers_multi_per_core,
+        adaptive_floor_cpu_cores_detected: runtime.adaptive_floor_cpu_cores_detected,
+        adaptive_floor_cpu_cores_effective: runtime.adaptive_floor_cpu_cores_effective,
+        adaptive_floor_global_cap_raw: runtime.adaptive_floor_global_cap_raw,
+        adaptive_floor_global_cap_effective: runtime.adaptive_floor_global_cap_effective,
+        adaptive_floor_target_writers_total: runtime.adaptive_floor_target_writers_total,
         me_keepalive_enabled: runtime.me_keepalive_enabled,
         me_keepalive_interval_secs: runtime.me_keepalive_interval_secs,
         me_keepalive_jitter_secs: runtime.me_keepalive_jitter_secs,

src/api/runtime_zero.rs

@@ -60,7 +60,12 @@ pub(super) struct EffectiveMiddleProxyLimits {
     pub(super) floor_mode: &'static str,
     pub(super) adaptive_floor_idle_secs: u64,
     pub(super) adaptive_floor_min_writers_single_endpoint: u8,
+    pub(super) adaptive_floor_min_writers_multi_endpoint: u8,
     pub(super) adaptive_floor_recover_grace_secs: u64,
+    pub(super) adaptive_floor_writers_per_core_total: u16,
+    pub(super) adaptive_floor_cpu_cores_override: u16,
+    pub(super) adaptive_floor_max_extra_writers_single_per_core: u16,
+    pub(super) adaptive_floor_max_extra_writers_multi_per_core: u16,
     pub(super) reconnect_max_concurrent_per_dc: u32,
     pub(super) reconnect_backoff_base_ms: u64,
     pub(super) reconnect_backoff_cap_ms: u64,
@@ -183,7 +188,22 @@ pub(super) fn build_limits_effective_data(cfg: &ProxyConfig) -> EffectiveLimitsD
             adaptive_floor_min_writers_single_endpoint: cfg
                 .general
                 .me_adaptive_floor_min_writers_single_endpoint,
+            adaptive_floor_min_writers_multi_endpoint: cfg
+                .general
+                .me_adaptive_floor_min_writers_multi_endpoint,
             adaptive_floor_recover_grace_secs: cfg.general.me_adaptive_floor_recover_grace_secs,
+            adaptive_floor_writers_per_core_total: cfg
+                .general
+                .me_adaptive_floor_writers_per_core_total,
+            adaptive_floor_cpu_cores_override: cfg
+                .general
+                .me_adaptive_floor_cpu_cores_override,
+            adaptive_floor_max_extra_writers_single_per_core: cfg
+                .general
+                .me_adaptive_floor_max_extra_writers_single_per_core,
+            adaptive_floor_max_extra_writers_multi_per_core: cfg
+                .general
+                .me_adaptive_floor_max_extra_writers_multi_per_core,
             reconnect_max_concurrent_per_dc: cfg.general.me_reconnect_max_concurrent_per_dc,
             reconnect_backoff_base_ms: cfg.general.me_reconnect_backoff_base_ms,
             reconnect_backoff_cap_ms: cfg.general.me_reconnect_backoff_cap_ms,

src/config/defaults.rs

@@ -11,7 +11,12 @@ const DEFAULT_ME_RECONNECT_FAST_RETRY_COUNT: u32 = 16;
 const DEFAULT_ME_SINGLE_ENDPOINT_SHADOW_WRITERS: u8 = 2;
 const DEFAULT_ME_ADAPTIVE_FLOOR_IDLE_SECS: u64 = 90;
 const DEFAULT_ME_ADAPTIVE_FLOOR_MIN_WRITERS_SINGLE_ENDPOINT: u8 = 1;
+const DEFAULT_ME_ADAPTIVE_FLOOR_MIN_WRITERS_MULTI_ENDPOINT: u8 = 1;
 const DEFAULT_ME_ADAPTIVE_FLOOR_RECOVER_GRACE_SECS: u64 = 180;
+const DEFAULT_ME_ADAPTIVE_FLOOR_WRITERS_PER_CORE_TOTAL: u16 = 48;
+const DEFAULT_ME_ADAPTIVE_FLOOR_CPU_CORES_OVERRIDE: u16 = 0;
+const DEFAULT_ME_ADAPTIVE_FLOOR_MAX_EXTRA_WRITERS_SINGLE_PER_CORE: u16 = 1;
+const DEFAULT_ME_ADAPTIVE_FLOOR_MAX_EXTRA_WRITERS_MULTI_PER_CORE: u16 = 2;
 const DEFAULT_USER_MAX_UNIQUE_IPS_WINDOW_SECS: u64 = 30;
 const DEFAULT_UPSTREAM_CONNECT_RETRY_ATTEMPTS: u32 = 2;
 const DEFAULT_UPSTREAM_UNHEALTHY_FAIL_THRESHOLD: u32 = 5;
@@ -247,10 +252,30 @@ pub(crate) fn default_me_adaptive_floor_min_writers_single_endpoint() -> u8 {
     DEFAULT_ME_ADAPTIVE_FLOOR_MIN_WRITERS_SINGLE_ENDPOINT
 }
 
+pub(crate) fn default_me_adaptive_floor_min_writers_multi_endpoint() -> u8 {
+    DEFAULT_ME_ADAPTIVE_FLOOR_MIN_WRITERS_MULTI_ENDPOINT
+}
+
 pub(crate) fn default_me_adaptive_floor_recover_grace_secs() -> u64 {
     DEFAULT_ME_ADAPTIVE_FLOOR_RECOVER_GRACE_SECS
 }
 
+pub(crate) fn default_me_adaptive_floor_writers_per_core_total() -> u16 {
+    DEFAULT_ME_ADAPTIVE_FLOOR_WRITERS_PER_CORE_TOTAL
+}
+
+pub(crate) fn default_me_adaptive_floor_cpu_cores_override() -> u16 {
+    DEFAULT_ME_ADAPTIVE_FLOOR_CPU_CORES_OVERRIDE
+}
+
+pub(crate) fn default_me_adaptive_floor_max_extra_writers_single_per_core() -> u16 {
+    DEFAULT_ME_ADAPTIVE_FLOOR_MAX_EXTRA_WRITERS_SINGLE_PER_CORE
+}
+
+pub(crate) fn default_me_adaptive_floor_max_extra_writers_multi_per_core() -> u16 {
+    DEFAULT_ME_ADAPTIVE_FLOOR_MAX_EXTRA_WRITERS_MULTI_PER_CORE
+}
+
 pub(crate) fn default_upstream_connect_retry_attempts() -> u32 {
     DEFAULT_UPSTREAM_CONNECT_RETRY_ATTEMPTS
 }

src/config/hot_reload.rs

@@ -78,7 +78,12 @@ pub struct HotFields {
     pub me_floor_mode: MeFloorMode,
     pub me_adaptive_floor_idle_secs: u64,
     pub me_adaptive_floor_min_writers_single_endpoint: u8,
+    pub me_adaptive_floor_min_writers_multi_endpoint: u8,
     pub me_adaptive_floor_recover_grace_secs: u64,
+    pub me_adaptive_floor_writers_per_core_total: u16,
+    pub me_adaptive_floor_cpu_cores_override: u16,
+    pub me_adaptive_floor_max_extra_writers_single_per_core: u16,
+    pub me_adaptive_floor_max_extra_writers_multi_per_core: u16,
     pub me_route_backpressure_base_timeout_ms: u64,
     pub me_route_backpressure_high_timeout_ms: u64,
     pub me_route_backpressure_high_watermark_pct: u8,
@@ -150,9 +155,24 @@ impl HotFields {
             me_adaptive_floor_min_writers_single_endpoint: cfg
                 .general
                 .me_adaptive_floor_min_writers_single_endpoint,
+            me_adaptive_floor_min_writers_multi_endpoint: cfg
+                .general
+                .me_adaptive_floor_min_writers_multi_endpoint,
             me_adaptive_floor_recover_grace_secs: cfg
                 .general
                 .me_adaptive_floor_recover_grace_secs,
+            me_adaptive_floor_writers_per_core_total: cfg
+                .general
+                .me_adaptive_floor_writers_per_core_total,
+            me_adaptive_floor_cpu_cores_override: cfg
+                .general
+                .me_adaptive_floor_cpu_cores_override,
+            me_adaptive_floor_max_extra_writers_single_per_core: cfg
+                .general
+                .me_adaptive_floor_max_extra_writers_single_per_core,
+            me_adaptive_floor_max_extra_writers_multi_per_core: cfg
+                .general
+                .me_adaptive_floor_max_extra_writers_multi_per_core,
             me_route_backpressure_base_timeout_ms: cfg.general.me_route_backpressure_base_timeout_ms,
             me_route_backpressure_high_timeout_ms: cfg.general.me_route_backpressure_high_timeout_ms,
             me_route_backpressure_high_watermark_pct: cfg.general.me_route_backpressure_high_watermark_pct,
@@ -273,8 +293,18 @@ fn overlay_hot_fields(old: &ProxyConfig, new: &ProxyConfig) -> ProxyConfig {
     cfg.general.me_adaptive_floor_idle_secs = new.general.me_adaptive_floor_idle_secs;
     cfg.general.me_adaptive_floor_min_writers_single_endpoint =
         new.general.me_adaptive_floor_min_writers_single_endpoint;
+    cfg.general.me_adaptive_floor_min_writers_multi_endpoint =
+        new.general.me_adaptive_floor_min_writers_multi_endpoint;
     cfg.general.me_adaptive_floor_recover_grace_secs =
         new.general.me_adaptive_floor_recover_grace_secs;
+    cfg.general.me_adaptive_floor_writers_per_core_total =
+        new.general.me_adaptive_floor_writers_per_core_total;
+    cfg.general.me_adaptive_floor_cpu_cores_override =
+        new.general.me_adaptive_floor_cpu_cores_override;
+    cfg.general.me_adaptive_floor_max_extra_writers_single_per_core =
+        new.general.me_adaptive_floor_max_extra_writers_single_per_core;
+    cfg.general.me_adaptive_floor_max_extra_writers_multi_per_core =
+        new.general.me_adaptive_floor_max_extra_writers_multi_per_core;
     cfg.general.me_route_backpressure_base_timeout_ms =
         new.general.me_route_backpressure_base_timeout_ms;
     cfg.general.me_route_backpressure_high_timeout_ms =
@@ -697,15 +727,30 @@ fn log_changes(
         || old_hot.me_adaptive_floor_idle_secs != new_hot.me_adaptive_floor_idle_secs
         || old_hot.me_adaptive_floor_min_writers_single_endpoint
             != new_hot.me_adaptive_floor_min_writers_single_endpoint
+        || old_hot.me_adaptive_floor_min_writers_multi_endpoint
+            != new_hot.me_adaptive_floor_min_writers_multi_endpoint
         || old_hot.me_adaptive_floor_recover_grace_secs
             != new_hot.me_adaptive_floor_recover_grace_secs
+        || old_hot.me_adaptive_floor_writers_per_core_total
+            != new_hot.me_adaptive_floor_writers_per_core_total
+        || old_hot.me_adaptive_floor_cpu_cores_override
+            != new_hot.me_adaptive_floor_cpu_cores_override
+        || old_hot.me_adaptive_floor_max_extra_writers_single_per_core
+            != new_hot.me_adaptive_floor_max_extra_writers_single_per_core
+        || old_hot.me_adaptive_floor_max_extra_writers_multi_per_core
+            != new_hot.me_adaptive_floor_max_extra_writers_multi_per_core
     {
         info!(
-            "config reload: me_floor: mode={:?} idle={}s min_single={} recover_grace={}s",
+            "config reload: me_floor: mode={:?} idle={}s min_single={} min_multi={} recover_grace={}s per_core_total={} cores_override={} extra_single_per_core={} extra_multi_per_core={}",
             new_hot.me_floor_mode,
             new_hot.me_adaptive_floor_idle_secs,
             new_hot.me_adaptive_floor_min_writers_single_endpoint,
+            new_hot.me_adaptive_floor_min_writers_multi_endpoint,
             new_hot.me_adaptive_floor_recover_grace_secs,
+            new_hot.me_adaptive_floor_writers_per_core_total,
+            new_hot.me_adaptive_floor_cpu_cores_override,
+            new_hot.me_adaptive_floor_max_extra_writers_single_per_core,
+            new_hot.me_adaptive_floor_max_extra_writers_multi_per_core,
         );
     }
 

src/config/load.rs

@@ -312,6 +312,21 @@ impl ProxyConfig {
             ));
         }
 
+        if config.general.me_adaptive_floor_min_writers_multi_endpoint == 0
+            || config.general.me_adaptive_floor_min_writers_multi_endpoint > 32
+        {
+            return Err(ProxyError::Config(
+                "general.me_adaptive_floor_min_writers_multi_endpoint must be within [1, 32]"
+                    .to_string(),
+            ));
+        }
+
+        if config.general.me_adaptive_floor_writers_per_core_total == 0 {
+            return Err(ProxyError::Config(
+                "general.me_adaptive_floor_writers_per_core_total must be > 0".to_string(),
+            ));
+        }
+
         if config.general.me_single_endpoint_outage_backoff_min_ms == 0 {
             return Err(ProxyError::Config(
                 "general.me_single_endpoint_outage_backoff_min_ms must be > 0".to_string(),

src/config/types.rs

@@ -520,10 +520,31 @@ pub struct GeneralConfig {
     #[serde(default = "default_me_adaptive_floor_min_writers_single_endpoint")]
     pub me_adaptive_floor_min_writers_single_endpoint: u8,
 
+    /// Minimum writer target for multi-endpoint DC groups in adaptive floor mode.
+    #[serde(default = "default_me_adaptive_floor_min_writers_multi_endpoint")]
+    pub me_adaptive_floor_min_writers_multi_endpoint: u8,
+
     /// Grace period in seconds to hold static floor after activity in adaptive mode.
     #[serde(default = "default_me_adaptive_floor_recover_grace_secs")]
     pub me_adaptive_floor_recover_grace_secs: u64,
 
+    /// Global ME writer budget per logical CPU core in adaptive mode.
+    #[serde(default = "default_me_adaptive_floor_writers_per_core_total")]
+    pub me_adaptive_floor_writers_per_core_total: u16,
+
+    /// Override logical CPU core count for adaptive floor calculations.
+    /// Set to 0 to use runtime auto-detection.
+    #[serde(default = "default_me_adaptive_floor_cpu_cores_override")]
+    pub me_adaptive_floor_cpu_cores_override: u16,
+
+    /// Per-core max extra writers above base required floor for single-endpoint DC groups.
+    #[serde(default = "default_me_adaptive_floor_max_extra_writers_single_per_core")]
+    pub me_adaptive_floor_max_extra_writers_single_per_core: u16,
+
+    /// Per-core max extra writers above base required floor for multi-endpoint DC groups.
+    #[serde(default = "default_me_adaptive_floor_max_extra_writers_multi_per_core")]
+    pub me_adaptive_floor_max_extra_writers_multi_per_core: u16,
+
     /// Connect attempts for the selected upstream before returning error/fallback.
     #[serde(default = "default_upstream_connect_retry_attempts")]
     pub upstream_connect_retry_attempts: u32,
@@ -775,7 +796,12 @@ impl Default for GeneralConfig {
             me_floor_mode: MeFloorMode::default(),
             me_adaptive_floor_idle_secs: default_me_adaptive_floor_idle_secs(),
             me_adaptive_floor_min_writers_single_endpoint: default_me_adaptive_floor_min_writers_single_endpoint(),
+            me_adaptive_floor_min_writers_multi_endpoint: default_me_adaptive_floor_min_writers_multi_endpoint(),
             me_adaptive_floor_recover_grace_secs: default_me_adaptive_floor_recover_grace_secs(),
+            me_adaptive_floor_writers_per_core_total: default_me_adaptive_floor_writers_per_core_total(),
+            me_adaptive_floor_cpu_cores_override: default_me_adaptive_floor_cpu_cores_override(),
+            me_adaptive_floor_max_extra_writers_single_per_core: default_me_adaptive_floor_max_extra_writers_single_per_core(),
+            me_adaptive_floor_max_extra_writers_multi_per_core: default_me_adaptive_floor_max_extra_writers_multi_per_core(),
             upstream_connect_retry_attempts: default_upstream_connect_retry_attempts(),
             upstream_connect_retry_backoff_ms: default_upstream_connect_retry_backoff_ms(),
             upstream_connect_budget_ms: default_upstream_connect_budget_ms(),

src/main.rs

@@ -786,7 +786,12 @@ async fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
                         config.general.me_floor_mode,
                         config.general.me_adaptive_floor_idle_secs,
                         config.general.me_adaptive_floor_min_writers_single_endpoint,
+                        config.general.me_adaptive_floor_min_writers_multi_endpoint,
                         config.general.me_adaptive_floor_recover_grace_secs,
+                        config.general.me_adaptive_floor_writers_per_core_total,
+                        config.general.me_adaptive_floor_cpu_cores_override,
+                        config.general.me_adaptive_floor_max_extra_writers_single_per_core,
+                        config.general.me_adaptive_floor_max_extra_writers_multi_per_core,
                         config.general.hardswap,
                         config.general.me_pool_drain_ttl_secs,
                         config.general.effective_me_pool_force_close_secs(),

src/metrics.rs

@@ -968,6 +968,133 @@ async fn render_metrics(stats: &Stats, config: &ProxyConfig, ip_tracker: &UserIp
             0
         }
     );
+    let _ = writeln!(
+        out,
+        "# HELP telemt_me_adaptive_floor_cpu_cores_detected Runtime detected logical CPU cores for adaptive floor"
+    );
+    let _ = writeln!(
+        out,
+        "# TYPE telemt_me_adaptive_floor_cpu_cores_detected gauge"
+    );
+    let _ = writeln!(
+        out,
+        "telemt_me_adaptive_floor_cpu_cores_detected {}",
+        if me_allows_normal {
+            stats.get_me_floor_cpu_cores_detected_gauge()
+        } else {
+            0
+        }
+    );
+    let _ = writeln!(
+        out,
+        "# HELP telemt_me_adaptive_floor_cpu_cores_effective Runtime effective logical CPU cores for adaptive floor"
+    );
+    let _ = writeln!(
+        out,
+        "# TYPE telemt_me_adaptive_floor_cpu_cores_effective gauge"
+    );
+    let _ = writeln!(
+        out,
+        "telemt_me_adaptive_floor_cpu_cores_effective {}",
+        if me_allows_normal {
+            stats.get_me_floor_cpu_cores_effective_gauge()
+        } else {
+            0
+        }
+    );
+    let _ = writeln!(
+        out,
+        "# HELP telemt_me_adaptive_floor_global_cap_raw Runtime raw global adaptive floor cap"
+    );
+    let _ = writeln!(
+        out,
+        "# TYPE telemt_me_adaptive_floor_global_cap_raw gauge"
+    );
+    let _ = writeln!(
+        out,
+        "telemt_me_adaptive_floor_global_cap_raw {}",
+        if me_allows_normal {
+            stats.get_me_floor_global_cap_raw_gauge()
+        } else {
+            0
+        }
+    );
+    let _ = writeln!(
+        out,
+        "# HELP telemt_me_adaptive_floor_global_cap_effective Runtime effective global adaptive floor cap"
+    );
+    let _ = writeln!(
+        out,
+        "# TYPE telemt_me_adaptive_floor_global_cap_effective gauge"
+    );
+    let _ = writeln!(
+        out,
+        "telemt_me_adaptive_floor_global_cap_effective {}",
+        if me_allows_normal {
+            stats.get_me_floor_global_cap_effective_gauge()
+        } else {
+            0
+        }
+    );
+    let _ = writeln!(
+        out,
+        "# HELP telemt_me_adaptive_floor_target_writers_total Runtime adaptive floor target writers total"
+    );
+    let _ = writeln!(
+        out,
+        "# TYPE telemt_me_adaptive_floor_target_writers_total gauge"
+    );
+    let _ = writeln!(
+        out,
+        "telemt_me_adaptive_floor_target_writers_total {}",
+        if me_allows_normal {
+            stats.get_me_floor_target_writers_total_gauge()
+        } else {
+            0
+        }
+    );
+    let _ = writeln!(
+        out,
+        "# HELP telemt_me_floor_cap_block_total Reconnect attempts blocked by adaptive floor caps"
+    );
+    let _ = writeln!(out, "# TYPE telemt_me_floor_cap_block_total counter");
+    let _ = writeln!(
+        out,
+        "telemt_me_floor_cap_block_total {}",
+        if me_allows_normal {
+            stats.get_me_floor_cap_block_total()
+        } else {
+            0
+        }
+    );
+    let _ = writeln!(
+        out,
+        "# HELP telemt_me_floor_swap_idle_total Adaptive floor cap recovery via idle writer swap"
+    );
+    let _ = writeln!(out, "# TYPE telemt_me_floor_swap_idle_total counter");
+    let _ = writeln!(
+        out,
+        "telemt_me_floor_swap_idle_total {}",
+        if me_allows_normal {
+            stats.get_me_floor_swap_idle_total()
+        } else {
+            0
+        }
+    );
+    let _ = writeln!(
+        out,
+        "# HELP telemt_me_floor_swap_idle_failed_total Failed idle swap attempts under adaptive floor caps"
+    );
+    let _ = writeln!(out, "# TYPE telemt_me_floor_swap_idle_failed_total counter");
+    let _ = writeln!(
+        out,
+        "telemt_me_floor_swap_idle_failed_total {}",
+        if me_allows_normal {
+            stats.get_me_floor_swap_idle_failed_total()
+        } else {
+            0
+        }
+    );
 
     let _ = writeln!(out, "# HELP telemt_secure_padding_invalid_total Invalid secure frame lengths");
     let _ = writeln!(out, "# TYPE telemt_secure_padding_invalid_total counter");

src/stats/mod.rs

@@ -75,6 +75,14 @@ pub struct Stats {
     me_floor_mode_switch_total: AtomicU64,
     me_floor_mode_switch_static_to_adaptive_total: AtomicU64,
     me_floor_mode_switch_adaptive_to_static_total: AtomicU64,
+    me_floor_cpu_cores_detected_gauge: AtomicU64,
+    me_floor_cpu_cores_effective_gauge: AtomicU64,
+    me_floor_global_cap_raw_gauge: AtomicU64,
+    me_floor_global_cap_effective_gauge: AtomicU64,
+    me_floor_target_writers_total_gauge: AtomicU64,
+    me_floor_cap_block_total: AtomicU64,
+    me_floor_swap_idle_total: AtomicU64,
+    me_floor_swap_idle_failed_total: AtomicU64,
     me_handshake_error_codes: DashMap<i32, AtomicU64>,
     me_route_drop_no_conn: AtomicU64,
     me_route_drop_channel_closed: AtomicU64,
@@ -676,6 +684,52 @@ impl Stats {
                 .fetch_add(1, Ordering::Relaxed);
         }
     }
+    pub fn set_me_floor_cpu_cores_detected_gauge(&self, value: u64) {
+        if self.telemetry_me_allows_normal() {
+            self.me_floor_cpu_cores_detected_gauge
+                .store(value, Ordering::Relaxed);
+        }
+    }
+    pub fn set_me_floor_cpu_cores_effective_gauge(&self, value: u64) {
+        if self.telemetry_me_allows_normal() {
+            self.me_floor_cpu_cores_effective_gauge
+                .store(value, Ordering::Relaxed);
+        }
+    }
+    pub fn set_me_floor_global_cap_raw_gauge(&self, value: u64) {
+        if self.telemetry_me_allows_normal() {
+            self.me_floor_global_cap_raw_gauge
+                .store(value, Ordering::Relaxed);
+        }
+    }
+    pub fn set_me_floor_global_cap_effective_gauge(&self, value: u64) {
+        if self.telemetry_me_allows_normal() {
+            self.me_floor_global_cap_effective_gauge
+                .store(value, Ordering::Relaxed);
+        }
+    }
+    pub fn set_me_floor_target_writers_total_gauge(&self, value: u64) {
+        if self.telemetry_me_allows_normal() {
+            self.me_floor_target_writers_total_gauge
+                .store(value, Ordering::Relaxed);
+        }
+    }
+    pub fn increment_me_floor_cap_block_total(&self) {
+        if self.telemetry_me_allows_normal() {
+            self.me_floor_cap_block_total.fetch_add(1, Ordering::Relaxed);
+        }
+    }
+    pub fn increment_me_floor_swap_idle_total(&self) {
+        if self.telemetry_me_allows_normal() {
+            self.me_floor_swap_idle_total.fetch_add(1, Ordering::Relaxed);
+        }
+    }
+    pub fn increment_me_floor_swap_idle_failed_total(&self) {
+        if self.telemetry_me_allows_normal() {
+            self.me_floor_swap_idle_failed_total
+                .fetch_add(1, Ordering::Relaxed);
+        }
+    }
     pub fn get_connects_all(&self) -> u64 { self.connects_all.load(Ordering::Relaxed) }
     pub fn get_connects_bad(&self) -> u64 { self.connects_bad.load(Ordering::Relaxed) }
     pub fn get_current_connections_direct(&self) -> u64 {
@@ -781,6 +835,34 @@ impl Stats {
         self.me_floor_mode_switch_adaptive_to_static_total
             .load(Ordering::Relaxed)
     }
+    pub fn get_me_floor_cpu_cores_detected_gauge(&self) -> u64 {
+        self.me_floor_cpu_cores_detected_gauge
+            .load(Ordering::Relaxed)
+    }
+    pub fn get_me_floor_cpu_cores_effective_gauge(&self) -> u64 {
+        self.me_floor_cpu_cores_effective_gauge
+            .load(Ordering::Relaxed)
+    }
+    pub fn get_me_floor_global_cap_raw_gauge(&self) -> u64 {
+        self.me_floor_global_cap_raw_gauge.load(Ordering::Relaxed)
+    }
+    pub fn get_me_floor_global_cap_effective_gauge(&self) -> u64 {
+        self.me_floor_global_cap_effective_gauge
+            .load(Ordering::Relaxed)
+    }
+    pub fn get_me_floor_target_writers_total_gauge(&self) -> u64 {
+        self.me_floor_target_writers_total_gauge
+            .load(Ordering::Relaxed)
+    }
+    pub fn get_me_floor_cap_block_total(&self) -> u64 {
+        self.me_floor_cap_block_total.load(Ordering::Relaxed)
+    }
+    pub fn get_me_floor_swap_idle_total(&self) -> u64 {
+        self.me_floor_swap_idle_total.load(Ordering::Relaxed)
+    }
+    pub fn get_me_floor_swap_idle_failed_total(&self) -> u64 {
+        self.me_floor_swap_idle_failed_total.load(Ordering::Relaxed)
+    }
     pub fn get_me_handshake_error_code_counts(&self) -> Vec<(i32, u64)> {
         let mut out: Vec<(i32, u64)> = self
             .me_handshake_error_codes

src/transport/middle_proxy/config_updater.rs

@@ -315,7 +315,12 @@ async fn run_update_cycle(
         cfg.general.me_floor_mode,
         cfg.general.me_adaptive_floor_idle_secs,
         cfg.general.me_adaptive_floor_min_writers_single_endpoint,
+        cfg.general.me_adaptive_floor_min_writers_multi_endpoint,
         cfg.general.me_adaptive_floor_recover_grace_secs,
+        cfg.general.me_adaptive_floor_writers_per_core_total,
+        cfg.general.me_adaptive_floor_cpu_cores_override,
+        cfg.general.me_adaptive_floor_max_extra_writers_single_per_core,
+        cfg.general.me_adaptive_floor_max_extra_writers_multi_per_core,
     );
 
     let required_cfg_snapshots = cfg.general.me_config_stable_snapshots.max(1);
@@ -527,7 +532,12 @@ pub async fn me_config_updater(
                     cfg.general.me_floor_mode,
                     cfg.general.me_adaptive_floor_idle_secs,
                     cfg.general.me_adaptive_floor_min_writers_single_endpoint,
+                    cfg.general.me_adaptive_floor_min_writers_multi_endpoint,
                     cfg.general.me_adaptive_floor_recover_grace_secs,
+                    cfg.general.me_adaptive_floor_writers_per_core_total,
+                    cfg.general.me_adaptive_floor_cpu_cores_override,
+                    cfg.general.me_adaptive_floor_max_extra_writers_single_per_core,
+                    cfg.general.me_adaptive_floor_max_extra_writers_multi_per_core,
                 );
                 let new_secs = cfg.general.effective_update_every_secs().max(1);
                 if new_secs == update_every_secs {

src/transport/middle_proxy/health.rs

@@ -23,6 +23,25 @@ const IDLE_REFRESH_TRIGGER_JITTER_SECS: u64 = 5;
 const IDLE_REFRESH_RETRY_SECS: u64 = 8;
 const IDLE_REFRESH_SUCCESS_GUARD_SECS: u64 = 5;
 
+#[derive(Debug, Clone)]
+struct DcFloorPlanEntry {
+    dc: i32,
+    endpoints: Vec<SocketAddr>,
+    alive: usize,
+    min_required: usize,
+    target_required: usize,
+    max_required: usize,
+    has_bound_clients: bool,
+    floor_capped: bool,
+}
+
+#[derive(Debug, Clone)]
+struct FamilyFloorPlan {
+    by_dc: HashMap<i32, DcFloorPlanEntry>,
+    global_cap_effective_total: usize,
+    target_writers_total: usize,
+}
+
 pub async fn me_health_monitor(pool: Arc<MePool>, rng: Arc<SecureRandom>, _min_connections: usize) {
     let mut backoff: HashMap<(i32, IpFamily), u64> = HashMap::new();
     let mut next_attempt: HashMap<(i32, IpFamily), Instant> = HashMap::new();
@@ -129,22 +148,33 @@ async fn check_family(
             .push(writer.id);
     }
     let writer_idle_since = pool.registry.writer_idle_since_snapshot().await;
+    let floor_plan = build_family_floor_plan(
+        pool,
+        family,
+        &dc_endpoints,
+        &live_addr_counts,
+        &live_writer_ids_by_addr,
+        adaptive_idle_since,
+        adaptive_recover_until,
+    )
+    .await;
+    pool.set_adaptive_floor_runtime_caps(
+        floor_plan.global_cap_effective_total,
+        floor_plan.target_writers_total,
+    );
 
     for (dc, endpoints) in dc_endpoints {
         if endpoints.is_empty() {
             continue;
         }
         let key = (dc, family);
-        let reduce_for_idle = should_reduce_floor_for_idle(
-            pool,
-            key,
-            &endpoints,
-            &live_writer_ids_by_addr,
-            adaptive_idle_since,
-            adaptive_recover_until,
-        )
-        .await;
-        let required = pool.required_writers_for_dc_with_floor_mode(endpoints.len(), reduce_for_idle);
+        let required = floor_plan
+            .by_dc
+            .get(&dc)
+            .map(|entry| entry.target_required)
+            .unwrap_or_else(|| {
+                pool.required_writers_for_dc_with_floor_mode(endpoints.len(), false)
+            });
         let alive = endpoints
             .iter()
             .map(|addr| *live_addr_counts.get(addr).unwrap_or(&0))
@@ -251,6 +281,36 @@ async fn check_family(
 
         let mut restored = 0usize;
         for _ in 0..missing {
+            if pool.floor_mode() == MeFloorMode::Adaptive
+                && pool.active_writer_count_total().await >= floor_plan.global_cap_effective_total
+            {
+                let swapped = maybe_swap_idle_writer_for_cap(
+                    pool,
+                    rng,
+                    dc,
+                    family,
+                    &endpoints,
+                    &live_writer_ids_by_addr,
+                    &writer_idle_since,
+                )
+                .await;
+                if swapped {
+                    pool.stats.increment_me_floor_swap_idle_total();
+                    restored += 1;
+                    continue;
+                }
+                pool.stats.increment_me_floor_cap_block_total();
+                pool.stats.increment_me_floor_swap_idle_failed_total();
+                debug!(
+                    dc = %dc,
+                    ?family,
+                    alive,
+                    required,
+                    global_cap_effective_total = floor_plan.global_cap_effective_total,
+                    "Adaptive floor cap reached, reconnect attempt blocked"
+                );
+                break;
+            }
             let res = tokio::time::timeout(
                 pool.me_one_timeout,
                 pool.connect_endpoints_round_robin(&endpoints, rng.as_ref()),
@@ -323,6 +383,280 @@ async fn check_family(
     }
 }
 
+fn adaptive_floor_class_min(
+    pool: &Arc<MePool>,
+    endpoint_count: usize,
+    base_required: usize,
+) -> usize {
+    if endpoint_count <= 1 {
+        let min_single = (pool
+            .me_adaptive_floor_min_writers_single_endpoint
+            .load(std::sync::atomic::Ordering::Relaxed) as usize)
+            .max(1);
+        min_single.min(base_required.max(1))
+    } else {
+        pool.adaptive_floor_min_writers_multi_endpoint()
+            .min(base_required.max(1))
+    }
+}
+
+fn adaptive_floor_class_max(
+    pool: &Arc<MePool>,
+    endpoint_count: usize,
+    base_required: usize,
+    cpu_cores: usize,
+) -> usize {
+    let extra_per_core = if endpoint_count <= 1 {
+        pool.adaptive_floor_max_extra_single_per_core()
+    } else {
+        pool.adaptive_floor_max_extra_multi_per_core()
+    };
+    base_required.saturating_add(cpu_cores.saturating_mul(extra_per_core))
+}
+
+fn list_writer_ids_for_endpoints(
+    endpoints: &[SocketAddr],
+    live_writer_ids_by_addr: &HashMap<SocketAddr, Vec<u64>>,
+) -> Vec<u64> {
+    let mut out = Vec::<u64>::new();
+    for endpoint in endpoints {
+        if let Some(ids) = live_writer_ids_by_addr.get(endpoint) {
+            out.extend(ids.iter().copied());
+        }
+    }
+    out
+}
+
+async fn build_family_floor_plan(
+    pool: &Arc<MePool>,
+    family: IpFamily,
+    dc_endpoints: &HashMap<i32, Vec<SocketAddr>>,
+    live_addr_counts: &HashMap<SocketAddr, usize>,
+    live_writer_ids_by_addr: &HashMap<SocketAddr, Vec<u64>>,
+    adaptive_idle_since: &mut HashMap<(i32, IpFamily), Instant>,
+    adaptive_recover_until: &mut HashMap<(i32, IpFamily), Instant>,
+) -> FamilyFloorPlan {
+    let mut entries = Vec::<DcFloorPlanEntry>::new();
+    let mut by_dc = HashMap::<i32, DcFloorPlanEntry>::new();
+    let mut family_active_total = 0usize;
+
+    let floor_mode = pool.floor_mode();
+    let is_adaptive = floor_mode == MeFloorMode::Adaptive;
+    let cpu_cores = pool.adaptive_floor_effective_cpu_cores().max(1);
+
+    for (dc, endpoints) in dc_endpoints {
+        if endpoints.is_empty() {
+            continue;
+        }
+        let key = (*dc, family);
+        let reduce_for_idle = should_reduce_floor_for_idle(
+            pool,
+            key,
+            endpoints,
+            live_writer_ids_by_addr,
+            adaptive_idle_since,
+            adaptive_recover_until,
+        )
+        .await;
+        let base_required = pool.required_writers_for_dc(endpoints.len()).max(1);
+        let min_required = if is_adaptive {
+            adaptive_floor_class_min(pool, endpoints.len(), base_required)
+        } else {
+            base_required
+        };
+        let mut max_required = if is_adaptive {
+            adaptive_floor_class_max(pool, endpoints.len(), base_required, cpu_cores)
+        } else {
+            base_required
+        };
+        if max_required < min_required {
+            max_required = min_required;
+        }
+        let desired_raw = if is_adaptive && reduce_for_idle {
+            min_required
+        } else {
+            base_required
+        };
+        let target_required = desired_raw.clamp(min_required, max_required);
+        let alive = endpoints
+            .iter()
+            .map(|endpoint| live_addr_counts.get(endpoint).copied().unwrap_or(0))
+            .sum::<usize>();
+        family_active_total = family_active_total.saturating_add(alive);
+        let writer_ids = list_writer_ids_for_endpoints(endpoints, live_writer_ids_by_addr);
+        let has_bound_clients = has_bound_clients_on_endpoint(pool, &writer_ids).await;
+
+        entries.push(DcFloorPlanEntry {
+            dc: *dc,
+            endpoints: endpoints.clone(),
+            alive,
+            min_required,
+            target_required,
+            max_required,
+            has_bound_clients,
+            floor_capped: false,
+        });
+    }
+
+    if entries.is_empty() {
+        return FamilyFloorPlan {
+            by_dc,
+            global_cap_effective_total: 0,
+            target_writers_total: 0,
+        };
+    }
+
+    if !is_adaptive {
+        let target_total = entries
+            .iter()
+            .map(|entry| entry.target_required)
+            .sum::<usize>();
+        let active_total = pool.active_writer_count_total().await;
+        for entry in entries {
+            by_dc.insert(entry.dc, entry);
+        }
+        return FamilyFloorPlan {
+            by_dc,
+            global_cap_effective_total: active_total.max(target_total),
+            target_writers_total: target_total,
+        };
+    }
+
+    let global_cap_raw = pool.adaptive_floor_global_cap_raw();
+    let total_active = pool.active_writer_count_total().await;
+    let other_active = total_active.saturating_sub(family_active_total);
+    let min_sum = entries
+        .iter()
+        .map(|entry| entry.min_required)
+        .sum::<usize>();
+    let mut target_sum = entries
+        .iter()
+        .map(|entry| entry.target_required)
+        .sum::<usize>();
+    let family_cap = global_cap_raw
+        .saturating_sub(other_active)
+        .max(min_sum);
+    if target_sum > family_cap {
+        entries.sort_by_key(|entry| {
+            (
+                entry.has_bound_clients,
+                std::cmp::Reverse(entry.target_required.saturating_sub(entry.min_required)),
+                std::cmp::Reverse(entry.alive),
+                entry.dc.abs(),
+                entry.dc,
+                entry.endpoints.len(),
+                entry.max_required,
+            )
+        });
+        let mut changed = true;
+        while target_sum > family_cap && changed {
+            changed = false;
+            for entry in &mut entries {
+                if target_sum <= family_cap {
+                    break;
+                }
+                if entry.target_required > entry.min_required {
+                    entry.target_required -= 1;
+                    entry.floor_capped = true;
+                    target_sum -= 1;
+                    changed = true;
+                }
+            }
+        }
+    }
+
+    for entry in entries {
+        by_dc.insert(entry.dc, entry);
+    }
+    let global_cap_effective_total = global_cap_raw.max(other_active.saturating_add(min_sum));
+    let target_writers_total = other_active.saturating_add(target_sum);
+    FamilyFloorPlan {
+        by_dc,
+        global_cap_effective_total,
+        target_writers_total,
+    }
+}
+
+async fn maybe_swap_idle_writer_for_cap(
+    pool: &Arc<MePool>,
+    rng: &Arc<SecureRandom>,
+    dc: i32,
+    family: IpFamily,
+    endpoints: &[SocketAddr],
+    live_writer_ids_by_addr: &HashMap<SocketAddr, Vec<u64>>,
+    writer_idle_since: &HashMap<u64, u64>,
+) -> bool {
+    let now_epoch_secs = MePool::now_epoch_secs();
+    let mut candidate: Option<(u64, SocketAddr, u64)> = None;
+    for endpoint in endpoints {
+        let Some(writer_ids) = live_writer_ids_by_addr.get(endpoint) else {
+            continue;
+        };
+        for writer_id in writer_ids {
+            if !pool.registry.is_writer_empty(*writer_id).await {
+                continue;
+            }
+            let Some(idle_since_epoch_secs) = writer_idle_since.get(writer_id).copied() else {
+                continue;
+            };
+            let idle_age_secs = now_epoch_secs.saturating_sub(idle_since_epoch_secs);
+            if candidate
+                .as_ref()
+                .map(|(_, _, age)| idle_age_secs > *age)
+                .unwrap_or(true)
+            {
+                candidate = Some((*writer_id, *endpoint, idle_age_secs));
+            }
+        }
+    }
+
+    let Some((old_writer_id, endpoint, idle_age_secs)) = candidate else {
+        return false;
+    };
+
+    let connected = match tokio::time::timeout(pool.me_one_timeout, pool.connect_one(endpoint, rng.as_ref())).await {
+        Ok(Ok(())) => true,
+        Ok(Err(error)) => {
+            debug!(
+                dc = %dc,
+                ?family,
+                %endpoint,
+                old_writer_id,
+                idle_age_secs,
+                %error,
+                "Adaptive floor cap swap connect failed"
+            );
+            false
+        }
+        Err(_) => {
+            debug!(
+                dc = %dc,
+                ?family,
+                %endpoint,
+                old_writer_id,
+                idle_age_secs,
+                "Adaptive floor cap swap connect timed out"
+            );
+            false
+        }
+    };
+    if !connected {
+        return false;
+    }
+
+    pool.mark_writer_draining_with_timeout(old_writer_id, pool.force_close_timeout(), false)
+        .await;
+    info!(
+        dc = %dc,
+        ?family,
+        %endpoint,
+        old_writer_id,
+        idle_age_secs,
+        "Adaptive floor cap swap: idle writer rotated"
+    );
+    true
+}
+
 async fn maybe_refresh_idle_writer_for_dc(
     pool: &Arc<MePool>,
     rng: &Arc<SecureRandom>,
@@ -438,19 +772,15 @@ async fn should_reduce_floor_for_idle(
     adaptive_idle_since: &mut HashMap<(i32, IpFamily), Instant>,
     adaptive_recover_until: &mut HashMap<(i32, IpFamily), Instant>,
 ) -> bool {
-    if endpoints.len() != 1 || pool.floor_mode() != MeFloorMode::Adaptive {
+    if pool.floor_mode() != MeFloorMode::Adaptive {
         adaptive_idle_since.remove(&key);
         adaptive_recover_until.remove(&key);
         return false;
     }
 
     let now = Instant::now();
-    let endpoint = endpoints[0];
-    let writer_ids = live_writer_ids_by_addr
-        .get(&endpoint)
-        .map(Vec::as_slice)
-        .unwrap_or(&[]);
-    let has_bound_clients = has_bound_clients_on_endpoint(pool, writer_ids).await;
+    let writer_ids = list_writer_ids_for_endpoints(endpoints, live_writer_ids_by_addr);
+    let has_bound_clients = has_bound_clients_on_endpoint(pool, &writer_ids).await;
     if has_bound_clients {
         adaptive_idle_since.remove(&key);
         adaptive_recover_until.insert(key, now + pool.adaptive_floor_recover_grace_duration());

src/transport/middle_proxy/pool.rs

@@ -111,7 +111,17 @@ pub struct MePool {
     pub(super) me_floor_mode: AtomicU8,
     pub(super) me_adaptive_floor_idle_secs: AtomicU64,
     pub(super) me_adaptive_floor_min_writers_single_endpoint: AtomicU8,
+    pub(super) me_adaptive_floor_min_writers_multi_endpoint: AtomicU8,
     pub(super) me_adaptive_floor_recover_grace_secs: AtomicU64,
+    pub(super) me_adaptive_floor_writers_per_core_total: AtomicU32,
+    pub(super) me_adaptive_floor_cpu_cores_override: AtomicU32,
+    pub(super) me_adaptive_floor_max_extra_writers_single_per_core: AtomicU32,
+    pub(super) me_adaptive_floor_max_extra_writers_multi_per_core: AtomicU32,
+    pub(super) me_adaptive_floor_cpu_cores_detected: AtomicU32,
+    pub(super) me_adaptive_floor_cpu_cores_effective: AtomicU32,
+    pub(super) me_adaptive_floor_global_cap_raw: AtomicU64,
+    pub(super) me_adaptive_floor_global_cap_effective: AtomicU64,
+    pub(super) me_adaptive_floor_target_writers_total: AtomicU64,
     pub(super) proxy_map_v4: Arc<RwLock<HashMap<i32, Vec<(IpAddr, u16)>>>>,
     pub(super) proxy_map_v6: Arc<RwLock<HashMap<i32, Vec<(IpAddr, u16)>>>>,
     pub(super) default_dc: AtomicI32,
@@ -217,7 +227,12 @@ impl MePool {
         me_floor_mode: MeFloorMode,
         me_adaptive_floor_idle_secs: u64,
         me_adaptive_floor_min_writers_single_endpoint: u8,
+        me_adaptive_floor_min_writers_multi_endpoint: u8,
         me_adaptive_floor_recover_grace_secs: u64,
+        me_adaptive_floor_writers_per_core_total: u16,
+        me_adaptive_floor_cpu_cores_override: u16,
+        me_adaptive_floor_max_extra_writers_single_per_core: u16,
+        me_adaptive_floor_max_extra_writers_multi_per_core: u16,
         hardswap: bool,
         me_pool_drain_ttl_secs: u64,
         me_pool_force_close_secs: u64,
@@ -314,9 +329,29 @@ impl MePool {
             me_adaptive_floor_min_writers_single_endpoint: AtomicU8::new(
                 me_adaptive_floor_min_writers_single_endpoint,
             ),
+            me_adaptive_floor_min_writers_multi_endpoint: AtomicU8::new(
+                me_adaptive_floor_min_writers_multi_endpoint,
+            ),
             me_adaptive_floor_recover_grace_secs: AtomicU64::new(
                 me_adaptive_floor_recover_grace_secs,
             ),
+            me_adaptive_floor_writers_per_core_total: AtomicU32::new(
+                me_adaptive_floor_writers_per_core_total as u32,
+            ),
+            me_adaptive_floor_cpu_cores_override: AtomicU32::new(
+                me_adaptive_floor_cpu_cores_override as u32,
+            ),
+            me_adaptive_floor_max_extra_writers_single_per_core: AtomicU32::new(
+                me_adaptive_floor_max_extra_writers_single_per_core as u32,
+            ),
+            me_adaptive_floor_max_extra_writers_multi_per_core: AtomicU32::new(
+                me_adaptive_floor_max_extra_writers_multi_per_core as u32,
+            ),
+            me_adaptive_floor_cpu_cores_detected: AtomicU32::new(1),
+            me_adaptive_floor_cpu_cores_effective: AtomicU32::new(1),
+            me_adaptive_floor_global_cap_raw: AtomicU64::new(0),
+            me_adaptive_floor_global_cap_effective: AtomicU64::new(0),
+            me_adaptive_floor_target_writers_total: AtomicU64::new(0),
             pool_size: 2,
             proxy_map_v4: Arc::new(RwLock::new(proxy_map_v4)),
             proxy_map_v6: Arc::new(RwLock::new(proxy_map_v6)),
@@ -399,7 +434,12 @@ impl MePool {
         floor_mode: MeFloorMode,
         adaptive_floor_idle_secs: u64,
         adaptive_floor_min_writers_single_endpoint: u8,
+        adaptive_floor_min_writers_multi_endpoint: u8,
         adaptive_floor_recover_grace_secs: u64,
+        adaptive_floor_writers_per_core_total: u16,
+        adaptive_floor_cpu_cores_override: u16,
+        adaptive_floor_max_extra_writers_single_per_core: u16,
+        adaptive_floor_max_extra_writers_multi_per_core: u16,
     ) {
         self.hardswap.store(hardswap, Ordering::Relaxed);
         self.me_pool_drain_ttl_secs
@@ -443,8 +483,24 @@ impl MePool {
             .store(adaptive_floor_idle_secs, Ordering::Relaxed);
         self.me_adaptive_floor_min_writers_single_endpoint
             .store(adaptive_floor_min_writers_single_endpoint, Ordering::Relaxed);
+        self.me_adaptive_floor_min_writers_multi_endpoint
+            .store(adaptive_floor_min_writers_multi_endpoint, Ordering::Relaxed);
         self.me_adaptive_floor_recover_grace_secs
             .store(adaptive_floor_recover_grace_secs, Ordering::Relaxed);
+        self.me_adaptive_floor_writers_per_core_total
+            .store(adaptive_floor_writers_per_core_total as u32, Ordering::Relaxed);
+        self.me_adaptive_floor_cpu_cores_override
+            .store(adaptive_floor_cpu_cores_override as u32, Ordering::Relaxed);
+        self.me_adaptive_floor_max_extra_writers_single_per_core
+            .store(
+                adaptive_floor_max_extra_writers_single_per_core as u32,
+                Ordering::Relaxed,
+            );
+        self.me_adaptive_floor_max_extra_writers_multi_per_core
+            .store(
+                adaptive_floor_max_extra_writers_multi_per_core as u32,
+                Ordering::Relaxed,
+            );
         if previous_floor_mode != floor_mode {
             self.stats.increment_me_floor_mode_switch_total();
             match (previous_floor_mode, floor_mode) {
@@ -515,6 +571,13 @@ impl MePool {
         self.proxy_secret.read().await.key_selector
     }
 
+    pub(super) async fn active_writer_count_total(&self) -> usize {
+        let ws = self.writers.read().await;
+        ws.iter()
+            .filter(|w| !w.draining.load(Ordering::Relaxed))
+            .count()
+    }
+
     pub(super) async fn secret_snapshot(&self) -> SecretSnapshot {
         self.proxy_secret.read().await.clone()
     }
@@ -551,6 +614,82 @@ impl MePool {
         )
     }
 
+    pub(super) fn adaptive_floor_min_writers_multi_endpoint(&self) -> usize {
+        (self
+            .me_adaptive_floor_min_writers_multi_endpoint
+            .load(Ordering::Relaxed) as usize)
+            .max(1)
+    }
+
+    pub(super) fn adaptive_floor_writers_per_core_total(&self) -> usize {
+        (self
+            .me_adaptive_floor_writers_per_core_total
+            .load(Ordering::Relaxed) as usize)
+            .max(1)
+    }
+
+    pub(super) fn adaptive_floor_max_extra_single_per_core(&self) -> usize {
+        self.me_adaptive_floor_max_extra_writers_single_per_core
+            .load(Ordering::Relaxed) as usize
+    }
+
+    pub(super) fn adaptive_floor_max_extra_multi_per_core(&self) -> usize {
+        self.me_adaptive_floor_max_extra_writers_multi_per_core
+            .load(Ordering::Relaxed) as usize
+    }
+
+    pub(super) fn adaptive_floor_detected_cpu_cores(&self) -> usize {
+        std::thread::available_parallelism()
+            .map(|value| value.get())
+            .unwrap_or(1)
+            .max(1)
+    }
+
+    pub(super) fn adaptive_floor_effective_cpu_cores(&self) -> usize {
+        let detected = self.adaptive_floor_detected_cpu_cores();
+        let override_cores = self
+            .me_adaptive_floor_cpu_cores_override
+            .load(Ordering::Relaxed) as usize;
+        let effective = if override_cores == 0 {
+            detected
+        } else {
+            override_cores.max(1)
+        };
+        self.me_adaptive_floor_cpu_cores_detected
+            .store(detected as u32, Ordering::Relaxed);
+        self.me_adaptive_floor_cpu_cores_effective
+            .store(effective as u32, Ordering::Relaxed);
+        self.stats
+            .set_me_floor_cpu_cores_detected_gauge(detected as u64);
+        self.stats
+            .set_me_floor_cpu_cores_effective_gauge(effective as u64);
+        effective
+    }
+
+    pub(super) fn adaptive_floor_global_cap_raw(&self) -> usize {
+        let cores = self.adaptive_floor_effective_cpu_cores();
+        let cap = cores.saturating_mul(self.adaptive_floor_writers_per_core_total());
+        self.me_adaptive_floor_global_cap_raw
+            .store(cap as u64, Ordering::Relaxed);
+        self.stats.set_me_floor_global_cap_raw_gauge(cap as u64);
+        cap
+    }
+
+    pub(super) fn set_adaptive_floor_runtime_caps(
+        &self,
+        global_cap_effective: usize,
+        target_writers_total: usize,
+    ) {
+        self.me_adaptive_floor_global_cap_effective
+            .store(global_cap_effective as u64, Ordering::Relaxed);
+        self.me_adaptive_floor_target_writers_total
+            .store(target_writers_total as u64, Ordering::Relaxed);
+        self.stats
+            .set_me_floor_global_cap_effective_gauge(global_cap_effective as u64);
+        self.stats
+            .set_me_floor_target_writers_total_gauge(target_writers_total as u64);
+    }
+
     pub(super) fn required_writers_for_dc_with_floor_mode(
         &self,
         endpoint_count: usize,
@@ -560,13 +699,20 @@ impl MePool {
         if !reduce_for_idle {
             return base_required;
         }
-        if endpoint_count != 1 || self.floor_mode() != MeFloorMode::Adaptive {
+        if self.floor_mode() != MeFloorMode::Adaptive {
             return base_required;
         }
-        let min_writers = (self
-            .me_adaptive_floor_min_writers_single_endpoint
-            .load(Ordering::Relaxed) as usize)
-            .max(1);
+        let min_writers = if endpoint_count == 1 {
+            (self
+                .me_adaptive_floor_min_writers_single_endpoint
+                .load(Ordering::Relaxed) as usize)
+                .max(1)
+        } else {
+            (self
+                .me_adaptive_floor_min_writers_multi_endpoint
+                .load(Ordering::Relaxed) as usize)
+                .max(1)
+        };
         base_required.min(min_writers)
     }
 

src/transport/middle_proxy/pool_status.rs

@@ -28,6 +28,10 @@ pub(crate) struct MeApiDcStatusSnapshot {
     pub available_endpoints: usize,
     pub available_pct: f64,
     pub required_writers: usize,
+    pub floor_min: usize,
+    pub floor_target: usize,
+    pub floor_max: usize,
+    pub floor_capped: bool,
     pub alive_writers: usize,
     pub coverage_pct: f64,
     pub rtt_ms: Option<f64>,
@@ -72,7 +76,17 @@ pub(crate) struct MeApiRuntimeSnapshot {
     pub floor_mode: &'static str,
     pub adaptive_floor_idle_secs: u64,
     pub adaptive_floor_min_writers_single_endpoint: u8,
+    pub adaptive_floor_min_writers_multi_endpoint: u8,
     pub adaptive_floor_recover_grace_secs: u64,
+    pub adaptive_floor_writers_per_core_total: u16,
+    pub adaptive_floor_cpu_cores_override: u16,
+    pub adaptive_floor_max_extra_writers_single_per_core: u16,
+    pub adaptive_floor_max_extra_writers_multi_per_core: u16,
+    pub adaptive_floor_cpu_cores_detected: u32,
+    pub adaptive_floor_cpu_cores_effective: u32,
+    pub adaptive_floor_global_cap_raw: u64,
+    pub adaptive_floor_global_cap_effective: u64,
+    pub adaptive_floor_target_writers_total: u64,
     pub me_keepalive_enabled: bool,
     pub me_keepalive_interval_secs: u64,
     pub me_keepalive_jitter_secs: u64,
@@ -275,14 +289,43 @@ impl MePool {
         let mut dcs = Vec::<MeApiDcStatusSnapshot>::with_capacity(endpoints_by_dc.len());
         let mut available_endpoints = 0usize;
         let mut alive_writers = 0usize;
+        let floor_mode = self.floor_mode();
+        let adaptive_cpu_cores = (self
+            .me_adaptive_floor_cpu_cores_effective
+            .load(Ordering::Relaxed) as usize)
+            .max(1);
         for (dc, endpoints) in endpoints_by_dc {
             let endpoint_count = endpoints.len();
             let dc_available_endpoints = endpoints
                 .iter()
                 .filter(|endpoint| live_writers_by_endpoint.contains_key(endpoint))
                 .count();
+            let base_required = self.required_writers_for_dc(endpoint_count);
             let dc_required_writers =
                 self.required_writers_for_dc_with_floor_mode(endpoint_count, false);
+            let floor_min = if endpoint_count <= 1 {
+                (self
+                    .me_adaptive_floor_min_writers_single_endpoint
+                    .load(Ordering::Relaxed) as usize)
+                    .max(1)
+                    .min(base_required.max(1))
+            } else {
+                (self
+                    .me_adaptive_floor_min_writers_multi_endpoint
+                    .load(Ordering::Relaxed) as usize)
+                    .max(1)
+                    .min(base_required.max(1))
+            };
+            let extra_per_core = if endpoint_count <= 1 {
+                self.me_adaptive_floor_max_extra_writers_single_per_core
+                    .load(Ordering::Relaxed) as usize
+            } else {
+                self.me_adaptive_floor_max_extra_writers_multi_per_core
+                    .load(Ordering::Relaxed) as usize
+            };
+            let floor_max = base_required.saturating_add(adaptive_cpu_cores.saturating_mul(extra_per_core));
+            let floor_capped = matches!(floor_mode, MeFloorMode::Adaptive)
+                && dc_required_writers < base_required;
             let dc_alive_writers = live_writers_by_dc.get(&dc).copied().unwrap_or(0);
             let dc_load = activity
                 .active_sessions_by_target_dc
@@ -302,6 +345,10 @@ impl MePool {
                 available_endpoints: dc_available_endpoints,
                 available_pct: ratio_pct(dc_available_endpoints, endpoint_count),
                 required_writers: dc_required_writers,
+                floor_min,
+                floor_target: dc_required_writers,
+                floor_max,
+                floor_capped,
                 alive_writers: dc_alive_writers,
                 coverage_pct: ratio_pct(dc_alive_writers, dc_required_writers),
                 rtt_ms: dc_rtt_ms,
@@ -378,9 +425,39 @@ impl MePool {
             adaptive_floor_min_writers_single_endpoint: self
                 .me_adaptive_floor_min_writers_single_endpoint
                 .load(Ordering::Relaxed),
+            adaptive_floor_min_writers_multi_endpoint: self
+                .me_adaptive_floor_min_writers_multi_endpoint
+                .load(Ordering::Relaxed),
             adaptive_floor_recover_grace_secs: self
                 .me_adaptive_floor_recover_grace_secs
                 .load(Ordering::Relaxed),
+            adaptive_floor_writers_per_core_total: self
+                .me_adaptive_floor_writers_per_core_total
+                .load(Ordering::Relaxed) as u16,
+            adaptive_floor_cpu_cores_override: self
+                .me_adaptive_floor_cpu_cores_override
+                .load(Ordering::Relaxed) as u16,
+            adaptive_floor_max_extra_writers_single_per_core: self
+                .me_adaptive_floor_max_extra_writers_single_per_core
+                .load(Ordering::Relaxed) as u16,
+            adaptive_floor_max_extra_writers_multi_per_core: self
+                .me_adaptive_floor_max_extra_writers_multi_per_core
+                .load(Ordering::Relaxed) as u16,
+            adaptive_floor_cpu_cores_detected: self
+                .me_adaptive_floor_cpu_cores_detected
+                .load(Ordering::Relaxed),
+            adaptive_floor_cpu_cores_effective: self
+                .me_adaptive_floor_cpu_cores_effective
+                .load(Ordering::Relaxed),
+            adaptive_floor_global_cap_raw: self
+                .me_adaptive_floor_global_cap_raw
+                .load(Ordering::Relaxed),
+            adaptive_floor_global_cap_effective: self
+                .me_adaptive_floor_global_cap_effective
+                .load(Ordering::Relaxed),
+            adaptive_floor_target_writers_total: self
+                .me_adaptive_floor_target_writers_total
+                .load(Ordering::Relaxed),
             me_keepalive_enabled: self.me_keepalive_enabled,
             me_keepalive_interval_secs: self.me_keepalive_interval.as_secs(),
             me_keepalive_jitter_secs: self.me_keepalive_jitter.as_secs(),

tools/aesdiag.py

@@ -0,0 +1,403 @@
+#!/usr/bin/env python3
+"""
+AES-CBC validation tool for telemt middle proxy logs with support for noop padding.
+
+Parses log lines containing:
+  - "ME diag: derived keys and handshake plaintext" (provides write_key, write_iv, hs_plain)
+  - "ME diag: handshake ciphertext" (provides hs_cipher)
+
+For each pair it:
+  - Decrypts the ciphertext using the provided key and IV.
+  - Compares the beginning of the decrypted data with hs_plain.
+  - Attempts to identify the actual padding scheme (PKCS#7, zero padding, noop padding).
+  - Re-encrypts with different paddings and reports mismatches block by block.
+  - Accumulates statistics for final summary.
+"""
+
+import sys
+import re
+from collections import defaultdict
+from Crypto.Cipher import AES
+
+# Constants
+NOOP_FRAME = bytes([0x04, 0x00, 0x00, 0x00])   # noop frame used for padding
+
+def hex_str_to_bytes(hex_str):
+    """Convert a hex string like 'aa bb cc' to bytes."""
+    return bytes.fromhex(hex_str.replace(' ', ''))
+
+def parse_params(line):
+    """Extract key=value pairs where value is a space-separated hex string."""
+    pattern = r'(\w+)=((?:[0-9a-f]{2} )*[0-9a-f]{2})'
+    return {key: val for key, val in re.findall(pattern, line)}
+
+def pkcs7_pad(data, block_size=16):
+    """Apply PKCS#7 padding to the given data."""
+    pad_len = block_size - (len(data) % block_size)
+    if pad_len == 0:
+        pad_len = block_size
+    return data + bytes([pad_len]) * pad_len
+
+def zero_pad(data, block_size=16):
+    """Pad with zeros to the next block boundary."""
+    pad_len = block_size - (len(data) % block_size)
+    if pad_len == block_size:
+        return data  # already full blocks, no zero padding needed
+    return data + bytes(pad_len)
+
+def noop_pad(data):
+    """
+    Pad with minimal number of noop frames (b'\\x04\\x00\\x00\\x00')
+    to reach a multiple of 16 bytes.
+    """
+    block_size = 16
+    frame_len = len(NOOP_FRAME)  # 4
+    remainder = len(data) % block_size
+    if remainder == 0:
+        return data  # no padding needed
+    # We need to add k frames such that (len(data) + k*frame_len) % block_size == 0
+    # => k*frame_len ≡ -remainder (mod block_size)
+    # Since frame_len=4 and block_size=16, we need k*4 ≡ (16-remainder) mod 16
+    # k must be an integer in {1,2,3} (because 4*4=16 ≡0 mod16, so k=4 gives remainder 0, but then total increase=16,
+    # but if remainder==0 we already handled; if remainder!=0, k=4 gives (len+16)%16 == remainder, not 0,
+    # so k=4 doesn't solve unless remainder=0. Actually 4*4=16 ≡0, so k=4 gives (len+16)%16 = remainder, so still not 0.
+    # The equation is k*4 ≡ (16-remainder) mod 16. Let r=16-remainder (1..15). Then k ≡ r*inv(4) mod 4? Since mod 16,
+    # 4 has no inverse modulo 16 because gcd(4,16)=4. So solutions exist only if r is multiple of 4.
+    # Therefore remainder must be 4,8,12 (so that r = 12,8,4). This matches the idea that noop padding is only added
+    # when the plaintext length mod 16 is 4,8,12. In our logs it's always 44 mod16=12, so r=4, so k=1 works.
+    # For safety, we compute k as (block_size - remainder) // frame_len, but this only works if that value is integer.
+    need = block_size - remainder
+    if need % frame_len != 0:
+        # This shouldn't happen by protocol, but if it does, fall back to adding full blocks of noop until multiple.
+        # We'll add ceil(need/frame_len) frames.
+        k = (need + frame_len - 1) // frame_len
+    else:
+        k = need // frame_len
+    return data + NOOP_FRAME * k
+
+def unpad_pkcs7(data):
+    """Remove PKCS#7 padding (assumes correct padding)."""
+    if not data:
+        return data
+    pad_len = data[-1]
+    if pad_len < 1 or pad_len > 16:
+        return data  # not valid PKCS#7, return as is
+    # Check that all padding bytes are equal to pad_len
+    if all(b == pad_len for b in data[-pad_len:]):
+        return data[:-pad_len]
+    return data
+
+def is_noop_padded(decrypted, plain_log):
+    """
+    Check if the extra bytes after plain_log in decrypted consist of one or more NOOP_FRAMEs.
+    Returns True if they do, False otherwise.
+    """
+    extra = decrypted[len(plain_log):]
+    if len(extra) == 0:
+        return False
+    # Split into chunks of 4
+    if len(extra) % 4 != 0:
+        return False
+    for i in range(0, len(extra), 4):
+        if extra[i:i+4] != NOOP_FRAME:
+            return False
+    return True
+
+def main():
+    derived_list = []   # entries from "derived keys and handshake plaintext"
+    cipher_list = []    # entries from "handshake ciphertext"
+
+    for line in sys.stdin:
+        if 'ME diag: derived keys and handshake plaintext' in line:
+            params = parse_params(line)
+            if all(k in params for k in ('write_key', 'write_iv', 'hs_plain')):
+                derived_list.append(params)
+        elif 'ME diag: handshake ciphertext' in line:
+            params = parse_params(line)
+            if 'hs_cipher' in params:
+                cipher_list.append(params)
+
+    # Warn about count mismatch but process as many pairs as possible
+    n_pairs = min(len(derived_list), len(cipher_list))
+    if len(derived_list) != len(cipher_list):
+        print(f"\n[WARN] Number of derived entries ({len(derived_list)}) "
+              f"differs from cipher entries ({len(cipher_list)}). "
+              f"Processing first {n_pairs} pairs.\n")
+
+    # Statistics accumulators
+    stats = {
+        'total': n_pairs,
+        'key_length_ok': 0,
+        'iv_length_ok': 0,
+        'cipher_aligned': 0,
+        'decryption_match_start': 0,      # first bytes equal hs_plain
+        'pkcs7_after_unpad_matches': 0,   # after removing PKCS7, equals hs_plain
+        'extra_bytes_all_zero': 0,         # extra bytes after hs_plain are zero
+        'extra_bytes_noop': 0,             # extra bytes are noop frames
+        'pkcs7_encrypt_ok': 0,             # re-encryption with PKCS7 matches ciphertext
+        'zero_encrypt_ok': 0,               # re-encryption with zero padding matches
+        'noop_encrypt_ok': 0,                # re-encryption with noop padding matches
+        'no_padding_encrypt_ok': 0,         # only if plaintext multiple of 16 and matches
+        'no_padding_applicable': 0,         # number of tests where plaintext len %16 ==0
+    }
+
+    detailed_results = []  # store per-test summary for final heuristic
+
+    for idx, (der, ciph) in enumerate(zip(derived_list[:n_pairs], cipher_list[:n_pairs]), 1):
+        print(f"\n{'='*60}")
+        print(f"Test #{idx}")
+        print(f"{'='*60}")
+
+        # Local stats for this test
+        test_stats = defaultdict(bool)
+
+        try:
+            key = hex_str_to_bytes(der['write_key'])
+            iv = hex_str_to_bytes(der['write_iv'])
+            plain_log = hex_str_to_bytes(der['hs_plain'])
+            ciphertext = hex_str_to_bytes(ciph['hs_cipher'])
+
+            # Basic sanity checks
+            print(f"[INFO] Key length       : {len(key)} bytes (expected 32)")
+            print(f"[INFO] IV length        : {len(iv)} bytes (expected 16)")
+            print(f"[INFO] hs_plain length  : {len(plain_log)} bytes")
+            print(f"[INFO] hs_cipher length : {len(ciphertext)} bytes")
+
+            if len(key) == 32:
+                stats['key_length_ok'] += 1
+                test_stats['key_ok'] = True
+            else:
+                print("[WARN] Key length is not 32 bytes – AES-256 requires 32-byte key.")
+
+            if len(iv) == 16:
+                stats['iv_length_ok'] += 1
+                test_stats['iv_ok'] = True
+            else:
+                print("[WARN] IV length is not 16 bytes – AES-CBC requires 16-byte IV.")
+
+            if len(ciphertext) % 16 == 0:
+                stats['cipher_aligned'] += 1
+                test_stats['cipher_aligned'] = True
+            else:
+                print("[ERROR] Ciphertext length is not a multiple of 16 – invalid AES-CBC block alignment.")
+                # Skip further processing for this test
+                detailed_results.append(test_stats)
+                continue
+
+            # --- Decryption test ---
+            cipher_dec = AES.new(key, AES.MODE_CBC, iv)
+            decrypted = cipher_dec.decrypt(ciphertext)
+            print(f"[INFO] Decrypted ({len(decrypted)} bytes): {decrypted.hex()}")
+
+            # Compare beginning with hs_plain
+            match_len = min(len(plain_log), len(decrypted))
+            if decrypted[:match_len] == plain_log[:match_len]:
+                print(f"[OK] First {match_len} bytes match hs_plain.")
+                stats['decryption_match_start'] += 1
+                test_stats['decrypt_start_ok'] = True
+            else:
+                print(f"[FAIL] First bytes do NOT match hs_plain.")
+                for i in range(match_len):
+                    if decrypted[i] != plain_log[i]:
+                        print(f"       First mismatch at byte {i}: hs_plain={plain_log[i]:02x}, decrypted={decrypted[i]:02x}")
+                        break
+                test_stats['decrypt_start_ok'] = False
+
+            # --- Try to identify actual padding ---
+            # Remove possible PKCS#7 padding from decrypted data
+            decrypted_unpadded = unpad_pkcs7(decrypted)
+            if decrypted_unpadded != decrypted:
+                print(f"[INFO] After removing PKCS#7 padding: {len(decrypted_unpadded)} bytes left.")
+                if decrypted_unpadded == plain_log:
+                    print("[OK] Decrypted data with PKCS#7 removed exactly matches hs_plain.")
+                    stats['pkcs7_after_unpad_matches'] += 1
+                    test_stats['pkcs7_unpad_matches'] = True
+                else:
+                    print("[INFO] Decrypted (PKCS#7 removed) does NOT match hs_plain.")
+                    test_stats['pkcs7_unpad_matches'] = False
+            else:
+                print("[INFO] No valid PKCS#7 padding detected in decrypted data.")
+                test_stats['pkcs7_unpad_matches'] = False
+
+            # Check if the extra bytes after hs_plain in decrypted are all zero (zero padding)
+            extra = decrypted[len(plain_log):]
+            if extra and all(b == 0 for b in extra):
+                print("[INFO] Extra bytes after hs_plain are all zeros – likely zero padding.")
+                stats['extra_bytes_all_zero'] += 1
+                test_stats['extra_zero'] = True
+            else:
+                test_stats['extra_zero'] = False
+
+            # Check for noop padding in extra bytes
+            if is_noop_padded(decrypted, plain_log):
+                print(f"[OK] Extra bytes after hs_plain consist of noop frames ({NOOP_FRAME.hex()}).")
+                stats['extra_bytes_noop'] += 1
+                test_stats['extra_noop'] = True
+            else:
+                test_stats['extra_noop'] = False
+                if extra:
+                    print(f"[INFO] Extra bytes after hs_plain (hex): {extra.hex()}")
+
+            # --- Re-encryption tests ---
+            # PKCS#7
+            padded_pkcs7 = pkcs7_pad(plain_log)
+            cipher_enc = AES.new(key, AES.MODE_CBC, iv)
+            computed_pkcs7 = cipher_enc.encrypt(padded_pkcs7)
+            if computed_pkcs7 == ciphertext:
+                print("[OK] PKCS#7 padding produces the expected ciphertext.")
+                stats['pkcs7_encrypt_ok'] += 1
+                test_stats['pkcs7_enc_ok'] = True
+            else:
+                print("[FAIL] PKCS#7 padding does NOT match the ciphertext.")
+                test_stats['pkcs7_enc_ok'] = False
+                # Show block where first difference occurs
+                block_size = 16
+                for blk in range(len(ciphertext)//block_size):
+                    start = blk*block_size
+                    exp = ciphertext[start:start+block_size]
+                    comp = computed_pkcs7[start:start+block_size]
+                    if exp != comp:
+                        print(f"       First difference in block {blk}:")
+                        print(f"         expected : {exp.hex()}")
+                        print(f"         computed : {comp.hex()}")
+                        break
+
+            # Zero padding
+            padded_zero = zero_pad(plain_log)
+            # Ensure multiple of 16
+            if len(padded_zero) % 16 != 0:
+                padded_zero += bytes(16 - (len(padded_zero)%16))
+            cipher_enc_zero = AES.new(key, AES.MODE_CBC, iv)
+            computed_zero = cipher_enc_zero.encrypt(padded_zero)
+            if computed_zero == ciphertext:
+                print("[OK] Zero padding produces the expected ciphertext.")
+                stats['zero_encrypt_ok'] += 1
+                test_stats['zero_enc_ok'] = True
+            else:
+                print("[INFO] Zero padding does NOT match (expected, unless log used PKCS#7).")
+                test_stats['zero_enc_ok'] = False
+
+            # Noop padding
+            padded_noop = noop_pad(plain_log)
+            # Ensure multiple of 16 (noop_pad already returns multiple of 16)
+            cipher_enc_noop = AES.new(key, AES.MODE_CBC, iv)
+            computed_noop = cipher_enc_noop.encrypt(padded_noop)
+            if computed_noop == ciphertext:
+                print("[OK] Noop padding produces the expected ciphertext.")
+                stats['noop_encrypt_ok'] += 1
+                test_stats['noop_enc_ok'] = True
+            else:
+                print("[FAIL] Noop padding does NOT match the ciphertext.")
+                test_stats['noop_enc_ok'] = False
+                # Show block difference if needed
+                for blk in range(len(ciphertext)//16):
+                    start = blk*16
+                    if computed_noop[start:start+16] != ciphertext[start:start+16]:
+                        print(f"       First difference in block {blk}:")
+                        print(f"         expected : {ciphertext[start:start+16].hex()}")
+                        print(f"         computed : {computed_noop[start:start+16].hex()}")
+                        break
+
+            # No padding (only possible if plaintext is already multiple of 16)
+            if len(plain_log) % 16 == 0:
+                stats['no_padding_applicable'] += 1
+                cipher_enc_nopad = AES.new(key, AES.MODE_CBC, iv)
+                computed_nopad = cipher_enc_nopad.encrypt(plain_log)
+                if computed_nopad == ciphertext:
+                    print("[OK] No padding (plaintext multiple of 16) matches.")
+                    stats['no_padding_encrypt_ok'] += 1
+                    test_stats['no_pad_enc_ok'] = True
+                else:
+                    print("[INFO] No padding does NOT match.")
+                    test_stats['no_pad_enc_ok'] = False
+            else:
+                print("[INFO] Skipping no‑padding test because plaintext length is not a multiple of 16.")
+
+        except Exception as e:
+            print(f"[EXCEPTION] {e}")
+            test_stats['exception'] = True
+
+        detailed_results.append(test_stats)
+
+    # --- Final statistics and heuristic summary ---
+    print("\n" + "="*60)
+    print("STATISTICS SUMMARY")
+    print("="*60)
+    print(f"Total tests processed          : {stats['total']}")
+    print(f"Key length OK (32)              : {stats['key_length_ok']}/{stats['total']}")
+    print(f"IV length OK (16)                : {stats['iv_length_ok']}/{stats['total']}")
+    print(f"Ciphertext 16-byte aligned       : {stats['cipher_aligned']}/{stats['total']}")
+    print(f"Decryption starts with hs_plain  : {stats['decryption_match_start']}/{stats['total']}")
+    print(f"After PKCS#7 removal matches     : {stats['pkcs7_after_unpad_matches']}/{stats['total']}")
+    print(f"Extra bytes after hs_plain are 0 : {stats['extra_bytes_all_zero']}/{stats['total']}")
+    print(f"Extra bytes are noop frames       : {stats['extra_bytes_noop']}/{stats['total']}")
+    print(f"PKCS#7 re-encryption OK           : {stats['pkcs7_encrypt_ok']}/{stats['total']}")
+    print(f"Zero padding re-encryption OK     : {stats['zero_encrypt_ok']}/{stats['total']}")
+    print(f"Noop padding re-encryption OK     : {stats['noop_encrypt_ok']}/{stats['total']}")
+    if stats['no_padding_applicable'] > 0:
+        print(f"No-padding applicable tests      : {stats['no_padding_applicable']}")
+        print(f"No-padding re-encryption OK       : {stats['no_padding_encrypt_ok']}/{stats['no_padding_applicable']}")
+
+    # Heuristic: determine most likely padding
+    print("\n" + "="*60)
+    print("HEURISTIC CONCLUSION")
+    print("="*60)
+
+    if stats['decryption_match_start'] == stats['total']:
+        print("✓ All tests: first bytes of decrypted data match hs_plain → keys and IV are correct.")
+    else:
+        print("✗ Some tests: first bytes mismatch → possible key/IV issues or corrupted ciphertext.")
+
+    # Guess padding based on re-encryption success and extra bytes
+    candidates = []
+    if stats['pkcs7_encrypt_ok'] == stats['total']:
+        candidates.append("PKCS#7")
+    if stats['zero_encrypt_ok'] == stats['total']:
+        candidates.append("zero padding")
+    if stats['noop_encrypt_ok'] == stats['total']:
+        candidates.append("noop padding")
+    if stats['no_padding_applicable'] == stats['total'] and stats['no_padding_encrypt_ok'] == stats['total']:
+        candidates.append("no padding")
+
+    if len(candidates) == 1:
+        print(f"✓ All tests consistent with padding scheme: {candidates[0]}.")
+    elif len(candidates) > 1:
+        print(f"⚠ Multiple padding schemes succeed in all tests: {', '.join(candidates)}. This is unusual.")
+    else:
+        # No scheme succeeded in all tests – look at ratios
+        print("Mixed padding results:")
+        total = stats['total']
+        pkcs7_ratio = stats['pkcs7_encrypt_ok'] / total if total else 0
+        zero_ratio = stats['zero_encrypt_ok'] / total if total else 0
+        noop_ratio = stats['noop_encrypt_ok'] / total if total else 0
+        print(f"  PKCS#7 success = {stats['pkcs7_encrypt_ok']}/{total} ({pkcs7_ratio*100:.1f}%)")
+        print(f"  Zero success   = {stats['zero_encrypt_ok']}/{total} ({zero_ratio*100:.1f}%)")
+        print(f"  Noop success   = {stats['noop_encrypt_ok']}/{total} ({noop_ratio*100:.1f}%)")
+
+        if noop_ratio > max(pkcs7_ratio, zero_ratio):
+            print("→ Noop padding is most frequent. Check if extra bytes are indeed noop frames.")
+        elif pkcs7_ratio > zero_ratio:
+            print("→ PKCS#7 is most frequent, but fails in some tests.")
+        elif zero_ratio > pkcs7_ratio:
+            print("→ Zero padding is most frequent, but fails in some tests.")
+        else:
+            print("→ No clear winner; possibly a different padding scheme or random data.")
+
+    # Additional heuristics based on extra bytes
+    if stats['extra_bytes_noop'] == stats['total']:
+        print("✓ All tests: extra bytes after hs_plain are noop frames → strongly indicates noop padding.")
+    if stats['extra_bytes_all_zero'] == stats['total']:
+        print("✓ All tests: extra bytes are zeros → suggests zero padding.")
+
+    # Final health check
+    if (stats['decryption_match_start'] == stats['total'] and
+        (stats['pkcs7_encrypt_ok'] == stats['total'] or
+         stats['zero_encrypt_ok'] == stats['total'] or
+         stats['noop_encrypt_ok'] == stats['total'] or
+         stats['no_padding_encrypt_ok'] == stats['no_padding_applicable'] == stats['total'])):
+        print("\n✅ OVERALL: All tests consistent. The encryption parameters and padding are correct.")
+    else:
+        print("\n⚠️ OVERALL: Inconsistencies detected. Review the detailed output for failing tests.")
+
+if __name__ == '__main__':
+    main()