[fix] improve authentication recovery after server restarts

.env.example

@@ -43,5 +43,14 @@ ADMIN_USERNAME=
 ADMIN_PASSWORD=
 
 # Signaling capacity limits; effective connections also depend on available FDs.
-CROSSDESK_NOFILE=8192
-CROSSDESK_MAX_CONNECTIONS=2048
+CROSSDESK_NOFILE=32768
+CROSSDESK_MAX_CONNECTIONS=8192
+# Credential processing: workers execute hashes, the bounded queue only waits.
+# Limits include active jobs/results; per-source keys are TCP peer IPs.
+CROSSDESK_AUTH_WORKERS=2
+CROSSDESK_AUTH_QUEUE_CAPACITY=4096
+CROSSDESK_AUTH_SOURCE_CAPACITY=256
+CROSSDESK_AUTH_SOURCE_PER_MINUTE=60
+# Maximum accepted-request lifetime; no-login sockets still expire after 15s.
+# Set from measured hash throughput and desired restart recovery time.
+CROSSDESK_AUTH_TIMEOUT_SECONDS=600

README.md

@@ -254,9 +254,13 @@ sudo docker compose up -d --no-build
 
 新注册与改密使用 **Argon2id v19，19 MiB 内存、2 次迭代、并行度 1、32 字节摘要**。现有 `password_hash` 列保存 `argon2id$v=19$m=19456,t=2,p=1$` 前缀和十六进制摘要，`password_salt` 保存十六进制盐。旧版加盐 SHA-256 记录在密码验证成功后原子升级，设备 ID 和六位密码不变；长期未验证的旧记录继续按设备保留期限清理。哈希升级不能使旧密码或已泄漏备份失效，需要时应通过客户端更换密码。旧版服务端无法验证升级后的记录，回滚必须使用升级前备份。
 
-密码计算由两个独立工作线程执行，最多 34 个任务（包括等待提交的结果）；不会在计算时持有数据库或失败限流锁。每个连接保持消息顺序，完成后复查连接身份及凭据快照，过期结果不能覆盖并发改密或恢复已删除身份。未知设备、旧格式错误密码和损坏的记录也执行一次 Argon2id，减少耗时差异。启动时验证本机 OpenSSL provider 支持上述算法和参数。
+密码计算默认由两个独立工作线程执行，使用按来源 IP 轮转的有界等待队列。默认最多接纳 4096 个请求（包含执行中和等待提交的结果），同一 IP 最多占用 256 个；等待请求不占用计算线程。不会在计算时持有数据库或失败限流锁。每个连接保持消息顺序，完成后复查连接身份及凭据快照，过期结果不能覆盖并发改密或恢复已删除身份。未知设备、旧格式错误密码和损坏的记录也执行一次 Argon2id，减少耗时差异。启动时验证本机 OpenSSL provider 支持上述算法和参数。
 
-注册、登录、查询、加入会话及新改密计算共用计算额度：每个来源 IP 每分钟最多 60 次，整个进程每分钟最多 300 次，成功操作同样计数。达到额度返回 `Too many credential requests`；计算队列满时返回 `Credential service busy`，均附 `retry_after` 秒数。六位密码的强度仍有限，计算额度与下述失败冷却共同限制在线猜测；慢哈希主要提高数据库泄漏后的离线猜测成本。
+注册、登录、查询、加入会话及新改密计算共用来源额度：默认每个来源 IP 每分钟最多启动 60 次，成功操作同样计数；不再设置整个进程每分钟 300 次的配额。仅在获得空闲计算线程后扣除额度，排队、断线取消和队列满拒绝都不提前扣费。来源额度用尽时在服务器内等待下一窗口，不阻塞其他来源。队列满或排队到期仍返回原协议的 `Credential service busy` 和 `retry_after`，失败冷却规则保持不变。
+
+已接纳的认证请求默认最多等待／执行 600 秒，排队期间不触发原来的 15 秒未登录断开；从未提交有效认证请求的连接仍按 15 秒规则清理。旧客户端无需识别新协议或实现 `retry_after` 重试，即可收到排队后的原登录响应。600 秒是最长保留期限，并非固定等待时间；断线会取消尚未执行的请求。重启时恢复中的远控记录保留窗口随认证期限延长，避免客户端还在排队时记录就被清理。
+
+六位密码的强度仍有限，来源计算额度与下述失败冷却共同限制在线猜测；慢哈希主要提高数据库泄漏后的离线猜测成本。
 
 同一 WebSocket 连接只允许一次成功登录。后续 `login` 请求返回 `Already authenticated`，不会再注册设备、验证密码或更新在线状态；如需重新登录，请建立新连接。首次登录失败后仍可在原连接重试。
 
@@ -268,6 +272,26 @@ sudo docker compose up -d --no-build
 
 注册需显式使用空设备 ID（原生客户端）或 `web`（Web 客户端）。携带不存在或已过期的 ID 登录会返回统一认证失败，不再自动分配新 ID；客户端需通过重置本地身份重新发起注册。
 
+### 在线容量与重启恢复调优
+
+在 `.env` 配置下列选项，并重新创建信令容器：
+
+| 配置 | 默认值 | 作用与范围 |
+| --- | --- | --- |
+| `CROSSDESK_MAX_CONNECTIONS` | `8192` | 信令连接总上限，1–65536；设备、`C-*` 控制端及未完成连接都占用名额 |
+| `CROSSDESK_NOFILE` | `32768` | Compose 容器文件描述符上限；实际连接容量还会扣除已用与预留 FD |
+| `CROSSDESK_AUTH_WORKERS` | `2` | 同时执行密码计算的线程数，1–32；影响 CPU、内存和认证吞吐量 |
+| `CROSSDESK_AUTH_QUEUE_CAPACITY` | `4096` | 总未完成认证请求数，1–16384；扩大它能接住突发，不提高计算速度 |
+| `CROSSDESK_AUTH_SOURCE_CAPACITY` | `256` | 同一来源 IP 的未完成请求数，1–4096；防止单一来源占满队列 |
+| `CROSSDESK_AUTH_SOURCE_PER_MINUTE` | `60` | 单 IP 每分钟启动认证数，1–6000；同一 NAT／反向代理后的客户端共享 |
+| `CROSSDESK_AUTH_TIMEOUT_SECONDS` | `600` | 接纳后的最大认证期限，15–900 秒 |
+
+2000 台设备保持在线不需要 2000 个认证线程；只有登录、加入会话等需要校验密码的操作使用线程。先保留 2 个线程，在目标机器测量认证吞吐量再增加，避免挤占 TLS、心跳和数据库的 CPU。每个正在计算的 Argon2id 任务需要约 19 MiB 算法内存，等待队列没有这项开销。集中恢复时间约为请求数除以实测完成速率，另受单 IP 配额约束；队列期限要覆盖预计恢复时间。大量设备共用出口时，应同时评估来源队列、来源额度和恢复期限。
+
+现有 `.env` 的值会覆盖新默认值，例如保留 `CROSSDESK_MAX_CONNECTIONS=2048` 不会自动变成 8192。服务端请求的 TCP listen backlog 为 4096，实际值受操作系统上限约束；同时处理的未打开连接仍限制为 128。上述参数用于信令与重启恢复，不能代替目标机器上的持续在线、并发远控及 TURN 带宽压测。
+
+`Credential capacity` 记录实际生效的认证配置；`Credential diagnostics` 每分钟记录 `active`、`waiting`、`queue_full_total`、`source_full_total`、`expired_total`、`source_rate_waits_total` 及计算耗时。`Connection diagnostics` 同时记录登录成功、认证失败、繁忙响应、失败冷却响应和认证超时的累计次数。队列持续增长表示处理速度跟不上请求；`source_rate_waits_total` 增长表示来源额度等待，单独增加线程无法解除它。
+
 ### 数据保存与自动清理
 
 | 数据 | 默认规则 |
@@ -330,6 +354,18 @@ xmake b -vy crossdesk_server
 
 首次构建前需确保 `xmake --version` 可用。其他系统的源码配置见 [xmake.lua](xmake.lua)，当前发布流程只提供 Linux 二进制与容器镜像。切换 Debug 使用 `xmake f -m debug`；`xmake r -d crossdesk_server` 表示使用调试器运行。
 
+认证队列回归与旧协议 WSS 集成测试（需要 Python 3、`websockets>=15` 和 `openssl` 命令）：
+
+```bash
+xmake b credential_queue_test credential_server_fixture
+xmake r credential_queue_test
+# 按当前平台、架构和构建模式替换 fixture 路径。
+python3 tests/credential_server_integration.py \
+  --fixture build/linux/x86_64/release/credential_server_fixture --clients 2100
+```
+
+测试在临时目录创建独立证书与数据库，验证来源等待超过 15 秒、断线释放名额、队列到期，以及集中登录、停服和重启后的再次登录。所有登录都执行真实 Argon2id 校验，模拟客户端不重试登录；TCP/TLS 握手并发限制为 64，避免本机 listen 队列先溢出。集中登录用例提高回环 IP 的来源额度，来源限额另有独立用例；结果不代表默认配置允许同一 IP 下 2100 台设备同时接纳，也不代表生产机器性能。
+
 ### 构建自己的运行镜像
 
 Dockerfile **只打包已编译的 Linux 二进制**。先在与目标镜像架构相同的 Ubuntu 22.04 环境完成 Release 编译，再在仓库根目录执行：

README_EN.md

@@ -236,9 +236,13 @@ Device passwords remain **6 alphanumeric characters**, generated using OpenSSL `
 
 Registration and password changes use **Argon2id v19 with 19 MiB memory, 2 iterations, parallelism 1 and a 32-byte digest**. The existing `password_hash` column contains the `argon2id$v=19$m=19456,t=2,p=1$` prefix and a hex digest; `password_salt` contains a hex salt. Legacy salted SHA-256 records migrate atomically after successful verification without changing the ID or six-character password. Unused legacy records remain subject to device retention. Rehashing does not invalidate old passwords or leaked backups; rotate passwords through the client when necessary. Older servers cannot verify upgraded records; rollback requires a pre-upgrade backup.
 
-Two dedicated workers perform password calculations, with at most 34 jobs including results awaiting commit. Hashing holds neither the database nor the failure-limiter lock. Messages stay ordered per connection; completion rechecks the connection identity and credential snapshot, so stale work cannot overwrite concurrent password changes or restore deleted identities. Missing devices, wrong legacy passwords and malformed records also run one Argon2id calculation to reduce timing differences. Startup checks the actual OpenSSL provider and parameters.
+Two dedicated workers perform password calculations by default, with a bounded queue scheduled round-robin by source IP. The defaults allow 4096 outstanding jobs, including running jobs and results awaiting commit, and 256 per source. Waiting jobs occupy no hashing worker. Hashing holds neither the database nor the failure-limiter lock. Messages stay ordered per connection; completion rechecks the connection identity and credential snapshot, so stale work cannot overwrite concurrent password changes or restore deleted identities. Missing devices, wrong legacy passwords and malformed records also run one Argon2id calculation to reduce timing differences. Startup checks the actual OpenSSL provider and parameters.
 
-Registration, login, queries, joins and new password-change calculations share admission limits of 60 attempts per source IP per minute and 300 per process per minute, including successful operations. Rate exhaustion returns `Too many credential requests`; a full calculation queue returns `Credential service busy`, both with `retry_after` seconds. Six-character passwords still have limited strength: these budgets and the failure cooldown below constrain online guessing, while slow hashing raises the cost of offline guessing after a database leak.
+Registration, login, queries, joins and new password-change calculations share a source budget: by default, 60 starts per IP per minute, including successful operations. There is no longer a process-wide 300-per-minute quota. A token is consumed only after a worker becomes available; enqueueing, canceling disconnected waiters and rejecting a full queue consume no token. Rate-limited sources wait on the server without blocking unrelated sources. Queue overflow or expiry returns the existing `Credential service busy` response with `retry_after`. Failure cooldowns remain unchanged.
+
+Accepted authentication requests have a default 600-second lifetime and are exempt from the ordinary 15-second unauthenticated-connection timeout while pending. Connections that never submit an accepted request still use the 15-second rule. Legacy clients receive the original login response after waiting, without a new protocol or client-side retry. The 600 seconds is a maximum, not a fixed delay. Disconnected waiters are canceled. The restart recovery window for remote-session records also extends with the authentication lifetime so records survive queued reconnects.
+
+Six-character passwords still have limited strength: source budgets and the failure cooldown below constrain online guessing, while slow hashing raises the cost of offline guessing after a database leak.
 
 Each WebSocket connection allows one successful login. Subsequent `login` requests return `Already authenticated` without registering devices, verifying passwords or updating presence. Open a new connection to log in again; an initial failed login can still be retried on the same connection.
 
@@ -250,6 +254,26 @@ The source is the TCP peer IP without its port; request fields and `X-Forwarded-
 
 Registration must explicitly use an empty device ID (native clients) or `web` (Web clients). Logging in with an unknown or expired ID returns the generic authentication failure instead of assigning a replacement ID. Clients must reset their local identity to register again.
 
+### Online capacity and restart recovery
+
+Configure these options in `.env` and recreate the signaling container:
+
+| Setting | Default | Effect and range |
+| --- | --- | --- |
+| `CROSSDESK_MAX_CONNECTIONS` | `8192` | Total signaling connections, 1–65536; includes devices, `C-*` controllers and unopened connections |
+| `CROSSDESK_NOFILE` | `32768` | Compose container file-descriptor limit; used and reserved FDs reduce effective capacity |
+| `CROSSDESK_AUTH_WORKERS` | `2` | Concurrent password workers, 1–32; controls CPU, memory and authentication throughput |
+| `CROSSDESK_AUTH_QUEUE_CAPACITY` | `4096` | Total outstanding requests, 1–16384; absorbs bursts without increasing calculation speed |
+| `CROSSDESK_AUTH_SOURCE_CAPACITY` | `256` | Outstanding requests per source IP, 1–4096; prevents one source from filling the queue |
+| `CROSSDESK_AUTH_SOURCE_PER_MINUTE` | `60` | Starts per IP per minute, 1–6000; shared behind NAT or a reverse proxy |
+| `CROSSDESK_AUTH_TIMEOUT_SECONDS` | `600` | Maximum accepted authentication lifetime, 15–900 seconds |
+
+Keeping 2000 devices online does not require 2000 authentication threads. Only operations that verify passwords use these workers. Start with two, measure throughput on the deployment machine, and increase only with CPU headroom for TLS, heartbeats and database work. Each active Argon2id calculation needs approximately 19 MiB of algorithm memory; waiting requests do not allocate it. Burst recovery time is approximately the request count divided by measured completion throughput, also constrained by source quotas. Choose a deadline covering that recovery time. Shared-egress deployments need source capacity, source rate and deadlines considered together.
+
+Existing `.env` settings override new defaults: an existing `CROSSDESK_MAX_CONNECTIONS=2048` remains 2048. The requested TCP listen backlog is 4096, subject to the OS limit, while concurrent unopened connections remain capped at 128. These settings cover signaling and reconnect recovery; sustained online operation, concurrent remote sessions and TURN bandwidth still need tests on the deployment machine.
+
+`Credential capacity` logs effective authentication settings. `Credential diagnostics` reports `active`, `waiting`, `queue_full_total`, `source_full_total`, `expired_total`, `source_rate_waits_total` and calculation timing every minute. `Connection diagnostics` also counts successful logins, authentication failures, busy responses, failure-cooldown responses and authentication timeouts. Persistent queue growth means arrivals exceed processing throughput; increasing workers alone cannot resolve source-rate waits.
+
 ### Retention and automatic cleanup
 
 | Data | Default policy |
@@ -311,6 +335,18 @@ xmake b -vy crossdesk_server
 
 Ensure `xmake --version` works before the first build. See [xmake.lua](xmake.lua) for other platform configurations; the current release workflow provides only Linux binaries and images. Select Debug with `xmake f -m debug`; `xmake r -d crossdesk_server` runs through a debugger.
 
+Credential queue regression and legacy WSS integration tests require Python 3, `websockets>=15` and the `openssl` command:
+
+```bash
+xmake b credential_queue_test credential_server_fixture
+xmake r credential_queue_test
+# Substitute the fixture path for your platform, architecture and build mode.
+python3 tests/credential_server_integration.py \
+  --fixture build/linux/x86_64/release/credential_server_fixture --clients 2100
+```
+
+The test creates isolated certificates and databases in a temporary directory. It checks source waits beyond 15 seconds, disconnect cancellation, expiry, burst login and another burst after restart. Every login performs real Argon2id verification with no client login retry. TCP/TLS handshake concurrency is capped at 64 to avoid overflowing the local listen queue first. The burst case raises the loopback source quota; separate cases enforce source limits. Results do not imply default admission of 2100 devices behind one IP or production-machine performance.
+
 ### Build a local runtime image
 
 The Dockerfile **packages an already compiled Linux binary**. Complete a Release build on Ubuntu 22.04 with the same architecture as the target image, then run from the repository root:

compose.yaml

@@ -10,11 +10,16 @@ services:
     stop_grace_period: 20s
     ulimits:
       nofile:
-        soft: ${CROSSDESK_NOFILE:-8192}
-        hard: ${CROSSDESK_NOFILE:-8192}
+        soft: ${CROSSDESK_NOFILE:-32768}
+        hard: ${CROSSDESK_NOFILE:-32768}
     environment:
       EXTERNAL_IP: "${EXTERNAL_IP:?Set EXTERNAL_IP in .env}"
-      CROSSDESK_MAX_CONNECTIONS: "${CROSSDESK_MAX_CONNECTIONS:-2048}"
+      CROSSDESK_MAX_CONNECTIONS: "${CROSSDESK_MAX_CONNECTIONS:-8192}"
+      CROSSDESK_AUTH_WORKERS: "${CROSSDESK_AUTH_WORKERS:-2}"
+      CROSSDESK_AUTH_QUEUE_CAPACITY: "${CROSSDESK_AUTH_QUEUE_CAPACITY:-4096}"
+      CROSSDESK_AUTH_SOURCE_CAPACITY: "${CROSSDESK_AUTH_SOURCE_CAPACITY:-256}"
+      CROSSDESK_AUTH_SOURCE_PER_MINUTE: "${CROSSDESK_AUTH_SOURCE_PER_MINUTE:-60}"
+      CROSSDESK_AUTH_TIMEOUT_SECONDS: "${CROSSDESK_AUTH_TIMEOUT_SECONDS:-600}"
       CROSSDESK_LOG_RETENTION_DAYS: "${CROSSDESK_LOG_RETENTION_DAYS:-185}"
       CROSSDESK_OFFLINE_RETENTION_DAYS: "${CROSSDESK_OFFLINE_RETENTION_DAYS:-180}"
       CROSSDESK_RETENTION_INTERVAL_SECONDS: "${CROSSDESK_RETENTION_INTERVAL_SECONDS:-3600}"

src/credential_queue.h

@@ -0,0 +1,181 @@
+#ifndef CROSSDESK_CREDENTIAL_QUEUE_H
+#define CROSSDESK_CREDENTIAL_QUEUE_H
+
+#include <algorithm>
+#include <chrono>
+#include <cstdint>
+#include <deque>
+#include <functional>
+#include <map>
+#include <optional>
+#include <stdexcept>
+#include <string>
+#include <unordered_set>
+#include <utility>
+#include <vector>
+
+// Owned exclusively by the application worker. Waiting requests do not occupy
+// a hashing worker. Round-robin sources prevent a NAT or reconnecting client
+// from keeping unrelated sources behind its rate-limited requests.
+class CredentialQueue {
+ public:
+  using Clock = std::chrono::steady_clock;
+  using Now = std::function<Clock::time_point()>;
+  struct Job {
+    uint64_t id;
+    std::string source;
+    Clock::time_point deadline;
+    std::function<bool()> alive;
+    std::function<int()> admit;  // Consume a source token; or retry seconds.
+    std::function<void(size_t)> start;  // Reserved, idle hashing worker index.
+    std::function<void()> expire;
+    Clock::time_point queued_at{};
+  };
+  struct Stats {
+    uint64_t accepted = 0, started = 0, completed = 0, canceled = 0;
+    uint64_t full = 0, source_full = 0, expired = 0, source_waits = 0;
+    int64_t max_wait_ms = 0, max_execution_ms = 0;
+  };
+
+  CredentialQueue(size_t workers, size_t capacity, size_t source_capacity,
+                  Now now = Clock::now)
+      : capacity_(capacity), source_capacity_(source_capacity),
+        now_(std::move(now)), running_(workers) {
+    if (!workers || !capacity || !source_capacity)
+      throw std::invalid_argument("Credential queue limits must be positive");
+  }
+
+  // Does not start work until Pump(), so callers can publish pending state.
+  bool Enqueue(Job job) {
+    if (ids_.size() >= capacity_) { ++stats_.full; return false; }
+    if (ids_.count(job.id)) return false;
+    if (job.source.empty()) job.source = "unknown";
+    auto found = sources_.find(job.source);
+    if (found != sources_.end() && found->second.outstanding >= source_capacity_) {
+      ++stats_.source_full;
+      return false;
+    }
+    job.queued_at = now_();
+    ids_.insert(job.id);
+    auto& source = sources_[job.source];
+    ++source.outstanding;
+    source.waiting.push_back(std::move(job));
+    ++stats_.accepted;
+    return true;
+  }
+
+  // Running hashes finish normally; their result must still check liveness.
+  void Cancel(uint64_t id) {
+    for (auto it = sources_.begin(); it != sources_.end(); ++it) {
+      auto& source = it->second;
+      auto job = std::find_if(source.waiting.begin(), source.waiting.end(),
+                              [id](const Job& value) { return value.id == id; });
+      if (job == source.waiting.end()) continue;
+      source.waiting.erase(job);
+      ids_.erase(id);
+      ++stats_.canceled;
+      if (--source.outstanding == 0) sources_.erase(it);
+      return;
+    }
+  }
+
+  void Complete(size_t worker) {
+    auto& active = running_.at(worker);
+    if (!active) throw std::logic_error("Credential worker completed twice");
+    stats_.max_execution_ms = std::max(stats_.max_execution_ms,
+                                      Millis(now_() - active->started_at));
+    ids_.erase(active->id);
+    auto source = sources_.find(active->source);
+    if (--source->second.outstanding == 0) sources_.erase(source);
+    active.reset();
+    ++stats_.completed;
+  }
+
+  void Pump() {
+    if (pumping_) return;
+    pumping_ = true;
+    struct Reset { bool& flag; ~Reset() { flag = false; } } reset{pumping_};
+    const auto now = now_();
+    std::vector<std::function<void()>> expired;
+    // Also prune when every worker is busy, bounding waiting credentials and
+    // preventing disconnected/expired requests from retaining capacity.
+    for (auto it = sources_.begin(); it != sources_.end();) {
+      auto& source = it->second;
+      for (auto job = source.waiting.begin(); job != source.waiting.end();) {
+        const bool alive = job->alive();
+        if (alive && now < job->deadline) { ++job; continue; }
+        if (alive) {
+          ++stats_.expired;
+          expired.push_back(std::move(job->expire));
+        } else {
+          ++stats_.canceled;
+        }
+        ids_.erase(job->id);
+        --source.outstanding;
+        job = source.waiting.erase(job);
+      }
+      if (source.outstanding == 0) it = sources_.erase(it);
+      else ++it;
+    }
+    // Callbacks may enqueue new work. No container references survive them.
+    for (auto& callback : expired) callback();
+    for (size_t worker = 0; worker < running_.size(); ++worker) {
+      if (running_[worker]) continue;
+      auto source = sources_.upper_bound(last_source_);
+      const auto count = sources_.size();
+      for (size_t attempt = 0; attempt < count; ++attempt) {
+        if (source == sources_.end()) source = sources_.begin();
+        auto& queue = source->second;
+        if (queue.waiting.empty() || now < queue.not_before) { ++source; continue; }
+        const auto delay = queue.waiting.front().admit();
+        if (delay > 0) {
+          queue.not_before = now + std::chrono::seconds(delay);
+          ++stats_.source_waits;
+          ++source;
+          continue;
+        }
+        auto job = std::move(queue.waiting.front());
+        queue.waiting.pop_front();
+        last_source_ = source->first;
+        running_[worker] = Running{job.id, job.source, now};
+        stats_.max_wait_ms = std::max(stats_.max_wait_ms, Millis(now - job.queued_at));
+        ++stats_.started;
+        job.start(worker);
+        break;
+      }
+    }
+  }
+
+  size_t Active() const {
+    return std::count_if(running_.begin(), running_.end(),
+                         [](const auto& slot) { return slot.has_value(); });
+  }
+  size_t Waiting() const { return ids_.size() - Active(); }
+  size_t Capacity() const { return capacity_; }
+  const Stats& GetStats() const { return stats_; }
+
+ private:
+  static int64_t Millis(Clock::duration value) {
+    return std::chrono::duration_cast<std::chrono::milliseconds>(value).count();
+  }
+  struct Source {
+    std::deque<Job> waiting;
+    size_t outstanding = 0;
+    Clock::time_point not_before{};
+  };
+  struct Running {
+    uint64_t id;
+    std::string source;
+    Clock::time_point started_at;
+  };
+  const size_t capacity_, source_capacity_;
+  Now now_;
+  std::map<std::string, Source> sources_;
+  std::unordered_set<uint64_t> ids_;
+  std::vector<std::optional<Running>> running_;
+  std::string last_source_;
+  Stats stats_;
+  bool pumping_ = false;
+};
+
+#endif

src/device_auth_limiter.cpp

@@ -106,18 +106,21 @@ int CredentialWorkLimiter::Admit(const std::string& source) {
   std::lock_guard<std::mutex> lock(mutex_);
   const auto now = now_();
   constexpr auto window = std::chrono::minutes(1);
-  for (auto it = sources_.begin(); it != sources_.end();) {
-    if (it->second.expires <= now)
-      it = sources_.erase(it);
-    else
-      ++it;
+  if (now >= next_cleanup_) {
+    for (auto it = sources_.begin(); it != sources_.end();) {
+      if (it->second.expires <= now)
+        it = sources_.erase(it);
+      else
+        ++it;
+    }
+    next_cleanup_ = now + window;
   }
-  if (global_.expires <= now) global_ = {0, now + window};
   const auto key = source.empty() ? "unknown" : source;
   if (!sources_.count(key) && sources_.size() >= DeviceAuthLimiter::kMaxEntries)
     return 60;
   auto& local =
       sources_.try_emplace(key, Window{0, now + window}).first->second;
+  if (local.expires <= now) local = {0, now + window};
   auto retry = [now](const Window& value, size_t limit) {
     return value.count < limit
                ? 0
@@ -125,9 +128,8 @@ int CredentialWorkLimiter::Admit(const std::string& source) {
                                       value.expires - now)
                                       .count());
   };
-  const int delay = std::max(retry(local, 60), retry(global_, 300));
+  const int delay = retry(local, source_limit_);
   if (delay > 0) return delay;
   ++local.count;
-  ++global_.count;
   return 0;
 }

src/device_auth_limiter.h

@@ -11,6 +11,7 @@
 #include <cstddef>
 #include <functional>
 #include <mutex>
+#include <stdexcept>
 #include <string>
 #include <unordered_map>
 #include <utility>
@@ -60,8 +61,12 @@ class DeviceAuthLimiter {
 class CredentialWorkLimiter {
  public:
   explicit CredentialWorkLimiter(
-      DeviceAuthLimiter::Now now = DeviceAuthLimiter::Clock::now)
-      : now_(std::move(now)) {}
+      DeviceAuthLimiter::Now now = DeviceAuthLimiter::Clock::now,
+      size_t source_limit = 60)
+      : now_(std::move(now)), source_limit_(source_limit) {
+    if (!source_limit_)
+      throw std::invalid_argument("Credential source limit must be positive");
+  }
   int Admit(const std::string& source);
 
  private:
@@ -71,7 +76,8 @@ class CredentialWorkLimiter {
   };
   DeviceAuthLimiter::Now now_;
   std::mutex mutex_;
-  Window global_;
+  const size_t source_limit_;
+  DeviceAuthLimiter::Clock::time_point next_cleanup_{};
   std::unordered_map<std::string, Window> sources_;
 };
 

src/session_recovery.cpp

@@ -38,7 +38,8 @@ struct Statement {
 }
 
 SessionRecovery::SessionRecovery(const std::string& path, DeviceDBManager* db,
-    std::shared_ptr<TransmissionManager> transmission, Send send)
+    std::shared_ptr<TransmissionManager> transmission, Send send,
+    int recovery_timeout_seconds)
     : db_(db), transmission_(std::move(transmission)), send_(std::move(send)) {
   if (sqlite3_open(path.c_str(), &sql_) != SQLITE_OK) {
     if (sql_) sqlite3_close(sql_);
@@ -61,7 +62,7 @@ SessionRecovery::SessionRecovery(const std::string& path, DeviceDBManager* db,
     Ticket ticket;
     ticket.tx = read.Text(0); ticket.host = read.Text(1);
     ticket.guest = read.Text(2); ticket.token = read.Text(3);
-    ticket.recovering = true; ticket.deadline = Now() + 120;
+    ticket.recovering = true; ticket.deadline = Now() + recovery_timeout_seconds;
     tickets_[{ticket.tx, ticket.guest}] = std::move(ticket);
   }
   } catch (...) {

src/session_recovery.h

@@ -21,7 +21,8 @@ class SessionRecovery {
  public:
   using Send = std::function<void(websocketpp::connection_hdl, nlohmann::json)>;
   SessionRecovery(const std::string& path, DeviceDBManager* db,
-                  std::shared_ptr<TransmissionManager> transmission, Send send);
+                  std::shared_ptr<TransmissionManager> transmission, Send send,
+                  int recovery_timeout_seconds = 120);
   ~SessionRecovery();
   void Login(websocketpp::connection_hdl hdl, const nlohmann::json& request);
   void Issue(const std::string& tx, const std::string& host,
@@ -53,4 +54,4 @@ class SessionRecovery {
            std::owner_less<websocketpp::connection_hdl>> capable_;
 };
 
-#endif
+#endif

src/signal_negotiation.cpp

@@ -74,14 +74,17 @@ SignalNegotiation::SignalNegotiation(
     DeviceDBManager* device_db,
     std::shared_ptr<TurnCredentialIssuer> turn_credential_issuer,
     std::shared_ptr<IceServerConfigIssuer> ice_config_issuer,
-    std::shared_ptr<DeviceAuthLimiter> device_auth_limiter)
+    std::shared_ptr<DeviceAuthLimiter> device_auth_limiter,
+    size_t source_credential_limit)
     : transmission_manager_(transmission_manager),
       device_db_manager_(device_db),
       turn_credential_issuer_(std::move(turn_credential_issuer)),
       ice_config_issuer_(std::move(ice_config_issuer)),
       device_auth_limiter_(device_auth_limiter
                                ? std::move(device_auth_limiter)
-                               : std::make_shared<DeviceAuthLimiter>()) {}
+                               : std::make_shared<DeviceAuthLimiter>()),
+      credential_work_limiter_(DeviceAuthLimiter::Clock::now,
+                               source_credential_limit) {}
 
 SignalNegotiation::~SignalNegotiation() {}
 
@@ -148,17 +151,25 @@ void SignalNegotiation::RunCredentialWork(websocketpp::connection_hdl hdl,
                                           const std::string& source_address,
                                           json failure, CredentialWork work,
                                           const CredentialDispatch& dispatch) {
-  const int delay = credential_work_limiter_.Admit(source_address);
+  auto admit = [this, source_address] {
+    return credential_work_limiter_.Admit(source_address);
+  };
+  if (dispatch) {
+    auto expired = [this, hdl, failure]() mutable {
+      failure["reason"] = "Credential service busy";
+      failure["retry_after"] = 1;
+      send_msg_(hdl, failure);
+    };
+    if (!dispatch(std::move(work), std::move(admit), expired)) expired();
+    return;
+  }
+  const int delay = admit();
   if (delay > 0) {
     failure["reason"] = "Too many credential requests";
     failure["retry_after"] = delay;
     send_msg_(hdl, failure);
-  } else if (!dispatch) {
+  } else {
     work()();
-  } else if (!dispatch(std::move(work))) {
-    failure["reason"] = "Credential service busy";
-    failure["retry_after"] = 1;
-    send_msg_(hdl, failure);
   }
 }
 

src/signal_negotiation.h

@@ -27,13 +27,17 @@ class SignalNegotiation {
   // Run computation on a credential worker; run its returned continuation on
   // the application worker after checking that the connection is still alive.
   using CredentialWork = std::function<std::function<void()>()>;
-  using CredentialDispatch = std::function<bool(CredentialWork)>;
+  // Admission is invoked only with a reserved execution slot, never on enqueue.
+  // Expiry sends the original protocol's failure response for legacy clients.
+  using CredentialDispatch = std::function<bool(
+      CredentialWork, std::function<int()>, std::function<void()>)>;
   SignalNegotiation(
       std::shared_ptr<TransmissionManager> transmission_manager,
       DeviceDBManager* device_db,
       std::shared_ptr<TurnCredentialIssuer> turn_credential_issuer = nullptr,
       std::shared_ptr<IceServerConfigIssuer> ice_config_issuer = nullptr,
-      std::shared_ptr<DeviceAuthLimiter> device_auth_limiter = nullptr);
+      std::shared_ptr<DeviceAuthLimiter> device_auth_limiter = nullptr,
+      size_t source_credential_limit = 60);
   ~SignalNegotiation();
 
   void SetSendMsgCallback(

src/signal_server.cpp

@@ -16,7 +16,6 @@
 
 namespace {
 
-constexpr long kRecoveredSessionCleanupDelayMs = 120000;
 constexpr int kDefaultTurnCredentialTtlSeconds = 3600;
 constexpr auto kDiagnosticInterval = std::chrono::seconds(60);
 constexpr auto kDiagnosticWarningInterval = std::chrono::seconds(30);
@@ -133,6 +132,22 @@ SignalServer::SignalServer(uint16_t port, std::string certs_dir,
       port_, certs_dir_, db_path);
 
   DevicePassword::CheckSupport();
+  credential_worker_count_ = EnvMillis("CROSSDESK_AUTH_WORKERS", 2, 1, 32);
+  credential_timeout_seconds_ =
+      EnvMillis("CROSSDESK_AUTH_TIMEOUT_SECONDS", 600, 15, 900);
+  const auto credential_capacity =
+      EnvMillis("CROSSDESK_AUTH_QUEUE_CAPACITY", 4096, 1, 16384);
+  const auto source_capacity =
+      EnvMillis("CROSSDESK_AUTH_SOURCE_CAPACITY", 256, 1, 4096);
+  const auto source_rate =
+      EnvMillis("CROSSDESK_AUTH_SOURCE_PER_MINUTE", 60, 1, 6000);
+  credential_queue_ = std::make_unique<CredentialQueue>(
+      credential_worker_count_, credential_capacity, source_capacity);
+  LOG_INFO("Credential capacity: workers={} outstanding_limit={} "
+           "source_outstanding_limit={} source_per_minute={} timeout_seconds={} "
+           "global_rate_limit=disabled", credential_worker_count_,
+           credential_capacity, source_capacity, source_rate,
+           credential_timeout_seconds_);
 
   server_.set_error_channels(websocketpp::log::elevel::none);
   server_.set_access_channels(websocketpp::log::alevel::none);
@@ -176,12 +191,13 @@ SignalServer::SignalServer(uint16_t port, std::string certs_dir,
       });
   session_recovery_ = std::make_unique<SessionRecovery>(db_path,
       device_db_manager_.get(), transmission_manager_,
-      [this](websocketpp::connection_hdl hdl, json msg) { SendMsg(hdl, std::move(msg)); });
+      [this](websocketpp::connection_hdl hdl, json msg) { SendMsg(hdl, std::move(msg)); },
+      std::max(120, credential_timeout_seconds_ + 15));
   RestorePersistedRemoteControlSessions(transmission_manager_,
                                         device_db_manager_.get(), session_recovery_.get());
   signal_negotiation_ = std::make_unique<SignalNegotiation>(
       transmission_manager_, device_db_manager_.get(), nullptr,
-      CreateIceServerConfigIssuer());
+      CreateIceServerConfigIssuer(), nullptr, source_rate);
   signal_negotiation_->SetSessionRecovery(session_recovery_.get());
   signal_negotiation_->SetSendMsgCallback(std::bind(&SignalServer::SendMsg,
                                                     this, std::placeholders::_1,
@@ -216,16 +232,17 @@ SignalServer::SignalServer(uint16_t port, std::string certs_dir,
       notification_read_service_.get());
   admin_controller_->SetSessionRecovery(session_recovery_.get());
   admin_read_controller_->SetSessionRecovery(session_recovery_.get());
-  max_connections_ = EnvMillis("CROSSDESK_MAX_CONNECTIONS", 2048, 1, 65536);
+  max_connections_ = EnvMillis("CROSSDESK_MAX_CONNECTIONS", 8192, 1, 65536);
   auto on_error = [this](std::exception_ptr error) { WorkerFailed(error); };
   application_worker_ =
       std::make_unique<BoundedExecutor>(
-          1024, max_connections_ + kMaxCredentialJobs + 4, on_error);
+          std::max<size_t>(1024, credential_capacity),
+          max_connections_ + credential_worker_count_ + 4, on_error);
   admin_worker_ = std::make_unique<BoundedExecutor>(32, 0, on_error);
   maintenance_worker_ = std::make_unique<BoundedExecutor>(2, 0, on_error);
-  for (size_t i = 0; i < kCredentialWorkers; ++i)
+  for (size_t i = 0; i < credential_worker_count_; ++i)
     credential_workers_.push_back(
-        std::make_unique<BoundedExecutor>(kCredentialQueueCapacity, 0, on_error));
+        std::make_unique<BoundedExecutor>(1, 0, on_error));
   server_.set_max_message_size(64 * 1024);
   server_.set_max_http_body_size(16 * 1024);
   server_.set_open_handshake_timeout(10000);
@@ -344,13 +361,17 @@ void SignalServer::LogConnectionDiagnostics() {
       "accept_failures_total={} init_resource_failures_total={} "
       "connection_limit_checks_total={} unopened_limit_checks_total={} "
       "fd_limit_checks_total={} fd_sample_failed_checks_total={} "
-      "event_loop_stalls_total={}",
+      "event_loop_stalls_total={} login_success_total={} "
+      "authentication_failed_total={} credential_busy_replies_total={} "
+      "authentication_cooldown_replies_total={} authentication_timeouts_total={}",
       DescribeConnectionResources(), diagnostics_.accepted, diagnostics_.opened,
       diagnostics_.preopen_failed,
       diagnostics_.accept_failed, diagnostics_.init_resource_failed,
       diagnostics_.connection_limit_checks, diagnostics_.unopened_limit_checks,
       diagnostics_.fd_limit_checks, diagnostics_.fd_sample_failed_checks,
-      diagnostics_.loop_stalls);
+      diagnostics_.loop_stalls, diagnostics_.login_success,
+      diagnostics_.authentication_failed, diagnostics_.credential_busy,
+      diagnostics_.authentication_throttled, diagnostics_.authentication_timeouts);
 }
 
 void SignalServer::AcceptNext() {
@@ -440,6 +461,7 @@ void SignalServer::AcceptNext() {
       con->get_raw_socket() = std::move(*socket);
       auto state = std::make_shared<ConnectionState>();
       state->id = next_connection_id_++;
+      state->handle = con->get_handle();
       // Read transport metadata on the network thread. Never use forwarded
       // headers or source ports as authentication rate-limit keys.
       asio::error_code peer_error;
@@ -498,7 +520,9 @@ void SignalServer::QueueSessionCleanup(
   ++pending_cleanup_;
   if (!application_worker_->Submit(
           [this, hdl, state] {
+            credential_queue_->Cancel(state->id);
             state->pending_messages -= state->deferred_messages.size();
+            deferred_message_count_ -= state->deferred_messages.size();
             state->deferred_messages.clear();
             // On process shutdown the durable session ledger is the checkpoint.
             // Releasing individual users here would erase it before restart.
@@ -679,7 +703,8 @@ void SignalServer::ScheduleRuntimeHeartbeat() {
 
 void SignalServer::ScheduleRecoveredSessionCleanup() {
   server_.set_timer(
-      kRecoveredSessionCleanupDelayMs, [this](websocketpp::lib::error_code ec) {
+      std::max(120, credential_timeout_seconds_ + 15) * 1000,
+      [this](websocketpp::lib::error_code ec) {
         if (ec || stopping_) return;
         if (!application_worker_->Submit(
                 [this] {
@@ -803,7 +828,11 @@ void SignalServer::ScheduleMaintenance() {
           CloseConnection(entry.first, "Heartbeat timeout",
                           websocketpp::close::status::going_away);
         } else if (state->opened && !state->authenticated &&
-                   lifetime > authentication_timeout_ms_) {
+                   lifetime > authentication_timeout_ms_ &&
+                   std::chrono::duration_cast<std::chrono::milliseconds>(
+                       now.time_since_epoch()).count() >=
+                       state->credential_deadline_ms.load()) {
+          ++diagnostics_.authentication_timeouts;
           CloseConnection(entry.first, "Authentication timeout",
                           websocketpp::close::status::policy_violation);
         } else if (state->pending_http && lifetime > http_timeout_ms_) {
@@ -880,9 +909,9 @@ void SignalServer::Run() {
   }
   LOG_INFO("Connection capacity: configured_max_connections={} "
            "effective_max_connections={} fd_open={} fd_soft_limit={} "
-           "fd_reserve={} unopened_limit={} accept_retry_ms=500 listen_backlog=128",
+           "fd_reserve={} unopened_limit={} accept_retry_ms=500 listen_backlog={}",
            configured_max_connections, max_connections_, fd_usage_.open,
-           fd_usage_.soft_limit, fd_reserve_, max_handshakes_);
+           fd_usage_.soft_limit, fd_reserve_, max_handshakes_, kListenBacklog);
   LOG_INFO("Retention policy: log_days={} offline_days={} interval_seconds={} "
            "device_credentials=expire_with_device", retention_policy_.log_days,
            retention_policy_.offline_days, retention_policy_.interval_seconds);
@@ -896,7 +925,9 @@ void SignalServer::Run() {
   acceptor_->open(asio::ip::tcp::v4());
   acceptor_->set_option(asio::ip::tcp::acceptor::reuse_address(true));
   acceptor_->bind(asio::ip::tcp::endpoint(asio::ip::tcp::v4(), port_));
-  acceptor_->listen(128);
+  // Absorb restart bursts in the kernel while keeping TLS handshakes bounded.
+  // The OS may clamp this request to its own listen queue limit.
+  acceptor_->listen(kListenBacklog);
   signals_ = std::make_unique<websocketpp::lib::asio::signal_set>(
       server_.get_io_service(), SIGINT, SIGTERM);
   signals_->async_wait([this](websocketpp::lib::error_code ec, int) {
@@ -904,6 +935,7 @@ void SignalServer::Run() {
   });
   AcceptNext();
   ScheduleMaintenance();
+  ScheduleCredentialPump();
   ScheduleRecoveredSessionCleanup();
   LOG_INFO("Signal server listening on port [{}], waiting for connections...",
            port_);
@@ -973,20 +1005,28 @@ void SignalServer::SendMsg(websocketpp::connection_hdl hdl, json message) {
     return;
   }
   std::string login_id;
+  const auto failure_reason = message.value("reason", "");
   if (type == "login" && message.value("status", "") == "success") {
     login_id = message.value("user_id", "");
     login_id = login_id.substr(0, login_id.find('@'));
   }
   auto payload = message.dump();
   server_.get_io_service().post(
-      [this, hdl, payload = std::move(payload), login_id] {
+      [this, hdl, payload = std::move(payload), login_id, failure_reason] {
         --pending_sends_;
         auto it = connections_.find(hdl);
         if (it == connections_.end() || !it->second->alive) return;
         if (!login_id.empty()) {
+          ++diagnostics_.login_success;
           it->second->device_id = login_id;
           it->second->authenticated = true;
         }
+        if (failure_reason == "Authentication failed")
+          ++diagnostics_.authentication_failed;
+        else if (failure_reason == "Credential service busy")
+          ++diagnostics_.credential_busy;
+        else if (failure_reason == "Too many authentication attempts")
+          ++diagnostics_.authentication_throttled;
         websocketpp::lib::error_code ec;
         auto con = server_.get_con_from_hdl(hdl, ec);
         if (ec) return;
@@ -1039,6 +1079,14 @@ void SignalServer::DispatchMessage(
     const std::shared_ptr<ConnectionState>& state) {
   if (state->alive && !stopping_) {
     if (state->credential_pending) {
+      // A large login queue must not multiply the per-connection message
+      // backlog into hundreds of thousands of retained JSON documents.
+      if (deferred_message_count_ >= 1024) {
+        --state->pending_messages;
+        RequestBackpressureClose(hdl);
+        return;
+      }
+      ++deferred_message_count_;
       state->deferred_messages.push_back(std::move(message));
       return;
     }
@@ -1050,44 +1098,121 @@ void SignalServer::DispatchMessage(
 bool SignalServer::SubmitCredentialWork(
     websocketpp::connection_hdl hdl,
     const std::shared_ptr<ConnectionState>& state,
-    SignalNegotiation::CredentialWork work) {
-  if (stopping_ || !state->alive || state->credential_pending ||
-      credential_jobs_ >= kMaxCredentialJobs)
-    return false;
-  auto job = [this, hdl, state, work = std::move(work)] {
+    SignalNegotiation::CredentialWork work, std::function<int()> admit,
+    std::function<void()> expire) {
+  if (stopping_ || !state->alive || state->credential_pending) return false;
+  const auto deadline = Clock::now() +
+                        std::chrono::seconds(credential_timeout_seconds_);
+  CredentialQueue::Job job{
+      state->id, state->source_address, deadline,
+      [this, state] { return !stopping_ && state->alive.load(); },
+      std::move(admit),
+      [this, hdl, state, work = std::move(work)](size_t worker) mutable {
+        StartCredentialWork(hdl, state, std::move(work), worker);
+      },
+      [this, state, expire = std::move(expire)] {
+        if (!stopping_ && state->alive) expire();
+        FinishCredentialRequest(state);
+      }};
+  if (!credential_queue_->Enqueue(std::move(job))) return false;
+  state->credential_pending = true;
+  // Keep deadlines distinct even for successive requests in one millisecond.
+  const auto deadline_ms = std::chrono::duration_cast<std::chrono::milliseconds>(
+      deadline.time_since_epoch()).count();
+  state->credential_deadline_ms =
+      std::max(deadline_ms, state->credential_deadline_ms.load() + 1);
+  credential_queue_->Pump();
+  return true;
+}
+
+void SignalServer::FinishCredentialRequest(
+    const std::shared_ptr<ConnectionState>& state) {
+  state->credential_pending = false;
+  const auto deadline = state->credential_deadline_ms.load();
+  // Successful login replies are posted before this reset. Do not reset the
+  // deadline of a subsequent request that was queued before this callback ran.
+  server_.get_io_service().post([state, deadline]() mutable {
+    auto expected = deadline;
+    state->credential_deadline_ms.compare_exchange_strong(expected, 0);
+  });
+  while (!state->credential_pending && !state->deferred_messages.empty()) {
+    auto message = std::move(state->deferred_messages.front());
+    state->deferred_messages.pop_front();
+    --deferred_message_count_;
+    DispatchMessage(state->handle, std::move(message), state);
+  }
+}
+
+void SignalServer::StartCredentialWork(
+    websocketpp::connection_hdl hdl,
+    const std::shared_ptr<ConnectionState>& state,
+    SignalNegotiation::CredentialWork work, size_t worker) {
+  auto job = [this, hdl, state, work = std::move(work), worker] {
     std::function<void()> finish;
-    if (!stopping_ && state->alive) finish = work();
-    // At most 34 completions can be outstanding; they have reserved slots so
-    // load cannot strand a connection after consuming a credential-work slot.
+    if (!stopping_ && state->alive) {
+      const auto started = Clock::now();
+      finish = work();
+      credential_work_us_.fetch_add(
+          std::chrono::duration_cast<std::chrono::microseconds>(
+              Clock::now() - started).count());
+      ++credential_work_count_;
+    }
+    // One reserved completion per worker, independent of the waiting capacity.
     if (!application_worker_->Submit(
-            [this, hdl, state, finish = std::move(finish)] {
-              --credential_jobs_;
-              state->credential_pending = false;
+            [this, state, finish = std::move(finish), worker] {
+              credential_queue_->Complete(worker);
               try {
                 if (!stopping_ && state->alive && finish) finish();
               } catch (const json::exception&) {
                 LOG_WARN("Invalid signaling fields on connection [{}]", state->id);
               }
-              while (!state->credential_pending &&
-                     !state->deferred_messages.empty()) {
-                auto message = std::move(state->deferred_messages.front());
-                state->deferred_messages.pop_front();
-                DispatchMessage(hdl, std::move(message), state);
-              }
-            },
-            true))
+              FinishCredentialRequest(state);
+              if (!stopping_) credential_queue_->Pump();
+            }, true))
       throw std::runtime_error("Credential completion reserve exhausted");
   };
-  for (size_t i = 0; i < credential_workers_.size(); ++i) {
-    auto& worker = credential_workers_[next_credential_worker_++ %
-                                       credential_workers_.size()];
-    if (worker->Submit(job)) {
-      ++credential_jobs_;
-      state->credential_pending = true;
-      return true;
+  // The scheduler assigns only idle workers. Failure indicates shutdown or an
+  // invariant violation, not normal overload; queued jobs consume no tokens.
+  if (!credential_workers_.at(worker)->Submit(std::move(job)))
+    throw std::runtime_error("Reserved credential worker unavailable");
+}
+
+void SignalServer::LogCredentialDiagnostics() {
+  const auto& stats = credential_queue_->GetStats();
+  const auto count = credential_work_count_.load();
+  LOG_INFO("Credential diagnostics: active={} waiting={} outstanding_limit={} deferred_messages={} "
+           "accepted_total={} started_total={} completed_total={} canceled_total={} "
+           "queue_full_total={} source_full_total={} expired_total={} "
+           "source_rate_waits_total={} max_queue_wait_ms={} max_execution_ms={} "
+           "work_count={} average_work_ms={}",
+           credential_queue_->Active(), credential_queue_->Waiting(),
+           credential_queue_->Capacity(), deferred_message_count_, stats.accepted, stats.started,
+           stats.completed, stats.canceled, stats.full, stats.source_full,
+           stats.expired, stats.source_waits, stats.max_wait_ms,
+           stats.max_execution_ms, count,
+           count ? credential_work_us_.load() / count / 1000 : 0);
+}
+
+void SignalServer::ScheduleCredentialPump() {
+  if (stopping_) return;
+  server_.set_timer(250, [this](websocketpp::lib::error_code ec) {
+    if (ec || stopping_) return;
+    if (!credential_pump_pending_) {
+      credential_pump_pending_ = true;
+      if (!application_worker_->Submit([this] {
+            if (!stopping_) {
+              credential_queue_->Pump();
+              const auto now = Clock::now();
+              if (now >= next_credential_diagnostics_) {
+                LogCredentialDiagnostics();
+                next_credential_diagnostics_ = now + kDiagnosticInterval;
+              }
+            }
+            server_.get_io_service().post([this] { credential_pump_pending_ = false; });
+          })) credential_pump_pending_ = false;
     }
-  }
-  return false;
+    ScheduleCredentialPump();
+  });
 }
 
 void SignalServer::ProcessMessage(
@@ -1103,8 +1228,11 @@ void SignalServer::ProcessMessage(
       }
       return;
     }
-    auto dispatch = [this, hdl, state](SignalNegotiation::CredentialWork work) {
-      return SubmitCredentialWork(hdl, state, std::move(work));
+    auto dispatch = [this, hdl, state](SignalNegotiation::CredentialWork work,
+                                        std::function<int()> admit,
+                                        std::function<void()> expire) {
+      return SubmitCredentialWork(hdl, state, std::move(work),
+                                  std::move(admit), std::move(expire));
     };
     switch (HASH_STRING_PIECE(type.c_str())) {
       case "login"_H:

src/signal_server.h

@@ -31,6 +31,7 @@
 #include "resource_monitor.h"
 #include "retention_policy.h"
 #include "session_recovery.h"
+#include "credential_queue.h"
 #include "signal_negotiation.h"
 
 using nlohmann::json;
@@ -64,12 +65,9 @@ class SignalServer {
 
  private:
   using Clock = std::chrono::steady_clock;
-  static constexpr size_t kCredentialWorkers = 2;
-  static constexpr size_t kCredentialQueueCapacity = 16;
-  static constexpr size_t kMaxCredentialJobs =
-      kCredentialWorkers * (kCredentialQueueCapacity + 1);
   struct ConnectionState {
     uint64_t id = 0;
+    websocketpp::connection_hdl handle;
     std::string source_address;  // Set before publication; transport peer IP.
     std::string device_id;  // Accessed only on the network thread.
     Clock::time_point accepted = Clock::now(), last_heartbeat = accepted;
@@ -77,6 +75,8 @@ class SignalServer {
     server::connection_ptr pending_http;  // Network thread; retains deferred HTTP.
     std::atomic<bool> alive{true};
     std::atomic<size_t> pending_messages{0};
+    // Steady-clock epoch milliseconds; application worker publishes, I/O reads.
+    std::atomic<int64_t> credential_deadline_ms{0};
     // Application worker only. Preserve message order across async hashing.
     bool credential_pending = false;
     std::deque<json> deferred_messages;
@@ -99,7 +99,15 @@ class SignalServer {
                        const std::shared_ptr<ConnectionState>& state);
   bool SubmitCredentialWork(websocketpp::connection_hdl hdl,
                             const std::shared_ptr<ConnectionState>& state,
-                            SignalNegotiation::CredentialWork work);
+                            SignalNegotiation::CredentialWork work,
+                            std::function<int()> admit,
+                            std::function<void()> expire);
+  void StartCredentialWork(websocketpp::connection_hdl hdl,
+                           const std::shared_ptr<ConnectionState>& state,
+                           SignalNegotiation::CredentialWork work, size_t worker);
+  void FinishCredentialRequest(const std::shared_ptr<ConnectionState>& state);
+  void ScheduleCredentialPump();
+  void LogCredentialDiagnostics();
   void CompleteHttp(websocketpp::connection_hdl hdl,
                     AdminHttpResponse response);
   void WorkerFailed(std::exception_ptr error);
@@ -121,8 +129,9 @@ class SignalServer {
   size_t fd_connections_at_sample_ = 0;
   uint64_t next_connection_id_ = 1;
   size_t pending_cleanup_ = 0;
-  size_t max_connections_ = 2048;
+  size_t max_connections_ = 8192;
   const size_t fd_reserve_ = 256, max_handshakes_ = 128;
+  static constexpr int kListenBacklog = 4096;
   const long heartbeat_timeout_ms_ = 30000, check_interval_ms_ = 5000;
   const long authentication_timeout_ms_ = 15000, http_timeout_ms_ = 10000;
   const long tls_reload_interval_ms_ = 30000;
@@ -134,6 +143,9 @@ class SignalServer {
     uint64_t accept_failed = 0, init_resource_failed = 0, loop_stalls = 0;
     uint64_t connection_limit_checks = 0, unopened_limit_checks = 0;
     uint64_t fd_limit_checks = 0, fd_sample_failed_checks = 0;
+    uint64_t login_success = 0, authentication_failed = 0;
+    uint64_t credential_busy = 0, authentication_throttled = 0;
+    uint64_t authentication_timeouts = 0;
   } diagnostics_;
   Clock::time_point next_diagnostics_{}, next_admission_warning_{};
   Clock::time_point next_loop_warning_{};
@@ -144,6 +156,7 @@ class SignalServer {
   std::atomic<bool> stopping_{false};
   bool accept_pending_ = false, accept_retry_pending_ = false;
   bool runtime_job_pending_ = false, tls_job_pending_ = false;
+  bool credential_pump_pending_ = false;  // I/O thread only.
   const RetentionPolicy retention_policy_ = RetentionPolicy::FromEnvironment();
   bool metadata_cleanup_pending_ = false, log_cleanup_pending_ = false;
   Clock::time_point next_metadata_cleanup_{}, next_log_cleanup_{};
@@ -163,9 +176,12 @@ class SignalServer {
   std::unique_ptr<BoundedExecutor> application_worker_, admin_worker_,
       maintenance_worker_;
   std::vector<std::unique_ptr<BoundedExecutor>> credential_workers_;
-  size_t next_credential_worker_ = 0;  // Application worker only.
-  size_t credential_jobs_ =
-      0;  // Includes completions waiting on the application worker.
+  std::unique_ptr<CredentialQueue> credential_queue_;  // Application worker only.
+  size_t credential_worker_count_ = 2;
+  int credential_timeout_seconds_ = 600;
+  Clock::time_point next_credential_diagnostics_{};
+  size_t deferred_message_count_ = 0;  // Application worker; bounded globally.
+  std::atomic<uint64_t> credential_work_us_{0}, credential_work_count_{0};
 
   std::shared_ptr<TransmissionManager> transmission_manager_;
   std::unique_ptr<DeviceDBManager> device_db_manager_;

tests/credential_queue_test.cpp

@@ -0,0 +1,129 @@
+#include "credential_queue.h"
+#include "device_auth_limiter.h"
+
+#include <iostream>
+#include <string>
+#include <vector>
+
+int main() {
+  int failures = 0;
+  auto expect = [&](bool value, const char* message) {
+    if (!value) { std::cerr << "FAIL: " << message << '\n'; ++failures; }
+  };
+  using Clock = CredentialQueue::Clock;
+  auto time = Clock::time_point{};
+  auto now = [&] { return time; };
+  CredentialWorkLimiter limiter(now);
+  for (int i = 0; i < 2000; ++i)
+    expect(limiter.Admit("source-" + std::to_string(i)) == 0,
+           "2000 independent clients have no shared minute quota");
+  for (int i = 0; i < 60; ++i)
+    expect(limiter.Admit("nat") == 0, "source permits its own budget");
+  expect(limiter.Admit("nat") == 60, "source throttle remains enforced");
+  time += std::chrono::seconds(61);
+  expect(limiter.Admit("nat") == 0, "expired source budget resets");
+
+  int admissions = 0, expirations = 0;
+  std::vector<uint64_t> started;
+  auto make_job = [&](uint64_t id, std::string source, bool* alive = nullptr) {
+    return CredentialQueue::Job{
+        id, std::move(source), time + std::chrono::seconds(120),
+        [alive] { return !alive || *alive; },
+        [&] { ++admissions; return 0; },
+        [&, id](size_t) { started.push_back(id); },
+        [&] { ++expirations; }};
+  };
+  {
+    CredentialQueue queue(2, 4, 2, now);
+    expect(queue.Enqueue(make_job(1, "a")), "first job queues");
+    expect(queue.Enqueue(make_job(2, "a")), "second source job queues");
+    expect(!queue.Enqueue(make_job(3, "a")), "per-source outstanding cap");
+    expect(queue.Enqueue(make_job(3, "b")), "unrelated source queues");
+    expect(queue.Enqueue(make_job(4, "c")), "global capacity includes waiting");
+    expect(!queue.Enqueue(make_job(5, "d")), "bounded global capacity");
+    expect(admissions == 0, "queued and rejected work consumes no source tokens");
+    queue.Pump();
+    expect(started == std::vector<uint64_t>({1, 3}), "workers fairly select sources");
+    expect(queue.Active() == 2 && queue.Waiting() == 2, "concurrency stays bounded");
+    queue.Cancel(4);
+    queue.Complete(0);
+    queue.Pump();
+    expect(started.back() == 2 && admissions == 3, "canceled work never consumes quota");
+    queue.Complete(0); queue.Complete(1);
+    expect(queue.Waiting() == 0 && queue.Active() == 0, "completion releases all capacity");
+  }
+  {
+    CredentialQueue queue(1, 4, 4, now);
+    auto delayed = make_job(10, "a");
+    const auto ready = time + std::chrono::seconds(60);
+    delayed.admit = [&] { return time < ready ? 60 : 0; };
+    queue.Enqueue(std::move(delayed)); queue.Enqueue(make_job(11, "b"));
+    queue.Pump();
+    expect(started.back() == 11, "rate-limited source does not block another source");
+    queue.Complete(0); queue.Pump();
+    expect(queue.Active() == 0 && queue.Waiting() == 1, "source retry waits server-side");
+    time = ready; queue.Pump();
+    expect(started.back() == 10, "legacy request resumes without a client retry");
+    queue.Complete(0);
+  }
+  {
+    CredentialQueue queue(1, 4, 4, now);
+    bool alive = true;
+    queue.Enqueue(make_job(20, "a")); queue.Enqueue(make_job(21, "b", &alive));
+    queue.Enqueue(make_job(22, "c")); queue.Pump();
+    const auto before = admissions;
+    alive = false; time += std::chrono::seconds(121); queue.Pump();
+    expect(queue.Active() == 1 && queue.Waiting() == 0,
+           "expiry and disconnect are pruned even when workers are full");
+    expect(expirations == 1 && admissions == before,
+           "only live expired work receives failure, without consuming quota");
+    queue.Complete(0);
+  }
+  {
+    CredentialQueue queue(2, 4096, 256, now);
+    const auto before = started.size();
+    for (int i = 0; i < 2000; ++i)
+      expect(queue.Enqueue(make_job(100 + i, std::to_string(i))),
+             "2000-client restart burst is accepted");
+    queue.Pump();
+    for (int i = 0; i < 1000; ++i) {
+      expect(queue.Active() == 2, "burst never exceeds hashing concurrency");
+      time += std::chrono::milliseconds(10);
+      queue.Complete(0); queue.Complete(1); queue.Pump();
+    }
+    expect(started.size() - before == 2000 && queue.Waiting() == 0,
+           "all legacy requests complete without reconnecting");
+    expect(queue.GetStats().full == 0 && queue.GetStats().expired == 0,
+           "restart burst does not reject or expire healthy work");
+  }
+  {
+    CredentialQueue queue(1, 1, 1, now);
+    auto job = make_job(30, "before");
+    job.expire = [&] {
+      expect(queue.Enqueue(make_job(30, "after")),
+             "expiry callback may enqueue next request on the same connection");
+      queue.Pump();
+    };
+    queue.Enqueue(std::move(job));
+    time += std::chrono::seconds(121);
+    queue.Pump();
+    expect(queue.Active() == 1 && started.back() == 30,
+           "reentrant expiry safely starts new request");
+    queue.Cancel(30);
+    expect(queue.Active() == 1 && !queue.Enqueue(make_job(30, "after")),
+           "cancel does not release a running worker or duplicate its connection");
+    queue.Complete(0);
+    expect(queue.Enqueue(make_job(30, "after")), "completion releases connection id");
+    queue.Cancel(30);
+  }
+  {
+    DeviceAuthLimiter auth(now);
+    int verifies = 0;
+    for (int i = 0; i < 5; ++i)
+      auth.Verify("attacker", "device", [&] { ++verifies; return false; });
+    auto result = auth.Verify("another-source", "device", [&] { ++verifies; return true; });
+    expect(!result.authenticated && result.retry_after == 900 && verifies == 5,
+           "device failure cooldown remains effective across source changes");
+  }
+  return failures ? 1 : 0;
+}

tests/credential_server_fixture.cpp

@@ -0,0 +1,17 @@
+#include "signal_server.h"
+#include "log.h"
+
+#include <iostream>
+
+int main(int argc, char** argv) {
+  if (argc != 3) return 2;
+  try {
+    InitLogger(std::string(argv[2]) + "/logs");
+    SignalServer server(static_cast<uint16_t>(std::stoi(argv[1])), argv[2],
+                        std::string(argv[2]) + "/test.db");
+    server.Run();
+  } catch (const std::exception& error) {
+    std::cerr << error.what() << '\n';
+    return 1;
+  }
+}

tests/credential_server_integration.py

@@ -0,0 +1,303 @@
+"""Legacy WSS protocol integration; requires Python websockets and openssl.
+
+xmake b credential_server_fixture
+python3 tests/credential_server_integration.py --clients 2000
+All databases, certificates and logs are isolated in a temporary directory.
+"""
+import argparse
+import asyncio
+import json
+import os
+from pathlib import Path
+import resource
+import signal
+import socket
+import sqlite3
+import ssl
+import statistics
+import subprocess
+import tempfile
+import time
+
+import websockets
+
+
+class Server:
+    def __init__(self, binary, directory, **settings):
+        self.binary = binary
+        self.directory = directory
+        directory.mkdir(parents=True, exist_ok=True)
+        self.settings = settings
+        self.generation = 0
+        self.proc = None
+        subprocess.run([
+            'openssl', 'req', '-x509', '-newkey', 'rsa:2048', '-nodes',
+            '-keyout', str(directory / 'api.crossdesk.cn.key'),
+            '-out', str(directory / 'api.crossdesk.cn_bundle.crt'),
+            '-days', '1', '-subj', '/CN=localhost',
+        ], check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+        self.ssl = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+        self.ssl.check_hostname = False
+        self.ssl.verify_mode = ssl.CERT_NONE  # Isolated, self-signed loopback fixture.
+
+    async def start(self):
+        with socket.socket() as sock:
+            sock.bind(('127.0.0.1', 0))
+            self.port = sock.getsockname()[1]
+        self.generation += 1
+        self.log = self.directory / f'stdout-{self.generation}.log'
+        self.stream = self.log.open('w')
+        env = os.environ.copy()
+        env.update(CROSSDESK_ICE_SERVERS='[]', ADMIN_USERNAME='', ADMIN_PASSWORD='')
+        env.update(CROSSDESK_MAX_CONNECTIONS='8192', CROSSDESK_AUTH_WORKERS='2',
+                   CROSSDESK_AUTH_QUEUE_CAPACITY='4096',
+                   CROSSDESK_AUTH_SOURCE_CAPACITY='256',
+                   CROSSDESK_AUTH_SOURCE_PER_MINUTE='60',
+                   CROSSDESK_AUTH_TIMEOUT_SECONDS='600')
+        env.update({key: str(value) for key, value in self.settings.items()})
+        self.proc = subprocess.Popen(
+            [self.binary, str(self.port), str(self.directory)], env=env,
+            stdout=self.stream, stderr=subprocess.STDOUT)
+        deadline = time.monotonic() + 15
+        while 'Signal server listening' not in self.log.read_text():
+            assert self.proc.poll() is None, self.log.read_text()[-5000:]
+            assert time.monotonic() < deadline, 'startup timeout'
+            await asyncio.sleep(.05)
+
+    async def connect(self):
+        return await websockets.connect(
+            f'wss://127.0.0.1:{self.port}', ssl=self.ssl,
+            open_timeout=30, close_timeout=3, ping_interval=3, ping_timeout=30,
+            proxy=None)
+
+    async def stop(self):
+        if self.proc is None:
+            return
+        if self.proc.poll() is None:
+            self.proc.send_signal(signal.SIGTERM)
+            try:
+                await asyncio.wait_for(asyncio.to_thread(self.proc.wait), 12)
+            except asyncio.TimeoutError:
+                self.proc.kill()
+                await asyncio.to_thread(self.proc.wait)
+                raise AssertionError('shutdown did not drain pending authentication')
+        self.stream.close()
+        assert self.proc.returncode == 0, self.log.read_text()[-5000:]
+        self.proc = None
+
+
+async def login(ws, identity='', timeout=90):
+    # No retry_after, new capability, or retry logic: old login request/response.
+    await ws.send(json.dumps({'type': 'login', 'user_id': identity}))
+    while True:
+        result = json.loads(await asyncio.wait_for(ws.recv(), timeout))
+        if result.get('type') == 'login':
+            return result
+
+
+async def source_wait(binary, output):
+    server = Server(binary, output / 'source-wait', CROSSDESK_AUTH_WORKERS=1,
+                    CROSSDESK_AUTH_QUEUE_CAPACITY=1,
+                    CROSSDESK_AUTH_SOURCE_CAPACITY=1,
+                    CROSSDESK_AUTH_SOURCE_PER_MINUTE=1,
+                    CROSSDESK_AUTH_TIMEOUT_SECONDS=75)
+    sockets = []
+    try:
+        await server.start()
+        first = await server.connect(); sockets.append(first)
+        assert (await login(first))['status'] == 'success'
+        # A disconnected waiter must relinquish the one queue slot.
+        abandoned = await server.connect(); sockets.append(abandoned)
+        await abandoned.send(json.dumps({'type': 'login', 'user_id': ''}))
+        await asyncio.sleep(.3)
+        await abandoned.close()
+        # The client close can finish before websocketpp finishes TLS shutdown.
+        # Wait for the server's close callback, which queues cancellation.
+        cleanup_deadline = time.monotonic() + 8
+        while 'connection [2|] closed' not in server.log.read_text():
+            assert time.monotonic() < cleanup_deadline, 'disconnected waiter retained'
+            await asyncio.sleep(.05)
+        waiting = await server.connect(); sockets.append(waiting)
+        begun = time.monotonic()
+        response = asyncio.create_task(login(waiting))
+        await asyncio.sleep(.4)
+        overflow = await server.connect(); sockets.append(overflow)
+        failure = await login(overflow)
+        assert failure['status'] == 'fail' and failure['reason'] == 'Credential service busy'
+        await overflow.close()
+        await waiting.send('{"type":"login","user_id":""}')
+        await asyncio.sleep(20)
+        assert not response.done(), ('legacy waiter failed early', response.result())
+        pong = await waiting.ping()
+        await asyncio.wait_for(pong, 3)
+        result = await response
+        elapsed = time.monotonic() - begun
+        assert result['status'] == 'success' and 45 < elapsed < 75, (result, elapsed)
+        duplicate = json.loads(await asyncio.wait_for(waiting.recv(), 3))
+        assert duplicate['reason'] == 'Already authenticated', duplicate
+        print(json.dumps({'case': 'source_wait_legacy_no_retry', 'result': 'PASS',
+                          'wait_seconds': round(elapsed, 2),
+                          'heartbeat_after_20s': 'PASS',
+                          'disconnect_releases_slot': 'PASS',
+                          'deferred_login_order': 'PASS',
+                          'queue_full_does_not_consume_quota': 'PASS'}), flush=True)
+    finally:
+        await asyncio.gather(*(ws.close() for ws in sockets), return_exceptions=True)
+        await server.stop()
+
+
+async def expiry(binary, output):
+    server = Server(binary, output / 'expiry', CROSSDESK_AUTH_SOURCE_PER_MINUTE=1,
+                    CROSSDESK_AUTH_TIMEOUT_SECONDS=15)
+    sockets = []
+    try:
+        await server.start()
+        first = await server.connect(); sockets.append(first)
+        assert (await login(first))['status'] == 'success'
+        waiting = await server.connect(); sockets.append(waiting)
+        start = time.monotonic()
+        try:
+            result = await login(waiting, timeout=23)
+            assert result['status'] == 'fail' and result['reason'] == 'Credential service busy'
+        except websockets.ConnectionClosed as error:
+            assert error.rcvd and error.rcvd.code == 1008
+        elapsed = time.monotonic() - start
+        assert 14 <= elapsed < 23, elapsed
+        print(json.dumps({'case': 'bounded_wait_expiry', 'result': 'PASS',
+                          'seconds': round(elapsed, 2)}), flush=True)
+    finally:
+        await asyncio.gather(*(ws.close() for ws in sockets), return_exceptions=True)
+        await server.stop()
+
+
+async def burst(binary, output, count):
+    server = Server(binary, output / 'burst', CROSSDESK_MAX_CONNECTIONS=8192,
+                    CROSSDESK_AUTH_WORKERS=2, CROSSDESK_AUTH_QUEUE_CAPACITY=4096,
+                    CROSSDESK_AUTH_SOURCE_CAPACITY=4096,
+                    CROSSDESK_AUTH_SOURCE_PER_MINUTE=6000,
+                    CROSSDESK_AUTH_TIMEOUT_SECONDS=120)
+    sockets = []
+    try:
+        await server.start()
+        seed = await server.connect()
+        credentials = (await login(seed))['user_id']
+        seed_id, password = credentials.split('@', 1)
+        await seed.close(); await server.stop()
+        # Isolated test identities reuse one test digest to avoid timing fixture
+        # creation. Every login still performs the real Argon2id verification.
+        with sqlite3.connect(server.directory / 'test.db') as db:
+            salt, digest = db.execute(
+                'SELECT password_salt,password_hash FROM devices WHERE device_id=?',
+                (seed_id,)).fetchone()
+            db.executemany(
+                'INSERT INTO devices(device_id,password_salt,password_hash,retention_started_at) '
+                'VALUES(?,?,?,?)',
+                [(str(600000000 + i), salt, digest, int(time.time())) for i in range(count)])
+        for phase in ['initial_burst', 'restart_burst']:
+            await server.start()
+            started = time.monotonic()
+            latencies = []
+            # Bound simultaneous TCP/TLS handshakes, not pending logins. macOS
+            # defaults to a 128-entry listen backlog; overwhelming it tests the
+            # OS accept queue instead of the credential recovery path.
+            handshakes = asyncio.Semaphore(64)
+            async def client(index):
+                async with handshakes:
+                    ws = await server.connect()
+                sockets.append(ws)
+                result = await login(ws, f'{600000000 + index}@{password}', timeout=125)
+                assert result['status'] == 'success', result
+                latencies.append(time.monotonic() - started)
+            await asyncio.gather(*(client(i) for i in range(count)))
+            await asyncio.sleep(5)
+            # Verify that authenticated connections also serve ordinary signaling.
+            await asyncio.gather(*(ws.send('{"type":"ping"}') for ws in sockets))
+            replies = await asyncio.gather(*(asyncio.wait_for(ws.recv(), 10) for ws in sockets))
+            assert all(json.loads(msg)['type'] == 'pong' for msg in replies)
+            await server.stop()  # Disconnect every client as a real restart does.
+            text = server.log.read_text()
+            assert f'authenticated={count}' in text, 'online snapshot missing'
+            assert 'authentication_timeouts_total=0' in text
+            print(json.dumps({'case': phase, 'result': 'PASS', 'clients': count,
+                              'p50_login_seconds': round(statistics.median(latencies), 2),
+                              'p95_login_seconds': round(sorted(latencies)[int(count*.95)-1], 2),
+                              'all_online_seconds': round(max(latencies), 2),
+                              'concurrent_handshakes': 64,
+                              'all_signaling_pings': 'PASS'}), flush=True)
+            await asyncio.gather(*(ws.close() for ws in sockets), return_exceptions=True)
+            sockets = []
+    finally:
+        await asyncio.gather(*(ws.close() for ws in sockets), return_exceptions=True)
+        await server.stop()
+
+
+async def deferred_backlog(binary, output):
+    server = Server(binary, output / 'deferred-backlog',
+                    CROSSDESK_AUTH_SOURCE_PER_MINUTE=1)
+    sockets = []
+    try:
+        await server.start()
+        seed = await server.connect(); sockets.append(seed)
+        assert (await login(seed))['status'] == 'success'
+        for _ in range(9):
+            ws = await server.connect(); sockets.append(ws)
+            await ws.send('{"type":"login","user_id":""}')
+            await asyncio.sleep(.05)
+            # Below each connection's cap, but collectively above the global
+            # deferred-message capacity behind pending authentication.
+            for _ in range(120):
+                await ws.send('{"type":"login","user_id":""}')
+            await asyncio.sleep(.05)
+        overloaded = sockets[-1]
+        await asyncio.wait_for(overloaded.wait_closed(), 8)
+        assert overloaded.close_code == 1013, overloaded.close_code
+        pong = await sockets[1].ping()
+        await asyncio.wait_for(pong, 3)
+        await asyncio.gather(*(ws.close() for ws in sockets[1:]))
+        # Allow server TLS teardown / periodic closing-state cleanup to finish.
+        await asyncio.sleep(5)
+        fresh = await server.connect(); sockets.append(fresh)
+        for _ in range(121):
+            await fresh.send('{"type":"login","user_id":""}')
+        await asyncio.sleep(1)
+        pong = await fresh.ping()
+        await asyncio.wait_for(pong, 3)
+        print(json.dumps({'case': 'bounded_deferred_backlog', 'result': 'PASS',
+                          'unrelated_waiter_alive': 'PASS',
+                          'cleanup_releases_capacity': 'PASS'}), flush=True)
+    finally:
+        await asyncio.gather(*(ws.close() for ws in sockets), return_exceptions=True)
+        await server.stop()
+
+
+async def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--fixture', help='Fixture binary; auto-detects a unique build')
+    parser.add_argument('--clients', type=int, default=128)
+    parser.add_argument('--case', choices=['all', 'limits', 'burst'], default='all')
+    args = parser.parse_args()
+    assert 1 <= args.clients <= 4000
+    if not args.fixture:
+        fixtures = list(Path('build').glob('*/*/*/credential_server_fixture'))
+        if len(fixtures) != 1:
+            parser.error('pass --fixture with the built credential_server_fixture path')
+        args.fixture = str(fixtures[0])
+    soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE)
+    resource.setrlimit(resource.RLIMIT_NOFILE, (min(32768, hard) if hard != resource.RLIM_INFINITY else 32768, hard))
+    output = Path(tempfile.mkdtemp(prefix='crossdesk-auth-integration-'))
+    print('Integration logs: ' + str(output), flush=True)
+    binary = str(Path(args.fixture).resolve())
+    cases = []
+    if args.case != 'burst':
+        cases += [source_wait(binary, output), expiry(binary, output),
+                  deferred_backlog(binary, output)]
+    if args.case != 'limits':
+        cases += [burst(binary, output, args.clients)]
+    results = await asyncio.gather(*cases, return_exceptions=True)
+    for result in results:
+        if isinstance(result, BaseException):
+            raise result
+
+
+if __name__ == '__main__':
+    asyncio.run(main())

xmake.lua

@@ -118,6 +118,19 @@ target("signal_negotiation_test")
     add_files("tests/signal_negotiation_test.cpp")
     add_includedirs("src", "src/device_db_manager", "src/log")
 
+target("credential_queue_test")
+    set_kind("binary")
+    set_default(false)
+    add_files("tests/credential_queue_test.cpp", "src/device_auth_limiter.cpp")
+    add_includedirs("src")
+
+target("credential_server_fixture")
+    set_kind("binary")
+    set_default(false)
+    add_deps("server")
+    add_files("tests/credential_server_fixture.cpp")
+    add_includedirs("src", "src/log", "src/device_db_manager")
+
 target("turn_credentials_test")
     set_kind("binary")
     set_default(false)